network-transport-tcp 0.5.1 → 0.8.6
raw patch · 6 files changed
Files
- ChangeLog +56/−0
- Setup.hs +0/−2
- network-transport-tcp.cabal +40/−25
- src/Network/Transport/TCP.hs +2199/−1800
- src/Network/Transport/TCP/Internal.hs +275/−51
- tests/TestTCP.hs +466/−137
ChangeLog view
@@ -1,3 +1,59 @@+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.
− Setup.hs
@@ -1,2 +0,0 @@-import Distribution.Simple-main = defaultMain
network-transport-tcp.cabal view
@@ -1,55 +1,72 @@+cabal-version: 3.0 Name: network-transport-tcp-Version: 0.5.1-Cabal-Version: >=1.10+Version: 0.8.6 Build-Type: Simple-License: BSD3+License: BSD-3-Clause License-file: LICENSE Copyright: Well-Typed LLP, Tweag I/O Limited Author: Duncan Coutts, Nicolas Wu, Edsko de Vries-maintainer: Facundo Domínguez <facundo.dominguez@tweag.io>+maintainer: The Distributed Haskell team Stability: experimental Homepage: http://haskell-distributed.github.com-Bug-Reports: https://cloud-haskell.atlassian.net/browse/NTTCP+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.6.3 GHC==7.8.4 GHC==7.10.3+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-source-files: ChangeLog+extra-doc-files: ChangeLog -Source-Repository head+source-repository head Type: git- Location: https://github.com/haskell-distributed/network-transport-tcp+ Location: https://github.com/haskell-distributed/distributed-process+ 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.4.1.0 && < 0.5,+ 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.6.2 && < 2.7+ 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 Default-Extensions: CPP default-language: Haskell2010 Other-Extensions: RecursiveDo- ghc-options: -Wall -fno-warn-unused-do-bind+ 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.2.1.0 && < 0.3,- network >= 2.3 && < 2.7,- network-transport >= 0.4.1.0 && < 0.5,+ 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@@ -60,13 +77,11 @@ CPP-Options: -DUSE_MOCK_NETWORK Test-Suite TestQC+ import: warnings Type: exitcode-stdio-1.0 Main-Is: TestQC.hs If flag(use-mock-network)- Build-Depends: base >= 4.3 && < 5,- test-framework,- test-framework-quickcheck2,- test-framework-hunit,+ Build-Depends: base >= 4.14 && < 5, QuickCheck, HUnit, network-transport,
src/Network/Transport/TCP.hs view
@@ -10,1806 +10,2205 @@ -- Applications that use the TCP transport should use -- 'Network.Socket.withSocketsDo' in their main function for Windows -- compatibility (see "Network.Socket").--{-# LANGUAGE RecursiveDo #-}-{-# LANGUAGE TupleSections #-}--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, NoDelay, UserTimeout, KeepAlive)- , isSupportedSocketOption- , 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- , threadDelay- , throwTo- )-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, mplus, (<=<))-import Control.Exception- ( IOException- , SomeException- , AsyncException- , handle- , throw- , throwIO- , try- , bracketOnError- , bracket- , fromException- , finally- , catch- , bracket- , mask_- )-import Data.IORef (IORef, newIORef, writeIORef, readIORef, writeIORef)-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.Maybe (isJust)-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.Traversable (traverse)-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).--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- -- | 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 ())- , 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- -- | Message sent to probe a socket- | ProbeSocket- -- | Acknowledgement of the ProbeSocket message- | ProbeSocketAck- 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- -- | Should we set TCP_NODELAY on connection sockets?- -- Defaults to True.- , tcpNoDelay :: Bool- -- | Value of TCP_USER_TIMEOUT in milliseconds- , tcpKeepAlive :: Bool- -- | Should we set TCP_KEEPALIVE on connection sockets?- , 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'.- , transportConnectTimeout :: Maybe Int- }---- | 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- }- 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 transport = TCPTransport { transportState = state- , transportHost = host- , transportPort = port'- , transportParams = params- }- bracketOnError (forkServer- host- port- (tcpBacklog params)- (tcpReuseServerAddr params)- (terminationHandler transport)- (handleConnectionRequest transport))- (\(_port', tid) -> killThread tid)- (\(port'', tid) -> (port'',) <$> mkTransport transport tid)- return result- 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- , tcpNoDelay = False- , tcpKeepAlive = False- , tcpUserTimeout = Nothing- , transportConnectTimeout = Nothing- }------------------------------------------------------------------------------------- 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 $ 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 [encodeInt32 CloseConnection, encodeInt32 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 (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"- 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 = 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- sched theirEndPoint $ do- void $ tryIO $ sendOn vst [ encodeInt32 CloseSocket- , encodeInt32 (vst ^. remoteMaxIncoming)- ]- -- Release probing resources if probing.- forM_ (remoteProbing vst) id- tryCloseSocket (remoteSocket vst)- return closed- RemoteEndPointClosing resolved vst -> do- -- Release probing resources if probing.- forM_ (remoteProbing vst) id- putMVar resolved ()- sched theirEndPoint $ tryCloseSocket (remoteSocket 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 -> N.Socket -> IO ()-handleConnectionRequest transport sock = 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- 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 Nothing-- if not isNew- then do- void $ tryIO $ sendMany sock [encodeInt32 ConnectionRequestCrossed]- probeIfValid theirEndPoint- tryCloseSocket sock- return Nothing- else do- sendLock <- newMVar ()- let vst = ValidRemoteEndPointState- { remoteSocket = sock- , remoteProbing = Nothing- , 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-- 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) [encodeInt32 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 :: 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- Just ProbeSocket -> do- forkIO $ sendMany sock [encodeInt32 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)- $ (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- -- 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' [ 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- -- Release probing resources if probing.- forM_ (remoteProbing vst) id- 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)-- -- 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- 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 vst -> do- -- Release probing resources if probing.- forM_ (remoteProbing vst) id- let code = EventConnectionLost (remoteAddress theirEndPoint)- writeChan ourChannel . 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' ->- 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 = 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-- 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 params (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 [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 :: a -> SomeException -> IO a- absorbAllExceptions a _ex =- return a---- | Set up a remote endpoint-setupRemoteEndPoint :: TCPParameters -> EndPointPair -> Maybe Int- -> IO (Maybe ConnectionRequestResponse)-setupRemoteEndPoint params (ourEndPoint, theirEndPoint) connTimeout = do- result <- socketToEndPoint ourAddress- theirAddress- (tcpReuseClientAddr params)- (tcpNoDelay params)- (tcpKeepAlive params)- (tcpUserTimeout params)- connTimeout- didAccept <- case result of- Right (sock, ConnectionRequestAccepted) -> do- sendLock <- newMVar ()- let vst = ValidRemoteEndPointState- { remoteSocket = sock- , remoteProbing = Nothing- , 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 $ void $ forkIO $- handleIncomingMessages (ourEndPoint, theirEndPoint)- return $ either (const Nothing) (Just . snd) result- 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 seq (foldr seq () msg)- 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, 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- readMVarTimeout mtimer 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 _ ->- 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)-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- -- Release probing resources if probing.- forM_ (remoteProbing vst) id- tryCloseSocket (remoteSocket vst)- let code = EventConnectionLost (remoteAddress theirEndPoint)- err = TransportError code (show ex)- writeChan (localChannel ourEndPoint) $ ErrorEvent err- 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-socketToEndPoint :: 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)- (N.Socket, ConnectionRequestResponse))-socketToEndPoint (EndPointAddress ourAddress) 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- 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)+{-# 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)
src/Network/Transport/TCP/Internal.hs view
@@ -1,17 +1,37 @@+{-# LANGUAGE CPP #-} -- | Utility functions for TCP sockets module Network.Transport.TCP.Internal- ( forkServer+ ( ControlHeader(..)+ , encodeControlHeader+ , decodeControlHeader+ , ConnectionRequestResponse(..)+ , encodeConnectionRequestResponse+ , decodeConnectionRequestResponse+ , forkServer , recvWithLength , recvExact- , recvInt32+ , 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@@ -19,6 +39,7 @@ import qualified Network.Socket as N #endif ( HostName+ , NameInfoFlag(NI_NUMERICHOST) , ServiceName , Socket , SocketType(Stream)@@ -26,15 +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@@ -43,44 +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 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.+-- 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 ()) -- ^ Termination handler- -> (N.Socket -> IO ()) -- ^ Request handler+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 terminationHandler requestHandler = do+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@@ -88,52 +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.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. (,) <$> fmap show (N.socketPort sock) <*> (mask_ $ forkIOWithUnmask $ \unmask ->- catch (unmask (forever $ acceptRequest sock)) $ \ex -> do+ catch (unmask (forever (catch acceptRequest errorHandler))) $ \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+-- | 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 --- | Receive a 32-bit integer-recvInt32 :: Num a => N.Socket -> IO a-recvInt32 sock = decodeInt32 . BS.concat <$> recvExact sock 4+-- | 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+-- | Close a socket, ignoring I/O exceptions. tryCloseSocket :: N.Socket -> IO () tryCloseSocket sock = void . tryIO $- N.sClose sock+ 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 :: [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 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"
tests/TestTCP.hs view
@@ -1,4 +1,6 @@ {-# LANGUAGE RebindableSyntax, TemplateHaskell #-}+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE CPP #-} {-# OPTIONS_GHC -fno-warn-orphans #-} module Main where @@ -7,56 +9,68 @@ , 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 #endif- ( sClose- , ServiceName+ ( close , Socket , AddrInfo , shutdown , ShutdownCmd(ShutdownSend)+ , SocketType(Stream)+ , AddrInfo(..)+ , getAddrInfo+ , defaultHints+ , defaultProtocol+ , socket+ , connect+ , close ) #ifdef USE_MOCK_NETWORK@@ -65,11 +79,12 @@ 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 @@ -105,7 +120,7 @@ server :: MVar EndPointAddress -> MVar EndPointAddress -> MVar () -> IO () server serverAddr clientAddr serverDone = do tlog "Server"- Right transport <- createTransport "127.0.0.1" "0" defaultTCPParameters+ Right transport <- createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters Right endpoint <- newEndPoint transport putMVar serverAddr (address endpoint) theirAddr <- readMVar clientAddr@@ -154,39 +169,45 @@ tlog "Client" -- Listen for incoming messages- (clientPort, _) <- forkServer "127.0.0.1" "0" 5 True throwIO $ \sock -> do+ (clientPort, _) <- forkServer "127.0.0.1" "0" 5 True throwIO throwIO $ \socketFree (sock, _) -> do -- Initial setup- 0 <- recvInt32 sock :: IO Int- _ <- recvWithLength sock- sendMany sock [encodeInt32 ConnectionRequestAccepted]+ 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+ 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"])+ 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 False False Nothing 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+ N.close sock -- | Test the behaviour of a premature CloseSocket request testEarlyCloseSocket :: IO ()@@ -204,7 +225,7 @@ server :: MVar EndPointAddress -> MVar EndPointAddress -> MVar () -> IO () server serverAddr clientAddr serverDone = do tlog "Server"- Right transport <- createTransport "127.0.0.1" "0" defaultTCPParameters+ Right transport <- createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters Right endpoint <- newEndPoint transport putMVar serverAddr (address endpoint) theirAddr <- readMVar clientAddr@@ -260,53 +281,65 @@ tlog "Client" -- Listen for incoming messages- (clientPort, _) <- forkServer "127.0.0.1" "0" 5 True throwIO $ \sock -> do+ (clientPort, _) <- forkServer "127.0.0.1" "0" 5 True throwIO throwIO $ \socketFree (sock, _) -> do -- Initial setup- 0 <- recvInt32 sock :: IO Int- _ <- recvWithLength sock- sendMany sock [encodeInt32 ConnectionRequestAccepted]+ 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+ 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"])+ 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 False False Nothing 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+ ] -- 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 () testInvalidAddress = do- Left _ <- createTransport "invalidHostName" "0" defaultTCPParameters+ Left _ <- createTransport (defaultTCPAddr "invalidHostName" "0") defaultTCPParameters return () -- | Test connecting to invalid or non-existing endpoints testInvalidConnect :: IO () testInvalidConnect = do- Right transport <- createTransport "127.0.0.1" "0" defaultTCPParameters+ Right transport <- createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters Right endpoint <- newEndPoint transport -- Syntax error in the endpoint address@@ -337,7 +370,7 @@ clientDone <- newEmptyMVar serverDone <- newEmptyMVar connectionEstablished <- newEmptyMVar- Right transport <- createTransport "127.0.0.1" "0" defaultTCPParameters+ Right transport <- createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters -- Server forkTry $ do@@ -375,33 +408,42 @@ theirAddress <- readMVar serverAddr -- Connect to the server- Right (sock, ConnectionRequestAccepted) <- socketToEndPoint ourAddress theirAddress True False False Nothing 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 () @@ -418,7 +460,7 @@ clientDone <- newEmptyMVar serverDone <- newEmptyMVar connectionEstablished <- newEmptyMVar- Right transport <- createTransport "127.0.0.1" "0" defaultTCPParameters+ Right transport <- createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters -- Server forkTry $ do@@ -452,20 +494,20 @@ theirAddress <- readMVar serverAddr -- Connect to the server- Right (sock, ConnectionRequestAccepted) <- socketToEndPoint ourAddress theirAddress True False False Nothing 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 @@ -473,7 +515,7 @@ -- 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 () @@ -482,7 +524,10 @@ 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@@ -495,13 +540,18 @@ serverAddr <- newEmptyMVar forkTry $ do- Right transport <- createTransport "127.0.0.1" "0" 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@@ -510,17 +560,17 @@ 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 False False Nothing 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 -- 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 ()@@ -534,11 +584,11 @@ -- | Test that we can create "many" transport instances testMany :: IO () testMany = do- Right masterTransport <- createTransport "127.0.0.1" "0" defaultTCPParameters+ Right masterTransport <- createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters Right masterEndPoint <- newEndPoint masterTransport replicateM_ 10 $ do- mTransport <- createTransport "127.0.0.1" "0" 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)@@ -555,15 +605,23 @@ -- | Test what happens when the transport breaks completely testBreakTransport :: IO () testBreakTransport = do- Right (transport, internals) <- createTransportExposeInternals "127.0.0.1" "0" defaultTCPParameters+ 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 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@@ -575,72 +633,110 @@ testReconnect = do serverDone <- 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 counter <- newMVar (0 :: Int) -- Server- (serverPort, _) <- forkServer "127.0.0.1" "0" 5 True throwIO $ \sock -> do+ (serverPort, _) <- forkServer "127.0.0.1" "0" 5 True throwIO throwIO $ \socketFree (sock, _) -> do -- Accept the connection- Right 0 <- tryIO $ (recvInt32 sock :: IO Int)- Right _ <- tryIO $ recvWithLength sock+ Right 0 <- tryIO $ recvWord32 sock+ Right _ <- tryIO $ recvWithLength maxBound sock -- 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- Right () <- tryIO $ sendMany sock [encodeInt32 ConnectionRequestAccepted]+ -- 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- Right CreatedNewConnection <- tryIO $ toEnum <$> (recvInt32 sock :: IO Int)- connId <- recvInt32 sock :: IO LightweightConnectionId- return ()+ -- 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 > 2) $ do- -- Client sends a message- Right connId' <- tryIO $ (recvInt32 sock :: IO LightweightConnectionId)+ -- 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 sock+ Right ["ping"] <- tryIO $ recvWithLength maxBound sock putMVar serverDone () - Right () <- tryIO $ N.sClose sock return () putMVar endpointCreated () -- Client forkTry $ do- Right transport <- createTransport "127.0.0.1" "0" 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 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 - resultConnect <- timeout 500000 $ connect endpoint theirAddr ReliableOrdered defaultConnectHints+ -- 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- Nothing -> return ()- Just (Left (TransportError ConnectFailed _)) -> return ()- Just (Left err) -> throwIO err- Just (Right c) -> do- ev <- send c ["foo"]+ 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" - -- The third attempt succeeds- Right conn1 <- connect endpoint theirAddr ReliableOrdered defaultConnectHints+ -- In any case, since a heavyweight connection was made, we'll get a+ -- connection lost event.+ ErrorEvent (TransportError (EventConnectionLost _) _) <- receive endpoint - -- But a send will fail because the server has closed the connection again- threadDelay 100000- Left (TransportError SendFailed _) <- send conn1 ["ping"]+ -- 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@@ -648,6 +744,9 @@ -- Send should now succeed Right () <- send conn2 ["ping"]++ WaitSocketFree wait <- takeMVar socketClosed+ wait return () takeMVar serverDone@@ -662,28 +761,32 @@ serverGotPing <- newEmptyMVar -- Server- (serverPort, _) <- forkServer "127.0.0.1" "0" 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+ Just CreatedNewConnection <- decodeControlHeader <$> recvWord32 sock+ connId <- recvWord32 sock :: IO LightweightConnectionId - connId' <- recvInt32 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) <- createTransportExposeInternals "127.0.0.1" "0" 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 @@ -734,11 +837,11 @@ putMVar clientDone () - takeMVar clientDone+ readMVar clientDone testInvalidCloseConnection :: IO () testInvalidCloseConnection = do- Right (transport, internals) <- createTransportExposeInternals "127.0.0.1" "0" defaultTCPParameters+ Right (transport, internals) <- createTransportExposeInternals (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters serverAddr <- newEmptyMVar clientDone <- newEmptyMVar serverDone <- newEmptyMVar@@ -767,7 +870,10 @@ -- 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)]+ sendMany sock [+ encodeWord32 (encodeControlHeader CloseConnection)+ , encodeWord32 (12345 :: LightweightConnectionId)+ ] putMVar clientDone () @@ -777,24 +883,237 @@ testUseRandomPort = do testDone <- newEmptyMVar forkTry $ do- Right transport1 <- createTransport "127.0.0.1" "0" defaultTCPParameters+ Right transport1 <- createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters Right ep1 <- newEndPoint transport1- Right transport2 <- createTransport "127.0.0.1" "0" defaultTCPParameters+ -- 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 tcpResult <- tryIO $ runTests- [ ("Use random port" , testUseRandomPort)+ [ ("Use random port", testUseRandomPort) , ("EarlyDisconnect", testEarlyDisconnect) , ("EarlyCloseSocket", testEarlyCloseSocket) , ("IgnoreCloseSocket", testIgnoreCloseSocket) , ("BlockAfterCloseSocket", testBlockAfterCloseSocket)- , ("UnnecessaryConnect", testUnnecessaryConnect 10)+ -- , ("UnnecessaryConnect", testUnnecessaryConnect 10) -- flaky: #91 , ("InvalidAddress", testInvalidAddress) , ("InvalidConnect", testInvalidConnect) , ("Many", testMany)@@ -802,11 +1121,21 @@ , ("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) <$>- createTransport "127.0.0.1" "0" defaultTCPParameters)+ 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+ ]