git.iko.soy/distributed-process
on master help sign in
Modify multicast API, test the in-memory transport
386ead7Edsko de Vries
doc/InternalsTransportMulticast.txtnew file+181
compare with master
unifiedsplit
@@ -0,0 +1,181 @@
1+Adding Support for Multicast to the Transport API
2+=================================================
3+
4+This document describes various design options for adding support for multicast
5+to the Transport API.
6+
7+Creating a new multicast group
8+------------------------------
9+
10+We can either have this as part of the transport
11+
12+ data Transport = Transport {
13+ ...
14+ , newMulticastGroup :: IO (Either Error MulticastGroup)
15+ }
16+
17+ data MulticastGroup = MulticastGroup {
18+ ...
19+ , multicastAddress :: MulticastAddress
20+ , deleteMulticastGroup :: IO ()
21+ }
22+
23+or as part of an endpoint:
24+
25+ data Transport = Transport {
26+ newEndPoint :: IO (Either Error EndPoint)
27+ }
28+
29+ data EndPoint = EndPoint {
30+ ...
31+ , newMulticastGroup :: IO (Either Error MulticastGroup)
32+ }
33+
34+It should probably be part of the `Transport`, as there is no real connection
35+between an endpoint and the creation of the multigroup (however, see section
36+"Sending messages to a multicast group").
37+
38+Subscribing to a multicast group
39+--------------------------------
40+
41+This should be part of an endpoint; subscribing basically means that the
42+endpoint wants to receive events when multicast messages are sent.
43+
44+We could reify a subscription:
45+
46+ data EndPoint = EndPoint {
47+ ...
48+ , multicastSubscribe :: MulticastAddress -> IO MulticastSubscription
49+ }
50+
51+ data MulticastSubscription = MulticastSubscription {
52+ ...
53+ , multicastSubscriptionClose :: IO ()
54+ }
55+
56+but this suggests that one might have multiple subscriptions to the same group
57+which can be distinguished, which is misleading. Probably better to have:
58+
59+ data EndPoint = EndPoint {
60+ multicastSubscribe :: MulticastAddress -> IO ()
61+ , multicastUnsubscribe :: MulticastAddress -> IO ()
62+ }
63+
64+Sending messages to a multicast group
65+-------------------------------------
66+
67+An important feature of the Transport API is that we are clear about which
68+operations are *lightweight* and which are not. For instance, creating new
69+endpoints is not lightweight, but opening new connections to endpoints is (as
70+light-weight as possible).
71+
72+Clearly the creation of a new multicast group is a heavyweight operation. It is
73+less evident however if we can support multiple lightweight "connections" to the
74+same multicast group, and if so, whether it is useful.
75+
76+If we decide that multiple lightweight connections to the multigroup is useful,
77+one option might be
78+
79+ data EndPoint = EndPoint {
80+ ...
81+ , connect :: Address -> Reliability -> IO (Either Error Connection)
82+ , multicastConnect :: MulticastAddress -> IO (Either Error Connection)
83+ }
84+
85+ data Connection = Connection {
86+ connectionId :: ConnectionId
87+ , send :: [ByteString] -> IO ()
88+ , close :: IO ()
89+ , maxMsgSize :: Maybe Int
90+ }
91+
92+ data Event =
93+ Receive ConnectionId [ByteString]
94+ | ConnectionClosed ConnectionId
95+ | ConnectionOpened ConnectionId ConnectionType Reliability Address
96+
97+The advantage of this approach is it's consistency with the rest of the
98+interface. The problem is that with multicast we cannot reliably send any
99+control messages, so we cannot make sure that the subscribers of the multicast
100+group will receive ConnectionOpened events when an endpoint creates a new
101+connection. Since we don't support these "connectionless connections" anywhere
102+else in the API this seems inconsistent with the rest of the design (this
103+implies that an "unreliable" Transport over UDP still needs to have reliable
104+control messages). (On the other hand, if we were going to support reliable
105+multicast protocols, then that would fit this design).
106+
107+If we don't want to support multiple lightweight connections to a multicast
108+group then a better design would be
109+
110+ data EndPoint = EndPoint {
111+ , connect :: Address -> Reliability -> IO (Either Error Connection)
112+ , multicastSend :: MulticastAddress -> [ByteString] -> IO ()
113+ }
114+
115+ data Event =
116+ ...
117+ | MulticastReceive Address [ByteString]
118+
119+or alternatively
120+
121+ data EndPoint = EndPoint {
122+ ...
123+ , resolveMulticastGroup :: MulticastAddress -> IO (Either Error MulticastGroup)
124+ }
125+
126+ data MulticastGroup = MulticastGroup {
127+ , ...
128+ , send :: [ByteString] -> IO ()
129+ }
130+
131+If we do this however we need to make sure that newGroup is part an `EndPoint`,
132+not the `Transport`, otherwise `send` will not know the source of the message.
133+The version with `resolveMulticastGroup` has the additional benefit that in
134+"real" implementations we will probably need to allocate some resources before
135+we can send to the multicast group, and need to deallocate these resources at
136+some point too.
137+
138+The current solution
139+--------------------
140+
141+The above considerations lead to the following tentative proposal:
142+
143+ data Transport = Transport {
144+ newEndPoint :: IO (Either Error EndPoint)
145+ }
146+
147+ data EndPoint = EndPoint {
148+ receive :: IO Event
149+ , address :: Address
150+ , connect :: Address -> Reliability -> IO (Either Error Connection)
151+ , newMulticastGroup :: IO (Either Error MulticastGroup)
152+ , resolveMulticastGroup :: MulticastAddress -> IO (Either Error MulticastGroup)
153+ }
154+
155+ data Connection = Connection {
156+ send :: [ByteString] -> IO ()
157+ , close :: IO ()
158+ }
159+
160+ data Event =
161+ Receive ConnectionId [ByteString]
162+ | ConnectionClosed ConnectionId
163+ | ConnectionOpened ConnectionId Reliability Address
164+ | MulticastReceive MulticastAddress [ByteString]
165+
166+ data MulticastGroup = MulticastGroup {
167+ multicastAddress :: MulticastAddress
168+ , deleteMulticastGroup :: IO ()
169+ , maxMsgSize :: Maybe Int
170+ , multicastSend :: [ByteString] -> IO ()
171+ , multicastSubscribe :: IO ()
172+ , multicastUnsubscribe :: IO ()
173+ , multicastClose :: IO ()
174+ }
175+
176+where `multicastClose` indicates to the runtime that this endpoint no longer
177+wishes to send to this multicast group, and we can therefore deallocate the
178+resources we needed to send to the group (these resources can be allocated on
179+`resolveMulticastGroup` or on the first `multicastSend`; the advantage of the
180+latter is that is somebody resolves a group only to subscribe to it, not to
181+send to it, we don't allocate any unneeded resources).
network-transport-2/network-transport.cabal+11
@@ -14,4 +14,15 @@ Library
14 Exposed-modules: Network.Transport,
14 Exposed-modules: Network.Transport,
15 Network.Transport.Chan
15 Network.Transport.Chan
16 extensions: EmptyDataDecls
16 extensions: EmptyDataDecls
17+ ghc-options: -Wall -fno-warn-unused-do-bind
17 HS-Source-Dirs: src
18 HS-Source-Dirs: src
19+
20+Test-Suite TestInMemory
21+ Type: exitcode-stdio-1.0
22+ Main-Is: TestInMemory.hs
23+ Build-Depends: base,
24+ containers,
25+ bytestring
26+ extensions: EmptyDataDecls OverloadedStrings
27+ ghc-options: -Wall -fno-warn-unused-do-bind
28+ HS-Source-Dirs: tests src
network-transport-2/src/Network/Transport.hs+59−25
@@ -1,63 +1,97 @@
1-- | Network Transport
1-- | Network Transport
2−--
3−-- Connections are as light-weight as possible. That is, for a UDP
4−-- implementation they are practically free (only a single socket is
5−-- created per end point). For a TCP implementation the Transport guarantees
6−-- that only a single socket pair will be created for every pair of end
7−-- points.
8module Network.Transport where
2module Network.Transport where
9
3
10import Data.ByteString (ByteString)
4import Data.ByteString (ByteString)
5+import Control.Concurrent (forkIO)
6+import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar)
11
7
12-- | To create a network abstraction layer, use one of the
8-- | To create a network abstraction layer, use one of the
13−-- Network.Transport.* packages.
9+-- @Network.Transport.*@ packages.
14data Transport = Transport {
10data Transport = Transport {
11+ -- | Create a new end point (heavyweight operation)
15− newEndPoint :: Identity -> IO (Either Error EndPoint)
12+ newEndPoint :: IO (Either Error EndPoint)
16−}
13+ }
17−
18−-- | Every end point must have a unique identity.
19−type Identity = ByteString
20
14
15+-- | Address of an endpoint.
21newtype Address = Address ByteString
16newtype Address = Address ByteString
17+ deriving Show
22
18
19+-- | Address of a multicast group.
23newtype MulticastAddress = MulticastAddress ByteString
20newtype MulticastAddress = MulticastAddress ByteString
21+ deriving Show
24
22
23+-- | Error codes (none defined at the moment).
25data ErrorCode
24data ErrorCode
26
25
26+-- | Error consisting of an error code and a human readable error string.
27data Error = Error ErrorCode String
27data Error = Error ErrorCode String
28
28
29+-- | Network endpoint.
29data EndPoint = EndPoint {
30data EndPoint = EndPoint {
31+ -- | Endpoints have a single shared receive queue.
30 receive :: IO Event
32 receive :: IO Event
33+ -- | Address of the endpoint.
31 , address :: Address
34 , address :: Address
35+ -- | Create a new lightweight connection.
32 , connect :: Address -> Reliability -> IO (Either Error Connection)
36 , connect :: Address -> Reliability -> IO (Either Error Connection)
37+ -- | Create a new multicast group.
33− , multicastNewGroup :: IO (Either Error MulticastGroup)
38+ , newMulticastGroup :: IO (Either Error MulticastGroup)
34− , multicastConnect :: MulticastAddress -> IO MulticastGroup
39+ -- | Resolve an address to a multicast group.
35− , multicastSubscribe :: MulticastAddress -> IO ()
40+ , resolveMulticastGroup :: MulticastAddress -> IO (Either Error MulticastGroup)
36− , multicastUnsubscribe :: MulticastAddress -> IO ()
37−}
41+ }
38
42
43+-- | Reliability guarantees of a connection.
39data Reliability =
44data Reliability =
40 ReliableOrdered
45 ReliableOrdered
41 | ReliableUnordered
46 | ReliableUnordered
42 | Unreliable
47 | Unreliable
48+ deriving Show
43
49
50+-- | Connection IDs enable receivers to distinguish one connection from another.
44type ConnectionId = Int
51type ConnectionId = Int
45
52
53+-- | Lightweight connection to an endpoint.
46data Connection = Connection {
54data Connection = Connection {
47− connId :: ConnectionId
55+ -- | Send a message on this connection.
48− , send :: [ByteString] -> IO ()
56+ send :: [ByteString] -> IO ()
57+ -- | Close the connection.
49− , close :: IO ()
58+ , close :: IO ()
50 }
59 }
51
60
61+-- | Event on an endpoint.
52data Event =
62data Event =
53 Receive ConnectionId [ByteString]
63 Receive ConnectionId [ByteString]
54 | ConnectionClosed ConnectionId
64 | ConnectionClosed ConnectionId
55− | ConnectionOpened ConnectionId Reliability
65+ | ConnectionOpened ConnectionId Reliability Address
56 | MulticastReceive MulticastAddress [ByteString]
66 | MulticastReceive MulticastAddress [ByteString]
67+ deriving Show
57
68
69+-- | Multicast group.
58data MulticastGroup = MulticastGroup {
70data MulticastGroup = MulticastGroup {
71+ -- | Address of the multicast group.
72+ multicastAddress :: MulticastAddress
73+ -- | Delete the multicast group completely.
74+ , deleteMulticastGroup :: IO ()
75+ -- | Maximum message size that we can send to this group.
76+ , maxMsgSize :: Maybe Int
77+ -- | Send a message to the group.
59− multicastSend :: [ByteString] -> IO ()
78+ , multicastSend :: [ByteString] -> IO ()
60− , multicastAddress :: MulticastAddress
79+ -- | Subscribe to the given multicast group (to start receiving messages from the group).
61− , multicastMaxMsgSize :: Maybe Int
80+ , multicastSubscribe :: IO ()
62− , multicastGroupClose :: IO ()
81+ -- | Unsubscribe from the given multicast group (to stop receiving messages from the group).
82+ , multicastUnsubscribe :: IO ()
83+ -- | Close the group (that is, indicate you no longer wish to send to the group).
84+ , multicastClose :: IO ()
63 }
85 }
86+
87+-- | Fork a new thread, create a new end point on that thread, and run the specified IO operation on that thread.
88+--
89+-- Returns the address of the new end point.
90+spawn :: Transport -> (EndPoint -> IO ()) -> IO Address
91+spawn transport proc = do
92+ addr <- newEmptyMVar
93+ forkIO $ do
94+ Right endpoint <- newEndPoint transport
95+ putMVar addr (address endpoint)
96+ proc endpoint
97+ takeMVar addr
network-transport-2/src/Network/Transport/Chan.hs+32−34
@@ -1,49 +1,47 @@
1−-- | Implementation of the Transport which creates a single Chan for every endpoint.
1+-- | In-memory implementation of the Transport API.
2module Network.Transport.Chan (createTransport) where
2module 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.Concurrent.MVar (MVar, newMVar, takeMVar, putMVar)
6+import Control.Concurrent.MVar (MVar, newMVar, modifyMVar)
7import Data.Map (Map)
7import Data.Map (Map)
8−import qualified Data.Map as Map (empty, insert, (!), member)
8+import qualified Data.Map as Map (empty, insert, (!))
9+import Data.ByteString (ByteString)
10+import qualified Data.ByteString.Char8 as BSC (pack)
9
11
10−-- Global state: mapping from endpoint identities to the number of the next available connection
12+-- Global state: next available "address", mapping from addresses to channels and next available connection
11−type TransportState = Map Identity Int
13+type TransportState = (Int, Map ByteString (Chan Event, Int))
12
14
13−-- | Create a new Transport
15+-- | Create a new Transport.
16+--
17+-- Only a single transport should be created per Haskell process
18+-- (threads can, and should, create their own endpoints though).
14createTransport :: IO Transport
19createTransport :: IO Transport
15createTransport = do
20createTransport = do
16− endpoints <- newMVar Map.empty
21+ endpoints <- newMVar (0, Map.empty)
17 return Transport { newEndPoint = chanNewEndPoint endpoints }
22 return Transport { newEndPoint = chanNewEndPoint endpoints }
18
23
19-- Create a new end point
24-- Create a new end point
20−chanNewEndPoint :: MVar TransportState -> Identity -> IO (Either Error EndPoint)
25+chanNewEndPoint :: MVar TransportState -> IO (Either Error EndPoint)
21−chanNewEndPoint endpoints identity = do
26+chanNewEndPoint endpoints = do
22− state <- takeMVar endpoints
27+ chan <- newChan
23− if identity `Map.member` state
28+ addr <- modifyMVar endpoints $ \(next, state) -> do
24− then do
29+ let addr = BSC.pack (show next)
25− putMVar endpoints state
30+ return ((next + 1, Map.insert addr (chan, 0) state), addr)
26− return . Left $ Error undefined ("Identifier " ++ show identity ++ " already used")
27− else do
28− putMVar endpoints (Map.insert identity 0 state)
29− chan <- newChan
30− return . Right $ EndPoint { receive = readChan chan
31+ return . Right $ EndPoint { receive = readChan chan
31− , address = Address identity
32+ , address = Address addr
32− , connect = chanConnect chan endpoints
33+ , connect = chanConnect (Address addr) endpoints
33− , multicastNewGroup = undefined
34+ , newMulticastGroup = undefined
34− , multicastConnect = undefined
35+ , resolveMulticastGroup = undefined
35− , multicastSubscribe = undefined
36− , multicastUnsubscribe = undefined
37− }
36+ }
38
37
39-- Create a new connection
38-- Create a new connection
40−chanConnect :: Chan Event -> MVar TransportState -> Address -> Reliability -> IO (Either Error Connection)
39+chanConnect :: Address -> MVar TransportState -> Address -> Reliability -> IO (Either Error Connection)
41−chanConnect chan endpoints (Address identity) _ = do
40+chanConnect myAddress endpoints (Address theirAddress) _ = do
42− state <- takeMVar endpoints
41+ (chan, conn) <- modifyMVar endpoints $ \(next, state) -> do
43− let conn = state Map.! identity
42+ let (chan, conn) = state Map.! theirAddress
44− putMVar endpoints (Map.insert identity (conn + 1) state)
43+ return ((next, Map.insert theirAddress (chan, conn + 1) state), (chan, conn))
45− writeChan chan $ ConnectionOpened conn ReliableOrdered
44+ writeChan chan $ ConnectionOpened conn ReliableOrdered myAddress
46− return . Right $ Connection { connId = conn
47− , send = writeChan chan . Receive conn
45+ return . Right $ Connection { send = writeChan chan . Receive conn
48− , close = writeChan chan $ ConnectionClosed conn
46+ , close = writeChan chan $ ConnectionClosed conn
49 }
47 }
network-transport-2/tests/TestInMemory.hsnew file+10
@@ -0,0 +1,10 @@
1+module Main where
2+
3+import TestTransport
4+import Network.Transport.Chan
5+import System.Exit (exitFailure, exitSuccess)
6+
7+main :: IO ()
8+main = do
9+ success <- createTransport >>= testTransport
10+ if success then exitSuccess else exitFailure
network-transport-2/tests/TestTransport.hsnew file+138
@@ -0,0 +1,138 @@
1+module TestTransport where
2+
3+import Control.Concurrent (forkIO)
4+import Control.Monad (liftM2, replicateM, replicateM_, when)
5+import Control.Applicative ((<$>))
6+import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar)
7+import Network.Transport
8+import Data.ByteString (ByteString)
9+import Data.ByteString.Char8 ()
10+import Data.Map (Map)
11+import qualified Data.Map as Map (empty, insert, (!), delete)
12+
13+type Test = IO Bool
14+
15+-- Server that echoes messages straight back to the origin endpoint.
16+echoServer :: EndPoint -> IO ()
17+echoServer endpoint = go Map.empty
18+ where
19+ go :: Map ConnectionId Connection -> IO ()
20+ go cs = do
21+ event <- receive endpoint
22+ case event of
23+ ConnectionOpened cid rel addr -> do
24+ Right conn <- connect endpoint addr rel
25+ go (Map.insert cid conn cs)
26+ Receive cid payload -> do
27+ send (cs Map.! cid) payload
28+ go cs
29+ ConnectionClosed cid -> do
30+ close (cs Map.! cid)
31+ go (Map.delete cid cs)
32+ MulticastReceive _ _ ->
33+ -- Ignore
34+ go cs
35+
36+ping :: EndPoint -> Address -> Int -> ByteString -> IO ()
37+ping endpoint server numPings msg = do
38+ -- Open connection to the server
39+ Right conn <- connect endpoint server ReliableOrdered
40+
41+ -- Wait for the server to open reply connection
42+ ConnectionOpened _ _ _ <- receive endpoint
43+
44+ -- Send pings and wait for reply
45+ replicateM_ numPings $ do
46+ send conn [msg]
47+ Receive _ [reply] <- receive endpoint
48+ when (reply /= msg) . error $ "Message mismatch: " ++ show msg ++ " /= " ++ show reply
49+
50+ -- Close the connection
51+ close conn
52+
53+-- Basic ping test
54+testPingPong :: Transport -> Int -> Test
55+testPingPong transport numPings = do
56+ server <- spawn transport echoServer
57+ result <- newEmptyMVar
58+
59+ -- Client
60+ forkIO $ do
61+ Right endpoint <- newEndPoint transport
62+ ping endpoint server numPings "ping"
63+ putStrLn $ "client did " ++ show numPings ++ " pings"
64+ putMVar result True
65+
66+ takeMVar result
67+
68+-- Test that endpoints don't get confused
69+testEndPoints :: Transport -> Int -> Test
70+testEndPoints transport numPings = do
71+ server <- spawn transport echoServer
72+ [resultA, resultB] <- replicateM 2 newEmptyMVar
73+
74+ -- Client A
75+ forkIO $ do
76+ Right endpoint <- newEndPoint transport
77+ ping endpoint server numPings "pingA"
78+ putStrLn $ "client A did " ++ show numPings ++ " pings"
79+ putMVar resultA True
80+
81+ -- Client B
82+ forkIO $ do
83+ Right endpoint <- newEndPoint transport
84+ ping endpoint server numPings "pingB"
85+ putStrLn $ "client B did " ++ show numPings ++ " pings"
86+ putMVar resultB True
87+
88+ and <$> mapM takeMVar [resultA, resultB]
89+
90+-- Test that connections don't get confused
91+testConnections :: Transport -> Int -> Test
92+testConnections transport numPings = do
93+ server <- spawn transport echoServer
94+ result <- newEmptyMVar
95+
96+ -- Client
97+ forkIO $ do
98+ Right endpoint <- newEndPoint transport
99+
100+ -- Open two connections to the server
101+ Right conn1 <- connect endpoint server ReliableOrdered
102+ ConnectionOpened serv1 _ _ <- receive endpoint
103+
104+ Right conn2 <- connect endpoint server ReliableOrdered
105+ ConnectionOpened serv2 _ _ <- receive endpoint
106+
107+ -- One thread to send "pingA" on the first connection
108+ forkIO $ do
109+ replicateM_ numPings $ send conn1 ["pingA"]
110+ putStrLn $ "client A did " ++ show numPings ++ " pings"
111+
112+ -- One thread to send "pingB" on the second connection
113+ forkIO $ do
114+ replicateM_ numPings $ send conn2 ["pingB"]
115+ putStrLn $ "client B did " ++ show numPings ++ " pings"
116+
117+ -- Verify server responses
118+ let verifyResponse 0 = putMVar result True
119+ verifyResponse n = do
120+ event <- receive endpoint
121+ case event of
122+ Receive cid [payload] -> do
123+ when (cid == serv1 && payload /= "pingA") $ error "Wrong message"
124+ when (cid == serv2 && payload /= "pingB") $ error "Wrong message"
125+ verifyResponse (n - 1)
126+ _ ->
127+ verifyResponse n
128+ verifyResponse (2 * numPings)
129+
130+ takeMVar result
131+
132+-- Transport tests
133+testTransport :: Transport -> IO Bool
134+testTransport transport =
135+ foldl (liftM2 (&&)) (return True) [ testPingPong transport 10000
136+ , testEndPoints transport 10000
137+ , testConnections transport 10000
138+ ]