553015e · distributed-process

git.iko.soy/distributed-process

on master help sign in

Better error handling when self-connecting

553015eEdsko de Vries14 years ago

network-transport/src/Network/Transport/TCP.hs+40−39

compare with master

unifiedsplit

@@ -259,10 +259,10 @@ data ValidLocalEndPointState = ValidLocalEndPointState

259-- | /----------------\

260-- | | |

261-- | v |

262−-- \---> Valid ----+---> Closing

262+-- +---> Valid ----+---> Closing

263−-- | |

263+-- | | |

264−-- | v

264+-- | | v

265−-- \---> Closed

265+-- \---------------+---> Closed

266--

267-- Init: There are two places where we create new remote endpoints: in

268-- requestConnectionTo (in response to an API 'connect' call) and in

@@ -565,16 +565,15 @@ apiCloseEndPoint transport evs ourEndPoint = do

565 case st of

566 RemoteEndPointInvalid _ ->

567 return st

568− RemoteEndPointInit _ ->

568+ RemoteEndPointInit resolved -> do

569− return st

569+ putMVar resolved ()

570+ return RemoteEndPointClosed

570 RemoteEndPointValid conn -> do

571 RemoteEndPointValid conn -> do

571− -- Try to send a CloseSocket request

572 tryIO $ sendOn conn [encodeInt32 CloseSocket]

573− -- .. but even if it fails, close the socket anyway

574 tryCloseSocket (remoteSocket conn)

573 tryCloseSocket (remoteSocket conn)

575 return RemoteEndPointClosed

574 return RemoteEndPointClosed

576− RemoteEndPointClosing _ conn -> do

575+ RemoteEndPointClosing resolved conn -> do

577− -- TODO: should we signal on resolved here?

576+ putMVar resolved ()

578 tryCloseSocket (remoteSocket conn)

577 tryCloseSocket (remoteSocket conn)

579 return RemoteEndPointClosed

578 return RemoteEndPointClosed

580 RemoteEndPointClosed ->

579 RemoteEndPointClosed ->

@@ -583,42 +582,44 @@ apiCloseEndPoint transport evs ourEndPoint = do

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

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

584connectToSelf :: LocalEndPoint

583connectToSelf :: LocalEndPoint

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

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

586−connectToSelf ourEndPoint = do

585+connectToSelf ourEndPoint = try . mapIOException connectFailed $ do

587− -- Here connAlive must an MVar because it is not protected by another lock

586+ connAlive <- newIORef True -- Protected by the local endpoint lock

588− connAlive <- newMVar True

589− mConnId <- tryIO (getNextConnectionId ourEndPoint)

587+ connId <- getNextConnectionId ourEndPoint

590− case mConnId of

591− Left err ->

592− return . Left $ TransportError ConnectNotFound (show err)

593− Right connId -> do

594− writeChan ourChan $

588+ writeChan ourChan $

595− ConnectionOpened connId ReliableOrdered (localAddress ourEndPoint)

589+ ConnectionOpened connId ReliableOrdered (localAddress ourEndPoint)

596− return . Right $ Connection

590+ return Connection

597− { send = selfSend connAlive connId

591+ { send = selfSend connAlive connId

598− , close = selfClose connAlive connId

592+ , close = selfClose connAlive connId

599− }

593+ }

600 where

594 where

601− ourChan :: Chan Event

602− ourChan = localChannel ourEndPoint

603−

604− selfSend :: MVar Bool

595+ selfSend :: IORef Bool

605 -> ConnectionId

596 -> ConnectionId

606 -> [ByteString]

597 -> [ByteString]

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

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

608 selfSend connAlive connId msg =

599 selfSend connAlive connId msg =

609− modifyMVar connAlive $ \alive ->

600+ try . withMVar ourState $ \st -> case st of

601+ LocalEndPointValid _ -> do

602+ alive <- readIORef connAlive

610− if alive

603+ if alive

611− then do

612− writeChan ourChan (Received connId msg)

604+ then writeChan ourChan (Received connId msg)

613− return (alive, Right ())

614− else

615− return (alive, Left (TransportError SendFailed "Connection closed"))

605+ else throwIO $ TransportError SendClosed "Connection closed"

606+ LocalEndPointClosed ->

607+ throwIO $ TransportError SendFailed "Endpoint closed"

616

608

617− selfClose :: MVar Bool -> ConnectionId -> IO ()

609+ selfClose :: IORef Bool -> ConnectionId -> IO ()

618 selfClose connAlive connId =

610 selfClose connAlive connId =

619− modifyMVar_ connAlive $ \alive -> do

611+ withMVar ourState $ \st -> case st of

612+ LocalEndPointValid _ -> do

613+ alive <- readIORef connAlive

614+ when alive $ do

620− when alive $ writeChan ourChan (ConnectionClosed connId)

