@@ -10,6 +10,7 @@ module Network.Transport.TCP ( -- * Main API
1010 createTransport
1111 , -- * TCP specific functionality
1212 EndPointId
13+ , encodeEndPointAddress
1314 , decodeEndPointAddress
1415 , ControlHeader(..)
1516 , ConnectionRequestResponse(..)
@@ -237,6 +238,12 @@ data LocalState = LocalState { _nextConnectionId :: ConnectionId
237238--
238239-- Since we don't do concurrent reads from the same socket we don't need to
239240-- take the lock when reading from the socket.
241+--
242+-- Moreover, we insist on the invariant (INV-CLOSE) that whenever we put an
243+-- endpoint in closed state we remove that endpoint from localConnections
244+-- first, so that if a concurrent thread reads the mvar, finds EndPointClosed,
245+-- and then looks up the endpoint in localConnections it is guaranteed to
246+-- either find a different remote endpoint, or else none at all.
240247
241248data RemoteState = RemoteEndPointInvalid (FailedWith ConnectErrorCode)
242249 | RemoteEndPointValid RemoteConnection
@@ -258,14 +265,14 @@ data ControlHeader =
258265 | CloseConnection -- ^ Tell the remote endpoint we will no longer be using a connection
259266 | ControlResponse -- ^ Respond to a control request _from_ the remote endpoint
260267 | CloseSocket -- ^ Request to close the connection (see module description)
261− deriving (Enum, Bounded)
268+ deriving (Enum, Bounded, Show)
262269
263270-- Response sent by /B/ to /A/ when /A/ tries to connect
264271data ConnectionRequestResponse =
265272 ConnectionRequestAccepted -- ^ /B/ accepts the connection
266273 | ConnectionRequestEndPointInvalid -- ^ /A/ requested an invalid endpoint
267274 | ConnectionRequestCrossed -- ^ /A/s request crossed with a request from /B/ (see protocols)
268− deriving (Enum, Bounded)
275+ deriving (Enum, Bounded, Show)
269276
270277--------------------------------------------------------------------------------
271278-- Top-level functionality --
@@ -393,7 +400,7 @@ apiCloseEndPoint transport ourEndPoint = do
393400 -- calls will block. We could change this so that all subsequent calls to
394401 -- receive return an error, but this would mean checking for some state on
395402 -- every call to receive, which is an unnecessary overhead.
396− writeChan (localChannel ourEndPoint) (ErrorEvent EventErrorEndPointClosed incoming)
403+ writeChan (localChannel ourEndPoint) (ErrorEvent $ ErrorEventEndPointClosed incoming)
397404 where
398405 remoteSocketOf :: RemoteEndPoint -> MaybeT IO N.Socket
399406 remoteSocketOf theirEndPoint = do
@@ -452,7 +459,7 @@ createLocalEndPoint transport = do
452459 }
453460 modifyMVar (transportState transport) $ \st -> do
454461 let ix = fromIntegral $ Map.size (st ^. localEndPoints)
455− let addr = encodeEndPointAddress transport ix
462+ let addr = encodeEndPointAddress (transportHost transport) (transportPort transport) ix
456463 let localEndPoint = LocalEndPoint { localAddress = addr
457464 , localChannel = chan
458465 , localState = state
@@ -472,7 +479,7 @@ requestConnectionTo ourEndPoint theirAddress = ErrorT go
472479 go = do
473480 let ourState = localState ourEndPoint
474481 (theirEndPoint, isNew) <- modifyMVar ourState $ \st ->
475− case st ^. connectionTo theirAddress of
482+ case st ^. localConnectionTo theirAddress of
476483 Just theirEndPoint ->
477484 return (st, (theirEndPoint, False))
478485 Nothing -> do
@@ -480,7 +487,7 @@ requestConnectionTo ourEndPoint theirAddress = ErrorT go
480487 let theirEndPoint = RemoteEndPoint { remoteAddress = theirAddress
481488 , remoteState = theirState
482489 }
483− return (connectionTo theirAddress ^= Just theirEndPoint $ st, (theirEndPoint, True))
490+ return (localConnectionTo theirAddress ^= Just theirEndPoint $ st, (theirEndPoint, True))
484491
485492 -- If this is a new endpoint, fork a thread to listen for incoming
486493 -- connections. We don't want to do this while we hold the lock, because
@@ -574,14 +581,14 @@ connectToRemoteEndPoint (ourEndPoint, theirEndPoint) = do
574581 -- waiting on the result of this call to connect will get the
575582 -- RemoteEndPointInvalid; subsequent threads will initiate a new
576583 -- connection requests.
577− modifyMVar_ ourState $ return . (connectionTo theirAddress ^= Nothing)
584+ modifyMVar_ ourState $ return . (localConnectionTo theirAddress ^= Nothing)
578585 putMVar theirState (RemoteEndPointInvalid (invalidAddress "Invalid endpoint"))
579586 N.sClose sock
580587 Right (ConnectionRequestCrossed, sock) -> do
581588 N.sClose sock
582589 Left err -> do
583590 -- See comment above
584− modifyMVar_ ourState $ return . (connectionTo theirAddress ^= Nothing)
591+ modifyMVar_ ourState $ return . (localConnectionTo theirAddress ^= Nothing)
585592 putMVar theirState (RemoteEndPointInvalid err)
586593 -- TODO: this isn't quite right; socket *might* have been opened here,
587594 -- in which case we should close it
@@ -613,9 +620,9 @@ connectToRemoteEndPoint (ourEndPoint, theirEndPoint) = do
613620 EndPointAddress ourAddress = localAddress ourEndPoint
614621
615622-- | Encode end point address
616−encodeEndPointAddress :: TCPTransport -> EndPointId -> EndPointAddress
617−encodeEndPointAddress transport ix = EndPointAddress . BSC.pack $
618− transportHost transport ++ ":" ++ transportPort transport ++ ":" ++ show ix
623+encodeEndPointAddress :: N.HostName -> N.ServiceName -> EndPointId -> EndPointAddress
624+encodeEndPointAddress host port ix = EndPointAddress . BSC.pack $
625+ host ++ ":" ++ port ++ ":" ++ show ix
619626
620627-- | Decode end point address
621628-- TODO: This uses regular expression parsing, which is nice, but unnecessary
@@ -677,7 +684,7 @@ handleConnectionRequest transport sock = do
677684 request <- tryIO $ do
678685 ourEndPointId <- recvInt32 sock
679686 theirAddress <- EndPointAddress . BS.concat <$> recvWithLength sock
680− let ourAddress = encodeEndPointAddress transport ourEndPointId
687+ let ourAddress = encodeEndPointAddress (transportHost transport) (transportPort transport) ourEndPointId
681688 ourEndPoint <- (^. localEndPointAt ourAddress) <$> readMVar (transportState transport)
682689 return (theirAddress, ourEndPoint)
683690 case request of
@@ -686,13 +693,13 @@ handleConnectionRequest transport sock = do
686693 N.sClose sock
687694 Right (theirAddress, Just ourEndPoint) -> forkEndPointThread ourEndPoint $ do
688695 (crossed, theirEndPoint) <- modifyMVar (localState ourEndPoint) $ \st ->
689− case st ^. connectionTo theirAddress of
696+ case st ^. localConnectionTo theirAddress of
690697 Nothing -> do
691698 theirState <- newEmptyMVar
692699 let theirEndPoint = RemoteEndPoint { remoteAddress = theirAddress
693700 , remoteState = theirState
694701 }
695− return (connectionTo theirAddress ^= Just theirEndPoint $ st, (False, theirEndPoint))
702+ return (localConnectionTo theirAddress ^= Just theirEndPoint $ st, (False, theirEndPoint))
696703 Just theirEndPoint ->
697704 return (st, (localAddress ourEndPoint < theirAddress, theirEndPoint))
698705 if crossed
@@ -733,28 +740,10 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
733740 _ ->
734741 error "handleIncomingMessages RELY violation"
735742
736− tryIO $ go sock
737−
738− -- Normally, there would be no unclosed connections, but if the client
739− -- exits prematurely there might be
740− mUnclosedConnections <- modifyMVar theirState $ \st ->
741− case st of
742− RemoteEndPointInvalid _ ->
743− error "handleIncomingMessages RELY violation"
744− RemoteEndPointValid remoteConn ->
745− return (RemoteEndPointClosed, Just $ remoteConn ^. remoteIncoming)
746− RemoteEndPointClosing _ _ ->
747− return (RemoteEndPointClosed, Nothing)
748− RemoteEndPointClosed ->
749− return (st, Nothing)
750−
751− -- TODO: we need to remote this endpoint from localConnections at this point
752−
753− case mUnclosedConnections of
754− Nothing -> return ()
755− Just unclosedConnections ->
756− writeChan ourChannel (ErrorEvent EventErrorEndPointClosed (IntSet.elems unclosedConnections))
757−
743+ mCleanExit <- tryIO $ go sock
744+ case mCleanExit of
745+ Right () -> return ()
746+ Left err -> prematureExit sock err
758747 where
759748 -- Dispatch
760749 go :: N.Socket -> IO ()
@@ -835,24 +824,45 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
835824 -- Close the socket (if we don't have any outgoing connections)
836825 closeSocket :: N.Socket -> IO Bool
837826 closeSocket sock = do
838− didClose <- modifyMVar theirState $ \st ->
827+ -- We need to check if we can close the socket (that is, if we don't have
828+ -- any outgoing connections), and once we are sure that we can put the
829+ -- endpoint in Closed state. However, by INV-CLOSE we can only put the
830+ -- endpoint in Closed state once we remove the endpoint from our local
831+ -- connections. But we can do that only once we are sure that we can
832+ -- close the endpoint. Catch-22. We can resolve the catch-22 by locking
833+ -- *both* our local state and the endpoint state, but at the cost of
834+ -- introducing a double lock and all the associated perils, or by putting
835+ -- the remote endpoint in Closing state first. We opt for the latter.
836+ canClose <- modifyMVar theirState $ \st ->
839837 case st of
840− RemoteEndPointValid remoteConn | remoteConn ^. remoteOutgoing == 0 -> do
841− sendOn remoteConn [encodeInt32 CloseSocket]
842− N.sClose sock
843− return (RemoteEndPointClosed, True)
844− RemoteEndPointClosing resolved _ -> do
845− -- If the socket is already in closing state, then we have already
846− -- sent a CloseSocket request; hence, we don't need to do it again
847− -- at this point
848− putMVar resolved ()
849− N.sClose sock
850− return (RemoteEndPointClosed, True)
838+ RemoteEndPointValid remoteConn -> do
839+ -- TODO: We check only for outgoing connections here. That is
840+ -- correct, but we need to makke sure that we send the right event
841+ -- to our endpoint for any incoming connections that are still live
842+ -- at this point.
843+ if remoteConn ^. remoteOutgoing == 0
844+ then do
845+ -- TODO: code duplication with closeIfUnused
846+ resolved <- newEmptyMVar
847+ sendOn remoteConn [encodeInt32 CloseSocket]
848+ return (RemoteEndPointClosing resolved remoteConn, Just resolved)
849+ else
850+ return (st, Nothing)
851+ RemoteEndPointClosing resolved _ ->
852+ return (st, Just resolved)
851853 _ ->
852− return (st, False)
853− when didClose $ modifyMVar_ ourState $ return . (connectionTo theirAddr ^= Nothing)
854− return didClose
855−
854+ error "handleIncomingConnections RELY violation"
855+
856+ case canClose of
857+ Nothing ->
858+ return False
859+ Just resolved -> do
860+ modifyMVar_ ourState $ return . (localConnectionTo theirAddr ^= Nothing)
861+ modifyMVar_ theirState $ return . const RemoteEndPointClosed
862+ N.sClose sock
863+ putMVar resolved ()
864+ return True
865+
856866 -- Read a message and output it on the endPoint's channel
857867 readMessage :: N.Socket -> ConnectionId -> IO ()
858868 readMessage sock connId = recvWithLength sock >>= writeChan ourChannel . Received connId
@@ -863,6 +873,29 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
863873 theirState = remoteState theirEndPoint
864874 theirAddr = remoteAddress theirEndPoint
865875
876+ -- Deal with a premature exit
877+ prematureExit :: N.Socket -> IOException -> IO ()
878+ prematureExit sock _ = do
879+ N.sClose sock
880+ modifyMVar_ ourState $ return . (localConnectionTo theirAddr ^= Nothing)
881+ mUnclosedConnections <- modifyMVar theirState $ \st ->
882+ case st of
883+ RemoteEndPointInvalid _ ->
884+ error "handleIncomingMessages RELY violation"
885+ RemoteEndPointValid remoteConn ->
886+ return (RemoteEndPointClosed, Just $ remoteConn ^. remoteIncoming)
887+ RemoteEndPointClosing _ _ ->
888+ return (RemoteEndPointClosed, Nothing)
889+ RemoteEndPointClosed ->
890+ return (st, Nothing)
891+
892+ -- We send the connection lost message even if unclosedConnections is the
893+ -- empty set, because *outgoing* connections will be broken too now
894+ case mUnclosedConnections of
895+ Nothing ->
896+ return ()
897+ Just unclosedConnections -> writeChan ourChannel . ErrorEvent $
898+ ErrorEventConnectionLost (remoteAddress theirEndPoint) (IntSet.elems unclosedConnections)
866899
867900
868901-- | Get the next connection ID
@@ -911,8 +944,8 @@ localEndPointAt addr = localEndPoints >>> mapLens addr
911944pendingCtrlRequestsAt :: ControlRequestId -> Lens LocalState (Maybe (MVar [ByteString]))
912945pendingCtrlRequestsAt ix = pendingCtrlRequests >>> intMapLens (fromIntegral ix)
913946
914−connectionTo :: EndPointAddress -> Lens LocalState (Maybe RemoteEndPoint)
915−connectionTo addr = localConnections >>> mapLens addr
947+localConnectionTo :: EndPointAddress -> Lens LocalState (Maybe RemoteEndPoint)
948+localConnectionTo addr = localConnections >>> mapLens addr
916949
917950remoteOutgoing :: Lens RemoteConnection Int
918951remoteOutgoing = lens _remoteOutgoing (\cs conn -> conn { _remoteOutgoing = cs })