b16f5d4 · distributed-process

78import Control.Monad (when, unless)

79import Control.Exception ( IOException

80 , SomeException

81+ , AsyncException

81 , handle

82 , handle

82 , throw

83 , throw

83 , throwIO

84 , throwIO

203-- Internal datatypes --

204-- Internal datatypes --

204--------------------------------------------------------------------------------

205--------------------------------------------------------------------------------

205

206

206−-- We use underscores for fields that we might update (using accessores)

207+-- We use underscores for fields that we might update (using accessors)

207--

208--

208-- All data types follow the same structure:

209-- All data types follow the same structure:

209--

210--

254--

255--

255-- Init ---+---> Invalid

256-- Init ---+---> Invalid

256-- |

257-- |

258+-- +---------------------------------------\

259+-- | |

257−-- | /--------------\

260+-- | /--------------\ |

258−-- | | |

261+-- | | | |

259−-- | v |

262+-- | v | v

260-- +---> Valid ---+---> Closing ---+---> Closed

263-- +---> Valid ---+---> Closing ---+---> Closed

261-- | | | |

264-- | | | |

262-- | | | v

265-- | | | v

282-- CloseSocket) or be put back in 'Valid' state if the remote endpoint denies

285-- CloseSocket) or be put back in 'Valid' state if the remote endpoint denies

283-- the request.

286-- the request.

284--

287--

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+--

285-- Closed: The endpoint is put in Closed state after a successful garbage

293-- Closed: The endpoint is put in Closed state after a successful garbage

286-- collection.

294-- collection.

287--

295--

481

489

482-- | Close the transport

490-- | Close the transport

483apiCloseTransport :: TCPTransport -> Maybe ThreadId -> [Event] -> IO ()

491apiCloseTransport :: TCPTransport -> Maybe ThreadId -> [Event] -> IO ()

484−apiCloseTransport transport mTransportThread evs = do

492+apiCloseTransport transport mTransportThread evs =

493+ asyncWhenCancelled return $ do

485− mTSt <- modifyMVar (transportState transport) $ \st -> case st of

494+ mTSt <- modifyMVar (transportState transport) $ \st -> case st of

486− TransportValid vst -> return (TransportClosed, Just vst)

495+ TransportValid vst -> return (TransportClosed, Just vst)

487− TransportClosed -> return (TransportClosed, Nothing)

496+ TransportClosed -> return (TransportClosed, Nothing)

488− forM_ mTSt $ mapM_ (apiCloseEndPoint transport evs) . (^. localEndPoints)

497+ forM_ mTSt $ mapM_ (apiCloseEndPoint transport evs) . (^. localEndPoints)

489− -- This will invoke the termination handler, which in turn will call

498+ -- This will invoke the termination handler, which in turn will call

490− -- apiCloseTransport again, but then the transport will already be closed and

499+ -- apiCloseTransport again, but then the transport will already be closed

491− -- we won't be passed a transport thread, so we terminate immmediate

500+ -- and we won't be passed a transport thread, so we terminate immmediate

492− forM_ mTransportThread killThread

501+ forM_ mTransportThread killThread

493

502

494-- | Create a new endpoint

503-- | Create a new endpoint

495apiNewEndPoint :: TCPTransport

504apiNewEndPoint :: TCPTransport

496 -> IO (Either (TransportError NewEndPointErrorCode) EndPoint)

505 -> IO (Either (TransportError NewEndPointErrorCode) EndPoint)

497−apiNewEndPoint transport = try $ do

506+apiNewEndPoint transport =

507+ try . asyncWhenCancelled closeEndPoint $ do

498− ourEndPoint <- createLocalEndPoint transport

508+ ourEndPoint <- createLocalEndPoint transport

499− return EndPoint

509+ return EndPoint

500− { receive = readChan (localChannel ourEndPoint)

510+ { receive = readChan (localChannel ourEndPoint)

501− , address = localAddress ourEndPoint

511+ , address = localAddress ourEndPoint

502− , connect = apiConnect (transportParams transport) ourEndPoint

512+ , connect = apiConnect (transportParams transport) ourEndPoint

503− , closeEndPoint = let evs = [ EndPointClosed

513+ , closeEndPoint = let evs = [ EndPointClosed

504− , throw $ userError "Endpoint closed"

514+ , throw $ userError "Endpoint closed"

505− ] in

515+ ] in

506− apiCloseEndPoint transport evs ourEndPoint

516+ apiCloseEndPoint transport evs ourEndPoint

507− , newMulticastGroup = return . Left $ newMulticastGroupError

517+ , newMulticastGroup = return . Left $ newMulticastGroupError

508− , resolveMulticastGroup = return . Left . const resolveMulticastGroupError

518+ , resolveMulticastGroup = return . Left . const resolveMulticastGroupError

509− }

519+ }

