git.iko.soy/distributed-process
on master help sign in
Merge remote-tracking branch 'origin/transport-close' into tcp-benching
624ed40Nicolas Wu
examples/DemoTCP.hs+6
compare with master
unifiedsplit
@@ -45,6 +45,9 @@ main = do
45 [clientMessage] <- receive masterTargetEnd
45 [clientMessage] <- receive masterTargetEnd
46 print clientMessage
46 print clientMessage
47
47
48+ closeTargetEnd masterTargetEnd
49+ putStrLn "master: close connections"
50+
48 -- Slave:
51 -- Slave:
49 -- Each slave provides its own host and service, as well as details of
52 -- Each slave provides its own host and service, as well as details of
50 -- the master it wishes to connect to.
53 -- the master it wishes to connect to.
@@ -58,6 +61,9 @@ main = do
58 send masterSourceEnd [BS.pack message]
61 send masterSourceEnd [BS.pack message]
59 putStrLn "slave: sent message to master"
62 putStrLn "slave: sent message to master"
60
63
64+ closeSourceEnd masterSourceEnd
65+ putStrLn "slave: close connection"
66+
61 _ -> error "Unexpected arguments"
67 _ -> error "Unexpected arguments"
62
68
63demoMaster :: Transport -- ^ Transport
69demoMaster :: Transport -- ^ Transport
examples/DemoTransport.hs+2
@@ -30,6 +30,8 @@ demo0 = do
30 mapM_ (\n -> send sourceEnd [BS.pack ("hello " ++ show n)]) [1 .. 10]
30 mapM_ (\n -> send sourceEnd [BS.pack ("hello " ++ show n)]) [1 .. 10]
31 threadDelay 100000
31 threadDelay 100000
32
32
33+ closeAll targetEnd
34+
33-- | Check endpoint serialization and deserialization.
35-- | Check endpoint serialization and deserialization.
34demo1 :: IO ()
36demo1 :: IO ()
35demo1 = do
37demo1 = do
network-transport/src/Network/Transport.hs+8−2
@@ -68,14 +68,20 @@ defaultSourceHints = SourceHints
68
68
69-- | A `SourceEnd` provides a `send` function that allows vectored messages
69-- | A `SourceEnd` provides a `send` function that allows vectored messages
70-- to be sent to the corresponding `TargetEnd`.
70-- to be sent to the corresponding `TargetEnd`.
71+-- The `close` function closes the connection between this source and the target
72+-- end. Connections between other sources the target end remain unaffected
71−newtype SourceEnd = SourceEnd
73+data SourceEnd = SourceEnd
72 { send :: [ByteString] -> IO ()
74 { send :: [ByteString] -> IO ()
75+ , closeSourceEnd :: IO ()
73 }
76 }
74
77
75-- | A `TargetEnd` provides a `receive` function that allows messages
78-- | A `TargetEnd` provides a `receive` function that allows messages
76-- to be received from the corresponding `SourceEnd`s.
79-- to be received from the corresponding `SourceEnd`s.
80+-- The `closeAll` function closes all connections to this target,
81+-- and all new connections will be refused.
77−newtype TargetEnd = TargetEnd
82+data TargetEnd = TargetEnd
78 { receive :: IO [ByteString]
83 { receive :: IO [ByteString]
84+ , closeTargetEnd :: IO ()
79 }
85 }
80
86
81newtype MulticastSourceEnd = MulticastSourceEnd
87newtype MulticastSourceEnd = MulticastSourceEnd
network-transport/src/Network/Transport/TCP.hs+21−12
@@ -8,11 +8,15 @@ import Network.Transport
8import Control.Concurrent (forkIO)
8import Control.Concurrent (forkIO)
9import Control.Concurrent.Chan
9import Control.Concurrent.Chan
10import Control.Concurrent.MVar
10import Control.Concurrent.MVar
11−import Control.Monad (forever, unless)
11+import Control.Monad (forever)
12import Data.ByteString.Lazy.Char8 (ByteString)
12import Data.ByteString.Lazy.Char8 (ByteString)
13import Data.IntMap (IntMap)
13import Data.IntMap (IntMap)
14import Data.Int
14import Data.Int
15import Network.Socket
15import Network.Socket
16+ ( AddrInfoFlag (AI_PASSIVE), HostName, ServiceName, Socket
17+ , SocketType (Stream), SocketOption (ReuseAddr)
18+ , accept, addrAddress, addrFlags, addrFamily, bindSocket, defaultProtocol
19+ , getAddrInfo, listen, setSocketOption, socket, sClose, withSocketsDo )
16import Safe
20import Safe
17
21
18import qualified Data.Binary as B
22import qualified Data.Binary as B
@@ -21,8 +25,8 @@ import qualified Data.IntMap as IntMap
21import qualified Network.Socket as N
25import qualified Network.Socket as N
22import qualified Network.Socket.ByteString.Lazy as NBS
26import qualified Network.Socket.ByteString.Lazy as NBS
23
27
24−type ChanId = Int
28+type ChanId = Int
25−type Chans = MVar (ChanId, IntMap (Chan ByteString))
29+type Chans = MVar (ChanId, IntMap (Chan ByteString, MVar [Socket]))
26
30
27-- | This deals with several different configuration properties:
31-- | This deals with several different configuration properties:
28-- * Buffer size, specified in Hints
32-- * Buffer size, specified in Hints
@@ -49,14 +53,15 @@ mkTransport (TCPConfig _hints host service) = withSocketsDo $ do
49 sock <- socket (addrFamily serverAddr) Stream defaultProtocol
53 sock <- socket (addrFamily serverAddr) Stream defaultProtocol
50 bindSocket sock (addrAddress serverAddr)
54 bindSocket sock (addrAddress serverAddr)
51 listen sock 5
55 listen sock 5
52− forkIO $ procConnections channels sock
56+ _ <- forkIO $ procConnections channels sock
53
57
54 return Transport
58 return Transport
55 { newConnectionWith = \_ -> do
59 { newConnectionWith = \_ -> do
56 (chanId, chanMap) <- takeMVar channels
60 (chanId, chanMap) <- takeMVar channels
57 chan <- newChan
61 chan <- newChan
62+ socks <- newMVar [sock]
58− putMVar channels (chanId + 1, IntMap.insert chanId chan chanMap)
63+ putMVar channels (chanId + 1, IntMap.insert chanId (chan, socks) chanMap)
59− return (mkSourceAddr host service chanId, mkTargetEnd chan)
64+ return (mkSourceAddr host service chanId, mkTargetEnd chan socks)
60 , newMulticastWith = error "newMulticastWith: not defined"
65 , newMulticastWith = error "newMulticastWith: not defined"
61 , deserialize = \bs ->
66 , deserialize = \bs ->
62 let (host, service, chanId) = B.decode bs in
67 let (host, service, chanId) = B.decode bs in
@@ -81,18 +86,20 @@ mkTransport (TCPConfig _hints host service) = withSocketsDo $ do
81 N.connect sock (addrAddress serverAddr)
86 N.connect sock (addrAddress serverAddr)
82 NBS.sendAll sock $ B.encode (fromIntegral chanId :: Int64)
87 NBS.sendAll sock $ B.encode (fromIntegral chanId :: Int64)
83 return SourceEnd
88 return SourceEnd
84− { Network.Transport.send = \bs -> do
89+ { send = \bss -> do
85− let size = fromIntegral (sum . map BS.length $ bs) :: Int64
90+ let size = fromIntegral (sum . map BS.length $ bss) :: Int64
86 NBS.sendAll sock (B.encode size)
91 NBS.sendAll sock (B.encode size)
87− mapM_ (NBS.sendAll sock) bs
92+ mapM_ (NBS.sendAll sock) bss
93+ , closeSourceEnd = sClose sock
88 }
94 }
89
95
90− mkTargetEnd :: Chan ByteString -> TargetEnd
96+ mkTargetEnd :: Chan ByteString -> MVar [Socket] -> TargetEnd
91− mkTargetEnd chan = TargetEnd
97+ mkTargetEnd chan socks = TargetEnd
92 { -- for now we will implement this as a Chan
98 { -- for now we will implement this as a Chan
93 receive = do
99 receive = do
94 bs <- readChan chan
100 bs <- readChan chan
95 return [bs]
101 return [bs]
102+ , closeTargetEnd = takeMVar socks >>= mapM_ sClose
96 }
103 }
97
104
98 procConnections :: Chans -> Socket -> IO ()
105 procConnections :: Chans -> Socket -> IO ()
@@ -104,7 +111,9 @@ mkTransport (TCPConfig _hints host service) = withSocketsDo $ do
104 (_, chanMap) <- readMVar chans
111 (_, chanMap) <- readMVar chans
105 case IntMap.lookup chanId chanMap of
112 case IntMap.lookup chanId chanMap of
106 Nothing -> error "procConnections: cannot find chanId"
113 Nothing -> error "procConnections: cannot find chanId"
114+ Just (chan, socks) -> do
115+ modifyMVar_ socks (return . (clientSock :))
107− Just chan -> forkIO $ procMessages chan clientSock
116+ forkIO $ procMessages chan clientSock
108
117
109 -- This function first extracts a header of type Int64, which determines
118 -- This function first extracts a header of type Int64, which determines
110 -- the size of the ByteString that follows. The ByteString is then
119 -- the size of the ByteString that follows. The ByteString is then