e9af460 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Don't reuse endpoint IDs. More error handling.

e9af460Edsko de Vries14 years ago

network-transport/src/Network/Transport/TCP.hs+29−17

compare with master

unifiedsplit

@@ -85,7 +85,7 @@ import qualified Data.IntMap as IntMap (empty)

85import Data.IntSet (IntSet)

86import qualified Data.IntSet as IntSet (empty, insert, elems, singleton, null, delete)

87import Data.Map (Map)

88−import qualified Data.Map as Map (empty, elems, size)

88+import qualified Data.Map as Map (empty, elems)

89import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))

90import qualified Data.Accessor.Container as DAC (mapMaybe, intMapMaybe)

91import System.IO (hPutStrLn, stderr)

@@ -209,6 +209,7 @@ data TransportState =

209

210data ValidTransportState = ValidTransportState

211 { _localEndPoints :: Map EndPointAddress LocalEndPoint

212+ , _nextEndPointId :: EndPointId

212 }

213 }

213

214

214data LocalEndPoint = LocalEndPoint

215data LocalEndPoint = LocalEndPoint

@@ -330,7 +331,10 @@ createTransport host port = do

330-- | You should probably not use this function (used for unit testing only)

331-- | You should probably not use this function (used for unit testing only)

331createTransportExposeInternals :: N.HostName -> N.ServiceName -> IO (Either IOException (Transport, ThreadId))

332createTransportExposeInternals :: N.HostName -> N.ServiceName -> IO (Either IOException (Transport, ThreadId))

332createTransportExposeInternals host port = do

333createTransportExposeInternals host port = do

333− state <- newMVar . TransportValid $ ValidTransportState { _localEndPoints = Map.empty }

334+ state <- newMVar . TransportValid $ ValidTransportState

335+ { _localEndPoints = Map.empty

336+ , _nextEndPointId = 0

337+ }

