72d1f65 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Distinguish between closeTransport and failure

72d1f65Edsko de Vries14 years ago

network-transport/src/Network/Transport/Internal/TCP.hs+16−12

compare with master

unifiedsplit

@@ -25,9 +25,9 @@ import qualified Network.Socket as N ( HostName

25 , sClose

26 )

27import qualified Network.Socket.ByteString as NBS (recv)

28−import Control.Concurrent (forkIO, ThreadId)

28+import Control.Concurrent (ThreadId, forkIOWithUnmask)

29import Control.Monad (liftM, forever)

30−import Control.Exception (SomeException, handle, bracketOnError, throw)

30+import Control.Exception (SomeException, catch, bracketOnError, throw, mask_)

31import Data.ByteString (ByteString)

32import qualified Data.ByteString as BS (length, concat, null)

33import Data.Int (Int32)

@@ -60,16 +60,20 @@ forkServer host port backlog terminationHandler requestHandler = do

60 -- return an empty list (but will throw an exception instead) and will return

61 -- the "best" address first, whatever that means

62 addr:_ <- N.getAddrInfo (Just N.defaultHints) (Just host) (Just port)

63− sock <- N.socket (N.addrFamily addr) N.Stream N.defaultProtocol

63+ bracketOnError (N.socket (N.addrFamily addr) N.Stream N.defaultProtocol)

64− -- putStrLn $ "Created server socket " ++ show sock ++ " for address " ++ host ++ ":" ++ port

64+ N.sClose $ \sock -> do

65− N.setSocketOption sock N.ReuseAddr 1

65+ N.setSocketOption sock N.ReuseAddr 1

66− N.bindSocket sock (N.addrAddress addr)

66+ N.bindSocket sock (N.addrAddress addr)

67− N.listen sock backlog

67+ N.listen sock backlog

68− -- TODO: when is this socket closed?

68+ mask_ $ forkIOWithUnmask $ \unmask ->

69+ catch (unmask (forever $ acceptRequest sock)) $ \ex -> do

70+ N.sClose sock

69− forkIO . handle terminationHandler . forever $

71+ terminationHandler ex

72+ where

73+ acceptRequest :: N.Socket -> IO ()

70− bracketOnError (N.accept sock)

74+ acceptRequest sock = bracketOnError (N.accept sock)

71− (N.sClose . fst)

75+ (N.sClose . fst)

72− (requestHandler . fst)

76+ (requestHandler . fst)

74-- | Read a length and then a payload of that length

78-- | Read a length and then a payload of that length

75recvWithLength :: N.Socket -> IO [ByteString]

79recvWithLength :: N.Socket -> IO [ByteString]

network-transport/src/Network/Transport/TCP.hs+52−24

@@ -15,8 +15,9 @@

15-- * Output exception on channel after endpoint is closed

