network-transport-tcp 0.4.2 → 0.5.0

raw patch · 4 files changed

+313/−153 lines, 4 filesdep ~networkdep ~network-transport-tcpPVP ok

version bump matches the API change (PVP)

Dependency ranges changed: network, network-transport-tcp

API changes (from Hackage documentation)

Files

• ChangeLog +13/−0
• network-transport-tcp.cabal +5/−5
• src/Network/Transport/TCP.hs +204/−42
• tests/TestTCP.hs +91/−106

@@ -1,3 +1,16 @@+2016-02-17 FacundoDominguez <facundo.dominguez@tweag.io> 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 <facundo.dominguez@tweag.io> 0.4.2  * Update dependencies.

@@ -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 <facundo.dominguez@tweag.io>+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,

@@ -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)

@@ -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