git.iko.soy/distributed-process
on master help sign in
strictified, switched back to cereal in preparation for using conduits
2ce3419Adam C. Foltzer
examples/DemoTCP.hs+8−4
compare with master
unifiedsplit
@@ -4,10 +4,12 @@ import Network.Transport
4import Network.Transport.TCP
4import Network.Transport.TCP
5
5
6import Control.Monad
6import Control.Monad
7+import Data.Char
8+import Data.String
7import Network.Socket (HostName, ServiceName)
9import Network.Socket (HostName, ServiceName)
8import System.Environment
10import System.Environment
9
11
10−import qualified Data.ByteString.Lazy.Char8 as BS
12+import qualified Data.ByteString as BS
11
13
12import Debug.Trace
14import Debug.Trace
13
15
@@ -39,7 +41,8 @@ main = do
39 transport <- mkTransport $ TCPConfig defaultHints host service
41 transport <- mkTransport $ TCPConfig defaultHints host service
40 (clientSourceEnds, masterTargetEnd) <- demoMaster transport sourceAddrFilePath numSlaves
42 (clientSourceEnds, masterTargetEnd) <- demoMaster transport sourceAddrFilePath numSlaves
41 zipWithM_
43 zipWithM_
44+ (\clientSourceEnd clientId ->
42− (\clientSourceEnd clientId -> send clientSourceEnd [BS.pack . show $ clientId])
45+ send clientSourceEnd [BS.pack . map (fromIntegral . ord) . show $ clientId])
43 clientSourceEnds [0 .. numSlaves-1]
46 clientSourceEnds [0 .. numSlaves-1]
44 replicateM_ numSlaves $ do
47 replicateM_ numSlaves $ do
45 [clientMessage] <- receive masterTargetEnd
48 [clientMessage] <- receive masterTargetEnd
@@ -56,9 +59,10 @@ main = do
56 (masterSourceEnd, slaveTargetEnd) <- demoSlave transport sourceAddrFilePath
59 (masterSourceEnd, slaveTargetEnd) <- demoSlave transport sourceAddrFilePath
57 [clientId] <- receive slaveTargetEnd
60 [clientId] <- receive slaveTargetEnd
58 putStrLn $ "slave: received clientId: " ++ show clientId
61 putStrLn $ "slave: received clientId: " ++ show clientId
59− let message = "Connected to slave: " ++ BS.unpack clientId
62+ let message = "Connected to slave: "
63+ ++ (map (chr . fromIntegral) $ BS.unpack clientId)
60
64
61− send masterSourceEnd [BS.pack message]
65+ send masterSourceEnd [BS.pack . map (fromIntegral . ord) $ message]
62 putStrLn "slave: sent message to master"
66 putStrLn "slave: sent message to master"
63
67
64 closeSourceEnd masterSourceEnd
68 closeSourceEnd masterSourceEnd
network-transport/network-transport.cabal+3
@@ -13,8 +13,11 @@ Library
13 Build-Depends: base >= 3 && < 5,
13 Build-Depends: base >= 3 && < 5,
14 binary >= 0.5,
14 binary >= 0.5,
15 bytestring >= 0.9,
15 bytestring >= 0.9,
16+ cereal >= 0.3,
17+ conduit >= 0.2,
16 containers >= 0.4,
18 containers >= 0.4,
17 network >= 2.3,
19 network >= 2.3,
20+ network-conduit >= 0.2,
18 safe >= 0.3
21 safe >= 0.3
19 Exposed-modules: Network.Transport,
22 Exposed-modules: Network.Transport,
20 Network.Transport.MVar,
23 Network.Transport.MVar,
network-transport/src/Network/Transport.hs+1−1
@@ -12,7 +12,7 @@ module Network.Transport
12 , newMulticast
12 , newMulticast
13 ) where
13 ) where
14
14
15−import Data.ByteString.Lazy.Char8 (ByteString)
15+import Data.ByteString (ByteString)
16
16
17-- | The `Hints` and `SourceHints` provide hints to the underlying transport
17-- | The `Hints` and `SourceHints` provide hints to the underlying transport
18-- about the kind of connection that is required. This might include details
18-- about the kind of connection that is required. This might include details
network-transport/src/Network/Transport/MVar.hs+7−6
@@ -6,10 +6,11 @@ module Network.Transport.MVar
6
6
7import Control.Concurrent.MVar
7import Control.Concurrent.MVar
8import Data.IntMap (IntMap)
8import Data.IntMap (IntMap)
9−import Data.ByteString.Lazy.Char8 (ByteString)
9+import Data.ByteString (ByteString)
10+import Data.Serialize
10
11
11import qualified Data.IntMap as IntMap
12import qualified Data.IntMap as IntMap
12−import qualified Data.ByteString.Lazy.Char8 as BS
13+import qualified Data.ByteString as BS
13
14
14import Network.Transport
15import Network.Transport
15
16
@@ -29,15 +30,15 @@ mkTransport = do
29 return (mkSourceAddr channels sourceAddr, mkTargetEnd receiveChan)
30 return (mkSourceAddr channels sourceAddr, mkTargetEnd receiveChan)
30 , newMulticastWith = undefined
31 , newMulticastWith = undefined
31 , deserialize = \bs ->
32 , deserialize = \bs ->
32− case BS.readInt bs of
33− Nothing -> error "dummyBackend.deserializeSourceEnd: cannot parse"
33+ either (error "dummyBackend.deserializeSourceEnd: cannot parse")
34− Just (n,_) -> Just . mkSourceAddr channels $ n
34+ (Just . mkSourceAddr channels)
35+ (decode bs)
35 }
36 }
36 where
37 where
37 mkSourceAddr :: Chans -> Int -> SourceAddr
38 mkSourceAddr :: Chans -> Int -> SourceAddr
38 mkSourceAddr channels addr = SourceAddr
39 mkSourceAddr channels addr = SourceAddr
39 { connectWith = \_ -> mkSourceEnd channels addr
40 { connectWith = \_ -> mkSourceEnd channels addr
40− , serialize = BS.pack (show addr)
41+ , serialize = encode addr
41 }
42 }
42
43
43 mkSourceEnd :: Chans -> Int -> IO SourceEnd
44 mkSourceEnd :: Chans -> Int -> IO SourceEnd
network-transport/src/Network/Transport/TCP.hs+46−26
@@ -5,26 +5,41 @@ module Network.Transport.TCP
5
5
6import Network.Transport
6import Network.Transport
7
7
8+import Control.Applicative
9+
8import Control.Concurrent (forkIO, ThreadId, killThread)
10import Control.Concurrent (forkIO, ThreadId, killThread)
9import Control.Concurrent.Chan
11import Control.Concurrent.Chan
10import Control.Concurrent.MVar
12import Control.Concurrent.MVar
11import Control.Monad (forever, forM_)
13import Control.Monad (forever, forM_)
14+
12−import Data.ByteString.Lazy.Char8 (ByteString)
15+import Data.ByteString (ByteString)
16+import qualified Data.ByteString as BS
17+
18+import Data.Conduit
19+import qualified Data.Conduit.Binary as CB
20+import Data.Conduit.Network
21+
13import Data.IntMap (IntMap)
22import Data.IntMap (IntMap)
23+import qualified Data.IntMap as IntMap
24+
14import Data.Int
25import Data.Int
26+import Data.Serialize
15import Data.Word
27import Data.Word
28+
16import Network.Socket
29import Network.Socket
17 ( AddrInfoFlag (AI_PASSIVE), HostName, ServiceName, Socket
30 ( AddrInfoFlag (AI_PASSIVE), HostName, ServiceName, Socket
18 , SocketType (Stream), SocketOption (ReuseAddr)
31 , SocketType (Stream), SocketOption (ReuseAddr)
19 , accept, addrAddress, addrFlags, addrFamily, bindSocket, defaultProtocol
32 , accept, addrAddress, addrFlags, addrFamily, bindSocket, defaultProtocol
20 , getAddrInfo, listen, setSocketOption, socket, sClose, withSocketsDo )
33 , getAddrInfo, listen, setSocketOption, socket, sClose, withSocketsDo )
34+import qualified Network.Socket as N
35+import qualified Network.Socket.ByteString as NBS
36+
37+import Text.Printf
38+
21import Safe
39import Safe
22
40
23import qualified Data.Binary as B
41import qualified Data.Binary as B
24−import qualified Data.ByteString.Lazy.Char8 as BS
42+
25−import qualified Data.IntMap as IntMap
26−import qualified Network.Socket as N
27−import qualified Network.Socket.ByteString.Lazy as NBS
28
43
29type ChanId = Int
44type ChanId = Int
30type Chans = MVar (ChanId, IntMap (Chan [ByteString], [(ThreadId, Socket)]))
45type Chans = MVar (ChanId, IntMap (Chan [ByteString], [(ThreadId, Socket)]))
@@ -65,7 +80,8 @@ mkTransport (TCPConfig _hints host service) = withSocketsDo $ do
65 return (mkSourceAddr host service chanId, mkTargetEnd chans chanId chan)
80 return (mkSourceAddr host service chanId, mkTargetEnd chans chanId chan)
66 , newMulticastWith = error "newMulticastWith: not defined"
81 , newMulticastWith = error "newMulticastWith: not defined"
67 , deserialize = {-# SCC "deserialize" #-} \bs ->
82 , deserialize = {-# SCC "deserialize" #-} \bs ->
68− let (host, service, chanId) = B.decode bs in
83+ let (host, service, chanId) =
84+ either (error $ printf "deserialize: %s") id $ decode bs in
69 Just $ mkSourceAddr host service chanId
85 Just $ mkSourceAddr host service chanId
70 , closeTransport = {-# SCC "closeTransport" #-} do
86 , closeTransport = {-# SCC "closeTransport" #-} do
71 -- Kill the transport channel process
87 -- Kill the transport channel process
@@ -82,28 +98,30 @@ mkTransport (TCPConfig _hints host service) = withSocketsDo $ do
82mkSourceAddr :: HostName -> ServiceName -> ChanId -> SourceAddr
98mkSourceAddr :: HostName -> ServiceName -> ChanId -> SourceAddr
83mkSourceAddr host service chanId = SourceAddr
99mkSourceAddr host service chanId = SourceAddr
84 { connectWith = {-# SCC "connectWith" #-} \_ -> mkSourceEnd host service chanId
100 { connectWith = {-# SCC "connectWith" #-} \_ -> mkSourceEnd host service chanId
85− , serialize = {-# SCC "serialize" #-} B.encode (host, service, chanId)
101+ , serialize = {-# SCC "serialize" #-} encode (host, service, chanId)
86 }
102 }
87
103
88mkSourceEnd :: HostName -> ServiceName -> ChanId -> IO SourceEnd
104mkSourceEnd :: HostName -> ServiceName -> ChanId -> IO SourceEnd
89mkSourceEnd host service chanId = withSocketsDo $ do
105mkSourceEnd host service chanId = withSocketsDo $ do
106+ let err = error $ printf "mkSourceEnd: %s"
90 serverAddrs <- getAddrInfo Nothing (Just host) (Just service)
107 serverAddrs <- getAddrInfo Nothing (Just host) (Just service)
91 let serverAddr = case serverAddrs of
108 let serverAddr = case serverAddrs of
92− [] -> error "mkSourceEnd: getAddrInfo returned []"
109+ [] -> err "getAddrInfo returned []"
93 as -> head as
110 as -> head as
94 sock <- socket (addrFamily serverAddr) Stream defaultProtocol
111 sock <- socket (addrFamily serverAddr) Stream defaultProtocol
95 setSocketOption sock ReuseAddr 1
112 setSocketOption sock ReuseAddr 1
96 N.connect sock (addrAddress serverAddr)
113 N.connect sock (addrAddress serverAddr)
114+-- let sink = sinkSocket sock :: Sink ByteString IO ()
97− NBS.sendAll sock $ B.encode (fromIntegral chanId :: Int64)
115+ NBS.sendAll sock $ encode (fromIntegral chanId :: Int64)
98 return SourceEnd
116 return SourceEnd
99 { send = {-# SCC "send" #-} \bss -> do
117 { send = {-# SCC "send" #-} \bss -> do
100 let size = fromIntegral (sum . map BS.length $ bss) :: Int64
118 let size = fromIntegral (sum . map BS.length $ bss) :: Int64
101 if size < 255
119 if size < 255
102 then
120 then
103− NBS.sendAll sock (B.encode (fromIntegral size :: Word8))
121+ NBS.sendAll sock (encode (fromIntegral size :: Word8))
104 else do
122 else do
105− NBS.sendAll sock (B.encode (255 :: Word8))
123+ NBS.sendAll sock (encode (255 :: Word8))
106− NBS.sendAll sock (B.encode size)
124+ NBS.sendAll sock (encode size)
107 mapM_ (NBS.sendAll sock) bss
125 mapM_ (NBS.sendAll sock) bss
108 , closeSourceEnd = {-# SCC "closeSourceEnd" #-} sClose sock
126 , closeSourceEnd = {-# SCC "closeSourceEnd" #-} sClose sock
109 }
127 }
@@ -128,19 +146,20 @@ mkTargetEnd chans chanId chan = TargetEnd
128-- for some reason, an error is raised.
146-- for some reason, an error is raised.
129procConnections :: Chans -> Socket -> IO ()
147procConnections :: Chans -> Socket -> IO ()
130procConnections chans sock = forever $ do
148procConnections chans sock = forever $ do
149+ let err = error $ printf "procConnections: %s"
131 (clientSock, _clientAddr) <- accept sock
150 (clientSock, _clientAddr) <- accept sock
132 -- decode the first message to find the correct chanId
151 -- decode the first message to find the correct chanId
133 bss <- recvExact clientSock 8
152 bss <- recvExact clientSock 8
134 case bss of
153 case bss of
135− [] -> error "procConnections: inbound chanId aborted"
154+ [] -> err "inbound chanId aborted"
136 bss -> do
155 bss -> do
137 let bs = BS.concat bss
156 let bs = BS.concat bss
138− let chanId = fromIntegral (B.decode bs :: Int64)
157+ let chanId = either err fromIntegral (decode bs :: Either String Int64)
139 (chanId', chanMap) <- takeMVar chans
158 (chanId', chanMap) <- takeMVar chans
140 case IntMap.lookup chanId chanMap of
159 case IntMap.lookup chanId chanMap of
141 Nothing -> do
160 Nothing -> do
142 putMVar chans (chanId', chanMap)
161 putMVar chans (chanId', chanMap)
143− error "procConnections: cannot find chanId"
162+ err "cannot find chanId"
144 Just (chan, socks) -> do
163 Just (chan, socks) -> do
145 threadId <- forkIO $ procMessages chans chanId chan clientSock
164 threadId <- forkIO $ procMessages chans chanId chan clientSock
146 let chanMap' = IntMap.insert chanId (chan, (threadId, clientSock):socks) chanMap
165 let chanMap' = IntMap.insert chanId (chan, (threadId, clientSock):socks) chanMap
@@ -159,29 +178,30 @@ procMessages chans chanId chan sock = do
159 case sizeBSs of
178 case sizeBSs of
160 [] -> closeSocket
179 [] -> closeSocket
161 _ -> do
180 _ -> do
162− let size = fromIntegral (B.decode . BS.concat $ sizeBSs :: Word8)
181+ case decode . BS.concat $ sizeBSs :: Either String Word8 of
163− if size == 255
182+ Right 255 -> do
164− then do
165 sizeBSs' <- recvExact sock 8
183 sizeBSs' <- recvExact sock 8
166 case sizeBSs' of
184 case sizeBSs' of
167 [] -> closeSocket
185 [] -> closeSocket
168− _ -> procMessage (B.decode . BS.concat $ sizeBSs' :: Int64)
186+ _ -> procMessage (decode . BS.concat $ sizeBSs')
169− else procMessage size
187+ esize -> procMessage (fromIntegral <$> esize)
170 where
188 where
189+ err = error $ printf "procMessages: %s"
171 closeSocket :: IO ()
190 closeSocket :: IO ()
172 closeSocket = do
191 closeSocket = do
173 (chanId', chanMap) <- takeMVar chans
192 (chanId', chanMap) <- takeMVar chans
174 case IntMap.lookup chanId chanMap of
193 case IntMap.lookup chanId chanMap of
175 Nothing -> do
194 Nothing -> do
176 putMVar chans (chanId', chanMap)
195 putMVar chans (chanId', chanMap)
177− error "procMessages: chanId not found."
196+ err "chanId not found."
178 Just (chan, socks) -> do
197 Just (chan, socks) -> do
179 let socks' = filter ((/= sock) . snd) socks
198 let socks' = filter ((/= sock) . snd) socks
180 let chanMap' = IntMap.insert chanId (chan, socks') chanMap
199 let chanMap' = IntMap.insert chanId (chan, socks') chanMap
181 putMVar chans (chanId', chanMap')
200 putMVar chans (chanId', chanMap')
182 sClose sock
201 sClose sock
183− procMessage :: Int64 -> IO ()
202+ procMessage :: Either String Int64 -> IO ()
184− procMessage size = do
203+ procMessage (Left msg) = err msg
204+ procMessage (Right size) = do
185 bss <- recvExact sock size
205 bss <- recvExact sock size
186 case bss of
206 case bss of
187 [] -> closeSocket
207 [] -> closeSocket
@@ -202,13 +222,13 @@ procMessages chans chanId chan sock = do
202-- indicates success. This hasn't been implemented, since a Transport `receive`
222-- indicates success. This hasn't been implemented, since a Transport `receive`
203-- represents an atomic receipt of a message.
223-- represents an atomic receipt of a message.
204recvExact :: Socket -> Int64 -> IO [ByteString]
224recvExact :: Socket -> Int64 -> IO [ByteString]
205−recvExact sock n = go [] sock n
225+recvExact sock n = go [] sock (fromIntegral n)
206 where
226 where
207− go :: [ByteString] -> Socket -> Int64 -> IO [ByteString]
227+ go :: [ByteString] -> Socket -> Int -> IO [ByteString]
208 go bss _ 0 = return (reverse bss)
228 go bss _ 0 = return (reverse bss)
209 go bss sock n = do
229 go bss sock n = do
210 bs <- NBS.recv sock n
230 bs <- NBS.recv sock n
211 if BS.null bs
231 if BS.null bs
212 then return []
232 then return []
213− else go (bs:bss) sock (n - BS.length bs)
233+ else go (bs:bss) sock (n - (fromIntegral $ BS.length bs))
214
234