334 let transport = TCPTransport { transportState = state

338 let transport = TCPTransport { transportState = state

335 , transportHost = host

339 , transportHost = host

336 , transportPort = port

340 , transportPort = port

@@ -378,7 +382,6 @@ apiCloseTransport transport mTransportThread evs = do

378 -- transport thread died (in which case we don't need to kill it)

382 -- transport thread died (in which case we don't need to kill it)

379 return ()

383 return ()

380

384

381−

382-- | Create a new endpoint

385-- | Create a new endpoint

383apiNewEndPoint :: TCPTransport -> IO (Either (TransportError NewEndPointErrorCode) EndPoint)

386apiNewEndPoint :: TCPTransport -> IO (Either (TransportError NewEndPointErrorCode) EndPoint)

384apiNewEndPoint transport = try $ do

387apiNewEndPoint transport = try $ do

@@ -418,8 +421,6 @@ apiConnect ourEndPoint theirAddress _reliability _hints = try $ do

418-- GUARANTEE: If the connection is alive on entry then the remote endpoint will

421-- GUARANTEE: If the connection is alive on entry then the remote endpoint will

419-- either be RemoteEndPointValid or RemoteEndPointClosing. Otherwise, the state

422-- either be RemoteEndPointValid or RemoteEndPointClosing. Otherwise, the state

420-- of the remote endpoint will not be changed.

423-- of the remote endpoint will not be changed.

421−--

422−-- TODO: We ignore errors during a close. Is that right?

423apiClose :: RemoteEndPoint -> ConnectionId -> IORef Bool -> IO ()

424apiClose :: RemoteEndPoint -> ConnectionId -> IORef Bool -> IO ()

424apiClose theirEndPoint connId connAlive = void . tryIO $ do

425apiClose theirEndPoint connId connAlive = void . tryIO $ do

425 modifyMVar_ (remoteState theirEndPoint) $ \st -> case st of

426 modifyMVar_ (remoteState theirEndPoint) $ \st -> case st of

@@ -428,6 +429,7 @@ apiClose theirEndPoint connId connAlive = void . tryIO $ do

428 if alive

429 if alive

429 then do

430 then do

430 writeIORef connAlive False

431 writeIORef connAlive False

432+ -- Possible exception caught top-level

431 sendOn vst [encodeInt32 CloseConnection, encodeInt32 connId]

433 sendOn vst [encodeInt32 CloseConnection, encodeInt32 connId]

432 return (RemoteEndPointValid . (remoteOutgoing ^: (\x -> x - 1)) $ vst)

434 return (RemoteEndPointValid . (remoteOutgoing ^: (\x -> x - 1)) $ vst)

433 else

435 else

@@ -454,7 +456,7 @@ apiSend theirEndPoint connId connAlive payload = do

454 then mapExceptionIO sendFailed $ sendOn vst (encodeInt32 connId : prependLength payload)

456 then mapExceptionIO sendFailed $ sendOn vst (encodeInt32 connId : prependLength payload)

455 else throw $ TransportError SendClosed "Connection closed"

457 else throw $ TransportError SendClosed "Connection closed"

456 RemoteEndPointClosed -> do

458 RemoteEndPointClosed -> do

457− return (Left $ TransportError SendClosed "Connection lost")

459+ return . Left $ TransportError SendClosed "Connection lost"

458 RemoteEndPointClosing _ _ ->

460 RemoteEndPointClosing _ _ ->

459 -- The only way for the endpoint to be in closing state, while a

461 -- The only way for the endpoint to be in closing state, while a

460 -- connection is still active, is for the application to call 'close'

462 -- connection is still active, is for the application to call 'close'

@@ -463,7 +465,7 @@ apiSend theirEndPoint connId connAlive payload = do

463 -- yet. Even if the remote endpoint comes back with "please don't

465 -- yet. Even if the remote endpoint comes back with "please don't

464 -- close", this still means that our *outgoing* connection has been

466 -- close", this still means that our *outgoing* connection has been

465 -- closed

467 -- closed

466− return (Left $ TransportError SendClosed "Connection lost")

468+ return . Left $ TransportError SendClosed "Connection lost"

467 RemoteEndPointInvalid _ ->

469 RemoteEndPointInvalid _ ->

468 error "apiSend RELY violation"

470 error "apiSend RELY violation"

469 where

471 where

@@ -520,12 +522,15 @@ connectToSelf :: LocalEndPoint -> IO (Either (TransportError ConnectErrorCode) C

520connectToSelf ourEndPoint = do

522connectToSelf ourEndPoint = do

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

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

522 connAlive <- newMVar True

524 connAlive <- newMVar True

523− -- TODO: catch exception

524− connId <- getNextConnectionId ourEndPoint

525+ mConnId <- tryIO (getNextConnectionId ourEndPoint)

526+ case mConnId of

527+ Left err ->

528+ return . Left $ TransportError ConnectNotFound (show err)

529+ Right connId -> do

525− writeChan ourChan (ConnectionOpened connId ReliableOrdered (localAddress ourEndPoint))

530+ writeChan ourChan (ConnectionOpened connId ReliableOrdered (localAddress ourEndPoint))

526− return . Right $ Connection { send = selfSend connAlive connId

531+ return . Right $ Connection { send = selfSend connAlive connId

527− , close = selfClose connAlive connId

532+ , close = selfClose connAlive connId

528− }

533+ }

529 where

534 where

530 ourChan :: Chan Event

535 ourChan :: Chan Event

531 ourChan = localChannel ourEndPoint

536 ourChan = localChannel ourEndPoint

@@ -565,13 +570,13 @@ createLocalEndPoint transport = do

565 }

570 }

566 modifyMVar (transportState transport) $ \st -> case st of

571 modifyMVar (transportState transport) $ \st -> case st of

567 TransportValid vst -> do

572 TransportValid vst -> do

568− let ix = fromIntegral $ Map.size (vst ^. localEndPoints)

573+ let ix = vst ^. nextEndPointId

569 let addr = encodeEndPointAddress (transportHost transport) (transportPort transport) ix

574 let addr = encodeEndPointAddress (transportHost transport) (transportPort transport) ix

570 let localEndPoint = LocalEndPoint { localAddress = addr

575 let localEndPoint = LocalEndPoint { localAddress = addr

571 , localChannel = chan

576 , localChannel = chan

572 , localState = state

577 , localState = state

573 }

578 }

574− return (TransportValid . (localEndPointAt addr ^= Just localEndPoint) $ vst, localEndPoint)

579+ return (TransportValid . (localEndPointAt addr ^= Just localEndPoint) . (nextEndPointId ^= ix + 1) $ vst, localEndPoint)

575 TransportClosed ->

580 TransportClosed ->

576 throw (TransportError NewEndPointFailed "Transport closed")

581 throw (TransportError NewEndPointFailed "Transport closed")

577

582

@@ -644,8 +649,10 @@ requestConnectionTo ourEndPoint theirAddress = go

644 -- point)

649 -- point)

645 reply <- handle failureHandler $ doRemoteRequest (ourEndPoint, theirEndPoint) RequestConnectionId

650 reply <- handle failureHandler $ doRemoteRequest (ourEndPoint, theirEndPoint) RequestConnectionId

646 case decodeInt32 . BS.concat $ reply of

651 case decodeInt32 . BS.concat $ reply of

652+ Nothing ->

647− Nothing -> failureHandler (userError "Invalid integer")

653+ failureHandler (userError "Invalid integer")

654+ Just cid -> do

648− Just cid -> return (theirEndPoint, cid)

655+ return (theirEndPoint, cid)

649

656

650 -- If this is a new endpoint, fork a thread to listen for incoming

657 -- If this is a new endpoint, fork a thread to listen for incoming

651 -- connections. We don't want to do this while we hold the lock, because

658 -- connections. We don't want to do this while we hold the lock, because

@@ -979,6 +986,8 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do

979 -- Create a new connection

986 -- Create a new connection

980 createNewConnection :: ControlRequestId -> IO ()

987 createNewConnection :: ControlRequestId -> IO ()

981 createNewConnection reqId = do

988 createNewConnection reqId = do

989+ -- getNextConnectionId throws an exception if ourEndPoint is closed; but

990+ -- if this endpoint is closed, this thread will soon die anyway

982 newId <- getNextConnectionId ourEndPoint

991 newId <- getNextConnectionId ourEndPoint

983 modifyMVar_ theirState $ \st -> do

992 modifyMVar_ theirState $ \st -> do

984 vst <- case st of

993 vst <- case st of

@@ -1135,6 +1144,9 @@ firstNonReservedConnectionId = 1024

1135localEndPoints :: Accessor ValidTransportState (Map EndPointAddress LocalEndPoint)

1144localEndPoints :: Accessor ValidTransportState (Map EndPointAddress LocalEndPoint)

1136localEndPoints = accessor _localEndPoints (\es st -> st { _localEndPoints = es })

1145localEndPoints = accessor _localEndPoints (\es st -> st { _localEndPoints = es })

1137

1146

1147+nextEndPointId :: Accessor ValidTransportState EndPointId

1148+nextEndPointId = accessor _nextEndPointId (\eid st -> st { _nextEndPointId = eid })

1149+

1138pendingCtrlRequests :: Accessor ValidLocalEndPointState (IntMap (MVar [ByteString]))

1150pendingCtrlRequests :: Accessor ValidLocalEndPointState (IntMap (MVar [ByteString]))

1139pendingCtrlRequests = accessor _pendingCtrlRequests (\rep st -> st { _pendingCtrlRequests = rep })

1151pendingCtrlRequests = accessor _pendingCtrlRequests (\rep st -> st { _pendingCtrlRequests = rep })

1140

1152

network-transport/tests/TestTransport.hs+46−13

@@ -16,6 +16,7 @@ import Data.Map (Map)

16import qualified Data.Map as Map (empty, insert, delete, findWithDefault, adjust, null, toList, map)

17import Data.String (fromString)

18import Traced

19+import Control.Concurrent (threadDelay)

20-- | We overload connect to always pass the default hints

21-- | We overload connect to always pass the default hints

21connect :: EndPoint -> EndPointAddress -> Reliability -> IO (Either (TransportError ConnectErrorCode) Connection)

22connect :: EndPoint -> EndPointAddress -> Reliability -> IO (Either (TransportError ConnectErrorCode) Connection)

@@ -702,25 +703,57 @@ testCloseTransport newTransport = do

702 putMVar clientDone ()

703 putMVar clientDone ()

703

704

704 mapM_ takeMVar [serverDone, clientDone]

705 mapM_ takeMVar [serverDone, clientDone]

706+

707+-- | Remote node attempts to connect to a closed local endpoint

708+testConnectClosedEndPoint :: Transport -> IO ()

709+testConnectClosedEndPoint transport = do

710+ serverAddr <- newEmptyMVar

711+ serverClosed <- newEmptyMVar

712+ clientDone <- newEmptyMVar

705

713

714+ -- Server

715+ forkTry $ do

716+ Right endpoint <- newEndPoint transport

717+ putMVar serverAddr (address endpoint)

718+

719+ closeEndPoint endpoint

720+ putMVar serverClosed ()

721+

722+ -- Client

723+ forkTry $ do

724+ Right endpoint <- newEndPoint transport

725+ readMVar serverClosed

726+

727+ -- Connect to a remote closed endpoint

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

729+

730+ -- Self-connect to a closed endpoint

731+ closeEndPoint endpoint

732+ Left (TransportError ConnectNotFound _) <- connect endpoint (address endpoint) ReliableOrdered

733+

734+ putMVar clientDone ()

735+

736+ takeMVar clientDone

737+

706-- Transport tests

738-- Transport tests

707testTransport :: IO (Either String Transport) -> IO ()

739testTransport :: IO (Either String Transport) -> IO ()

708testTransport newTransport = do

740testTransport newTransport = do

709 Right transport <- newTransport

741 Right transport <- newTransport

710 runTests

742 runTests

711− [ ("PingPong", testPingPong transport numPings)

743+ [ ("PingPong", testPingPong transport numPings)

712− , ("EndPoints", testEndPoints transport numPings)

744+ , ("EndPoints", testEndPoints transport numPings)

713− , ("Connections", testConnections transport numPings)

745+ , ("Connections", testConnections transport numPings)

714− , ("CloseOneConnection", testCloseOneConnection transport numPings)

746+ , ("CloseOneConnection", testCloseOneConnection transport numPings)

715− , ("CloseOneDirection", testCloseOneDirection transport numPings)

747+ , ("CloseOneDirection", testCloseOneDirection transport numPings)

716− , ("CloseReopen", testCloseReopen transport numPings)

748+ , ("CloseReopen", testCloseReopen transport numPings)

717− , ("ParallelConnects", testParallelConnects transport numPings)

749+ , ("ParallelConnects", testParallelConnects transport numPings)

718− , ("SendAfterClose", testSendAfterClose transport 100)

750+ , ("SendAfterClose", testSendAfterClose transport 100)

719− , ("CloseTwice", testCloseTwice transport 100)

751+ , ("CloseTwice", testCloseTwice transport 100)

720− , ("ConnectToSelf", testConnectToSelf transport numPings)

752+ , ("ConnectToSelf", testConnectToSelf transport numPings)

721− , ("ConnectToSelfTwice", testConnectToSelfTwice transport numPings)

753+ , ("ConnectToSelfTwice", testConnectToSelfTwice transport numPings)

722− , ("CloseEndPoint", testCloseEndPoint transport numPings)

754+ , ("CloseEndPoint", testCloseEndPoint transport numPings)

723− , ("CloseTransport", testCloseTransport newTransport)

755+ , ("CloseTransport", testCloseTransport newTransport)

756+ , ("ConnectClosedEndPoint", testConnectClosedEndPoint transport)

724 ]

757 ]

725 where

758 where

726 numPings = 10000 :: Int

759 numPings = 10000 :: Int