git.iko.soy/distributed-process
on master help sign in
Merge remote-tracking branch 'origin/transport-close' into tcp-benching

Conflicts: network-transport/src/Network/Transport/TCP.hs

commit624ed4005ae6c034a60dcc43ac0e45e0197a3a79 authorNicolas Wu <nick@well-typed.com> dateMon 23 Jan 2012, 15:06 (15 years ago) parentd231bff929e832
4 files changed+37−14
compare with master
unifiedsplit
examples/DemoTCP.hs+6 examples/DemoTransport.hs+2 network-transport/src/Network/Transport.hs+8−2 network-transport/src/Network/Transport/TCP.hs+21−12
examples/DemoTCP.hs+6
@@ -45,6 +45,9 @@ main = do
4545 [clientMessage] <- receive masterTargetEnd
4646 print clientMessage
4747
48+ closeTargetEnd masterTargetEnd
49+ putStrLn "master: close connections"
50+
4851 -- Slave:
4952 -- Each slave provides its own host and service, as well as details of
5053 -- the master it wishes to connect to.
@@ -58,6 +61,9 @@ main = do
5861 send masterSourceEnd [BS.pack message]
5962 putStrLn "slave: sent message to master"
6063
64+ closeSourceEnd masterSourceEnd
65+ putStrLn "slave: close connection"
66+
6167 _ -> error "Unexpected arguments"
6268
6369demoMaster :: Transport -- ^ Transport
examples/DemoTransport.hs+2
@@ -30,6 +30,8 @@ demo0 = do
3030 mapM_ (\n -> send sourceEnd [BS.pack ("hello " ++ show n)]) [1 .. 10]
3131 threadDelay 100000
3232
33+ closeAll targetEnd
34+
3335-- | Check endpoint serialization and deserialization.
3436demo1 :: IO ()
3537demo1 = do
network-transport/src/Network/Transport.hs+8−2
@@ -68,14 +68,20 @@ defaultSourceHints = SourceHints
6868
6969-- | A `SourceEnd` provides a `send` function that allows vectored messages
7070-- to be sent to the corresponding `TargetEnd`.
71−newtype SourceEnd = SourceEnd
71+-- The `close` function closes the connection between this source and the target
72+-- end. Connections between other sources the target end remain unaffected
73+data SourceEnd = SourceEnd
7274 { send :: [ByteString] -> IO ()
75+ , closeSourceEnd :: IO ()
7376 }
7477
7578-- | A `TargetEnd` provides a `receive` function that allows messages
7679-- to be received from the corresponding `SourceEnd`s.
77−newtype TargetEnd = TargetEnd
80+-- The `closeAll` function closes all connections to this target,
81+-- and all new connections will be refused.
82+data TargetEnd = TargetEnd
7883 { receive :: IO [ByteString]
84+ , closeTargetEnd :: IO ()
7985 }
8086
8187newtype MulticastSourceEnd = MulticastSourceEnd
network-transport/src/Network/Transport/TCP.hs+21−12
@@ -8,11 +8,15 @@ import Network.Transport
88import Control.Concurrent (forkIO)
99import Control.Concurrent.Chan
1010import Control.Concurrent.MVar
11−import Control.Monad (forever, unless)
11+import Control.Monad (forever)
1212import Data.ByteString.Lazy.Char8 (ByteString)
1313import Data.IntMap (IntMap)
1414import Data.Int
1515import 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 )
1620import Safe
1721
1822import qualified Data.Binary as B
@@ -21,8 +25,8 @@ import qualified Data.IntMap as IntMap
2125import qualified Network.Socket as N
2226import qualified Network.Socket.ByteString.Lazy as NBS
2327
24−type ChanId = Int
25−type Chans = MVar (ChanId, IntMap (Chan ByteString))
28+type ChanId = Int
29+type Chans = MVar (ChanId, IntMap (Chan ByteString, MVar [Socket]))
2630
2731-- | This deals with several different configuration properties:
2832-- * Buffer size, specified in Hints
@@ -49,14 +53,15 @@ mkTransport (TCPConfig _hints host service) = withSocketsDo $ do
4953 sock <- socket (addrFamily serverAddr) Stream defaultProtocol
5054 bindSocket sock (addrAddress serverAddr)
5155 listen sock 5
52− forkIO $ procConnections channels sock
56+ _ <- forkIO $ procConnections channels sock
5357
5458 return Transport
5559 { newConnectionWith = \_ -> do
5660 (chanId, chanMap) <- takeMVar channels
5761 chan <- newChan
58− putMVar channels (chanId + 1, IntMap.insert chanId chan chanMap)
59− return (mkSourceAddr host service chanId, mkTargetEnd chan)
62+ socks <- newMVar [sock]
63+ putMVar channels (chanId + 1, IntMap.insert chanId (chan, socks) chanMap)
64+ return (mkSourceAddr host service chanId, mkTargetEnd chan socks)
6065 , newMulticastWith = error "newMulticastWith: not defined"
6166 , deserialize = \bs ->
6267 let (host, service, chanId) = B.decode bs in
@@ -81,18 +86,20 @@ mkTransport (TCPConfig _hints host service) = withSocketsDo $ do
8186 N.connect sock (addrAddress serverAddr)
8287 NBS.sendAll sock $ B.encode (fromIntegral chanId :: Int64)
8388 return SourceEnd
84− { Network.Transport.send = \bs -> do
85− let size = fromIntegral (sum . map BS.length $ bs) :: Int64
89+ { send = \bss -> do
90+ let size = fromIntegral (sum . map BS.length $ bss) :: Int64
8691 NBS.sendAll sock (B.encode size)
87− mapM_ (NBS.sendAll sock) bs
92+ mapM_ (NBS.sendAll sock) bss
93+ , closeSourceEnd = sClose sock
8894 }
8995
90− mkTargetEnd :: Chan ByteString -> TargetEnd
91− mkTargetEnd chan = TargetEnd
96+ mkTargetEnd :: Chan ByteString -> MVar [Socket] -> TargetEnd
97+ mkTargetEnd chan socks = TargetEnd
9298 { -- for now we will implement this as a Chan
9399 receive = do
94100 bs <- readChan chan
95101 return [bs]
102+ , closeTargetEnd = takeMVar socks >>= mapM_ sClose
96103 }
97104
98105 procConnections :: Chans -> Socket -> IO ()
@@ -104,7 +111,9 @@ mkTransport (TCPConfig _hints host service) = withSocketsDo $ do
104111 (_, chanMap) <- readMVar chans
105112 case IntMap.lookup chanId chanMap of
106113 Nothing -> error "procConnections: cannot find chanId"
107− Just chan -> forkIO $ procMessages chan clientSock
114+ Just (chan, socks) -> do
115+ modifyMVar_ socks (return . (clientSock :))
116+ forkIO $ procMessages chan clientSock
108117
109118 -- This function first extracts a header of type Int64, which determines
110119 -- the size of the ByteString that follows. The ByteString is then