diff --git a/ChangeLog b/ChangeLog --- a/ChangeLog +++ b/ChangeLog @@ -1,6 +1,32 @@ -next release +2025-02-04 Laurent P. René de Cotret 0.8.6 -* +* Ported test suite to use `tasty` rather than `test-framework`. +* Addressed some compilation warnings (#467) + +2024-09-03 Laurent P. René de Cotret 0.8.5 + +* Bumped dependency bounds to support GHC 8.10.7 - GHC 9.10.1 +* Updated links to point to Distributed Haskell monorepo + +2024-07-09 Laurent P. René de Cotret 0.8.4 + +* Set TCP_NODELAY by default. The documentation erroneously reported it as the default. (#66) + +2023-03-26 David Simmons-Duffin 0.8.3 + +* Disable a flaky test + +2023-03-25 David Simmons-Duffin 0.8.2 + +* Bump bytestring version to build with GHC 9.8. + +2022-10-12 FacundoDominguez 0.8.1 + +* Fix imports for ghc 9 (#93). + +2020-10-09 FacundoDominguez 0.8.0 + +* Move to network 3. 2019-12-31 FacundoDominguez 0.7.0 diff --git a/Setup.hs b/Setup.hs deleted file mode 100644 --- a/Setup.hs +++ /dev/null @@ -1,2 +0,0 @@ -import Distribution.Simple -main = defaultMain diff --git a/network-transport-tcp.cabal b/network-transport-tcp.cabal --- a/network-transport-tcp.cabal +++ b/network-transport-tcp.cabal @@ -1,44 +1,57 @@ +cabal-version: 3.0 Name: network-transport-tcp -Version: 0.6.1 -Cabal-Version: >=1.10 +Version: 0.8.6 Build-Type: Simple -License: BSD3 +License: BSD-3-Clause License-file: LICENSE Copyright: Well-Typed LLP, Tweag I/O Limited Author: Duncan Coutts, Nicolas Wu, Edsko de Vries -maintainer: Facundo Domínguez +maintainer: The Distributed Haskell team Stability: experimental Homepage: http://haskell-distributed.github.com -Bug-Reports: https://github.com/haskell-distributed/network-transport-tcp/issues +Bug-Reports: https://github.com/haskell-distributed/distributed-process/issues Synopsis: TCP instantiation of Network.Transport -Description: TCP instantiation of Network.Transport -Tested-With: GHC==7.6.3 GHC==7.8.4 GHC==7.10.3 +Description: TCP instantiation of Network.Transport, which can be used to create Cloud Haskell backends +tested-with: GHC==8.10.7 GHC==9.0.2 GHC==9.2.8 GHC==9.4.5 GHC==9.6.4 GHC==9.8.2 GHC==9.10.1 GHC==9.12.1 Category: Network -extra-source-files: ChangeLog +extra-doc-files: ChangeLog -Source-Repository head +source-repository head Type: git - Location: https://github.com/haskell-distributed/network-transport-tcp + Location: https://github.com/haskell-distributed/distributed-process + SubDir: packages/network-transport-tcp +common warnings + ghc-options: -Wall + -Wcompat + -Widentities + -Wincomplete-uni-patterns + -Wincomplete-record-updates + -Wredundant-constraints + -fhide-source-paths + -Wpartial-fields + -Wunused-packages + Flag use-mock-network Description: Use mock network implementation (for testing) Default: False Library - Build-Depends: base >= 4.3 && < 5, + import: warnings + Build-Depends: base >= 4.14 && < 5, async >= 2.2 && < 2.3, network-transport >= 0.5 && < 0.6, data-accessor >= 0.2 && < 0.3, - containers >= 0.4 && < 0.7, - bytestring >= 0.9 && < 0.11, - network >= 2.6.2 && < 2.9, + containers >= 0.6 && < 0.8, + bytestring >= 0.10 && < 0.13, + network >= 3.1 && < 3.3, uuid >= 1.3 && < 1.4 Exposed-modules: Network.Transport.TCP, Network.Transport.TCP.Internal Default-Extensions: CPP default-language: Haskell2010 Other-Extensions: RecursiveDo - ghc-options: -Wall -fno-warn-unused-do-bind + ghc-options: -fno-warn-unused-do-bind HS-Source-Dirs: src If flag(use-mock-network) CPP-Options: -DUSE_MOCK_NETWORK @@ -46,12 +59,13 @@ Network.Transport.TCP.Mock.Socket.ByteString Test-Suite TestTCP + import: warnings 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.9, + Build-Depends: base >= 4.14 && < 5, + bytestring >= 0.10, + network-transport-tests >= 0.3 && < 0.4, + network >= 3.1, network-transport, network-transport-tcp ghc-options: -threaded -rtsopts -with-rtsopts=-N @@ -63,13 +77,11 @@ CPP-Options: -DUSE_MOCK_NETWORK Test-Suite TestQC + import: warnings Type: exitcode-stdio-1.0 Main-Is: TestQC.hs If flag(use-mock-network) - Build-Depends: base >= 4.3 && < 5, - test-framework, - test-framework-quickcheck2, - test-framework-hunit, + Build-Depends: base >= 4.14 && < 5, QuickCheck, HUnit, network-transport, 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 @@ -10,7 +10,7 @@ -- Applications that use the TCP transport should use -- 'Network.Socket.withSocketsDo' in their main function for Windows -- compatibility (see "Network.Socket"). - +{-# LANGUAGE CPP #-} {-# LANGUAGE RecursiveDo #-} {-# LANGUAGE TupleSections #-} {-# LANGUAGE RankNTypes #-} @@ -19,6 +19,9 @@ module Network.Transport.TCP ( -- * Main API createTransport + , TCPAddr(..) + , defaultTCPAddr + , TCPAddrInfo(..) , TCPParameters(..) , defaultTCPParameters -- * Internals (exposed for unit tests) @@ -40,9 +43,6 @@ import Prelude hiding ( mapM_ -#if ! MIN_VERSION_base(4,6,0) - , catch -#endif ) import Network.Transport @@ -65,12 +65,12 @@ , encodeEndPointAddress , decodeEndPointAddress , currentProtocolVersion + , randomEndPointAddress ) import Network.Transport.Internal ( prependLength , mapIOException , tryIO - , tryToEnum , void , timeoutMaybe , asyncWhenCancelled @@ -85,6 +85,7 @@ , ServiceName , Socket , getAddrInfo + , maxListenQueue , socket , addrFamily , addrAddress @@ -94,7 +95,6 @@ , SocketOption(ReuseAddr, NoDelay, UserTimeout, KeepAlive) , isSupportedSocketOption , connect - , sOMAXCONN , AddrInfo , SockAddr(..) ) @@ -129,7 +129,6 @@ ) import Control.Concurrent.Async (async, wait) import Control.Category ((>>>)) -import Control.Applicative ((<$>)) import Control.Monad (when, unless, join, mplus, (<=<)) import Control.Exception ( IOException @@ -150,10 +149,10 @@ ) import Data.IORef (IORef, newIORef, writeIORef, readIORef, writeIORef) import Data.ByteString (ByteString) -import qualified Data.ByteString as BS (concat) +import qualified Data.ByteString as BS (concat, null) import qualified Data.ByteString.Char8 as BSC (pack, unpack) import Data.Bits (shiftL, (.|.)) -import Data.Maybe (isJust, isNothing, fromJust) +import Data.Maybe (isJust, isNothing) import Data.Word (Word32) import Data.Set (Set) import qualified Data.Set as Set @@ -167,12 +166,10 @@ ) import Data.Map (Map) import qualified Data.Map as Map (empty) -import Data.Traversable (traverse) import Data.Accessor (Accessor, accessor, (^.), (^=), (^:)) import qualified Data.Accessor.Container as DAC (mapMaybe) import Data.Foldable (forM_, mapM_) import qualified System.Timeout (timeout) -import qualified Data.ByteString as BS (length) -- $design -- @@ -285,11 +282,21 @@ -- about the state (ValidTransportState, ValidLocalEndPointState, -- ValidRemoteEndPointState). -data TCPTransport = TCPTransport +-- | Information about the network addresses of a transport: the external +-- host/port as well as the bound host/port, which are not necessarily the +-- same. +data TransportAddrInfo = TransportAddrInfo { transportHost :: !N.HostName , transportPort :: !N.ServiceName , transportBindHost :: !N.HostName , transportBindPort :: !N.ServiceName + } + +data TCPTransport = TCPTransport + { transportAddrInfo :: !(Maybe TransportAddrInfo) + -- ^ This is 'Nothing' in case the transport is not addressable from the + -- network: peers cannot connect to it unless it has a connection to the + -- peer. , transportState :: !(MVar TransportState) , transportParams :: !TCPParameters } @@ -299,16 +306,17 @@ | TransportClosed data ValidTransportState = ValidTransportState - { _localEndPoints :: !(Map EndPointAddress LocalEndPoint) + { _localEndPoints :: !(Map EndPointId LocalEndPoint) , _nextEndPointId :: !EndPointId } data LocalEndPoint = LocalEndPoint - { localAddress :: !EndPointAddress - , localState :: !(MVar LocalEndPointState) + { localAddress :: !EndPointAddress + , localEndPointId :: !EndPointId + , localState :: !(MVar LocalEndPointState) -- | A 'QDisc' is held here rather than on the 'ValidLocalEndPointState' -- because even closed 'LocalEndPoint's can have queued input data. - , localQueue :: !(QDisc Event) + , localQueue :: !(QDisc Event) } data LocalEndPointState = @@ -474,6 +482,28 @@ -- 'LightweightConnectionId'. type HeavyweightConnectionId = Word32 +-- | A transport which is addressable from the network must give a host/port +-- on which to bind/listen, and determine its external address (host/port) from +-- the actual port (which may not be known, in case 0 is used for the bind +-- port). +data TCPAddrInfo = TCPAddrInfo { + tcpBindHost :: N.HostName + , tcpBindPort :: N.ServiceName + , tcpExternalAddress :: N.ServiceName -> (N.HostName, N.ServiceName) + } + +-- | Addressability of a transport. If your transport cannot be connected +-- to, for instance because it runs behind NAT, use Unaddressable. +data TCPAddr = Addressable TCPAddrInfo | Unaddressable + +-- | The bind and external host/port are the same. +defaultTCPAddr :: N.HostName -> N.ServiceName -> TCPAddr +defaultTCPAddr host port = Addressable $ TCPAddrInfo { + tcpBindHost = host + , tcpBindPort = port + , tcpExternalAddress = (,) host + } + -- | Parameters for setting up the TCP transport data TCPParameters = TCPParameters { -- | Backlog for 'listen'. @@ -489,6 +519,7 @@ -- Defaults to True. , tcpNoDelay :: Bool -- | Should we set TCP_KEEPALIVE on connection sockets? + -- Defaults to False. , tcpKeepAlive :: Bool -- | Value of TCP_USER_TIMEOUT in milliseconds , tcpUserTimeout :: Maybe Int @@ -500,6 +531,8 @@ -- -- Connection requests to this transport will also timeout if they don't -- send the required data before this many microseconds. + -- + -- Defaults to Nothing (no timeout). , transportConnectTimeout :: Maybe Int -- | Create a QDisc for an EndPoint. , tcpNewQDisc :: forall t . IO (QDisc t) @@ -518,17 +551,19 @@ -- This is useful when operating on untrusted networks, because the peer -- could otherwise deny service to some victim by claiming the victim's -- address. + -- Defaults to False. , tcpCheckPeerHost :: Bool -- | What to do if there's an exception when accepting a new TCP -- connection. Throwing an exception here will cause the server to -- terminate. + -- Defaults to `throwIO`. , tcpServerExceptionHandler :: SomeException -> IO () } -- | Internal functionality we expose for unit testing data TransportInternals = TransportInternals { -- | The ID of the thread that listens for new incoming connections - transportThread :: ThreadId + transportThread :: Maybe ThreadId -- | A variant of newEndPoint in which the QDisc determined by the -- transport's TCPParameters can be optionally overridden. , newEndPointInternal :: (forall t . Maybe (QDisc t)) @@ -544,74 +579,79 @@ -------------------------------------------------------------------------------- -- | Create a TCP transport -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 bindHost bindPort mkExternal params = - either Left (Right . fst) <$> - createTransportExposeInternals bindHost bindPort mkExternal params +createTransport + :: TCPAddr + -> TCPParameters + -> IO (Either IOException Transport) +createTransport addr params = + either Left (Right . fst) <$> createTransportExposeInternals addr params -- | You should probably not use this function (used for unit testing only) createTransportExposeInternals - :: N.HostName - -> N.ServiceName - -> (N.ServiceName -> (N.HostName, N.ServiceName)) + :: TCPAddr -> TCPParameters -> IO (Either IOException (Transport, TransportInternals)) -createTransportExposeInternals bindHost bindPort mkExternal params = do +createTransportExposeInternals addr params = do state <- newMVar . TransportValid $ ValidTransportState { _localEndPoints = Map.empty , _nextEndPointId = 0 } - tryIO $ mdo - when ( isJust (tcpUserTimeout params) && - not (N.isSupportedSocketOption N.UserTimeout) - ) $ - throwIO $ userError $ "Network.Transport.TCP.createTransport: " ++ - "the parameter tcpUserTimeout is unsupported " ++ - "in this system." - -- We don't know for sure the actual port 'forkServer' binded until it - -- completes (see description of 'forkServer'), yet we need the port to - -- construct a transport. So we tie a recursive knot. - (port', result) <- do - let (externalHost, externalPort) = mkExternal port' - let transport = TCPTransport { transportState = state - , transportHost = externalHost - , transportPort = externalPort - , transportBindHost = bindHost - , transportBindPort = port' - , transportParams = params - } - bracketOnError (forkServer - bindHost - bindPort - (tcpBacklog params) - (tcpReuseServerAddr params) - (errorHandler transport) - (terminationHandler transport) - (handleConnectionRequest transport)) - (\(_port', tid) -> killThread tid) - (\(port'', tid) -> (port'',) <$> mkTransport transport tid) - return result + case addr of + + Unaddressable -> + let transport = TCPTransport { transportState = state + , transportAddrInfo = Nothing + , transportParams = params + } + in fmap Right (mkTransport transport Nothing) + + Addressable (TCPAddrInfo bindHost bindPort mkExternal) -> tryIO $ mdo + when ( isJust (tcpUserTimeout params) && + not (N.isSupportedSocketOption N.UserTimeout) + ) $ + throwIO $ userError $ "Network.Transport.TCP.createTransport: " ++ + "the parameter tcpUserTimeout is unsupported " ++ + "in this system." + -- We don't know for sure the actual port 'forkServer' binded until it + -- completes (see description of 'forkServer'), yet we need the port to + -- construct a transport. So we tie a recursive knot. + (port', result) <- do + let (externalHost, externalPort) = mkExternal port' + let addrInfo = TransportAddrInfo { transportHost = externalHost + , transportPort = externalPort + , transportBindHost = bindHost + , transportBindPort = port' + } + let transport = TCPTransport { transportState = state + , transportAddrInfo = Just addrInfo + , transportParams = params + } + bracketOnError (forkServer + bindHost + bindPort + (tcpBacklog params) + (tcpReuseServerAddr params) + (errorHandler transport) + (terminationHandler transport) + (handleConnectionRequest transport)) + (\(_port', tid) -> killThread tid) + (\(port'', tid) -> (port'',) <$> mkTransport transport (Just tid)) + return result where mkTransport :: TCPTransport - -> ThreadId + -> Maybe ThreadId -> IO (Transport, TransportInternals) - mkTransport transport tid = do + mkTransport transport mtid = do return ( Transport { newEndPoint = do qdisc <- tcpNewQDisc params apiNewEndPoint transport qdisc , closeTransport = let evs = [ EndPointClosed ] - in apiCloseTransport transport (Just tid) evs + in apiCloseTransport transport mtid evs } , TransportInternals - { transportThread = tid + { transportThread = mtid , socketBetween = internalSocketBetween transport , newEndPointInternal = \mqdisc -> case mqdisc of Just qdisc -> apiNewEndPoint transport qdisc @@ -634,10 +674,10 @@ -- | Default TCP parameters defaultTCPParameters :: TCPParameters defaultTCPParameters = TCPParameters { - tcpBacklog = N.sOMAXCONN + tcpBacklog = N.maxListenQueue , tcpReuseServerAddr = True , tcpReuseClientAddr = True - , tcpNoDelay = False + , tcpNoDelay = True , tcpKeepAlive = False , tcpUserTimeout = Nothing , tcpNewQDisc = simpleUnboundedQDisc @@ -675,7 +715,7 @@ return EndPoint { receive = qdiscDequeue (localQueue ourEndPoint) , address = localAddress ourEndPoint - , connect = apiConnect (transportParams transport) ourEndPoint + , connect = apiConnect transport ourEndPoint , closeEndPoint = let evs = [ EndPointClosed ] in apiCloseEndPoint transport evs ourEndPoint , newMulticastGroup = return . Left $ newMulticastGroupError @@ -757,20 +797,20 @@ } -- | Connnect to an endpoint -apiConnect :: TCPParameters -- ^ Parameters +apiConnect :: TCPTransport -> LocalEndPoint -- ^ Local end point -> EndPointAddress -- ^ Remote address -> Reliability -- ^ Reliability (ignored) -> ConnectHints -- ^ Hints -> IO (Either (TransportError ConnectErrorCode) Connection) -apiConnect params ourEndPoint theirAddress _reliability hints = +apiConnect transport ourEndPoint theirAddress _reliability hints = try . asyncWhenCancelled close $ if localAddress ourEndPoint == theirAddress then connectToSelf ourEndPoint else do resetIfBroken ourEndPoint theirAddress (theirEndPoint, connId) <- - createConnectionTo params ourEndPoint theirAddress hints + createConnectionTo transport ourEndPoint theirAddress hints -- connAlive can be an IORef rather than an MVar because it is protected -- by the remoteState MVar. We don't need the overhead of locking twice. connAlive <- newIORef True @@ -980,40 +1020,48 @@ handleConnectionRequestV0 :: (N.Socket, N.SockAddr) -> IO (Maybe (IO ())) handleConnectionRequestV0 (sock, sockAddr) = do -- Get the OS-determined host and port. - (numericHost, resolvedHost, actualPort) <- + (numericHost, resolvedHost, _actualPort) <- resolveSockAddr sockAddr >>= maybe (throwIO (userError "handleConnectionRequest: invalid socket address")) return - - (ourEndPointId, theirAddress) <- do + -- The peer must send our identifier and their address promptly, if a + -- timeout is set. + (ourEndPointId, theirAddress, mTheirHost) <- do ourEndPointId <- recvWord32 sock let maxAddressLength = tcpMaxAddressLength $ transportParams transport - 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) + mTheirAddress <- BS.concat <$> recvWithLength maxAddressLength sock + -- Sending a length = 0 address means unaddressable. + if BS.null mTheirAddress + then do + theirAddress <- randomEndPointAddress + return (ourEndPointId, theirAddress, Nothing) + else do + let theirAddress = EndPointAddress mTheirAddress + (theirHost, _, _) + <- maybe (throwIO (userError "handleConnectionRequest: peer gave malformed address")) + return + (decodeEndPointAddress theirAddress) + return (ourEndPointId, theirAddress, Just theirHost) let checkPeerHost = tcpCheckPeerHost (transportParams transport) - if checkPeerHost && (theirHost /= resolvedHost) && (theirHost /= numericHost) + continue <- case (mTheirHost, checkPeerHost) of + (Just theirHost, True) -> do + -- If the OS-determined host doesn't match the host that the peer gave us, + -- then we have no choice but to reject the connection. It's because we + -- use the EndPointAddress to key the remote end points (localConnections) + -- and we don't want to allow a peer to deny service to other peers by + -- claiming to have their host and port. + if theirHost == numericHost || theirHost == resolvedHost + then return True + else do + sendMany sock $ + encodeWord32 (encodeConnectionRequestResponse ConnectionRequestHostMismatch) + : (prependLength [BSC.pack theirHost] ++ prependLength [BSC.pack numericHost] ++ prependLength [BSC.pack resolvedHost]) + return False + _ -> return True + if continue then do - -- 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 theirHost] ++ prependLength [BSC.pack numericHost] ++ prependLength [BSC.pack resolvedHost]) - - return Nothing - else do ourEndPoint <- withMVar (transportState transport) $ \st -> case st of TransportValid vst -> - case vst ^. localEndPointAt ourAddress of + case vst ^. localEndPointAt ourEndPointId of Nothing -> do sendMany sock [encodeWord32 (encodeConnectionRequestResponse ConnectionRequestInvalid)] throwIO $ userError "handleConnectionRequest: Invalid endpoint" @@ -1022,11 +1070,13 @@ TransportClosed -> throwIO $ userError "Transport closed" return (Just (go ourEndPoint theirAddress)) + else return Nothing where go :: LocalEndPoint -> EndPointAddress -> IO () go ourEndPoint theirAddress = handle handleException $ do + resetIfBroken ourEndPoint theirAddress (theirEndPoint, isNew) <- findRemoteEndPoint ourEndPoint theirAddress RequestedByThem Nothing @@ -1249,7 +1299,7 @@ -- Close the socket (if we don't have any outgoing connections) closeSocket :: N.Socket -> LightweightConnectionId -> IO Bool - closeSocket sock lastReceivedId = do + closeSocket _sock lastReceivedId = do mAct <- modifyMVar theirState $ \st -> do case st of RemoteEndPointInvalid _ -> @@ -1333,8 +1383,8 @@ "handleIncomingMessages:closeSocket (closed)" case mAct of Nothing -> return False - Just act -> do - runScheduledAction (ourEndPoint, theirEndPoint) act + Just act' -> do + runScheduledAction (ourEndPoint, theirEndPoint) act' return True -- Read a message and output it on the endPoint's channel. By rights we @@ -1425,12 +1475,13 @@ -- block until that is resolved. -- -- May throw a TransportError ConnectErrorCode exception. -createConnectionTo :: TCPParameters - -> LocalEndPoint - -> EndPointAddress - -> ConnectHints - -> IO (RemoteEndPoint, LightweightConnectionId) -createConnectionTo params ourEndPoint theirAddress hints = do +createConnectionTo + :: TCPTransport + -> LocalEndPoint + -> EndPointAddress + -> ConnectHints + -> IO (RemoteEndPoint, LightweightConnectionId) +createConnectionTo transport ourEndPoint theirAddress hints = do -- @timer@ is an IO action that completes when the timeout expires. timer <- case connTimeout of Just t -> do @@ -1442,6 +1493,7 @@ where + params = transportParams transport connTimeout = connectTimeout hints `mplus` transportConnectTimeout params -- The second argument indicates the response obtained to the last @@ -1462,7 +1514,7 @@ if isNew then do mr' <- handle (absorbAllExceptions Nothing) $ - setupRemoteEndPoint params (ourEndPoint, theirEndPoint) connTimeout + setupRemoteEndPoint transport (ourEndPoint, theirEndPoint) connTimeout go timer (fmap ((,) theirEndPoint) mr') else do -- 'findRemoteEndPoint' will have increased 'remoteOutgoing' @@ -1502,10 +1554,14 @@ return a -- | Set up a remote endpoint -setupRemoteEndPoint :: TCPParameters -> EndPointPair -> Maybe Int - -> IO (Maybe ConnectionRequestResponse) -setupRemoteEndPoint params (ourEndPoint, theirEndPoint) connTimeout = do - result <- socketToEndPoint ourAddress +setupRemoteEndPoint + :: TCPTransport + -> EndPointPair + -> Maybe Int + -> IO (Maybe ConnectionRequestResponse) +setupRemoteEndPoint transport (ourEndPoint, theirEndPoint) connTimeout = do + let mOurAddress = const ourAddress <$> transportAddrInfo transport + result <- socketToEndPoint mOurAddress theirAddress (tcpReuseClientAddr params) (tcpNoDelay params) @@ -1587,6 +1643,7 @@ (tryCloseSocket sock `finally` putMVar socketClosed ()) return $ either (const Nothing) (Just . (\(_,_,x) -> x)) result where + params = transportParams transport ourAddress = localAddress ourEndPoint theirAddress = remoteAddress theirEndPoint invalidAddress = TransportError ConnectNotFound @@ -1736,15 +1793,19 @@ modifyMVar (transportState transport) $ \st -> case st of TransportValid vst -> do let ix = vst ^. nextEndPointId - let addr = encodeEndPointAddress (transportHost transport) - (transportPort transport) - ix - let localEndPoint = LocalEndPoint { localAddress = addr - , localQueue = qdisc - , localState = state + addr <- case transportAddrInfo transport of + Nothing -> randomEndPointAddress + Just addrInfo -> return $ + encodeEndPointAddress (transportHost addrInfo) + (transportPort addrInfo) + ix + let localEndPoint = LocalEndPoint { localAddress = addr + , localEndPointId = ix + , localQueue = qdisc + , localState = state } return ( TransportValid - . (localEndPointAt addr ^= Just localEndPoint) + . (localEndPointAt ix ^= Just localEndPoint) . (nextEndPointId ^= ix + 1) $ vst , localEndPoint @@ -1785,7 +1846,7 @@ modifyMVar_ (transportState transport) $ \st -> case st of TransportValid vst -> return ( TransportValid - . (localEndPointAt (localAddress ourEndPoint) ^= Nothing) + . (localEndPointAt (localEndPointId ourEndPoint) ^= Nothing) $ vst ) TransportClosed -> @@ -1947,7 +2008,7 @@ -- However, it will then wait until @p@ is executed (by this call to -- 'runScheduledAction' or by another). runScheduledAction :: EndPointPair -> Action a -> IO a -runScheduledAction (ourEndPoint, theirEndPoint) mvar = do +runScheduledAction (_ourEndPoint, theirEndPoint) mvar = do join $ readChan (remoteScheduled theirEndPoint) ma <- readMVar mvar case ma of @@ -1994,16 +2055,16 @@ -- 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? - -> Bool -- ^ Use TCP_NODELAY - -> Bool -- ^ Use TCP_KEEPALIVE - -> Maybe Int -- ^ Maybe TCP_USER_TIMEOUT - -> Maybe Int -- ^ Timeout for connect +socketToEndPoint :: Maybe EndPointAddress -- ^ Our address + -> EndPointAddress -- ^ Their address + -> Bool -- ^ Use SO_REUSEADDR? + -> Bool -- ^ Use TCP_NODELAY + -> Bool -- ^ Use TCP_KEEPALIVE + -> Maybe Int -- ^ Maybe TCP_USER_TIMEOUT + -> Maybe Int -- ^ Timeout for connect -> IO (Either (TransportError ConnectErrorCode) (MVar (), N.Socket, ConnectionRequestResponse)) -socketToEndPoint (EndPointAddress ourAddress) theirAddress reuseAddr noDelay keepAlive +socketToEndPoint mOurAddress theirAddress reuseAddr noDelay keepAlive mUserTimeout timeout = try $ do (host, port, theirEndPointId) <- case decodeEndPointAddress theirAddress of @@ -2024,11 +2085,15 @@ mapIOException invalidAddress $ N.connect sock (N.addrAddress addr) mapIOException failed $ do - sendMany sock $ - -- The version. - encodeWord32 currentProtocolVersion - -- The V0 handshake data with the length prepended. - : prependLength (encodeWord32 theirEndPointId : prependLength [ourAddress]) + case mOurAddress of + Just (EndPointAddress ourAddress) -> + sendMany sock $ + encodeWord32 currentProtocolVersion + : prependLength (encodeWord32 theirEndPointId : prependLength [ourAddress]) + Nothing -> + sendMany sock $ + encodeWord32 currentProtocolVersion + : prependLength ([encodeWord32 theirEndPointId, encodeWord32 0]) recvWord32 sock case decodeConnectionRequestResponse response of Nothing -> throwIO (failed . userError $ "Unexpected response") @@ -2064,11 +2129,14 @@ -> EndPointAddress -- ^ Remote endpoint -> IO N.Socket internalSocketBetween transport ourAddress theirAddress = do + ourEndPointId <- case decodeEndPointAddress ourAddress of + Just (_, _, eid) -> return eid + _ -> throwIO $ userError "Malformed local EndPointAddress" ourEndPoint <- withMVar (transportState transport) $ \st -> case st of TransportClosed -> throwIO $ userError "Transport closed" TransportValid vst -> - case vst ^. localEndPointAt ourAddress of + case vst ^. localEndPointAt ourEndPointId of Nothing -> throwIO $ userError "Local endpoint not found" Just ep -> return ep theirEndPoint <- withMVar (localState ourEndPoint) $ \st -> case st of @@ -2091,6 +2159,7 @@ throwIO err RemoteEndPointFailed err -> throwIO err + where -------------------------------------------------------------------------------- -- Constants -- @@ -2112,7 +2181,7 @@ -- Accessor definitions -- -------------------------------------------------------------------------------- -localEndPoints :: Accessor ValidTransportState (Map EndPointAddress LocalEndPoint) +localEndPoints :: Accessor ValidTransportState (Map EndPointId LocalEndPoint) localEndPoints = accessor _localEndPoints (\es st -> st { _localEndPoints = es }) nextEndPointId :: Accessor ValidTransportState EndPointId @@ -2139,7 +2208,7 @@ remoteNextConnOutId :: Accessor ValidRemoteEndPointState LightweightConnectionId remoteNextConnOutId = accessor _remoteNextConnOutId (\cix st -> st { _remoteNextConnOutId = cix }) -localEndPointAt :: EndPointAddress -> Accessor ValidTransportState (Maybe LocalEndPoint) +localEndPointAt :: EndPointId -> Accessor ValidTransportState (Maybe LocalEndPoint) localEndPointAt addr = localEndPoints >>> DAC.mapMaybe addr localConnectionTo :: EndPointAddress -> Accessor ValidLocalEndPointState (Maybe RemoteEndPoint) 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,3 +1,4 @@ +{-# LANGUAGE CPP #-} -- | Utility functions for TCP sockets module Network.Transport.TCP.Internal ( ControlHeader(..) @@ -17,14 +18,11 @@ , EndPointId , encodeEndPointAddress , decodeEndPointAddress + , randomEndPointAddress , ProtocolVersion , currentProtocolVersion ) where -#if ! MIN_VERSION_base(4,6,0) -import Prelude hiding (catch) -#endif - import Network.Transport.Internal ( decodeWord32 , encodeWord32 @@ -41,6 +39,7 @@ import qualified Network.Socket as N #endif ( HostName + , NameInfoFlag(NI_NUMERICHOST) , ServiceName , Socket , SocketType(Stream) @@ -48,19 +47,18 @@ , getAddrInfo , defaultHints , socket - , bindSocket + , bind , listen , addrFamily , addrAddress , defaultProtocol , setSocketOption , accept - , sClose + , close , socketPort , shutdown , ShutdownCmd(ShutdownBoth) , SockAddr(..) - , inet_ntoa , getNameInfo ) @@ -81,24 +79,17 @@ , putMVar , readMVar ) -import Control.Monad (forever, when) import Control.Exception - ( SomeException - , catch - , bracketOnError - , throwIO - , mask_ - , 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.ByteString.Lazy.Internal (smallChunkSize) import qualified Data.ByteString.Char8 as BSC (unpack, pack) +import qualified Data.UUID as UUID +import qualified Data.UUID.V4 as UUID -- | Local identifier for an endpoint within this transport type EndPointId = Word32 @@ -177,6 +168,14 @@ ConnectionRequestCrossed -> 0x00000002 ConnectionRequestHostMismatch -> 0x00000003 +-- | Generate an EndPointAddress which does not encode a host/port/endpointid. +-- Such addresses are used for unreachable endpoints, and for ephemeral +-- addresses when such endpoints establish new heavyweight connections. +randomEndPointAddress :: IO EndPointAddress +randomEndPointAddress = do + uuid <- UUID.nextRandom + return $ EndPointAddress (UUID.toASCIIBytes uuid) + -- | Start a server at the specified address. -- -- This sets up a server socket for the specified host and port. Exceptions @@ -221,13 +220,13 @@ bracketOnError (N.socket (N.addrFamily addr) N.Stream N.defaultProtocol) tryCloseSocket $ \sock -> do when reuseAddr $ N.setSocketOption sock N.ReuseAddr 1 - N.bindSocket sock (N.addrAddress addr) + N.bind sock (N.addrAddress addr) N.listen sock backlog -- Close up and fill the synchonizing MVar. let release :: ((N.Socket, N.SockAddr), MVar ()) -> IO () release ((sock, _), socketClosed) = - N.sClose sock `finally` putMVar socketClosed () + N.close sock `finally` putMVar socketClosed () -- Run the request handler. let act restore (sock, sockAddr) = do @@ -249,7 +248,7 @@ -- 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 + handler _ = N.close sock catch (act restore (sock, sockAddr)) handler -- We start listening for incoming requests in a separate thread. When @@ -282,7 +281,7 @@ -- | Close a socket, ignoring I/O exceptions. tryCloseSocket :: N.Socket -> IO () tryCloseSocket sock = void . tryIO $ - N.sClose sock + N.close sock -- | Shutdown socket sends and receives, ignoring I/O exceptions. tryShutdownSocketBoth :: N.Socket -> IO () @@ -312,11 +311,11 @@ -- Will only give 'Just' for IPv4 addresses. resolveSockAddr :: N.SockAddr -> IO (Maybe (N.HostName, N.HostName, N.ServiceName)) resolveSockAddr sockAddr = case sockAddr of - N.SockAddrInet port host -> do + N.SockAddrInet port _ -> do (mResolvedHost, mResolvedPort) <- N.getNameInfo [] True False sockAddr case (mResolvedHost, mResolvedPort) of (Just resolvedHost, Nothing) -> do - numericHost <- N.inet_ntoa host + (Just numericHost, _) <- N.getNameInfo [N.NI_NUMERICHOST] True False sockAddr return $ Just (numericHost, resolvedHost, show port) _ -> error $ concat [ "decodeSockAddr: unexpected resolution " diff --git a/tests/TestTCP.hs b/tests/TestTCP.hs --- a/tests/TestTCP.hs +++ b/tests/TestTCP.hs @@ -1,5 +1,6 @@ {-# LANGUAGE RebindableSyntax, TemplateHaskell #-} {-# LANGUAGE BangPatterns #-} +{-# LANGUAGE CPP #-} {-# OPTIONS_GHC -fno-warn-orphans #-} module Main where @@ -8,9 +9,6 @@ , return , fail , (>>) -#if ! MIN_VERSION_base(4,6,0) - , catch -#endif ) import Network.Transport import Network.Transport.TCP ( createTransport @@ -19,6 +17,9 @@ , TCPParameters(..) , defaultTCPParameters , LightweightConnectionId + , TCPAddrInfo(..) + , TCPAddr(..) + , defaultTCPAddr ) import Control.Concurrent (threadDelay, killThread) import Control.Concurrent.MVar ( MVar @@ -26,14 +27,11 @@ , putMVar , takeMVar , readMVar - , isEmptyMVar , newMVar , modifyMVar , modifyMVar_ - , swapMVar ) -import Control.Monad (replicateM, guard, forM_, replicateM_, when) -import Control.Applicative ((<$>)) +import Control.Monad (replicateM, guard, replicateM_, when) import Control.Exception (throwIO, try, SomeException) import Network.Transport.TCP ( socketToEndPoint ) import Network.Transport.Internal ( prependLength @@ -47,7 +45,6 @@ , decodeControlHeader , ConnectionRequestResponse(..) , encodeConnectionRequestResponse - , decodeConnectionRequestResponse , encodeWord32 , recvWord32 , forkServer @@ -61,13 +58,11 @@ #else import qualified Network.Socket as N #endif - ( sClose - , ServiceName + ( close , Socket , AddrInfo , shutdown , ShutdownCmd(ShutdownSend) - , SockAddr(..) , SocketType(Stream) , AddrInfo(..) , getAddrInfo @@ -89,7 +84,7 @@ import GHC.IO.Exception (ioe_errno) import Foreign.C.Error (Errno(..), eADDRNOTAVAIL) import System.Timeout (timeout) -import Network.Transport.Tests (testTransport) +import Network.Transport.Tests (testTransportWithFilter) import Network.Transport.Tests.Auxiliary (forkTry, runTests) import Network.Transport.Tests.Traced @@ -125,7 +120,7 @@ server :: MVar EndPointAddress -> MVar EndPointAddress -> MVar () -> IO () server serverAddr clientAddr serverDone = do tlog "Server" - Right transport <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters + Right transport <- createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters Right endpoint <- newEndPoint transport putMVar serverAddr (address endpoint) theirAddr <- readMVar clientAddr @@ -196,13 +191,13 @@ sendMany sock (encodeWord32 10002 : prependLength ["pong"]) -- Close the socket - N.sClose sock + N.close sock let ourAddress = encodeEndPointAddress "127.0.0.1" clientPort 0 putMVar clientAddr ourAddress -- Connect to the server - Right (_, sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr True False False Nothing Nothing + Right (_, sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= \addr -> socketToEndPoint (Just ourAddress) addr True False False Nothing Nothing -- Open a new connection sendMany sock [ @@ -212,7 +207,7 @@ -- Close the socket without closing the connection explicitly -- The server should receive an error event - N.sClose sock + N.close sock -- | Test the behaviour of a premature CloseSocket request testEarlyCloseSocket :: IO () @@ -230,7 +225,7 @@ server :: MVar EndPointAddress -> MVar EndPointAddress -> MVar () -> IO () server serverAddr clientAddr serverDone = do tlog "Server" - Right transport <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters + Right transport <- createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters Right endpoint <- newEndPoint transport putMVar serverAddr (address endpoint) theirAddr <- readMVar clientAddr @@ -313,13 +308,13 @@ encodeWord32 (encodeControlHeader CloseSocket) , encodeWord32 1024 ] - N.sClose sock + N.close sock let ourAddress = encodeEndPointAddress "127.0.0.1" clientPort 0 putMVar clientAddr ourAddress -- Connect to the server - Right (_, sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr True False False Nothing Nothing + Right (_, sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= \addr -> socketToEndPoint (Just ourAddress) addr True False False Nothing Nothing -- Open a new connection sendMany sock [ @@ -333,18 +328,18 @@ encodeWord32 (encodeControlHeader CloseSocket) , encodeWord32 0 ] - N.sClose sock + N.close sock -- | Test the creation of a transport with an invalid address testInvalidAddress :: IO () testInvalidAddress = do - Left _ <- createTransport "invalidHostName" "0" ((,) "invalidHostName") defaultTCPParameters + Left _ <- createTransport (defaultTCPAddr "invalidHostName" "0") defaultTCPParameters return () -- | Test connecting to invalid or non-existing endpoints testInvalidConnect :: IO () testInvalidConnect = do - Right transport <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters + Right transport <- createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters Right endpoint <- newEndPoint transport -- Syntax error in the endpoint address @@ -375,7 +370,7 @@ clientDone <- newEmptyMVar serverDone <- newEmptyMVar connectionEstablished <- newEmptyMVar - Right transport <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters + Right transport <- createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters -- Server forkTry $ do @@ -413,7 +408,7 @@ theirAddress <- readMVar serverAddr -- Connect to the server - Right (_, sock, ConnectionRequestAccepted) <- socketToEndPoint ourAddress theirAddress True False False Nothing Nothing + Right (_, sock, ConnectionRequestAccepted) <- socketToEndPoint (Just ourAddress) theirAddress True False False Nothing Nothing putMVar connectionEstablished () -- Server connects to us, and then closes the connection @@ -448,7 +443,7 @@ encodeWord32 (encodeControlHeader CloseSocket) , encodeWord32 1024 ] - N.sClose sock + N.close sock putMVar clientDone () @@ -465,7 +460,7 @@ clientDone <- newEmptyMVar serverDone <- newEmptyMVar connectionEstablished <- newEmptyMVar - Right transport <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters + Right transport <- createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters -- Server forkTry $ do @@ -499,7 +494,7 @@ theirAddress <- readMVar serverAddr -- Connect to the server - Right (_, sock, ConnectionRequestAccepted) <- socketToEndPoint ourAddress theirAddress True False False Nothing Nothing + Right (_, sock, ConnectionRequestAccepted) <- socketToEndPoint (Just ourAddress) theirAddress True False False Nothing Nothing putMVar connectionEstablished () -- Server connects to us, and then closes the connection @@ -545,7 +540,7 @@ serverAddr <- newEmptyMVar forkTry $ do - Right transport <- createTransport "127.0.0.1" "0" ((,) "128.0.0.1") defaultTCPParameters + Right transport <- createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters Right endpoint <- newEndPoint transport -- 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 @@ -565,7 +560,7 @@ forkTry $ do -- It is possible that the remote endpoint just rejects the request by closing the socket -- immediately (depending on far the remote endpoint got with the initialization) - response <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr True False False Nothing Nothing + response <- readMVar serverAddr >>= \addr -> socketToEndPoint (Just ourAddress) addr True False False Nothing Nothing case response of Right (_, _, ConnectionRequestAccepted) -> -- We don't close this socket because we want to keep this connection open @@ -573,9 +568,9 @@ -- We might get either Invalid or Crossed (the transport does not -- maintain enough history to be able to tell) Right (_, sock, ConnectionRequestInvalid) -> - N.sClose sock + N.close sock Right (_, sock, ConnectionRequestCrossed) -> - N.sClose sock + N.close sock Left _ -> return () putMVar done () @@ -589,11 +584,11 @@ -- | Test that we can create "many" transport instances testMany :: IO () testMany = do - Right masterTransport <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters + Right masterTransport <- createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters Right masterEndPoint <- newEndPoint masterTransport replicateM_ 10 $ do - mTransport <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters + mTransport <- createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters case mTransport of Left ex -> do putStrLn $ "IOException: " ++ show ex ++ "; errno = " ++ show (ioe_errno ex) @@ -610,10 +605,11 @@ -- | Test what happens when the transport breaks completely testBreakTransport :: IO () testBreakTransport = do - Right (transport, internals) <- createTransportExposeInternals "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters + Right (transport, internals) <- createTransportExposeInternals (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters Right endpoint <- newEndPoint transport - killThread (transportThread internals) -- Uh oh + let Just tid = transportThread internals + killThread tid -- Uh oh ErrorEvent (TransportError EventTransportFailed _) <- receive endpoint @@ -686,7 +682,7 @@ -- Client forkTry $ do - Right transport <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters + Right transport <- createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters Right endpoint <- newEndPoint transport let theirAddr = encodeEndPointAddress "127.0.0.1" serverPort 0 @@ -790,7 +786,7 @@ -- Client forkTry $ do - Right (transport, internals) <- createTransportExposeInternals "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters + Right (transport, internals) <- createTransportExposeInternals (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters Right endpoint <- newEndPoint transport let theirAddr = encodeEndPointAddress "127.0.0.1" serverPort 0 @@ -845,7 +841,7 @@ testInvalidCloseConnection :: IO () testInvalidCloseConnection = do - Right (transport, internals) <- createTransportExposeInternals "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters + Right (transport, internals) <- createTransportExposeInternals (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters serverAddr <- newEmptyMVar clientDone <- newEmptyMVar serverDone <- newEmptyMVar @@ -887,10 +883,10 @@ testUseRandomPort = do testDone <- newEmptyMVar forkTry $ do - Right transport1 <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters + Right transport1 <- createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters Right ep1 <- newEndPoint transport1 -- Same as transport1, but is strict in the port. - Right transport2 <- createTransport "127.0.0.1" "0" (\(!port) -> ("127.0.0.1", port)) defaultTCPParameters + Right transport2 <- createTransport (Addressable (TCPAddrInfo "127.0.0.1" "0" (\(!port) -> ("127.0.0.1", port)))) defaultTCPParameters Right ep2 <- newEndPoint transport2 Right conn1 <- connect ep2 (address ep1) ReliableOrdered defaultConnectHints ConnectionOpened _ _ _ <- receive ep1 @@ -903,7 +899,7 @@ testMaxLength :: IO () testMaxLength = do - Right serverTransport <- createTransport "127.0.0.1" "9998" ((,) "127.0.0.1") $ defaultTCPParameters { + Right serverTransport <- createTransport (defaultTCPAddr "127.0.0.1" "9998") $ defaultTCPParameters { -- 17 bytes should fit every valid address at 127.0.0.1. -- Port is at most 5 bytes (65536) and id is a base-10 Word32 so -- at most 10 bytes. We'll have one client with a 5-byte port to push it @@ -911,8 +907,8 @@ 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 + Right goodClientTransport <- createTransport (defaultTCPAddr "127.0.0.1" "9999") defaultTCPParameters + Right badClientTransport <- createTransport (defaultTCPAddr "127.0.0.1" "10000") defaultTCPParameters serverAddress <- newEmptyMVar testDone <- newEmptyMVar @@ -975,7 +971,7 @@ -- 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 transport <- createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters Right ep <- newEndPoint transport putMVar serverAddress (address ep) ConnectionOpened _ _ _ <- receive ep @@ -1019,10 +1015,10 @@ testCheckPeerHostReject = do let params = defaultTCPParameters { tcpCheckPeerHost = True } - Right transport1 <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") params + Right transport1 <- createTransport (defaultTCPAddr "127.0.0.1" "0") params -- This transport claims 127.0.0.2 as its host, but connections from it to -- an EndPoint on transport1 will show 127.0.0.1 as the socket's source host. - Right transport2 <- createTransport "127.0.0.1" "0" ((,) "127.0.0.2") defaultTCPParameters + Right transport2 <- createTransport (Addressable (TCPAddrInfo "127.0.0.1" "0" ((,) "127.0.0.2"))) defaultTCPParameters Right ep1 <- newEndPoint transport1 Right ep2 <- newEndPoint transport2 @@ -1036,13 +1032,21 @@ -- | Ensure that if peer host checking works through name resolution: if the -- peer claims "localhost", and connects to a transport also on localhost, -- it should be accepted. +-- +-- This test fails on some systems which resolve to +-- "localhost.localdomain" instead of "localhost". Disabling it for +-- now. testCheckPeerHostResolve :: IO () testCheckPeerHostResolve = do - let params = defaultTCPParameters { tcpCheckPeerHost = True } - Right transport1 <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") params + let + params = defaultTCPParameters { tcpCheckPeerHost = True } + -- This test only passes with this choice on some systems + -- localHostName = "localhost.localdomain" + localHostName = "localhost" + Right transport1 <- createTransport (defaultTCPAddr "127.0.0.1" "0") params -- EndPoints on this transport have addresses with "localhost" host part. - Right transport2 <- createTransport "127.0.0.1" "0" ((,) "localhost") defaultTCPParameters + Right transport2 <- createTransport (Addressable (TCPAddrInfo "127.0.0.1" "0" ((,) localHostName))) defaultTCPParameters Right ep1 <- newEndPoint transport1 Right ep2 <- newEndPoint transport2 @@ -1053,6 +1057,54 @@ return () +-- | Test that an unreachable EndPoint can use its own address to connect +-- to itself. +testUnreachableSelfConnect :: IO () +testUnreachableSelfConnect = do + Right transport <- createTransport Unaddressable defaultTCPParameters + Right ep <- newEndPoint transport + Right conn <- connect ep (address ep) ReliableOrdered defaultConnectHints + ConnectionOpened connid ReliableOrdered _ <- receive ep + Right () <- send conn ["ping"] + Received connid' bytes <- receive ep + _ <- close conn + ConnectionClosed connid'' <- receive ep + closeEndPoint ep + closeTransport transport + +-- | Test that +-- +-- 1. Connecting to an unreachable EndPoint's address gives ConnectFailed +-- 2. An unreachable EndPoint can successfully connect to a reachable EndPoint +-- 3. The address given in the ConnectionOpened event at the reachable EndPoint +-- can be used to connect to the unreachable EndPoint, so long as there is +-- at least one lightweight connection open between the two. +testUnreachableConnect :: IO () +testUnreachableConnect = do + Right rtransport <- createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters + Right utransport <- createTransport Unaddressable defaultTCPParameters + Right rep <- newEndPoint rtransport + Right uep <- newEndPoint utransport + -- Reachable endpoint connects to the unreachable endpoint, but it fails. + -- NB ConnectNotFound isn't the error; that would mean the address makes + -- sense but the host could not be found. + Left (TransportError ConnectFailed _) <- connect rep (address uep) ReliableOrdered defaultConnectHints + -- Unreachable endpoint connects to the reachable endpoint. + Right conn <- connect uep (address rep) ReliableOrdered defaultConnectHints + -- Reachable endpoint now has an address at which it can connect to the + -- unreachable + ConnectionOpened _ _ addr <- receive rep + Right conn' <- connect rep addr ReliableOrdered defaultConnectHints + ConnectionOpened _ _ addr' <- receive uep + close conn + ConnectionClosed _ <- receive rep + close conn' + ConnectionClosed _ <- receive uep + closeEndPoint rep + closeEndPoint uep + closeTransport rtransport + closeTransport utransport + main :: IO () main = do tcpResult <- tryIO $ runTests @@ -1061,7 +1113,7 @@ , ("EarlyCloseSocket", testEarlyCloseSocket) , ("IgnoreCloseSocket", testIgnoreCloseSocket) , ("BlockAfterCloseSocket", testBlockAfterCloseSocket) - , ("UnnecessaryConnect", testUnnecessaryConnect 10) + -- , ("UnnecessaryConnect", testUnnecessaryConnect 10) -- flaky: #91 , ("InvalidAddress", testInvalidAddress) , ("InvalidConnect", testInvalidConnect) , ("Many", testMany) @@ -1072,12 +1124,18 @@ , ("MaxLength", testMaxLength) , ("CloseEndPoint", testCloseEndPoint) , ("CheckPeerHostReject", testCheckPeerHostReject) - , ("CheckPeerHostResolve", testCheckPeerHostResolve) + -- , ("CheckPeerHostResolve", testCheckPeerHostResolve) -- flaky + , ("UnreachableSelfConnect", testUnreachableSelfConnect) + , ("UnreachableConnect", testUnreachableConnect) ] -- Run the generic tests even if the TCP specific tests failed.. - testTransport (either (Left . show) (Right) <$> - createTransport "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters) + testTransportWithFilter (`notElem` flakies) (either (Left . show) (Right) <$> + createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters) -- ..but if the generic tests pass, still fail if the specific tests did not case tcpResult of Left err -> throwIO err Right () -> return () + where + flakies = + [ "ParallelConnects" -- #92 + ]