510 where

520 where

511 newMulticastGroupError =

521 newMulticastGroupError =

512 TransportError NewMulticastGroupUnsupported "Multicast not supported"

522 TransportError NewMulticastGroupUnsupported "Multicast not supported"

521 -> ConnectHints -- ^ Hints

531 -> ConnectHints -- ^ Hints

522 -> IO (Either (TransportError ConnectErrorCode) Connection)

532 -> IO (Either (TransportError ConnectErrorCode) Connection)

523apiConnect params ourEndPoint theirAddress _reliability hints =

533apiConnect params ourEndPoint theirAddress _reliability hints =

534+ try . asyncWhenCancelled close $

524− if localAddress ourEndPoint == theirAddress

535+ if localAddress ourEndPoint == theirAddress

525− then connectToSelf ourEndPoint

536+ then connectToSelf ourEndPoint

526− else try . asyncWhenCancelled close $ do

537+ else do

527 resetIfBroken ourEndPoint theirAddress

538 resetIfBroken ourEndPoint theirAddress

528 (theirEndPoint, connId) <-

539 (theirEndPoint, connId) <-

529− uRequestConnectionTo params ourEndPoint theirAddress hints

540+ requestConnectionTo params ourEndPoint theirAddress hints

530 -- connAlive can be an IORef rather than an MVar because it is protected

541 -- connAlive can be an IORef rather than an MVar because it is protected

531 -- by the remoteState MVar. We don't need the overhead of locking twice.

542 -- by the remoteState MVar. We don't need the overhead of locking twice.

532 connAlive <- newIORef True

543 connAlive <- newIORef True

562 -> [ByteString] -- ^ Payload

573 -> [ByteString] -- ^ Payload

563 -> IO (Either (TransportError SendErrorCode) ())

574 -> IO (Either (TransportError SendErrorCode) ())

564apiSend (ourEndPoint, theirEndPoint) connId connAlive payload =

575apiSend (ourEndPoint, theirEndPoint) connId connAlive payload =

576+ -- We don't need the overhead of asyncWhenCancelled here

565 try . mapIOException sendFailed $

577 try . mapIOException sendFailed $

566 withRemoteState (ourEndPoint, theirEndPoint) RemoteStatePatternMatch

578 withRemoteState (ourEndPoint, theirEndPoint) RemoteStatePatternMatch

