305bd0a · distributed-process

git.iko.soy/distributed-process

on master help sign in

Continue work on error handling

305bd0aEdsko de Vries14 years ago

network-transport-2/src/Network/Transport.hs+10−5

compare with master

unifiedsplit

@@ -16,7 +16,7 @@ module Network.Transport ( -- * Types

16 , NewMulticastGroupErrorCode(..)

17 , ResolveMulticastGroupErrorCode(..)

18 , SendErrorCode(..)

19− , EventErrorCode(..)

19+ , ErrorEventInfo(..)

20 ) where

22import Data.ByteString (ByteString)

@@ -70,7 +70,7 @@ data Event =

70 -- | Received multicast

71 | ReceivedMulticast MulticastAddress [ByteString]

72 -- | Error that caused a bunch of connections to close (possibly none)

73− | ErrorEvent EventErrorCode [ConnectionId]

73+ | ErrorEvent ErrorEventInfo

74 deriving Show

76-- | Connection IDs enable receivers to distinguish one connection from another.

@@ -122,6 +122,7 @@ instance Error (FailedWith error) where

122-- | Errors during the creation of an endpoint (currently, there are none)

123data NewEndPointErrorCode =

124 NewEndPointTransportFailure -- ^ Transport-wide failure

125+ deriving Show

125

126

126-- | Connection failure

127-- | Connection failure

127data ConnectErrorCode =

128data ConnectErrorCode =

@@ -148,7 +149,11 @@ data SendErrorCode =

148 deriving Show

149 deriving Show

149

150

150-- | Error codes used when reporting errors to endpoints (through receive)

151-- | Error codes used when reporting errors to endpoints (through receive)

151−data EventErrorCode =

152+data ErrorEventInfo =

152− EventErrorEndPointClosed -- ^ Endpoint was closed (manually or because of an error)

153+ -- | Endpoint was closed (manually or because of an error)

154+ ErrorEventEndPointClosed [ConnectionId]

153− | EventErrorTransportFailure -- ^ Transport-wide fatal error

155+ -- | Transport-wide fatal error

156+ | ErrorEventTransportFailure

157+ -- | Connection to a remote endpoint was lost

158+ | ErrorEventConnectionLost EndPointAddress [ConnectionId]

154 deriving Show

159 deriving Show

network-transport-2/src/Network/Transport/TCP.hs+87−54

@@ -10,6 +10,7 @@ module Network.Transport.TCP ( -- * Main API

10 createTransport

11 , -- * TCP specific functionality

12 EndPointId

13+ , encodeEndPointAddress

13 , decodeEndPointAddress

14 , decodeEndPointAddress

14 , ControlHeader(..)

15 , ControlHeader(..)

15 , ConnectionRequestResponse(..)

16 , ConnectionRequestResponse(..)

@@ -237,6 +238,12 @@ data LocalState = LocalState { _nextConnectionId :: ConnectionId

237--

238--

238-- Since we don't do concurrent reads from the same socket we don't need to

239-- Since we don't do concurrent reads from the same socket we don't need to

239-- take the lock when reading from the socket.

240-- take the lock when reading from the socket.

241+--

242+-- Moreover, we insist on the invariant (INV-CLOSE) that whenever we put an

243+-- endpoint in closed state we remove that endpoint from localConnections

244+-- first, so that if a concurrent thread reads the mvar, finds EndPointClosed,

245+-- and then looks up the endpoint in localConnections it is guaranteed to

246+-- either find a different remote endpoint, or else none at all.

240

247

241data RemoteState = RemoteEndPointInvalid (FailedWith ConnectErrorCode)

248data RemoteState = RemoteEndPointInvalid (FailedWith ConnectErrorCode)

242 | RemoteEndPointValid RemoteConnection

249 | RemoteEndPointValid RemoteConnection

@@ -258,14 +265,14 @@ data ControlHeader =

258 | CloseConnection -- ^ Tell the remote endpoint we will no longer be using a connection

265 | CloseConnection -- ^ Tell the remote endpoint we will no longer be using a connection

259 | ControlResponse -- ^ Respond to a control request _from_ the remote endpoint

266 | ControlResponse -- ^ Respond to a control request _from_ the remote endpoint

260 | CloseSocket -- ^ Request to close the connection (see module description)

267 | CloseSocket -- ^ Request to close the connection (see module description)

261− deriving (Enum, Bounded)

268+ deriving (Enum, Bounded, Show)

262

269

263-- Response sent by /B/ to /A/ when /A/ tries to connect

270-- Response sent by /B/ to /A/ when /A/ tries to connect

264data ConnectionRequestResponse =

271data ConnectionRequestResponse =

265 ConnectionRequestAccepted -- ^ /B/ accepts the connection

272 ConnectionRequestAccepted -- ^ /B/ accepts the connection

266 | ConnectionRequestEndPointInvalid -- ^ /A/ requested an invalid endpoint

273 | ConnectionRequestEndPointInvalid -- ^ /A/ requested an invalid endpoint

267 | ConnectionRequestCrossed -- ^ /A/s request crossed with a request from /B/ (see protocols)

274 | ConnectionRequestCrossed -- ^ /A/s request crossed with a request from /B/ (see protocols)

268− deriving (Enum, Bounded)

275+ deriving (Enum, Bounded, Show)

269

276

270--------------------------------------------------------------------------------

277--------------------------------------------------------------------------------

271-- Top-level functionality --

278-- Top-level functionality --

@@ -393,7 +400,7 @@ apiCloseEndPoint transport ourEndPoint = do

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

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

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

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

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

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

396− writeChan (localChannel ourEndPoint) (ErrorEvent EventErrorEndPointClosed incoming)

403+ writeChan (localChannel ourEndPoint) (ErrorEvent $ ErrorEventEndPointClosed incoming)

397 where

404 where

398 remoteSocketOf :: RemoteEndPoint -> MaybeT IO N.Socket

405 remoteSocketOf :: RemoteEndPoint -> MaybeT IO N.Socket

399 remoteSocketOf theirEndPoint = do

406 remoteSocketOf theirEndPoint = do

@@ -452,7 +459,7 @@ createLocalEndPoint transport = do

452 }

459 }

