f46269a · distributed-process

git.iko.soy/distributed-process

on master help sign in

Switch to asynchronous connect

f46269aEdsko de Vries14 years ago

network-transport-tcp/ChangeLog+1

compare with master

unifiedsplit

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

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

3* Implement new disconnection semantics

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

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

62012-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

23 , decodeEndPointAddress

24 , ControlHeader(..)

25 , ConnectionRequestResponse(..)

26− , firstNonReservedConnectionId

26+ , firstNonReservedLightweightConnectionId

27+ , firstNonReservedHeavyweightConnectionId

27 , socketToEndPoint

28 , socketToEndPoint

29+ , LightweightConnectionId

28 -- * Design notes

30 -- * Design notes

29 -- $design

31 -- $design

30 ) where

32 ) where

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

45 )

47 )

46import Network.Transport.Internal

48import Network.Transport.Internal

47 ( encodeInt32

49 ( encodeInt32

48− , decodeInt32

49 , prependLength

50 , prependLength

50 , mapIOException

51 , mapIOException

51 , tryIO

52 , tryIO

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

79 , modifyMVar

80 , modifyMVar

80 , modifyMVar_

81 , modifyMVar_

81 , readMVar

82 , readMVar

82− , takeMVar

83 , putMVar

84 , newEmptyMVar

85 , withMVar

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

103import Data.ByteString (ByteString)

104import qualified Data.ByteString as BS (concat)

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

106−import Data.Int (Int32)

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

107−import Data.IntMap (IntMap)

107+import Data.Word (Word32)

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

109−import Data.IntSet (IntSet)

108+import Data.Set (Set)

110−import qualified Data.IntSet as IntSet

109+import qualified Data.Set as Set

111 ( empty

110 ( empty

112 , insert

111 , insert

113 , elems

112 , elems

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

119import Data.Map (Map)

118import Data.Map (Map)

120import qualified Data.Map as Map (empty)

119import qualified Data.Map as Map (empty)

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

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

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

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

123import Data.Foldable (forM_, mapM_)

122import Data.Foldable (forM_, mapM_)

124

123

125-- $design

124-- $design

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

256 | LocalEndPointClosed

255 | LocalEndPointClosed

257

256

258data ValidLocalEndPointState = ValidLocalEndPointState

257data ValidLocalEndPointState = ValidLocalEndPointState

259− { _nextConnectionId :: !ConnectionId

258+ { _nextConnOutId :: !LightweightConnectionId

259+ , _nextConnInId :: !HeavyweightConnectionId

260 , _localConnections :: !(Map EndPointAddress RemoteEndPoint)

261− , _nextRemoteId :: !Int

262 }

261 }

263

262

264-- REMOTE ENDPOINTS

263-- REMOTE ENDPOINTS

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

278-- \-------+----------+--------> Failed

277-- \-------+----------+--------> Failed

279--

278--

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

279-- 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

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

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

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

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

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

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

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

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

286--

285--

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

286-- 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.

289--

288--

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

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

291--

290--

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

346data RemoteEndPoint = RemoteEndPoint

345data RemoteEndPoint = RemoteEndPoint

347 { remoteAddress :: !EndPointAddress

346 { remoteAddress :: !EndPointAddress

348 , remoteState :: !(MVar RemoteState)

347 , remoteState :: !(MVar RemoteState)

349− , remoteId :: !Int

348+ , remoteId :: !HeavyweightConnectionId

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

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

351 }

350 }

352

351

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

368 -- using a closeTransport or closeEndPoint API call

367 -- using a closeTransport or closeEndPoint API call

369 | RemoteEndPointFailed !IOException

368 | RemoteEndPointFailed !IOException

370

369

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'

371data ValidRemoteEndPointState = ValidRemoteEndPointState

375data ValidRemoteEndPointState = ValidRemoteEndPointState

372 { _remoteOutgoing :: !Int

376 { _remoteOutgoing :: !Int

373− , _remoteIncoming :: !IntSet

377+ , _remoteIncoming :: !(Set LightweightConnectionId)

378+ , _remoteMaxIncoming :: !LightweightConnectionId

374 , remoteSocket :: !N.Socket

379 , remoteSocket :: !N.Socket

375 , remoteSendLock :: !(MVar ())

380 , remoteSendLock :: !(MVar ())

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

377− , _nextCtrlRequestId :: !ControlRequestId

378 }

381 }

379

382

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

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

381−type EndPointId = Int32

384+type EndPointId = Word32

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

388

385

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

386-- | 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

391

400

392-- | Control headers

401-- | Control headers

393data ControlHeader =

402data ControlHeader =

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

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

395− RequestConnectionId

404+ CreatedNewConnection

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

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

397 | CloseConnection

406 | CloseConnection

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

399− | ControlResponse

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

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

401 | CloseSocket

408 | CloseSocket

402 deriving (Enum, Bounded, Show)

409 deriving (Enum, Bounded, Show)

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

558 else do

565 else do

559 resetIfBroken ourEndPoint theirAddress

566 resetIfBroken ourEndPoint theirAddress

560 (theirEndPoint, connId) <-

567 (theirEndPoint, connId) <-

561− requestConnectionTo params ourEndPoint theirAddress hints

568+ createConnectionTo params ourEndPoint theirAddress hints

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

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

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

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

564 connAlive <- newIORef True

571 connAlive <- newIORef True

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

568 }

575 }

569

576

570-- | Close a connection

577-- | Close a connection

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

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

572apiClose (ourEndPoint, theirEndPoint) connId connAlive =

579apiClose (ourEndPoint, theirEndPoint) connId connAlive =

573 void . tryIO . asyncWhenCancelled return $ do

580 void . tryIO . asyncWhenCancelled return $ do

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

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

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

593

600

594

601

595-- | Send data across a connection

602-- | Send data across a connection

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

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

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

604+ -> LightweightConnectionId -- ^ Connection ID

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

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

599− -> [ByteString] -- ^ Payload

606+ -> [ByteString] -- ^ Payload

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

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

601apiSend (ourEndPoint, theirEndPoint) connId connAlive payload =

608apiSend (ourEndPoint, theirEndPoint) connId connAlive payload =

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

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

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

665 return (closed, Nothing)

672 return (closed, Nothing)

666 RemoteEndPointValid vst -> do

673 RemoteEndPointValid vst -> do

667 act <- schedule theirEndPoint $ do

674 act <- schedule theirEndPoint $ do

668− tryIO $ sendOn vst [encodeInt32 CloseSocket]

675+ tryIO $ sendOn vst [ encodeInt32 CloseSocket

676+ , encodeInt32 (vst ^. remoteMaxIncoming)

677+ ]

669 tryCloseSocket (remoteSocket vst)

678 tryCloseSocket (remoteSocket vst)

670 return (closed, Just act)

679 return (closed, Just act)

671 RemoteEndPointClosing resolved vst -> do

680 RemoteEndPointClosing resolved vst -> do

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

729 else do

738 else do

730 sendLock <- newMVar ()

739 sendLock <- newMVar ()

731 let vst = ValidRemoteEndPointState

740 let vst = ValidRemoteEndPointState

732− { remoteSocket = sock

741+ { remoteSocket = sock

733− , remoteSendLock = sendLock

742+ , remoteSendLock = sendLock

734− , _remoteOutgoing = 0

743+ , _remoteOutgoing = 0

735− , _remoteIncoming = IntSet.empty

744+ , _remoteIncoming = Set.empty

736− , _pendingCtrlRequests = IntMap.empty

745+ , _remoteMaxIncoming = 0

737− , _nextCtrlRequestId = 0

738 }

746 }

739 sendMany sock [encodeInt32 ConnectionRequestAccepted]

747 sendMany sock [encodeInt32 ConnectionRequestAccepted]

740 resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst)

748 resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst)

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

