87b8a75 · distributed-process

git.iko.soy/distributed-process

on master help sign in

More error handling (not yet done)

87b8a75Edsko de Vries14 years ago

network-transport/src/Network/Transport/Internal/TCP.hs+4−3

compare with master

unifiedsplit

@@ -87,7 +87,8 @@ recvInt32 sock = decodeInt32 . BS.concat <$> recvExact sock 4

88-- | Close a socket, ignoring I/O exceptions

89tryCloseSocket :: N.Socket -> IO ()

90−tryCloseSocket = void . tryIO . N.sClose

90+tryCloseSocket sock = void . tryIO $

91+ N.sClose sock

92-- | Read an exact number of bytes from a socket

93-- | Read an exact number of bytes from a socket

93--

94--

@@ -96,7 +97,7 @@ tryCloseSocket = void . tryIO . N.sClose

96recvExact :: N.Socket -- ^ Socket to read from

97recvExact :: N.Socket -- ^ Socket to read from

97 -> Int32 -- ^ Number of bytes to read

98 -> Int32 -- ^ Number of bytes to read

98 -> IO [ByteString]

99 -> IO [ByteString]

99−recvExact _ len | len <= 0 = throwIO (userError "Negative length")

100+recvExact _ len | len <= 0 = throwIO (userError "recvExact: Negative length")

100recvExact sock len = go [] len

101recvExact sock len = go [] len

101 where

102 where

102 go :: [ByteString] -> Int32 -> IO [ByteString]

103 go :: [ByteString] -> Int32 -> IO [ByteString]

@@ -104,5 +105,5 @@ recvExact sock len = go [] len

104 go acc l = do

105 go acc l = do

105 bs <- NBS.recv sock (fromIntegral l `min` 4096)

106 bs <- NBS.recv sock (fromIntegral l `min` 4096)

106 if BS.null bs

107 if BS.null bs

107− then throwIO (userError "Socket closed")

108+ then throwIO (userError "recvExact: Socket closed")

108 else go (bs : acc) (l - fromIntegral (BS.length bs))

109 else go (bs : acc) (l - fromIntegral (BS.length bs))

network-transport/src/Network/Transport/TCP.hs+210−154

@@ -84,6 +84,7 @@ import Control.Exception ( IOException

84 , mask_

85 , onException

86 , fromException

87+ , catch

87 )

88 )

88import Data.IORef (IORef, newIORef, writeIORef, readIORef)

89import Data.IORef (IORef, newIORef, writeIORef, readIORef)

89import Data.ByteString (ByteString)

90import Data.ByteString (ByteString)

@@ -106,7 +107,6 @@ import qualified Data.Map as Map (empty)

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

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

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

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

108import Data.Foldable (forM_, mapM_)

109import Data.Foldable (forM_, mapM_)

109−import Data.Maybe (isNothing)

110

111-- $design

112--

@@ -241,12 +241,10 @@ data LocalEndPointState =

241 | LocalEndPointClosed

242

243data ValidLocalEndPointState = ValidLocalEndPointState

244− { _nextConnectionId :: !ConnectionId

244+ { _nextConnectionId :: !ConnectionId

245− , _pendingCtrlRequests :: IntMap (MVar [ByteString])

246− , _nextCtrlRequestId :: !ControlRequestId

247− , _localConnections :: Map EndPointAddress RemoteEndPoint

245+ , _localConnections :: Map EndPointAddress RemoteEndPoint

248− , _internalThreads :: [ThreadId]

246+ , _internalThreads :: [ThreadId]

249− , _nextRemoteId :: !Int

247+ , _nextRemoteId :: !Int

250 }

248 }

251

249

252-- REMOTE ENDPOINTS

250-- REMOTE ENDPOINTS

@@ -327,19 +325,22 @@ data RemoteEndPoint = RemoteEndPoint

327 }

325 }

328

326

329data Origin = Local | Remote

327data Origin = Local | Remote

328+ deriving (Eq, Show)

330

329

331data RemoteState =

330data RemoteState =

332 RemoteEndPointInvalid (TransportError ConnectErrorCode)

331 RemoteEndPointInvalid (TransportError ConnectErrorCode)

333 | RemoteEndPointInit (MVar ()) Origin

332 | RemoteEndPointInit (MVar ()) Origin

334 | RemoteEndPointValid ValidRemoteEndPointState

333 | RemoteEndPointValid ValidRemoteEndPointState

335 | RemoteEndPointClosing (MVar ()) ValidRemoteEndPointState

334 | RemoteEndPointClosing (MVar ()) ValidRemoteEndPointState

336− | RemoteEndPointClosed

335+ | RemoteEndPointClosed (Maybe IOException)

337

336

338data ValidRemoteEndPointState = ValidRemoteEndPointState

337data ValidRemoteEndPointState = ValidRemoteEndPointState

339− { _remoteOutgoing :: !Int

338+ { _remoteOutgoing :: !Int

340− , _remoteIncoming :: IntSet

339+ , _remoteIncoming :: IntSet

341− , remoteSocket :: N.Socket

340+ , remoteSocket :: N.Socket

342− , sendOn :: [ByteString] -> IO ()

341+ , sendOn :: [ByteString] -> IO ()

342+ , _pendingCtrlRequests :: IntMap (MVar (Either IOException [ByteString]))

343+ , _nextCtrlRequestId :: !ControlRequestId

343 }

344 }

344

345

345type EndPointId = Int32

346type EndPointId = Int32

@@ -476,6 +477,7 @@ apiConnect ourEndPoint theirAddress _reliability _hints =

476 if localAddress ourEndPoint == theirAddress

477 if localAddress ourEndPoint == theirAddress

477 then connectToSelf ourEndPoint

478 then connectToSelf ourEndPoint

478 else try $ do

479 else try $ do

480+ resetIfInvalid ourEndPoint theirAddress

479 (theirEndPoint, connId) <- requestConnectionTo ourEndPoint theirAddress

481 (theirEndPoint, connId) <- requestConnectionTo ourEndPoint theirAddress

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

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

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

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

@@ -525,12 +527,12 @@ apiSend (ourEndPoint, theirEndPoint) connId connAlive payload =

525 , caseClosing = \_ _ -> do

527 , caseClosing = \_ _ -> do

526 alive <- readIORef connAlive

528 alive <- readIORef connAlive

527 if alive

529 if alive

528− then throwIO $ TransportError SendFailed "Connection lost"

530+ then relyViolation (ourEndPoint, theirEndPoint) "apiSend"

529 else throwIO $ TransportError SendClosed "Connection closed"

531 else throwIO $ TransportError SendClosed "Connection closed"

530− , caseClosed = do