567 { caseInvalid = \_ ->

579 { caseInvalid = \_ ->

597 -> [Event] -- ^ Events used to report closure

609 -> [Event] -- ^ Events used to report closure

598 -> LocalEndPoint -- ^ Local endpoint

610 -> LocalEndPoint -- ^ Local endpoint

599 -> IO ()

611 -> IO ()

600−apiCloseEndPoint transport evs ourEndPoint = do

612+apiCloseEndPoint transport evs ourEndPoint =

613+ asyncWhenCancelled return $ do

601− -- Remove the reference from the transport state

614+ -- Remove the reference from the transport state

602− removeLocalEndPoint transport ourEndPoint

615+ removeLocalEndPoint transport ourEndPoint

603− -- Close the local endpoint

616+ -- Close the local endpoint

604− mOurState <- modifyMVar (localState ourEndPoint) $ \st ->

617+ mOurState <- modifyMVar (localState ourEndPoint) $ \st ->

605− case st of

618+ case st of

606− LocalEndPointValid vst ->

619+ LocalEndPointValid vst ->

607− return (LocalEndPointClosed, Just vst)

620+ return (LocalEndPointClosed, Just vst)

608− LocalEndPointClosed ->

621+ LocalEndPointClosed ->

609− return (LocalEndPointClosed, Nothing)

622+ return (LocalEndPointClosed, Nothing)

610− forM_ mOurState $ \vst -> do

623+ forM_ mOurState $ \vst -> do

611− forM_ (vst ^. localConnections) tryCloseRemoteSocket

624+ forM_ (vst ^. localConnections) tryCloseRemoteSocket

612− forM_ evs $ writeChan (localChannel ourEndPoint)

625+ forM_ evs $ writeChan (localChannel ourEndPoint)

613 where

626 where

614 -- Close the remote socket and return the set of all incoming connections

627 -- Close the remote socket and return the set of all incoming connections

615 tryCloseRemoteSocket :: RemoteEndPoint -> IO ()

628 tryCloseRemoteSocket :: RemoteEndPoint -> IO ()

637 RemoteEndPointFailed err ->

650 RemoteEndPointFailed err ->

638 return $ RemoteEndPointFailed err

651 return $ RemoteEndPointFailed err

639

652

640−-- | Special case of 'apiConnect': connect an endpoint to itself

653+--------------------------------------------------------------------------------

641−connectToSelf :: LocalEndPoint

654+-- As soon as a remote connection fails, we want to put notify our endpoint --

642− -> IO (Either (TransportError ConnectErrorCode) Connection)

655+-- and put it into a closed state. Since this may happen in many places, we --

643−connectToSelf ourEndPoint = try . mapIOException connectFailed $ do

656+-- provide some abstractions. --

644− connAlive <- newIORef True -- Protected by the local endpoint lock

657+--------------------------------------------------------------------------------

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"

666

658

667− selfClose :: IORef Bool -> ConnectionId -> IO ()

659+data RemoteStatePatternMatch a = RemoteStatePatternMatch

668− selfClose connAlive connId =

660+ { caseInvalid :: TransportError ConnectErrorCode -> IO a

669− withMVar ourState $ \st -> case st of

661+ , caseInit :: MVar () -> RequestedBy -> IO a

670− LocalEndPointValid _ -> do

662+ , caseValid :: ValidRemoteEndPointState -> IO a

671− alive <- readIORef connAlive

663+ , caseClosing :: MVar () -> ValidRemoteEndPointState -> IO a

672− when alive $ do

664+ , caseClosed :: IO a

673− writeChan ourChan (ConnectionClosed connId)

665+ , caseFailed :: IOException -> IO a

674− writeIORef connAlive False

666+ }

675− LocalEndPointClosed ->

676− return ()

677

667

678− ourChan = localChannel ourEndPoint

668+remoteStateIdentity :: RemoteStatePatternMatch RemoteState

679− ourState = localState ourEndPoint

669+remoteStateIdentity =

680− connectFailed = TransportError ConnectFailed . show

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+ }

681

678

682−--------------------------------------------------------------------------------

679+-- | Like modifyMVar, but if an I/O exception occurs don't restore the remote

683−-- Functions from TransportInternals --

680+-- endpoint to its original value but close it instead

684−--------------------------------------------------------------------------------

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

685

729

686−-- Find a socket between two endpoints

730+ handleIOException :: IOException -> ValidRemoteEndPointState -> IO ()

687−--

731+ handleIOException ex vst = do

688−-- Throws an IO exception if the socket could not be found.

732+ tryCloseSocket (remoteSocket vst)

689−internalSocketBetween :: TCPTransport -- ^ Transport

733+ putMVar theirState (RemoteEndPointFailed ex)

690− -> EndPointAddress -- ^ Local endpoint

734+ let incoming = IntSet.elems $ vst ^. remoteIncoming

691− -> EndPointAddress -- ^ Remote endpoint

735+ code = EventConnectionLost (remoteAddress theirEndPoint) incoming

692− -> IO N.Socket

736+ err = TransportError code (show ex)

693−internalSocketBetween transport ourAddress theirAddress = do

737+ writeChan (localChannel ourEndPoint) $ ErrorEvent err

694− ourEndPoint <- withMVar (transportState transport) $ \st -> case st of

738+

695− TransportClosed ->

739+-- | Like 'modifyRemoteState' but without a return value

696− throwIO $ userError "Transport closed"

740+modifyRemoteState_ :: EndPointPair

697− TransportValid vst ->

741+ -> RemoteStatePatternMatch RemoteState

698− case vst ^. localEndPointAt ourAddress of

742+ -> IO ()

699− Nothing -> throwIO $ userError "Local endpoint not found"

743+modifyRemoteState_ (ourEndPoint, theirEndPoint) match =

700− Just ep -> return ep

744+ modifyRemoteState (ourEndPoint, theirEndPoint)

701− theirEndPoint <- withMVar (localState ourEndPoint) $ \st -> case st of

745+ RemoteStatePatternMatch

702− LocalEndPointClosed ->

746+ { caseInvalid = u . caseInvalid match

703− throwIO $ userError "Local endpoint closed"

747+ , caseInit = \resolved origin -> u $ caseInit match resolved origin

704− LocalEndPointValid vst ->

748+ , caseValid = u . caseValid match

705− case vst ^. localConnectionTo theirAddress of

749+ , caseClosing = \resolved vst -> u $ caseClosing match resolved vst

706− Nothing -> throwIO $ userError "Remote endpoint not found"

750+ , caseClosed = u $ caseClosed match

707− Just ep -> return ep

751+ , caseFailed = u . caseFailed match

708− withMVar (remoteState theirEndPoint) $ \st -> case st of

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

709− RemoteEndPointInit _ _ ->

769+ return (RemoteEndPointInit resolved origin, a)

710− throwIO $ userError "Remote endpoint not yet initialized"

770+ , caseValid = \vst -> do

771+ a <- caseValid match vst

711− RemoteEndPointValid vst ->

772+ return (RemoteEndPointValid vst, a)

712− return $ remoteSocket vst

773+ , caseClosing = \resolved vst -> do

774+ a <- caseClosing match resolved vst

713− RemoteEndPointClosing _ vst ->

775+ return (RemoteEndPointClosing resolved vst, a)

714− return $ remoteSocket vst

776+ , caseClosed = do

777+ a <- caseClosed match

715− RemoteEndPointClosed ->

778+ return (RemoteEndPointClosed, a)

716− throwIO $ userError "Remote endpoint closed"

779+ , caseFailed = \err -> do

717− RemoteEndPointInvalid err ->

780+ a <- caseFailed match err

718− throwIO err

719− RemoteEndPointFailed err ->

781+ return (RemoteEndPointFailed err, a)

720− throwIO err

782+ }

721

783

722--------------------------------------------------------------------------------

784--------------------------------------------------------------------------------

723−-- Lower level functionality --

785+-- Incoming requests --

724--------------------------------------------------------------------------------

786--------------------------------------------------------------------------------

725

787

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

788+-- | Handle a connection request (that is, a remote endpoint that is trying to

1078−-- establish a TCP connection with us)

789+-- establish a TCP connection with us)