790 -- exception thrown by 'recv'.

798 -- exception thrown by 'recv'.

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

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

792 go sock = do

800 go sock = do

793− connId <- recvInt32 sock

801+ lcid <- recvInt32 sock :: IO LightweightConnectionId

794− if connId >= firstNonReservedConnectionId

802+ if lcid >= firstNonReservedLightweightConnectionId

795 then do

803 then do

796− readMessage sock connId

804+ readMessage sock lcid

797 go sock

805 go sock

798 else

806 else

799− case tryToEnum (fromIntegral connId) of

807+ case tryToEnum (fromIntegral lcid) of

800− Just RequestConnectionId -> do

808+ Just CreatedNewConnection -> do

801− recvInt32 sock >>= createNewConnection

809+ recvInt32 sock >>= createdNewConnection

802 go sock

810 go sock

803− Just ControlResponse -> do

804− recvInt32 sock >>= readControlResponse sock

805− go sock

806 Just CloseConnection -> do

811 Just CloseConnection -> do

807 recvInt32 sock >>= closeConnection

812 recvInt32 sock >>= closeConnection

808 go sock

813 go sock

809 Just CloseSocket -> do

814 Just CloseSocket -> do

810− didClose <- closeSocket sock

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

811 unless didClose $ go sock

816 unless didClose $ go sock

812 Nothing ->

817 Nothing ->

813 throwIO $ userError "Invalid control request"

818 throwIO $ userError "Invalid control request"

814

819

815 -- Create a new connection

820 -- Create a new connection

816− createNewConnection :: ControlRequestId -> IO ()

821+ createdNewConnection :: LightweightConnectionId -> IO ()

817− createNewConnection reqId = do

822+ createdNewConnection lcid = do

818− newId <- getNextConnectionId ourEndPoint

819 modifyMVar_ theirState $ \st -> do

823 modifyMVar_ theirState $ \st -> do

820 vst <- case st of

824 vst <- case st of

821 RemoteEndPointInvalid _ ->

825 RemoteEndPointInvalid _ ->

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

825 relyViolation (ourEndPoint, theirEndPoint)

829 relyViolation (ourEndPoint, theirEndPoint)

826 "handleIncomingMessages:createNewConnection (init)"

830 "handleIncomingMessages:createNewConnection (init)"

827 RemoteEndPointValid vst ->

831 RemoteEndPointValid vst ->

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

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

833+ $ (remoteMaxIncoming ^= lcid)

834+ vst

835+ )

829 RemoteEndPointClosing resolved vst -> do

836 RemoteEndPointClosing resolved vst -> do

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

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

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

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

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

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

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

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

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

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

835− -- to RemoteEndPointValid

842+ -- RemoteEndPointValid

836 putMVar resolved ()

843 putMVar resolved ()

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

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

845+ . (remoteMaxIncoming ^= lcid)

