network-transport-tcp 0.3.1 → 0.4.0
raw patch · 7 files changed
+984/−726 lines, 7 filesdep ~network-transportdep ~network-transport-tcpnew-uploaderPVP ok
version bump matches the API change (PVP)
Dependency ranges changed: network-transport, network-transport-tcp
API changes (from Hackage documentation)
Files
- ChangeLog +34/−0
- network-transport-tcp.cabal +30/−26
- src/Network/Transport/TCP.hs +368/−368
- src/Network/Transport/TCP/Internal.hs +22/−22
- src/Network/Transport/TCP/Mock/Socket.hs +47/−47
- tests/TestQC.hs +385/−165
- tests/TestTCP.hs +98/−98
+ ChangeLog view
@@ -0,0 +1,34 @@+2014-05-30 Tim Watson <watson.timothy@gmail.com> 0.4.0++* Update dependencies++2012-10-19 Edsko de Vries <edsko@well-typed.com> 0.3.1++* Bugfix. Reconnecting between endpoints did not work under certain+circumstances.++2012-10-03 Edsko de Vries <edsko@well-typed.com> 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 <edsko@well-typed.com> 0.2.0.3++* Allow for colons in hostnames (for IPv6)++2012-07-16 Edsko de Vries <edsko@well-typed.com> 0.2.0.2++* Base 4.6 compatibility+* Relax package contraints on bytestring and containers++2012-07-12 Edsko de Vries <edsko@well-typed.com> 0.2.0.1++* Fix bug in recvExact++2012-07-07 Edsko de Vries <edsko@well-typed.com> 0.2.0++* Initial release.
network-transport-tcp.cabal view
@@ -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
src/Network/Transport/TCP.hs view
@@ -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
src/Network/Transport/TCP/Internal.hs view
@@ -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))
src/Network/Transport/TCP/Mock/Socket.hs view
@@ -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 --
tests/TestQC.hs view
@@ -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) --
tests/TestTCP.hs view
@@ -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