git.iko.soy/distributed-process
on master help sign in
Add multicast support to in-memory transport
53e0889Edsko de Vries
network-transport-2/network-transport.cabal+16−2
compare with master
unifiedsplit
@@ -10,19 +10,33 @@ Cabal-Version: >=1.9.2
10Library
10Library
11 Build-Depends: base,
11 Build-Depends: base,
12 bytestring,
12 bytestring,
13− containers
13+ containers,
14+ data-lens
14 Exposed-modules: Network.Transport,
15 Exposed-modules: Network.Transport,
15 Network.Transport.Chan
16 Network.Transport.Chan
16 extensions: EmptyDataDecls
17 extensions: EmptyDataDecls
17 ghc-options: -Wall -fno-warn-unused-do-bind
18 ghc-options: -Wall -fno-warn-unused-do-bind
18 HS-Source-Dirs: src
19 HS-Source-Dirs: src
19
20
21+Test-Suite TestMulticastInMemory
22+ Type: exitcode-stdio-1.0
23+ Main-Is: TestMulticastInMemory.hs
24+ Build-Depends: base,
25+ containers,
26+ bytestring,
27+ data-lens
28+ extensions: EmptyDataDecls OverloadedStrings
29+ ghc-options: -Wall -fno-warn-unused-do-bind
30+ HS-Source-Dirs: tests src
31+
20Test-Suite TestInMemory
32Test-Suite TestInMemory
21 Type: exitcode-stdio-1.0
33 Type: exitcode-stdio-1.0
22 Main-Is: TestInMemory.hs
34 Main-Is: TestInMemory.hs
23 Build-Depends: base,
35 Build-Depends: base,
24 containers,
36 containers,
25− bytestring
37+ bytestring,
38+ data-lens
26 extensions: EmptyDataDecls OverloadedStrings
39 extensions: EmptyDataDecls OverloadedStrings
27 ghc-options: -Wall -fno-warn-unused-do-bind
40 ghc-options: -Wall -fno-warn-unused-do-bind
28 HS-Source-Dirs: tests src
41 HS-Source-Dirs: tests src
42+
network-transport-2/src/Network/Transport.hs+4−4
@@ -14,11 +14,11 @@ data Transport = Transport {
14
14
15-- | Address of an endpoint.
15-- | Address of an endpoint.
16newtype Address = Address ByteString
16newtype Address = Address ByteString
17− deriving Show
17+ deriving (Show, Eq, Ord)
18
18
19-- | Address of a multicast group.
19-- | Address of a multicast group.
20newtype MulticastAddress = MulticastAddress ByteString
20newtype MulticastAddress = MulticastAddress ByteString
21− deriving Show
21+ deriving (Show, Eq, Ord)
22
22
23-- | Error codes (none defined at the moment).
23-- | Error codes (none defined at the moment).
24data ErrorCode
24data ErrorCode
@@ -60,10 +60,10 @@ data Connection = Connection {
60
60
61-- | Event on an endpoint.
61-- | Event on an endpoint.
62data Event =
62data Event =
63− Receive ConnectionId [ByteString]
63+ Received ConnectionId [ByteString]
64 | ConnectionClosed ConnectionId
64 | ConnectionClosed ConnectionId
65 | ConnectionOpened ConnectionId Reliability Address
65 | ConnectionOpened ConnectionId Reliability Address
66− | MulticastReceive MulticastAddress [ByteString]
66+ | ReceivedMulticast MulticastAddress [ByteString]
67 deriving Show
67 deriving Show
68
68
69-- | Multicast group.
69-- | Multicast group.
network-transport-2/src/Network/Transport/Chan.hs+91−20
@@ -3,14 +3,41 @@ module Network.Transport.Chan (createTransport) where
3
3
4import Network.Transport
4import Network.Transport
5import Control.Concurrent.Chan (Chan, newChan, readChan, writeChan)
5import Control.Concurrent.Chan (Chan, newChan, readChan, writeChan)
6+import Control.Applicative ((<$>))
7+import Control.Category ((>>>))
6−import Control.Concurrent.MVar (MVar, newMVar, modifyMVar)
8+import Control.Concurrent.MVar (MVar, newMVar, modifyMVar, modifyMVar_, readMVar)
9+import Control.Monad (forM_)
7import Data.Map (Map)
10import Data.Map (Map)
8−import qualified Data.Map as Map (empty, insert, (!))
11+import qualified Data.Map as Map (empty, insert, (!), size, delete)
9−import Data.ByteString (ByteString)
10import qualified Data.ByteString.Char8 as BSC (pack)
12import qualified Data.ByteString.Char8 as BSC (pack)
13+import Data.Lens.Lazy (Lens, lens, (^.), (^%=), (^=), (^+=), mapLens)
11
14
12-- Global state: next available "address", mapping from addresses to channels and next available connection
15-- Global state: next available "address", mapping from addresses to channels and next available connection
13−type TransportState = (Int, Map ByteString (Chan Event, Int))
16+data TransportState = State { _channels :: Map Address (Chan Event)
17+ , _connections :: Map Address Int
18+ , _multigroups :: Map MulticastAddress (MVar [Chan Event])
19+ }
20+
21+channels :: Lens TransportState (Map Address (Chan Event))
22+channels = lens _channels (\ch st -> st { _channels = ch })
23+
24+connections :: Lens TransportState (Map Address Int)
25+connections = lens _connections (\conn st -> st { _connections = conn })
26+
27+multigroups :: Lens TransportState (Map MulticastAddress (MVar [Chan Event]))
28+multigroups = lens _multigroups (\gs st -> st { _multigroups = gs })
29+
30+at :: Ord k => k -> Lens (Map k v) v
31+at k = lens (Map.! k) (Map.insert k)
32+
33+channelAt :: Address -> Lens TransportState (Chan Event)
34+channelAt addr = channels >>> at addr
35+
36+connectionAt :: Address -> Lens TransportState Int
37+connectionAt addr = connections >>> at addr
38+
39+multigroupAt :: MulticastAddress -> Lens TransportState (MVar [Chan Event])
40+multigroupAt addr = multigroups >>> at addr
14
41
15-- | Create a new Transport.
42-- | Create a new Transport.
16--
43--
@@ -18,30 +45,74 @@ type TransportState = (Int, Map ByteString (Chan Event, Int))
18-- (threads can, and should, create their own endpoints though).
45-- (threads can, and should, create their own endpoints though).
19createTransport :: IO Transport
46createTransport :: IO Transport
20createTransport = do
47createTransport = do
21− endpoints <- newMVar (0, Map.empty)
48+ state <- newMVar $ State { _channels = Map.empty
49+ , _connections = Map.empty
50+ , _multigroups = Map.empty
51+ }
22− return Transport { newEndPoint = chanNewEndPoint endpoints }
52+ return Transport { newEndPoint = chanNewEndPoint state }
23
53
24-- Create a new end point
54-- Create a new end point
25chanNewEndPoint :: MVar TransportState -> IO (Either Error EndPoint)
55chanNewEndPoint :: MVar TransportState -> IO (Either Error EndPoint)
26−chanNewEndPoint endpoints = do
56+chanNewEndPoint state = do
27 chan <- newChan
57 chan <- newChan
28− addr <- modifyMVar endpoints $ \(next, state) -> do
58+ addr <- modifyMVar state $ \st -> do
29− let addr = BSC.pack (show next)
59+ let addr = Address . BSC.pack . show . Map.size $ st ^. channels
30− return ((next + 1, Map.insert addr (chan, 0) state), addr)
60+ return ((channelAt addr ^= chan) . (connectionAt addr ^= 0) $ st, addr)
31 return . Right $ EndPoint { receive = readChan chan
61 return . Right $ EndPoint { receive = readChan chan
32− , address = Address addr
62+ , address = addr
33− , connect = chanConnect (Address addr) endpoints
63+ , connect = chanConnect addr state
34− , newMulticastGroup = undefined
64+ , newMulticastGroup = chanNewMulticastGroup state chan
35− , resolveMulticastGroup = undefined
65+ , resolveMulticastGroup = chanResolveMulticastGroup state chan
36 }
66 }
37
67
38-- Create a new connection
68-- Create a new connection
39chanConnect :: Address -> MVar TransportState -> Address -> Reliability -> IO (Either Error Connection)
69chanConnect :: Address -> MVar TransportState -> Address -> Reliability -> IO (Either Error Connection)
40−chanConnect myAddress endpoints (Address theirAddress) _ = do
70+chanConnect myAddress state theirAddress _ = do
41− (chan, conn) <- modifyMVar endpoints $ \(next, state) -> do
71+ (chan, conn) <- modifyMVar state $ \st -> do
72+ let chan = st ^. channelAt theirAddress
42− let (chan, conn) = state Map.! theirAddress
73+ let conn = st ^. connectionAt theirAddress
43− return ((next, Map.insert theirAddress (chan, conn + 1) state), (chan, conn))
74+ return (connectionAt theirAddress ^+= 1 $ st, (chan, conn))
44 writeChan chan $ ConnectionOpened conn ReliableOrdered myAddress
75 writeChan chan $ ConnectionOpened conn ReliableOrdered myAddress
45− return . Right $ Connection { send = writeChan chan . Receive conn
76+ return . Right $ Connection { send = writeChan chan . Received conn
46− , close = writeChan chan $ ConnectionClosed conn
77+ , close = writeChan chan $ ConnectionClosed conn
47 }
78 }
79+
80+-- Create a new multicast group
81+chanNewMulticastGroup :: MVar TransportState -> Chan Event -> IO (Either Error MulticastGroup)
82+chanNewMulticastGroup state endpoint = do
83+ group <- newMVar []
84+ addr <- modifyMVar state $ \st -> do
85+ let addr = MulticastAddress . BSC.pack . show . Map.size $ st ^. multigroups
86+ return (multigroupAt addr ^= group $ st, addr)
87+ return . Right $ chanMulticastGroup state endpoint addr group
88+
89+-- Construct a multicast group
90+--
91+-- When the group is deleted some endpoints may still receive messages, but
92+-- subsequent calls to resolveMulticastGroup will fail. This mimicks the fact
93+-- that some multicast messages may still be in transit when the group is
94+-- deleted.
95+--
96+-- TODO: subscribing twice means you will receive messages twice, but as soon
97+-- as you unsubscribe you will stop receiving all messages. Not sure if this
98+-- is the right behaviour.
99+chanMulticastGroup :: MVar TransportState -> Chan Event -> MulticastAddress -> MVar [Chan Event] -> MulticastGroup
100+chanMulticastGroup state endpoint addr group =
101+ MulticastGroup { multicastAddress = addr
102+ , deleteMulticastGroup = modifyMVar_ state $ return . (multigroups ^%= Map.delete addr)
103+ , maxMsgSize = Nothing
104+ , multicastSend = \payload -> do
105+ cs <- readMVar group
106+ forM_ cs $ \ch -> writeChan ch (ReceivedMulticast addr payload)
107+ , multicastSubscribe = modifyMVar_ group $ return . (endpoint :)
108+ , multicastUnsubscribe = modifyMVar_ group $ return . (filter (/= endpoint))
109+ , multicastClose = return ()
110+ }
111+
112+-- Resolve a multicast group
113+chanResolveMulticastGroup :: MVar TransportState -> Chan Event -> MulticastAddress -> IO (Either Error MulticastGroup)
114+chanResolveMulticastGroup state endpoint addr = do
115+ group <- (^. (multigroups >>> mapLens addr)) <$> readMVar state
116+ case group of
117+ Nothing -> return . Left $ Error undefined ("Group " ++ show addr ++ " not found")
118+ Just mvar -> return . Right $ chanMulticastGroup state endpoint addr mvar
network-transport-2/tests/TestMulticast.hsnew file+70
@@ -0,0 +1,70 @@
1+module TestMulticast where
2+
3+import Network.Transport
4+import Control.Monad (replicateM, replicateM_, forM_, when)
5+import Control.Concurrent (forkIO)
6+import Control.Concurrent.MVar (MVar, newEmptyMVar, takeMVar, putMVar, readMVar)
7+import Control.Applicative ((<$>))
8+import Data.ByteString (ByteString)
9+import Data.List (elemIndex)
10+
11+-- | Node for the "No confusion" test
12+noConfusionNode :: Transport -- ^ Transport
13+ -> [MVar MulticastAddress] -- ^ my group : groups to subscribe to
14+ -> [MVar ()] -- ^ I'm ready : others ready
15+ -> Int -- ^ number of pings
16+ -> [ByteString] -- ^ my message : messages from subscribed groups (same order as 'groups to subscribe to')
17+ -> MVar Bool -- ^ I'm done
18+ -> IO ()
19+noConfusionNode transport groups ready numPings msgs done = do
20+ -- Create a new endpoint
21+ Right endpoint <- newEndPoint transport
22+
23+ -- Create a new multicast group and broadcast its address
24+ Right myGroup <- newMulticastGroup endpoint
25+ putMVar (head groups) (multicastAddress myGroup)
26+
27+ -- Subscribe to the given multicast groups
28+ addrs <- mapM readMVar (tail groups)
29+ forM_ addrs $ \addr -> do Right group <- resolveMulticastGroup endpoint addr
30+ multicastSubscribe group
31+
32+ -- Indicate that we're ready and wait for everybody else to be ready
33+ putMVar (head ready) ()
34+ mapM_ readMVar (tail ready)
35+
36+ -- Send messages..
37+ forkIO . replicateM_ numPings $ multicastSend myGroup [head msgs]
38+
39+ -- ..while checking that the messages we receive are the right ones
40+ replicateM_ (2 * numPings) $ do
41+ event <- receive endpoint
42+ case event of
43+ ReceivedMulticast addr [msg] ->
44+ let mix = addr `elemIndex` addrs in
45+ case mix of
46+ Nothing -> error "Message from unexpected source"
47+ Just ix -> when (msgs !! (ix + 1) /= msg) $ error "Unexpected message"
48+ _ ->
49+ error "Unexpected event"
50+
51+ -- Success
52+ putMVar done True
53+
54+-- | Test that distinct multicast groups are not confused
55+testNoConfusion :: Transport -> Int -> IO Bool
56+testNoConfusion transport numPings = do
57+ [group1, group2, group3] <- replicateM 3 newEmptyMVar
58+ [readyA, readyB, readyC] <- replicateM 3 newEmptyMVar
59+ [doneA, doneB, doneC] <- replicateM 3 newEmptyMVar
60+ let [msgA, msgB, msgC] = ["A says hi", "B says hi", "C says hi"]
61+
62+ forkIO $ noConfusionNode transport [group1, group1, group2] [readyA, readyB, readyC] numPings [msgA, msgA, msgB] doneA
63+ forkIO $ noConfusionNode transport [group2, group1, group3] [readyB, readyC, readyA] numPings [msgB, msgA, msgC] doneB
64+ forkIO $ noConfusionNode transport [group3, group2, group3] [readyC, readyA, readyB] numPings [msgC, msgB, msgC] doneC
65+
66+ and <$> mapM takeMVar [doneA, doneB, doneC]
67+
68+-- | Test multicast
69+testMulticast :: Transport -> IO Bool
70+testMulticast transport = testNoConfusion transport 10000
network-transport-2/tests/TestMulticastInMemory.hsnew file+10
@@ -0,0 +1,10 @@
1+module Main where
2+
3+import TestMulticast
4+import Network.Transport.Chan
5+import System.Exit (exitFailure, exitSuccess)
6+
7+main :: IO ()
8+main = do
9+ success <- createTransport >>= testMulticast
10+ if success then exitSuccess else exitFailure
network-transport-2/tests/TestTransport.hs+4−4
@@ -23,13 +23,13 @@ echoServer endpoint = go Map.empty
23 ConnectionOpened cid rel addr -> do
23 ConnectionOpened cid rel addr -> do
24 Right conn <- connect endpoint addr rel
24 Right conn <- connect endpoint addr rel
25 go (Map.insert cid conn cs)
25 go (Map.insert cid conn cs)
26− Receive cid payload -> do
26+ Received cid payload -> do
27 send (cs Map.! cid) payload
27 send (cs Map.! cid) payload
28 go cs
28 go cs
29 ConnectionClosed cid -> do
29 ConnectionClosed cid -> do
30 close (cs Map.! cid)
30 close (cs Map.! cid)
31 go (Map.delete cid cs)
31 go (Map.delete cid cs)
32− MulticastReceive _ _ ->
32+ ReceivedMulticast _ _ ->
33 -- Ignore
33 -- Ignore
34 go cs
34 go cs
35
35
@@ -44,7 +44,7 @@ ping endpoint server numPings msg = do
44 -- Send pings and wait for reply
44 -- Send pings and wait for reply
45 replicateM_ numPings $ do
45 replicateM_ numPings $ do
46 send conn [msg]
46 send conn [msg]
47− Receive _ [reply] <- receive endpoint
47+ Received _ [reply] <- receive endpoint
48 when (reply /= msg) . error $ "Message mismatch: " ++ show msg ++ " /= " ++ show reply
48 when (reply /= msg) . error $ "Message mismatch: " ++ show msg ++ " /= " ++ show reply
49
49
50 -- Close the connection
50 -- Close the connection
@@ -119,7 +119,7 @@ testConnections transport numPings = do
119 verifyResponse n = do
119 verifyResponse n = do
120 event <- receive endpoint
120 event <- receive endpoint
121 case event of
121 case event of
122− Receive cid [payload] -> do
122+ Received cid [payload] -> do
123 when (cid == serv1 && payload /= "pingA") $ error "Wrong message"
123 when (cid == serv1 && payload /= "pingA") $ error "Wrong message"
124 when (cid == serv2 && payload /= "pingB") $ error "Wrong message"
124 when (cid == serv2 && payload /= "pingB") $ error "Wrong message"
125 verifyResponse (n - 1)
125 verifyResponse (n - 1)