1079--

790--

1080-- 'handleConnectionRequest' runs in the context of the transport thread, which

791-- 'handleConnectionRequest' runs in the context of the transport thread, which

1081-- can be killed asynchronously by 'closeTransport'. We fork a separate thread

792-- can be killed asynchronously by 'closeTransport'. We fork a separate thread

1106 where

817 where

1107 go :: LocalEndPoint -> EndPointAddress -> IO ()

818 go :: LocalEndPoint -> EndPointAddress -> IO ()

1108 go ourEndPoint theirAddress = do

819 go ourEndPoint theirAddress = do

820+ -- This runs in a thread that will never be killed

1109 mEndPoint <- handle ((>> return Nothing) . handleException) $ do

821 mEndPoint <- handle ((>> return Nothing) . handleException) $ do

1110 resetIfBroken ourEndPoint theirAddress

822 resetIfBroken ourEndPoint theirAddress

1111 (theirEndPoint, isNew) <-

823 (theirEndPoint, isNew) <-

1131 -- If we left the scope of the exception handler with a return value of

843 -- If we left the scope of the exception handler with a return value of

1132 -- Nothing then the socket is already closed; otherwise, the socket has

844 -- Nothing then the socket is already closed; otherwise, the socket has

1133 -- been recorded as part of the remote endpoint. Either way, we no longer

845 -- 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

846+ -- have to worry about closing the socket on receiving an asynchronous

1135− -- exception from this point forward.

847+ -- exception from this point forward.

1136− forM_ mEndPoint $ handleIncomingMessages . (,) ourEndPoint

848+ forM_ mEndPoint $ handleIncomingMessages . (,) ourEndPoint

1137−

849+

1138− handleException :: SomeException -> IO ()

850+ handleException :: SomeException -> IO ()

1139− handleException _ex =

851+ handleException ex = do

1140− -- putStrLn $ "handleConnectionRequest " ++ show _ex

1141− tryCloseSocket sock

852+ tryCloseSocket sock

1142−

853+ rethrowIfAsync (fromException ex)

1143−-- | Find a remote endpoint. If the remote endpoint does not yet exist we

854+

1144−-- create it in Init state. Returns if the endpoint was new.

855+ rethrowIfAsync :: Maybe AsyncException -> IO ()

1145−findRemoteEndPoint

856+ rethrowIfAsync = mapM_ throwIO

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

1227

