network-transport-tcp 0.3.1 → 0.8.6
raw patch · 9 files changed
Files
- ChangeLog +113/−0
- LICENSE +1/−0
- Setup.hs +0/−2
- network-transport-tcp.cabal +66/−43
- src/Network/Transport/TCP.hs +2232/−1650
- src/Network/Transport/TCP/Internal.hs +297/−67
- src/Network/Transport/TCP/Mock/Socket.hs +47/−47
- tests/TestQC.hs +385/−165
- tests/TestTCP.hs +606/−265
+ ChangeLog view
@@ -0,0 +1,113 @@+2025-02-04 Laurent P. René de Cotret <laurent.decotret@outlook.com> 0.8.6++* Ported test suite to use `tasty` rather than `test-framework`.+* Addressed some compilation warnings (#467)++2024-09-03 Laurent P. René de Cotret <laurent.decotret@outlook.com> 0.8.5++* Bumped dependency bounds to support GHC 8.10.7 - GHC 9.10.1+* Updated links to point to Distributed Haskell monorepo++2024-07-09 Laurent P. René de Cotret <laurent.decotret@outlook.com> 0.8.4++* Set TCP_NODELAY by default. The documentation erroneously reported it as the default. (#66)++2023-03-26 David Simmons-Duffin <dsd@caltech.edu> 0.8.3++* Disable a flaky test++2023-03-25 David Simmons-Duffin <dsd@caltech.edu> 0.8.2++* Bump bytestring version to build with GHC 9.8.++2022-10-12 FacundoDominguez <facundo.dominguez@tweag.io> 0.8.1++* Fix imports for ghc 9 (#93).++2020-10-09 FacundoDominguez <facundo.dominguez@tweag.io> 0.8.0++* Move to network 3.++2019-12-31 FacundoDominguez <facundo.dominguez@tweag.io> 0.7.0++* Added support for unaddressable endpoints. (#61)++2019-12-31 FacundoDominguez <facundo.dominguez@tweag.io> 0.6.1++* Relax dependency bounds to build with ghc-8.6.5+* apiSend RELY violation is removed for closed remote endpoints (#70)+* The server no longer needs crash if accept throws an exception.+* Check peer-reported host against socket host (#54)+* Fix possible endless waiting on the 'crossed' MVar (#74)+* Fix possible msg corruption on a busy network (#85)++2017-08-21 FacundoDominguez <facundo.dominguez@tweag.io> 0.6.0++* Implemented protocol versioning (#55)+* Extend interface so queue policies and lengths can be configured.+* Test improvements+* Fix races when an EndPoint is closed or failed (#60)+* Fix timeout socket connections (#53)+* Use equality rather than ordering in socket close (#56)+* apiCloseEndPoint blocks until no longer reciving+* Shutdown sockets when closing endpoints+* Allow computing the external address from the chosen bind port (#50)+* Discard remote endpoints when they close or fail (#43)++2016-02-17 FacundoDominguez <facundo.dominguez@tweag.io> 0.5.0++* Add TCP_KEEPALIVE support for sockets.+* Run nearly all tests on non hardcoded ports.+* Remove obsolete top-level Makefile.+* Yield an error when tcpUserTimeout is set in unsupported systems.+* Fix for NTTCP-10. Have apiConnect respect timeouts.+* Make schedule - runScheduled pair exception safe.+* Allow to specify a default timeout for all connect calls.+* Allow to set TCP_USER_TIMEOUT on tcp connections.+* Implement configuration parameter to set TCP_NODELAY.+* Fix for NTTCP-9 / #23. Handle network failures when connection requests cross.++2015-06-15 FacundoDominguez <facundo.dominguez@tweag.io> 0.4.2++* Update dependencies.+* Fixes in test-suite.+* Bug fixes DP-109, NTTCP-11.++2014-12-09 Tim Watson <watson.timothy@gmail.com> 0.4.1++* Update dependencies++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.
LICENSE view
@@ -1,4 +1,5 @@ Copyright Well-Typed LLP, 2011-2012+Copyright Tweag I/O Limited, 2015 All rights reserved.
− Setup.hs
@@ -1,2 +0,0 @@-import Distribution.Simple-main = defaultMain
network-transport-tcp.cabal view
@@ -1,84 +1,107 @@+cabal-version: 3.0 Name: network-transport-tcp-Version: 0.3.1-Cabal-Version: >=1.8+Version: 0.8.6 Build-Type: Simple-License: BSD3 +License: BSD-3-Clause License-file: LICENSE-Copyright: Well-Typed LLP+Copyright: Well-Typed LLP, Tweag I/O Limited Author: Duncan Coutts, Nicolas Wu, Edsko de Vries-Maintainer: edsko@well-typed.com, dcoutts@well-typed.com+maintainer: The Distributed Haskell team 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://github.com/haskell-distributed/distributed-process/issues 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 +Description: TCP instantiation of Network.Transport, which can be used to create Cloud Haskell backends+tested-with: GHC==8.10.7 GHC==9.0.2 GHC==9.2.8 GHC==9.4.5 GHC==9.6.4 GHC==9.8.2 GHC==9.10.1 GHC==9.12.1+Category: Network+extra-doc-files: ChangeLog -Source-Repository head+source-repository head Type: git Location: https://github.com/haskell-distributed/distributed-process- SubDir: network-transport-tcp+ SubDir: packages/network-transport-tcp +common warnings+ ghc-options: -Wall+ -Wcompat+ -Widentities+ -Wincomplete-uni-patterns+ -Wincomplete-record-updates+ -Wredundant-constraints+ -fhide-source-paths+ -Wpartial-fields+ -Wunused-packages+ Flag use-mock-network Description: Use mock network implementation (for testing) Default: False Library- Build-Depends: base >= 4.3 && < 5,- network-transport >= 0.3 && < 0.4,+ import: warnings+ Build-Depends: base >= 4.14 && < 5,+ async >= 2.2 && < 2.3,+ network-transport >= 0.5 && < 0.6, data-accessor >= 0.2 && < 0.3,- containers >= 0.4 && < 0.6,- bytestring >= 0.9 && < 0.11,- network >= 2.3 && < 2.5+ containers >= 0.6 && < 0.8,+ bytestring >= 0.10 && < 0.13,+ network >= 3.1 && < 3.3,+ uuid >= 1.3 && < 1.4 Exposed-modules: Network.Transport.TCP, Network.Transport.TCP.Internal- Extensions: CPP- ghc-options: -Wall -fno-warn-unused-do-bind+ Default-Extensions: CPP+ default-language: Haskell2010+ Other-Extensions: RecursiveDo+ ghc-options: -fno-warn-unused-do-bind HS-Source-Dirs: src If flag(use-mock-network) CPP-Options: -DUSE_MOCK_NETWORK Exposed-modules: Network.Transport.TCP.Mock.Socket Network.Transport.TCP.Mock.Socket.ByteString -Test-Suite TestTCP +Test-Suite TestTCP+ import: warnings Type: exitcode-stdio-1.0 Main-Is: TestTCP.hs- 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+ Build-Depends: base >= 4.14 && < 5,+ bytestring >= 0.10,+ network-transport-tests >= 0.3 && < 0.4,+ network >= 3.1,+ network-transport,+ network-transport-tcp ghc-options: -threaded -rtsopts -with-rtsopts=-N HS-Source-Dirs: tests- Extensions: CPP,+ default-extensions: CPP, OverloadedStrings+ default-language: Haskell2010 If flag(use-mock-network) CPP-Options: -DUSE_MOCK_NETWORK Test-Suite TestQC+ import: warnings 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.14 && < 5,+ 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+ default-extensions: TypeSynonymInstances FlexibleInstances OverlappingInstances OverloadedStrings DeriveDataTypeable+ MultiParamTypeClasses+ default-language: Haskell2010
src/Network/Transport/TCP.hs view
@@ -1,1651 +1,2233 @@--- | 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. ------ 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 - ( -- * Main API- createTransport- , TCPParameters(..)- , defaultTCPParameters- -- * Internals (exposed for unit tests) - , createTransportExposeInternals - , TransportInternals(..)- , EndPointId- , encodeEndPointAddress- , decodeEndPointAddress- , ControlHeader(..)- , ConnectionRequestResponse(..)- , firstNonReservedLightweightConnectionId- , firstNonReservedHeavyweightConnectionId- , socketToEndPoint - , LightweightConnectionId- -- * Design notes- -- $design- ) where--import Prelude hiding - ( mapM_-#if ! MIN_VERSION_base(4,6,0)- , catch-#endif- )- -import Network.Transport-import Network.Transport.TCP.Internal - ( forkServer- , recvWithLength- , recvInt32- , tryCloseSocket- )-import Network.Transport.Internal - ( encodeInt32- , prependLength- , mapIOException- , tryIO- , tryToEnum- , void- , timeoutMaybe- , asyncWhenCancelled- )--#ifdef USE_MOCK_NETWORK-import qualified Network.Transport.TCP.Mock.Socket as N-#else-import qualified Network.Socket as N -#endif- ( HostName- , ServiceName- , Socket- , getAddrInfo- , socket- , addrFamily- , addrAddress- , SocketType(Stream)- , defaultProtocol- , setSocketOption- , SocketOption(ReuseAddr) - , connect- , sOMAXCONN- , AddrInfo- )--#ifdef USE_MOCK_NETWORK-import Network.Transport.TCP.Mock.Socket.ByteString (sendMany)-#else-import Network.Socket.ByteString (sendMany)-#endif--import Control.Concurrent (forkIO, ThreadId, killThread, myThreadId)-import Control.Concurrent.Chan (Chan, newChan, readChan, writeChan)-import Control.Concurrent.MVar - ( MVar- , newMVar- , modifyMVar- , modifyMVar_- , readMVar- , putMVar- , newEmptyMVar- , withMVar- )-import Control.Category ((>>>))-import Control.Applicative ((<$>))-import Control.Monad (when, unless, join)-import Control.Exception - ( IOException- , SomeException- , AsyncException- , handle- , throw- , throwIO- , try- , bracketOnError- , fromException- , catch- )-import Data.IORef (IORef, newIORef, writeIORef, readIORef)-import Data.ByteString (ByteString)-import qualified Data.ByteString as BS (concat)-import qualified Data.ByteString.Char8 as BSC (pack, unpack)-import Data.Bits (shiftL, (.|.))-import Data.Word (Word32)-import Data.Set (Set)-import qualified Data.Set as Set - ( empty- , insert- , elems- , singleton- , null- , delete- , member- )-import Data.Map (Map)-import qualified Data.Map as Map (empty)-import Data.Accessor (Accessor, accessor, (^.), (^=), (^:)) -import qualified Data.Accessor.Container as DAC (mapMaybe)-import Data.Foldable (forM_, mapM_)---- $design ------ [Goals]------ The TCP transport maps multiple logical connections between /A/ and /B/ (in--- either direction) to a single TCP connection:------ > +-------+ +-------+--- > | A |==========================| B |--- > | |>~~~~~~~~~~~~~~~~~~~~~~~~~|~~~\ |--- > | Q |>~~~~~~~~~~~~~~~~~~~~~~~~~|~~~Q |--- > | \~~~|~~~~~~~~~~~~~~~~~~~~~~~~~<| |--- > | |==========================| |--- > +-------+ +-------+--- --- 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--- closed.------ [Connecting]------ Let /A/, /B/ be two endpoints without any connections. When /A/ wants to--- connect to /B/, it locally records that it is trying to connect to /B/ and--- sends a request to /B/. As part of the request /A/ sends its own endpoint--- address to /B/ (so that /B/ can reuse the connection in the other direction).------ When /B/ receives the connection request it first checks if it did not--- already initiate a connection request to /A/. If not it will acknowledge the--- connection request by sending 'ConnectionRequestAccepted' to /A/ and record--- that it has a TCP connection to /A/.------ The tricky case arises when /A/ sends a connection request to /B/ and /B/--- finds that it had already sent a connection request to /A/. In this case /B/--- will accept the connection request from /A/ if /A/s endpoint address is--- smaller (lexicographically) than /B/s, and reject it otherwise. If it rejects--- it, it sends a 'ConnectionRequestCrossed' message to /A/. (The--- lexicographical ordering is an arbitrary but convenient way to break the--- tie.)------ When it receives a 'ConnectionRequestCrossed' message the /A/ thread that--- initiated the request just needs to wait until the /A/ thread that is dealing--- 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/:------ > +-------+ +-------+--- > | A |==========================| B |--- > | |>~~~~~~~~~~~~~~~~~~~~~~~~~|~~~\ |--- > | | | Q |--- > | #| | |--- > | |==========================| |--- > +-------+ +-------+------ 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). ------ 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.) ------ 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--- requests to /B/ after having sent a 'CloseSocket' request, and will actually--- close its end of the socket only when receiving the 'CloseSocket' message--- 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. ------------------------------------------------------------------------------------- Internal datatypes --------------------------------------------------------------------------------------- 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. --- * For the case of "valid" we use third data structure to give more details--- about the state (ValidTransportState, ValidLocalEndPointState,--- ValidRemoteEndPointState).--data TCPTransport = TCPTransport - { transportHost :: !N.HostName- , transportPort :: !N.ServiceName- , transportState :: !(MVar TransportState)- , transportParams :: !TCPParameters- }--data TransportState = - TransportValid !ValidTransportState- | TransportClosed--data ValidTransportState = ValidTransportState - { _localEndPoints :: !(Map EndPointAddress LocalEndPoint)- , _nextEndPointId :: !EndPointId - }--data LocalEndPoint = LocalEndPoint- { localAddress :: !EndPointAddress- , localChannel :: !(Chan Event)- , localState :: !(MVar LocalEndPointState)- }--data LocalEndPointState = - LocalEndPointValid !ValidLocalEndPointState- | LocalEndPointClosed--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- , _nextConnInId :: !HeavyweightConnectionId- -- Currently active outgoing heavyweight connections- , _localConnections :: !(Map EndPointAddress RemoteEndPoint)- }---- REMOTE ENDPOINTS ------ Remote endpoints (basically, TCP connections) have the following lifecycle:------ Init ---+---> Invalid--- |--- +-------------------------------\--- | |--- | /----------\ |--- | | | |--- | v | v--- +---> Valid ---> Closing ---> Closed--- | | | |--- | | | v--- \-------+----------+--------> Failed------ Init: There are two places where we create new remote endpoints: in--- createConnectionTo (in response to an API 'connect' call) and in--- handleConnectionRequest (when a remote node tries to connect to us).--- 'Init' carries an MVar () 'resolved' which concurrent threads can use to--- wait for the remote endpoint to finish initialization. We record who--- requested the connection (the local endpoint or the remote endpoint).------ Invalid: We put the remote endpoint in invalid state only during--- 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--- concurrent threads can use to wait for the remote endpoint to either be--- closed fully (if the communication parnet responds with another--- CloseSocket) or be put back in 'Valid' state if the remote endpoint denies--- the request.------ We also put the endpoint in Closed state, directly from Init, if we our--- outbound connection request crossed an inbound connection request and we--- decide to keep the inbound (i.e., the remote endpoint sent us a--- ConnectionRequestCrossed message).------ Closed: The endpoint is put in Closed state after a successful garbage--- collection.------ Failed: If the connection to the remote endpoint is lost, or the local--- endpoint (or the whole transport) is closed manually, the remote endpoint is--- put in Failed state, and we record the reason.------ Invariants for dealing with remote endpoints:------ INV-SEND: Whenever we send data the remote endpoint must be locked (to avoid--- interleaving bits of payload).------ INV-CLOSE: Local endpoints should never point to remote endpoint in closed--- state. Whenever we put an endpoint in Closed state we remove that--- endpoint from localConnections first, so that if a concurrent thread reads--- the MVar, finds RemoteEndPointClosed, and then looks up the endpoint in--- 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'. ------ 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. ------ 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--- acceptable, of course -- as we are waiting for the remote endpoint to--- confirm or deny the request).------ INV-LOCK-ORDER: Remote endpoint must be locked before their local endpoints.--- In other words: it is okay to call modifyMVar on a local endpoint inside a--- modifyMVar on a remote endpoint, but not the other way around. In--- particular, it is okay to call removeRemoteEndPoint inside--- modifyRemoteState.--data RemoteEndPoint = RemoteEndPoint - { remoteAddress :: !EndPointAddress- , remoteState :: !(MVar RemoteState)- , remoteId :: !HeavyweightConnectionId- , remoteScheduled :: !(Chan (IO ()))- }--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 - -- | "Normal" working endpoint- | RemoteEndPointValid !ValidRemoteEndPointState - -- | The remote endpoint is being closed (garbage collected)- | RemoteEndPointClosing !(MVar ()) !ValidRemoteEndPointState- -- | The remote endpoint has been closed (garbage collected)- | RemoteEndPointClosed- -- | The remote endpoint has failed, or has been forcefully shutdown- -- using a closeTransport or closeEndPoint API call- | RemoteEndPointFailed !IOException---- 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 - { _remoteOutgoing :: !Int- , _remoteIncoming :: !(Set LightweightConnectionId)- , _remoteMaxIncoming :: !LightweightConnectionId- , _remoteNextConnOutId :: !LightweightConnectionId- , remoteSocket :: !N.Socket- , remoteSendLock :: !(MVar ())- }---- | Local identifier for an endpoint within this transport-type EndPointId = Word32---- | Pair of local and a remote endpoint (for conciseness in signatures)-type EndPointPair = (LocalEndPoint, RemoteEndPoint)---- | Lightweight connection ID (sender allocated)------ A ConnectionId is the concentation of a 'HeavyweightConnectionId' and a--- 'LightweightConnectionId'.-type LightweightConnectionId = Word32---- | Heavyweight connection ID (recipient allocated)------ A ConnectionId is the concentation of a 'HeavyweightConnectionId' and a--- 'LightweightConnectionId'.-type HeavyweightConnectionId = Word32---- | Control headers -data ControlHeader = - -- | Tell the remote endpoint that we created a new connection- CreatedNewConnection - -- | Tell the remote endpoint we will no longer be using a connection- | CloseConnection - -- | Request to close the connection (see module description)- | CloseSocket - deriving (Enum, Bounded, Show)---- | Response sent by /B/ to /A/ when /A/ tries to connect-data ConnectionRequestResponse =- -- | /B/ accepts the connection- ConnectionRequestAccepted - -- | /A/ requested an invalid endpoint- | ConnectionRequestInvalid - -- | /A/s request crossed with a request from /B/ (see protocols)- | ConnectionRequestCrossed - deriving (Enum, Bounded, Show)---- | Parameters for setting up the TCP transport-data TCPParameters = TCPParameters {- -- | Backlog for 'listen'.- -- Defaults to SOMAXCONN.- tcpBacklog :: Int- -- | Should we set SO_REUSEADDR on the server socket? - -- Defaults to True.- , tcpReuseServerAddr :: Bool- -- | Should we set SO_REUSEADDR on client sockets?- -- Defaults to True.- , tcpReuseClientAddr :: Bool- }---- | Internal functionality we expose for unit testing-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 - }------------------------------------------------------------------------------------- Top-level functionality --------------------------------------------------------------------------------------- | Create a TCP transport-createTransport :: N.HostName - -> N.ServiceName - -> TCPParameters- -> IO (Either IOException Transport)-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 - -> TCPParameters- -> 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) - (tcpReuseServerAddr params)- (terminationHandler transport) - (handleConnectionRequest transport))- killThread- (mkTransport transport)- where- mkTransport :: TCPTransport - -> ThreadId - -> IO (Transport, TransportInternals)- mkTransport transport tid = return- ( Transport - { newEndPoint = apiNewEndPoint transport - , closeTransport = let evs = [ EndPointClosed- , throw $ userError "Transport closed"- ] in- apiCloseTransport transport (Just tid) evs - } - , TransportInternals - { transportThread = tid- , socketBetween = internalSocketBetween transport- }- )-- terminationHandler :: TCPTransport -> SomeException -> IO ()- terminationHandler transport ex = do - let evs = [ ErrorEvent (TransportError EventTransportFailed (show ex))- , throw $ userError "Transport closed" - ]- apiCloseTransport transport Nothing evs ---- | Default TCP parameters-defaultTCPParameters :: TCPParameters-defaultTCPParameters = TCPParameters {- tcpBacklog = N.sOMAXCONN- , tcpReuseServerAddr = True- , tcpReuseClientAddr = True- }------------------------------------------------------------------------------------- API functions --------------------------------------------------------------------------------------- | Close the transport-apiCloseTransport :: TCPTransport -> Maybe ThreadId -> [Event] -> IO ()-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) - -- 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 - -> IO (Either (TransportError NewEndPointErrorCode) EndPoint)-apiNewEndPoint transport = - try . asyncWhenCancelled closeEndPoint $ do- ourEndPoint <- createLocalEndPoint transport- return EndPoint - { receive = readChan (localChannel ourEndPoint)- , address = localAddress ourEndPoint- , connect = apiConnect (transportParams transport) ourEndPoint - , closeEndPoint = let evs = [ EndPointClosed- , throw $ userError "Endpoint closed" - ] in- apiCloseEndPoint transport evs ourEndPoint- , newMulticastGroup = return . Left $ newMulticastGroupError - , resolveMulticastGroup = return . Left . const resolveMulticastGroupError- }- where- newMulticastGroupError = - TransportError NewMulticastGroupUnsupported "Multicast not supported" - resolveMulticastGroupError = - TransportError ResolveMulticastGroupUnsupported "Multicast not supported" ---- | Connnect to an endpoint-apiConnect :: TCPParameters -- ^ Parameters- -> LocalEndPoint -- ^ Local end point- -> EndPointAddress -- ^ Remote address- -> Reliability -- ^ Reliability (ignored)- -> ConnectHints -- ^ Hints- -> IO (Either (TransportError ConnectErrorCode) Connection)-apiConnect params ourEndPoint theirAddress _reliability hints =- try . asyncWhenCancelled close $ - if localAddress ourEndPoint == theirAddress - then connectToSelf ourEndPoint - else do- resetIfBroken ourEndPoint theirAddress- (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 - }---- | Close a connection-apiClose :: EndPointPair -> LightweightConnectionId -> IORef Bool -> IO ()-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 - then do- writeIORef connAlive False- act <- schedule theirEndPoint $ - sendOn vst [encodeInt32 CloseConnection, encodeInt32 connId] - return ( RemoteEndPointValid - . (remoteOutgoing ^: (\x -> x - 1)) - $ vst- , Just act- )- else- return (RemoteEndPointValid vst, Nothing)- _ ->- return (st, Nothing)- forM_ mAct $ runScheduledAction (ourEndPoint, theirEndPoint)- closeIfUnused (ourEndPoint, theirEndPoint)----- | Send data across a connection-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 = - -- We don't need the overhead of asyncWhenCancelled here- try . mapIOException sendFailed $ do- act <- withMVar (remoteState theirEndPoint) $ \st -> case st of- RemoteEndPointInvalid _ ->- relyViolation (ourEndPoint, theirEndPoint) "apiSend"- RemoteEndPointInit _ _ _ ->- relyViolation (ourEndPoint, theirEndPoint) "apiSend"- RemoteEndPointValid vst -> do- alive <- readIORef connAlive- if alive - then schedule theirEndPoint $ - sendOn vst (encodeInt32 connId : prependLength payload)- else throwIO $ TransportError SendClosed "Connection closed"- RemoteEndPointClosing _ _ -> do - alive <- readIORef connAlive- 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" - else throwIO $ TransportError SendClosed "Connection closed"- RemoteEndPointFailed err -> do- alive <- readIORef connAlive- 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 - -> 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 - mOurState <- modifyMVar (localState ourEndPoint) $ \st ->- case st of- LocalEndPointValid vst -> - return (LocalEndPointClosed, Just vst)- LocalEndPointClosed ->- return (LocalEndPointClosed, Nothing)- forM_ mOurState $ \vst -> do- forM_ (vst ^. localConnections) tryCloseRemoteSocket- forM_ evs $ writeChan (localChannel ourEndPoint) - where- -- Close the remote socket and return the set of all incoming connections- tryCloseRemoteSocket :: RemoteEndPoint -> IO () - tryCloseRemoteSocket theirEndPoint = do- -- 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 _ -> - return (st, Nothing)- RemoteEndPointInit resolved _ _ -> do- putMVar resolved ()- return (closed, Nothing)- 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 - putMVar resolved ()- act <- schedule theirEndPoint $ tryCloseSocket (remoteSocket vst)- return (closed, Just act)- RemoteEndPointClosed ->- return (st, Nothing)- RemoteEndPointFailed err ->- return (RemoteEndPointFailed err, Nothing)- forM_ mAct $ runScheduledAction (ourEndPoint, theirEndPoint)- ------------------------------------------------------------------------------------- Incoming requests --------------------------------------------------------------------------------------- | Handle a connection request (that is, a remote endpoint that is trying to--- establish a TCP connection with us)------ 'handleConnectionRequest' runs in the context of the transport thread, which--- can be killed asynchronously by 'closeTransport'. We fork a separate thread--- as soon as we have located the lcoal endpoint that the remote endpoint is--- interested in. We cannot fork any sooner because then we have no way of--- storing the thread ID and hence no way of killing the thread when we take--- 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 - ourEndPointId <- recvInt32 sock- theirAddress <- EndPointAddress . BS.concat <$> recvWithLength sock - let ourAddress = encodeEndPointAddress (transportHost transport) - (transportPort transport)- ourEndPointId- ourEndPoint <- withMVar (transportState transport) $ \st -> case st of- TransportValid vst ->- case vst ^. localEndPointAt ourAddress of- Nothing -> do- sendMany sock [encodeInt32 ConnectionRequestInvalid]- throwIO $ userError "handleConnectionRequest: Invalid endpoint"- Just ourEndPoint ->- return ourEndPoint- TransportClosed -> - throwIO $ userError "Transport closed"- void . forkIO $ go ourEndPoint theirAddress- where- go :: LocalEndPoint -> EndPointAddress -> IO ()- go ourEndPoint theirAddress = do - -- This runs in a thread that will never be killed- mEndPoint <- handle ((>> return Nothing) . handleException) $ do - resetIfBroken ourEndPoint theirAddress- (theirEndPoint, isNew) <- - findRemoteEndPoint ourEndPoint theirAddress RequestedByThem- - if not isNew - then do- tryIO $ sendMany sock [encodeInt32 ConnectionRequestCrossed]- tryCloseSocket sock- return Nothing - else do- sendLock <- newMVar ()- let vst = ValidRemoteEndPointState - { remoteSocket = sock- , remoteSendLock = sendLock- , _remoteOutgoing = 0- , _remoteIncoming = Set.empty- , _remoteMaxIncoming = 0- , _remoteNextConnOutId = firstNonReservedLightweightConnectionId - }- sendMany sock [encodeInt32 ConnectionRequestAccepted]- resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst)- return (Just theirEndPoint)- -- If we left the scope of the exception handler with a return value of- -- 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. - forM_ mEndPoint $ handleIncomingMessages . (,) ourEndPoint-- handleException :: SomeException -> IO ()- handleException ex = do - tryCloseSocket sock - rethrowIfAsync (fromException ex)-- 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 (ourEndPoint, theirEndPoint) = do- mSock <- withMVar theirState $ \st ->- case st of- RemoteEndPointInvalid _ ->- relyViolation (ourEndPoint, theirEndPoint) - "handleIncomingMessages (invalid)" - RemoteEndPointInit _ _ _ ->- relyViolation (ourEndPoint, theirEndPoint) - "handleIncomingMessages (init)"- RemoteEndPointValid ep ->- return . Just $ remoteSocket ep- RemoteEndPointClosing _ ep ->- return . Just $ remoteSocket ep- RemoteEndPointClosed ->- return Nothing- RemoteEndPointFailed _ -> - return Nothing- - forM_ mSock $ \sock -> - tryIO (go sock) >>= either (prematureExit sock) return- where- -- 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- -- endpoint is put into a Closed (or Closing) state by a concurrent thread- -- (because a 'send' failed) -- the individual handlers below will throw a- -- user exception which is then caught and handled the same way as an- -- exception thrown by 'recv'.- go :: N.Socket -> IO ()- go sock = do- lcid <- recvInt32 sock :: IO LightweightConnectionId- if lcid >= firstNonReservedLightweightConnectionId - then do- readMessage sock lcid - go sock- else - case tryToEnum (fromIntegral lcid) of- Just CreatedNewConnection -> do- recvInt32 sock >>= createdNewConnection - go sock - Just CloseConnection -> do- recvInt32 sock >>= closeConnection - go 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 lcid = do- modifyMVar_ theirState $ \st -> do- vst <- case st of- RemoteEndPointInvalid _ ->- relyViolation (ourEndPoint, theirEndPoint) - "handleIncomingMessages:createNewConnection (invalid)"- RemoteEndPointInit _ _ _ ->- relyViolation (ourEndPoint, theirEndPoint) - "handleIncomingMessages:createNewConnection (init)"- RemoteEndPointValid vst ->- return ( (remoteIncoming ^: Set.insert lcid)- $ (remoteMaxIncoming ^= lcid) - vst- )- RemoteEndPointClosing resolved vst -> do- -- If the endpoint is in closing state that means we send a- -- CloseSocket request to the remote endpoint. If the remote- -- endpoint replies that it created a new connection, it either- -- ignored our request or it sent the request before it got ours.- -- Either way, at this point we simply restore the endpoint to- -- RemoteEndPointValid- putMVar resolved ()- return ( (remoteIncoming ^= Set.singleton lcid)- . (remoteMaxIncoming ^= lcid)- $ vst- )- RemoteEndPointFailed err -> - throwIO err- RemoteEndPointClosed ->- relyViolation (ourEndPoint, theirEndPoint) - "createNewConnection (closed)"- return (RemoteEndPointValid vst)- writeChan ourChannel (ConnectionOpened (connId lcid) ReliableOrdered theirAddr) -- -- 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 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)) $ - throwIO $ userError "Invalid CloseConnection"- 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" - RemoteEndPointFailed err ->- throwIO err- RemoteEndPointClosed ->- relyViolation (ourEndPoint, theirEndPoint) "closeConnection (closed)"- writeChan ourChannel (ConnectionClosed (connId lcid))-- -- Close the socket (if we don't have any outgoing connections)- closeSocket :: N.Socket -> LightweightConnectionId -> IO Bool - closeSocket sock lastReceivedId = do- mAct <- modifyMVar theirState $ \st -> do- case st of- RemoteEndPointInvalid _ ->- relyViolation (ourEndPoint, theirEndPoint)- "handleIncomingMessages:closeSocket (invalid)"- RemoteEndPointInit _ _ _ ->- 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) $ - writeChan ourChannel . ConnectionClosed . connId- 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 - then- return (RemoteEndPointValid vst', Nothing)- 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 - 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 - -- 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- -- that case.- if lastReceivedId < lastSentId vst- then do- return (RemoteEndPointClosing resolved vst, Nothing)- else do- removeRemoteEndPoint (ourEndPoint, theirEndPoint)- act <- schedule theirEndPoint $ tryCloseSocket sock - putMVar resolved ()- return (RemoteEndPointClosed, Just act)- RemoteEndPointFailed err ->- throwIO err- RemoteEndPointClosed ->- 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) -- -- Arguments- ourChannel = localChannel ourEndPoint - theirState = remoteState theirEndPoint- theirAddr = remoteAddress theirEndPoint-- -- Deal with a premature exit- prematureExit :: N.Socket -> IOException -> IO ()- prematureExit sock err = do- tryCloseSocket sock- modifyMVar_ theirState $ \st ->- case st of- RemoteEndPointInvalid _ ->- relyViolation (ourEndPoint, theirEndPoint) - "handleIncomingMessages:prematureExit"- RemoteEndPointInit _ _ _ ->- relyViolation (ourEndPoint, theirEndPoint) - "handleIncomingMessages:prematureExit"- RemoteEndPointValid _ -> do- 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) - "handleIncomingMessages:prematureExit"- RemoteEndPointFailed err' ->- return (RemoteEndPointFailed err')-- -- Construct a connection ID- connId :: LightweightConnectionId -> ConnectionId- connId = createConnectionId (remoteId theirEndPoint)-- -- The ID of the last connection _we_ created (or 0 for none)- lastSentId :: ValidRemoteEndPointState -> LightweightConnectionId- lastSentId vst = - if vst ^. remoteNextConnOutId == firstNonReservedLightweightConnectionId- then 0- else (vst ^. remoteNextConnOutId) - 1------------------------------------------------------------------------------------- Uninterruptable auxiliary functions ----- ----- All these functions assume they are running in a thread which will never ----- be killed.------------------------------------------------------------------------------------- | Create a connection to a remote endpoint------ If the remote endpoint is in 'RemoteEndPointClosing' state then we will--- block until that is resolved. ------ May throw a TransportError ConnectErrorCode exception.-createConnectionTo :: TCPParameters - -> LocalEndPoint - -> EndPointAddress - -> ConnectHints- -> IO (RemoteEndPoint, LightweightConnectionId)-createConnectionTo params ourEndPoint theirAddress hints = go- where- go = do- (theirEndPoint, isNew) <- mapIOException connectFailed $- findRemoteEndPoint ourEndPoint theirAddress RequestedByUs-- if isNew - then do- 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 - act <- schedule theirEndPoint $ do- sendOn vst [encodeInt32 CreatedNewConnection, encodeInt32 connId]- return connId- return ( RemoteEndPointValid - $ remoteNextConnOutId ^= connId + 1 - $ vst- , act - )- -- Error cases- RemoteEndPointInvalid err -> - throwIO err- RemoteEndPointFailed err ->- throwIO err- -- Algorithmic errors- _ ->- relyViolation (ourEndPoint, theirEndPoint) "createConnectionTo"- -- 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 = - return ()---- | Set up a remote endpoint-setupRemoteEndPoint :: TCPParameters -> EndPointPair -> ConnectHints -> IO () -setupRemoteEndPoint params (ourEndPoint, theirEndPoint) hints = do- result <- socketToEndPoint ourAddress - theirAddress - (tcpReuseClientAddr params)- (connectTimeout hints)- didAccept <- case result of- Right (sock, ConnectionRequestAccepted) -> do - sendLock <- newMVar () - let vst = ValidRemoteEndPointState - { remoteSocket = sock- , remoteSendLock = sendLock- , _remoteOutgoing = 0 - , _remoteIncoming = Set.empty- , _remoteMaxIncoming = 0- , _remoteNextConnOutId = firstNonReservedLightweightConnectionId - }- resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst)- return True- Right (sock, ConnectionRequestInvalid) -> do- let err = invalidAddress "setupRemoteEndPoint: Invalid endpoint"- resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointInvalid err)- tryCloseSocket sock- return False- Right (sock, ConnectionRequestCrossed) -> do- withMVar (remoteState theirEndPoint) $ \st -> case st of- RemoteEndPointInit _ crossed _ -> - putMVar crossed ()- RemoteEndPointFailed ex ->- throwIO ex- _ ->- relyViolation (ourEndPoint, theirEndPoint) "setupRemoteEndPoint: Crossed"- tryCloseSocket sock- return False- Left err -> do - resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointInvalid err)- return False-- when didAccept $ handleIncomingMessages (ourEndPoint, theirEndPoint) - where- ourAddress = localAddress ourEndPoint- theirAddress = remoteAddress theirEndPoint- invalidAddress = TransportError ConnectNotFound---- | Send a CloseSocket request if the remote endpoint is unused-closeIfUnused :: EndPointPair -> IO ()-closeIfUnused (ourEndPoint, theirEndPoint) = do - mAct <- modifyMVar (remoteState theirEndPoint) $ \st -> case st of- RemoteEndPointValid vst ->- if vst ^. remoteOutgoing == 0 && Set.null (vst ^. remoteIncoming) - then do - resolved <- newEmptyMVar- act <- schedule theirEndPoint $ - sendOn vst [ encodeInt32 CloseSocket- , encodeInt32 (vst ^. remoteMaxIncoming)- ]- return (RemoteEndPointClosing resolved vst, Just act)- else - return (RemoteEndPointValid vst, Nothing)- _ ->- return (st, Nothing)- 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------ - a user calls the API function 'connect', or and the remote endpoint is--- - an inbound connection request comes in from this remote address------ we remove the remote endpoint first.------ Throws a TransportError ConnectFailed exception if the local endpoint is--- closed.-resetIfBroken :: LocalEndPoint -> EndPointAddress -> IO ()-resetIfBroken ourEndPoint theirAddress = do- mTheirEndPoint <- withMVar (localState ourEndPoint) $ \st -> case st of- LocalEndPointValid vst ->- return (vst ^. localConnectionTo theirAddress)- LocalEndPointClosed ->- throwIO $ TransportError ConnectFailed "Endpoint closed"- forM_ mTheirEndPoint $ \theirEndPoint -> - withMVar (remoteState theirEndPoint) $ \st -> case st of- RemoteEndPointInvalid _ ->- removeRemoteEndPoint (ourEndPoint, theirEndPoint)- RemoteEndPointFailed _ ->- removeRemoteEndPoint (ourEndPoint, theirEndPoint)- _ ->- return ()---- | Special case of 'apiConnect': connect an endpoint to itself--- --- May throw a TransportError ConnectErrorCode (if the local endpoint is closed)-connectToSelf :: LocalEndPoint - -> IO Connection-connectToSelf ourEndPoint = do - connAlive <- newIORef True -- Protected by the local endpoint lock- lconnId <- mapIOException connectFailed $ getLocalNextConnOutId ourEndPoint - let connId = createConnectionId heavyweightSelfConnectionId lconnId- writeChan ourChan $- ConnectionOpened connId ReliableOrdered (localAddress ourEndPoint)- return Connection - { send = selfSend connAlive connId - , close = selfClose connAlive connId- }- where- selfSend :: IORef Bool - -> ConnectionId - -> [ByteString] - -> IO (Either (TransportError SendErrorCode) ())- selfSend connAlive connId msg = - try . withMVar ourState $ \st -> case st of- LocalEndPointValid _ -> do- alive <- readIORef connAlive- if alive- then writeChan ourChan (Received connId msg)- else throwIO $ TransportError SendClosed "Connection closed"- LocalEndPointClosed ->- throwIO $ TransportError SendFailed "Endpoint closed"-- selfClose :: IORef Bool -> ConnectionId -> IO ()- selfClose connAlive connId = - withMVar ourState $ \st -> case st of- LocalEndPointValid _ -> do- alive <- readIORef connAlive- when alive $ do- writeChan ourChan (ConnectionClosed connId) - writeIORef connAlive False- LocalEndPointClosed ->- return () -- ourChan = localChannel ourEndPoint- ourState = localState ourEndPoint- connectFailed = TransportError ConnectFailed . show ---- | Resolve an endpoint currently in 'Init' state-resolveInit :: EndPointPair -> RemoteState -> IO ()-resolveInit (ourEndPoint, theirEndPoint) newState =- modifyMVar_ (remoteState theirEndPoint) $ \st -> case st of- RemoteEndPointInit resolved _ _ -> do- putMVar resolved ()- case newState of - RemoteEndPointClosed -> - removeRemoteEndPoint (ourEndPoint, theirEndPoint)- _ ->- return ()- return newState- 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 = - modifyMVar (localState ourEndpoint) $ \st -> case st of- LocalEndPointValid vst -> do- 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 - chan <- newChan- state <- newMVar . LocalEndPointValid $ ValidLocalEndPointState - { _localNextConnOutId = firstNonReservedLightweightConnectionId- , _localConnections = Map.empty- , _nextConnInId = firstNonReservedHeavyweightConnectionId - }- modifyMVar (transportState transport) $ \st -> case st of- TransportValid vst -> do- let ix = vst ^. nextEndPointId- let addr = encodeEndPointAddress (transportHost transport) - (transportPort transport)- ix - let localEndPoint = LocalEndPoint { localAddress = addr- , localChannel = chan- , localState = state- }- return ( TransportValid - . (localEndPointAt addr ^= Just localEndPoint) - . (nextEndPointId ^= ix + 1) - $ vst- , localEndPoint- )- TransportClosed ->- throwIO (TransportError NewEndPointFailed "Transport closed")----- | Remove reference to a remote endpoint from a local endpoint------ If the local endpoint is closed, do nothing-removeRemoteEndPoint :: EndPointPair -> IO ()-removeRemoteEndPoint (ourEndPoint, theirEndPoint) =- modifyMVar_ ourState $ \st -> case st of- LocalEndPointValid vst ->- case vst ^. localConnectionTo theirAddress of- Nothing -> - return st- Just remoteEndPoint' ->- if remoteId remoteEndPoint' == remoteId theirEndPoint - then return - ( LocalEndPointValid - . (localConnectionTo (remoteAddress theirEndPoint) ^= Nothing) - $ vst- )- else return st- LocalEndPointClosed ->- return LocalEndPointClosed- where- ourState = localState ourEndPoint- theirAddress = remoteAddress theirEndPoint---- | Remove reference to a local endpoint from the transport state------ Does nothing if the transport is closed-removeLocalEndPoint :: TCPTransport -> LocalEndPoint -> IO ()-removeLocalEndPoint transport ourEndPoint = - modifyMVar_ (transportState transport) $ \st -> case st of - TransportValid vst ->- 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 - :: LocalEndPoint- -> EndPointAddress- -> RequestedBy - -> IO (RemoteEndPoint, Bool) -findRemoteEndPoint ourEndPoint theirAddress findOrigin = go- where- go = do- (theirEndPoint, isNew) <- modifyMVar ourState $ \st -> case st of- LocalEndPointValid vst -> case vst ^. localConnectionTo theirAddress of- Just theirEndPoint ->- return (st, (theirEndPoint, False))- Nothing -> do- resolved <- newEmptyMVar- crossed <- newEmptyMVar- theirState <- newMVar (RemoteEndPointInit resolved crossed findOrigin)- scheduled <- newChan- let theirEndPoint = RemoteEndPoint- { remoteAddress = theirAddress- , remoteState = theirState- , remoteId = vst ^. nextConnInId- , remoteScheduled = scheduled - }- return ( LocalEndPointValid - . (localConnectionTo theirAddress ^= Just theirEndPoint) - . (nextConnInId ^: (+ 1)) - $ vst- , (theirEndPoint, True) - )- LocalEndPointClosed ->- throwIO $ userError "Local endpoint closed"- - if isNew - then- return (theirEndPoint, True)- else do- let theirState = remoteState theirEndPoint- snapshot <- modifyMVar theirState $ \st -> case st of- RemoteEndPointValid vst -> - case findOrigin of- RequestedByUs -> do- let st' = RemoteEndPointValid - . (remoteOutgoing ^: (+ 1)) - $ vst - return (st', st')- RequestedByThem ->- 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') - case snapshot of- RemoteEndPointInvalid err ->- throwIO err- RemoteEndPointInit resolved crossed initOrigin ->- case (findOrigin, initOrigin) of- (RequestedByUs, RequestedByUs) -> - readMVar resolved >> go - (RequestedByUs, RequestedByThem) -> - readMVar resolved >> go- (RequestedByThem, RequestedByUs) -> - if ourAddress > theirAddress - then do- -- Wait for the Crossed message- readMVar crossed - return (theirEndPoint, True)- else- return (theirEndPoint, False)- (RequestedByThem, RequestedByThem) -> - throwIO $ userError "Already connected"- RemoteEndPointValid _ ->- -- We assume that the request crossed if we find the endpoint in- -- Valid state. It is possible that this is really an invalid- -- request, but only in the case of a broken client (we don't- -- maintain enough history to be able to tell the difference).- return (theirEndPoint, False)- RemoteEndPointClosing resolved _ ->- readMVar resolved >> go- RemoteEndPointClosed ->- go- RemoteEndPointFailed err -> - throwIO err- - 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) $ \() -> - sendMany (remoteSocket vst) bs------------------------------------------------------------------------------------- Scheduling actions --------------------------------------------------------------------------------------- | See 'schedule'/'runScheduledAction'-type Action a = MVar (Either SomeException a)---- | Schedule an action to be executed (see also 'runScheduledAction')-schedule :: RemoteEndPoint -> IO a -> IO (Action a)-schedule theirEndPoint act = do- mvar <- newEmptyMVar- writeChan (remoteScheduled theirEndPoint) $ - catch (act >>= putMVar mvar . Right) (putMVar mvar . Left)- return mvar---- | Run a scheduled action. Every call to 'schedule' should be paired with a--- call to 'runScheduledAction' so that every scheduled action is run. Note--- however that the there is no guarantee that in------ > do act <- schedule p--- > 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 --- 'runScheduledAction' or by another).-runScheduledAction :: EndPointPair -> Action a -> IO a-runScheduledAction (ourEndPoint, theirEndPoint) mvar = do- join $ readChan (remoteScheduled theirEndPoint)- ma <- readMVar mvar- case ma of- Right a -> return a- Left e -> do- forM_ (fromException e) $ \ioe -> - modifyMVar_ (remoteState theirEndPoint) $ \st ->- case st of - RemoteEndPointValid vst -> handleIOException ioe vst- _ -> return (RemoteEndPointFailed ioe)- throwIO e- where- handleIOException :: IOException - -> ValidRemoteEndPointState - -> IO RemoteState - handleIOException ex vst = do- tryCloseSocket (remoteSocket vst)- let code = EventConnectionLost (remoteAddress theirEndPoint) - err = TransportError code (show ex)- writeChan (localChannel ourEndPoint) $ ErrorEvent err - return (RemoteEndPointFailed ex)------------------------------------------------------------------------------------- "Stateless" (MVar free) functions --------------------------------------------------------------------------------------- | Establish a connection to a remote endpoint------ Maybe throw a TransportError-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 - Nothing -> throwIO (failed . userError $ "Could not parse")- Just dec -> return dec- addr:_ <- mapIOException invalidAddress $ - N.getAddrInfo Nothing (Just host) (Just port)- bracketOnError (createSocket addr) tryCloseSocket $ \sock -> do- when reuseAddr $ - mapIOException failed $ N.setSocketOption sock N.ReuseAddr 1- mapIOException invalidAddress $ - timeoutMaybe timeout timeoutError $ - N.connect sock (N.addrAddress addr) - response <- mapIOException failed $ do- sendMany sock (encodeInt32 theirEndPointId : prependLength [ourAddress])- recvInt32 sock- case tryToEnum response of- Nothing -> throwIO (failed . userError $ "Unexpected response")- Just r -> return (sock, r)- where- createSocket :: N.AddrInfo -> IO N.Socket- createSocket addr = mapIOException insufficientResources $ - N.socket (N.addrFamily addr) N.Stream N.defaultProtocol-- 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 - -> EndPointAddress-encodeEndPointAddress host port ix = EndPointAddress . BSC.pack $- host ++ ":" ++ port ++ ":" ++ show ix ---- | Decode end point address-decodeEndPointAddress :: EndPointAddress - -> Maybe (N.HostName, N.ServiceName, EndPointId)-decodeEndPointAddress (EndPointAddress bs) = - case splitMaxFromEnd (== ':') 2 $ BSC.unpack bs of- [host, port, endPointIdStr] -> - case reads endPointIdStr of - [(endPointId, "")] -> Just (host, port, endPointId)- _ -> Nothing- _ ->- Nothing---- | Construct a ConnectionId-createConnectionId :: HeavyweightConnectionId - -> LightweightConnectionId - -> ConnectionId-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 - where- -- go :: [[a]] -> Int -> [a] -> [[a]]- 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- else go ((x : acc) : accs) n xs- go _ _ _ = error "Bug in splitMaxFromEnd"------------------------------------------------------------------------------------- Functions from TransportInternals --------------------------------------------------------------------------------------- Find a socket between two endpoints--- --- Throws an IO exception if the socket could not be found.-internalSocketBetween :: TCPTransport -- ^ Transport - -> EndPointAddress -- ^ Local endpoint- -> EndPointAddress -- ^ Remote endpoint- -> IO N.Socket -internalSocketBetween transport ourAddress theirAddress = do- ourEndPoint <- withMVar (transportState transport) $ \st -> case st of- 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 -> - 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 -> - return $ remoteSocket vst- RemoteEndPointClosing _ vst ->- return $ remoteSocket vst - RemoteEndPointClosed ->- throwIO $ userError "Remote endpoint closed"- RemoteEndPointInvalid err ->- throwIO err - RemoteEndPointFailed err ->- throwIO err ------------------------------------------------------------------------------------- Constants --------------------------------------------------------------------------------------- | We reserve a bunch of connection IDs for control messages-firstNonReservedLightweightConnectionId :: LightweightConnectionId-firstNonReservedLightweightConnectionId = 1024---- | Self-connection-heavyweightSelfConnectionId :: HeavyweightConnectionId-heavyweightSelfConnectionId = 0---- | We reserve some connection IDs for special heavyweight connections-firstNonReservedHeavyweightConnectionId :: HeavyweightConnectionId-firstNonReservedHeavyweightConnectionId = 1------------------------------------------------------------------------------------- Accessor definitions -------------------------------------------------------------------------------------localEndPoints :: Accessor ValidTransportState (Map EndPointAddress LocalEndPoint)-localEndPoints = accessor _localEndPoints (\es st -> st { _localEndPoints = es })--nextEndPointId :: Accessor ValidTransportState EndPointId-nextEndPointId = accessor _nextEndPointId (\eid st -> st { _nextEndPointId = eid })--localNextConnOutId :: Accessor ValidLocalEndPointState LightweightConnectionId-localNextConnOutId = accessor _localNextConnOutId (\cix st -> st { _localNextConnOutId = cix })--localConnections :: Accessor ValidLocalEndPointState (Map EndPointAddress RemoteEndPoint)-localConnections = accessor _localConnections (\es st -> st { _localConnections = es })--nextConnInId :: Accessor ValidLocalEndPointState HeavyweightConnectionId -nextConnInId = accessor _nextConnInId (\rid st -> st { _nextConnInId = rid })--remoteOutgoing :: Accessor ValidRemoteEndPointState Int-remoteOutgoing = accessor _remoteOutgoing (\cs conn -> conn { _remoteOutgoing = cs })--remoteIncoming :: Accessor ValidRemoteEndPointState (Set LightweightConnectionId)-remoteIncoming = accessor _remoteIncoming (\cs conn -> conn { _remoteIncoming = cs })--remoteMaxIncoming :: Accessor ValidRemoteEndPointState LightweightConnectionId-remoteMaxIncoming = accessor _remoteMaxIncoming (\lcid st -> st { _remoteMaxIncoming = lcid })--remoteNextConnOutId :: Accessor ValidRemoteEndPointState LightweightConnectionId-remoteNextConnOutId = accessor _remoteNextConnOutId (\cix st -> st { _remoteNextConnOutId = cix })--localEndPointAt :: EndPointAddress -> Accessor ValidTransportState (Maybe LocalEndPoint)-localEndPointAt addr = localEndPoints >>> DAC.mapMaybe addr --localConnectionTo :: EndPointAddress -> Accessor ValidLocalEndPointState (Maybe RemoteEndPoint)-localConnectionTo addr = localConnections >>> DAC.mapMaybe addr ------------------------------------------------------------------------------------ Debugging ------------------------------------------------------------------------------------relyViolation :: EndPointPair -> String -> IO a-relyViolation (ourEndPoint, theirEndPoint) str = do- 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) - ++ "(" ++ show (remoteId theirEndPoint) ++ ")"- ++ "/" ++ show tid +-- | 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.+--+-- Applications that use the TCP transport should use+-- 'Network.Socket.withSocketsDo' in their main function for Windows+-- compatibility (see "Network.Socket").+{-# LANGUAGE CPP #-}+{-# LANGUAGE RecursiveDo #-}+{-# LANGUAGE TupleSections #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE LambdaCase #-}++module Network.Transport.TCP+ ( -- * Main API+ createTransport+ , TCPAddr(..)+ , defaultTCPAddr+ , TCPAddrInfo(..)+ , TCPParameters(..)+ , defaultTCPParameters+ -- * Internals (exposed for unit tests)+ , createTransportExposeInternals+ , TransportInternals(..)+ , EndPointId+ , ControlHeader(..)+ , ConnectionRequestResponse(..)+ , firstNonReservedLightweightConnectionId+ , firstNonReservedHeavyweightConnectionId+ , socketToEndPoint+ , LightweightConnectionId+ , QDisc(..)+ , simpleUnboundedQDisc+ , simpleOnePlaceQDisc+ -- * Design notes+ -- $design+ ) where++import Prelude hiding+ ( mapM_+ )++import Network.Transport+import Network.Transport.TCP.Internal+ ( ControlHeader(..)+ , encodeControlHeader+ , decodeControlHeader+ , ConnectionRequestResponse(..)+ , encodeConnectionRequestResponse+ , decodeConnectionRequestResponse+ , forkServer+ , recvWithLength+ , recvExact+ , recvWord32+ , encodeWord32+ , tryCloseSocket+ , tryShutdownSocketBoth+ , resolveSockAddr+ , EndPointId+ , encodeEndPointAddress+ , decodeEndPointAddress+ , currentProtocolVersion+ , randomEndPointAddress+ )+import Network.Transport.Internal+ ( prependLength+ , mapIOException+ , tryIO+ , void+ , timeoutMaybe+ , asyncWhenCancelled+ )++#ifdef USE_MOCK_NETWORK+import qualified Network.Transport.TCP.Mock.Socket as N+#else+import qualified Network.Socket as N+#endif+ ( HostName+ , ServiceName+ , Socket+ , getAddrInfo+ , maxListenQueue+ , socket+ , addrFamily+ , addrAddress+ , SocketType(Stream)+ , defaultProtocol+ , setSocketOption+ , SocketOption(ReuseAddr, NoDelay, UserTimeout, KeepAlive)+ , isSupportedSocketOption+ , connect+ , AddrInfo+ , SockAddr(..)+ )++#ifdef USE_MOCK_NETWORK+import Network.Transport.TCP.Mock.Socket.ByteString (sendMany)+#else+import Network.Socket.ByteString (sendMany)+#endif++import Control.Concurrent+ ( forkIO+ , ThreadId+ , killThread+ , myThreadId+ , threadDelay+ , throwTo+ )+import Control.Concurrent.Chan (Chan, newChan, readChan, writeChan)+import Control.Concurrent.MVar+ ( MVar+ , newMVar+ , modifyMVar+ , modifyMVar_+ , readMVar+ , tryReadMVar+ , takeMVar+ , putMVar+ , tryPutMVar+ , newEmptyMVar+ , withMVar+ )+import Control.Concurrent.Async (async, wait)+import Control.Category ((>>>))+import Control.Monad (when, unless, join, mplus, (<=<))+import Control.Exception+ ( IOException+ , SomeException+ , AsyncException+ , handle+ , throw+ , throwIO+ , try+ , bracketOnError+ , bracket+ , fromException+ , finally+ , catch+ , bracket+ , mask+ , mask_+ )+import Data.IORef (IORef, newIORef, writeIORef, readIORef, writeIORef)+import Data.ByteString (ByteString)+import qualified Data.ByteString as BS (concat, null)+import qualified Data.ByteString.Char8 as BSC (pack, unpack)+import Data.Bits (shiftL, (.|.))+import Data.Maybe (isJust, isNothing)+import Data.Word (Word32)+import Data.Set (Set)+import qualified Data.Set as Set+ ( empty+ , insert+ , elems+ , singleton+ , null+ , delete+ , member+ )+import Data.Map (Map)+import qualified Data.Map as Map (empty)+import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))+import qualified Data.Accessor.Container as DAC (mapMaybe)+import Data.Foldable (forM_, mapM_)+import qualified System.Timeout (timeout)++-- $design+--+-- [Goals]+--+-- The TCP transport maps multiple logical connections between /A/ and /B/ (in+-- either direction) to a single TCP connection:+--+-- > +-------+ +-------++-- > | A |==========================| B |+-- > | |>~~~~~~~~~~~~~~~~~~~~~~~~~|~~~\ |+-- > | Q |>~~~~~~~~~~~~~~~~~~~~~~~~~|~~~Q |+-- > | \~~~|~~~~~~~~~~~~~~~~~~~~~~~~~<| |+-- > | |==========================| |+-- > +-------+ +-------++--+-- 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+-- closed.+--+-- [Connecting]+--+-- Let /A/, /B/ be two endpoints without any connections. When /A/ wants to+-- connect to /B/, it locally records that it is trying to connect to /B/ and+-- sends a request to /B/. As part of the request /A/ sends its own endpoint+-- address to /B/ (so that /B/ can reuse the connection in the other direction).+--+-- When /B/ receives the connection request it first checks if it did not+-- already initiate a connection request to /A/. If not it will acknowledge the+-- connection request by sending 'ConnectionRequestAccepted' to /A/ and record+-- that it has a TCP connection to /A/.+--+-- The tricky case arises when /A/ sends a connection request to /B/ and /B/+-- finds that it had already sent a connection request to /A/. In this case /B/+-- will accept the connection request from /A/ if /A/s endpoint address is+-- smaller (lexicographically) than /B/s, and reject it otherwise. If it rejects+-- it, it sends a 'ConnectionRequestCrossed' message to /A/. The+-- lexicographical ordering is an arbitrary but convenient way to break the+-- tie. If a connection exists between /A/ and /B/ when /B/ rejects the request,+-- /B/ will probe the connection to make sure it is healthy. If /A/ does not+-- answer timely to the probe, /B/ will discard the connection.+--+-- When it receives a 'ConnectionRequestCrossed' message the /A/ thread that+-- initiated the request just needs to wait until the /A/ thread that is dealing+-- with /B/'s connection request completes, unless there is a network failure.+-- If there is a network failure, the initiator thread would timeout and return+-- an error.+--+-- [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/:+--+-- > +-------+ +-------++-- > | A |==========================| B |+-- > | |>~~~~~~~~~~~~~~~~~~~~~~~~~|~~~\ |+-- > | | | Q |+-- > | #| | |+-- > | |==========================| |+-- > +-------+ +-------++--+-- 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).+--+-- 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.)+--+-- 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+-- requests to /B/ after having sent a 'CloseSocket' request, and will actually+-- close its end of the socket only when receiving the 'CloseSocket' message+-- 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.++--------------------------------------------------------------------------------+-- Internal datatypes --+--------------------------------------------------------------------------------++-- 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.+-- * For the case of "valid" we use third data structure to give more details+-- about the state (ValidTransportState, ValidLocalEndPointState,+-- ValidRemoteEndPointState).++-- | Information about the network addresses of a transport: the external+-- host/port as well as the bound host/port, which are not necessarily the+-- same.+data TransportAddrInfo = TransportAddrInfo+ { transportHost :: !N.HostName+ , transportPort :: !N.ServiceName+ , transportBindHost :: !N.HostName+ , transportBindPort :: !N.ServiceName+ }++data TCPTransport = TCPTransport+ { transportAddrInfo :: !(Maybe TransportAddrInfo)+ -- ^ This is 'Nothing' in case the transport is not addressable from the+ -- network: peers cannot connect to it unless it has a connection to the+ -- peer.+ , transportState :: !(MVar TransportState)+ , transportParams :: !TCPParameters+ }++data TransportState =+ TransportValid !ValidTransportState+ | TransportClosed++data ValidTransportState = ValidTransportState+ { _localEndPoints :: !(Map EndPointId LocalEndPoint)+ , _nextEndPointId :: !EndPointId+ }++data LocalEndPoint = LocalEndPoint+ { localAddress :: !EndPointAddress+ , localEndPointId :: !EndPointId+ , localState :: !(MVar LocalEndPointState)+ -- | A 'QDisc' is held here rather than on the 'ValidLocalEndPointState'+ -- because even closed 'LocalEndPoint's can have queued input data.+ , localQueue :: !(QDisc Event)+ }++data LocalEndPointState =+ LocalEndPointValid !ValidLocalEndPointState+ | LocalEndPointClosed++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+ , _nextConnInId :: !HeavyweightConnectionId+ -- Currently active outgoing heavyweight connections+ , _localConnections :: !(Map EndPointAddress RemoteEndPoint)+ }++-- REMOTE ENDPOINTS+--+-- Remote endpoints (basically, TCP connections) have the following lifecycle:+--+-- Init ---+---> Invalid+-- |+-- +-------------------------------\+-- | |+-- | /----------\ |+-- | | | |+-- | v | v+-- +---> Valid ---> Closing ---> Closed+-- | | | |+-- | | | v+-- \-------+----------+--------> Failed+--+-- Init: There are two places where we create new remote endpoints: in+-- createConnectionTo (in response to an API 'connect' call) and in+-- handleConnectionRequest (when a remote node tries to connect to us).+-- 'Init' carries an MVar () 'resolved' which concurrent threads can use to+-- wait for the remote endpoint to finish initialization. We record who+-- requested the connection (the local endpoint or the remote endpoint).+--+-- Invalid: We put the remote endpoint in invalid state only during+-- 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+-- concurrent threads can use to wait for the remote endpoint to either be+-- closed fully (if the communication parnet responds with another+-- CloseSocket) or be put back in 'Valid' state if the remote endpoint denies+-- the request.+--+-- We also put the endpoint in Closed state, directly from Init, if we our+-- outbound connection request crossed an inbound connection request and we+-- decide to keep the inbound (i.e., the remote endpoint sent us a+-- ConnectionRequestCrossed message).+--+-- Closed: The endpoint is put in Closed state after a successful garbage+-- collection.+--+-- Failed: If the connection to the remote endpoint is lost, or the local+-- endpoint (or the whole transport) is closed manually, the remote endpoint is+-- put in Failed state, and we record the reason.+--+-- Invariants for dealing with remote endpoints:+--+-- INV-SEND: Whenever we send data the remote endpoint must be locked (to avoid+-- interleaving bits of payload).+--+-- INV-CLOSE: Local endpoints should never point to remote endpoint in closed+-- state. Whenever we put an endpoint in Closed state we remove that+-- endpoint from localConnections first, so that if a concurrent thread reads+-- the MVar, finds RemoteEndPointClosed, and then looks up the endpoint in+-- 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'.+--+-- 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.+--+-- 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+-- acceptable, of course -- as we are waiting for the remote endpoint to+-- confirm or deny the request).+--+-- INV-LOCK-ORDER: Remote endpoint must be locked before their local endpoints.+-- In other words: it is okay to call modifyMVar on a local endpoint inside a+-- modifyMVar on a remote endpoint, but not the other way around. In+-- particular, it is okay to call removeRemoteEndPoint inside+-- modifyRemoteState.++data RemoteEndPoint = RemoteEndPoint+ { remoteAddress :: !EndPointAddress+ , remoteState :: !(MVar RemoteState)+ , remoteId :: !HeavyweightConnectionId+ , remoteScheduled :: !(Chan (IO ()))+ }++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+ -- | "Normal" working endpoint+ | RemoteEndPointValid !ValidRemoteEndPointState+ -- | The remote endpoint is being closed (garbage collected)+ | RemoteEndPointClosing !(MVar ()) !ValidRemoteEndPointState+ -- | The remote endpoint has been closed (garbage collected)+ | RemoteEndPointClosed+ -- | The remote endpoint has failed, or has been forcefully shutdown+ -- using a closeTransport or closeEndPoint API call+ | RemoteEndPointFailed !IOException++-- 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+ { _remoteOutgoing :: !Int+ , _remoteIncoming :: !(Set LightweightConnectionId)+ , _remoteLastIncoming :: !LightweightConnectionId+ , _remoteNextConnOutId :: !LightweightConnectionId+ , remoteSocket :: !N.Socket+ -- | When the connection is being probed, yields an IO action that can be+ -- used to release any resources dedicated to the probing.+ , remoteProbing :: Maybe (IO ())+ -- | MVar protects the socket usage by the concurrent threads and+ -- prohibits its usage after SomeException.+ --+ -- Nothing allows the socket usage. @Just e@ is set on an+ -- exception after which the socket should not be used (see 'sendOn').+ , remoteSendLock :: !(MVar (Maybe SomeException))+ -- | An IO which returns when the socket (remoteSocket) has been closed.+ -- The program/thread which created the socket is always responsible+ -- for closing it, but sometimes other threads need to know when this+ -- happens.+ , remoteSocketClosed :: !(IO ())+ }++-- | Pair of local and a remote endpoint (for conciseness in signatures)+type EndPointPair = (LocalEndPoint, RemoteEndPoint)++-- | Lightweight connection ID (sender allocated)+--+-- A ConnectionId is the concentation of a 'HeavyweightConnectionId' and a+-- 'LightweightConnectionId'.+type LightweightConnectionId = Word32++-- | Heavyweight connection ID (recipient allocated)+--+-- A ConnectionId is the concentation of a 'HeavyweightConnectionId' and a+-- 'LightweightConnectionId'.+type HeavyweightConnectionId = Word32++-- | A transport which is addressable from the network must give a host/port+-- on which to bind/listen, and determine its external address (host/port) from+-- the actual port (which may not be known, in case 0 is used for the bind+-- port).+data TCPAddrInfo = TCPAddrInfo {+ tcpBindHost :: N.HostName+ , tcpBindPort :: N.ServiceName+ , tcpExternalAddress :: N.ServiceName -> (N.HostName, N.ServiceName)+ }++-- | Addressability of a transport. If your transport cannot be connected+-- to, for instance because it runs behind NAT, use Unaddressable.+data TCPAddr = Addressable TCPAddrInfo | Unaddressable++-- | The bind and external host/port are the same.+defaultTCPAddr :: N.HostName -> N.ServiceName -> TCPAddr+defaultTCPAddr host port = Addressable $ TCPAddrInfo {+ tcpBindHost = host+ , tcpBindPort = port+ , tcpExternalAddress = (,) host+ }++-- | Parameters for setting up the TCP transport+data TCPParameters = TCPParameters {+ -- | Backlog for 'listen'.+ -- Defaults to SOMAXCONN.+ tcpBacklog :: Int+ -- | Should we set SO_REUSEADDR on the server socket?+ -- Defaults to True.+ , tcpReuseServerAddr :: Bool+ -- | Should we set SO_REUSEADDR on client sockets?+ -- Defaults to True.+ , tcpReuseClientAddr :: Bool+ -- | Should we set TCP_NODELAY on connection sockets?+ -- Defaults to True.+ , tcpNoDelay :: Bool+ -- | Should we set TCP_KEEPALIVE on connection sockets?+ -- Defaults to False.+ , tcpKeepAlive :: Bool+ -- | Value of TCP_USER_TIMEOUT in milliseconds+ , tcpUserTimeout :: Maybe Int+ -- | A connect timeout for all 'connect' calls of the transport+ -- in microseconds+ --+ -- This can be overriden for each connect call with+ -- 'ConnectHints'.'connectTimeout'.+ --+ -- Connection requests to this transport will also timeout if they don't+ -- send the required data before this many microseconds.+ --+ -- Defaults to Nothing (no timeout).+ , transportConnectTimeout :: Maybe Int+ -- | Create a QDisc for an EndPoint.+ , tcpNewQDisc :: forall t . IO (QDisc t)+ -- | Maximum length (in bytes) for a peer's address.+ -- If a peer attempts to send an address of length exceeding the limit,+ -- the connection will be refused (socket will close).+ , tcpMaxAddressLength :: Word32+ -- | Maximum length (in bytes) to receive from a peer.+ -- If a peer attempts to send data on a lightweight connection exceeding+ -- the limit, the heavyweight connection which carries that lightweight+ -- connection will go down. The peer and the local node will get an+ -- EventConnectionLost.+ , tcpMaxReceiveLength :: Word32+ -- | If True, new connections will be accepted only if the socket's host+ -- matches the host that the peer claims in its EndPointAddress.+ -- This is useful when operating on untrusted networks, because the peer+ -- could otherwise deny service to some victim by claiming the victim's+ -- address.+ -- Defaults to False.+ , tcpCheckPeerHost :: Bool+ -- | What to do if there's an exception when accepting a new TCP+ -- connection. Throwing an exception here will cause the server to+ -- terminate.+ -- Defaults to `throwIO`.+ , tcpServerExceptionHandler :: SomeException -> IO ()+ }++-- | Internal functionality we expose for unit testing+data TransportInternals = TransportInternals+ { -- | The ID of the thread that listens for new incoming connections+ transportThread :: Maybe ThreadId+ -- | A variant of newEndPoint in which the QDisc determined by the+ -- transport's TCPParameters can be optionally overridden.+ , newEndPointInternal :: (forall t . Maybe (QDisc t))+ -> IO (Either (TransportError NewEndPointErrorCode) EndPoint)+ -- | Find the socket between a local and a remote endpoint+ , socketBetween :: EndPointAddress+ -> EndPointAddress+ -> IO N.Socket+ }++--------------------------------------------------------------------------------+-- Top-level functionality --+--------------------------------------------------------------------------------++-- | Create a TCP transport+createTransport+ :: TCPAddr+ -> TCPParameters+ -> IO (Either IOException Transport)+createTransport addr params =+ either Left (Right . fst) <$> createTransportExposeInternals addr params++-- | You should probably not use this function (used for unit testing only)+createTransportExposeInternals+ :: TCPAddr+ -> TCPParameters+ -> IO (Either IOException (Transport, TransportInternals))+createTransportExposeInternals addr params = do+ state <- newMVar . TransportValid $ ValidTransportState+ { _localEndPoints = Map.empty+ , _nextEndPointId = 0+ }+ case addr of++ Unaddressable ->+ let transport = TCPTransport { transportState = state+ , transportAddrInfo = Nothing+ , transportParams = params+ }+ in fmap Right (mkTransport transport Nothing)++ Addressable (TCPAddrInfo bindHost bindPort mkExternal) -> tryIO $ mdo+ when ( isJust (tcpUserTimeout params) &&+ not (N.isSupportedSocketOption N.UserTimeout)+ ) $+ throwIO $ userError $ "Network.Transport.TCP.createTransport: " +++ "the parameter tcpUserTimeout is unsupported " +++ "in this system."+ -- We don't know for sure the actual port 'forkServer' binded until it+ -- completes (see description of 'forkServer'), yet we need the port to+ -- construct a transport. So we tie a recursive knot.+ (port', result) <- do+ let (externalHost, externalPort) = mkExternal port'+ let addrInfo = TransportAddrInfo { transportHost = externalHost+ , transportPort = externalPort+ , transportBindHost = bindHost+ , transportBindPort = port'+ }+ let transport = TCPTransport { transportState = state+ , transportAddrInfo = Just addrInfo+ , transportParams = params+ }+ bracketOnError (forkServer+ bindHost+ bindPort+ (tcpBacklog params)+ (tcpReuseServerAddr params)+ (errorHandler transport)+ (terminationHandler transport)+ (handleConnectionRequest transport))+ (\(_port', tid) -> killThread tid)+ (\(port'', tid) -> (port'',) <$> mkTransport transport (Just tid))+ return result+ where+ mkTransport :: TCPTransport+ -> Maybe ThreadId+ -> IO (Transport, TransportInternals)+ mkTransport transport mtid = do+ return+ ( Transport+ { newEndPoint = do+ qdisc <- tcpNewQDisc params+ apiNewEndPoint transport qdisc+ , closeTransport = let evs = [ EndPointClosed ]+ in apiCloseTransport transport mtid evs+ }+ , TransportInternals+ { transportThread = mtid+ , socketBetween = internalSocketBetween transport+ , newEndPointInternal = \mqdisc -> case mqdisc of+ Just qdisc -> apiNewEndPoint transport qdisc+ Nothing -> do+ qdisc <- tcpNewQDisc params+ apiNewEndPoint transport qdisc+ }+ )++ errorHandler :: TCPTransport -> SomeException -> IO ()+ errorHandler _ = tcpServerExceptionHandler params++ terminationHandler :: TCPTransport -> SomeException -> IO ()+ terminationHandler transport ex = do+ let evs = [ ErrorEvent (TransportError EventTransportFailed (show ex))+ , throw $ userError "Transport closed"+ ]+ apiCloseTransport transport Nothing evs++-- | Default TCP parameters+defaultTCPParameters :: TCPParameters+defaultTCPParameters = TCPParameters {+ tcpBacklog = N.maxListenQueue+ , tcpReuseServerAddr = True+ , tcpReuseClientAddr = True+ , tcpNoDelay = True+ , tcpKeepAlive = False+ , tcpUserTimeout = Nothing+ , tcpNewQDisc = simpleUnboundedQDisc+ , transportConnectTimeout = Nothing+ , tcpMaxAddressLength = maxBound+ , tcpMaxReceiveLength = maxBound+ , tcpCheckPeerHost = False+ , tcpServerExceptionHandler = throwIO+ }++--------------------------------------------------------------------------------+-- API functions --+--------------------------------------------------------------------------------++-- | Close the transport+apiCloseTransport :: TCPTransport -> Maybe ThreadId -> [Event] -> IO ()+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)+ -- 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+ -> QDisc Event+ -> IO (Either (TransportError NewEndPointErrorCode) EndPoint)+apiNewEndPoint transport qdisc =+ try . asyncWhenCancelled closeEndPoint $ do+ ourEndPoint <- createLocalEndPoint transport qdisc+ return EndPoint+ { receive = qdiscDequeue (localQueue ourEndPoint)+ , address = localAddress ourEndPoint+ , connect = apiConnect transport ourEndPoint+ , closeEndPoint = let evs = [ EndPointClosed ]+ in apiCloseEndPoint transport evs ourEndPoint+ , newMulticastGroup = return . Left $ newMulticastGroupError+ , resolveMulticastGroup = return . Left . const resolveMulticastGroupError+ }+ where+ newMulticastGroupError =+ TransportError NewMulticastGroupUnsupported "Multicast not supported"+ resolveMulticastGroupError =+ TransportError ResolveMulticastGroupUnsupported "Multicast not supported"++-- | Abstraction of a queue for an 'EndPoint'.+--+-- A value of type @QDisc t@ is a queue of events of an abstract type @t@.+--+-- This specifies which 'Event's will come from+-- 'receive :: EndPoint -> IO Event' and when. It is highly general so that+-- the simple yet potentially very fast implementation backed by a single+-- unbounded channel can be used, without excluding more nuanced policies+-- like class-based queueing with bounded buffers for each peer, which may be+-- faster in certain conditions but probably has lower maximal throughput.+--+-- A 'QDisc' must satisfy some properties in order for the semantics of+-- network-transport to hold true. In general, an event fed with+-- 'qdiscEnqueue' must not be dropped. i.e. provided that no other event in+-- the QDisc has higher priority, the event should eventually be returned by+-- 'qdiscDequeue'. An exception to this are 'Receive' events of unreliable+-- connections.+--+-- Every call to 'receive' is just 'qdiscDequeue' on that 'EndPoint's+-- 'QDisc'. Whenever an event arises from a socket, `qdiscEnqueue` is called+-- with the relevant metadata in the same thread that reads from the socket.+-- You can be clever about when to block here, so as to control network+-- ingress. This applies also to loopback connections (an 'EndPoint' connects+-- to itself), in which case blocking on the enqueue would only block some+-- thread in your program rather than some chatty network peer. The 'Event'+-- which is to be enqueued is given to 'qdiscEnqueue' so that the 'QDisc'+-- can know about open connections, their identifiers and peer addresses, etc.+data QDisc t = QDisc {+ -- | Dequeue an event.+ qdiscDequeue :: IO t+ -- | @qdiscEnqueue ep ev t@ enqueues and event @t@, originated from the+ -- given remote endpoint @ep@ and with data @ev@.+ --+ -- @ep@ might be the local endpoint if it relates to a self-connection.+ --+ -- @ev@ might be in practice the value given as @t@. It is passed in+ -- the abstract form @t@ to enforce it is dequeued unmodified, but the+ -- 'QDisc' implementation can still observe the concrete form @ev@ to+ -- make prioritization decisions.+ , qdiscEnqueue :: EndPointAddress -> Event -> t -> IO ()+ }++-- | Post an 'Event' using a 'QDisc'.+qdiscEnqueue' :: QDisc Event -> EndPointAddress -> Event -> IO ()+qdiscEnqueue' qdisc addr event = qdiscEnqueue qdisc addr event event++-- | A very simple QDisc backed by an unbounded channel.+simpleUnboundedQDisc :: forall t . IO (QDisc t)+simpleUnboundedQDisc = do+ eventChan <- newChan+ return $ QDisc {+ qdiscDequeue = readChan eventChan+ , qdiscEnqueue = const (const (writeChan eventChan))+ }++-- | A very simple QDisc backed by a 1-place queue (MVar).+-- With this QDisc, all threads reading from sockets will try to put their+-- events into the same MVar. That MVar will be cleared by calls to+-- 'receive'. Thus the rate at which data is read from the wire is directly+-- related to the rate at which data is pulled from the EndPoint by+-- 'receive'.+simpleOnePlaceQDisc :: forall t . IO (QDisc t)+simpleOnePlaceQDisc = do+ mvar <- newEmptyMVar+ return $ QDisc {+ qdiscDequeue = takeMVar mvar+ , qdiscEnqueue = const (const (putMVar mvar))+ }++-- | Connnect to an endpoint+apiConnect :: TCPTransport+ -> LocalEndPoint -- ^ Local end point+ -> EndPointAddress -- ^ Remote address+ -> Reliability -- ^ Reliability (ignored)+ -> ConnectHints -- ^ Hints+ -> IO (Either (TransportError ConnectErrorCode) Connection)+apiConnect transport ourEndPoint theirAddress _reliability hints =+ try . asyncWhenCancelled close $+ if localAddress ourEndPoint == theirAddress+ then connectToSelf ourEndPoint+ else do+ resetIfBroken ourEndPoint theirAddress+ (theirEndPoint, connId) <-+ createConnectionTo transport 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+ }++-- | Close a connection+apiClose :: EndPointPair -> LightweightConnectionId -> IORef Bool -> IO ()+apiClose (ourEndPoint, theirEndPoint) connId connAlive =+ void . tryIO . asyncWhenCancelled return $ finally+ (withScheduledAction ourEndPoint $ \sched -> do+ modifyMVar_ (remoteState theirEndPoint) $ \st -> case st of+ RemoteEndPointValid vst -> do+ alive <- readIORef connAlive+ if alive+ then do+ writeIORef connAlive False+ sched theirEndPoint $+ sendOn vst [+ encodeWord32 (encodeControlHeader CloseConnection)+ , encodeWord32 connId+ ]+ return ( RemoteEndPointValid+ . (remoteOutgoing ^: (\x -> x - 1))+ $ vst+ )+ else+ return (RemoteEndPointValid vst)+ _ ->+ return st)+ (closeIfUnused (ourEndPoint, theirEndPoint))+++-- | Send data across a connection+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 =+ -- We don't need the overhead of asyncWhenCancelled here+ try . mapIOException sendFailed $ withScheduledAction ourEndPoint $ \sched -> do+ withMVar (remoteState theirEndPoint) $ \st -> case st of+ RemoteEndPointInvalid _ ->+ relyViolation (ourEndPoint, theirEndPoint) "apiSend"+ RemoteEndPointInit _ _ _ ->+ relyViolation (ourEndPoint, theirEndPoint) "apiSend"+ RemoteEndPointValid vst -> do+ alive <- readIORef connAlive+ if alive+ then sched theirEndPoint $+ sendOn vst (encodeWord32 connId : prependLength payload)+ else throwIO $ TransportError SendClosed "Connection closed"+ RemoteEndPointClosing _ _ -> do+ alive <- readIORef connAlive+ if alive+ -- RemoteEndPointClosing is only entered by 'closeIfUnused',+ -- which guarantees that there are no alive connections.+ then relyViolation (ourEndPoint, theirEndPoint) "apiSend RemoteEndPointClosing"+ else throwIO $ TransportError SendClosed "Connection closed"+ RemoteEndPointClosed -> do+ alive <- readIORef connAlive+ if alive+ -- This is normal. If the remote endpoint closes up while we have+ -- an outgoing connection (CloseEndPoint or CloseSocket message),+ -- we'll post the connection lost event but we won't update these+ -- 'connAlive' IORefs.+ then throwIO $ TransportError SendFailed "Remote endpoint closed"+ else throwIO $ TransportError SendClosed "Connection closed"+ RemoteEndPointFailed err -> do+ alive <- readIORef connAlive+ if alive+ then throwIO $ TransportError SendFailed (show err)+ else throwIO $ TransportError SendClosed "Connection closed"+ where+ sendFailed = TransportError SendFailed . show++-- | Force-close the endpoint+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+ mOurState <- modifyMVar (localState ourEndPoint) $ \st ->+ case st of+ LocalEndPointValid vst ->+ return (LocalEndPointClosed, Just vst)+ LocalEndPointClosed ->+ return (LocalEndPointClosed, Nothing)+ forM_ mOurState $ \vst -> do+ forM_ (vst ^. localConnections) tryCloseRemoteSocket+ let qdisc = localQueue ourEndPoint+ forM_ evs (qdiscEnqueue' qdisc (localAddress ourEndPoint))+ where+ -- Close the remote socket and return the set of all incoming connections+ tryCloseRemoteSocket :: RemoteEndPoint -> IO ()+ tryCloseRemoteSocket theirEndPoint = withScheduledAction ourEndPoint $ \sched -> do+ -- We make an attempt to close the connection nicely+ -- (by sending a CloseSocket first)+ let closed = RemoteEndPointFailed . userError $ "apiCloseEndPoint"+ modifyMVar_ (remoteState theirEndPoint) $ \st ->+ case st of+ RemoteEndPointInvalid _ ->+ return st+ RemoteEndPointInit resolved _ _ -> do+ putMVar resolved ()+ return closed+ RemoteEndPointValid vst -> do+ -- Schedule an action to send a CloseEndPoint message and then+ -- wait for the socket to actually close (meaning that this+ -- end point is no longer receiving from it).+ -- Since we replace the state in this MVar with 'closed', it's+ -- guaranteed that no other actions will be scheduled after this+ -- one.+ sched theirEndPoint $ do+ void $ tryIO $ sendOn vst+ [ encodeWord32 (encodeControlHeader CloseEndPoint) ]+ -- Release probing resources if probing.+ forM_ (remoteProbing vst) id+ tryShutdownSocketBoth (remoteSocket vst)+ remoteSocketClosed vst+ return closed+ RemoteEndPointClosing resolved vst -> do+ -- Release probing resources if probing.+ forM_ (remoteProbing vst) id+ putMVar resolved ()+ -- Schedule an action to wait for the socket to actually close (this+ -- end point is no longer receiving from it).+ -- Since we replace the state in this MVar with 'closed', it's+ -- guaranteed that no other actions will be scheduled after this+ -- one.+ sched theirEndPoint $ do+ tryShutdownSocketBoth (remoteSocket vst)+ remoteSocketClosed vst+ return closed+ RemoteEndPointClosed ->+ return st+ RemoteEndPointFailed err ->+ return (RemoteEndPointFailed err)+++--------------------------------------------------------------------------------+-- Incoming requests --+--------------------------------------------------------------------------------++-- | Handle a connection request (that is, a remote endpoint that is trying to+-- establish a TCP connection with us)+--+-- 'handleConnectionRequest' runs in the context of the transport thread, which+-- can be killed asynchronously by 'closeTransport'. We fork a separate thread+-- as soon as we have located the lcoal endpoint that the remote endpoint is+-- interested in. We cannot fork any sooner because then we have no way of+-- storing the thread ID and hence no way of killing the thread when we take+-- 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 -> IO () -> (N.Socket, N.SockAddr) -> IO ()+handleConnectionRequest transport socketClosed (sock, sockAddr) = handle handleException $ do+ when (tcpNoDelay $ transportParams transport) $+ N.setSocketOption sock N.NoDelay 1+ when (tcpKeepAlive $ transportParams transport) $+ N.setSocketOption sock N.KeepAlive 1+ forM_ (tcpUserTimeout $ transportParams transport) $+ N.setSocketOption sock N.UserTimeout+ let handleVersioned = do+ -- Always receive the protocol version and a handshake (content of the+ -- handshake is version-dependent, but the length is always sent,+ -- regardless of the version).+ protocolVersion <- recvWord32 sock+ handshakeLength <- recvWord32 sock+ -- For now we support only version 0.0.0.0.+ case protocolVersion of+ 0x00000000 -> handleConnectionRequestV0 (sock, sockAddr)+ _ -> do+ -- Inform the peer that we want version 0x00000000+ sendMany sock [+ encodeWord32 (encodeConnectionRequestResponse ConnectionRequestUnsupportedVersion)+ , encodeWord32 0x00000000+ ]+ -- Clear the socket of the unsupported handshake data.+ _ <- recvExact sock handshakeLength+ handleVersioned+ -- The handshake must complete within the optional timeout duration.+ -- No socket 'recv's are to be run outside the timeout. The continuation+ -- returned may 'send', but not 'recv'.+ let connTimeout = transportConnectTimeout (transportParams transport)+ outcome <- maybe (fmap Just) System.Timeout.timeout connTimeout handleVersioned+ case outcome of+ Nothing -> throwIO (userError "handleConnectionRequest: timed out")+ Just act -> forM_ act id++ where++ handleException :: SomeException -> IO ()+ handleException ex = do+ rethrowIfAsync (fromException ex)++ rethrowIfAsync :: Maybe AsyncException -> IO ()+ rethrowIfAsync = mapM_ throwIO++ handleConnectionRequestV0 :: (N.Socket, N.SockAddr) -> IO (Maybe (IO ()))+ handleConnectionRequestV0 (sock, sockAddr) = do+ -- Get the OS-determined host and port.+ (numericHost, resolvedHost, _actualPort) <-+ resolveSockAddr sockAddr >>=+ maybe (throwIO (userError "handleConnectionRequest: invalid socket address")) return+ -- The peer must send our identifier and their address promptly, if a+ -- timeout is set.+ (ourEndPointId, theirAddress, mTheirHost) <- do+ ourEndPointId <- recvWord32 sock+ let maxAddressLength = tcpMaxAddressLength $ transportParams transport+ mTheirAddress <- BS.concat <$> recvWithLength maxAddressLength sock+ -- Sending a length = 0 address means unaddressable.+ if BS.null mTheirAddress+ then do+ theirAddress <- randomEndPointAddress+ return (ourEndPointId, theirAddress, Nothing)+ else do+ let theirAddress = EndPointAddress mTheirAddress+ (theirHost, _, _)+ <- maybe (throwIO (userError "handleConnectionRequest: peer gave malformed address"))+ return+ (decodeEndPointAddress theirAddress)+ return (ourEndPointId, theirAddress, Just theirHost)+ let checkPeerHost = tcpCheckPeerHost (transportParams transport)+ continue <- case (mTheirHost, checkPeerHost) of+ (Just theirHost, True) -> do+ -- If the OS-determined host doesn't match the host that the peer gave us,+ -- then we have no choice but to reject the connection. It's because we+ -- use the EndPointAddress to key the remote end points (localConnections)+ -- and we don't want to allow a peer to deny service to other peers by+ -- claiming to have their host and port.+ if theirHost == numericHost || theirHost == resolvedHost+ then return True+ else do+ sendMany sock $+ encodeWord32 (encodeConnectionRequestResponse ConnectionRequestHostMismatch)+ : (prependLength [BSC.pack theirHost] ++ prependLength [BSC.pack numericHost] ++ prependLength [BSC.pack resolvedHost])+ return False+ _ -> return True+ if continue+ then do+ ourEndPoint <- withMVar (transportState transport) $ \st -> case st of+ TransportValid vst ->+ case vst ^. localEndPointAt ourEndPointId of+ Nothing -> do+ sendMany sock [encodeWord32 (encodeConnectionRequestResponse ConnectionRequestInvalid)]+ throwIO $ userError "handleConnectionRequest: Invalid endpoint"+ Just ourEndPoint ->+ return ourEndPoint+ TransportClosed ->+ throwIO $ userError "Transport closed"+ return (Just (go ourEndPoint theirAddress))+ else return Nothing++ where++ go :: LocalEndPoint -> EndPointAddress -> IO ()+ go ourEndPoint theirAddress = handle handleException $ do++ resetIfBroken ourEndPoint theirAddress+ (theirEndPoint, isNew) <-+ findRemoteEndPoint ourEndPoint theirAddress RequestedByThem Nothing++ if not isNew+ then do+ void $ tryIO $ sendMany sock+ [encodeWord32 (encodeConnectionRequestResponse ConnectionRequestCrossed)]+ probeIfValid theirEndPoint+ else do+ sendLock <- newMVar Nothing+ let vst = ValidRemoteEndPointState+ { remoteSocket = sock+ , remoteSocketClosed = socketClosed+ , remoteProbing = Nothing+ , remoteSendLock = sendLock+ , _remoteOutgoing = 0+ , _remoteIncoming = Set.empty+ , _remoteLastIncoming = 0+ , _remoteNextConnOutId = firstNonReservedLightweightConnectionId+ }+ sendMany sock [encodeWord32 (encodeConnectionRequestResponse ConnectionRequestAccepted)]+ -- resolveInit will update the shared state, and handleIncomingMessages+ -- will always ultimately clean up after it.+ -- Closing up the socket is also out of our hands. It will happen+ -- when handleIncomingMessages finishes.+ resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst)+ `finally`+ handleIncomingMessages (transportParams transport) (ourEndPoint, theirEndPoint)++ probeIfValid :: RemoteEndPoint -> IO ()+ probeIfValid theirEndPoint = modifyMVar_ (remoteState theirEndPoint) $+ \st -> case st of+ RemoteEndPointValid+ vst@(ValidRemoteEndPointState { remoteProbing = Nothing }) -> do+ tid <- forkIO $ do+ -- send probe+ let params = transportParams transport+ void $ tryIO $ System.Timeout.timeout+ (maybe (-1) id $ transportConnectTimeout params) $ do+ sendMany (remoteSocket vst)+ [encodeWord32 (encodeControlHeader ProbeSocket)]+ threadDelay maxBound+ -- Discard the connection if this thread is not killed (i.e. the+ -- probe ack does not arrive on time).+ --+ -- The thread handling incoming messages will detect the socket is+ -- closed and will report the failure upwards.+ tryCloseSocket (remoteSocket vst)+ -- Waiting the probe ack and closing the socket is only needed in+ -- platforms where TCP_USER_TIMEOUT is not available or when the+ -- user does not set it. Otherwise the ack would be handled at the+ -- TCP level and the the thread handling incoming messages would+ -- get the error.++ return $ RemoteEndPointValid+ vst { remoteProbing = Just (killThread tid) }+ _ -> return st++-- | 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 :: TCPParameters -> EndPointPair -> IO ()+handleIncomingMessages params (ourEndPoint, theirEndPoint) =+ bracket acquire release act++ where++ -- Use shared remote endpoint state to get a socket, or an appropriate+ -- exception in case it's neither valid nor closing.+ acquire :: IO (Either IOError N.Socket)+ acquire = withMVar theirState $ \st -> case st of+ RemoteEndPointInvalid _ ->+ relyViolation (ourEndPoint, theirEndPoint)+ "handleIncomingMessages (invalid)"+ RemoteEndPointInit _ _ _ ->+ relyViolation (ourEndPoint, theirEndPoint)+ "handleIncomingMessages (init)"+ RemoteEndPointValid ep ->+ return . Right $ remoteSocket ep+ RemoteEndPointClosing _ ep ->+ return . Right $ remoteSocket ep+ RemoteEndPointClosed ->+ return . Left $ userError "handleIncomingMessages (already closed)"+ RemoteEndPointFailed _ ->+ return . Left $ userError "handleIncomingMessages (failed)"++ -- 'Right' is the normal case in which there still is a live socket to+ -- the remote endpoint, and so 'act' was run and installed its own+ -- exception handler.+ release :: Either IOError N.Socket -> IO ()+ release (Left err) = prematureExit err+ release (Right _) = return ()++ act :: Either IOError N.Socket -> IO ()+ act (Left _) = return ()+ act (Right sock) = go sock `catch` prematureExit++ -- 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+ -- endpoint is put into a Closed (or Closing) state by a concurrent thread+ -- (because a 'send' failed) -- the individual handlers below will throw a+ -- user exception which is then caught and handled the same way as an+ -- exception thrown by 'recv'.+ go :: N.Socket -> IO ()+ go sock = do+ lcid <- recvWord32 sock :: IO LightweightConnectionId+ if lcid >= firstNonReservedLightweightConnectionId+ then do+ readMessage sock lcid+ go sock+ else+ case decodeControlHeader lcid of+ Just CreatedNewConnection -> do+ recvWord32 sock >>= createdNewConnection+ go sock+ Just CloseConnection -> do+ recvWord32 sock >>= closeConnection+ go sock+ Just CloseSocket -> do+ didClose <- recvWord32 sock >>= closeSocket sock+ unless didClose $ go sock+ Just CloseEndPoint -> do+ let closeRemoteEndPoint vst = do+ forM_ (remoteProbing vst) id+ -- close incoming connections+ forM_ (Set.elems $ vst ^. remoteIncoming) $+ qdiscEnqueue' ourQueue theirAddr . ConnectionClosed . connId+ -- report the endpoint as gone if we have any outgoing+ -- connections+ when (vst ^. remoteOutgoing > 0) $ do+ let code = EventConnectionLost (remoteAddress theirEndPoint)+ qdiscEnqueue' ourQueue theirAddr . ErrorEvent $+ TransportError code "The remote endpoint was closed."+ removeRemoteEndPoint (ourEndPoint, theirEndPoint)+ modifyMVar_ theirState $ \s -> case s of+ RemoteEndPointValid vst -> do+ closeRemoteEndPoint vst+ return RemoteEndPointClosed+ RemoteEndPointClosing resolved vst -> do+ closeRemoteEndPoint vst+ putMVar resolved ()+ return RemoteEndPointClosed+ _ -> return s+ Just ProbeSocket -> do+ forkIO $ sendMany sock [encodeWord32 (encodeControlHeader ProbeSocketAck)]+ go sock+ Just ProbeSocketAck -> do+ stopProbing+ go sock+ Nothing ->+ throwIO $ userError "Invalid control request"++ -- Create a new connection+ createdNewConnection :: LightweightConnectionId -> IO ()+ createdNewConnection lcid = do+ modifyMVar_ theirState $ \st -> do+ vst <- case st of+ RemoteEndPointInvalid _ ->+ relyViolation (ourEndPoint, theirEndPoint)+ "handleIncomingMessages:createNewConnection (invalid)"+ RemoteEndPointInit _ _ _ ->+ relyViolation (ourEndPoint, theirEndPoint)+ "handleIncomingMessages:createNewConnection (init)"+ RemoteEndPointValid vst ->+ return ( (remoteIncoming ^: Set.insert lcid)+ $ (remoteLastIncoming ^= lcid)+ vst+ )+ RemoteEndPointClosing resolved vst -> do+ -- If the endpoint is in closing state that means we send a+ -- CloseSocket request to the remote endpoint. If the remote+ -- endpoint replies that it created a new connection, it either+ -- ignored our request or it sent the request before it got ours.+ -- Either way, at this point we simply restore the endpoint to+ -- RemoteEndPointValid+ putMVar resolved ()+ return ( (remoteIncoming ^= Set.singleton lcid)+ . (remoteLastIncoming ^= lcid)+ $ vst+ )+ RemoteEndPointFailed err ->+ throwIO err+ RemoteEndPointClosed ->+ relyViolation (ourEndPoint, theirEndPoint)+ "createNewConnection (closed)"+ return (RemoteEndPointValid vst)+ qdiscEnqueue' ourQueue theirAddr (ConnectionOpened (connId lcid) ReliableOrdered theirAddr)++ -- 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 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)) $+ throwIO $ userError "Invalid CloseConnection"+ 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"+ RemoteEndPointFailed err ->+ throwIO err+ RemoteEndPointClosed ->+ relyViolation (ourEndPoint, theirEndPoint) "closeConnection (closed)"+ qdiscEnqueue' ourQueue theirAddr (ConnectionClosed (connId lcid))++ -- Close the socket (if we don't have any outgoing connections)+ closeSocket :: N.Socket -> LightweightConnectionId -> IO Bool+ closeSocket _sock lastReceivedId = do+ mAct <- modifyMVar theirState $ \st -> do+ case st of+ RemoteEndPointInvalid _ ->+ relyViolation (ourEndPoint, theirEndPoint)+ "handleIncomingMessages:closeSocket (invalid)"+ RemoteEndPointInit _ _ _ ->+ 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) $+ qdiscEnqueue' ourQueue theirAddr . ConnectionClosed . connId+ let vst' = remoteIncoming ^= Set.empty $ vst+ -- The peer sends the connection id of the last connection which+ -- they accepted from us.+ --+ -- If it's not the same as the id of the last connection that we+ -- have made to them (assuming we haven't cycled through all+ -- identifiers so fast) then they hadn't seen the request before+ -- they tried to close the socket. In that case, we don't close the+ -- socket. They'll see our in-flight connection request and then+ -- abandon their attempt to close the socket.+ --+ -- It's possible that a local connection is coming up but has not+ -- yet sent CreatedNewConnection (see createConnectionTo), in+ -- which case remoteOutgoing is positive, but the sent and received+ -- ids do match. In this case we don't close the socket, because+ -- that connection will soon sent the message and bump the lastSentId.+ --+ -- Both disjuncts are needed: it's possible that remoteOutgoing is+ -- 0 and the ids do not match, in case we have created and closed+ -- a connection but the peer has not yet heard of it.+ if vst ^. remoteOutgoing > 0 || lastReceivedId /= lastSentId vst+ then+ return (RemoteEndPointValid vst', Nothing)+ else do+ -- Release probing resources if probing.+ forM_ (remoteProbing vst) id+ removeRemoteEndPoint (ourEndPoint, theirEndPoint)+ -- Attempt to reply (but don't insist)+ act <- schedule theirEndPoint $ do+ void $ tryIO $ sendOn vst'+ [ encodeWord32 (encodeControlHeader CloseSocket)+ , encodeWord32 (vst ^. remoteLastIncoming)+ ]+ 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+ -- way this may happen is when we sent a ConnectionCreated,+ -- ConnectionClosed, and CloseSocket message, none of which have+ -- yet been received. It's sufficient to check that the peer has+ -- not seen the ConnectionCreated message. In case they have seen+ -- it (so that lastReceivedId == lastSendId vst) then they must+ -- have seen the other messages or else they would not have sent+ -- CloseSocket.+ -- We leave the endpoint in closing state in that case.+ if lastReceivedId /= lastSentId vst+ then do+ return (RemoteEndPointClosing resolved vst, Nothing)+ else do+ -- Release probing resources if probing.+ when (vst ^. remoteOutgoing > 0) $ do+ let code = EventConnectionLost (remoteAddress theirEndPoint)+ let msg = "socket closed prematurely by peer"+ qdiscEnqueue' ourQueue theirAddr . ErrorEvent $ TransportError code msg+ forM_ (remoteProbing vst) id+ removeRemoteEndPoint (ourEndPoint, theirEndPoint)+ -- Nothing to do, but we want to indicate that the socket+ -- really did close.+ act <- schedule theirEndPoint $ return ()+ putMVar resolved ()+ return (RemoteEndPointClosed, Just act)+ RemoteEndPointFailed err ->+ throwIO err+ RemoteEndPointClosed ->+ 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 recvLimit sock >>=+ qdiscEnqueue' ourQueue theirAddr . Received (connId lcid)++ -- Stop probing a connection as a result of receiving a probe ack.+ stopProbing :: IO ()+ stopProbing = modifyMVar_ theirState $ \st -> case st of+ RemoteEndPointValid+ vst@(ValidRemoteEndPointState { remoteProbing = Just stop }) -> do+ stop+ return $ RemoteEndPointValid vst { remoteProbing = Nothing }+ _ -> return st++ -- Arguments+ ourQueue = localQueue ourEndPoint+ ourState = localState ourEndPoint+ theirState = remoteState theirEndPoint+ theirAddr = remoteAddress theirEndPoint+ recvLimit = tcpMaxReceiveLength params++ -- Deal with a premature exit+ prematureExit :: IOException -> IO ()+ prematureExit err = do+ modifyMVar_ theirState $ \st ->+ case st of+ RemoteEndPointInvalid _ ->+ relyViolation (ourEndPoint, theirEndPoint)+ "handleIncomingMessages:prematureExit"+ RemoteEndPointInit _ _ _ ->+ relyViolation (ourEndPoint, theirEndPoint)+ "handleIncomingMessages:prematureExit"+ RemoteEndPointValid vst -> do+ -- Release probing resources if probing.+ forM_ (remoteProbing vst) id+ let code = EventConnectionLost (remoteAddress theirEndPoint)+ qdiscEnqueue' ourQueue theirAddr . ErrorEvent $ TransportError code (show err)+ return (RemoteEndPointFailed err)+ RemoteEndPointClosing resolved vst -> do+ -- Release probing resources if probing.+ forM_ (remoteProbing vst) id+ putMVar resolved ()+ return (RemoteEndPointFailed err)+ RemoteEndPointClosed ->+ relyViolation (ourEndPoint, theirEndPoint)+ "handleIncomingMessages:prematureExit"+ RemoteEndPointFailed err' -> do+ -- Here we post a connection-lost event, but only if the+ -- local endpoint is not closed; if it's closed, the EndPointClosed+ -- event will be posted without connection-lost events, and this is+ -- part of the network-transport specification (there's a test+ -- case for it).+ modifyMVar_ ourState $ \st' -> case st' of+ LocalEndPointClosed -> return st'+ LocalEndPointValid _ -> do+ let code = EventConnectionLost (remoteAddress theirEndPoint)+ err = TransportError code (show err')+ qdiscEnqueue' ourQueue theirAddr (ErrorEvent err)+ return st'+ return (RemoteEndPointFailed err')++ -- Construct a connection ID+ connId :: LightweightConnectionId -> ConnectionId+ connId = createConnectionId (remoteId theirEndPoint)++ -- The ID of the last connection _we_ created (or 0 for none)+ lastSentId :: ValidRemoteEndPointState -> LightweightConnectionId+ lastSentId vst =+ if vst ^. remoteNextConnOutId == firstNonReservedLightweightConnectionId+ then 0+ else (vst ^. remoteNextConnOutId) - 1++--------------------------------------------------------------------------------+-- Uninterruptable auxiliary functions --+-- --+-- All these functions assume they are running in a thread which will never --+-- be killed.+--------------------------------------------------------------------------------++-- | Create a connection to a remote endpoint+--+-- If the remote endpoint is in 'RemoteEndPointClosing' state then we will+-- block until that is resolved.+--+-- May throw a TransportError ConnectErrorCode exception.+createConnectionTo+ :: TCPTransport+ -> LocalEndPoint+ -> EndPointAddress+ -> ConnectHints+ -> IO (RemoteEndPoint, LightweightConnectionId)+createConnectionTo transport ourEndPoint theirAddress hints = do+ -- @timer@ is an IO action that completes when the timeout expires.+ timer <- case connTimeout of+ Just t -> do+ mv <- newEmptyMVar+ _ <- forkIO $ threadDelay t >> putMVar mv ()+ return $ Just $ readMVar mv+ _ -> return Nothing+ go timer Nothing++ where++ params = transportParams transport+ connTimeout = connectTimeout hints `mplus` transportConnectTimeout params++ -- The second argument indicates the response obtained to the last+ -- connection request and the remote endpoint that was used.+ go timer mr = do+ (theirEndPoint, isNew) <- mapIOException connectFailed+ (findRemoteEndPoint ourEndPoint theirAddress RequestedByUs timer)+ `finally` case mr of+ Just (theirEndPoint, ConnectionRequestCrossed) ->+ modifyMVar_ (remoteState theirEndPoint) $+ \rst -> case rst of+ RemoteEndPointInit resolved _ _ -> do+ putMVar resolved ()+ removeRemoteEndPoint (ourEndPoint, theirEndPoint)+ return RemoteEndPointClosed+ _ -> return rst+ _ -> return ()+ if isNew+ then do+ mr' <- handle (absorbAllExceptions Nothing) $+ setupRemoteEndPoint transport (ourEndPoint, theirEndPoint) connTimeout+ go timer (fmap ((,) theirEndPoint) mr')+ 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+ act <- schedule theirEndPoint $ do+ sendOn vst [+ encodeWord32 (encodeControlHeader CreatedNewConnection)+ , encodeWord32 connId+ ]+ return connId+ return ( RemoteEndPointValid+ $ remoteNextConnOutId ^= connId + 1+ $ vst+ , act+ )+ -- Error cases+ RemoteEndPointInvalid err ->+ throwIO err+ RemoteEndPointFailed err ->+ throwIO err+ -- Algorithmic errors+ _ ->+ relyViolation (ourEndPoint, theirEndPoint) "createConnectionTo"+ -- TODO: deal with exception case?+ connId <- runScheduledAction (ourEndPoint, theirEndPoint) act+ return (theirEndPoint, connId)+++ connectFailed :: IOException -> TransportError ConnectErrorCode+ connectFailed = TransportError ConnectFailed . show++ absorbAllExceptions :: a -> SomeException -> IO a+ absorbAllExceptions a _ex =+ return a++-- | Set up a remote endpoint+setupRemoteEndPoint+ :: TCPTransport+ -> EndPointPair+ -> Maybe Int+ -> IO (Maybe ConnectionRequestResponse)+setupRemoteEndPoint transport (ourEndPoint, theirEndPoint) connTimeout = do+ let mOurAddress = const ourAddress <$> transportAddrInfo transport+ result <- socketToEndPoint mOurAddress+ theirAddress+ (tcpReuseClientAddr params)+ (tcpNoDelay params)+ (tcpKeepAlive params)+ (tcpUserTimeout params)+ connTimeout+ didAccept <- case result of+ -- Since a socket was created, we are now responsible for closing it.+ --+ -- In case the connection was accepted, we have some work to do.+ -- We'll remember how to wait for the socket to close+ -- (readMVar socketClosedVar), and we'll take care of closing it up+ -- once handleIncomingMessages has finished.+ Right (socketClosedVar, sock, ConnectionRequestAccepted) -> do+ sendLock <- newMVar Nothing+ let vst = ValidRemoteEndPointState+ { remoteSocket = sock+ , remoteSocketClosed = readMVar socketClosedVar+ , remoteProbing = Nothing+ , remoteSendLock = sendLock+ , _remoteOutgoing = 0+ , _remoteIncoming = Set.empty+ , _remoteLastIncoming = 0+ , _remoteNextConnOutId = firstNonReservedLightweightConnectionId+ }+ resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst)+ return (Just (socketClosedVar, sock))+ Right (socketClosedVar, sock, ConnectionRequestUnsupportedVersion) -> do+ -- If the peer doesn't support V0 then there's nothing we can do, for+ -- it's the only version we support.+ let err = connectFailed "setupRemoteEndPoint: unsupported version"+ resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointInvalid err)+ tryCloseSocket sock `finally` putMVar socketClosedVar ()+ return Nothing+ Right (socketClosedVar, sock, ConnectionRequestInvalid) -> do+ let err = invalidAddress "setupRemoteEndPoint: Invalid endpoint"+ resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointInvalid err)+ tryCloseSocket sock `finally` putMVar socketClosedVar ()+ return Nothing+ Right (socketClosedVar, sock, ConnectionRequestCrossed) -> do+ withMVar (remoteState theirEndPoint) $ \st -> case st of+ RemoteEndPointInit _ crossed _ ->+ putMVar crossed ()+ RemoteEndPointFailed ex ->+ throwIO ex+ _ ->+ relyViolation (ourEndPoint, theirEndPoint) "setupRemoteEndPoint: Crossed"+ tryCloseSocket sock `finally` putMVar socketClosedVar ()+ return Nothing+ Right (socketClosedVar, sock, ConnectionRequestHostMismatch) -> do+ let handler :: SomeException -> IO (TransportError ConnectErrorCode)+ handler err = return (TransportError ConnectFailed (show err))+ err <- handle handler $ do+ claimedHost <- recvWithLength (tcpMaxReceiveLength params) sock+ actualNumericHost <- recvWithLength (tcpMaxReceiveLength params) sock+ actualResolvedHost <- recvWithLength (tcpMaxReceiveLength params) sock+ let reason = concat [+ "setupRemoteEndPoint: Host mismatch"+ , ". Claimed: "+ , BSC.unpack (BS.concat claimedHost)+ , "; Numeric: "+ , BSC.unpack (BS.concat actualNumericHost)+ , "; Resolved: "+ , BSC.unpack (BS.concat actualResolvedHost)+ ]+ return (TransportError ConnectFailed reason)+ resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointInvalid err)+ tryCloseSocket sock `finally` putMVar socketClosedVar ()+ return Nothing+ Left err -> do+ resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointInvalid err)+ return Nothing++ -- We handle incoming messages in a separate thread, and are careful to+ -- always close the socket once that thread is finished.+ forM_ didAccept $ \(socketClosed, sock) -> void $ forkIO $+ handleIncomingMessages params (ourEndPoint, theirEndPoint)+ `finally`+ (tryCloseSocket sock `finally` putMVar socketClosed ())+ return $ either (const Nothing) (Just . (\(_,_,x) -> x)) result+ where+ params = transportParams transport+ ourAddress = localAddress ourEndPoint+ theirAddress = remoteAddress theirEndPoint+ invalidAddress = TransportError ConnectNotFound+ connectFailed = TransportError ConnectFailed++-- | Send a CloseSocket request if the remote endpoint is unused+closeIfUnused :: EndPointPair -> IO ()+closeIfUnused (ourEndPoint, theirEndPoint) = do+ mAct <- modifyMVar (remoteState theirEndPoint) $ \st -> case st of+ RemoteEndPointValid vst ->+ if vst ^. remoteOutgoing == 0 && Set.null (vst ^. remoteIncoming)+ then do+ resolved <- newEmptyMVar+ act <- schedule theirEndPoint $+ sendOn vst [ encodeWord32 (encodeControlHeader CloseSocket)+ , encodeWord32 (vst ^. remoteLastIncoming)+ ]+ return (RemoteEndPointClosing resolved vst, Just act)+ else+ return (RemoteEndPointValid vst, Nothing)+ _ ->+ return (st, Nothing)+ 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+--+-- - a user calls the API function 'connect', or and the remote endpoint is+-- - an inbound connection request comes in from this remote address+--+-- we remove the remote endpoint first.+--+-- Throws a TransportError ConnectFailed exception if the local endpoint is+-- closed.+resetIfBroken :: LocalEndPoint -> EndPointAddress -> IO ()+resetIfBroken ourEndPoint theirAddress = do+ mTheirEndPoint <- withMVar (localState ourEndPoint) $ \st -> case st of+ LocalEndPointValid vst ->+ return (vst ^. localConnectionTo theirAddress)+ LocalEndPointClosed ->+ throwIO $ TransportError ConnectFailed "Endpoint closed"+ forM_ mTheirEndPoint $ \theirEndPoint ->+ withMVar (remoteState theirEndPoint) $ \st -> case st of+ RemoteEndPointInvalid _ ->+ removeRemoteEndPoint (ourEndPoint, theirEndPoint)+ RemoteEndPointFailed _ ->+ removeRemoteEndPoint (ourEndPoint, theirEndPoint)+ _ ->+ return ()++-- | Special case of 'apiConnect': connect an endpoint to itself+--+-- May throw a TransportError ConnectErrorCode (if the local endpoint is closed)+connectToSelf :: LocalEndPoint+ -> IO Connection+connectToSelf ourEndPoint = do+ connAlive <- newIORef True -- Protected by the local endpoint lock+ lconnId <- mapIOException connectFailed $ getLocalNextConnOutId ourEndPoint+ let connId = createConnectionId heavyweightSelfConnectionId lconnId+ qdiscEnqueue' ourQueue ourAddress $+ ConnectionOpened connId ReliableOrdered (localAddress ourEndPoint)+ return Connection+ { send = selfSend connAlive connId+ , close = selfClose connAlive connId+ }+ where+ selfSend :: IORef Bool+ -> ConnectionId+ -> [ByteString]+ -> IO (Either (TransportError SendErrorCode) ())+ selfSend connAlive connId msg =+ try . withMVar ourState $ \st -> case st of+ LocalEndPointValid _ -> do+ alive <- readIORef connAlive+ if alive+ then seq (foldr seq () msg)+ qdiscEnqueue' ourQueue ourAddress (Received connId msg)+ else throwIO $ TransportError SendClosed "Connection closed"+ LocalEndPointClosed ->+ throwIO $ TransportError SendFailed "Endpoint closed"++ selfClose :: IORef Bool -> ConnectionId -> IO ()+ selfClose connAlive connId =+ withMVar ourState $ \st -> case st of+ LocalEndPointValid _ -> do+ alive <- readIORef connAlive+ when alive $ do+ qdiscEnqueue' ourQueue ourAddress (ConnectionClosed connId)+ writeIORef connAlive False+ LocalEndPointClosed ->+ return ()++ ourQueue = localQueue ourEndPoint+ ourState = localState ourEndPoint+ connectFailed = TransportError ConnectFailed . show+ ourAddress = localAddress ourEndPoint++-- | Resolve an endpoint currently in 'Init' state+resolveInit :: EndPointPair -> RemoteState -> IO ()+resolveInit (ourEndPoint, theirEndPoint) newState =+ modifyMVar_ (remoteState theirEndPoint) $ \st -> case st of+ RemoteEndPointInit resolved crossed _ -> do+ putMVar resolved ()+ -- Unblock the reader (if any) if the ConnectionRequestCrossed+ -- message did not come within the connection timeout.+ tryPutMVar crossed ()+ case newState of+ RemoteEndPointClosed ->+ removeRemoteEndPoint (ourEndPoint, theirEndPoint)+ _ ->+ return ()+ return newState+ 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 =+ modifyMVar (localState ourEndpoint) $ \st -> case st of+ LocalEndPointValid vst -> do+ 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+ -> QDisc Event+ -> IO LocalEndPoint+createLocalEndPoint transport qdisc = do+ state <- newMVar . LocalEndPointValid $ ValidLocalEndPointState+ { _localNextConnOutId = firstNonReservedLightweightConnectionId+ , _localConnections = Map.empty+ , _nextConnInId = firstNonReservedHeavyweightConnectionId+ }+ modifyMVar (transportState transport) $ \st -> case st of+ TransportValid vst -> do+ let ix = vst ^. nextEndPointId+ addr <- case transportAddrInfo transport of+ Nothing -> randomEndPointAddress+ Just addrInfo -> return $+ encodeEndPointAddress (transportHost addrInfo)+ (transportPort addrInfo)+ ix+ let localEndPoint = LocalEndPoint { localAddress = addr+ , localEndPointId = ix+ , localQueue = qdisc+ , localState = state+ }+ return ( TransportValid+ . (localEndPointAt ix ^= Just localEndPoint)+ . (nextEndPointId ^= ix + 1)+ $ vst+ , localEndPoint+ )+ TransportClosed ->+ throwIO (TransportError NewEndPointFailed "Transport closed")+++-- | Remove reference to a remote endpoint from a local endpoint+--+-- If the local endpoint is closed, do nothing+removeRemoteEndPoint :: EndPointPair -> IO ()+removeRemoteEndPoint (ourEndPoint, theirEndPoint) =+ modifyMVar_ ourState $ \st -> case st of+ LocalEndPointValid vst ->+ case vst ^. localConnectionTo theirAddress of+ Nothing ->+ return st+ Just remoteEndPoint' ->+ if remoteId remoteEndPoint' == remoteId theirEndPoint+ then return+ ( LocalEndPointValid+ . (localConnectionTo (remoteAddress theirEndPoint) ^= Nothing)+ $ vst+ )+ else return st+ LocalEndPointClosed ->+ return LocalEndPointClosed+ where+ ourState = localState ourEndPoint+ theirAddress = remoteAddress theirEndPoint++-- | Remove reference to a local endpoint from the transport state+--+-- Does nothing if the transport is closed+removeLocalEndPoint :: TCPTransport -> LocalEndPoint -> IO ()+removeLocalEndPoint transport ourEndPoint =+ modifyMVar_ (transportState transport) $ \st -> case st of+ TransportValid vst ->+ return ( TransportValid+ . (localEndPointAt (localEndPointId 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, or 'Nothing' if+-- it times out.+findRemoteEndPoint+ :: LocalEndPoint+ -> EndPointAddress+ -> RequestedBy+ -> Maybe (IO ()) -- ^ an action which completes when the time is up+ -> IO (RemoteEndPoint, Bool)+findRemoteEndPoint ourEndPoint theirAddress findOrigin mtimer = go+ where+ go = do+ (theirEndPoint, isNew) <- modifyMVar ourState $ \st -> case st of+ LocalEndPointValid vst -> case vst ^. localConnectionTo theirAddress of+ Just theirEndPoint ->+ return (st, (theirEndPoint, False))+ Nothing -> do+ resolved <- newEmptyMVar+ crossed <- newEmptyMVar+ theirState <- newMVar (RemoteEndPointInit resolved crossed findOrigin)+ scheduled <- newChan+ let theirEndPoint = RemoteEndPoint+ { remoteAddress = theirAddress+ , remoteState = theirState+ , remoteId = vst ^. nextConnInId+ , remoteScheduled = scheduled+ }+ return ( LocalEndPointValid+ . (localConnectionTo theirAddress ^= Just theirEndPoint)+ . (nextConnInId ^: (+ 1))+ $ vst+ , (theirEndPoint, True)+ )+ LocalEndPointClosed ->+ throwIO $ userError "Local endpoint closed"++ if isNew+ then+ return (theirEndPoint, True)+ else do+ let theirState = remoteState theirEndPoint+ snapshot <- modifyMVar theirState $ \st -> case st of+ RemoteEndPointValid vst ->+ case findOrigin of+ RequestedByUs -> do+ let st' = RemoteEndPointValid+ . (remoteOutgoing ^: (+ 1))+ $ vst+ return (st', st')+ RequestedByThem ->+ 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')+ case snapshot of+ RemoteEndPointInvalid err ->+ throwIO err+ RemoteEndPointInit resolved crossed initOrigin ->+ case (findOrigin, initOrigin) of+ (RequestedByUs, RequestedByUs) ->+ readMVarTimeout mtimer resolved >> go+ (RequestedByUs, RequestedByThem) ->+ readMVarTimeout mtimer resolved >> go+ (RequestedByThem, RequestedByUs) ->+ if ourAddress > theirAddress+ then do+ -- Wait for the Crossed message and recheck the state+ -- of the remote endpoint after this (it may well be+ -- invalid already in case of a timeout).+ tryReadMVar crossed >>= \case+ Nothing -> readMVarTimeout mtimer crossed >> go+ _ -> return (theirEndPoint, True)+ else+ return (theirEndPoint, False)+ (RequestedByThem, RequestedByThem) ->+ throwIO $ userError "Already connected"+ RemoteEndPointValid _ ->+ -- We assume that the request crossed if we find the endpoint in+ -- Valid state. It is possible that this is really an invalid+ -- request, but only in the case of a broken client (we don't+ -- maintain enough history to be able to tell the difference).+ return (theirEndPoint, False)+ RemoteEndPointClosing resolved _ ->+ readMVarTimeout mtimer resolved >> go+ RemoteEndPointClosed ->+ go+ RemoteEndPointFailed err ->+ throwIO err++ ourState = localState ourEndPoint+ ourAddress = localAddress ourEndPoint++ -- | Like 'readMVar' but it throws an exception if the timer expires.+ readMVarTimeout Nothing mv = readMVar mv+ readMVarTimeout (Just timer) mv = do+ let connectTimedout = TransportError ConnectTimeout "Timed out"+ tid <- myThreadId+ bracket (forkIO $ timer >> throwTo tid connectTimedout) killThread $+ const $ readMVar mv++-- | Send a payload over a heavyweight connection (thread safe)+--+-- The socket cannot be used for sending after the non-atomic 'sendMany'+-- is interrupted - otherwise, the other side may get the msg corrupted.+--+-- There are two types of possible exceptions here:+-- 1) Outer asynchronous exceptions (like 'ProcessLinkException').+-- 2) Synchronous exceptions (inner or outer).+-- On a synchronous exception the remote endpoint is failed (see 'runScheduledAction',+-- for example) and its socket is not supposed to be used again.+--+-- With 'async' the code is run in a new thread which is not-targeted (and+-- thus, not interrupted) by the 1st type of exceptions. With 'remoteSendLock'+-- we protect the socket usage by the concurrent threads, as well as prevent+-- that usage after SomeException.+sendOn :: ValidRemoteEndPointState -> [ByteString] -> IO ()+sendOn vst bs = (wait =<<) $ async $+ mask $ \restore -> do+ let lock = remoteSendLock vst+ maybeException <- takeMVar lock+ when (isNothing maybeException) $+ restore (sendMany (remoteSocket vst) bs) `catch` \ex -> do+ putMVar lock (Just ex)+ throwIO ex+ putMVar lock maybeException+ forM_ maybeException $ \e ->+ throwIO $ userError $ "sendOn failed earlier with: " ++ show e++--------------------------------------------------------------------------------+-- Scheduling actions --+--------------------------------------------------------------------------------++-- | See 'schedule'/'runScheduledAction'+type Action a = MVar (Either SomeException a)++-- | Schedule an action to be executed (see also 'runScheduledAction')+schedule :: RemoteEndPoint -> IO a -> IO (Action a)+schedule theirEndPoint act = do+ mvar <- newEmptyMVar+ writeChan (remoteScheduled theirEndPoint) $+ catch (act >>= putMVar mvar . Right) (putMVar mvar . Left)+ return mvar++-- | Run a scheduled action. Every call to 'schedule' should be paired with a+-- call to 'runScheduledAction' so that every scheduled action is run. Note+-- however that the there is no guarantee that in+--+-- > do act <- schedule p+-- > 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+-- 'runScheduledAction' or by another).+runScheduledAction :: EndPointPair -> Action a -> IO a+runScheduledAction (_ourEndPoint, theirEndPoint) mvar = do+ join $ readChan (remoteScheduled theirEndPoint)+ ma <- readMVar mvar+ case ma of+ Right a -> return a+ Left e -> do+ forM_ (fromException e) $ \ioe ->+ modifyMVar_ (remoteState theirEndPoint) $ \st ->+ case st of+ RemoteEndPointValid vst -> handleIOException ioe vst+ _ -> return (RemoteEndPointFailed ioe)+ throwIO e+ where+ handleIOException :: IOException+ -> ValidRemoteEndPointState+ -> IO RemoteState+ handleIOException ex vst = do+ -- Release probing resources if probing.+ forM_ (remoteProbing vst) id+ -- Must shut down the socket here, so that the other end will realize+ -- we lost the connection+ tryShutdownSocketBoth (remoteSocket vst)+ -- Eventually, handleIncomingMessages will fail while trying to+ -- receive, and ultimately enqueue the 'EventConnectionLost'.+ return (RemoteEndPointFailed ex)++-- | Use 'schedule' action 'runScheduled' action in a safe way, it's assumed that+-- callback is used only once, otherwise guarantees of runScheduledAction are not+-- respected.+withScheduledAction :: LocalEndPoint -> ((RemoteEndPoint -> IO a -> IO ()) -> IO ()) -> IO ()+withScheduledAction ourEndPoint f =+ bracket (newIORef Nothing)+ (traverse (\(tp, a) -> runScheduledAction (ourEndPoint, tp) a) <=< readIORef)+ (\ref -> f (\rp g -> mask_ $ schedule rp g >>= \x -> writeIORef ref (Just (rp,x)) ))++--------------------------------------------------------------------------------+-- "Stateless" (MVar free) functions --+--------------------------------------------------------------------------------++-- | Establish a connection to a remote endpoint+--+-- Maybe throw a TransportError+--+-- If a socket is created and returned (Right is given) then the caller is+-- responsible for eventually closing the socket and filling the MVar (which+-- is empty). The MVar must be filled immediately after, and never before,+-- the socket is closed.+socketToEndPoint :: Maybe EndPointAddress -- ^ Our address+ -> EndPointAddress -- ^ Their address+ -> Bool -- ^ Use SO_REUSEADDR?+ -> Bool -- ^ Use TCP_NODELAY+ -> Bool -- ^ Use TCP_KEEPALIVE+ -> Maybe Int -- ^ Maybe TCP_USER_TIMEOUT+ -> Maybe Int -- ^ Timeout for connect+ -> IO (Either (TransportError ConnectErrorCode)+ (MVar (), N.Socket, ConnectionRequestResponse))+socketToEndPoint mOurAddress theirAddress reuseAddr noDelay keepAlive+ mUserTimeout timeout =+ try $ do+ (host, port, theirEndPointId) <- case decodeEndPointAddress theirAddress of+ Nothing -> throwIO (failed . userError $ "Could not parse")+ Just dec -> return dec+ addr:_ <- mapIOException invalidAddress $+ N.getAddrInfo Nothing (Just host) (Just port)+ bracketOnError (createSocket addr) tryCloseSocket $ \sock -> do+ when reuseAddr $+ mapIOException failed $ N.setSocketOption sock N.ReuseAddr 1+ when noDelay $+ mapIOException failed $ N.setSocketOption sock N.NoDelay 1+ when keepAlive $+ mapIOException failed $ N.setSocketOption sock N.KeepAlive 1+ forM_ mUserTimeout $+ mapIOException failed . N.setSocketOption sock N.UserTimeout+ response <- timeoutMaybe timeout timeoutError $ do+ mapIOException invalidAddress $+ N.connect sock (N.addrAddress addr)+ mapIOException failed $ do+ case mOurAddress of+ Just (EndPointAddress ourAddress) ->+ sendMany sock $+ encodeWord32 currentProtocolVersion+ : prependLength (encodeWord32 theirEndPointId : prependLength [ourAddress])+ Nothing ->+ sendMany sock $+ encodeWord32 currentProtocolVersion+ : prependLength ([encodeWord32 theirEndPointId, encodeWord32 0])+ recvWord32 sock+ case decodeConnectionRequestResponse response of+ Nothing -> throwIO (failed . userError $ "Unexpected response")+ Just r -> do+ socketClosedVar <- newEmptyMVar+ return (socketClosedVar, sock, r)+ where+ createSocket :: N.AddrInfo -> IO N.Socket+ createSocket addr = mapIOException insufficientResources $+ N.socket (N.addrFamily addr) N.Stream N.defaultProtocol++ invalidAddress = TransportError ConnectNotFound . show+ insufficientResources = TransportError ConnectInsufficientResources . show+ failed = TransportError ConnectFailed . show+ timeoutError = TransportError ConnectTimeout "Timed out"++-- | Construct a ConnectionId+createConnectionId :: HeavyweightConnectionId+ -> LightweightConnectionId+ -> ConnectionId+createConnectionId hcid lcid =+ (fromIntegral hcid `shiftL` 32) .|. fromIntegral lcid++--------------------------------------------------------------------------------+-- Functions from TransportInternals --+--------------------------------------------------------------------------------++-- Find a socket between two endpoints+--+-- Throws an IO exception if the socket could not be found.+internalSocketBetween :: TCPTransport -- ^ Transport+ -> EndPointAddress -- ^ Local endpoint+ -> EndPointAddress -- ^ Remote endpoint+ -> IO N.Socket+internalSocketBetween transport ourAddress theirAddress = do+ ourEndPointId <- case decodeEndPointAddress ourAddress of+ Just (_, _, eid) -> return eid+ _ -> throwIO $ userError "Malformed local EndPointAddress"+ ourEndPoint <- withMVar (transportState transport) $ \st -> case st of+ TransportClosed ->+ throwIO $ userError "Transport closed"+ TransportValid vst ->+ case vst ^. localEndPointAt ourEndPointId 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 ->+ 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 ->+ return $ remoteSocket vst+ RemoteEndPointClosing _ vst ->+ return $ remoteSocket vst+ RemoteEndPointClosed ->+ throwIO $ userError "Remote endpoint closed"+ RemoteEndPointInvalid err ->+ throwIO err+ RemoteEndPointFailed err ->+ throwIO err+ where++--------------------------------------------------------------------------------+-- Constants --+--------------------------------------------------------------------------------++-- | We reserve a bunch of connection IDs for control messages+firstNonReservedLightweightConnectionId :: LightweightConnectionId+firstNonReservedLightweightConnectionId = 1024++-- | Self-connection+heavyweightSelfConnectionId :: HeavyweightConnectionId+heavyweightSelfConnectionId = 0++-- | We reserve some connection IDs for special heavyweight connections+firstNonReservedHeavyweightConnectionId :: HeavyweightConnectionId+firstNonReservedHeavyweightConnectionId = 1++--------------------------------------------------------------------------------+-- Accessor definitions --+--------------------------------------------------------------------------------++localEndPoints :: Accessor ValidTransportState (Map EndPointId LocalEndPoint)+localEndPoints = accessor _localEndPoints (\es st -> st { _localEndPoints = es })++nextEndPointId :: Accessor ValidTransportState EndPointId+nextEndPointId = accessor _nextEndPointId (\eid st -> st { _nextEndPointId = eid })++localNextConnOutId :: Accessor ValidLocalEndPointState LightweightConnectionId+localNextConnOutId = accessor _localNextConnOutId (\cix st -> st { _localNextConnOutId = cix })++localConnections :: Accessor ValidLocalEndPointState (Map EndPointAddress RemoteEndPoint)+localConnections = accessor _localConnections (\es st -> st { _localConnections = es })++nextConnInId :: Accessor ValidLocalEndPointState HeavyweightConnectionId+nextConnInId = accessor _nextConnInId (\rid st -> st { _nextConnInId = rid })++remoteOutgoing :: Accessor ValidRemoteEndPointState Int+remoteOutgoing = accessor _remoteOutgoing (\cs conn -> conn { _remoteOutgoing = cs })++remoteIncoming :: Accessor ValidRemoteEndPointState (Set LightweightConnectionId)+remoteIncoming = accessor _remoteIncoming (\cs conn -> conn { _remoteIncoming = cs })++remoteLastIncoming :: Accessor ValidRemoteEndPointState LightweightConnectionId+remoteLastIncoming = accessor _remoteLastIncoming (\lcid st -> st { _remoteLastIncoming = lcid })++remoteNextConnOutId :: Accessor ValidRemoteEndPointState LightweightConnectionId+remoteNextConnOutId = accessor _remoteNextConnOutId (\cix st -> st { _remoteNextConnOutId = cix })++localEndPointAt :: EndPointId -> Accessor ValidTransportState (Maybe LocalEndPoint)+localEndPointAt addr = localEndPoints >>> DAC.mapMaybe addr++localConnectionTo :: EndPointAddress -> Accessor ValidLocalEndPointState (Maybe RemoteEndPoint)+localConnectionTo addr = localConnections >>> DAC.mapMaybe addr++-------------------------------------------------------------------------------+-- Debugging --+-------------------------------------------------------------------------------++relyViolation :: EndPointPair -> String -> IO a+relyViolation (ourEndPoint, theirEndPoint) str = do+ 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)+ ++ "(" ++ show (remoteId theirEndPoint) ++ ")"+ ++ "/" ++ show tid ++ ": " ++ msg
src/Network/Transport/TCP/Internal.hs view
@@ -1,24 +1,45 @@--- | Utility functions for TCP sockets -module Network.Transport.TCP.Internal - ( forkServer+{-# LANGUAGE CPP #-}+-- | Utility functions for TCP sockets+module Network.Transport.TCP.Internal+ ( ControlHeader(..)+ , encodeControlHeader+ , decodeControlHeader+ , ConnectionRequestResponse(..)+ , encodeConnectionRequestResponse+ , decodeConnectionRequestResponse+ , forkServer , recvWithLength- , recvExact - , recvInt32+ , recvExact+ , recvWord32+ , encodeWord32 , tryCloseSocket+ , tryShutdownSocketBoth+ , resolveSockAddr+ , EndPointId+ , encodeEndPointAddress+ , decodeEndPointAddress+ , randomEndPointAddress+ , ProtocolVersion+ , currentProtocolVersion ) where -#if ! MIN_VERSION_base(4,6,0)-import Prelude hiding (catch)-#endif+import Network.Transport.Internal+ ( decodeWord32+ , encodeWord32+ , void+ , tryIO+ , forkIOWithUnmask+ ) -import Network.Transport.Internal (decodeInt32, void, tryIO, forkIOWithUnmask)+import Network.Transport ( EndPointAddress(..) ) #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+ , NameInfoFlag(NI_NUMERICHOST) , ServiceName , Socket , SocketType(Stream)@@ -26,14 +47,19 @@ , getAddrInfo , defaultHints , socket- , bindSocket+ , bind , listen , addrFamily , addrAddress , defaultProtocol , setSocketOption , accept- , sClose+ , close+ , socketPort+ , shutdown+ , ShutdownCmd(ShutdownBoth)+ , SockAddr(..)+ , getNameInfo ) #ifdef USE_MOCK_NETWORK@@ -42,40 +68,151 @@ import qualified Network.Socket.ByteString as NBS (recv) #endif -import Control.Concurrent (ThreadId)+import Data.Word (Word32)+ import Control.Monad (forever, when) import Control.Exception (SomeException, catch, bracketOnError, throwIO, mask_)-import Control.Applicative ((<$>))+import Control.Concurrent (ThreadId, forkIO)+import Control.Concurrent.MVar+ ( MVar+ , newEmptyMVar+ , putMVar+ , readMVar+ )+import Control.Exception+ ( mask+ , finally+ )+ 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)+import qualified Data.ByteString.Char8 as BSC (unpack, pack)+import qualified Data.UUID as UUID+import qualified Data.UUID.V4 as UUID +-- | Local identifier for an endpoint within this transport+type EndPointId = Word32++-- | Identifies the version of the network-transport-tcp protocol.+-- It's the first piece of data sent when a new heavyweight connection is+-- established.+type ProtocolVersion = Word32++currentProtocolVersion :: ProtocolVersion+currentProtocolVersion = 0x00000000++-- | Control headers+data ControlHeader =+ -- | Tell the remote endpoint that we created a new connection+ CreatedNewConnection+ -- | Tell the remote endpoint we will no longer be using a connection+ | CloseConnection+ -- | Request to close the connection (see module description)+ | CloseSocket+ -- | Sent by an endpoint when it is closed.+ | CloseEndPoint+ -- | Message sent to probe a socket+ | ProbeSocket+ -- | Acknowledgement of the ProbeSocket message+ | ProbeSocketAck+ deriving (Show)++decodeControlHeader :: Word32 -> Maybe ControlHeader+decodeControlHeader w32 = case w32 of+ 0 -> Just CreatedNewConnection+ 1 -> Just CloseConnection+ 2 -> Just CloseSocket+ 3 -> Just CloseEndPoint+ 4 -> Just ProbeSocket+ 5 -> Just ProbeSocketAck+ _ -> Nothing++encodeControlHeader :: ControlHeader -> Word32+encodeControlHeader ch = case ch of+ CreatedNewConnection -> 0+ CloseConnection -> 1+ CloseSocket -> 2+ CloseEndPoint -> 3+ ProbeSocket -> 4+ ProbeSocketAck -> 5++-- | Response sent by /B/ to /A/ when /A/ tries to connect+data ConnectionRequestResponse =+ -- | /B/ does not support the protocol version requested by /A/.+ ConnectionRequestUnsupportedVersion+ -- | /B/ accepts the connection+ | ConnectionRequestAccepted+ -- | /A/ requested an invalid endpoint+ | ConnectionRequestInvalid+ -- | /A/s request crossed with a request from /B/ (see protocols)+ | ConnectionRequestCrossed+ -- | /A/ gave an incorrect host (did not match the host that /B/ observed).+ | ConnectionRequestHostMismatch+ deriving (Show)++decodeConnectionRequestResponse :: Word32 -> Maybe ConnectionRequestResponse+decodeConnectionRequestResponse w32 = case w32 of+ 0xFFFFFFFF -> Just ConnectionRequestUnsupportedVersion+ 0x00000000 -> Just ConnectionRequestAccepted+ 0x00000001 -> Just ConnectionRequestInvalid+ 0x00000002 -> Just ConnectionRequestCrossed+ 0x00000003 -> Just ConnectionRequestHostMismatch+ _ -> Nothing++encodeConnectionRequestResponse :: ConnectionRequestResponse -> Word32+encodeConnectionRequestResponse crr = case crr of+ ConnectionRequestUnsupportedVersion -> 0xFFFFFFFF+ ConnectionRequestAccepted -> 0x00000000+ ConnectionRequestInvalid -> 0x00000001+ ConnectionRequestCrossed -> 0x00000002+ ConnectionRequestHostMismatch -> 0x00000003++-- | Generate an EndPointAddress which does not encode a host/port/endpointid.+-- Such addresses are used for unreachable endpoints, and for ephemeral+-- addresses when such endpoints establish new heavyweight connections.+randomEndPointAddress :: IO EndPointAddress+randomEndPointAddress = do+ uuid <- UUID.nextRandom+ return $ EndPointAddress (UUID.toASCIIBytes uuid)+ -- | 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. -- -- Once the socket is created we spawn a new thread which repeatedly accepts--- incoming connections and executes the given request handler. If any--- exception occurs the thread terminates and calls the terminationHandler.--- This exception may occur because of a call to 'N.accept', because the thread--- was explicitly killed, or because of a synchronous exception thrown by the--- request handler. Typically, you should avoid the last case by catching any--- relevant exceptions in the request handler.+-- incoming connections and executes the given request handler in another+-- thread. If any exception occurs the accepting thread terminates and calls+-- the terminationHandler. Threads spawned for previous accepted connections+-- are not killed.+-- This exception may occur because of a call to 'N.accept', or because the+-- thread was explicitly killed. ----- 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. -forkServer :: N.HostName -- ^ Host- -> N.ServiceName -- ^ Port - -> Int -- ^ Backlog (maximum number of queued connections)- -> Bool -- ^ Set ReuseAddr option?- -> (SomeException -> IO ()) -- ^ Termination handler- -> (N.Socket -> IO ()) -- ^ Request handler - -> IO ThreadId-forkServer host port backlog reuseAddr terminationHandler requestHandler = do +-- The request handler is not responsible for closing the socket. It will be+-- closed once that handler returns. Take care to ensure that the socket is not+-- used after the handler returns, or you will get undefined behavior+-- (the file descriptor may be re-used).+--+-- The return value includes the port was bound to. This is not always the same+-- port as that given in the argument. For example, binding to port 0 actually+-- binds to a random port, selected by the OS.+forkServer :: N.HostName -- ^ Host+ -> N.ServiceName -- ^ Port+ -> Int -- ^ Backlog (maximum number of queued connections)+ -> Bool -- ^ Set ReuseAddr option?+ -> (SomeException -> IO ()) -- ^ Error handler. Called with an+ -- exception raised when+ -- accepting a connection.+ -> (SomeException -> IO ()) -- ^ Termination handler. Called+ -- when the error handler throws+ -- an exception.+ -> (IO () -> (N.Socket, N.SockAddr) -> IO ())+ -- ^ Request handler. Gets an+ -- action which completes when+ -- the socket is closed.+ -> IO (N.ServiceName, ThreadId)+forkServer host port backlog reuseAddr errorHandler 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@@ -83,51 +220,144 @@ bracketOnError (N.socket (N.addrFamily addr) N.Stream N.defaultProtocol) tryCloseSocket $ \sock -> do when reuseAddr $ N.setSocketOption sock N.ReuseAddr 1- N.bindSocket sock (N.addrAddress addr)- N.listen sock backlog + N.bind sock (N.addrAddress addr)+ N.listen sock backlog++ -- Close up and fill the synchonizing MVar.+ let release :: ((N.Socket, N.SockAddr), MVar ()) -> IO ()+ release ((sock, _), socketClosed) =+ N.close sock `finally` putMVar socketClosed ()++ -- Run the request handler.+ let act restore (sock, sockAddr) = do+ socketClosed <- newEmptyMVar+ void $ forkIO $ restore $ do+ requestHandler (readMVar socketClosed) (sock, sockAddr)+ `finally`+ release ((sock, sockAddr), socketClosed)++ let acceptRequest :: IO ()+ acceptRequest = mask $ \restore -> do+ -- Async exceptions are masked so that, if accept does give a+ -- socket, we'll always deliver it to the handler before the+ -- exception is raised.+ -- If it's a Right handler then it will eventually be closed.+ -- If it's a Left handler then we assume the handler itself will+ -- close it.+ (sock, sockAddr) <- N.accept sock+ -- Looks like 'act' will never throw an exception, but to be+ -- safe we'll close the socket if it does.+ let handler :: SomeException -> IO ()+ handler _ = N.close sock+ catch (act restore (sock, sockAddr)) handler+ -- 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+ -- handler is run. 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 -> - catch (unmask (forever $ acceptRequest sock)) $ \ex -> do- tryCloseSocket sock- terminationHandler ex- where- acceptRequest :: N.Socket -> IO ()- 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+ (,) <$> fmap show (N.socketPort sock) <*>+ (mask_ $ forkIOWithUnmask $ \unmask ->+ catch (unmask (forever (catch acceptRequest errorHandler))) $ \ex -> do+ tryCloseSocket sock+ terminationHandler ex) --- | Receive a 32-bit integer-recvInt32 :: Num a => N.Socket -> IO a -recvInt32 sock = decodeInt32 . BS.concat <$> recvExact sock 4 +-- | Read a length and then a payload of that length, subject to a limit+-- on the length.+-- If the length (first 'Word32' received) is greater than the limit then+-- an exception is thrown.+recvWithLength :: Word32 -> N.Socket -> IO [ByteString]+recvWithLength limit sock = do+ len <- recvWord32 sock+ when (len > limit) $+ throwIO (userError "recvWithLength: limit exceeded")+ recvExact sock len --- | Close a socket, ignoring I/O exceptions+-- | Receive a 32-bit unsigned integer+recvWord32 :: N.Socket -> IO Word32+recvWord32 = fmap (decodeWord32 . BS.concat) . flip recvExact 4++-- | Close a socket, ignoring I/O exceptions. tryCloseSocket :: N.Socket -> IO ()-tryCloseSocket sock = void . tryIO $ - N.sClose sock+tryCloseSocket sock = void . tryIO $+ N.close sock +-- | Shutdown socket sends and receives, ignoring I/O exceptions.+tryShutdownSocketBoth :: N.Socket -> IO ()+tryShutdownSocketBoth sock = void . tryIO $+ N.shutdown sock N.ShutdownBoth+ -- | 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 - -> Int32 -- ^ Number of bytes to read- -> IO [ByteString]-recvExact _ len | len < 0 = throwIO (userError "recvExact: Negative length") +recvExact :: N.Socket -- ^ Socket to read from+ -> Word32 -- ^ Number of bytes to read+ -> IO [ByteString] -- ^ Data read recvExact sock len = go [] len where- go :: [ByteString] -> Int32 -> IO [ByteString] - go acc 0 = return (reverse acc) + go :: [ByteString] -> Word32 -> 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))++-- | Get the numeric host, resolved host (via getNameInfo), and port from a+-- SockAddr. The numeric host is first, then resolved host (which may be the+-- same as the numeric host).+-- Will only give 'Just' for IPv4 addresses.+resolveSockAddr :: N.SockAddr -> IO (Maybe (N.HostName, N.HostName, N.ServiceName))+resolveSockAddr sockAddr = case sockAddr of+ N.SockAddrInet port _ -> do+ (mResolvedHost, mResolvedPort) <- N.getNameInfo [] True False sockAddr+ case (mResolvedHost, mResolvedPort) of+ (Just resolvedHost, Nothing) -> do+ (Just numericHost, _) <- N.getNameInfo [N.NI_NUMERICHOST] True False sockAddr+ return $ Just (numericHost, resolvedHost, show port)+ _ -> error $ concat [+ "decodeSockAddr: unexpected resolution "+ , show sockAddr+ , " -> "+ , show mResolvedHost+ , ", "+ , show mResolvedPort+ ]+ _ -> return Nothing++-- | Encode end point address+encodeEndPointAddress :: N.HostName+ -> N.ServiceName+ -> EndPointId+ -> EndPointAddress+encodeEndPointAddress host port ix = EndPointAddress . BSC.pack $+ host ++ ":" ++ port ++ ":" ++ show ix++-- | Decode end point address+decodeEndPointAddress :: EndPointAddress+ -> Maybe (N.HostName, N.ServiceName, EndPointId)+decodeEndPointAddress (EndPointAddress bs) =+ case splitMaxFromEnd (== ':') 2 $ BSC.unpack bs of+ [host, port, endPointIdStr] ->+ case reads endPointIdStr of+ [(endPointId, "")] -> Just (host, port, endPointId)+ _ -> Nothing+ _ ->+ Nothing++-- | @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+ where+ -- go :: [[a]] -> Int -> [a] -> [[a]]+ 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+ else go ((x : acc) : accs) n xs+ go _ _ _ = error "Bug in splitMaxFromEnd"
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
@@ -1,62 +1,76 @@ {-# LANGUAGE RebindableSyntax, TemplateHaskell #-}+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE CPP #-} {-# OPTIONS_GHC -fno-warn-orphans #-} module Main where -import Prelude hiding +import Prelude hiding ( (>>=) , return , fail , (>>)-#if ! MIN_VERSION_base(4,6,0)- , catch-#endif ) import Network.Transport import Network.Transport.TCP ( createTransport , createTransportExposeInternals , TransportInternals(..)- , encodeEndPointAddress+ , TCPParameters(..) , defaultTCPParameters , LightweightConnectionId+ , TCPAddrInfo(..)+ , TCPAddr(..)+ , defaultTCPAddr )-import Data.Int (Int32) import Control.Concurrent (threadDelay, killThread) import Control.Concurrent.MVar ( MVar , newEmptyMVar , putMVar , takeMVar , readMVar- , isEmptyMVar , newMVar , modifyMVar , modifyMVar_ )-import Control.Monad (replicateM, guard, forM_, replicateM_, when)-import Control.Applicative ((<$>))+import Control.Monad (replicateM, guard, replicateM_, when) import Control.Exception (throwIO, try, SomeException)-import Network.Transport.TCP ( ControlHeader(..)- , ConnectionRequestResponse(..)- , socketToEndPoint- )-import Network.Transport.Internal ( encodeInt32- , prependLength+import Network.Transport.TCP ( socketToEndPoint )+import Network.Transport.Internal ( prependLength , tlog , tryIO , void )-import Network.Transport.TCP.Internal (recvInt32, forkServer, recvWithLength)+import Network.Transport.TCP.Internal+ ( ControlHeader(..)+ , encodeControlHeader+ , decodeControlHeader+ , ConnectionRequestResponse(..)+ , encodeConnectionRequestResponse+ , encodeWord32+ , recvWord32+ , forkServer+ , recvWithLength+ , encodeEndPointAddress+ , decodeEndPointAddress+ ) #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+ ( close , Socket , AddrInfo , shutdown , ShutdownCmd(ShutdownSend)+ , SocketType(Stream)+ , AddrInfo(..)+ , getAddrInfo+ , defaultHints+ , defaultProtocol+ , socket+ , connect+ , close ) #ifdef USE_MOCK_NETWORK@@ -65,13 +79,14 @@ import Network.Socket.ByteString (sendMany) #endif +import qualified Data.ByteString as BS (length, concat) import Data.String (fromString) 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 (testTransportWithFilter) 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 +95,7 @@ trace = traceShow instance Traceable N.Socket where- trace = traceShow + trace = traceShow instance Traceable N.AddrInfo where trace = traceShow@@ -90,8 +105,8 @@ -- Test that the server gets a ConnectionClosed message when the client closes -- the socket without sending an explicit control message to the server first-testEarlyDisconnect :: IO N.ServiceName -> IO ()-testEarlyDisconnect nextPort = do+testEarlyDisconnect :: IO ()+testEarlyDisconnect = do clientAddr <- newEmptyMVar serverAddr <- newEmptyMVar serverDone <- newEmptyMVar@@ -105,24 +120,24 @@ server :: MVar EndPointAddress -> MVar EndPointAddress -> MVar () -> IO () server serverAddr clientAddr serverDone = do tlog "Server"- Right transport <- nextPort >>= \port -> createTransport "127.0.0.1" port defaultTCPParameters+ Right transport <- createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters Right endpoint <- newEndPoint transport putMVar serverAddr (address endpoint) theirAddr <- readMVar clientAddr -- 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,66 +146,72 @@ -- 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 - + -- Listen for incoming messages- forkServer "127.0.0.1" clientPort 5 True throwIO $ \sock -> do- -- Initial setup - 0 <- recvInt32 sock :: IO Int- _ <- recvWithLength sock - sendMany sock [encodeInt32 ConnectionRequestAccepted]+ (clientPort, _) <- forkServer "127.0.0.1" "0" 5 True throwIO throwIO $ \socketFree (sock, _) -> do+ -- Initial setup+ 0 <- recvWord32 sock+ _ <- recvWithLength maxBound sock+ sendMany sock [encodeWord32 (encodeConnectionRequestResponse ConnectionRequestAccepted)] - -- Server opens a logical connection - CreatedNewConnection <- toEnum <$> (recvInt32 sock :: IO Int)- 1024 <- recvInt32 sock :: IO LightweightConnectionId+ -- Server opens a logical connection+ Just CreatedNewConnection <- decodeControlHeader <$> recvWord32 sock+ 1024 <- recvWord32 sock :: IO LightweightConnectionId -- Server sends a message- 1024 <- recvInt32 sock :: IO Int- ["ping"] <- recvWithLength sock+ 1024 <- recvWord32 sock+ ["ping"] <- recvWithLength maxBound sock - -- Reply - sendMany sock [encodeInt32 CreatedNewConnection, encodeInt32 (10002 :: Int)]- sendMany sock (encodeInt32 10002 : prependLength ["pong"]) + -- Reply+ sendMany sock [+ encodeWord32 (encodeControlHeader CreatedNewConnection)+ , encodeWord32 10002+ ]+ sendMany sock (encodeWord32 10002 : prependLength ["pong"]) -- Close the socket- N.sClose sock- + N.close sock++ let ourAddress = encodeEndPointAddress "127.0.0.1" clientPort 0+ putMVar clientAddr ourAddress+ -- Connect to the server- Right (sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr True Nothing- + Right (_, sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= \addr -> socketToEndPoint (Just ourAddress) addr True False False Nothing Nothing+ -- Open a new connection- sendMany sock [encodeInt32 CreatedNewConnection, encodeInt32 (10003 :: Int)]- + sendMany sock [+ encodeWord32 (encodeControlHeader CreatedNewConnection)+ , encodeWord32 10003+ ]+ -- Close the socket without closing the connection explicitly- -- The server should receive an error event - N.sClose sock+ -- The server should receive an error event+ N.close sock -- | Test the behaviour of a premature CloseSocket request-testEarlyCloseSocket :: IO N.ServiceName -> IO ()-testEarlyCloseSocket nextPort = do+testEarlyCloseSocket :: IO ()+testEarlyCloseSocket = do clientAddr <- newEmptyMVar serverAddr <- newEmptyMVar serverDone <- newEmptyMVar@@ -204,26 +225,26 @@ server :: MVar EndPointAddress -> MVar EndPointAddress -> MVar () -> IO () server serverAddr clientAddr serverDone = do tlog "Server"- Right transport <- nextPort >>= \port -> createTransport "127.0.0.1" port defaultTCPParameters+ Right transport <- createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters Right endpoint <- newEndPoint transport putMVar serverAddr (address endpoint) theirAddr <- readMVar clientAddr -- 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,111 +255,122 @@ -- 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 - + -- Listen for incoming messages- forkServer "127.0.0.1" clientPort 5 True throwIO $ \sock -> do- -- Initial setup - 0 <- recvInt32 sock :: IO Int- _ <- recvWithLength sock - sendMany sock [encodeInt32 ConnectionRequestAccepted]+ (clientPort, _) <- forkServer "127.0.0.1" "0" 5 True throwIO throwIO $ \socketFree (sock, _) -> do+ -- Initial setup+ 0 <- recvWord32 sock+ _ <- recvWithLength maxBound sock+ sendMany sock [encodeWord32 (encodeConnectionRequestResponse ConnectionRequestAccepted)] - -- Server opens a logical connection - CreatedNewConnection <- toEnum <$> (recvInt32 sock :: IO Int)- 1024 <- recvInt32 sock :: IO LightweightConnectionId + -- Server opens a logical connection+ Just CreatedNewConnection <- decodeControlHeader <$> recvWord32 sock+ 1024 <- recvWord32 sock :: IO LightweightConnectionId -- Server sends a message- 1024 <- recvInt32 sock :: IO Int- ["ping"] <- recvWithLength sock+ 1024 <- recvWord32 sock+ ["ping"] <- recvWithLength maxBound sock - -- Reply - sendMany sock [encodeInt32 CreatedNewConnection, encodeInt32 (10002 :: Int)]- sendMany sock (encodeInt32 (10002 :: Int) : prependLength ["pong"]) + -- Reply+ sendMany sock [+ encodeWord32 (encodeControlHeader CreatedNewConnection)+ , encodeWord32 10002+ ]+ sendMany sock (encodeWord32 10002 : prependLength ["pong"]) -- Send a CloseSocket even though there are still connections *in both -- directions*- sendMany sock [encodeInt32 CloseSocket, encodeInt32 (1024 :: Int)]- N.sClose sock- + sendMany sock [+ encodeWord32 (encodeControlHeader CloseSocket)+ , encodeWord32 1024+ ]+ N.close sock++ let ourAddress = encodeEndPointAddress "127.0.0.1" clientPort 0+ putMVar clientAddr ourAddress+ -- Connect to the server- Right (sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr True Nothing- + Right (_, sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= \addr -> socketToEndPoint (Just ourAddress) addr True False False Nothing Nothing+ -- Open a new connection- sendMany sock [encodeInt32 CreatedNewConnection, encodeInt32 (10003 :: Int)]- - -- Send a CloseSocket without sending a closeconnecton + sendMany sock [+ encodeWord32 (encodeControlHeader CreatedNewConnection)+ , encodeWord32 10003+ ]++ -- 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+ sendMany sock [+ encodeWord32 (encodeControlHeader CloseSocket)+ , encodeWord32 0+ ]+ N.close sock -- | Test the creation of a transport with an invalid address-testInvalidAddress :: IO N.ServiceName -> IO ()-testInvalidAddress nextPort = do- Left _ <- nextPort >>= \port -> createTransport "invalidHostName" port defaultTCPParameters+testInvalidAddress :: IO ()+testInvalidAddress = do+ Left _ <- createTransport (defaultTCPAddr "invalidHostName" "0") defaultTCPParameters return () -- | Test connecting to invalid or non-existing endpoints-testInvalidConnect :: IO N.ServiceName -> IO ()-testInvalidConnect nextPort = do- port <- nextPort- Right transport <- createTransport "127.0.0.1" port defaultTCPParameters+testInvalidConnect :: IO ()+testInvalidConnect = do+ Right transport <- createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters 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 _) <- - connect endpoint (encodeEndPointAddress "127.0.0.1" port 1) ReliableOrdered defaultConnectHints+ Left (TransportError ConnectNotFound _) <-+ connect endpoint (encodeEndPointAddress "127.0.0.1" "0" 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) -testIgnoreCloseSocket :: IO N.ServiceName -> IO ()-testIgnoreCloseSocket nextPort = do+-- receiving an (already underway) CloseSocket request)+testIgnoreCloseSocket :: IO ()+testIgnoreCloseSocket = do serverAddr <- newEmptyMVar clientAddr <- newEmptyMVar clientDone <- newEmptyMVar serverDone <- newEmptyMVar connectionEstablished <- newEmptyMVar- Right transport <- nextPort >>= \port -> createTransport "127.0.0.1" port defaultTCPParameters+ Right transport <- createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters -- Server forkTry $ do@@ -376,33 +408,42 @@ theirAddress <- readMVar serverAddr -- Connect to the server- Right (sock, ConnectionRequestAccepted) <- socketToEndPoint ourAddress theirAddress True Nothing+ Right (_, sock, ConnectionRequestAccepted) <- socketToEndPoint (Just ourAddress) theirAddress True False False Nothing Nothing putMVar connectionEstablished () -- Server connects to us, and then closes the connection- CreatedNewConnection <- toEnum <$> (recvInt32 sock :: IO Int)- 1024 <- recvInt32 sock :: IO LightweightConnectionId+ Just CreatedNewConnection <- decodeControlHeader <$> recvWord32 sock+ 1024 <- recvWord32 sock :: IO LightweightConnectionId - CloseConnection <- toEnum <$> (recvInt32 sock :: IO Int)- 1024 <- recvInt32 sock :: IO LightweightConnectionId+ Just CloseConnection <- decodeControlHeader <$> recvWord32 sock+ 1024 <- recvWord32 sock :: IO LightweightConnectionId -- Server will now send a CloseSocket request as its refcount reached 0 tlog "Waiting for CloseSocket request"- CloseSocket <- toEnum <$> recvInt32 sock- _ <- recvInt32 sock :: IO LightweightConnectionId+ Just CloseSocket <- decodeControlHeader <$> recvWord32 sock+ _ <- recvWord32 sock :: IO LightweightConnectionId -- But we ignore it and request another connection in the other direction tlog "Ignoring it, requesting another connection"- sendMany sock [encodeInt32 CreatedNewConnection, encodeInt32 (1024 :: Int)]+ sendMany sock [+ encodeWord32 (encodeControlHeader CreatedNewConnection)+ , encodeWord32 1024+ ] -- Close it again tlog "Closing connection"- sendMany sock [encodeInt32 CloseConnection, encodeInt32 (1024 :: Int)] + sendMany sock [+ encodeWord32 (encodeControlHeader CloseConnection)+ , encodeWord32 1024+ ] -- And close the connection completely tlog "Closing socket"- sendMany sock [encodeInt32 CloseSocket, encodeInt32 (1024 :: Int)]- N.sClose sock+ sendMany sock [+ encodeWord32 (encodeControlHeader CloseSocket)+ , encodeWord32 1024+ ]+ N.close sock putMVar clientDone () @@ -412,14 +453,14 @@ -- | Like 'testIgnoreSocket', but now the server requests a connection after the -- client closed their connection. In the meantime, the server will have sent a -- CloseSocket request to the client, and must block until the client responds.-testBlockAfterCloseSocket :: IO N.ServiceName -> IO ()-testBlockAfterCloseSocket nextPort = do+testBlockAfterCloseSocket :: IO ()+testBlockAfterCloseSocket = do serverAddr <- newEmptyMVar clientAddr <- newEmptyMVar clientDone <- newEmptyMVar serverDone <- newEmptyMVar connectionEstablished <- newEmptyMVar- Right transport <- nextPort >>= \port -> createTransport "127.0.0.1" port defaultTCPParameters+ Right transport <- createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters -- Server forkTry $ do@@ -453,28 +494,28 @@ theirAddress <- readMVar serverAddr -- Connect to the server- Right (sock, ConnectionRequestAccepted) <- socketToEndPoint ourAddress theirAddress True Nothing+ Right (_, sock, ConnectionRequestAccepted) <- socketToEndPoint (Just ourAddress) theirAddress True False False Nothing Nothing putMVar connectionEstablished () -- Server connects to us, and then closes the connection- CreatedNewConnection <- toEnum <$> (recvInt32 sock :: IO Int)- 1024 <- recvInt32 sock :: IO LightweightConnectionId+ Just CreatedNewConnection <- decodeControlHeader <$> recvWord32 sock+ 1024 <- recvWord32 sock :: IO LightweightConnectionId - CloseConnection <- toEnum <$> (recvInt32 sock :: IO Int)- 1024 <- recvInt32 sock :: IO LightweightConnectionId+ Just CloseConnection <- decodeControlHeader <$> recvWord32 sock+ 1024 <- recvWord32 sock :: IO LightweightConnectionId -- Server will now send a CloseSocket request as its refcount reached 0 tlog "Waiting for CloseSocket request"- CloseSocket <- toEnum <$> recvInt32 sock- _ <- recvInt32 sock :: IO LightweightConnectionId+ Just CloseSocket <- decodeControlHeader <$> recvWord32 sock+ _ <- recvWord32 sock :: IO LightweightConnectionId 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+ recvWord32 sock readMVar unblocked >>= guard putMVar clientDone () @@ -483,45 +524,53 @@ tlog "Client ignores close socket and sends connection request" tlog "This should unblock the server" modifyMVar_ unblocked $ \_ -> return True- sendMany sock [encodeInt32 CreatedNewConnection, encodeInt32 (1024 :: Int)]+ sendMany sock [+ encodeWord32 (encodeControlHeader CreatedNewConnection)+ , encodeWord32 1024+ ] takeMVar clientDone takeMVar serverDone -- | 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 nextPort numThreads = do+testUnnecessaryConnect :: Int -> IO ()+testUnnecessaryConnect numThreads = do clientDone <- newEmptyMVar serverAddr <- newEmptyMVar forkTry $ do- Right transport <- nextPort >>= \port -> createTransport "127.0.0.1" port defaultTCPParameters+ Right transport <- createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters Right endpoint <- newEndPoint transport- putMVar serverAddr (address endpoint)+ -- Since we're lying about the server's address, we have to manually+ -- construct the proper address. If we used its actual address, the clients+ -- would try to resolve "128.0.0.1" and then would fail due to invalid+ -- address.+ Just (_, port, epid) <- return $ decodeEndPointAddress (address endpoint)+ putMVar serverAddr $ encodeEndPointAddress "127.0.0.1" port epid forkTry $ do- -- 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 pick an address < 128.0.0.1 so that this is not rejected purely because of the "crossed" check+ let ourAddress = encodeEndPointAddress "127.0.0.1" "1234" 0 -- 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 (Just ourAddress) addr True False False Nothing 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- Right (sock, ConnectionRequestCrossed) ->- N.sClose sock+ Right (_, sock, ConnectionRequestInvalid) ->+ N.close sock+ Right (_, sock, ConnectionRequestCrossed) ->+ N.close sock Left _ -> return () putMVar done ()@@ -533,18 +582,18 @@ takeMVar clientDone -- | Test that we can create "many" transport instances-testMany :: IO N.ServiceName -> IO ()-testMany nextPort = do- Right masterTransport <- nextPort >>= \port -> createTransport "127.0.0.1" port defaultTCPParameters- Right masterEndPoint <- newEndPoint masterTransport +testMany :: IO ()+testMany = do+ Right masterTransport <- createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters+ Right masterEndPoint <- newEndPoint masterTransport replicateM_ 10 $ do- mTransport <- nextPort >>= \port -> createTransport "127.0.0.1" port defaultTCPParameters+ mTransport <- createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters case mTransport of 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 ->@@ -554,17 +603,25 @@ return () -- | Test what happens when the transport breaks completely-testBreakTransport :: IO N.ServiceName -> IO ()-testBreakTransport nextPort = do- Right (transport, internals) <- nextPort >>= \port -> createTransportExposeInternals "127.0.0.1" port defaultTCPParameters+testBreakTransport :: IO ()+testBreakTransport = do+ Right (transport, internals) <- createTransportExposeInternals (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters Right endpoint <- newEndPoint transport - killThread (transportThread internals) -- Uh oh+ let Just tid = transportThread internals+ killThread tid -- Uh oh - ErrorEvent (TransportError EventTransportFailed _) <- receive endpoint + ErrorEvent (TransportError EventTransportFailed _) <- receive endpoint return () +-- Used in testReconnect to block until a socket is closed. newtype is needed+-- for the Traceable instance.+newtype WaitSocketFree = WaitSocketFree (IO ())++instance Traceable WaitSocketFree where+ trace = const Nothing+ -- | Test that a second call to 'connect' might succeed even if the first -- failed. This is a TCP specific test rather than an endpoint specific test -- because we must manually create the endpoint address to match an endpoint we@@ -572,81 +629,124 @@ -- Then test that we get a connection lost message after the remote endpoint -- suddenly closes the socket, and that a subsequent 'connect' allows us to -- re-establish a connection to the same endpoint-testReconnect :: IO N.ServiceName -> IO ()-testReconnect nextPort = do- serverPort <- nextPort+testReconnect :: IO ()+testReconnect = do serverDone <- newEmptyMVar- firstAttempt <- newEmptyMVar endpointCreated <- newEmptyMVar+ -- The server will put the 'socketFree' IO in here, so that the client can+ -- block until the server has closed the socket.+ socketClosed <- newEmptyMVar - -- Server- forkTry $ do - -- Wait for the client to do its first attempt- readMVar firstAttempt + counter <- newMVar (0 :: Int) - counter <- newMVar (0 :: Int) + -- Server+ (serverPort, _) <- forkServer "127.0.0.1" "0" 5 True throwIO throwIO $ \socketFree (sock, _) -> do+ -- Accept the connection+ Right 0 <- tryIO $ recvWord32 sock+ Right _ <- tryIO $ recvWithLength maxBound sock - forkServer "127.0.0.1" serverPort 5 True throwIO $ \sock -> do- -- Accept the connection - Right 0 <- tryIO $ (recvInt32 sock :: IO Int)- Right _ <- tryIO $ recvWithLength sock - Right () <- tryIO $ sendMany sock [encodeInt32 ConnectionRequestAccepted]+ -- The first time we close the socket before accepting the logical connection+ count <- modifyMVar counter $ \i -> return (i + 1, i) - -- The first time we close the socket before accepting the logical connection- count <- modifyMVar counter $ \i -> return (i + 1, i)+ -- Wait 100ms after the socket closes, to (hopefully) ensure that the client+ -- knows the connection is closed, and sending on that socket will therefore+ -- fail.+ putMVar socketClosed (WaitSocketFree (socketFree >> threadDelay 100000)) - when (count > 0) $ do- -- Client requests a logical connection- Right CreatedNewConnection <- tryIO $ toEnum <$> (recvInt32 sock :: IO Int)- connId <- recvInt32 sock :: IO LightweightConnectionId- return ()+ when (count > 0) $ do+ -- The second, third, and fourth connections are accepted according to the+ -- protocol.+ -- On the second request, the socket then closes.+ Right () <- tryIO $ sendMany sock [+ encodeWord32 (encodeConnectionRequestResponse ConnectionRequestAccepted)+ ]+ -- Client requests a logical connection+ when (count > 1) $ do+ -- On the third and fourth requests, a new logical connection is+ -- accepted.+ -- On the third request the socket then closes.+ Right (Just CreatedNewConnection) <- tryIO $ decodeControlHeader <$> recvWord32 sock+ connId <- recvWord32 sock :: IO LightweightConnectionId - when (count > 1) $ do- -- Client sends a message- Right connId' <- tryIO $ (recvInt32 sock :: IO LightweightConnectionId)- True <- return $ connId == connId' - Right ["ping"] <- tryIO $ recvWithLength sock+ when (count > 2) $ do+ -- On the fourth request, a message is received and then the socket+ -- is closed.+ Right connId' <- tryIO $ (recvWord32 sock :: IO LightweightConnectionId)+ True <- return $ connId == connId'+ Right ["ping"] <- tryIO $ recvWithLength maxBound sock putMVar serverDone ()- - Right () <- tryIO $ N.sClose sock- return () - putMVar endpointCreated ()+ return () + putMVar endpointCreated ()+ -- Client forkTry $ do- Right transport <- nextPort >>= \port -> createTransport "127.0.0.1" port defaultTCPParameters+ Right transport <- createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters Right endpoint <- newEndPoint transport let theirAddr = encodeEndPointAddress "127.0.0.1" serverPort 0 - -- The first attempt will fail because no endpoint is yet set up- Left (TransportError ConnectNotFound _) <- connect endpoint theirAddr ReliableOrdered defaultConnectHints- putMVar firstAttempt ()-- -- 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 ++ -- First attempt: fails because the server closes the socket without+ -- doing the handshake.+ resultConnect <- 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" + Left (TransportError ConnectFailed _) -> return ()+ Left err -> throwIO err+ Right _ -> throwIO $ userError "testConnect: unexpected connect success"+ WaitSocketFree wait <- takeMVar socketClosed+ wait - -- 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"]+ -- Second attempt: server accepts the connection but then closes the socket.+ -- We expect a failed connection if the socket is closed *before*+ -- CreatedNewConnection is sent, or a successful connection such that a+ -- subsequent send will fail in case CreatedNewConnection was sent before+ -- the close.+ resultConnect <- connect endpoint theirAddr ReliableOrdered defaultConnectHints+ -- We must be sure that the socket has closed before trying to send.+ WaitSocketFree wait <- takeMVar socketClosed+ wait+ case resultConnect of+ Left (TransportError ConnectFailed _) -> return ()+ Left err -> throwIO err+ Right c -> do+ ev <- send c ["ping"]+ case ev of+ Left _ -> return ()+ Right _ -> throwIO $ userError "testConnect: unexpected send success"++ -- In any case, since a heavyweight connection was made, we'll get a+ -- connection lost event. ErrorEvent (TransportError (EventConnectionLost _) _) <- receive endpoint + -- Third attempt: server accepts the heavyweight and the lightweight+ -- connection (CreatedNewConnection) but then closes the socket.+ -- The connection must succeed, but sending after the socket is closed+ -- must fail.+ resultConnect <- connect endpoint theirAddr ReliableOrdered defaultConnectHints+ -- Wait until close before trying to send.+ WaitSocketFree wait <- takeMVar socketClosed+ wait+ case resultConnect of+ Left err -> throwIO err+ Right c -> do+ ev <- send c ["ping"]+ case ev of+ Left (TransportError SendFailed _) -> return ()+ Left err -> throwIO err+ Right _ -> throwIO $ userError "testConnect: unexpected send success"++ ErrorEvent (TransportError (EventConnectionLost _) _) <- receive endpoint+ -- But a subsequent call to connect should reestablish the connection Right conn2 <- connect endpoint theirAddr ReliableOrdered defaultConnectHints -- Send should now succeed Right () <- send conn2 ["ping"]++ WaitSocketFree wait <- takeMVar socketClosed+ wait return () takeMVar serverDone@@ -655,35 +755,38 @@ -- 'recv' in 'handleIncomingMessages' will not fail, but a 'send' or 'connect' -- *will* fail. We are testing that error handling everywhere does the right -- thing.-testUnidirectionalError :: IO N.ServiceName -> IO ()-testUnidirectionalError nextPort = do+testUnidirectionalError :: IO ()+testUnidirectionalError = do clientDone <- newEmptyMVar- serverPort <- nextPort serverGotPing <- newEmptyMVar -- Server- forkServer "127.0.0.1" serverPort 5 True throwIO $ \sock -> do+ (serverPort, _) <- forkServer "127.0.0.1" "0" 5 True throwIO throwIO $ \socketFree (sock, _) -> do -- We accept connections, but when an exception occurs we don't do -- anything (in particular, we don't close the socket). This is important -- because when we shutdown one direction of the socket a recv here will -- fail, but we don't want to close that socket at that point (which -- would shutdown the socket in the other direction) void . (try :: IO () -> IO (Either SomeException ())) $ do- 0 <- recvInt32 sock :: IO Int- _ <- recvWithLength sock- () <- sendMany sock [encodeInt32 ConnectionRequestAccepted]+ 0 <- recvWord32 sock+ _ <- recvWithLength maxBound sock+ () <- sendMany sock [encodeWord32 (encodeConnectionRequestResponse ConnectionRequestAccepted)] - CreatedNewConnection <- toEnum <$> (recvInt32 sock :: IO Int)- connId <- recvInt32 sock :: IO LightweightConnectionId- - connId' <- recvInt32 sock :: IO LightweightConnectionId+ Just CreatedNewConnection <- decodeControlHeader <$> recvWord32 sock+ connId <- recvWord32 sock :: IO LightweightConnectionId++ connId' <- recvWord32 sock :: IO LightweightConnectionId True <- return $ connId == connId'- ["ping"] <- recvWithLength sock+ ["ping"] <- recvWithLength maxBound sock putMVar serverGotPing ()- ++ -- Must read the clientDone MVar so that we don't close the socket+ -- (forkServer will close it once this action ends).+ readMVar clientDone+ -- Client forkTry $ do- Right (transport, internals) <- nextPort >>= \port -> createTransportExposeInternals "127.0.0.1" port defaultTCPParameters+ Right (transport, internals) <- createTransportExposeInternals (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters Right endpoint <- newEndPoint transport let theirAddr = encodeEndPointAddress "127.0.0.1" serverPort 0 @@ -693,12 +796,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 +812,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 +823,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@@ -734,11 +837,11 @@ putMVar clientDone () - takeMVar clientDone+ readMVar clientDone -testInvalidCloseConnection :: IO N.ServiceName -> IO ()-testInvalidCloseConnection nextPort = do- Right (transport, internals) <- nextPort >>= \port -> createTransportExposeInternals "127.0.0.1" port defaultTCPParameters+testInvalidCloseConnection :: IO ()+testInvalidCloseConnection = do+ Right (transport, internals) <- createTransportExposeInternals (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters serverAddr <- newEmptyMVar clientDone <- newEmptyMVar serverDone <- newEmptyMVar@@ -754,7 +857,7 @@ -- connection ErrorEvent (TransportError (EventConnectionLost _) _) <- receive endpoint - putMVar serverDone () + putMVar serverDone () -- Client forkTry $ do@@ -766,35 +869,273 @@ 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 - sendMany sock [encodeInt32 CloseConnection, encodeInt32 (12345 :: Int)]+ sock <- socketBetween internals ourAddr theirAddr+ sendMany sock [+ encodeWord32 (encodeControlHeader CloseConnection)+ , encodeWord32 (12345 :: LightweightConnectionId)+ ] putMVar clientDone () mapM_ takeMVar [clientDone, serverDone] +testUseRandomPort :: IO ()+testUseRandomPort = do+ testDone <- newEmptyMVar+ forkTry $ do+ Right transport1 <- createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters+ Right ep1 <- newEndPoint transport1+ -- Same as transport1, but is strict in the port.+ Right transport2 <- createTransport (Addressable (TCPAddrInfo "127.0.0.1" "0" (\(!port) -> ("127.0.0.1", port)))) defaultTCPParameters+ Right ep2 <- newEndPoint transport2+ Right conn1 <- connect ep2 (address ep1) ReliableOrdered defaultConnectHints+ ConnectionOpened _ _ _ <- receive ep1+ putMVar testDone ()+ takeMVar testDone++-- | Verify that if a peer sends an address or data which exceeds the maximum+-- length, that peer's connection will be terminated, but other peers will+-- not be affected.+testMaxLength :: IO ()+testMaxLength = do++ Right serverTransport <- createTransport (defaultTCPAddr "127.0.0.1" "9998") $ defaultTCPParameters {+ -- 17 bytes should fit every valid address at 127.0.0.1.+ -- Port is at most 5 bytes (65536) and id is a base-10 Word32 so+ -- at most 10 bytes. We'll have one client with a 5-byte port to push it+ -- over the chosen limit of 16+ tcpMaxAddressLength = 16+ , tcpMaxReceiveLength = 8+ }+ Right goodClientTransport <- createTransport (defaultTCPAddr "127.0.0.1" "9999") defaultTCPParameters+ Right badClientTransport <- createTransport (defaultTCPAddr "127.0.0.1" "10000") defaultTCPParameters++ serverAddress <- newEmptyMVar+ testDone <- newEmptyMVar+ goodClientConnected <- newEmptyMVar+ goodClientDone <- newEmptyMVar+ badClientDone <- newEmptyMVar++ forkTry $ do+ Right serverEp <- newEndPoint serverTransport+ putMVar serverAddress (address serverEp)+ readMVar badClientDone+ ConnectionOpened _ _ _ <- receive serverEp+ Received _ _ <- receive serverEp+ -- Will lose the connection when the good client sends 9 bytes.+ ErrorEvent (TransportError (EventConnectionLost _) _) <- receive serverEp+ readMVar goodClientDone+ putMVar testDone ()++ forkTry $ do+ Right badClientEp <- newEndPoint badClientTransport+ address <- readMVar serverAddress+ -- Wait until the good client connects, then try to connect. It'll fail,+ -- but the good client should still be OK.+ readMVar goodClientConnected+ Left (TransportError ConnectFailed _)+ <- connect badClientEp address ReliableOrdered defaultConnectHints+ closeEndPoint badClientEp+ putMVar badClientDone ()++ forkTry $ do+ Right goodClientEp <- newEndPoint goodClientTransport+ address <- readMVar serverAddress+ Right conn <- connect goodClientEp address ReliableOrdered defaultConnectHints+ putMVar goodClientConnected ()+ -- Wait until the bad client has tried and failed to connect before+ -- attempting a send, to ensure that its failure did not affect us.+ readMVar badClientDone+ Right () <- send conn ["00000000"]+ -- The send which breaches the limit does not appear to fail, but the+ -- (heavyweight) connection is now severed. We can reliably determine that+ -- by receiving.+ Right () <- send conn ["000000000"]+ ErrorEvent (TransportError (EventConnectionLost _) _) <- receive goodClientEp+ closeEndPoint goodClientEp+ putMVar goodClientDone ()++ readMVar testDone+ closeTransport badClientTransport+ closeTransport goodClientTransport+ closeTransport serverTransport++-- | Ensure that an end point closes up OK even if the peer disobeys the+-- protocol.+testCloseEndPoint :: IO ()+testCloseEndPoint = do++ serverAddress <- newEmptyMVar+ serverFinished <- newEmptyMVar++ -- A server which accepts one connection and then attempts to close the+ -- end point.+ forkTry $ do+ Right transport <- createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters+ Right ep <- newEndPoint transport+ putMVar serverAddress (address ep)+ ConnectionOpened _ _ _ <- receive ep+ Just () <- timeout 5000000 (closeEndPoint ep)+ putMVar serverFinished ()+ return ()++ -- A nefarious client which connects to the server then stops responding.+ forkTry $ do+ Just (hostName, serviceName, endPointId) <- decodeEndPointAddress <$> readMVar serverAddress+ addr:_ <- N.getAddrInfo (Just N.defaultHints) (Just hostName) (Just serviceName)+ sock <- N.socket (N.addrFamily addr) N.Stream N.defaultProtocol+ N.connect sock (N.addrAddress addr)+ let endPointAddress = "127.0.0.1:0:0"+ -- Version 0x00000000 handshake data.+ v0handshake = [+ encodeWord32 endPointId+ , encodeWord32 (fromIntegral (BS.length endPointAddress))+ , endPointAddress+ ]+ -- Version, and total length of the versioned handshake.+ handshake = [+ encodeWord32 0x00000000+ , encodeWord32 (fromIntegral (BS.length (BS.concat v0handshake)))+ ]+ sendMany sock $+ handshake+ ++ v0handshake+ ++ [ -- Create a lightweight connection.+ encodeWord32 (encodeControlHeader CreatedNewConnection)+ , encodeWord32 1024+ ]+ readMVar serverFinished+ N.close sock++ readMVar serverFinished++-- | Ensure that if the peer's claimed host doesn't match its actual host,+-- the connection is rejected (when tcpCheckPeerHost is enabled).+testCheckPeerHostReject :: IO ()+testCheckPeerHostReject = do++ let params = defaultTCPParameters { tcpCheckPeerHost = True }+ Right transport1 <- createTransport (defaultTCPAddr "127.0.0.1" "0") params+ -- This transport claims 127.0.0.2 as its host, but connections from it to+ -- an EndPoint on transport1 will show 127.0.0.1 as the socket's source host.+ Right transport2 <- createTransport (Addressable (TCPAddrInfo "127.0.0.1" "0" ((,) "127.0.0.2"))) defaultTCPParameters++ Right ep1 <- newEndPoint transport1+ Right ep2 <- newEndPoint transport2++ Left err <- connect ep2 (address ep1) ReliableOrdered defaultConnectHints++ TransportError ConnectFailed _ <- return err++ return ()++-- | Ensure that if peer host checking works through name resolution: if the+-- peer claims "localhost", and connects to a transport also on localhost,+-- it should be accepted.+--+-- This test fails on some systems which resolve to+-- "localhost.localdomain" instead of "localhost". Disabling it for+-- now.+testCheckPeerHostResolve :: IO ()+testCheckPeerHostResolve = do++ let+ params = defaultTCPParameters { tcpCheckPeerHost = True }+ -- This test only passes with this choice on some systems+ -- localHostName = "localhost.localdomain"+ localHostName = "localhost"+ Right transport1 <- createTransport (defaultTCPAddr "127.0.0.1" "0") params+ -- EndPoints on this transport have addresses with "localhost" host part.+ Right transport2 <- createTransport (Addressable (TCPAddrInfo "127.0.0.1" "0" ((,) localHostName))) defaultTCPParameters++ Right ep1 <- newEndPoint transport1+ Right ep2 <- newEndPoint transport2++ Right conn <- connect ep2 (address ep1) ReliableOrdered defaultConnectHints++ close conn++ return ()++-- | Test that an unreachable EndPoint can use its own address to connect+-- to itself.+testUnreachableSelfConnect :: IO ()+testUnreachableSelfConnect = do+ Right transport <- createTransport Unaddressable defaultTCPParameters+ Right ep <- newEndPoint transport+ Right conn <- connect ep (address ep) ReliableOrdered defaultConnectHints+ ConnectionOpened connid ReliableOrdered _ <- receive ep+ Right () <- send conn ["ping"]+ Received connid' bytes <- receive ep+ _ <- close conn+ ConnectionClosed connid'' <- receive ep+ closeEndPoint ep+ closeTransport transport++-- | Test that+--+-- 1. Connecting to an unreachable EndPoint's address gives ConnectFailed+-- 2. An unreachable EndPoint can successfully connect to a reachable EndPoint+-- 3. The address given in the ConnectionOpened event at the reachable EndPoint+-- can be used to connect to the unreachable EndPoint, so long as there is+-- at least one lightweight connection open between the two.+testUnreachableConnect :: IO ()+testUnreachableConnect = do+ Right rtransport <- createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters+ Right utransport <- createTransport Unaddressable defaultTCPParameters+ Right rep <- newEndPoint rtransport+ Right uep <- newEndPoint utransport+ -- Reachable endpoint connects to the unreachable endpoint, but it fails.+ -- NB ConnectNotFound isn't the error; that would mean the address makes+ -- sense but the host could not be found.+ Left (TransportError ConnectFailed _) <- connect rep (address uep) ReliableOrdered defaultConnectHints+ -- Unreachable endpoint connects to the reachable endpoint.+ Right conn <- connect uep (address rep) ReliableOrdered defaultConnectHints+ -- Reachable endpoint now has an address at which it can connect to the+ -- unreachable+ ConnectionOpened _ _ addr <- receive rep+ Right conn' <- connect rep addr ReliableOrdered defaultConnectHints+ ConnectionOpened _ _ addr' <- receive uep+ close conn+ ConnectionClosed _ <- receive rep+ close conn'+ ConnectionClosed _ <- receive uep+ closeEndPoint rep+ closeEndPoint uep+ closeTransport rtransport+ closeTransport utransport+ main :: IO () main = do- portMVar <- newEmptyMVar- 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) - , ("Many", testMany nextPort)- , ("BreakTransport", testBreakTransport nextPort)- , ("Reconnect", testReconnect nextPort)- , ("UnidirectionalError", testUnidirectionalError nextPort)- , ("InvalidCloseConnection", testInvalidCloseConnection nextPort)+ tcpResult <- tryIO $ runTests+ [ ("Use random port", testUseRandomPort)+ , ("EarlyDisconnect", testEarlyDisconnect)+ , ("EarlyCloseSocket", testEarlyCloseSocket)+ , ("IgnoreCloseSocket", testIgnoreCloseSocket)+ , ("BlockAfterCloseSocket", testBlockAfterCloseSocket)+ -- , ("UnnecessaryConnect", testUnnecessaryConnect 10) -- flaky: #91+ , ("InvalidAddress", testInvalidAddress)+ , ("InvalidConnect", testInvalidConnect)+ , ("Many", testMany)+ , ("BreakTransport", testBreakTransport)+ , ("Reconnect", testReconnect)+ , ("UnidirectionalError", testUnidirectionalError)+ , ("InvalidCloseConnection", testInvalidCloseConnection)+ , ("MaxLength", testMaxLength)+ , ("CloseEndPoint", testCloseEndPoint)+ , ("CheckPeerHostReject", testCheckPeerHostReject)+ -- , ("CheckPeerHostResolve", testCheckPeerHostResolve) -- flaky+ , ("UnreachableSelfConnect", testUnreachableSelfConnect)+ , ("UnreachableConnect", testUnreachableConnect) ]- -- 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)+ -- Run the generic tests even if the TCP specific tests failed..+ testTransportWithFilter (`notElem` flakies) (either (Left . show) (Right) <$>+ createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters) -- ..but if the generic tests pass, still fail if the specific tests did not case tcpResult of Left err -> throwIO err Right () -> return ()+ where+ flakies =+ [ "ParallelConnects" -- #92+ ]