d8aceac · distributed-process

git.iko.soy/distributed-process

on master help sign in

Output exceptional event after endpoint is closed

d8aceacEdsko de Vries14 years ago

network-transport/src/Network/Transport/TCP.hs+15−10

compare with master

unifiedsplit

@@ -10,9 +10,6 @@

10-- Applications that use the TCP transport should use

11-- 'Network.Socket.withSocketsDo' in their main function for Windows

12-- compatibility (see "Network.Socket").

13−--

14−-- TODOs:

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

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

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

17 createTransport

14 createTransport

18 -- * Internals (exposed for unit tests)

15 -- * Internals (exposed for unit tests)

@@ -80,6 +77,7 @@ import Control.Monad.Error (ErrorT(..), runErrorT)

80import Control.Exception ( IOException

77import Control.Exception ( IOException

81 , SomeException

78 , SomeException

82 , handle

79 , handle

80+ , throw

83 , throwIO

81 , throwIO

84 , try

82 , try

85 , bracketOnError

83 , bracketOnError

@@ -293,7 +291,7 @@ data ValidLocalEndPointState = ValidLocalEndPointState

293--

291--

294-- INV-CLOSE: Whenever we put an endpoint in closed state we remove that

292-- INV-CLOSE: Whenever we put an endpoint in closed state we remove that

295-- endpoint from localConnections first, so that if a concurrent thread reads

293-- endpoint from localConnections first, so that if a concurrent thread reads

296−-- the mvar, finds EndPointClosed, and then looks up the endpoint in

294+-- the mvar, finds RemoteEndPointClosed, and then looks up the endpoint in

297-- localConnections it is guaranteed to either find a different remote

295-- localConnections it is guaranteed to either find a different remote

298-- endpoint, or else none at all.

296-- endpoint, or else none at all.

299-- INV-RESOLVE: Whenever we move a endpoint from Closing to Closed state, we

297-- INV-RESOLVE: Whenever we move a endpoint from Closing to Closed state, we

@@ -399,7 +397,9 @@ createTransportExposeInternals host port = do

399 return

397 return

400 ( Transport

398 ( Transport

401 { newEndPoint = apiNewEndPoint transport

399 { newEndPoint = apiNewEndPoint transport

402− , closeTransport = let evs = [EndPointClosed] in

400+ , closeTransport = let evs = [ EndPointClosed

401+ , throw $ userError "Transport closed"

402+ ] in

403 apiCloseTransport transport (Just tid) evs

404 }

405 , TransportInternals

@@ -410,8 +410,10 @@ createTransportExposeInternals host port = do

410 where

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

412 terminationHandler transport ex = do

413− let ev = ErrorEvent (TransportError EventTransportFailed (show ex))

413+ let evs = [ ErrorEvent (TransportError EventTransportFailed (show ex))

414+ , throw $ userError "Transport closed"

415+ ]

414− apiCloseTransport transport Nothing [ev]

416+ apiCloseTransport transport Nothing evs

415

417

416--------------------------------------------------------------------------------

418--------------------------------------------------------------------------------

417-- API functions --

419-- API functions --

@@ -438,7 +440,10 @@ apiNewEndPoint transport = try $ do

438 { receive = readChan (localChannel ourEndPoint)

440 { receive = readChan (localChannel ourEndPoint)

439 , address = localAddress ourEndPoint

441 , address = localAddress ourEndPoint

440 , connect = apiConnect ourEndPoint

442 , connect = apiConnect ourEndPoint

441− , closeEndPoint = apiCloseEndPoint transport [EndPointClosed] ourEndPoint

443+ , closeEndPoint = let evs = [ EndPointClosed

444+ , throw $ userError "Endpoint closed"

445+ ] in

446+ apiCloseEndPoint transport evs ourEndPoint

442 , newMulticastGroup = return . Left $ newMulticastGroupError

447 , newMulticastGroup = return . Left $ newMulticastGroupError

443 , resolveMulticastGroup = return . Left . const resolveMulticastGroupError

448 , resolveMulticastGroup = return . Left . const resolveMulticastGroupError

444 }

449 }

@@ -501,7 +506,7 @@ apiSend :: EndPointPair -- ^ Local and remote endpoint

501 -> [ByteString] -- ^ Payload

506 -> [ByteString] -- ^ Payload

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

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

503apiSend (ourEndPoint, theirEndPoint) connId connAlive payload =

508apiSend (ourEndPoint, theirEndPoint) connId connAlive payload =

504− -- The 'join' joins the inner exception that we explicitly return) for

509+ -- The 'join' joins the inner exception (which we explicitly return, for

505 -- instance if the connection is closed) with the outer exception (which is

510 -- instance if the connection is closed) with the outer exception (which is

506 -- returned by 'try' when an exception is thrown by 'sendOn', and handled

511 -- returned by 'try' when an exception is thrown by 'sendOn', and handled

507 -- by 'withRemoteState')

512 -- by 'withRemoteState')

@@ -733,7 +738,7 @@ requestConnectionTo ourEndPoint theirAddress = go

733 readMVar resolved >> go

738 readMVar resolved >> go

734

739

735 RemoteEndPointClosed ->

740 RemoteEndPointClosed ->

736− -- EndPointClosed indicates that a concurrent thread was in the

741+ -- RemoteEndPointClosed indicates that a concurrent thread was in the

737 -- process of closing the TCP connection to the remote endpoint when

742 -- process of closing the TCP connection to the remote endpoint when

738 -- we obtained a reference to it. By INV-CLOSE we can assume that the

743 -- we obtained a reference to it. By INV-CLOSE we can assume that the

739 -- remote endpoint will now have been removed from ourState, so we

744 -- remote endpoint will now have been removed from ourState, so we

network-transport/tests/TestAuxiliary.hs+6

@@ -80,3 +80,9 @@ instance Show err => Traceable (TransportError err) where

81instance Traceable EndPointAddress where

82 trace = traceShow . endPointAddressToByteString

83+

84+instance Traceable SomeException where

85+ trace = traceShow

86+

87+instance Traceable ThreadId where

88+ trace = const Nothing

network-transport/tests/TestTCP.hs−6

@@ -57,15 +57,9 @@ instance Traceable N.Socket where

57instance Traceable N.AddrInfo where

58 trace = traceShow

60−instance Traceable ThreadId where

61− trace = const Nothing

62−

63instance Traceable TransportInternals where

60instance Traceable TransportInternals where

64 trace = const Nothing

61 trace = const Nothing

66−instance Traceable SomeException where

67− trace = traceShow

68−

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

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

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

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

71testEarlyDisconnect :: IO N.ServiceName -> IO ()

65testEarlyDisconnect :: IO N.ServiceName -> IO ()

network-transport/tests/TestTransport.hs+23−1

@@ -4,11 +4,12 @@ module TestTransport where

4import Prelude hiding (catch, (>>=), (>>), return, fail)

5import TestAuxiliary (forkTry, runTests)

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

7+import Control.Exception (evaluate, try, SomeException)

7import Control.Monad (replicateM, replicateM_, when, guard, forM_)

8import Control.Monad (replicateM, replicateM_, when, guard, forM_)

8import Control.Monad.Error ()

9import Control.Monad.Error ()

9import Network.Transport hiding (connect)

10import Network.Transport hiding (connect)

10import qualified Network.Transport as NT

11import qualified Network.Transport as NT

11−import Network.Transport.Internal (tlog)

12+import Network.Transport.Internal (tlog, tryIO)

12import Network.Transport.Util (spawn)

13import Network.Transport.Util (spawn)

13import Data.ByteString (ByteString)

14import Data.ByteString (ByteString)

14import Data.ByteString.Char8 (pack)

15import Data.ByteString.Char8 (pack)

@@ -734,6 +735,26 @@ testConnectClosedEndPoint transport = do

734

735

735 takeMVar clientDone

736 takeMVar clientDone

736

737

738+-- | We should receive an exception when doing a 'receive' after we have been

739+-- notified that an endpoint has been closed

740+testExceptionOnReceive :: IO (Either String Transport) -> IO ()

741+testExceptionOnReceive newTransport = do

742+ Right transport <- newTransport

743+

744+ -- Test one: when we close an endpoint specifically

745+ Right endpoint1 <- newEndPoint transport

746+ closeEndPoint endpoint1

747+ EndPointClosed <- receive endpoint1

748+ Left _ <- (try :: IO a -> IO (Either SomeException a)) (receive endpoint1 >>= evaluate)

749+

750+ -- Test two: when we close the entire transport

751+ Right endpoint2 <- newEndPoint transport

752+ closeTransport transport

753+ EndPointClosed <- receive endpoint2

754+ Left _ <- (try :: IO a -> IO (Either SomeException a)) (receive endpoint2 >>= evaluate)

755+

756+ return ()

757+

737-- Transport tests

758-- Transport tests

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

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

739testTransport newTransport = do

760testTransport newTransport = do

@@ -753,6 +774,7 @@ testTransport newTransport = do

753 , ("CloseEndPoint", testCloseEndPoint transport numPings)

774 , ("CloseEndPoint", testCloseEndPoint transport numPings)

754 , ("CloseTransport", testCloseTransport newTransport)

775 , ("CloseTransport", testCloseTransport newTransport)

755 , ("ConnectClosedEndPoint", testConnectClosedEndPoint transport)

776 , ("ConnectClosedEndPoint", testConnectClosedEndPoint transport)

777+ , ("ExceptionOnReceive", testExceptionOnReceive newTransport)

756 ]

778 ]

757 where

779 where

758 numPings = 10000 :: Int

780 numPings = 10000 :: Int