16module Network.Transport.TCP ( -- * Main API

17 createTransport

18− , -- * TCP specific functionality (exported mostly for testing purposes_

18+ -- * Internals (exposed for unit tests)

19+ , createTransportExposeInternals

19− EndPointId

20+ , EndPointId

20 , encodeEndPointAddress

21 , encodeEndPointAddress

21 , decodeEndPointAddress

22 , decodeEndPointAddress

22 , ControlHeader(..)

23 , ControlHeader(..)

@@ -319,6 +320,16 @@ data ConnectionRequestResponse =

319-- TODOs: deal with hints

320-- TODOs: deal with hints

320createTransport :: N.HostName -> N.ServiceName -> IO (Either IOException Transport)

321createTransport :: N.HostName -> N.ServiceName -> IO (Either IOException Transport)

321createTransport host port = do

322createTransport host port = do

323+ result <- createTransportExposeInternals host port

324+ case result of

325+ Left err ->

326+ return $ Left err

327+ Right (transport, _) ->

328+ return $ Right transport

329+

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

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

332+createTransportExposeInternals host port = do

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

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

323 let transport = TCPTransport { transportState = state

334 let transport = TCPTransport { transportState = state

324 , transportHost = host

335 , transportHost = host

@@ -329,27 +340,45 @@ createTransport host port = do

329 -- http://tangentsoft.net/wskfaq/advanced.html#backlog

340 -- http://tangentsoft.net/wskfaq/advanced.html#backlog

330 -- http://www.linuxjournal.com/files/linuxjournal.com/linuxjournal/articles/023/2333/2333s2.html

341 -- http://www.linuxjournal.com/files/linuxjournal.com/linuxjournal/articles/023/2333/2333s2.html

331 transportThread <- forkServer host port N.sOMAXCONN (terminationHandler transport) (handleConnectionRequest transport)

342 transportThread <- forkServer host port N.sOMAXCONN (terminationHandler transport) (handleConnectionRequest transport)

332− return Transport { newEndPoint = apiNewEndPoint transport

343+ return ( Transport { newEndPoint = apiNewEndPoint transport

333− , closeTransport = killThread transportThread -- This will invoke the termination handler

344+ , closeTransport = apiCloseTransport transport (Just transportThread) [EndPointClosed]

334− }

345+ }

346+ , transportThread

347+ )

335 where

348 where

336 terminationHandler :: TCPTransport -> SomeException -> IO ()

349 terminationHandler :: TCPTransport -> SomeException -> IO ()

337− terminationHandler transport _ = do

350+ terminationHandler transport ex = do

338− -- TODO: we currently don't make a distinction between endpoint failure and manual closure

351+ let ev = ErrorEvent (TransportError EventTransportFailed (show ex))

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

352+ apiCloseTransport transport Nothing [ev]

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

341− TransportClosed -> return (TransportClosed, Nothing)

342− case mTSt of

343− Nothing ->

344− -- Transport already closed

345− return ()

346− Just tSt ->

347− mapM_ (apiCloseEndPoint transport) (Map.elems $ tSt ^. localEndPoints)

348

353

349--------------------------------------------------------------------------------

354--------------------------------------------------------------------------------

350-- API functions --

355-- API functions --

351--------------------------------------------------------------------------------

356--------------------------------------------------------------------------------

352

357

358+-- | Close the transport

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

360+apiCloseTransport transport mTransportThread evs = do

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

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

363+ TransportClosed -> return (TransportClosed, Nothing)

364+ case mTSt of

365+ Nothing ->

366+ -- Transport already closed

367+ return ()

368+ Just tSt -> do

369+ mapM_ (apiCloseEndPoint transport evs) (Map.elems $ tSt ^. localEndPoints)

370+ case mTransportThread of

371+ Just transportThread ->

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

373+ -- apiCloseTransport again, but at that point the transport is

374+ -- already in closed state so at that point we return immediately

375+ killThread transportThread

376+ Nothing ->

377+ -- This will happen is apiCloseTransport was invoked *because* the

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

379+ return ()

380+

381+

353-- | Create a new endpoint

382-- | Create a new endpoint

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

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

355apiNewEndPoint transport = try $ do

384apiNewEndPoint transport = try $ do

@@ -358,7 +387,7 @@ apiNewEndPoint transport = try $ do

358 { receive = readChan (localChannel ourEndPoint)

387 { receive = readChan (localChannel ourEndPoint)

359 , address = localAddress ourEndPoint

388 , address = localAddress ourEndPoint

360 , connect = apiConnect ourEndPoint

389 , connect = apiConnect ourEndPoint

361− , closeEndPoint = apiCloseEndPoint transport ourEndPoint

390+ , closeEndPoint = apiCloseEndPoint transport [EndPointClosed] ourEndPoint

362 , newMulticastGroup = return . Left $ newMulticastGroupError

391 , newMulticastGroup = return . Left $ newMulticastGroupError

363 , resolveMulticastGroup = return . Left . const resolveMulticastGroupError

392 , resolveMulticastGroup = return . Left . const resolveMulticastGroupError

364 }

393 }

@@ -442,8 +471,11 @@ apiSend theirEndPoint connId connAlive payload = do

442 sendFailed = TransportError SendFailed . show

471 sendFailed = TransportError SendFailed . show

443

472

444-- | Force-close the endpoint

473-- | Force-close the endpoint

445−apiCloseEndPoint :: TCPTransport -> LocalEndPoint -> IO ()

474+apiCloseEndPoint :: TCPTransport -- ^ Transport

475+ -> [Event] -- ^ Events to output on the endpoint to indicate closure

476+ -> LocalEndPoint -- ^ Local endpoint

477+ -> IO ()

446−apiCloseEndPoint transport ourEndPoint = do

478+apiCloseEndPoint transport evs ourEndPoint = do

447 -- Remove the reference from the transport state

479 -- Remove the reference from the transport state

448 removeLocalEndPoint transport ourEndPoint

480 removeLocalEndPoint transport ourEndPoint

449 -- Close the local endpoint

481 -- Close the local endpoint

@@ -460,11 +492,7 @@ apiCloseEndPoint transport ourEndPoint = do

460 -- Close all endpoints and kill all threads

492 -- Close all endpoints and kill all threads

461 forM_ (Map.elems $ vst ^. localConnections) $ tryCloseRemoteSocket

493 forM_ (Map.elems $ vst ^. localConnections) $ tryCloseRemoteSocket

462 forM_ (vst ^. internalThreads) killThread

494 forM_ (vst ^. internalThreads) killThread

463− -- We send a single message to the endpoint that it is closed. Subsequent

464− -- calls will block. We could change this so that all subsequent calls to

465− -- receive return an error, but this would mean checking for some state on

466− -- every call to receive, which is an unnecessary overhead.

467− writeChan (localChannel ourEndPoint) EndPointClosed

495+ forM_ evs $ writeChan (localChannel ourEndPoint)

468 where

496 where

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

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

470 tryCloseRemoteSocket :: RemoteEndPoint -> IO ()

498 tryCloseRemoteSocket :: RemoteEndPoint -> IO ()

network-transport/tests/TestTCP.hs+18−3

@@ -6,9 +6,9 @@ import Prelude hiding (catch, (>>=), (>>), return, fail)

6import TestTransport (testTransport)

7import TestAuxiliary (forkTry, runTests)

8import Network.Transport

9−import Network.Transport.TCP (createTransport, encodeEndPointAddress)

9+import Network.Transport.TCP (createTransport, createTransportExposeInternals, encodeEndPointAddress)

10import Data.Int (Int32)

11−import Control.Concurrent (threadDelay)

11+import Control.Concurrent (threadDelay, ThreadId, killThread)

12import Control.Concurrent.MVar (MVar, newEmptyMVar, putMVar, takeMVar, readMVar, isEmptyMVar)

13import Control.Monad (replicateM, guard, forM_, replicateM_)

14import Control.Applicative ((<$>))

@@ -41,7 +41,9 @@ instance Traceable N.Socket where

42instance Traceable N.AddrInfo where

43 trace = traceShow

44−

44+

45+instance Traceable ThreadId where

46+ trace = const Nothing

46-- Test that the server gets a ConnectionClosed message when the client closes

48-- Test that the server gets a ConnectionClosed message when the client closes

47-- the socket without sending an explicit control message to the server first

49-- the socket without sending an explicit control message to the server first

@@ -493,6 +495,18 @@ testMany nextPort = do

493 Right _ <- connect endpoint (address masterEndPoint) ReliableOrdered defaultConnectHints

495 Right _ <- connect endpoint (address masterEndPoint) ReliableOrdered defaultConnectHints

494 return ()

496 return ()

495

497

498+-- | Test what happens when the transport breaks completely

499+testBreakTransport :: IO N.ServiceName -> IO ()

500+testBreakTransport nextPort = do

501+ Right (transport, transportThread) <- nextPort >>= createTransportExposeInternals "127.0.0.1"

502+ Right endpoint <- newEndPoint transport

503+

504+ killThread transportThread -- Uh oh

505+

506+ ErrorEvent (TransportError EventTransportFailed _) <- receive endpoint

507+

508+ return ()

509+

496main :: IO ()

510main :: IO ()

497main = do

511main = do

498 portMVar <- newEmptyMVar

512 portMVar <- newEmptyMVar

@@ -507,5 +521,6 @@ main = do

507 , ("InvalidAddress", testInvalidAddress nextPort)

521 , ("InvalidAddress", testInvalidAddress nextPort)

508 , ("InvalidConnect", testInvalidConnect nextPort)

522 , ("InvalidConnect", testInvalidConnect nextPort)

509 , ("TestMany", testMany nextPort)

523 , ("TestMany", testMany nextPort)

524+ , ("TestBreakTransport", testBreakTransport nextPort)

510 ]

525 ]

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

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