846+ $ vst

847+ )

838 RemoteEndPointFailed err ->

848 RemoteEndPointFailed err ->

839 throwIO err

849 throwIO err

840 RemoteEndPointClosed ->

850 RemoteEndPointClosed ->

841 relyViolation (ourEndPoint, theirEndPoint)

851 relyViolation (ourEndPoint, theirEndPoint)

842 "createNewConnection (closed)"

852 "createNewConnection (closed)"

843− sendOn vst ( encodeInt32 ControlResponse

844− : encodeInt32 reqId

845− : prependLength [encodeInt32 newId]

846− )

847 return (RemoteEndPointValid vst)

853 return (RemoteEndPointValid vst)

848− writeChan ourChannel (ConnectionOpened newId ReliableOrdered theirAddr)

854+ writeChan ourChannel (ConnectionOpened (connId lcid) 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)

879

855

880 -- Close a connection

856 -- Close a connection

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

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

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

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

883− closeConnection :: ConnectionId -> IO ()

859+ closeConnection :: LightweightConnectionId -> IO ()

884− closeConnection cid = do

860+ closeConnection lcid = do

885 modifyMVar_ theirState $ \st -> case st of

861 modifyMVar_ theirState $ \st -> case st of

886 RemoteEndPointInvalid _ ->

862 RemoteEndPointInvalid _ ->

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

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

888 RemoteEndPointInit _ _ _ ->

864 RemoteEndPointInit _ _ _ ->

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

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

890 RemoteEndPointValid vst -> do

866 RemoteEndPointValid vst -> do

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

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

892 throwIO $ userError "Invalid CloseConnection"

868 throwIO $ userError "Invalid CloseConnection"

893 return ( RemoteEndPointValid

869 return ( RemoteEndPointValid

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

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

895 $ vst

871 $ vst

896 )

872 )

897 RemoteEndPointClosing _ _ ->

873 RemoteEndPointClosing _ _ ->

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

903 throwIO err

879 throwIO err

904 RemoteEndPointClosed ->

880 RemoteEndPointClosed ->

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

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

906− writeChan ourChannel (ConnectionClosed cid)

882+ writeChan ourChannel (ConnectionClosed (connId lcid))

907− closeIfUnused (ourEndPoint, theirEndPoint)

908

883

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

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

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

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

911− closeSocket sock = do

886+ closeSocket sock lastReceivedId = do

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

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

888+ lastSentId <- getLastConnOutId ourEndPoint

913 case st of

889 case st of

914 RemoteEndPointInvalid _ ->

890 RemoteEndPointInvalid _ ->

915 relyViolation (ourEndPoint, theirEndPoint)

891 relyViolation (ourEndPoint, theirEndPoint)

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

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

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

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

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

923 -- now properly closed

899 -- now properly closed

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

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

925− writeChan ourChannel . ConnectionClosed

901+ writeChan ourChannel . ConnectionClosed . connId

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

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

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

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.

928− if vst' ^. remoteOutgoing == 0

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

929− then do

911+ then

912+ return (RemoteEndPointValid vst', Nothing)

913+ else do

930 removeRemoteEndPoint (ourEndPoint, theirEndPoint)

914 removeRemoteEndPoint (ourEndPoint, theirEndPoint)

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

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

932 act <- schedule theirEndPoint $ do

916 act <- schedule theirEndPoint $ do

933− tryIO $ sendOn vst' [encodeInt32 CloseSocket]

917+ tryIO $ sendOn vst' [ encodeInt32 CloseSocket

918+ , encodeInt32 (vst ^. remoteMaxIncoming)

919+ ]

934 tryCloseSocket sock

920 tryCloseSocket sock

935 return (RemoteEndPointClosed, Just act)

921 return (RemoteEndPointClosed, Just act)

936− else