615+ writeChan ourChan (ConnectionClosed connId)

621− return False

616+ writeIORef connAlive False

617+ LocalEndPointClosed ->

618+ return ()

619+

620+ ourChan = localChannel ourEndPoint

621+ ourState = localState ourEndPoint

622+ connectFailed = TransportError ConnectFailed . show

622

623

623--------------------------------------------------------------------------------

624--------------------------------------------------------------------------------

624-- Functions from TransportInternals --

625-- Functions from TransportInternals --

network-transport/tests/TestTCP.hs−1

@@ -754,4 +754,3 @@ main = do

754 , ("InvalidCloseConnection", testInvalidCloseConnection nextPort)

755 ]

756 testTransport (either (Left . show) (Right) <$> nextPort >>= createTransport "127.0.0.1")

757− return ()

network-transport/tests/TestTransport.hs+31−5

@@ -533,6 +533,36 @@ testConnectToSelfTwice transport numPings = do

533

534 takeMVar done

535

536+-- | Test that we self-connections no longer work once we close our endpoint

537+-- or our transport

538+testCloseSelf :: IO (Either String Transport) -> IO ()

539+testCloseSelf newTransport = do

540+ Right transport <- newTransport

541+ Right endpoint1 <- newEndPoint transport

542+ Right endpoint2 <- newEndPoint transport

543+ Right conn1 <- connect endpoint1 (address endpoint1) ReliableOrdered

544+ Right conn2 <- connect endpoint1 (address endpoint1) ReliableOrdered

545+ Right conn3 <- connect endpoint2 (address endpoint2) ReliableOrdered

546+

547+ -- Close the conneciton and try to send

548+ close conn1

549+ Left (TransportError SendClosed _) <- send conn1 ["ping"]

550+

551+ -- Close the first endpoint. We should not be able to use the first

552+ -- connection anymore, or open more self connections, but the self connection

553+ -- to the second endpoint should still be fine

554+ closeEndPoint endpoint1

555+ Left (TransportError SendFailed _) <- send conn2 ["ping"]

556+ Left (TransportError ConnectFailed _) <- connect endpoint1 (address endpoint1) ReliableOrdered

557+ Right () <- send conn3 ["ping"]

558+

559+ -- Close the transport; now the second should no longer work

560+ closeTransport transport

561+ Left (TransportError SendFailed _) <- send conn3 ["ping"]

562+ Left (TransportError ConnectFailed _) <- connect endpoint2 (address endpoint2) ReliableOrdered

563+

564+ return ()

565+

536-- | Test various aspects of 'closeEndPoint'

566-- | Test various aspects of 'closeEndPoint'

537testCloseEndPoint :: Transport -> Int -> IO ()

567testCloseEndPoint :: Transport -> Int -> IO ()

538testCloseEndPoint transport _ = do

568testCloseEndPoint transport _ = do

@@ -726,13 +756,8 @@ testConnectClosedEndPoint transport = do

726 Right endpoint <- newEndPoint transport

756 Right endpoint <- newEndPoint transport

727 readMVar serverClosed

757 readMVar serverClosed

728

758

729− -- Connect to a remote closed endpoint

730 Left (TransportError ConnectNotFound _) <- readMVar serverAddr >>= \addr -> connect endpoint addr ReliableOrdered

759 Left (TransportError ConnectNotFound _) <- readMVar serverAddr >>= \addr -> connect endpoint addr ReliableOrdered

731

760

732− -- Self-connect to a closed endpoint

733− closeEndPoint endpoint

734− Left (TransportError ConnectNotFound _) <- connect endpoint (address endpoint) ReliableOrdered

735−

736 putMVar clientDone ()

761 putMVar clientDone ()

737

762

738 takeMVar clientDone

763 takeMVar clientDone

@@ -850,6 +875,7 @@ testTransport newTransport = do

850 , ("CloseTwice", testCloseTwice transport 100)

875 , ("CloseTwice", testCloseTwice transport 100)

851 , ("ConnectToSelf", testConnectToSelf transport numPings)

876 , ("ConnectToSelf", testConnectToSelf transport numPings)

852 , ("ConnectToSelfTwice", testConnectToSelfTwice transport numPings)

877 , ("ConnectToSelfTwice", testConnectToSelfTwice transport numPings)

878+ , ("CloseSelf", testCloseSelf newTransport)

853 , ("CloseEndPoint", testCloseEndPoint transport numPings)

879 , ("CloseEndPoint", testCloseEndPoint transport numPings)

854 , ("CloseTransport", testCloseTransport newTransport)

880 , ("CloseTransport", testCloseTransport newTransport)

855 , ("ConnectClosedEndPoint", testConnectClosedEndPoint transport)

881 , ("ConnectClosedEndPoint", testConnectClosedEndPoint transport)