@@ -266,8 +266,12 @@ data LocalEndPointState =
266 | LocalEndPointClosed
266 | LocalEndPointClosed
268data ValidLocalEndPointState = ValidLocalEndPointState
268data ValidLocalEndPointState = ValidLocalEndPointState
269+ { -- Next available ID for an outgoing lightweight self-connection
270+ -- (see also remoteNextConnOutId)
269− { _nextConnOutId :: !LightweightConnectionId
271+ _localNextConnOutId :: !LightweightConnectionId
272+ -- Next available ID for an incoming heavyweight connection
270− , _nextConnInId :: !HeavyweightConnectionId
273+ , _nextConnInId :: !HeavyweightConnectionId
274+ -- Currently active outgoing heavyweight connections
271 , _localConnections :: !(Map EndPointAddress RemoteEndPoint)
275 , _localConnections :: !(Map EndPointAddress RemoteEndPoint)
@@ -387,6 +391,7 @@ data ValidRemoteEndPointState = ValidRemoteEndPointState
387 { _remoteOutgoing :: !Int
391 { _remoteOutgoing :: !Int
388 , _remoteIncoming :: !(Set LightweightConnectionId)
392 , _remoteIncoming :: !(Set LightweightConnectionId)
389 , _remoteMaxIncoming :: !LightweightConnectionId
393 , _remoteMaxIncoming :: !LightweightConnectionId
394+ , _remoteNextConnOutId :: !LightweightConnectionId
390 , remoteSocket :: !N.Socket
395 , remoteSocket :: !N.Socket
391 , remoteSendLock :: !(MVar ())
396 , remoteSendLock :: !(MVar ())
@@ -627,7 +632,8 @@ apiSend (ourEndPoint, theirEndPoint) connId connAlive payload =
627 RemoteEndPointValid vst -> do
632 RemoteEndPointValid vst -> do
628 alive <- readIORef connAlive
633 alive <- readIORef connAlive
629 if alive
634 if alive
635+ then schedule theirEndPoint $
630− then schedule theirEndPoint $ sendOn vst (encodeInt32 connId : prependLength payload)
636+ sendOn vst (encodeInt32 connId : prependLength payload)
631 else throwIO $ TransportError SendClosed "Connection closed"
637 else throwIO $ TransportError SendClosed "Connection closed"
632 RemoteEndPointClosing _ _ -> do
638 RemoteEndPointClosing _ _ -> do
633 alive <- readIORef connAlive
639 alive <- readIORef connAlive
@@ -749,11 +755,12 @@ handleConnectionRequest transport sock = handle handleException $ do
750 sendLock <- newMVar ()
756 sendLock <- newMVar ()
751 let vst = ValidRemoteEndPointState
757 let vst = ValidRemoteEndPointState
752− { remoteSocket = sock
758+ { remoteSocket = sock
753− , remoteSendLock = sendLock
759+ , remoteSendLock = sendLock
754− , _remoteOutgoing = 0
760+ , _remoteOutgoing = 0
755− , _remoteIncoming = Set.empty
761+ , _remoteIncoming = Set.empty
756− , _remoteMaxIncoming = 0
762+ , _remoteMaxIncoming = 0
763+ , _remoteNextConnOutId = firstNonReservedLightweightConnectionId
758 sendMany sock [encodeInt32 ConnectionRequestAccepted]
765 sendMany sock [encodeInt32 ConnectionRequestAccepted]
759 resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst)
766 resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst)
@@ -896,7 +903,6 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
896 closeSocket :: N.Socket -> LightweightConnectionId -> IO Bool
903 closeSocket :: N.Socket -> LightweightConnectionId -> IO Bool
897 closeSocket sock lastReceivedId = do
904 closeSocket sock lastReceivedId = do
898 mAct <- modifyMVar theirState $ \st -> do
905 mAct <- modifyMVar theirState $ \st -> do
899− lastSentId <- getLastConnOutId ourEndPoint
900 case st of
906 case st of
901 RemoteEndPointInvalid _ ->
907 RemoteEndPointInvalid _ ->
902 relyViolation (ourEndPoint, theirEndPoint)
908 relyViolation (ourEndPoint, theirEndPoint)
@@ -918,7 +924,7 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
918 -- then we sent a ConnectionCreated *AND* a ConnectionClosed
924 -- then we sent a ConnectionCreated *AND* a ConnectionClosed
919 -- message to the remote endpoint, *both of which* it did not yet
925 -- message to the remote endpoint, *both of which* it did not yet
920 -- receive before sending the CloseSocket request.
926 -- receive before sending the CloseSocket request.
921− if vst' ^. remoteOutgoing > 0 || lastReceivedId < lastSentId
927+ if vst' ^. remoteOutgoing > 0 || lastReceivedId < lastSentId vst
923 return (RemoteEndPointValid vst', Nothing)
929 return (RemoteEndPointValid vst', Nothing)
@@ -930,7 +936,7 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
931 tryCloseSocket sock
937 tryCloseSocket sock
932 return (RemoteEndPointClosed, Just act)
938 return (RemoteEndPointClosed, Just act)
933− RemoteEndPointClosing resolved vst ->
939+ RemoteEndPointClosing resolved vst -> do
934 -- Like above, we need to check if there is a ConnectionCreated
940 -- Like above, we need to check if there is a ConnectionCreated
935 -- message that we sent but that the remote endpoint has not yet
941 -- message that we sent but that the remote endpoint has not yet
936 -- received. However, since we are in 'closing' state, the only
942 -- received. However, since we are in 'closing' state, the only
@@ -938,8 +944,9 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
938 -- ConnectionClosed, and CloseSocket message, none of which have
944 -- ConnectionClosed, and CloseSocket message, none of which have
939 -- yet been received. We leave the endpoint in closing state in
945 -- yet been received. We leave the endpoint in closing state in
940 -- that case.
946 -- that case.
941− if lastReceivedId < lastSentId
947+ if lastReceivedId < lastSentId vst
949+ putStrLn "This really shouldn't happen"
943 return (RemoteEndPointClosing resolved vst, Nothing)
950 return (RemoteEndPointClosing resolved vst, Nothing)
945 removeRemoteEndPoint (ourEndPoint, theirEndPoint)
952 removeRemoteEndPoint (ourEndPoint, theirEndPoint)
@@ -998,6 +1005,13 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
998 connId :: LightweightConnectionId -> ConnectionId
1005 connId :: LightweightConnectionId -> ConnectionId
999 connId = createConnectionId (remoteId theirEndPoint)
1006 connId = createConnectionId (remoteId theirEndPoint)
1008+ -- The ID of the last connection _we_ created (or 0 for none)
1009+ lastSentId :: ValidRemoteEndPointState -> LightweightConnectionId
1011+ if vst ^. remoteNextConnOutId == firstNonReservedLightweightConnectionId
1013+ else (vst ^. remoteNextConnOutId) - 1
1001--------------------------------------------------------------------------------
1015--------------------------------------------------------------------------------
1002-- Uninterruptable auxiliary functions --
1016-- Uninterruptable auxiliary functions --
@@ -1030,12 +1044,17 @@ createConnectionTo params ourEndPoint theirAddress hints = go
1031 -- 'findRemoteEndPoint' will have increased 'remoteOutgoing'
1045 -- 'findRemoteEndPoint' will have increased 'remoteOutgoing'
1032 mapIOException connectFailed $ do
1046 mapIOException connectFailed $ do
1033− act <- withMVar (remoteState theirEndPoint) $ \st -> case st of
1047+ act <- modifyMVar (remoteState theirEndPoint) $ \st -> case st of
1034 RemoteEndPointValid vst -> do
1048 RemoteEndPointValid vst -> do
1035− connId <- getNextConnOutId ourEndPoint
1049+ let connId = vst ^. remoteNextConnOutId
1036− schedule theirEndPoint $ do
1050+ act <- schedule theirEndPoint $ do
1037 sendOn vst [encodeInt32 CreatedNewConnection, encodeInt32 connId]
1051 sendOn vst [encodeInt32 CreatedNewConnection, encodeInt32 connId]
1038 return connId
1052 return connId
1053+ return ( RemoteEndPointValid
1054+ $ remoteNextConnOutId ^= connId + 1
1039 -- Error cases
1058 -- Error cases
1040 RemoteEndPointInvalid err ->
1059 RemoteEndPointInvalid err ->
1041 throwIO err
1060 throwIO err
@@ -1067,11 +1086,12 @@ setupRemoteEndPoint params (ourEndPoint, theirEndPoint) hints = do
1067 Right (sock, ConnectionRequestAccepted) -> do
1086 Right (sock, ConnectionRequestAccepted) -> do
1068 sendLock <- newMVar ()
1087 sendLock <- newMVar ()
1069 let vst = ValidRemoteEndPointState
1088 let vst = ValidRemoteEndPointState
1070− { remoteSocket = sock
1089+ { remoteSocket = sock
1071− , remoteSendLock = sendLock
1090+ , remoteSendLock = sendLock
1072− , _remoteOutgoing = 0
1091+ , _remoteOutgoing = 0
1073− , _remoteIncoming = Set.empty
1092+ , _remoteIncoming = Set.empty
1074− , _remoteMaxIncoming = 0
1093+ , _remoteMaxIncoming = 0
1094+ , _remoteNextConnOutId = firstNonReservedLightweightConnectionId
1076 resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst)
1096 resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst)
1077 return True
1097 return True
@@ -1153,7 +1173,7 @@ connectToSelf :: LocalEndPoint
1153 -> IO Connection
1173 -> IO Connection
1154connectToSelf ourEndPoint = do
1174connectToSelf ourEndPoint = do
1155 connAlive <- newIORef True -- Protected by the local endpoint lock
1175 connAlive <- newIORef True -- Protected by the local endpoint lock
1156− lconnId <- mapIOException connectFailed $ getNextConnOutId ourEndPoint
1176+ lconnId <- mapIOException connectFailed $ getLocalNextConnOutId ourEndPoint
1157 let connId = createConnectionId heavyweightSelfConnectionId lconnId
1177 let connId = createConnectionId heavyweightSelfConnectionId lconnId
1158 writeChan ourChan $
1178 writeChan ourChan $
1159 ConnectionOpened connId ReliableOrdered (localAddress ourEndPoint)
1179 ConnectionOpened connId ReliableOrdered (localAddress ourEndPoint)
@@ -1208,33 +1228,21 @@ resolveInit (ourEndPoint, theirEndPoint) newState =
1209 relyViolation (ourEndPoint, theirEndPoint) "resolveInit"
1229 relyViolation (ourEndPoint, theirEndPoint) "resolveInit"
1211−-- | Get the next outgoing connection ID
1231+-- | Get the next outgoing self-connection ID
1213-- Throws an IO exception when the endpoint is closed.
1233-- Throws an IO exception when the endpoint is closed.
1214−getNextConnOutId :: LocalEndPoint -> IO LightweightConnectionId
1234+getLocalNextConnOutId :: LocalEndPoint -> IO LightweightConnectionId
1215−getNextConnOutId ourEndpoint =
1235+getLocalNextConnOutId ourEndpoint =
1216 modifyMVar (localState ourEndpoint) $ \st -> case st of
1236 modifyMVar (localState ourEndpoint) $ \st -> case st of
1217 LocalEndPointValid vst -> do
1237 LocalEndPointValid vst -> do
1218− let connId = vst ^. nextConnOutId
1238+ let connId = vst ^. localNextConnOutId
1219 return ( LocalEndPointValid
1239 return ( LocalEndPointValid
1220− . (nextConnOutId ^= connId + 1)
1240+ . (localNextConnOutId ^= connId + 1)
1222 , connId)
1242 , connId)
1223 LocalEndPointClosed ->
1243 LocalEndPointClosed ->
1224 throwIO $ userError "Local endpoint closed"
1244 throwIO $ userError "Local endpoint closed"
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
1234− else return (nextId - 1)
1235− LocalEndPointClosed ->
1236− throwIO $ userError "Local endpoint closed"
1238-- | Create a new local endpoint
1246-- | Create a new local endpoint
1240-- May throw a TransportError NewEndPointErrorCode exception if the transport
1248-- May throw a TransportError NewEndPointErrorCode exception if the transport
@@ -1243,9 +1251,9 @@ createLocalEndPoint :: TCPTransport -> IO LocalEndPoint
1243createLocalEndPoint transport = do
1251createLocalEndPoint transport = do
1244 chan <- newChan
1252 chan <- newChan
1245 state <- newMVar . LocalEndPointValid $ ValidLocalEndPointState
1253 state <- newMVar . LocalEndPointValid $ ValidLocalEndPointState
1246− { _nextConnOutId = firstNonReservedLightweightConnectionId
1254+ { _localNextConnOutId = firstNonReservedLightweightConnectionId
1247− , _nextConnInId = firstNonReservedHeavyweightConnectionId
1248− , _localConnections = Map.empty
1255+ , _localConnections = Map.empty
1256+ , _nextConnInId = firstNonReservedHeavyweightConnectionId
1250 modifyMVar (transportState transport) $ \st -> case st of
1258 modifyMVar (transportState transport) $ \st -> case st of
1251 TransportValid vst -> do
1259 TransportValid vst -> do
@@ -1598,8 +1606,8 @@ localEndPoints = accessor _localEndPoints (\es st -> st { _localEndPoints = es }
1598nextEndPointId :: Accessor ValidTransportState EndPointId
1606nextEndPointId :: Accessor ValidTransportState EndPointId
1599nextEndPointId = accessor _nextEndPointId (\eid st -> st { _nextEndPointId = eid })
1607nextEndPointId = accessor _nextEndPointId (\eid st -> st { _nextEndPointId = eid })
1601−nextConnOutId :: Accessor ValidLocalEndPointState LightweightConnectionId
1609+localNextConnOutId :: Accessor ValidLocalEndPointState LightweightConnectionId
1602−nextConnOutId = accessor _nextConnOutId (\cix st -> st { _nextConnOutId = cix })
1610+localNextConnOutId = accessor _localNextConnOutId (\cix st -> st { _localNextConnOutId = cix })
1604localConnections :: Accessor ValidLocalEndPointState (Map EndPointAddress RemoteEndPoint)
1612localConnections :: Accessor ValidLocalEndPointState (Map EndPointAddress RemoteEndPoint)
1605localConnections = accessor _localConnections (\es st -> st { _localConnections = es })
1613localConnections = accessor _localConnections (\es st -> st { _localConnections = es })
@@ -1616,6 +1624,9 @@ remoteIncoming = accessor _remoteIncoming (\cs conn -> conn { _remoteIncoming =
1616remoteMaxIncoming :: Accessor ValidRemoteEndPointState LightweightConnectionId
1624remoteMaxIncoming :: Accessor ValidRemoteEndPointState LightweightConnectionId
1617remoteMaxIncoming = accessor _remoteMaxIncoming (\lcid st -> st { _remoteMaxIncoming = lcid })
1625remoteMaxIncoming = accessor _remoteMaxIncoming (\lcid st -> st { _remoteMaxIncoming = lcid })
1627+remoteNextConnOutId :: Accessor ValidRemoteEndPointState LightweightConnectionId
1628+remoteNextConnOutId = accessor _remoteNextConnOutId (\cix st -> st { _remoteNextConnOutId = cix })
1619localEndPointAt :: EndPointAddress -> Accessor ValidTransportState (Maybe LocalEndPoint)
1630localEndPointAt :: EndPointAddress -> Accessor ValidTransportState (Maybe LocalEndPoint)
1620localEndPointAt addr = localEndPoints >>> DAC.mapMaybe addr
1631localEndPointAt addr = localEndPoints >>> DAC.mapMaybe addr