f46269a · distributed-process

git.iko.soy/distributed-process

Switch to asynchronous connect

In the original protocol we had recipient-allocated connection IDs. Hence, when doing a connect, we would send a connection request to the remote endpoint (provided that a heavyweight connection is already in place) and then wait for the recipient to reply with a new connection ID. This however makes connection much slower than it needs to be. We now use sender-allocated connection IDs, so that connect turns into a simple message to the remote endpoint that we allocated a new connection, and we no longer have to wait for a response. This means that all operations, with the sole exception of creating the heavyweight connection itself, are now synchronous. Hence, we have completely removed infrastructure relating to 'remote requests' (doRemoteRequest, ConnectionRequestResponse, etc.). What's not obvious is that we also had to change the 'CloseSocket' message. A sends a CloseSocket message to B when A thinks that the heavyweight connection to B is no longer required. A doesn't immediately close the socket, because it is possible that B has already sent a new connection request (now a CreatedNewConnection message) to A, even though A has not yet received it. Instead, it puts the heavyweight connection in 'Closing' state and waits for B to confirm or deny the request. When B receives A's CloseSocket message, it checks if it agrees that the heavyweight connection can be closed. If not, it simply ignores the request and A will forget it ever sent a CloseSocket request when it receives a CreatedNewConnection message from B (that is, it moves the endpoint from Closing to Valid state). If B does agree, then it will reply with a CloseSocket message, and then actually close the socket. When A receives a CloseSocket message having sent a CloseSocket message it closes the socket immediately. However, now that we have asynchronous connection requests we might have the following potential ordering of messages: A sends CloseSocket request, and puts endpoint in closing state B sends CreatedNewConnection, CloseConnection, and CloseSocket requests, putting its own representation of the endpoint also in closing state. B receives the CloseSocket message from A. Since it's endpoint is in Closing state, it immediately closes the socket. A receives B CreatedNewConnection message, hence *cancelling* the CloseSocket message it sent to be and putting the endpoint back in Valid state -- *even though* B did closed the socket. To avoid this problem, we now send the ID of the last created connection along with the close socket request. Then when B receives a CloseSocket request, it can compare this ID with the ID of the last connection it opened. If this does not match, B knows that there is still a CreatedNewConnection message underway to A, which will cancel the CloseSocket, and hence it will know to ignore the CloseSocket request from A.

commitf46269a8c17da692d20254812413921f3c948842 authorEdsko de Vries <edsko@edsko.net> dateMon 17 Sep 2012, 13:19 (14 years ago) parentb44da5a

7 files changed+400−381

compare with master

unifiedsplit

network-transport-tcp/ChangeLog+1 network-transport-tcp/src/Network/Transport/TCP.hs+209−206 network-transport-tcp/tests/TestTCP.hs+165−171 network-transport-tests/src/Network/Transport/Tests.hs+4−2 network-transport-tests/src/Network/Transport/Tests/Traced.hs+11−1 network-transport/ChangeLog+1 network-transport/src/Network/Transport.hs+9−1

network-transport-tcp/ChangeLog+1

@@ -1,6 +1,7 @@

11<<date>> Edsko de Vries <edsko@well-typed.com> 0.3.0

33* Implement new disconnection semantics

4+* Make 'connect' asynchronous (sender allocated connection IDs)

562012-08-20 Edsko de Vries <edsko@well-typed.com> 0.2.0.3

network-transport-tcp/src/Network/Transport/TCP.hs+209−206

@@ -23,8 +23,10 @@ module Network.Transport.TCP

2323 , decodeEndPointAddress

2424 , ControlHeader(..)

2525 , ConnectionRequestResponse(..)

26− , firstNonReservedConnectionId

26+ , firstNonReservedLightweightConnectionId

27+ , firstNonReservedHeavyweightConnectionId

2728 , socketToEndPoint

29+ , LightweightConnectionId

2830 -- * Design notes

2931 -- $design

3032 ) where

@@ -45,7 +47,6 @@ import Network.Transport.TCP.Internal

4547 )

4648import Network.Transport.Internal

4749 ( encodeInt32

48− , decodeInt32

4950 , prependLength

5051 , mapIOException

5152 , tryIO

@@ -79,7 +80,6 @@ import Control.Concurrent.MVar

7980 , modifyMVar

8081 , modifyMVar_

8182 , readMVar

82− , takeMVar

8383 , putMVar

8484 , newEmptyMVar

8585 , withMVar

@@ -103,11 +103,10 @@ import Data.IORef (IORef, newIORef, writeIORef, readIORef)

103103import Data.ByteString (ByteString)

104104import qualified Data.ByteString as BS (concat)

105105import qualified Data.ByteString.Char8 as BSC (pack, unpack)

106−import Data.Int (Int32)

107−import Data.IntMap (IntMap)

108−import qualified Data.IntMap as IntMap (empty)

109−import Data.IntSet (IntSet)

110−import qualified Data.IntSet as IntSet

106+import Data.Bits (shiftL, (.|.))

107+import Data.Word (Word32)

108+import Data.Set (Set)

109+import qualified Data.Set as Set

111110 ( empty

112111 , insert

113112 , elems

@@ -119,7 +118,7 @@ import qualified Data.IntSet as IntSet

119118import Data.Map (Map)

120119import qualified Data.Map as Map (empty)

121120import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))

122−import qualified Data.Accessor.Container as DAC (mapMaybe, intMapMaybe)

121+import qualified Data.Accessor.Container as DAC (mapMaybe)

123122import Data.Foldable (forM_, mapM_)

124123

125124-- $design

@@ -256,9 +255,9 @@ data LocalEndPointState =

256255 | LocalEndPointClosed

257256

258257data ValidLocalEndPointState = ValidLocalEndPointState

259− { _nextConnectionId :: !ConnectionId

258+ { _nextConnOutId :: !LightweightConnectionId

259+ , _nextConnInId :: !HeavyweightConnectionId

260260 , _localConnections :: !(Map EndPointAddress RemoteEndPoint)

261− , _nextRemoteId :: !Int

262261 }

263262

264263-- REMOTE ENDPOINTS

@@ -278,14 +277,14 @@ data ValidLocalEndPointState = ValidLocalEndPointState

278277-- \-------+----------+--------> Failed

279278--

280279-- Init: There are two places where we create new remote endpoints: in

281−-- requestConnectionTo (in response to an API 'connect' call) and in

280+-- createConnectionTo (in response to an API 'connect' call) and in

282281-- handleConnectionRequest (when a remote node tries to connect to us).

283282-- 'Init' carries an MVar () 'resolved' which concurrent threads can use to

284283-- wait for the remote endpoint to finish initialization. We record who

285284-- requested the connection (the local endpoint or the remote endpoint).

286285--

287286-- Invalid: We put the remote endpoint in invalid state only during

288−-- requestConnectionTo when we fail to connect.

287+-- createConnectionTo when we fail to connect.

289288--

290289-- Valid: This is the "normal" state for a working remote endpoint.

291290--

@@ -346,7 +345,7 @@ data ValidLocalEndPointState = ValidLocalEndPointState

346345data RemoteEndPoint = RemoteEndPoint

347346 { remoteAddress :: !EndPointAddress

348347 , remoteState :: !(MVar RemoteState)

349− , remoteId :: !Int

348+ , remoteId :: !HeavyweightConnectionId

350349 , remoteScheduled :: !(Chan (IO ()))

351350 }

352351

@@ -368,35 +367,43 @@ data RemoteState =

368367 -- using a closeTransport or closeEndPoint API call

369368 | RemoteEndPointFailed !IOException

370369

370+-- TODO: we might want to replace Set (here and elsewhere) by faster

371+-- containers

372+--

373+-- TODO: we could get rid of 'remoteIncoming' (and maintain less state) if

374+-- we introduce a new event 'AllConnectionsClosed'

371375data ValidRemoteEndPointState = ValidRemoteEndPointState

372376 { _remoteOutgoing :: !Int

373− , _remoteIncoming :: !IntSet

377+ , _remoteIncoming :: !(Set LightweightConnectionId)

378+ , _remoteMaxIncoming :: !LightweightConnectionId

374379 , remoteSocket :: !N.Socket

375380 , remoteSendLock :: !(MVar ())

376− , _pendingCtrlRequests :: !(IntMap (MVar (Either IOException [ByteString])))

377− , _nextCtrlRequestId :: !ControlRequestId

378381 }

379382

380383-- | Local identifier for an endpoint within this transport

381−type EndPointId = Int32

382−

383−-- | Control request ID

384−--

385−-- Control requests are asynchronous; the request ID makes it possible to match

386−-- requests and replies

387−type ControlRequestId = Int32

384+type EndPointId = Word32

388385

389386-- | Pair of local and a remote endpoint (for conciseness in signatures)

390−type EndPointPair = (LocalEndPoint, RemoteEndPoint)

387+type EndPointPair = (LocalEndPoint, RemoteEndPoint)

388+

389+-- | Lightweight connection ID (sender allocated)

390+--

391+-- A ConnectionId is the concentation of a 'HeavyweightConnectionId' and a

392+-- 'LightweightConnectionId'.

393+type LightweightConnectionId = Word32

394+

395+-- | Heavyweight connection ID (recipient allocated)

396+--

397+-- A ConnectionId is the concentation of a 'HeavyweightConnectionId' and a

398+-- 'LightweightConnectionId'.

399+type HeavyweightConnectionId = Word32

391400

392401-- | Control headers

393402data ControlHeader =

394− -- | Request a new connection ID from the remote endpoint

395− RequestConnectionId

403+ -- | Tell the remote endpoint that we created a new connection

404+ CreatedNewConnection

396405 -- | Tell the remote endpoint we will no longer be using a connection

397406 | CloseConnection

398− -- | Respond to a control request /from/ the remote endpoint

399− | ControlResponse

400407 -- | Request to close the connection (see module description)

401408 | CloseSocket

402409 deriving (Enum, Bounded, Show)

@@ -558,7 +565,7 @@ apiConnect params ourEndPoint theirAddress _reliability hints =

558565 else do

559566 resetIfBroken ourEndPoint theirAddress

560567 (theirEndPoint, connId) <-

561− requestConnectionTo params ourEndPoint theirAddress hints

568+ createConnectionTo params ourEndPoint theirAddress hints

562569 -- connAlive can be an IORef rather than an MVar because it is protected

563570 -- by the remoteState MVar. We don't need the overhead of locking twice.

564571 connAlive <- newIORef True

@@ -568,7 +575,7 @@ apiConnect params ourEndPoint theirAddress _reliability hints =

568575 }