532+ , caseClosed = \err -> do

531 alive <- readIORef connAlive

533 alive <- readIORef connAlive

532 if alive

534 if alive

533− then throwIO $ TransportError SendFailed "Connection lost"

535+ then throwIO $ TransportError SendFailed (show err)

534 else throwIO $ TransportError SendClosed "Connection closed"

536 else throwIO $ TransportError SendClosed "Connection closed"

535 }

537 }

536 where

538 where

@@ -568,17 +570,17 @@ apiCloseEndPoint transport evs ourEndPoint = do

568 return st

570 return st

569 RemoteEndPointInit resolved _ -> do

571 RemoteEndPointInit resolved _ -> do

570 putMVar resolved ()

572 putMVar resolved ()

571− return RemoteEndPointClosed

573+ return $ RemoteEndPointClosed (Just . userError $ "apiCloseEndPoint")

572 RemoteEndPointValid conn -> do

574 RemoteEndPointValid conn -> do

573 tryIO $ sendOn conn [encodeInt32 CloseSocket]

575 tryIO $ sendOn conn [encodeInt32 CloseSocket]

574 tryCloseSocket (remoteSocket conn)

576 tryCloseSocket (remoteSocket conn)

575− return RemoteEndPointClosed

577+ return $ RemoteEndPointClosed (Just . userError $ "apiCloseEndPoint")

576 RemoteEndPointClosing resolved conn -> do

578 RemoteEndPointClosing resolved conn -> do

577 putMVar resolved ()

579 putMVar resolved ()

578 tryCloseSocket (remoteSocket conn)

580 tryCloseSocket (remoteSocket conn)

579− return RemoteEndPointClosed

581+ return $ RemoteEndPointClosed (Just . userError $ "apiCloseEndPoint")

580− RemoteEndPointClosed ->

582+ RemoteEndPointClosed err ->

581− return RemoteEndPointClosed

583+ return $ RemoteEndPointClosed err

582

584

583-- | Special case of 'apiConnect': connect an endpoint to itself

585-- | Special case of 'apiConnect': connect an endpoint to itself

584connectToSelf :: LocalEndPoint

586connectToSelf :: LocalEndPoint

@@ -655,7 +657,7 @@ internalSocketBetween transport ourAddress theirAddress = do

655 return $ remoteSocket vst

657 return $ remoteSocket vst

656 RemoteEndPointClosing _ vst ->

658 RemoteEndPointClosing _ vst ->

657 return $ remoteSocket vst

659 return $ remoteSocket vst

658− RemoteEndPointClosed ->

660+ RemoteEndPointClosed _ ->

659 throwIO $ userError "Remote endpoint closed"

661 throwIO $ userError "Remote endpoint closed"

660 RemoteEndPointInvalid _ ->

662 RemoteEndPointInvalid _ ->

661 throwIO $ userError "Remote endpoint invalid"

663 throwIO $ userError "Remote endpoint invalid"

@@ -664,6 +666,28 @@ internalSocketBetween transport ourAddress theirAddress = do

664-- Lower level functionality --

666-- Lower level functionality --

665--------------------------------------------------------------------------------

667--------------------------------------------------------------------------------

666

668

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

670+--

671+-- If a user calls the API function 'connect' and the remote endpoint is

672+-- currently in Invalid state, we remove the remote endpoint first because a

673+-- new attempt to connect might succeed even if the previous one failed.

674+--

675+-- Throws a TransportError ConnectFailed exception if the local endpoint is

676+-- closed.

677+resetIfInvalid :: LocalEndPoint -> EndPointAddress -> IO ()

678+resetIfInvalid ourEndPoint theirAddress = do

679+ mTheirEndPoint <- withMVar (localState ourEndPoint) $ \st -> case st of

680+ LocalEndPointValid vst ->

681+ return (vst ^. localConnectionTo theirAddress)

682+ LocalEndPointClosed ->

683+ throwIO $ TransportError ConnectFailed "Endpoint closed"

684+ forM_ mTheirEndPoint $ \theirEndPoint ->

685+ withMVar (remoteState theirEndPoint) $ \st -> case st of

686+ RemoteEndPointInvalid _ ->

687+ removeRemoteEndPoint (ourEndPoint, theirEndPoint)

688+ _ ->

689+ return ()

690+

667-- | Create a new local endpoint

691-- | Create a new local endpoint

668--

692--

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

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

@@ -673,8 +697,6 @@ createLocalEndPoint transport = do

673 chan <- newChan

697 chan <- newChan

674 state <- newMVar . LocalEndPointValid $ ValidLocalEndPointState

698 state <- newMVar . LocalEndPointValid $ ValidLocalEndPointState

675 { _nextConnectionId = firstNonReservedConnectionId

699 { _nextConnectionId = firstNonReservedConnectionId

676− , _pendingCtrlRequests = IntMap.empty

677− , _nextCtrlRequestId = 0

678 , _localConnections = Map.empty

700 , _localConnections = Map.empty

679 , _internalThreads = []

701 , _internalThreads = []

680 , _nextRemoteId = 0

702 , _nextRemoteId = 0

@@ -708,47 +730,43 @@ createLocalEndPoint transport = do

708requestConnectionTo :: LocalEndPoint

730requestConnectionTo :: LocalEndPoint

709 -> EndPointAddress

731 -> EndPointAddress

710 -> IO (RemoteEndPoint, ConnectionId)

732 -> IO (RemoteEndPoint, ConnectionId)

711−requestConnectionTo ourEndPoint theirAddress = do

733+requestConnectionTo ourEndPoint theirAddress = go

712− (theirEndPoint, isNew) <- mapIOException connectFailed $

713− findRemoteEndPoint ourEndPoint theirAddress Local

714−

715− -- If this is a new endpoint, start the initialization process and wait

716− forM_ isNew $ \resolved -> do

717− void . forkEndPointThread ourEndPoint $ do

718− let outgoing = 1

719− setupRemoteEndPoint (ourEndPoint, theirEndPoint) outgoing

720− readMVar resolved

721−

722− reply <- mapIOException connectFailed $

723− doRemoteRequest (ourEndPoint, theirEndPoint) RequestConnectionId

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

725 where

734 where

735+ go = do

736+ (theirEndPoint, isNew, _) <- mapIOException connectFailed $

737+ findRemoteEndPoint ourEndPoint theirAddress Local

738+

739+ if isNew

740+ then do

741+ void . forkEndPointThread ourEndPoint $

742+ catch (setupRemoteEndPoint (ourEndPoint, theirEndPoint))

743+ (\err -> let _ = err :: SomeException in return ())

744+ go

745+ else do

746+ reply <- mapIOException connectFailed $

747+ doRemoteRequest (ourEndPoint, theirEndPoint) RequestConnectionId

748+ return (theirEndPoint, decodeInt32 . BS.concat $ reply)

749+

726 connectFailed = TransportError ConnectFailed . show

750 connectFailed = TransportError ConnectFailed . show

727

751

728-- | Set up a remote endpoint

752-- | Set up a remote endpoint

729−setupRemoteEndPoint :: EndPointPair -- ^ Local endpoint, remote endpoint

753+setupRemoteEndPoint :: EndPointPair -> IO ()

730− -> Int -- ^ Initial (outgoing) refcount

731− -> IO ()

732−setupRemoteEndPoint (ourEndPoint, theirEndPoint) outgoing = do

754+setupRemoteEndPoint (ourEndPoint, theirEndPoint) = do

733 didAccept <- bracketOnError (socketToEndPoint ourAddress theirAddress)

755 didAccept <- bracketOnError (socketToEndPoint ourAddress theirAddress)

734 onError $ \result ->

756 onError $ \result ->

735 case result of

757 case result of

736 Right (sock, ConnectionRequestAccepted) -> do

758 Right (sock, ConnectionRequestAccepted) -> do

737 let vst = ValidRemoteEndPointState

759 let vst = ValidRemoteEndPointState

738− { remoteSocket = sock

760+ { remoteSocket = sock

739− , _remoteOutgoing = outgoing

761+ , _remoteOutgoing = 0

740− , _remoteIncoming = IntSet.empty

762+ , _remoteIncoming = IntSet.empty

741− , sendOn = sendMany sock

763+ , sendOn = sendMany sock

764+ , _pendingCtrlRequests = IntMap.empty

765+ , _nextCtrlRequestId = 0

742 }

766 }

743 resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst)

767 resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst)