453 modifyMVar (transportState transport) $ \st -> do

460 modifyMVar (transportState transport) $ \st -> do

454 let ix = fromIntegral $ Map.size (st ^. localEndPoints)

461 let ix = fromIntegral $ Map.size (st ^. localEndPoints)

455− let addr = encodeEndPointAddress transport ix

462+ let addr = encodeEndPointAddress (transportHost transport) (transportPort transport) ix

456 let localEndPoint = LocalEndPoint { localAddress = addr

463 let localEndPoint = LocalEndPoint { localAddress = addr

457 , localChannel = chan

464 , localChannel = chan

458 , localState = state

465 , localState = state

@@ -472,7 +479,7 @@ requestConnectionTo ourEndPoint theirAddress = ErrorT go

472 go = do

479 go = do

473 let ourState = localState ourEndPoint

480 let ourState = localState ourEndPoint

474 (theirEndPoint, isNew) <- modifyMVar ourState $ \st ->

481 (theirEndPoint, isNew) <- modifyMVar ourState $ \st ->

475− case st ^. connectionTo theirAddress of

482+ case st ^. localConnectionTo theirAddress of

476 Just theirEndPoint ->

483 Just theirEndPoint ->

477 return (st, (theirEndPoint, False))

484 return (st, (theirEndPoint, False))

478 Nothing -> do

485 Nothing -> do

@@ -480,7 +487,7 @@ requestConnectionTo ourEndPoint theirAddress = ErrorT go

480 let theirEndPoint = RemoteEndPoint { remoteAddress = theirAddress

487 let theirEndPoint = RemoteEndPoint { remoteAddress = theirAddress

481 , remoteState = theirState

488 , remoteState = theirState

482 }

489 }

483− return (connectionTo theirAddress ^= Just theirEndPoint $ st, (theirEndPoint, True))

490+ return (localConnectionTo theirAddress ^= Just theirEndPoint $ st, (theirEndPoint, True))

484

491

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

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

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

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

@@ -574,14 +581,14 @@ connectToRemoteEndPoint (ourEndPoint, theirEndPoint) = do

574 -- waiting on the result of this call to connect will get the

581 -- waiting on the result of this call to connect will get the

575 -- RemoteEndPointInvalid; subsequent threads will initiate a new

582 -- RemoteEndPointInvalid; subsequent threads will initiate a new

576 -- connection requests.

583 -- connection requests.

577− modifyMVar_ ourState $ return . (connectionTo theirAddress ^= Nothing)

584+ modifyMVar_ ourState $ return . (localConnectionTo theirAddress ^= Nothing)

578 putMVar theirState (RemoteEndPointInvalid (invalidAddress "Invalid endpoint"))

585 putMVar theirState (RemoteEndPointInvalid (invalidAddress "Invalid endpoint"))

579 N.sClose sock

586 N.sClose sock

580 Right (ConnectionRequestCrossed, sock) -> do

587 Right (ConnectionRequestCrossed, sock) -> do

581 N.sClose sock

588 N.sClose sock

582 Left err -> do

589 Left err -> do

583 -- See comment above

590 -- See comment above

584− modifyMVar_ ourState $ return . (connectionTo theirAddress ^= Nothing)

591+ modifyMVar_ ourState $ return . (localConnectionTo theirAddress ^= Nothing)

585 putMVar theirState (RemoteEndPointInvalid err)

592 putMVar theirState (RemoteEndPointInvalid err)

586 -- TODO: this isn't quite right; socket *might* have been opened here,

593 -- TODO: this isn't quite right; socket *might* have been opened here,

587 -- in which case we should close it

594 -- in which case we should close it

@@ -613,9 +620,9 @@ connectToRemoteEndPoint (ourEndPoint, theirEndPoint) = do

613 EndPointAddress ourAddress = localAddress ourEndPoint

620 EndPointAddress ourAddress = localAddress ourEndPoint

614

621

615-- | Encode end point address

622-- | Encode end point address

616−encodeEndPointAddress :: TCPTransport -> EndPointId -> EndPointAddress

623+encodeEndPointAddress :: N.HostName -> N.ServiceName -> EndPointId -> EndPointAddress

617−encodeEndPointAddress transport ix = EndPointAddress . BSC.pack $

624+encodeEndPointAddress host port ix = EndPointAddress . BSC.pack $

618− transportHost transport ++ ":" ++ transportPort transport ++ ":" ++ show ix

625+ host ++ ":" ++ port ++ ":" ++ show ix

619

626

620-- | Decode end point address

627-- | Decode end point address

621-- TODO: This uses regular expression parsing, which is nice, but unnecessary

628-- TODO: This uses regular expression parsing, which is nice, but unnecessary

@@ -677,7 +684,7 @@ handleConnectionRequest transport sock = do

677 request <- tryIO $ do

684 request <- tryIO $ do

678 ourEndPointId <- recvInt32 sock

685 ourEndPointId <- recvInt32 sock

679 theirAddress <- EndPointAddress . BS.concat <$> recvWithLength sock

686 theirAddress <- EndPointAddress . BS.concat <$> recvWithLength sock

680− let ourAddress = encodeEndPointAddress transport ourEndPointId

687+ let ourAddress = encodeEndPointAddress (transportHost transport) (transportPort transport) ourEndPointId

681 ourEndPoint <- (^. localEndPointAt ourAddress) <$> readMVar (transportState transport)

688 ourEndPoint <- (^. localEndPointAt ourAddress) <$> readMVar (transportState transport)

682 return (theirAddress, ourEndPoint)

689 return (theirAddress, ourEndPoint)

683 case request of

690 case request of

@@ -686,13 +693,13 @@ handleConnectionRequest transport sock = do

686 N.sClose sock

693 N.sClose sock

687 Right (theirAddress, Just ourEndPoint) -> forkEndPointThread ourEndPoint $ do

694 Right (theirAddress, Just ourEndPoint) -> forkEndPointThread ourEndPoint $ do

688 (crossed, theirEndPoint) <- modifyMVar (localState ourEndPoint) $ \st ->

695 (crossed, theirEndPoint) <- modifyMVar (localState ourEndPoint) $ \st ->

689− case st ^. connectionTo theirAddress of

696+ case st ^. localConnectionTo theirAddress of

690 Nothing -> do

697 Nothing -> do

691 theirState <- newEmptyMVar

698 theirState <- newEmptyMVar

692 let theirEndPoint = RemoteEndPoint { remoteAddress = theirAddress

699 let theirEndPoint = RemoteEndPoint { remoteAddress = theirAddress

693 , remoteState = theirState

700 , remoteState = theirState

694 }

701 }

695− return (connectionTo theirAddress ^= Just theirEndPoint $ st, (False, theirEndPoint))

702+ return (localConnectionTo theirAddress ^= Just theirEndPoint $ st, (False, theirEndPoint))

696 Just theirEndPoint ->

703 Just theirEndPoint ->

697 return (st, (localAddress ourEndPoint < theirAddress, theirEndPoint))

704 return (st, (localAddress ourEndPoint < theirAddress, theirEndPoint))

698 if crossed

705 if crossed

@@ -733,28 +740,10 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do

733 _ ->

740 _ ->

734 error "handleIncomingMessages RELY violation"

741 error "handleIncomingMessages RELY violation"

735

742

736− tryIO $ go sock

743+ mCleanExit <- tryIO $ go sock

737−

738− -- Normally, there would be no unclosed connections, but if the client

739− -- exits prematurely there might be

740− mUnclosedConnections <- modifyMVar theirState $ \st ->

741− case st of

744+ case mCleanExit of

742− RemoteEndPointInvalid _ ->

743− error "handleIncomingMessages RELY violation"

744− RemoteEndPointValid remoteConn ->

745− return (RemoteEndPointClosed, Just $ remoteConn ^. remoteIncoming)

746− RemoteEndPointClosing _ _ ->

747− return (RemoteEndPointClosed, Nothing)

748− RemoteEndPointClosed ->

749− return (st, Nothing)

750−

751− -- TODO: we need to remote this endpoint from localConnections at this point

752−

753− case mUnclosedConnections of

754− Nothing -> return ()

745+ Right () -> return ()

755− Just unclosedConnections ->

746+ Left err -> prematureExit sock err

756− writeChan ourChannel (ErrorEvent EventErrorEndPointClosed (IntSet.elems unclosedConnections))

757−

758 where

747 where

759 -- Dispatch

748 -- Dispatch

760 go :: N.Socket -> IO ()

749 go :: N.Socket -> IO ()

@@ -835,24 +824,45 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do

835 -- Close the socket (if we don't have any outgoing connections)

824 -- Close the socket (if we don't have any outgoing connections)

836 closeSocket :: N.Socket -> IO Bool

825 closeSocket :: N.Socket -> IO Bool

837 closeSocket sock = do

826 closeSocket sock = do

827+ -- We need to check if we can close the socket (that is, if we don't have

828+ -- any outgoing connections), and once we are sure that we can put the

829+ -- endpoint in Closed state. However, by INV-CLOSE we can only put the

830+ -- endpoint in Closed state once we remove the endpoint from our local

831+ -- connections. But we can do that only once we are sure that we can

832+ -- close the endpoint. Catch-22. We can resolve the catch-22 by locking

833+ -- *both* our local state and the endpoint state, but at the cost of

834+ -- introducing a double lock and all the associated perils, or by putting

835+ -- the remote endpoint in Closing state first. We opt for the latter.

838− didClose <- modifyMVar theirState $ \st ->

836+ canClose <- modifyMVar theirState $ \st ->

839 case st of

837 case st of

840− RemoteEndPointValid remoteConn | remoteConn ^. remoteOutgoing == 0 -> do

838+ RemoteEndPointValid remoteConn -> do

839+ -- TODO: We check only for outgoing connections here. That is

840+ -- correct, but we need to makke sure that we send the right event

841+ -- to our endpoint for any incoming connections that are still live

842+ -- at this point.

843+ if remoteConn ^. remoteOutgoing == 0

844+ then do

845+ -- TODO: code duplication with closeIfUnused

846+ resolved <- newEmptyMVar

841− sendOn remoteConn [encodeInt32 CloseSocket]

847+ sendOn remoteConn [encodeInt32 CloseSocket]

842− N.sClose sock

848+ return (RemoteEndPointClosing resolved remoteConn, Just resolved)

843− return (RemoteEndPointClosed, True)

849+ else

850+ return (st, Nothing)

844− RemoteEndPointClosing resolved _ -> do

851+ RemoteEndPointClosing resolved _ ->

845− -- If the socket is already in closing state, then we have already

852+ return (st, Just resolved)

846− -- sent a CloseSocket request; hence, we don't need to do it again

847− -- at this point

848− putMVar resolved ()

849− N.sClose sock

850− return (RemoteEndPointClosed, True)

851 _ ->

853 _ ->

854+ error "handleIncomingConnections RELY violation"

855+

856+ case canClose of

857+ Nothing ->

852− return (st, False)

858+ return False

859+ Just resolved -> do

853− when didClose $ modifyMVar_ ourState $ return . (connectionTo theirAddr ^= Nothing)

860+ modifyMVar_ ourState $ return . (localConnectionTo theirAddr ^= Nothing)

861+ modifyMVar_ theirState $ return . const RemoteEndPointClosed

862+ N.sClose sock

863+ putMVar resolved ()

854− return didClose

864+ return True

855−

865+

856 -- Read a message and output it on the endPoint's channel

866 -- Read a message and output it on the endPoint's channel

857 readMessage :: N.Socket -> ConnectionId -> IO ()

867 readMessage :: N.Socket -> ConnectionId -> IO ()

858 readMessage sock connId = recvWithLength sock >>= writeChan ourChannel . Received connId

868 readMessage sock connId = recvWithLength sock >>= writeChan ourChannel . Received connId

@@ -863,6 +873,29 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do

863 theirState = remoteState theirEndPoint

873 theirState = remoteState theirEndPoint

864 theirAddr = remoteAddress theirEndPoint

874 theirAddr = remoteAddress theirEndPoint

865

875

876+ -- Deal with a premature exit

877+ prematureExit :: N.Socket -> IOException -> IO ()

878+ prematureExit sock _ = do

879+ N.sClose sock

880+ modifyMVar_ ourState $ return . (localConnectionTo theirAddr ^= Nothing)

881+ mUnclosedConnections <- modifyMVar theirState $ \st ->

882+ case st of

883+ RemoteEndPointInvalid _ ->

884+ error "handleIncomingMessages RELY violation"

885+ RemoteEndPointValid remoteConn ->

886+ return (RemoteEndPointClosed, Just $ remoteConn ^. remoteIncoming)

887+ RemoteEndPointClosing _ _ ->

888+ return (RemoteEndPointClosed, Nothing)

889+ RemoteEndPointClosed ->

890+ return (st, Nothing)

891+

892+ -- We send the connection lost message even if unclosedConnections is the

893+ -- empty set, because *outgoing* connections will be broken too now

894+ case mUnclosedConnections of

895+ Nothing ->

896+ return ()

897+ Just unclosedConnections -> writeChan ourChannel . ErrorEvent $

898+ ErrorEventConnectionLost (remoteAddress theirEndPoint) (IntSet.elems unclosedConnections)

866

899

867

900

868-- | Get the next connection ID

901-- | Get the next connection ID

@@ -911,8 +944,8 @@ localEndPointAt addr = localEndPoints >>> mapLens addr

911pendingCtrlRequestsAt :: ControlRequestId -> Lens LocalState (Maybe (MVar [ByteString]))

944pendingCtrlRequestsAt :: ControlRequestId -> Lens LocalState (Maybe (MVar [ByteString]))

912pendingCtrlRequestsAt ix = pendingCtrlRequests >>> intMapLens (fromIntegral ix)

945pendingCtrlRequestsAt ix = pendingCtrlRequests >>> intMapLens (fromIntegral ix)

913

946

914−connectionTo :: EndPointAddress -> Lens LocalState (Maybe RemoteEndPoint)

947+localConnectionTo :: EndPointAddress -> Lens LocalState (Maybe RemoteEndPoint)

915−connectionTo addr = localConnections >>> mapLens addr

948+localConnectionTo addr = localConnections >>> mapLens addr

916

949

917remoteOutgoing :: Lens RemoteConnection Int

950remoteOutgoing :: Lens RemoteConnection Int

918remoteOutgoing = lens _remoteOutgoing (\cs conn -> conn { _remoteOutgoing = cs })

951remoteOutgoing = lens _remoteOutgoing (\cs conn -> conn { _remoteOutgoing = cs })

network-transport-2/tests/TestAuxiliary.hs+6−1

@@ -1,4 +1,9 @@

1−module TestAuxiliary (runTest, runTests, forkTry) where

1+module TestAuxiliary ( -- Running tests

2+ runTest

3+ , runTests

4+ -- Writing tests

5+ , forkTry

6+ ) where

3import Prelude hiding (catch)

8import Prelude hiding (catch)

4import Control.Concurrent (myThreadId, forkIO, ThreadId, throwTo)

9import Control.Concurrent (myThreadId, forkIO, ThreadId, throwTo)

network-transport-2/tests/TestTCP.hs+86−28

@@ -1,22 +1,23 @@

1+{-# LANGUAGE RebindableSyntax #-}

1module Main where

2module Main where

3−import Prelude hiding (catch)

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

4import TestTransport (testTransport)

5import TestTransport (testTransport)

5import TestAuxiliary (forkTry, runTests)

6import TestAuxiliary (forkTry, runTests)

6import Network.Transport

7import Network.Transport

7−import Network.Transport.TCP (createTransport)

8+import Network.Transport.TCP (createTransport, encodeEndPointAddress)

8import Data.Int (Int32)

9import Data.Int (Int32)

9import Data.Maybe (fromJust)

10import Data.Maybe (fromJust)

10import Control.Concurrent (threadDelay)

11import Control.Concurrent (threadDelay)

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

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

12−import Control.Monad (replicateM, guard)

13+import Control.Monad (replicateM)

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

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

14import Network.Transport.TCP ( decodeEndPointAddress

15import Network.Transport.TCP ( decodeEndPointAddress

15 , EndPointId

16 , EndPointId

16 , ControlHeader(..)

17 , ControlHeader(..)

17 , ConnectionRequestResponse(..)

18 , ConnectionRequestResponse(..)

18 )

19 )

19−import Network.Transport.Internal (encodeInt32, prependLength, tlog)

20+import Network.Transport.Internal (encodeInt32, prependLength, tlog, tryIO)

20import Network.Transport.Internal.TCP (recvInt32)

21import Network.Transport.Internal.TCP (recvInt32)

21import qualified Network.Socket as N ( getAddrInfo

22import qualified Network.Socket as N ( getAddrInfo

22 , socket

23 , socket

@@ -32,41 +33,97 @@ import qualified Network.Socket as N ( getAddrInfo

32 , ServiceName

33 , ServiceName

33 )

34 )

34import Network.Socket.ByteString (sendMany)

35import Network.Socket.ByteString (sendMany)

36+import Data.String (fromString)

37+import Traced

38+

39+instance Show (MVar a) where

40+ show _ = "<<mvar>>"

41+

42+instance Show EndPoint where

43+ show _ = "<<endpoint>>"

44+

45+instance Show Transport where

46+ show _ = "<<transport>>"

47+

48+instance Show Connection where

49+ show _ = "<<connection>>"

50+

51+trlog :: String -> Traced ()

52+trlog = liftIO . tlog

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

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

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

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

38testEarlyDisconnect :: IO ()

56testEarlyDisconnect :: IO ()

39testEarlyDisconnect = do

57testEarlyDisconnect = do

58+ clientAddr <- newEmptyMVar

40 serverAddr <- newEmptyMVar

59 serverAddr <- newEmptyMVar

41 serverDone <- newEmptyMVar

60 serverDone <- newEmptyMVar

42− Right transport <- createTransport "127.0.0.1" "8081"

61+

43−

62+ tlog "testEarlyDisconnect"

44− forkTry $ server transport serverAddr serverDone

63+ forkTry $ server serverAddr clientAddr serverDone

45− forkTry $ client transport serverAddr

64+ forkTry $ client serverAddr clientAddr

47 takeMVar serverDone

66 takeMVar serverDone

48 where

67 where

49− server :: Transport -> MVar (N.HostName, N.ServiceName, EndPointId) -> MVar () -> IO ()

68+ server :: MVar (N.HostName, N.ServiceName, EndPointId) -> MVar EndPointAddress -> MVar () -> IO ()

50− server transport serverAddr serverDone = do

69+ server serverAddr clientAddr serverDone = do

70+ tlog "Server"

71+ Right transport <- createTransport "127.0.0.1" "8081"

51 Right endpoint <- newEndPoint transport

72 Right endpoint <- newEndPoint transport

52 putMVar serverAddr (fromJust . decodeEndPointAddress . address $ endpoint)

73 putMVar serverAddr (fromJust . decodeEndPointAddress . address $ endpoint)

74+ theirAddr <- readMVar clientAddr

75+

76+ runTraced $ do

77+ -- First test: they connect to us, then drop the connection

53− ConnectionOpened _ _ _ <- receive endpoint

78+ ConnectionOpened cid _ addr <- liftIO $ receive endpoint

54− ConnectionClosed _ <- receive endpoint

79+ True <- return $ addr == theirAddr

80+

81+ ErrorEvent (ErrorEventConnectionLost addr [cid']) <- liftIO $ receive endpoint

82+ True <- return $ addr == theirAddr && cid' == cid

83+

84+ -- Second test: after they dropped their connection to us, we now try to

85+ -- establish a connection to them. This should re-establish the broken

86+ -- TCP connection.

87+ trlog "Trying to connect to client"

88+ Right conn <- liftIO $ connect endpoint theirAddr ReliableOrdered

89+

90+

91+ return ()

92+

55 putMVar serverDone ()

93 putMVar serverDone ()

95+{-

96+ -- To test the connection, we do a simple ping test; as before, however,

97+ -- the remote client won't close the connection nicely but just closes

98+ -- the socket

99+ do

100+ send conn ["ping"]

101+ cid <- expect endpoint (\(ConnectionOpened cid _ addr) -> (addr == theirAddr, cid))

102+ expect' endpoint (\(Received cid' [msg]) -> cid' == cid && msg == "pong")

103+ expect' endpoint (\(ErrorEvent (ErrorEventConnectionLost addr [cid'])) -> addr == theirAddr && cid' == cid)

104+

105+ -- TODO: we now need to decide (and test) and should happen on a subsequent 'send'

106+

107+-}

108+

109+

110+

57− client :: Transport -> MVar (N.HostName, N.ServiceName, EndPointId) -> IO ()

111+ client :: MVar (N.HostName, N.ServiceName, EndPointId) -> MVar EndPointAddress -> IO ()

58− client transport serverAddr = do

112+ client serverAddr clientAddr = do

59− Right endpoint <- newEndPoint transport

113+ tlog "Client"

60− let EndPointAddress myAddress = address endpoint

114+ let ourAddress = encodeEndPointAddress "127.0.0.1" "8082" 0

61−

115+ putMVar clientAddr ourAddress

116+ (_, _, endPointIx) <- readMVar serverAddr

117+

118+

119+

62 -- Connect to the server

120 -- Connect to the server

63 addr:_ <- N.getAddrInfo Nothing (Just "127.0.0.1") (Just "8081")

121 addr:_ <- N.getAddrInfo Nothing (Just "127.0.0.1") (Just "8081")

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

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

65 N.setSocketOption sock N.ReuseAddr 1

123 N.setSocketOption sock N.ReuseAddr 1

66 N.connect sock (N.addrAddress addr)

124 N.connect sock (N.addrAddress addr)

125

68− (_, _, endPointIx) <- readMVar serverAddr

69− sendMany sock (encodeInt32 endPointIx : prependLength [myAddress])

126+ sendMany sock (encodeInt32 endPointIx : prependLength [endPointAddressToByteString ourAddress])

127

71 -- Wait for acknowledgement

128 -- Wait for acknowledgement

72 ConnectionRequestAccepted <- toEnum <$> recvInt32 sock

129 ConnectionRequestAccepted <- toEnum <$> recvInt32 sock

@@ -87,24 +144,24 @@ testInvalidAddress = do

144

88-- | Test connecting to invalid or non-existing endpoints

145-- | Test connecting to invalid or non-existing endpoints

89testInvalidConnect :: IO ()

146testInvalidConnect :: IO ()

90−testInvalidConnect = do

147+testInvalidConnect = runTraced $ do

91− Right transport <- createTransport "127.0.0.1" "8083"

148+ Right transport <- liftIO $ createTransport "127.0.0.1" "8083"

92− Right endpoint <- newEndPoint transport

149+ Right endpoint <- liftIO $ newEndPoint transport

150

94 -- Syntax error in the endpoint address

151 -- Syntax error in the endpoint address

95− Left (FailedWith ConnectInvalidAddress _) <-

152+ Left (FailedWith ConnectInvalidAddress _) <- liftIO $

96 connect endpoint (EndPointAddress "InvalidAddress") ReliableOrdered

153 connect endpoint (EndPointAddress "InvalidAddress") ReliableOrdered

154

98 -- Syntax connect, but invalid hostname (TCP address lookup failure)

155 -- Syntax connect, but invalid hostname (TCP address lookup failure)

99− Left (FailedWith ConnectInvalidAddress _) <-

156+ Left (FailedWith ConnectInvalidAddress _) <- liftIO $

100 connect endpoint (EndPointAddress "invalidHost:port:0") ReliableOrdered

157 connect endpoint (EndPointAddress "invalidHost:port:0") ReliableOrdered

101

158

102 -- TCP address correct, but nobody home at that address

159 -- TCP address correct, but nobody home at that address

103− Left (FailedWith ConnectFailed _) <-

160+ Left (FailedWith ConnectFailed _) <- liftIO $

104 connect endpoint (EndPointAddress "127.0.0.1:9000:0") ReliableOrdered

161 connect endpoint (EndPointAddress "127.0.0.1:9000:0") ReliableOrdered

105

162

106 -- Valid TCP address but invalid endpoint number

163 -- Valid TCP address but invalid endpoint number

107− Left (FailedWith ConnectInvalidAddress _) <-

164+ Left (FailedWith ConnectInvalidAddress _) <- liftIO $

108 connect endpoint (EndPointAddress "127.0.0.1:8083:1") ReliableOrdered

165 connect endpoint (EndPointAddress "127.0.0.1:8083:1") ReliableOrdered

109

166

110 return ()

167 return ()

@@ -276,7 +333,7 @@ testBlockAfterCloseSocket = do

276 -- respond by sending a ConnectionRequest)

333 -- respond by sending a ConnectionRequest)

277 forkTry $ do

334 forkTry $ do

278 recvInt32 sock :: IO Int32

335 recvInt32 sock :: IO Int32

279− isEmptyMVar unblocked >>= guard . not

336+ isEmptyMVar unblocked >>= (guard . not)

280 putMVar clientDone ()

337 putMVar clientDone ()

281

338

282 threadDelay 1000000

339 threadDelay 1000000

@@ -289,7 +346,8 @@ testBlockAfterCloseSocket = do

289

346

290main :: IO ()

347main :: IO ()

291main = do

348main = do

349+ tryIO $ runTests

292− runTests [ ("EarlyDisconnect", testEarlyDisconnect)

350+ [ ("EarlyDisconnect", testEarlyDisconnect)

293 , ("InvalidAddress", testInvalidAddress)

351 , ("InvalidAddress", testInvalidAddress)

294 , ("InvalidConnect", testInvalidConnect)

352 , ("InvalidConnect", testInvalidConnect)

295 , ("IgnoreCloseSocket", testIgnoreCloseSocket)

353 , ("IgnoreCloseSocket", testIgnoreCloseSocket)

network-transport-2/tests/TestTransport.hs+2−2

@@ -37,7 +37,7 @@ echoServer endpoint = do

37 ReceivedMulticast _ _ ->

38 -- Ignore

39 go cs

40− ErrorEvent _ _ ->

40+ ErrorEvent _ ->

41 fail (show event)

43-- | Wait for an event, throw an exception if the given predicate returns false

@@ -298,7 +298,7 @@ collect endPoint numEvents = go numEvents Map.empty Map.empty

298 go (n - 1) (Map.adjust (msg :) cid open) closed

299 ReceivedMulticast _ _ ->

300 fail "Unexpected multicast"

301− ErrorEvent _ _ ->

301+ ErrorEvent _ ->

302 fail "Unexpected error"

303

304-- | Open connection, close it, then reopen it

network-transport-2/tests/Traced.hsnew file+161

@@ -0,0 +1,161 @@

1+{-# LANGUAGE RebindableSyntax #-}

2+-- | Monad to trace all terms that get bound

3+--

4+-- Examples:

5+--

6+-- > test1 :: IO Int

7+-- > test1 = runTraced $ do

8+-- > Left x <- return (Left 1 :: Either Int Int)

9+-- > Left y <- return (Left 2 :: Either Int Int)

10+-- > return (x + y)

11+--

12+-- outputs

13+--

14+-- > 3

15+--

16+-- as expected. Things get more interesting when we have a pattern match failure:

17+--

18+-- > test2 :: IO Int

19+-- > test2 = runTraced $ do

20+-- > Left x <- return (Left 1 :: Either Int Int)

21+-- > Right y <- return (Left 2 :: Either Int Int)

22+-- > return (x + y)

23+--

24+-- outputs

25+--

26+-- > *** Exception: user error (Pattern match failure in do expression at Traced.hs:96:3-9

27+-- > Trace:

28+-- > Left 1

29+-- > Left 2

30+-- > )

31+--

32+-- The module also exports a 'guard' function which is used in similar fashion

33+-- to the 'guard' from "Control.Monad":

34+--

35+-- > test3 :: IO Int

36+-- > test3 = runTraced $ do

37+-- > Left x <- return (Left 1 :: Either Int Int)

38+-- > Left y <- return (Left 2 :: Either Int Int)

39+-- > guard (x == 3)

40+-- > return (x + y)

41+--

42+-- outputs

43+--

44+-- > *** Exception: user error (Guard failed

45+-- > Trace:

46+-- > Left 1

47+-- > Left 2

48+-- > )

49+--

50+-- Note that we get no line-number info this way; we can fix this by saying

51+--

52+-- > test4 :: IO Int

53+-- > test4 = runTraced $ do

54+-- > Left x <- return (Left 1 :: Either Int Int)

55+-- > Left y <- return (Left 2 :: Either Int Int)

56+-- > True <- return (x == 3)

57+-- > return (x + y)

58+--

59+-- which outputs

60+--

61+-- > *** Exception: user error (Pattern match failure in do expression at Traced.hs:110:3-6

62+-- > Trace:

63+-- > Left 1

64+-- > Left 2

65+-- > False

66+-- > )

67+--

68+-- The module also exports a 'liftIO', and does not trace ignored unit values (from '>>'):

69+--

70+-- > test5 :: IO Int

71+-- > test5 = runTraced $ do

72+-- > Left x <- return (Left 1 :: Either Int Int)

73+-- > liftIO $ putStrLn "hi"

74+-- > Left y <- return (Left 2 :: Either Int Int)

75+-- > True <- return (y == 1)

76+-- > return (x + y)

77+--

78+-- outputs

79+--

80+-- > hi

81+-- > *** Exception: user error (Pattern match failure in do expression at Traced.hs:137:3-6

82+-- > Trace:

83+-- > Left 1

84+-- > Left 2

85+-- > False

86+-- > )

87+--

88+-- Every Monad is also a MonadS, but not vice versa.

89+

90+module Traced ( -- Subclass of monads where bind is defined only for showable arguments

91+ MonadS(..)

92+ , return

93+ , (>>=)

94+ , (>>)

95+ , fail

96+ , guard

97+ -- MonadS instance that shows a trace of all bound terms when an error occurs

98+ , Traced

99+ , runTraced

100+ , liftIO

101+ ) where

102+

103+import Prelude hiding ((>>=), return, fail, catch, (>>))

104+import qualified Prelude

105+import Data.String (fromString)

106+

107+class MonadS m where

108+ returnS :: a -> m a

109+ bindS :: Show a => m a -> (a -> m b) -> m b

110+ failS :: String -> m a

111+ seqS :: Show a => m a -> m b -> m b

112+

113+instance MonadS IO where

114+ returnS = Prelude.return

115+ bindS = (Prelude.>>=)

116+ failS = Prelude.fail

117+ seqS = (Prelude.>>)

118+

119+newtype Traced a = Traced { unTraced :: IO ([String], Either String a) }

120+

121+instance MonadS Traced where

122+ returnS x = Traced $ Prelude.return ([], Right x)

123+ x `bindS` f = Traced $ unTraced x Prelude.>>= \(trace, result) ->

124+ case result of

125+ Left err -> Prelude.return (trace, Left err)

126+ Right a -> unTraced (f a) Prelude.>>= \(trace', result') ->

127+ Prelude.return (trace ++ show a : trace', result')

128+ x `seqS` y = Traced $ unTraced x Prelude.>>= \(trace, result) ->

129+ case result of

130+ Left err -> Prelude.return (trace, Left err)

131+ Right a -> unTraced y Prelude.>>= \(trace', result') ->

132+ Prelude.return (trace ++ trace', result')

133+ failS err = Traced $ Prelude.return ([], Left err)

134+

135+runTraced :: Traced a -> IO a

136+runTraced (Traced x) = x Prelude.>>= \(trace, ma) ->

137+ case ma of

138+ Left err -> Prelude.fail $ err ++ "\nTrace:\n" ++ showTrace trace

139+ Right a -> Prelude.return a

140+

141+showTrace :: [String] -> String

142+showTrace = unlines . map (\xs -> ' ' : ' ' : xs)

143+

144+guard :: MonadS m => Bool -> m ()

145+guard True = returnS ()

146+guard False = failS "Guard failed"

147+

148+(>>=) :: (MonadS m, Show a) => m a -> (a -> m b) -> m b

149+(>>=) = bindS

150+

151+(>>) :: (MonadS m, Show a) => m a -> m b -> m b

152+(>>) = seqS

153+

154+return :: MonadS m => a -> m a

155+return = returnS

156+

157+fail :: MonadS m => String -> m a

158+fail = failS

159+

160+liftIO :: IO a -> Traced a

161+liftIO x = Traced $ x Prelude.>>= \a -> Prelude.return ([], Right a)