857

1228-- | Handle requests from a remote endpoint.

858-- | Handle requests from a remote endpoint.

1229--

859--

1230-- Returns only if the remote party closes the socket or if an error occurs.

860-- 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.

1231handleIncomingMessages :: EndPointPair -> IO ()

862handleIncomingMessages :: EndPointPair -> IO ()

1232handleIncomingMessages (ourEndPoint, theirEndPoint) = do

863handleIncomingMessages (ourEndPoint, theirEndPoint) = do

1233 mSock <- withMVar theirState $ \st ->

864 mSock <- withMVar theirState $ \st ->

1258 -- (because a 'send' failed) -- the individual handlers below will throw a

889 -- (because a 'send' failed) -- the individual handlers below will throw a

1259 -- user exception which is then caught and handled the same way as an

890 -- user exception which is then caught and handled the same way as an

1260 -- exception thrown by 'recv'.

891 -- 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.

1266 go :: N.Socket -> IO ()

892 go :: N.Socket -> IO ()

1267 go sock = do

893 go sock = do

1268 connId <- recvInt32 sock

894 connId <- recvInt32 sock

1290 -- Create a new connection

916 -- Create a new connection

1291 createNewConnection :: ControlRequestId -> IO ()

917 createNewConnection :: ControlRequestId -> IO ()

1292 createNewConnection reqId = do

918 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

1295 newId <- getNextConnectionId ourEndPoint

919 newId <- getNextConnectionId ourEndPoint

1296 modifyMVar_ theirState $ \st -> do

920 modifyMVar_ theirState $ \st -> do

1297 vst <- case st of

921 vst <- case st of

1321 : encodeInt32 reqId

945 : encodeInt32 reqId

1322 : prependLength [encodeInt32 newId]

946 : prependLength [encodeInt32 newId]

1323 )

947 )

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)

1327 return (RemoteEndPointValid vst)

948 return (RemoteEndPointValid vst)

1328 writeChan ourChannel (ConnectionOpened newId ReliableOrdered theirAddr)

949 writeChan ourChannel (ConnectionOpened newId ReliableOrdered theirAddr)

1329

950

1425 relyViolation (ourEndPoint, theirEndPoint)

1046 relyViolation (ourEndPoint, theirEndPoint)

1426 "handleIncomingMessages:closeSocket (closed)"

1047 "handleIncomingMessages:closeSocket (closed)"

1427

1048

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

1429 -- should verify that the connection ID is valid, but this is unnecessary

1050 -- should verify that the connection ID is valid, but this is unnecessary

1430 -- overhead

1051 -- overhead

1431 readMessage :: N.Socket -> ConnectionId -> IO ()

1052 readMessage :: N.Socket -> ConnectionId -> IO ()

1465 RemoteEndPointFailed err' ->

1086 RemoteEndPointFailed err' ->

