9ad8f79 · distributed-process

473 -> Reliability -- ^ Reliability (ignored)

474 -> ConnectHints -- ^ Hints (ignored for now)

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

476−apiConnect ourEndPoint theirAddress _reliability hints = do

476+apiConnect ourEndPoint theirAddress _reliability hints =

477 if localAddress ourEndPoint == theirAddress

478 then connectToSelf ourEndPoint

479 else try $ do

562 where

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

564 tryCloseRemoteSocket :: RemoteEndPoint -> IO ()

565− tryCloseRemoteSocket theirEndPoint =

565+ tryCloseRemoteSocket theirEndPoint = do

566 -- We make an attempt to close the connection nicely

567 -- (by sending a CloseSocket first)

568+ let closed = RemoteEndPointClosed .Just . userError $ "apiCloseEndPoint"

568 modifyMVar_ (remoteState theirEndPoint) $ \st ->

569 modifyMVar_ (remoteState theirEndPoint) $ \st ->

569 case st of

570 case st of

570 RemoteEndPointInvalid _ ->

571 RemoteEndPointInvalid _ ->

571 return st

572 return st

572 RemoteEndPointInit resolved _ -> do

573 RemoteEndPointInit resolved _ -> do

573 putMVar resolved ()

574 putMVar resolved ()

574− return $ RemoteEndPointClosed (Just . userError $ "apiCloseEndPoint")

575+ return closed

575 RemoteEndPointValid conn -> do

576 RemoteEndPointValid conn -> do

576 tryIO $ sendOn conn [encodeInt32 CloseSocket]

577 tryIO $ sendOn conn [encodeInt32 CloseSocket]

577 tryCloseSocket (remoteSocket conn)

578 tryCloseSocket (remoteSocket conn)

578− return $ RemoteEndPointClosed (Just . userError $ "apiCloseEndPoint")

579+ return closed

579 RemoteEndPointClosing resolved conn -> do

580 RemoteEndPointClosing resolved conn -> do

580 putMVar resolved ()

581 putMVar resolved ()

581 tryCloseSocket (remoteSocket conn)

582 tryCloseSocket (remoteSocket conn)

582− return $ RemoteEndPointClosed (Just . userError $ "apiCloseEndPoint")

583+ return closed

583 RemoteEndPointClosed err ->

584 RemoteEndPointClosed err ->

584 return $ RemoteEndPointClosed err

585 return $ RemoteEndPointClosed err

585

586

760setupRemoteEndPoint :: EndPointPair -> ConnectHints -> IO ()

761setupRemoteEndPoint :: EndPointPair -> ConnectHints -> IO ()

761setupRemoteEndPoint (ourEndPoint, theirEndPoint) hints = do

762setupRemoteEndPoint (ourEndPoint, theirEndPoint) hints = do

762 didAccept <- bracketOnError (socketToEndPoint ourAddress theirAddress hints)

763 didAccept <- bracketOnError (socketToEndPoint ourAddress theirAddress hints)

763− onError $ \result ->

764+ onError $ \result -> case result of

764− case result of

765− Right (sock, ConnectionRequestAccepted) -> do

765+ Right (sock, ConnectionRequestAccepted) -> do

766− let vst = ValidRemoteEndPointState

766+ let vst = ValidRemoteEndPointState

767− { remoteSocket = sock

767+ { remoteSocket = sock

768− , _remoteOutgoing = 0

768+ , _remoteOutgoing = 0

769− , _remoteIncoming = IntSet.empty

769+ , _remoteIncoming = IntSet.empty

770− , sendOn = sendMany sock

770+ , sendOn = sendMany sock

771− , _pendingCtrlRequests = IntMap.empty

771+ , _pendingCtrlRequests = IntMap.empty

772− , _nextCtrlRequestId = 0

772+ , _nextCtrlRequestId = 0

773− }

773+ }

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

774+ resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst)

775− return True

775+ return True

776− Right (sock, ConnectionRequestInvalid) -> do

776+ Right (sock, ConnectionRequestInvalid) -> do

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

777+ let err = invalidAddress "setupRemoteEndPoint: Invalid endpoint"

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

778+ resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointInvalid err)

779− tryCloseSocket sock

779+ tryCloseSocket sock

780− return False

780+ return False

781− Right (sock, ConnectionRequestCrossed) -> do

781+ Right (sock, ConnectionRequestCrossed) -> do

782− -- We leave the endpoint in Init state, handleConnectionRequest will

782+ -- We leave the endpoint in Init state, handleConnectionRequest will

783− -- take care of it

783+ -- take care of it

784− resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointClosed Nothing)

784+ resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointClosed Nothing)

785− tryCloseSocket sock

785+ tryCloseSocket sock

786− return False

786+ return False

787− Left err -> do

787+ Left err -> do

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

788+ resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointInvalid err)

789− return False

789+ return False

790

791 -- If we get to this point without an exception, then

792 --

794 -- if didAccept is True, the socket has been stored as part of the remote

795 -- state so we no longer need to worry about closing it when an

796 -- asynchronous exception occurs

797− when didAccept $ handleIncomingMessages (ourEndPoint, theirEndPoint)

797+ when didAccept $ handleIncomingMessages (ourEndPoint, theirEndPoint)

798 where

799 -- If an asynchronous exception occurs while we set up the remote endpoint

800 -- we need to make sure to close the socket. It is also useful to

857 timeoutMaybe (connectTimeout hints) timeoutError $

858 N.connect sock (N.addrAddress addr)

859 response <- mapIOException failed $ do

860− sendMany sock (encodeInt32 theirEndPointId : prependLength [ourAddress])

860+ sendMany sock (encodeInt32 theirEndPointId : prependLength [ourAddress])

861 recvInt32 sock

862 case tryToEnum response of

863 Nothing -> throwIO (failed . userError $ "Unexpected response")

947 $ vst

948 )

949 -- Error cases

950− , caseInvalid = throwIO

950+ , caseInvalid =

951+ throwIO

952+ , caseInit = \_ _ ->

951− , caseInit = \_ _ -> relyViolation (ourEndPoint, theirEndPoint) "doRemoteRequest (init)"

953+ relyViolation (ourEndPoint, theirEndPoint) "doRemoteRequest (init)"

954+ , caseClosing = \_ _ ->

952− , caseClosing = \_ _ -> relyViolation (ourEndPoint, theirEndPoint) "doRemoteRequest (closing)"

955+ relyViolation (ourEndPoint, theirEndPoint) "doRemoteRequest (closing)"

956+ , caseClosed = \_ ->

953− , caseClosed = \_ -> throwIO $ userError "doRemoteRequest: Remote endpoint closed"

957+ throwIO $ userError "doRemoteRequest: Remote endpoint closed"

954 }

958 }

955 mReply <- takeMVar replyMVar

959 mReply <- takeMVar replyMVar

956 case mReply of

960 case mReply of

1014 ourState = localState ourEndPoint

1018 ourState = localState ourEndPoint

1015

1019

1016--------------------------------------------------------------------------------

1020--------------------------------------------------------------------------------

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

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

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

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

1019-- provide some abstractions. --

1023-- provide some abstractions. --

1020--------------------------------------------------------------------------------

1024--------------------------------------------------------------------------------

1296 mSock <- withMVar theirState $ \st ->

1300 mSock <- withMVar theirState $ \st ->

1297 case st of

1301 case st of

1298 RemoteEndPointInvalid _ ->

1302 RemoteEndPointInvalid _ ->

1299− relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages (invalid)"

1303+ relyViolation (ourEndPoint, theirEndPoint)

1304+ "handleIncomingMessages (invalid)"

1300 RemoteEndPointInit _ _ ->

1305 RemoteEndPointInit _ _ ->

1301− relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages (init)"

1306+ relyViolation (ourEndPoint, theirEndPoint)

1307+ "handleIncomingMessages (init)"

1302 RemoteEndPointValid ep ->

1308 RemoteEndPointValid ep ->

1303 return . Just $ remoteSocket ep

1309 return . Just $ remoteSocket ep

1304 RemoteEndPointClosing _ ep ->

1310 RemoteEndPointClosing _ ep ->

1505 modifyMVar_ theirState $ \st ->

1511 modifyMVar_ theirState $ \st ->

1506 case st of

1512 case st of

1507 RemoteEndPointInvalid _ ->

1513 RemoteEndPointInvalid _ ->

1508− relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:prematureExit"

1514+ relyViolation (ourEndPoint, theirEndPoint)

1515+ "handleIncomingMessages:prematureExit"

1509 RemoteEndPointInit _ _ ->

1516 RemoteEndPointInit _ _ ->

1510− relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:prematureExit"

1517+ relyViolation (ourEndPoint, theirEndPoint)

1518+ "handleIncomingMessages:prematureExit"

1511 RemoteEndPointValid vst -> do

1519 RemoteEndPointValid vst -> do

1512 let code = EventConnectionLost

1520 let code = EventConnectionLost

1513 (remoteAddress theirEndPoint)

1521 (remoteAddress theirEndPoint)