937− return (RemoteEndPointValid vst', Nothing)

938− RemoteEndPointClosing resolved _ -> do

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

939− removeRemoteEndPoint (ourEndPoint, theirEndPoint)

934+ removeRemoteEndPoint (ourEndPoint, theirEndPoint)

940− act <- schedule theirEndPoint $ tryCloseSocket sock

935+ act <- schedule theirEndPoint $ tryCloseSocket sock

941− putMVar resolved ()

936+ putMVar resolved ()

942− return (RemoteEndPointClosed, Just act)

937+ return (RemoteEndPointClosed, Just act)

943 RemoteEndPointFailed err ->

938 RemoteEndPointFailed err ->

944 throwIO err

939 throwIO err

945 RemoteEndPointClosed ->

940 RemoteEndPointClosed ->

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

950 Just act -> do

945 Just act -> do

951 runScheduledAction (ourEndPoint, theirEndPoint) act

946 runScheduledAction (ourEndPoint, theirEndPoint) act

952 return True

947 return True

953−

948+

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

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

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

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

956 -- overhead

951 -- overhead

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

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

958− readMessage sock connId =

953+ readMessage sock lcid =

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

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

960

955

961 -- Arguments

956 -- Arguments

962 ourChannel = localChannel ourEndPoint

957 ourChannel = localChannel ourEndPoint

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

975 RemoteEndPointInit _ _ _ ->

970 RemoteEndPointInit _ _ _ ->

976 relyViolation (ourEndPoint, theirEndPoint)

971 relyViolation (ourEndPoint, theirEndPoint)

977 "handleIncomingMessages:prematureExit"

972 "handleIncomingMessages:prematureExit"

978− RemoteEndPointValid vst -> do

973+ RemoteEndPointValid _ -> do

979 let code = EventConnectionLost (remoteAddress theirEndPoint)

974 let code = EventConnectionLost (remoteAddress theirEndPoint)

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

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

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

982 return (RemoteEndPointFailed err)

976 return (RemoteEndPointFailed err)

983 RemoteEndPointClosing resolved _ -> do

977 RemoteEndPointClosing resolved _ -> do

984 putMVar resolved ()

978 putMVar resolved ()

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

989 RemoteEndPointFailed err' ->

983 RemoteEndPointFailed err' ->

990 return (RemoteEndPointFailed err')

984 return (RemoteEndPointFailed err')

991

985

986+ -- Construct a connection ID

987+ connId :: LightweightConnectionId -> ConnectionId

988+ connId = createConnectionId (remoteId theirEndPoint)

989+

992--------------------------------------------------------------------------------

990--------------------------------------------------------------------------------

993-- Uninterruptable auxiliary functions --

991-- Uninterruptable auxiliary functions --

994-- --

992-- --

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

996-- be killed.

994-- be killed.

997--------------------------------------------------------------------------------

995--------------------------------------------------------------------------------

998

996

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

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

1000--

998--

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

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

1003−-- additionally block until that is resolved.

1000+-- block until that is resolved.

1004--

1001--

1005-- May throw a TransportError ConnectErrorCode exception.

1002-- May throw a TransportError ConnectErrorCode exception.

1006−requestConnectionTo :: TCPParameters

1003+createConnectionTo :: TCPParameters

1007 -> LocalEndPoint

1004 -> LocalEndPoint

1008 -> EndPointAddress

1005 -> EndPointAddress

1009 -> ConnectHints

1006 -> ConnectHints

1010− -> IO (RemoteEndPoint, ConnectionId)

1007+ -> IO (RemoteEndPoint, LightweightConnectionId)

1011−requestConnectionTo params ourEndPoint theirAddress hints = go

1008+createConnectionTo params ourEndPoint theirAddress hints = go

1012 where

1009 where

1013 go = do

1010 go = do

1014 (theirEndPoint, isNew) <- mapIOException connectFailed $

1011 (theirEndPoint, isNew) <- mapIOException connectFailed $

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

1020 setupRemoteEndPoint params (ourEndPoint, theirEndPoint) hints

1017 setupRemoteEndPoint params (ourEndPoint, theirEndPoint) hints

1021 go

1018 go

1022 else do

1019 else do

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

1023− reply <- mapIOException connectFailed $

1021+ mapIOException connectFailed $ do

1024− doRemoteRequest (ourEndPoint, theirEndPoint) RequestConnectionId

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

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

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+

1026

1040

1027 connectFailed :: IOException -> TransportError ConnectErrorCode

1041 connectFailed :: IOException -> TransportError ConnectErrorCode

1028 connectFailed = TransportError ConnectFailed . show

1042 connectFailed = TransportError ConnectFailed . show

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

1042 Right (sock, ConnectionRequestAccepted) -> do

1056 Right (sock, ConnectionRequestAccepted) -> do

1043 sendLock <- newMVar ()

1057 sendLock <- newMVar ()

1044 let vst = ValidRemoteEndPointState

1058 let vst = ValidRemoteEndPointState

1045− { remoteSocket = sock

1059+ { remoteSocket = sock

1046− , remoteSendLock = sendLock

1060+ , remoteSendLock = sendLock

1047− , _remoteOutgoing = 0

1061+ , _remoteOutgoing = 0

1048− , _remoteIncoming = IntSet.empty

1062+ , _remoteIncoming = Set.empty

1049− , _pendingCtrlRequests = IntMap.empty

1063+ , _remoteMaxIncoming = 0

1050− , _nextCtrlRequestId = 0

1051 }

1064 }

1052 resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst)

1065 resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst)

1053 return True

1066 return True

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

1076 theirAddress = remoteAddress theirEndPoint

1089 theirAddress = remoteAddress theirEndPoint

1077 invalidAddress = TransportError ConnectNotFound

1090 invalidAddress = TransportError ConnectNotFound

1078

1091

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−

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

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

1114closeIfUnused :: EndPointPair -> IO ()

1093closeIfUnused :: EndPointPair -> IO ()

1115closeIfUnused (ourEndPoint, theirEndPoint) = do

1094closeIfUnused (ourEndPoint, theirEndPoint) = do

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

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

1117 RemoteEndPointValid vst ->

1096 RemoteEndPointValid vst ->

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

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

1119 then do

1098 then do

1120 resolved <- newEmptyMVar

1099 resolved <- newEmptyMVar

1100+ act <- schedule theirEndPoint $

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

1101+ sendOn vst [ encodeInt32 CloseSocket

1102+ , encodeInt32 (vst ^. remoteMaxIncoming)

1103+ ]

1122 return (RemoteEndPointClosing resolved vst, Just act)

1104 return (RemoteEndPointClosing resolved vst, Just act)

1123 else

1105 else

1124 return (RemoteEndPointValid vst, Nothing)

1106 return (RemoteEndPointValid vst, Nothing)

1125 _ ->

1107 _ ->

1126 return (st, Nothing)

1108 return (st, Nothing)

1127 forM_ mAct $ runScheduledAction (ourEndPoint, theirEndPoint)

1109 forM_ mAct $ runScheduledAction (ourEndPoint, theirEndPoint)

1128−

1110+

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

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

1130--

1112--

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

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

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

1160 -> IO Connection

1142 -> IO Connection

1161connectToSelf ourEndPoint = do

1143connectToSelf ourEndPoint = do

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

1144 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

1164 writeChan ourChan $

1147 writeChan ourChan $

1165 ConnectionOpened connId ReliableOrdered (localAddress ourEndPoint)

1148 ConnectionOpened connId ReliableOrdered (localAddress ourEndPoint)

1166 return Connection

1149 return Connection

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

1214 _ ->

1197 _ ->

1215 relyViolation (ourEndPoint, theirEndPoint) "resolveInit"

1198 relyViolation (ourEndPoint, theirEndPoint) "resolveInit"

1216

1199

1217−-- | Get the next connection ID

1200+-- | Get the next outgoing connection ID

1218--

1201--

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

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

1220−getNextConnectionId :: LocalEndPoint -> IO ConnectionId

1203+getNextConnOutId :: LocalEndPoint -> IO LightweightConnectionId

1221−getNextConnectionId ourEndpoint =

1204+getNextConnOutId ourEndpoint =

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

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

1223 LocalEndPointValid vst -> do

1206 LocalEndPointValid vst -> do

1224− let connId = vst ^. nextConnectionId

1207+ let connId = vst ^. nextConnOutId

1225 return ( LocalEndPointValid

1208 return ( LocalEndPointValid

1226− . (nextConnectionId ^= connId + 1)

1209+ . (nextConnOutId ^= connId + 1)

1227 $ vst

1210 $ vst

1228 , connId)

1211 , connId)

1229 LocalEndPointClosed ->

1212 LocalEndPointClosed ->

1230 throwIO $ userError "Local endpoint closed"

1213 throwIO $ userError "Local endpoint closed"

1231

1214

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+

1232-- | Create a new local endpoint

1227-- | Create a new local endpoint

1233--

1228--

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

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

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

1237createLocalEndPoint transport = do

1232createLocalEndPoint transport = do

1238 chan <- newChan

1233 chan <- newChan

1239 state <- newMVar . LocalEndPointValid $ ValidLocalEndPointState

1234 state <- newMVar . LocalEndPointValid $ ValidLocalEndPointState

1240− { _nextConnectionId = firstNonReservedConnectionId

1235+ { _nextConnOutId = firstNonReservedLightweightConnectionId

1236+ , _nextConnInId = firstNonReservedHeavyweightConnectionId

1241− , _localConnections = Map.empty

1237+ , _localConnections = Map.empty

1242− , _nextRemoteId = 0

1243 }

1238 }

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

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

1245 TransportValid vst -> do

1240 TransportValid vst -> do

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

1321 let theirEndPoint = RemoteEndPoint

1316 let theirEndPoint = RemoteEndPoint

1322 { remoteAddress = theirAddress

1317 { remoteAddress = theirAddress

1323 , remoteState = theirState

1318 , remoteState = theirState

1324− , remoteId = vst ^. nextRemoteId

1319+ , remoteId = vst ^. nextConnInId

1325 , remoteScheduled = scheduled

1320 , remoteScheduled = scheduled

1326 }

1321 }

1327 return ( LocalEndPointValid

1322 return ( LocalEndPointValid

1328 . (localConnectionTo theirAddress ^= Just theirEndPoint)

1323 . (localConnectionTo theirAddress ^= Just theirEndPoint)

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

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

1330 $ vst

1325 $ vst

1331 , (theirEndPoint, True)

1326 , (theirEndPoint, True)

1332 )

1327 )

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

1504 _ ->

1499 _ ->

1505 Nothing

1500 Nothing

1506

1501

1502+-- | Construct a ConnectionId

1503+createConnectionId :: HeavyweightConnectionId

1504+ -> LightweightConnectionId

1505+ -> ConnectionId

1506+createConnectionId hcid lcid =

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

1508+

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

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

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

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

1509--

1511--

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

1564--------------------------------------------------------------------------------

1566--------------------------------------------------------------------------------

1565

1567

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

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

1567−firstNonReservedConnectionId :: ConnectionId

1569+firstNonReservedLightweightConnectionId :: LightweightConnectionId

1568−firstNonReservedConnectionId = 1024

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

1569

1579

1570--------------------------------------------------------------------------------

1580--------------------------------------------------------------------------------

1571-- Accessor definitions --

1581-- Accessor definitions --

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

1577nextEndPointId :: Accessor ValidTransportState EndPointId

1587nextEndPointId :: Accessor ValidTransportState EndPointId

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

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

1579

1589

1580−nextConnectionId :: Accessor ValidLocalEndPointState ConnectionId

1590+nextConnOutId :: Accessor ValidLocalEndPointState LightweightConnectionId

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

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

1582

1592

1583localConnections :: Accessor ValidLocalEndPointState (Map EndPointAddress RemoteEndPoint)

1593localConnections :: Accessor ValidLocalEndPointState (Map EndPointAddress RemoteEndPoint)

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

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

1585

1595

1586−nextRemoteId :: Accessor ValidLocalEndPointState Int

1596+nextConnInId :: Accessor ValidLocalEndPointState HeavyweightConnectionId

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

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

1588

1598

1589remoteOutgoing :: Accessor ValidRemoteEndPointState Int

1599remoteOutgoing :: Accessor ValidRemoteEndPointState Int

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

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

1591

1601

1592−remoteIncoming :: Accessor ValidRemoteEndPointState IntSet

1602+remoteIncoming :: Accessor ValidRemoteEndPointState (Set LightweightConnectionId)

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

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

1594

1604

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

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

1597−

1598−nextCtrlRequestId :: Accessor ValidRemoteEndPointState ControlRequestId

1605+remoteMaxIncoming :: Accessor ValidRemoteEndPointState LightweightConnectionId

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

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

1600

1607

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

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

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

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

1603

1610

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)

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

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

