diff --git a/ChangeLog b/ChangeLog new file mode 100644 --- /dev/null +++ b/ChangeLog @@ -0,0 +1,34 @@ +2014-05-30 Tim Watson 0.4.0 + +* Update dependencies + +2012-10-19 Edsko de Vries 0.3.1 + +* Bugfix. Reconnecting between endpoints did not work under certain +circumstances. + +2012-10-03 Edsko de Vries 0.3.0 + +* Implement new disconnection semantics +* Make 'connect' asynchronous (sender allocated connection IDs) +* Fix distributed deadlock +* Optimize treatment of crossed connection requests +* Relax upper bound on network +* Fix memory leaks + +2012-08-20 Edsko de Vries 0.2.0.3 + +* Allow for colons in hostnames (for IPv6) + +2012-07-16 Edsko de Vries 0.2.0.2 + +* Base 4.6 compatibility +* Relax package contraints on bytestring and containers + +2012-07-12 Edsko de Vries 0.2.0.1 + +* Fix bug in recvExact + +2012-07-07 Edsko de Vries 0.2.0 + +* Initial release. diff --git a/network-transport-tcp.cabal b/network-transport-tcp.cabal --- a/network-transport-tcp.cabal +++ b/network-transport-tcp.cabal @@ -1,24 +1,24 @@ Name: network-transport-tcp -Version: 0.3.1 +Version: 0.4.0 Cabal-Version: >=1.8 Build-Type: Simple License: BSD3 License-file: LICENSE Copyright: Well-Typed LLP Author: Duncan Coutts, Nicolas Wu, Edsko de Vries -Maintainer: edsko@well-typed.com, dcoutts@well-typed.com +Maintainer: edsko@well-typed.com, duncan@well-typed.com, watson.timothy@gmail.com Stability: experimental -Homepage: http://github.com/haskell-distributed/distributed-process -Bug-Reports: mailto:edsko@well-typed.com +Homepage: http://haskell-distributed.github.com +Bug-Reports: https://cloud-haskell.atlassian.net/browse/NTTCP Synopsis: TCP instantiation of Network.Transport Description: TCP instantiation of Network.Transport Tested-With: GHC==7.0.4 GHC==7.2.2 GHC==7.4.1 GHC==7.4.2 -Category: Network +Category: Network +extra-source-files: ChangeLog Source-Repository head Type: git - Location: https://github.com/haskell-distributed/distributed-process - SubDir: network-transport-tcp + Location: https://github.com/haskell-distributed/network-transport-tcp Flag use-mock-network Description: Use mock network implementation (for testing) @@ -26,7 +26,7 @@ Library Build-Depends: base >= 4.3 && < 5, - network-transport >= 0.3 && < 0.4, + network-transport >= 0.4.0.0 && < 0.5, data-accessor >= 0.2 && < 0.3, containers >= 0.4 && < 0.6, bytestring >= 0.9 && < 0.11, @@ -47,8 +47,8 @@ Build-Depends: base >= 4.3 && < 5, network-transport-tests >= 0.1.0.1 && < 0.2, network >= 2.3 && < 2.5, - network-transport >= 0.3 && < 0.4, - network-transport-tcp >= 0.3 && < 0.4 + network-transport >= 0.4.0.0 && < 0.5, + network-transport-tcp >= 0.3 && < 0.5 ghc-options: -threaded -rtsopts -with-rtsopts=-N HS-Source-Dirs: tests Extensions: CPP, @@ -59,22 +59,25 @@ Test-Suite TestQC Type: exitcode-stdio-1.0 Main-Is: TestQC.hs - Build-Depends: base >= 4.3 && < 5, - test-framework, - test-framework-quickcheck2, - test-framework-hunit, - QuickCheck, - HUnit, - network-transport, - network-transport-tcp, - containers, - bytestring, - pretty, - data-accessor, - data-accessor-transformers, - mtl, - transformers, - lockfree-queue + If flag(use-mock-network) + Build-Depends: base >= 4.3 && < 5, + test-framework, + test-framework-quickcheck2, + test-framework-hunit, + QuickCheck, + HUnit, + network-transport, + network-transport-tcp, + containers, + bytestring, + pretty, + data-accessor, + data-accessor-transformers, + mtl, + transformers, + lockfree-queue + Else + Buildable: False ghc-options: -threaded -Wall -fno-warn-orphans HS-Source-Dirs: tests Extensions: TypeSynonymInstances @@ -82,3 +85,4 @@ OverlappingInstances OverloadedStrings DeriveDataTypeable + MultiParamTypeClasses diff --git a/src/Network/Transport/TCP.hs b/src/Network/Transport/TCP.hs --- a/src/Network/Transport/TCP.hs +++ b/src/Network/Transport/TCP.hs @@ -1,22 +1,22 @@ --- | TCP implementation of the transport layer. --- +-- | TCP implementation of the transport layer. +-- -- The TCP implementation guarantees that only a single TCP connection (socket) -- will be used between endpoints, provided that the addresses specified are -- canonical. If /A/ connects to /B/ and reports its address as -- @192.168.0.1:8080@ and /B/ subsequently connects tries to connect to /A/ as -- @client1.local:http-alt@ then the transport layer will not realize that the --- TCP connection can be reused. +-- TCP connection can be reused. -- -- Applications that use the TCP transport should use -- 'Network.Socket.withSocketsDo' in their main function for Windows -- compatibility (see "Network.Socket"). -module Network.Transport.TCP +module Network.Transport.TCP ( -- * Main API createTransport , TCPParameters(..) , defaultTCPParameters - -- * Internals (exposed for unit tests) - , createTransportExposeInternals + -- * Internals (exposed for unit tests) + , createTransportExposeInternals , TransportInternals(..) , EndPointId , encodeEndPointAddress @@ -25,27 +25,27 @@ , ConnectionRequestResponse(..) , firstNonReservedLightweightConnectionId , firstNonReservedHeavyweightConnectionId - , socketToEndPoint + , socketToEndPoint , LightweightConnectionId -- * Design notes -- $design ) where -import Prelude hiding +import Prelude hiding ( mapM_ #if ! MIN_VERSION_base(4,6,0) , catch #endif ) - + import Network.Transport -import Network.Transport.TCP.Internal +import Network.Transport.TCP.Internal ( forkServer , recvWithLength , recvInt32 , tryCloseSocket ) -import Network.Transport.Internal +import Network.Transport.Internal ( encodeInt32 , prependLength , mapIOException @@ -59,7 +59,7 @@ #ifdef USE_MOCK_NETWORK import qualified Network.Transport.TCP.Mock.Socket as N #else -import qualified Network.Socket as N +import qualified Network.Socket as N #endif ( HostName , ServiceName @@ -71,7 +71,7 @@ , SocketType(Stream) , defaultProtocol , setSocketOption - , SocketOption(ReuseAddr) + , SocketOption(ReuseAddr) , connect , sOMAXCONN , AddrInfo @@ -85,7 +85,7 @@ import Control.Concurrent (forkIO, ThreadId, killThread, myThreadId) import Control.Concurrent.Chan (Chan, newChan, readChan, writeChan) -import Control.Concurrent.MVar +import Control.Concurrent.MVar ( MVar , newMVar , modifyMVar @@ -98,7 +98,7 @@ import Control.Category ((>>>)) import Control.Applicative ((<$>)) import Control.Monad (when, unless, join) -import Control.Exception +import Control.Exception ( IOException , SomeException , AsyncException @@ -117,7 +117,7 @@ import Data.Bits (shiftL, (.|.)) import Data.Word (Word32) import Data.Set (Set) -import qualified Data.Set as Set +import qualified Data.Set as Set ( empty , insert , elems @@ -128,11 +128,11 @@ ) import Data.Map (Map) import qualified Data.Map as Map (empty) -import Data.Accessor (Accessor, accessor, (^.), (^=), (^:)) +import Data.Accessor (Accessor, accessor, (^.), (^=), (^:)) import qualified Data.Accessor.Container as DAC (mapMaybe) import Data.Foldable (forM_, mapM_) --- $design +-- $design -- -- [Goals] -- @@ -146,7 +146,7 @@ -- > | \~~~|~~~~~~~~~~~~~~~~~~~~~~~~~<| | -- > | |==========================| | -- > +-------+ +-------+ --- +-- -- Ignoring the complications detailed below, the TCP connection is set up is -- when the first lightweight connection is created (in either direction), and -- torn down when the last lightweight connection (in either direction) is @@ -177,7 +177,7 @@ -- with /B/'s connection request completes. -- -- [Disconnecting] --- +-- -- The TCP connection is created as soon as the first logical connection from -- /A/ to /B/ (or /B/ to /A/) is established. At this point a thread (@#@) is -- spawned that listens for incoming connections from /B/: @@ -193,21 +193,21 @@ -- The question is when the TCP connection can be closed again. Conceptually, -- we want to do reference counting: when there are no logical connections left -- between /A/ and /B/ we want to close the socket (possibly after some --- timeout). +-- timeout). -- -- However, /A/ and /B/ need to agree that the refcount has reached zero. It -- might happen that /B/ sends a connection request over the existing socket at -- the same time that /A/ closes its logical connection to /B/ and closes the -- socket. This will cause a failure in /B/ (which will have to retry) which is -- not caused by a network failure, which is unfortunate. (Note that the --- connection request from /B/ might succeed even if /A/ closes the socket.) +-- connection request from /B/ might succeed even if /A/ closes the socket.) -- -- Instead, when /A/ is ready to close the socket it sends a 'CloseSocket' -- request to /B/ and records that its connection to /B/ is closing. If /A/ -- receives a new connection request from /B/ after having sent the -- 'CloseSocket' request it simply forgets that it sent a 'CloseSocket' request -- and increments the reference count of the connection again. --- +-- -- When /B/ receives a 'CloseSocket' message and it too is ready to close the -- connection, it will respond with a reciprocal 'CloseSocket' request to /A/ -- and then actually close the socket. /A/ meanwhile will not send any more @@ -216,11 +216,11 @@ -- from /B/. (Since /A/ recorded that its connection to /B/ is in closing state -- after sending a 'CloseSocket' request to /B/, it knows not to reciprocate /B/ -- reciprocal 'CloseSocket' message.) --- +-- -- If there is a concurrent thread in /A/ waiting to connect to /B/ after /A/ -- has sent a 'CloseSocket' request then this thread will block until /A/ knows -- whether to reuse the old socket (if /B/ sends a new connection request --- instead of acknowledging the 'CloseSocket') or to set up a new socket. +-- instead of acknowledging the 'CloseSocket') or to set up a new socket. -------------------------------------------------------------------------------- -- Internal datatypes -- @@ -229,30 +229,30 @@ -- We use underscores for fields that we might update (using accessors) -- -- All data types follow the same structure: --- +-- -- * A top-level data type describing static properties (TCPTransport, -- LocalEndPoint, RemoteEndPoint) -- * The 'static' properties include an MVar containing a data structure for -- the dynamic properties (TransportState, LocalEndPointState, --- RemoteEndPointState). The state could be invalid/valid/closed,/etc. +-- RemoteEndPointState). The state could be invalid/valid/closed,/etc. -- * For the case of "valid" we use third data structure to give more details -- about the state (ValidTransportState, ValidLocalEndPointState, -- ValidRemoteEndPointState). -data TCPTransport = TCPTransport +data TCPTransport = TCPTransport { transportHost :: !N.HostName , transportPort :: !N.ServiceName , transportState :: !(MVar TransportState) , transportParams :: !TCPParameters } -data TransportState = +data TransportState = TransportValid !ValidTransportState | TransportClosed -data ValidTransportState = ValidTransportState +data ValidTransportState = ValidTransportState { _localEndPoints :: !(Map EndPointAddress LocalEndPoint) - , _nextEndPointId :: !EndPointId + , _nextEndPointId :: !EndPointId } data LocalEndPoint = LocalEndPoint @@ -261,12 +261,12 @@ , localState :: !(MVar LocalEndPointState) } -data LocalEndPointState = +data LocalEndPointState = LocalEndPointValid !ValidLocalEndPointState | LocalEndPointClosed -data ValidLocalEndPointState = ValidLocalEndPointState - { -- Next available ID for an outgoing lightweight self-connection +data ValidLocalEndPointState = ValidLocalEndPointState + { -- Next available ID for an outgoing lightweight self-connection -- (see also remoteNextConnOutId) _localNextConnOutId :: !LightweightConnectionId -- Next available ID for an incoming heavyweight connection @@ -275,7 +275,7 @@ , _localConnections :: !(Map EndPointAddress RemoteEndPoint) } --- REMOTE ENDPOINTS +-- REMOTE ENDPOINTS -- -- Remote endpoints (basically, TCP connections) have the following lifecycle: -- @@ -302,7 +302,7 @@ -- createConnectionTo when we fail to connect. -- -- Valid: This is the "normal" state for a working remote endpoint. --- +-- -- Closing: When we detect that a remote endpoint is no longer used, we send a -- CloseSocket request across the connection and put the remote endpoint in -- closing state. As with Init, 'Closing' carries an MVar () 'resolved' which @@ -335,16 +335,16 @@ -- localConnections it is guaranteed to either find a different remote -- endpoint, or else none at all (if we don't insist in this order some -- threads might start spinning). --- +-- -- INV-RESOLVE: We should only signal on 'resolved' while the remote endpoint is -- locked, and the remote endpoint must be in Valid or Closed state once -- unlocked. This guarantees that there will not be two threads attempting to --- both signal on 'resolved'. +-- both signal on 'resolved'. -- -- INV-LOST: If a send or recv fails, or a socket is closed unexpectedly, we -- first put the remote endpoint in Closed state, and then send a -- EventConnectionLost event. This guarantees that we only send this event --- once. +-- once. -- -- INV-CLOSING: An endpoint in closing state is for all intents and purposes -- closed; that is, we shouldn't do any 'send's on it (although 'recv' is @@ -357,23 +357,23 @@ -- particular, it is okay to call removeRemoteEndPoint inside -- modifyRemoteState. -data RemoteEndPoint = RemoteEndPoint +data RemoteEndPoint = RemoteEndPoint { remoteAddress :: !EndPointAddress , remoteState :: !(MVar RemoteState) , remoteId :: !HeavyweightConnectionId , remoteScheduled :: !(Chan (IO ())) } -data RequestedBy = RequestedByUs | RequestedByThem +data RequestedBy = RequestedByUs | RequestedByThem deriving (Eq, Show) data RemoteState = -- | Invalid remote endpoint (for example, invalid address) RemoteEndPointInvalid !(TransportError ConnectErrorCode) -- | The remote endpoint is being initialized - | RemoteEndPointInit !(MVar ()) !(MVar ()) !RequestedBy + | RemoteEndPointInit !(MVar ()) !(MVar ()) !RequestedBy -- | "Normal" working endpoint - | RemoteEndPointValid !ValidRemoteEndPointState + | RemoteEndPointValid !ValidRemoteEndPointState -- | The remote endpoint is being closed (garbage collected) | RemoteEndPointClosing !(MVar ()) !ValidRemoteEndPointState -- | The remote endpoint has been closed (garbage collected) @@ -382,12 +382,12 @@ -- using a closeTransport or closeEndPoint API call | RemoteEndPointFailed !IOException --- TODO: we might want to replace Set (here and elsewhere) by faster +-- TODO: we might want to replace Set (here and elsewhere) by faster -- containers -- -- TODO: we could get rid of 'remoteIncoming' (and maintain less state) if --- we introduce a new event 'AllConnectionsClosed' -data ValidRemoteEndPointState = ValidRemoteEndPointState +-- we introduce a new event 'AllConnectionsClosed' +data ValidRemoteEndPointState = ValidRemoteEndPointState { _remoteOutgoing :: !Int , _remoteIncoming :: !(Set LightweightConnectionId) , _remoteMaxIncoming :: !LightweightConnectionId @@ -414,24 +414,24 @@ -- 'LightweightConnectionId'. type HeavyweightConnectionId = Word32 --- | Control headers -data ControlHeader = +-- | Control headers +data ControlHeader = -- | Tell the remote endpoint that we created a new connection - CreatedNewConnection + CreatedNewConnection -- | Tell the remote endpoint we will no longer be using a connection - | CloseConnection + | CloseConnection -- | Request to close the connection (see module description) - | CloseSocket + | CloseSocket deriving (Enum, Bounded, Show) -- | Response sent by /B/ to /A/ when /A/ tries to connect data ConnectionRequestResponse = -- | /B/ accepts the connection - ConnectionRequestAccepted + ConnectionRequestAccepted -- | /A/ requested an invalid endpoint - | ConnectionRequestInvalid + | ConnectionRequestInvalid -- | /A/s request crossed with a request from /B/ (see protocols) - | ConnectionRequestCrossed + | ConnectionRequestCrossed deriving (Enum, Bounded, Show) -- | Parameters for setting up the TCP transport @@ -439,7 +439,7 @@ -- | Backlog for 'listen'. -- Defaults to SOMAXCONN. tcpBacklog :: Int - -- | Should we set SO_REUSEADDR on the server socket? + -- | Should we set SO_REUSEADDR on the server socket? -- Defaults to True. , tcpReuseServerAddr :: Bool -- | Should we set SO_REUSEADDR on client sockets? @@ -448,13 +448,13 @@ } -- | Internal functionality we expose for unit testing -data TransportInternals = TransportInternals +data TransportInternals = TransportInternals { -- | The ID of the thread that listens for new incoming connections transportThread :: ThreadId -- | Find the socket between a local and a remote endpoint - , socketBetween :: EndPointAddress - -> EndPointAddress - -> IO N.Socket + , socketBetween :: EndPointAddress + -> EndPointAddress + -> IO N.Socket } -------------------------------------------------------------------------------- @@ -462,62 +462,62 @@ -------------------------------------------------------------------------------- -- | Create a TCP transport -createTransport :: N.HostName - -> N.ServiceName +createTransport :: N.HostName + -> N.ServiceName -> TCPParameters -> IO (Either IOException Transport) -createTransport host port params = +createTransport host port params = either Left (Right . fst) <$> createTransportExposeInternals host port params -- | You should probably not use this function (used for unit testing only) -createTransportExposeInternals - :: N.HostName - -> N.ServiceName +createTransportExposeInternals + :: N.HostName + -> N.ServiceName -> TCPParameters - -> IO (Either IOException (Transport, TransportInternals)) -createTransportExposeInternals host port params = do - state <- newMVar . TransportValid $ ValidTransportState - { _localEndPoints = Map.empty - , _nextEndPointId = 0 + -> IO (Either IOException (Transport, TransportInternals)) +createTransportExposeInternals host port params = do + state <- newMVar . TransportValid $ ValidTransportState + { _localEndPoints = Map.empty + , _nextEndPointId = 0 } let transport = TCPTransport { transportState = state , transportHost = host , transportPort = port , transportParams = params } - tryIO $ bracketOnError (forkServer - host - port - (tcpBacklog params) + tryIO $ bracketOnError (forkServer + host + port + (tcpBacklog params) (tcpReuseServerAddr params) - (terminationHandler transport) + (terminationHandler transport) (handleConnectionRequest transport)) killThread (mkTransport transport) where - mkTransport :: TCPTransport - -> ThreadId + mkTransport :: TCPTransport + -> ThreadId -> IO (Transport, TransportInternals) mkTransport transport tid = return - ( Transport - { newEndPoint = apiNewEndPoint transport + ( Transport + { newEndPoint = apiNewEndPoint transport , closeTransport = let evs = [ EndPointClosed , throw $ userError "Transport closed" ] in - apiCloseTransport transport (Just tid) evs - } - , TransportInternals + apiCloseTransport transport (Just tid) evs + } + , TransportInternals { transportThread = tid , socketBetween = internalSocketBetween transport } ) terminationHandler :: TCPTransport -> SomeException -> IO () - terminationHandler transport ex = do + terminationHandler transport ex = do let evs = [ ErrorEvent (TransportError EventTransportFailed (show ex)) - , throw $ userError "Transport closed" + , throw $ userError "Transport closed" ] - apiCloseTransport transport Nothing evs + apiCloseTransport transport Nothing evs -- | Default TCP parameters defaultTCPParameters :: TCPParameters @@ -533,39 +533,39 @@ -- | Close the transport apiCloseTransport :: TCPTransport -> Maybe ThreadId -> [Event] -> IO () -apiCloseTransport transport mTransportThread evs = +apiCloseTransport transport mTransportThread evs = asyncWhenCancelled return $ do mTSt <- modifyMVar (transportState transport) $ \st -> case st of TransportValid vst -> return (TransportClosed, Just vst) TransportClosed -> return (TransportClosed, Nothing) - forM_ mTSt $ mapM_ (apiCloseEndPoint transport evs) . (^. localEndPoints) + forM_ mTSt $ mapM_ (apiCloseEndPoint transport evs) . (^. localEndPoints) -- This will invoke the termination handler, which in turn will call -- apiCloseTransport again, but then the transport will already be closed -- and we won't be passed a transport thread, so we terminate immmediate - forM_ mTransportThread killThread - --- | Create a new endpoint -apiNewEndPoint :: TCPTransport + forM_ mTransportThread killThread + +-- | Create a new endpoint +apiNewEndPoint :: TCPTransport -> IO (Either (TransportError NewEndPointErrorCode) EndPoint) -apiNewEndPoint transport = +apiNewEndPoint transport = try . asyncWhenCancelled closeEndPoint $ do ourEndPoint <- createLocalEndPoint transport - return EndPoint + return EndPoint { receive = readChan (localChannel ourEndPoint) , address = localAddress ourEndPoint - , connect = apiConnect (transportParams transport) ourEndPoint + , connect = apiConnect (transportParams transport) ourEndPoint , closeEndPoint = let evs = [ EndPointClosed - , throw $ userError "Endpoint closed" + , throw $ userError "Endpoint closed" ] in apiCloseEndPoint transport evs ourEndPoint - , newMulticastGroup = return . Left $ newMulticastGroupError + , newMulticastGroup = return . Left $ newMulticastGroupError , resolveMulticastGroup = return . Left . const resolveMulticastGroupError } where - newMulticastGroupError = - TransportError NewMulticastGroupUnsupported "Multicast not supported" - resolveMulticastGroupError = - TransportError ResolveMulticastGroupUnsupported "Multicast not supported" + newMulticastGroupError = + TransportError NewMulticastGroupUnsupported "Multicast not supported" + resolveMulticastGroupError = + TransportError ResolveMulticastGroupUnsupported "Multicast not supported" -- | Connnect to an endpoint apiConnect :: TCPParameters -- ^ Parameters @@ -575,35 +575,35 @@ -> ConnectHints -- ^ Hints -> IO (Either (TransportError ConnectErrorCode) Connection) apiConnect params ourEndPoint theirAddress _reliability hints = - try . asyncWhenCancelled close $ - if localAddress ourEndPoint == theirAddress - then connectToSelf ourEndPoint + try . asyncWhenCancelled close $ + if localAddress ourEndPoint == theirAddress + then connectToSelf ourEndPoint else do resetIfBroken ourEndPoint theirAddress - (theirEndPoint, connId) <- + (theirEndPoint, connId) <- createConnectionTo params ourEndPoint theirAddress hints -- connAlive can be an IORef rather than an MVar because it is protected -- by the remoteState MVar. We don't need the overhead of locking twice. connAlive <- newIORef True - return Connection - { send = apiSend (ourEndPoint, theirEndPoint) connId connAlive - , close = apiClose (ourEndPoint, theirEndPoint) connId connAlive + return Connection + { send = apiSend (ourEndPoint, theirEndPoint) connId connAlive + , close = apiClose (ourEndPoint, theirEndPoint) connId connAlive } -- | Close a connection apiClose :: EndPointPair -> LightweightConnectionId -> IORef Bool -> IO () -apiClose (ourEndPoint, theirEndPoint) connId connAlive = - void . tryIO . asyncWhenCancelled return $ do +apiClose (ourEndPoint, theirEndPoint) connId connAlive = + void . tryIO . asyncWhenCancelled return $ do mAct <- modifyMVar (remoteState theirEndPoint) $ \st -> case st of RemoteEndPointValid vst -> do alive <- readIORef connAlive - if alive + if alive then do writeIORef connAlive False - act <- schedule theirEndPoint $ - sendOn vst [encodeInt32 CloseConnection, encodeInt32 connId] - return ( RemoteEndPointValid - . (remoteOutgoing ^: (\x -> x - 1)) + act <- schedule theirEndPoint $ + sendOn vst [encodeInt32 CloseConnection, encodeInt32 connId] + return ( RemoteEndPointValid + . (remoteOutgoing ^: (\x -> x - 1)) $ vst , Just act ) @@ -616,12 +616,12 @@ -- | Send data across a connection -apiSend :: EndPointPair -- ^ Local and remote endpoint +apiSend :: EndPointPair -- ^ Local and remote endpoint -> LightweightConnectionId -- ^ Connection ID -> IORef Bool -- ^ Is the connection still alive? -> [ByteString] -- ^ Payload -> IO (Either (TransportError SendErrorCode) ()) -apiSend (ourEndPoint, theirEndPoint) connId connAlive payload = +apiSend (ourEndPoint, theirEndPoint) connId connAlive payload = -- We don't need the overhead of asyncWhenCancelled here try . mapIOException sendFailed $ do act <- withMVar (remoteState theirEndPoint) $ \st -> case st of @@ -631,70 +631,70 @@ relyViolation (ourEndPoint, theirEndPoint) "apiSend" RemoteEndPointValid vst -> do alive <- readIORef connAlive - if alive - then schedule theirEndPoint $ + if alive + then schedule theirEndPoint $ sendOn vst (encodeInt32 connId : prependLength payload) else throwIO $ TransportError SendClosed "Connection closed" - RemoteEndPointClosing _ _ -> do + RemoteEndPointClosing _ _ -> do alive <- readIORef connAlive - if alive - then relyViolation (ourEndPoint, theirEndPoint) "apiSend" + if alive + then relyViolation (ourEndPoint, theirEndPoint) "apiSend" else throwIO $ TransportError SendClosed "Connection closed" RemoteEndPointClosed -> do alive <- readIORef connAlive - if alive - then relyViolation (ourEndPoint, theirEndPoint) "apiSend" + if alive + then relyViolation (ourEndPoint, theirEndPoint) "apiSend" else throwIO $ TransportError SendClosed "Connection closed" RemoteEndPointFailed err -> do alive <- readIORef connAlive - if alive - then throwIO $ TransportError SendFailed (show err) + if alive + then throwIO $ TransportError SendFailed (show err) else throwIO $ TransportError SendClosed "Connection closed" runScheduledAction (ourEndPoint, theirEndPoint) act where sendFailed = TransportError SendFailed . show -- | Force-close the endpoint -apiCloseEndPoint :: TCPTransport -- ^ Transport - -> [Event] -- ^ Events used to report closure +apiCloseEndPoint :: TCPTransport -- ^ Transport + -> [Event] -- ^ Events used to report closure -> LocalEndPoint -- ^ Local endpoint -> IO () apiCloseEndPoint transport evs ourEndPoint = asyncWhenCancelled return $ do -- Remove the reference from the transport state removeLocalEndPoint transport ourEndPoint - -- Close the local endpoint + -- Close the local endpoint mOurState <- modifyMVar (localState ourEndPoint) $ \st -> case st of - LocalEndPointValid vst -> + LocalEndPointValid vst -> return (LocalEndPointClosed, Just vst) LocalEndPointClosed -> return (LocalEndPointClosed, Nothing) forM_ mOurState $ \vst -> do forM_ (vst ^. localConnections) tryCloseRemoteSocket - forM_ evs $ writeChan (localChannel ourEndPoint) + forM_ evs $ writeChan (localChannel ourEndPoint) where -- Close the remote socket and return the set of all incoming connections - tryCloseRemoteSocket :: RemoteEndPoint -> IO () + tryCloseRemoteSocket :: RemoteEndPoint -> IO () tryCloseRemoteSocket theirEndPoint = do - -- We make an attempt to close the connection nicely + -- We make an attempt to close the connection nicely -- (by sending a CloseSocket first) let closed = RemoteEndPointFailed . userError $ "apiCloseEndPoint" mAct <- modifyMVar (remoteState theirEndPoint) $ \st -> case st of - RemoteEndPointInvalid _ -> + RemoteEndPointInvalid _ -> return (st, Nothing) RemoteEndPointInit resolved _ _ -> do putMVar resolved () return (closed, Nothing) - RemoteEndPointValid vst -> do + RemoteEndPointValid vst -> do act <- schedule theirEndPoint $ do tryIO $ sendOn vst [ encodeInt32 CloseSocket , encodeInt32 (vst ^. remoteMaxIncoming) ] tryCloseSocket (remoteSocket vst) return (closed, Just act) - RemoteEndPointClosing resolved vst -> do + RemoteEndPointClosing resolved vst -> do putMVar resolved () act <- schedule theirEndPoint $ tryCloseSocket (remoteSocket vst) return (closed, Just act) @@ -703,8 +703,8 @@ RemoteEndPointFailed err -> return (RemoteEndPointFailed err, Nothing) forM_ mAct $ runScheduledAction (ourEndPoint, theirEndPoint) - + -------------------------------------------------------------------------------- -- Incoming requests -- -------------------------------------------------------------------------------- @@ -720,11 +720,11 @@ -- the transport down. We must be careful to close the socket when a (possibly -- asynchronous, ThreadKilled) exception occurs. (If an exception escapes from -- handleConnectionRequest the transport will be shut down.) -handleConnectionRequest :: TCPTransport -> N.Socket -> IO () -handleConnectionRequest transport sock = handle handleException $ do +handleConnectionRequest :: TCPTransport -> N.Socket -> IO () +handleConnectionRequest transport sock = handle handleException $ do ourEndPointId <- recvInt32 sock - theirAddress <- EndPointAddress . BS.concat <$> recvWithLength sock - let ourAddress = encodeEndPointAddress (transportHost transport) + theirAddress <- EndPointAddress . BS.concat <$> recvWithLength sock + let ourAddress = encodeEndPointAddress (transportHost transport) (transportPort transport) ourEndPointId ourEndPoint <- withMVar (transportState transport) $ \st -> case st of @@ -735,32 +735,32 @@ throwIO $ userError "handleConnectionRequest: Invalid endpoint" Just ourEndPoint -> return ourEndPoint - TransportClosed -> + TransportClosed -> throwIO $ userError "Transport closed" void . forkIO $ go ourEndPoint theirAddress where go :: LocalEndPoint -> EndPointAddress -> IO () - go ourEndPoint theirAddress = do + go ourEndPoint theirAddress = do -- This runs in a thread that will never be killed - mEndPoint <- handle ((>> return Nothing) . handleException) $ do + mEndPoint <- handle ((>> return Nothing) . handleException) $ do resetIfBroken ourEndPoint theirAddress - (theirEndPoint, isNew) <- + (theirEndPoint, isNew) <- findRemoteEndPoint ourEndPoint theirAddress RequestedByThem - - if not isNew + + if not isNew then do tryIO $ sendMany sock [encodeInt32 ConnectionRequestCrossed] tryCloseSocket sock - return Nothing + return Nothing else do sendLock <- newMVar () - let vst = ValidRemoteEndPointState + let vst = ValidRemoteEndPointState { remoteSocket = sock , remoteSendLock = sendLock , _remoteOutgoing = 0 , _remoteIncoming = Set.empty , _remoteMaxIncoming = 0 - , _remoteNextConnOutId = firstNonReservedLightweightConnectionId + , _remoteNextConnOutId = firstNonReservedLightweightConnectionId } sendMany sock [encodeInt32 ConnectionRequestAccepted] resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst) @@ -769,30 +769,30 @@ -- Nothing then the socket is already closed; otherwise, the socket has -- been recorded as part of the remote endpoint. Either way, we no longer -- have to worry about closing the socket on receiving an asynchronous - -- exception from this point forward. + -- exception from this point forward. forM_ mEndPoint $ handleIncomingMessages . (,) ourEndPoint handleException :: SomeException -> IO () - handleException ex = do - tryCloseSocket sock + handleException ex = do + tryCloseSocket sock rethrowIfAsync (fromException ex) - rethrowIfAsync :: Maybe AsyncException -> IO () - rethrowIfAsync = mapM_ throwIO + rethrowIfAsync :: Maybe AsyncException -> IO () + rethrowIfAsync = mapM_ throwIO -- | Handle requests from a remote endpoint. --- +-- -- Returns only if the remote party closes the socket or if an error occurs. -- This runs in a thread that will never be killed. -handleIncomingMessages :: EndPointPair -> IO () +handleIncomingMessages :: EndPointPair -> IO () handleIncomingMessages (ourEndPoint, theirEndPoint) = do mSock <- withMVar theirState $ \st -> case st of RemoteEndPointInvalid _ -> - relyViolation (ourEndPoint, theirEndPoint) - "handleIncomingMessages (invalid)" + relyViolation (ourEndPoint, theirEndPoint) + "handleIncomingMessages (invalid)" RemoteEndPointInit _ _ _ -> - relyViolation (ourEndPoint, theirEndPoint) + relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages (init)" RemoteEndPointValid ep -> return . Just $ remoteSocket ep @@ -800,13 +800,13 @@ return . Just $ remoteSocket ep RemoteEndPointClosed -> return Nothing - RemoteEndPointFailed _ -> + RemoteEndPointFailed _ -> return Nothing - - forM_ mSock $ \sock -> + + forM_ mSock $ \sock -> tryIO (go sock) >>= either (prematureExit sock) return where - -- Dispatch + -- Dispatch -- -- If a recv throws an exception this will be caught top-level and -- 'prematureExit' will be invoked. The same will happen if the remote @@ -817,38 +817,38 @@ go :: N.Socket -> IO () go sock = do lcid <- recvInt32 sock :: IO LightweightConnectionId - if lcid >= firstNonReservedLightweightConnectionId + if lcid >= firstNonReservedLightweightConnectionId then do - readMessage sock lcid + readMessage sock lcid go sock - else + else case tryToEnum (fromIntegral lcid) of Just CreatedNewConnection -> do - recvInt32 sock >>= createdNewConnection - go sock + recvInt32 sock >>= createdNewConnection + go sock Just CloseConnection -> do - recvInt32 sock >>= closeConnection + recvInt32 sock >>= closeConnection go sock - Just CloseSocket -> do - didClose <- recvInt32 sock >>= closeSocket sock + Just CloseSocket -> do + didClose <- recvInt32 sock >>= closeSocket sock unless didClose $ go sock Nothing -> throwIO $ userError "Invalid control request" - + -- Create a new connection - createdNewConnection :: LightweightConnectionId -> IO () + createdNewConnection :: LightweightConnectionId -> IO () createdNewConnection lcid = do modifyMVar_ theirState $ \st -> do vst <- case st of RemoteEndPointInvalid _ -> - relyViolation (ourEndPoint, theirEndPoint) + relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:createNewConnection (invalid)" RemoteEndPointInit _ _ _ -> - relyViolation (ourEndPoint, theirEndPoint) + relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:createNewConnection (init)" RemoteEndPointValid vst -> return ( (remoteIncoming ^: Set.insert lcid) - $ (remoteMaxIncoming ^= lcid) + $ (remoteMaxIncoming ^= lcid) vst ) RemoteEndPointClosing resolved vst -> do @@ -863,36 +863,36 @@ . (remoteMaxIncoming ^= lcid) $ vst ) - RemoteEndPointFailed err -> + RemoteEndPointFailed err -> throwIO err RemoteEndPointClosed -> - relyViolation (ourEndPoint, theirEndPoint) + relyViolation (ourEndPoint, theirEndPoint) "createNewConnection (closed)" return (RemoteEndPointValid vst) - writeChan ourChannel (ConnectionOpened (connId lcid) ReliableOrdered theirAddr) + writeChan ourChannel (ConnectionOpened (connId lcid) ReliableOrdered theirAddr) - -- Close a connection + -- Close a connection -- It is important that we verify that the connection is in fact open, -- because otherwise we should not decrement the reference count - closeConnection :: LightweightConnectionId -> IO () + closeConnection :: LightweightConnectionId -> IO () closeConnection lcid = do modifyMVar_ theirState $ \st -> case st of RemoteEndPointInvalid _ -> relyViolation (ourEndPoint, theirEndPoint) "closeConnection (invalid)" RemoteEndPointInit _ _ _ -> relyViolation (ourEndPoint, theirEndPoint) "closeConnection (init)" - RemoteEndPointValid vst -> do - unless (Set.member lcid (vst ^. remoteIncoming)) $ + RemoteEndPointValid vst -> do + unless (Set.member lcid (vst ^. remoteIncoming)) $ throwIO $ userError "Invalid CloseConnection" - return ( RemoteEndPointValid - . (remoteIncoming ^: Set.delete lcid) + return ( RemoteEndPointValid + . (remoteIncoming ^: Set.delete lcid) $ vst ) RemoteEndPointClosing _ _ -> -- If the remote endpoint is in Closing state, that means that are as -- far as we are concerned there are no incoming connections. This -- means that a CloseConnection request at this point is invalid. - throwIO $ userError "Invalid CloseConnection request" + throwIO $ userError "Invalid CloseConnection request" RemoteEndPointFailed err -> throwIO err RemoteEndPointClosed -> @@ -900,7 +900,7 @@ writeChan ourChannel (ConnectionClosed (connId lcid)) -- Close the socket (if we don't have any outgoing connections) - closeSocket :: N.Socket -> LightweightConnectionId -> IO Bool + closeSocket :: N.Socket -> LightweightConnectionId -> IO Bool closeSocket sock lastReceivedId = do mAct <- modifyMVar theirState $ \st -> do case st of @@ -908,38 +908,38 @@ relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:closeSocket (invalid)" RemoteEndPointInit _ _ _ -> - relyViolation (ourEndPoint, theirEndPoint) + relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:closeSocket (init)" RemoteEndPointValid vst -> do -- We regard a CloseSocket message as an (optimized) way for the -- remote endpoint to indicate that all its connections to us are -- now properly closed - forM_ (Set.elems $ vst ^. remoteIncoming) $ + forM_ (Set.elems $ vst ^. remoteIncoming) $ writeChan ourChannel . ConnectionClosed . connId - let vst' = remoteIncoming ^= Set.empty $ vst + let vst' = remoteIncoming ^= Set.empty $ vst -- If we still have outgoing connections then we ignore the -- CloseSocket request (we sent a ConnectionCreated message to the -- remote endpoint, but it did not receive it before sending the -- CloseSocket request). Similarly, if lastReceivedId < lastSentId -- then we sent a ConnectionCreated *AND* a ConnectionClosed -- message to the remote endpoint, *both of which* it did not yet - -- receive before sending the CloseSocket request. - if vst' ^. remoteOutgoing > 0 || lastReceivedId < lastSentId vst + -- receive before sending the CloseSocket request. + if vst' ^. remoteOutgoing > 0 || lastReceivedId < lastSentId vst then return (RemoteEndPointValid vst', Nothing) - else do + else do removeRemoteEndPoint (ourEndPoint, theirEndPoint) -- Attempt to reply (but don't insist) act <- schedule theirEndPoint $ do tryIO $ sendOn vst' [ encodeInt32 CloseSocket , encodeInt32 (vst ^. remoteMaxIncoming) ] - tryCloseSocket sock + tryCloseSocket sock return (RemoteEndPointClosed, Just act) RemoteEndPointClosing resolved vst -> do -- Like above, we need to check if there is a ConnectionCreated - -- message that we sent but that the remote endpoint has not yet - -- received. However, since we are in 'closing' state, the only + -- message that we sent but that the remote endpoint has not yet + -- received. However, since we are in 'closing' state, the only -- way this may happen is when we sent a ConnectionCreated, -- ConnectionClosed, and CloseSocket message, none of which have -- yet been received. We leave the endpoint in closing state in @@ -949,29 +949,29 @@ return (RemoteEndPointClosing resolved vst, Nothing) else do removeRemoteEndPoint (ourEndPoint, theirEndPoint) - act <- schedule theirEndPoint $ tryCloseSocket sock + act <- schedule theirEndPoint $ tryCloseSocket sock putMVar resolved () return (RemoteEndPointClosed, Just act) RemoteEndPointFailed err -> throwIO err RemoteEndPointClosed -> - relyViolation (ourEndPoint, theirEndPoint) + relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:closeSocket (closed)" case mAct of Nothing -> return False Just act -> do runScheduledAction (ourEndPoint, theirEndPoint) act return True - + -- Read a message and output it on the endPoint's channel. By rights we -- should verify that the connection ID is valid, but this is unnecessary -- overhead - readMessage :: N.Socket -> LightweightConnectionId -> IO () - readMessage sock lcid = - recvWithLength sock >>= writeChan ourChannel . Received (connId lcid) + readMessage :: N.Socket -> LightweightConnectionId -> IO () + readMessage sock lcid = + recvWithLength sock >>= writeChan ourChannel . Received (connId lcid) -- Arguments - ourChannel = localChannel ourEndPoint + ourChannel = localChannel ourEndPoint theirState = remoteState theirEndPoint theirAddr = remoteAddress theirEndPoint @@ -982,20 +982,20 @@ modifyMVar_ theirState $ \st -> case st of RemoteEndPointInvalid _ -> - relyViolation (ourEndPoint, theirEndPoint) + relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:prematureExit" RemoteEndPointInit _ _ _ -> - relyViolation (ourEndPoint, theirEndPoint) + relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:prematureExit" RemoteEndPointValid _ -> do - let code = EventConnectionLost (remoteAddress theirEndPoint) + let code = EventConnectionLost (remoteAddress theirEndPoint) writeChan ourChannel . ErrorEvent $ TransportError code (show err) return (RemoteEndPointFailed err) RemoteEndPointClosing resolved _ -> do putMVar resolved () return (RemoteEndPointFailed err) RemoteEndPointClosed -> - relyViolation (ourEndPoint, theirEndPoint) + relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:prematureExit" RemoteEndPointFailed err' -> return (RemoteEndPointFailed err') @@ -1006,7 +1006,7 @@ -- The ID of the last connection _we_ created (or 0 for none) lastSentId :: ValidRemoteEndPointState -> LightweightConnectionId - lastSentId vst = + lastSentId vst = if vst ^. remoteNextConnOutId == firstNonReservedLightweightConnectionId then 0 else (vst ^. remoteNextConnOutId) - 1 @@ -1021,12 +1021,12 @@ -- | Create a connection to a remote endpoint -- -- If the remote endpoint is in 'RemoteEndPointClosing' state then we will --- block until that is resolved. +-- block until that is resolved. -- -- May throw a TransportError ConnectErrorCode exception. -createConnectionTo :: TCPParameters - -> LocalEndPoint - -> EndPointAddress +createConnectionTo :: TCPParameters + -> LocalEndPoint + -> EndPointAddress -> ConnectHints -> IO (RemoteEndPoint, LightweightConnectionId) createConnectionTo params ourEndPoint theirAddress hints = go @@ -1035,27 +1035,27 @@ (theirEndPoint, isNew) <- mapIOException connectFailed $ findRemoteEndPoint ourEndPoint theirAddress RequestedByUs - if isNew + if isNew then do - forkIO . handle absorbAllExceptions $ - setupRemoteEndPoint params (ourEndPoint, theirEndPoint) hints + forkIO . handle absorbAllExceptions $ + setupRemoteEndPoint params (ourEndPoint, theirEndPoint) hints go else do -- 'findRemoteEndPoint' will have increased 'remoteOutgoing' mapIOException connectFailed $ do act <- modifyMVar (remoteState theirEndPoint) $ \st -> case st of - RemoteEndPointValid vst -> do - let connId = vst ^. remoteNextConnOutId + RemoteEndPointValid vst -> do + let connId = vst ^. remoteNextConnOutId act <- schedule theirEndPoint $ do sendOn vst [encodeInt32 CreatedNewConnection, encodeInt32 connId] return connId - return ( RemoteEndPointValid - $ remoteNextConnOutId ^= connId + 1 + return ( RemoteEndPointValid + $ remoteNextConnOutId ^= connId + 1 $ vst - , act + , act ) -- Error cases - RemoteEndPointInvalid err -> + RemoteEndPointInvalid err -> throwIO err RemoteEndPointFailed err -> throwIO err @@ -1065,32 +1065,32 @@ -- TODO: deal with exception case? connId <- runScheduledAction (ourEndPoint, theirEndPoint) act return (theirEndPoint, connId) - + connectFailed :: IOException -> TransportError ConnectErrorCode connectFailed = TransportError ConnectFailed . show absorbAllExceptions :: SomeException -> IO () - absorbAllExceptions _ex = + absorbAllExceptions _ex = return () -- | Set up a remote endpoint -setupRemoteEndPoint :: TCPParameters -> EndPointPair -> ConnectHints -> IO () +setupRemoteEndPoint :: TCPParameters -> EndPointPair -> ConnectHints -> IO () setupRemoteEndPoint params (ourEndPoint, theirEndPoint) hints = do - result <- socketToEndPoint ourAddress - theirAddress + result <- socketToEndPoint ourAddress + theirAddress (tcpReuseClientAddr params) (connectTimeout hints) didAccept <- case result of - Right (sock, ConnectionRequestAccepted) -> do - sendLock <- newMVar () - let vst = ValidRemoteEndPointState + Right (sock, ConnectionRequestAccepted) -> do + sendLock <- newMVar () + let vst = ValidRemoteEndPointState { remoteSocket = sock , remoteSendLock = sendLock - , _remoteOutgoing = 0 + , _remoteOutgoing = 0 , _remoteIncoming = Set.empty , _remoteMaxIncoming = 0 - , _remoteNextConnOutId = firstNonReservedLightweightConnectionId + , _remoteNextConnOutId = firstNonReservedLightweightConnectionId } resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst) return True @@ -1101,7 +1101,7 @@ return False Right (sock, ConnectionRequestCrossed) -> do withMVar (remoteState theirEndPoint) $ \st -> case st of - RemoteEndPointInit _ crossed _ -> + RemoteEndPointInit _ crossed _ -> putMVar crossed () RemoteEndPointFailed ex -> throwIO ex @@ -1109,11 +1109,11 @@ relyViolation (ourEndPoint, theirEndPoint) "setupRemoteEndPoint: Crossed" tryCloseSocket sock return False - Left err -> do + Left err -> do resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointInvalid err) return False - when didAccept $ handleIncomingMessages (ourEndPoint, theirEndPoint) + when didAccept $ handleIncomingMessages (ourEndPoint, theirEndPoint) where ourAddress = localAddress ourEndPoint theirAddress = remoteAddress theirEndPoint @@ -1121,23 +1121,23 @@ -- | Send a CloseSocket request if the remote endpoint is unused closeIfUnused :: EndPointPair -> IO () -closeIfUnused (ourEndPoint, theirEndPoint) = do +closeIfUnused (ourEndPoint, theirEndPoint) = do mAct <- modifyMVar (remoteState theirEndPoint) $ \st -> case st of RemoteEndPointValid vst -> - if vst ^. remoteOutgoing == 0 && Set.null (vst ^. remoteIncoming) - then do + if vst ^. remoteOutgoing == 0 && Set.null (vst ^. remoteIncoming) + then do resolved <- newEmptyMVar - act <- schedule theirEndPoint $ + act <- schedule theirEndPoint $ sendOn vst [ encodeInt32 CloseSocket , encodeInt32 (vst ^. remoteMaxIncoming) ] return (RemoteEndPointClosing resolved vst, Just act) - else + else return (RemoteEndPointValid vst, Nothing) _ -> return (st, Nothing) - forM_ mAct $ runScheduledAction (ourEndPoint, theirEndPoint) - + forM_ mAct $ runScheduledAction (ourEndPoint, theirEndPoint) + -- | Reset a remote endpoint if it is in Invalid mode -- -- If the remote endpoint is currently in broken state, and @@ -1156,7 +1156,7 @@ return (vst ^. localConnectionTo theirAddress) LocalEndPointClosed -> throwIO $ TransportError ConnectFailed "Endpoint closed" - forM_ mTheirEndPoint $ \theirEndPoint -> + forM_ mTheirEndPoint $ \theirEndPoint -> withMVar (remoteState theirEndPoint) $ \st -> case st of RemoteEndPointInvalid _ -> removeRemoteEndPoint (ourEndPoint, theirEndPoint) @@ -1166,26 +1166,26 @@ return () -- | Special case of 'apiConnect': connect an endpoint to itself --- +-- -- May throw a TransportError ConnectErrorCode (if the local endpoint is closed) -connectToSelf :: LocalEndPoint +connectToSelf :: LocalEndPoint -> IO Connection -connectToSelf ourEndPoint = do +connectToSelf ourEndPoint = do connAlive <- newIORef True -- Protected by the local endpoint lock - lconnId <- mapIOException connectFailed $ getLocalNextConnOutId ourEndPoint + lconnId <- mapIOException connectFailed $ getLocalNextConnOutId ourEndPoint let connId = createConnectionId heavyweightSelfConnectionId lconnId writeChan ourChan $ ConnectionOpened connId ReliableOrdered (localAddress ourEndPoint) - return Connection - { send = selfSend connAlive connId + return Connection + { send = selfSend connAlive connId , close = selfClose connAlive connId } where - selfSend :: IORef Bool - -> ConnectionId - -> [ByteString] + selfSend :: IORef Bool + -> ConnectionId + -> [ByteString] -> IO (Either (TransportError SendErrorCode) ()) - selfSend connAlive connId msg = + selfSend connAlive connId msg = try . withMVar ourState $ \st -> case st of LocalEndPointValid _ -> do alive <- readIORef connAlive @@ -1196,19 +1196,19 @@ throwIO $ TransportError SendFailed "Endpoint closed" selfClose :: IORef Bool -> ConnectionId -> IO () - selfClose connAlive connId = + selfClose connAlive connId = withMVar ourState $ \st -> case st of LocalEndPointValid _ -> do alive <- readIORef connAlive when alive $ do - writeChan ourChan (ConnectionClosed connId) + writeChan ourChan (ConnectionClosed connId) writeIORef connAlive False LocalEndPointClosed -> - return () + return () ourChan = localChannel ourEndPoint ourState = localState ourEndPoint - connectFailed = TransportError ConnectFailed . show + connectFailed = TransportError ConnectFailed . show -- | Resolve an endpoint currently in 'Init' state resolveInit :: EndPointPair -> RemoteState -> IO () @@ -1216,57 +1216,57 @@ modifyMVar_ (remoteState theirEndPoint) $ \st -> case st of RemoteEndPointInit resolved _ _ -> do putMVar resolved () - case newState of - RemoteEndPointClosed -> + case newState of + RemoteEndPointClosed -> removeRemoteEndPoint (ourEndPoint, theirEndPoint) _ -> return () return newState - RemoteEndPointFailed ex -> + RemoteEndPointFailed ex -> throwIO ex _ -> relyViolation (ourEndPoint, theirEndPoint) "resolveInit" -- | Get the next outgoing self-connection ID --- +-- -- Throws an IO exception when the endpoint is closed. getLocalNextConnOutId :: LocalEndPoint -> IO LightweightConnectionId -getLocalNextConnOutId ourEndpoint = +getLocalNextConnOutId ourEndpoint = modifyMVar (localState ourEndpoint) $ \st -> case st of LocalEndPointValid vst -> do - let connId = vst ^. localNextConnOutId - return ( LocalEndPointValid - . (localNextConnOutId ^= connId + 1) + let connId = vst ^. localNextConnOutId + return ( LocalEndPointValid + . (localNextConnOutId ^= connId + 1) $ vst , connId) LocalEndPointClosed -> throwIO $ userError "Local endpoint closed" -- | Create a new local endpoint --- +-- -- May throw a TransportError NewEndPointErrorCode exception if the transport -- is closed. createLocalEndPoint :: TCPTransport -> IO LocalEndPoint -createLocalEndPoint transport = do +createLocalEndPoint transport = do chan <- newChan - state <- newMVar . LocalEndPointValid $ ValidLocalEndPointState + state <- newMVar . LocalEndPointValid $ ValidLocalEndPointState { _localNextConnOutId = firstNonReservedLightweightConnectionId , _localConnections = Map.empty - , _nextConnInId = firstNonReservedHeavyweightConnectionId + , _nextConnInId = firstNonReservedHeavyweightConnectionId } modifyMVar (transportState transport) $ \st -> case st of TransportValid vst -> do let ix = vst ^. nextEndPointId - let addr = encodeEndPointAddress (transportHost transport) + let addr = encodeEndPointAddress (transportHost transport) (transportPort transport) - ix + ix let localEndPoint = LocalEndPoint { localAddress = addr , localChannel = chan , localState = state } - return ( TransportValid - . (localEndPointAt addr ^= Just localEndPoint) - . (nextEndPointId ^= ix + 1) + return ( TransportValid + . (localEndPointAt addr ^= Just localEndPoint) + . (nextEndPointId ^= ix + 1) $ vst , localEndPoint ) @@ -1282,13 +1282,13 @@ modifyMVar_ ourState $ \st -> case st of LocalEndPointValid vst -> case vst ^. localConnectionTo theirAddress of - Nothing -> + Nothing -> return st Just remoteEndPoint' -> - if remoteId remoteEndPoint' == remoteId theirEndPoint - then return - ( LocalEndPointValid - . (localConnectionTo (remoteAddress theirEndPoint) ^= Nothing) + if remoteId remoteEndPoint' == remoteId theirEndPoint + then return + ( LocalEndPointValid + . (localConnectionTo (remoteAddress theirEndPoint) ^= Nothing) $ vst ) else return st @@ -1302,23 +1302,23 @@ -- -- Does nothing if the transport is closed removeLocalEndPoint :: TCPTransport -> LocalEndPoint -> IO () -removeLocalEndPoint transport ourEndPoint = - modifyMVar_ (transportState transport) $ \st -> case st of +removeLocalEndPoint transport ourEndPoint = + modifyMVar_ (transportState transport) $ \st -> case st of TransportValid vst -> - return ( TransportValid - . (localEndPointAt (localAddress ourEndPoint) ^= Nothing) + return ( TransportValid + . (localEndPointAt (localAddress ourEndPoint) ^= Nothing) $ vst ) TransportClosed -> return TransportClosed -- | Find a remote endpoint. If the remote endpoint does not yet exist we --- create it in Init state. Returns if the endpoint was new. -findRemoteEndPoint +-- create it in Init state. Returns if the endpoint was new. +findRemoteEndPoint :: LocalEndPoint -> EndPointAddress - -> RequestedBy - -> IO (RemoteEndPoint, Bool) + -> RequestedBy + -> IO (RemoteEndPoint, Bool) findRemoteEndPoint ourEndPoint theirAddress findOrigin = go where go = do @@ -1335,55 +1335,55 @@ { remoteAddress = theirAddress , remoteState = theirState , remoteId = vst ^. nextConnInId - , remoteScheduled = scheduled + , remoteScheduled = scheduled } - return ( LocalEndPointValid - . (localConnectionTo theirAddress ^= Just theirEndPoint) - . (nextConnInId ^: (+ 1)) + return ( LocalEndPointValid + . (localConnectionTo theirAddress ^= Just theirEndPoint) + . (nextConnInId ^: (+ 1)) $ vst - , (theirEndPoint, True) + , (theirEndPoint, True) ) LocalEndPointClosed -> throwIO $ userError "Local endpoint closed" - - if isNew + + if isNew then return (theirEndPoint, True) else do let theirState = remoteState theirEndPoint snapshot <- modifyMVar theirState $ \st -> case st of - RemoteEndPointValid vst -> + RemoteEndPointValid vst -> case findOrigin of RequestedByUs -> do - let st' = RemoteEndPointValid - . (remoteOutgoing ^: (+ 1)) - $ vst + let st' = RemoteEndPointValid + . (remoteOutgoing ^: (+ 1)) + $ vst return (st', st') RequestedByThem -> - return (st, st) + return (st, st) _ -> return (st, st) -- The snapshot may no longer be up to date at this point, but if we - -- increased the refcount then it can only either be Valid or Failed - -- (after an explicit call to 'closeEndPoint' or 'closeTransport') + -- increased the refcount then it can only either be Valid or Failed + -- (after an explicit call to 'closeEndPoint' or 'closeTransport') case snapshot of RemoteEndPointInvalid err -> throwIO err RemoteEndPointInit resolved crossed initOrigin -> case (findOrigin, initOrigin) of - (RequestedByUs, RequestedByUs) -> - readMVar resolved >> go - (RequestedByUs, RequestedByThem) -> + (RequestedByUs, RequestedByUs) -> readMVar resolved >> go - (RequestedByThem, RequestedByUs) -> - if ourAddress > theirAddress + (RequestedByUs, RequestedByThem) -> + readMVar resolved >> go + (RequestedByThem, RequestedByUs) -> + if ourAddress > theirAddress then do -- Wait for the Crossed message - readMVar crossed + readMVar crossed return (theirEndPoint, True) else return (theirEndPoint, False) - (RequestedByThem, RequestedByThem) -> + (RequestedByThem, RequestedByThem) -> throwIO $ userError "Already connected" RemoteEndPointValid _ -> -- We assume that the request crossed if we find the endpoint in @@ -1395,15 +1395,15 @@ readMVar resolved >> go RemoteEndPointClosed -> go - RemoteEndPointFailed err -> + RemoteEndPointFailed err -> throwIO err - - ourState = localState ourEndPoint + + ourState = localState ourEndPoint ourAddress = localAddress ourEndPoint -- | Send a payload over a heavyweight connection (thread safe) -sendOn :: ValidRemoteEndPointState -> [ByteString] -> IO () -sendOn vst bs = withMVar (remoteSendLock vst) $ \() -> +sendOn :: ValidRemoteEndPointState -> [ByteString] -> IO () +sendOn vst bs = withMVar (remoteSendLock vst) $ \() -> sendMany (remoteSocket vst) bs -------------------------------------------------------------------------------- @@ -1417,7 +1417,7 @@ schedule :: RemoteEndPoint -> IO a -> IO (Action a) schedule theirEndPoint act = do mvar <- newEmptyMVar - writeChan (remoteScheduled theirEndPoint) $ + writeChan (remoteScheduled theirEndPoint) $ catch (act >>= putMVar mvar . Right) (putMVar mvar . Left) return mvar @@ -1429,7 +1429,7 @@ -- > runScheduledAction -- -- 'runScheduledAction' will run @p@ (it might run some other scheduled action). --- However, it will then wait until @p@ is executed (by this call to +-- However, it will then wait until @p@ is executed (by this call to -- 'runScheduledAction' or by another). runScheduledAction :: EndPointPair -> Action a -> IO a runScheduledAction (ourEndPoint, theirEndPoint) mvar = do @@ -1438,21 +1438,21 @@ case ma of Right a -> return a Left e -> do - forM_ (fromException e) $ \ioe -> + forM_ (fromException e) $ \ioe -> modifyMVar_ (remoteState theirEndPoint) $ \st -> - case st of + case st of RemoteEndPointValid vst -> handleIOException ioe vst _ -> return (RemoteEndPointFailed ioe) throwIO e where - handleIOException :: IOException - -> ValidRemoteEndPointState - -> IO RemoteState + handleIOException :: IOException + -> ValidRemoteEndPointState + -> IO RemoteState handleIOException ex vst = do tryCloseSocket (remoteSocket vst) - let code = EventConnectionLost (remoteAddress theirEndPoint) + let code = EventConnectionLost (remoteAddress theirEndPoint) err = TransportError code (show ex) - writeChan (localChannel ourEndPoint) $ ErrorEvent err + writeChan (localChannel ourEndPoint) $ ErrorEvent err return (RemoteEndPointFailed ex) -------------------------------------------------------------------------------- @@ -1462,25 +1462,25 @@ -- | Establish a connection to a remote endpoint -- -- Maybe throw a TransportError -socketToEndPoint :: EndPointAddress -- ^ Our address +socketToEndPoint :: EndPointAddress -- ^ Our address -> EndPointAddress -- ^ Their address -> Bool -- ^ Use SO_REUSEADDR? - -> Maybe Int -- ^ Timeout for connect - -> IO (Either (TransportError ConnectErrorCode) - (N.Socket, ConnectionRequestResponse)) -socketToEndPoint (EndPointAddress ourAddress) theirAddress reuseAddr timeout = - try $ do - (host, port, theirEndPointId) <- case decodeEndPointAddress theirAddress of + -> Maybe Int -- ^ Timeout for connect + -> IO (Either (TransportError ConnectErrorCode) + (N.Socket, ConnectionRequestResponse)) +socketToEndPoint (EndPointAddress ourAddress) theirAddress reuseAddr timeout = + try $ do + (host, port, theirEndPointId) <- case decodeEndPointAddress theirAddress of Nothing -> throwIO (failed . userError $ "Could not parse") Just dec -> return dec - addr:_ <- mapIOException invalidAddress $ + addr:_ <- mapIOException invalidAddress $ N.getAddrInfo Nothing (Just host) (Just port) bracketOnError (createSocket addr) tryCloseSocket $ \sock -> do - when reuseAddr $ + when reuseAddr $ mapIOException failed $ N.setSocketOption sock N.ReuseAddr 1 - mapIOException invalidAddress $ - timeoutMaybe timeout timeoutError $ - N.connect sock (N.addrAddress addr) + mapIOException invalidAddress $ + timeoutMaybe timeout timeoutError $ + N.connect sock (N.addrAddress addr) response <- mapIOException failed $ do sendMany sock (encodeInt32 theirEndPointId : prependLength [ourAddress]) recvInt32 sock @@ -1489,50 +1489,50 @@ Just r -> return (sock, r) where createSocket :: N.AddrInfo -> IO N.Socket - createSocket addr = mapIOException insufficientResources $ + createSocket addr = mapIOException insufficientResources $ N.socket (N.addrFamily addr) N.Stream N.defaultProtocol - invalidAddress = TransportError ConnectNotFound . show - insufficientResources = TransportError ConnectInsufficientResources . show + invalidAddress = TransportError ConnectNotFound . show + insufficientResources = TransportError ConnectInsufficientResources . show failed = TransportError ConnectFailed . show timeoutError = TransportError ConnectTimeout "Timed out" -- | Encode end point address -encodeEndPointAddress :: N.HostName - -> N.ServiceName - -> EndPointId +encodeEndPointAddress :: N.HostName + -> N.ServiceName + -> EndPointId -> EndPointAddress encodeEndPointAddress host port ix = EndPointAddress . BSC.pack $ - host ++ ":" ++ port ++ ":" ++ show ix + host ++ ":" ++ port ++ ":" ++ show ix -- | Decode end point address -decodeEndPointAddress :: EndPointAddress +decodeEndPointAddress :: EndPointAddress -> Maybe (N.HostName, N.ServiceName, EndPointId) -decodeEndPointAddress (EndPointAddress bs) = +decodeEndPointAddress (EndPointAddress bs) = case splitMaxFromEnd (== ':') 2 $ BSC.unpack bs of - [host, port, endPointIdStr] -> - case reads endPointIdStr of + [host, port, endPointIdStr] -> + case reads endPointIdStr of [(endPointId, "")] -> Just (host, port, endPointId) _ -> Nothing _ -> Nothing -- | Construct a ConnectionId -createConnectionId :: HeavyweightConnectionId - -> LightweightConnectionId +createConnectionId :: HeavyweightConnectionId + -> LightweightConnectionId -> ConnectionId -createConnectionId hcid lcid = +createConnectionId hcid lcid = (fromIntegral hcid `shiftL` 32) .|. fromIntegral lcid -- | @spltiMaxFromEnd p n xs@ splits list @xs@ at elements matching @p@, -- returning at most @p@ segments -- counting from the /end/ --- +-- -- > splitMaxFromEnd (== ':') 2 "ab:cd:ef:gh" == ["ab:cd", "ef", "gh"] splitMaxFromEnd :: (a -> Bool) -> Int -> [a] -> [[a]] -splitMaxFromEnd p = \n -> go [[]] n . reverse +splitMaxFromEnd p = \n -> go [[]] n . reverse where -- go :: [[a]] -> Int -> [a] -> [[a]] - go accs _ [] = accs + go accs _ [] = accs go ([] : accs) 0 xs = reverse xs : accs go (acc : accs) n (x:xs) = if p x then go ([] : acc : accs) (n - 1) xs @@ -1544,40 +1544,40 @@ -------------------------------------------------------------------------------- -- Find a socket between two endpoints --- +-- -- Throws an IO exception if the socket could not be found. -internalSocketBetween :: TCPTransport -- ^ Transport +internalSocketBetween :: TCPTransport -- ^ Transport -> EndPointAddress -- ^ Local endpoint -> EndPointAddress -- ^ Remote endpoint - -> IO N.Socket + -> IO N.Socket internalSocketBetween transport ourAddress theirAddress = do ourEndPoint <- withMVar (transportState transport) $ \st -> case st of - TransportClosed -> - throwIO $ userError "Transport closed" - TransportValid vst -> + TransportClosed -> + throwIO $ userError "Transport closed" + TransportValid vst -> case vst ^. localEndPointAt ourAddress of Nothing -> throwIO $ userError "Local endpoint not found" Just ep -> return ep theirEndPoint <- withMVar (localState ourEndPoint) $ \st -> case st of LocalEndPointClosed -> throwIO $ userError "Local endpoint closed" - LocalEndPointValid vst -> + LocalEndPointValid vst -> case vst ^. localConnectionTo theirAddress of Nothing -> throwIO $ userError "Remote endpoint not found" Just ep -> return ep withMVar (remoteState theirEndPoint) $ \st -> case st of RemoteEndPointInit _ _ _ -> throwIO $ userError "Remote endpoint not yet initialized" - RemoteEndPointValid vst -> + RemoteEndPointValid vst -> return $ remoteSocket vst RemoteEndPointClosing _ vst -> - return $ remoteSocket vst + return $ remoteSocket vst RemoteEndPointClosed -> throwIO $ userError "Remote endpoint closed" RemoteEndPointInvalid err -> - throwIO err + throwIO err RemoteEndPointFailed err -> - throwIO err + throwIO err -------------------------------------------------------------------------------- -- Constants -- @@ -1611,7 +1611,7 @@ localConnections :: Accessor ValidLocalEndPointState (Map EndPointAddress RemoteEndPoint) localConnections = accessor _localConnections (\es st -> st { _localConnections = es }) -nextConnInId :: Accessor ValidLocalEndPointState HeavyweightConnectionId +nextConnInId :: Accessor ValidLocalEndPointState HeavyweightConnectionId nextConnInId = accessor _nextConnInId (\rid st -> st { _nextConnInId = rid }) remoteOutgoing :: Accessor ValidRemoteEndPointState Int @@ -1627,10 +1627,10 @@ remoteNextConnOutId = accessor _remoteNextConnOutId (\cix st -> st { _remoteNextConnOutId = cix }) localEndPointAt :: EndPointAddress -> Accessor ValidTransportState (Maybe LocalEndPoint) -localEndPointAt addr = localEndPoints >>> DAC.mapMaybe addr +localEndPointAt addr = localEndPoints >>> DAC.mapMaybe addr localConnectionTo :: EndPointAddress -> Accessor ValidLocalEndPointState (Maybe RemoteEndPoint) -localConnectionTo addr = localConnections >>> DAC.mapMaybe addr +localConnectionTo addr = localConnections >>> DAC.mapMaybe addr ------------------------------------------------------------------------------- -- Debugging -- @@ -1638,14 +1638,14 @@ relyViolation :: EndPointPair -> String -> IO a relyViolation (ourEndPoint, theirEndPoint) str = do - elog (ourEndPoint, theirEndPoint) (str ++ " RELY violation") + elog (ourEndPoint, theirEndPoint) (str ++ " RELY violation") fail (str ++ " RELY violation") elog :: EndPointPair -> String -> IO () elog (ourEndPoint, theirEndPoint) msg = do tid <- myThreadId - putStrLn $ show (localAddress ourEndPoint) - ++ "/" ++ show (remoteAddress theirEndPoint) + putStrLn $ show (localAddress ourEndPoint) + ++ "/" ++ show (remoteAddress theirEndPoint) ++ "(" ++ show (remoteId theirEndPoint) ++ ")" - ++ "/" ++ show tid + ++ "/" ++ show tid ++ ": " ++ msg diff --git a/src/Network/Transport/TCP/Internal.hs b/src/Network/Transport/TCP/Internal.hs --- a/src/Network/Transport/TCP/Internal.hs +++ b/src/Network/Transport/TCP/Internal.hs @@ -1,8 +1,8 @@ --- | Utility functions for TCP sockets -module Network.Transport.TCP.Internal +-- | Utility functions for TCP sockets +module Network.Transport.TCP.Internal ( forkServer , recvWithLength - , recvExact + , recvExact , recvInt32 , tryCloseSocket ) where @@ -16,7 +16,7 @@ #ifdef USE_MOCK_NETWORK import qualified Network.Transport.TCP.Mock.Socket as N #else -import qualified Network.Socket as N +import qualified Network.Socket as N #endif ( HostName , ServiceName @@ -49,10 +49,10 @@ import Data.ByteString (ByteString) import qualified Data.ByteString as BS (length, concat, null) import Data.Int (Int32) -import Data.ByteString.Lazy.Internal (smallChunkSize) +import Data.ByteString.Lazy.Internal (smallChunkSize) -- | Start a server at the specified address. --- +-- -- This sets up a server socket for the specified host and port. Exceptions -- thrown during setup are not caught. -- @@ -67,15 +67,15 @@ -- The request handler should spawn threads to handle each individual request -- or the server will block. Once a thread has been spawned it will be the -- responsibility of the new thread to close the socket when an exception --- occurs. +-- occurs. forkServer :: N.HostName -- ^ Host - -> N.ServiceName -- ^ Port + -> N.ServiceName -- ^ Port -> Int -- ^ Backlog (maximum number of queued connections) -> Bool -- ^ Set ReuseAddr option? -> (SomeException -> IO ()) -- ^ Termination handler - -> (N.Socket -> IO ()) -- ^ Request handler + -> (N.Socket -> IO ()) -- ^ Request handler -> IO ThreadId -forkServer host port backlog reuseAddr terminationHandler requestHandler = do +forkServer host port backlog reuseAddr terminationHandler requestHandler = do -- Resolve the specified address. By specification, getAddrInfo will never -- return an empty list (but will throw an exception instead) and will return -- the "best" address first, whatever that means @@ -84,14 +84,14 @@ tryCloseSocket $ \sock -> do when reuseAddr $ N.setSocketOption sock N.ReuseAddr 1 N.bindSocket sock (N.addrAddress addr) - N.listen sock backlog + N.listen sock backlog -- We start listening for incoming requests in a separate thread. When -- that thread is killed, we close the server socket and the termination -- handler. We have to make sure that the exception handler is installed -- /before/ any asynchronous exception occurs. So we mask_, then fork -- (the child thread inherits the masked state from the parent), then -- unmask only inside the catch. - mask_ $ forkIOWithUnmask $ \unmask -> + mask_ $ forkIOWithUnmask $ \unmask -> catch (unmask (forever $ acceptRequest sock)) $ \ex -> do tryCloseSocket sock terminationHandler ex @@ -100,34 +100,34 @@ acceptRequest sock = bracketOnError (N.accept sock) (tryCloseSocket . fst) (requestHandler . fst) - + -- | Read a length and then a payload of that length recvWithLength :: N.Socket -> IO [ByteString] recvWithLength sock = recvInt32 sock >>= recvExact sock -- | Receive a 32-bit integer -recvInt32 :: Num a => N.Socket -> IO a -recvInt32 sock = decodeInt32 . BS.concat <$> recvExact sock 4 +recvInt32 :: Num a => N.Socket -> IO a +recvInt32 sock = decodeInt32 . BS.concat <$> recvExact sock 4 -- | Close a socket, ignoring I/O exceptions tryCloseSocket :: N.Socket -> IO () -tryCloseSocket sock = void . tryIO $ +tryCloseSocket sock = void . tryIO $ N.sClose sock -- | Read an exact number of bytes from a socket --- +-- -- Throws an I/O exception if the socket closes before the specified -- number of bytes could be read -recvExact :: N.Socket -- ^ Socket to read from +recvExact :: N.Socket -- ^ Socket to read from -> Int32 -- ^ Number of bytes to read -> IO [ByteString] -recvExact _ len | len < 0 = throwIO (userError "recvExact: Negative length") +recvExact _ len | len < 0 = throwIO (userError "recvExact: Negative length") recvExact sock len = go [] len where - go :: [ByteString] -> Int32 -> IO [ByteString] - go acc 0 = return (reverse acc) + go :: [ByteString] -> Int32 -> IO [ByteString] + go acc 0 = return (reverse acc) go acc l = do bs <- NBS.recv sock (fromIntegral l `min` smallChunkSize) - if BS.null bs + if BS.null bs then throwIO (userError "recvExact: Socket closed") else go (bs : acc) (l - fromIntegral (BS.length bs)) diff --git a/src/Network/Transport/TCP/Mock/Socket.hs b/src/Network/Transport/TCP/Mock/Socket.hs --- a/src/Network/Transport/TCP/Mock/Socket.hs +++ b/src/Network/Transport/TCP/Mock/Socket.hs @@ -38,7 +38,7 @@ import qualified Data.Map as Map import Control.Exception (throwIO) import Control.Category ((>>>)) -import Control.Concurrent.MVar +import Control.Concurrent.MVar import Control.Concurrent.Chan import System.IO.Unsafe (unsafePerformIO) import Data.Accessor (Accessor, accessor, (^=), (^.), (^:)) @@ -70,7 +70,7 @@ get acc = timeoutThrow mvarThreshold $ withMVar mockState $ return . (^. acc) set :: Accessor MockState a -> a -> IO () -set acc val = timeoutThrow mvarThreshold $ modifyMVar_ mockState $ return . (acc ^= val) +set acc val = timeoutThrow mvarThreshold $ modifyMVar_ mockState $ return . (acc ^= val) boundSockets :: Accessor MockState (Map SockAddr Socket) boundSockets = accessor _boundSockets (\bs st -> st { _boundSockets = bs }) @@ -90,34 +90,34 @@ type HostName = String type ServiceName = String -type PortNumber = String -type HostAddress = String +type PortNumber = String +type HostAddress = String -data SocketType = Stream +data SocketType = Stream data SocketOption = ReuseAddr data ShutdownCmd = ShutdownSend data Family data ProtocolNumber -data Socket = Socket { +data Socket = Socket { socketState :: MVar SocketState , socketDescription :: String } -data SocketState = +data SocketState = Uninit - | BoundSocket { - socketBacklog :: Chan (Socket, SockAddr, MVar Socket) + | BoundSocket { + socketBacklog :: Chan (Socket, SockAddr, MVar Socket) } - | Connected { - socketBuff :: Chan Message + | Connected { + socketBuff :: Chan Message , _socketPeer :: Maybe Socket , _scheduledReadActions :: [(Int, IO ())] } | Closed -data Message = +data Message = Payload Word8 | CloseSocket @@ -145,43 +145,43 @@ getAddrInfo _ (Just host) (Just port) = do validHosts <- get validHostnames if host `elem` validHosts - then return . return $ AddrInfo { - addrFamily = error "Family unused" - , addrAddress = SockAddrInet port host + then return . return $ AddrInfo { + addrFamily = error "Family unused" + , addrAddress = SockAddrInet port host } else throwSocketError $ "getAddrInfo: invalid hostname '" ++ host ++ "'" getAddrInfo _ _ _ = error "getAddrInfo: unsupported arguments" defaultHints :: AddrInfo -defaultHints = error "defaultHints not implemented" +defaultHints = error "defaultHints not implemented" socket :: Family -> SocketType -> ProtocolNumber -> IO Socket socket _ Stream _ = do state <- newMVar Uninit sid <- get nextSocketId set nextSocketId (sid + 1) - return Socket { + return Socket { socketState = state , socketDescription = show sid } - + bindSocket :: Socket -> SockAddr -> IO () bindSocket sock addr = do timeoutThrow mvarThreshold $ modifyMVar_ (socketState sock) $ \st -> case st of Uninit -> do backlog <- newChan - return BoundSocket { - socketBacklog = backlog + return BoundSocket { + socketBacklog = backlog } _ -> throwSocketError "bind: socket already initialized" set (boundSocketAt addr) (Just sock) - + listen :: Socket -> Int -> IO () -listen _ _ = return () +listen _ _ = return () defaultProtocol :: ProtocolNumber -defaultProtocol = error "defaultProtocol not implemented" +defaultProtocol = error "defaultProtocol not implemented" setSocketOption :: Socket -> SocketOption -> Int -> IO () setSocketOption _ ReuseAddr 1 = return () @@ -190,13 +190,13 @@ accept :: Socket -> IO (Socket, SockAddr) accept serverSock = do backlog <- timeoutThrow mvarThreshold $ withMVar (socketState serverSock) $ \st -> case st of - BoundSocket {} -> + BoundSocket {} -> return (socketBacklog st) _ -> throwSocketError "accept: socket not bound" - (theirSocket, theirAddress, reply) <- readChan backlog + (theirSocket, theirAddress, reply) <- readChan backlog ourBuff <- newChan - ourState <- newMVar Connected { + ourState <- newMVar Connected { socketBuff = ourBuff , _socketPeer = Just theirSocket , _scheduledReadActions = [] @@ -205,13 +205,13 @@ socketState = ourState , socketDescription = "" } - timeoutThrow mvarThreshold $ putMVar reply ourSocket + timeoutThrow mvarThreshold $ putMVar reply ourSocket return (ourSocket, theirAddress) sClose :: Socket -> IO () sClose sock = do -- Close the peer socket - writeSocket sock CloseSocket + writeSocket sock CloseSocket -- Close our socket timeoutThrow mvarThreshold $ modifyMVar_ (socketState sock) $ \st -> @@ -220,7 +220,7 @@ -- In case there is a parallel read stuck on a readChan writeChan (socketBuff st) CloseSocket return Closed - _ -> + _ -> return Closed connect :: Socket -> SockAddr -> IO () @@ -235,13 +235,13 @@ throwSocketError "connect: server socket not bound" reply <- newEmptyMVar writeChan serverBacklog (us, SockAddrInet "" "", reply) - them <- timeoutThrow mvarThreshold $ readMVar reply + them <- timeoutThrow mvarThreshold $ readMVar reply timeoutThrow mvarThreshold $ modifyMVar_ (socketState us) $ \st -> case st of - Uninit -> do + Uninit -> do buff <- newChan - return Connected { + return Connected { socketBuff = buff - , _socketPeer = Just them + , _socketPeer = Just them , _scheduledReadActions = [] } _ -> @@ -249,7 +249,7 @@ Nothing -> throwSocketError "connect: unknown address" sOMAXCONN :: Int -sOMAXCONN = error "sOMAXCONN not implemented" +sOMAXCONN = error "sOMAXCONN not implemented" shutdown :: Socket -> ShutdownCmd -> IO () shutdown sock ShutdownSend = do @@ -267,7 +267,7 @@ peerBuffer :: Socket -> IO (Either String (Chan Message)) peerBuffer sock = do mPeer <- timeoutThrow mvarThreshold $ withMVar (socketState sock) $ \st -> case st of - Connected {} -> + Connected {} -> return (st ^. socketPeer) _ -> return Nothing @@ -276,9 +276,9 @@ Connected {} -> return (Right (socketBuff st)) _ -> - return (Left "Peer socket closed") - Nothing -> - return (Left "Socket closed") + return (Left "Peer socket closed") + Nothing -> + return (Left "Socket closed") throwSocketError :: String -> IO a throwSocketError = throwIO . userError @@ -287,8 +287,8 @@ writeSocket sock msg = do theirBuff <- peerBuffer sock case theirBuff of - Right buff -> writeChan buff msg - Left err -> case msg of Payload _ -> throwSocketError $ "writeSocket: " ++ err + Right buff -> writeChan buff msg + Left err -> case msg of Payload _ -> throwSocketError $ "writeSocket: " ++ err CloseSocket -> return () readSocket :: Socket -> IO (Maybe Word8) @@ -304,7 +304,7 @@ case mBuff of Just (buff, actions) -> do sequence actions - msg <- timeoutThrow readSocketThreshold $ readChan buff + msg <- timeoutThrow readSocketThreshold $ readChan buff case msg of Payload w -> return (Just w) CloseSocket -> timeoutThrow mvarThreshold $ modifyMVar (socketState sock) $ \st -> case st of @@ -312,7 +312,7 @@ return (Closed, Nothing) _ -> throwSocketError "readSocket: socket in unexpected state" - Nothing -> + Nothing -> return Nothing -- | Given a list of scheduled actions, reduce all delays by 1, and return the @@ -321,7 +321,7 @@ tick = go [] [] where go later now [] = (reverse later, reverse now) - go later now ((n, action) : actions) + go later now ((n, action) : actions) | n == 0 = go later (action : now) actions | otherwise = go ((n - 1, action) : later) now actions @@ -330,16 +330,16 @@ -------------------------------------------------------------------------------- -- | Schedule an action to be executed after /n/ reads on this socket --- +-- -- If /n/ is zero we execute the action immediately. scheduleReadAction :: Socket -> Int -> IO () -> IO () scheduleReadAction _ 0 action = action -scheduleReadAction sock n action = - modifyMVar_ (socketState sock) $ \st -> case st of +scheduleReadAction sock n action = + modifyMVar_ (socketState sock) $ \st -> case st of Connected {} -> return (scheduledReadActions ^: ((n, action) :) $ st) _ -> - throwSocketError "scheduleReadAction: socket not connected" + throwSocketError "scheduleReadAction: socket not connected" -------------------------------------------------------------------------------- -- Util -- diff --git a/tests/TestQC.hs b/tests/TestQC.hs --- a/tests/TestQC.hs +++ b/tests/TestQC.hs @@ -1,17 +1,24 @@ -module Main +-- Test the TCP transport using QuickCheck generated scripts +-- +-- TODO: This is not quite working yet. The main problem, I think, is the +-- allocation of "bundle ID"s to connections. The problem is exposed by the +-- aptly-named regression test script_Foo (to be renamed once I figure out what +-- bug that test is actually exposing :) +module Main ( main -- Shush the compiler about unused definitions , log , logShow , forAllShrink , inits + , expectedFailure ) where import Prelude hiding (log) import Test.Framework (Test, TestName, defaultMain, testGroup) import Test.Framework.Providers.QuickCheck2 (testProperty) import Test.Framework.Providers.HUnit (testCase) -import Test.QuickCheck +import Test.QuickCheck ( Gen , choose , suchThatMaybe @@ -25,12 +32,11 @@ import Data.Map (Map) import qualified Data.Map as Map import Control.Category ((>>>)) -import Control.Arrow (second) import Control.Applicative ((<$>)) import Control.Exception (Exception, throwIO, try) import Control.Concurrent (forkIO, threadDelay, ThreadId, killThread) -import Control.Monad (replicateM, forever, guard) -import Control.Monad.State.Lazy (StateT, execStateT) +import Control.Monad (MonadPlus(..), replicateM, forever, guard) +import Control.Monad.State (StateT, execStateT) import Control.Monad.IO.Class (MonadIO, liftIO) import Data.Typeable (Typeable) import Data.Maybe (isJust) @@ -38,19 +44,23 @@ import Data.ByteString (ByteString) import Data.ByteString.Char8 (pack) import Data.Accessor (Accessor, accessor, (^.)) -import Data.Accessor.Monad.Trans.State (get, modify) -import qualified Data.Accessor.Container as DAC (mapDefault) +import Data.Accessor.Monad.Trans.State (get, set, modify) +import qualified Data.Accessor.Container as DAC (set, mapDefault) import qualified Data.ByteString as BSS (concat) import qualified Text.PrettyPrint as PP import Data.Unique (Unique, newUnique, hashUnique) import Data.Concurrent.Queue.MichaelScott (newQ, pushL, tryPopR) +import Data.Set (Set) +import qualified Data.Set as Set +import GHC.Stack (currentCallStack, renderStack) import Network.Transport -import Network.Transport.TCP +import Network.Transport.TCP ( createTransportExposeInternals , defaultTCPParameters - , TransportInternals + , TransportInternals(socketBetween) ) +import Network.Transport.TCP.Mock.Socket (Socket, scheduleReadAction, sClose) -------------------------------------------------------------------------------- -- Script infrastructure -- @@ -64,26 +74,26 @@ -- | We randomly generate /scripts/ which are essentially a deep embedding of -- the Transport API. These scripts are then executed and the results compared -- against an abstract interpreter. -data ScriptCmd = +data ScriptCmd = -- | Create a new endpoint NewEndPoint -- | @Connect i j@ creates a connection from endpoint @i@ to endpoint @j@, -- where @i@ and @j@ are indices and refer to the @i@th and @j@th endpoint -- created by NewEndPoint - | Connect SourceEndPointIx TargetEndPointIx + | Connect SourceEndPointIx TargetEndPointIx -- | @Close i@ closes the @i@ connection created using 'Connect'. Note that -- closing a connection does not shift other indices; in other words, in -- @[Connect 0 0, Close 0, Connect 0 0, Close 0]@ the second 'Close' -- refers to the first (already closed) connection - | Close ConnectionIx - -- | @Send i bs@ sends payload @bs@ on the @i@ connection created + | Close ConnectionIx + -- | @Send i bs@ sends payload @bs@ on the @i@ connection created | Send ConnectionIx [ByteString] -- | @BreakAfterReads n i j@ force-closes the socket between endpoints @i@ - -- and @j@ after @n@ reads by @i@ - -- + -- and @j@ after @n@ reads by @i@ + -- -- We should have @i /= j@ because the TCP transport does not use sockets -- for connections from an endpoint to itself - | BreakAfterReads Int SourceEndPointIx TargetEndPointIx + | BreakAfterReads Int SourceEndPointIx TargetEndPointIx deriving Show type Script = [ScriptCmd] @@ -99,101 +109,180 @@ show (Value x) = show x show (Variable u) = "<<" ++ show (hashUnique u) ++ ">>" -data ExpEvent = - ExpConnectionOpened (Variable ConnectionId) - | ExpConnectionClosed (Variable ConnectionId) - | ExpReceived (Variable ConnectionId) [ByteString] +-- | In the implementation "bundles" are purely a conceptual idea, but in the +-- verifier we need to concretize this notion +type BundleId = Int + +data ConnectionInfo = ConnectionInfo { + source :: EndPointAddress + , target :: EndPointAddress + , connectionId :: Variable ConnectionId + , connectionBundle :: BundleId + } deriving Show -type TargetAddress = EndPointAddress +data ExpEvent = + ExpConnectionOpened ConnectionInfo + | ExpConnectionClosed ConnectionInfo + | ExpReceived ConnectionInfo [ByteString] + | ExpConnectionLost BundleId EndPointAddress + deriving Show data RunState = RunState { - _endPoints :: [EndPoint] - , _connections :: [(TargetAddress, Connection, Variable ConnectionId)] - , _expectedEvents :: Map EndPointAddress [ExpEvent] + _endPoints :: [EndPoint] + , _connections :: [(Connection, ConnectionInfo)] + , _expectedEvents :: Map EndPointAddress [ExpEvent] + -- | For each endpoint we create we create a thread that forwards the events + -- of that endpoint to a central channel. We collect the thread IDs so that + -- we can kill these thread when we are done. , _forwardingThreads :: [ThreadId] + -- | When a connection from A to be may break, we add both (A, B, n) + -- and (B, A, n) to _mayBreak. Then once we detect that from A to B + -- has in fact broken we move (A, B, n), *and/or* (B, A, n), from _mayBreak + -- to _broken. Note that we can detect that a connection has been broken + -- in one direction even if we haven't yet detected that the connection + -- has broken in the other direction. + -- + -- | Invariant: not mayBreak && broken + , _mayBreak :: Set (EndPointAddress, EndPointAddress, BundleId) + , _broken :: Set (EndPointAddress, EndPointAddress, BundleId) + -- | Current bundle ID between two endpoints + -- + -- Invariant: For all keys (A, B), A <= B + , _currentBundle :: Map (EndPointAddress, EndPointAddress) BundleId } -initialRunState :: RunState +initialRunState :: RunState initialRunState = RunState { - _endPoints = [] - , _connections = [] - , _expectedEvents = Map.empty + _endPoints = [] + , _connections = [] + , _expectedEvents = Map.empty , _forwardingThreads = [] + , _mayBreak = Set.empty + , _broken = Set.empty + , _currentBundle = Map.empty } verify :: (Transport, TransportInternals) -> Script -> IO (Either String ()) -verify (transport, _transportInternals) script = do +verify (transport, transportInternals) script = do allEvents <- newQ let runScript :: Script -> StateT RunState IO () - runScript = mapM_ runCmd + runScript = mapM_ runCmd - runCmd :: ScriptCmd -> StateT RunState IO () + runCmd :: ScriptCmd -> StateT RunState IO () runCmd NewEndPoint = do mEndPoint <- liftIO $ newEndPoint transport case mEndPoint of Right endPoint -> do - modify endPoints (snoc endPoint) - tid <- liftIO $ forkIO (forward endPoint) - modify forwardingThreads (tid :) + tid <- liftIO $ forkIO (forward endPoint) + append endPoints endPoint + append forwardingThreads tid + set (expectedEventsAt (address endPoint)) [] Left err -> liftIO $ throwIO err runCmd (Connect i j) = do endPointA <- get (endPointAtIx i) endPointB <- address <$> get (endPointAtIx j) mConn <- liftIO $ connect endPointA endPointB ReliableOrdered defaultConnectHints + let bundleId = currentBundle (address endPointA) endPointB + connBroken = broken (address endPointA) endPointB + connMayBreak = mayBreak (address endPointA) endPointB case mConn of Right conn -> do - connId <- Variable <$> liftIO newUnique - modify connections (snoc (endPointB, conn, connId)) - modify (expectedEventsAt endPointB) (snoc (ExpConnectionOpened connId)) - Left err -> - liftIO $ throwIO err + bundleBroken <- get bundleId >>= get . connBroken + currentBundleId <- if bundleBroken + then modify bundleId (+ 1) >> get bundleId + else get bundleId + connId <- Variable <$> liftIO newUnique + let connInfo = ConnectionInfo { + source = address endPointA + , target = endPointB + , connectionId = connId + , connectionBundle = currentBundleId + } + append connections (conn, connInfo) + append (expectedEventsAt endPointB) (ExpConnectionOpened connInfo) + Left err -> do + currentBundleId <- get bundleId + expectingBreak <- get $ connMayBreak currentBundleId + if expectingBreak + then do + set (connMayBreak currentBundleId) False + set (connBroken currentBundleId) True + else + liftIO $ throwIO err runCmd (Close i) = do - (target, conn, connId) <- get (connectionAt i) + (conn, connInfo) <- get (connectionAt i) liftIO $ close conn - modify (expectedEventsAt target) (snoc (ExpConnectionClosed connId)) + append (expectedEventsAt (target connInfo)) (ExpConnectionClosed connInfo) runCmd (Send i payload) = do - (target, conn, connId) <- get (connectionAt i) + (conn, connInfo) <- get (connectionAt i) mResult <- liftIO $ send conn payload + let connMayBreak = mayBreak (source connInfo) (target connInfo) (connectionBundle connInfo) + connBroken = broken (source connInfo) (target connInfo) (connectionBundle connInfo) case mResult of Right () -> return () - Left err -> liftIO $ throwIO err - modify (expectedEventsAt target) (snoc (ExpReceived connId payload)) - runCmd (BreakAfterReads _n _i _j) = - expectedFailure "BreakAfterReads not implemented" - + Left err -> do + expectingBreak <- get connMayBreak + isBroken <- get connBroken + if expectingBreak || isBroken + then do + set connMayBreak False + set connBroken True + else + liftIO $ throwIO err + append (expectedEventsAt (target connInfo)) (ExpReceived connInfo payload) + -- TODO: This will only work if a connection between 'i' and 'j' has + -- already been established. We would need to modify the mock network + -- layer to support breaking "future" connections + runCmd (BreakAfterReads n i j) = do + endPointA <- address <$> get (endPointAtIx i) + endPointB <- address <$> get (endPointAtIx j) + liftIO $ do + sock <- socketBetween transportInternals endPointA endPointB + scheduleReadAction sock n $ breakSocket sock + currentBundleId <- get (currentBundle endPointA endPointB) + set (mayBreak endPointA endPointB currentBundleId) True + set (mayBreak endPointB endPointA currentBundleId) True + append (expectedEventsAt endPointA) (ExpConnectionLost currentBundleId endPointB) + append (expectedEventsAt endPointB) (ExpConnectionLost currentBundleId endPointA) + forward :: EndPoint -> IO () forward endPoint = forever $ do ev <- receive endPoint pushL allEvents (address endPoint, ev) - collectEvents :: RunState -> IO (Map EndPointAddress [Event]) + collectEvents :: RunState -> IO (Map EndPointAddress [Event]) collectEvents st = do threadDelay 10000 mapM_ killThread (st ^. forwardingThreads) evs <- go [] - return (groupByKey evs) + return $ groupByKey (map address (st ^. endPoints)) evs where go acc = do mEv <- tryPopR allEvents case mEv of Just ev -> go (ev : acc) - Nothing -> return acc - - st <- execStateT (runScript script) initialRunState + Nothing -> return (reverse acc) + + st <- execStateT (runScript script) initialRunState actualEvents <- collectEvents st - - let eventsMatch = and . map (uncurry match) $ + + let eventsMatch = all (uncurry match) $ zip (Map.elems (st ^. expectedEvents)) (Map.elems actualEvents) - return $ if eventsMatch - then Right () + return $ if eventsMatch + then Right () else Left ("Could not match " ++ show (st ^. expectedEvents) ++ " and " ++ show actualEvents) +breakSocket :: Socket -> IO () +breakSocket sock = do + currentCallStack >>= putStrLn . renderStack + sClose sock + -------------------------------------------------------------------------------- -- Match expected and actual events -- -------------------------------------------------------------------------------- @@ -201,69 +290,112 @@ -- | Match a list of expected events to a list of actual events, taking into -- account that events may be reordered match :: [ExpEvent] -> [Event] -> Bool -match expected actual = or (map (isJust . flip unify actual) (reorder expected)) +match expected actual = any (`canUnify` actual) (possibleTraces expected) --- | Match a list of expected events to a list of actual events, without doing --- reordering -unify :: [ExpEvent] -> [Event] -> Maybe () -unify [] [] = return () -unify (ExpConnectionOpened connId : expected) (ConnectionOpened connId' _ _ : actual) = do - subst <- unifyConnectionId connId connId' - unify (apply subst expected) actual -unify (ExpConnectionClosed connId : expected) (ConnectionClosed connId' : actual) = do - subst <- unifyConnectionId connId connId' - unify (apply subst expected) actual -unify (ExpReceived connId payload : expected) (Received connId' payload' : actual) = do - guard (BSS.concat payload == BSS.concat payload') - subst <- unifyConnectionId connId connId' - unify (apply subst expected) actual -unify _ _ = fail "Cannot unify" +possibleTraces :: [ExpEvent] -> [[ExpEvent]] +possibleTraces = go + where + go [] = [[]] + go (ev@(ExpConnectionLost _ _) : evs) = + [ trace | evs' <- possibleTraces evs, trace <- insertConnectionLost ev evs' ] + go (ev : evs) = + [ trace | evs' <- possibleTraces evs, trace <- insertEvent ev evs' ] -type Substitution a = Map Unique a + -- We don't know when exactly the error will occur (indeed, it may never + -- happen at all), but it must occur before any future connection lost + -- event to the same destination. + -- If it occurs now, then all other events on this bundle will not happen. + insertConnectionLost :: ExpEvent -> [ExpEvent] -> [[ExpEvent]] + insertConnectionLost ev [] = [[ev], []] + insertConnectionLost ev@(ExpConnectionLost bid addr) (ev' : evs) = + (ev : removeBundle bid (ev' : evs)) : + case ev' of + ExpConnectionLost _ addr' | addr == addr' -> [] + _ -> [ev' : evs' | evs' <- insertConnectionLost ev evs] + insertConnectionLost _ _ = error "The impossible happened" --- | Match two connection IDs -unifyConnectionId :: Variable ConnectionId -> ConnectionId -> Maybe (Substitution ConnectionId) -unifyConnectionId (Variable x) connId = Just $ Map.singleton x connId -unifyConnectionId (Value connId') connId | connId == connId' = Just Map.empty - | otherwise = Nothing + -- All other events can be arbitrarily reordered /across/ connections, but + -- never /within/ connections + insertEvent :: ExpEvent -> [ExpEvent] -> [[ExpEvent]] + insertEvent ev [] = [[ev]] + insertEvent ev (ev' : evs) = + (ev : ev' : evs) : + if eventConnId ev == eventConnId ev' + then [] + else [ev' : evs' | evs' <- insertEvent ev evs] --- | Apply a substitution -apply :: Substitution ConnectionId -> [ExpEvent] -> [ExpEvent] -apply subst = map applyEvent - where - applyEvent :: ExpEvent -> ExpEvent - applyEvent (ExpConnectionOpened connId) = ExpConnectionOpened (applyVar connId) - applyEvent (ExpConnectionClosed connId) = ExpConnectionClosed (applyVar connId) - applyEvent (ExpReceived connId payload) = ExpReceived (applyVar connId) payload + removeBundle :: BundleId -> [ExpEvent] -> [ExpEvent] + removeBundle bid = filter ((/= bid) . eventBundleId) - applyVar :: Variable ConnectionId -> Variable ConnectionId - applyVar (Value connId) = Value connId - applyVar (Variable x) = case Map.lookup x subst of - Just connId -> Value connId - Nothing -> Variable x + eventBundleId :: ExpEvent -> BundleId + eventBundleId (ExpConnectionOpened connInfo) = connectionBundle connInfo + eventBundleId (ExpConnectionClosed connInfo) = connectionBundle connInfo + eventBundleId (ExpReceived connInfo _) = connectionBundle connInfo + eventBundleId (ExpConnectionLost bid _) = bid --- | Return all possible reorderings of a list of expected events --- --- Events from different connections can be reordered, but events from the --- same connection cannot. -reorder :: [ExpEvent] -> [[ExpEvent]] -reorder = go - where - go :: [ExpEvent] -> [[ExpEvent]] - go [] = [[]] - go (ev : evs) = concat [insert ev evs' | evs' <- reorder evs] + eventConnId :: ExpEvent -> Maybe (Variable ConnectionId) + eventConnId (ExpConnectionOpened connInfo) = Just $ connectionId connInfo + eventConnId (ExpConnectionClosed connInfo) = Just $ connectionId connInfo + eventConnId (ExpReceived connInfo _) = Just $ connectionId connInfo + eventConnId (ExpConnectionLost _ _) = Nothing - insert :: ExpEvent -> [ExpEvent] -> [[ExpEvent]] - insert ev [] = [[ev]] - insert ev (ev' : evs') - | connectionId ev == connectionId ev' = [ev : ev' : evs'] - | otherwise = (ev : ev' : evs') : [ev' : evs'' | evs'' <- insert ev evs'] +-------------------------------------------------------------------------------- +-- Unification -- +-------------------------------------------------------------------------------- - connectionId :: ExpEvent -> Variable ConnectionId - connectionId (ExpConnectionOpened connId) = connId - connectionId (ExpConnectionClosed connId) = connId - connectionId (ExpReceived connId _) = connId +type Substitution = Map Unique ConnectionId +newtype Unifier a = Unifier { + runUnifier :: Substitution -> Maybe (a, Substitution) + } + +instance Monad Unifier where + return x = Unifier $ \subst -> Just (x, subst) + x >>= f = Unifier $ \subst -> case runUnifier x subst of + Nothing -> Nothing + Just (a, subst') -> runUnifier (f a) subst' + fail _str = mzero + +instance MonadPlus Unifier where + mzero = Unifier $ const Nothing + f `mplus` g = Unifier $ \subst -> case runUnifier f subst of + Nothing -> runUnifier g subst + Just (a, subst') -> Just (a, subst') + +class Unify a b where + unify :: a -> b -> Unifier () + +canUnify :: Unify a b => a -> b -> Bool +canUnify a b = isJust $ runUnifier (unify a b) Map.empty + +instance Unify Unique ConnectionId where + unify x cid = Unifier $ \subst -> + case Map.lookup x subst of + Just cid' -> if cid == cid' then Just ((), subst) + else Nothing + Nothing -> Just ((), Map.insert x cid subst) + +instance Unify (Variable ConnectionId) ConnectionId where + unify (Variable x) connId = unify x connId + unify (Value connId') connId = guard $ connId' == connId + +instance Unify ExpEvent Event where + unify (ExpConnectionOpened connInfo) (ConnectionOpened connId _ _) = + unify (connectionId connInfo) connId + unify (ExpConnectionClosed connInfo) (ConnectionClosed connId) = + unify (connectionId connInfo) connId + unify (ExpReceived connInfo payload) (Received connId payload') = do + guard $ BSS.concat payload == BSS.concat payload' + unify (connectionId connInfo) connId + unify (ExpConnectionLost _ addr) (ErrorEvent (TransportError (EventConnectionLost addr') _)) = + guard $ addr == addr' + unify _ _ = fail "Cannot unify" + +instance Unify a b => Unify [a] [b] where + unify [] [] = return () + unify (x:xs) (y:ys) = unify x y >> unify xs ys + unify _ _ = fail "Cannot unify" + -------------------------------------------------------------------------------- -- Script generators -- -------------------------------------------------------------------------------- @@ -290,7 +422,7 @@ script_ConnectClose :: Int -> Gen Script script_ConnectClose numEndPoints = do - script <- go Map.empty + script <- go Map.empty return (replicate numEndPoints NewEndPoint ++ script) where go :: Map Int Bool -> Gen Script @@ -300,15 +432,15 @@ 0 -> do fr <- choose (0, numEndPoints - 1) to <- choose (0, numEndPoints - 1) - cmds <- go (Map.insert (Map.size conns) True conns) + cmds <- go (Map.insert (Map.size conns) True conns) return (Connect fr to : cmds) 1 -> do - mConn <- choose (0, Map.size conns - 1) `suchThatMaybe` isOpen conns - case mConn of + mConn <- choose (0, Map.size conns - 1) `suchThatMaybe` isOpen conns + case mConn of Nothing -> go conns Just conn -> do cmds <- go (Map.insert conn False conns) - return (Close conn : cmds) + return (Close conn : cmds) _ -> return [] @@ -317,7 +449,7 @@ script_ConnectSendClose :: Int -> Gen Script script_ConnectSendClose numEndPoints = do - script <- go Map.empty + script <- go Map.empty return (replicate numEndPoints NewEndPoint ++ script) where go :: Map Int Bool -> Gen Script @@ -327,24 +459,24 @@ 0 -> do fr <- choose (0, numEndPoints - 1) to <- choose (0, numEndPoints - 1) - cmds <- go (Map.insert (Map.size conns) True conns) + cmds <- go (Map.insert (Map.size conns) True conns) return (Connect fr to : cmds) 1 -> do - mConn <- choose (0, Map.size conns - 1) `suchThatMaybe` isOpen conns - case mConn of + mConn <- choose (0, Map.size conns - 1) `suchThatMaybe` isOpen conns + case mConn of Nothing -> go conns Just conn -> do numSegments <- choose (0, 2) - payload <- replicateM numSegments arbitrary - cmds <- go conns - return (Send conn payload : cmds) + payload <- replicateM numSegments arbitrary + cmds <- go conns + return (Send conn payload : cmds) 2 -> do - mConn <- choose (0, Map.size conns - 1) `suchThatMaybe` isOpen conns - case mConn of + mConn <- choose (0, Map.size conns - 1) `suchThatMaybe` isOpen conns + case mConn of Nothing -> go conns Just conn -> do cmds <- go (Map.insert conn False conns) - return (Close conn : cmds) + return (Close conn : cmds) _ -> return [] @@ -360,7 +492,7 @@ insert <- arbitrary if insert && n > 0 then do - numReads <- chooseFrom' (NormalD { mean = 5, stdDev = 10 }) (0, 100) + numReads <- chooseFrom' NormalD { mean = 5, stdDev = 10 } (0, 100) swap <- arbitrary if swap then return $ Connect i j : BreakAfterReads numReads j i : cmds @@ -373,7 +505,7 @@ return $ cmd : cmds' -------------------------------------------------------------------------------- --- Individual scripts to test specific bugs -- +-- Individual scripts to test specific bugs -- -------------------------------------------------------------------------------- -- | Bug #1 @@ -381,15 +513,15 @@ -- When process A wants to close the heavyweight connection to process B it -- sends a CloseSocket request together with the ID of the last connection from -- B. When B receives the CloseSocket request it can compare this ID to the last --- connection it created; if they don't match, B knows that there are some --- messages still on the way from B to A (in particular, a CreatedConnection --- message) which will cancel the CloseSocket request from A. Hence, it will +-- connection it created; if they don't match, B knows that there are some +-- messages still on the way from B to A (in particular, a CreatedConnection +-- message) which will cancel the CloseSocket request from A. Hence, it will -- know to ignore the CloseSocket request from A. -- -- The bug was that we recorded the last _created_ outgoing connection on the -- local endpoint, but the last _received_ incoming connection on the state of -- the heavyweight connection. So, in the script below, the following happened: --- +-- -- A connects to B, records "last connection ID is 1024" -- A closes the lightweight connection, sends [CloseConnection 1024] -- A closes the heivyweight connection, sends [CloseSocket 0] @@ -409,7 +541,7 @@ -- A receives the [CloseSocket 0] request, compares it to the last recorded -- outgoing ID (1024), sees that they are not equal, and concludes that this -- must mean that there is still a CreatedConnection message on the way from A --- to B. +-- to B. -- -- This of course is not the case, so B will wait forever for A to confirm -- the CloseSocket request, and deadlock arises. (This deadlock doesn't become @@ -428,7 +560,19 @@ , Connect 1 0 ] --- | Simulate broken network connection during send +-- | Test ordering of sends +script_MultipleSends :: Script +script_MultipleSends = [ + NewEndPoint + , Connect 0 0 + , Send 0 ["A"] + , Send 0 ["B"] + , Send 0 ["C"] + , Send 0 ["D"] + , Send 0 ["E"] + ] + +-- | Simulate broken network connection during send script_BreakSend :: Script script_BreakSend = [ NewEndPoint @@ -439,8 +583,8 @@ ] -- | Simulate broken network connection during connect -script_BreakConnect :: Script -script_BreakConnect = [ +script_BreakConnect1 :: Script +script_BreakConnect1 = [ NewEndPoint , NewEndPoint , Connect 0 1 @@ -448,6 +592,38 @@ , Connect 0 1 ] +-- | Simulate broken network connection during connect +script_BreakConnect2 :: Script +script_BreakConnect2 = [ + NewEndPoint + , NewEndPoint + , Connect 0 1 + , BreakAfterReads 1 0 1 + , Connect 0 1 + ] + +-- | Simulate broken send, then reconnect +script_BreakSendReconnect :: Script +script_BreakSendReconnect = [ + NewEndPoint + , NewEndPoint + , Connect 0 1 + , BreakAfterReads 1 1 0 + , Send 0 ["ping1"] + , Connect 0 1 + , Send 1 ["ping2"] + ] + +script_Foo :: Script +script_Foo = [ + NewEndPoint + , NewEndPoint + , Connect 1 0 + , BreakAfterReads 2 0 1 + , Send 0 ["pingpong"] + , Connect 0 1 + ] + -------------------------------------------------------------------------------- -- Main application driver -- -------------------------------------------------------------------------------- @@ -462,15 +638,26 @@ tests :: (Transport, TransportInternals) -> [Test] tests transport = [ - testGroup "Specific scripts" [ - testOne "Bug1" transport script_Bug1 - , testOne "BreakSend" transport script_BreakSend - , testOne "BreakConnect" transport script_BreakConnect - ] - , testGroup "One endpoint, with delays" (basicTests transport 1 id) - , testGroup "Two endpoints, with delays" (basicTests transport 2 id) - , testGroup "Three endpoints, with delays" (basicTests transport 3 id) - , testGroup "Four endpoints, with delay, single error" (basicTests transport 4 (withErrors 1)) + testGroup "Regression tests" [ + testOne "Bug1" transport script_Bug1 + ] + , testGroup "Specific scripts" [ + testOne "BreakMultipleSends" transport script_MultipleSends + , testOne "BreakSend" transport script_BreakSend + , testOne "BreakConnect1" transport script_BreakConnect1 + , testOne "BreakConnect2" transport script_BreakConnect2 + , testOne "BreakSendReconnect" transport script_BreakSendReconnect + , testOne "Foo" transport script_Foo + ] + , testGroup "Without errors" [ + testGroup "One endpoint, with delays" (basicTests transport 1 id) + , testGroup "Two endpoints, with delays" (basicTests transport 2 id) + , testGroup "Three endpoints, with delays" (basicTests transport 3 id) + ] + , testGroup "Single error" [ + testGroup "Two endpoints, with delays" (basicTests transport 2 (withErrors 1)) + , testGroup "Three endpoints, with delays" (basicTests transport 3 (withErrors 1)) + ] ] where @@ -478,7 +665,7 @@ testOne label transport script = testCase label (testScript transport script) testGen :: TestName -> (Transport, TransportInternals) -> Gen Script -> Test -testGen label transport script = testProperty label (testScriptGen transport script) +testGen label transport script = testProperty label (testScriptGen transport script) main :: IO () main = do @@ -490,10 +677,10 @@ -------------------------------------------------------------------------------- testScriptGen :: (Transport, TransportInternals) -> Gen Script -> Property -testScriptGen transport scriptGen = - forAll scriptGen $ \script -> - morallyDubiousIOProperty $ do - logShow script +testScriptGen transport scriptGen = + forAll scriptGen $ \script -> + morallyDubiousIOProperty $ do + logShow script mErr <- try $ verify transport script return $ case mErr of Left (ExpectedFailure str) -> @@ -509,12 +696,12 @@ testScript :: (Transport, TransportInternals) -> Script -> Assertion testScript transport script = do - logShow script + logShow script mErr <- try $ verify transport script case mErr of - Left (ExpectedFailure _str) -> + Left (ExpectedFailure _str) -> return () - Right (Left err) -> + Right (Left err) -> assertFailure $ "Failed with script " ++ show script ++ ": " ++ err ++ "\n" Right (Right ()) -> return () @@ -529,10 +716,10 @@ endPointAtIx :: EndPointIx -> Accessor RunState EndPoint endPointAtIx i = endPoints >>> listAccessor i -connections :: Accessor RunState [(TargetAddress, Connection, Variable ConnectionId)] +connections :: Accessor RunState [(Connection, ConnectionInfo)] connections = accessor _connections (\cs st -> st { _connections = cs }) -connectionAt :: ConnectionIx -> Accessor RunState (TargetAddress, Connection, Variable ConnectionId) +connectionAt :: ConnectionIx -> Accessor RunState (Connection, ConnectionInfo) connectionAt i = connections >>> listAccessor i expectedEvents :: Accessor RunState (Map EndPointAddress [ExpEvent]) @@ -544,6 +731,23 @@ forwardingThreads :: Accessor RunState [ThreadId] forwardingThreads = accessor _forwardingThreads (\ts st -> st { _forwardingThreads = ts }) +mayBreak :: EndPointAddress -> EndPointAddress -> BundleId -> Accessor RunState Bool +mayBreak a b bid = aux >>> DAC.set (a, b, bid) + where + aux = accessor _mayBreak (\bs st -> st { _mayBreak = bs }) + +broken :: EndPointAddress -> EndPointAddress -> BundleId -> Accessor RunState Bool +broken a b bid = aux >>> DAC.set (a, b, bid) + where + aux = accessor _broken (\bs st -> st { _broken = bs }) + +currentBundle :: EndPointAddress -> EndPointAddress -> Accessor RunState BundleId +currentBundle i j = aux >>> if i < j then DAC.mapDefault 0 (i, j) + else DAC.mapDefault 0 (j, i) + where + aux :: Accessor RunState (Map (EndPointAddress, EndPointAddress) BundleId) + aux = accessor _currentBundle (\mp st -> st { _currentBundle = mp }) + -------------------------------------------------------------------------------- -- Pretty-printing -- -------------------------------------------------------------------------------- @@ -552,7 +756,7 @@ verticalList = PP.brackets . PP.vcat . map (PP.text . show) instance Show Script where - show = ("\n" ++) . show . verticalList + show = ("\n" ++) . show . verticalList instance Show [Event] where show = ("\n" ++) . show . verticalList @@ -560,6 +764,16 @@ instance Show [ExpEvent] where show = ("\n" ++) . show . verticalList +instance Show (Map EndPointAddress [ExpEvent]) where + show = ("\n" ++) . show . PP.brackets . PP.vcat + . map (\(addr, evs) -> PP.hcat . PP.punctuate PP.comma $ [PP.text (show addr), verticalList evs]) + . Map.toList + +instance Show (Map EndPointAddress [Event]) where + show = ("\n" ++) . show . PP.brackets . PP.vcat + . map (\(addr, evs) -> PP.hcat . PP.punctuate PP.comma $ [PP.text (show addr), verticalList evs]) + . Map.toList + -------------------------------------------------------------------------------- -- Draw random values from probability distributions -- -------------------------------------------------------------------------------- @@ -570,25 +784,25 @@ probabilityOf :: d -> Double -> Double instance Distribution NormalD where - probabilityOf d x = a * exp (-0.5 * b * b) + probabilityOf d x = a * exp (-0.5 * b * b) where a = 1 / (stdDev d * sqrt (2 * pi)) b = (x - mean d) / stdDev d --- | Choose from a distribution +-- | Choose from a distribution chooseFrom :: Distribution d => d -> (Double, Double) -> Gen Double -chooseFrom d (lo, hi) = findCandidate +chooseFrom d (lo, hi) = findCandidate where - findCandidate :: Gen Double + findCandidate :: Gen Double findCandidate = do candidate <- choose (lo, hi) uniformSample <- choose (0, 1) if uniformSample < probabilityOf d candidate then return candidate - else findCandidate + else findCandidate chooseFrom' :: Distribution d => d -> (Int, Int) -> Gen Int -chooseFrom' d (lo, hi) = +chooseFrom' d (lo, hi) = round <$> chooseFrom d (fromIntegral lo, fromIntegral hi) -------------------------------------------------------------------------------- @@ -603,18 +817,24 @@ instance Arbitrary ByteString where arbitrary = do - len <- chooseFrom' (NormalD { mean = 5, stdDev = 10 }) (0, 100) + len <- chooseFrom' NormalD { mean = 5, stdDev = 10 } (0, 100) xs <- replicateM len arbitrary return (pack xs) listAccessor :: Int -> Accessor [a] a -listAccessor i = accessor (!! i) (error "listAccessor.set not defined") +listAccessor i = accessor (!! i) (error "listAccessor.set not defined") +append :: Monad m => Accessor st [a] -> a -> StateT st m () +append acc x = modify acc (snoc x) + snoc :: a -> [a] -> [a] snoc x xs = xs ++ [x] -groupByKey :: Ord a => [(a, b)] -> Map a [b] -groupByKey = Map.fromListWith (++) . map (second return) +groupByKey :: Ord a => [a] -> [(a, b)] -> Map a [b] +groupByKey keys = go (Map.fromList [(key, []) | key <- keys]) + where + go acc [] = Map.map reverse acc + go acc ((key, val) : rest) = go (Map.adjust (val :) key acc) rest -------------------------------------------------------------------------------- -- Expected failures (can't find explicit support for this in test-framework) -- diff --git a/tests/TestTCP.hs b/tests/TestTCP.hs --- a/tests/TestTCP.hs +++ b/tests/TestTCP.hs @@ -2,7 +2,7 @@ {-# OPTIONS_GHC -fno-warn-orphans #-} module Main where -import Prelude hiding +import Prelude hiding ( (>>=) , return , fail @@ -49,7 +49,7 @@ #ifdef USE_MOCK_NETWORK import qualified Network.Transport.TCP.Mock.Socket as N #else -import qualified Network.Socket as N +import qualified Network.Socket as N #endif ( sClose , ServiceName @@ -69,9 +69,9 @@ import GHC.IO.Exception (ioe_errno) import Foreign.C.Error (Errno(..), eADDRNOTAVAIL) import System.Timeout (timeout) -import Network.Transport.Tests (testTransport) +import Network.Transport.Tests (testTransport) import Network.Transport.Tests.Auxiliary (forkTry, runTests) -import Network.Transport.Tests.Traced +import Network.Transport.Tests.Traced instance Traceable ControlHeader where trace = traceShow @@ -80,7 +80,7 @@ trace = traceShow instance Traceable N.Socket where - trace = traceShow + trace = traceShow instance Traceable N.AddrInfo where trace = traceShow @@ -112,17 +112,17 @@ -- TEST 1: they connect to us, then drop the connection do - ConnectionOpened _ _ addr <- receive endpoint + ConnectionOpened _ _ addr <- receive endpoint True <- return $ addr == theirAddr - - ErrorEvent (TransportError (EventConnectionLost addr') _) <- receive endpoint - True <- return $ addr' == theirAddr + ErrorEvent (TransportError (EventConnectionLost addr') _) <- receive endpoint + True <- return $ addr' == theirAddr + return () -- TEST 2: after they dropped their connection to us, we now try to -- establish a connection to them. This should re-establish the broken - -- TCP connection. + -- TCP connection. tlog "Trying to connect to client" Right conn <- connect endpoint theirAddr ReliableOrdered defaultConnectHints @@ -131,39 +131,39 @@ -- closes the socket do Right () <- send conn ["ping"] - - ConnectionOpened cid _ addr <- receive endpoint + + ConnectionOpened cid _ addr <- receive endpoint True <- return $ addr == theirAddr - + Received cid' ["pong"] <- receive endpoint True <- return $ cid == cid' - + ErrorEvent (TransportError (EventConnectionLost addr') _) <- receive endpoint - True <- return $ addr' == theirAddr + True <- return $ addr' == theirAddr return () -- TEST 4: A subsequent send on an already-open connection will now break Left (TransportError SendFailed _) <- send conn ["ping2"] - -- *Pfew* + -- *Pfew* putMVar serverDone () client :: MVar EndPointAddress -> MVar EndPointAddress -> IO () client serverAddr clientAddr = do tlog "Client" clientPort <- nextPort - let ourAddress = encodeEndPointAddress "127.0.0.1" clientPort 0 - putMVar clientAddr ourAddress - + let ourAddress = encodeEndPointAddress "127.0.0.1" clientPort 0 + putMVar clientAddr ourAddress + -- Listen for incoming messages forkServer "127.0.0.1" clientPort 5 True throwIO $ \sock -> do - -- Initial setup + -- Initial setup 0 <- recvInt32 sock :: IO Int - _ <- recvWithLength sock + _ <- recvWithLength sock sendMany sock [encodeInt32 ConnectionRequestAccepted] - -- Server opens a logical connection + -- Server opens a logical connection CreatedNewConnection <- toEnum <$> (recvInt32 sock :: IO Int) 1024 <- recvInt32 sock :: IO LightweightConnectionId @@ -171,21 +171,21 @@ 1024 <- recvInt32 sock :: IO Int ["ping"] <- recvWithLength sock - -- Reply + -- Reply sendMany sock [encodeInt32 CreatedNewConnection, encodeInt32 (10002 :: Int)] - sendMany sock (encodeInt32 10002 : prependLength ["pong"]) + sendMany sock (encodeInt32 10002 : prependLength ["pong"]) -- Close the socket N.sClose sock - + -- Connect to the server Right (sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr True Nothing - + -- Open a new connection sendMany sock [encodeInt32 CreatedNewConnection, encodeInt32 (10003 :: Int)] - + -- Close the socket without closing the connection explicitly - -- The server should receive an error event + -- The server should receive an error event N.sClose sock -- | Test the behaviour of a premature CloseSocket request @@ -211,19 +211,19 @@ -- TEST 1: they connect to us, then send a CloseSocket. Since we don't -- have any outgoing connections, this means we will agree to close the - -- socket + -- socket do - ConnectionOpened cid _ addr <- receive endpoint + ConnectionOpened cid _ addr <- receive endpoint True <- return $ addr == theirAddr - - ConnectionClosed cid' <- receive endpoint + + ConnectionClosed cid' <- receive endpoint True <- return $ cid' == cid return () -- TEST 2: after they dropped their connection to us, we now try to -- establish a connection to them. This should re-establish the broken - -- TCP connection. + -- TCP connection. tlog "Trying to connect to client" Right conn <- connect endpoint theirAddr ReliableOrdered defaultConnectHints @@ -234,65 +234,65 @@ -- the socket gets closed do Right () <- send conn ["ping"] - + ConnectionOpened cid _ addr <- receive endpoint True <- return $ addr == theirAddr - + Received cid' ["pong"] <- receive endpoint True <- return $ cid' == cid - + ConnectionClosed cid'' <- receive endpoint True <- return $ cid'' == cid - + ErrorEvent (TransportError (EventConnectionLost addr') _) <- receive endpoint - True <- return $ addr' == theirAddr - + True <- return $ addr' == theirAddr + return () -- TEST 4: A subsequent send on an already-open connection will now break Left (TransportError SendFailed _) <- send conn ["ping2"] - -- *Pfew* + -- *Pfew* putMVar serverDone () client :: MVar EndPointAddress -> MVar EndPointAddress -> IO () client serverAddr clientAddr = do tlog "Client" clientPort <- nextPort - let ourAddress = encodeEndPointAddress "127.0.0.1" clientPort 0 - putMVar clientAddr ourAddress - + let ourAddress = encodeEndPointAddress "127.0.0.1" clientPort 0 + putMVar clientAddr ourAddress + -- Listen for incoming messages forkServer "127.0.0.1" clientPort 5 True throwIO $ \sock -> do - -- Initial setup + -- Initial setup 0 <- recvInt32 sock :: IO Int - _ <- recvWithLength sock + _ <- recvWithLength sock sendMany sock [encodeInt32 ConnectionRequestAccepted] - -- Server opens a logical connection + -- Server opens a logical connection CreatedNewConnection <- toEnum <$> (recvInt32 sock :: IO Int) - 1024 <- recvInt32 sock :: IO LightweightConnectionId + 1024 <- recvInt32 sock :: IO LightweightConnectionId -- Server sends a message 1024 <- recvInt32 sock :: IO Int ["ping"] <- recvWithLength sock - -- Reply + -- Reply sendMany sock [encodeInt32 CreatedNewConnection, encodeInt32 (10002 :: Int)] - sendMany sock (encodeInt32 (10002 :: Int) : prependLength ["pong"]) + sendMany sock (encodeInt32 (10002 :: Int) : prependLength ["pong"]) -- Send a CloseSocket even though there are still connections *in both -- directions* sendMany sock [encodeInt32 CloseSocket, encodeInt32 (1024 :: Int)] N.sClose sock - + -- Connect to the server Right (sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr True Nothing - + -- Open a new connection sendMany sock [encodeInt32 CreatedNewConnection, encodeInt32 (10003 :: Int)] - - -- Send a CloseSocket without sending a closeconnecton + + -- Send a CloseSocket without sending a closeconnecton -- The server should still receive a ConnectionClosed message sendMany sock [encodeInt32 CloseSocket, encodeInt32 (0 :: Int)] N.sClose sock @@ -311,26 +311,26 @@ Right endpoint <- newEndPoint transport -- Syntax error in the endpoint address - Left (TransportError ConnectFailed _) <- + Left (TransportError ConnectFailed _) <- connect endpoint (EndPointAddress "InvalidAddress") ReliableOrdered defaultConnectHints - + -- Syntax connect, but invalid hostname (TCP address lookup failure) - Left (TransportError ConnectNotFound _) <- + Left (TransportError ConnectNotFound _) <- connect endpoint (encodeEndPointAddress "invalidHost" "port" 0) ReliableOrdered defaultConnectHints - + -- TCP address correct, but nobody home at that address - Left (TransportError ConnectNotFound _) <- + Left (TransportError ConnectNotFound _) <- connect endpoint (encodeEndPointAddress "127.0.0.1" "9000" 0) ReliableOrdered defaultConnectHints - + -- Valid TCP address but invalid endpoint number - Left (TransportError ConnectNotFound _) <- + Left (TransportError ConnectNotFound _) <- connect endpoint (encodeEndPointAddress "127.0.0.1" port 1) ReliableOrdered defaultConnectHints return () -- | Test that an endpoint can ignore CloseSocket requests (in "reality" this -- would happen when the endpoint sends a new connection request before --- receiving an (already underway) CloseSocket request) +-- receiving an (already underway) CloseSocket request) testIgnoreCloseSocket :: IO N.ServiceName -> IO () testIgnoreCloseSocket nextPort = do serverAddr <- newEmptyMVar @@ -397,7 +397,7 @@ -- Close it again tlog "Closing connection" - sendMany sock [encodeInt32 CloseConnection, encodeInt32 (1024 :: Int)] + sendMany sock [encodeInt32 CloseConnection, encodeInt32 (1024 :: Int)] -- And close the connection completely tlog "Closing socket" @@ -471,7 +471,7 @@ unblocked <- newMVar False -- We should not hear from the server until we unblock him by - -- responding to the CloseSocket request (in this case, we + -- responding to the CloseSocket request (in this case, we -- respond by sending a ConnectionRequest) forkTry $ do recvInt32 sock :: IO Int32 @@ -490,7 +490,7 @@ -- | Test what happens when a remote endpoint sends a connection request to our -- transport for an endpoint it already has a connection to -testUnnecessaryConnect :: IO N.ServiceName -> Int -> IO () +testUnnecessaryConnect :: IO N.ServiceName -> Int -> IO () testUnnecessaryConnect nextPort numThreads = do clientDone <- newEmptyMVar serverAddr <- newEmptyMVar @@ -501,22 +501,22 @@ putMVar serverAddr (address endpoint) forkTry $ do - -- We pick an address < 127.0.0.1 so that this is not rejected purely because of the "crossed" check + -- We pick an address < 127.0.0.1 so that this is not rejected purely because of the "crossed" check let ourAddress = EndPointAddress "126.0.0.1" -- We should only get a single 'Accepted' reply gotAccepted <- newEmptyMVar dones <- replicateM numThreads $ do - done <- newEmptyMVar + done <- newEmptyMVar forkTry $ do -- It is possible that the remote endpoint just rejects the request by closing the socket -- immediately (depending on far the remote endpoint got with the initialization) - response <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr True Nothing + response <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr True Nothing case response of - Right (_, ConnectionRequestAccepted) -> + Right (_, ConnectionRequestAccepted) -> -- We don't close this socket because we want to keep this connection open putMVar gotAccepted () - -- We might get either Invalid or Crossed (the transport does not + -- We might get either Invalid or Crossed (the transport does not -- maintain enough history to be able to tell) Right (sock, ConnectionRequestInvalid) -> N.sClose sock @@ -536,7 +536,7 @@ testMany :: IO N.ServiceName -> IO () testMany nextPort = do Right masterTransport <- nextPort >>= \port -> createTransport "127.0.0.1" port defaultTCPParameters - Right masterEndPoint <- newEndPoint masterTransport + Right masterEndPoint <- newEndPoint masterTransport replicateM_ 10 $ do mTransport <- nextPort >>= \port -> createTransport "127.0.0.1" port defaultTCPParameters @@ -544,7 +544,7 @@ Left ex -> do putStrLn $ "IOException: " ++ show ex ++ "; errno = " ++ show (ioe_errno ex) case (ioe_errno ex) of - Just no | Errno no == eADDRNOTAVAIL -> putStrLn "(ADDRNOTAVAIL)" + Just no | Errno no == eADDRNOTAVAIL -> putStrLn "(ADDRNOTAVAIL)" _ -> return () throwIO ex Right transport -> @@ -561,7 +561,7 @@ killThread (transportThread internals) -- Uh oh - ErrorEvent (TransportError EventTransportFailed _) <- receive endpoint + ErrorEvent (TransportError EventTransportFailed _) <- receive endpoint return () @@ -580,16 +580,16 @@ endpointCreated <- newEmptyMVar -- Server - forkTry $ do + forkTry $ do -- Wait for the client to do its first attempt - readMVar firstAttempt + readMVar firstAttempt - counter <- newMVar (0 :: Int) + counter <- newMVar (0 :: Int) forkServer "127.0.0.1" serverPort 5 True throwIO $ \sock -> do - -- Accept the connection + -- Accept the connection Right 0 <- tryIO $ (recvInt32 sock :: IO Int) - Right _ <- tryIO $ recvWithLength sock + Right _ <- tryIO $ recvWithLength sock Right () <- tryIO $ sendMany sock [encodeInt32 ConnectionRequestAccepted] -- The first time we close the socket before accepting the logical connection @@ -604,10 +604,10 @@ when (count > 1) $ do -- Client sends a message Right connId' <- tryIO $ (recvInt32 sock :: IO LightweightConnectionId) - True <- return $ connId == connId' + True <- return $ connId == connId' Right ["ping"] <- tryIO $ recvWithLength sock putMVar serverDone () - + Right () <- tryIO $ N.sClose sock return () @@ -626,17 +626,17 @@ -- The second attempt will fail because the server closes the socket before we can request a connection takeMVar endpointCreated -- This might time out or not, depending on whether the server closes the - -- socket before or after we can send the RequestConnectionId request - resultConnect <- timeout 500000 $ connect endpoint theirAddr ReliableOrdered defaultConnectHints + -- socket before or after we can send the RequestConnectionId request + resultConnect <- timeout 500000 $ connect endpoint theirAddr ReliableOrdered defaultConnectHints case resultConnect of Nothing -> return () Just (Left (TransportError ConnectFailed _)) -> return () Just (Left err) -> throwIO err - Just (Right _) -> throwIO $ userError "testConnect: unexpected connect success" + Just (Right _) -> throwIO $ userError "testConnect: unexpected connect success" -- The third attempt succeeds Right conn1 <- connect endpoint theirAddr ReliableOrdered defaultConnectHints - + -- But a send will fail because the server has closed the connection again threadDelay 100000 Left (TransportError SendFailed _) <- send conn1 ["ping"] @@ -675,12 +675,12 @@ CreatedNewConnection <- toEnum <$> (recvInt32 sock :: IO Int) connId <- recvInt32 sock :: IO LightweightConnectionId - + connId' <- recvInt32 sock :: IO LightweightConnectionId True <- return $ connId == connId' ["ping"] <- recvWithLength sock putMVar serverGotPing () - + -- Client forkTry $ do Right (transport, internals) <- nextPort >>= \port -> createTransportExposeInternals "127.0.0.1" port defaultTCPParameters @@ -693,12 +693,12 @@ takeMVar serverGotPing -- Close the *outgoing* part of the socket only - sock <- socketBetween internals (address endpoint) theirAddr + sock <- socketBetween internals (address endpoint) theirAddr N.shutdown sock N.ShutdownSend -- At this point we cannot notice the problem yet so we shouldn't receive an event yet Nothing <- timeout 500000 $ receive endpoint - + -- But when we send we find the error Left (TransportError SendFailed _) <- send conn1 ["ping"] ErrorEvent (TransportError (EventConnectionLost _) _) <- receive endpoint @@ -709,7 +709,7 @@ takeMVar serverGotPing -- Again, close the outgoing part of the socket - sock' <- socketBetween internals (address endpoint) theirAddr + sock' <- socketBetween internals (address endpoint) theirAddr N.shutdown sock' N.ShutdownSend -- We now find the error when we attempt to close the connection @@ -720,10 +720,10 @@ send conn3 ["ping"] takeMVar serverGotPing - -- We repeat once more. - sock'' <- socketBetween internals (address endpoint) theirAddr + -- We repeat once more. + sock'' <- socketBetween internals (address endpoint) theirAddr N.shutdown sock'' N.ShutdownSend - + -- Now we notice the problem when we try to connect Nothing <- timeout 500000 $ receive endpoint Left (TransportError ConnectFailed _) <- connect endpoint theirAddr ReliableOrdered defaultConnectHints @@ -754,7 +754,7 @@ -- connection ErrorEvent (TransportError (EventConnectionLost _) _) <- receive endpoint - putMVar serverDone () + putMVar serverDone () -- Client forkTry $ do @@ -766,7 +766,7 @@ Right _ <- connect endpoint theirAddr ReliableOrdered defaultConnectHints -- Get a handle on the TCP connection and manually send an invalid CloseConnection request - sock <- socketBetween internals ourAddr theirAddr + sock <- socketBetween internals ourAddr theirAddr sendMany sock [encodeInt32 CloseConnection, encodeInt32 (12345 :: Int)] putMVar clientDone () @@ -776,23 +776,23 @@ main :: IO () main = do portMVar <- newEmptyMVar - forkTry $ forM_ ([10080 ..] :: [Int]) $ putMVar portMVar . show - let nextPort = takeMVar portMVar - tcpResult <- tryIO $ runTests + forkTry $ forM_ ([10080 ..] :: [Int]) $ putMVar portMVar . show + let nextPort = takeMVar portMVar + tcpResult <- tryIO $ runTests [ ("EarlyDisconnect", testEarlyDisconnect nextPort) , ("EarlyCloseSocket", testEarlyCloseSocket nextPort) , ("IgnoreCloseSocket", testIgnoreCloseSocket nextPort) , ("BlockAfterCloseSocket", testBlockAfterCloseSocket nextPort) , ("UnnecessaryConnect", testUnnecessaryConnect nextPort 10) , ("InvalidAddress", testInvalidAddress nextPort) - , ("InvalidConnect", testInvalidConnect nextPort) + , ("InvalidConnect", testInvalidConnect nextPort) , ("Many", testMany nextPort) , ("BreakTransport", testBreakTransport nextPort) , ("Reconnect", testReconnect nextPort) , ("UnidirectionalError", testUnidirectionalError nextPort) , ("InvalidCloseConnection", testInvalidCloseConnection nextPort) ] - -- Run the generic tests even if the TCP specific tests failed.. + -- Run the generic tests even if the TCP specific tests failed.. testTransport (either (Left . show) (Right) <$> nextPort >>= \port -> createTransport "127.0.0.1" port defaultTCPParameters) -- ..but if the generic tests pass, still fail if the specific tests did not case tcpResult of