744 return True

768 return True

745 Right (sock, ConnectionRequestInvalid) -> do

769 Right (sock, ConnectionRequestInvalid) -> do

746− -- We remove the endpoint from our local state again because the next

747− -- call to 'connect' might give a different result. Threads that were

748− -- waiting on the result of this call to connect will get the

749− -- RemoteEndPointInvalid; subsequent threads will initiate a new

750− -- connection requests.

751− removeRemoteEndPoint (ourEndPoint, theirEndPoint)

752 let err = invalidAddress "Invalid endpoint"

770 let err = invalidAddress "Invalid endpoint"

753 resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointInvalid err)

771 resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointInvalid err)

754 tryCloseSocket sock

772 tryCloseSocket sock

@@ -759,7 +777,6 @@ setupRemoteEndPoint (ourEndPoint, theirEndPoint) outgoing = do

759 tryCloseSocket sock

777 tryCloseSocket sock

760 return False

778 return False

761 Left err -> do

779 Left err -> do

762− removeRemoteEndPoint (ourEndPoint, theirEndPoint) -- See comment above

763 resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointInvalid err)

780 resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointInvalid err)

764 return False

781 return False

765

782

@@ -780,8 +797,7 @@ setupRemoteEndPoint (ourEndPoint, theirEndPoint) outgoing = do

780 onError :: Either (TransportError ConnectErrorCode)

797 onError :: Either (TransportError ConnectErrorCode)

781 (N.Socket, ConnectionRequestResponse)

798 (N.Socket, ConnectionRequestResponse)

782 -> IO ()

799 -> IO ()

783− onError result = do

800+ onError result =

784− removeRemoteEndPoint (ourEndPoint, theirEndPoint)

785 case result of

801 case result of

786 Left err ->

802 Left err ->

787 void $ tryPutMVar theirState (RemoteEndPointInvalid err)

803 void $ tryPutMVar theirState (RemoteEndPointInvalid err)

@@ -796,21 +812,23 @@ setupRemoteEndPoint (ourEndPoint, theirEndPoint) outgoing = do

796 theirState = remoteState theirEndPoint

812 theirState = remoteState theirEndPoint

797 invalidAddress = TransportError ConnectNotFound

813 invalidAddress = TransportError ConnectNotFound

798

814

815+-- TODO: Document

816+-- TODO: is this what we want? modifyRemoteState only does something special

817+-- when an exception is thrown in the Valid case.

799resolveInit :: EndPointPair -> RemoteState -> IO ()

818resolveInit :: EndPointPair -> RemoteState -> IO ()

800−resolveInit (ourEndPoint, theirEndPoint) newState =

819+resolveInit (ourEndPoint, theirEndPoint) newState =

801− modifyMVar_ (remoteState theirEndPoint) $ \st -> case st of

820+ modifyRemoteState_ (ourEndPoint, theirEndPoint) $ remoteStateIdentity

802− RemoteEndPointInvalid _ ->

803− relyViolation (ourEndPoint, theirEndPoint) "resolveInit (invalid)"

804− RemoteEndPointInit resolved _ -> do

821+ { caseInit = \resolved _ -> do

805− putMVar resolved ()

822+ putMVar resolved ()

806− return newState

823+ return newState

807− RemoteEndPointClosed ->

824+ , caseInvalid = \_ -> throwException

808− throwIO $ userError "Remote endpoint closed"

825+ , caseValid = \_ -> throwException

809− RemoteEndPointClosing _ _ ->

826+ , caseClosing = \_ _ -> throwException

810− relyViolation (ourEndPoint, theirEndPoint) "resolveInit (closing)"

827+ , caseClosed = \_ -> throwException

811− RemoteEndPointValid _ ->

828+ }

812− relyViolation (ourEndPoint, theirEndPoint) "resolveInit (valid)"

829+ where

813−

830+ throwException = throwIO $ userError "resolveInit: not in Init state"

831+

814-- | Establish a connection to a remote endpoint

832-- | Establish a connection to a remote endpoint

815socketToEndPoint :: EndPointAddress -- ^ Our address

833socketToEndPoint :: EndPointAddress -- ^ Our address

816 -> EndPointAddress -- ^ Their address

834 -> EndPointAddress -- ^ Their address

@@ -904,26 +922,26 @@ decodeEndPointAddress (EndPointAddress bs) =

904-- or if the send fails.

922-- or if the send fails.

905doRemoteRequest :: EndPointPair -> ControlHeader -> IO [ByteString]

923doRemoteRequest :: EndPointPair -> ControlHeader -> IO [ByteString]

906doRemoteRequest (ourEndPoint, theirEndPoint) header = do

924doRemoteRequest (ourEndPoint, theirEndPoint) header = do

907− reply <- newEmptyMVar

925+ replyMVar <- newEmptyMVar

908− reqId <- modifyMVar (localState ourEndPoint) $ \st -> case st of