1610

1613

1611-------------------------------------------------------------------------------

1614-------------------------------------------------------------------------------

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

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

1−{-# LANGUAGE RebindableSyntax #-}

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

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

3module Main where

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

17 , TransportInternals(..)

18 , encodeEndPointAddress

19 , defaultTCPParameters

20+ , LightweightConnectionId

20 )

21 )

21import Data.Int (Int32)

22import Data.Int (Int32)

22import Control.Concurrent (threadDelay, killThread)

23import Control.Concurrent (threadDelay, killThread)

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

28 , isEmptyMVar

29 , isEmptyMVar

29 , newMVar

30 , newMVar

30 , modifyMVar

31 , modifyMVar

32+ , modifyMVar_

31 )

33 )

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

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

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

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

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

149 _ <- recvWithLength sock

151 _ <- recvWithLength sock

150 sendMany sock [encodeInt32 ConnectionRequestAccepted]

152 sendMany sock [encodeInt32 ConnectionRequestAccepted]

151

153

152− -- Server requests a logical connection

154+ -- Server opens a logical connection

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

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

154− reqId <- recvInt32 sock :: IO Int

156+ 1024 <- recvInt32 sock :: IO LightweightConnectionId

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

156

157

157 -- Server sends a message

158 -- Server sends a message

