git.iko.soy/distributed-process
on master help sign in
Add connection closing.
commite17b03ea7c1bd138be36c6db49985f3dd0ccbbf9 authorNicolas Wu <nick@well-typed.com> dateThu 12 Jan 2012, 16:09 (15 years ago) parent817179f
4 files changed+31−12
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+15−10
examples/DemoTCP.hs+6
@@ -45,6 +45,9 @@ main = do
4545 [clientMessage] <- receive masterTargetEnd
4646 print clientMessage
4747
48+ closeAll 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+ close 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+ , close :: 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+ , closeAll :: IO ()
7985 }
8086
8187newtype MulticastSourceEnd = MulticastSourceEnd
network-transport/src/Network/Transport/TCP.hs+15−10
@@ -15,7 +15,7 @@ import Network.Socket
1515 ( AddrInfoFlag (AI_PASSIVE), HostName, ServiceName, Socket
1616 , SocketType (Stream), SocketOption (ReuseAddr)
1717 , accept, addrAddress, addrFlags, addrFamily, bindSocket, defaultProtocol
18− , getAddrInfo, listen, setSocketOption, socket, withSocketsDo )
18+ , getAddrInfo, listen, setSocketOption, socket, sClose, withSocketsDo )
1919import Safe
2020
2121import qualified Data.ByteString.Char8 as BS
@@ -23,7 +23,7 @@ import qualified Data.IntMap as IntMap
2323import qualified Network.Socket as N
2424import qualified Network.Socket.ByteString as NBS
2525
26−type Chans = MVar (Int, IntMap (Chan ByteString))
26+type Chans = MVar (Int, IntMap (Chan ByteString, MVar [Socket]))
2727type ChanId = Int
2828
2929-- | This deals with several different configuration properties:
@@ -51,14 +51,15 @@ mkTransport (TCPConfig _hints host service) = withSocketsDo $ do
5151 sock <- socket (addrFamily serverAddr) Stream defaultProtocol
5252 bindSocket sock (addrAddress serverAddr)
5353 listen sock 5
54− forkIO $ procConnections channels sock
54+ _ <- forkIO $ procConnections channels sock
5555
5656 return Transport
5757 { newConnectionWith = \_ -> do
5858 (chanId, chanMap) <- takeMVar channels
5959 chan <- newChan
60− putMVar channels (chanId + 1, IntMap.insert chanId chan chanMap)
61− return (mkSourceAddr host service chanId, mkTargetEnd chan)
60+ socks <- newMVar [sock]
61+ putMVar channels (chanId + 1, IntMap.insert chanId (chan, socks) chanMap)
62+ return (mkSourceAddr host service chanId, mkTargetEnd chan socks)
6263 , newMulticastWith = undefined
6364 , deserialize = \bs ->
6465 case readMay . BS.unpack $ bs of
@@ -85,14 +86,16 @@ mkTransport (TCPConfig _hints host service) = withSocketsDo $ do
8586 NBS.sendAll sock $ BS.pack . show $ chanId
8687 return $ SourceEnd
8788 { send = \bss -> NBS.sendMany sock bss
89+ , close = sClose sock
8890 }
8991
90− mkTargetEnd :: Chan ByteString -> TargetEnd
91− mkTargetEnd chan = TargetEnd
92+ mkTargetEnd :: Chan ByteString -> MVar [Socket] -> TargetEnd
93+ mkTargetEnd chan socks = TargetEnd
9294 { -- for now we will implement this as a Chan
9395 receive = do
9496 bs <- readChan chan
9597 return [bs]
98+ , closeAll = takeMVar socks >>= mapM_ sClose
9699 }
97100
98101 procConnections :: Chans -> Socket -> IO ()
@@ -107,9 +110,11 @@ mkTransport (TCPConfig _hints host service) = withSocketsDo $ do
107110 (_, chanMap) <- readMVar chans
108111 case IntMap.lookup chanId chanMap of
109112 Nothing -> error "procConnections: cannot find chanId"
110− Just chan -> forkIO $ do
111− unless (BS.null bs') $ writeChan chan bs'
112− procMessages chan clientSock
113+ Just (chan, socks) -> do
114+ modifyMVar_ socks (return . (clientSock :))
115+ forkIO $ do
116+ unless (BS.null bs') $ writeChan chan bs'
117+ procMessages chan clientSock
113118
114119 procMessages :: Chan ByteString -> Socket -> IO ()
115120 procMessages chan sock = forever $ do