@@ -326,9 +326,11 @@ data RemoteEndPoint = RemoteEndPoint
326326 , remoteId :: Int
327327 }
328328
329+data Origin = Local | Remote
330+
329331data RemoteState =
330332 RemoteEndPointInvalid (TransportError ConnectErrorCode)
331− | RemoteEndPointInit (MVar ())
333+ | RemoteEndPointInit (MVar ()) Origin
332334 | RemoteEndPointValid ValidRemoteEndPointState
333335 | RemoteEndPointClosing (MVar ()) ValidRemoteEndPointState
334336 | RemoteEndPointClosed
@@ -513,7 +515,7 @@ apiSend (ourEndPoint, theirEndPoint) connId connAlive payload =
513515 withRemoteState (ourEndPoint, theirEndPoint) RemoteStatePatternMatch
514516 { caseInvalid = \_ ->
515517 relyViolation (ourEndPoint, theirEndPoint) "apiSend"
516− , caseInit = \_ ->
518+ , caseInit = \_ _ ->
517519 relyViolation (ourEndPoint, theirEndPoint) "apiSend"
518520 , caseValid = \vst -> do
519521 alive <- readIORef connAlive
@@ -564,7 +566,7 @@ apiCloseEndPoint transport evs ourEndPoint = do
564566 case st of
565567 RemoteEndPointInvalid _ ->
566568 return st
567− RemoteEndPointInit resolved -> do
569+ RemoteEndPointInit resolved _ -> do
568570 putMVar resolved ()
569571 return RemoteEndPointClosed
570572 RemoteEndPointValid conn -> do
@@ -647,7 +649,7 @@ internalSocketBetween transport ourAddress theirAddress = do
647649 Nothing -> throwIO $ userError "Remote endpoint not found"
648650 Just ep -> return ep
649651 withMVar (remoteState theirEndPoint) $ \st -> case st of
650− RemoteEndPointInit _ ->
652+ RemoteEndPointInit _ _ ->
651653 throwIO $ userError "Remote endpoint not yet initialized"
652654 RemoteEndPointValid vst ->
653655 return $ remoteSocket vst
@@ -708,8 +710,7 @@ requestConnectionTo :: LocalEndPoint
708710 -> IO (RemoteEndPoint, ConnectionId)
709711requestConnectionTo ourEndPoint theirAddress = do
710712 (theirEndPoint, isNew) <- mapIOException connectFailed $
711− findRemoteEndPoint ourEndPoint theirAddress BlockOnInit
712− (RemoteEndPointValid . (remoteOutgoing ^: (+ 1)))
713+ findRemoteEndPoint ourEndPoint theirAddress Local
713714
714715 -- If this is a new endpoint, start the initialization process and wait
715716 forM_ isNew $ \resolved -> do
@@ -800,7 +801,7 @@ resolveInit (ourEndPoint, theirEndPoint) newState =
800801 modifyMVar_ (remoteState theirEndPoint) $ \st -> case st of
801802 RemoteEndPointInvalid _ ->
802803 relyViolation (ourEndPoint, theirEndPoint) "resolveInit (invalid)"
803− RemoteEndPointInit resolved -> do
804+ RemoteEndPointInit resolved _ -> do
804805 putMVar resolved ()
805806 return newState
806807 RemoteEndPointClosed ->
@@ -917,7 +918,7 @@ doRemoteRequest (ourEndPoint, theirEndPoint) header = do
917918 throwIO (userError "Local endpoint closed")
918919 withRemoteState (ourEndPoint, theirEndPoint) RemoteStatePatternMatch
919920 { caseInvalid = throwIO
920− , caseInit = \_ -> relyViolation (ourEndPoint, theirEndPoint) "doRemoteRequest (init)"
921+ , caseInit = \_ _ -> relyViolation (ourEndPoint, theirEndPoint) "doRemoteRequest (init)"
921922 , caseValid = \vst -> sendOn vst [encodeInt32 header, encodeInt32 reqId]
922923 , caseClosing = \_ _ -> relyViolation (ourEndPoint, theirEndPoint) "doRemoteRequest (closing)"
923924 , caseClosed = throwIO $ userError "Remote endpoint closed"
@@ -988,7 +989,7 @@ forkEndPointThread ourEndPoint p =
988989
989990data RemoteStatePatternMatch a = RemoteStatePatternMatch
990991 { caseInvalid :: TransportError ConnectErrorCode -> IO a
991− , caseInit :: MVar () -> IO a
992+ , caseInit :: MVar () -> Origin -> IO a
992993 , caseValid :: ValidRemoteEndPointState -> IO a
993994 , caseClosing :: MVar () -> ValidRemoteEndPointState -> IO a
994995 , caseClosed :: IO a
@@ -998,7 +999,7 @@ remoteStateIdentity :: RemoteStatePatternMatch RemoteState
998999remoteStateIdentity =
9991000 RemoteStatePatternMatch
10001001 { caseInvalid = return . RemoteEndPointInvalid
1001− , caseInit = return . RemoteEndPointInit
1002+ , caseInit = (return .) . RemoteEndPointInit
10021003 , caseValid = return . RemoteEndPointValid
10031004 , caseClosing = (return .) . RemoteEndPointClosing
10041005 , caseClosed = return RemoteEndPointClosed
@@ -1026,8 +1027,8 @@ modifyRemoteState (ourEndPoint, theirEndPoint) match =
10261027 throwIO ex
10271028 -- The other cases are less interesting, because unless the endpoint is
10281029 -- in Valid state we're not supposed to do any IO on it
1029− RemoteEndPointInit resolved -> do
1030− (st', a) <- onException (restore $ caseInit match resolved)
1030+ RemoteEndPointInit resolved origin -> do
1031+ (st', a) <- onException (restore $ caseInit match resolved origin)
10311032 (putMVar theirState st)
10321033 putMVar theirState st'
10331034 return a
@@ -1068,7 +1069,7 @@ modifyRemoteState_ (ourEndPoint, theirEndPoint) match =
10681069 modifyRemoteState (ourEndPoint, theirEndPoint)
10691070 RemoteStatePatternMatch
10701071 { caseInvalid = u . caseInvalid match
1071− , caseInit = u . caseInit match
1072+ , caseInit = \resolved origin -> u $ caseInit match resolved origin
10721073 , caseValid = u . caseValid match
10731074 , caseClosing = \resolved vst -> u $ caseClosing match resolved vst
10741075 , caseClosed = u $ caseClosed match
@@ -1087,9 +1088,9 @@ withRemoteState (ourEndPoint, theirEndPoint) match =
10871088 { caseInvalid = \err -> do
10881089 a <- caseInvalid match err
10891090 return (RemoteEndPointInvalid err, a)
1090− , caseInit = \resolved -> do
1091− a <- caseInit match resolved
1092− return (RemoteEndPointInit resolved, a)
1091+ , caseInit = \resolved origin -> do
1092+ a <- caseInit match resolved origin
1093+ return (RemoteEndPointInit resolved origin, a)
10931094 , caseValid = \vst -> do
10941095 a <- caseValid match vst
10951096 return (RemoteEndPointValid vst, a)
@@ -1139,11 +1140,9 @@ handleConnectionRequest transport sock = handle handleException $ do
11391140 go ourEndPoint theirAddress = do
11401141 mEndPoint <- handle ((>> return Nothing) . invalidEndPoint) $ do
11411142 (theirEndPoint, isNew) <-
1142− findRemoteEndPoint ourEndPoint theirAddress DontBlockOnInit
1143− (throw $ userError "Already connected")
1143+ findRemoteEndPoint ourEndPoint theirAddress Remote
11441144
11451145 let ourAddress = localAddress ourEndPoint
1146− theirState = remoteState theirEndPoint
11471146 crossed = isNothing isNew && ourAddress < theirAddress
11481147
11491148 if crossed
@@ -1158,8 +1157,8 @@ handleConnectionRequest transport sock = handle handleException $ do
11581157 , _remoteIncoming = IntSet.empty
11591158 , sendOn = sendMany sock
11601159 }
1161− resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst)
11621160 sendMany sock [encodeInt32 ConnectionRequestAccepted]
1161+ resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst)
11631162 return (Just theirEndPoint)
11641163 -- If we left the scope of the exception handler with a return value of
11651164 -- Nothing then the socket is already closed; otherwise, the socket has
@@ -1176,17 +1175,14 @@ handleConnectionRequest transport sock = handle handleException $ do
11761175 tryIO $ sendMany sock [encodeInt32 ConnectionRequestInvalid]
11771176 tryCloseSocket sock
11781177
1179−data BlockOnInit = BlockOnInit | DontBlockOnInit
1180−
11811178-- | Find a remote endpoint. If the remote endpoint does not yet exist we
11821179-- create it in Init state and return the corresponding 'resolved' MVar.
11831180findRemoteEndPoint
11841181 :: LocalEndPoint
11851182 -> EndPointAddress
1186− -> BlockOnInit
1187− -> (ValidRemoteEndPointState -> RemoteState) -- ^ Apply if in Valid state
1183+ -> Origin
11881184 -> IO (RemoteEndPoint, Maybe (MVar ()))
1189−findRemoteEndPoint ourEndPoint theirAddress blockOnInit whenValid = go
1185+findRemoteEndPoint ourEndPoint theirAddress findOrigin = go
11901186 where
11911187 go = do
11921188 (theirEndPoint, isNew) <- modifyMVar ourState $ \st -> case st of
@@ -1195,7 +1191,7 @@ findRemoteEndPoint ourEndPoint theirAddress blockOnInit whenValid = go
11951191 return (st, (theirEndPoint, Nothing))
11961192 Nothing -> do
11971193 resolved <- newEmptyMVar
1198− theirState <- newMVar (RemoteEndPointInit resolved)
1194+ theirState <- newMVar (RemoteEndPointInit resolved findOrigin)
11991195 let theirEndPoint = RemoteEndPoint
12001196 { remoteAddress = theirAddress
12011197 , remoteState = theirState
@@ -1216,18 +1212,29 @@ findRemoteEndPoint ourEndPoint theirAddress blockOnInit whenValid = go
12161212 Nothing -> do
12171213 let theirState = remoteState theirEndPoint
12181214 snapshot <- modifyMVar theirState $ \st -> case st of
1219− RemoteEndPointValid vst -> do
1220− let st' = whenValid vst
1221− return (st', st')
1222− _ ->
1215+ RemoteEndPointValid vst ->
1216+ case findOrigin of
1217+ Local -> do
1218+ let st' = RemoteEndPointValid
1219+ . (remoteOutgoing ^: (+ 1))
1220+ $ vst
1221+ return (st', st')
1222+ Remote ->
1223+ throwIO $ userError "Already connected"
1224+ _ ->
12231225 return (st, st)
1226+ -- The snapshot may no longer be up to date at this point, but if we
1227+ -- increased the refcount then it can only either be valid or closed
1228+ -- (after an explicit call to 'closeEndPoint' or 'closeTransport')
12241229 case snapshot of
12251230 RemoteEndPointInvalid err ->
12261231 throwIO err
1227− RemoteEndPointInit resolved ->
1228− case blockOnInit of
1229− BlockOnInit -> readMVar resolved >> go
1230− DontBlockOnInit -> return (theirEndPoint, Nothing)
1232+ RemoteEndPointInit resolved initOrigin ->
1233+ case (findOrigin, initOrigin) of
1234+ (Local, Local) -> readMVar resolved >> go
1235+ (Local, Remote) -> readMVar resolved >> go
1236+ (Remote, Local) -> return (theirEndPoint, Nothing)
1237+ (Remote, Remote) -> throwIO $ userError "Already connected"
12311238 RemoteEndPointValid _ ->
12321239 return (theirEndPoint, Nothing)
12331240 RemoteEndPointClosing resolved _ ->
@@ -1246,7 +1253,7 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
12461253 case st of
12471254 RemoteEndPointInvalid _ ->
12481255 relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages"
1249− RemoteEndPointInit _ ->
1256+ RemoteEndPointInit _ _ ->
12501257 relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages"
12511258 RemoteEndPointValid ep ->
12521259 return . Just $ remoteSocket ep
@@ -1307,7 +1314,7 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
13071314 vst <- case st of
13081315 RemoteEndPointInvalid _ ->
13091316 relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:createNewConnection"
1310− RemoteEndPointInit _ ->
1317+ RemoteEndPointInit _ _ ->
13111318 relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:createNewConnection"
13121319 RemoteEndPointValid vst ->
13131320 return (remoteIncoming ^: IntSet.insert newId $ vst)
@@ -1359,7 +1366,7 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
13591366 modifyMVar_ theirState $ \st -> case st of
13601367 RemoteEndPointInvalid _ ->
13611368 relyViolation (ourEndPoint, theirEndPoint) "closeConnection"
1362− RemoteEndPointInit _ ->
1369+ RemoteEndPointInit _ _ ->
13631370 relyViolation (ourEndPoint, theirEndPoint) "closeConnection"
13641371 RemoteEndPointValid vst -> do
13651372 unless (IntSet.member cid (vst ^. remoteIncoming)) $
@@ -1385,7 +1392,7 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
13851392 case st of
13861393 RemoteEndPointInvalid _ ->
13871394 relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:closeSocket"
1388− RemoteEndPointInit _ ->
1395+ RemoteEndPointInit _ _ ->
13891396 relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:closeSocket"
13901397 RemoteEndPointValid vst -> do
13911398 -- We regard a CloseSocket message as an (optimized) way for the
@@ -1434,7 +1441,7 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
14341441 case st of
14351442 RemoteEndPointInvalid _ ->
14361443 relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:prematureExit"
1437− RemoteEndPointInit _ ->
1444+ RemoteEndPointInit _ _ ->
14381445 relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:prematureExit"
14391446 RemoteEndPointValid vst -> do
14401447 let code = EventConnectionLost