926+ modifyRemoteState_ (ourEndPoint, theirEndPoint) RemoteStatePatternMatch

909− LocalEndPointValid vst -> do

927+ { caseValid = \vst -> do

910− let reqId = vst ^. nextCtrlRequestId

928+ let reqId = vst ^. nextCtrlRequestId

911− return ( LocalEndPointValid

929+ sendOn vst [encodeInt32 header, encodeInt32 reqId]

930+ return ( RemoteEndPointValid

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

931+ . (nextCtrlRequestId ^: (+ 1))

913− . (pendingCtrlRequestsAt reqId ^= Just reply)

932+ . (pendingCtrlRequestsAt reqId ^= Just replyMVar)

914− $ vst

933+ $ vst

915− , reqId

916− )

934+ )

917− LocalEndPointClosed ->

935+ -- Error cases

918− throwIO (userError "Local endpoint closed")

919− withRemoteState (ourEndPoint, theirEndPoint) RemoteStatePatternMatch

920− { caseInvalid = throwIO

936+ , caseInvalid = throwIO

921 , caseInit = \_ _ -> relyViolation (ourEndPoint, theirEndPoint) "doRemoteRequest (init)"

937 , caseInit = \_ _ -> relyViolation (ourEndPoint, theirEndPoint) "doRemoteRequest (init)"

922− , caseValid = \vst -> sendOn vst [encodeInt32 header, encodeInt32 reqId]

923 , caseClosing = \_ _ -> relyViolation (ourEndPoint, theirEndPoint) "doRemoteRequest (closing)"

938 , caseClosing = \_ _ -> relyViolation (ourEndPoint, theirEndPoint) "doRemoteRequest (closing)"

924− , caseClosed = throwIO $ userError "Remote endpoint closed"

939+ , caseClosed = \_ -> throwIO $ userError "doRemoteRequest: Remote endpoint closed"

925 }

940 }

926− takeMVar reply

941+ mReply <- takeMVar replyMVar

942+ case mReply of

943+ Left err -> throwIO err

944+ Right reply -> return reply

927

945

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

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

929closeIfUnused :: EndPointPair -> IO ()

947closeIfUnused :: EndPointPair -> IO ()

@@ -992,7 +1010,7 @@ data RemoteStatePatternMatch a = RemoteStatePatternMatch

992 , caseInit :: MVar () -> Origin -> IO a

1010 , caseInit :: MVar () -> Origin -> IO a

993 , caseValid :: ValidRemoteEndPointState -> IO a

1011 , caseValid :: ValidRemoteEndPointState -> IO a

994 , caseClosing :: MVar () -> ValidRemoteEndPointState -> IO a

1012 , caseClosing :: MVar () -> ValidRemoteEndPointState -> IO a

995− , caseClosed :: IO a

1013+ , caseClosed :: Maybe IOException -> IO a

996 }

1014 }

997

1015

998remoteStateIdentity :: RemoteStatePatternMatch RemoteState

1016remoteStateIdentity :: RemoteStatePatternMatch RemoteState

@@ -1002,7 +1020,7 @@ remoteStateIdentity =

1002 , caseInit = (return .) . RemoteEndPointInit

1020 , caseInit = (return .) . RemoteEndPointInit

1003 , caseValid = return . RemoteEndPointValid

1021 , caseValid = return . RemoteEndPointValid

1004 , caseClosing = (return .) . RemoteEndPointClosing

1022 , caseClosing = (return .) . RemoteEndPointClosing

1005− , caseClosed = return RemoteEndPointClosed

1023+ , caseClosed = return . RemoteEndPointClosed

1006 }

1024 }

1007

1025

1008-- | Like modifyMVar, but if an I/O exception occurs don't restore the remote

1026-- | Like modifyMVar, but if an I/O exception occurs don't restore the remote

@@ -1042,8 +1060,8 @@ modifyRemoteState (ourEndPoint, theirEndPoint) match =

1042 (putMVar theirState st)

1060 (putMVar theirState st)

1043 putMVar theirState st'

1061 putMVar theirState st'

1044 return a

1062 return a

1045− RemoteEndPointClosed -> do

1063+ RemoteEndPointClosed err -> do

1046− (st', a) <- onException (restore $ caseClosed match)

1064+ (st', a) <- onException (restore $ caseClosed match err)

1047 (putMVar theirState st)

1065 (putMVar theirState st)

1048 putMVar theirState st'

1066 putMVar theirState st'

1049 return a

1067 return a

@@ -1055,7 +1073,7 @@ modifyRemoteState (ourEndPoint, theirEndPoint) match =

1055 handleIOException ex vst = do

1073 handleIOException ex vst = do

1056 tryCloseSocket (remoteSocket vst)

1074 tryCloseSocket (remoteSocket vst)

1057 removeRemoteEndPoint (ourEndPoint, theirEndPoint)

1075 removeRemoteEndPoint (ourEndPoint, theirEndPoint)

1058− putMVar theirState RemoteEndPointClosed

1076+ putMVar theirState (RemoteEndPointClosed $ Just ex)

1059 let incoming = IntSet.elems $ vst ^. remoteIncoming

1077 let incoming = IntSet.elems $ vst ^. remoteIncoming

1060 code = EventConnectionLost (remoteAddress theirEndPoint) incoming

1078 code = EventConnectionLost (remoteAddress theirEndPoint) incoming

1061 err = TransportError code (show ex)

1079 err = TransportError code (show ex)

@@ -1072,7 +1090,7 @@ modifyRemoteState_ (ourEndPoint, theirEndPoint) match =

1072 , caseInit = \resolved origin -> u $ caseInit match resolved origin

1090 , caseInit = \resolved origin -> u $ caseInit match resolved origin

1073 , caseValid = u . caseValid match

1091 , caseValid = u . caseValid match

1074 , caseClosing = \resolved vst -> u $ caseClosing match resolved vst

1092 , caseClosing = \resolved vst -> u $ caseClosing match resolved vst

1075− , caseClosed = u $ caseClosed match

1093+ , caseClosed = u . caseClosed match

1076 }

1094 }

1077 where

1095 where

1078 u :: IO a -> IO (a, ())

1096 u :: IO a -> IO (a, ())

@@ -1097,9 +1115,9 @@ withRemoteState (ourEndPoint, theirEndPoint) match =

1097 , caseClosing = \resolved vst -> do

1115 , caseClosing = \resolved vst -> do

1098 a <- caseClosing match resolved vst

1116 a <- caseClosing match resolved vst

1099 return (RemoteEndPointClosing resolved vst, a)

1117 return (RemoteEndPointClosing resolved vst, a)

1100− , caseClosed = do

1118+ , caseClosed = \err -> do