158− 10001 <- recvInt32 sock :: IO Int

159+ 1024 <- recvInt32 sock :: IO Int

159 ["ping"] <- recvWithLength sock

160 ["ping"] <- recvWithLength sock

160

161

161 -- Reply

162 -- Reply

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

163+ sendMany sock [encodeInt32 CreatedNewConnection, 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"])

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

167

165

168 -- Close the socket

166 -- Close the socket

169 N.sClose sock

167 N.sClose sock

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

171 -- Connect to the server

169 -- Connect to the server

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

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

173

171

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

172+ -- Open a new connection

175− let reqId = 0 :: Int32

176− sendMany sock [encodeInt32 RequestConnectionId, encodeInt32 reqId]

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

177

174

178 -- Close the socket without closing the connection explicitly

175 -- Close the socket without closing the connection explicitly

179 -- The server should receive an error event

176 -- The server should receive an error event

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

260 _ <- recvWithLength sock

257 _ <- recvWithLength sock

261 sendMany sock [encodeInt32 ConnectionRequestAccepted]

258 sendMany sock [encodeInt32 ConnectionRequestAccepted]

262

259

263− -- Server requests a logical connection

260+ -- Server opens a logical connection

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

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

265− reqId <- recvInt32 sock :: IO Int

262+ 1024 <- recvInt32 sock :: IO LightweightConnectionId

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

267

263

268 -- Server sends a message

264 -- Server sends a message

269− 10001 <- recvInt32 sock :: IO Int

265+ 1024 <- recvInt32 sock :: IO Int

270 ["ping"] <- recvWithLength sock

266 ["ping"] <- recvWithLength sock

271

267

272 -- Reply

268 -- Reply

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

269+ sendMany sock [encodeInt32 CreatedNewConnection, 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"])

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

278

271

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

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

280 -- directions*

273 -- directions*

281− sendMany sock [encodeInt32 CloseSocket]

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

282 N.sClose sock

275 N.sClose sock

283

276

284 -- Connect to the server

277 -- Connect to the server

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

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

286

279

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

280+ -- Open a new connection

288− let reqId = 0 :: Int32

289− sendMany sock [encodeInt32 RequestConnectionId, encodeInt32 reqId]

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

290

282

291 -- Send a CloseSocket without sending a closeconnecton

283 -- Send a CloseSocket without sending a closeconnecton

292 -- The server should still receive a ConnectionClosed message

284 -- The server should still receive a ConnectionClosed message

293− sendMany sock [encodeInt32 CloseSocket]

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

294 N.sClose sock

286 N.sClose sock

295

287

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

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

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

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

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

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

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

331testIgnoreCloseSocket nextPort = do

323testIgnoreCloseSocket nextPort = do

332− serverAddr <- newEmptyMVar

324+ serverAddr <- newEmptyMVar

325+ clientAddr <- newEmptyMVar

333− clientDone <- newEmptyMVar

326+ clientDone <- newEmptyMVar

327+ serverDone <- newEmptyMVar

328+ connectionEstablished <- newEmptyMVar

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

329+ 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

340

330

341− where

331+ -- Server

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

332+ forkTry $ do

343− server transport serverAddr = do

344− tlog "Server"

333+ tlog "Server"

345− Right endpoint <- newEndPoint transport

334+ Right endpoint <- newEndPoint transport

346− putMVar serverAddr (address endpoint)

335+ putMVar serverAddr (address endpoint)

347

336

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

337+ let ourAddress = address endpoint

349− tlog "Waiting for ConnectionOpened"

338+ theirAddress <- readMVar clientAddr

350− ConnectionOpened _ _ _ <- receive endpoint

351− tlog "Waiting for ConnectionClosed"

352− ConnectionClosed _ <- receive endpoint

353

339

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

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

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

341+ takeMVar connectionEstablished

356− -- another connection

357− tlog "Waiting for ConnectionOpened"

358− ConnectionOpened _ _ _ <- receive endpoint

359− tlog "Waiting for ConnectionClosed"

360− ConnectionClosed _ <- receive endpoint

361

342

362−

343+ -- Connect then disconnect to the client

363− tlog "Server waiting.."

344+ Right conn <- connect endpoint theirAddress ReliableOrdered defaultConnectHints

345+ close conn

364

346

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

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

366− client transport serverAddr clientDone = do

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

367− tlog "Client"

349+ -- another connection

368− Right endpoint <- newEndPoint transport

350+ tlog "Waiting for ConnectionOpened"

369− let ourAddress = address endpoint

351+ ConnectionOpened _ _ _ <- receive endpoint

352+ tlog "Waiting for ConnectionClosed"

353+ ConnectionClosed _ <- receive endpoint

370

354

371− -- Connect to the server

355+ putMVar serverDone ()

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

373

356

374− -- Request a new connection

357+ -- Client

375− tlog "Requesting connection"

358+ forkTry $ do

376− let reqId = 0 :: Int32

359+ tlog "Client"

377− sendMany sock [encodeInt32 RequestConnectionId, encodeInt32 reqId]

360+ Right endpoint <- newEndPoint transport

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

361+ putMVar clientAddr (address endpoint)

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

405

362

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+

406− putMVar clientDone ()

395+ putMVar clientDone ()

396+

397+ takeMVar clientDone

398+ takeMVar serverDone

407

399

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

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

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

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

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

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

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

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

412testBlockAfterCloseSocket nextPort = do

404testBlockAfterCloseSocket nextPort = do

413− serverAddr <- newEmptyMVar

405+ serverAddr <- newEmptyMVar

414− clientAddr <- newEmptyMVar

406+ clientAddr <- newEmptyMVar

415− clientDone <- newEmptyMVar

407+ clientDone <- newEmptyMVar

416− port <- nextPort

408+ serverDone <- newEmptyMVar

409+ connectionEstablished <- newEmptyMVar

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

410+ Right transport <- nextPort >>= \port -> 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

423

411

424− where

412+ -- Server

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

413+ forkTry $ do

426− server transport serverAddr clientAddr = do

427− tlog "Server"

414+ tlog "Server"

428− Right endpoint <- newEndPoint transport

415+ Right endpoint <- newEndPoint transport

429− putMVar serverAddr (address endpoint)

416+ putMVar serverAddr (address endpoint)

430

417

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

418+ let ourAddress = address endpoint

432− tlog "Waiting for ConnectionOpened"

419+ theirAddress <- readMVar clientAddr

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.."

443

420

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

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

445− client transport serverAddr clientAddr clientDone = do

422+ takeMVar connectionEstablished

446− tlog "Client"

447− Right endpoint <- newEndPoint transport

448− putMVar clientAddr (address endpoint)

449− let ourAddress = address endpoint

450

423

451− -- Connect to the server

424+ -- Connect then disconnect to the client

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

425+ Right conn <- connect endpoint theirAddress ReliableOrdered defaultConnectHints

426+ close conn

453

427

454− -- Request a new connection

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

455− tlog "Requesting connection"

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

456− let reqId = 0 :: Int32

430+ Right conn <- connect endpoint theirAddress ReliableOrdered defaultConnectHints

457− sendMany sock [encodeInt32 RequestConnectionId, encodeInt32 reqId]

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

459

431

460− -- Close the connection again

432+ putMVar serverDone ()

461− tlog "Closing connection"

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

463

433

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

434+ -- Client

465− tlog "Waiting for CloseSocket request"

435+ forkTry $ do

466− CloseSocket <- toEnum <$> recvInt32 sock

436+ tlog "Client"

437+ Right endpoint <- newEndPoint transport

438+ putMVar clientAddr (address endpoint)

467

439

468− unblocked <- newEmptyMVar

440+ let ourAddress = address endpoint

441+ theirAddress <- readMVar serverAddr

469

442

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

443+ -- Connect to the server

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

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

472− -- respond by sending a ConnectionRequest)

445+ putMVar connectionEstablished ()

473− forkTry $ do

446+

474− recvInt32 sock :: IO Int32

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

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

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

476− putMVar clientDone ()

449+ 1024 <- recvInt32 sock :: IO LightweightConnectionId

477

450

478− threadDelay 1000000

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

452+ 1024 <- recvInt32 sock :: IO LightweightConnectionId

479

453

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+

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

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

481− tlog "This should unblock the server"

472+ tlog "This should unblock the server"

482− putMVar unblocked ()

473+ modifyMVar_ unblocked $ \_ -> return True

483− let reqId' = 1 :: Int32

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

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

475+

476+ takeMVar clientDone

477+ takeMVar serverDone

485

478

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

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

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

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

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

592

585

593 when (count > 0) $ do

586 when (count > 0) $ do

594 -- Client requests a logical connection

587 -- Client requests a logical connection

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

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

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

589+ connId <- recvInt32 sock :: IO LightweightConnectionId

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

598 return ()

590 return ()

599

591

600− when (count > 1) $ do

592+ when (count > 1) $ do

601− -- Client sends a message

593+ -- Client sends a message

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

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

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

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

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

604− putMVar serverDone ()

597+ putMVar serverDone ()

605

598

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

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

607 return ()

600 return ()

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

615 let theirAddr = encodeEndPointAddress "127.0.0.1" serverPort 0

608 let theirAddr = encodeEndPointAddress "127.0.0.1" serverPort 0

616

609

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

610 -- 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

619 putMVar firstAttempt ()

612 putMVar firstAttempt ()

620

613

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

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

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

633 Right conn1 <- connect endpoint theirAddr ReliableOrdered defaultConnectHints

626 Right conn1 <- connect endpoint theirAddr ReliableOrdered defaultConnectHints

634

627

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

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

629+ threadDelay 100000

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

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

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

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

638

632

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

667 _ <- recvWithLength sock

661 _ <- recvWithLength sock

668 () <- sendMany sock [encodeInt32 ConnectionRequestAccepted]

662 () <- sendMany sock [encodeInt32 ConnectionRequestAccepted]

669

663

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

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

671− reqId <- recvInt32 sock :: IO Int

665+ connId <- recvInt32 sock :: IO LightweightConnectionId

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

673

666

674− 10001 <- recvInt32 sock :: IO Int

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

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

675 ["ping"] <- recvWithLength sock

669 ["ping"] <- recvWithLength sock

676 putMVar serverGotPing ()

670 putMVar serverGotPing ()

677

671

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

777 , ("EarlyCloseSocket", testEarlyCloseSocket nextPort)

771 , ("EarlyCloseSocket", testEarlyCloseSocket nextPort)

778 , ("IgnoreCloseSocket", testIgnoreCloseSocket nextPort)

772 , ("IgnoreCloseSocket", testIgnoreCloseSocket nextPort)

779 , ("BlockAfterCloseSocket", testBlockAfterCloseSocket nextPort)

773 , ("BlockAfterCloseSocket", testBlockAfterCloseSocket nextPort)

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

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

781 , ("InvalidAddress", testInvalidAddress nextPort)

775 , ("InvalidAddress", testInvalidAddress nextPort)

782 , ("InvalidConnect", testInvalidConnect nextPort)

776 , ("InvalidConnect", testInvalidConnect nextPort)

783 , ("Many", testMany nextPort)

777 , ("Many", testMany nextPort)

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

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

691 send conn ["pong"]

692

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

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

694− evs <- replicateM 3 $ receive endpoint

695+ evs <- replicateM 4 $ receive endpoint

695 let expected = [ ConnectionClosed cid1

696 let expected = [ ConnectionClosed cid1

696 , ConnectionClosed cid2

697 , ConnectionClosed cid2

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

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

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

698 ]

700 ]

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

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

700

702

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

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

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

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

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

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

66import Data.Typeable (Typeable)

67import Data.Maybe (catMaybes)

68import Data.ByteString (ByteString)

69−import Data.Int (Int32)

69+import Data.Int (Int32, Int64)

70+import Data.Word (Word32, Word64)

70import Control.Concurrent.MVar (MVar)

71import Control.Concurrent.MVar (MVar)

72--------------------------------------------------------------------------------

73--------------------------------------------------------------------------------

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

152instance Traceable Int32 where

153instance Traceable Int32 where

153 trace = traceShow

154 trace = traceShow

154

155

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+

155instance Traceable Bool where

165instance Traceable Bool where

156 trace = const Nothing

166 trace = const Nothing

157

167

network-transport/ChangeLog+1

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

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

3* Clarify disconnection

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

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

62012-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)

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