569576

570577-- | Close a connection

571−apiClose :: EndPointPair -> ConnectionId -> IORef Bool -> IO ()

578+apiClose :: EndPointPair -> LightweightConnectionId -> IORef Bool -> IO ()

572579apiClose (ourEndPoint, theirEndPoint) connId connAlive =

573580 void . tryIO . asyncWhenCancelled return $ do

574581 mAct <- modifyMVar (remoteState theirEndPoint) $ \st -> case st of

@@ -593,10 +600,10 @@ apiClose (ourEndPoint, theirEndPoint) connId connAlive =

593600

594601

595602-- | Send data across a connection

596−apiSend :: EndPointPair -- ^ Local and remote endpoint

597− -> ConnectionId -- ^ Connection ID (supplied by remote endpoint)

598− -> IORef Bool -- ^ Is the connection still alive?

599− -> [ByteString] -- ^ Payload

603+apiSend :: EndPointPair -- ^ Local and remote endpoint

604+ -> LightweightConnectionId -- ^ Connection ID

605+ -> IORef Bool -- ^ Is the connection still alive?

606+ -> [ByteString] -- ^ Payload

600607 -> IO (Either (TransportError SendErrorCode) ())

601608apiSend (ourEndPoint, theirEndPoint) connId connAlive payload =

602609 -- We don't need the overhead of asyncWhenCancelled here

@@ -665,7 +672,9 @@ apiCloseEndPoint transport evs ourEndPoint =

665672 return (closed, Nothing)

666673 RemoteEndPointValid vst -> do

667674 act <- schedule theirEndPoint $ do

668− tryIO $ sendOn vst [encodeInt32 CloseSocket]

675+ tryIO $ sendOn vst [ encodeInt32 CloseSocket

676+ , encodeInt32 (vst ^. remoteMaxIncoming)

677+ ]

669678 tryCloseSocket (remoteSocket vst)

670679 return (closed, Just act)

671680 RemoteEndPointClosing resolved vst -> do

@@ -729,12 +738,11 @@ handleConnectionRequest transport sock = handle handleException $ do

729738 else do

730739 sendLock <- newMVar ()

731740 let vst = ValidRemoteEndPointState

732− { remoteSocket = sock

733− , remoteSendLock = sendLock

734− , _remoteOutgoing = 0

735− , _remoteIncoming = IntSet.empty

736− , _pendingCtrlRequests = IntMap.empty

737− , _nextCtrlRequestId = 0

741+ { remoteSocket = sock

742+ , remoteSendLock = sendLock

743+ , _remoteOutgoing = 0

744+ , _remoteIncoming = Set.empty

745+ , _remoteMaxIncoming = 0

738746 }

739747 sendMany sock [encodeInt32 ConnectionRequestAccepted]

740748 resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst)

@@ -790,32 +798,28 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do

790798 -- exception thrown by 'recv'.

791799 go :: N.Socket -> IO ()

792800 go sock = do

793− connId <- recvInt32 sock

794− if connId >= firstNonReservedConnectionId

801+ lcid <- recvInt32 sock :: IO LightweightConnectionId

802+ if lcid >= firstNonReservedLightweightConnectionId

795803 then do

796− readMessage sock connId

804+ readMessage sock lcid

797805 go sock

798806 else

799− case tryToEnum (fromIntegral connId) of

800− Just RequestConnectionId -> do

801− recvInt32 sock >>= createNewConnection

807+ case tryToEnum (fromIntegral lcid) of

808+ Just CreatedNewConnection -> do

809+ recvInt32 sock >>= createdNewConnection

802810 go sock

803− Just ControlResponse -> do

804− recvInt32 sock >>= readControlResponse sock

805− go sock

806811 Just CloseConnection -> do

807812 recvInt32 sock >>= closeConnection

808813 go sock

809814 Just CloseSocket -> do

810− didClose <- closeSocket sock

815+ didClose <- recvInt32 sock >>= closeSocket sock

811816 unless didClose $ go sock

812817 Nothing ->

813818 throwIO $ userError "Invalid control request"

814819

815820 -- Create a new connection

816− createNewConnection :: ControlRequestId -> IO ()

817− createNewConnection reqId = do

818− newId <- getNextConnectionId ourEndPoint

821+ createdNewConnection :: LightweightConnectionId -> IO ()

822+ createdNewConnection lcid = do

819823 modifyMVar_ theirState $ \st -> do

820824 vst <- case st of

821825 RemoteEndPointInvalid _ ->

@@ -825,73 +829,45 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do

825829 relyViolation (ourEndPoint, theirEndPoint)

826830 "handleIncomingMessages:createNewConnection (init)"

827831 RemoteEndPointValid vst ->

828− return (remoteIncoming ^: IntSet.insert newId $ vst)

832+ return ( (remoteIncoming ^: Set.insert lcid)

833+ $ (remoteMaxIncoming ^= lcid)

834+ vst

835+ )

829836 RemoteEndPointClosing resolved vst -> do

830837 -- If the endpoint is in closing state that means we send a

831838 -- CloseSocket request to the remote endpoint. If the remote

832− -- endpoint replies with the request to create a new connection, it

833− -- either ignored our request or it sent the request before it got

834− -- ours. Either way, at this point we simply restore the endpoint

835− -- to RemoteEndPointValid

839+ -- endpoint replies that it created a new connection, it either

840+ -- ignored our request or it sent the request before it got ours.

841+ -- Either way, at this point we simply restore the endpoint to

842+ -- RemoteEndPointValid

836843 putMVar resolved ()

837− return (remoteIncoming ^= IntSet.singleton newId $ vst)

844+ return ( (remoteIncoming ^= Set.singleton lcid)

845+ . (remoteMaxIncoming ^= lcid)

846+ $ vst

847+ )

838848 RemoteEndPointFailed err ->

839849 throwIO err

840850 RemoteEndPointClosed ->

841851 relyViolation (ourEndPoint, theirEndPoint)

842852 "createNewConnection (closed)"

843− sendOn vst ( encodeInt32 ControlResponse

844− : encodeInt32 reqId

845− : prependLength [encodeInt32 newId]

846− )

847853 return (RemoteEndPointValid vst)

848− writeChan ourChannel (ConnectionOpened newId ReliableOrdered theirAddr)

849−

850− -- Read a control response

851− readControlResponse :: N.Socket -> ControlRequestId -> IO ()

852− readControlResponse sock reqId = do

853− response <- recvWithLength sock

854− mmvar <- modifyMVar theirState $ \st -> case st of