1101− a <- caseClosed match

1119+ a <- caseClosed match err

1102− return (RemoteEndPointClosed, a)

1120+ return (RemoteEndPointClosed err, a)

1103 }

1121 }

1104

1122

1105--------------------------------------------------------------------------------

1123--------------------------------------------------------------------------------

@@ -1139,12 +1157,9 @@ handleConnectionRequest transport sock = handle handleException $ do

1139 go :: LocalEndPoint -> EndPointAddress -> IO ()

1157 go :: LocalEndPoint -> EndPointAddress -> IO ()

1140 go ourEndPoint theirAddress = do

1158 go ourEndPoint theirAddress = do

1141 mEndPoint <- handle ((>> return Nothing) . invalidEndPoint) $ do

1159 mEndPoint <- handle ((>> return Nothing) . invalidEndPoint) $ do

1142− (theirEndPoint, isNew) <-

1160+ (theirEndPoint, _, crossed) <-

1143 findRemoteEndPoint ourEndPoint theirAddress Remote

1161 findRemoteEndPoint ourEndPoint theirAddress Remote

1144−

1162+

1145− let ourAddress = localAddress ourEndPoint

1146− crossed = isNothing isNew && ourAddress < theirAddress

1147−

1148 if crossed

1163 if crossed

1149 then do

1164 then do

1150 tryIO $ sendMany sock [encodeInt32 ConnectionRequestCrossed]

1165 tryIO $ sendMany sock [encodeInt32 ConnectionRequestCrossed]

@@ -1152,10 +1167,12 @@ handleConnectionRequest transport sock = handle handleException $ do

1152 return Nothing

1167 return Nothing

1153 else do

1168 else do

1154 let vst = ValidRemoteEndPointState

1169 let vst = ValidRemoteEndPointState

1155− { remoteSocket = sock

1170+ { remoteSocket = sock

1156− , _remoteOutgoing = 0

1171+ , _remoteOutgoing = 0

1157− , _remoteIncoming = IntSet.empty

1172+ , _remoteIncoming = IntSet.empty

1158− , sendOn = sendMany sock

1173+ , sendOn = sendMany sock

1174+ , _pendingCtrlRequests = IntMap.empty

1175+ , _nextCtrlRequestId = 0

1159 }

1176 }

1160 sendMany sock [encodeInt32 ConnectionRequestAccepted]

1177 sendMany sock [encodeInt32 ConnectionRequestAccepted]

1161 resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst)

1178 resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst)

@@ -1176,19 +1193,20 @@ handleConnectionRequest transport sock = handle handleException $ do

1176 tryCloseSocket sock

1193 tryCloseSocket sock

1177

1194

1178-- | Find a remote endpoint. If the remote endpoint does not yet exist we

1195-- | Find a remote endpoint. If the remote endpoint does not yet exist we

1179−-- create it in Init state and return the corresponding 'resolved' MVar.

1196+-- create it in Init state. Returns if the endpoint was new and, for requests

1197+-- of Remote origin, whether the request crossed.

1180findRemoteEndPoint

1198findRemoteEndPoint

1181 :: LocalEndPoint

1199 :: LocalEndPoint

1182 -> EndPointAddress

1200 -> EndPointAddress

1183 -> Origin

1201 -> Origin

1184− -> IO (RemoteEndPoint, Maybe (MVar ()))

1202+ -> IO (RemoteEndPoint, Bool, Bool)

1185findRemoteEndPoint ourEndPoint theirAddress findOrigin = go

1203findRemoteEndPoint ourEndPoint theirAddress findOrigin = go

1186 where

1204 where

1187 go = do

1205 go = do

1188 (theirEndPoint, isNew) <- modifyMVar ourState $ \st -> case st of

1206 (theirEndPoint, isNew) <- modifyMVar ourState $ \st -> case st of

1189 LocalEndPointValid vst -> case vst ^. localConnectionTo theirAddress of

1207 LocalEndPointValid vst -> case vst ^. localConnectionTo theirAddress of

1190 Just theirEndPoint ->

1208 Just theirEndPoint ->

1191− return (st, (theirEndPoint, Nothing))

1209+ return (st, (theirEndPoint, False))

1192 Nothing -> do

1210 Nothing -> do

1193 resolved <- newEmptyMVar

1211 resolved <- newEmptyMVar

1194 theirState <- newMVar (RemoteEndPointInit resolved findOrigin)

1212 theirState <- newMVar (RemoteEndPointInit resolved findOrigin)

@@ -1201,15 +1219,15 @@ findRemoteEndPoint ourEndPoint theirAddress findOrigin = go

1201 . (localConnectionTo theirAddress ^= Just theirEndPoint)

1219 . (localConnectionTo theirAddress ^= Just theirEndPoint)

1202 . (nextRemoteId ^: (+ 1))

1220 . (nextRemoteId ^: (+ 1))

1203 $ vst

1221 $ vst

1204− , (theirEndPoint, Just resolved)

1222+ , (theirEndPoint, True)

1205 )

1223 )

1206 LocalEndPointClosed ->

1224 LocalEndPointClosed ->

1207 throwIO $ userError "Local endpoint closed"

1225 throwIO $ userError "Local endpoint closed"

1208−

1226+

1209− case isNew of

1227+ if isNew

1210− Just resolved ->

1228+ then

1211− return (theirEndPoint, Just resolved)

1229+ return (theirEndPoint, True, False)

1212− Nothing -> do

1230+ else do

1213 let theirState = remoteState theirEndPoint

1231 let theirState = remoteState theirEndPoint

1214 snapshot <- modifyMVar theirState $ \st -> case st of

1232 snapshot <- modifyMVar theirState $ \st -> case st of

1215 RemoteEndPointValid vst ->

1233 RemoteEndPointValid vst ->

@@ -1220,7 +1238,7 @@ findRemoteEndPoint ourEndPoint theirAddress findOrigin = go

1220 $ vst

1238 $ vst

1221 return (st', st')

1239 return (st', st')

1222 Remote ->

1240 Remote ->

1223− throwIO $ userError "Already connected"

1241+ return (st, st)

1224 _ ->

1242 _ ->

1225 return (st, st)

1243 return (st, st)

1226 -- The snapshot may no longer be up to date at this point, but if we

1244 -- The snapshot may no longer be up to date at this point, but if we

@@ -1233,16 +1251,27 @@ findRemoteEndPoint ourEndPoint theirAddress findOrigin = go

1233 case (findOrigin, initOrigin) of

1251 case (findOrigin, initOrigin) of

1234 (Local, Local) -> readMVar resolved >> go

1252 (Local, Local) -> readMVar resolved >> go

