git.iko.soy/distributed-process
on master help sign in
Use 32-bit integres for connection IDs
85860aeEdsko de Vries
network-transport-2/src/Network/Transport.hs+2−1
compare with master
unifiedsplit
@@ -24,6 +24,7 @@ import Data.ByteString (ByteString)
24import Control.Concurrent (forkIO)
24import Control.Concurrent (forkIO)
25import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar)
25import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar)
26import Control.Monad.Error (Error(..))
26import Control.Monad.Error (Error(..))
27+import Data.Int (Int32)
27
28
28-- | To create a network abstraction layer, use one of the
29-- | To create a network abstraction layer, use one of the
29-- @Network.Transport.*@ packages.
30-- @Network.Transport.*@ packages.
@@ -87,7 +88,7 @@ data Reliability =
87 deriving Show
88 deriving Show
88
89
89-- | Connection IDs enable receivers to distinguish one connection from another.
90-- | Connection IDs enable receivers to distinguish one connection from another.
90−type ConnectionId = Int
91+type ConnectionId = Int32
91
92
92-- | Lightweight connection to an endpoint.
93-- | Lightweight connection to an endpoint.
93data Connection = Connection {
94data Connection = Connection {
network-transport-2/src/Network/Transport/Chan.hs+13−13
@@ -13,16 +13,16 @@ import qualified Data.ByteString.Char8 as BSC (pack)
13import Data.Lens.Lazy (Lens, lens, (^.), (^%=), (^=), (^+=), mapLens)
13import Data.Lens.Lazy (Lens, lens, (^.), (^%=), (^=), (^+=), mapLens)
14
14
15-- 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
16−data TransportState = State { _channels :: Map EndPointAddress (Chan Event)
16+data TransportState = State { _channels :: Map EndPointAddress (Chan Event)
17− , _connections :: Map EndPointAddress Int
17+ , _nextConnectionId :: Map EndPointAddress ConnectionId
18− , _multigroups :: Map MulticastAddress (MVar [Chan Event])
18+ , _multigroups :: Map MulticastAddress (MVar [Chan Event])
19 }
19 }
20
20
21channels :: Lens TransportState (Map EndPointAddress (Chan Event))
21channels :: Lens TransportState (Map EndPointAddress (Chan Event))
22channels = lens _channels (\ch st -> st { _channels = ch })
22channels = lens _channels (\ch st -> st { _channels = ch })
23
23
24−connections :: Lens TransportState (Map EndPointAddress Int)
24+nextConnectionId :: Lens TransportState (Map EndPointAddress ConnectionId)
25−connections = lens _connections (\conn st -> st { _connections = conn })
25+nextConnectionId = lens _nextConnectionId (\cid st -> st { _nextConnectionId = cid })
26
26
27multigroups :: Lens TransportState (Map MulticastAddress (MVar [Chan Event]))
27multigroups :: Lens TransportState (Map MulticastAddress (MVar [Chan Event]))
28multigroups = lens _multigroups (\gs st -> st { _multigroups = gs })
28multigroups = lens _multigroups (\gs st -> st { _multigroups = gs })
@@ -33,8 +33,8 @@ at k = lens (Map.! k) (Map.insert k)
33channelAt :: EndPointAddress -> Lens TransportState (Chan Event)
33channelAt :: EndPointAddress -> Lens TransportState (Chan Event)
34channelAt addr = channels >>> at addr
34channelAt addr = channels >>> at addr
35
35
36−connectionAt :: EndPointAddress -> Lens TransportState Int
36+nextConnectionIdAt :: EndPointAddress -> Lens TransportState ConnectionId
37−connectionAt addr = connections >>> at addr
37+nextConnectionIdAt addr = nextConnectionId >>> at addr
38
38
39multigroupAt :: MulticastAddress -> Lens TransportState (MVar [Chan Event])
39multigroupAt :: MulticastAddress -> Lens TransportState (MVar [Chan Event])
40multigroupAt addr = multigroups >>> at addr
40multigroupAt addr = multigroups >>> at addr
@@ -45,9 +45,9 @@ multigroupAt addr = multigroups >>> at addr
45-- (threads can, and should, create their own endpoints though).
45-- (threads can, and should, create their own endpoints though).
46createTransport :: IO Transport
46createTransport :: IO Transport
47createTransport = do
47createTransport = do
48− state <- newMVar State { _channels = Map.empty
48+ state <- newMVar State { _channels = Map.empty
49− , _connections = Map.empty
49+ , _nextConnectionId = Map.empty
50− , _multigroups = Map.empty
50+ , _multigroups = Map.empty
51 }
51 }
52 return Transport { newEndPoint = chanNewEndPoint state }
52 return Transport { newEndPoint = chanNewEndPoint state }
53
53
@@ -57,7 +57,7 @@ chanNewEndPoint state = do
57 chan <- newChan
57 chan <- newChan
58 addr <- modifyMVar state $ \st -> do
58 addr <- modifyMVar state $ \st -> do
59 let addr = EndPointAddress . BSC.pack . show . Map.size $ st ^. channels
59 let addr = EndPointAddress . BSC.pack . show . Map.size $ st ^. channels
60− return ((channelAt addr ^= chan) . (connectionAt addr ^= 0) $ st, addr)
60+ return ((channelAt addr ^= chan) . (nextConnectionIdAt addr ^= 1) $ st, addr)
61 return . Right $ EndPoint { receive = readChan chan
61 return . Right $ EndPoint { receive = readChan chan
62 , address = addr
62 , address = addr
63 , connect = chanConnect addr state
63 , connect = chanConnect addr state
@@ -70,8 +70,8 @@ chanConnect :: EndPointAddress -> MVar TransportState -> EndPointAddress -> Reli
70chanConnect myAddress state theirAddress _ = do
70chanConnect myAddress state theirAddress _ = do
71 (chan, conn) <- modifyMVar state $ \st -> do
71 (chan, conn) <- modifyMVar state $ \st -> do
72 let chan = st ^. channelAt theirAddress
72 let chan = st ^. channelAt theirAddress
73− let conn = st ^. connectionAt theirAddress
73+ let conn = st ^. nextConnectionIdAt theirAddress
74− return (connectionAt theirAddress ^+= 1 $ st, (chan, conn))
74+ return (nextConnectionIdAt theirAddress ^+= 1 $ st, (chan, conn))
75 writeChan chan $ ConnectionOpened conn ReliableOrdered myAddress
75 writeChan chan $ ConnectionOpened conn ReliableOrdered myAddress
76 return . Right $ Connection { send = \msg -> do writeChan chan (Received conn msg)
76 return . Right $ Connection { send = \msg -> do writeChan chan (Received conn msg)
77 return (Right ())
77 return (Right ())
network-transport-2/src/Network/Transport/Internal/TCP.hs+24−1
@@ -5,11 +5,17 @@ module Network.Transport.Internal.TCP ( forkServer
5 , recvWithLength
5 , recvWithLength
6 , sendMany
6 , sendMany
7 , recvExact
7 , recvExact
8+ , recvInt16
9+ , recvInt32
10+ , sendInt16
11+ , sendInt32
8 ) where
12 ) where
9
13
10import Prelude hiding (catch)
14import Prelude hiding (catch)
11import Network.Transport.Internal ( encodeInt32
15import Network.Transport.Internal ( encodeInt32
12 , decodeInt32
16 , decodeInt32
17+ , encodeInt16
18+ , decodeInt16
13 )
19 )
14import qualified Network.Socket as N ( HostName
20import qualified Network.Socket as N ( HostName
15 , ServiceName
21 , ServiceName
@@ -32,9 +38,10 @@ import Control.Concurrent (forkIO, ThreadId)
32import Control.Monad (mzero, MonadPlus, unless)
38import Control.Monad (mzero, MonadPlus, unless)
33import Control.Monad.IO.Class (MonadIO, liftIO)
39import Control.Monad.IO.Class (MonadIO, liftIO)
34import Control.Exception (catch, IOException)
40import Control.Exception (catch, IOException)
41+import Control.Applicative (pure)
35import Data.ByteString (ByteString)
42import Data.ByteString (ByteString)
36import qualified Data.ByteString as BS (length, concat, null)
43import qualified Data.ByteString as BS (length, concat, null)
37−import Data.Int (Int32)
44+import Data.Int (Int16, Int32)
38
45
39-- | Start a server at the specified address
46-- | Start a server at the specified address
40--
47--
@@ -110,3 +117,19 @@ recvExact sock len = do
110 if BS.null bs
117 if BS.null bs
111 then return (True, reverse acc)
118 then return (True, reverse acc)
112 else go (bs : acc) (l - fromIntegral (BS.length bs))
119 else go (bs : acc) (l - fromIntegral (BS.length bs))
120+
121+-- | Receive a 16-bit integer
122+recvInt16 :: (MonadIO m, MonadPlus m) => N.Socket -> m Int16
123+recvInt16 sock = recvExact sock 2 >>= decodeInt16 . BS.concat
124+
125+-- | Receive a 32-bit integer
126+recvInt32 :: (MonadIO m, MonadPlus m) => N.Socket -> m Int32
127+recvInt32 sock = recvExact sock 4 >>= decodeInt32 . BS.concat
128+
129+-- | Send a 16-bit integer
130+sendInt16 :: (MonadIO m, MonadPlus m) => N.Socket -> Int16 -> m ()
131+sendInt16 sock n = encodeInt16 n >>= sendMany sock . pure
132+
133+-- | Send a 32-bit integer
134+sendInt32 :: (MonadIO m, MonadPlus m) => N.Socket -> Int32 -> m ()
135+sendInt32 sock n = encodeInt32 n >>= sendMany sock . pure
network-transport-2/src/Network/Transport/TCP.hs+48−51
@@ -7,7 +7,7 @@
7module Network.Transport.TCP ( -- * Main API
7module Network.Transport.TCP ( -- * Main API
8 createTransport
8 createTransport
9 , -- TCP specific functionality
9 , -- TCP specific functionality
10− EndPointIx
10+ EndPointId
11 , decodeEndPointAddress
11 , decodeEndPointAddress
12 ) where
12 ) where
13
13
@@ -15,12 +15,13 @@ import Network.Transport
15import Network.Transport.Internal.TCP ( forkServer
15import Network.Transport.Internal.TCP ( forkServer
16 , connectTo
16 , connectTo
17 , recvExact
17 , recvExact
18− , sendMany
19 , recvWithLength
18 , recvWithLength
20 , sendWithLength
19 , sendWithLength
20+ , recvInt16
21+ , recvInt32
22+ , sendInt32
21 )
23 )
22import Network.Transport.Internal ( encodeInt16
24import Network.Transport.Internal ( encodeInt16
23− , decodeInt16
24 , maybeToErrorT
25 , maybeToErrorT
25 , maybeTToErrorT
26 , maybeTToErrorT
26 )
27 )
@@ -34,7 +35,7 @@ import Control.Concurrent (forkIO)
34import Control.Concurrent.Chan (Chan, newChan, readChan, writeChan)
35import Control.Concurrent.Chan (Chan, newChan, readChan, writeChan)
35import Control.Concurrent.MVar (MVar, newMVar, modifyMVar, readMVar)
36import Control.Concurrent.MVar (MVar, newMVar, modifyMVar, readMVar)
36import Control.Category ((>>>))
37import Control.Category ((>>>))
37−import Control.Applicative ((<*>), (*>), (<$>), pure)
38+import Control.Applicative ((<*>), (*>), (<$>))
38import Control.Monad (forever, forM_)
39import Control.Monad (forever, forM_)
39import Control.Monad.Error (ErrorT, liftIO, runErrorT)
40import Control.Monad.Error (ErrorT, liftIO, runErrorT)
40import Control.Monad.Trans.Maybe (MaybeT(MaybeT), runMaybeT)
41import Control.Monad.Trans.Maybe (MaybeT(MaybeT), runMaybeT)
@@ -43,24 +44,24 @@ import Control.Monad.Writer.Class (tell)
43import Data.ByteString (ByteString)
44import Data.ByteString (ByteString)
44import qualified Data.ByteString as BS (concat)
45import qualified Data.ByteString as BS (concat)
45import qualified Data.ByteString.Char8 as BSC (pack, unpack)
46import qualified Data.ByteString.Char8 as BSC (pack, unpack)
47+import Data.Int (Int16)
46import Data.IntMap (IntMap)
48import Data.IntMap (IntMap)
47import qualified Data.IntMap as IntMap (empty, size)
49import qualified Data.IntMap as IntMap (empty, size)
48import Data.Lens.Lazy (Lens, lens, intMapLens, (^.), (^=), (^+=))
50import Data.Lens.Lazy (Lens, lens, intMapLens, (^.), (^=), (^+=))
49import Text.Regex.Applicative (RE, few, anySym, sym, many, match)
51import Text.Regex.Applicative (RE, few, anySym, sym, many, match)
50
52
51data TransportState = TransportState { _endPoints :: IntMap (Chan Event, MVar EndPointState) }
53data TransportState = TransportState { _endPoints :: IntMap (Chan Event, MVar EndPointState) }
52−data EndPointState = EndPointState { _nextConnectionIx :: ConnectionIx }
54+data EndPointState = EndPointState { _nextConnectionId :: ConnectionId }
53−type EndPointIx = Int
55+type EndPointId = Int16
54−type ConnectionIx = Int
55
56
56endPoints :: Lens TransportState (IntMap (Chan Event, MVar EndPointState))
57endPoints :: Lens TransportState (IntMap (Chan Event, MVar EndPointState))
57endPoints = lens _endPoints (\es st -> st { _endPoints = es })
58endPoints = lens _endPoints (\es st -> st { _endPoints = es })
58
59
59−nextConnectionIx :: Lens EndPointState ConnectionIx
60+nextConnectionId :: Lens EndPointState ConnectionId
60−nextConnectionIx = lens _nextConnectionIx (\cix st -> st { _nextConnectionIx = cix })
61+nextConnectionId = lens _nextConnectionId (\cix st -> st { _nextConnectionId = cix })
61
62
62−endPointAt :: EndPointIx -> Lens TransportState (Maybe (Chan Event, MVar EndPointState))
63+endPointAt :: EndPointId -> Lens TransportState (Maybe (Chan Event, MVar EndPointState))
63−endPointAt ix = endPoints >>> intMapLens ix
64+endPointAt ix = endPoints >>> intMapLens (fromIntegral ix)
64
65
65-- | Create a TCP transport
66-- | Create a TCP transport
66--
67--
@@ -92,18 +93,14 @@ transportServer state sock = forever $ do
92handleClient :: MVar TransportState -> N.Socket -> IO ()
93handleClient :: MVar TransportState -> N.Socket -> IO ()
93handleClient state sock = do
94handleClient state sock = do
94 mendpoint <- runMaybeT $ do
95 mendpoint <- runMaybeT $ do
95− -- Our endpoint the other side is interested in
96+ ourEndPointId <- recvInt16 sock
96− endPointBS <- recvExact sock 2
97− endPointIx <- decodeInt16 (BS.concat endPointBS)
98− -- Full endpoint address of the other side
99− theirAddress <- BS.concat <$> recvWithLength sock
97+ theirAddress <- BS.concat <$> recvWithLength sock
100− -- Find our endpoint
101− (endPointCh, endPointSt) <- MaybeT $ (^. endPointAt (fromIntegral endPointIx)) <$> readMVar state
98+ ourEndPoint <- MaybeT $ (^. endPointAt (fromIntegral ourEndPointId)) <$> readMVar state
102− return (endPointCh, endPointSt, theirAddress)
99+ return (ourEndPoint, EndPointAddress theirAddress)
103 case mendpoint of
100 case mendpoint of
104− Just (endPointCh, endPointSt, theirAddress) -> do
101+ Just ((endPointCh, endPointSt), theirAddress) -> do
105− openConnections <- endpointServer sock (EndPointAddress theirAddress) endPointCh endPointSt
102+ unclosedConnections <- endpointServer sock theirAddress endPointCh endPointSt
106− forM_ openConnections $ \cix -> writeChan endPointCh (ConnectionClosed cix)
103+ forM_ unclosedConnections $ \cix -> writeChan endPointCh (ConnectionClosed cix)
107 Nothing ->
104 Nothing ->
108 -- Invalid request
105 -- Invalid request
109 return ()
106 return ()
@@ -111,41 +108,39 @@ handleClient state sock = do
111-- | Handle requests for one endpoint. Returns only if an error occurs,
108-- | Handle requests for one endpoint. Returns only if an error occurs,
112-- in which case it returns the connections that are still open (normally,
109-- in which case it returns the connections that are still open (normally,
113-- this should be an empty list).
110-- this should be an empty list).
114−endpointServer :: N.Socket -> EndPointAddress -> Chan Event -> MVar EndPointState -> IO [ConnectionIx]
111+endpointServer :: N.Socket -> EndPointAddress -> Chan Event -> MVar EndPointState -> IO [ConnectionId]
115endpointServer sock theirAddress chan state = execWriterT . runMaybeT . forever $ do
112endpointServer sock theirAddress chan state = execWriterT . runMaybeT . forever $ do
116− connBS <- recvExact sock 2
113+ connId <- recvInt32 sock
117− connIx <- fromIntegral <$> decodeInt16 (BS.concat connBS)
114+ case connId of
118− if connIx == 0
119− then createNewConnection
115+ 0 -> createNewConnection
120− else readMessage connIx
116+ _ -> readMessage connId
121 where
117 where
122 -- A brief note on the types:
118 -- A brief note on the types:
123 -- These functions may fail (connection error): hence MaybeT
119 -- These functions may fail (connection error): hence MaybeT
124 -- We also need to keep track of which connections are still open: hence WriterT
120 -- We also need to keep track of which connections are still open: hence WriterT
125
121
126 -- Create a new connection
122 -- Create a new connection
127− createNewConnection :: MaybeT (WriterT [ConnectionIx] IO) ()
123+ createNewConnection :: MaybeT (WriterT [ConnectionId] IO) ()
128 createNewConnection = do
124 createNewConnection = do
129− newIx <- liftIO $ modifyMVar state $ \st -> return (nextConnectionIx ^+= 1 $ st, st ^. nextConnectionIx)
125+ newId <- liftIO $ modifyMVar state $ \st -> return (nextConnectionId ^+= 1 $ st, st ^. nextConnectionId)
130− newBs <- encodeInt16 (fromIntegral newIx)
126+ sendInt32 sock newId
131− sendMany sock [newBs]
132− liftIO $ writeChan chan (ConnectionOpened newIx ReliableOrdered theirAddress)
127+ liftIO $ writeChan chan (ConnectionOpened newId ReliableOrdered theirAddress)
128+ -- We add the new connection ID to the list of open connections only once the
129+ -- endpoint has been notified of the new connection (sendInt32 may fail)
133− tell [newIx]
130+ tell [newId]
134
131
135 -- Read a message and output it on the endPoint's channel
132 -- Read a message and output it on the endPoint's channel
136− readMessage :: ConnectionIx -> MaybeT (WriterT [ConnectionIx] IO) ()
133+ readMessage :: ConnectionId -> MaybeT (WriterT [ConnectionId] IO) ()
137− readMessage connIx = do
138− payload <- recvWithLength sock
139− liftIO $ writeChan chan (Received connIx payload)
134+ readMessage connId = recvWithLength sock >>= liftIO . writeChan chan . Received connId
140
135
141-- | Create a new endpoint
136-- | Create a new endpoint
142tcpNewEndPoint :: MVar TransportState -> N.HostName -> N.ServiceName -> IO (Either (FailedWith NewEndPointErrorCode) EndPoint)
137tcpNewEndPoint :: MVar TransportState -> N.HostName -> N.ServiceName -> IO (Either (FailedWith NewEndPointErrorCode) EndPoint)
143tcpNewEndPoint transportState host port = do
138tcpNewEndPoint transportState host port = do
144 chan <- newChan
139 chan <- newChan
145− endPointState <- newMVar EndPointState { _nextConnectionIx = 1 } -- Connection id 0 is reserved for requesting new connections
140+ endPointState <- newMVar EndPointState { _nextConnectionId = firstNonReservedConnectionId }
146− endPointIx <- modifyMVar transportState $ \st -> do let ix = IntMap.size (st ^. endPoints)
141+ endPointId <- modifyMVar transportState $ \st -> do let ix = fromIntegral $ IntMap.size (st ^. endPoints)
147 return ((endPointAt ix ^= Just (chan, endPointState)) $ st, ix)
142 return ((endPointAt ix ^= Just (chan, endPointState)) $ st, ix)
148− let addr = EndPointAddress . BSC.pack $ host ++ ":" ++ port ++ ":" ++ show endPointIx
143+ let addr = EndPointAddress . BSC.pack $ host ++ ":" ++ port ++ ":" ++ show endPointId
149 return . Right $ EndPoint { receive = readChan chan
144 return . Right $ EndPoint { receive = readChan chan
150 , address = addr
145 , address = addr
151 , connect = tcpConnect addr
146 , connect = tcpConnect addr
@@ -155,24 +150,26 @@ tcpNewEndPoint transportState host port = do
155 where
150 where
156 newMulticastGroupError = FailedWith NewMulticastGroupUnsupported "TCP does not support multicast"
151 newMulticastGroupError = FailedWith NewMulticastGroupUnsupported "TCP does not support multicast"
157 resolveMulticastGroupError = FailedWith ResolveMulticastGroupUnsupported "TCP does not support multicast"
152 resolveMulticastGroupError = FailedWith ResolveMulticastGroupUnsupported "TCP does not support multicast"
158−
153+
154+-- | We reserve a bunch of connection IDs for control messages
155+firstNonReservedConnectionId :: ConnectionId
156+firstNonReservedConnectionId = 1024
159
157
160-- | Connnect to an endpoint
158-- | Connnect to an endpoint
161tcpConnect :: EndPointAddress -> EndPointAddress -> Reliability -> IO (Either (FailedWith ConnectErrorCode) Connection)
159tcpConnect :: EndPointAddress -> EndPointAddress -> Reliability -> IO (Either (FailedWith ConnectErrorCode) Connection)
162tcpConnect myAddress theirAddress _ = runErrorT $ do
160tcpConnect myAddress theirAddress _ = runErrorT $ do
163 sock <- socketForEndPoint myAddress theirAddress
161 sock <- socketForEndPoint myAddress theirAddress
164− connIx <- maybeTToErrorT undefined $ requestNewConnection sock
162+ connId <- maybeTToErrorT undefined $ requestNewConnection sock
165 return Connection { send = runErrorT .
163 return Connection { send = runErrorT .
166 maybeTToErrorT (FailedWith SendFailed "Send failed") .
164 maybeTToErrorT (FailedWith SendFailed "Send failed") .
167− sendWithLength sock (Just connIx)
165+ sendWithLength sock (Just connId)
168 , close = N.sClose sock
166 , close = N.sClose sock
169 }
167 }
170
168
171-- | Request a new connection
169-- | Request a new connection
170+-- Returns the new connection ID in serialized form
172requestNewConnection :: N.Socket -> MaybeT IO ByteString
171requestNewConnection :: N.Socket -> MaybeT IO ByteString
173−requestNewConnection sock = do
172+requestNewConnection sock = sendInt32 sock 0 >> BS.concat <$> recvExact sock 4
174− liftIO (pure <$> encodeInt16 0) >>= sendMany sock
175− BS.concat <$> recvExact sock 2
176
173
177-- | Find a socket to the specified endpoint
174-- | Find a socket to the specified endpoint
178--
175--
@@ -183,9 +180,9 @@ socketForEndPoint :: EndPointAddress -- ^ Our address
183 -> EndPointAddress -- ^ Their address
180 -> EndPointAddress -- ^ Their address
184 -> ErrorT (FailedWith ConnectErrorCode) IO N.Socket
181 -> ErrorT (FailedWith ConnectErrorCode) IO N.Socket
185socketForEndPoint (EndPointAddress myAddress) theirAddress = do
182socketForEndPoint (EndPointAddress myAddress) theirAddress = do
186− (host, port, endPointIx) <- maybeToErrorT invalidAddress (decodeEndPointAddress theirAddress)
183+ (host, port, endPointId) <- maybeToErrorT invalidAddress (decodeEndPointAddress theirAddress)
187 sock <- maybeTToErrorT undefined (connectTo host port)
184 sock <- maybeTToErrorT undefined (connectTo host port)
188− endPointBS <- liftIO $ encodeInt16 (fromIntegral endPointIx)
185+ endPointBS <- encodeInt16 (fromIntegral endPointId)
189 maybeTToErrorT undefined $ sendWithLength sock (Just endPointBS) [myAddress]
186 maybeTToErrorT undefined $ sendWithLength sock (Just endPointBS) [myAddress]
190 return sock
187 return sock
191 where
188 where
@@ -193,8 +190,8 @@ socketForEndPoint (EndPointAddress myAddress) theirAddress = do
193
190
194-- | Decode end point address
191-- | Decode end point address
195-- TODO: This uses regular expression parsing, which is nice, but unnecessary
192-- TODO: This uses regular expression parsing, which is nice, but unnecessary
196−decodeEndPointAddress :: EndPointAddress -> Maybe (N.HostName, N.ServiceName, EndPointIx)
193+decodeEndPointAddress :: EndPointAddress -> Maybe (N.HostName, N.ServiceName, EndPointId)
197decodeEndPointAddress (EndPointAddress bs) = match endPointAddressRE (BSC.unpack bs)
194decodeEndPointAddress (EndPointAddress bs) = match endPointAddressRE (BSC.unpack bs)
198 where
195 where
199− endPointAddressRE :: RE Char (N.HostName, N.ServiceName, EndPointIx)
196+ endPointAddressRE :: RE Char (N.HostName, N.ServiceName, EndPointId)
200 endPointAddressRE = (,,) <$> few anySym <*> (sym ':' *> few anySym) <*> (sym ':' *> (read <$> many anySym))
197 endPointAddressRE = (,,) <$> few anySym <*> (sym ':' *> few anySym) <*> (sym ':' *> (read <$> many anySym))
network-transport-2/tests/TestTCP.hs+48−61
@@ -2,13 +2,15 @@ module Main where
2
2
3import Network.Transport
3import Network.Transport
4import Network.Transport.Internal (encodeInt16, decodeInt16)
4import Network.Transport.Internal (encodeInt16, decodeInt16)
5−import Network.Transport.Internal.TCP (sendWithLength, recvExact, sendMany)
5+import Network.Transport.Internal.TCP ( sendWithLength
6+ , recvExact
7+ , sendInt32
8+ )
9+import Network.Transport.TCP (createTransport, decodeEndPointAddress, EndPointId)
6import TestTransport (testTransport)
10import TestTransport (testTransport)
7−import Network.Transport.TCP (createTransport, decodeEndPointAddress)
8import System.Exit (exitFailure, exitSuccess)
11import System.Exit (exitFailure, exitSuccess)
9−import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar, readMVar)
12+import Control.Concurrent.MVar (MVar, newEmptyMVar, putMVar, takeMVar, readMVar)
10import Control.Concurrent (forkIO)
13import Control.Concurrent (forkIO)
11−-- import Control.Monad.IO.Class (liftIO)
12import Control.Monad.Trans.Maybe (runMaybeT)
14import Control.Monad.Trans.Maybe (runMaybeT)
13import Data.Maybe (fromJust)
15import Data.Maybe (fromJust)
14import qualified Network.Socket as N ( getAddrInfo
16import qualified Network.Socket as N ( getAddrInfo
@@ -21,68 +23,53 @@ import qualified Network.Socket as N ( getAddrInfo
21 , defaultProtocol
23 , defaultProtocol
22 , setSocketOption
24 , setSocketOption
23 , sClose
25 , sClose
26+ , HostName
27+ , ServiceName
24 )
28 )
25
29
26testEarlyDisconnect :: IO ()
30testEarlyDisconnect :: IO ()
27testEarlyDisconnect = do
31testEarlyDisconnect = do
28− serverAddr <- newEmptyMVar
32+ serverAddr <- newEmptyMVar
29− serverDone <- newEmptyMVar
33+ serverDone <- newEmptyMVar
30− transport <- createTransport "127.0.0.1" "8081"
34+ transport <- createTransport "127.0.0.1" "8081"
35+
36+ forkIO $ server transport serverAddr serverDone
37+ forkIO $ client transport serverAddr
38+
39+ takeMVar serverDone
40+ where
41+ server :: Transport -> MVar (N.HostName, N.ServiceName, EndPointId) -> MVar () -> IO ()
42+ server transport serverAddr serverDone = do
43+ Right endpoint <- newEndPoint transport
44+ putMVar serverAddr (fromJust . decodeEndPointAddress . address $ endpoint)
45+ ConnectionOpened _ _ _ <- receive endpoint
46+ ConnectionClosed _ <- receive endpoint
47+ putMVar serverDone ()
48+
49+ client :: Transport -> MVar (N.HostName, N.ServiceName, EndPointId) -> IO ()
50+ client transport serverAddr = do
51+ Right endpoint <- newEndPoint transport
52+ let EndPointAddress myAddress = address endpoint
31
53
32− -- Server
33− forkIO $ do
34− Right endpoint <- newEndPoint transport
35− putMVar serverAddr (fromJust . decodeEndPointAddress . address $ endpoint)
36−
37− putStrLn "Server waiting for client to connect"
38−
39− -- Wait for an incoming connection
40− ConnectionOpened _ _ _ <- receive endpoint
41−
42− putStrLn "Server waiting for client to disconnect"
43−
44− -- Wait for the client to disconnect
45− ConnectionClosed _ <- receive endpoint
46−
47− putStrLn "Server exiting"
48−
49− putMVar serverDone ()
50−
51− -- Client
52− --
53− -- We manually connect to the server to request a new connection,
54− -- but then close the socket before the server gets the chance to
55− -- respond
56− forkIO $ do
57− Right endpoint <- newEndPoint transport
58− let EndPointAddress myAddress = address endpoint
59−
60− -- Connect to the server
54+ -- Connect to the server
61− addr:_ <- N.getAddrInfo Nothing (Just "127.0.0.1") (Just "8081")
55+ addr:_ <- N.getAddrInfo Nothing (Just "127.0.0.1") (Just "8081")
62− sock <- N.socket (N.addrFamily addr) N.Stream N.defaultProtocol
56+ sock <- N.socket (N.addrFamily addr) N.Stream N.defaultProtocol
63− N.setSocketOption sock N.ReuseAddr 1
57+ N.setSocketOption sock N.ReuseAddr 1
64− N.connect sock (N.addrAddress addr)
58+ N.connect sock (N.addrAddress addr)
65−
59+
66− (_, _, endPointIx) <- readMVar serverAddr
60+ (_, _, endPointIx) <- readMVar serverAddr
67− endPointBS <- encodeInt16 (fromIntegral endPointIx)
61+ endPointBS <- encodeInt16 (fromIntegral endPointIx)
68− runMaybeT $ sendWithLength sock (Just endPointBS) [myAddress]
62+ runMaybeT $ sendWithLength sock (Just endPointBS) [myAddress]
69−
63+
70− -- Request a new connection
64+ -- Request a new connection
71− zero <- encodeInt16 0
72− runMaybeT $ sendMany sock [zero]
65+ connIx <- runMaybeT $ do
73−
66+ sendInt32 sock 0
74− -- Request a new connection
75− connIx <- runMaybeT $ do
76− [connBs] <- recvExact sock 2
67+ [connBs] <- recvExact sock 2
77− decodeInt16 connBs
68+ decodeInt16 connBs
78− putStrLn $ "Client got connection ID " ++ show connIx
69+
79−
80− -- Close the socket without closing the connection explicitly
70+ -- Close the socket without closing the connection explicitly
81− -- The server should still receive a ConnectionClosed message
71+ -- The server should still receive a ConnectionClosed message
82− N.sClose sock
72+ N.sClose sock
83−
84− -- Wait for the server to finish
85− takeMVar serverDone
86
73
87main :: IO ()
74main :: IO ()
88main = do
75main = do