@@ -266,8 +266,12 @@ data LocalEndPointState =
266266 | LocalEndPointClosed
267267
268268data ValidLocalEndPointState = ValidLocalEndPointState
269− { _nextConnOutId :: !LightweightConnectionId
270− , _nextConnInId :: !HeavyweightConnectionId
269+ { -- Next available ID for an outgoing lightweight self-connection
270+ -- (see also remoteNextConnOutId)
271+ _localNextConnOutId :: !LightweightConnectionId
272+ -- Next available ID for an incoming heavyweight connection
273+ , _nextConnInId :: !HeavyweightConnectionId
274+ -- Currently active outgoing heavyweight connections
271275 , _localConnections :: !(Map EndPointAddress RemoteEndPoint)
272276 }
273277
@@ -387,6 +391,7 @@ data ValidRemoteEndPointState = ValidRemoteEndPointState
387391 { _remoteOutgoing :: !Int
388392 , _remoteIncoming :: !(Set LightweightConnectionId)
389393 , _remoteMaxIncoming :: !LightweightConnectionId
394+ , _remoteNextConnOutId :: !LightweightConnectionId
390395 , remoteSocket :: !N.Socket
391396 , remoteSendLock :: !(MVar ())
392397 }
@@ -627,7 +632,8 @@ apiSend (ourEndPoint, theirEndPoint) connId connAlive payload =
627632 RemoteEndPointValid vst -> do
628633 alive <- readIORef connAlive
629634 if alive
630− then schedule theirEndPoint $ sendOn vst (encodeInt32 connId : prependLength payload)
635+ then schedule theirEndPoint $
636+ sendOn vst (encodeInt32 connId : prependLength payload)
631637 else throwIO $ TransportError SendClosed "Connection closed"
632638 RemoteEndPointClosing _ _ -> do
633639 alive <- readIORef connAlive
@@ -749,11 +755,12 @@ handleConnectionRequest transport sock = handle handleException $ do
749755 else do
750756 sendLock <- newMVar ()
751757 let vst = ValidRemoteEndPointState
752− { remoteSocket = sock
753− , remoteSendLock = sendLock
754− , _remoteOutgoing = 0
755− , _remoteIncoming = Set.empty
756− , _remoteMaxIncoming = 0
758+ { remoteSocket = sock
759+ , remoteSendLock = sendLock
760+ , _remoteOutgoing = 0
761+ , _remoteIncoming = Set.empty
762+ , _remoteMaxIncoming = 0
763+ , _remoteNextConnOutId = firstNonReservedLightweightConnectionId
757764 }
758765 sendMany sock [encodeInt32 ConnectionRequestAccepted]
759766 resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst)
@@ -896,7 +903,6 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
896903 closeSocket :: N.Socket -> LightweightConnectionId -> IO Bool
897904 closeSocket sock lastReceivedId = do
898905 mAct <- modifyMVar theirState $ \st -> do
899− lastSentId <- getLastConnOutId ourEndPoint
900906 case st of
901907 RemoteEndPointInvalid _ ->
902908 relyViolation (ourEndPoint, theirEndPoint)
@@ -918,7 +924,7 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
918924 -- then we sent a ConnectionCreated *AND* a ConnectionClosed
919925 -- message to the remote endpoint, *both of which* it did not yet
920926 -- receive before sending the CloseSocket request.
921− if vst' ^. remoteOutgoing > 0 || lastReceivedId < lastSentId
927+ if vst' ^. remoteOutgoing > 0 || lastReceivedId < lastSentId vst
922928 then
923929 return (RemoteEndPointValid vst', Nothing)
924930 else do
@@ -930,7 +936,7 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
930936 ]
931937 tryCloseSocket sock
932938 return (RemoteEndPointClosed, Just act)
933− RemoteEndPointClosing resolved vst ->
939+ RemoteEndPointClosing resolved vst -> do
934940 -- Like above, we need to check if there is a ConnectionCreated
935941 -- message that we sent but that the remote endpoint has not yet
936942 -- received. However, since we are in 'closing' state, the only
@@ -938,8 +944,9 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
938944 -- ConnectionClosed, and CloseSocket message, none of which have
939945 -- yet been received. We leave the endpoint in closing state in
940946 -- that case.
941− if lastReceivedId < lastSentId
942− then
947+ if lastReceivedId < lastSentId vst
948+ then do
949+ putStrLn "This really shouldn't happen"
943950 return (RemoteEndPointClosing resolved vst, Nothing)
944951 else do
945952 removeRemoteEndPoint (ourEndPoint, theirEndPoint)
@@ -998,6 +1005,13 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
9981005 connId :: LightweightConnectionId -> ConnectionId
9991006 connId = createConnectionId (remoteId theirEndPoint)
10001007
1008+ -- The ID of the last connection _we_ created (or 0 for none)
1009+ lastSentId :: ValidRemoteEndPointState -> LightweightConnectionId
1010+ lastSentId vst =
1011+ if vst ^. remoteNextConnOutId == firstNonReservedLightweightConnectionId
1012+ then 0
1013+ else (vst ^. remoteNextConnOutId) - 1
1014+
10011015--------------------------------------------------------------------------------
10021016-- Uninterruptable auxiliary functions --
10031017-- --
@@ -1030,12 +1044,17 @@ createConnectionTo params ourEndPoint theirAddress hints = go
10301044 else do
10311045 -- 'findRemoteEndPoint' will have increased 'remoteOutgoing'
10321046 mapIOException connectFailed $ do
1033− act <- withMVar (remoteState theirEndPoint) $ \st -> case st of
1047+ act <- modifyMVar (remoteState theirEndPoint) $ \st -> case st of
10341048 RemoteEndPointValid vst -> do
1035− connId <- getNextConnOutId ourEndPoint
1036− schedule theirEndPoint $ do
1049+ let connId = vst ^. remoteNextConnOutId
1050+ act <- schedule theirEndPoint $ do
10371051 sendOn vst [encodeInt32 CreatedNewConnection, encodeInt32 connId]
10381052 return connId
1053+ return ( RemoteEndPointValid
1054+ $ remoteNextConnOutId ^= connId + 1
1055+ $ vst
1056+ , act
1057+ )
10391058 -- Error cases
10401059 RemoteEndPointInvalid err ->
10411060 throwIO err
@@ -1067,11 +1086,12 @@ setupRemoteEndPoint params (ourEndPoint, theirEndPoint) hints = do
10671086 Right (sock, ConnectionRequestAccepted) -> do
10681087 sendLock <- newMVar ()
10691088 let vst = ValidRemoteEndPointState
1070− { remoteSocket = sock
1071− , remoteSendLock = sendLock
1072− , _remoteOutgoing = 0
1073− , _remoteIncoming = Set.empty
1074− , _remoteMaxIncoming = 0
1089+ { remoteSocket = sock
1090+ , remoteSendLock = sendLock
1091+ , _remoteOutgoing = 0
1092+ , _remoteIncoming = Set.empty
1093+ , _remoteMaxIncoming = 0
1094+ , _remoteNextConnOutId = firstNonReservedLightweightConnectionId
10751095 }
10761096 resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst)
10771097 return True
@@ -1153,7 +1173,7 @@ connectToSelf :: LocalEndPoint
11531173 -> IO Connection
11541174connectToSelf ourEndPoint = do
11551175 connAlive <- newIORef True -- Protected by the local endpoint lock
1156− lconnId <- mapIOException connectFailed $ getNextConnOutId ourEndPoint
1176+ lconnId <- mapIOException connectFailed $ getLocalNextConnOutId ourEndPoint
11571177 let connId = createConnectionId heavyweightSelfConnectionId lconnId
11581178 writeChan ourChan $
11591179 ConnectionOpened connId ReliableOrdered (localAddress ourEndPoint)
@@ -1208,33 +1228,21 @@ resolveInit (ourEndPoint, theirEndPoint) newState =
12081228 _ ->
12091229 relyViolation (ourEndPoint, theirEndPoint) "resolveInit"
12101230
1211−-- | Get the next outgoing connection ID
1231+-- | Get the next outgoing self-connection ID
12121232--
12131233-- Throws an IO exception when the endpoint is closed.
1214−getNextConnOutId :: LocalEndPoint -> IO LightweightConnectionId
1215−getNextConnOutId ourEndpoint =
1234+getLocalNextConnOutId :: LocalEndPoint -> IO LightweightConnectionId
1235+getLocalNextConnOutId ourEndpoint =
12161236 modifyMVar (localState ourEndpoint) $ \st -> case st of
12171237 LocalEndPointValid vst -> do
1218− let connId = vst ^. nextConnOutId
1238+ let connId = vst ^. localNextConnOutId
12191239 return ( LocalEndPointValid
1220− . (nextConnOutId ^= connId + 1)
1240+ . (localNextConnOutId ^= connId + 1)
12211241 $ vst
12221242 , connId)
12231243 LocalEndPointClosed ->
12241244 throwIO $ userError "Local endpoint closed"
12251245
1226−-- | The last outgoing connection ID we created, or zero if we never created any
1227−getLastConnOutId :: LocalEndPoint -> IO LightweightConnectionId
1228−getLastConnOutId ourEndPoint =
1229− withMVar (localState ourEndPoint) $ \st -> case st of
1230− LocalEndPointValid vst ->
1231− let nextId = vst ^. nextConnOutId in
1232− if nextId == firstNonReservedLightweightConnectionId
1233− then return 0
1234− else return (nextId - 1)
1235− LocalEndPointClosed ->
1236− throwIO $ userError "Local endpoint closed"
1237−
12381246-- | Create a new local endpoint
12391247--
12401248-- May throw a TransportError NewEndPointErrorCode exception if the transport
@@ -1243,9 +1251,9 @@ createLocalEndPoint :: TCPTransport -> IO LocalEndPoint
12431251createLocalEndPoint transport = do
12441252 chan <- newChan
12451253 state <- newMVar . LocalEndPointValid $ ValidLocalEndPointState
1246− { _nextConnOutId = firstNonReservedLightweightConnectionId
1247− , _nextConnInId = firstNonReservedHeavyweightConnectionId
1248− , _localConnections = Map.empty
1254+ { _localNextConnOutId = firstNonReservedLightweightConnectionId
1255+ , _localConnections = Map.empty
1256+ , _nextConnInId = firstNonReservedHeavyweightConnectionId
12491257 }
12501258 modifyMVar (transportState transport) $ \st -> case st of
12511259 TransportValid vst -> do
@@ -1598,8 +1606,8 @@ localEndPoints = accessor _localEndPoints (\es st -> st { _localEndPoints = es }
15981606nextEndPointId :: Accessor ValidTransportState EndPointId
15991607nextEndPointId = accessor _nextEndPointId (\eid st -> st { _nextEndPointId = eid })
16001608
1601−nextConnOutId :: Accessor ValidLocalEndPointState LightweightConnectionId
1602−nextConnOutId = accessor _nextConnOutId (\cix st -> st { _nextConnOutId = cix })
1609+localNextConnOutId :: Accessor ValidLocalEndPointState LightweightConnectionId
1610+localNextConnOutId = accessor _localNextConnOutId (\cix st -> st { _localNextConnOutId = cix })
16031611
16041612localConnections :: Accessor ValidLocalEndPointState (Map EndPointAddress RemoteEndPoint)
16051613localConnections = accessor _localConnections (\es st -> st { _localConnections = es })
@@ -1616,6 +1624,9 @@ remoteIncoming = accessor _remoteIncoming (\cs conn -> conn { _remoteIncoming =
16161624remoteMaxIncoming :: Accessor ValidRemoteEndPointState LightweightConnectionId
16171625remoteMaxIncoming = accessor _remoteMaxIncoming (\lcid st -> st { _remoteMaxIncoming = lcid })
16181626
1627+remoteNextConnOutId :: Accessor ValidRemoteEndPointState LightweightConnectionId
1628+remoteNextConnOutId = accessor _remoteNextConnOutId (\cix st -> st { _remoteNextConnOutId = cix })
1629+
16191630localEndPointAt :: EndPointAddress -> Accessor ValidTransportState (Maybe LocalEndPoint)
16201631localEndPointAt addr = localEndPoints >>> DAC.mapMaybe addr
16211632