31import Data.Typeable (Typeable)

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

33+import Data.Word (Word64)

34--------------------------------------------------------------------------------

35--------------------------------------------------------------------------------

35-- Main API --

36-- Main API --

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

51 -- | EndPointAddress of the endpoint.

52 -- | EndPointAddress of the endpoint.

52 , address :: EndPointAddress

53 , address :: EndPointAddress

53 -- | Create a new lightweight connection.

54 -- | 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.

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

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

55 -- | Create a new multicast group.

61 -- | Create a new multicast group.

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

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

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

80 -- | Connection closed

86 -- | Connection closed

81 | ConnectionClosed ConnectionId

87 | ConnectionClosed ConnectionId

82 -- | Connection opened

88 -- | Connection opened

89+ --

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

83 | ConnectionOpened ConnectionId Reliability EndPointAddress

91 | ConnectionOpened ConnectionId Reliability EndPointAddress

84 -- | Received multicast

92 -- | Received multicast

85 | ReceivedMulticast MulticastAddress [ByteString]

93 | ReceivedMulticast MulticastAddress [ByteString]

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

90 deriving (Show, Eq)

98 deriving (Show, Eq)

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

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

93−type ConnectionId = Int

101+type ConnectionId = Word64

102

95-- | Reliability guarantees of a connection.

103-- | Reliability guarantees of a connection.

96data Reliability =

104data Reliability =