network-transport-tcp 0.6.1 → 0.8.6
raw patch · 6 files changed
Files
- ChangeLog +28/−2
- Setup.hs +0/−2
- network-transport-tcp.cabal +35/−23
- src/Network/Transport/TCP.hs +205/−136
- src/Network/Transport/TCP/Internal.hs +22/−23
- tests/TestTCP.hs +111/−53
ChangeLog view
@@ -1,6 +1,32 @@-next release+2025-02-04 Laurent P. René de Cotret <laurent.decotret@outlook.com> 0.8.6 -*+* Ported test suite to use `tasty` rather than `test-framework`.+* Addressed some compilation warnings (#467)++2024-09-03 Laurent P. René de Cotret <laurent.decotret@outlook.com> 0.8.5++* Bumped dependency bounds to support GHC 8.10.7 - GHC 9.10.1+* Updated links to point to Distributed Haskell monorepo++2024-07-09 Laurent P. René de Cotret <laurent.decotret@outlook.com> 0.8.4++* Set TCP_NODELAY by default. The documentation erroneously reported it as the default. (#66)++2023-03-26 David Simmons-Duffin <dsd@caltech.edu> 0.8.3++* Disable a flaky test++2023-03-25 David Simmons-Duffin <dsd@caltech.edu> 0.8.2++* Bump bytestring version to build with GHC 9.8.++2022-10-12 FacundoDominguez <facundo.dominguez@tweag.io> 0.8.1++* Fix imports for ghc 9 (#93).++2020-10-09 FacundoDominguez <facundo.dominguez@tweag.io> 0.8.0++* Move to network 3. 2019-12-31 FacundoDominguez <facundo.dominguez@tweag.io> 0.7.0
− Setup.hs
@@ -1,2 +0,0 @@-import Distribution.Simple-main = defaultMain
network-transport-tcp.cabal view
@@ -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 <facundo.dominguez@tweag.io>+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,
src/Network/Transport/TCP.hs view
@@ -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)
src/Network/Transport/TCP/Internal.hs view
@@ -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 "
tests/TestTCP.hs view
@@ -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+ ]