diff --git a/ChangeLog b/ChangeLog --- a/ChangeLog +++ b/ChangeLog @@ -1,3 +1,16 @@ +2016-02-17 FacundoDominguez 0.5.0 + +* Add TCP_KEEPALIVE support for sockets. +* Run nearly all tests on non hardcoded ports. +* Remove obsolete top-level Makefile. +* Yield an error when tcpUserTimeout is set in unsupported systems. +* Fix for NTTCP-10. Have apiConnect respect timeouts. +* Make schedule - runScheduled pair exception safe. +* Allow to specify a default timeout for all connect calls. +* Allow to set TCP_USER_TIMEOUT on tcp connections. +* Implement configuration parameter to set TCP_NODELAY. +* Fix for NTTCP-9 / #23. Handle network failures when connection requests cross. + 2015-06-15 FacundoDominguez 0.4.2 * Update dependencies. diff --git a/network-transport-tcp.cabal b/network-transport-tcp.cabal --- a/network-transport-tcp.cabal +++ b/network-transport-tcp.cabal @@ -1,18 +1,18 @@ Name: network-transport-tcp -Version: 0.4.2 +Version: 0.5.0 Cabal-Version: >=1.10 Build-Type: Simple License: BSD3 License-file: LICENSE Copyright: Well-Typed LLP, Tweag I/O Limited Author: Duncan Coutts, Nicolas Wu, Edsko de Vries -Maintainer: Facundo Domínguez +Maintainer: edsko@well-typed.com, duncan@well-typed.com, watson.timothy@gmail.com Stability: experimental Homepage: http://haskell-distributed.github.com Bug-Reports: https://cloud-haskell.atlassian.net/browse/NTTCP Synopsis: TCP instantiation of Network.Transport Description: TCP instantiation of Network.Transport -Tested-With: GHC==7.4.2 GHC==7.6.3 GHC==7.8.4 GHC==7.10.1 +Tested-With: GHC==7.0.4 GHC==7.2.2 GHC==7.4.1 GHC==7.4.2 GHC==7.6.2 Category: Network extra-source-files: ChangeLog @@ -30,7 +30,7 @@ data-accessor >= 0.2 && < 0.3, containers >= 0.4 && < 0.6, bytestring >= 0.9 && < 0.11, - network >= 2.3 && < 2.7 + network >= 2.6.2 && < 2.7 Exposed-modules: Network.Transport.TCP, Network.Transport.TCP.Internal Default-Extensions: CPP @@ -50,7 +50,7 @@ network-transport-tests >= 0.2.1.0 && < 0.3, network >= 2.3 && < 2.7, network-transport >= 0.4.1.0 && < 0.5, - network-transport-tcp >= 0.3 && < 0.5 + network-transport-tcp ghc-options: -threaded -rtsopts -with-rtsopts=-N HS-Source-Dirs: tests default-extensions: CPP, 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 @@ -75,7 +75,8 @@ , SocketType(Stream) , defaultProtocol , setSocketOption - , SocketOption(ReuseAddr) + , SocketOption(ReuseAddr, NoDelay, UserTimeout, KeepAlive) + , isSupportedSocketOption , connect , sOMAXCONN , AddrInfo @@ -87,7 +88,14 @@ import Network.Socket.ByteString (sendMany) #endif -import Control.Concurrent (forkIO, ThreadId, killThread, myThreadId) +import Control.Concurrent + ( forkIO + , ThreadId + , killThread + , myThreadId + , threadDelay + , throwTo + ) import Control.Concurrent.Chan (Chan, newChan, readChan, writeChan) import Control.Concurrent.MVar ( MVar @@ -101,7 +109,7 @@ ) import Control.Category ((>>>)) import Control.Applicative ((<$>)) -import Control.Monad (when, unless, join, (<=<)) +import Control.Monad (when, unless, join, mplus, (<=<)) import Control.Exception ( IOException , SomeException @@ -111,6 +119,7 @@ , throwIO , try , bracketOnError + , bracket , fromException , finally , catch @@ -122,6 +131,7 @@ import qualified Data.ByteString as BS (concat) import qualified Data.ByteString.Char8 as BSC (pack, unpack) import Data.Bits (shiftL, (.|.)) +import Data.Maybe (isJust) import Data.Word (Word32) import Data.Set (Set) import qualified Data.Set as Set @@ -139,7 +149,9 @@ import Data.Accessor (Accessor, accessor, (^.), (^=), (^:)) import qualified Data.Accessor.Container as DAC (mapMaybe) import Data.Foldable (forM_, mapM_) +import qualified System.Timeout (timeout) + -- $design -- -- [Goals] @@ -176,13 +188,17 @@ -- finds that it had already sent a connection request to /A/. In this case /B/ -- will accept the connection request from /A/ if /A/s endpoint address is -- smaller (lexicographically) than /B/s, and reject it otherwise. If it rejects --- it, it sends a 'ConnectionRequestCrossed' message to /A/. (The +-- it, it sends a 'ConnectionRequestCrossed' message to /A/. The -- lexicographical ordering is an arbitrary but convenient way to break the --- tie.) +-- tie. If a connection exists between /A/ and /B/ when /B/ rejects the request, +-- /B/ will probe the connection to make sure it is healthy. If /A/ does not +-- answer timely to the probe, /B/ will discard the connection. -- -- When it receives a 'ConnectionRequestCrossed' message the /A/ thread that -- initiated the request just needs to wait until the /A/ thread that is dealing --- with /B/'s connection request completes. +-- with /B/'s connection request completes, unless there is a network failure. +-- If there is a network failure, the initiator thread would timeout and return +-- an error. -- -- [Disconnecting] -- @@ -401,6 +417,9 @@ , _remoteMaxIncoming :: !LightweightConnectionId , _remoteNextConnOutId :: !LightweightConnectionId , remoteSocket :: !N.Socket + -- | When the connection is being probed, yields an IO action that can be + -- used to release any resources dedicated to the probing. + , remoteProbing :: Maybe (IO ()) , remoteSendLock :: !(MVar ()) } @@ -430,6 +449,10 @@ | CloseConnection -- | Request to close the connection (see module description) | CloseSocket + -- | Message sent to probe a socket + | ProbeSocket + -- | Acknowledgement of the ProbeSocket message + | ProbeSocketAck deriving (Enum, Bounded, Show) -- | Response sent by /B/ to /A/ when /A/ tries to connect @@ -453,6 +476,19 @@ -- | Should we set SO_REUSEADDR on client sockets? -- Defaults to True. , tcpReuseClientAddr :: Bool + -- | Should we set TCP_NODELAY on connection sockets? + -- Defaults to True. + , tcpNoDelay :: Bool + -- | Value of TCP_USER_TIMEOUT in milliseconds + , tcpKeepAlive :: Bool + -- | Should we set TCP_KEEPALIVE on connection sockets? + , tcpUserTimeout :: Maybe Int + -- | A connect timeout for all 'connect' calls of the transport + -- in microseconds + -- + -- This can be overriden for each connect call with + -- 'ConnectHints'.'connectTimeout'. + , transportConnectTimeout :: Maybe Int } -- | Internal functionality we expose for unit testing @@ -489,6 +525,12 @@ , _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. @@ -539,6 +581,10 @@ tcpBacklog = N.sOMAXCONN , tcpReuseServerAddr = True , tcpReuseClientAddr = True + , tcpNoDelay = False + , tcpKeepAlive = False + , tcpUserTimeout = Nothing + , transportConnectTimeout = Nothing } -------------------------------------------------------------------------------- @@ -701,12 +747,16 @@ return closed RemoteEndPointValid vst -> do sched theirEndPoint $ do - tryIO $ sendOn vst [ encodeInt32 CloseSocket - , encodeInt32 (vst ^. remoteMaxIncoming) - ] + void $ tryIO $ sendOn vst [ encodeInt32 CloseSocket + , encodeInt32 (vst ^. remoteMaxIncoming) + ] + -- Release probing resources if probing. + forM_ (remoteProbing vst) id tryCloseSocket (remoteSocket vst) return closed RemoteEndPointClosing resolved vst -> do + -- Release probing resources if probing. + forM_ (remoteProbing vst) id putMVar resolved () sched theirEndPoint $ tryCloseSocket (remoteSocket vst) return closed @@ -733,6 +783,12 @@ -- handleConnectionRequest the transport will be shut down.) handleConnectionRequest :: TCPTransport -> N.Socket -> IO () handleConnectionRequest transport sock = handle handleException $ do + when (tcpNoDelay $ transportParams transport) $ + N.setSocketOption sock N.NoDelay 1 + when (tcpKeepAlive $ transportParams transport) $ + N.setSocketOption sock N.KeepAlive 1 + forM_ (tcpUserTimeout $ transportParams transport) $ + N.setSocketOption sock N.UserTimeout ourEndPointId <- recvInt32 sock theirAddress <- EndPointAddress . BS.concat <$> recvWithLength sock let ourAddress = encodeEndPointAddress (transportHost transport) @@ -756,17 +812,19 @@ mEndPoint <- handle ((>> return Nothing) . handleException) $ do resetIfBroken ourEndPoint theirAddress (theirEndPoint, isNew) <- - findRemoteEndPoint ourEndPoint theirAddress RequestedByThem + findRemoteEndPoint ourEndPoint theirAddress RequestedByThem Nothing if not isNew then do - tryIO $ sendMany sock [encodeInt32 ConnectionRequestCrossed] + void $ tryIO $ sendMany sock [encodeInt32 ConnectionRequestCrossed] + probeIfValid theirEndPoint tryCloseSocket sock return Nothing else do sendLock <- newMVar () let vst = ValidRemoteEndPointState { remoteSocket = sock + , remoteProbing = Nothing , remoteSendLock = sendLock , _remoteOutgoing = 0 , _remoteIncoming = Set.empty @@ -791,6 +849,34 @@ rethrowIfAsync :: Maybe AsyncException -> IO () rethrowIfAsync = mapM_ throwIO + probeIfValid :: RemoteEndPoint -> IO () + probeIfValid theirEndPoint = modifyMVar_ (remoteState theirEndPoint) $ + \st -> case st of + RemoteEndPointValid + vst@(ValidRemoteEndPointState { remoteProbing = Nothing }) -> do + tid <- forkIO $ do + -- send probe + let params = transportParams transport + void $ tryIO $ System.Timeout.timeout + (maybe (-1) id $ transportConnectTimeout params) $ do + sendMany (remoteSocket vst) [encodeInt32 ProbeSocket] + threadDelay maxBound + -- Discard the connection if this thread is not killed (i.e. the + -- probe ack does not arrive on time). + -- + -- The thread handling incoming messages will detect the socket is + -- closed and will report the failure upwards. + tryCloseSocket (remoteSocket vst) + -- Waiting the probe ack and closing the socket is only needed in + -- platforms where TCP_USER_TIMEOUT is not available or when the + -- user does not set it. Otherwise the ack would be handled at the + -- TCP level and the the thread handling incoming messages would + -- get the error. + + return $ RemoteEndPointValid + vst { remoteProbing = Just (killThread tid) } + _ -> return st + -- | Handle requests from a remote endpoint. -- -- Returns only if the remote party closes the socket or if an error occurs. @@ -843,6 +929,12 @@ Just CloseSocket -> do didClose <- recvInt32 sock >>= closeSocket sock unless didClose $ go sock + Just ProbeSocket -> do + forkIO $ sendMany sock [encodeInt32 ProbeSocketAck] + go sock + Just ProbeSocketAck -> do + stopProbing + go sock Nothing -> throwIO $ userError "Invalid control request" @@ -939,12 +1031,14 @@ then return (RemoteEndPointValid vst', Nothing) else do + -- Release probing resources if probing. + forM_ (remoteProbing vst) id removeRemoteEndPoint (ourEndPoint, theirEndPoint) -- Attempt to reply (but don't insist) act <- schedule theirEndPoint $ do - tryIO $ sendOn vst' [ encodeInt32 CloseSocket - , encodeInt32 (vst ^. remoteMaxIncoming) - ] + void $ tryIO $ sendOn vst' [ encodeInt32 CloseSocket + , encodeInt32 (vst ^. remoteMaxIncoming) + ] tryCloseSocket sock return (RemoteEndPointClosed, Just act) RemoteEndPointClosing resolved vst -> do @@ -959,6 +1053,8 @@ then do return (RemoteEndPointClosing resolved vst, Nothing) else do + -- Release probing resources if probing. + forM_ (remoteProbing vst) id removeRemoteEndPoint (ourEndPoint, theirEndPoint) act <- schedule theirEndPoint $ tryCloseSocket sock putMVar resolved () @@ -981,6 +1077,15 @@ readMessage sock lcid = recvWithLength sock >>= writeChan ourChannel . Received (connId lcid) + -- Stop probing a connection as a result of receiving a probe ack. + stopProbing :: IO () + stopProbing = modifyMVar_ theirState $ \st -> case st of + RemoteEndPointValid + vst@(ValidRemoteEndPointState { remoteProbing = Just stop }) -> do + stop + return $ RemoteEndPointValid vst { remoteProbing = Nothing } + _ -> return st + -- Arguments ourChannel = localChannel ourEndPoint theirState = remoteState theirEndPoint @@ -998,11 +1103,15 @@ RemoteEndPointInit _ _ _ -> relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:prematureExit" - RemoteEndPointValid _ -> do + RemoteEndPointValid vst -> do + -- Release probing resources if probing. + forM_ (remoteProbing vst) id let code = EventConnectionLost (remoteAddress theirEndPoint) writeChan ourChannel . ErrorEvent $ TransportError code (show err) return (RemoteEndPointFailed err) - RemoteEndPointClosing resolved _ -> do + RemoteEndPointClosing resolved vst -> do + -- Release probing resources if probing. + forM_ (remoteProbing vst) id putMVar resolved () return (RemoteEndPointFailed err) RemoteEndPointClosed -> @@ -1040,17 +1149,40 @@ -> EndPointAddress -> ConnectHints -> IO (RemoteEndPoint, LightweightConnectionId) -createConnectionTo params ourEndPoint theirAddress hints = go +createConnectionTo params ourEndPoint theirAddress hints = do + -- @timer@ is an IO action that completes when the timeout expires. + timer <- case connTimeout of + Just t -> do + mv <- newEmptyMVar + _ <- forkIO $ threadDelay t >> putMVar mv () + return $ Just $ readMVar mv + _ -> return Nothing + go timer Nothing + where - go = do - (theirEndPoint, isNew) <- mapIOException connectFailed $ - findRemoteEndPoint ourEndPoint theirAddress RequestedByUs + connTimeout = connectTimeout hints `mplus` transportConnectTimeout params + + -- The second argument indicates the response obtained to the last + -- connection request and the remote endpoint that was used. + go timer mr = do + (theirEndPoint, isNew) <- mapIOException connectFailed + (findRemoteEndPoint ourEndPoint theirAddress RequestedByUs timer) + `finally` case mr of + Just (theirEndPoint, ConnectionRequestCrossed) -> + modifyMVar_ (remoteState theirEndPoint) $ + \rst -> case rst of + RemoteEndPointInit resolved _ _ -> do + putMVar resolved () + removeRemoteEndPoint (ourEndPoint, theirEndPoint) + return RemoteEndPointClosed + _ -> return rst + _ -> return () if isNew then do - forkIO . handle absorbAllExceptions $ - setupRemoteEndPoint params (ourEndPoint, theirEndPoint) hints - go + mr' <- handle (absorbAllExceptions Nothing) $ + setupRemoteEndPoint params (ourEndPoint, theirEndPoint) connTimeout + go timer (fmap ((,) theirEndPoint) mr') else do -- 'findRemoteEndPoint' will have increased 'remoteOutgoing' mapIOException connectFailed $ do @@ -1081,22 +1213,27 @@ connectFailed :: IOException -> TransportError ConnectErrorCode connectFailed = TransportError ConnectFailed . show - absorbAllExceptions :: SomeException -> IO () - absorbAllExceptions _ex = - return () + absorbAllExceptions :: a -> SomeException -> IO a + absorbAllExceptions a _ex = + return a -- | Set up a remote endpoint -setupRemoteEndPoint :: TCPParameters -> EndPointPair -> ConnectHints -> IO () -setupRemoteEndPoint params (ourEndPoint, theirEndPoint) hints = do +setupRemoteEndPoint :: TCPParameters -> EndPointPair -> Maybe Int + -> IO (Maybe ConnectionRequestResponse) +setupRemoteEndPoint params (ourEndPoint, theirEndPoint) connTimeout = do result <- socketToEndPoint ourAddress theirAddress (tcpReuseClientAddr params) - (connectTimeout hints) + (tcpNoDelay params) + (tcpKeepAlive params) + (tcpUserTimeout params) + connTimeout didAccept <- case result of Right (sock, ConnectionRequestAccepted) -> do sendLock <- newMVar () let vst = ValidRemoteEndPointState { remoteSocket = sock + , remoteProbing = Nothing , remoteSendLock = sendLock , _remoteOutgoing = 0 , _remoteIncoming = Set.empty @@ -1124,7 +1261,9 @@ resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointInvalid err) return False - when didAccept $ handleIncomingMessages (ourEndPoint, theirEndPoint) + when didAccept $ void $ forkIO $ + handleIncomingMessages (ourEndPoint, theirEndPoint) + return $ either (const Nothing) (Just . snd) result where ourAddress = localAddress ourEndPoint theirAddress = remoteAddress theirEndPoint @@ -1325,13 +1464,15 @@ return TransportClosed -- | Find a remote endpoint. If the remote endpoint does not yet exist we --- create it in Init state. Returns if the endpoint was new. +-- create it in Init state. Returns if the endpoint was new, or 'Nothing' if +-- it times out. findRemoteEndPoint :: LocalEndPoint -> EndPointAddress -> RequestedBy + -> Maybe (IO ()) -- ^ an action which completes when the time is up -> IO (RemoteEndPoint, Bool) -findRemoteEndPoint ourEndPoint theirAddress findOrigin = go +findRemoteEndPoint ourEndPoint theirAddress findOrigin mtimer = go where go = do (theirEndPoint, isNew) <- modifyMVar ourState $ \st -> case st of @@ -1384,14 +1525,14 @@ RemoteEndPointInit resolved crossed initOrigin -> case (findOrigin, initOrigin) of (RequestedByUs, RequestedByUs) -> - readMVar resolved >> go + readMVarTimeout mtimer resolved >> go (RequestedByUs, RequestedByThem) -> - readMVar resolved >> go + readMVarTimeout mtimer resolved >> go (RequestedByThem, RequestedByUs) -> if ourAddress > theirAddress then do -- Wait for the Crossed message - readMVar crossed + readMVarTimeout mtimer crossed return (theirEndPoint, True) else return (theirEndPoint, False) @@ -1404,7 +1545,7 @@ -- maintain enough history to be able to tell the difference). return (theirEndPoint, False) RemoteEndPointClosing resolved _ -> - readMVar resolved >> go + readMVarTimeout mtimer resolved >> go RemoteEndPointClosed -> go RemoteEndPointFailed err -> @@ -1413,6 +1554,14 @@ ourState = localState ourEndPoint ourAddress = localAddress ourEndPoint + -- | Like 'readMVar' but it throws an exception if the timer expires. + readMVarTimeout Nothing mv = readMVar mv + readMVarTimeout (Just timer) mv = do + let connectTimedout = TransportError ConnectTimeout "Timed out" + tid <- myThreadId + bracket (forkIO $ timer >> throwTo tid connectTimedout) killThread $ + const $ readMVar mv + -- | Send a payload over a heavyweight connection (thread safe) sendOn :: ValidRemoteEndPointState -> [ByteString] -> IO () sendOn vst bs = withMVar (remoteSendLock vst) $ \() -> @@ -1461,6 +1610,8 @@ -> ValidRemoteEndPointState -> IO RemoteState handleIOException ex vst = do + -- Release probing resources if probing. + forM_ (remoteProbing vst) id tryCloseSocket (remoteSocket vst) let code = EventConnectionLost (remoteAddress theirEndPoint) err = TransportError code (show ex) @@ -1486,10 +1637,14 @@ socketToEndPoint :: EndPointAddress -- ^ Our address -> EndPointAddress -- ^ Their address -> Bool -- ^ Use SO_REUSEADDR? + -> Bool -- ^ Use TCP_NODELAY + -> Bool -- ^ Use TCP_KEEPALIVE + -> Maybe Int -- ^ Maybe TCP_USER_TIMEOUT -> Maybe Int -- ^ Timeout for connect -> IO (Either (TransportError ConnectErrorCode) (N.Socket, ConnectionRequestResponse)) -socketToEndPoint (EndPointAddress ourAddress) theirAddress reuseAddr timeout = +socketToEndPoint (EndPointAddress ourAddress) theirAddress reuseAddr noDelay keepAlive + mUserTimeout timeout = try $ do (host, port, theirEndPointId) <- case decodeEndPointAddress theirAddress of Nothing -> throwIO (failed . userError $ "Could not parse") @@ -1499,12 +1654,19 @@ bracketOnError (createSocket addr) tryCloseSocket $ \sock -> do when reuseAddr $ mapIOException failed $ N.setSocketOption sock N.ReuseAddr 1 - mapIOException invalidAddress $ - timeoutMaybe timeout timeoutError $ + when noDelay $ + mapIOException failed $ N.setSocketOption sock N.NoDelay 1 + when keepAlive $ + mapIOException failed $ N.setSocketOption sock N.KeepAlive 1 + forM_ mUserTimeout $ + mapIOException failed . N.setSocketOption sock N.UserTimeout + response <- timeoutMaybe timeout timeoutError $ do + mapIOException invalidAddress $ N.connect sock (N.addrAddress addr) - response <- mapIOException failed $ do - sendMany sock (encodeInt32 theirEndPointId : prependLength [ourAddress]) - recvInt32 sock + mapIOException failed $ do + sendMany sock + (encodeInt32 theirEndPointId : prependLength [ourAddress]) + recvInt32 sock case tryToEnum response of Nothing -> throwIO (failed . userError $ "Unexpected response") Just r -> return (sock, r) diff --git a/tests/TestTCP.hs b/tests/TestTCP.hs --- a/tests/TestTCP.hs +++ b/tests/TestTCP.hs @@ -90,8 +90,8 @@ -- Test that the server gets a ConnectionClosed message when the client closes -- the socket without sending an explicit control message to the server first -testEarlyDisconnect :: IO N.ServiceName -> IO () -testEarlyDisconnect nextPort = do +testEarlyDisconnect :: IO () +testEarlyDisconnect = do clientAddr <- newEmptyMVar serverAddr <- newEmptyMVar serverDone <- newEmptyMVar @@ -105,7 +105,7 @@ server :: MVar EndPointAddress -> MVar EndPointAddress -> MVar () -> IO () server serverAddr clientAddr serverDone = do tlog "Server" - Right transport <- nextPort >>= \port -> createTransport "127.0.0.1" port defaultTCPParameters + Right transport <- createTransport "127.0.0.1" "0" defaultTCPParameters Right endpoint <- newEndPoint transport putMVar serverAddr (address endpoint) theirAddr <- readMVar clientAddr @@ -152,12 +152,9 @@ client :: MVar EndPointAddress -> MVar EndPointAddress -> IO () client serverAddr clientAddr = do tlog "Client" - clientPort <- nextPort - let ourAddress = encodeEndPointAddress "127.0.0.1" clientPort 0 - putMVar clientAddr ourAddress -- Listen for incoming messages - forkServer "127.0.0.1" clientPort 5 True throwIO $ \sock -> do + (clientPort, _) <- forkServer "127.0.0.1" "0" 5 True throwIO $ \sock -> do -- Initial setup 0 <- recvInt32 sock :: IO Int _ <- recvWithLength sock @@ -178,8 +175,11 @@ -- Close the socket N.sClose sock + let ourAddress = encodeEndPointAddress "127.0.0.1" clientPort 0 + putMVar clientAddr ourAddress + -- Connect to the server - Right (sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr True Nothing + Right (sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr True False False Nothing Nothing -- Open a new connection sendMany sock [encodeInt32 CreatedNewConnection, encodeInt32 (10003 :: Int)] @@ -189,8 +189,8 @@ N.sClose sock -- | Test the behaviour of a premature CloseSocket request -testEarlyCloseSocket :: IO N.ServiceName -> IO () -testEarlyCloseSocket nextPort = do +testEarlyCloseSocket :: IO () +testEarlyCloseSocket = do clientAddr <- newEmptyMVar serverAddr <- newEmptyMVar serverDone <- newEmptyMVar @@ -204,7 +204,7 @@ server :: MVar EndPointAddress -> MVar EndPointAddress -> MVar () -> IO () server serverAddr clientAddr serverDone = do tlog "Server" - Right transport <- nextPort >>= \port -> createTransport "127.0.0.1" port defaultTCPParameters + Right transport <- createTransport "127.0.0.1" "0" defaultTCPParameters Right endpoint <- newEndPoint transport putMVar serverAddr (address endpoint) theirAddr <- readMVar clientAddr @@ -258,12 +258,9 @@ client :: MVar EndPointAddress -> MVar EndPointAddress -> IO () client serverAddr clientAddr = do tlog "Client" - clientPort <- nextPort - let ourAddress = encodeEndPointAddress "127.0.0.1" clientPort 0 - putMVar clientAddr ourAddress -- Listen for incoming messages - forkServer "127.0.0.1" clientPort 5 True throwIO $ \sock -> do + (clientPort, _) <- forkServer "127.0.0.1" "0" 5 True throwIO $ \sock -> do -- Initial setup 0 <- recvInt32 sock :: IO Int _ <- recvWithLength sock @@ -286,8 +283,11 @@ sendMany sock [encodeInt32 CloseSocket, encodeInt32 (1024 :: Int)] N.sClose sock + let ourAddress = encodeEndPointAddress "127.0.0.1" clientPort 0 + putMVar clientAddr ourAddress + -- Connect to the server - Right (sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr True Nothing + Right (sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr True False False Nothing Nothing -- Open a new connection sendMany sock [encodeInt32 CreatedNewConnection, encodeInt32 (10003 :: Int)] @@ -298,16 +298,15 @@ N.sClose sock -- | Test the creation of a transport with an invalid address -testInvalidAddress :: IO N.ServiceName -> IO () -testInvalidAddress nextPort = do - Left _ <- nextPort >>= \port -> createTransport "invalidHostName" port defaultTCPParameters +testInvalidAddress :: IO () +testInvalidAddress = do + Left _ <- createTransport "invalidHostName" "0" defaultTCPParameters return () -- | Test connecting to invalid or non-existing endpoints -testInvalidConnect :: IO N.ServiceName -> IO () -testInvalidConnect nextPort = do - port <- nextPort - Right transport <- createTransport "127.0.0.1" port defaultTCPParameters +testInvalidConnect :: IO () +testInvalidConnect = do + Right transport <- createTransport "127.0.0.1" "0" defaultTCPParameters Right endpoint <- newEndPoint transport -- Syntax error in the endpoint address @@ -324,21 +323,21 @@ -- Valid TCP address but invalid endpoint number Left (TransportError ConnectNotFound _) <- - connect endpoint (encodeEndPointAddress "127.0.0.1" port 1) ReliableOrdered defaultConnectHints + connect endpoint (encodeEndPointAddress "127.0.0.1" "0" 1) ReliableOrdered defaultConnectHints return () -- | Test that an endpoint can ignore CloseSocket requests (in "reality" this -- would happen when the endpoint sends a new connection request before -- receiving an (already underway) CloseSocket request) -testIgnoreCloseSocket :: IO N.ServiceName -> IO () -testIgnoreCloseSocket nextPort = do +testIgnoreCloseSocket :: IO () +testIgnoreCloseSocket = do serverAddr <- newEmptyMVar clientAddr <- newEmptyMVar clientDone <- newEmptyMVar serverDone <- newEmptyMVar connectionEstablished <- newEmptyMVar - Right transport <- nextPort >>= \port -> createTransport "127.0.0.1" port defaultTCPParameters + Right transport <- createTransport "127.0.0.1" "0" defaultTCPParameters -- Server forkTry $ do @@ -376,7 +375,7 @@ theirAddress <- readMVar serverAddr -- Connect to the server - Right (sock, ConnectionRequestAccepted) <- socketToEndPoint ourAddress theirAddress True Nothing + Right (sock, ConnectionRequestAccepted) <- socketToEndPoint ourAddress theirAddress True False False Nothing Nothing putMVar connectionEstablished () -- Server connects to us, and then closes the connection @@ -412,14 +411,14 @@ -- | Like 'testIgnoreSocket', but now the server requests a connection after the -- client closed their connection. In the meantime, the server will have sent a -- CloseSocket request to the client, and must block until the client responds. -testBlockAfterCloseSocket :: IO N.ServiceName -> IO () -testBlockAfterCloseSocket nextPort = do +testBlockAfterCloseSocket :: IO () +testBlockAfterCloseSocket = do serverAddr <- newEmptyMVar clientAddr <- newEmptyMVar clientDone <- newEmptyMVar serverDone <- newEmptyMVar connectionEstablished <- newEmptyMVar - Right transport <- nextPort >>= \port -> createTransport "127.0.0.1" port defaultTCPParameters + Right transport <- createTransport "127.0.0.1" "0" defaultTCPParameters -- Server forkTry $ do @@ -453,7 +452,7 @@ theirAddress <- readMVar serverAddr -- Connect to the server - Right (sock, ConnectionRequestAccepted) <- socketToEndPoint ourAddress theirAddress True Nothing + Right (sock, ConnectionRequestAccepted) <- socketToEndPoint ourAddress theirAddress True False False Nothing Nothing putMVar connectionEstablished () -- Server connects to us, and then closes the connection @@ -490,13 +489,13 @@ -- | Test what happens when a remote endpoint sends a connection request to our -- transport for an endpoint it already has a connection to -testUnnecessaryConnect :: IO N.ServiceName -> Int -> IO () -testUnnecessaryConnect nextPort numThreads = do +testUnnecessaryConnect :: Int -> IO () +testUnnecessaryConnect numThreads = do clientDone <- newEmptyMVar serverAddr <- newEmptyMVar forkTry $ do - Right transport <- nextPort >>= \port -> createTransport "127.0.0.1" port defaultTCPParameters + Right transport <- createTransport "127.0.0.1" "0" defaultTCPParameters Right endpoint <- newEndPoint transport putMVar serverAddr (address endpoint) @@ -511,7 +510,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 Nothing + response <- readMVar serverAddr >>= \addr -> socketToEndPoint 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 @@ -533,13 +532,13 @@ takeMVar clientDone -- | Test that we can create "many" transport instances -testMany :: IO N.ServiceName -> IO () -testMany nextPort = do - Right masterTransport <- nextPort >>= \port -> createTransport "127.0.0.1" port defaultTCPParameters +testMany :: IO () +testMany = do + Right masterTransport <- createTransport "127.0.0.1" "0" defaultTCPParameters Right masterEndPoint <- newEndPoint masterTransport replicateM_ 10 $ do - mTransport <- nextPort >>= \port -> createTransport "127.0.0.1" port defaultTCPParameters + mTransport <- createTransport "127.0.0.1" "0" defaultTCPParameters case mTransport of Left ex -> do putStrLn $ "IOException: " ++ show ex ++ "; errno = " ++ show (ioe_errno ex) @@ -554,9 +553,9 @@ return () -- | Test what happens when the transport breaks completely -testBreakTransport :: IO N.ServiceName -> IO () -testBreakTransport nextPort = do - Right (transport, internals) <- nextPort >>= \port -> createTransportExposeInternals "127.0.0.1" port defaultTCPParameters +testBreakTransport :: IO () +testBreakTransport = do + Right (transport, internals) <- createTransportExposeInternals "127.0.0.1" "0" defaultTCPParameters Right endpoint <- newEndPoint transport killThread (transportThread internals) -- Uh oh @@ -572,58 +571,48 @@ -- Then test that we get a connection lost message after the remote endpoint -- suddenly closes the socket, and that a subsequent 'connect' allows us to -- re-establish a connection to the same endpoint -testReconnect :: IO N.ServiceName -> IO () -testReconnect nextPort = do - serverPort <- nextPort +testReconnect :: IO () +testReconnect = do serverDone <- newEmptyMVar - firstAttempt <- newEmptyMVar endpointCreated <- newEmptyMVar - -- Server - forkTry $ do - -- Wait for the client to do its first attempt - readMVar firstAttempt - - counter <- newMVar (0 :: Int) + counter <- newMVar (0 :: Int) - forkServer "127.0.0.1" serverPort 5 True throwIO $ \sock -> do - -- Accept the connection - Right 0 <- tryIO $ (recvInt32 sock :: IO Int) - Right _ <- tryIO $ recvWithLength sock + -- Server + (serverPort, _) <- forkServer "127.0.0.1" "0" 5 True throwIO $ \sock -> do + -- Accept the connection + Right 0 <- tryIO $ (recvInt32 sock :: IO Int) + Right _ <- tryIO $ recvWithLength sock - -- The first time we close the socket before accepting the logical connection - count <- modifyMVar counter $ \i -> return (i + 1, i) + -- The first time we close the socket before accepting the logical connection + count <- modifyMVar counter $ \i -> return (i + 1, i) - when (count > 0) $ do - Right () <- tryIO $ sendMany sock [encodeInt32 ConnectionRequestAccepted] - -- Client requests a logical connection - when (count > 1) $ do - Right CreatedNewConnection <- tryIO $ toEnum <$> (recvInt32 sock :: IO Int) - connId <- recvInt32 sock :: IO LightweightConnectionId - return () + when (count > 0) $ do + Right () <- tryIO $ sendMany sock [encodeInt32 ConnectionRequestAccepted] + -- Client requests a logical connection + when (count > 1) $ do + Right CreatedNewConnection <- tryIO $ toEnum <$> (recvInt32 sock :: IO Int) + connId <- recvInt32 sock :: IO LightweightConnectionId + return () - when (count > 2) $ do - -- Client sends a message - Right connId' <- tryIO $ (recvInt32 sock :: IO LightweightConnectionId) - True <- return $ connId == connId' - Right ["ping"] <- tryIO $ recvWithLength sock - putMVar serverDone () + when (count > 2) $ do + -- Client sends a message + Right connId' <- tryIO $ (recvInt32 sock :: IO LightweightConnectionId) + True <- return $ connId == connId' + Right ["ping"] <- tryIO $ recvWithLength sock + putMVar serverDone () - Right () <- tryIO $ N.sClose sock - return () + Right () <- tryIO $ N.sClose sock + return () - putMVar endpointCreated () + putMVar endpointCreated () -- Client forkTry $ do - Right transport <- nextPort >>= \port -> createTransport "127.0.0.1" port defaultTCPParameters + Right transport <- createTransport "127.0.0.1" "0" defaultTCPParameters Right endpoint <- newEndPoint transport let theirAddr = encodeEndPointAddress "127.0.0.1" serverPort 0 - -- The first attempt will fail because no endpoint is yet set up - Left (TransportError ConnectNotFound _) <- connect endpoint theirAddr ReliableOrdered defaultConnectHints - putMVar firstAttempt () - -- The second attempt will fail because the server closes the socket before we can request a connection takeMVar endpointCreated -- This might time out or not, depending on whether the server closes the @@ -667,14 +656,13 @@ -- 'recv' in 'handleIncomingMessages' will not fail, but a 'send' or 'connect' -- *will* fail. We are testing that error handling everywhere does the right -- thing. -testUnidirectionalError :: IO N.ServiceName -> IO () -testUnidirectionalError nextPort = do +testUnidirectionalError :: IO () +testUnidirectionalError = do clientDone <- newEmptyMVar - serverPort <- nextPort serverGotPing <- newEmptyMVar -- Server - forkServer "127.0.0.1" serverPort 5 True throwIO $ \sock -> do + (serverPort, _) <- forkServer "127.0.0.1" "0" 5 True throwIO $ \sock -> do -- We accept connections, but when an exception occurs we don't do -- anything (in particular, we don't close the socket). This is important -- because when we shutdown one direction of the socket a recv here will @@ -695,7 +683,7 @@ -- Client forkTry $ do - Right (transport, internals) <- nextPort >>= \port -> createTransportExposeInternals "127.0.0.1" port defaultTCPParameters + Right (transport, internals) <- createTransportExposeInternals "127.0.0.1" "0" defaultTCPParameters Right endpoint <- newEndPoint transport let theirAddr = encodeEndPointAddress "127.0.0.1" serverPort 0 @@ -748,9 +736,9 @@ takeMVar clientDone -testInvalidCloseConnection :: IO N.ServiceName -> IO () -testInvalidCloseConnection nextPort = do - Right (transport, internals) <- nextPort >>= \port -> createTransportExposeInternals "127.0.0.1" port defaultTCPParameters +testInvalidCloseConnection :: IO () +testInvalidCloseConnection = do + Right (transport, internals) <- createTransportExposeInternals "127.0.0.1" "0" defaultTCPParameters serverAddr <- newEmptyMVar clientDone <- newEmptyMVar serverDone <- newEmptyMVar @@ -798,29 +786,26 @@ putMVar testDone () takeMVar testDone - main :: IO () main = do - portMVar <- newEmptyMVar - forkTry $ forM_ ([10080 ..] :: [Int]) $ putMVar portMVar . show - let nextPort = takeMVar portMVar tcpResult <- tryIO $ runTests - [ ("EarlyDisconnect", testEarlyDisconnect nextPort) - , ("EarlyCloseSocket", testEarlyCloseSocket nextPort) - , ("IgnoreCloseSocket", testIgnoreCloseSocket nextPort) - , ("BlockAfterCloseSocket", testBlockAfterCloseSocket nextPort) - , ("UnnecessaryConnect", testUnnecessaryConnect nextPort 10) - , ("InvalidAddress", testInvalidAddress nextPort) - , ("InvalidConnect", testInvalidConnect nextPort) - , ("Many", testMany nextPort) - , ("BreakTransport", testBreakTransport nextPort) - , ("Reconnect", testReconnect nextPort) - , ("UnidirectionalError", testUnidirectionalError nextPort) - , ("InvalidCloseConnection", testInvalidCloseConnection nextPort) - , ("Use random port" , testUseRandomPort) + [ ("Use random port" , testUseRandomPort) + , ("EarlyDisconnect", testEarlyDisconnect) + , ("EarlyCloseSocket", testEarlyCloseSocket) + , ("IgnoreCloseSocket", testIgnoreCloseSocket) + , ("BlockAfterCloseSocket", testBlockAfterCloseSocket) + , ("UnnecessaryConnect", testUnnecessaryConnect 10) + , ("InvalidAddress", testInvalidAddress) + , ("InvalidConnect", testInvalidConnect) + , ("Many", testMany) + , ("BreakTransport", testBreakTransport) + , ("Reconnect", testReconnect) + , ("UnidirectionalError", testUnidirectionalError) + , ("InvalidCloseConnection", testInvalidCloseConnection) ] -- Run the generic tests even if the TCP specific tests failed.. - testTransport (either (Left . show) (Right) <$> nextPort >>= \port -> createTransport "127.0.0.1" port defaultTCPParameters) + testTransport (either (Left . show) (Right) <$> + createTransport "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