git.iko.soy/distributed-process
on master help sign in
Add TCP transport implementation.
f62c152Nicolas Wu
network-transport/src/Network/Transport/TCP.hs+97−11
compare with master
unifiedsplit
@@ -1,28 +1,114 @@
1module Network.Transport.TCP
1module Network.Transport.TCP
2 ( mkTransport
2 ( mkTransport
3+ , TCPConfig (..)
3 ) where
4 ) where
4
5
5import Network.Transport
6import Network.Transport
7+
8+import Control.Concurrent (forkIO)
9+import Control.Concurrent.Chan
10+import Control.Concurrent.MVar
11+import Control.Monad (forever, unless)
12+import Data.ByteString.Char8 (ByteString)
13+import Data.IntMap (IntMap)
6import Data.Word
14import Data.Word
15+import Network.Socket
16+import Safe
17+
18+import qualified Data.ByteString.Char8 as BS
19+import qualified Data.IntMap as IntMap
20+import qualified Network.Socket as N
21+import qualified Network.Socket.ByteString as NBS
7
22
8−data Address = Address String
23+type Chans = MVar (Int, IntMap (Chan ByteString))
9−type Port = Word16
24+type ChanId = Int
10
25
11-- | This deals with several different configuration properties:
26-- | This deals with several different configuration properties:
12-- * Buffer size, specified in Hints
27-- * Buffer size, specified in Hints
13-- * LAN/WAN, since we can inspect the addresses
28-- * LAN/WAN, since we can inspect the addresses
29+-- Note that HostName could be an IP address, and ServiceName could be
30+-- a port number
14−data TCPConfig = TCPConfig Hints Port Address Address
31+data TCPConfig = TCPConfig Hints HostName ServiceName
15
32
33+-- | This creates a TCP connection between a server and a number of
34+-- clients. Behind the scenes, the server hostname is passed as the SendAddr
35+-- and when a connection is made, messages sent down the SendEnd go
36+-- via a socket made for the client that connected.
37+-- Messages are all queued using an unbounded Chan.
16mkTransport :: TCPConfig -> IO Transport
38mkTransport :: TCPConfig -> IO Transport
17−mkTransport tcpConfig =
39+mkTransport (TCPConfig _hints host service) = withSocketsDo $ do
18− -- create buffer
40+ channels <- newMVar (0, IntMap.empty)
19− -- allocate listening socket
41+ serverAddrs <- getAddrInfo
20− -- fork thread on that socket
42+ (Just (N.defaultHints { addrFlags = [AI_PASSIVE] }))
43+ Nothing
44+ (Just service)
45+ let serverAddr = case serverAddrs of
46+ [] -> error "mkTransport: getAddrInfo returned []"
47+ as -> head as
48+ sock <- socket (addrFamily serverAddr) Stream defaultProtocol
49+ bindSocket sock (addrAddress serverAddr)
50+ listen sock 5
51+ forkIO $ procConnections channels sock
52+
21 return Transport
53 return Transport
22− { newConnectionWith = undefined
54+ { newConnectionWith = \_ -> do
23− -- create listening mailbox
55+ (chanId, chanMap) <- takeMVar channels
24− -- send back mailbox details
56+ chan <- newChan
57+ putMVar channels (chanId + 1, IntMap.insert chanId chan chanMap)
58+ return (mkSendAddr host service chanId, mkReceiveEnd chan)
25 , newMulticastWith = undefined
59 , newMulticastWith = undefined
26− , deserialize = undefined
60+ , deserialize = \bs ->
61+ case readMay . BS.unpack $ bs of
62+ Nothing -> error "deserialize: cannot parse"
63+ Just (host, service, chanId) -> Just $ mkSendAddr host service chanId
27 }
64 }
28
65
66+ where
67+ mkSendAddr :: HostName -> ServiceName -> ChanId -> SendAddr
68+ mkSendAddr host service chanId = SendAddr
69+ { connectWith = \_ -> mkSendEnd chanId
70+ , serialize = BS.pack . show $ (host, service, chanId)
71+ }
72+
73+ mkSendEnd :: ChanId -> IO SendEnd
74+ mkSendEnd chanId = withSocketsDo $ do
75+ serverAddrs <- getAddrInfo Nothing (Just host) (Just service)
76+ let serverAddr = case serverAddrs of
77+ [] -> error "mkSendEnd: getAddrInfo returned []"
78+ as -> head as
79+ sock <- socket (addrFamily serverAddr) Stream defaultProtocol
80+ setSocketOption sock KeepAlive 1
81+ N.connect sock (addrAddress serverAddr)
82+ NBS.sendAll sock $ BS.pack . show $ chanId
83+ return $ SendEnd
84+ { Network.Transport.send = \bss -> NBS.sendMany sock bss
85+ }
86+
87+ mkReceiveEnd :: Chan ByteString -> ReceiveEnd
88+ mkReceiveEnd chan = ReceiveEnd
89+ { -- for now we will implement this as a Chan
90+ receive = do
91+ bs <- readChan chan
92+ return [bs]
93+ }
94+
95+ procConnections :: Chans -> Socket -> IO ()
96+ procConnections chans sock = forever $ do
97+ (clientSock, _clientAddr) <- accept sock
98+ -- decode the first message to find the correct chanId
99+ bs <- NBS.recv clientSock 4096
100+ case BS.readInt bs of
101+ Nothing -> error "procConnections: cannot parse chanId"
102+ Just (chanId, bs') -> do
103+ -- lookup the channel
104+ (_, chanMap) <- readMVar chans
105+ case IntMap.lookup chanId chanMap of
106+ Nothing -> error "procConnections: cannot find chanId"
107+ Just chan -> forkIO $ do
108+ unless (BS.null bs') $ writeChan chan bs'
109+ procMessages chan clientSock
110+
111+ procMessages :: Chan ByteString -> Socket -> IO ()
112+ procMessages chan sock = forever $ do
113+ bs <- NBS.recv sock 4096
114+ writeChan chan bs