@@ -84,6 +84,7 @@ import Control.Exception ( IOException
8484 , mask_
8585 , onException
8686 , fromException
87+ , catch
8788 )
8889import Data.IORef (IORef, newIORef, writeIORef, readIORef)
8990import Data.ByteString (ByteString)
@@ -106,7 +107,6 @@ import qualified Data.Map as Map (empty)
106107import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))
107108import qualified Data.Accessor.Container as DAC (mapMaybe, intMapMaybe)
108109import Data.Foldable (forM_, mapM_)
109−import Data.Maybe (isNothing)
110110
111111-- $design
112112--
@@ -241,12 +241,10 @@ data LocalEndPointState =
241241 | LocalEndPointClosed
242242
243243data ValidLocalEndPointState = ValidLocalEndPointState
244− { _nextConnectionId :: !ConnectionId
245− , _pendingCtrlRequests :: IntMap (MVar [ByteString])
246− , _nextCtrlRequestId :: !ControlRequestId
247− , _localConnections :: Map EndPointAddress RemoteEndPoint
248− , _internalThreads :: [ThreadId]
249− , _nextRemoteId :: !Int
244+ { _nextConnectionId :: !ConnectionId
245+ , _localConnections :: Map EndPointAddress RemoteEndPoint
246+ , _internalThreads :: [ThreadId]
247+ , _nextRemoteId :: !Int
250248 }
251249
252250-- REMOTE ENDPOINTS
@@ -327,19 +325,22 @@ data RemoteEndPoint = RemoteEndPoint
327325 }
328326
329327data Origin = Local | Remote
328+ deriving (Eq, Show)
330329
331330data RemoteState =
332331 RemoteEndPointInvalid (TransportError ConnectErrorCode)
333332 | RemoteEndPointInit (MVar ()) Origin
334333 | RemoteEndPointValid ValidRemoteEndPointState
335334 | RemoteEndPointClosing (MVar ()) ValidRemoteEndPointState
336− | RemoteEndPointClosed
335+ | RemoteEndPointClosed (Maybe IOException)
337336
338337data ValidRemoteEndPointState = ValidRemoteEndPointState
339− { _remoteOutgoing :: !Int
340− , _remoteIncoming :: IntSet
341− , remoteSocket :: N.Socket
342− , sendOn :: [ByteString] -> IO ()
338+ { _remoteOutgoing :: !Int
339+ , _remoteIncoming :: IntSet
340+ , remoteSocket :: N.Socket
341+ , sendOn :: [ByteString] -> IO ()
342+ , _pendingCtrlRequests :: IntMap (MVar (Either IOException [ByteString]))
343+ , _nextCtrlRequestId :: !ControlRequestId
343344 }
344345
345346type EndPointId = Int32
@@ -476,6 +477,7 @@ apiConnect ourEndPoint theirAddress _reliability _hints =
476477 if localAddress ourEndPoint == theirAddress
477478 then connectToSelf ourEndPoint
478479 else try $ do
480+ resetIfInvalid ourEndPoint theirAddress
479481 (theirEndPoint, connId) <- requestConnectionTo ourEndPoint theirAddress
480482 -- connAlive can be an IORef rather than an MVar because it is protected
481483 -- by the remoteState MVar. We don't need the overhead of locking twice.
@@ -525,12 +527,12 @@ apiSend (ourEndPoint, theirEndPoint) connId connAlive payload =
525527 , caseClosing = \_ _ -> do
526528 alive <- readIORef connAlive
527529 if alive
528− then throwIO $ TransportError SendFailed "Connection lost"
530+ then relyViolation (ourEndPoint, theirEndPoint) "apiSend"
529531 else throwIO $ TransportError SendClosed "Connection closed"
530− , caseClosed = do
532+ , caseClosed = \err -> do
531533 alive <- readIORef connAlive
532534 if alive
533− then throwIO $ TransportError SendFailed "Connection lost"
535+ then throwIO $ TransportError SendFailed (show err)
534536 else throwIO $ TransportError SendClosed "Connection closed"
535537 }
536538 where
@@ -568,17 +570,17 @@ apiCloseEndPoint transport evs ourEndPoint = do
568570 return st
569571 RemoteEndPointInit resolved _ -> do
570572 putMVar resolved ()
571− return RemoteEndPointClosed
573+ return $ RemoteEndPointClosed (Just . userError $ "apiCloseEndPoint")
572574 RemoteEndPointValid conn -> do
573575 tryIO $ sendOn conn [encodeInt32 CloseSocket]
574576 tryCloseSocket (remoteSocket conn)
575− return RemoteEndPointClosed
577+ return $ RemoteEndPointClosed (Just . userError $ "apiCloseEndPoint")
576578 RemoteEndPointClosing resolved conn -> do
577579 putMVar resolved ()
578580 tryCloseSocket (remoteSocket conn)
579− return RemoteEndPointClosed
580− RemoteEndPointClosed ->
581− return RemoteEndPointClosed
581+ return $ RemoteEndPointClosed (Just . userError $ "apiCloseEndPoint")
582+ RemoteEndPointClosed err ->
583+ return $ RemoteEndPointClosed err
582584
583585-- | Special case of 'apiConnect': connect an endpoint to itself
584586connectToSelf :: LocalEndPoint
@@ -655,7 +657,7 @@ internalSocketBetween transport ourAddress theirAddress = do
655657 return $ remoteSocket vst
656658 RemoteEndPointClosing _ vst ->
657659 return $ remoteSocket vst
658− RemoteEndPointClosed ->
660+ RemoteEndPointClosed _ ->
659661 throwIO $ userError "Remote endpoint closed"
660662 RemoteEndPointInvalid _ ->
661663 throwIO $ userError "Remote endpoint invalid"
@@ -664,6 +666,28 @@ internalSocketBetween transport ourAddress theirAddress = do
664666-- Lower level functionality --
665667--------------------------------------------------------------------------------
666668
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+
667691-- | Create a new local endpoint
668692--
669693-- May throw a TransportError NewEndPointErrorCode exception if the transport
@@ -673,8 +697,6 @@ createLocalEndPoint transport = do
673697 chan <- newChan
674698 state <- newMVar . LocalEndPointValid $ ValidLocalEndPointState
675699 { _nextConnectionId = firstNonReservedConnectionId
676− , _pendingCtrlRequests = IntMap.empty
677− , _nextCtrlRequestId = 0
678700 , _localConnections = Map.empty
679701 , _internalThreads = []
680702 , _nextRemoteId = 0
@@ -708,47 +730,43 @@ createLocalEndPoint transport = do
708730requestConnectionTo :: LocalEndPoint
709731 -> EndPointAddress
710732 -> IO (RemoteEndPoint, ConnectionId)
711−requestConnectionTo ourEndPoint theirAddress = do
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)
733+requestConnectionTo ourEndPoint theirAddress = go
725734 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+
726750 connectFailed = TransportError ConnectFailed . show
727751
728752-- | Set up a remote endpoint
729−setupRemoteEndPoint :: EndPointPair -- ^ Local endpoint, remote endpoint
730− -> Int -- ^ Initial (outgoing) refcount
731− -> IO ()
732−setupRemoteEndPoint (ourEndPoint, theirEndPoint) outgoing = do
753+setupRemoteEndPoint :: EndPointPair -> IO ()
754+setupRemoteEndPoint (ourEndPoint, theirEndPoint) = do
733755 didAccept <- bracketOnError (socketToEndPoint ourAddress theirAddress)
734756 onError $ \result ->
735757 case result of
736758 Right (sock, ConnectionRequestAccepted) -> do
737759 let vst = ValidRemoteEndPointState
738− { remoteSocket = sock
739− , _remoteOutgoing = outgoing
740− , _remoteIncoming = IntSet.empty
741− , sendOn = sendMany sock
760+ { remoteSocket = sock
761+ , _remoteOutgoing = 0
762+ , _remoteIncoming = IntSet.empty
763+ , sendOn = sendMany sock
764+ , _pendingCtrlRequests = IntMap.empty
765+ , _nextCtrlRequestId = 0
742766 }
743767 resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst)
744768 return True
745769 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)
752770 let err = invalidAddress "Invalid endpoint"
753771 resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointInvalid err)
754772 tryCloseSocket sock
@@ -759,7 +777,6 @@ setupRemoteEndPoint (ourEndPoint, theirEndPoint) outgoing = do
759777 tryCloseSocket sock
760778 return False
761779 Left err -> do
762− removeRemoteEndPoint (ourEndPoint, theirEndPoint) -- See comment above
763780 resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointInvalid err)
764781 return False
765782
@@ -780,8 +797,7 @@ setupRemoteEndPoint (ourEndPoint, theirEndPoint) outgoing = do
780797 onError :: Either (TransportError ConnectErrorCode)
781798 (N.Socket, ConnectionRequestResponse)
782799 -> IO ()
783− onError result = do
784− removeRemoteEndPoint (ourEndPoint, theirEndPoint)
800+ onError result =
785801 case result of
786802 Left err ->
787803 void $ tryPutMVar theirState (RemoteEndPointInvalid err)
@@ -796,21 +812,23 @@ setupRemoteEndPoint (ourEndPoint, theirEndPoint) outgoing = do
796812 theirState = remoteState theirEndPoint
797813 invalidAddress = TransportError ConnectNotFound
798814
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.
799818resolveInit :: EndPointPair -> RemoteState -> IO ()
800−resolveInit (ourEndPoint, theirEndPoint) newState =
801− modifyMVar_ (remoteState theirEndPoint) $ \st -> case st of
802− RemoteEndPointInvalid _ ->
803− relyViolation (ourEndPoint, theirEndPoint) "resolveInit (invalid)"
804− RemoteEndPointInit resolved _ -> do
805− putMVar resolved ()
806− return newState
807− RemoteEndPointClosed ->
808− throwIO $ userError "Remote endpoint closed"
809− RemoteEndPointClosing _ _ ->
810− relyViolation (ourEndPoint, theirEndPoint) "resolveInit (closing)"
811− RemoteEndPointValid _ ->
812− relyViolation (ourEndPoint, theirEndPoint) "resolveInit (valid)"
813−
819+resolveInit (ourEndPoint, theirEndPoint) newState =
820+ modifyRemoteState_ (ourEndPoint, theirEndPoint) $ remoteStateIdentity
821+ { caseInit = \resolved _ -> do
822+ putMVar resolved ()
823+ return newState
824+ , caseInvalid = \_ -> throwException
825+ , caseValid = \_ -> throwException
826+ , caseClosing = \_ _ -> throwException
827+ , caseClosed = \_ -> throwException
828+ }
829+ where
830+ throwException = throwIO $ userError "resolveInit: not in Init state"
831+
814832-- | Establish a connection to a remote endpoint
815833socketToEndPoint :: EndPointAddress -- ^ Our address
816834 -> EndPointAddress -- ^ Their address
@@ -904,26 +922,26 @@ decodeEndPointAddress (EndPointAddress bs) =
904922-- or if the send fails.
905923doRemoteRequest :: EndPointPair -> ControlHeader -> IO [ByteString]
906924doRemoteRequest (ourEndPoint, theirEndPoint) header = do
907− reply <- newEmptyMVar
908− reqId <- modifyMVar (localState ourEndPoint) $ \st -> case st of
909− LocalEndPointValid vst -> do
910− let reqId = vst ^. nextCtrlRequestId
911− return ( LocalEndPointValid
912− . (nextCtrlRequestId ^: (+ 1))
913− . (pendingCtrlRequestsAt reqId ^= Just reply)
914− $ vst
915− , reqId
916− )
917− LocalEndPointClosed ->
918− throwIO (userError "Local endpoint closed")
919− withRemoteState (ourEndPoint, theirEndPoint) RemoteStatePatternMatch
920− { caseInvalid = throwIO
925+ replyMVar <- newEmptyMVar
926+ modifyRemoteState_ (ourEndPoint, theirEndPoint) RemoteStatePatternMatch
927+ { caseValid = \vst -> do
928+ let reqId = vst ^. nextCtrlRequestId
929+ sendOn vst [encodeInt32 header, encodeInt32 reqId]
930+ return ( RemoteEndPointValid
931+ . (nextCtrlRequestId ^: (+ 1))
932+ . (pendingCtrlRequestsAt reqId ^= Just replyMVar)
933+ $ vst
934+ )
935+ -- Error cases
936+ , caseInvalid = throwIO
921937 , caseInit = \_ _ -> relyViolation (ourEndPoint, theirEndPoint) "doRemoteRequest (init)"
922− , caseValid = \vst -> sendOn vst [encodeInt32 header, encodeInt32 reqId]
923938 , caseClosing = \_ _ -> relyViolation (ourEndPoint, theirEndPoint) "doRemoteRequest (closing)"
924− , caseClosed = throwIO $ userError "Remote endpoint closed"
939+ , caseClosed = \_ -> throwIO $ userError "doRemoteRequest: Remote endpoint closed"
925940 }
926− takeMVar reply
941+ mReply <- takeMVar replyMVar
942+ case mReply of
943+ Left err -> throwIO err
944+ Right reply -> return reply
927945
928946-- | Send a CloseSocket request if the remote endpoint is unused
929947closeIfUnused :: EndPointPair -> IO ()
@@ -992,7 +1010,7 @@ data RemoteStatePatternMatch a = RemoteStatePatternMatch
9921010 , caseInit :: MVar () -> Origin -> IO a
9931011 , caseValid :: ValidRemoteEndPointState -> IO a
9941012 , caseClosing :: MVar () -> ValidRemoteEndPointState -> IO a
995− , caseClosed :: IO a
1013+ , caseClosed :: Maybe IOException -> IO a
9961014 }
9971015
9981016remoteStateIdentity :: RemoteStatePatternMatch RemoteState
@@ -1002,7 +1020,7 @@ remoteStateIdentity =
10021020 , caseInit = (return .) . RemoteEndPointInit
10031021 , caseValid = return . RemoteEndPointValid
10041022 , caseClosing = (return .) . RemoteEndPointClosing
1005− , caseClosed = return RemoteEndPointClosed
1023+ , caseClosed = return . RemoteEndPointClosed
10061024 }
10071025
10081026-- | Like modifyMVar, but if an I/O exception occurs don't restore the remote
@@ -1042,8 +1060,8 @@ modifyRemoteState (ourEndPoint, theirEndPoint) match =
10421060 (putMVar theirState st)
10431061 putMVar theirState st'
10441062 return a
1045− RemoteEndPointClosed -> do
1046− (st', a) <- onException (restore $ caseClosed match)
1063+ RemoteEndPointClosed err -> do
1064+ (st', a) <- onException (restore $ caseClosed match err)
10471065 (putMVar theirState st)
10481066 putMVar theirState st'
10491067 return a
@@ -1055,7 +1073,7 @@ modifyRemoteState (ourEndPoint, theirEndPoint) match =
10551073 handleIOException ex vst = do
10561074 tryCloseSocket (remoteSocket vst)
10571075 removeRemoteEndPoint (ourEndPoint, theirEndPoint)
1058− putMVar theirState RemoteEndPointClosed
1076+ putMVar theirState (RemoteEndPointClosed $ Just ex)
10591077 let incoming = IntSet.elems $ vst ^. remoteIncoming
10601078 code = EventConnectionLost (remoteAddress theirEndPoint) incoming
10611079 err = TransportError code (show ex)
@@ -1072,7 +1090,7 @@ modifyRemoteState_ (ourEndPoint, theirEndPoint) match =
10721090 , caseInit = \resolved origin -> u $ caseInit match resolved origin
10731091 , caseValid = u . caseValid match
10741092 , caseClosing = \resolved vst -> u $ caseClosing match resolved vst
1075− , caseClosed = u $ caseClosed match
1093+ , caseClosed = u . caseClosed match
10761094 }
10771095 where
10781096 u :: IO a -> IO (a, ())
@@ -1097,9 +1115,9 @@ withRemoteState (ourEndPoint, theirEndPoint) match =
10971115 , caseClosing = \resolved vst -> do
10981116 a <- caseClosing match resolved vst
10991117 return (RemoteEndPointClosing resolved vst, a)
1100− , caseClosed = do
1101− a <- caseClosed match
1102− return (RemoteEndPointClosed, a)
1118+ , caseClosed = \err -> do
1119+ a <- caseClosed match err
1120+ return (RemoteEndPointClosed err, a)
11031121 }
11041122
11051123--------------------------------------------------------------------------------
@@ -1139,12 +1157,9 @@ handleConnectionRequest transport sock = handle handleException $ do
11391157 go :: LocalEndPoint -> EndPointAddress -> IO ()
11401158 go ourEndPoint theirAddress = do
11411159 mEndPoint <- handle ((>> return Nothing) . invalidEndPoint) $ do
1142− (theirEndPoint, isNew) <-
1160+ (theirEndPoint, _, crossed) <-
11431161 findRemoteEndPoint ourEndPoint theirAddress Remote
1144−
1145− let ourAddress = localAddress ourEndPoint
1146− crossed = isNothing isNew && ourAddress < theirAddress
1147−
1162+
11481163 if crossed
11491164 then do
11501165 tryIO $ sendMany sock [encodeInt32 ConnectionRequestCrossed]
@@ -1152,10 +1167,12 @@ handleConnectionRequest transport sock = handle handleException $ do
11521167 return Nothing
11531168 else do
11541169 let vst = ValidRemoteEndPointState
1155− { remoteSocket = sock
1156− , _remoteOutgoing = 0
1157− , _remoteIncoming = IntSet.empty
1158− , sendOn = sendMany sock
1170+ { remoteSocket = sock
1171+ , _remoteOutgoing = 0
1172+ , _remoteIncoming = IntSet.empty
1173+ , sendOn = sendMany sock
1174+ , _pendingCtrlRequests = IntMap.empty
1175+ , _nextCtrlRequestId = 0
11591176 }
11601177 sendMany sock [encodeInt32 ConnectionRequestAccepted]
11611178 resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst)
@@ -1176,19 +1193,20 @@ handleConnectionRequest transport sock = handle handleException $ do
11761193 tryCloseSocket sock
11771194
11781195-- | 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.
11801198findRemoteEndPoint
11811199 :: LocalEndPoint
11821200 -> EndPointAddress
11831201 -> Origin
1184− -> IO (RemoteEndPoint, Maybe (MVar ()))
1202+ -> IO (RemoteEndPoint, Bool, Bool)
11851203findRemoteEndPoint ourEndPoint theirAddress findOrigin = go
11861204 where
11871205 go = do
11881206 (theirEndPoint, isNew) <- modifyMVar ourState $ \st -> case st of
11891207 LocalEndPointValid vst -> case vst ^. localConnectionTo theirAddress of
11901208 Just theirEndPoint ->
1191− return (st, (theirEndPoint, Nothing))
1209+ return (st, (theirEndPoint, False))
11921210 Nothing -> do
11931211 resolved <- newEmptyMVar
11941212 theirState <- newMVar (RemoteEndPointInit resolved findOrigin)
@@ -1201,15 +1219,15 @@ findRemoteEndPoint ourEndPoint theirAddress findOrigin = go
12011219 . (localConnectionTo theirAddress ^= Just theirEndPoint)
12021220 . (nextRemoteId ^: (+ 1))
12031221 $ vst
1204− , (theirEndPoint, Just resolved)
1222+ , (theirEndPoint, True)
12051223 )
12061224 LocalEndPointClosed ->
12071225 throwIO $ userError "Local endpoint closed"
1208−
1209− case isNew of
1210− Just resolved ->
1211− return (theirEndPoint, Just resolved)
1212− Nothing -> do
1226+
1227+ if isNew
1228+ then
1229+ return (theirEndPoint, True, False)
1230+ else do
12131231 let theirState = remoteState theirEndPoint
12141232 snapshot <- modifyMVar theirState $ \st -> case st of
12151233 RemoteEndPointValid vst ->
@@ -1220,7 +1238,7 @@ findRemoteEndPoint ourEndPoint theirAddress findOrigin = go
12201238 $ vst
12211239 return (st', st')
12221240 Remote ->
1223− throwIO $ userError "Already connected"
1241+ return (st, st)
12241242 _ ->
12251243 return (st, st)
12261244 -- The snapshot may no longer be up to date at this point, but if we
@@ -1233,16 +1251,27 @@ findRemoteEndPoint ourEndPoint theirAddress findOrigin = go
12331251 case (findOrigin, initOrigin) of
12341252 (Local, Local) -> readMVar resolved >> go
12351253 (Local, Remote) -> readMVar resolved >> go
1236− (Remote, Local) -> return (theirEndPoint, Nothing)
1237− (Remote, Remote) -> throwIO $ userError "Already connected"
1254+ (Remote, Local) -> return (theirEndPoint, False, ourAddress < theirAddress)
1255+ (Remote, Remote) -> throwIO $ userError "Already connected (B)"
12381256 RemoteEndPointValid _ ->
1239− return (theirEndPoint, Nothing)
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).
1261+ return (theirEndPoint, False, True)
12401262 RemoteEndPointClosing resolved _ ->
12411263 readMVar resolved >> go
1242− RemoteEndPointClosed ->
1264+ RemoteEndPointClosed Nothing ->
12431265 go
1266+ RemoteEndPointClosed (Just err) ->
1267+ throwIO err
12441268
1245− ourState = localState ourEndPoint
1269+ ourState = localState ourEndPoint
1270+ ourAddress = localAddress ourEndPoint
1271+
1272+
1273+
1274+
12461275
12471276-- | Handle requests from a remote endpoint.
12481277--
@@ -1252,14 +1281,14 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
12521281 mSock <- withMVar theirState $ \st ->
12531282 case st of
12541283 RemoteEndPointInvalid _ ->
1255− relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages"
1284+ relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages (invalid)"
12561285 RemoteEndPointInit _ _ ->
1257− relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages"
1286+ relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages (init)"
12581287 RemoteEndPointValid ep ->
12591288 return . Just $ remoteSocket ep
12601289 RemoteEndPointClosing _ ep ->
12611290 return . Just $ remoteSocket ep
1262− RemoteEndPointClosed ->
1291+ RemoteEndPointClosed _ ->
12631292 -- The remote endpoint got closed before we got a chance to start
12641293 -- dealing with incoming messages
12651294 return Nothing
@@ -1313,9 +1342,11 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
13131342 modifyMVar_ theirState $ \st -> do
13141343 vst <- case st of
13151344 RemoteEndPointInvalid _ ->
1316− relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:createNewConnection"
1345+ relyViolation (ourEndPoint, theirEndPoint)
1346+ "handleIncomingMessages:createNewConnection (invalid)"
13171347 RemoteEndPointInit _ _ ->
1318− relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:createNewConnection"
1348+ relyViolation (ourEndPoint, theirEndPoint)
1349+ "handleIncomingMessages:createNewConnection (init)"
13191350 RemoteEndPointValid vst ->
13201351 return (remoteIncoming ^: IntSet.insert newId $ vst)
13211352 RemoteEndPointClosing resolved vst -> do
@@ -1327,8 +1358,11 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
13271358 -- to RemoteEndPointValid
13281359 putMVar resolved ()
13291360 return (remoteIncoming ^= IntSet.singleton newId $ vst)
1330− RemoteEndPointClosed ->
1331− throwIO $ userError "Remote endpoint closed"
1361+ RemoteEndPointClosed (Just err) ->
1362+ throwIO err
1363+ RemoteEndPointClosed Nothing ->
1364+ relyViolation (ourEndPoint, theirEndPoint)
1365+ "createNewConnection (closed)"
13321366 sendOn vst ( encodeInt32 ControlResponse
13331367 : encodeInt32 reqId
13341368 : prependLength [encodeInt32 newId]
@@ -1343,20 +1377,31 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
13431377 readControlResponse :: N.Socket -> ControlRequestId -> IO ()
13441378 readControlResponse sock reqId = do
13451379 response <- recvWithLength sock
1346− mmvar <- modifyMVar ourState $ \st -> case st of
1347− LocalEndPointValid vst ->
1348− return ( LocalEndPointValid
1380+ mmvar <- modifyMVar theirState $ \st -> case st of
1381+ RemoteEndPointInvalid _ ->
1382+ relyViolation (ourEndPoint, theirEndPoint)
1383+ "readControlResponse (invalid)"
1384+ RemoteEndPointInit _ _ ->
1385+ relyViolation (ourEndPoint, theirEndPoint)
1386+ "readControlResponse (init)"
1387+ RemoteEndPointValid vst ->
1388+ return ( RemoteEndPointValid
13491389 . (pendingCtrlRequestsAt reqId ^= Nothing)
13501390 $ vst
13511391 , vst ^. pendingCtrlRequestsAt reqId
13521392 )
1353− LocalEndPointClosed ->
1354− throwIO $ userError "Local endpoint closed"
1393+ RemoteEndPointClosing _ _ ->
1394+ throwIO $ userError "Invalid control response"
1395+ RemoteEndPointClosed (Just err) ->
1396+ throwIO err
1397+ RemoteEndPointClosed Nothing ->
1398+ relyViolation (ourEndPoint, theirEndPoint)
1399+ "readControlResponse (closed)"
13551400 case mmvar of
13561401 Nothing ->
13571402 throwIO $ userError "Invalid request ID"
13581403 Just mvar ->
1359− putMVar mvar response
1404+ putMVar mvar (Right response)
13601405
13611406 -- Close a connection
13621407 -- It is important that we verify that the connection is in fact open,
@@ -1365,9 +1410,9 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
13651410 closeConnection cid = do
13661411 modifyMVar_ theirState $ \st -> case st of
13671412 RemoteEndPointInvalid _ ->
1368− relyViolation (ourEndPoint, theirEndPoint) "closeConnection"
1413+ relyViolation (ourEndPoint, theirEndPoint) "closeConnection (invalid)"
13691414 RemoteEndPointInit _ _ ->
1370− relyViolation (ourEndPoint, theirEndPoint) "closeConnection"
1415+ relyViolation (ourEndPoint, theirEndPoint) "closeConnection (init)"
13711416 RemoteEndPointValid vst -> do
13721417 unless (IntSet.member cid (vst ^. remoteIncoming)) $
13731418 throwIO $ userError "Invalid CloseConnection"
@@ -1380,8 +1425,10 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
13801425 -- far as we are concerned there are no incoming connections. This
13811426 -- means that a CloseConnection request at this point is invalid.
13821427 throwIO $ userError "Invalid CloseConnection request"
1383− RemoteEndPointClosed ->
1384− throwIO $ userError "Remote endpoint closed"
1428+ RemoteEndPointClosed (Just err) ->
1429+ throwIO err
1430+ RemoteEndPointClosed Nothing ->
1431+ relyViolation (ourEndPoint, theirEndPoint) "closeConnection (closed)"
13851432 writeChan ourChannel (ConnectionClosed cid)
13861433 closeIfUnused (ourEndPoint, theirEndPoint)
13871434
@@ -1391,9 +1438,11 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
13911438 modifyMVar theirState $ \st ->
13921439 case st of
13931440 RemoteEndPointInvalid _ ->
1394− relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:closeSocket"
1441+ relyViolation (ourEndPoint, theirEndPoint)
1442+ "handleIncomingMessages:closeSocket (invalid)"
13951443 RemoteEndPointInit _ _ ->
1396− relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:closeSocket"
1444+ relyViolation (ourEndPoint, theirEndPoint)
1445+ "handleIncomingMessages:closeSocket (init)"
13971446 RemoteEndPointValid vst -> do
13981447 -- We regard a CloseSocket message as an (optimized) way for the
13991448 -- remote endpoint to indicate that all its connections to us are
@@ -1408,16 +1457,19 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
14081457 -- Attempt to reply (but don't insist)
14091458 tryIO $ sendOn vst' [encodeInt32 CloseSocket]
14101459 tryCloseSocket sock
1411− return (RemoteEndPointClosed, True)
1460+ return (RemoteEndPointClosed Nothing, True)
14121461 else
14131462 return (RemoteEndPointValid vst', False)
14141463 RemoteEndPointClosing resolved _ -> do
14151464 removeRemoteEndPoint (ourEndPoint, theirEndPoint)
14161465 tryCloseSocket sock
14171466 putMVar resolved ()
1418− return (RemoteEndPointClosed, True)
1419− RemoteEndPointClosed ->
1420− throwIO $ userError "Remote endpoint closed"
1467+ return (RemoteEndPointClosed Nothing, True)
1468+ RemoteEndPointClosed (Just err) ->
1469+ throwIO err
1470+ RemoteEndPointClosed Nothing ->
1471+ relyViolation (ourEndPoint, theirEndPoint)
1472+ "handleIncomingMessages:closeSocket (closed)"
14211473
14221474 -- Read a message and output it on the endPoint's channel By rights we
14231475 -- should verify that the connection ID is valid, but this is unnecessary
@@ -1428,7 +1480,6 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
14281480
14291481 -- Arguments
14301482 ourChannel = localChannel ourEndPoint
1431− ourState = localState ourEndPoint
14321483 theirState = remoteState theirEndPoint
14331484 theirAddr = remoteAddress theirEndPoint
14341485
@@ -1448,12 +1499,13 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
14481499 (remoteAddress theirEndPoint)
14491500 (IntSet.elems $ vst ^. remoteIncoming)
14501501 writeChan ourChannel . ErrorEvent $ TransportError code (show err)
1451− return RemoteEndPointClosed
1502+ forM_ (vst ^. pendingCtrlRequests) $ flip putMVar (Left err)
1503+ return (RemoteEndPointClosed $ Just err)
14521504 RemoteEndPointClosing resolved _ -> do
14531505 putMVar resolved ()
1454− return RemoteEndPointClosed
1455− RemoteEndPointClosed ->
1456− return RemoteEndPointClosed
1506+ return (RemoteEndPointClosed $ Just err)
1507+ RemoteEndPointClosed err' ->
1508+ return (RemoteEndPointClosed err')
14571509
14581510-- | Get the next connection ID
14591511--
@@ -1488,12 +1540,6 @@ localEndPoints = accessor _localEndPoints (\es st -> st { _localEndPoints = es }
14881540nextEndPointId :: Accessor ValidTransportState EndPointId
14891541nextEndPointId = accessor _nextEndPointId (\eid st -> st { _nextEndPointId = eid })
14901542
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−
14971543nextConnectionId :: Accessor ValidLocalEndPointState ConnectionId
14981544nextConnectionId = accessor _nextConnectionId (\cix st -> st { _nextConnectionId = cix })
14991545
@@ -1512,10 +1558,16 @@ remoteOutgoing = accessor _remoteOutgoing (\cs conn -> conn { _remoteOutgoing =
15121558remoteIncoming :: Accessor ValidRemoteEndPointState IntSet
15131559remoteIncoming = accessor _remoteIncoming (\cs conn -> conn { _remoteIncoming = cs })
15141560
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+
15151567localEndPointAt :: EndPointAddress -> Accessor ValidTransportState (Maybe LocalEndPoint)
15161568localEndPointAt addr = localEndPoints >>> DAC.mapMaybe addr
15171569
1518−pendingCtrlRequestsAt :: ControlRequestId -> Accessor ValidLocalEndPointState (Maybe (MVar [ByteString]))
1570+pendingCtrlRequestsAt :: ControlRequestId -> Accessor ValidRemoteEndPointState (Maybe (MVar (Either IOException [ByteString])))
15191571pendingCtrlRequestsAt ix = pendingCtrlRequests >>> DAC.intMapMaybe (fromIntegral ix)
15201572
15211573localConnectionTo :: EndPointAddress
@@ -1534,4 +1586,8 @@ relyViolation (ourEndPoint, theirEndPoint) str = do
15341586elog :: EndPointPair -> String -> IO ()
15351587elog (ourEndPoint, theirEndPoint) msg = do
15361588 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