855− RemoteEndPointInvalid _ ->

856− relyViolation (ourEndPoint, theirEndPoint)

857− "readControlResponse (invalid)"

858− RemoteEndPointInit _ _ _ ->

859− relyViolation (ourEndPoint, theirEndPoint)

860− "readControlResponse (init)"

861− RemoteEndPointValid vst ->

862− return ( RemoteEndPointValid

863− . (pendingCtrlRequestsAt reqId ^= Nothing)

864− $ vst

865− , vst ^. pendingCtrlRequestsAt reqId

866− )

867− RemoteEndPointClosing _ _ ->

868− throwIO $ userError "Invalid control response"

869− RemoteEndPointFailed err ->

870− throwIO err

871− RemoteEndPointClosed ->

872− relyViolation (ourEndPoint, theirEndPoint)

873− "readControlResponse (closed)"

874− case mmvar of

875− Nothing ->

876− throwIO $ userError "Invalid request ID"

877− Just mvar ->

878− putMVar mvar (Right response)

854+ writeChan ourChannel (ConnectionOpened (connId lcid) ReliableOrdered theirAddr)

879855

880856 -- Close a connection

881857 -- It is important that we verify that the connection is in fact open,

882858 -- because otherwise we should not decrement the reference count

883− closeConnection :: ConnectionId -> IO ()

884− closeConnection cid = do

859+ closeConnection :: LightweightConnectionId -> IO ()

860+ closeConnection lcid = do

885861 modifyMVar_ theirState $ \st -> case st of

886862 RemoteEndPointInvalid _ ->

887863 relyViolation (ourEndPoint, theirEndPoint) "closeConnection (invalid)"

888864 RemoteEndPointInit _ _ _ ->

889865 relyViolation (ourEndPoint, theirEndPoint) "closeConnection (init)"

890866 RemoteEndPointValid vst -> do

891− unless (IntSet.member cid (vst ^. remoteIncoming)) $

867+ unless (Set.member lcid (vst ^. remoteIncoming)) $

892868 throwIO $ userError "Invalid CloseConnection"

893869 return ( RemoteEndPointValid

894− . (remoteIncoming ^: IntSet.delete cid)

870+ . (remoteIncoming ^: Set.delete lcid)

895871 $ vst

896872 )

897873 RemoteEndPointClosing _ _ ->

@@ -903,13 +879,13 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do

903879 throwIO err

904880 RemoteEndPointClosed ->

905881 relyViolation (ourEndPoint, theirEndPoint) "closeConnection (closed)"

906− writeChan ourChannel (ConnectionClosed cid)

907− closeIfUnused (ourEndPoint, theirEndPoint)

882+ writeChan ourChannel (ConnectionClosed (connId lcid))

908883

