git.iko.soy/distributed-process
on master help sign in
Add connection closing.
e17b03eNicolas 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+ closeAll 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+ close 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+ , close :: 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+ , closeAll :: IO ()
79 }
85 }
80
86
81newtype MulticastSourceEnd = MulticastSourceEnd
87newtype MulticastSourceEnd = MulticastSourceEnd
network-transport/src/Network/Transport/TCP.hs+15−10
@@ -15,7 +15,7 @@ import Network.Socket
15 ( AddrInfoFlag (AI_PASSIVE), HostName, ServiceName, Socket
15 ( AddrInfoFlag (AI_PASSIVE), HostName, ServiceName, Socket
16 , SocketType (Stream), SocketOption (ReuseAddr)
16 , SocketType (Stream), SocketOption (ReuseAddr)
17 , accept, addrAddress, addrFlags, addrFamily, bindSocket, defaultProtocol
17 , accept, addrAddress, addrFlags, addrFamily, bindSocket, defaultProtocol
18− , getAddrInfo, listen, setSocketOption, socket, withSocketsDo )
18+ , getAddrInfo, listen, setSocketOption, socket, sClose, withSocketsDo )
19import Safe
19import Safe
20
20
21import qualified Data.ByteString.Char8 as BS
21import qualified Data.ByteString.Char8 as BS
@@ -23,7 +23,7 @@ import qualified Data.IntMap as IntMap
23import qualified Network.Socket as N
23import qualified Network.Socket as N
24import qualified Network.Socket.ByteString as NBS
24import qualified Network.Socket.ByteString as NBS
25
25
26−type Chans = MVar (Int, IntMap (Chan ByteString))
26+type Chans = MVar (Int, IntMap (Chan ByteString, MVar [Socket]))
27type ChanId = Int
27type ChanId = Int
28
28
29-- | This deals with several different configuration properties:
29-- | This deals with several different configuration properties:
@@ -51,14 +51,15 @@ mkTransport (TCPConfig _hints host service) = withSocketsDo $ do
51 sock <- socket (addrFamily serverAddr) Stream defaultProtocol
51 sock <- socket (addrFamily serverAddr) Stream defaultProtocol
52 bindSocket sock (addrAddress serverAddr)
52 bindSocket sock (addrAddress serverAddr)
53 listen sock 5
53 listen sock 5
54− forkIO $ procConnections channels sock
54+ _ <- forkIO $ procConnections channels sock
55
55
56 return Transport
56 return Transport
57 { newConnectionWith = \_ -> do
57 { newConnectionWith = \_ -> do
58 (chanId, chanMap) <- takeMVar channels
58 (chanId, chanMap) <- takeMVar channels
59 chan <- newChan
59 chan <- newChan
60+ socks <- newMVar [sock]
60− putMVar channels (chanId + 1, IntMap.insert chanId chan chanMap)
61+ putMVar channels (chanId + 1, IntMap.insert chanId (chan, socks) chanMap)
61− return (mkSourceAddr host service chanId, mkTargetEnd chan)
62+ return (mkSourceAddr host service chanId, mkTargetEnd chan socks)
62 , newMulticastWith = undefined
63 , newMulticastWith = undefined
63 , deserialize = \bs ->
64 , deserialize = \bs ->
64 case readMay . BS.unpack $ bs of
65 case readMay . BS.unpack $ bs of
@@ -85,14 +86,16 @@ mkTransport (TCPConfig _hints host service) = withSocketsDo $ do
85 NBS.sendAll sock $ BS.pack . show $ chanId
86 NBS.sendAll sock $ BS.pack . show $ chanId
86 return $ SourceEnd
87 return $ SourceEnd
87 { send = \bss -> NBS.sendMany sock bss
88 { send = \bss -> NBS.sendMany sock bss
89+ , close = sClose sock
88 }
90 }
89
91
90− mkTargetEnd :: Chan ByteString -> TargetEnd
92+ mkTargetEnd :: Chan ByteString -> MVar [Socket] -> TargetEnd
91− mkTargetEnd chan = TargetEnd
93+ mkTargetEnd chan socks = TargetEnd
92 { -- for now we will implement this as a Chan
94 { -- for now we will implement this as a Chan
93 receive = do
95 receive = do
94 bs <- readChan chan
96 bs <- readChan chan
95 return [bs]
97 return [bs]
98+ , closeAll = takeMVar socks >>= mapM_ sClose
96 }
99 }
97
100
98 procConnections :: Chans -> Socket -> IO ()
101 procConnections :: Chans -> Socket -> IO ()
@@ -107,9 +110,11 @@ mkTransport (TCPConfig _hints host service) = withSocketsDo $ do
107 (_, chanMap) <- readMVar chans
110 (_, chanMap) <- readMVar chans
108 case IntMap.lookup chanId chanMap of
111 case IntMap.lookup chanId chanMap of
109 Nothing -> error "procConnections: cannot find chanId"
112 Nothing -> error "procConnections: cannot find chanId"
113+ Just (chan, socks) -> do
114+ modifyMVar_ socks (return . (clientSock :))
110− Just chan -> forkIO $ do
115+ forkIO $ do
111− unless (BS.null bs') $ writeChan chan bs'
116+ unless (BS.null bs') $ writeChan chan bs'
112− procMessages chan clientSock
117+ procMessages chan clientSock
113
118
114 procMessages :: Chan ByteString -> Socket -> IO ()
119 procMessages :: Chan ByteString -> Socket -> IO ()
115 procMessages chan sock = forever $ do
120 procMessages chan sock = forever $ do