1235 (Local, Remote) -> readMVar resolved >> go

1253 (Local, Remote) -> readMVar resolved >> go

1236− (Remote, Local) -> return (theirEndPoint, Nothing)

1254+ (Remote, Local) -> return (theirEndPoint, False, ourAddress < theirAddress)

1237− (Remote, Remote) -> throwIO $ userError "Already connected"

1255+ (Remote, Remote) -> throwIO $ userError "Already connected (B)"

1238 RemoteEndPointValid _ ->

1256 RemoteEndPointValid _ ->

1257+ -- We assume that the request crossed if we find the endpoint in

1258+ -- Valid state. It is possible that this is really an invalid

1259+ -- request, but only in the case of a broken client (we don't

1260+ -- maintain enough history to be able to tell the difference).

1239− return (theirEndPoint, Nothing)

1261+ return (theirEndPoint, False, True)

1240 RemoteEndPointClosing resolved _ ->

1262 RemoteEndPointClosing resolved _ ->

1241 readMVar resolved >> go

1263 readMVar resolved >> go

1242− RemoteEndPointClosed ->

1264+ RemoteEndPointClosed Nothing ->

1243 go

1265 go

1266+ RemoteEndPointClosed (Just err) ->

1267+ throwIO err

1244

1268

1245− ourState = localState ourEndPoint

1269+ ourState = localState ourEndPoint

1270+ ourAddress = localAddress ourEndPoint

1271+

1272+

1273+

1274+

1246

1275

1247-- | Handle requests from a remote endpoint.

1276-- | Handle requests from a remote endpoint.

1248--

1277--

@@ -1252,14 +1281,14 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do

1252 mSock <- withMVar theirState $ \st ->

1281 mSock <- withMVar theirState $ \st ->

1253 case st of

1282 case st of

1254 RemoteEndPointInvalid _ ->

1283 RemoteEndPointInvalid _ ->

1255− relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages"

1284+ relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages (invalid)"

1256 RemoteEndPointInit _ _ ->

1285 RemoteEndPointInit _ _ ->

1257− relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages"

1286+ relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages (init)"

1258 RemoteEndPointValid ep ->

1287 RemoteEndPointValid ep ->

1259 return . Just $ remoteSocket ep

1288 return . Just $ remoteSocket ep

1260 RemoteEndPointClosing _ ep ->

1289 RemoteEndPointClosing _ ep ->

1261 return . Just $ remoteSocket ep

1290 return . Just $ remoteSocket ep

1262− RemoteEndPointClosed ->

1291+ RemoteEndPointClosed _ ->

1263 -- The remote endpoint got closed before we got a chance to start

1292 -- The remote endpoint got closed before we got a chance to start

1264 -- dealing with incoming messages

1293 -- dealing with incoming messages

1265 return Nothing

1294 return Nothing

@@ -1313,9 +1342,11 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do

1313 modifyMVar_ theirState $ \st -> do

1342 modifyMVar_ theirState $ \st -> do

1314 vst <- case st of

1343 vst <- case st of

1315 RemoteEndPointInvalid _ ->

1344 RemoteEndPointInvalid _ ->

1345+ relyViolation (ourEndPoint, theirEndPoint)

1316− relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:createNewConnection"

1346+ "handleIncomingMessages:createNewConnection (invalid)"

1317 RemoteEndPointInit _ _ ->

1347 RemoteEndPointInit _ _ ->

1348+ relyViolation (ourEndPoint, theirEndPoint)

1318− relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:createNewConnection"

1349+ "handleIncomingMessages:createNewConnection (init)"

1319 RemoteEndPointValid vst ->

1350 RemoteEndPointValid vst ->

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

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

1321 RemoteEndPointClosing resolved vst -> do

1352 RemoteEndPointClosing resolved vst -> do

@@ -1327,8 +1358,11 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do

1327 -- to RemoteEndPointValid

1358 -- to RemoteEndPointValid

1328 putMVar resolved ()

1359 putMVar resolved ()

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

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

1330− RemoteEndPointClosed ->

1361+ RemoteEndPointClosed (Just err) ->

1331− throwIO $ userError "Remote endpoint closed"

1362+ throwIO err

1363+ RemoteEndPointClosed Nothing ->

1364+ relyViolation (ourEndPoint, theirEndPoint)

1365+ "createNewConnection (closed)"