909884 -- Close the socket (if we don't have any outgoing connections)

910− closeSocket :: N.Socket -> IO Bool

911− closeSocket sock = do

912− mAct <- modifyMVar theirState $ \st ->

885+ closeSocket :: N.Socket -> LightweightConnectionId -> IO Bool

886+ closeSocket sock lastReceivedId = do

887+ mAct <- modifyMVar theirState $ \st -> do

888+ lastSentId <- getLastConnOutId ourEndPoint

913889 case st of

914890 RemoteEndPointInvalid _ ->

915891 relyViolation (ourEndPoint, theirEndPoint)

@@ -921,25 +897,44 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do

921897 -- We regard a CloseSocket message as an (optimized) way for the

922898 -- remote endpoint to indicate that all its connections to us are

923899 -- now properly closed

924− forM_ (IntSet.elems $ vst ^. remoteIncoming) $

925− writeChan ourChannel . ConnectionClosed

926− let vst' = remoteIncoming ^= IntSet.empty $ vst

927− -- Check if we agree that the connection should be closed

928− if vst' ^. remoteOutgoing == 0

929− then do

900+ forM_ (Set.elems $ vst ^. remoteIncoming) $

901+ writeChan ourChannel . ConnectionClosed . connId

902+ let vst' = remoteIncoming ^= Set.empty $ vst

903+ -- If we still have outgoing connections then we ignore the

904+ -- CloseSocket request (we sent a ConnectionCreated message to the

905+ -- remote endpoint, but it did not receive it before sending the

906+ -- CloseSocket request). Similarly, if lastReceivedId < lastSentId

907+ -- then we sent a ConnectionCreated *AND* a ConnectionClosed

908+ -- message to the remote endpoint, *both of which* it did not yet

909+ -- receive before sending the CloseSocket request.

910+ if vst' ^. remoteOutgoing > 0 || lastReceivedId < lastSentId

911+ then

912+ return (RemoteEndPointValid vst', Nothing)

913+ else do

930914 removeRemoteEndPoint (ourEndPoint, theirEndPoint)

931915 -- Attempt to reply (but don't insist)

932916 act <- schedule theirEndPoint $ do

933− tryIO $ sendOn vst' [encodeInt32 CloseSocket]

917+ tryIO $ sendOn vst' [ encodeInt32 CloseSocket

918+ , encodeInt32 (vst ^. remoteMaxIncoming)

919+ ]

934920 tryCloseSocket sock

935921 return (RemoteEndPointClosed, Just act)

936− else

937− return (RemoteEndPointValid vst', Nothing)

938− RemoteEndPointClosing resolved _ -> do

939− removeRemoteEndPoint (ourEndPoint, theirEndPoint)

940− act <- schedule theirEndPoint $ tryCloseSocket sock

941− putMVar resolved ()

942− return (RemoteEndPointClosed, Just act)

922+ RemoteEndPointClosing resolved vst ->

923+ -- Like above, we need to check if there is a ConnectionCreated

924+ -- message that we sent but that the remote endpoint has not yet

925+ -- received. However, since we are in 'closing' state, the only

926+ -- way this may happen is when we sent a ConnectionCreated,

927+ -- ConnectionClosed, and CloseSocket message, none of which have

928+ -- yet been received. We leave the endpoint in closing state in

929+ -- that case.

930+ if lastReceivedId < lastSentId

931+ then

932+ return (RemoteEndPointClosing resolved vst, Nothing)

933+ else do

934+ removeRemoteEndPoint (ourEndPoint, theirEndPoint)

935+ act <- schedule theirEndPoint $ tryCloseSocket sock

936+ putMVar resolved ()

937+ return (RemoteEndPointClosed, Just act)

943938 RemoteEndPointFailed err ->

944939 throwIO err

945940 RemoteEndPointClosed ->

@@ -950,13 +945,13 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do

950945 Just act -> do

951946 runScheduledAction (ourEndPoint, theirEndPoint) act

952947 return True

953−

948+

954949 -- Read a message and output it on the endPoint's channel. By rights we

955950 -- should verify that the connection ID is valid, but this is unnecessary

956951 -- overhead

957− readMessage :: N.Socket -> ConnectionId -> IO ()

958− readMessage sock connId =

959− recvWithLength sock >>= writeChan ourChannel . Received connId

952+ readMessage :: N.Socket -> LightweightConnectionId -> IO ()

953+ readMessage sock lcid =

954+ recvWithLength sock >>= writeChan ourChannel . Received (connId lcid)

960955

961956 -- Arguments

962957 ourChannel = localChannel ourEndPoint

@@ -975,10 +970,9 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do

975970 RemoteEndPointInit _ _ _ ->

976971 relyViolation (ourEndPoint, theirEndPoint)

977972 "handleIncomingMessages:prematureExit"

978− RemoteEndPointValid vst -> do

973+ RemoteEndPointValid _ -> do

979974 let code = EventConnectionLost (remoteAddress theirEndPoint)

980975 writeChan ourChannel . ErrorEvent $ TransportError code (show err)

981− forM_ (vst ^. pendingCtrlRequests) $ flip putMVar (Left err)

982976 return (RemoteEndPointFailed err)

983977 RemoteEndPointClosing resolved _ -> do

984978 putMVar resolved ()

@@ -989,6 +983,10 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do

989983 RemoteEndPointFailed err' ->

990984 return (RemoteEndPointFailed err')

991985

986+ -- Construct a connection ID

987+ connId :: LightweightConnectionId -> ConnectionId

988+ connId = createConnectionId (remoteId theirEndPoint)

989+

992990--------------------------------------------------------------------------------

993991-- Uninterruptable auxiliary functions --

994992-- --

@@ -996,19 +994,18 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do

996994-- be killed.

997995--------------------------------------------------------------------------------

998996

999−-- | Request a connection to a remote endpoint

997+-- | Create a connection to a remote endpoint

1000998--

1001−-- This will block until we get a connection ID from the remote endpoint; if

1002−-- the remote endpoint was in 'RemoteEndPointClosing' state then we will

1003−-- additionally block until that is resolved.

999+-- If the remote endpoint is in 'RemoteEndPointClosing' state then we will

1000+-- block until that is resolved.

10041001--

10051002-- May throw a TransportError ConnectErrorCode exception.

1006−requestConnectionTo :: TCPParameters

1003+createConnectionTo :: TCPParameters

10071004 -> LocalEndPoint

10081005 -> EndPointAddress

10091006 -> ConnectHints

1010− -> IO (RemoteEndPoint, ConnectionId)

1011−requestConnectionTo params ourEndPoint theirAddress hints = go

1007+ -> IO (RemoteEndPoint, LightweightConnectionId)

1008+createConnectionTo params ourEndPoint theirAddress hints = go

10121009 where

10131010 go = do

10141011 (theirEndPoint, isNew) <- mapIOException connectFailed $

@@ -1020,9 +1017,26 @@ requestConnectionTo params ourEndPoint theirAddress hints = go

10201017 setupRemoteEndPoint params (ourEndPoint, theirEndPoint) hints

10211018 go

10221019 else do

1023− reply <- mapIOException connectFailed $

1024− doRemoteRequest (ourEndPoint, theirEndPoint) RequestConnectionId

1025− return (theirEndPoint, decodeInt32 . BS.concat $ reply)

1020+ -- 'findRemoteEndPoint' will have increased 'remoteOutgoing'

1021+ mapIOException connectFailed $ do

1022+ act <- withMVar (remoteState theirEndPoint) $ \st -> case st of

1023+ RemoteEndPointValid vst -> do

1024+ connId <- getNextConnOutId ourEndPoint

1025+ schedule theirEndPoint $ do

1026+ sendOn vst [encodeInt32 CreatedNewConnection, encodeInt32 connId]

1027+ return connId

1028+ -- Error cases

1029+ RemoteEndPointInvalid err ->

1030+ throwIO err

1031+ RemoteEndPointFailed err ->

1032+ throwIO err

1033+ -- Algorithmic errors

1034+ _ ->

1035+ relyViolation (ourEndPoint, theirEndPoint) "createConnectionTo"

1036+ -- TODO: deal with exception case?

1037+ connId <- runScheduledAction (ourEndPoint, theirEndPoint) act

1038+ return (theirEndPoint, connId)

1039+

10261040

10271041 connectFailed :: IOException -> TransportError ConnectErrorCode

10281042 connectFailed = TransportError ConnectFailed . show

@@ -1042,12 +1056,11 @@ setupRemoteEndPoint params (ourEndPoint, theirEndPoint) hints = do

10421056 Right (sock, ConnectionRequestAccepted) -> do

10431057 sendLock <- newMVar ()

10441058 let vst = ValidRemoteEndPointState

1045− { remoteSocket = sock

1046− , remoteSendLock = sendLock

1047− , _remoteOutgoing = 0

1048− , _remoteIncoming = IntSet.empty

1049− , _pendingCtrlRequests = IntMap.empty

1050− , _nextCtrlRequestId = 0

1059+ { remoteSocket = sock

1060+ , remoteSendLock = sendLock

1061+ , _remoteOutgoing = 0

1062+ , _remoteIncoming = Set.empty

1063+ , _remoteMaxIncoming = 0

10511064 }

10521065 resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst)

10531066 return True

@@ -1076,56 +1089,25 @@ setupRemoteEndPoint params (ourEndPoint, theirEndPoint) hints = do

10761089 theirAddress = remoteAddress theirEndPoint

10771090 invalidAddress = TransportError ConnectNotFound

10781091

1079−-- | Do a (blocking) remote request

1080−--

1081−-- May throw IO (user) exception if the local or the remote endpoint is closed,

1082−-- if the send fails, or if the remote endpoint fails before it replies.

1083−doRemoteRequest :: EndPointPair -> ControlHeader -> IO [ByteString]

1084−doRemoteRequest (ourEndPoint, theirEndPoint) header = do

1085− replyMVar <- newEmptyMVar

1086− act <- modifyMVar (remoteState theirEndPoint) $ \st -> case st of

1087− RemoteEndPointValid vst -> do

1088− let reqId = vst ^. nextCtrlRequestId

1089− act <- schedule theirEndPoint $ sendOn vst [encodeInt32 header, encodeInt32 reqId]

1090− return ( RemoteEndPointValid

1091− . (nextCtrlRequestId ^: (+ 1))

1092− . (pendingCtrlRequestsAt reqId ^= Just replyMVar)

1093− $ vst

1094− , act

1095− )

1096− -- Error cases

1097− RemoteEndPointInvalid err ->

1098− throwIO err

1099− RemoteEndPointInit _ _ _ ->

1100− relyViolation (ourEndPoint, theirEndPoint) "doRemoteRequest (init)"

1101− RemoteEndPointClosing _ _ ->

1102− relyViolation (ourEndPoint, theirEndPoint) "doRemoteRequest (closing)"

1103− RemoteEndPointClosed ->

1104− relyViolation (ourEndPoint, theirEndPoint) "doRemoteRequest (closed)"

1105− RemoteEndPointFailed err ->

1106− throwIO err

1107− runScheduledAction (ourEndPoint, theirEndPoint) act

1108− mReply <- takeMVar replyMVar

1109− case mReply of

1110− Left err -> throwIO err

1111− Right reply -> return reply

1112−

11131092-- | Send a CloseSocket request if the remote endpoint is unused

11141093closeIfUnused :: EndPointPair -> IO ()

11151094closeIfUnused (ourEndPoint, theirEndPoint) = do

11161095 mAct <- modifyMVar (remoteState theirEndPoint) $ \st -> case st of

11171096 RemoteEndPointValid vst ->

1118− if vst ^. remoteOutgoing == 0 && IntSet.null (vst ^. remoteIncoming)

1097+ if vst ^. remoteOutgoing == 0 && Set.null (vst ^. remoteIncoming)

11191098 then do

11201099 resolved <- newEmptyMVar

1121− act <- schedule theirEndPoint $ sendOn vst [encodeInt32 CloseSocket]

1100+ act <- schedule theirEndPoint $

1101+ sendOn vst [ encodeInt32 CloseSocket

1102+ , encodeInt32 (vst ^. remoteMaxIncoming)

1103+ ]

11221104 return (RemoteEndPointClosing resolved vst, Just act)

11231105 else

11241106 return (RemoteEndPointValid vst, Nothing)

11251107 _ ->

11261108 return (st, Nothing)

11271109 forM_ mAct $ runScheduledAction (ourEndPoint, theirEndPoint)

1128−

1110+

11291111-- | Reset a remote endpoint if it is in Invalid mode

11301112--

11311113-- If the remote endpoint is currently in broken state, and

@@ -1160,7 +1142,8 @@ connectToSelf :: LocalEndPoint

11601142 -> IO Connection

11611143connectToSelf ourEndPoint = do

11621144 connAlive <- newIORef True -- Protected by the local endpoint lock

1163− connId <- mapIOException connectFailed $ getNextConnectionId ourEndPoint

1145+ lconnId <- mapIOException connectFailed $ getNextConnOutId ourEndPoint

1146+ let connId = createConnectionId heavyweightSelfConnectionId lconnId

11641147 writeChan ourChan $

11651148 ConnectionOpened connId ReliableOrdered (localAddress ourEndPoint)

11661149 return Connection

@@ -1214,21 +1197,33 @@ resolveInit (ourEndPoint, theirEndPoint) newState =

12141197 _ ->

12151198 relyViolation (ourEndPoint, theirEndPoint) "resolveInit"

12161199

1217−-- | Get the next connection ID

1200+-- | Get the next outgoing connection ID

12181201--

12191202-- Throws an IO exception when the endpoint is closed.

1220−getNextConnectionId :: LocalEndPoint -> IO ConnectionId

1221−getNextConnectionId ourEndpoint =

1203+getNextConnOutId :: LocalEndPoint -> IO LightweightConnectionId

1204+getNextConnOutId ourEndpoint =

12221205 modifyMVar (localState ourEndpoint) $ \st -> case st of

12231206 LocalEndPointValid vst -> do

1224− let connId = vst ^. nextConnectionId

1207+ let connId = vst ^. nextConnOutId

12251208 return ( LocalEndPointValid

1226− . (nextConnectionId ^= connId + 1)

1209+ . (nextConnOutId ^= connId + 1)

12271210 $ vst

12281211 , connId)

12291212 LocalEndPointClosed ->

12301213 throwIO $ userError "Local endpoint closed"

12311214

1215+-- | The last outgoing connection ID we created, or zero if we never created any

1216+getLastConnOutId :: LocalEndPoint -> IO LightweightConnectionId

1217+getLastConnOutId ourEndPoint =

1218+ withMVar (localState ourEndPoint) $ \st -> case st of

1219+ LocalEndPointValid vst ->

1220+ let nextId = vst ^. nextConnOutId in

1221+ if nextId == firstNonReservedLightweightConnectionId

1222+ then return 0

1223+ else return (nextId - 1)

1224+ LocalEndPointClosed ->

1225+ throwIO $ userError "Local endpoint closed"

1226+

12321227-- | Create a new local endpoint

12331228--

12341229-- May throw a TransportError NewEndPointErrorCode exception if the transport

@@ -1237,9 +1232,9 @@ createLocalEndPoint :: TCPTransport -> IO LocalEndPoint

12371232createLocalEndPoint transport = do

12381233 chan <- newChan

12391234 state <- newMVar . LocalEndPointValid $ ValidLocalEndPointState

1240− { _nextConnectionId = firstNonReservedConnectionId

1241− , _localConnections = Map.empty

1242− , _nextRemoteId = 0

1235+ { _nextConnOutId = firstNonReservedLightweightConnectionId

1236+ , _nextConnInId = firstNonReservedHeavyweightConnectionId

1237+ , _localConnections = Map.empty

12431238 }

12441239 modifyMVar (transportState transport) $ \st -> case st of

12451240 TransportValid vst -> do

@@ -1321,12 +1316,12 @@ findRemoteEndPoint ourEndPoint theirAddress findOrigin = go

13211316 let theirEndPoint = RemoteEndPoint

13221317 { remoteAddress = theirAddress

13231318 , remoteState = theirState

1324− , remoteId = vst ^. nextRemoteId

1319+ , remoteId = vst ^. nextConnInId

13251320 , remoteScheduled = scheduled

13261321 }

13271322 return ( LocalEndPointValid

13281323 . (localConnectionTo theirAddress ^= Just theirEndPoint)

1329− . (nextRemoteId ^: (+ 1))

1324+ . (nextConnInId ^: (+ 1))

13301325 $ vst

13311326 , (theirEndPoint, True)

13321327 )

@@ -1504,6 +1499,13 @@ decodeEndPointAddress (EndPointAddress bs) =

15041499 _ ->

15051500 Nothing

15061501

1502+-- | Construct a ConnectionId

1503+createConnectionId :: HeavyweightConnectionId

1504+ -> LightweightConnectionId

1505+ -> ConnectionId

1506+createConnectionId hcid lcid =

1507+ (fromIntegral hcid `shiftL` 32) .|. fromIntegral lcid

1508+

15071509-- | @spltiMaxFromEnd p n xs@ splits list @xs@ at elements matching @p@,

15081510-- returning at most @p@ segments -- counting from the /end/

15091511--

@@ -1564,8 +1566,16 @@ internalSocketBetween transport ourAddress theirAddress = do

15641566--------------------------------------------------------------------------------

15651567

15661568-- | We reserve a bunch of connection IDs for control messages

1567−firstNonReservedConnectionId :: ConnectionId

1568−firstNonReservedConnectionId = 1024

1569+firstNonReservedLightweightConnectionId :: LightweightConnectionId

1570+firstNonReservedLightweightConnectionId = 1024

1571+

1572+-- | Self-connection

1573+heavyweightSelfConnectionId :: HeavyweightConnectionId

1574+heavyweightSelfConnectionId = 0

1575+

1576+-- | We reserve some connection IDs for special heavyweight connections

1577+firstNonReservedHeavyweightConnectionId :: HeavyweightConnectionId

1578+firstNonReservedHeavyweightConnectionId = 1

15691579

15701580--------------------------------------------------------------------------------

15711581-- Accessor definitions --

@@ -1577,35 +1587,28 @@ localEndPoints = accessor _localEndPoints (\es st -> st { _localEndPoints = es }

15771587nextEndPointId :: Accessor ValidTransportState EndPointId

15781588nextEndPointId = accessor _nextEndPointId (\eid st -> st { _nextEndPointId = eid })

15791589

1580−nextConnectionId :: Accessor ValidLocalEndPointState ConnectionId

1581−nextConnectionId = accessor _nextConnectionId (\cix st -> st { _nextConnectionId = cix })

1590+nextConnOutId :: Accessor ValidLocalEndPointState LightweightConnectionId

1591+nextConnOutId = accessor _nextConnOutId (\cix st -> st { _nextConnOutId = cix })

15821592

15831593localConnections :: Accessor ValidLocalEndPointState (Map EndPointAddress RemoteEndPoint)

15841594localConnections = accessor _localConnections (\es st -> st { _localConnections = es })

15851595

1586−nextRemoteId :: Accessor ValidLocalEndPointState Int

1587−nextRemoteId = accessor _nextRemoteId (\rid st -> st { _nextRemoteId = rid })

1596+nextConnInId :: Accessor ValidLocalEndPointState HeavyweightConnectionId

1597+nextConnInId = accessor _nextConnInId (\rid st -> st { _nextConnInId = rid })

15881598

15891599remoteOutgoing :: Accessor ValidRemoteEndPointState Int

15901600remoteOutgoing = accessor _remoteOutgoing (\cs conn -> conn { _remoteOutgoing = cs })

15911601

1592−remoteIncoming :: Accessor ValidRemoteEndPointState IntSet

1602+remoteIncoming :: Accessor ValidRemoteEndPointState (Set LightweightConnectionId)

15931603remoteIncoming = accessor _remoteIncoming (\cs conn -> conn { _remoteIncoming = cs })

15941604

1595−pendingCtrlRequests :: Accessor ValidRemoteEndPointState (IntMap (MVar (Either IOException [ByteString])))

1596−pendingCtrlRequests = accessor _pendingCtrlRequests (\rep st -> st { _pendingCtrlRequests = rep })

1597−

1598−nextCtrlRequestId :: Accessor ValidRemoteEndPointState ControlRequestId

1599−nextCtrlRequestId = accessor _nextCtrlRequestId (\cid st -> st { _nextCtrlRequestId = cid })

1605+remoteMaxIncoming :: Accessor ValidRemoteEndPointState LightweightConnectionId

1606+remoteMaxIncoming = accessor _remoteMaxIncoming (\lcid st -> st { _remoteMaxIncoming = lcid })

16001607

16011608localEndPointAt :: EndPointAddress -> Accessor ValidTransportState (Maybe LocalEndPoint)

16021609localEndPointAt addr = localEndPoints >>> DAC.mapMaybe addr

16031610

1604−pendingCtrlRequestsAt :: ControlRequestId -> Accessor ValidRemoteEndPointState (Maybe (MVar (Either IOException [ByteString])))

1605−pendingCtrlRequestsAt ix = pendingCtrlRequests >>> DAC.intMapMaybe (fromIntegral ix)

1606−

1607−localConnectionTo :: EndPointAddress

1608− -> Accessor ValidLocalEndPointState (Maybe RemoteEndPoint)

1611+localConnectionTo :: EndPointAddress -> Accessor ValidLocalEndPointState (Maybe RemoteEndPoint)

16091612localConnectionTo addr = localConnections >>> DAC.mapMaybe addr

16101613

16111614-------------------------------------------------------------------------------

network-transport-tcp/tests/TestTCP.hs+165−171

@@ -1,4 +1,4 @@

1−{-# LANGUAGE RebindableSyntax #-}

1+{-# LANGUAGE RebindableSyntax, TemplateHaskell #-}

22{-# OPTIONS_GHC -fno-warn-orphans #-}

33module Main where

@@ -17,6 +17,7 @@ import Network.Transport.TCP ( createTransport

1717 , TransportInternals(..)

1818 , encodeEndPointAddress

1919 , defaultTCPParameters

20+ , LightweightConnectionId

2021 )

2122import Data.Int (Int32)

2223import Control.Concurrent (threadDelay, killThread)

@@ -28,6 +29,7 @@ import Control.Concurrent.MVar ( MVar

2829 , isEmptyMVar

2930 , newMVar

3031 , modifyMVar

32+ , modifyMVar_

3133 )

3234import Control.Monad (replicateM, guard, forM_, replicateM_, when)

3335import Control.Applicative ((<$>))

@@ -149,21 +151,17 @@ testEarlyDisconnect nextPort = do

149151 _ <- recvWithLength sock

150152 sendMany sock [encodeInt32 ConnectionRequestAccepted]

151153

152− -- Server requests a logical connection

153− RequestConnectionId <- toEnum <$> (recvInt32 sock :: IO Int)

154− reqId <- recvInt32 sock :: IO Int

155− sendMany sock (encodeInt32 ControlResponse : encodeInt32 reqId : prependLength [encodeInt32 (10001 :: Int)])

154+ -- Server opens a logical connection

155+ CreatedNewConnection <- toEnum <$> (recvInt32 sock :: IO Int)

156+ 1024 <- recvInt32 sock :: IO LightweightConnectionId

156157

157158 -- Server sends a message

158− 10001 <- recvInt32 sock :: IO Int

159+ 1024 <- recvInt32 sock :: IO Int

159160 ["ping"] <- recvWithLength sock

160161

161162 -- Reply

162− sendMany sock [encodeInt32 RequestConnectionId, encodeInt32 (10002 :: Int)]

163− ControlResponse <- toEnum <$> (recvInt32 sock :: IO Int)

164− 10002 <- recvInt32 sock :: IO Int

165− [cid] <- recvWithLength sock

166− sendMany sock (cid : prependLength ["pong"])

163+ sendMany sock [encodeInt32 CreatedNewConnection, encodeInt32 (10002 :: Int)]

164+ sendMany sock (encodeInt32 10002 : prependLength ["pong"])

167165

168166 -- Close the socket

169167 N.sClose sock

@@ -171,9 +169,8 @@ testEarlyDisconnect nextPort = do

171169 -- Connect to the server

172170 Right (sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr True Nothing

173171

174− -- Request a new connection, but don't wait for the response

175− let reqId = 0 :: Int32

176− sendMany sock [encodeInt32 RequestConnectionId, encodeInt32 reqId]

172+ -- Open a new connection

173+ sendMany sock [encodeInt32 CreatedNewConnection, encodeInt32 (10003 :: Int)]

177174

178175 -- Close the socket without closing the connection explicitly

179176 -- The server should receive an error event

@@ -260,37 +257,32 @@ testEarlyCloseSocket nextPort = do

260257 _ <- recvWithLength sock

261258 sendMany sock [encodeInt32 ConnectionRequestAccepted]

262259

263− -- Server requests a logical connection

264− RequestConnectionId <- toEnum <$> (recvInt32 sock :: IO Int)

265− reqId <- recvInt32 sock :: IO Int

266− sendMany sock (encodeInt32 ControlResponse : encodeInt32 reqId : prependLength [encodeInt32 (10001 :: Int)])

260+ -- Server opens a logical connection

261+ CreatedNewConnection <- toEnum <$> (recvInt32 sock :: IO Int)

262+ 1024 <- recvInt32 sock :: IO LightweightConnectionId

267263

268264 -- Server sends a message

269− 10001 <- recvInt32 sock :: IO Int

265+ 1024 <- recvInt32 sock :: IO Int

270266 ["ping"] <- recvWithLength sock

271267

272268 -- Reply

273− sendMany sock [encodeInt32 RequestConnectionId, encodeInt32 (10002 :: Int)]

274− ControlResponse <- toEnum <$> (recvInt32 sock :: IO Int)

275− 10002 <- recvInt32 sock :: IO Int

276− [cid] <- recvWithLength sock

277− sendMany sock (cid : prependLength ["pong"])

269+ sendMany sock [encodeInt32 CreatedNewConnection, encodeInt32 (10002 :: Int)]

270+ sendMany sock (encodeInt32 (10002 :: Int) : prependLength ["pong"])

278271

279272 -- Send a CloseSocket even though there are still connections *in both

280273 -- directions*

281− sendMany sock [encodeInt32 CloseSocket]

274+ sendMany sock [encodeInt32 CloseSocket, encodeInt32 (1024 :: Int)]

282275 N.sClose sock

283276

284277 -- Connect to the server

285278 Right (sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr True Nothing

286279

287− -- Request a new connection, but don't wait for the response

288− let reqId = 0 :: Int32

289− sendMany sock [encodeInt32 RequestConnectionId, encodeInt32 reqId]

280+ -- Open a new connection

281+ sendMany sock [encodeInt32 CreatedNewConnection, encodeInt32 (10003 :: Int)]

290282

291283 -- Send a CloseSocket without sending a closeconnecton

292284 -- The server should still receive a ConnectionClosed message

293− sendMany sock [encodeInt32 CloseSocket]

285+ sendMany sock [encodeInt32 CloseSocket, encodeInt32 (0 :: Int)]

294286 N.sClose sock

295287

296288-- | Test the creation of a transport with an invalid address

@@ -329,159 +321,160 @@ testInvalidConnect nextPort = do

329321-- receiving an (already underway) CloseSocket request)

330322testIgnoreCloseSocket :: IO N.ServiceName -> IO ()

331323testIgnoreCloseSocket nextPort = do

332− serverAddr <- newEmptyMVar

333− clientDone <- newEmptyMVar

334− Right transport <- nextPort >>= \port -> createTransport "127.0.0.1" port defaultTCPParameters

335−

336− forkTry $ server transport serverAddr

337− forkTry $ client transport serverAddr clientDone

338−

339− takeMVar clientDone

324+ serverAddr <- newEmptyMVar

325+ clientAddr <- newEmptyMVar

326+ clientDone <- newEmptyMVar

327+ serverDone <- newEmptyMVar

328+ connectionEstablished <- newEmptyMVar

329+ Right transport <- nextPort >>= \port -> createTransport "127.0.0.1" port defaultTCPParameters

340330

341− where

342− server :: Transport -> MVar EndPointAddress -> IO ()

343− server transport serverAddr = do

344− tlog "Server"

345− Right endpoint <- newEndPoint transport

346− putMVar serverAddr (address endpoint)

331+ -- Server

332+ forkTry $ do

333+ tlog "Server"

334+ Right endpoint <- newEndPoint transport

335+ putMVar serverAddr (address endpoint)

347336

348− -- Wait for the client to connect and disconnect

349− tlog "Waiting for ConnectionOpened"

350− ConnectionOpened _ _ _ <- receive endpoint

351− tlog "Waiting for ConnectionClosed"

352− ConnectionClosed _ <- receive endpoint

337+ let ourAddress = address endpoint

338+ theirAddress <- readMVar clientAddr

353339

354− -- At this point the server will have sent a CloseSocket request to the

355− -- client, which however ignores it, instead it requests and closes

356− -- another connection

357− tlog "Waiting for ConnectionOpened"

358− ConnectionOpened _ _ _ <- receive endpoint

359− tlog "Waiting for ConnectionClosed"

360− ConnectionClosed _ <- receive endpoint

340+ -- Wait for the client to set up the TCP connection to us

341+ takeMVar connectionEstablished

361342

362−

363− tlog "Server waiting.."

343+ -- Connect then disconnect to the client

344+ Right conn <- connect endpoint theirAddress ReliableOrdered defaultConnectHints

345+ close conn

364346

365− client :: Transport -> MVar EndPointAddress -> MVar () -> IO ()

366− client transport serverAddr clientDone = do

367− tlog "Client"

368− Right endpoint <- newEndPoint transport

369− let ourAddress = address endpoint

347+ -- At this point the server will have sent a CloseSocket request to the

348+ -- client, which however ignores it, instead it requests and closes

349+ -- another connection

350+ tlog "Waiting for ConnectionOpened"

351+ ConnectionOpened _ _ _ <- receive endpoint

352+ tlog "Waiting for ConnectionClosed"

353+ ConnectionClosed _ <- receive endpoint

370354

371− -- Connect to the server

372− Right (sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr True Nothing

355+ putMVar serverDone ()

373356

374− -- Request a new connection

375− tlog "Requesting connection"

376− let reqId = 0 :: Int32

377− sendMany sock [encodeInt32 RequestConnectionId, encodeInt32 reqId]

378− response <- replicateM 4 $ recvInt32 sock :: IO [Int32]

379−

380− -- Close the connection again

381− tlog "Closing connection"

382− sendMany sock [encodeInt32 CloseConnection, encodeInt32 (response !! 3)]

383−

384− -- Server will now send a CloseSocket request as its refcount reached 0

385− tlog "Waiting for CloseSocket request"

386− CloseSocket <- toEnum <$> recvInt32 sock

387−

388− -- But we ignore it and request another connection

389− tlog "Ignoring it, requesting another connection"

390− let reqId' = 1 :: Int32

391− sendMany sock [encodeInt32 RequestConnectionId, encodeInt32 reqId']

392− response' <- replicateM 4 $ recvInt32 sock :: IO [Int32]

393−

394− -- Close it again

395− tlog "Closing connection"

396− sendMany sock [encodeInt32 CloseConnection, encodeInt32 (response' !! 3)]

397−

398− -- We now get a CloseSocket again, and this time we heed it

399− tlog "Waiting for second CloseSocket request"

400− CloseSocket <- toEnum <$> recvInt32 sock

401−

402− tlog "Closing socket"

403− sendMany sock [encodeInt32 CloseSocket]

404− N.sClose sock

357+ -- Client

358+ forkTry $ do

359+ tlog "Client"

360+ Right endpoint <- newEndPoint transport

361+ putMVar clientAddr (address endpoint)

405362

406− putMVar clientDone ()

363+ let ourAddress = address endpoint

364+ theirAddress <- readMVar serverAddr

365+

366+ -- Connect to the server

367+ Right (sock, ConnectionRequestAccepted) <- socketToEndPoint ourAddress theirAddress True Nothing

368+ putMVar connectionEstablished ()

369+

370+ -- Server connects to us, and then closes the connection

371+ CreatedNewConnection <- toEnum <$> (recvInt32 sock :: IO Int)

372+ 1024 <- recvInt32 sock :: IO LightweightConnectionId

373+

374+ CloseConnection <- toEnum <$> (recvInt32 sock :: IO Int)

375+ 1024 <- recvInt32 sock :: IO LightweightConnectionId

376+

377+ -- Server will now send a CloseSocket request as its refcount reached 0

378+ tlog "Waiting for CloseSocket request"

379+ CloseSocket <- toEnum <$> recvInt32 sock

380+ _ <- recvInt32 sock :: IO LightweightConnectionId

381+

382+ -- But we ignore it and request another connection in the other direction

383+ tlog "Ignoring it, requesting another connection"

384+ sendMany sock [encodeInt32 CreatedNewConnection, encodeInt32 (1024 :: Int)]

385+

386+ -- Close it again

387+ tlog "Closing connection"

388+ sendMany sock [encodeInt32 CloseConnection, encodeInt32 (1024 :: Int)]

389+

390+ -- And close the connection completely

391+ tlog "Closing socket"

392+ sendMany sock [encodeInt32 CloseSocket, encodeInt32 (1024 :: Int)]

393+ N.sClose sock

394+

395+ putMVar clientDone ()

396+

397+ takeMVar clientDone

398+ takeMVar serverDone

407399

408400-- | Like 'testIgnoreSocket', but now the server requests a connection after the

409401-- client closed their connection. In the meantime, the server will have sent a

410402-- CloseSocket request to the client, and must block until the client responds.

411403testBlockAfterCloseSocket :: IO N.ServiceName -> IO ()

412404testBlockAfterCloseSocket nextPort = do

413− serverAddr <- newEmptyMVar

414− clientAddr <- newEmptyMVar

415− clientDone <- newEmptyMVar

416− port <- nextPort

417− Right transport <- createTransport "127.0.0.1" port defaultTCPParameters

418−

419− forkTry $ server transport serverAddr clientAddr

420− forkTry $ client transport serverAddr clientAddr clientDone

421−

422− takeMVar clientDone

405+ serverAddr <- newEmptyMVar

406+ clientAddr <- newEmptyMVar

407+ clientDone <- newEmptyMVar

408+ serverDone <- newEmptyMVar

409+ connectionEstablished <- newEmptyMVar

410+ Right transport <- nextPort >>= \port -> createTransport "127.0.0.1" port defaultTCPParameters

423411

424− where

425− server :: Transport -> MVar EndPointAddress -> MVar EndPointAddress -> IO ()

426− server transport serverAddr clientAddr = do

427− tlog "Server"

428− Right endpoint <- newEndPoint transport

429− putMVar serverAddr (address endpoint)

412+ -- Server

413+ forkTry $ do

414+ tlog "Server"

415+ Right endpoint <- newEndPoint transport

416+ putMVar serverAddr (address endpoint)

430417

431− -- Wait for the client to connect and disconnect

432− tlog "Waiting for ConnectionOpened"

433− ConnectionOpened _ _ _ <- receive endpoint

434− tlog "Waiting for ConnectionClosed"

435− ConnectionClosed _ <- receive endpoint

436−

437− -- At this point the server will have sent a CloseSocket request to the

438− -- client, and must block until the client responds

439− tlog "Server waiting to connect to the client.."

440− Right _ <- readMVar clientAddr >>= \addr -> connect endpoint addr ReliableOrdered defaultConnectHints

441−

442− tlog "Server waiting.."

418+ let ourAddress = address endpoint

419+ theirAddress <- readMVar clientAddr

443420

444− client :: Transport -> MVar EndPointAddress -> MVar EndPointAddress -> MVar () -> IO ()

445− client transport serverAddr clientAddr clientDone = do

446− tlog "Client"

447− Right endpoint <- newEndPoint transport

448− putMVar clientAddr (address endpoint)

449− let ourAddress = address endpoint

421+ -- Wait for the client to set up the TCP connection to us

422+ takeMVar connectionEstablished

450423

451− -- Connect to the server

452− Right (sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr True Nothing

424+ -- Connect then disconnect to the client

425+ Right conn <- connect endpoint theirAddress ReliableOrdered defaultConnectHints

426+ close conn

453427

454− -- Request a new connection

455− tlog "Requesting connection"

456− let reqId = 0 :: Int32

457− sendMany sock [encodeInt32 RequestConnectionId, encodeInt32 reqId]

458− response <- replicateM 4 $ recvInt32 sock :: IO [Int32]

428+ -- At this point the server will have sent a CloseSocket request to the

429+ -- client, and must block until the client responds

430+ Right conn <- connect endpoint theirAddress ReliableOrdered defaultConnectHints

459431

460− -- Close the connection again

461− tlog "Closing connection"

462− sendMany sock [encodeInt32 CloseConnection, encodeInt32 (response !! 3)]

432+ putMVar serverDone ()

463433

464− -- Server will now send a CloseSocket request as its refcount reached 0

465− tlog "Waiting for CloseSocket request"

466− CloseSocket <- toEnum <$> recvInt32 sock

434+ -- Client

435+ forkTry $ do

436+ tlog "Client"

437+ Right endpoint <- newEndPoint transport

438+ putMVar clientAddr (address endpoint)

467439

468− unblocked <- newEmptyMVar

440+ let ourAddress = address endpoint

441+ theirAddress <- readMVar serverAddr

469442

470− -- We should not hear from the server until we unblock him by

471− -- responding to the CloseSocket request (in this case, we

472− -- respond by sending a ConnectionRequest)

473− forkTry $ do

474− recvInt32 sock :: IO Int32

475− isEmptyMVar unblocked >>= (guard . not)

476− putMVar clientDone ()

443+ -- Connect to the server

444+ Right (sock, ConnectionRequestAccepted) <- socketToEndPoint ourAddress theirAddress True Nothing

445+ putMVar connectionEstablished ()

446+

447+ -- Server connects to us, and then closes the connection

448+ CreatedNewConnection <- toEnum <$> (recvInt32 sock :: IO Int)

449+ 1024 <- recvInt32 sock :: IO LightweightConnectionId

477450

478− threadDelay 1000000

451+ CloseConnection <- toEnum <$> (recvInt32 sock :: IO Int)

452+ 1024 <- recvInt32 sock :: IO LightweightConnectionId

479453

480− tlog "Client ignores close socket and sends connection request"

481− tlog "This should unblock the server"

482− putMVar unblocked ()

483− let reqId' = 1 :: Int32

484− sendMany sock [encodeInt32 RequestConnectionId, encodeInt32 reqId']

454+ -- Server will now send a CloseSocket request as its refcount reached 0

455+ tlog "Waiting for CloseSocket request"

456+ CloseSocket <- toEnum <$> recvInt32 sock

457+ _ <- recvInt32 sock :: IO LightweightConnectionId

458+

459+ unblocked <- newMVar False

460+

461+ -- We should not hear from the server until we unblock him by

462+ -- responding to the CloseSocket request (in this case, we

463+ -- respond by sending a ConnectionRequest)

464+ forkTry $ do

465+ recvInt32 sock :: IO Int32

466+ readMVar unblocked >>= guard

467+ putMVar clientDone ()

468+

469+ threadDelay 1000000

470+

471+ tlog "Client ignores close socket and sends connection request"

472+ tlog "This should unblock the server"

473+ modifyMVar_ unblocked $ \_ -> return True

474+ sendMany sock [encodeInt32 CreatedNewConnection, encodeInt32 (1024 :: Int)]

475+

476+ takeMVar clientDone

477+ takeMVar serverDone

485478

486479-- | Test what happens when a remote endpoint sends a connection request to our

487480-- transport for an endpoint it already has a connection to

@@ -592,16 +585,16 @@ testReconnect nextPort = do

592585

593586 when (count > 0) $ do

594587 -- Client requests a logical connection

595− Right RequestConnectionId <- tryIO $ toEnum <$> (recvInt32 sock :: IO Int)

596− Right reqId <- tryIO $ (recvInt32 sock :: IO Int)

597− Right () <- tryIO $ sendMany sock (encodeInt32 ControlResponse : encodeInt32 reqId : prependLength [encodeInt32 (10001 :: Int)])

588+ Right CreatedNewConnection <- tryIO $ toEnum <$> (recvInt32 sock :: IO Int)

589+ connId <- recvInt32 sock :: IO LightweightConnectionId

598590 return ()

599591

600− when (count > 1) $ do

601− -- Client sends a message

602− Right 10001 <- tryIO $ (recvInt32 sock :: IO Int)

603− Right ["ping"] <- tryIO $ recvWithLength sock

604− putMVar serverDone ()

592+ when (count > 1) $ do

593+ -- Client sends a message

594+ Right connId' <- tryIO $ (recvInt32 sock :: IO LightweightConnectionId)

595+ True <- return $ connId == connId'

596+ Right ["ping"] <- tryIO $ recvWithLength sock

597+ putMVar serverDone ()

605598

606599 Right () <- tryIO $ N.sClose sock

607600 return ()

@@ -615,7 +608,7 @@ testReconnect nextPort = do

615608 let theirAddr = encodeEndPointAddress "127.0.0.1" serverPort 0

616609

617610 -- The first attempt will fail because no endpoint is yet set up

618− -- Left (TransportError ConnectNotFound _) <- connect endpoint theirAddr ReliableOrdered defaultConnectHints

611+ Left (TransportError ConnectNotFound _) <- connect endpoint theirAddr ReliableOrdered defaultConnectHints

619612 putMVar firstAttempt ()

620613

621614 -- The second attempt will fail because the server closes the socket before we can request a connection

@@ -633,6 +626,7 @@ testReconnect nextPort = do

633626 Right conn1 <- connect endpoint theirAddr ReliableOrdered defaultConnectHints

634627

635628 -- But a send will fail because the server has closed the connection again

629+ threadDelay 100000

636630 Left (TransportError SendFailed _) <- send conn1 ["ping"]

637631 ErrorEvent (TransportError (EventConnectionLost _) _) <- receive endpoint

638632

@@ -667,11 +661,11 @@ testUnidirectionalError nextPort = do

667661 _ <- recvWithLength sock

668662 () <- sendMany sock [encodeInt32 ConnectionRequestAccepted]

669663

670− RequestConnectionId <- toEnum <$> (recvInt32 sock :: IO Int)

671− reqId <- recvInt32 sock :: IO Int

672− sendMany sock (encodeInt32 ControlResponse : encodeInt32 reqId : prependLength [encodeInt32 (10001 :: Int)])

664+ CreatedNewConnection <- toEnum <$> (recvInt32 sock :: IO Int)

665+ connId <- recvInt32 sock :: IO LightweightConnectionId

673666

674− 10001 <- recvInt32 sock :: IO Int

667+ connId' <- recvInt32 sock :: IO LightweightConnectionId

668+ True <- return $ connId == connId'

675669 ["ping"] <- recvWithLength sock

676670 putMVar serverGotPing ()

677671

@@ -777,7 +771,7 @@ main = do

777771 , ("EarlyCloseSocket", testEarlyCloseSocket nextPort)

778772 , ("IgnoreCloseSocket", testIgnoreCloseSocket nextPort)

779773 , ("BlockAfterCloseSocket", testBlockAfterCloseSocket nextPort)

780− , ("TestUnnecessaryConnect", testUnnecessaryConnect nextPort 10)

774+ , ("UnnecessaryConnect", testUnnecessaryConnect nextPort 10)

781775 , ("InvalidAddress", testInvalidAddress nextPort)

782776 , ("InvalidConnect", testInvalidConnect nextPort)

783777 , ("Many", testMany nextPort)

network-transport-tests/src/Network/Transport/Tests.hs+4−2

@@ -691,12 +691,14 @@ testCloseTransport newTransport = do

691691 send conn ["pong"]

692692

693693 -- Client now closes down its transport. We should receive connection closed messages (we don't know the precise order, however)

694− evs <- replicateM 3 $ receive endpoint

694+ -- TODO: should we get an EventConnectionLost for theirAddr1? We have no outgoing connections

695+ evs <- replicateM 4 $ receive endpoint

695696 let expected = [ ConnectionClosed cid1

696697 , ConnectionClosed cid2

698+ , ErrorEvent (TransportError (EventConnectionLost theirAddr1) "")

697699 , ErrorEvent (TransportError (EventConnectionLost theirAddr2) "")

698700 ]

699− True <- return $ any (== expected) (permutations evs)

701+ True <- return $ expected `elem` permutations evs

700702

701703 -- An attempt to send to the endpoint should now fail

702704 Left (TransportError SendFailed _) <- send conn ["pong2"]

network-transport-tests/src/Network/Transport/Tests/Traced.hs+11−1

@@ -66,7 +66,8 @@ import Control.Applicative ((<$>))

6666import Data.Typeable (Typeable)

6767import Data.Maybe (catMaybes)

6868import Data.ByteString (ByteString)

69−import Data.Int (Int32)

69+import Data.Int (Int32, Int64)

70+import Data.Word (Word32, Word64)

7071import Control.Concurrent.MVar (MVar)

7172

7273--------------------------------------------------------------------------------

@@ -152,6 +153,15 @@ instance Traceable Int where

152153instance Traceable Int32 where

153154 trace = traceShow

154155

156+instance Traceable Int64 where

157+ trace = traceShow

158+

159+instance Traceable Word32 where

160+ trace = traceShow

161+

162+instance Traceable Word64 where

163+ trace = traceShow

164+

155165instance Traceable Bool where

156166 trace = const Nothing

157167

network-transport/ChangeLog+1

@@ -1,6 +1,7 @@

11<<date> Edsko de Vries <edsko@well-typed.com> 0.3.0

33* Clarify disconnection

4+* Require that 'connect' be "as asynchronous as possible"

562012-07-16 Edsko de Vries <edsko@well-typed.com> 0.2.0.2

network-transport/src/Network/Transport.hs+9−1

@@ -30,6 +30,7 @@ import Control.Exception (Exception)

3030import Control.Applicative ((<$>))

3131import Data.Typeable (Typeable)

3232import Data.Binary (Binary(get, put))

33+import Data.Word (Word64)

3334

3435--------------------------------------------------------------------------------

3536-- Main API --

@@ -51,6 +52,11 @@ data EndPoint = EndPoint {

5152 -- | EndPointAddress of the endpoint.

5253 , address :: EndPointAddress

5354 -- | Create a new lightweight connection.

55+ --

56+ -- 'connect' should be as asynchronous as possible; for instance, in

57+ -- Transport implementations based on some heavy-weight underlying network

58+ -- protocol (TCP, ssh), a call to 'connect' should be asynchronous when a

59+ -- heavyweight connection has already been established.

5460 , connect :: EndPointAddress -> Reliability -> ConnectHints -> IO (Either (TransportError ConnectErrorCode) Connection)

5561 -- | Create a new multicast group.

5662 , newMulticastGroup :: IO (Either (TransportError NewMulticastGroupErrorCode) MulticastGroup)

@@ -80,6 +86,8 @@ data Event =

8086 -- | Connection closed

8187 | ConnectionClosed ConnectionId

8288 -- | Connection opened

89+ --

90+ -- 'ConnectionId's need not be allocated contiguously.

8391 | ConnectionOpened ConnectionId Reliability EndPointAddress

8492 -- | Received multicast

8593 | ReceivedMulticast MulticastAddress [ByteString]

@@ -90,7 +98,7 @@ data Event =

9098 deriving (Show, Eq)

9199

92100-- | Connection data ConnectHintsIDs enable receivers to distinguish one connection from another.

93−type ConnectionId = Int

101+type ConnectionId = Word64

94102

95103-- | Reliability guarantees of a connection.

96104data Reliability =