diff --git a/ChangeLog b/ChangeLog --- a/ChangeLog +++ b/ChangeLog @@ -1,3 +1,17 @@ +2017-08-21 FacundoDominguez 0.6.0 + +* Implemented protocol versioning (#55) +* Extend interface so queue policies and lengths can be configured. +* Check peer-reported host against socket host (#54) +* Test improvements +* Fix races when an EndPoint is closed or failed (#60) +* Fix timeout socket connections (#53) +* Use equality rather than ordering in socket close (#56) +* apiCloseEndPoint blocks until no longer reciving +* Shutdown sockets when closing endpoints +* Allow computing the external address from the chosen bind port (#50) +* Discard remote endpoints when they close or fail (#43) + 2016-02-17 FacundoDominguez 0.5.0 * Add TCP_KEEPALIVE support for sockets. diff --git a/network-transport-tcp.cabal b/network-transport-tcp.cabal --- a/network-transport-tcp.cabal +++ b/network-transport-tcp.cabal @@ -1,5 +1,5 @@ Name: network-transport-tcp -Version: 0.5.1 +Version: 0.6.0 Cabal-Version: >=1.10 Build-Type: Simple License: BSD3 @@ -26,7 +26,7 @@ Library Build-Depends: base >= 4.3 && < 5, - network-transport >= 0.4.1.0 && < 0.5, + network-transport >= 0.5 && < 0.6, data-accessor >= 0.2 && < 0.3, containers >= 0.4 && < 0.6, bytestring >= 0.9 && < 0.11, @@ -47,9 +47,10 @@ Type: exitcode-stdio-1.0 Main-Is: TestTCP.hs Build-Depends: base >= 4.3 && < 5, + bytestring >= 0.9 && < 0.11, network-transport-tests >= 0.2.1.0 && < 0.3, network >= 2.3 && < 2.7, - network-transport >= 0.4.1.0 && < 0.5, + network-transport, network-transport-tcp ghc-options: -threaded -rtsopts -with-rtsopts=-N HS-Source-Dirs: tests diff --git a/src/Network/Transport/TCP.hs b/src/Network/Transport/TCP.hs --- a/src/Network/Transport/TCP.hs +++ b/src/Network/Transport/TCP.hs @@ -13,6 +13,7 @@ {-# LANGUAGE RecursiveDo #-} {-# LANGUAGE TupleSections #-} +{-# LANGUAGE RankNTypes #-} module Network.Transport.TCP ( -- * Main API @@ -23,14 +24,15 @@ , createTransportExposeInternals , TransportInternals(..) , EndPointId - , encodeEndPointAddress - , decodeEndPointAddress , ControlHeader(..) , ConnectionRequestResponse(..) , firstNonReservedLightweightConnectionId , firstNonReservedHeavyweightConnectionId , socketToEndPoint , LightweightConnectionId + , QDisc(..) + , simpleUnboundedQDisc + , simpleOnePlaceQDisc -- * Design notes -- $design ) where @@ -44,14 +46,27 @@ import Network.Transport import Network.Transport.TCP.Internal - ( forkServer + ( ControlHeader(..) + , encodeControlHeader + , decodeControlHeader + , ConnectionRequestResponse(..) + , encodeConnectionRequestResponse + , decodeConnectionRequestResponse + , forkServer , recvWithLength - , recvInt32 + , recvExact + , recvWord32 + , encodeWord32 , tryCloseSocket + , tryShutdownSocketBoth + , decodeSockAddr + , EndPointId + , encodeEndPointAddress + , decodeEndPointAddress + , currentProtocolVersion ) import Network.Transport.Internal - ( encodeInt32 - , prependLength + ( prependLength , mapIOException , tryIO , tryToEnum @@ -80,6 +95,7 @@ , connect , sOMAXCONN , AddrInfo + , SockAddr(..) ) #ifdef USE_MOCK_NETWORK @@ -103,6 +119,7 @@ , modifyMVar , modifyMVar_ , readMVar + , takeMVar , putMVar , newEmptyMVar , withMVar @@ -150,7 +167,7 @@ import qualified Data.Accessor.Container as DAC (mapMaybe) import Data.Foldable (forM_, mapM_) import qualified System.Timeout (timeout) - +import qualified Data.ByteString as BS (length) -- $design -- @@ -264,10 +281,12 @@ -- ValidRemoteEndPointState). data TCPTransport = TCPTransport - { transportHost :: !N.HostName - , transportPort :: !N.ServiceName - , transportState :: !(MVar TransportState) - , transportParams :: !TCPParameters + { transportHost :: !N.HostName + , transportPort :: !N.ServiceName + , transportBindHost :: !N.HostName + , transportBindPort :: !N.ServiceName + , transportState :: !(MVar TransportState) + , transportParams :: !TCPParameters } data TransportState = @@ -281,8 +300,10 @@ data LocalEndPoint = LocalEndPoint { localAddress :: !EndPointAddress - , localChannel :: !(Chan Event) , localState :: !(MVar LocalEndPointState) + -- | A 'QDisc' is held here rather than on the 'ValidLocalEndPointState' + -- because even closed 'LocalEndPoint's can have queued input data. + , localQueue :: !(QDisc Event) } data LocalEndPointState = @@ -414,18 +435,20 @@ data ValidRemoteEndPointState = ValidRemoteEndPointState { _remoteOutgoing :: !Int , _remoteIncoming :: !(Set LightweightConnectionId) - , _remoteMaxIncoming :: !LightweightConnectionId + , _remoteLastIncoming :: !LightweightConnectionId , _remoteNextConnOutId :: !LightweightConnectionId , remoteSocket :: !N.Socket -- | When the connection is being probed, yields an IO action that can be -- used to release any resources dedicated to the probing. , remoteProbing :: Maybe (IO ()) , remoteSendLock :: !(MVar ()) + -- | An IO which returns when the socket (remoteSocket) has been closed. + -- The program/thread which created the socket is always responsible + -- for closing it, but sometimes other threads need to know when this + -- happens. + , remoteSocketClosed :: !(IO ()) } --- | Local identifier for an endpoint within this transport -type EndPointId = Word32 - -- | Pair of local and a remote endpoint (for conciseness in signatures) type EndPointPair = (LocalEndPoint, RemoteEndPoint) @@ -441,30 +464,6 @@ -- 'LightweightConnectionId'. type HeavyweightConnectionId = Word32 --- | Control headers -data ControlHeader = - -- | Tell the remote endpoint that we created a new connection - CreatedNewConnection - -- | Tell the remote endpoint we will no longer be using a connection - | CloseConnection - -- | Request to close the connection (see module description) - | CloseSocket - -- | Message sent to probe a socket - | ProbeSocket - -- | Acknowledgement of the ProbeSocket message - | ProbeSocketAck - deriving (Enum, Bounded, Show) - --- | Response sent by /B/ to /A/ when /A/ tries to connect -data ConnectionRequestResponse = - -- | /B/ accepts the connection - ConnectionRequestAccepted - -- | /A/ requested an invalid endpoint - | ConnectionRequestInvalid - -- | /A/s request crossed with a request from /B/ (see protocols) - | ConnectionRequestCrossed - deriving (Enum, Bounded, Show) - -- | Parameters for setting up the TCP transport data TCPParameters = TCPParameters { -- | Backlog for 'listen'. @@ -479,26 +478,51 @@ -- | Should we set TCP_NODELAY on connection sockets? -- Defaults to True. , tcpNoDelay :: Bool - -- | Value of TCP_USER_TIMEOUT in milliseconds - , tcpKeepAlive :: Bool -- | Should we set TCP_KEEPALIVE on connection sockets? + , tcpKeepAlive :: Bool + -- | Value of TCP_USER_TIMEOUT in milliseconds , tcpUserTimeout :: Maybe Int -- | A connect timeout for all 'connect' calls of the transport -- in microseconds -- -- This can be overriden for each connect call with -- 'ConnectHints'.'connectTimeout'. + -- + -- Connection requests to this transport will also timeout if they don't + -- send the required data before this many microseconds. , transportConnectTimeout :: Maybe Int + -- | Create a QDisc for an EndPoint. + , tcpNewQDisc :: forall t . IO (QDisc t) + -- | Maximum length (in bytes) for a peer's address. + -- If a peer attempts to send an address of length exceeding the limit, + -- the connection will be refused (socket will close). + , tcpMaxAddressLength :: Word32 + -- | Maximum length (in bytes) to receive from a peer. + -- If a peer attempts to send data on a lightweight connection exceeding + -- the limit, the heavyweight connection which carries that lightweight + -- connection will go down. The peer and the local node will get an + -- EventConnectionLost. + , tcpMaxReceiveLength :: Word32 + -- | If True, new connections will be accepted only if the socket's host + -- matches the host that the peer claims in its EndPointAddress. + -- This is useful when operating on untrusted networks, because the peer + -- could otherwise deny service to some victim by claiming the victim's + -- address. + , tcpCheckPeerHost :: Bool } -- | Internal functionality we expose for unit testing data TransportInternals = TransportInternals { -- | The ID of the thread that listens for new incoming connections - transportThread :: ThreadId + transportThread :: ThreadId + -- | A variant of newEndPoint in which the QDisc determined by the + -- transport's TCPParameters can be optionally overridden. + , newEndPointInternal :: (forall t . Maybe (QDisc t)) + -> IO (Either (TransportError NewEndPointErrorCode) EndPoint) -- | Find the socket between a local and a remote endpoint - , socketBetween :: EndPointAddress - -> EndPointAddress - -> IO N.Socket + , socketBetween :: EndPointAddress + -> EndPointAddress + -> IO N.Socket } -------------------------------------------------------------------------------- @@ -506,20 +530,25 @@ -------------------------------------------------------------------------------- -- | Create a TCP transport -createTransport :: N.HostName - -> N.ServiceName +createTransport :: N.HostName -- ^ Bind host name. + -> N.ServiceName -- ^ Bind port. + -> (N.ServiceName -> (N.HostName, N.ServiceName)) + -- ^ External address host name and port, computed from the + -- actual bind port. -> TCPParameters -> IO (Either IOException Transport) -createTransport host port params = - either Left (Right . fst) <$> createTransportExposeInternals host port params +createTransport bindHost bindPort mkExternal params = + either Left (Right . fst) <$> + createTransportExposeInternals bindHost bindPort mkExternal params -- | You should probably not use this function (used for unit testing only) createTransportExposeInternals :: N.HostName -> N.ServiceName + -> (N.ServiceName -> (N.HostName, N.ServiceName)) -> TCPParameters -> IO (Either IOException (Transport, TransportInternals)) -createTransportExposeInternals host port params = do +createTransportExposeInternals bindHost bindPort mkExternal params = do state <- newMVar . TransportValid $ ValidTransportState { _localEndPoints = Map.empty , _nextEndPointId = 0 @@ -535,16 +564,20 @@ -- completes (see description of 'forkServer'), yet we need the port to -- construct a transport. So we tie a recursive knot. (port', result) <- do - let transport = TCPTransport { transportState = state - , transportHost = host - , transportPort = port' - , transportParams = params + let (externalHost, externalPort) = mkExternal port' + let transport = TCPTransport { transportState = state + , transportHost = externalHost + , transportPort = externalPort + , transportBindHost = bindHost + , transportBindPort = port' + , transportParams = params } bracketOnError (forkServer - host - port + bindHost + bindPort (tcpBacklog params) (tcpReuseServerAddr params) + (errorHandler transport) (terminationHandler transport) (handleConnectionRequest transport)) (\(_port', tid) -> killThread tid) @@ -554,20 +587,29 @@ 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 - } - ) + mkTransport transport tid = do + return + ( Transport + { newEndPoint = do + qdisc <- tcpNewQDisc params + apiNewEndPoint transport qdisc + , closeTransport = let evs = [ EndPointClosed ] + in apiCloseTransport transport (Just tid) evs + } + , TransportInternals + { transportThread = tid + , socketBetween = internalSocketBetween transport + , newEndPointInternal = \mqdisc -> case mqdisc of + Just qdisc -> apiNewEndPoint transport qdisc + Nothing -> do + qdisc <- tcpNewQDisc params + apiNewEndPoint transport qdisc + } + ) + errorHandler :: TCPTransport -> SomeException -> IO () + errorHandler _ = throwIO + terminationHandler :: TCPTransport -> SomeException -> IO () terminationHandler transport ex = do let evs = [ ErrorEvent (TransportError EventTransportFailed (show ex)) @@ -584,7 +626,11 @@ , tcpNoDelay = False , tcpKeepAlive = False , tcpUserTimeout = Nothing + , tcpNewQDisc = simpleUnboundedQDisc , transportConnectTimeout = Nothing + , tcpMaxAddressLength = maxBound + , tcpMaxReceiveLength = maxBound + , tcpCheckPeerHost = False } -------------------------------------------------------------------------------- @@ -606,18 +652,17 @@ -- | Create a new endpoint apiNewEndPoint :: TCPTransport + -> QDisc Event -> IO (Either (TransportError NewEndPointErrorCode) EndPoint) -apiNewEndPoint transport = +apiNewEndPoint transport qdisc = try . asyncWhenCancelled closeEndPoint $ do - ourEndPoint <- createLocalEndPoint transport + ourEndPoint <- createLocalEndPoint transport qdisc return EndPoint - { receive = readChan (localChannel ourEndPoint) + { receive = qdiscDequeue (localQueue ourEndPoint) , address = localAddress ourEndPoint , connect = apiConnect (transportParams transport) ourEndPoint - , closeEndPoint = let evs = [ EndPointClosed - , throw $ userError "Endpoint closed" - ] in - apiCloseEndPoint transport evs ourEndPoint + , closeEndPoint = let evs = [ EndPointClosed ] + in apiCloseEndPoint transport evs ourEndPoint , newMulticastGroup = return . Left $ newMulticastGroupError , resolveMulticastGroup = return . Left . const resolveMulticastGroupError } @@ -627,6 +672,75 @@ resolveMulticastGroupError = TransportError ResolveMulticastGroupUnsupported "Multicast not supported" +-- | Abstraction of a queue for an 'EndPoint'. +-- +-- A value of type @QDisc t@ is a queue of events of an abstract type @t@. +-- +-- This specifies which 'Event's will come from +-- 'receive :: EndPoint -> IO Event' and when. It is highly general so that +-- the simple yet potentially very fast implementation backed by a single +-- unbounded channel can be used, without excluding more nuanced policies +-- like class-based queueing with bounded buffers for each peer, which may be +-- faster in certain conditions but probably has lower maximal throughput. +-- +-- A 'QDisc' must satisfy some properties in order for the semantics of +-- network-transport to hold true. In general, an event fed with +-- 'qdiscEnqueue' must not be dropped. i.e. provided that no other event in +-- the QDisc has higher priority, the event should eventually be returned by +-- 'qdiscDequeue'. An exception to this are 'Receive' events of unreliable +-- connections. +-- +-- Every call to 'receive' is just 'qdiscDequeue' on that 'EndPoint's +-- 'QDisc'. Whenever an event arises from a socket, `qdiscEnqueue` is called +-- with the relevant metadata in the same thread that reads from the socket. +-- You can be clever about when to block here, so as to control network +-- ingress. This applies also to loopback connections (an 'EndPoint' connects +-- to itself), in which case blocking on the enqueue would only block some +-- thread in your program rather than some chatty network peer. The 'Event' +-- which is to be enqueued is given to 'qdiscEnqueue' so that the 'QDisc' +-- can know about open connections, their identifiers and peer addresses, etc. +data QDisc t = QDisc { + -- | Dequeue an event. + qdiscDequeue :: IO t + -- | @qdiscEnqueue ep ev t@ enqueues and event @t@, originated from the + -- given remote endpoint @ep@ and with data @ev@. + -- + -- @ep@ might be the local endpoint if it relates to a self-connection. + -- + -- @ev@ might be in practice the value given as @t@. It is passed in + -- the abstract form @t@ to enforce it is dequeued unmodified, but the + -- 'QDisc' implementation can still observe the concrete form @ev@ to + -- make prioritization decisions. + , qdiscEnqueue :: EndPointAddress -> Event -> t -> IO () + } + +-- | Post an 'Event' using a 'QDisc'. +qdiscEnqueue' :: QDisc Event -> EndPointAddress -> Event -> IO () +qdiscEnqueue' qdisc addr event = qdiscEnqueue qdisc addr event event + +-- | A very simple QDisc backed by an unbounded channel. +simpleUnboundedQDisc :: forall t . IO (QDisc t) +simpleUnboundedQDisc = do + eventChan <- newChan + return $ QDisc { + qdiscDequeue = readChan eventChan + , qdiscEnqueue = const (const (writeChan eventChan)) + } + +-- | A very simple QDisc backed by a 1-place queue (MVar). +-- With this QDisc, all threads reading from sockets will try to put their +-- events into the same MVar. That MVar will be cleared by calls to +-- 'receive'. Thus the rate at which data is read from the wire is directly +-- related to the rate at which data is pulled from the EndPoint by +-- 'receive'. +simpleOnePlaceQDisc :: forall t . IO (QDisc t) +simpleOnePlaceQDisc = do + mvar <- newEmptyMVar + return $ QDisc { + qdiscDequeue = takeMVar mvar + , qdiscEnqueue = const (const (putMVar mvar)) + } + -- | Connnect to an endpoint apiConnect :: TCPParameters -- ^ Parameters -> LocalEndPoint -- ^ Local end point @@ -662,7 +776,10 @@ then do writeIORef connAlive False sched theirEndPoint $ - sendOn vst [encodeInt32 CloseConnection, encodeInt32 connId] + sendOn vst [ + encodeWord32 (encodeControlHeader CloseConnection) + , encodeWord32 connId + ] return ( RemoteEndPointValid . (remoteOutgoing ^: (\x -> x - 1)) $ vst @@ -692,7 +809,7 @@ alive <- readIORef connAlive if alive then sched theirEndPoint $ - sendOn vst (encodeInt32 connId : prependLength payload) + sendOn vst (encodeWord32 connId : prependLength payload) else throwIO $ TransportError SendClosed "Connection closed" RemoteEndPointClosing _ _ -> do alive <- readIORef connAlive @@ -730,7 +847,8 @@ return (LocalEndPointClosed, Nothing) forM_ mOurState $ \vst -> do forM_ (vst ^. localConnections) tryCloseRemoteSocket - forM_ evs $ writeChan (localChannel ourEndPoint) + let qdisc = localQueue ourEndPoint + forM_ evs (qdiscEnqueue' qdisc (localAddress ourEndPoint)) where -- Close the remote socket and return the set of all incoming connections tryCloseRemoteSocket :: RemoteEndPoint -> IO () @@ -746,19 +864,32 @@ putMVar resolved () return closed RemoteEndPointValid vst -> do + -- Schedule an action to send a CloseEndPoint message and then + -- wait for the socket to actually close (meaning that this + -- end point is no longer receiving from it). + -- Since we replace the state in this MVar with 'closed', it's + -- guaranteed that no other actions will be scheduled after this + -- one. sched theirEndPoint $ do - void $ tryIO $ sendOn vst [ encodeInt32 CloseSocket - , encodeInt32 (vst ^. remoteMaxIncoming) - ] + void $ tryIO $ sendOn vst + [ encodeWord32 (encodeControlHeader CloseEndPoint) ] -- Release probing resources if probing. forM_ (remoteProbing vst) id - tryCloseSocket (remoteSocket vst) + tryShutdownSocketBoth (remoteSocket vst) + remoteSocketClosed vst return closed RemoteEndPointClosing resolved vst -> do -- Release probing resources if probing. forM_ (remoteProbing vst) id putMVar resolved () - sched theirEndPoint $ tryCloseSocket (remoteSocket vst) + -- Schedule an action to wait for the socket to actually close (this + -- end point is no longer receiving from it). + -- Since we replace the state in this MVar with 'closed', it's + -- guaranteed that no other actions will be scheduled after this + -- one. + sched theirEndPoint $ do + tryShutdownSocketBoth (remoteSocket vst) + remoteSocketClosed vst return closed RemoteEndPointClosed -> return st @@ -781,128 +912,197 @@ -- the transport down. We must be careful to close the socket when a (possibly -- asynchronous, ThreadKilled) exception occurs. (If an exception escapes from -- handleConnectionRequest the transport will be shut down.) -handleConnectionRequest :: TCPTransport -> N.Socket -> IO () -handleConnectionRequest transport sock = handle handleException $ do +handleConnectionRequest :: TCPTransport -> IO () -> (N.Socket, N.SockAddr) -> IO () +handleConnectionRequest transport socketClosed (sock, sockAddr) = handle handleException $ do when (tcpNoDelay $ transportParams transport) $ N.setSocketOption sock N.NoDelay 1 when (tcpKeepAlive $ transportParams transport) $ N.setSocketOption sock N.KeepAlive 1 forM_ (tcpUserTimeout $ transportParams transport) $ N.setSocketOption sock N.UserTimeout - 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 + let handleVersioned = do + -- Always receive the protocol version and a handshake (content of the + -- handshake is version-dependent, but the length is always sent, + -- regardless of the version). + protocolVersion <- recvWord32 sock + handshakeLength <- recvWord32 sock + -- For now we support only version 0.0.0.0. + case protocolVersion of + 0x00000000 -> handleConnectionRequestV0 (sock, sockAddr) + _ -> do + -- Inform the peer that we want version 0x00000000 + sendMany sock [ + encodeWord32 (encodeConnectionRequestResponse ConnectionRequestUnsupportedVersion) + , encodeWord32 0x00000000 + ] + -- Clear the socket of the unsupported handshake data. + _ <- recvExact sock handshakeLength + handleVersioned + -- The handshake must complete within the optional timeout duration. + -- No socket 'recv's are to be run outside the timeout. The continuation + -- returned may 'send', but not 'recv'. + let connTimeout = transportConnectTimeout (transportParams transport) + outcome <- maybe (fmap Just) System.Timeout.timeout connTimeout handleVersioned + case outcome of + Nothing -> throwIO (userError "handleConnectionRequest: timed out") + Just act -> forM_ act id + where - go :: LocalEndPoint -> EndPointAddress -> IO () - go ourEndPoint theirAddress = do - -- This runs in a thread that will never be killed - mEndPoint <- handle ((>> return Nothing) . handleException) $ do + + handleException :: SomeException -> IO () + handleException ex = do + rethrowIfAsync (fromException ex) + + rethrowIfAsync :: Maybe AsyncException -> IO () + rethrowIfAsync = mapM_ throwIO + + handleConnectionRequestV0 :: (N.Socket, N.SockAddr) -> IO (Maybe (IO ())) + handleConnectionRequestV0 (sock, sockAddr) = do + -- Get the OS-determined host and port. + (actualHost, actualPort) <- + decodeSockAddr sockAddr >>= + maybe (throwIO (userError "handleConnectionRequest: invalid socket address")) return + (ourEndPointId, theirAddress) <- do + ourEndPointId <- recvWord32 sock + let maxAddressLength = tcpMaxAddressLength $ transportParams transport + theirAddress <- EndPointAddress . BS.concat <$> + recvWithLength maxAddressLength sock + return (ourEndPointId, theirAddress) + let ourAddress = encodeEndPointAddress (transportHost transport) + (transportPort transport) + ourEndPointId + (theirHost, _, _) + <- maybe (throwIO (userError "handleConnectionRequest: peer gave malformed address")) + return + (decodeEndPointAddress theirAddress) + let checkPeerHost = tcpCheckPeerHost (transportParams transport) + if checkPeerHost && (theirHost /= actualHost) + then do + -- If the OS-determined host doesn't match the host that the peer gave us, + -- then we have no choice but to reject the connection. It's because we + -- use the EndPointAddress to key the remote end points (localConnections) + -- and we don't want to allow a peer to deny service to other peers by + -- claiming to have their host and port. + sendMany sock $ + encodeWord32 (encodeConnectionRequestResponse ConnectionRequestHostMismatch) + : prependLength [BSC.pack actualHost] + return Nothing + else do + ourEndPoint <- withMVar (transportState transport) $ \st -> case st of + TransportValid vst -> + case vst ^. localEndPointAt ourAddress of + Nothing -> do + sendMany sock [encodeWord32 (encodeConnectionRequestResponse ConnectionRequestInvalid)] + throwIO $ userError "handleConnectionRequest: Invalid endpoint" + Just ourEndPoint -> + return ourEndPoint + TransportClosed -> + throwIO $ userError "Transport closed" + return (Just (go ourEndPoint theirAddress)) + + where + + go :: LocalEndPoint -> EndPointAddress -> IO () + go ourEndPoint theirAddress = handle handleException $ do resetIfBroken ourEndPoint theirAddress (theirEndPoint, isNew) <- findRemoteEndPoint ourEndPoint theirAddress RequestedByThem Nothing if not isNew then do - void $ tryIO $ sendMany sock [encodeInt32 ConnectionRequestCrossed] + void $ tryIO $ sendMany sock + [encodeWord32 (encodeConnectionRequestResponse ConnectionRequestCrossed)] probeIfValid theirEndPoint - tryCloseSocket sock - return Nothing else do sendLock <- newMVar () let vst = ValidRemoteEndPointState { remoteSocket = sock + , remoteSocketClosed = socketClosed , remoteProbing = Nothing , remoteSendLock = sendLock , _remoteOutgoing = 0 , _remoteIncoming = Set.empty - , _remoteMaxIncoming = 0 + , _remoteLastIncoming = 0 , _remoteNextConnOutId = firstNonReservedLightweightConnectionId } - sendMany sock [encodeInt32 ConnectionRequestAccepted] + sendMany sock [encodeWord32 (encodeConnectionRequestResponse ConnectionRequestAccepted)] + -- resolveInit will update the shared state, and handleIncomingMessages + -- will always ultimately clean up after it. + -- Closing up the socket is also out of our hands. It will happen + -- when handleIncomingMessages finishes. resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst) - 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 + `finally` + handleIncomingMessages (transportParams transport) (ourEndPoint, theirEndPoint) - probeIfValid :: RemoteEndPoint -> IO () - probeIfValid theirEndPoint = modifyMVar_ (remoteState theirEndPoint) $ - \st -> case st of - RemoteEndPointValid - vst@(ValidRemoteEndPointState { remoteProbing = Nothing }) -> do - tid <- forkIO $ do - -- send probe - let params = transportParams transport - void $ tryIO $ System.Timeout.timeout - (maybe (-1) id $ transportConnectTimeout params) $ do - sendMany (remoteSocket vst) [encodeInt32 ProbeSocket] - threadDelay maxBound - -- Discard the connection if this thread is not killed (i.e. the - -- probe ack does not arrive on time). - -- - -- The thread handling incoming messages will detect the socket is - -- closed and will report the failure upwards. - tryCloseSocket (remoteSocket vst) - -- Waiting the probe ack and closing the socket is only needed in - -- platforms where TCP_USER_TIMEOUT is not available or when the - -- user does not set it. Otherwise the ack would be handled at the - -- TCP level and the the thread handling incoming messages would - -- get the error. + probeIfValid :: RemoteEndPoint -> IO () + probeIfValid theirEndPoint = modifyMVar_ (remoteState theirEndPoint) $ + \st -> case st of + RemoteEndPointValid + vst@(ValidRemoteEndPointState { remoteProbing = Nothing }) -> do + tid <- forkIO $ do + -- send probe + let params = transportParams transport + void $ tryIO $ System.Timeout.timeout + (maybe (-1) id $ transportConnectTimeout params) $ do + sendMany (remoteSocket vst) + [encodeWord32 (encodeControlHeader ProbeSocket)] + threadDelay maxBound + -- Discard the connection if this thread is not killed (i.e. the + -- probe ack does not arrive on time). + -- + -- The thread handling incoming messages will detect the socket is + -- closed and will report the failure upwards. + tryCloseSocket (remoteSocket vst) + -- Waiting the probe ack and closing the socket is only needed in + -- platforms where TCP_USER_TIMEOUT is not available or when the + -- user does not set it. Otherwise the ack would be handled at the + -- TCP level and the the thread handling incoming messages would + -- get the error. - return $ RemoteEndPointValid - vst { remoteProbing = Just (killThread tid) } - _ -> return st + return $ RemoteEndPointValid + vst { remoteProbing = Just (killThread tid) } + _ -> return st -- | Handle requests from a remote endpoint. -- -- Returns only if the remote party closes the socket or if an error occurs. -- This runs in a thread that will never be killed. -handleIncomingMessages :: EndPointPair -> IO () -handleIncomingMessages (ourEndPoint, theirEndPoint) = do - mSock <- withMVar theirState $ \st -> - case st of - RemoteEndPointInvalid _ -> - relyViolation (ourEndPoint, theirEndPoint) - "handleIncomingMessages (invalid)" - RemoteEndPointInit _ _ _ -> - relyViolation (ourEndPoint, theirEndPoint) - "handleIncomingMessages (init)" - RemoteEndPointValid ep -> - return . Just $ remoteSocket ep - RemoteEndPointClosing _ ep -> - return . Just $ remoteSocket ep - RemoteEndPointClosed -> - return Nothing - RemoteEndPointFailed _ -> - return Nothing +handleIncomingMessages :: TCPParameters -> EndPointPair -> IO () +handleIncomingMessages params (ourEndPoint, theirEndPoint) = + bracket acquire release act - forM_ mSock $ \sock -> - tryIO (go sock) >>= either (prematureExit sock) return where + + -- Use shared remote endpoint state to get a socket, or an appropriate + -- exception in case it's neither valid nor closing. + acquire :: IO (Either IOError N.Socket) + acquire = withMVar theirState $ \st -> case st of + RemoteEndPointInvalid _ -> + relyViolation (ourEndPoint, theirEndPoint) + "handleIncomingMessages (invalid)" + RemoteEndPointInit _ _ _ -> + relyViolation (ourEndPoint, theirEndPoint) + "handleIncomingMessages (init)" + RemoteEndPointValid ep -> + return . Right $ remoteSocket ep + RemoteEndPointClosing _ ep -> + return . Right $ remoteSocket ep + RemoteEndPointClosed -> + return . Left $ userError "handleIncomingMessages (already closed)" + RemoteEndPointFailed _ -> + return . Left $ userError "handleIncomingMessages (failed)" + + -- 'Right' is the normal case in which there still is a live socket to + -- the remote endpoint, and so 'act' was run and installed its own + -- exception handler. + release :: Either IOError N.Socket -> IO () + release (Left err) = prematureExit err + release (Right _) = return () + + act :: Either IOError N.Socket -> IO () + act (Left _) = return () + act (Right sock) = go sock `catch` prematureExit + -- Dispatch -- -- If a recv throws an exception this will be caught top-level and @@ -913,24 +1113,46 @@ -- exception thrown by 'recv'. go :: N.Socket -> IO () go sock = do - lcid <- recvInt32 sock :: IO LightweightConnectionId + lcid <- recvWord32 sock :: IO LightweightConnectionId if lcid >= firstNonReservedLightweightConnectionId then do readMessage sock lcid go sock else - case tryToEnum (fromIntegral lcid) of + case decodeControlHeader lcid of Just CreatedNewConnection -> do - recvInt32 sock >>= createdNewConnection + recvWord32 sock >>= createdNewConnection go sock Just CloseConnection -> do - recvInt32 sock >>= closeConnection + recvWord32 sock >>= closeConnection go sock Just CloseSocket -> do - didClose <- recvInt32 sock >>= closeSocket sock + didClose <- recvWord32 sock >>= closeSocket sock unless didClose $ go sock + Just CloseEndPoint -> do + let closeRemoteEndPoint vst = do + forM_ (remoteProbing vst) id + -- close incoming connections + forM_ (Set.elems $ vst ^. remoteIncoming) $ + qdiscEnqueue' ourQueue theirAddr . ConnectionClosed . connId + -- report the endpoint as gone if we have any outgoing + -- connections + when (vst ^. remoteOutgoing > 0) $ do + let code = EventConnectionLost (remoteAddress theirEndPoint) + qdiscEnqueue' ourQueue theirAddr . ErrorEvent $ + TransportError code "The remote endpoint was closed." + removeRemoteEndPoint (ourEndPoint, theirEndPoint) + modifyMVar_ theirState $ \s -> case s of + RemoteEndPointValid vst -> do + closeRemoteEndPoint vst + return RemoteEndPointClosed + RemoteEndPointClosing resolved vst -> do + closeRemoteEndPoint vst + putMVar resolved () + return RemoteEndPointClosed + _ -> return s Just ProbeSocket -> do - forkIO $ sendMany sock [encodeInt32 ProbeSocketAck] + forkIO $ sendMany sock [encodeWord32 (encodeControlHeader ProbeSocketAck)] go sock Just ProbeSocketAck -> do stopProbing @@ -951,7 +1173,7 @@ "handleIncomingMessages:createNewConnection (init)" RemoteEndPointValid vst -> return ( (remoteIncoming ^: Set.insert lcid) - $ (remoteMaxIncoming ^= lcid) + $ (remoteLastIncoming ^= lcid) vst ) RemoteEndPointClosing resolved vst -> do @@ -963,7 +1185,7 @@ -- RemoteEndPointValid putMVar resolved () return ( (remoteIncoming ^= Set.singleton lcid) - . (remoteMaxIncoming ^= lcid) + . (remoteLastIncoming ^= lcid) $ vst ) RemoteEndPointFailed err -> @@ -972,7 +1194,7 @@ relyViolation (ourEndPoint, theirEndPoint) "createNewConnection (closed)" return (RemoteEndPointValid vst) - writeChan ourChannel (ConnectionOpened (connId lcid) ReliableOrdered theirAddr) + qdiscEnqueue' ourQueue theirAddr (ConnectionOpened (connId lcid) ReliableOrdered theirAddr) -- Close a connection -- It is important that we verify that the connection is in fact open, @@ -1000,7 +1222,7 @@ throwIO err RemoteEndPointClosed -> relyViolation (ourEndPoint, theirEndPoint) "closeConnection (closed)" - writeChan ourChannel (ConnectionClosed (connId lcid)) + qdiscEnqueue' ourQueue theirAddr (ConnectionClosed (connId lcid)) -- Close the socket (if we don't have any outgoing connections) closeSocket :: N.Socket -> LightweightConnectionId -> IO Bool @@ -1018,16 +1240,28 @@ -- remote endpoint to indicate that all its connections to us are -- now properly closed forM_ (Set.elems $ vst ^. remoteIncoming) $ - writeChan ourChannel . ConnectionClosed . connId + qdiscEnqueue' ourQueue theirAddr . ConnectionClosed . connId let vst' = remoteIncoming ^= Set.empty $ vst - -- If we still have outgoing connections then we ignore the - -- CloseSocket request (we sent a ConnectionCreated message to the - -- remote endpoint, but it did not receive it before sending the - -- CloseSocket request). Similarly, if lastReceivedId < lastSentId - -- then we sent a ConnectionCreated *AND* a ConnectionClosed - -- message to the remote endpoint, *both of which* it did not yet - -- receive before sending the CloseSocket request. - if vst' ^. remoteOutgoing > 0 || lastReceivedId < lastSentId vst + -- The peer sends the connection id of the last connection which + -- they accepted from us. + -- + -- If it's not the same as the id of the last connection that we + -- have made to them (assuming we haven't cycled through all + -- identifiers so fast) then they hadn't seen the request before + -- they tried to close the socket. In that case, we don't close the + -- socket. They'll see our in-flight connection request and then + -- abandon their attempt to close the socket. + -- + -- It's possible that a local connection is coming up but has not + -- yet sent CreatedNewConnection (see createConnectionTo), in + -- which case remoteOutgoing is positive, but the sent and received + -- ids do match. In this case we don't close the socket, because + -- that connection will soon sent the message and bump the lastSentId. + -- + -- Both disjuncts are needed: it's possible that remoteOutgoing is + -- 0 and the ids do not match, in case we have created and closed + -- a connection but the peer has not yet heard of it. + if vst ^. remoteOutgoing > 0 || lastReceivedId /= lastSentId vst then return (RemoteEndPointValid vst', Nothing) else do @@ -1036,10 +1270,10 @@ removeRemoteEndPoint (ourEndPoint, theirEndPoint) -- Attempt to reply (but don't insist) act <- schedule theirEndPoint $ do - void $ tryIO $ sendOn vst' [ encodeInt32 CloseSocket - , encodeInt32 (vst ^. remoteMaxIncoming) - ] - tryCloseSocket sock + void $ tryIO $ sendOn vst' + [ encodeWord32 (encodeControlHeader CloseSocket) + , encodeWord32 (vst ^. remoteLastIncoming) + ] return (RemoteEndPointClosed, Just act) RemoteEndPointClosing resolved vst -> do -- Like above, we need to check if there is a ConnectionCreated @@ -1047,16 +1281,26 @@ -- received. However, since we are in 'closing' state, the only -- way this may happen is when we sent a ConnectionCreated, -- ConnectionClosed, and CloseSocket message, none of which have - -- yet been received. We leave the endpoint in closing state in - -- that case. - if lastReceivedId < lastSentId vst + -- yet been received. It's sufficient to check that the peer has + -- not seen the ConnectionCreated message. In case they have seen + -- it (so that lastReceivedId == lastSendId vst) then they must + -- have seen the other messages or else they would not have sent + -- CloseSocket. + -- We leave the endpoint in closing state in that case. + if lastReceivedId /= lastSentId vst then do return (RemoteEndPointClosing resolved vst, Nothing) else do -- Release probing resources if probing. + when (vst ^. remoteOutgoing > 0) $ do + let code = EventConnectionLost (remoteAddress theirEndPoint) + let msg = "socket closed prematurely by peer" + qdiscEnqueue' ourQueue theirAddr . ErrorEvent $ TransportError code msg forM_ (remoteProbing vst) id removeRemoteEndPoint (ourEndPoint, theirEndPoint) - act <- schedule theirEndPoint $ tryCloseSocket sock + -- Nothing to do, but we want to indicate that the socket + -- really did close. + act <- schedule theirEndPoint $ return () putMVar resolved () return (RemoteEndPointClosed, Just act) RemoteEndPointFailed err -> @@ -1075,7 +1319,8 @@ -- overhead readMessage :: N.Socket -> LightweightConnectionId -> IO () readMessage sock lcid = - recvWithLength sock >>= writeChan ourChannel . Received (connId lcid) + recvWithLength recvLimit sock >>= + qdiscEnqueue' ourQueue theirAddr . Received (connId lcid) -- Stop probing a connection as a result of receiving a probe ack. stopProbing :: IO () @@ -1087,14 +1332,15 @@ _ -> return st -- Arguments - ourChannel = localChannel ourEndPoint + ourQueue = localQueue ourEndPoint + ourState = localState ourEndPoint theirState = remoteState theirEndPoint theirAddr = remoteAddress theirEndPoint + recvLimit = tcpMaxReceiveLength params -- Deal with a premature exit - prematureExit :: N.Socket -> IOException -> IO () - prematureExit sock err = do - tryCloseSocket sock + prematureExit :: IOException -> IO () + prematureExit err = do modifyMVar_ theirState $ \st -> case st of RemoteEndPointInvalid _ -> @@ -1107,7 +1353,7 @@ -- Release probing resources if probing. forM_ (remoteProbing vst) id let code = EventConnectionLost (remoteAddress theirEndPoint) - writeChan ourChannel . ErrorEvent $ TransportError code (show err) + qdiscEnqueue' ourQueue theirAddr . ErrorEvent $ TransportError code (show err) return (RemoteEndPointFailed err) RemoteEndPointClosing resolved vst -> do -- Release probing resources if probing. @@ -1117,7 +1363,19 @@ RemoteEndPointClosed -> relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:prematureExit" - RemoteEndPointFailed err' -> + RemoteEndPointFailed err' -> do + -- Here we post a connection-lost event, but only if the + -- local endpoint is not closed; if it's closed, the EndPointClosed + -- event will be posted without connection-lost events, and this is + -- part of the network-transport specification (there's a test + -- case for it). + modifyMVar_ ourState $ \st' -> case st' of + LocalEndPointClosed -> return st' + LocalEndPointValid _ -> do + let code = EventConnectionLost (remoteAddress theirEndPoint) + err = TransportError code (show err') + qdiscEnqueue' ourQueue theirAddr (ErrorEvent err) + return st' return (RemoteEndPointFailed err') -- Construct a connection ID @@ -1190,7 +1448,10 @@ RemoteEndPointValid vst -> do let connId = vst ^. remoteNextConnOutId act <- schedule theirEndPoint $ do - sendOn vst [encodeInt32 CreatedNewConnection, encodeInt32 connId] + sendOn vst [ + encodeWord32 (encodeControlHeader CreatedNewConnection) + , encodeWord32 connId + ] return connId return ( RemoteEndPointValid $ remoteNextConnOutId ^= connId + 1 @@ -1229,25 +1490,39 @@ (tcpUserTimeout params) connTimeout didAccept <- case result of - Right (sock, ConnectionRequestAccepted) -> do + -- Since a socket was created, we are now responsible for closing it. + -- + -- In case the connection was accepted, we have some work to do. + -- We'll remember how to wait for the socket to close + -- (readMVar socketClosedVar), and we'll take care of closing it up + -- once handleIncomingMessages has finished. + Right (socketClosedVar, sock, ConnectionRequestAccepted) -> do sendLock <- newMVar () let vst = ValidRemoteEndPointState { remoteSocket = sock + , remoteSocketClosed = readMVar socketClosedVar , remoteProbing = Nothing , remoteSendLock = sendLock , _remoteOutgoing = 0 , _remoteIncoming = Set.empty - , _remoteMaxIncoming = 0 + , _remoteLastIncoming = 0 , _remoteNextConnOutId = firstNonReservedLightweightConnectionId } resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst) - return True - Right (sock, ConnectionRequestInvalid) -> do + return (Just (socketClosedVar, sock)) + Right (socketClosedVar, sock, ConnectionRequestUnsupportedVersion) -> do + -- If the peer doesn't support V0 then there's nothing we can do, for + -- it's the only version we support. + let err = connectFailed "setupRemoteEndPoint: unsupported version" + resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointInvalid err) + tryCloseSocket sock `finally` putMVar socketClosedVar () + return Nothing + Right (socketClosedVar, sock, ConnectionRequestInvalid) -> do let err = invalidAddress "setupRemoteEndPoint: Invalid endpoint" resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointInvalid err) - tryCloseSocket sock - return False - Right (sock, ConnectionRequestCrossed) -> do + tryCloseSocket sock `finally` putMVar socketClosedVar () + return Nothing + Right (socketClosedVar, sock, ConnectionRequestCrossed) -> do withMVar (remoteState theirEndPoint) $ \st -> case st of RemoteEndPointInit _ crossed _ -> putMVar crossed () @@ -1255,19 +1530,33 @@ throwIO ex _ -> relyViolation (ourEndPoint, theirEndPoint) "setupRemoteEndPoint: Crossed" - tryCloseSocket sock - return False + tryCloseSocket sock `finally` putMVar socketClosedVar () + return Nothing + Right (socketClosedVar, sock, ConnectionRequestHostMismatch) -> do + let handler :: SomeException -> IO (TransportError ConnectErrorCode) + handler err = return (TransportError ConnectFailed (show err)) + err <- handle handler $ do + actualHost <- recvWithLength (tcpMaxReceiveLength params) sock + return (TransportError ConnectFailed ("setupRemoteEndPoint: Host mismatch " ++ BSC.unpack (BS.concat actualHost))) + resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointInvalid err) + tryCloseSocket sock `finally` putMVar socketClosedVar () + return Nothing Left err -> do resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointInvalid err) - return False + return Nothing - when didAccept $ void $ forkIO $ - handleIncomingMessages (ourEndPoint, theirEndPoint) - return $ either (const Nothing) (Just . snd) result + -- We handle incoming messages in a separate thread, and are careful to + -- always close the socket once that thread is finished. + forM_ didAccept $ \(socketClosed, sock) -> void $ forkIO $ + handleIncomingMessages params (ourEndPoint, theirEndPoint) + `finally` + (tryCloseSocket sock `finally` putMVar socketClosed ()) + return $ either (const Nothing) (Just . (\(_,_,x) -> x)) result where ourAddress = localAddress ourEndPoint theirAddress = remoteAddress theirEndPoint invalidAddress = TransportError ConnectNotFound + connectFailed = TransportError ConnectFailed -- | Send a CloseSocket request if the remote endpoint is unused closeIfUnused :: EndPointPair -> IO () @@ -1278,8 +1567,8 @@ then do resolved <- newEmptyMVar act <- schedule theirEndPoint $ - sendOn vst [ encodeInt32 CloseSocket - , encodeInt32 (vst ^. remoteMaxIncoming) + sendOn vst [ encodeWord32 (encodeControlHeader CloseSocket) + , encodeWord32 (vst ^. remoteLastIncoming) ] return (RemoteEndPointClosing resolved vst, Just act) else @@ -1324,7 +1613,7 @@ connAlive <- newIORef True -- Protected by the local endpoint lock lconnId <- mapIOException connectFailed $ getLocalNextConnOutId ourEndPoint let connId = createConnectionId heavyweightSelfConnectionId lconnId - writeChan ourChan $ + qdiscEnqueue' ourQueue ourAddress $ ConnectionOpened connId ReliableOrdered (localAddress ourEndPoint) return Connection { send = selfSend connAlive connId @@ -1341,7 +1630,7 @@ alive <- readIORef connAlive if alive then seq (foldr seq () msg) - writeChan ourChan (Received connId msg) + qdiscEnqueue' ourQueue ourAddress (Received connId msg) else throwIO $ TransportError SendClosed "Connection closed" LocalEndPointClosed -> throwIO $ TransportError SendFailed "Endpoint closed" @@ -1352,14 +1641,15 @@ LocalEndPointValid _ -> do alive <- readIORef connAlive when alive $ do - writeChan ourChan (ConnectionClosed connId) + qdiscEnqueue' ourQueue ourAddress (ConnectionClosed connId) writeIORef connAlive False LocalEndPointClosed -> return () - ourChan = localChannel ourEndPoint + ourQueue = localQueue ourEndPoint ourState = localState ourEndPoint connectFailed = TransportError ConnectFailed . show + ourAddress = localAddress ourEndPoint -- | Resolve an endpoint currently in 'Init' state resolveInit :: EndPointPair -> RemoteState -> IO () @@ -1397,9 +1687,10 @@ -- -- May throw a TransportError NewEndPointErrorCode exception if the transport -- is closed. -createLocalEndPoint :: TCPTransport -> IO LocalEndPoint -createLocalEndPoint transport = do - chan <- newChan +createLocalEndPoint :: TCPTransport + -> QDisc Event + -> IO LocalEndPoint +createLocalEndPoint transport qdisc = do state <- newMVar . LocalEndPointValid $ ValidLocalEndPointState { _localNextConnOutId = firstNonReservedLightweightConnectionId , _localConnections = Map.empty @@ -1411,9 +1702,9 @@ let addr = encodeEndPointAddress (transportHost transport) (transportPort transport) ix - let localEndPoint = LocalEndPoint { localAddress = addr - , localChannel = chan - , localState = state + let localEndPoint = LocalEndPoint { localAddress = addr + , localQueue = qdisc + , localState = state } return ( TransportValid . (localEndPointAt addr ^= Just localEndPoint) @@ -1612,10 +1903,11 @@ handleIOException ex vst = do -- Release probing resources if probing. forM_ (remoteProbing vst) id - tryCloseSocket (remoteSocket vst) - let code = EventConnectionLost (remoteAddress theirEndPoint) - err = TransportError code (show ex) - writeChan (localChannel ourEndPoint) $ ErrorEvent err + -- Must shut down the socket here, so that the other end will realize + -- we lost the connection + tryShutdownSocketBoth (remoteSocket vst) + -- Eventually, handleIncomingMessages will fail while trying to + -- receive, and ultimately enqueue the 'EventConnectionLost'. return (RemoteEndPointFailed ex) -- | Use 'schedule' action 'runScheduled' action in a safe way, it's assumed that @@ -1634,6 +1926,11 @@ -- | Establish a connection to a remote endpoint -- -- Maybe throw a TransportError +-- +-- If a socket is created and returned (Right is given) then the caller is +-- responsible for eventually closing the socket and filling the MVar (which +-- is empty). The MVar must be filled immediately after, and never before, +-- the socket is closed. socketToEndPoint :: EndPointAddress -- ^ Our address -> EndPointAddress -- ^ Their address -> Bool -- ^ Use SO_REUSEADDR? @@ -1642,7 +1939,7 @@ -> Maybe Int -- ^ Maybe TCP_USER_TIMEOUT -> Maybe Int -- ^ Timeout for connect -> IO (Either (TransportError ConnectErrorCode) - (N.Socket, ConnectionRequestResponse)) + (MVar (), N.Socket, ConnectionRequestResponse)) socketToEndPoint (EndPointAddress ourAddress) theirAddress reuseAddr noDelay keepAlive mUserTimeout timeout = try $ do @@ -1664,12 +1961,17 @@ mapIOException invalidAddress $ N.connect sock (N.addrAddress addr) mapIOException failed $ do - sendMany sock - (encodeInt32 theirEndPointId : prependLength [ourAddress]) - recvInt32 sock - case tryToEnum response of + sendMany sock $ + -- The version. + encodeWord32 currentProtocolVersion + -- The V0 handshake data with the length prepended. + : prependLength (encodeWord32 theirEndPointId : prependLength [ourAddress]) + recvWord32 sock + case decodeConnectionRequestResponse response of Nothing -> throwIO (failed . userError $ "Unexpected response") - Just r -> return (sock, r) + Just r -> do + socketClosedVar <- newEmptyMVar + return (socketClosedVar, sock, r) where createSocket :: N.AddrInfo -> IO N.Socket createSocket addr = mapIOException insufficientResources $ @@ -1680,26 +1982,6 @@ failed = TransportError ConnectFailed . show timeoutError = TransportError ConnectTimeout "Timed out" --- | Encode end point address -encodeEndPointAddress :: N.HostName - -> N.ServiceName - -> EndPointId - -> EndPointAddress -encodeEndPointAddress host port ix = EndPointAddress . BSC.pack $ - host ++ ":" ++ port ++ ":" ++ show ix - --- | Decode end point address -decodeEndPointAddress :: EndPointAddress - -> Maybe (N.HostName, N.ServiceName, EndPointId) -decodeEndPointAddress (EndPointAddress bs) = - case splitMaxFromEnd (== ':') 2 $ BSC.unpack bs of - [host, port, endPointIdStr] -> - case reads endPointIdStr of - [(endPointId, "")] -> Just (host, port, endPointId) - _ -> Nothing - _ -> - Nothing - -- | Construct a ConnectionId createConnectionId :: HeavyweightConnectionId -> LightweightConnectionId @@ -1707,21 +1989,6 @@ createConnectionId hcid lcid = (fromIntegral hcid `shiftL` 32) .|. fromIntegral lcid --- | @spltiMaxFromEnd p n xs@ splits list @xs@ at elements matching @p@, --- returning at most @p@ segments -- counting from the /end/ --- --- > splitMaxFromEnd (== ':') 2 "ab:cd:ef:gh" == ["ab:cd", "ef", "gh"] -splitMaxFromEnd :: (a -> Bool) -> Int -> [a] -> [[a]] -splitMaxFromEnd p = \n -> go [[]] n . reverse - where - -- go :: [[a]] -> Int -> [a] -> [[a]] - go accs _ [] = accs - go ([] : accs) 0 xs = reverse xs : accs - go (acc : accs) n (x:xs) = - if p x then go ([] : acc : accs) (n - 1) xs - else go ((x : acc) : accs) n xs - go _ _ _ = error "Bug in splitMaxFromEnd" - -------------------------------------------------------------------------------- -- Functions from TransportInternals -- -------------------------------------------------------------------------------- @@ -1803,8 +2070,8 @@ remoteIncoming :: Accessor ValidRemoteEndPointState (Set LightweightConnectionId) remoteIncoming = accessor _remoteIncoming (\cs conn -> conn { _remoteIncoming = cs }) -remoteMaxIncoming :: Accessor ValidRemoteEndPointState LightweightConnectionId -remoteMaxIncoming = accessor _remoteMaxIncoming (\lcid st -> st { _remoteMaxIncoming = lcid }) +remoteLastIncoming :: Accessor ValidRemoteEndPointState LightweightConnectionId +remoteLastIncoming = accessor _remoteLastIncoming (\lcid st -> st { _remoteLastIncoming = lcid }) remoteNextConnOutId :: Accessor ValidRemoteEndPointState LightweightConnectionId remoteNextConnOutId = accessor _remoteNextConnOutId (\cix st -> st { _remoteNextConnOutId = cix }) diff --git a/src/Network/Transport/TCP/Internal.hs b/src/Network/Transport/TCP/Internal.hs --- a/src/Network/Transport/TCP/Internal.hs +++ b/src/Network/Transport/TCP/Internal.hs @@ -1,18 +1,40 @@ -- | Utility functions for TCP sockets module Network.Transport.TCP.Internal - ( forkServer + ( ControlHeader(..) + , encodeControlHeader + , decodeControlHeader + , ConnectionRequestResponse(..) + , encodeConnectionRequestResponse + , decodeConnectionRequestResponse + , forkServer , recvWithLength , recvExact - , recvInt32 + , recvWord32 + , encodeWord32 , tryCloseSocket + , tryShutdownSocketBoth + , decodeSockAddr + , EndPointId + , encodeEndPointAddress + , decodeEndPointAddress + , ProtocolVersion + , currentProtocolVersion ) where #if ! MIN_VERSION_base(4,6,0) import Prelude hiding (catch) #endif -import Network.Transport.Internal (decodeInt32, void, tryIO, forkIOWithUnmask) +import Network.Transport.Internal + ( decodeWord32 + , encodeWord32 + , void + , tryIO + , forkIOWithUnmask + ) +import Network.Transport ( EndPointAddress(..) ) + #ifdef USE_MOCK_NETWORK import qualified Network.Transport.TCP.Mock.Socket as N #else @@ -35,6 +57,10 @@ , accept , sClose , socketPort + , shutdown + , ShutdownCmd(ShutdownBoth) + , SockAddr(..) + , inet_ntoa ) #ifdef USE_MOCK_NETWORK @@ -43,44 +69,150 @@ import qualified Network.Socket.ByteString as NBS (recv) #endif -import Control.Concurrent (ThreadId) +import Data.Word (Word32) + import Control.Monad (forever, when) import Control.Exception (SomeException, catch, bracketOnError, throwIO, mask_) +import Control.Concurrent (ThreadId, forkIO) +import Control.Concurrent.MVar + ( MVar + , newEmptyMVar + , putMVar + , readMVar + ) +import Control.Monad (forever, when) +import Control.Exception + ( SomeException + , catch + , bracketOnError + , throwIO + , mask_ + , mask + , finally + , onException + ) + import Control.Applicative ((<$>), (<*>)) +import Data.Word (Word32) import Data.ByteString (ByteString) import qualified Data.ByteString as BS (length, concat, null) -import Data.Int (Int32) import Data.ByteString.Lazy.Internal (smallChunkSize) +import qualified Data.ByteString.Char8 as BSC (unpack, pack) +-- | Local identifier for an endpoint within this transport +type EndPointId = Word32 + +-- | Identifies the version of the network-transport-tcp protocol. +-- It's the first piece of data sent when a new heavyweight connection is +-- established. +type ProtocolVersion = Word32 + +currentProtocolVersion :: ProtocolVersion +currentProtocolVersion = 0x00000000 + +-- | Control headers +data ControlHeader = + -- | Tell the remote endpoint that we created a new connection + CreatedNewConnection + -- | Tell the remote endpoint we will no longer be using a connection + | CloseConnection + -- | Request to close the connection (see module description) + | CloseSocket + -- | Sent by an endpoint when it is closed. + | CloseEndPoint + -- | Message sent to probe a socket + | ProbeSocket + -- | Acknowledgement of the ProbeSocket message + | ProbeSocketAck + deriving (Show) + +decodeControlHeader :: Word32 -> Maybe ControlHeader +decodeControlHeader w32 = case w32 of + 0 -> Just CreatedNewConnection + 1 -> Just CloseConnection + 2 -> Just CloseSocket + 3 -> Just CloseEndPoint + 4 -> Just ProbeSocket + 5 -> Just ProbeSocketAck + _ -> Nothing + +encodeControlHeader :: ControlHeader -> Word32 +encodeControlHeader ch = case ch of + CreatedNewConnection -> 0 + CloseConnection -> 1 + CloseSocket -> 2 + CloseEndPoint -> 3 + ProbeSocket -> 4 + ProbeSocketAck -> 5 + +-- | Response sent by /B/ to /A/ when /A/ tries to connect +data ConnectionRequestResponse = + -- | /B/ does not support the protocol version requested by /A/. + ConnectionRequestUnsupportedVersion + -- | /B/ accepts the connection + | ConnectionRequestAccepted + -- | /A/ requested an invalid endpoint + | ConnectionRequestInvalid + -- | /A/s request crossed with a request from /B/ (see protocols) + | ConnectionRequestCrossed + -- | /A/ gave an incorrect host (did not match the host that /B/ observed). + | ConnectionRequestHostMismatch + deriving (Show) + +decodeConnectionRequestResponse :: Word32 -> Maybe ConnectionRequestResponse +decodeConnectionRequestResponse w32 = case w32 of + 0xFFFFFFFF -> Just ConnectionRequestUnsupportedVersion + 0x00000000 -> Just ConnectionRequestAccepted + 0x00000001 -> Just ConnectionRequestInvalid + 0x00000002 -> Just ConnectionRequestCrossed + 0x00000003 -> Just ConnectionRequestHostMismatch + _ -> Nothing + +encodeConnectionRequestResponse :: ConnectionRequestResponse -> Word32 +encodeConnectionRequestResponse crr = case crr of + ConnectionRequestUnsupportedVersion -> 0xFFFFFFFF + ConnectionRequestAccepted -> 0x00000000 + ConnectionRequestInvalid -> 0x00000001 + ConnectionRequestCrossed -> 0x00000002 + ConnectionRequestHostMismatch -> 0x00000003 + -- | Start a server at the specified address. -- -- This sets up a server socket for the specified host and port. Exceptions -- thrown during setup are not caught. -- -- Once the socket is created we spawn a new thread which repeatedly accepts --- incoming connections and executes the given request handler. If any --- exception occurs the thread terminates and calls the terminationHandler. --- This exception may occur because of a call to 'N.accept', because the thread --- was explicitly killed, or because of a synchronous exception thrown by the --- request handler. Typically, you should avoid the last case by catching any --- relevant exceptions in the request handler. +-- incoming connections and executes the given request handler in another +-- thread. If any exception occurs the accepting thread terminates and calls +-- the terminationHandler. Threads spawned for previous accepted connections +-- are not killed. +-- This exception may occur because of a call to 'N.accept', or because the +-- thread was explicitly killed. -- --- The request handler should spawn threads to handle each individual request --- or the server will block. Once a thread has been spawned it will be the --- responsibility of the new thread to close the socket when an exception --- occurs. +-- The request handler is not responsible for closing the socket. It will be +-- closed once that handler returns. Take care to ensure that the socket is not +-- used after the handler returns, or you will get undefined behavior +-- (the file descriptor may be re-used). -- -- The return value includes the port was bound to. This is not always the same -- port as that given in the argument. For example, binding to port 0 actually -- binds to a random port, selected by the OS. -forkServer :: N.HostName -- ^ Host - -> N.ServiceName -- ^ Port - -> Int -- ^ Backlog (maximum number of queued connections) - -> Bool -- ^ Set ReuseAddr option? - -> (SomeException -> IO ()) -- ^ Termination handler - -> (N.Socket -> IO ()) -- ^ Request handler +forkServer :: N.HostName -- ^ Host + -> N.ServiceName -- ^ Port + -> Int -- ^ Backlog (maximum number of queued connections) + -> Bool -- ^ Set ReuseAddr option? + -> (SomeException -> IO ()) -- ^ Error handler. Called with an + -- exception raised when + -- accepting a connection. + -> (SomeException -> IO ()) -- ^ Termination handler. Called + -- when the error handler throws + -- an exception. + -> (IO () -> (N.Socket, N.SockAddr) -> IO ()) + -- ^ Request handler. Gets an + -- action which completes when + -- the socket is closed. -> IO (N.ServiceName, ThreadId) -forkServer host port backlog reuseAddr terminationHandler requestHandler = do +forkServer host port backlog reuseAddr errorHandler terminationHandler requestHandler = do -- Resolve the specified address. By specification, getAddrInfo will never -- return an empty list (but will throw an exception instead) and will return -- the "best" address first, whatever that means @@ -90,50 +222,129 @@ when reuseAddr $ N.setSocketOption sock N.ReuseAddr 1 N.bindSocket sock (N.addrAddress addr) N.listen sock backlog + + -- Close up and fill the synchonizing MVar. + let release :: ((N.Socket, N.SockAddr), MVar ()) -> IO () + release ((sock, _), socketClosed) = + N.sClose sock `finally` putMVar socketClosed () + + -- Run the request handler. + let act restore (sock, sockAddr) = do + socketClosed <- newEmptyMVar + void $ forkIO $ restore $ do + requestHandler (readMVar socketClosed) (sock, sockAddr) + `finally` + release ((sock, sockAddr), socketClosed) + + let acceptRequest :: IO () + acceptRequest = mask $ \restore -> do + -- Async exceptions are masked so that, if accept does give a + -- socket, we'll always deliver it to the handler before the + -- exception is raised. + -- If it's a Right handler then it will eventually be closed. + -- If it's a Left handler then we assume the handler itself will + -- close it. + (sock, sockAddr) <- N.accept sock + -- Looks like 'act' will never throw an exception, but to be + -- safe we'll close the socket if it does. + let handler :: SomeException -> IO () + handler _ = N.sClose sock + catch (act restore (sock, sockAddr)) handler + -- We start listening for incoming requests in a separate thread. When -- that thread is killed, we close the server socket and the termination - -- handler. We have to make sure that the exception handler is installed - -- /before/ any asynchronous exception occurs. So we mask_, then fork - -- (the child thread inherits the masked state from the parent), then + -- handler is run. We have to make sure that the exception handler is + -- installed /before/ any asynchronous exception occurs. So we mask_, then + -- fork (the child thread inherits the masked state from the parent), then -- unmask only inside the catch. (,) <$> fmap show (N.socketPort sock) <*> (mask_ $ forkIOWithUnmask $ \unmask -> - catch (unmask (forever $ acceptRequest sock)) $ \ex -> do + catch (unmask (forever (catch acceptRequest errorHandler))) $ \ex -> do tryCloseSocket sock terminationHandler ex) - where - acceptRequest :: N.Socket -> IO () - acceptRequest sock = bracketOnError (N.accept sock) - (tryCloseSocket . fst) - (requestHandler . fst) --- | Read a length and then a payload of that length -recvWithLength :: N.Socket -> IO [ByteString] -recvWithLength sock = recvInt32 sock >>= recvExact sock +-- | Read a length and then a payload of that length, subject to a limit +-- on the length. +-- If the length (first 'Word32' received) is greater than the limit then +-- an exception is thrown. +recvWithLength :: Word32 -> N.Socket -> IO [ByteString] +recvWithLength limit sock = do + len <- recvWord32 sock + when (len > limit) $ + throwIO (userError "recvWithLength: limit exceeded") + recvExact sock len --- | Receive a 32-bit integer -recvInt32 :: Num a => N.Socket -> IO a -recvInt32 sock = decodeInt32 . BS.concat <$> recvExact sock 4 +-- | Receive a 32-bit unsigned integer +recvWord32 :: N.Socket -> IO Word32 +recvWord32 = fmap (decodeWord32 . BS.concat) . flip recvExact 4 --- | Close a socket, ignoring I/O exceptions +-- | Close a socket, ignoring I/O exceptions. tryCloseSocket :: N.Socket -> IO () tryCloseSocket sock = void . tryIO $ N.sClose sock +-- | Shutdown socket sends and receives, ignoring I/O exceptions. +tryShutdownSocketBoth :: N.Socket -> IO () +tryShutdownSocketBoth sock = void . tryIO $ + N.shutdown sock N.ShutdownBoth + -- | Read an exact number of bytes from a socket -- -- Throws an I/O exception if the socket closes before the specified -- number of bytes could be read -recvExact :: N.Socket -- ^ Socket to read from - -> Int32 -- ^ Number of bytes to read - -> IO [ByteString] -recvExact _ len | len < 0 = throwIO (userError "recvExact: Negative length") +recvExact :: N.Socket -- ^ Socket to read from + -> Word32 -- ^ Number of bytes to read + -> IO [ByteString] -- ^ Data read recvExact sock len = go [] len where - go :: [ByteString] -> Int32 -> IO [ByteString] + go :: [ByteString] -> Word32 -> IO [ByteString] go acc 0 = return (reverse acc) go acc l = do bs <- NBS.recv sock (fromIntegral l `min` smallChunkSize) if BS.null bs then throwIO (userError "recvExact: Socket closed") else go (bs : acc) (l - fromIntegral (BS.length bs)) + +-- | Produce a HostName and ServiceName from a SockAddr. Only gives 'Just' for +-- IPv4 addresses. +decodeSockAddr :: N.SockAddr -> IO (Maybe (N.HostName, N.ServiceName)) +decodeSockAddr sockAddr = case sockAddr of + N.SockAddrInet port host -> do + hostString <- N.inet_ntoa host + return $ Just (hostString, show port) + _ -> return Nothing + +-- | Encode end point address +encodeEndPointAddress :: N.HostName + -> N.ServiceName + -> EndPointId + -> EndPointAddress +encodeEndPointAddress host port ix = EndPointAddress . BSC.pack $ + host ++ ":" ++ port ++ ":" ++ show ix + +-- | Decode end point address +decodeEndPointAddress :: EndPointAddress + -> Maybe (N.HostName, N.ServiceName, EndPointId) +decodeEndPointAddress (EndPointAddress bs) = + case splitMaxFromEnd (== ':') 2 $ BSC.unpack bs of + [host, port, endPointIdStr] -> + case reads endPointIdStr of + [(endPointId, "")] -> Just (host, port, endPointId) + _ -> Nothing + _ -> + Nothing + +-- | @spltiMaxFromEnd p n xs@ splits list @xs@ at elements matching @p@, +-- returning at most @p@ segments -- counting from the /end/ +-- +-- > splitMaxFromEnd (== ':') 2 "ab:cd:ef:gh" == ["ab:cd", "ef", "gh"] +splitMaxFromEnd :: (a -> Bool) -> Int -> [a] -> [[a]] +splitMaxFromEnd p = \n -> go [[]] n . reverse + where + -- go :: [[a]] -> Int -> [a] -> [[a]] + go accs _ [] = accs + go ([] : accs) 0 xs = reverse xs : accs + go (acc : accs) n (x:xs) = + if p x then go ([] : acc : accs) (n - 1) xs + else go ((x : acc) : accs) n xs + go _ _ _ = error "Bug in splitMaxFromEnd" diff --git a/tests/TestTCP.hs b/tests/TestTCP.hs --- a/tests/TestTCP.hs +++ b/tests/TestTCP.hs @@ -1,4 +1,5 @@ {-# LANGUAGE RebindableSyntax, TemplateHaskell #-} +{-# LANGUAGE BangPatterns #-} {-# OPTIONS_GHC -fno-warn-orphans #-} module Main where @@ -15,11 +16,10 @@ import Network.Transport.TCP ( createTransport , createTransportExposeInternals , TransportInternals(..) - , encodeEndPointAddress + , TCPParameters(..) , defaultTCPParameters , LightweightConnectionId ) -import Data.Int (Int32) import Control.Concurrent (threadDelay, killThread) import Control.Concurrent.MVar ( MVar , newEmptyMVar @@ -30,21 +30,31 @@ , newMVar , modifyMVar , modifyMVar_ + , swapMVar ) 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 +import Network.Transport.TCP ( socketToEndPoint ) +import Network.Transport.Internal ( prependLength , tlog , tryIO , void ) -import Network.Transport.TCP.Internal (recvInt32, forkServer, recvWithLength) +import Network.Transport.TCP.Internal + ( ControlHeader(..) + , encodeControlHeader + , decodeControlHeader + , ConnectionRequestResponse(..) + , encodeConnectionRequestResponse + , decodeConnectionRequestResponse + , encodeWord32 + , recvWord32 + , forkServer + , recvWithLength + , encodeEndPointAddress + , decodeEndPointAddress + ) #ifdef USE_MOCK_NETWORK import qualified Network.Transport.TCP.Mock.Socket as N @@ -57,6 +67,15 @@ , AddrInfo , shutdown , ShutdownCmd(ShutdownSend) + , SockAddr(..) + , SocketType(Stream) + , AddrInfo(..) + , getAddrInfo + , defaultHints + , defaultProtocol + , socket + , connect + , close ) #ifdef USE_MOCK_NETWORK @@ -65,6 +84,7 @@ import Network.Socket.ByteString (sendMany) #endif +import qualified Data.ByteString as BS (length, concat) import Data.String (fromString) import GHC.IO.Exception (ioe_errno) import Foreign.C.Error (Errno(..), eADDRNOTAVAIL) @@ -105,7 +125,7 @@ server :: MVar EndPointAddress -> MVar EndPointAddress -> MVar () -> IO () server serverAddr clientAddr serverDone = do tlog "Server" - Right transport <- createTransport "127.0.0.1" "0" defaultTCPParameters + Right transport <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters Right endpoint <- newEndPoint transport putMVar serverAddr (address endpoint) theirAddr <- readMVar clientAddr @@ -154,23 +174,26 @@ tlog "Client" -- Listen for incoming messages - (clientPort, _) <- forkServer "127.0.0.1" "0" 5 True throwIO $ \sock -> do + (clientPort, _) <- forkServer "127.0.0.1" "0" 5 True throwIO throwIO $ \socketFree (sock, _) -> do -- Initial setup - 0 <- recvInt32 sock :: IO Int - _ <- recvWithLength sock - sendMany sock [encodeInt32 ConnectionRequestAccepted] + 0 <- recvWord32 sock + _ <- recvWithLength maxBound sock + sendMany sock [encodeWord32 (encodeConnectionRequestResponse ConnectionRequestAccepted)] -- Server opens a logical connection - CreatedNewConnection <- toEnum <$> (recvInt32 sock :: IO Int) - 1024 <- recvInt32 sock :: IO LightweightConnectionId + Just CreatedNewConnection <- decodeControlHeader <$> recvWord32 sock + 1024 <- recvWord32 sock :: IO LightweightConnectionId -- Server sends a message - 1024 <- recvInt32 sock :: IO Int - ["ping"] <- recvWithLength sock + 1024 <- recvWord32 sock + ["ping"] <- recvWithLength maxBound sock -- Reply - sendMany sock [encodeInt32 CreatedNewConnection, encodeInt32 (10002 :: Int)] - sendMany sock (encodeInt32 10002 : prependLength ["pong"]) + sendMany sock [ + encodeWord32 (encodeControlHeader CreatedNewConnection) + , encodeWord32 10002 + ] + sendMany sock (encodeWord32 10002 : prependLength ["pong"]) -- Close the socket N.sClose sock @@ -179,10 +202,13 @@ putMVar clientAddr ourAddress -- Connect to the server - Right (sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr True False False Nothing Nothing + Right (_, sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr True False False Nothing Nothing -- Open a new connection - sendMany sock [encodeInt32 CreatedNewConnection, encodeInt32 (10003 :: Int)] + sendMany sock [ + encodeWord32 (encodeControlHeader CreatedNewConnection) + , encodeWord32 10003 + ] -- Close the socket without closing the connection explicitly -- The server should receive an error event @@ -204,7 +230,7 @@ server :: MVar EndPointAddress -> MVar EndPointAddress -> MVar () -> IO () server serverAddr clientAddr serverDone = do tlog "Server" - Right transport <- createTransport "127.0.0.1" "0" defaultTCPParameters + Right transport <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters Right endpoint <- newEndPoint transport putMVar serverAddr (address endpoint) theirAddr <- readMVar clientAddr @@ -260,53 +286,65 @@ tlog "Client" -- Listen for incoming messages - (clientPort, _) <- forkServer "127.0.0.1" "0" 5 True throwIO $ \sock -> do + (clientPort, _) <- forkServer "127.0.0.1" "0" 5 True throwIO throwIO $ \socketFree (sock, _) -> do -- Initial setup - 0 <- recvInt32 sock :: IO Int - _ <- recvWithLength sock - sendMany sock [encodeInt32 ConnectionRequestAccepted] + 0 <- recvWord32 sock + _ <- recvWithLength maxBound sock + sendMany sock [encodeWord32 (encodeConnectionRequestResponse ConnectionRequestAccepted)] -- Server opens a logical connection - CreatedNewConnection <- toEnum <$> (recvInt32 sock :: IO Int) - 1024 <- recvInt32 sock :: IO LightweightConnectionId + Just CreatedNewConnection <- decodeControlHeader <$> recvWord32 sock + 1024 <- recvWord32 sock :: IO LightweightConnectionId -- Server sends a message - 1024 <- recvInt32 sock :: IO Int - ["ping"] <- recvWithLength sock + 1024 <- recvWord32 sock + ["ping"] <- recvWithLength maxBound sock -- Reply - sendMany sock [encodeInt32 CreatedNewConnection, encodeInt32 (10002 :: Int)] - sendMany sock (encodeInt32 (10002 :: Int) : prependLength ["pong"]) + sendMany sock [ + encodeWord32 (encodeControlHeader CreatedNewConnection) + , encodeWord32 10002 + ] + sendMany sock (encodeWord32 10002 : prependLength ["pong"]) -- Send a CloseSocket even though there are still connections *in both -- directions* - sendMany sock [encodeInt32 CloseSocket, encodeInt32 (1024 :: Int)] + sendMany sock [ + encodeWord32 (encodeControlHeader CloseSocket) + , encodeWord32 1024 + ] N.sClose sock let ourAddress = encodeEndPointAddress "127.0.0.1" clientPort 0 putMVar clientAddr ourAddress -- Connect to the server - Right (sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr True False False Nothing Nothing + Right (_, sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr True False False Nothing Nothing -- Open a new connection - sendMany sock [encodeInt32 CreatedNewConnection, encodeInt32 (10003 :: Int)] + sendMany sock [ + encodeWord32 (encodeControlHeader CreatedNewConnection) + , encodeWord32 10003 + ] -- Send a CloseSocket without sending a closeconnecton -- The server should still receive a ConnectionClosed message - sendMany sock [encodeInt32 CloseSocket, encodeInt32 (0 :: Int)] + sendMany sock [ + encodeWord32 (encodeControlHeader CloseSocket) + , encodeWord32 0 + ] N.sClose sock -- | Test the creation of a transport with an invalid address testInvalidAddress :: IO () testInvalidAddress = do - Left _ <- createTransport "invalidHostName" "0" defaultTCPParameters + Left _ <- createTransport "invalidHostName" "0" ((,) "invalidHostName") defaultTCPParameters return () -- | Test connecting to invalid or non-existing endpoints testInvalidConnect :: IO () testInvalidConnect = do - Right transport <- createTransport "127.0.0.1" "0" defaultTCPParameters + Right transport <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters Right endpoint <- newEndPoint transport -- Syntax error in the endpoint address @@ -337,7 +375,7 @@ clientDone <- newEmptyMVar serverDone <- newEmptyMVar connectionEstablished <- newEmptyMVar - Right transport <- createTransport "127.0.0.1" "0" defaultTCPParameters + Right transport <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters -- Server forkTry $ do @@ -375,32 +413,41 @@ theirAddress <- readMVar serverAddr -- Connect to the server - Right (sock, ConnectionRequestAccepted) <- socketToEndPoint ourAddress theirAddress True False False Nothing Nothing + Right (_, sock, ConnectionRequestAccepted) <- socketToEndPoint ourAddress theirAddress True False False Nothing Nothing putMVar connectionEstablished () -- Server connects to us, and then closes the connection - CreatedNewConnection <- toEnum <$> (recvInt32 sock :: IO Int) - 1024 <- recvInt32 sock :: IO LightweightConnectionId + Just CreatedNewConnection <- decodeControlHeader <$> recvWord32 sock + 1024 <- recvWord32 sock :: IO LightweightConnectionId - CloseConnection <- toEnum <$> (recvInt32 sock :: IO Int) - 1024 <- recvInt32 sock :: IO LightweightConnectionId + Just CloseConnection <- decodeControlHeader <$> recvWord32 sock + 1024 <- recvWord32 sock :: IO LightweightConnectionId -- Server will now send a CloseSocket request as its refcount reached 0 tlog "Waiting for CloseSocket request" - CloseSocket <- toEnum <$> recvInt32 sock - _ <- recvInt32 sock :: IO LightweightConnectionId + Just CloseSocket <- decodeControlHeader <$> recvWord32 sock + _ <- recvWord32 sock :: IO LightweightConnectionId -- But we ignore it and request another connection in the other direction tlog "Ignoring it, requesting another connection" - sendMany sock [encodeInt32 CreatedNewConnection, encodeInt32 (1024 :: Int)] + sendMany sock [ + encodeWord32 (encodeControlHeader CreatedNewConnection) + , encodeWord32 1024 + ] -- Close it again tlog "Closing connection" - sendMany sock [encodeInt32 CloseConnection, encodeInt32 (1024 :: Int)] + sendMany sock [ + encodeWord32 (encodeControlHeader CloseConnection) + , encodeWord32 1024 + ] -- And close the connection completely tlog "Closing socket" - sendMany sock [encodeInt32 CloseSocket, encodeInt32 (1024 :: Int)] + sendMany sock [ + encodeWord32 (encodeControlHeader CloseSocket) + , encodeWord32 1024 + ] N.sClose sock putMVar clientDone () @@ -418,7 +465,7 @@ clientDone <- newEmptyMVar serverDone <- newEmptyMVar connectionEstablished <- newEmptyMVar - Right transport <- createTransport "127.0.0.1" "0" defaultTCPParameters + Right transport <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters -- Server forkTry $ do @@ -452,20 +499,20 @@ theirAddress <- readMVar serverAddr -- Connect to the server - Right (sock, ConnectionRequestAccepted) <- socketToEndPoint ourAddress theirAddress True False False Nothing Nothing + Right (_, sock, ConnectionRequestAccepted) <- socketToEndPoint ourAddress theirAddress True False False Nothing Nothing putMVar connectionEstablished () -- Server connects to us, and then closes the connection - CreatedNewConnection <- toEnum <$> (recvInt32 sock :: IO Int) - 1024 <- recvInt32 sock :: IO LightweightConnectionId + Just CreatedNewConnection <- decodeControlHeader <$> recvWord32 sock + 1024 <- recvWord32 sock :: IO LightweightConnectionId - CloseConnection <- toEnum <$> (recvInt32 sock :: IO Int) - 1024 <- recvInt32 sock :: IO LightweightConnectionId + Just CloseConnection <- decodeControlHeader <$> recvWord32 sock + 1024 <- recvWord32 sock :: IO LightweightConnectionId -- Server will now send a CloseSocket request as its refcount reached 0 tlog "Waiting for CloseSocket request" - CloseSocket <- toEnum <$> recvInt32 sock - _ <- recvInt32 sock :: IO LightweightConnectionId + Just CloseSocket <- decodeControlHeader <$> recvWord32 sock + _ <- recvWord32 sock :: IO LightweightConnectionId unblocked <- newMVar False @@ -473,7 +520,7 @@ -- responding to the CloseSocket request (in this case, we -- respond by sending a ConnectionRequest) forkTry $ do - recvInt32 sock :: IO Int32 + recvWord32 sock readMVar unblocked >>= guard putMVar clientDone () @@ -482,7 +529,10 @@ tlog "Client ignores close socket and sends connection request" tlog "This should unblock the server" modifyMVar_ unblocked $ \_ -> return True - sendMany sock [encodeInt32 CreatedNewConnection, encodeInt32 (1024 :: Int)] + sendMany sock [ + encodeWord32 (encodeControlHeader CreatedNewConnection) + , encodeWord32 1024 + ] takeMVar clientDone takeMVar serverDone @@ -495,13 +545,18 @@ serverAddr <- newEmptyMVar forkTry $ do - Right transport <- createTransport "127.0.0.1" "0" defaultTCPParameters + Right transport <- createTransport "127.0.0.1" "0" ((,) "128.0.0.1") defaultTCPParameters Right endpoint <- newEndPoint transport - putMVar serverAddr (address endpoint) + -- Since we're lying about the server's address, we have to manually + -- construct the proper address. If we used its actual address, the clients + -- would try to resolve "128.0.0.1" and then would fail due to invalid + -- address. + Just (_, port, epid) <- return $ decodeEndPointAddress (address endpoint) + putMVar serverAddr $ encodeEndPointAddress "127.0.0.1" port epid forkTry $ do - -- We pick an address < 127.0.0.1 so that this is not rejected purely because of the "crossed" check - let ourAddress = EndPointAddress "126.0.0.1" + -- We pick an address < 128.0.0.1 so that this is not rejected purely because of the "crossed" check + let ourAddress = encodeEndPointAddress "127.0.0.1" "1234" 0 -- We should only get a single 'Accepted' reply gotAccepted <- newEmptyMVar @@ -512,14 +567,14 @@ -- immediately (depending on far the remote endpoint got with the initialization) response <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr True False False Nothing Nothing case response of - Right (_, ConnectionRequestAccepted) -> + Right (_, _, ConnectionRequestAccepted) -> -- We don't close this socket because we want to keep this connection open putMVar gotAccepted () -- We might get either Invalid or Crossed (the transport does not -- maintain enough history to be able to tell) - Right (sock, ConnectionRequestInvalid) -> + Right (_, sock, ConnectionRequestInvalid) -> N.sClose sock - Right (sock, ConnectionRequestCrossed) -> + Right (_, sock, ConnectionRequestCrossed) -> N.sClose sock Left _ -> return () @@ -534,11 +589,11 @@ -- | Test that we can create "many" transport instances testMany :: IO () testMany = do - Right masterTransport <- createTransport "127.0.0.1" "0" defaultTCPParameters + Right masterTransport <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters Right masterEndPoint <- newEndPoint masterTransport replicateM_ 10 $ do - mTransport <- createTransport "127.0.0.1" "0" defaultTCPParameters + mTransport <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters case mTransport of Left ex -> do putStrLn $ "IOException: " ++ show ex ++ "; errno = " ++ show (ioe_errno ex) @@ -555,7 +610,7 @@ -- | Test what happens when the transport breaks completely testBreakTransport :: IO () testBreakTransport = do - Right (transport, internals) <- createTransportExposeInternals "127.0.0.1" "0" defaultTCPParameters + Right (transport, internals) <- createTransportExposeInternals "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters Right endpoint <- newEndPoint transport killThread (transportThread internals) -- Uh oh @@ -564,6 +619,13 @@ return () +-- Used in testReconnect to block until a socket is closed. newtype is needed +-- for the Traceable instance. +newtype WaitSocketFree = WaitSocketFree (IO ()) + +instance Traceable WaitSocketFree where + trace = const Nothing + -- | Test that a second call to 'connect' might succeed even if the first -- failed. This is a TCP specific test rather than an endpoint specific test -- because we must manually create the endpoint address to match an endpoint we @@ -575,72 +637,110 @@ testReconnect = do serverDone <- newEmptyMVar endpointCreated <- newEmptyMVar + -- The server will put the 'socketFree' IO in here, so that the client can + -- block until the server has closed the socket. + socketClosed <- newEmptyMVar counter <- newMVar (0 :: Int) -- Server - (serverPort, _) <- forkServer "127.0.0.1" "0" 5 True throwIO $ \sock -> do + (serverPort, _) <- forkServer "127.0.0.1" "0" 5 True throwIO throwIO $ \socketFree (sock, _) -> do -- Accept the connection - Right 0 <- tryIO $ (recvInt32 sock :: IO Int) - Right _ <- tryIO $ recvWithLength sock + Right 0 <- tryIO $ recvWord32 sock + Right _ <- tryIO $ recvWithLength maxBound sock -- The first time we close the socket before accepting the logical connection count <- modifyMVar counter $ \i -> return (i + 1, i) + -- Wait 100ms after the socket closes, to (hopefully) ensure that the client + -- knows the connection is closed, and sending on that socket will therefore + -- fail. + putMVar socketClosed (WaitSocketFree (socketFree >> threadDelay 100000)) + when (count > 0) $ do - Right () <- tryIO $ sendMany sock [encodeInt32 ConnectionRequestAccepted] + -- The second, third, and fourth connections are accepted according to the + -- protocol. + -- On the second request, the socket then closes. + Right () <- tryIO $ sendMany sock [ + encodeWord32 (encodeConnectionRequestResponse ConnectionRequestAccepted) + ] -- Client requests a logical connection when (count > 1) $ do - Right CreatedNewConnection <- tryIO $ toEnum <$> (recvInt32 sock :: IO Int) - connId <- recvInt32 sock :: IO LightweightConnectionId - return () + -- On the third and fourth requests, a new logical connection is + -- accepted. + -- On the third request the socket then closes. + Right (Just CreatedNewConnection) <- tryIO $ decodeControlHeader <$> recvWord32 sock + connId <- recvWord32 sock :: IO LightweightConnectionId when (count > 2) $ do - -- Client sends a message - Right connId' <- tryIO $ (recvInt32 sock :: IO LightweightConnectionId) + -- On the fourth request, a message is received and then the socket + -- is closed. + Right connId' <- tryIO $ (recvWord32 sock :: IO LightweightConnectionId) True <- return $ connId == connId' - Right ["ping"] <- tryIO $ recvWithLength sock + Right ["ping"] <- tryIO $ recvWithLength maxBound sock putMVar serverDone () - Right () <- tryIO $ N.sClose sock return () putMVar endpointCreated () -- Client forkTry $ do - Right transport <- createTransport "127.0.0.1" "0" defaultTCPParameters + Right transport <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters Right endpoint <- newEndPoint transport let theirAddr = encodeEndPointAddress "127.0.0.1" serverPort 0 - -- The second attempt will fail because the server closes the socket before we can request a connection takeMVar endpointCreated - -- This might time out or not, depending on whether the server closes the - -- socket before or after we can send the RequestConnectionId request - resultConnect <- timeout 500000 $ connect endpoint theirAddr ReliableOrdered defaultConnectHints + + -- First attempt: fails because the server closes the socket without + -- doing the handshake. + resultConnect <- connect endpoint theirAddr ReliableOrdered defaultConnectHints case resultConnect of - Nothing -> return () - Just (Left (TransportError ConnectFailed _)) -> return () - Just (Left err) -> throwIO err - Just (Right _) -> throwIO $ userError "testConnect: unexpected connect success" + Left (TransportError ConnectFailed _) -> return () + Left err -> throwIO err + Right _ -> throwIO $ userError "testConnect: unexpected connect success" + WaitSocketFree wait <- takeMVar socketClosed + wait - resultConnect <- timeout 500000 $ connect endpoint theirAddr ReliableOrdered defaultConnectHints + -- Second attempt: server accepts the connection but then closes the socket. + -- We expect a failed connection if the socket is closed *before* + -- CreatedNewConnection is sent, or a successful connection such that a + -- subsequent send will fail in case CreatedNewConnection was sent before + -- the close. + resultConnect <- connect endpoint theirAddr ReliableOrdered defaultConnectHints + -- We must be sure that the socket has closed before trying to send. + WaitSocketFree wait <- takeMVar socketClosed + wait case resultConnect of - Nothing -> return () - Just (Left (TransportError ConnectFailed _)) -> return () - Just (Left err) -> throwIO err - Just (Right c) -> do - ev <- send c ["foo"] + Left (TransportError ConnectFailed _) -> return () + Left err -> throwIO err + Right c -> do + ev <- send c ["ping"] case ev of Left _ -> return () Right _ -> throwIO $ userError "testConnect: unexpected send success" - -- The third attempt succeeds - Right conn1 <- connect endpoint theirAddr ReliableOrdered defaultConnectHints + -- In any case, since a heavyweight connection was made, we'll get a + -- connection lost event. + ErrorEvent (TransportError (EventConnectionLost _) _) <- receive endpoint - -- But a send will fail because the server has closed the connection again - threadDelay 100000 - Left (TransportError SendFailed _) <- send conn1 ["ping"] + -- Third attempt: server accepts the heavyweight and the lightweight + -- connection (CreatedNewConnection) but then closes the socket. + -- The connection must succeed, but sending after the socket is closed + -- must fail. + resultConnect <- connect endpoint theirAddr ReliableOrdered defaultConnectHints + -- Wait until close before trying to send. + WaitSocketFree wait <- takeMVar socketClosed + wait + case resultConnect of + Left err -> throwIO err + Right c -> do + ev <- send c ["ping"] + case ev of + Left (TransportError SendFailed _) -> return () + Left err -> throwIO err + Right _ -> throwIO $ userError "testConnect: unexpected send success" + ErrorEvent (TransportError (EventConnectionLost _) _) <- receive endpoint -- But a subsequent call to connect should reestablish the connection @@ -648,6 +748,9 @@ -- Send should now succeed Right () <- send conn2 ["ping"] + + WaitSocketFree wait <- takeMVar socketClosed + wait return () takeMVar serverDone @@ -662,28 +765,32 @@ serverGotPing <- newEmptyMVar -- Server - (serverPort, _) <- forkServer "127.0.0.1" "0" 5 True throwIO $ \sock -> do + (serverPort, _) <- forkServer "127.0.0.1" "0" 5 True throwIO throwIO $ \socketFree (sock, _) -> do -- We accept connections, but when an exception occurs we don't do -- anything (in particular, we don't close the socket). This is important -- because when we shutdown one direction of the socket a recv here will -- fail, but we don't want to close that socket at that point (which -- would shutdown the socket in the other direction) void . (try :: IO () -> IO (Either SomeException ())) $ do - 0 <- recvInt32 sock :: IO Int - _ <- recvWithLength sock - () <- sendMany sock [encodeInt32 ConnectionRequestAccepted] + 0 <- recvWord32 sock + _ <- recvWithLength maxBound sock + () <- sendMany sock [encodeWord32 (encodeConnectionRequestResponse ConnectionRequestAccepted)] - CreatedNewConnection <- toEnum <$> (recvInt32 sock :: IO Int) - connId <- recvInt32 sock :: IO LightweightConnectionId + Just CreatedNewConnection <- decodeControlHeader <$> recvWord32 sock + connId <- recvWord32 sock :: IO LightweightConnectionId - connId' <- recvInt32 sock :: IO LightweightConnectionId + connId' <- recvWord32 sock :: IO LightweightConnectionId True <- return $ connId == connId' - ["ping"] <- recvWithLength sock + ["ping"] <- recvWithLength maxBound sock putMVar serverGotPing () + -- Must read the clientDone MVar so that we don't close the socket + -- (forkServer will close it once this action ends). + readMVar clientDone + -- Client forkTry $ do - Right (transport, internals) <- createTransportExposeInternals "127.0.0.1" "0" defaultTCPParameters + Right (transport, internals) <- createTransportExposeInternals "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters Right endpoint <- newEndPoint transport let theirAddr = encodeEndPointAddress "127.0.0.1" serverPort 0 @@ -734,11 +841,11 @@ putMVar clientDone () - takeMVar clientDone + readMVar clientDone testInvalidCloseConnection :: IO () testInvalidCloseConnection = do - Right (transport, internals) <- createTransportExposeInternals "127.0.0.1" "0" defaultTCPParameters + Right (transport, internals) <- createTransportExposeInternals "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters serverAddr <- newEmptyMVar clientDone <- newEmptyMVar serverDone <- newEmptyMVar @@ -767,7 +874,10 @@ -- Get a handle on the TCP connection and manually send an invalid CloseConnection request sock <- socketBetween internals ourAddr theirAddr - sendMany sock [encodeInt32 CloseConnection, encodeInt32 (12345 :: Int)] + sendMany sock [ + encodeWord32 (encodeControlHeader CloseConnection) + , encodeWord32 (12345 :: LightweightConnectionId) + ] putMVar clientDone () @@ -777,19 +887,157 @@ testUseRandomPort = do testDone <- newEmptyMVar forkTry $ do - Right transport1 <- createTransport "127.0.0.1" "0" defaultTCPParameters + Right transport1 <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters Right ep1 <- newEndPoint transport1 - Right transport2 <- createTransport "127.0.0.1" "0" defaultTCPParameters + -- Same as transport1, but is strict in the port. + Right transport2 <- createTransport "127.0.0.1" "0" (\(!port) -> ("127.0.0.1", port)) defaultTCPParameters Right ep2 <- newEndPoint transport2 Right conn1 <- connect ep2 (address ep1) ReliableOrdered defaultConnectHints ConnectionOpened _ _ _ <- receive ep1 putMVar testDone () takeMVar testDone +-- | Verify that if a peer sends an address or data which exceeds the maximum +-- length, that peer's connection will be terminated, but other peers will +-- not be affected. +testMaxLength :: IO () +testMaxLength = do + + Right serverTransport <- createTransport "127.0.0.1" "9998" ((,) "127.0.0.1") $ defaultTCPParameters { + -- 17 bytes should fit every valid address at 127.0.0.1. + -- Port is at most 5 bytes (65536) and id is a base-10 Word32 so + -- at most 10 bytes. We'll have one client with a 5-byte port to push it + -- over the chosen limit of 16 + tcpMaxAddressLength = 16 + , tcpMaxReceiveLength = 8 + } + Right goodClientTransport <- createTransport "127.0.0.1" "9999" ((,) "127.0.0.1") defaultTCPParameters + Right badClientTransport <- createTransport "127.0.0.1" "10000" ((,) "127.0.0.1") defaultTCPParameters + + serverAddress <- newEmptyMVar + testDone <- newEmptyMVar + goodClientConnected <- newEmptyMVar + goodClientDone <- newEmptyMVar + badClientDone <- newEmptyMVar + + forkTry $ do + Right serverEp <- newEndPoint serverTransport + putMVar serverAddress (address serverEp) + readMVar badClientDone + ConnectionOpened _ _ _ <- receive serverEp + Received _ _ <- receive serverEp + -- Will lose the connection when the good client sends 9 bytes. + ErrorEvent (TransportError (EventConnectionLost _) _) <- receive serverEp + readMVar goodClientDone + putMVar testDone () + + forkTry $ do + Right badClientEp <- newEndPoint badClientTransport + address <- readMVar serverAddress + -- Wait until the good client connects, then try to connect. It'll fail, + -- but the good client should still be OK. + readMVar goodClientConnected + Left (TransportError ConnectFailed _) + <- connect badClientEp address ReliableOrdered defaultConnectHints + closeEndPoint badClientEp + putMVar badClientDone () + + forkTry $ do + Right goodClientEp <- newEndPoint goodClientTransport + address <- readMVar serverAddress + Right conn <- connect goodClientEp address ReliableOrdered defaultConnectHints + putMVar goodClientConnected () + -- Wait until the bad client has tried and failed to connect before + -- attempting a send, to ensure that its failure did not affect us. + readMVar badClientDone + Right () <- send conn ["00000000"] + -- The send which breaches the limit does not appear to fail, but the + -- (heavyweight) connection is now severed. We can reliably determine that + -- by receiving. + Right () <- send conn ["000000000"] + ErrorEvent (TransportError (EventConnectionLost _) _) <- receive goodClientEp + closeEndPoint goodClientEp + putMVar goodClientDone () + + readMVar testDone + closeTransport badClientTransport + closeTransport goodClientTransport + closeTransport serverTransport + +-- | Ensure that an end point closes up OK even if the peer disobeys the +-- protocol. +testCloseEndPoint :: IO () +testCloseEndPoint = do + + serverAddress <- newEmptyMVar + serverFinished <- newEmptyMVar + + -- A server which accepts one connection and then attempts to close the + -- end point. + forkTry $ do + Right transport <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters + Right ep <- newEndPoint transport + putMVar serverAddress (address ep) + ConnectionOpened _ _ _ <- receive ep + Just () <- timeout 5000000 (closeEndPoint ep) + putMVar serverFinished () + return () + + -- A nefarious client which connects to the server then stops responding. + forkTry $ do + Just (hostName, serviceName, endPointId) <- decodeEndPointAddress <$> readMVar serverAddress + addr:_ <- N.getAddrInfo (Just N.defaultHints) (Just hostName) (Just serviceName) + sock <- N.socket (N.addrFamily addr) N.Stream N.defaultProtocol + N.connect sock (N.addrAddress addr) + let endPointAddress = "127.0.0.1:0:0" + -- Version 0x00000000 handshake data. + v0handshake = [ + encodeWord32 endPointId + , encodeWord32 (fromIntegral (BS.length endPointAddress)) + , endPointAddress + ] + -- Version, and total length of the versioned handshake. + handshake = [ + encodeWord32 0x00000000 + , encodeWord32 (fromIntegral (BS.length (BS.concat v0handshake))) + ] + sendMany sock $ + handshake + ++ v0handshake + ++ [ -- Create a lightweight connection. + encodeWord32 (encodeControlHeader CreatedNewConnection) + , encodeWord32 1024 + ] + readMVar serverFinished + N.close sock + + readMVar serverFinished + +-- | Ensure that if the peer's claimed host doesn't match its actual host, +-- the connection is rejected (when tcpCheckPeerHost is enabled). +testCheckPeerHost :: IO () +testCheckPeerHost = do + + let params = defaultTCPParameters { tcpCheckPeerHost = True } + Right transport1 <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") params + -- This transport claims 127.0.0.2 as its host, but connections from it to + -- an EndPoint on transport1 will show 127.0.0.1 as the socket's source host. + Right transport2 <- createTransport "127.0.0.1" "0" ((,) "127.0.0.2") defaultTCPParameters + + Right ep1 <- newEndPoint transport1 + Right ep2 <- newEndPoint transport2 + + Left err <- connect ep2 (address ep1) ReliableOrdered defaultConnectHints + + TransportError ConnectFailed "setupRemoteEndPoint: Host mismatch 127.0.0.1" + <- return err + + return () + main :: IO () main = do tcpResult <- tryIO $ runTests - [ ("Use random port" , testUseRandomPort) + [ ("Use random port", testUseRandomPort) , ("EarlyDisconnect", testEarlyDisconnect) , ("EarlyCloseSocket", testEarlyCloseSocket) , ("IgnoreCloseSocket", testIgnoreCloseSocket) @@ -802,10 +1050,13 @@ , ("Reconnect", testReconnect) , ("UnidirectionalError", testUnidirectionalError) , ("InvalidCloseConnection", testInvalidCloseConnection) + , ("MaxLength", testMaxLength) + , ("CloseEndPoint", testCloseEndPoint) + , ("CheckPeerHost", testCheckPeerHost) ] -- Run the generic tests even if the TCP specific tests failed.. testTransport (either (Left . show) (Right) <$> - createTransport "127.0.0.1" "0" defaultTCPParameters) + createTransport "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters) -- ..but if the generic tests pass, still fail if the specific tests did not case tcpResult of Left err -> throwIO err