network-transport-tcp (empty) → 0.2.0
raw patch · 6 files changed
+2638/−0 lines, 6 filesdep +ansi-terminaldep +basedep +bytestringsetup-changed
Dependencies added: ansi-terminal, base, bytestring, containers, data-accessor, mtl, network, network-transport, random
Files
- LICENSE +31/−0
- Setup.hs +2/−0
- network-transport-tcp.cabal +54/−0
- src/Network/Transport/TCP.hs +1649/−0
- src/Network/Transport/TCP/Internal.hs +116/−0
- tests/TestTCP.hs +786/−0
+ LICENSE view
@@ -0,0 +1,31 @@+Copyright Well-Typed LLP, 2011-2012++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++ * Redistributions of source code must retain the above copyright+ notice, this list of conditions and the following disclaimer.++ * Redistributions in binary form must reproduce the above+ copyright notice, this list of conditions and the following+ disclaimer in the documentation and/or other materials provided+ with the distribution.++ * Neither the name of the owner nor the names of other+ contributors may be used to endorse or promote products derived+ from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.+
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ network-transport-tcp.cabal view
@@ -0,0 +1,54 @@+Name: network-transport-tcp+Version: 0.2.0+Cabal-Version: >=1.8+Build-Type: Simple+License: BSD3 +License-file: LICENSE+Copyright: Well-Typed LLP+Author: Duncan Coutts, Nicolas Wu, Edsko de Vries+Maintainer: edsko@well-typed.com, dcoutts@well-typed.com+Stability: experimental+Homepage: http://github.com/haskell-distributed/distributed-process+Bug-Reports: mailto:edsko@well-typed.com+Synopsis: TCP instantation of Network.Transport+Description: TCP instantation of Network.Transport +Tested-With: GHC==7.0.4 GHC==7.2.2 GHC==7.4.1 GHC==7.4.2+Category: Network ++Source-Repository head+ Type: git+ Location: https://github.com/haskell-distributed/distributed-process+ SubDir: network-transport-tcp++Library+ Build-Depends: base >= 4.3 && < 5,+ network-transport >= 0.2 && < 0.3,+ data-accessor >= 0.2 && < 0.3,+ containers >= 0.4 && < 0.5,+ bytestring >= 0.9 && < 0.10,+ network >= 2.3 && < 2.4+ Exposed-modules: Network.Transport.TCP,+ Network.Transport.TCP.Internal+ ghc-options: -Wall -fno-warn-unused-do-bind+ HS-Source-Dirs: src++Test-Suite TestTCP + Type: exitcode-stdio-1.0+ Main-Is: TestTCP.hs+ Build-Depends: base >= 4.3 && < 5,+ network-transport >= 0.2 && < 0.3,+ data-accessor >= 0.2 && < 0.3,+ containers >= 0.4 && < 0.5,+ bytestring >= 0.9 && < 0.10,+ network >= 2.3 && < 2.4,+ random >= 1.0 && < 1.1,+ ansi-terminal >= 0.5 && < 0.6,+ mtl >= 2.0 && < 2.2+ ghc-options: -Wall -fno-warn-unused-do-bind -threaded -rtsopts -with-rtsopts=-N+ Extensions: ExistentialQuantification,+ FlexibleInstances,+ DeriveDataTypeable,+ RankNTypes,+ OverlappingInstances,+ OverloadedStrings+ HS-Source-Dirs: tests src
+ src/Network/Transport/TCP.hs view
@@ -0,0 +1,1649 @@+-- | TCP implementation of the transport layer. +-- +-- The TCP implementation guarantees that only a single TCP connection (socket)+-- will be used between endpoints, provided that the addresses specified are+-- canonical. If /A/ connects to /B/ and reports its address as+-- @192.168.0.1:8080@ and /B/ subsequently connects tries to connect to /A/ as+-- @client1.local:http-alt@ then the transport layer will not realize that the+-- TCP connection can be reused. +--+-- Applications that use the TCP transport should use+-- 'Network.Socket.withSocketsDo' in their main function for Windows+-- compatibility (see "Network.Socket").+module Network.Transport.TCP ( -- * Main API+ createTransport+ , TCPParameters(..)+ , defaultTCPParameters+ -- * Internals (exposed for unit tests) + , createTransportExposeInternals + , TransportInternals(..)+ , EndPointId+ , encodeEndPointAddress+ , decodeEndPointAddress+ , ControlHeader(..)+ , ConnectionRequestResponse(..)+ , firstNonReservedConnectionId+ , socketToEndPoint + -- * Design notes+ -- $design+ ) where++import Prelude hiding (catch, mapM_)+import Network.Transport+import Network.Transport.TCP.Internal ( forkServer+ , recvWithLength+ , recvInt32+ , tryCloseSocket+ )+import Network.Transport.Internal ( encodeInt32+ , decodeInt32+ , prependLength+ , mapIOException+ , tryIO+ , tryToEnum+ , void+ , timeoutMaybe+ , asyncWhenCancelled+ )+import qualified Network.Socket as N ( HostName+ , ServiceName+ , Socket+ , getAddrInfo+ , socket+ , addrFamily+ , addrAddress+ , SocketType(Stream)+ , defaultProtocol+ , setSocketOption+ , SocketOption(ReuseAddr) + , connect+ , sOMAXCONN+ , AddrInfo+ )+import Network.Socket.ByteString (sendMany)+import Control.Concurrent (forkIO, ThreadId, killThread, myThreadId)+import Control.Concurrent.Chan (Chan, newChan, readChan, writeChan)+import Control.Concurrent.MVar ( MVar+ , newMVar+ , modifyMVar+ , modifyMVar_+ , readMVar+ , takeMVar+ , putMVar+ , newEmptyMVar+ , withMVar+ )+import Control.Category ((>>>))+import Control.Applicative ((<$>))+import Control.Monad (when, unless)+import Control.Exception ( IOException+ , SomeException+ , AsyncException+ , handle+ , throw+ , throwIO+ , try+ , bracketOnError+ , mask+ , onException+ , fromException+ )+import Data.IORef (IORef, newIORef, writeIORef, readIORef)+import Data.ByteString (ByteString)+import qualified Data.ByteString as BS (concat)+import qualified Data.ByteString.Char8 as BSC (pack, unpack, split)+import Data.Int (Int32)+import Data.IntMap (IntMap)+import qualified Data.IntMap as IntMap (empty)+import Data.IntSet (IntSet)+import qualified Data.IntSet as IntSet ( 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, intMapMaybe)+import Data.Foldable (forM_, mapM_)++-- $design +--+-- [Goals]+--+-- The TCP transport maps multiple logical connections between /A/ and /B/ (in+-- either direction) to a single TCP connection:+--+-- > +-------+ +-------++-- > | A |==========================| B |+-- > | |>~~~~~~~~~~~~~~~~~~~~~~~~~|~~~\ |+-- > | Q |>~~~~~~~~~~~~~~~~~~~~~~~~~|~~~Q |+-- > | \~~~|~~~~~~~~~~~~~~~~~~~~~~~~~<| |+-- > | |==========================| |+-- > +-------+ +-------++-- +-- Ignoring the complications detailed below, the TCP connection is set up is+-- when the first lightweight connection is created (in either direction), and+-- torn down when the last lightweight connection (in either direction) is+-- closed.+--+-- [Connecting]+--+-- Let /A/, /B/ be two endpoints without any connections. When /A/ wants to+-- connect to /B/, it locally records that it is trying to connect to /B/ and+-- sends a request to /B/. As part of the request /A/ sends its own endpoint+-- address to /B/ (so that /B/ can reuse the connection in the other direction).+--+-- When /B/ receives the connection request it first checks if it did not+-- already initiate a connection request to /A/. If not it will acknowledge the+-- connection request by sending 'ConnectionRequestAccepted' to /A/ and record+-- that it has a TCP connection to /A/.+--+-- The tricky case arises when /A/ sends a connection request to /B/ and /B/+-- finds that it had already sent a connection request to /A/. In this case /B/+-- will accept the connection request from /A/ if /A/s endpoint address is+-- smaller (lexicographically) than /B/s, and reject it otherwise. If it rejects+-- it, it sends a 'ConnectionRequestCrossed' message to /A/. (The+-- lexicographical ordering is an arbitrary but convenient way to break the+-- tie.)+--+-- When it receives a 'ConnectionRequestCrossed' message the /A/ thread that+-- initiated the request just needs to wait until the /A/ thread that is dealing+-- with /B/'s connection request completes.+--+-- [Disconnecting]+-- +-- The TCP connection is created as soon as the first logical connection from+-- /A/ to /B/ (or /B/ to /A/) is established. At this point a thread (@#@) is+-- spawned that listens for incoming connections from /B/:+--+-- > +-------+ +-------++-- > | A |==========================| B |+-- > | |>~~~~~~~~~~~~~~~~~~~~~~~~~|~~~\ |+-- > | | | Q |+-- > | #| | |+-- > | |==========================| |+-- > +-------+ +-------++--+-- The question is when the TCP connection can be closed again. Conceptually,+-- we want to do reference counting: when there are no logical connections left+-- between /A/ and /B/ we want to close the socket (possibly after some+-- timeout). +--+-- However, /A/ and /B/ need to agree that the refcount has reached zero. It+-- might happen that /B/ sends a connection request over the existing socket at+-- the same time that /A/ closes its logical connection to /B/ and closes the+-- socket. This will cause a failure in /B/ (which will have to retry) which is+-- not caused by a network failure, which is unfortunate. (Note that the+-- connection request from /B/ might succeed even if /A/ closes the socket.) +--+-- Instead, when /A/ is ready to close the socket it sends a 'CloseSocket'+-- request to /B/ and records that its connection to /B/ is closing. If /A/+-- receives a new connection request from /B/ after having sent the+-- 'CloseSocket' request it simply forgets that it sent a 'CloseSocket' request+-- and increments the reference count of the connection again.+-- +-- When /B/ receives a 'CloseSocket' message and it too is ready to close the+-- connection, it will respond with a reciprocal 'CloseSocket' request to /A/+-- and then actually close the socket. /A/ meanwhile will not send any more+-- requests to /B/ after having sent a 'CloseSocket' request, and will actually+-- close its end of the socket only when receiving the 'CloseSocket' message+-- from /B/. (Since /A/ recorded that its connection to /B/ is in closing state+-- after sending a 'CloseSocket' request to /B/, it knows not to reciprocate /B/+-- reciprocal 'CloseSocket' message.)+-- +-- If there is a concurrent thread in /A/ waiting to connect to /B/ after /A/+-- has sent a 'CloseSocket' request then this thread will block until /A/ knows+-- whether to reuse the old socket (if /B/ sends a new connection request+-- instead of acknowledging the 'CloseSocket') or to set up a new socket. ++--------------------------------------------------------------------------------+-- Internal datatypes --+--------------------------------------------------------------------------------++-- We use underscores for fields that we might update (using accessors)+--+-- All data types follow the same structure:+-- +-- * A top-level data type describing static properties (TCPTransport,+-- LocalEndPoint, RemoteEndPoint)+-- * The 'static' properties include an MVar containing a data structure for+-- the dynamic properties (TransportState, LocalEndPointState,+-- RemoteEndPointState). The state could be invalid/valid/closed,/etc. +-- * For the case of "valid" we use third data structure to give more details+-- about the state (ValidTransportState, ValidLocalEndPointState,+-- ValidRemoteEndPointState).++data TCPTransport = TCPTransport + { transportHost :: N.HostName+ , transportPort :: N.ServiceName+ , transportState :: MVar TransportState + , transportParams :: TCPParameters+ }++data TransportState = + TransportValid ValidTransportState+ | TransportClosed++data ValidTransportState = ValidTransportState + { _localEndPoints :: Map EndPointAddress LocalEndPoint + , _nextEndPointId :: EndPointId + }++data LocalEndPoint = LocalEndPoint+ { localAddress :: EndPointAddress+ , localChannel :: Chan Event+ , localState :: MVar LocalEndPointState + }++data LocalEndPointState = + LocalEndPointValid ValidLocalEndPointState+ | LocalEndPointClosed++data ValidLocalEndPointState = ValidLocalEndPointState + { _nextConnectionId :: !ConnectionId + , _localConnections :: Map EndPointAddress RemoteEndPoint + , _nextRemoteId :: !Int+ }++-- 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+-- requestConnectionTo (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+-- requestConnectionTo 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 :: Int+ }++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 ()) 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++data ValidRemoteEndPointState = ValidRemoteEndPointState + { _remoteOutgoing :: !Int+ , _remoteIncoming :: IntSet+ , remoteSocket :: N.Socket+ , sendOn :: [ByteString] -> IO ()+ , _pendingCtrlRequests :: IntMap (MVar (Either IOException [ByteString]))+ , _nextCtrlRequestId :: !ControlRequestId + }++-- | Local identifier for an endpoint within this transport+type EndPointId = Int32++-- | Control request ID+-- +-- Control requests are asynchronous; the request ID makes it possible to match+-- requests and replies+type ControlRequestId = Int32++-- | Pair of local and a remote endpoint (for conciseness in signatures)+type EndPointPair = (LocalEndPoint, RemoteEndPoint)++-- | Control headers +data ControlHeader = + -- | Request a new connection ID from the remote endpoint+ RequestConnectionId + -- | Tell the remote endpoint we will no longer be using a connection+ | CloseConnection + -- | Respond to a control request /from/ the remote endpoint+ | ControlResponse + -- | Request to close the connection (see module description)+ | CloseSocket + deriving (Enum, Bounded, Show)++-- | Response sent by /B/ to /A/ when /A/ tries to connect+data ConnectionRequestResponse =+ -- | /B/ accepts the connection+ ConnectionRequestAccepted + -- | /A/ requested an invalid endpoint+ | ConnectionRequestInvalid + -- | /A/s request crossed with a request from /B/ (see protocols)+ | ConnectionRequestCrossed + deriving (Enum, Bounded, Show)++-- | Parameters for setting up the TCP transport+data TCPParameters = TCPParameters {+ -- | Backlog for 'listen'.+ -- Defaults to SOMAXCONN.+ tcpBacklog :: Int+ -- | Should we set SO_REUSEADDR on the server socket? + -- Defaults to True.+ , tcpReuseServerAddr :: Bool+ -- | Should we set SO_REUSEADDR on client sockets?+ -- Defaults to True.+ , tcpReuseClientAddr :: Bool+ }++-- | Internal functionality we expose for unit testing+data TransportInternals = TransportInternals + { -- | The ID of the thread that listens for new incoming connections+ transportThread :: ThreadId+ -- | Find the socket between a local and a remote endpoint+ , socketBetween :: EndPointAddress + -> EndPointAddress + -> IO N.Socket + }++--------------------------------------------------------------------------------+-- Top-level functionality --+--------------------------------------------------------------------------------++-- | Create a TCP transport+createTransport :: N.HostName + -> N.ServiceName + -> TCPParameters+ -> IO (Either IOException Transport)+createTransport host port params = + either Left (Right . fst) <$> createTransportExposeInternals host port params++-- | You should probably not use this function (used for unit testing only)+createTransportExposeInternals + :: N.HostName + -> N.ServiceName + -> TCPParameters+ -> IO (Either IOException (Transport, TransportInternals)) +createTransportExposeInternals host port params = do + state <- newMVar . TransportValid $ ValidTransportState + { _localEndPoints = Map.empty + , _nextEndPointId = 0 + }+ let transport = TCPTransport { transportState = state+ , transportHost = host+ , transportPort = port+ , transportParams = params+ }+ tryIO $ bracketOnError (forkServer + host + port + (tcpBacklog params) + (tcpReuseServerAddr params)+ (terminationHandler transport) + (handleConnectionRequest transport))+ killThread+ (mkTransport transport)+ where+ mkTransport :: TCPTransport + -> ThreadId + -> IO (Transport, TransportInternals)+ mkTransport transport tid = return+ ( Transport + { newEndPoint = apiNewEndPoint transport + , closeTransport = let evs = [ EndPointClosed+ , throw $ userError "Transport closed"+ ] in+ apiCloseTransport transport (Just tid) evs + } + , TransportInternals + { transportThread = tid+ , socketBetween = internalSocketBetween transport+ }+ )++ terminationHandler :: TCPTransport -> SomeException -> IO ()+ terminationHandler transport ex = do + let evs = [ ErrorEvent (TransportError EventTransportFailed (show ex))+ , throw $ userError "Transport closed" + ]+ apiCloseTransport transport Nothing evs ++-- | Default TCP parameters+defaultTCPParameters :: TCPParameters+defaultTCPParameters = TCPParameters {+ tcpBacklog = N.sOMAXCONN+ , tcpReuseServerAddr = True+ , tcpReuseClientAddr = True+ }++--------------------------------------------------------------------------------+-- API functions --+--------------------------------------------------------------------------------++-- | Close the transport+apiCloseTransport :: TCPTransport -> Maybe ThreadId -> [Event] -> IO ()+apiCloseTransport transport mTransportThread evs = + asyncWhenCancelled return $ do+ mTSt <- modifyMVar (transportState transport) $ \st -> case st of+ TransportValid vst -> return (TransportClosed, Just vst)+ TransportClosed -> return (TransportClosed, Nothing)+ forM_ mTSt $ mapM_ (apiCloseEndPoint transport evs) . (^. localEndPoints) + -- This will invoke the termination handler, which in turn will call+ -- apiCloseTransport again, but then the transport will already be closed+ -- and we won't be passed a transport thread, so we terminate immmediate+ forM_ mTransportThread killThread + +-- | Create a new endpoint +apiNewEndPoint :: TCPTransport + -> IO (Either (TransportError NewEndPointErrorCode) EndPoint)+apiNewEndPoint transport = + try . asyncWhenCancelled closeEndPoint $ do+ ourEndPoint <- createLocalEndPoint transport+ return EndPoint + { receive = readChan (localChannel ourEndPoint)+ , address = localAddress ourEndPoint+ , connect = apiConnect (transportParams transport) ourEndPoint + , closeEndPoint = let evs = [ EndPointClosed+ , throw $ userError "Endpoint closed" + ] in+ apiCloseEndPoint transport evs ourEndPoint+ , newMulticastGroup = return . Left $ newMulticastGroupError + , resolveMulticastGroup = return . Left . const resolveMulticastGroupError+ }+ where+ newMulticastGroupError = + TransportError NewMulticastGroupUnsupported "Multicast not supported" + resolveMulticastGroupError = + TransportError ResolveMulticastGroupUnsupported "Multicast not supported" ++-- | Connnect to an endpoint+apiConnect :: TCPParameters -- ^ Parameters+ -> LocalEndPoint -- ^ Local end point+ -> EndPointAddress -- ^ Remote address+ -> Reliability -- ^ Reliability (ignored)+ -> ConnectHints -- ^ Hints+ -> IO (Either (TransportError ConnectErrorCode) Connection)+apiConnect params ourEndPoint theirAddress _reliability hints =+ try . asyncWhenCancelled close $ + if localAddress ourEndPoint == theirAddress + then connectToSelf ourEndPoint + else do+ resetIfBroken ourEndPoint theirAddress+ (theirEndPoint, connId) <- + requestConnectionTo 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 -> ConnectionId -> IORef Bool -> IO ()+apiClose (ourEndPoint, theirEndPoint) connId connAlive = + void . tryIO . asyncWhenCancelled return $ do + modifyRemoteState_ (ourEndPoint, theirEndPoint) remoteStateIdentity + { caseValid = \vst -> do+ alive <- readIORef connAlive+ if alive + then do+ writeIORef connAlive False+ sendOn vst [encodeInt32 CloseConnection, encodeInt32 connId] + return ( RemoteEndPointValid + . (remoteOutgoing ^: (\x -> x - 1)) + $ vst+ )+ else+ return (RemoteEndPointValid vst)+ }+ closeIfUnused (ourEndPoint, theirEndPoint)++-- | Send data across a connection+apiSend :: EndPointPair -- ^ Local and remote endpoint + -> ConnectionId -- ^ Connection ID (supplied by remote endpoint)+ -> 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 $ + withRemoteState (ourEndPoint, theirEndPoint) RemoteStatePatternMatch+ { caseInvalid = \_ -> + relyViolation (ourEndPoint, theirEndPoint) "apiSend"+ , caseInit = \_ _ ->+ relyViolation (ourEndPoint, theirEndPoint) "apiSend"+ , caseValid = \vst -> do+ alive <- readIORef connAlive+ if alive + then sendOn vst (encodeInt32 connId : prependLength payload)+ else throwIO $ TransportError SendClosed "Connection closed"+ , caseClosing = \_ _ -> do+ alive <- readIORef connAlive+ if alive + then relyViolation (ourEndPoint, theirEndPoint) "apiSend" + else throwIO $ TransportError SendClosed "Connection closed"+ , caseClosed = do+ alive <- readIORef connAlive+ if alive + then relyViolation (ourEndPoint, theirEndPoint) "apiSend" + else throwIO $ TransportError SendClosed "Connection closed"+ , caseFailed = \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 = 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 conn -> do + tryIO $ sendOn conn [encodeInt32 CloseSocket]+ tryCloseSocket (remoteSocket conn)+ return closed + RemoteEndPointClosing resolved conn -> do + putMVar resolved ()+ tryCloseSocket (remoteSocket conn)+ return closed + RemoteEndPointClosed ->+ return st+ RemoteEndPointFailed err ->+ return $ RemoteEndPointFailed err ++--------------------------------------------------------------------------------+-- As soon as a remote connection fails, we want to put notify our endpoint --+-- and put it into a closed state. Since this may happen in many places, we --+-- provide some abstractions. --+--------------------------------------------------------------------------------++data RemoteStatePatternMatch a = RemoteStatePatternMatch + { caseInvalid :: TransportError ConnectErrorCode -> IO a+ , caseInit :: MVar () -> RequestedBy -> IO a+ , caseValid :: ValidRemoteEndPointState -> IO a+ , caseClosing :: MVar () -> ValidRemoteEndPointState -> IO a+ , caseClosed :: IO a+ , caseFailed :: IOException -> IO a+ }++remoteStateIdentity :: RemoteStatePatternMatch RemoteState+remoteStateIdentity =+ RemoteStatePatternMatch + { caseInvalid = return . RemoteEndPointInvalid+ , caseInit = (return .) . RemoteEndPointInit+ , caseValid = return . RemoteEndPointValid+ , caseClosing = (return .) . RemoteEndPointClosing + , caseClosed = return RemoteEndPointClosed+ , caseFailed = return . RemoteEndPointFailed+ }++-- | Like modifyMVar, but if an I/O exception occurs don't restore the remote+-- endpoint to its original value but close it instead +modifyRemoteState :: EndPointPair + -> RemoteStatePatternMatch (RemoteState, a) + -> IO a+modifyRemoteState (ourEndPoint, theirEndPoint) match = + mask $ \restore -> do+ st <- takeMVar theirState+ case st of+ RemoteEndPointValid vst -> do+ mResult <- try $ restore (caseValid match vst) + case mResult of+ Right (st', a) -> do+ putMVar theirState st'+ return a+ Left ex -> do+ case fromException ex of+ Just ioEx -> handleIOException ioEx vst + Nothing -> putMVar theirState st + throwIO ex+ -- The other cases are less interesting, because unless the endpoint is+ -- in Valid state we're not supposed to do any IO on it+ RemoteEndPointInit resolved origin -> do+ (st', a) <- onException (restore $ caseInit match resolved origin)+ (putMVar theirState st)+ putMVar theirState st'+ return a+ RemoteEndPointClosing resolved vst -> do + (st', a) <- onException (restore $ caseClosing match resolved vst)+ (putMVar theirState st)+ putMVar theirState st'+ return a+ RemoteEndPointInvalid err -> do+ (st', a) <- onException (restore $ caseInvalid match err)+ (putMVar theirState st)+ putMVar theirState st'+ return a+ RemoteEndPointClosed -> do+ (st', a) <- onException (restore $ caseClosed match)+ (putMVar theirState st)+ putMVar theirState st'+ return a+ RemoteEndPointFailed err -> do+ (st', a) <- onException (restore $ caseFailed match err)+ (putMVar theirState st)+ putMVar theirState st'+ return a+ where+ theirState :: MVar RemoteState+ theirState = remoteState theirEndPoint++ handleIOException :: IOException -> ValidRemoteEndPointState -> IO () + handleIOException ex vst = do+ tryCloseSocket (remoteSocket vst)+ putMVar theirState (RemoteEndPointFailed ex)+ let incoming = IntSet.elems $ vst ^. remoteIncoming+ code = EventConnectionLost (Just $ remoteAddress theirEndPoint) incoming+ err = TransportError code (show ex)+ writeChan (localChannel ourEndPoint) $ ErrorEvent err ++-- | Like 'modifyRemoteState' but without a return value+modifyRemoteState_ :: EndPointPair+ -> RemoteStatePatternMatch RemoteState + -> IO ()+modifyRemoteState_ (ourEndPoint, theirEndPoint) match =+ modifyRemoteState (ourEndPoint, theirEndPoint) + RemoteStatePatternMatch + { caseInvalid = u . caseInvalid match+ , caseInit = \resolved origin -> u $ caseInit match resolved origin+ , caseValid = u . caseValid match+ , caseClosing = \resolved vst -> u $ caseClosing match resolved vst+ , caseClosed = u $ caseClosed match+ , caseFailed = u . caseFailed match+ }+ where+ u :: IO a -> IO (a, ())+ u p = p >>= \a -> return (a, ())++-- | Like 'modifyRemoteState' but without the ability to change the state+withRemoteState :: EndPointPair+ -> RemoteStatePatternMatch a+ -> IO a +withRemoteState (ourEndPoint, theirEndPoint) match =+ modifyRemoteState (ourEndPoint, theirEndPoint)+ RemoteStatePatternMatch + { caseInvalid = \err -> do+ a <- caseInvalid match err+ return (RemoteEndPointInvalid err, a)+ , caseInit = \resolved origin -> do+ a <- caseInit match resolved origin+ return (RemoteEndPointInit resolved origin, a)+ , caseValid = \vst -> do+ a <- caseValid match vst+ return (RemoteEndPointValid vst, a)+ , caseClosing = \resolved vst -> do + a <- caseClosing match resolved vst+ return (RemoteEndPointClosing resolved vst, a)+ , caseClosed = do + a <- caseClosed match+ return (RemoteEndPointClosed, a)+ , caseFailed = \err -> do + a <- caseFailed match err+ return (RemoteEndPointFailed err, a)+ }++--------------------------------------------------------------------------------+-- Incoming requests --+--------------------------------------------------------------------------------++-- | Handle a connection request (that is, a remote endpoint that is trying to+-- establish a TCP connection with us)+--+-- 'handleConnectionRequest' runs in the context of the transport thread, which+-- can be killed asynchronously by 'closeTransport'. We fork a separate thread+-- as soon as we have located the lcoal endpoint that the remote endpoint is+-- interested in. We cannot fork any sooner because then we have no way of+-- storing the thread ID and hence no way of killing the thread when we take+-- the transport down. We must be careful to close the socket when a (possibly+-- asynchronous, ThreadKilled) exception occurs. (If an exception escapes from+-- handleConnectionRequest the transport will be shut down.)+handleConnectionRequest :: TCPTransport -> N.Socket -> IO () +handleConnectionRequest transport sock = handle handleException $ do + ourEndPointId <- recvInt32 sock+ theirAddress <- EndPointAddress . BS.concat <$> recvWithLength sock + let ourAddress = encodeEndPointAddress (transportHost transport) + (transportPort transport)+ ourEndPointId+ ourEndPoint <- withMVar (transportState transport) $ \st -> case st of+ TransportValid vst ->+ case vst ^. localEndPointAt ourAddress of+ Nothing -> do+ sendMany sock [encodeInt32 ConnectionRequestInvalid]+ throwIO $ userError "handleConnectionRequest: Invalid endpoint"+ Just ourEndPoint ->+ return ourEndPoint+ TransportClosed -> + throwIO $ userError "Transport closed"+ void . forkIO $ go ourEndPoint theirAddress+ where+ go :: LocalEndPoint -> EndPointAddress -> IO ()+ go ourEndPoint theirAddress = do + -- This runs in a thread that will never be killed+ mEndPoint <- handle ((>> return Nothing) . handleException) $ do + resetIfBroken ourEndPoint theirAddress+ (theirEndPoint, isNew) <- + findRemoteEndPoint ourEndPoint theirAddress RequestedByThem+ + if not isNew + then do+ tryIO $ sendMany sock [encodeInt32 ConnectionRequestCrossed]+ tryCloseSocket sock+ return Nothing + else do+ let vst = ValidRemoteEndPointState + { remoteSocket = sock+ , _remoteOutgoing = 0+ , _remoteIncoming = IntSet.empty+ , sendOn = sendMany sock+ , _pendingCtrlRequests = IntMap.empty+ , _nextCtrlRequestId = 0+ }+ sendMany sock [encodeInt32 ConnectionRequestAccepted]+ resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst)+ return (Just theirEndPoint)+ -- If we left the scope of the exception handler with a return value of+ -- Nothing then the socket is already closed; otherwise, the socket has+ -- been recorded as part of the remote endpoint. Either way, we no longer+ -- have to worry about closing the socket on receiving an asynchronous+ -- exception from this point forward. + forM_ mEndPoint $ handleIncomingMessages . (,) ourEndPoint++ handleException :: SomeException -> IO ()+ handleException ex = do + tryCloseSocket sock + rethrowIfAsync (fromException ex)++ rethrowIfAsync :: Maybe AsyncException -> IO () + rethrowIfAsync = mapM_ throwIO ++-- | Handle requests from a remote endpoint.+-- +-- Returns only if the remote party closes the socket or if an error occurs.+-- This runs in a thread that will never be killed.+handleIncomingMessages :: EndPointPair -> IO () +handleIncomingMessages (ourEndPoint, theirEndPoint) = do+ mSock <- withMVar theirState $ \st ->+ case st of+ RemoteEndPointInvalid _ ->+ relyViolation (ourEndPoint, theirEndPoint) + "handleIncomingMessages (invalid)" + RemoteEndPointInit _ _ ->+ relyViolation (ourEndPoint, theirEndPoint) + "handleIncomingMessages (init)"+ RemoteEndPointValid ep ->+ return . Just $ remoteSocket ep+ RemoteEndPointClosing _ ep ->+ return . Just $ remoteSocket ep+ RemoteEndPointClosed ->+ return Nothing+ RemoteEndPointFailed _ -> + return Nothing+ + forM_ mSock $ \sock -> + tryIO (go sock) >>= either (prematureExit sock) return+ where+ -- Dispatch + --+ -- If a recv throws an exception this will be caught top-level and+ -- 'prematureExit' will be invoked. The same will happen if the remote+ -- endpoint is put into a Closed (or Closing) state by a concurrent thread+ -- (because a 'send' failed) -- the individual handlers below will throw a+ -- user exception which is then caught and handled the same way as an+ -- exception thrown by 'recv'.+ go :: N.Socket -> IO ()+ go sock = do+ connId <- recvInt32 sock + if connId >= firstNonReservedConnectionId + then do+ readMessage sock connId+ go sock+ else + case tryToEnum (fromIntegral connId) of+ Just RequestConnectionId -> do+ recvInt32 sock >>= createNewConnection + go sock + Just ControlResponse -> do + recvInt32 sock >>= readControlResponse sock + go sock+ Just CloseConnection -> do+ recvInt32 sock >>= closeConnection + go sock+ Just CloseSocket -> do + didClose <- closeSocket sock + unless didClose $ go sock+ Nothing ->+ throwIO $ userError "Invalid control request"+ + -- Create a new connection+ createNewConnection :: ControlRequestId -> IO () + createNewConnection reqId = do + newId <- getNextConnectionId ourEndPoint+ 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 ^: IntSet.insert newId $ 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 with the request to create 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 ^= IntSet.singleton newId $ vst)+ RemoteEndPointFailed err -> + throwIO err+ RemoteEndPointClosed ->+ relyViolation (ourEndPoint, theirEndPoint) + "createNewConnection (closed)"+ sendOn vst ( encodeInt32 ControlResponse + : encodeInt32 reqId + : prependLength [encodeInt32 newId] + )+ return (RemoteEndPointValid vst)+ writeChan ourChannel (ConnectionOpened newId ReliableOrdered theirAddr) ++ -- Read a control response + readControlResponse :: N.Socket -> ControlRequestId -> IO () + readControlResponse sock reqId = do+ response <- recvWithLength sock+ mmvar <- modifyMVar theirState $ \st -> case st of+ RemoteEndPointInvalid _ ->+ relyViolation (ourEndPoint, theirEndPoint)+ "readControlResponse (invalid)"+ RemoteEndPointInit _ _ ->+ relyViolation (ourEndPoint, theirEndPoint)+ "readControlResponse (init)"+ RemoteEndPointValid vst ->+ return ( RemoteEndPointValid + . (pendingCtrlRequestsAt reqId ^= Nothing) + $ vst+ , vst ^. pendingCtrlRequestsAt reqId+ )+ RemoteEndPointClosing _ _ ->+ throwIO $ userError "Invalid control response"+ RemoteEndPointFailed err ->+ throwIO err+ RemoteEndPointClosed ->+ relyViolation (ourEndPoint, theirEndPoint) + "readControlResponse (closed)"+ case mmvar of+ Nothing -> + throwIO $ userError "Invalid request ID"+ Just mvar -> + putMVar mvar (Right response)++ -- 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 :: ConnectionId -> IO () + closeConnection cid = do+ modifyMVar_ theirState $ \st -> case st of+ RemoteEndPointInvalid _ ->+ relyViolation (ourEndPoint, theirEndPoint) "closeConnection (invalid)"+ RemoteEndPointInit _ _ ->+ relyViolation (ourEndPoint, theirEndPoint) "closeConnection (init)"+ RemoteEndPointValid vst -> do + unless (IntSet.member cid (vst ^. remoteIncoming)) $ + throwIO $ userError "Invalid CloseConnection"+ return ( RemoteEndPointValid + . (remoteIncoming ^: IntSet.delete cid) + $ 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 cid)+ closeIfUnused (ourEndPoint, theirEndPoint)++ -- Close the socket (if we don't have any outgoing connections)+ closeSocket :: N.Socket -> IO Bool + closeSocket sock =+ modifyMVar theirState $ \st ->+ 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_ (IntSet.elems $ vst ^. remoteIncoming) $ + writeChan ourChannel . ConnectionClosed + let vst' = remoteIncoming ^= IntSet.empty $ vst + -- Check if we agree that the connection should be closed+ if vst' ^. remoteOutgoing == 0 + then do + removeRemoteEndPoint (ourEndPoint, theirEndPoint)+ -- Attempt to reply (but don't insist)+ tryIO $ sendOn vst' [encodeInt32 CloseSocket]+ tryCloseSocket sock + return (RemoteEndPointClosed, True)+ else + return (RemoteEndPointValid vst', False)+ RemoteEndPointClosing resolved _ -> do+ removeRemoteEndPoint (ourEndPoint, theirEndPoint)+ tryCloseSocket sock + putMVar resolved ()+ return (RemoteEndPointClosed, True)+ RemoteEndPointFailed err ->+ throwIO err+ RemoteEndPointClosed ->+ relyViolation (ourEndPoint, theirEndPoint) + "handleIncomingMessages:closeSocket (closed)"+ + -- 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 -> ConnectionId -> IO () + readMessage sock connId = + recvWithLength sock >>= writeChan ourChannel . Received connId++ -- 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+ let code = EventConnectionLost + (Just $ remoteAddress theirEndPoint) + (IntSet.elems $ vst ^. remoteIncoming)+ writeChan ourChannel . ErrorEvent $ TransportError code (show err)+ forM_ (vst ^. pendingCtrlRequests) $ flip putMVar (Left err) + return (RemoteEndPointFailed err)+ RemoteEndPointClosing resolved _ -> do+ putMVar resolved ()+ return (RemoteEndPointFailed err)+ RemoteEndPointClosed ->+ relyViolation (ourEndPoint, theirEndPoint) + "handleIncomingMessages:prematureExit"+ RemoteEndPointFailed err' ->+ return (RemoteEndPointFailed err')++--------------------------------------------------------------------------------+-- Uninterruptable auxiliary functions --+-- --+-- All these functions assume they are running in a thread which will never --+-- be killed.+--------------------------------------------------------------------------------++-- | Request a connection to a remote endpoint+--+-- This will block until we get a connection ID from the remote endpoint; if+-- the remote endpoint was in 'RemoteEndPointClosing' state then we will+-- additionally block until that is resolved. +--+-- May throw a TransportError ConnectErrorCode exception.+requestConnectionTo :: TCPParameters + -> LocalEndPoint + -> EndPointAddress + -> ConnectHints+ -> IO (RemoteEndPoint, ConnectionId)+requestConnectionTo params ourEndPoint theirAddress hints = go+ where+ go = do+ (theirEndPoint, isNew) <- mapIOException connectFailed $+ findRemoteEndPoint ourEndPoint theirAddress RequestedByUs++ if isNew + then do+ forkIO . handle absorbAllExceptions $ + setupRemoteEndPoint params (ourEndPoint, theirEndPoint) hints + go+ else do+ reply <- mapIOException connectFailed $ + doRemoteRequest (ourEndPoint, theirEndPoint) RequestConnectionId + return (theirEndPoint, decodeInt32 . BS.concat $ reply)++ connectFailed :: IOException -> TransportError ConnectErrorCode+ connectFailed = TransportError ConnectFailed . show++ absorbAllExceptions :: SomeException -> IO ()+ absorbAllExceptions _ex = + return ()++-- | Set up a remote endpoint+setupRemoteEndPoint :: TCPParameters -> EndPointPair -> ConnectHints -> IO () +setupRemoteEndPoint params (ourEndPoint, theirEndPoint) hints = do+ result <- socketToEndPoint ourAddress + theirAddress + (tcpReuseClientAddr params)+ (connectTimeout hints)+ didAccept <- case result of+ Right (sock, ConnectionRequestAccepted) -> do + let vst = ValidRemoteEndPointState + { remoteSocket = sock+ , _remoteOutgoing = 0 + , _remoteIncoming = IntSet.empty+ , sendOn = sendMany sock + , _pendingCtrlRequests = IntMap.empty+ , _nextCtrlRequestId = 0+ }+ 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+ resolveInit (ourEndPoint, theirEndPoint) RemoteEndPointClosed+ tryCloseSocket sock+ return False+ Left err -> do + resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointInvalid err)+ return False++ when didAccept $ handleIncomingMessages (ourEndPoint, theirEndPoint) + where+ ourAddress = localAddress ourEndPoint+ theirAddress = remoteAddress theirEndPoint+ invalidAddress = TransportError ConnectNotFound++-- | Do a (blocking) remote request +-- +-- May throw IO (user) exception if the local or the remote endpoint is closed,+-- if the send fails, or if the remote endpoint fails before it replies.+doRemoteRequest :: EndPointPair -> ControlHeader -> IO [ByteString]+doRemoteRequest (ourEndPoint, theirEndPoint) header = do + replyMVar <- newEmptyMVar+ modifyRemoteState_ (ourEndPoint, theirEndPoint) RemoteStatePatternMatch + { caseValid = \vst -> do + let reqId = vst ^. nextCtrlRequestId+ sendOn vst [encodeInt32 header, encodeInt32 reqId]+ return ( RemoteEndPointValid+ . (nextCtrlRequestId ^: (+ 1))+ . (pendingCtrlRequestsAt reqId ^= Just replyMVar)+ $ vst+ ) + -- Error cases+ , caseInvalid = + throwIO + , caseInit = \_ _ -> + relyViolation (ourEndPoint, theirEndPoint) "doRemoteRequest (init)"+ , caseClosing = \_ _ -> + relyViolation (ourEndPoint, theirEndPoint) "doRemoteRequest (closing)" + , caseClosed =+ relyViolation (ourEndPoint, theirEndPoint) "doRemoteRequest (closed)" + , caseFailed =+ throwIO+ }+ mReply <- takeMVar replyMVar+ case mReply of+ Left err -> throwIO err+ Right reply -> return reply++-- | Send a CloseSocket request if the remote endpoint is unused+closeIfUnused :: EndPointPair -> IO ()+closeIfUnused (ourEndPoint, theirEndPoint) =+ modifyRemoteState_ (ourEndPoint, theirEndPoint) remoteStateIdentity + { caseValid = \vst -> + if vst ^. remoteOutgoing == 0 && IntSet.null (vst ^. remoteIncoming) + then do + sendOn vst [encodeInt32 CloseSocket]+ resolved <- newEmptyMVar+ return $ RemoteEndPointClosing resolved vst + else + return $ RemoteEndPointValid vst+ }++-- | 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+ connId <- mapIOException connectFailed $ getNextConnectionId ourEndPoint + writeChan ourChan $+ ConnectionOpened connId ReliableOrdered (localAddress ourEndPoint)+ return Connection + { send = selfSend connAlive connId + , close = selfClose connAlive connId+ }+ where+ selfSend :: IORef Bool + -> ConnectionId + -> [ByteString] + -> IO (Either (TransportError SendErrorCode) ())+ selfSend connAlive connId msg = + try . withMVar ourState $ \st -> case st of+ LocalEndPointValid _ -> do+ alive <- readIORef connAlive+ if alive+ then writeChan ourChan (Received connId msg)+ else throwIO $ TransportError SendClosed "Connection closed"+ LocalEndPointClosed ->+ throwIO $ TransportError SendFailed "Endpoint closed"++ selfClose :: IORef Bool -> ConnectionId -> IO ()+ selfClose connAlive connId = + withMVar ourState $ \st -> case st of+ LocalEndPointValid _ -> do+ alive <- readIORef connAlive+ when alive $ do+ writeChan ourChan (ConnectionClosed connId) + writeIORef connAlive False+ LocalEndPointClosed ->+ return () ++ ourChan = localChannel ourEndPoint+ ourState = localState ourEndPoint+ connectFailed = TransportError ConnectFailed . show ++-- | Resolve an endpoint currently in 'Init' state+resolveInit :: EndPointPair -> RemoteState -> IO ()+resolveInit (ourEndPoint, theirEndPoint) newState =+ modifyMVar_ (remoteState theirEndPoint) $ \st -> case st of+ RemoteEndPointInit resolved _ -> do+ putMVar resolved ()+ case newState of + RemoteEndPointClosed -> + removeRemoteEndPoint (ourEndPoint, theirEndPoint)+ _ ->+ return ()+ return newState+ RemoteEndPointFailed ex -> + throwIO ex+ _ ->+ relyViolation (ourEndPoint, theirEndPoint) "resolveInit"++-- | Get the next connection ID+-- +-- Throws an IO exception when the endpoint is closed.+getNextConnectionId :: LocalEndPoint -> IO ConnectionId+getNextConnectionId ourEndpoint = + modifyMVar (localState ourEndpoint) $ \st -> case st of+ LocalEndPointValid vst -> do+ let connId = vst ^. nextConnectionId + return ( LocalEndPointValid + . (nextConnectionId ^= 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 + { _nextConnectionId = firstNonReservedConnectionId + , _localConnections = Map.empty+ , _nextRemoteId = 0+ }+ modifyMVar (transportState transport) $ \st -> case st of+ TransportValid vst -> do+ let ix = vst ^. nextEndPointId+ let addr = encodeEndPointAddress (transportHost transport) + (transportPort transport)+ ix + let localEndPoint = LocalEndPoint { localAddress = addr+ , localChannel = chan+ , localState = state+ }+ return ( TransportValid + . (localEndPointAt addr ^= Just localEndPoint) + . (nextEndPointId ^= ix + 1) + $ vst+ , localEndPoint+ )+ TransportClosed ->+ throwIO (TransportError NewEndPointFailed "Transport closed")+++-- | Remove reference to a remote endpoint from a local endpoint+--+-- If the local endpoint is closed, do nothing+removeRemoteEndPoint :: EndPointPair -> IO ()+removeRemoteEndPoint (ourEndPoint, theirEndPoint) =+ modifyMVar_ ourState $ \st -> case st of+ LocalEndPointValid vst ->+ case vst ^. localConnectionTo theirAddress of+ Nothing -> + return st+ Just remoteEndPoint' ->+ if remoteId remoteEndPoint' == remoteId theirEndPoint + then return + ( LocalEndPointValid + . (localConnectionTo (remoteAddress theirEndPoint) ^= Nothing) + $ vst+ )+ else return st+ LocalEndPointClosed ->+ return LocalEndPointClosed+ where+ ourState = localState ourEndPoint+ theirAddress = remoteAddress theirEndPoint++-- | Remove reference to a local endpoint from the transport state+--+-- Does nothing if the transport is closed+removeLocalEndPoint :: TCPTransport -> LocalEndPoint -> IO ()+removeLocalEndPoint transport ourEndPoint = + modifyMVar_ (transportState transport) $ \st -> case st of + TransportValid vst ->+ return ( TransportValid + . (localEndPointAt (localAddress ourEndPoint) ^= Nothing) + $ vst+ )+ TransportClosed ->+ return TransportClosed++-- | Find a remote endpoint. If the remote endpoint does not yet exist we+-- create it in Init state. Returns if the endpoint was new. +findRemoteEndPoint + :: LocalEndPoint+ -> EndPointAddress+ -> RequestedBy + -> IO (RemoteEndPoint, Bool) +findRemoteEndPoint ourEndPoint theirAddress findOrigin = go+ where+ go = do+ (theirEndPoint, isNew) <- modifyMVar ourState $ \st -> case st of+ LocalEndPointValid vst -> case vst ^. localConnectionTo theirAddress of+ Just theirEndPoint ->+ return (st, (theirEndPoint, False))+ Nothing -> do+ resolved <- newEmptyMVar+ theirState <- newMVar (RemoteEndPointInit resolved findOrigin)+ let theirEndPoint = RemoteEndPoint+ { remoteAddress = theirAddress+ , remoteState = theirState+ , remoteId = vst ^. nextRemoteId+ }+ return ( LocalEndPointValid + . (localConnectionTo theirAddress ^= Just theirEndPoint) + . (nextRemoteId ^: (+ 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 initOrigin ->+ case (findOrigin, initOrigin) of+ (RequestedByUs, RequestedByUs) -> + readMVar resolved >> go + (RequestedByUs, RequestedByThem) -> + readMVar resolved >> go+ (RequestedByThem, RequestedByUs) -> + if ourAddress > theirAddress + then+ -- Wait for the Crossed message+ readMVar resolved >> go + else+ return (theirEndPoint, False)+ (RequestedByThem, RequestedByThem) -> + throwIO $ userError "Already connected"+ RemoteEndPointValid _ ->+ -- We assume that the request crossed if we find the endpoint in+ -- Valid state. It is possible that this is really an invalid+ -- request, but only in the case of a broken client (we don't+ -- maintain enough history to be able to tell the difference).+ return (theirEndPoint, False)+ RemoteEndPointClosing resolved _ ->+ readMVar resolved >> go+ RemoteEndPointClosed ->+ go+ RemoteEndPointFailed err -> + throwIO err+ + ourState = localState ourEndPoint + ourAddress = localAddress ourEndPoint++--------------------------------------------------------------------------------+-- "Stateless" (MVar free) functions --+--------------------------------------------------------------------------------++-- | Establish a connection to a remote endpoint+--+-- Maybe throw a TransportError+socketToEndPoint :: EndPointAddress -- ^ Our address + -> EndPointAddress -- ^ Their address+ -> Bool -- ^ Use SO_REUSEADDR?+ -> Maybe Int -- ^ Timeout for connect + -> IO (Either (TransportError ConnectErrorCode) + (N.Socket, ConnectionRequestResponse)) +socketToEndPoint (EndPointAddress ourAddress) theirAddress reuseAddr timeout = + try $ do + (host, port, theirEndPointId) <- case decodeEndPointAddress theirAddress of + Nothing -> throwIO (failed . userError $ "Could not parse")+ Just dec -> return dec+ addr:_ <- mapIOException invalidAddress $ + N.getAddrInfo Nothing (Just host) (Just port)+ bracketOnError (createSocket addr) tryCloseSocket $ \sock -> do+ when reuseAddr $ + mapIOException failed $ N.setSocketOption sock N.ReuseAddr 1+ mapIOException invalidAddress $ + timeoutMaybe timeout timeoutError $ + N.connect sock (N.addrAddress addr) + response <- mapIOException failed $ do+ sendMany sock (encodeInt32 theirEndPointId : prependLength [ourAddress])+ recvInt32 sock+ case tryToEnum response of+ Nothing -> throwIO (failed . userError $ "Unexpected response")+ Just r -> return (sock, r)+ where+ createSocket :: N.AddrInfo -> IO N.Socket+ createSocket addr = mapIOException insufficientResources $ + N.socket (N.addrFamily addr) N.Stream N.defaultProtocol++ invalidAddress = TransportError ConnectNotFound . show + insufficientResources = TransportError ConnectInsufficientResources . show + failed = TransportError ConnectFailed . show+ timeoutError = TransportError ConnectTimeout "Timed out"++-- | Encode end point address+encodeEndPointAddress :: N.HostName + -> N.ServiceName + -> EndPointId + -> EndPointAddress+encodeEndPointAddress host port ix = EndPointAddress . BSC.pack $+ host ++ ":" ++ port ++ ":" ++ show ix ++-- | Decode end point address+decodeEndPointAddress :: EndPointAddress + -> Maybe (N.HostName, N.ServiceName, EndPointId)+decodeEndPointAddress (EndPointAddress bs) = + case map BSC.unpack $ BSC.split ':' bs of+ [host, port, endPointIdStr] -> + case reads endPointIdStr of + [(endPointId, "")] -> Just (host, port, endPointId)+ _ -> Nothing+ _ ->+ Nothing++--------------------------------------------------------------------------------+-- 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+firstNonReservedConnectionId :: ConnectionId+firstNonReservedConnectionId = 1024++--------------------------------------------------------------------------------+-- 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 })++nextConnectionId :: Accessor ValidLocalEndPointState ConnectionId+nextConnectionId = accessor _nextConnectionId (\cix st -> st { _nextConnectionId = cix })++localConnections :: Accessor ValidLocalEndPointState (Map EndPointAddress RemoteEndPoint)+localConnections = accessor _localConnections (\es st -> st { _localConnections = es })++nextRemoteId :: Accessor ValidLocalEndPointState Int+nextRemoteId = accessor _nextRemoteId (\rid st -> st { _nextRemoteId = rid })++remoteOutgoing :: Accessor ValidRemoteEndPointState Int+remoteOutgoing = accessor _remoteOutgoing (\cs conn -> conn { _remoteOutgoing = cs })++remoteIncoming :: Accessor ValidRemoteEndPointState IntSet+remoteIncoming = accessor _remoteIncoming (\cs conn -> conn { _remoteIncoming = cs })++pendingCtrlRequests :: Accessor ValidRemoteEndPointState (IntMap (MVar (Either IOException [ByteString])))+pendingCtrlRequests = accessor _pendingCtrlRequests (\rep st -> st { _pendingCtrlRequests = rep })++nextCtrlRequestId :: Accessor ValidRemoteEndPointState ControlRequestId +nextCtrlRequestId = accessor _nextCtrlRequestId (\cid st -> st { _nextCtrlRequestId = cid })++localEndPointAt :: EndPointAddress -> Accessor ValidTransportState (Maybe LocalEndPoint)+localEndPointAt addr = localEndPoints >>> DAC.mapMaybe addr ++pendingCtrlRequestsAt :: ControlRequestId -> Accessor ValidRemoteEndPointState (Maybe (MVar (Either IOException [ByteString])))+pendingCtrlRequestsAt ix = pendingCtrlRequests >>> DAC.intMapMaybe (fromIntegral ix)++localConnectionTo :: EndPointAddress + -> Accessor ValidLocalEndPointState (Maybe RemoteEndPoint)+localConnectionTo addr = localConnections >>> DAC.mapMaybe addr ++-------------------------------------------------------------------------------+-- Debugging --+-------------------------------------------------------------------------------++relyViolation :: EndPointPair -> String -> IO a+relyViolation (ourEndPoint, theirEndPoint) str = do+ elog (ourEndPoint, theirEndPoint) (str ++ " RELY violation") + fail (str ++ " RELY violation")++elog :: EndPointPair -> String -> IO ()+elog (ourEndPoint, theirEndPoint) msg = do+ tid <- myThreadId+ putStrLn $ show (localAddress ourEndPoint) + ++ "/" ++ show (remoteAddress theirEndPoint) + ++ "(" ++ show (remoteId theirEndPoint) ++ ")"+ ++ "/" ++ show tid + ++ ": " ++ msg
+ src/Network/Transport/TCP/Internal.hs view
@@ -0,0 +1,116 @@+-- | Utility functions for TCP sockets +module Network.Transport.TCP.Internal ( forkServer+ , recvWithLength+ , recvExact + , recvInt32+ , tryCloseSocket+ ) where++import Prelude hiding (catch)+import Network.Transport.Internal (decodeInt32, void, tryIO, forkIOWithUnmask)+import qualified Network.Socket as N ( HostName+ , ServiceName+ , Socket+ , SocketType(Stream)+ , SocketOption(ReuseAddr)+ , getAddrInfo+ , defaultHints+ , socket+ , bindSocket+ , listen+ , addrFamily+ , addrAddress+ , defaultProtocol+ , setSocketOption+ , accept+ , sClose+ )+import qualified Network.Socket.ByteString as NBS (recv)+import Control.Concurrent (ThreadId)+import Control.Monad (forever, when)+import Control.Exception (SomeException, catch, bracketOnError, throwIO, mask_)+import Control.Applicative ((<$>))+import Data.ByteString (ByteString)+import qualified Data.ByteString as BS (length, concat, null)+import Data.Int (Int32)++-- | 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.+--+-- The request handler should spawn threads to handle each individual request+-- or the server will block. Once a thread has been spawned it will be the+-- responsibility of the new thread to close the socket when an exception+-- occurs. +forkServer :: N.HostName -- ^ Host+ -> N.ServiceName -- ^ Port + -> Int -- ^ Backlog (maximum number of queued connections)+ -> Bool -- ^ Set ReuseAddr option?+ -> (SomeException -> IO ()) -- ^ Termination handler+ -> (N.Socket -> IO ()) -- ^ Request handler + -> IO ThreadId+forkServer host port backlog reuseAddr terminationHandler requestHandler = do + -- 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+ addr:_ <- N.getAddrInfo (Just N.defaultHints) (Just host) (Just port)+ bracketOnError (N.socket (N.addrFamily addr) N.Stream N.defaultProtocol)+ tryCloseSocket $ \sock -> do+ when reuseAddr $ N.setSocketOption sock N.ReuseAddr 1+ N.bindSocket sock (N.addrAddress addr)+ N.listen sock backlog + -- We start listening for incoming requests in a separate thread. When+ -- that thread is killed, we close the server socket and the termination+ -- handler. We have to make sure that the exception handler is installed+ -- /before/ any asynchronous exception occurs. So we mask_, then fork+ -- (the child thread inherits the masked state from the parent), then+ -- unmask only inside the catch.+ mask_ $ forkIOWithUnmask $ \unmask -> + catch (unmask (forever $ acceptRequest sock)) $ \ex -> do+ tryCloseSocket sock+ terminationHandler ex+ where+ acceptRequest :: N.Socket -> IO ()+ acceptRequest sock = bracketOnError (N.accept sock)+ (tryCloseSocket . fst)+ (requestHandler . fst)+ +-- | Read a length and then a payload of that length+recvWithLength :: N.Socket -> IO [ByteString]+recvWithLength sock = recvInt32 sock >>= recvExact sock++-- | Receive a 32-bit integer+recvInt32 :: Num a => N.Socket -> IO a +recvInt32 sock = decodeInt32 . BS.concat <$> recvExact sock 4 ++-- | Close a socket, ignoring I/O exceptions+tryCloseSocket :: N.Socket -> IO ()+tryCloseSocket sock = void . tryIO $ + N.sClose sock++-- | Read an exact number of bytes from a socket+-- +-- Throws an I/O exception if the socket closes before the specified+-- number of bytes could be read+recvExact :: N.Socket -- ^ Socket to read from + -> Int32 -- ^ Number of bytes to read+ -> IO [ByteString]+recvExact _ len | len <= 0 = throwIO (userError "recvExact: Negative length") +recvExact sock len = go [] len+ where+ go :: [ByteString] -> Int32 -> IO [ByteString] + go acc 0 = return (reverse acc) + go acc l = do+ bs <- NBS.recv sock (fromIntegral l `min` 4096)+ if BS.null bs + then throwIO (userError "recvExact: Socket closed")+ else go (bs : acc) (l - fromIntegral (BS.length bs))
+ tests/TestTCP.hs view
@@ -0,0 +1,786 @@+{-# LANGUAGE RebindableSyntax #-}+{-# OPTIONS_GHC -fno-warn-orphans #-}+module Main where++import Prelude hiding (catch, (>>=), (>>), return, fail)+import TestTransport (testTransport) +import TestAuxiliary (forkTry, runTests)+import Network.Transport+import Network.Transport.TCP ( createTransport+ , createTransportExposeInternals+ , TransportInternals(..)+ , encodeEndPointAddress+ , defaultTCPParameters+ )+import Data.Int (Int32)+import Control.Concurrent (threadDelay, killThread)+import Control.Concurrent.MVar ( MVar+ , newEmptyMVar+ , putMVar+ , takeMVar+ , readMVar+ , isEmptyMVar+ , newMVar+ , modifyMVar+ )+import Control.Monad (replicateM, guard, forM_, replicateM_, when)+import Control.Applicative ((<$>))+import Control.Exception (throwIO, try, SomeException)+import Network.Transport.TCP ( ControlHeader(..)+ , ConnectionRequestResponse(..)+ , socketToEndPoint+ )+import Network.Transport.Internal ( encodeInt32+ , prependLength+ , tlog+ , tryIO+ , void+ )+import Network.Transport.TCP.Internal (recvInt32, forkServer, recvWithLength)+import qualified Network.Socket as N ( sClose+ , ServiceName+ , Socket+ , AddrInfo+ , shutdown+ , ShutdownCmd(ShutdownSend)+ )+import Network.Socket.ByteString (sendMany) +import Data.String (fromString)+import Traced +import GHC.IO.Exception (ioe_errno)+import Foreign.C.Error (Errno(..), eADDRNOTAVAIL)+import System.Timeout (timeout)++instance Traceable ControlHeader where+ trace = traceShow++instance Traceable ConnectionRequestResponse where+ trace = traceShow++instance Traceable N.Socket where+ trace = traceShow ++instance Traceable N.AddrInfo where+ trace = traceShow++instance Traceable TransportInternals where+ trace = const Nothing++-- Test that the server gets a ConnectionClosed message when the client closes+-- the socket without sending an explicit control message to the server first+testEarlyDisconnect :: IO N.ServiceName -> IO ()+testEarlyDisconnect nextPort = do+ clientAddr <- newEmptyMVar+ serverAddr <- newEmptyMVar+ serverDone <- newEmptyMVar++ tlog "testEarlyDisconnect"+ forkTry $ server serverAddr clientAddr serverDone+ forkTry $ client serverAddr clientAddr++ takeMVar serverDone+ where+ server :: MVar EndPointAddress -> MVar EndPointAddress -> MVar () -> IO ()+ server serverAddr clientAddr serverDone = do+ tlog "Server"+ Right transport <- nextPort >>= \port -> createTransport "127.0.0.1" port defaultTCPParameters+ Right endpoint <- newEndPoint transport+ putMVar serverAddr (address endpoint)+ theirAddr <- readMVar clientAddr++ -- TEST 1: they connect to us, then drop the connection+ do+ ConnectionOpened cid _ addr <- receive endpoint + True <- return $ addr == theirAddr+ + ErrorEvent (TransportError (EventConnectionLost (Just addr') [cid']) _) <- receive endpoint + True <- return $ addr' == theirAddr && cid' == cid++ return ()++ -- TEST 2: after they dropped their connection to us, we now try to+ -- establish a connection to them. This should re-establish the broken+ -- TCP connection. + tlog "Trying to connect to client"+ Right conn <- connect endpoint theirAddr ReliableOrdered defaultConnectHints++ -- TEST 3: To test the connection, we do a simple ping test; as before,+ -- however, the remote client won't close the connection nicely but just+ -- closes the socket+ do+ Right () <- send conn ["ping"]+ + ConnectionOpened cid _ addr <- receive endpoint + True <- return $ addr == theirAddr+ + Received cid' ["pong"] <- receive endpoint+ True <- return $ cid == cid'+ + ErrorEvent (TransportError (EventConnectionLost (Just addr') [cid'']) _) <- receive endpoint+ True <- return $ addr' == theirAddr && cid'' == cid++ return ()++ -- TEST 4: A subsequent send on an already-open connection will now break+ Left (TransportError SendFailed _) <- send conn ["ping2"]++ -- *Pfew* + putMVar serverDone ()++ client :: MVar EndPointAddress -> MVar EndPointAddress -> IO ()+ client serverAddr clientAddr = do+ tlog "Client"+ clientPort <- nextPort+ let ourAddress = encodeEndPointAddress "127.0.0.1" clientPort 0 + putMVar clientAddr ourAddress + + -- Listen for incoming messages+ forkServer "127.0.0.1" clientPort 5 True throwIO $ \sock -> do+ -- Initial setup + 0 <- recvInt32 sock :: IO Int+ _ <- recvWithLength sock + sendMany sock [encodeInt32 ConnectionRequestAccepted]++ -- Server requests a logical connection + RequestConnectionId <- toEnum <$> (recvInt32 sock :: IO Int)+ reqId <- recvInt32 sock :: IO Int+ sendMany sock (encodeInt32 ControlResponse : encodeInt32 reqId : prependLength [encodeInt32 (10001 :: Int)])++ -- Server sends a message+ 10001 <- recvInt32 sock :: IO Int+ ["ping"] <- recvWithLength sock++ -- Reply + sendMany sock [encodeInt32 RequestConnectionId, encodeInt32 (10002 :: Int)]+ ControlResponse <- toEnum <$> (recvInt32 sock :: IO Int)+ 10002 <- recvInt32 sock :: IO Int + [cid] <- recvWithLength sock+ sendMany sock (cid : prependLength ["pong"]) ++ -- Close the socket+ N.sClose sock+ + -- Connect to the server+ Right (sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr True Nothing+ + -- Request a new connection, but don't wait for the response+ let reqId = 0 :: Int32+ sendMany sock [encodeInt32 RequestConnectionId, encodeInt32 reqId]+ + -- Close the socket without closing the connection explicitly+ -- The server should receive an error event + N.sClose sock++-- | Test the behaviour of a premature CloseSocket request+testEarlyCloseSocket :: IO N.ServiceName -> IO ()+testEarlyCloseSocket nextPort = do+ clientAddr <- newEmptyMVar+ serverAddr <- newEmptyMVar+ serverDone <- newEmptyMVar++ tlog "testEarlyDisconnect"+ forkTry $ server serverAddr clientAddr serverDone+ forkTry $ client serverAddr clientAddr++ takeMVar serverDone+ where+ server :: MVar EndPointAddress -> MVar EndPointAddress -> MVar () -> IO ()+ server serverAddr clientAddr serverDone = do+ tlog "Server"+ Right transport <- nextPort >>= \port -> createTransport "127.0.0.1" port defaultTCPParameters+ Right endpoint <- newEndPoint transport+ putMVar serverAddr (address endpoint)+ theirAddr <- readMVar clientAddr++ -- TEST 1: they connect to us, then send a CloseSocket. Since we don't+ -- have any outgoing connections, this means we will agree to close the+ -- socket + do+ ConnectionOpened cid _ addr <- receive endpoint + True <- return $ addr == theirAddr+ + ConnectionClosed cid' <- receive endpoint + True <- return $ cid' == cid++ return ()++ -- TEST 2: after they dropped their connection to us, we now try to+ -- establish a connection to them. This should re-establish the broken+ -- TCP connection. + tlog "Trying to connect to client"+ Right conn <- connect endpoint theirAddr ReliableOrdered defaultConnectHints++ -- TEST 3: To test the connection, we do a simple ping test; as before,+ -- however, the remote client won't close the connection nicely but just+ -- sends a CloseSocket -- except that now we *do* have outgoing+ -- connections, so we won't agree and hence will receive an error when+ -- the socket gets closed+ do+ Right () <- send conn ["ping"]+ + ConnectionOpened cid _ addr <- receive endpoint+ True <- return $ addr == theirAddr+ + Received cid' ["pong"] <- receive endpoint+ True <- return $ cid' == cid+ + ConnectionClosed cid'' <- receive endpoint+ True <- return $ cid'' == cid+ + ErrorEvent (TransportError (EventConnectionLost (Just addr') []) _) <- receive endpoint+ True <- return $ addr' == theirAddr + + return ()++ -- TEST 4: A subsequent send on an already-open connection will now break+ Left (TransportError SendFailed _) <- send conn ["ping2"]++ -- *Pfew* + putMVar serverDone ()++ client :: MVar EndPointAddress -> MVar EndPointAddress -> IO ()+ client serverAddr clientAddr = do+ tlog "Client"+ clientPort <- nextPort+ let ourAddress = encodeEndPointAddress "127.0.0.1" clientPort 0 + putMVar clientAddr ourAddress + + -- Listen for incoming messages+ forkServer "127.0.0.1" clientPort 5 True throwIO $ \sock -> do+ -- Initial setup + 0 <- recvInt32 sock :: IO Int+ _ <- recvWithLength sock + sendMany sock [encodeInt32 ConnectionRequestAccepted]++ -- Server requests a logical connection + RequestConnectionId <- toEnum <$> (recvInt32 sock :: IO Int)+ reqId <- recvInt32 sock :: IO Int+ sendMany sock (encodeInt32 ControlResponse : encodeInt32 reqId : prependLength [encodeInt32 (10001 :: Int)])++ -- Server sends a message+ 10001 <- recvInt32 sock :: IO Int+ ["ping"] <- recvWithLength sock++ -- Reply + sendMany sock [encodeInt32 RequestConnectionId, encodeInt32 (10002 :: Int)]+ ControlResponse <- toEnum <$> (recvInt32 sock :: IO Int)+ 10002 <- recvInt32 sock :: IO Int + [cid] <- recvWithLength sock+ sendMany sock (cid : prependLength ["pong"]) ++ -- Send a CloseSocket even though there are still connections *in both+ -- directions*+ sendMany sock [encodeInt32 CloseSocket]+ N.sClose sock+ + -- Connect to the server+ Right (sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr True Nothing+ + -- Request a new connection, but don't wait for the response+ let reqId = 0 :: Int32+ sendMany sock [encodeInt32 RequestConnectionId, encodeInt32 reqId]+ + -- Send a CloseSocket without sending a closeconnecton + -- The server should still receive a ConnectionClosed message+ sendMany sock [encodeInt32 CloseSocket]+ N.sClose sock++-- | Test the creation of a transport with an invalid address+testInvalidAddress :: IO N.ServiceName -> IO ()+testInvalidAddress nextPort = do+ Left _ <- nextPort >>= \port -> createTransport "invalidHostName" port defaultTCPParameters+ return ()++-- | Test connecting to invalid or non-existing endpoints+testInvalidConnect :: IO N.ServiceName -> IO ()+testInvalidConnect nextPort = do+ port <- nextPort+ Right transport <- createTransport "127.0.0.1" port defaultTCPParameters+ Right endpoint <- newEndPoint transport++ -- Syntax error in the endpoint address+ Left (TransportError ConnectFailed _) <- + connect endpoint (EndPointAddress "InvalidAddress") ReliableOrdered defaultConnectHints+ + -- Syntax connect, but invalid hostname (TCP address lookup failure)+ Left (TransportError ConnectNotFound _) <- + connect endpoint (encodeEndPointAddress "invalidHost" "port" 0) ReliableOrdered defaultConnectHints+ + -- TCP address correct, but nobody home at that address+ Left (TransportError ConnectNotFound _) <- + connect endpoint (encodeEndPointAddress "127.0.0.1" "9000" 0) ReliableOrdered defaultConnectHints+ + -- Valid TCP address but invalid endpoint number+ Left (TransportError ConnectNotFound _) <- + connect endpoint (encodeEndPointAddress "127.0.0.1" port 1) ReliableOrdered defaultConnectHints++ return ()++-- | Test that an endpoint can ignore CloseSocket requests (in "reality" this+-- would happen when the endpoint sends a new connection request before+-- receiving an (already underway) CloseSocket request) +testIgnoreCloseSocket :: IO N.ServiceName -> IO ()+testIgnoreCloseSocket nextPort = do+ serverAddr <- newEmptyMVar+ clientDone <- newEmptyMVar+ Right transport <- nextPort >>= \port -> createTransport "127.0.0.1" port defaultTCPParameters+ + forkTry $ server transport serverAddr+ forkTry $ client transport serverAddr clientDone + + takeMVar clientDone++ where+ server :: Transport -> MVar EndPointAddress -> IO ()+ server transport serverAddr = do+ tlog "Server"+ Right endpoint <- newEndPoint transport+ putMVar serverAddr (address endpoint)++ -- Wait for the client to connect and disconnect+ tlog "Waiting for ConnectionOpened"+ ConnectionOpened _ _ _ <- receive endpoint+ tlog "Waiting for ConnectionClosed"+ ConnectionClosed _ <- receive endpoint++ -- At this point the server will have sent a CloseSocket request to the+ -- client, which however ignores it, instead it requests and closes+ -- another connection+ tlog "Waiting for ConnectionOpened"+ ConnectionOpened _ _ _ <- receive endpoint+ tlog "Waiting for ConnectionClosed"+ ConnectionClosed _ <- receive endpoint++ + tlog "Server waiting.."++ client :: Transport -> MVar EndPointAddress -> MVar () -> IO ()+ client transport serverAddr clientDone = do+ tlog "Client"+ Right endpoint <- newEndPoint transport+ let ourAddress = address endpoint++ -- Connect to the server+ Right (sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr True Nothing++ -- Request a new connection+ tlog "Requesting connection"+ let reqId = 0 :: Int32+ sendMany sock [encodeInt32 RequestConnectionId, encodeInt32 reqId]+ response <- replicateM 4 $ recvInt32 sock :: IO [Int32] ++ -- Close the connection again+ tlog "Closing connection"+ sendMany sock [encodeInt32 CloseConnection, encodeInt32 (response !! 3)] ++ -- Server will now send a CloseSocket request as its refcount reached 0+ tlog "Waiting for CloseSocket request"+ CloseSocket <- toEnum <$> recvInt32 sock++ -- But we ignore it and request another connection+ tlog "Ignoring it, requesting another connection"+ let reqId' = 1 :: Int32+ sendMany sock [encodeInt32 RequestConnectionId, encodeInt32 reqId']+ response' <- replicateM 4 $ recvInt32 sock :: IO [Int32] ++ -- Close it again+ tlog "Closing connection"+ sendMany sock [encodeInt32 CloseConnection, encodeInt32 (response' !! 3)] ++ -- We now get a CloseSocket again, and this time we heed it+ tlog "Waiting for second CloseSocket request"+ CloseSocket <- toEnum <$> recvInt32 sock+ + tlog "Closing socket"+ sendMany sock [encodeInt32 CloseSocket]+ N.sClose sock++ putMVar clientDone ()++-- | Like 'testIgnoreSocket', but now the server requests a connection after the+-- client closed their connection. In the meantime, the server will have sent a+-- CloseSocket request to the client, and must block until the client responds.+testBlockAfterCloseSocket :: IO N.ServiceName -> IO ()+testBlockAfterCloseSocket nextPort = do+ serverAddr <- newEmptyMVar+ clientAddr <- newEmptyMVar+ clientDone <- newEmptyMVar+ port <- nextPort + Right transport <- createTransport "127.0.0.1" port defaultTCPParameters+ + forkTry $ server transport serverAddr clientAddr+ forkTry $ client transport serverAddr clientAddr clientDone + + takeMVar clientDone++ where+ server :: Transport -> MVar EndPointAddress -> MVar EndPointAddress -> IO ()+ server transport serverAddr clientAddr = do+ tlog "Server"+ Right endpoint <- newEndPoint transport+ putMVar serverAddr (address endpoint)++ -- Wait for the client to connect and disconnect+ tlog "Waiting for ConnectionOpened"+ ConnectionOpened _ _ _ <- receive endpoint+ tlog "Waiting for ConnectionClosed"+ ConnectionClosed _ <- receive endpoint+ + -- At this point the server will have sent a CloseSocket request to the+ -- client, and must block until the client responds+ tlog "Server waiting to connect to the client.."+ Right _ <- readMVar clientAddr >>= \addr -> connect endpoint addr ReliableOrdered defaultConnectHints+ + tlog "Server waiting.."++ client :: Transport -> MVar EndPointAddress -> MVar EndPointAddress -> MVar () -> IO ()+ client transport serverAddr clientAddr clientDone = do+ tlog "Client"+ Right endpoint <- newEndPoint transport+ putMVar clientAddr (address endpoint)+ let ourAddress = address endpoint++ -- Connect to the server+ Right (sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr True Nothing ++ -- Request a new connection+ tlog "Requesting connection"+ let reqId = 0 :: Int32+ sendMany sock [encodeInt32 RequestConnectionId, encodeInt32 reqId]+ response <- replicateM 4 $ recvInt32 sock :: IO [Int32] ++ -- Close the connection again+ tlog "Closing connection"+ sendMany sock [encodeInt32 CloseConnection, encodeInt32 (response !! 3)] ++ -- Server will now send a CloseSocket request as its refcount reached 0+ tlog "Waiting for CloseSocket request"+ CloseSocket <- toEnum <$> recvInt32 sock++ unblocked <- newEmptyMVar++ -- We should not hear from the server until we unblock him by+ -- responding to the CloseSocket request (in this case, we + -- respond by sending a ConnectionRequest)+ forkTry $ do+ recvInt32 sock :: IO Int32+ isEmptyMVar unblocked >>= (guard . not)+ putMVar clientDone ()++ threadDelay 1000000++ tlog "Client ignores close socket and sends connection request"+ tlog "This should unblock the server"+ putMVar unblocked ()+ let reqId' = 1 :: Int32+ sendMany sock [encodeInt32 RequestConnectionId, encodeInt32 reqId']++-- | Test what happens when a remote endpoint sends a connection request to our+-- transport for an endpoint it already has a connection to+testUnnecessaryConnect :: IO N.ServiceName -> Int -> IO () +testUnnecessaryConnect nextPort numThreads = do+ clientDone <- newEmptyMVar+ serverAddr <- newEmptyMVar++ forkTry $ do+ Right transport <- nextPort >>= \port -> createTransport "127.0.0.1" port defaultTCPParameters+ Right endpoint <- newEndPoint transport+ putMVar serverAddr (address endpoint)++ forkTry $ do+ -- We pick an address < 127.0.0.1 so that this is not rejected purely because of the "crossed" check + let ourAddress = EndPointAddress "126.0.0.1"++ -- We should only get a single 'Accepted' reply+ gotAccepted <- newEmptyMVar+ dones <- replicateM numThreads $ do+ done <- newEmptyMVar + forkTry $ do+ -- It is possible that the remote endpoint just rejects the request by closing the socket+ -- immediately (depending on far the remote endpoint got with the initialization)+ response <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr True Nothing + case response of+ 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+ Left _ ->+ return ()+ putMVar done ()+ return done++ mapM_ readMVar (gotAccepted : dones)+ putMVar clientDone ()++ takeMVar clientDone++-- | Test that we can create "many" transport instances+testMany :: IO N.ServiceName -> IO ()+testMany nextPort = do+ Right masterTransport <- nextPort >>= \port -> createTransport "127.0.0.1" port defaultTCPParameters+ Right masterEndPoint <- newEndPoint masterTransport ++ replicateM_ 10 $ do+ mTransport <- nextPort >>= \port -> createTransport "127.0.0.1" port defaultTCPParameters+ case mTransport of+ Left ex -> do+ putStrLn $ "IOException: " ++ show ex ++ "; errno = " ++ show (ioe_errno ex)+ case (ioe_errno ex) of+ Just no | Errno no == eADDRNOTAVAIL -> putStrLn "(ADDRNOTAVAIL)" + _ -> return ()+ throwIO ex+ Right transport ->+ replicateM_ 2 $ do+ Right endpoint <- newEndPoint transport+ Right _ <- connect endpoint (address masterEndPoint) ReliableOrdered defaultConnectHints+ return ()++-- | Test what happens when the transport breaks completely+testBreakTransport :: IO N.ServiceName -> IO ()+testBreakTransport nextPort = do+ Right (transport, internals) <- nextPort >>= \port -> createTransportExposeInternals "127.0.0.1" port defaultTCPParameters+ Right endpoint <- newEndPoint transport++ killThread (transportThread internals) -- Uh oh++ ErrorEvent (TransportError EventTransportFailed _) <- receive endpoint ++ return ()++-- | 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+-- have yet to set up.+-- Then test that we get a connection lost message after the remote endpoint+-- suddenly closes the socket, and that a subsequent 'connect' allows us to+-- re-establish a connection to the same endpoint+testReconnect :: IO N.ServiceName -> IO ()+testReconnect nextPort = do+ serverPort <- nextPort+ serverDone <- newEmptyMVar+ firstAttempt <- newEmptyMVar+ endpointCreated <- newEmptyMVar++ -- Server+ forkTry $ do + -- Wait for the client to do its first attempt+ readMVar firstAttempt ++ counter <- newMVar (0 :: Int) ++ forkServer "127.0.0.1" serverPort 5 True throwIO $ \sock -> do+ -- Accept the connection + Right 0 <- tryIO $ (recvInt32 sock :: IO Int)+ Right _ <- tryIO $ recvWithLength sock + Right () <- tryIO $ sendMany sock [encodeInt32 ConnectionRequestAccepted]++ -- The first time we close the socket before accepting the logical connection+ count <- modifyMVar counter $ \i -> return (i + 1, i)++ when (count > 0) $ do+ -- Client requests a logical connection+ Right RequestConnectionId <- tryIO $ toEnum <$> (recvInt32 sock :: IO Int)+ Right reqId <- tryIO $ (recvInt32 sock :: IO Int)+ Right () <- tryIO $ sendMany sock (encodeInt32 ControlResponse : encodeInt32 reqId : prependLength [encodeInt32 (10001 :: Int)])+ return ()++ when (count > 1) $ do+ -- Client sends a message+ Right 10001 <- tryIO $ (recvInt32 sock :: IO Int)+ Right ["ping"] <- tryIO $ recvWithLength sock+ putMVar serverDone ()+ + Right () <- tryIO $ N.sClose sock+ return ()++ putMVar endpointCreated ()++ -- Client+ forkTry $ do+ Right transport <- nextPort >>= \port -> createTransport "127.0.0.1" port defaultTCPParameters+ Right endpoint <- newEndPoint transport+ let theirAddr = encodeEndPointAddress "127.0.0.1" serverPort 0++ -- The first attempt will fail because no endpoint is yet set up+ -- Left (TransportError ConnectNotFound _) <- connect endpoint theirAddr ReliableOrdered defaultConnectHints+ putMVar firstAttempt ()++ -- The second attempt will fail because the server closes the socket before we can request a connection+ takeMVar endpointCreated+ -- This might time out or not, depending on whether the server closes the+ -- socket before or after we can send the RequestConnectionId request + resultConnect <- timeout 500000 $ connect endpoint theirAddr ReliableOrdered defaultConnectHints + case resultConnect of+ Nothing -> return ()+ Just (Left (TransportError ConnectFailed _)) -> return ()+ Just (Left err) -> throwIO err+ Just (Right _) -> throwIO $ userError "testConnect: unexpected connect success" ++ -- The third attempt succeeds+ Right conn1 <- connect endpoint theirAddr ReliableOrdered defaultConnectHints+ + -- But a send will fail because the server has closed the connection again+ Left (TransportError SendFailed _) <- send conn1 ["ping"]+ ErrorEvent (TransportError (EventConnectionLost _ []) _) <- receive endpoint++ -- But a subsequent call to connect should reestablish the connection+ Right conn2 <- connect endpoint theirAddr ReliableOrdered defaultConnectHints++ -- Send should now succeed+ Right () <- send conn2 ["ping"]+ return ()++ takeMVar serverDone++-- Test what happens if we close the socket one way only. This means that the+-- 'recv' in 'handleIncomingMessages' will not fail, but a 'send' or 'connect'+-- *will* fail. We are testing that error handling everywhere does the right+-- thing.+testUnidirectionalError :: IO N.ServiceName -> IO ()+testUnidirectionalError nextPort = do+ clientDone <- newEmptyMVar+ serverPort <- nextPort+ serverGotPing <- newEmptyMVar++ -- Server+ forkServer "127.0.0.1" serverPort 5 True throwIO $ \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]++ RequestConnectionId <- toEnum <$> (recvInt32 sock :: IO Int)+ reqId <- recvInt32 sock :: IO Int+ sendMany sock (encodeInt32 ControlResponse : encodeInt32 reqId : prependLength [encodeInt32 (10001 :: Int)])+ + 10001 <- recvInt32 sock :: IO Int+ ["ping"] <- recvWithLength sock+ putMVar serverGotPing ()+ + -- Client+ forkTry $ do+ Right (transport, internals) <- nextPort >>= \port -> createTransportExposeInternals "127.0.0.1" port defaultTCPParameters+ Right endpoint <- newEndPoint transport+ let theirAddr = encodeEndPointAddress "127.0.0.1" serverPort 0++ -- Establish a connection to the server+ Right conn1 <- connect endpoint theirAddr ReliableOrdered defaultConnectHints+ send conn1 ["ping"]+ takeMVar serverGotPing++ -- Close the *outgoing* part of the socket only+ sock <- socketBetween internals (address endpoint) theirAddr + N.shutdown sock N.ShutdownSend++ -- At this point we cannot notice the problem yet so we shouldn't receive an event yet+ Nothing <- timeout 500000 $ receive endpoint+ + -- But when we send we find the error+ Left (TransportError SendFailed _) <- send conn1 ["ping"]+ ErrorEvent (TransportError (EventConnectionLost _ []) _) <- receive endpoint++ -- A call to connect should now re-establish the connection+ Right conn2 <- connect endpoint theirAddr ReliableOrdered defaultConnectHints+ send conn2 ["ping"]+ takeMVar serverGotPing++ -- Again, close the outgoing part of the socket+ sock' <- socketBetween internals (address endpoint) theirAddr + N.shutdown sock' N.ShutdownSend++ -- We now find the error when we attempt to close the connection+ Nothing <- timeout 500000 $ receive endpoint+ close conn2+ ErrorEvent (TransportError (EventConnectionLost _ []) _) <- receive endpoint+ Right conn3 <- connect endpoint theirAddr ReliableOrdered defaultConnectHints+ send conn3 ["ping"]+ takeMVar serverGotPing++ -- We repeat once more. + sock'' <- socketBetween internals (address endpoint) theirAddr + N.shutdown sock'' N.ShutdownSend+ + -- Now we notice the problem when we try to connect+ Nothing <- timeout 500000 $ receive endpoint+ Left (TransportError ConnectFailed _) <- connect endpoint theirAddr ReliableOrdered defaultConnectHints+ ErrorEvent (TransportError (EventConnectionLost _ []) _) <- receive endpoint+ Right conn4 <- connect endpoint theirAddr ReliableOrdered defaultConnectHints+ send conn4 ["ping"]+ takeMVar serverGotPing++ putMVar clientDone ()++ takeMVar clientDone++testInvalidCloseConnection :: IO N.ServiceName -> IO ()+testInvalidCloseConnection nextPort = do+ Right (transport, internals) <- nextPort >>= \port -> createTransportExposeInternals "127.0.0.1" port defaultTCPParameters+ serverAddr <- newEmptyMVar+ clientDone <- newEmptyMVar+ serverDone <- newEmptyMVar++ -- Server+ forkTry $ do+ Right endpoint <- newEndPoint transport+ putMVar serverAddr (address endpoint)++ ConnectionOpened _ _ _ <- receive endpoint++ -- At this point the client sends an invalid request, so we terminate the+ -- connection+ ErrorEvent (TransportError (EventConnectionLost _ [_]) _) <- receive endpoint++ putMVar serverDone () ++ -- Client+ forkTry $ do+ Right endpoint <- newEndPoint transport+ let ourAddr = address endpoint++ -- Connect so that we have a TCP connection+ theirAddr <- readMVar serverAddr+ Right _ <- connect endpoint theirAddr ReliableOrdered defaultConnectHints++ -- Get a handle on the TCP connection and manually send an invalid CloseConnection request+ sock <- socketBetween internals ourAddr theirAddr + sendMany sock [encodeInt32 CloseConnection, encodeInt32 (12345 :: Int)]++ putMVar clientDone ()++ mapM_ takeMVar [clientDone, serverDone]++main :: IO ()+main = do+ portMVar <- newEmptyMVar+ forkTry $ forM_ ([10080 ..] :: [Int]) $ putMVar portMVar . show + let nextPort = takeMVar portMVar + tcpResult <- tryIO $ runTests + [ ("EarlyDisconnect", testEarlyDisconnect nextPort)+ , ("EarlyCloseSocket", testEarlyCloseSocket nextPort)+ , ("IgnoreCloseSocket", testIgnoreCloseSocket nextPort)+ , ("BlockAfterCloseSocket", testBlockAfterCloseSocket nextPort)+ , ("TestUnnecessaryConnect", testUnnecessaryConnect nextPort 10)+ , ("InvalidAddress", testInvalidAddress nextPort)+ , ("InvalidConnect", testInvalidConnect nextPort) + , ("Many", testMany nextPort)+ , ("BreakTransport", testBreakTransport nextPort)+ , ("Reconnect", testReconnect nextPort)+ , ("UnidirectionalError", testUnidirectionalError nextPort)+ , ("InvalidCloseConnection", testInvalidCloseConnection nextPort)+ ]+ -- Run the generic tests even if the TCP specific tests failed.. + testTransport (either (Left . show) (Right) <$> nextPort >>= \port -> createTransport "127.0.0.1" port defaultTCPParameters)+ -- ..but if the generic tests pass, still fail if the specific tests did not+ case tcpResult of+ Left err -> throwIO err+ Right () -> return ()