@@ -78,6 +78,7 @@ import Control.Applicative ((<$>))
7878import Control.Monad (when, unless)
7979import Control.Exception ( IOException
8080 , SomeException
81+ , AsyncException
8182 , handle
8283 , throw
8384 , throwIO
@@ -203,7 +204,7 @@ import Data.Foldable (forM_, mapM_)
203204-- Internal datatypes --
204205--------------------------------------------------------------------------------
205206
206−-- We use underscores for fields that we might update (using accessores)
207+-- We use underscores for fields that we might update (using accessors)
207208--
208209-- All data types follow the same structure:
209210--
@@ -254,9 +255,11 @@ data ValidLocalEndPointState = ValidLocalEndPointState
254255--
255256-- Init ---+---> Invalid
256257-- |
257−-- | /--------------\
258−-- | | |
259−-- | v |
258+-- +---------------------------------------\
259+-- | |
260+-- | /--------------\ |
261+-- | | | |
262+-- | v | v
260263-- +---> Valid ---+---> Closing ---+---> Closed
261264-- | | | |
262265-- | | | v
@@ -282,6 +285,11 @@ data ValidLocalEndPointState = ValidLocalEndPointState
282285-- CloseSocket) or be put back in 'Valid' state if the remote endpoint denies
283286-- the request.
284287--
288+-- We also put the endpoint in Closed state, directly from Init, if we our
289+-- outbound connection request crossed an inbound connection request and we
290+-- decide to keep the inbound (i.e., the remote endpoint sent us a
291+-- ConnectionRequestCrossed message).
292+--
285293-- Closed: The endpoint is put in Closed state after a successful garbage
286294-- collection.
287295--
@@ -481,32 +489,34 @@ defaultTCPParameters = TCPParameters {
481489
482490-- | Close the transport
483491apiCloseTransport :: TCPTransport -> Maybe ThreadId -> [Event] -> IO ()
484−apiCloseTransport transport mTransportThread evs = do
485− mTSt <- modifyMVar (transportState transport) $ \st -> case st of
486− TransportValid vst -> return (TransportClosed, Just vst)
487− TransportClosed -> return (TransportClosed, Nothing)
488− forM_ mTSt $ mapM_ (apiCloseEndPoint transport evs) . (^. localEndPoints)
489− -- This will invoke the termination handler, which in turn will call
490− -- apiCloseTransport again, but then the transport will already be closed and
491− -- we won't be passed a transport thread, so we terminate immmediate
492− forM_ mTransportThread killThread
492+apiCloseTransport transport mTransportThread evs =
493+ asyncWhenCancelled return $ do
494+ mTSt <- modifyMVar (transportState transport) $ \st -> case st of
495+ TransportValid vst -> return (TransportClosed, Just vst)
496+ TransportClosed -> return (TransportClosed, Nothing)
497+ forM_ mTSt $ mapM_ (apiCloseEndPoint transport evs) . (^. localEndPoints)
498+ -- This will invoke the termination handler, which in turn will call
499+ -- apiCloseTransport again, but then the transport will already be closed
500+ -- and we won't be passed a transport thread, so we terminate immmediate
501+ forM_ mTransportThread killThread
493502
494503-- | Create a new endpoint
495504apiNewEndPoint :: TCPTransport
496505 -> IO (Either (TransportError NewEndPointErrorCode) EndPoint)
497−apiNewEndPoint transport = try $ do
498− ourEndPoint <- createLocalEndPoint transport
499− return EndPoint
500− { receive = readChan (localChannel ourEndPoint)
501− , address = localAddress ourEndPoint
502− , connect = apiConnect (transportParams transport) ourEndPoint
503− , closeEndPoint = let evs = [ EndPointClosed
504− , throw $ userError "Endpoint closed"
505− ] in
506− apiCloseEndPoint transport evs ourEndPoint
507− , newMulticastGroup = return . Left $ newMulticastGroupError
508− , resolveMulticastGroup = return . Left . const resolveMulticastGroupError
509− }
506+apiNewEndPoint transport =
507+ try . asyncWhenCancelled closeEndPoint $ do
508+ ourEndPoint <- createLocalEndPoint transport
509+ return EndPoint
510+ { receive = readChan (localChannel ourEndPoint)
511+ , address = localAddress ourEndPoint
512+ , connect = apiConnect (transportParams transport) ourEndPoint
513+ , closeEndPoint = let evs = [ EndPointClosed
514+ , throw $ userError "Endpoint closed"
515+ ] in
516+ apiCloseEndPoint transport evs ourEndPoint
517+ , newMulticastGroup = return . Left $ newMulticastGroupError
518+ , resolveMulticastGroup = return . Left . const resolveMulticastGroupError
519+ }
510520 where
511521 newMulticastGroupError =
512522 TransportError NewMulticastGroupUnsupported "Multicast not supported"
@@ -521,12 +531,13 @@ apiConnect :: TCPParameters -- ^ Parameters
521531 -> ConnectHints -- ^ Hints
522532 -> IO (Either (TransportError ConnectErrorCode) Connection)
523533apiConnect params ourEndPoint theirAddress _reliability hints =
524− if localAddress ourEndPoint == theirAddress
525− then connectToSelf ourEndPoint
526− else try . asyncWhenCancelled close $ do
534+ try . asyncWhenCancelled close $
535+ if localAddress ourEndPoint == theirAddress
536+ then connectToSelf ourEndPoint
537+ else do
527538 resetIfBroken ourEndPoint theirAddress
528539 (theirEndPoint, connId) <-
529− uRequestConnectionTo params ourEndPoint theirAddress hints
540+ requestConnectionTo params ourEndPoint theirAddress hints
530541 -- connAlive can be an IORef rather than an MVar because it is protected
531542 -- by the remoteState MVar. We don't need the overhead of locking twice.
532543 connAlive <- newIORef True
@@ -562,6 +573,7 @@ apiSend :: EndPointPair -- ^ Local and remote endpoint
562573 -> [ByteString] -- ^ Payload
563574 -> IO (Either (TransportError SendErrorCode) ())
564575apiSend (ourEndPoint, theirEndPoint) connId connAlive payload =
576+ -- We don't need the overhead of asyncWhenCancelled here
565577 try . mapIOException sendFailed $
566578 withRemoteState (ourEndPoint, theirEndPoint) RemoteStatePatternMatch
567579 { caseInvalid = \_ ->
@@ -597,19 +609,20 @@ apiCloseEndPoint :: TCPTransport -- ^ Transport
597609 -> [Event] -- ^ Events used to report closure
598610 -> LocalEndPoint -- ^ Local endpoint
599611 -> IO ()
600−apiCloseEndPoint transport evs ourEndPoint = do
601− -- Remove the reference from the transport state
602− removeLocalEndPoint transport ourEndPoint
603− -- Close the local endpoint
604− mOurState <- modifyMVar (localState ourEndPoint) $ \st ->
605− case st of
606− LocalEndPointValid vst ->
607− return (LocalEndPointClosed, Just vst)
608− LocalEndPointClosed ->
609− return (LocalEndPointClosed, Nothing)
610− forM_ mOurState $ \vst -> do
611− forM_ (vst ^. localConnections) tryCloseRemoteSocket
612− forM_ evs $ writeChan (localChannel ourEndPoint)
612+apiCloseEndPoint transport evs ourEndPoint =
613+ asyncWhenCancelled return $ do
614+ -- Remove the reference from the transport state
615+ removeLocalEndPoint transport ourEndPoint
616+ -- Close the local endpoint
617+ mOurState <- modifyMVar (localState ourEndPoint) $ \st ->
618+ case st of
619+ LocalEndPointValid vst ->
620+ return (LocalEndPointClosed, Just vst)
621+ LocalEndPointClosed ->
622+ return (LocalEndPointClosed, Nothing)
623+ forM_ mOurState $ \vst -> do
624+ forM_ (vst ^. localConnections) tryCloseRemoteSocket
625+ forM_ evs $ writeChan (localChannel ourEndPoint)
613626 where
614627 -- Close the remote socket and return the set of all incoming connections
615628 tryCloseRemoteSocket :: RemoteEndPoint -> IO ()
@@ -637,445 +650,143 @@ apiCloseEndPoint transport evs ourEndPoint = do
637650 RemoteEndPointFailed err ->
638651 return $ RemoteEndPointFailed err
639652
640−-- | Special case of 'apiConnect': connect an endpoint to itself
641−connectToSelf :: LocalEndPoint
642− -> IO (Either (TransportError ConnectErrorCode) Connection)
643−connectToSelf ourEndPoint = try . mapIOException connectFailed $ do
644− connAlive <- newIORef True -- Protected by the local endpoint lock
645− connId <- getNextConnectionId ourEndPoint
646− writeChan ourChan $
647− ConnectionOpened connId ReliableOrdered (localAddress ourEndPoint)
648− return Connection
649− { send = selfSend connAlive connId
650− , close = selfClose connAlive connId
651− }
652− where
653− selfSend :: IORef Bool
654− -> ConnectionId
655− -> [ByteString]
656− -> IO (Either (TransportError SendErrorCode) ())
657− selfSend connAlive connId msg =
658− try . withMVar ourState $ \st -> case st of
659− LocalEndPointValid _ -> do
660− alive <- readIORef connAlive
661− if alive
662− then writeChan ourChan (Received connId msg)
663− else throwIO $ TransportError SendClosed "Connection closed"
664− LocalEndPointClosed ->
665− throwIO $ TransportError SendFailed "Endpoint closed"
653+--------------------------------------------------------------------------------
654+-- As soon as a remote connection fails, we want to put notify our endpoint --
655+-- and put it into a closed state. Since this may happen in many places, we --
656+-- provide some abstractions. --
657+--------------------------------------------------------------------------------
666658
667− selfClose :: IORef Bool -> ConnectionId -> IO ()
668− selfClose connAlive connId =
669− withMVar ourState $ \st -> case st of
670− LocalEndPointValid _ -> do
671− alive <- readIORef connAlive
672− when alive $ do
673− writeChan ourChan (ConnectionClosed connId)
674− writeIORef connAlive False
675− LocalEndPointClosed ->
676− return ()
659+data RemoteStatePatternMatch a = RemoteStatePatternMatch
660+ { caseInvalid :: TransportError ConnectErrorCode -> IO a
661+ , caseInit :: MVar () -> RequestedBy -> IO a
662+ , caseValid :: ValidRemoteEndPointState -> IO a
663+ , caseClosing :: MVar () -> ValidRemoteEndPointState -> IO a
664+ , caseClosed :: IO a
665+ , caseFailed :: IOException -> IO a
666+ }
677667
678− ourChan = localChannel ourEndPoint
679− ourState = localState ourEndPoint
680− connectFailed = TransportError ConnectFailed . show
668+remoteStateIdentity :: RemoteStatePatternMatch RemoteState
669+remoteStateIdentity =
670+ RemoteStatePatternMatch
671+ { caseInvalid = return . RemoteEndPointInvalid
672+ , caseInit = (return .) . RemoteEndPointInit
673+ , caseValid = return . RemoteEndPointValid
674+ , caseClosing = (return .) . RemoteEndPointClosing
675+ , caseClosed = return RemoteEndPointClosed
676+ , caseFailed = return . RemoteEndPointFailed
677+ }
681678
682−--------------------------------------------------------------------------------
683−-- Functions from TransportInternals --
684−--------------------------------------------------------------------------------
679+-- | Like modifyMVar, but if an I/O exception occurs don't restore the remote
680+-- endpoint to its original value but close it instead
681+modifyRemoteState :: EndPointPair
682+ -> RemoteStatePatternMatch (RemoteState, a)
683+ -> IO a
684+modifyRemoteState (ourEndPoint, theirEndPoint) match =
685+ mask $ \restore -> do
686+ st <- takeMVar theirState
687+ case st of
688+ RemoteEndPointValid vst -> do
689+ mResult <- try $ restore (caseValid match vst)
690+ case mResult of
691+ Right (st', a) -> do
692+ putMVar theirState st'
693+ return a
694+ Left ex -> do
695+ case fromException ex of
696+ Just ioEx -> handleIOException ioEx vst
697+ Nothing -> putMVar theirState st
698+ throwIO ex
699+ -- The other cases are less interesting, because unless the endpoint is
700+ -- in Valid state we're not supposed to do any IO on it
701+ RemoteEndPointInit resolved origin -> do
702+ (st', a) <- onException (restore $ caseInit match resolved origin)
703+ (putMVar theirState st)
704+ putMVar theirState st'
705+ return a
706+ RemoteEndPointClosing resolved vst -> do
707+ (st', a) <- onException (restore $ caseClosing match resolved vst)
708+ (putMVar theirState st)
709+ putMVar theirState st'
710+ return a
711+ RemoteEndPointInvalid err -> do
712+ (st', a) <- onException (restore $ caseInvalid match err)
713+ (putMVar theirState st)
714+ putMVar theirState st'
715+ return a
716+ RemoteEndPointClosed -> do
717+ (st', a) <- onException (restore $ caseClosed match)
718+ (putMVar theirState st)
719+ putMVar theirState st'
720+ return a
721+ RemoteEndPointFailed err -> do
722+ (st', a) <- onException (restore $ caseFailed match err)
723+ (putMVar theirState st)
724+ putMVar theirState st'
725+ return a
726+ where
727+ theirState :: MVar RemoteState
728+ theirState = remoteState theirEndPoint
685729
686−-- Find a socket between two endpoints
687−--
688−-- Throws an IO exception if the socket could not be found.
689−internalSocketBetween :: TCPTransport -- ^ Transport
690− -> EndPointAddress -- ^ Local endpoint
691− -> EndPointAddress -- ^ Remote endpoint
692− -> IO N.Socket
693−internalSocketBetween transport ourAddress theirAddress = do
694− ourEndPoint <- withMVar (transportState transport) $ \st -> case st of
695− TransportClosed ->
696− throwIO $ userError "Transport closed"
697− TransportValid vst ->
698− case vst ^. localEndPointAt ourAddress of
699− Nothing -> throwIO $ userError "Local endpoint not found"
700− Just ep -> return ep
701− theirEndPoint <- withMVar (localState ourEndPoint) $ \st -> case st of
702− LocalEndPointClosed ->
703− throwIO $ userError "Local endpoint closed"
704− LocalEndPointValid vst ->
705− case vst ^. localConnectionTo theirAddress of
706− Nothing -> throwIO $ userError "Remote endpoint not found"
707− Just ep -> return ep
708− withMVar (remoteState theirEndPoint) $ \st -> case st of
709− RemoteEndPointInit _ _ ->
710− throwIO $ userError "Remote endpoint not yet initialized"
711− RemoteEndPointValid vst ->
712− return $ remoteSocket vst
713− RemoteEndPointClosing _ vst ->
714− return $ remoteSocket vst
715− RemoteEndPointClosed ->
716− throwIO $ userError "Remote endpoint closed"
717− RemoteEndPointInvalid err ->
718− throwIO err
719− RemoteEndPointFailed err ->
720− throwIO err
730+ handleIOException :: IOException -> ValidRemoteEndPointState -> IO ()
731+ handleIOException ex vst = do
732+ tryCloseSocket (remoteSocket vst)
733+ putMVar theirState (RemoteEndPointFailed ex)
734+ let incoming = IntSet.elems $ vst ^. remoteIncoming
735+ code = EventConnectionLost (remoteAddress theirEndPoint) incoming
736+ err = TransportError code (show ex)
737+ writeChan (localChannel ourEndPoint) $ ErrorEvent err
738+
739+-- | Like 'modifyRemoteState' but without a return value
740+modifyRemoteState_ :: EndPointPair
741+ -> RemoteStatePatternMatch RemoteState
742+ -> IO ()
743+modifyRemoteState_ (ourEndPoint, theirEndPoint) match =
744+ modifyRemoteState (ourEndPoint, theirEndPoint)
745+ RemoteStatePatternMatch
746+ { caseInvalid = u . caseInvalid match
747+ , caseInit = \resolved origin -> u $ caseInit match resolved origin
748+ , caseValid = u . caseValid match
749+ , caseClosing = \resolved vst -> u $ caseClosing match resolved vst
750+ , caseClosed = u $ caseClosed match
751+ , caseFailed = u . caseFailed match
752+ }
753+ where
754+ u :: IO a -> IO (a, ())
755+ u p = p >>= \a -> return (a, ())
756+
757+-- | Like 'modifyRemoteState' but without the ability to change the state
758+withRemoteState :: EndPointPair
759+ -> RemoteStatePatternMatch a
760+ -> IO a
761+withRemoteState (ourEndPoint, theirEndPoint) match =
762+ modifyRemoteState (ourEndPoint, theirEndPoint)
763+ RemoteStatePatternMatch
764+ { caseInvalid = \err -> do
765+ a <- caseInvalid match err
766+ return (RemoteEndPointInvalid err, a)
767+ , caseInit = \resolved origin -> do
768+ a <- caseInit match resolved origin
769+ return (RemoteEndPointInit resolved origin, a)
770+ , caseValid = \vst -> do
771+ a <- caseValid match vst
772+ return (RemoteEndPointValid vst, a)
773+ , caseClosing = \resolved vst -> do
774+ a <- caseClosing match resolved vst
775+ return (RemoteEndPointClosing resolved vst, a)
776+ , caseClosed = do
777+ a <- caseClosed match
778+ return (RemoteEndPointClosed, a)
779+ , caseFailed = \err -> do
780+ a <- caseFailed match err
781+ return (RemoteEndPointFailed err, a)
782+ }
721783
722784--------------------------------------------------------------------------------
723−-- Lower level functionality --
785+-- Incoming requests --
724786--------------------------------------------------------------------------------
725787
726−-- | Reset a remote endpoint if it is in Invalid mode
727−--
728−-- If a user calls the API function 'connect' and the remote endpoint is
729−-- currently in broken state, we remove the remote endpoint first because a
730−-- new attempt to connect might succeed even if the previous one failed.
731−--
732−-- Throws a TransportError ConnectFailed exception if the local endpoint is
733−-- closed.
734−resetIfBroken :: LocalEndPoint -> EndPointAddress -> IO ()
735−resetIfBroken ourEndPoint theirAddress = do
736− mTheirEndPoint <- withMVar (localState ourEndPoint) $ \st -> case st of
737− LocalEndPointValid vst ->
738− return (vst ^. localConnectionTo theirAddress)
739− LocalEndPointClosed ->
740− throwIO $ TransportError ConnectFailed "Endpoint closed"
741− forM_ mTheirEndPoint $ \theirEndPoint ->
742− withMVar (remoteState theirEndPoint) $ \st -> case st of
743− RemoteEndPointInvalid _ ->
744− removeRemoteEndPoint (ourEndPoint, theirEndPoint)
745− RemoteEndPointFailed _ ->
746− removeRemoteEndPoint (ourEndPoint, theirEndPoint)
747− _ ->
748− return ()
749−
750−-- | Create a new local endpoint
751−--
752−-- May throw a TransportError NewEndPointErrorCode exception if the transport
753−-- is closed.
754−createLocalEndPoint :: TCPTransport -> IO LocalEndPoint
755−createLocalEndPoint transport = do
756− chan <- newChan
757− state <- newMVar . LocalEndPointValid $ ValidLocalEndPointState
758− { _nextConnectionId = firstNonReservedConnectionId
759− , _localConnections = Map.empty
760− , _nextRemoteId = 0
761− }
762− modifyMVar (transportState transport) $ \st -> case st of
763− TransportValid vst -> do
764− let ix = vst ^. nextEndPointId
765− let addr = encodeEndPointAddress (transportHost transport)
766− (transportPort transport)
767− ix
768− let localEndPoint = LocalEndPoint { localAddress = addr
769− , localChannel = chan
770− , localState = state
771− }
772− return ( TransportValid
773− . (localEndPointAt addr ^= Just localEndPoint)
774− . (nextEndPointId ^= ix + 1)
775− $ vst
776− , localEndPoint
777− )
778− TransportClosed ->
779− throwIO (TransportError NewEndPointFailed "Transport closed")
780−
781−
782−-- | Resolve an endpoint currently in 'Init' state
783−resolveInit :: EndPointPair -> RemoteState -> IO ()
784−resolveInit (ourEndPoint, theirEndPoint) newState =
785− modifyMVar_ (remoteState theirEndPoint) $ \st -> case st of
786− RemoteEndPointInit resolved _ -> do
787− putMVar resolved ()
788− case newState of
789− RemoteEndPointClosed ->
790− removeRemoteEndPoint (ourEndPoint, theirEndPoint)
791− _ ->
792− return ()
793− return newState
794− RemoteEndPointFailed ex ->
795− throwIO ex
796− _ ->
797− relyViolation (ourEndPoint, theirEndPoint) "resolveInit"
798−
799−-- | Establish a connection to a remote endpoint
800−--
801−-- Maybe throw a TransportError
802−socketToEndPoint :: EndPointAddress -- ^ Our address
803− -> EndPointAddress -- ^ Their address
804− -> Bool -- ^ Use SO_REUSEADDR?
805− -> Maybe Int -- ^ Timeout for connect
806− -> IO (Either (TransportError ConnectErrorCode)
807− (N.Socket, ConnectionRequestResponse))
808−socketToEndPoint (EndPointAddress ourAddress) theirAddress reuseAddr timeout =
809− try $ do
810− (host, port, theirEndPointId) <- case decodeEndPointAddress theirAddress of
811− Nothing -> throwIO (failed . userError $ "Could not parse")
812− Just dec -> return dec
813− addr:_ <- mapIOException invalidAddress $
814− N.getAddrInfo Nothing (Just host) (Just port)
815− bracketOnError (createSocket addr) tryCloseSocket $ \sock -> do
816− when reuseAddr $
817− mapIOException failed $ N.setSocketOption sock N.ReuseAddr 1
818− mapIOException invalidAddress $
819− timeoutMaybe timeout timeoutError $
820− N.connect sock (N.addrAddress addr)
821− response <- mapIOException failed $ do
822− sendMany sock (encodeInt32 theirEndPointId : prependLength [ourAddress])
823− recvInt32 sock
824− case tryToEnum response of
825− Nothing -> throwIO (failed . userError $ "Unexpected response")
826− Just r -> return (sock, r)
827− where
828− createSocket :: N.AddrInfo -> IO N.Socket
829− createSocket addr = mapIOException insufficientResources $
830− N.socket (N.addrFamily addr) N.Stream N.defaultProtocol
831−
832− invalidAddress = TransportError ConnectNotFound . show
833− insufficientResources = TransportError ConnectInsufficientResources . show
834− failed = TransportError ConnectFailed . show
835− timeoutError = TransportError ConnectTimeout "Timed out"
836−
837−-- | Remove reference to a remote endpoint from a local endpoint
838−--
839−-- If the local endpoint is closed, do nothing
840−removeRemoteEndPoint :: EndPointPair -> IO ()
841−removeRemoteEndPoint (ourEndPoint, theirEndPoint) =
842− modifyMVar_ ourState $ \st -> case st of
843− LocalEndPointValid vst ->
844− case vst ^. localConnectionTo theirAddress of
845− Nothing ->
846− return st
847− Just remoteEndPoint' ->
848− if remoteId remoteEndPoint' == remoteId theirEndPoint
849− then return
850− ( LocalEndPointValid
851− . (localConnectionTo (remoteAddress theirEndPoint) ^= Nothing)
852− $ vst
853− )
854− else return st
855− LocalEndPointClosed ->
856− return LocalEndPointClosed
857− where
858− ourState = localState ourEndPoint
859− theirAddress = remoteAddress theirEndPoint
860−
861−-- | Remove reference to a local endpoint from the transport state
862−--
863−-- Does nothing if the transport is closed
864−removeLocalEndPoint :: TCPTransport -> LocalEndPoint -> IO ()
865−removeLocalEndPoint transport ourEndPoint =
866− modifyMVar_ (transportState transport) $ \st -> case st of
867− TransportValid vst ->
868− return ( TransportValid
869− . (localEndPointAt (localAddress ourEndPoint) ^= Nothing)
870− $ vst
871− )
872− TransportClosed ->
873− return TransportClosed
874−
875−-- | Encode end point address
876−encodeEndPointAddress :: N.HostName
877− -> N.ServiceName
878− -> EndPointId
879− -> EndPointAddress
880−encodeEndPointAddress host port ix = EndPointAddress . BSC.pack $
881− host ++ ":" ++ port ++ ":" ++ show ix
882−
883−-- | Decode end point address
884−decodeEndPointAddress :: EndPointAddress
885− -> Maybe (N.HostName, N.ServiceName, EndPointId)
886−decodeEndPointAddress (EndPointAddress bs) =
887− case map BSC.unpack $ BSC.split ':' bs of
888− [host, port, endPointIdStr] ->
889− case reads endPointIdStr of
890− [(endPointId, "")] -> Just (host, port, endPointId)
891− _ -> Nothing
892− _ ->
893− Nothing
894−
895−-- | Do a (blocking) remote request
896−--
897−-- May throw IO (user) exception if the local or the remote endpoint is closed,
898−-- or if the send fails.
899−doRemoteRequest :: EndPointPair -> ControlHeader -> IO [ByteString]
900−doRemoteRequest (ourEndPoint, theirEndPoint) header = do
901− replyMVar <- newEmptyMVar
902− modifyRemoteState_ (ourEndPoint, theirEndPoint) RemoteStatePatternMatch
903− { caseValid = \vst -> do
904− let reqId = vst ^. nextCtrlRequestId
905− sendOn vst [encodeInt32 header, encodeInt32 reqId]
906− return ( RemoteEndPointValid
907− . (nextCtrlRequestId ^: (+ 1))
908− . (pendingCtrlRequestsAt reqId ^= Just replyMVar)
909− $ vst
910− )
911− -- Error cases
912− , caseInvalid =
913− throwIO
914− , caseInit = \_ _ ->
915− relyViolation (ourEndPoint, theirEndPoint) "doRemoteRequest (init)"
916− , caseClosing = \_ _ ->
917− relyViolation (ourEndPoint, theirEndPoint) "doRemoteRequest (closing)"
918− , caseClosed =
919− relyViolation (ourEndPoint, theirEndPoint) "doRemoteRequest (closed)"
920− , caseFailed =
921− throwIO
922− }
923− mReply <- takeMVar replyMVar
924− case mReply of
925− Left err -> throwIO err
926− Right reply -> return reply
927−
928−-- | Send a CloseSocket request if the remote endpoint is unused
929−closeIfUnused :: EndPointPair -> IO ()
930−closeIfUnused (ourEndPoint, theirEndPoint) =
931− modifyRemoteState_ (ourEndPoint, theirEndPoint) remoteStateIdentity
932− { caseValid = \vst ->
933− if vst ^. remoteOutgoing == 0 && IntSet.null (vst ^. remoteIncoming)
934− then do
935− sendOn vst [encodeInt32 CloseSocket]
936− resolved <- newEmptyMVar
937− return $ RemoteEndPointClosing resolved vst
938− else
939− return $ RemoteEndPointValid vst
940− }
941−
942−--------------------------------------------------------------------------------
943−-- As soon as a remote connection fails, we want to put notify our endpoint --
944−-- and put it into a closed state. Since this may happen in many places, we --
945−-- provide some abstractions. --
946−--------------------------------------------------------------------------------
947−
948−data RemoteStatePatternMatch a = RemoteStatePatternMatch
949− { caseInvalid :: TransportError ConnectErrorCode -> IO a
950− , caseInit :: MVar () -> RequestedBy -> IO a
951− , caseValid :: ValidRemoteEndPointState -> IO a
952− , caseClosing :: MVar () -> ValidRemoteEndPointState -> IO a
953− , caseClosed :: IO a
954− , caseFailed :: IOException -> IO a
955− }
956−
957−remoteStateIdentity :: RemoteStatePatternMatch RemoteState
958−remoteStateIdentity =
959− RemoteStatePatternMatch
960− { caseInvalid = return . RemoteEndPointInvalid
961− , caseInit = (return .) . RemoteEndPointInit
962− , caseValid = return . RemoteEndPointValid
963− , caseClosing = (return .) . RemoteEndPointClosing
964− , caseClosed = return RemoteEndPointClosed
965− , caseFailed = return . RemoteEndPointFailed
966− }
967−
968−-- | Like modifyMVar, but if an I/O exception occurs don't restore the remote
969−-- endpoint to its original value but close it instead
970−modifyRemoteState :: EndPointPair
971− -> RemoteStatePatternMatch (RemoteState, a)
972− -> IO a
973−modifyRemoteState (ourEndPoint, theirEndPoint) match =
974− mask $ \restore -> do
975− st <- takeMVar theirState
976− case st of
977− RemoteEndPointValid vst -> do
978− mResult <- try $ restore (caseValid match vst)
979− case mResult of
980− Right (st', a) -> do
981− putMVar theirState st'
982− return a
983− Left ex -> do
984− case fromException ex of
985− Just ioEx -> handleIOException ioEx vst
986− Nothing -> putMVar theirState st
987− throwIO ex
988− -- The other cases are less interesting, because unless the endpoint is
989− -- in Valid state we're not supposed to do any IO on it
990− RemoteEndPointInit resolved origin -> do
991− (st', a) <- onException (restore $ caseInit match resolved origin)
992− (putMVar theirState st)
993− putMVar theirState st'
994− return a
995− RemoteEndPointClosing resolved vst -> do
996− (st', a) <- onException (restore $ caseClosing match resolved vst)
997− (putMVar theirState st)
998− putMVar theirState st'
999− return a
1000− RemoteEndPointInvalid err -> do
1001− (st', a) <- onException (restore $ caseInvalid match err)
1002− (putMVar theirState st)
1003− putMVar theirState st'
1004− return a
1005− RemoteEndPointClosed -> do
1006− (st', a) <- onException (restore $ caseClosed match)
1007− (putMVar theirState st)
1008− putMVar theirState st'
1009− return a
1010− RemoteEndPointFailed err -> do
1011− (st', a) <- onException (restore $ caseFailed match err)
1012− (putMVar theirState st)
1013− putMVar theirState st'
1014− return a
1015− where
1016− theirState :: MVar RemoteState
1017− theirState = remoteState theirEndPoint
1018−
1019− handleIOException :: IOException -> ValidRemoteEndPointState -> IO ()
1020− handleIOException ex vst = do
1021− tryCloseSocket (remoteSocket vst)
1022− putMVar theirState (RemoteEndPointFailed ex)
1023− let incoming = IntSet.elems $ vst ^. remoteIncoming
1024− code = EventConnectionLost (remoteAddress theirEndPoint) incoming
1025− err = TransportError code (show ex)
1026− writeChan (localChannel ourEndPoint) $ ErrorEvent err
1027−
1028−-- | Like 'modifyRemoteState' but without a return value
1029−modifyRemoteState_ :: EndPointPair
1030− -> RemoteStatePatternMatch RemoteState
1031− -> IO ()
1032−modifyRemoteState_ (ourEndPoint, theirEndPoint) match =
1033− modifyRemoteState (ourEndPoint, theirEndPoint)
1034− RemoteStatePatternMatch
1035− { caseInvalid = u . caseInvalid match
1036− , caseInit = \resolved origin -> u $ caseInit match resolved origin
1037− , caseValid = u . caseValid match
1038− , caseClosing = \resolved vst -> u $ caseClosing match resolved vst
1039− , caseClosed = u $ caseClosed match
1040− , caseFailed = u . caseFailed match
1041− }
1042− where
1043− u :: IO a -> IO (a, ())
1044− u p = p >>= \a -> return (a, ())
1045−
1046−-- | Like 'modifyRemoteState' but without the ability to change the state
1047−withRemoteState :: EndPointPair
1048− -> RemoteStatePatternMatch a
1049− -> IO a
1050−withRemoteState (ourEndPoint, theirEndPoint) match =
1051− modifyRemoteState (ourEndPoint, theirEndPoint)
1052− RemoteStatePatternMatch
1053− { caseInvalid = \err -> do
1054− a <- caseInvalid match err
1055− return (RemoteEndPointInvalid err, a)
1056− , caseInit = \resolved origin -> do
1057− a <- caseInit match resolved origin
1058− return (RemoteEndPointInit resolved origin, a)
1059− , caseValid = \vst -> do
1060− a <- caseValid match vst
1061− return (RemoteEndPointValid vst, a)
1062− , caseClosing = \resolved vst -> do
1063− a <- caseClosing match resolved vst
1064− return (RemoteEndPointClosing resolved vst, a)
1065− , caseClosed = do
1066− a <- caseClosed match
1067− return (RemoteEndPointClosed, a)
1068− , caseFailed = \err -> do
1069− a <- caseFailed match err
1070− return (RemoteEndPointFailed err, a)
1071− }
1072−
1073−--------------------------------------------------------------------------------
1074−-- Incoming requests --
1075−--------------------------------------------------------------------------------
1076−
1077−-- | Handle a connection request (that is, a remote endpoint that is trying to
1078−-- establish a TCP connection with us)
788+-- | Handle a connection request (that is, a remote endpoint that is trying to
789+-- establish a TCP connection with us)
1079790--
1080791-- 'handleConnectionRequest' runs in the context of the transport thread, which
1081792-- can be killed asynchronously by 'closeTransport'. We fork a separate thread
@@ -1106,6 +817,7 @@ handleConnectionRequest transport sock = handle handleException $ do
1106817 where
1107818 go :: LocalEndPoint -> EndPointAddress -> IO ()
1108819 go ourEndPoint theirAddress = do
820+ -- This runs in a thread that will never be killed
1109821 mEndPoint <- handle ((>> return Nothing) . handleException) $ do
1110822 resetIfBroken ourEndPoint theirAddress
1111823 (theirEndPoint, isNew) <-
@@ -1131,103 +843,22 @@ handleConnectionRequest transport sock = handle handleException $ do
1131843 -- If we left the scope of the exception handler with a return value of
1132844 -- Nothing then the socket is already closed; otherwise, the socket has
1133845 -- been recorded as part of the remote endpoint. Either way, we no longer
1134− -- have to worry about closing the socket on receiving an asynchronous
1135− -- exception from this point forward.
1136− forM_ mEndPoint $ handleIncomingMessages . (,) ourEndPoint
1137−
1138− handleException :: SomeException -> IO ()
1139− handleException _ex =
1140− -- putStrLn $ "handleConnectionRequest " ++ show _ex
1141− tryCloseSocket sock
1142−
1143−-- | Find a remote endpoint. If the remote endpoint does not yet exist we
1144−-- create it in Init state. Returns if the endpoint was new.
1145−findRemoteEndPoint
1146− :: LocalEndPoint
1147− -> EndPointAddress
1148− -> RequestedBy
1149− -> IO (RemoteEndPoint, Bool)
1150−findRemoteEndPoint ourEndPoint theirAddress findOrigin = go
1151− where
1152− go = do
1153− (theirEndPoint, isNew) <- modifyMVar ourState $ \st -> case st of
1154− LocalEndPointValid vst -> case vst ^. localConnectionTo theirAddress of
1155− Just theirEndPoint ->
1156− return (st, (theirEndPoint, False))
1157− Nothing -> do
1158− resolved <- newEmptyMVar
1159− theirState <- newMVar (RemoteEndPointInit resolved findOrigin)
1160− let theirEndPoint = RemoteEndPoint
1161− { remoteAddress = theirAddress
1162− , remoteState = theirState
1163− , remoteId = vst ^. nextRemoteId
1164− }
1165− return ( LocalEndPointValid
1166− . (localConnectionTo theirAddress ^= Just theirEndPoint)
1167− . (nextRemoteId ^: (+ 1))
1168− $ vst
1169− , (theirEndPoint, True)
1170− )
1171− LocalEndPointClosed ->
1172− throwIO $ userError "Local endpoint closed"
1173−
1174− if isNew
1175− then
1176− return (theirEndPoint, True)
1177− else do
1178− let theirState = remoteState theirEndPoint
1179− snapshot <- modifyMVar theirState $ \st -> case st of
1180− RemoteEndPointValid vst ->
1181− case findOrigin of
1182− RequestedByUs -> do
1183− let st' = RemoteEndPointValid
1184− . (remoteOutgoing ^: (+ 1))
1185− $ vst
1186− return (st', st')
1187− RequestedByThem ->
1188− return (st, st)
1189− _ ->
1190− return (st, st)
1191− -- The snapshot may no longer be up to date at this point, but if we
1192− -- increased the refcount then it can only either be valid or closed
1193− -- (after an explicit call to 'closeEndPoint' or 'closeTransport')
1194− case snapshot of
1195− RemoteEndPointInvalid err ->
1196− throwIO err
1197− RemoteEndPointInit resolved initOrigin ->
1198− case (findOrigin, initOrigin) of
1199− (RequestedByUs, RequestedByUs) ->
1200− readMVar resolved >> go
1201− (RequestedByUs, RequestedByThem) ->
1202− readMVar resolved >> go
1203− (RequestedByThem, RequestedByUs) ->
1204− if ourAddress > theirAddress
1205− then
1206− -- Wait for the Crossed message
1207− readMVar resolved >> go
1208− else
1209− return (theirEndPoint, False)
1210− (RequestedByThem, RequestedByThem) ->
1211− throwIO $ userError "Already connected (B)"
1212− RemoteEndPointValid _ ->
1213− -- We assume that the request crossed if we find the endpoint in
1214− -- Valid state. It is possible that this is really an invalid
1215− -- request, but only in the case of a broken client (we don't
1216− -- maintain enough history to be able to tell the difference).
1217− return (theirEndPoint, False)
1218− RemoteEndPointClosing resolved _ ->
1219− readMVar resolved >> go
1220− RemoteEndPointClosed ->
1221− go
1222− RemoteEndPointFailed err ->
1223− throwIO err
1224−
1225− ourState = localState ourEndPoint
1226− ourAddress = localAddress ourEndPoint
846+ -- have to worry about closing the socket on receiving an asynchronous
847+ -- exception from this point forward.
848+ forM_ mEndPoint $ handleIncomingMessages . (,) ourEndPoint
849+
850+ handleException :: SomeException -> IO ()
851+ handleException ex = do
852+ tryCloseSocket sock
853+ rethrowIfAsync (fromException ex)
854+
855+ rethrowIfAsync :: Maybe AsyncException -> IO ()
856+ rethrowIfAsync = mapM_ throwIO
1227857
1228858-- | Handle requests from a remote endpoint.
1229859--
1230860-- Returns only if the remote party closes the socket or if an error occurs.
861+-- This runs in a thread that will never be killed.
1231862handleIncomingMessages :: EndPointPair -> IO ()
1232863handleIncomingMessages (ourEndPoint, theirEndPoint) = do
1233864 mSock <- withMVar theirState $ \st ->
@@ -1258,11 +889,6 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
1258889 -- (because a 'send' failed) -- the individual handlers below will throw a
1259890 -- user exception which is then caught and handled the same way as an
1260891 -- exception thrown by 'recv'.
1261− --
1262− -- Note: modifyRemoteState closes the socket before putting the remote
1263− -- endpoint in closing state, so it is not possible that modifyRemoteState
1264− -- puts the remote endpoint in Closing state only for it to be reset to
1265− -- Valid by the RequestConnectionId handler below.
1266892 go :: N.Socket -> IO ()
1267893 go sock = do
1268894 connId <- recvInt32 sock
@@ -1290,8 +916,6 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
1290916 -- Create a new connection
1291917 createNewConnection :: ControlRequestId -> IO ()
1292918 createNewConnection reqId = do
1293− -- getNextConnectionId throws an exception if ourEndPoint is closed; but
1294− -- if this endpoint is closed, this thread will soon die anyway
1295919 newId <- getNextConnectionId ourEndPoint
1296920 modifyMVar_ theirState $ \st -> do
1297921 vst <- case st of
@@ -1321,9 +945,6 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
1321945 : encodeInt32 reqId
1322946 : prependLength [encodeInt32 newId]
1323947 )
1324− -- We add the new connection ID to the list of open connections only
1325− -- once the endpoint has been notified of the new connection (sendOn
1326− -- may fail)
1327948 return (RemoteEndPointValid vst)
1328949 writeChan ourChannel (ConnectionOpened newId ReliableOrdered theirAddr)
1329950
@@ -1425,7 +1046,7 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
14251046 relyViolation (ourEndPoint, theirEndPoint)
14261047 "handleIncomingMessages:closeSocket (closed)"
14271048
1428− -- Read a message and output it on the endPoint's channel By rights we
1049+ -- Read a message and output it on the endPoint's channel. By rights we
14291050 -- should verify that the connection ID is valid, but this is unnecessary
14301051 -- overhead
14311052 readMessage :: N.Socket -> ConnectionId -> IO ()
@@ -1465,6 +1086,221 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
14651086 RemoteEndPointFailed err' ->
14661087 return (RemoteEndPointFailed err')
14671088
1089+--------------------------------------------------------------------------------
1090+-- Uninterruptable auxiliary functions --
1091+-- --
1092+-- All these functions assume they are running in a thread which will never --
1093+-- be killed.
1094+--------------------------------------------------------------------------------
1095+
1096+-- | Request a connection to a remote endpoint
1097+--
1098+-- This will block until we get a connection ID from the remote endpoint; if
1099+-- the remote endpoint was in 'RemoteEndPointClosing' state then we will
1100+-- additionally block until that is resolved.
1101+--
1102+-- May throw a TransportError ConnectErrorCode exception.
1103+requestConnectionTo :: TCPParameters
1104+ -> LocalEndPoint
1105+ -> EndPointAddress
1106+ -> ConnectHints
1107+ -> IO (RemoteEndPoint, ConnectionId)
1108+requestConnectionTo params ourEndPoint theirAddress hints = go
1109+ where
1110+ go = do
1111+ (theirEndPoint, isNew) <- mapIOException connectFailed $
1112+ findRemoteEndPoint ourEndPoint theirAddress RequestedByUs
1113+
1114+ if isNew
1115+ then do
1116+ forkIO . handle absorbAllExceptions $
1117+ setupRemoteEndPoint params (ourEndPoint, theirEndPoint) hints
1118+ go
1119+ else do
1120+ reply <- mapIOException connectFailed $
1121+ doRemoteRequest (ourEndPoint, theirEndPoint) RequestConnectionId
1122+ return (theirEndPoint, decodeInt32 . BS.concat $ reply)
1123+
1124+ connectFailed :: IOException -> TransportError ConnectErrorCode
1125+ connectFailed = TransportError ConnectFailed . show
1126+
1127+ absorbAllExceptions :: SomeException -> IO ()
1128+ absorbAllExceptions _ex =
1129+ return ()
1130+
1131+-- | Set up a remote endpoint
1132+setupRemoteEndPoint :: TCPParameters -> EndPointPair -> ConnectHints -> IO ()
1133+setupRemoteEndPoint params (ourEndPoint, theirEndPoint) hints = do
1134+ result <- socketToEndPoint ourAddress
1135+ theirAddress
1136+ (tcpReuseClientAddr params)
1137+ (connectTimeout hints)
1138+ didAccept <- case result of
1139+ Right (sock, ConnectionRequestAccepted) -> do
1140+ let vst = ValidRemoteEndPointState
1141+ { remoteSocket = sock
1142+ , _remoteOutgoing = 0
1143+ , _remoteIncoming = IntSet.empty
1144+ , sendOn = sendMany sock
1145+ , _pendingCtrlRequests = IntMap.empty
1146+ , _nextCtrlRequestId = 0
1147+ }
1148+ resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst)
1149+ return True
1150+ Right (sock, ConnectionRequestInvalid) -> do
1151+ let err = invalidAddress "setupRemoteEndPoint: Invalid endpoint"
1152+ resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointInvalid err)
1153+ tryCloseSocket sock
1154+ return False
1155+ Right (sock, ConnectionRequestCrossed) -> do
1156+ resolveInit (ourEndPoint, theirEndPoint) RemoteEndPointClosed
1157+ tryCloseSocket sock
1158+ return False
1159+ Left err -> do
1160+ resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointInvalid err)
1161+ return False
1162+
1163+ when didAccept $ handleIncomingMessages (ourEndPoint, theirEndPoint)
1164+ where
1165+ ourAddress = localAddress ourEndPoint
1166+ theirAddress = remoteAddress theirEndPoint
1167+ invalidAddress = TransportError ConnectNotFound
1168+
1169+-- | Do a (blocking) remote request
1170+--
1171+-- May throw IO (user) exception if the local or the remote endpoint is closed,
1172+-- if the send fails, or if the remote endpoint fails before it replies.
1173+doRemoteRequest :: EndPointPair -> ControlHeader -> IO [ByteString]
1174+doRemoteRequest (ourEndPoint, theirEndPoint) header = do
1175+ replyMVar <- newEmptyMVar
1176+ modifyRemoteState_ (ourEndPoint, theirEndPoint) RemoteStatePatternMatch
1177+ { caseValid = \vst -> do
1178+ let reqId = vst ^. nextCtrlRequestId
1179+ sendOn vst [encodeInt32 header, encodeInt32 reqId]
1180+ return ( RemoteEndPointValid
1181+ . (nextCtrlRequestId ^: (+ 1))
1182+ . (pendingCtrlRequestsAt reqId ^= Just replyMVar)
1183+ $ vst
1184+ )
1185+ -- Error cases
1186+ , caseInvalid =
1187+ throwIO
1188+ , caseInit = \_ _ ->
1189+ relyViolation (ourEndPoint, theirEndPoint) "doRemoteRequest (init)"
1190+ , caseClosing = \_ _ ->
1191+ relyViolation (ourEndPoint, theirEndPoint) "doRemoteRequest (closing)"
1192+ , caseClosed =
1193+ relyViolation (ourEndPoint, theirEndPoint) "doRemoteRequest (closed)"
1194+ , caseFailed =
1195+ throwIO
1196+ }
1197+ mReply <- takeMVar replyMVar
1198+ case mReply of
1199+ Left err -> throwIO err
1200+ Right reply -> return reply
1201+
1202+-- | Send a CloseSocket request if the remote endpoint is unused
1203+closeIfUnused :: EndPointPair -> IO ()
1204+closeIfUnused (ourEndPoint, theirEndPoint) =
1205+ modifyRemoteState_ (ourEndPoint, theirEndPoint) remoteStateIdentity
1206+ { caseValid = \vst ->
1207+ if vst ^. remoteOutgoing == 0 && IntSet.null (vst ^. remoteIncoming)
1208+ then do
1209+ sendOn vst [encodeInt32 CloseSocket]
1210+ resolved <- newEmptyMVar
1211+ return $ RemoteEndPointClosing resolved vst
1212+ else
1213+ return $ RemoteEndPointValid vst
1214+ }
1215+
1216+-- | Reset a remote endpoint if it is in Invalid mode
1217+--
1218+-- If the remote endpoint is currently in broken state, and
1219+--
1220+-- - a user calls the API function 'connect', or and the remote endpoint is
1221+-- - an inbound connection request comes in from this remote address
1222+--
1223+-- we remove the remote endpoint first.
1224+--
1225+-- Throws a TransportError ConnectFailed exception if the local endpoint is
1226+-- closed.
1227+resetIfBroken :: LocalEndPoint -> EndPointAddress -> IO ()
1228+resetIfBroken ourEndPoint theirAddress = do
1229+ mTheirEndPoint <- withMVar (localState ourEndPoint) $ \st -> case st of
1230+ LocalEndPointValid vst ->
1231+ return (vst ^. localConnectionTo theirAddress)
1232+ LocalEndPointClosed ->
1233+ throwIO $ TransportError ConnectFailed "Endpoint closed"
1234+ forM_ mTheirEndPoint $ \theirEndPoint ->
1235+ withMVar (remoteState theirEndPoint) $ \st -> case st of
1236+ RemoteEndPointInvalid _ ->
1237+ removeRemoteEndPoint (ourEndPoint, theirEndPoint)
1238+ RemoteEndPointFailed _ ->
1239+ removeRemoteEndPoint (ourEndPoint, theirEndPoint)
1240+ _ ->
1241+ return ()
1242+
1243+-- | Special case of 'apiConnect': connect an endpoint to itself
1244+--
1245+-- May throw a TransportError ConnectErrorCode (if the local endpoint is closed)
1246+connectToSelf :: LocalEndPoint
1247+ -> IO Connection
1248+connectToSelf ourEndPoint = do
1249+ connAlive <- newIORef True -- Protected by the local endpoint lock
1250+ connId <- mapIOException connectFailed $ getNextConnectionId ourEndPoint
1251+ writeChan ourChan $
1252+ ConnectionOpened connId ReliableOrdered (localAddress ourEndPoint)
1253+ return Connection
1254+ { send = selfSend connAlive connId
1255+ , close = selfClose connAlive connId
1256+ }
1257+ where
1258+ selfSend :: IORef Bool
1259+ -> ConnectionId
1260+ -> [ByteString]
1261+ -> IO (Either (TransportError SendErrorCode) ())
1262+ selfSend connAlive connId msg =
1263+ try . withMVar ourState $ \st -> case st of
1264+ LocalEndPointValid _ -> do
1265+ alive <- readIORef connAlive
1266+ if alive
1267+ then writeChan ourChan (Received connId msg)
1268+ else throwIO $ TransportError SendClosed "Connection closed"
1269+ LocalEndPointClosed ->
1270+ throwIO $ TransportError SendFailed "Endpoint closed"
1271+
1272+ selfClose :: IORef Bool -> ConnectionId -> IO ()
1273+ selfClose connAlive connId =
1274+ withMVar ourState $ \st -> case st of
1275+ LocalEndPointValid _ -> do
1276+ alive <- readIORef connAlive
1277+ when alive $ do
1278+ writeChan ourChan (ConnectionClosed connId)
1279+ writeIORef connAlive False
1280+ LocalEndPointClosed ->
1281+ return ()
1282+
1283+ ourChan = localChannel ourEndPoint
1284+ ourState = localState ourEndPoint
1285+ connectFailed = TransportError ConnectFailed . show
1286+
1287+-- | Resolve an endpoint currently in 'Init' state
1288+resolveInit :: EndPointPair -> RemoteState -> IO ()
1289+resolveInit (ourEndPoint, theirEndPoint) newState =
1290+ modifyMVar_ (remoteState theirEndPoint) $ \st -> case st of
1291+ RemoteEndPointInit resolved _ -> do
1292+ putMVar resolved ()
1293+ case newState of
1294+ RemoteEndPointClosed ->
1295+ removeRemoteEndPoint (ourEndPoint, theirEndPoint)
1296+ _ ->
1297+ return ()
1298+ return newState
1299+ RemoteEndPointFailed ex ->
1300+ throwIO ex
1301+ _ ->
1302+ relyViolation (ourEndPoint, theirEndPoint) "resolveInit"
1303+
14681304-- | Get the next connection ID
14691305--
14701306-- Throws an IO exception when the endpoint is closed.
@@ -1476,96 +1312,266 @@ getNextConnectionId ourEndpoint =
14761312 return ( LocalEndPointValid
14771313 . (nextConnectionId ^= connId + 1)
14781314 $ vst
1479− , connId)
1480− LocalEndPointClosed ->
1481− throwIO $ userError "Local endpoint closed"
1315+ , connId)
1316+ LocalEndPointClosed ->
1317+ throwIO $ userError "Local endpoint closed"
1318+
1319+-- | Create a new local endpoint
1320+--
1321+-- May throw a TransportError NewEndPointErrorCode exception if the transport
1322+-- is closed.
1323+createLocalEndPoint :: TCPTransport -> IO LocalEndPoint
1324+createLocalEndPoint transport = do
1325+ chan <- newChan
1326+ state <- newMVar . LocalEndPointValid $ ValidLocalEndPointState
1327+ { _nextConnectionId = firstNonReservedConnectionId
1328+ , _localConnections = Map.empty
1329+ , _nextRemoteId = 0
1330+ }
1331+ modifyMVar (transportState transport) $ \st -> case st of
1332+ TransportValid vst -> do
1333+ let ix = vst ^. nextEndPointId
1334+ let addr = encodeEndPointAddress (transportHost transport)
1335+ (transportPort transport)
1336+ ix
1337+ let localEndPoint = LocalEndPoint { localAddress = addr
1338+ , localChannel = chan
1339+ , localState = state
1340+ }
1341+ return ( TransportValid
1342+ . (localEndPointAt addr ^= Just localEndPoint)
1343+ . (nextEndPointId ^= ix + 1)
1344+ $ vst
1345+ , localEndPoint
1346+ )
1347+ TransportClosed ->
1348+ throwIO (TransportError NewEndPointFailed "Transport closed")
1349+
1350+
1351+-- | Remove reference to a remote endpoint from a local endpoint
1352+--
1353+-- If the local endpoint is closed, do nothing
1354+removeRemoteEndPoint :: EndPointPair -> IO ()
1355+removeRemoteEndPoint (ourEndPoint, theirEndPoint) =
1356+ modifyMVar_ ourState $ \st -> case st of
1357+ LocalEndPointValid vst ->
1358+ case vst ^. localConnectionTo theirAddress of
1359+ Nothing ->
1360+ return st
1361+ Just remoteEndPoint' ->
1362+ if remoteId remoteEndPoint' == remoteId theirEndPoint
1363+ then return
1364+ ( LocalEndPointValid
1365+ . (localConnectionTo (remoteAddress theirEndPoint) ^= Nothing)
1366+ $ vst
1367+ )
1368+ else return st
1369+ LocalEndPointClosed ->
1370+ return LocalEndPointClosed
1371+ where
1372+ ourState = localState ourEndPoint
1373+ theirAddress = remoteAddress theirEndPoint
1374+
1375+-- | Remove reference to a local endpoint from the transport state
1376+--
1377+-- Does nothing if the transport is closed
1378+removeLocalEndPoint :: TCPTransport -> LocalEndPoint -> IO ()
1379+removeLocalEndPoint transport ourEndPoint =
1380+ modifyMVar_ (transportState transport) $ \st -> case st of
1381+ TransportValid vst ->
1382+ return ( TransportValid
1383+ . (localEndPointAt (localAddress ourEndPoint) ^= Nothing)
1384+ $ vst
1385+ )
1386+ TransportClosed ->
1387+ return TransportClosed
1388+
1389+-- | Find a remote endpoint. If the remote endpoint does not yet exist we
1390+-- create it in Init state. Returns if the endpoint was new.
1391+findRemoteEndPoint
1392+ :: LocalEndPoint
1393+ -> EndPointAddress
1394+ -> RequestedBy
1395+ -> IO (RemoteEndPoint, Bool)
1396+findRemoteEndPoint ourEndPoint theirAddress findOrigin = go
1397+ where
1398+ go = do
1399+ (theirEndPoint, isNew) <- modifyMVar ourState $ \st -> case st of
1400+ LocalEndPointValid vst -> case vst ^. localConnectionTo theirAddress of
1401+ Just theirEndPoint ->
1402+ return (st, (theirEndPoint, False))
1403+ Nothing -> do
1404+ resolved <- newEmptyMVar
1405+ theirState <- newMVar (RemoteEndPointInit resolved findOrigin)
1406+ let theirEndPoint = RemoteEndPoint
1407+ { remoteAddress = theirAddress
1408+ , remoteState = theirState
1409+ , remoteId = vst ^. nextRemoteId
1410+ }
1411+ return ( LocalEndPointValid
1412+ . (localConnectionTo theirAddress ^= Just theirEndPoint)
1413+ . (nextRemoteId ^: (+ 1))
1414+ $ vst
1415+ , (theirEndPoint, True)
1416+ )
1417+ LocalEndPointClosed ->
1418+ throwIO $ userError "Local endpoint closed"
1419+
1420+ if isNew
1421+ then
1422+ return (theirEndPoint, True)
1423+ else do
1424+ let theirState = remoteState theirEndPoint
1425+ snapshot <- modifyMVar theirState $ \st -> case st of
1426+ RemoteEndPointValid vst ->
1427+ case findOrigin of
1428+ RequestedByUs -> do
1429+ let st' = RemoteEndPointValid
1430+ . (remoteOutgoing ^: (+ 1))
1431+ $ vst
1432+ return (st', st')
1433+ RequestedByThem ->
1434+ return (st, st)
1435+ _ ->
1436+ return (st, st)
1437+ -- The snapshot may no longer be up to date at this point, but if we
1438+ -- increased the refcount then it can only either be Valid or Failed
1439+ -- (after an explicit call to 'closeEndPoint' or 'closeTransport')
1440+ case snapshot of
1441+ RemoteEndPointInvalid err ->
1442+ throwIO err
1443+ RemoteEndPointInit resolved initOrigin ->
1444+ case (findOrigin, initOrigin) of
1445+ (RequestedByUs, RequestedByUs) ->
1446+ readMVar resolved >> go
1447+ (RequestedByUs, RequestedByThem) ->
1448+ readMVar resolved >> go
1449+ (RequestedByThem, RequestedByUs) ->
1450+ if ourAddress > theirAddress
1451+ then
1452+ -- Wait for the Crossed message
1453+ readMVar resolved >> go
1454+ else
1455+ return (theirEndPoint, False)
1456+ (RequestedByThem, RequestedByThem) ->
1457+ throwIO $ userError "Already connected"
1458+ RemoteEndPointValid _ ->
1459+ -- We assume that the request crossed if we find the endpoint in
1460+ -- Valid state. It is possible that this is really an invalid
1461+ -- request, but only in the case of a broken client (we don't
1462+ -- maintain enough history to be able to tell the difference).
1463+ return (theirEndPoint, False)
1464+ RemoteEndPointClosing resolved _ ->
1465+ readMVar resolved >> go
1466+ RemoteEndPointClosed ->
1467+ go
1468+ RemoteEndPointFailed err ->
1469+ throwIO err
1470+
1471+ ourState = localState ourEndPoint
1472+ ourAddress = localAddress ourEndPoint
14821473
14831474--------------------------------------------------------------------------------
1484−-- Uninterruptable auxiliary functions --
1485−-- --
1486−-- All these functions assume they are running in a thread which will never --
1487−-- be killed, and are designed to be used in conjunction with --
1488−-- 'asyncWhenCancelled'. To be explicit about this assumption the function --
1489−-- names are prefixed with 'u'. --
1475+-- "Stateless" (MVar free) functions --
14901476--------------------------------------------------------------------------------
14911477
1492−-- | Request a connection to a remote endpoint
1493−--
1494−-- This will block until we get a connection ID from the remote endpoint; if
1495−-- the remote endpoint was in 'RemoteEndPointClosing' state then we will
1496−-- additionally block until that is resolved.
1478+-- | Establish a connection to a remote endpoint
14971479--
1498−-- May throw a TransportError ConnectErrorCode exception.
1499−uRequestConnectionTo :: TCPParameters
1500− -> LocalEndPoint
1501− -> EndPointAddress
1502− -> ConnectHints
1503− -> IO (RemoteEndPoint, ConnectionId)
1504−uRequestConnectionTo params ourEndPoint theirAddress hints = go
1480+-- Maybe throw a TransportError
1481+socketToEndPoint :: EndPointAddress -- ^ Our address
1482+ -> EndPointAddress -- ^ Their address
1483+ -> Bool -- ^ Use SO_REUSEADDR?
1484+ -> Maybe Int -- ^ Timeout for connect
1485+ -> IO (Either (TransportError ConnectErrorCode)
1486+ (N.Socket, ConnectionRequestResponse))
1487+socketToEndPoint (EndPointAddress ourAddress) theirAddress reuseAddr timeout =
1488+ try $ do
1489+ (host, port, theirEndPointId) <- case decodeEndPointAddress theirAddress of
1490+ Nothing -> throwIO (failed . userError $ "Could not parse")
1491+ Just dec -> return dec
1492+ addr:_ <- mapIOException invalidAddress $
1493+ N.getAddrInfo Nothing (Just host) (Just port)
1494+ bracketOnError (createSocket addr) tryCloseSocket $ \sock -> do
1495+ when reuseAddr $
1496+ mapIOException failed $ N.setSocketOption sock N.ReuseAddr 1
1497+ mapIOException invalidAddress $
1498+ timeoutMaybe timeout timeoutError $
1499+ N.connect sock (N.addrAddress addr)
1500+ response <- mapIOException failed $ do
1501+ sendMany sock (encodeInt32 theirEndPointId : prependLength [ourAddress])
1502+ recvInt32 sock
1503+ case tryToEnum response of
1504+ Nothing -> throwIO (failed . userError $ "Unexpected response")
1505+ Just r -> return (sock, r)
15051506 where
1506− go = do
1507− (theirEndPoint, isNew) <- mapIOException connectFailed $
1508− findRemoteEndPoint ourEndPoint theirAddress RequestedByUs
1507+ createSocket :: N.AddrInfo -> IO N.Socket
1508+ createSocket addr = mapIOException insufficientResources $
1509+ N.socket (N.addrFamily addr) N.Stream N.defaultProtocol
15091510
1510− if isNew
1511− then do
1512− forkIO . handle absorbAllExceptions $
1513− uSetupRemoteEndPoint params (ourEndPoint, theirEndPoint) hints
1514− go
1515− else do
1516− reply <- mapIOException connectFailed $
1517− doRemoteRequest (ourEndPoint, theirEndPoint) RequestConnectionId
1518− return (theirEndPoint, decodeInt32 . BS.concat $ reply)
1511+ invalidAddress = TransportError ConnectNotFound . show
1512+ insufficientResources = TransportError ConnectInsufficientResources . show
1513+ failed = TransportError ConnectFailed . show
1514+ timeoutError = TransportError ConnectTimeout "Timed out"
15191515
1520− connectFailed = TransportError ConnectFailed . show
1516+-- | Encode end point address
1517+encodeEndPointAddress :: N.HostName
1518+ -> N.ServiceName
1519+ -> EndPointId
1520+ -> EndPointAddress
1521+encodeEndPointAddress host port ix = EndPointAddress . BSC.pack $
1522+ host ++ ":" ++ port ++ ":" ++ show ix
15211523
1522− absorbAllExceptions :: SomeException -> IO ()
1523− absorbAllExceptions _ex =
1524− return ()
1524+-- | Decode end point address
1525+decodeEndPointAddress :: EndPointAddress
1526+ -> Maybe (N.HostName, N.ServiceName, EndPointId)
1527+decodeEndPointAddress (EndPointAddress bs) =
1528+ case map BSC.unpack $ BSC.split ':' bs of
1529+ [host, port, endPointIdStr] ->
1530+ case reads endPointIdStr of
1531+ [(endPointId, "")] -> Just (host, port, endPointId)
1532+ _ -> Nothing
1533+ _ ->
1534+ Nothing
15251535
1526−-- | Set up a remote endpoint
1527−uSetupRemoteEndPoint :: TCPParameters -> EndPointPair -> ConnectHints -> IO ()
1528−uSetupRemoteEndPoint params (ourEndPoint, theirEndPoint) hints = do
1529− result <- socketToEndPoint ourAddress
1530− theirAddress
1531− (tcpReuseClientAddr params)
1532− (connectTimeout hints)
1533− didAccept <- case result of
1534− Right (sock, ConnectionRequestAccepted) -> do
1535− let vst = ValidRemoteEndPointState
1536− { remoteSocket = sock
1537− , _remoteOutgoing = 0
1538− , _remoteIncoming = IntSet.empty
1539− , sendOn = sendMany sock
1540− , _pendingCtrlRequests = IntMap.empty
1541− , _nextCtrlRequestId = 0
1542− }
1543− resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst)
1544− return True
1545− Right (sock, ConnectionRequestInvalid) -> do
1546− let err = invalidAddress "setupRemoteEndPoint: Invalid endpoint"
1547− resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointInvalid err)
1548− tryCloseSocket sock
1549− return False
1550− Right (sock, ConnectionRequestCrossed) -> do
1551− resolveInit (ourEndPoint, theirEndPoint) RemoteEndPointClosed
1552− tryCloseSocket sock
1553− return False
1554− Left err -> do
1555− resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointInvalid err)
1556− return False
1557−
1558− -- If we get to this point without an exception, then
1559− --
1560− -- if didAccept is False the socket has already been closed
1561− -- if didAccept is True, the socket has been stored as part of the remote
1562− -- state so we no longer need to worry about closing it when an
1563− -- asynchronous exception occurs
1564− when didAccept $ handleIncomingMessages (ourEndPoint, theirEndPoint)
1565− where
1566− ourAddress = localAddress ourEndPoint
1567− theirAddress = remoteAddress theirEndPoint
1568− invalidAddress = TransportError ConnectNotFound
1536+--------------------------------------------------------------------------------
1537+-- Functions from TransportInternals --
1538+--------------------------------------------------------------------------------
1539+
1540+-- Find a socket between two endpoints
1541+--
1542+-- Throws an IO exception if the socket could not be found.
1543+internalSocketBetween :: TCPTransport -- ^ Transport
1544+ -> EndPointAddress -- ^ Local endpoint
1545+ -> EndPointAddress -- ^ Remote endpoint
1546+ -> IO N.Socket
1547+internalSocketBetween transport ourAddress theirAddress = do
1548+ ourEndPoint <- withMVar (transportState transport) $ \st -> case st of
1549+ TransportClosed ->
1550+ throwIO $ userError "Transport closed"
1551+ TransportValid vst ->
1552+ case vst ^. localEndPointAt ourAddress of
1553+ Nothing -> throwIO $ userError "Local endpoint not found"
1554+ Just ep -> return ep
1555+ theirEndPoint <- withMVar (localState ourEndPoint) $ \st -> case st of
1556+ LocalEndPointClosed ->
1557+ throwIO $ userError "Local endpoint closed"
1558+ LocalEndPointValid vst ->
1559+ case vst ^. localConnectionTo theirAddress of
1560+ Nothing -> throwIO $ userError "Remote endpoint not found"
1561+ Just ep -> return ep
1562+ withMVar (remoteState theirEndPoint) $ \st -> case st of
1563+ RemoteEndPointInit _ _ ->
1564+ throwIO $ userError "Remote endpoint not yet initialized"
1565+ RemoteEndPointValid vst ->
1566+ return $ remoteSocket vst
1567+ RemoteEndPointClosing _ vst ->
1568+ return $ remoteSocket vst
1569+ RemoteEndPointClosed ->
1570+ throwIO $ userError "Remote endpoint closed"
1571+ RemoteEndPointInvalid err ->
1572+ throwIO err
1573+ RemoteEndPointFailed err ->
1574+ throwIO err
15691575
15701576--------------------------------------------------------------------------------
15711577-- Constants --