1332 sendOn vst ( encodeInt32 ControlResponse

1366 sendOn vst ( encodeInt32 ControlResponse

1333 : encodeInt32 reqId

1367 : encodeInt32 reqId

1334 : prependLength [encodeInt32 newId]

1368 : prependLength [encodeInt32 newId]

@@ -1343,20 +1377,31 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do

1343 readControlResponse :: N.Socket -> ControlRequestId -> IO ()

1377 readControlResponse :: N.Socket -> ControlRequestId -> IO ()

1344 readControlResponse sock reqId = do

1378 readControlResponse sock reqId = do

1345 response <- recvWithLength sock

1379 response <- recvWithLength sock

1346− mmvar <- modifyMVar ourState $ \st -> case st of

1380+ mmvar <- modifyMVar theirState $ \st -> case st of

1347− LocalEndPointValid vst ->

1381+ RemoteEndPointInvalid _ ->

1348− return ( LocalEndPointValid

1382+ relyViolation (ourEndPoint, theirEndPoint)

1383+ "readControlResponse (invalid)"

1384+ RemoteEndPointInit _ _ ->

1385+ relyViolation (ourEndPoint, theirEndPoint)

1386+ "readControlResponse (init)"

1387+ RemoteEndPointValid vst ->

1388+ return ( RemoteEndPointValid

1349 . (pendingCtrlRequestsAt reqId ^= Nothing)

1389 . (pendingCtrlRequestsAt reqId ^= Nothing)

1350 $ vst

1390 $ vst

1351 , vst ^. pendingCtrlRequestsAt reqId

1391 , vst ^. pendingCtrlRequestsAt reqId

1352 )

1392 )

1353− LocalEndPointClosed ->

1393+ RemoteEndPointClosing _ _ ->

1354− throwIO $ userError "Local endpoint closed"

1394+ throwIO $ userError "Invalid control response"

1395+ RemoteEndPointClosed (Just err) ->

1396+ throwIO err

1397+ RemoteEndPointClosed Nothing ->

1398+ relyViolation (ourEndPoint, theirEndPoint)

1399+ "readControlResponse (closed)"

1355 case mmvar of

1400 case mmvar of

1356 Nothing ->

1401 Nothing ->

1357 throwIO $ userError "Invalid request ID"

1402 throwIO $ userError "Invalid request ID"

1358 Just mvar ->

1403 Just mvar ->

1359− putMVar mvar response

1404+ putMVar mvar (Right response)

1360

1405

1361 -- Close a connection

1406 -- Close a connection

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

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

@@ -1365,9 +1410,9 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do

1365 closeConnection cid = do

1410 closeConnection cid = do

1366 modifyMVar_ theirState $ \st -> case st of

1411 modifyMVar_ theirState $ \st -> case st of

1367 RemoteEndPointInvalid _ ->

1412 RemoteEndPointInvalid _ ->

1368− relyViolation (ourEndPoint, theirEndPoint) "closeConnection"

1413+ relyViolation (ourEndPoint, theirEndPoint) "closeConnection (invalid)"

1369 RemoteEndPointInit _ _ ->

1414 RemoteEndPointInit _ _ ->

1370− relyViolation (ourEndPoint, theirEndPoint) "closeConnection"

1415+ relyViolation (ourEndPoint, theirEndPoint) "closeConnection (init)"

1371 RemoteEndPointValid vst -> do

1416 RemoteEndPointValid vst -> do

1372 unless (IntSet.member cid (vst ^. remoteIncoming)) $

1417 unless (IntSet.member cid (vst ^. remoteIncoming)) $

1373 throwIO $ userError "Invalid CloseConnection"

1418 throwIO $ userError "Invalid CloseConnection"

@@ -1380,8 +1425,10 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do

1380 -- far as we are concerned there are no incoming connections. This

1425 -- far as we are concerned there are no incoming connections. This

1381 -- means that a CloseConnection request at this point is invalid.

1426 -- means that a CloseConnection request at this point is invalid.

1382 throwIO $ userError "Invalid CloseConnection request"

1427 throwIO $ userError "Invalid CloseConnection request"

1383− RemoteEndPointClosed ->

1428+ RemoteEndPointClosed (Just err) ->

1384− throwIO $ userError "Remote endpoint closed"

1429+ throwIO err

1430+ RemoteEndPointClosed Nothing ->

1431+ relyViolation (ourEndPoint, theirEndPoint) "closeConnection (closed)"

1385 writeChan ourChannel (ConnectionClosed cid)

1432 writeChan ourChannel (ConnectionClosed cid)

1386 closeIfUnused (ourEndPoint, theirEndPoint)

1433 closeIfUnused (ourEndPoint, theirEndPoint)

1387

1434

@@ -1391,9 +1438,11 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do

1391 modifyMVar theirState $ \st ->

1438 modifyMVar theirState $ \st ->

1392 case st of

1439 case st of

1393 RemoteEndPointInvalid _ ->

1440 RemoteEndPointInvalid _ ->

1394− relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:closeSocket"

1441+ relyViolation (ourEndPoint, theirEndPoint)

1442+ "handleIncomingMessages:closeSocket (invalid)"

1395 RemoteEndPointInit _ _ ->

1443 RemoteEndPointInit _ _ ->

1396− relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:closeSocket"

1444+ relyViolation (ourEndPoint, theirEndPoint)

1445+ "handleIncomingMessages:closeSocket (init)"

1397 RemoteEndPointValid vst -> do

1446 RemoteEndPointValid vst -> do

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

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

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

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

@@ -1408,16 +1457,19 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do

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

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

1409 tryIO $ sendOn vst' [encodeInt32 CloseSocket]

1458 tryIO $ sendOn vst' [encodeInt32 CloseSocket]

1410 tryCloseSocket sock

1459 tryCloseSocket sock

1411− return (RemoteEndPointClosed, True)

1460+ return (RemoteEndPointClosed Nothing, True)

1412 else

1461 else

1413 return (RemoteEndPointValid vst', False)

1462 return (RemoteEndPointValid vst', False)

1414 RemoteEndPointClosing resolved _ -> do

1463 RemoteEndPointClosing resolved _ -> do

1415 removeRemoteEndPoint (ourEndPoint, theirEndPoint)

1464 removeRemoteEndPoint (ourEndPoint, theirEndPoint)

1416 tryCloseSocket sock

1465 tryCloseSocket sock

1417 putMVar resolved ()

1466 putMVar resolved ()

1418− return (RemoteEndPointClosed, True)

1467+ return (RemoteEndPointClosed Nothing, True)

1419− RemoteEndPointClosed ->

1468+ RemoteEndPointClosed (Just err) ->

1420− throwIO $ userError "Remote endpoint closed"

1469+ throwIO err

1470+ RemoteEndPointClosed Nothing ->

1471+ relyViolation (ourEndPoint, theirEndPoint)

1472+ "handleIncomingMessages:closeSocket (closed)"

1421

1473

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

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

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

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

@@ -1428,7 +1480,6 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do

1428

1480

1429 -- Arguments

1481 -- Arguments

1430 ourChannel = localChannel ourEndPoint

1482 ourChannel = localChannel ourEndPoint

1431− ourState = localState ourEndPoint

1432 theirState = remoteState theirEndPoint

1483 theirState = remoteState theirEndPoint

1433 theirAddr = remoteAddress theirEndPoint

1484 theirAddr = remoteAddress theirEndPoint

1434

1485

@@ -1448,12 +1499,13 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do

1448 (remoteAddress theirEndPoint)

1499 (remoteAddress theirEndPoint)

1449 (IntSet.elems $ vst ^. remoteIncoming)

1500 (IntSet.elems $ vst ^. remoteIncoming)

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

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

1502+ forM_ (vst ^. pendingCtrlRequests) $ flip putMVar (Left err)

1451− return RemoteEndPointClosed

1503+ return (RemoteEndPointClosed $ Just err)

1452 RemoteEndPointClosing resolved _ -> do

1504 RemoteEndPointClosing resolved _ -> do

1453 putMVar resolved ()

1505 putMVar resolved ()

1454− return RemoteEndPointClosed

1506+ return (RemoteEndPointClosed $ Just err)

1455− RemoteEndPointClosed ->

1507+ RemoteEndPointClosed err' ->

1456− return RemoteEndPointClosed

1508+ return (RemoteEndPointClosed err')

1457

1509

1458-- | Get the next connection ID

1510-- | Get the next connection ID

1459--

1511--

@@ -1488,12 +1540,6 @@ localEndPoints = accessor _localEndPoints (\es st -> st { _localEndPoints = es }

1488nextEndPointId :: Accessor ValidTransportState EndPointId

1540nextEndPointId :: Accessor ValidTransportState EndPointId

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

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

1490

1542

1491−pendingCtrlRequests :: Accessor ValidLocalEndPointState (IntMap (MVar [ByteString]))

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

1493−

1494−nextCtrlRequestId :: Accessor ValidLocalEndPointState ControlRequestId

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

1496−

1497nextConnectionId :: Accessor ValidLocalEndPointState ConnectionId

1543nextConnectionId :: Accessor ValidLocalEndPointState ConnectionId

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

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

1499

1545

@@ -1512,10 +1558,16 @@ remoteOutgoing = accessor _remoteOutgoing (\cs conn -> conn { _remoteOutgoing =

1512remoteIncoming :: Accessor ValidRemoteEndPointState IntSet

1558remoteIncoming :: Accessor ValidRemoteEndPointState IntSet

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

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

1514

1560

1561+pendingCtrlRequests :: Accessor ValidRemoteEndPointState (IntMap (MVar (Either IOException [ByteString])))

1562+pendingCtrlRequests = accessor _pendingCtrlRequests (\rep st -> st { _pendingCtrlRequests = rep })

1563+

1564+nextCtrlRequestId :: Accessor ValidRemoteEndPointState ControlRequestId

1565+nextCtrlRequestId = accessor _nextCtrlRequestId (\cid st -> st { _nextCtrlRequestId = cid })

1566+

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

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

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

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

1517

1569

1518−pendingCtrlRequestsAt :: ControlRequestId -> Accessor ValidLocalEndPointState (Maybe (MVar [ByteString]))

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

1519pendingCtrlRequestsAt ix = pendingCtrlRequests >>> DAC.intMapMaybe (fromIntegral ix)

1571pendingCtrlRequestsAt ix = pendingCtrlRequests >>> DAC.intMapMaybe (fromIntegral ix)

1520

1572

1521localConnectionTo :: EndPointAddress

1573localConnectionTo :: EndPointAddress

@@ -1534,4 +1586,8 @@ relyViolation (ourEndPoint, theirEndPoint) str = do

1534elog :: EndPointPair -> String -> IO ()

1586elog :: EndPointPair -> String -> IO ()

1535elog (ourEndPoint, theirEndPoint) msg = do

1587elog (ourEndPoint, theirEndPoint) msg = do

1536 tid <- myThreadId

1588 tid <- myThreadId

1537− putStrLn $ show (localAddress ourEndPoint) ++ "/" ++ show (remoteAddress theirEndPoint) ++ "/" ++ show tid ++ ": " ++ msg

1589+ putStrLn $ show (localAddress ourEndPoint)

1590+ ++ "/" ++ show (remoteAddress theirEndPoint)

1591+ ++ "(" ++ show (remoteId theirEndPoint) ++ ")"

1592+ ++ "/" ++ show tid

1593+ ++ ": " ++ msg

network-transport/tests/TestTCP.hs+3−1

@@ -499,10 +499,12 @@ testUnnecessaryConnect nextPort numThreads = do

499 case reply of

500 ConnectionRequestAccepted ->

501 putMVar gotAccepted ()

502+ -- We might get either Invalid or Crossed (the transport does not

503+ -- maintain enough history to be able to tell)

502 ConnectionRequestInvalid ->

504 ConnectionRequestInvalid ->

503 return ()

505 return ()

504 ConnectionRequestCrossed ->

506 ConnectionRequestCrossed ->

505− throwIO $ userError "Unexpected response (Crossed)"

507+ return ()

506 putMVar done ()

508 putMVar done ()

507 return done

509 return done

508

510

network-transport/tests/TestTransport.hs+42−2

@@ -3,7 +3,7 @@ module TestTransport where

4import Prelude hiding (catch, (>>=), (>>), return, fail)

5import TestAuxiliary (forkTry, runTests, trySome, randomThreadDelay)

6−import Control.Concurrent (forkIO, killThread)

6+import Control.Concurrent (forkIO, killThread, yield)

7import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar, readMVar)

8import Control.Exception (evaluate, throw)

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

@@ -859,6 +859,45 @@ testKill newTransport numThreads = do

859

860 takeMVar killerDone

861

862+-- | Set up conditions with a high likelyhood of "crossing" (for transports

863+-- that multiplex lightweight connections across heavyweight connections)

864+testCrossing :: Transport -> Int -> IO ()

865+testCrossing transport numRepeats = do

866+ [aAddr, bAddr] <- replicateM 2 newEmptyMVar

867+ [aDone, bDone] <- replicateM 2 newEmptyMVar

868+ go <- newEmptyMVar

869+

870+ -- A

871+ forkTry $ do

872+ Right endpoint <- newEndPoint transport

873+ putMVar aAddr (address endpoint)

874+ theirAddress <- readMVar bAddr

875+

876+ replicateM_ numRepeats $ do

877+ takeMVar go >> yield

878+ Right conn <- connect endpoint theirAddress ReliableOrdered

879+ close conn

880+ putMVar aDone ()

881+

882+ -- B

883+ forkTry $ do

884+ Right endpoint <- newEndPoint transport

885+ putMVar bAddr (address endpoint)

886+ theirAddress <- readMVar aAddr

887+

888+ replicateM_ numRepeats $ do

889+ takeMVar go >> yield

890+ Right conn <- connect endpoint theirAddress ReliableOrdered

891+ close conn

892+ putMVar bDone ()

893+

894+ -- Driver

895+ forM_ [1 .. numRepeats] $ \_i -> do

896+ putMVar go ()

897+ putMVar go ()

898+ takeMVar aDone

899+ takeMVar bDone

900+

862-- Transport tests

901-- Transport tests

863testTransport :: IO (Either String Transport) -> IO ()

902testTransport :: IO (Either String Transport) -> IO ()

864testTransport newTransport = do

903testTransport newTransport = do

@@ -871,7 +910,8 @@ testTransport newTransport = do

871 , ("CloseOneDirection", testCloseOneDirection transport numPings)

910 , ("CloseOneDirection", testCloseOneDirection transport numPings)

872 , ("CloseReopen", testCloseReopen transport numPings)

911 , ("CloseReopen", testCloseReopen transport numPings)

873 , ("ParallelConnects", testParallelConnects transport numPings)

912 , ("ParallelConnects", testParallelConnects transport numPings)

874− , ("SendAfterClose", testSendAfterClose transport 100)

913+ , ("SendAfterClose", testSendAfterClose transport 1000)

914+ , ("Crossing", testCrossing transport 1000)

875 , ("CloseTwice", testCloseTwice transport 100)

915 , ("CloseTwice", testCloseTwice transport 100)

876 , ("ConnectToSelf", testConnectToSelf transport numPings)

916 , ("ConnectToSelf", testConnectToSelf transport numPings)

877 , ("ConnectToSelfTwice", testConnectToSelfTwice transport numPings)

917 , ("ConnectToSelfTwice", testConnectToSelfTwice transport numPings)