1466 return (RemoteEndPointFailed err')

1087 return (RemoteEndPointFailed err')

1467

1088

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+

1468-- | Get the next connection ID

1304-- | Get the next connection ID

1469--

1305--

1470-- Throws an IO exception when the endpoint is closed.

1306-- Throws an IO exception when the endpoint is closed.

1476 return ( LocalEndPointValid

1312 return ( LocalEndPointValid

1477 . (nextConnectionId ^= connId + 1)

1313 . (nextConnectionId ^= connId + 1)

1478 $ vst

1314 $ vst

1479− , connId)

1315+ , connId)

1480− LocalEndPointClosed ->

1316+ LocalEndPointClosed ->

1481− throwIO $ userError "Local endpoint closed"

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

1482

1473

1483--------------------------------------------------------------------------------

1474--------------------------------------------------------------------------------

1484−-- Uninterruptable auxiliary functions --

1475+-- "Stateless" (MVar free) 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'. --

1490--------------------------------------------------------------------------------

1476--------------------------------------------------------------------------------

1491

1477

1492−-- | Request a connection to a remote endpoint

1478+-- | Establish 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.

1497--

1479--

1498−-- May throw a TransportError ConnectErrorCode exception.

1480+-- Maybe throw a TransportError

1499−uRequestConnectionTo :: TCPParameters

1481+socketToEndPoint :: EndPointAddress -- ^ Our address

1500− -> LocalEndPoint

1482+ -> EndPointAddress -- ^ Their address

1501− -> EndPointAddress

1483+ -> Bool -- ^ Use SO_REUSEADDR?

1502− -> ConnectHints

1484+ -> Maybe Int -- ^ Timeout for connect

1503− -> IO (RemoteEndPoint, ConnectionId)

1485+ -> IO (Either (TransportError ConnectErrorCode)

1504−uRequestConnectionTo params ourEndPoint theirAddress hints = go

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)

1505 where

1506 where

1506− go = do

1507+ createSocket :: N.AddrInfo -> IO N.Socket

1507− (theirEndPoint, isNew) <- mapIOException connectFailed $

1508+ createSocket addr = mapIOException insufficientResources $

1508− findRemoteEndPoint ourEndPoint theirAddress RequestedByUs

1509+ N.socket (N.addrFamily addr) N.Stream N.defaultProtocol

1509

1510

1510− if isNew

1511+ invalidAddress = TransportError ConnectNotFound . show

1511− then do

1512+ insufficientResources = TransportError ConnectInsufficientResources . show

1512− forkIO . handle absorbAllExceptions $

1513+ failed = TransportError ConnectFailed . show

1513− uSetupRemoteEndPoint params (ourEndPoint, theirEndPoint) hints

1514+ timeoutError = TransportError ConnectTimeout "Timed out"

1514− go

1515− else do

1516− reply <- mapIOException connectFailed $

1517− doRemoteRequest (ourEndPoint, theirEndPoint) RequestConnectionId

1518− return (theirEndPoint, decodeInt32 . BS.concat $ reply)

1519

1515

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

1521

1523

1522− absorbAllExceptions :: SomeException -> IO ()

1524+-- | Decode end point address

1523− absorbAllExceptions _ex =

1525+decodeEndPointAddress :: EndPointAddress

1524− return ()

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

1525

1535

1526−-- | Set up a remote endpoint

1536+--------------------------------------------------------------------------------

1527−uSetupRemoteEndPoint :: TCPParameters -> EndPointPair -> ConnectHints -> IO ()

1537+-- Functions from TransportInternals --

1528−uSetupRemoteEndPoint params (ourEndPoint, theirEndPoint) hints = do

1538+--------------------------------------------------------------------------------

1529− result <- socketToEndPoint ourAddress

1539+

1530− theirAddress

1540+-- Find a socket between two endpoints

1531− (tcpReuseClientAddr params)

1541+--

1532− (connectTimeout hints)

1542+-- Throws an IO exception if the socket could not be found.

1533− didAccept <- case result of

1543+internalSocketBetween :: TCPTransport -- ^ Transport

1534− Right (sock, ConnectionRequestAccepted) -> do

1544+ -> EndPointAddress -- ^ Local endpoint

1535− let vst = ValidRemoteEndPointState

1545+ -> EndPointAddress -- ^ Remote endpoint

1536− { remoteSocket = sock

1546+ -> IO N.Socket

1537− , _remoteOutgoing = 0

1547+internalSocketBetween transport ourAddress theirAddress = do

1538− , _remoteIncoming = IntSet.empty

1548+ ourEndPoint <- withMVar (transportState transport) $ \st -> case st of

1539− , sendOn = sendMany sock

1549+ TransportClosed ->

1540− , _pendingCtrlRequests = IntMap.empty

1550+ throwIO $ userError "Transport closed"

1541− , _nextCtrlRequestId = 0

1551+ TransportValid vst ->

1542− }

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"

1543− resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst)

1565+ RemoteEndPointValid vst ->

1544− return True

1566+ return $ remoteSocket vst

1545− Right (sock, ConnectionRequestInvalid) -> do

1567+ RemoteEndPointClosing _ vst ->

1546− let err = invalidAddress "setupRemoteEndPoint: Invalid endpoint"

1568+ return $ remoteSocket vst

1547− resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointInvalid err)

1548− tryCloseSocket sock

1549− return False

1550− Right (sock, ConnectionRequestCrossed) -> do

1551− resolveInit (ourEndPoint, theirEndPoint) RemoteEndPointClosed

1569+ RemoteEndPointClosed ->

1552− tryCloseSocket sock

1570+ throwIO $ userError "Remote endpoint closed"

1553− return False

1571+ RemoteEndPointInvalid err ->

1554− Left err -> do

1572+ throwIO err

1555− resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointInvalid err)

1573+ RemoteEndPointFailed err ->

1556− return False

1574+ throwIO err

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

1569

1575

1570--------------------------------------------------------------------------------

1576--------------------------------------------------------------------------------

1571-- Constants --

1577-- Constants --