network-transport-tcp 0.5.1 → 0.6.0

raw patch · 5 files changed

+1187/−443 lines, 5 filesdep ~bytestringdep ~network-transportPVP ok

version bump matches the API change (PVP)

Dependency ranges changed: bytestring, network-transport

API changes (from Hackage documentation)

Files

• ChangeLog +14/−0
• network-transport-tcp.cabal +4/−3
• src/Network/Transport/TCP.hs +547/−280
• src/Network/Transport/TCP/Internal.hs +254/−43
• tests/TestTCP.hs +368/−117

@@ -1,3 +1,17 @@+2017-08-21 FacundoDominguez <facundo.dominguez@tweag.io> 0.6.0++* Implemented protocol versioning (#55)+* Extend interface so queue policies and lengths can be configured.+* Check peer-reported host against socket host (#54)+* Test improvements+* Fix races when an EndPoint is closed or failed (#60)+* Fix timeout socket connections (#53)+* Use equality rather than ordering in socket close (#56)+* apiCloseEndPoint blocks until no longer reciving+* Shutdown sockets when closing endpoints+* Allow computing the external address from the chosen bind port (#50)+* Discard remote endpoints when they close or fail (#43)+ 2016-02-17 FacundoDominguez <facundo.dominguez@tweag.io> 0.5.0  * Add TCP_KEEPALIVE support for sockets.

@@ -1,5 +1,5 @@ Name:          network-transport-tcp-Version:       0.5.1+Version:       0.6.0 Cabal-Version: >=1.10 Build-Type:    Simple License:       BSD3@@ -26,7 +26,7 @@  Library   Build-Depends:   base >= 4.3 && < 5,-                   network-transport >= 0.4.1.0 && < 0.5,+                   network-transport >= 0.5 && < 0.6,                    data-accessor >= 0.2 && < 0.3,                    containers >= 0.4 && < 0.6,                    bytestring >= 0.9 && < 0.11,@@ -47,9 +47,10 @@   Type:            exitcode-stdio-1.0   Main-Is:         TestTCP.hs   Build-Depends:   base >= 4.3 && < 5,+                   bytestring >= 0.9 && < 0.11,                    network-transport-tests >= 0.2.1.0 && < 0.3,                    network >= 2.3 && < 2.7,-                   network-transport >= 0.4.1.0 && < 0.5,+                   network-transport,                    network-transport-tcp   ghc-options:     -threaded -rtsopts -with-rtsopts=-N   HS-Source-Dirs:  tests

@@ -13,6 +13,7 @@  {-# LANGUAGE RecursiveDo #-} {-# LANGUAGE TupleSections #-}+{-# LANGUAGE RankNTypes #-}  module Network.Transport.TCP   ( -- * Main API@@ -23,14 +24,15 @@   , createTransportExposeInternals   , TransportInternals(..)   , EndPointId-  , encodeEndPointAddress-  , decodeEndPointAddress   , ControlHeader(..)   , ConnectionRequestResponse(..)   , firstNonReservedLightweightConnectionId   , firstNonReservedHeavyweightConnectionId   , socketToEndPoint   , LightweightConnectionId+  , QDisc(..)+  , simpleUnboundedQDisc+  , simpleOnePlaceQDisc     -- * Design notes     -- $design   ) where@@ -44,14 +46,27 @@  import Network.Transport import Network.Transport.TCP.Internal-  ( forkServer+  ( ControlHeader(..)+  , encodeControlHeader+  , decodeControlHeader+  , ConnectionRequestResponse(..)+  , encodeConnectionRequestResponse+  , decodeConnectionRequestResponse+  , forkServer   , recvWithLength-  , recvInt32+  , recvExact+  , recvWord32+  , encodeWord32   , tryCloseSocket+  , tryShutdownSocketBoth+  , decodeSockAddr+  , EndPointId+  , encodeEndPointAddress+  , decodeEndPointAddress+  , currentProtocolVersion   ) import Network.Transport.Internal-  ( encodeInt32-  , prependLength+  ( prependLength   , mapIOException   , tryIO   , tryToEnum@@ -80,6 +95,7 @@   , connect   , sOMAXCONN   , AddrInfo+  , SockAddr(..)   )  #ifdef USE_MOCK_NETWORK@@ -103,6 +119,7 @@   , modifyMVar   , modifyMVar_   , readMVar+  , takeMVar   , putMVar   , newEmptyMVar   , withMVar@@ -150,7 +167,7 @@ import qualified Data.Accessor.Container as DAC (mapMaybe) import Data.Foldable (forM_, mapM_) import qualified System.Timeout (timeout)-+import qualified Data.ByteString as BS (length)  -- $design --@@ -264,10 +281,12 @@ --   ValidRemoteEndPointState).  data TCPTransport = TCPTransport-  { transportHost   :: !N.HostName-  , transportPort   :: !N.ServiceName-  , transportState  :: !(MVar TransportState)-  , transportParams :: !TCPParameters+  { transportHost     :: !N.HostName+  , transportPort     :: !N.ServiceName+  , transportBindHost :: !N.HostName+  , transportBindPort :: !N.ServiceName+  , transportState    :: !(MVar TransportState)+  , transportParams   :: !TCPParameters   }  data TransportState =@@ -281,8 +300,10 @@  data LocalEndPoint = LocalEndPoint   { localAddress :: !EndPointAddress-  , localChannel :: !(Chan Event)   , localState   :: !(MVar LocalEndPointState)+    -- | A 'QDisc' is held here rather than on the 'ValidLocalEndPointState'+    --   because even closed 'LocalEndPoint's can have queued input data.+  , localQueue   :: !(QDisc Event)   }  data LocalEndPointState =@@ -414,18 +435,20 @@ data ValidRemoteEndPointState = ValidRemoteEndPointState   { _remoteOutgoing      :: !Int   , _remoteIncoming      :: !(Set LightweightConnectionId)-  , _remoteMaxIncoming   :: !LightweightConnectionId+  , _remoteLastIncoming  :: !LightweightConnectionId   , _remoteNextConnOutId :: !LightweightConnectionId   ,  remoteSocket        :: !N.Socket      -- | When the connection is being probed, yields an IO action that can be      -- used to release any resources dedicated to the probing.   ,  remoteProbing       :: Maybe (IO ())   ,  remoteSendLock      :: !(MVar ())+     -- | An IO which returns when the socket (remoteSocket) has been closed.+     --   The program/thread which created the socket is always responsible+     --   for closing it, but sometimes other threads need to know when this+     --   happens.+  ,  remoteSocketClosed  :: !(IO ())   } --- | Local identifier for an endpoint within this transport-type EndPointId = Word32- -- | Pair of local and a remote endpoint (for conciseness in signatures) type EndPointPair = (LocalEndPoint, RemoteEndPoint) @@ -441,30 +464,6 @@ -- 'LightweightConnectionId'. type HeavyweightConnectionId = Word32 --- | Control headers-data ControlHeader =-    -- | Tell the remote endpoint that we created a new connection-    CreatedNewConnection-    -- | Tell the remote endpoint we will no longer be using a connection-  | CloseConnection-    -- | Request to close the connection (see module description)-  | CloseSocket-    -- | Message sent to probe a socket-  | ProbeSocket-    -- | Acknowledgement of the ProbeSocket message-  | ProbeSocketAck-  deriving (Enum, Bounded, Show)---- | Response sent by /B/ to /A/ when /A/ tries to connect-data ConnectionRequestResponse =-    -- | /B/ accepts the connection-    ConnectionRequestAccepted-    -- | /A/ requested an invalid endpoint-  | ConnectionRequestInvalid-    -- | /A/s request crossed with a request from /B/ (see protocols)-  | ConnectionRequestCrossed-  deriving (Enum, Bounded, Show)- -- | Parameters for setting up the TCP transport data TCPParameters = TCPParameters {     -- | Backlog for 'listen'.@@ -479,26 +478,51 @@     -- | Should we set TCP_NODELAY on connection sockets?     -- Defaults to True.   , tcpNoDelay :: Bool-    -- | Value of TCP_USER_TIMEOUT in milliseconds-  , tcpKeepAlive :: Bool     -- | Should we set TCP_KEEPALIVE on connection sockets?+  , tcpKeepAlive :: Bool+    -- | Value of TCP_USER_TIMEOUT in milliseconds   , tcpUserTimeout :: Maybe Int     -- | A connect timeout for all 'connect' calls of the transport     -- in microseconds     --     -- This can be overriden for each connect call with     -- 'ConnectHints'.'connectTimeout'.+    --+    -- Connection requests to this transport will also timeout if they don't+    -- send the required data before this many microseconds.   , transportConnectTimeout :: Maybe Int+    -- | Create a QDisc for an EndPoint.+  , tcpNewQDisc :: forall t . IO (QDisc t)+    -- | Maximum length (in bytes) for a peer's address.+    -- If a peer attempts to send an address of length exceeding the limit,+    -- the connection will be refused (socket will close).+  , tcpMaxAddressLength :: Word32+    -- | Maximum length (in bytes) to receive from a peer.+    -- If a peer attempts to send data on a lightweight connection exceeding+    -- the limit, the heavyweight connection which carries that lightweight+    -- connection will go down. The peer and the local node will get an+    -- EventConnectionLost.+  , tcpMaxReceiveLength :: Word32+    -- | If True, new connections will be accepted only if the socket's host+    -- matches the host that the peer claims in its EndPointAddress.+    -- This is useful when operating on untrusted networks, because the peer+    -- could otherwise deny service to some victim by claiming the victim's+    -- address.+  , tcpCheckPeerHost :: Bool   }  -- | Internal functionality we expose for unit testing data TransportInternals = TransportInternals   { -- | The ID of the thread that listens for new incoming connections-    transportThread :: ThreadId+    transportThread     :: ThreadId+    -- | A variant of newEndPoint in which the QDisc determined by the+    -- transport's TCPParameters can be optionally overridden.+  , newEndPointInternal :: (forall t . Maybe (QDisc t))+                        -> IO (Either (TransportError NewEndPointErrorCode) EndPoint)     -- | Find the socket between a local and a remote endpoint-  , socketBetween :: EndPointAddress-                  -> EndPointAddress-                  -> IO N.Socket+  , socketBetween       :: EndPointAddress+                        -> EndPointAddress+                        -> IO N.Socket   }  --------------------------------------------------------------------------------@@ -506,20 +530,25 @@ --------------------------------------------------------------------------------  -- | Create a TCP transport-createTransport :: N.HostName-                -> N.ServiceName+createTransport :: N.HostName    -- ^ Bind host name.+                -> N.ServiceName -- ^ Bind port.+                -> (N.ServiceName -> (N.HostName, N.ServiceName))+                   -- ^ External address host name and port, computed from the+                   --   actual bind port.                 -> TCPParameters                 -> IO (Either IOException Transport)-createTransport host port params =-  either Left (Right . fst) <$> createTransportExposeInternals host port params+createTransport bindHost bindPort mkExternal params =+  either Left (Right . fst) <$>+    createTransportExposeInternals bindHost bindPort mkExternal params  -- | You should probably not use this function (used for unit testing only) createTransportExposeInternals   :: N.HostName   -> N.ServiceName+  -> (N.ServiceName -> (N.HostName, N.ServiceName))   -> TCPParameters   -> IO (Either IOException (Transport, TransportInternals))-createTransportExposeInternals host port params = do+createTransportExposeInternals bindHost bindPort mkExternal params = do     state <- newMVar . TransportValid $ ValidTransportState       { _localEndPoints = Map.empty       , _nextEndPointId = 0@@ -535,16 +564,20 @@        -- completes (see description of 'forkServer'), yet we need the port to        -- construct a transport. So we tie a recursive knot.        (port', result) <- do-         let transport = TCPTransport { transportState  = state-                                      , transportHost   = host-                                      , transportPort   = port'-                                      , transportParams = params+         let (externalHost, externalPort) = mkExternal port'+         let transport = TCPTransport { transportState    = state+                                      , transportHost     = externalHost+                                      , transportPort     = externalPort+                                      , transportBindHost = bindHost+                                      , transportBindPort = port'+                                      , transportParams   = params                                       }          bracketOnError (forkServer-                             host-                             port+                             bindHost+                             bindPort                              (tcpBacklog params)                              (tcpReuseServerAddr params)+                             (errorHandler transport)                              (terminationHandler transport)                              (handleConnectionRequest transport))                       (\(_port', tid) -> killThread tid)@@ -554,20 +587,29 @@     mkTransport :: TCPTransport                 -> ThreadId                 -> IO (Transport, TransportInternals)-    mkTransport transport tid = return-      ( Transport-          { newEndPoint    = apiNewEndPoint transport-          , closeTransport = let evs = [ EndPointClosed-                                       , throw $ userError "Transport closed"-                                       ] in-                             apiCloseTransport transport (Just tid) evs-          }-      , TransportInternals-          { transportThread = tid-          , socketBetween   = internalSocketBetween transport-          }-      )+    mkTransport transport tid = do+      return+        ( Transport+            { newEndPoint = do+                qdisc <- tcpNewQDisc params+                apiNewEndPoint transport qdisc+            , closeTransport = let evs = [ EndPointClosed ]+                               in apiCloseTransport transport (Just tid) evs+            }+        , TransportInternals+            { transportThread     = tid+            , socketBetween       = internalSocketBetween transport+            , newEndPointInternal = \mqdisc -> case mqdisc of+                Just qdisc -> apiNewEndPoint transport qdisc+                Nothing -> do+                  qdisc <- tcpNewQDisc params+                  apiNewEndPoint transport qdisc+            }+        ) +    errorHandler :: TCPTransport -> SomeException -> IO ()+    errorHandler _ = throwIO+     terminationHandler :: TCPTransport -> SomeException -> IO ()     terminationHandler transport ex = do       let evs = [ ErrorEvent (TransportError EventTransportFailed (show ex))@@ -584,7 +626,11 @@   , tcpNoDelay         = False   , tcpKeepAlive       = False   , tcpUserTimeout     = Nothing+  , tcpNewQDisc        = simpleUnboundedQDisc   , transportConnectTimeout = Nothing+  , tcpMaxAddressLength = maxBound+  , tcpMaxReceiveLength = maxBound+  , tcpCheckPeerHost   = False   }  --------------------------------------------------------------------------------@@ -606,18 +652,17 @@  -- | Create a new endpoint apiNewEndPoint :: TCPTransport+               -> QDisc Event                -> IO (Either (TransportError NewEndPointErrorCode) EndPoint)-apiNewEndPoint transport =+apiNewEndPoint transport qdisc =   try . asyncWhenCancelled closeEndPoint $ do-    ourEndPoint <- createLocalEndPoint transport+    ourEndPoint <- createLocalEndPoint transport qdisc     return EndPoint-      { receive       = readChan (localChannel ourEndPoint)+      { receive       = qdiscDequeue (localQueue ourEndPoint)       , address       = localAddress ourEndPoint       , connect       = apiConnect (transportParams transport) ourEndPoint-      , closeEndPoint = let evs = [ EndPointClosed-                                  , throw $ userError "Endpoint closed"-                                  ] in-                        apiCloseEndPoint transport evs ourEndPoint+      , closeEndPoint = let evs = [ EndPointClosed ]+                        in  apiCloseEndPoint transport evs ourEndPoint       , newMulticastGroup     = return . Left $ newMulticastGroupError       , resolveMulticastGroup = return . Left . const resolveMulticastGroupError       }@@ -627,6 +672,75 @@     resolveMulticastGroupError =       TransportError ResolveMulticastGroupUnsupported "Multicast not supported" +-- | Abstraction of a queue for an 'EndPoint'.+--+--   A value of type @QDisc t@ is a queue of events of an abstract type @t@.+--+--   This specifies which 'Event's will come from+--   'receive :: EndPoint -> IO Event' and when. It is highly general so that+--   the simple yet potentially very fast implementation backed by a single+--   unbounded channel can be used, without excluding more nuanced policies+--   like class-based queueing with bounded buffers for each peer, which may be+--   faster in certain conditions but probably has lower maximal throughput.+--+--   A 'QDisc' must satisfy some properties in order for the semantics of+--   network-transport to hold true. In general, an event fed with+--   'qdiscEnqueue' must not be dropped. i.e. provided that no other event in+--   the QDisc has higher priority, the event should eventually be returned by+--   'qdiscDequeue'. An exception to this are 'Receive' events of unreliable+--   connections.+--+--   Every call to 'receive' is just 'qdiscDequeue' on that 'EndPoint's+--   'QDisc'. Whenever an event arises from a socket, `qdiscEnqueue` is called+--   with the relevant metadata in the same thread that reads from the socket.+--   You can be clever about when to block here, so as to control network+--   ingress. This applies also to loopback connections (an 'EndPoint' connects+--   to itself), in which case blocking on the enqueue would only block some+--   thread in your program rather than some chatty network peer. The 'Event'+--   which is to be enqueued is given to 'qdiscEnqueue' so that the 'QDisc'+--   can know about open connections, their identifiers and peer addresses, etc.+data QDisc t = QDisc {+    -- | Dequeue an event.+    qdiscDequeue :: IO t+    -- | @qdiscEnqueue ep ev t@ enqueues and event @t@, originated from the+    -- given remote endpoint @ep@ and with data @ev@.+    --+    -- @ep@ might be the local endpoint if it relates to a self-connection.+    --+    -- @ev@ might be in practice the value given as @t@. It is passed in+    -- the abstract form @t@ to enforce it is dequeued unmodified, but the+    -- 'QDisc' implementation can still observe the concrete form @ev@ to+    -- make prioritization decisions.+  , qdiscEnqueue :: EndPointAddress -> Event -> t -> IO ()+  }++-- | Post an 'Event' using a 'QDisc'.+qdiscEnqueue' :: QDisc Event -> EndPointAddress -> Event -> IO ()+qdiscEnqueue' qdisc addr event = qdiscEnqueue qdisc addr event event++-- | A very simple QDisc backed by an unbounded channel.+simpleUnboundedQDisc :: forall t . IO (QDisc t)+simpleUnboundedQDisc = do+  eventChan <- newChan+  return $ QDisc {+      qdiscDequeue = readChan eventChan+    , qdiscEnqueue = const (const (writeChan eventChan))+    }++-- | A very simple QDisc backed by a 1-place queue (MVar).+--   With this QDisc, all threads reading from sockets will try to put their+--   events into the same MVar. That MVar will be cleared by calls to+--   'receive'. Thus the rate at which data is read from the wire is directly+--   related to the rate at which data is pulled from the EndPoint by+--   'receive'.+simpleOnePlaceQDisc :: forall t . IO (QDisc t)+simpleOnePlaceQDisc = do+  mvar <- newEmptyMVar+  return $ QDisc {+      qdiscDequeue = takeMVar mvar+    , qdiscEnqueue = const (const (putMVar mvar))+    }+ -- | Connnect to an endpoint apiConnect :: TCPParameters    -- ^ Parameters            -> LocalEndPoint    -- ^ Local end point@@ -662,7 +776,10 @@             then do               writeIORef connAlive False               sched theirEndPoint $-                sendOn vst [encodeInt32 CloseConnection, encodeInt32 connId]+                sendOn vst [+                    encodeWord32 (encodeControlHeader CloseConnection)+                  , encodeWord32 connId+                  ]               return ( RemoteEndPointValid                      . (remoteOutgoing ^: (\x -> x - 1))                      $ vst@@ -692,7 +809,7 @@           alive <- readIORef connAlive           if alive             then sched theirEndPoint $-              sendOn vst (encodeInt32 connId : prependLength payload)+              sendOn vst (encodeWord32 connId : prependLength payload)             else throwIO $ TransportError SendClosed "Connection closed"         RemoteEndPointClosing _ _ -> do           alive <- readIORef connAlive@@ -730,7 +847,8 @@           return (LocalEndPointClosed, Nothing)     forM_ mOurState $ \vst -> do       forM_ (vst ^. localConnections) tryCloseRemoteSocket-      forM_ evs $ writeChan (localChannel ourEndPoint)+      let qdisc = localQueue ourEndPoint+      forM_ evs (qdiscEnqueue' qdisc (localAddress ourEndPoint))   where     -- Close the remote socket and return the set of all incoming connections     tryCloseRemoteSocket :: RemoteEndPoint -> IO ()@@ -746,19 +864,32 @@             putMVar resolved ()             return closed           RemoteEndPointValid vst -> do+            -- Schedule an action to send a CloseEndPoint message and then+            -- wait for the socket to actually close (meaning that this+            -- end point is no longer receiving from it).+            -- Since we replace the state in this MVar with 'closed', it's+            -- guaranteed that no other actions will be scheduled after this+            -- one.             sched theirEndPoint $ do-              void $ tryIO $ sendOn vst [ encodeInt32 CloseSocket-                                        , encodeInt32 (vst ^. remoteMaxIncoming)-                                        ]+              void $ tryIO $ sendOn vst+                [ encodeWord32 (encodeControlHeader CloseEndPoint) ]               -- Release probing resources if probing.               forM_ (remoteProbing vst) id-              tryCloseSocket (remoteSocket vst)+              tryShutdownSocketBoth (remoteSocket vst)+              remoteSocketClosed vst             return closed           RemoteEndPointClosing resolved vst -> do             -- Release probing resources if probing.             forM_ (remoteProbing vst) id             putMVar resolved ()-            sched theirEndPoint $ tryCloseSocket (remoteSocket vst)+            -- Schedule an action to wait for the socket to actually close (this+            -- end point is no longer receiving from it).+            -- Since we replace the state in this MVar with 'closed', it's+            -- guaranteed that no other actions will be scheduled after this+            -- one.+            sched theirEndPoint $ do+              tryShutdownSocketBoth (remoteSocket vst)+              remoteSocketClosed vst             return closed           RemoteEndPointClosed ->             return st@@ -781,128 +912,197 @@ -- the transport down. We must be careful to close the socket when a (possibly -- asynchronous, ThreadKilled) exception occurs. (If an exception escapes from -- handleConnectionRequest the transport will be shut down.)-handleConnectionRequest :: TCPTransport -> N.Socket -> IO ()-handleConnectionRequest transport sock = handle handleException $ do+handleConnectionRequest :: TCPTransport -> IO () -> (N.Socket, N.SockAddr) -> IO ()+handleConnectionRequest transport socketClosed (sock, sockAddr) = handle handleException $ do     when (tcpNoDelay $ transportParams transport) $       N.setSocketOption sock N.NoDelay 1     when (tcpKeepAlive $ transportParams transport) $       N.setSocketOption sock N.KeepAlive 1     forM_ (tcpUserTimeout $ transportParams transport) $       N.setSocketOption sock N.UserTimeout-    ourEndPointId <- recvInt32 sock-    theirAddress  <- EndPointAddress . BS.concat <$> recvWithLength sock-    let ourAddress = encodeEndPointAddress (transportHost transport)-                                           (transportPort transport)-                                           ourEndPointId-    ourEndPoint <- withMVar (transportState transport) $ \st -> case st of-      TransportValid vst ->-        case vst ^. localEndPointAt ourAddress of-          Nothing -> do-            sendMany sock [encodeInt32 ConnectionRequestInvalid]-            throwIO $ userError "handleConnectionRequest: Invalid endpoint"-          Just ourEndPoint ->-            return ourEndPoint-      TransportClosed ->-        throwIO $ userError "Transport closed"-    void . forkIO $ go ourEndPoint theirAddress+    let handleVersioned = do+          -- Always receive the protocol version and a handshake (content of the+          -- handshake is version-dependent, but the length is always sent,+          -- regardless of the version).+          protocolVersion <- recvWord32 sock+          handshakeLength <- recvWord32 sock+          -- For now we support only version 0.0.0.0.+          case protocolVersion of+            0x00000000 -> handleConnectionRequestV0 (sock, sockAddr)+            _ -> do+              -- Inform the peer that we want version 0x00000000+              sendMany sock [+                  encodeWord32 (encodeConnectionRequestResponse ConnectionRequestUnsupportedVersion)+                , encodeWord32 0x00000000+                ]+              -- Clear the socket of the unsupported handshake data.+              _ <- recvExact sock handshakeLength+              handleVersioned+    -- The handshake must complete within the optional timeout duration.+    -- No socket 'recv's are to be run outside the timeout. The continuation+    -- returned may 'send', but not 'recv'.+    let connTimeout = transportConnectTimeout (transportParams transport)+    outcome <- maybe (fmap Just) System.Timeout.timeout connTimeout handleVersioned+    case outcome of+      Nothing -> throwIO (userError "handleConnectionRequest: timed out")+      Just act -> forM_ act id+   where-    go :: LocalEndPoint -> EndPointAddress -> IO ()-    go ourEndPoint theirAddress = do-      -- This runs in a thread that will never be killed-      mEndPoint <- handle ((>> return Nothing) . handleException) $ do++    handleException :: SomeException -> IO ()+    handleException ex = do+      rethrowIfAsync (fromException ex)++    rethrowIfAsync :: Maybe AsyncException -> IO ()+    rethrowIfAsync = mapM_ throwIO++    handleConnectionRequestV0 :: (N.Socket, N.SockAddr) -> IO (Maybe (IO ()))+    handleConnectionRequestV0 (sock, sockAddr) = do+      -- Get the OS-determined host and port.+      (actualHost, actualPort) <-+        decodeSockAddr sockAddr >>=+          maybe (throwIO (userError "handleConnectionRequest: invalid socket address")) return+      (ourEndPointId, theirAddress) <- do+        ourEndPointId <- recvWord32 sock+        let maxAddressLength = tcpMaxAddressLength $ transportParams transport+        theirAddress <- EndPointAddress . BS.concat <$>+          recvWithLength maxAddressLength sock+        return (ourEndPointId, theirAddress)+      let ourAddress = encodeEndPointAddress (transportHost transport)+                                             (transportPort transport)+                                             ourEndPointId+      (theirHost, _, _)+        <- maybe (throwIO (userError "handleConnectionRequest: peer gave malformed address"))+                 return+                 (decodeEndPointAddress theirAddress)+      let checkPeerHost = tcpCheckPeerHost (transportParams transport)+      if checkPeerHost && (theirHost /= actualHost)+      then do+        -- If the OS-determined host doesn't match the host that the peer gave us,+        -- then we have no choice but to reject the connection. It's because we+        -- use the EndPointAddress to key the remote end points (localConnections)+        -- and we don't want to allow a peer to deny service to other peers by+        -- claiming to have their host and port.+        sendMany sock $+            encodeWord32 (encodeConnectionRequestResponse ConnectionRequestHostMismatch)+          : prependLength [BSC.pack actualHost]+        return Nothing+      else do+        ourEndPoint <- withMVar (transportState transport) $ \st -> case st of+          TransportValid vst ->+            case vst ^. localEndPointAt ourAddress of+              Nothing -> do+                sendMany sock [encodeWord32 (encodeConnectionRequestResponse ConnectionRequestInvalid)]+                throwIO $ userError "handleConnectionRequest: Invalid endpoint"+              Just ourEndPoint ->+                return ourEndPoint+          TransportClosed ->+            throwIO $ userError "Transport closed"+        return (Just (go ourEndPoint theirAddress))++      where++      go :: LocalEndPoint -> EndPointAddress -> IO ()+      go ourEndPoint theirAddress = handle handleException $ do         resetIfBroken ourEndPoint theirAddress         (theirEndPoint, isNew) <-           findRemoteEndPoint ourEndPoint theirAddress RequestedByThem Nothing          if not isNew           then do-            void $ tryIO $ sendMany sock [encodeInt32 ConnectionRequestCrossed]+            void $ tryIO $ sendMany sock+              [encodeWord32 (encodeConnectionRequestResponse ConnectionRequestCrossed)]             probeIfValid theirEndPoint-            tryCloseSocket sock-            return Nothing           else do             sendLock <- newMVar ()             let vst = ValidRemoteEndPointState                         {  remoteSocket        = sock+                        ,  remoteSocketClosed  = socketClosed                         ,  remoteProbing       = Nothing                         ,  remoteSendLock      = sendLock                         , _remoteOutgoing      = 0                         , _remoteIncoming      = Set.empty-                        , _remoteMaxIncoming   = 0+                        , _remoteLastIncoming  = 0                         , _remoteNextConnOutId = firstNonReservedLightweightConnectionId                         }-            sendMany sock [encodeInt32 ConnectionRequestAccepted]+            sendMany sock [encodeWord32 (encodeConnectionRequestResponse ConnectionRequestAccepted)]+            -- resolveInit will update the shared state, and handleIncomingMessages+            -- will always ultimately clean up after it.+            -- Closing up the socket is also out of our hands. It will happen+            -- when handleIncomingMessages finishes.             resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst)-            return (Just theirEndPoint)-      -- If we left the scope of the exception handler with a return value of-      -- Nothing then the socket is already closed; otherwise, the socket has-      -- been recorded as part of the remote endpoint. Either way, we no longer-      -- have to worry about closing the socket on receiving an asynchronous-      -- exception from this point forward.-      forM_ mEndPoint $ handleIncomingMessages . (,) ourEndPoint--    handleException :: SomeException -> IO ()-    handleException ex = do-      tryCloseSocket sock-      rethrowIfAsync (fromException ex)--    rethrowIfAsync :: Maybe AsyncException -> IO ()-    rethrowIfAsync = mapM_ throwIO+              `finally`+              handleIncomingMessages (transportParams transport) (ourEndPoint, theirEndPoint) -    probeIfValid :: RemoteEndPoint -> IO ()-    probeIfValid theirEndPoint = modifyMVar_ (remoteState theirEndPoint) $-      \st -> case st of-        RemoteEndPointValid-          vst@(ValidRemoteEndPointState { remoteProbing = Nothing }) -> do-            tid <- forkIO $ do-              -- send probe-              let params = transportParams transport-              void $ tryIO $ System.Timeout.timeout-                  (maybe (-1) id $ transportConnectTimeout params) $ do-                sendMany (remoteSocket vst) [encodeInt32 ProbeSocket]-                threadDelay maxBound-              -- Discard the connection if this thread is not killed (i.e. the-              -- probe ack does not arrive on time).-              ---              -- The thread handling incoming messages will detect the socket is-              -- closed and will report the failure upwards.-              tryCloseSocket (remoteSocket vst)-              -- Waiting the probe ack and closing the socket is only needed in-              -- platforms where TCP_USER_TIMEOUT is not available or when the-              -- user does not set it. Otherwise the ack would be handled at the-              -- TCP level and the the thread handling incoming messages would-              -- get the error.+      probeIfValid :: RemoteEndPoint -> IO ()+      probeIfValid theirEndPoint = modifyMVar_ (remoteState theirEndPoint) $+        \st -> case st of+          RemoteEndPointValid+            vst@(ValidRemoteEndPointState { remoteProbing = Nothing }) -> do+              tid <- forkIO $ do+                -- send probe+                let params = transportParams transport+                void $ tryIO $ System.Timeout.timeout+                    (maybe (-1) id $ transportConnectTimeout params) $ do+                  sendMany (remoteSocket vst)+                    [encodeWord32 (encodeControlHeader ProbeSocket)]+                  threadDelay maxBound+                -- Discard the connection if this thread is not killed (i.e. the+                -- probe ack does not arrive on time).+                --+                -- The thread handling incoming messages will detect the socket is+                -- closed and will report the failure upwards.+                tryCloseSocket (remoteSocket vst)+                -- Waiting the probe ack and closing the socket is only needed in+                -- platforms where TCP_USER_TIMEOUT is not available or when the+                -- user does not set it. Otherwise the ack would be handled at the+                -- TCP level and the the thread handling incoming messages would+                -- get the error. -            return $ RemoteEndPointValid-              vst { remoteProbing = Just (killThread tid) }-        _                       -> return st+              return $ RemoteEndPointValid+                vst { remoteProbing = Just (killThread tid) }+          _                       -> return st  -- | Handle requests from a remote endpoint. -- -- Returns only if the remote party closes the socket or if an error occurs. -- This runs in a thread that will never be killed.-handleIncomingMessages :: EndPointPair -> IO ()-handleIncomingMessages (ourEndPoint, theirEndPoint) = do-    mSock <- withMVar theirState $ \st ->-      case st of-        RemoteEndPointInvalid _ ->-          relyViolation (ourEndPoint, theirEndPoint)-            "handleIncomingMessages (invalid)"-        RemoteEndPointInit _ _ _ ->-          relyViolation (ourEndPoint, theirEndPoint)-            "handleIncomingMessages (init)"-        RemoteEndPointValid ep ->-          return . Just $ remoteSocket ep-        RemoteEndPointClosing _ ep ->-          return . Just $ remoteSocket ep-        RemoteEndPointClosed ->-          return Nothing-        RemoteEndPointFailed _ ->-          return Nothing+handleIncomingMessages :: TCPParameters -> EndPointPair -> IO ()+handleIncomingMessages params (ourEndPoint, theirEndPoint) =+    bracket acquire release act -    forM_ mSock $ \sock ->-      tryIO (go sock) >>= either (prematureExit sock) return   where++    -- Use shared remote endpoint state to get a socket, or an appropriate+    -- exception in case it's neither valid nor closing.+    acquire :: IO (Either IOError N.Socket)+    acquire = withMVar theirState $ \st -> case st of+      RemoteEndPointInvalid _ ->+        relyViolation (ourEndPoint, theirEndPoint)+          "handleIncomingMessages (invalid)"+      RemoteEndPointInit _ _ _ ->+        relyViolation (ourEndPoint, theirEndPoint)+          "handleIncomingMessages (init)"+      RemoteEndPointValid ep ->+        return . Right $ remoteSocket ep+      RemoteEndPointClosing _ ep ->+        return . Right $ remoteSocket ep+      RemoteEndPointClosed ->+        return . Left $ userError "handleIncomingMessages (already closed)"+      RemoteEndPointFailed _ ->+        return . Left $ userError "handleIncomingMessages (failed)"++    -- 'Right' is the normal case in which there still is a live socket to+    -- the remote endpoint, and so 'act' was run and installed its own+    -- exception handler.+    release :: Either IOError N.Socket -> IO ()+    release (Left err) = prematureExit err+    release (Right _) = return ()++    act :: Either IOError N.Socket -> IO ()+    act (Left _) = return ()+    act (Right sock) = go sock `catch` prematureExit+     -- Dispatch     --     -- If a recv throws an exception this will be caught top-level and@@ -913,24 +1113,46 @@     -- exception thrown by 'recv'.     go :: N.Socket -> IO ()     go sock = do-      lcid <- recvInt32 sock :: IO LightweightConnectionId+      lcid <- recvWord32 sock :: IO LightweightConnectionId       if lcid >= firstNonReservedLightweightConnectionId         then do           readMessage sock lcid           go sock         else-          case tryToEnum (fromIntegral lcid) of+          case decodeControlHeader lcid of             Just CreatedNewConnection -> do-              recvInt32 sock >>= createdNewConnection+              recvWord32 sock >>= createdNewConnection               go sock             Just CloseConnection -> do-              recvInt32 sock >>= closeConnection+              recvWord32 sock >>= closeConnection               go sock             Just CloseSocket -> do-              didClose <- recvInt32 sock >>= closeSocket sock+              didClose <- recvWord32 sock >>= closeSocket sock               unless didClose $ go sock+            Just CloseEndPoint -> do+              let closeRemoteEndPoint vst = do+                    forM_ (remoteProbing vst) id+                    -- close incoming connections+                    forM_ (Set.elems $ vst ^. remoteIncoming) $+                      qdiscEnqueue' ourQueue theirAddr . ConnectionClosed . connId+                    -- report the endpoint as gone if we have any outgoing+                    -- connections+                    when (vst ^. remoteOutgoing > 0) $ do+                      let code = EventConnectionLost (remoteAddress theirEndPoint)+                      qdiscEnqueue' ourQueue theirAddr . ErrorEvent $+                        TransportError code "The remote endpoint was closed."+              removeRemoteEndPoint (ourEndPoint, theirEndPoint)+              modifyMVar_ theirState $ \s -> case s of+                RemoteEndPointValid vst     -> do+                  closeRemoteEndPoint vst+                  return RemoteEndPointClosed+                RemoteEndPointClosing resolved vst -> do+                  closeRemoteEndPoint vst+                  putMVar resolved ()+                  return RemoteEndPointClosed+                _                           -> return s             Just ProbeSocket -> do-              forkIO $ sendMany sock [encodeInt32 ProbeSocketAck]+              forkIO $ sendMany sock [encodeWord32 (encodeControlHeader ProbeSocketAck)]               go sock             Just ProbeSocketAck -> do               stopProbing@@ -951,7 +1173,7 @@               "handleIncomingMessages:createNewConnection (init)"           RemoteEndPointValid vst ->             return ( (remoteIncoming ^: Set.insert lcid)-                   $ (remoteMaxIncoming ^= lcid)+                   $ (remoteLastIncoming ^= lcid)                    vst                    )           RemoteEndPointClosing resolved vst -> do@@ -963,7 +1185,7 @@             -- RemoteEndPointValid             putMVar resolved ()             return ( (remoteIncoming ^= Set.singleton lcid)-                   . (remoteMaxIncoming ^= lcid)+                   . (remoteLastIncoming ^= lcid)                    $ vst                    )           RemoteEndPointFailed err ->@@ -972,7 +1194,7 @@             relyViolation (ourEndPoint, theirEndPoint)               "createNewConnection (closed)"         return (RemoteEndPointValid vst)-      writeChan ourChannel (ConnectionOpened (connId lcid) ReliableOrdered theirAddr)+      qdiscEnqueue' ourQueue theirAddr (ConnectionOpened (connId lcid) ReliableOrdered theirAddr)      -- Close a connection     -- It is important that we verify that the connection is in fact open,@@ -1000,7 +1222,7 @@           throwIO err         RemoteEndPointClosed ->           relyViolation (ourEndPoint, theirEndPoint) "closeConnection (closed)"-      writeChan ourChannel (ConnectionClosed (connId lcid))+      qdiscEnqueue' ourQueue theirAddr (ConnectionClosed (connId lcid))      -- Close the socket (if we don't have any outgoing connections)     closeSocket :: N.Socket -> LightweightConnectionId -> IO Bool@@ -1018,16 +1240,28 @@             -- remote endpoint to indicate that all its connections to us are             -- now properly closed             forM_ (Set.elems $ vst ^. remoteIncoming) $-              writeChan ourChannel . ConnectionClosed . connId+              qdiscEnqueue' ourQueue theirAddr . ConnectionClosed . connId             let vst' = remoteIncoming ^= Set.empty $ vst-            -- If we still have outgoing connections then we ignore the-            -- CloseSocket request (we sent a ConnectionCreated message to the-            -- remote endpoint, but it did not receive it before sending the-            -- CloseSocket request). Similarly, if lastReceivedId < lastSentId-            -- then we sent a ConnectionCreated *AND* a ConnectionClosed-            -- message to the remote endpoint, *both of which* it did not yet-            -- receive before sending the CloseSocket request.-            if vst' ^. remoteOutgoing > 0 || lastReceivedId < lastSentId vst+            -- The peer sends the connection id of the last connection which+            -- they accepted from us.+            --+            -- If it's not the same as the id of the last connection that we+            -- have made to them (assuming we haven't cycled through all+            -- identifiers so fast) then they hadn't seen the request before+            -- they tried to close the socket. In that case, we don't close the+            -- socket. They'll see our in-flight connection request and then+            -- abandon their attempt to close the socket.+            --+            -- It's possible that a local connection is coming up but has not+            -- yet sent CreatedNewConnection (see createConnectionTo), in+            -- which case remoteOutgoing is positive, but the sent and received+            -- ids do match. In this case we don't close the socket, because+            -- that connection will soon sent the message and bump the lastSentId.+            --+            -- Both disjuncts are needed: it's possible that remoteOutgoing is+            -- 0 and the ids do not match, in case we have created and closed+            -- a connection but the peer has not yet heard of it.+            if vst ^. remoteOutgoing > 0 || lastReceivedId /= lastSentId vst               then                 return (RemoteEndPointValid vst', Nothing)               else do@@ -1036,10 +1270,10 @@                 removeRemoteEndPoint (ourEndPoint, theirEndPoint)                 -- Attempt to reply (but don't insist)                 act <- schedule theirEndPoint $ do-                  void $ tryIO $ sendOn vst' [ encodeInt32 CloseSocket-                                             , encodeInt32 (vst ^. remoteMaxIncoming)-                                             ]-                  tryCloseSocket sock+                  void $ tryIO $ sendOn vst'+                    [ encodeWord32 (encodeControlHeader CloseSocket)+                    , encodeWord32 (vst ^. remoteLastIncoming)+                    ]                 return (RemoteEndPointClosed, Just act)           RemoteEndPointClosing resolved vst ->  do             -- Like above, we need to check if there is a ConnectionCreated@@ -1047,16 +1281,26 @@             -- received. However, since we are in 'closing' state, the only             -- way this may happen is when we sent a ConnectionCreated,             -- ConnectionClosed, and CloseSocket message, none of which have-            -- yet been received. We leave the endpoint in closing state in-            -- that case.-            if lastReceivedId < lastSentId vst+            -- yet been received. It's sufficient to check that the peer has+            -- not seen the ConnectionCreated message. In case they have seen+            -- it (so that lastReceivedId == lastSendId vst) then they must+            -- have seen the other messages or else they would not have sent+            -- CloseSocket.+            -- We leave the endpoint in closing state in that case.+            if lastReceivedId /= lastSentId vst               then do                 return (RemoteEndPointClosing resolved vst, Nothing)               else do                 -- Release probing resources if probing.+                when (vst ^. remoteOutgoing > 0) $ do+                  let code = EventConnectionLost (remoteAddress theirEndPoint)+                  let msg  = "socket closed prematurely by peer"+                  qdiscEnqueue' ourQueue theirAddr . ErrorEvent $ TransportError code msg                 forM_ (remoteProbing vst) id                 removeRemoteEndPoint (ourEndPoint, theirEndPoint)-                act <- schedule theirEndPoint $ tryCloseSocket sock+                -- Nothing to do, but we want to indicate that the socket+                -- really did close.+                act <- schedule theirEndPoint $ return ()                 putMVar resolved ()                 return (RemoteEndPointClosed, Just act)           RemoteEndPointFailed err ->@@ -1075,7 +1319,8 @@     -- overhead     readMessage :: N.Socket -> LightweightConnectionId -> IO ()     readMessage sock lcid =-      recvWithLength sock >>= writeChan ourChannel . Received (connId lcid)+      recvWithLength recvLimit sock >>=+        qdiscEnqueue' ourQueue theirAddr . Received (connId lcid)      -- Stop probing a connection as a result of receiving a probe ack.     stopProbing :: IO ()@@ -1087,14 +1332,15 @@       _ -> return st      -- Arguments-    ourChannel  = localChannel ourEndPoint+    ourQueue    = localQueue ourEndPoint+    ourState    = localState ourEndPoint     theirState  = remoteState theirEndPoint     theirAddr   = remoteAddress theirEndPoint+    recvLimit   = tcpMaxReceiveLength params      -- Deal with a premature exit-    prematureExit :: N.Socket -> IOException -> IO ()-    prematureExit sock err = do-      tryCloseSocket sock+    prematureExit :: IOException -> IO ()+    prematureExit err = do       modifyMVar_ theirState $ \st ->         case st of           RemoteEndPointInvalid _ ->@@ -1107,7 +1353,7 @@             -- Release probing resources if probing.             forM_ (remoteProbing vst) id             let code = EventConnectionLost (remoteAddress theirEndPoint)-            writeChan ourChannel . ErrorEvent $ TransportError code (show err)+            qdiscEnqueue' ourQueue theirAddr . ErrorEvent $ TransportError code (show err)             return (RemoteEndPointFailed err)           RemoteEndPointClosing resolved vst -> do             -- Release probing resources if probing.@@ -1117,7 +1363,19 @@           RemoteEndPointClosed ->             relyViolation (ourEndPoint, theirEndPoint)               "handleIncomingMessages:prematureExit"-          RemoteEndPointFailed err' ->+          RemoteEndPointFailed err' -> do+            -- Here we post a connection-lost event, but only if the+            -- local endpoint is not closed; if it's closed, the EndPointClosed+            -- event will be posted without connection-lost events, and this is+            -- part of the network-transport specification (there's a test+            -- case for it).+            modifyMVar_ ourState $ \st' -> case st' of+              LocalEndPointClosed -> return st'+              LocalEndPointValid _ -> do+                let code = EventConnectionLost (remoteAddress theirEndPoint)+                    err  = TransportError code (show err')+                qdiscEnqueue' ourQueue theirAddr (ErrorEvent err)+                return st'             return (RemoteEndPointFailed err')      -- Construct a connection ID@@ -1190,7 +1448,10 @@               RemoteEndPointValid vst -> do                 let connId = vst ^. remoteNextConnOutId                 act <- schedule theirEndPoint $ do-                  sendOn vst [encodeInt32 CreatedNewConnection, encodeInt32 connId]+                  sendOn vst [+                      encodeWord32 (encodeControlHeader CreatedNewConnection)+                    , encodeWord32 connId+                    ]                   return connId                 return ( RemoteEndPointValid                        $ remoteNextConnOutId ^= connId + 1@@ -1229,25 +1490,39 @@                                (tcpUserTimeout params)                                connTimeout     didAccept <- case result of-      Right (sock, ConnectionRequestAccepted) -> do+      -- Since a socket was created, we are now responsible for closing it.+      --+      -- In case the connection was accepted, we have some work to do.+      -- We'll remember how to wait for the socket to close+      -- (readMVar socketClosedVar), and we'll take care of closing it up+      -- once handleIncomingMessages has finished.+      Right (socketClosedVar, sock, ConnectionRequestAccepted) -> do         sendLock <- newMVar ()         let vst = ValidRemoteEndPointState                     {  remoteSocket        = sock+                    ,  remoteSocketClosed  = readMVar socketClosedVar                     ,  remoteProbing       = Nothing                     ,  remoteSendLock      = sendLock                     , _remoteOutgoing      = 0                     , _remoteIncoming      = Set.empty-                    , _remoteMaxIncoming   = 0+                    , _remoteLastIncoming  = 0                     , _remoteNextConnOutId = firstNonReservedLightweightConnectionId                     }         resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst)-        return True-      Right (sock, ConnectionRequestInvalid) -> do+        return (Just (socketClosedVar, sock))+      Right (socketClosedVar, sock, ConnectionRequestUnsupportedVersion) -> do+        -- If the peer doesn't support V0 then there's nothing we can do, for+        -- it's the only version we support.+        let err = connectFailed "setupRemoteEndPoint: unsupported version"+        resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointInvalid err)+        tryCloseSocket sock `finally` putMVar socketClosedVar ()+        return Nothing+      Right (socketClosedVar, sock, ConnectionRequestInvalid) -> do         let err = invalidAddress "setupRemoteEndPoint: Invalid endpoint"         resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointInvalid err)-        tryCloseSocket sock-        return False-      Right (sock, ConnectionRequestCrossed) -> do+        tryCloseSocket sock `finally` putMVar socketClosedVar ()+        return Nothing+      Right (socketClosedVar, sock, ConnectionRequestCrossed) -> do         withMVar (remoteState theirEndPoint) $ \st -> case st of           RemoteEndPointInit _ crossed _ ->             putMVar crossed ()@@ -1255,19 +1530,33 @@             throwIO ex           _ ->             relyViolation (ourEndPoint, theirEndPoint) "setupRemoteEndPoint: Crossed"-        tryCloseSocket sock-        return False+        tryCloseSocket sock `finally` putMVar socketClosedVar ()+        return Nothing+      Right (socketClosedVar, sock, ConnectionRequestHostMismatch) -> do+        let handler :: SomeException -> IO (TransportError ConnectErrorCode)+            handler err = return (TransportError ConnectFailed (show err))+        err <- handle handler $ do+          actualHost <- recvWithLength (tcpMaxReceiveLength params) sock+          return (TransportError ConnectFailed ("setupRemoteEndPoint: Host mismatch " ++ BSC.unpack (BS.concat actualHost)))+        resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointInvalid err)+        tryCloseSocket sock `finally` putMVar socketClosedVar ()+        return Nothing       Left err -> do         resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointInvalid err)-        return False+        return Nothing -    when didAccept $ void $ forkIO $-      handleIncomingMessages (ourEndPoint, theirEndPoint)-    return $ either (const Nothing) (Just . snd) result+    -- We handle incoming messages in a separate thread, and are careful to+    -- always close the socket once that thread is finished.+    forM_ didAccept $ \(socketClosed, sock) -> void $ forkIO $+      handleIncomingMessages params (ourEndPoint, theirEndPoint)+      `finally`+      (tryCloseSocket sock `finally` putMVar socketClosed ())+    return $ either (const Nothing) (Just . (\(_,_,x) -> x)) result   where     ourAddress      = localAddress ourEndPoint     theirAddress    = remoteAddress theirEndPoint     invalidAddress  = TransportError ConnectNotFound+    connectFailed   = TransportError ConnectFailed  -- | Send a CloseSocket request if the remote endpoint is unused closeIfUnused :: EndPointPair -> IO ()@@ -1278,8 +1567,8 @@         then do           resolved <- newEmptyMVar           act <- schedule theirEndPoint $-            sendOn vst [ encodeInt32 CloseSocket-                       , encodeInt32 (vst ^. remoteMaxIncoming)+            sendOn vst [ encodeWord32 (encodeControlHeader CloseSocket)+                       , encodeWord32 (vst ^. remoteLastIncoming)                        ]           return (RemoteEndPointClosing resolved vst, Just act)         else@@ -1324,7 +1613,7 @@     connAlive <- newIORef True  -- Protected by the local endpoint lock     lconnId   <- mapIOException connectFailed $ getLocalNextConnOutId ourEndPoint     let connId = createConnectionId heavyweightSelfConnectionId lconnId-    writeChan ourChan $+    qdiscEnqueue' ourQueue ourAddress $       ConnectionOpened connId ReliableOrdered (localAddress ourEndPoint)     return Connection       { send  = selfSend connAlive connId@@ -1341,7 +1630,7 @@           alive <- readIORef connAlive           if alive             then seq (foldr seq () msg)-                   writeChan ourChan (Received connId msg)+                   qdiscEnqueue' ourQueue ourAddress (Received connId msg)             else throwIO $ TransportError SendClosed "Connection closed"         LocalEndPointClosed ->           throwIO $ TransportError SendFailed "Endpoint closed"@@ -1352,14 +1641,15 @@         LocalEndPointValid _ -> do           alive <- readIORef connAlive           when alive $ do-            writeChan ourChan (ConnectionClosed connId)+            qdiscEnqueue' ourQueue ourAddress (ConnectionClosed connId)             writeIORef connAlive False         LocalEndPointClosed ->           return () -    ourChan  = localChannel ourEndPoint+    ourQueue = localQueue ourEndPoint     ourState = localState ourEndPoint     connectFailed = TransportError ConnectFailed . show+    ourAddress = localAddress ourEndPoint  -- | Resolve an endpoint currently in 'Init' state resolveInit :: EndPointPair -> RemoteState -> IO ()@@ -1397,9 +1687,10 @@ -- -- May throw a TransportError NewEndPointErrorCode exception if the transport -- is closed.-createLocalEndPoint :: TCPTransport -> IO LocalEndPoint-createLocalEndPoint transport = do-    chan  <- newChan+createLocalEndPoint :: TCPTransport+                    -> QDisc Event+                    -> IO LocalEndPoint+createLocalEndPoint transport qdisc = do     state <- newMVar . LocalEndPointValid $ ValidLocalEndPointState       { _localNextConnOutId = firstNonReservedLightweightConnectionId       , _localConnections   = Map.empty@@ -1411,9 +1702,9 @@         let addr = encodeEndPointAddress (transportHost transport)                                          (transportPort transport)                                          ix-        let localEndPoint = LocalEndPoint { localAddress  = addr-                                          , localChannel  = chan-                                          , localState    = state+        let localEndPoint = LocalEndPoint { localAddress = addr+                                          , localQueue   = qdisc+                                          , localState   = state                                           }         return ( TransportValid                . (localEndPointAt addr ^= Just localEndPoint)@@ -1612,10 +1903,11 @@     handleIOException ex vst = do       -- Release probing resources if probing.       forM_ (remoteProbing vst) id-      tryCloseSocket (remoteSocket vst)-      let code     = EventConnectionLost (remoteAddress theirEndPoint)-          err      = TransportError code (show ex)-      writeChan (localChannel ourEndPoint) $ ErrorEvent err+      -- Must shut down the socket here, so that the other end will realize+      -- we lost the connection+      tryShutdownSocketBoth (remoteSocket vst)+      -- Eventually, handleIncomingMessages will fail while trying to+      -- receive, and ultimately enqueue the 'EventConnectionLost'.       return (RemoteEndPointFailed ex)  -- | Use 'schedule' action 'runScheduled' action in a safe way, it's assumed that@@ -1634,6 +1926,11 @@ -- | Establish a connection to a remote endpoint -- -- Maybe throw a TransportError+--+-- If a socket is created and returned (Right is given) then the caller is+-- responsible for eventually closing the socket and filling the MVar (which+-- is empty). The MVar must be filled immediately after, and never before,+-- the socket is closed. socketToEndPoint :: EndPointAddress -- ^ Our address                  -> EndPointAddress -- ^ Their address                  -> Bool            -- ^ Use SO_REUSEADDR?@@ -1642,7 +1939,7 @@                  -> Maybe Int       -- ^ Maybe TCP_USER_TIMEOUT                  -> Maybe Int       -- ^ Timeout for connect                  -> IO (Either (TransportError ConnectErrorCode)-                               (N.Socket, ConnectionRequestResponse))+                               (MVar (), N.Socket, ConnectionRequestResponse)) socketToEndPoint (EndPointAddress ourAddress) theirAddress reuseAddr noDelay keepAlive                  mUserTimeout timeout =   try $ do@@ -1664,12 +1961,17 @@         mapIOException invalidAddress $           N.connect sock (N.addrAddress addr)         mapIOException failed $ do-          sendMany sock-                   (encodeInt32 theirEndPointId : prependLength [ourAddress])-          recvInt32 sock-      case tryToEnum response of+          sendMany sock $+              -- The version.+              encodeWord32 currentProtocolVersion+              -- The V0 handshake data with the length prepended.+            : prependLength (encodeWord32 theirEndPointId : prependLength [ourAddress])+          recvWord32 sock+      case decodeConnectionRequestResponse response of         Nothing -> throwIO (failed . userError $ "Unexpected response")-        Just r  -> return (sock, r)+        Just r  -> do+          socketClosedVar <- newEmptyMVar+          return (socketClosedVar, sock, r)   where     createSocket :: N.AddrInfo -> IO N.Socket     createSocket addr = mapIOException insufficientResources $@@ -1680,26 +1982,6 @@     failed                = TransportError ConnectFailed . show     timeoutError          = TransportError ConnectTimeout "Timed out" --- | Encode end point address-encodeEndPointAddress :: N.HostName-                      -> N.ServiceName-                      -> EndPointId-                      -> EndPointAddress-encodeEndPointAddress host port ix = EndPointAddress . BSC.pack $-  host ++ ":" ++ port ++ ":" ++ show ix---- | Decode end point address-decodeEndPointAddress :: EndPointAddress-                      -> Maybe (N.HostName, N.ServiceName, EndPointId)-decodeEndPointAddress (EndPointAddress bs) =-  case splitMaxFromEnd (== ':') 2 $ BSC.unpack bs of-    [host, port, endPointIdStr] ->-      case reads endPointIdStr of-        [(endPointId, "")] -> Just (host, port, endPointId)-        _                  -> Nothing-    _ ->-      Nothing- -- | Construct a ConnectionId createConnectionId :: HeavyweightConnectionId                    -> LightweightConnectionId@@ -1707,21 +1989,6 @@ createConnectionId hcid lcid =   (fromIntegral hcid `shiftL` 32) .|. fromIntegral lcid --- | @spltiMaxFromEnd p n xs@ splits list @xs@ at elements matching @p@,--- returning at most @p@ segments -- counting from the /end/------ > splitMaxFromEnd (== ':') 2 "ab:cd:ef:gh" == ["ab:cd", "ef", "gh"]-splitMaxFromEnd :: (a -> Bool) -> Int -> [a] -> [[a]]-splitMaxFromEnd p = \n -> go [[]] n . reverse-  where-    -- go :: [[a]] -> Int -> [a] -> [[a]]-    go accs         _ []     = accs-    go ([]  : accs) 0 xs     = reverse xs : accs-    go (acc : accs) n (x:xs) =-      if p x then go ([] : acc : accs) (n - 1) xs-             else go ((x : acc) : accs) n xs-    go _ _ _ = error "Bug in splitMaxFromEnd"- -------------------------------------------------------------------------------- -- Functions from TransportInternals                                          -- --------------------------------------------------------------------------------@@ -1803,8 +2070,8 @@ remoteIncoming :: Accessor ValidRemoteEndPointState (Set LightweightConnectionId) remoteIncoming = accessor _remoteIncoming (\cs conn -> conn { _remoteIncoming = cs }) -remoteMaxIncoming :: Accessor ValidRemoteEndPointState LightweightConnectionId-remoteMaxIncoming = accessor _remoteMaxIncoming (\lcid st -> st { _remoteMaxIncoming = lcid })+remoteLastIncoming :: Accessor ValidRemoteEndPointState LightweightConnectionId+remoteLastIncoming = accessor _remoteLastIncoming (\lcid st -> st { _remoteLastIncoming = lcid })  remoteNextConnOutId :: Accessor ValidRemoteEndPointState LightweightConnectionId remoteNextConnOutId = accessor _remoteNextConnOutId (\cix st -> st { _remoteNextConnOutId = cix })

@@ -1,18 +1,40 @@ -- | Utility functions for TCP sockets module Network.Transport.TCP.Internal-  ( forkServer+  ( ControlHeader(..)+  , encodeControlHeader+  , decodeControlHeader+  , ConnectionRequestResponse(..)+  , encodeConnectionRequestResponse+  , decodeConnectionRequestResponse+  , forkServer   , recvWithLength   , recvExact-  , recvInt32+  , recvWord32+  , encodeWord32   , tryCloseSocket+  , tryShutdownSocketBoth+  , decodeSockAddr+  , EndPointId+  , encodeEndPointAddress+  , decodeEndPointAddress+  , ProtocolVersion+  , currentProtocolVersion   ) where  #if ! MIN_VERSION_base(4,6,0) import Prelude hiding (catch) #endif -import Network.Transport.Internal (decodeInt32, void, tryIO, forkIOWithUnmask)+import Network.Transport.Internal+  ( decodeWord32+  , encodeWord32+  , void+  , tryIO+  , forkIOWithUnmask+  ) +import Network.Transport ( EndPointAddress(..) )+ #ifdef USE_MOCK_NETWORK import qualified Network.Transport.TCP.Mock.Socket as N #else@@ -35,6 +57,10 @@   , accept   , sClose   , socketPort+  , shutdown+  , ShutdownCmd(ShutdownBoth)+  , SockAddr(..)+  , inet_ntoa   )  #ifdef USE_MOCK_NETWORK@@ -43,44 +69,150 @@ import qualified Network.Socket.ByteString as NBS (recv) #endif -import Control.Concurrent (ThreadId)+import Data.Word (Word32)+ import Control.Monad (forever, when) import Control.Exception (SomeException, catch, bracketOnError, throwIO, mask_)+import Control.Concurrent (ThreadId, forkIO)+import Control.Concurrent.MVar+  ( MVar+  , newEmptyMVar+  , putMVar+  , readMVar+  )+import Control.Monad (forever, when)+import Control.Exception+  ( SomeException+  , catch+  , bracketOnError+  , throwIO+  , mask_+  , mask+  , finally+  , onException+  )+ import Control.Applicative ((<$>), (<*>))+import Data.Word (Word32) import Data.ByteString (ByteString) import qualified Data.ByteString as BS (length, concat, null)-import Data.Int (Int32) import Data.ByteString.Lazy.Internal (smallChunkSize)+import qualified Data.ByteString.Char8 as BSC (unpack, pack) +-- | Local identifier for an endpoint within this transport+type EndPointId = Word32++-- | Identifies the version of the network-transport-tcp protocol.+-- It's the first piece of data sent when a new heavyweight connection is+-- established.+type ProtocolVersion = Word32++currentProtocolVersion :: ProtocolVersion+currentProtocolVersion = 0x00000000++-- | Control headers+data ControlHeader =+    -- | Tell the remote endpoint that we created a new connection+    CreatedNewConnection+    -- | Tell the remote endpoint we will no longer be using a connection+  | CloseConnection+    -- | Request to close the connection (see module description)+  | CloseSocket+    -- | Sent by an endpoint when it is closed.+  | CloseEndPoint+    -- | Message sent to probe a socket+  | ProbeSocket+    -- | Acknowledgement of the ProbeSocket message+  | ProbeSocketAck+  deriving (Show)++decodeControlHeader :: Word32 -> Maybe ControlHeader+decodeControlHeader w32 = case w32 of+  0 -> Just CreatedNewConnection+  1 -> Just CloseConnection+  2 -> Just CloseSocket+  3 -> Just CloseEndPoint+  4 -> Just ProbeSocket+  5 -> Just ProbeSocketAck+  _ -> Nothing++encodeControlHeader :: ControlHeader -> Word32+encodeControlHeader ch = case ch of+  CreatedNewConnection -> 0+  CloseConnection      -> 1+  CloseSocket          -> 2+  CloseEndPoint        -> 3+  ProbeSocket          -> 4+  ProbeSocketAck       -> 5++-- | Response sent by /B/ to /A/ when /A/ tries to connect+data ConnectionRequestResponse =+    -- | /B/ does not support the protocol version requested by /A/.+    ConnectionRequestUnsupportedVersion+    -- | /B/ accepts the connection+  | ConnectionRequestAccepted+    -- | /A/ requested an invalid endpoint+  | ConnectionRequestInvalid+    -- | /A/s request crossed with a request from /B/ (see protocols)+  | ConnectionRequestCrossed+    -- | /A/ gave an incorrect host (did not match the host that /B/ observed).+  | ConnectionRequestHostMismatch+  deriving (Show)++decodeConnectionRequestResponse :: Word32 -> Maybe ConnectionRequestResponse+decodeConnectionRequestResponse w32 = case w32 of+  0xFFFFFFFF -> Just ConnectionRequestUnsupportedVersion+  0x00000000 -> Just ConnectionRequestAccepted+  0x00000001 -> Just ConnectionRequestInvalid+  0x00000002 -> Just ConnectionRequestCrossed+  0x00000003 -> Just ConnectionRequestHostMismatch+  _          -> Nothing++encodeConnectionRequestResponse :: ConnectionRequestResponse -> Word32+encodeConnectionRequestResponse crr = case crr of+  ConnectionRequestUnsupportedVersion -> 0xFFFFFFFF+  ConnectionRequestAccepted           -> 0x00000000+  ConnectionRequestInvalid            -> 0x00000001+  ConnectionRequestCrossed            -> 0x00000002+  ConnectionRequestHostMismatch       -> 0x00000003+ -- | Start a server at the specified address. -- -- This sets up a server socket for the specified host and port. Exceptions -- thrown during setup are not caught. -- -- Once the socket is created we spawn a new thread which repeatedly accepts--- incoming connections and executes the given request handler. If any--- exception occurs the thread terminates and calls the terminationHandler.--- This exception may occur because of a call to 'N.accept', because the thread--- was explicitly killed, or because of a synchronous exception thrown by the--- request handler. Typically, you should avoid the last case by catching any--- relevant exceptions in the request handler.+-- incoming connections and executes the given request handler in another+-- thread. If any exception occurs the accepting thread terminates and calls+-- the terminationHandler. Threads spawned for previous accepted connections+-- are not killed.+-- This exception may occur because of a call to 'N.accept', or because the+-- thread was explicitly killed. ----- The request handler should spawn threads to handle each individual request--- or the server will block. Once a thread has been spawned it will be the--- responsibility of the new thread to close the socket when an exception--- occurs.+-- The request handler is not responsible for closing the socket. It will be+-- closed once that handler returns. Take care to ensure that the socket is not+-- used after the handler returns, or you will get undefined behavior+-- (the file descriptor may be re-used). -- -- The return value includes the port was bound to. This is not always the same -- port as that given in the argument. For example, binding to port 0 actually -- binds to a random port, selected by the OS.-forkServer :: N.HostName               -- ^ Host-           -> N.ServiceName            -- ^ Port-           -> Int                      -- ^ Backlog (maximum number of queued connections)-           -> Bool                     -- ^ Set ReuseAddr option?-           -> (SomeException -> IO ()) -- ^ Termination handler-           -> (N.Socket -> IO ())      -- ^ Request handler+forkServer :: N.HostName                     -- ^ Host+           -> N.ServiceName                  -- ^ Port+           -> Int                            -- ^ Backlog (maximum number of queued connections)+           -> Bool                           -- ^ Set ReuseAddr option?+           -> (SomeException -> IO ())       -- ^ Error handler. Called with an+                                             --   exception raised when+                                             --   accepting a connection.+           -> (SomeException -> IO ())       -- ^ Termination handler. Called+                                             --   when the error handler throws+                                             --   an exception.+           -> (IO () -> (N.Socket, N.SockAddr) -> IO ())+                                             -- ^ Request handler. Gets an+                                             --   action which completes when+                                             --   the socket is closed.            -> IO (N.ServiceName, ThreadId)-forkServer host port backlog reuseAddr terminationHandler requestHandler = do+forkServer host port backlog reuseAddr errorHandler terminationHandler requestHandler = do     -- Resolve the specified address. By specification, getAddrInfo will never     -- return an empty list (but will throw an exception instead) and will return     -- the "best" address first, whatever that means@@ -90,50 +222,129 @@       when reuseAddr $ N.setSocketOption sock N.ReuseAddr 1       N.bindSocket sock (N.addrAddress addr)       N.listen sock backlog++      -- Close up and fill the synchonizing MVar.+      let release :: ((N.Socket, N.SockAddr), MVar ()) -> IO ()+          release ((sock, _), socketClosed) =+            N.sClose sock `finally` putMVar socketClosed ()++      -- Run the request handler.+      let act restore (sock, sockAddr) = do+            socketClosed <- newEmptyMVar+            void $ forkIO $ restore $ do+              requestHandler (readMVar socketClosed) (sock, sockAddr)+              `finally`+              release ((sock, sockAddr), socketClosed)++      let acceptRequest :: IO ()+          acceptRequest = mask $ \restore -> do+            -- Async exceptions are masked so that, if accept does give a+            -- socket, we'll always deliver it to the handler before the+            -- exception is raised.+            -- If it's a Right handler then it will eventually be closed.+            -- If it's a Left handler then we assume the handler itself will+            -- close it.+            (sock, sockAddr) <- N.accept sock+            -- Looks like 'act' will never throw an exception, but to be+            -- safe we'll close the socket if it does.+            let handler :: SomeException -> IO ()+                handler _ = N.sClose sock+            catch (act restore (sock, sockAddr)) handler+       -- We start listening for incoming requests in a separate thread. When       -- that thread is killed, we close the server socket and the termination-      -- handler. We have to make sure that the exception handler is installed-      -- /before/ any asynchronous exception occurs. So we mask_, then fork-      -- (the child thread inherits the masked state from the parent), then+      -- handler is run. We have to make sure that the exception handler is+      -- installed /before/ any asynchronous exception occurs. So we mask_, then+      -- fork (the child thread inherits the masked state from the parent), then       -- unmask only inside the catch.       (,) <$> fmap show (N.socketPort sock) <*>         (mask_ $ forkIOWithUnmask $ \unmask ->-          catch (unmask (forever $ acceptRequest sock)) $ \ex -> do+          catch (unmask (forever (catch acceptRequest errorHandler))) $ \ex -> do             tryCloseSocket sock             terminationHandler ex)-  where-    acceptRequest :: N.Socket -> IO ()-    acceptRequest sock = bracketOnError (N.accept sock)-                                        (tryCloseSocket . fst)-                                        (requestHandler . fst) --- | Read a length and then a payload of that length-recvWithLength :: N.Socket -> IO [ByteString]-recvWithLength sock = recvInt32 sock >>= recvExact sock+-- | Read a length and then a payload of that length, subject to a limit+--   on the length.+--   If the length (first 'Word32' received) is greater than the limit then+--   an exception is thrown.+recvWithLength :: Word32 -> N.Socket -> IO [ByteString]+recvWithLength limit sock = do+  len <- recvWord32 sock+  when (len > limit) $+    throwIO (userError "recvWithLength: limit exceeded")+  recvExact sock len --- | Receive a 32-bit integer-recvInt32 :: Num a => N.Socket -> IO a-recvInt32 sock = decodeInt32 . BS.concat <$> recvExact sock 4+-- | Receive a 32-bit unsigned integer+recvWord32 :: N.Socket -> IO Word32+recvWord32 = fmap (decodeWord32 . BS.concat) . flip recvExact 4 --- | Close a socket, ignoring I/O exceptions+-- | Close a socket, ignoring I/O exceptions. tryCloseSocket :: N.Socket -> IO () tryCloseSocket sock = void . tryIO $   N.sClose sock +-- | Shutdown socket sends and receives, ignoring I/O exceptions.+tryShutdownSocketBoth :: N.Socket -> IO ()+tryShutdownSocketBoth sock = void . tryIO $+  N.shutdown sock N.ShutdownBoth+ -- | Read an exact number of bytes from a socket -- -- Throws an I/O exception if the socket closes before the specified -- number of bytes could be read-recvExact :: N.Socket                -- ^ Socket to read from-          -> Int32                   -- ^ Number of bytes to read-          -> IO [ByteString]-recvExact _ len | len < 0 = throwIO (userError "recvExact: Negative length")+recvExact :: N.Socket        -- ^ Socket to read from+          -> Word32          -- ^ Number of bytes to read+          -> IO [ByteString] -- ^ Data read recvExact sock len = go [] len   where-    go :: [ByteString] -> Int32 -> IO [ByteString]+    go :: [ByteString] -> Word32 -> IO [ByteString]     go acc 0 = return (reverse acc)     go acc l = do       bs <- NBS.recv sock (fromIntegral l `min` smallChunkSize)       if BS.null bs         then throwIO (userError "recvExact: Socket closed")         else go (bs : acc) (l - fromIntegral (BS.length bs))++-- | Produce a HostName and ServiceName from a SockAddr. Only gives 'Just' for+-- IPv4 addresses.+decodeSockAddr :: N.SockAddr -> IO (Maybe (N.HostName, N.ServiceName))+decodeSockAddr sockAddr = case sockAddr of+  N.SockAddrInet port host -> do+    hostString <- N.inet_ntoa host+    return $ Just (hostString, show port)+  _ -> return Nothing++-- | Encode end point address+encodeEndPointAddress :: N.HostName+                      -> N.ServiceName+                      -> EndPointId+                      -> EndPointAddress+encodeEndPointAddress host port ix = EndPointAddress . BSC.pack $+  host ++ ":" ++ port ++ ":" ++ show ix++-- | Decode end point address+decodeEndPointAddress :: EndPointAddress+                      -> Maybe (N.HostName, N.ServiceName, EndPointId)+decodeEndPointAddress (EndPointAddress bs) =+  case splitMaxFromEnd (== ':') 2 $ BSC.unpack bs of+    [host, port, endPointIdStr] ->+      case reads endPointIdStr of+        [(endPointId, "")] -> Just (host, port, endPointId)+        _                  -> Nothing+    _ ->+      Nothing++-- | @spltiMaxFromEnd p n xs@ splits list @xs@ at elements matching @p@,+-- returning at most @p@ segments -- counting from the /end/+--+-- > splitMaxFromEnd (== ':') 2 "ab:cd:ef:gh" == ["ab:cd", "ef", "gh"]+splitMaxFromEnd :: (a -> Bool) -> Int -> [a] -> [[a]]+splitMaxFromEnd p = \n -> go [[]] n . reverse+  where+    -- go :: [[a]] -> Int -> [a] -> [[a]]+    go accs         _ []     = accs+    go ([]  : accs) 0 xs     = reverse xs : accs+    go (acc : accs) n (x:xs) =+      if p x then go ([] : acc : accs) (n - 1) xs+             else go ((x : acc) : accs) n xs+    go _ _ _ = error "Bug in splitMaxFromEnd"

@@ -1,4 +1,5 @@ {-# LANGUAGE RebindableSyntax, TemplateHaskell #-}+{-# LANGUAGE BangPatterns #-} {-# OPTIONS_GHC -fno-warn-orphans #-} module Main where @@ -15,11 +16,10 @@ import Network.Transport.TCP ( createTransport                              , createTransportExposeInternals                              , TransportInternals(..)-                             , encodeEndPointAddress+                             , TCPParameters(..)                              , defaultTCPParameters                              , LightweightConnectionId                              )-import Data.Int (Int32) import Control.Concurrent (threadDelay, killThread) import Control.Concurrent.MVar ( MVar                                , newEmptyMVar@@ -30,21 +30,31 @@                                , newMVar                                , modifyMVar                                , modifyMVar_+                               , swapMVar                                ) import Control.Monad (replicateM, guard, forM_, replicateM_, when) import Control.Applicative ((<$>)) import Control.Exception (throwIO, try, SomeException)-import Network.Transport.TCP ( ControlHeader(..)-                             , ConnectionRequestResponse(..)-                             , socketToEndPoint-                             )-import Network.Transport.Internal ( encodeInt32-                                  , prependLength+import Network.Transport.TCP ( socketToEndPoint )+import Network.Transport.Internal ( prependLength                                   , tlog                                   , tryIO                                   , void                                   )-import Network.Transport.TCP.Internal (recvInt32, forkServer, recvWithLength)+import Network.Transport.TCP.Internal+  ( ControlHeader(..)+  , encodeControlHeader+  , decodeControlHeader+  , ConnectionRequestResponse(..)+  , encodeConnectionRequestResponse+  , decodeConnectionRequestResponse+  , encodeWord32+  , recvWord32+  , forkServer+  , recvWithLength+  , encodeEndPointAddress+  , decodeEndPointAddress+  )  #ifdef USE_MOCK_NETWORK import qualified Network.Transport.TCP.Mock.Socket as N@@ -57,6 +67,15 @@   , AddrInfo   , shutdown   , ShutdownCmd(ShutdownSend)+  , SockAddr(..)+  , SocketType(Stream)+  , AddrInfo(..)+  , getAddrInfo+  , defaultHints+  , defaultProtocol+  , socket+  , connect+  , close   )  #ifdef USE_MOCK_NETWORK@@ -65,6 +84,7 @@ import Network.Socket.ByteString (sendMany) #endif +import qualified Data.ByteString as BS (length, concat) import Data.String (fromString) import GHC.IO.Exception (ioe_errno) import Foreign.C.Error (Errno(..), eADDRNOTAVAIL)@@ -105,7 +125,7 @@     server :: MVar EndPointAddress -> MVar EndPointAddress -> MVar () -> IO ()     server serverAddr clientAddr serverDone = do       tlog "Server"-      Right transport <- createTransport "127.0.0.1" "0" defaultTCPParameters+      Right transport <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters       Right endpoint  <- newEndPoint transport       putMVar serverAddr (address endpoint)       theirAddr <- readMVar clientAddr@@ -154,23 +174,26 @@       tlog "Client"        -- Listen for incoming messages-      (clientPort, _) <- forkServer "127.0.0.1" "0" 5 True throwIO $ \sock -> do+      (clientPort, _) <- forkServer "127.0.0.1" "0" 5 True throwIO throwIO $ \socketFree (sock, _) -> do         -- Initial setup-        0 <- recvInt32 sock :: IO Int-        _ <- recvWithLength sock-        sendMany sock [encodeInt32 ConnectionRequestAccepted]+        0 <- recvWord32 sock+        _ <- recvWithLength maxBound sock+        sendMany sock [encodeWord32 (encodeConnectionRequestResponse ConnectionRequestAccepted)]          -- Server opens  a logical connection-        CreatedNewConnection <- toEnum <$> (recvInt32 sock :: IO Int)-        1024 <- recvInt32 sock :: IO LightweightConnectionId+        Just CreatedNewConnection <- decodeControlHeader <$> recvWord32 sock+        1024 <- recvWord32 sock :: IO LightweightConnectionId          -- Server sends a message-        1024 <- recvInt32 sock :: IO Int-        ["ping"] <- recvWithLength sock+        1024 <- recvWord32 sock+        ["ping"] <- recvWithLength maxBound sock          -- Reply-        sendMany sock [encodeInt32 CreatedNewConnection, encodeInt32 (10002 :: Int)]-        sendMany sock (encodeInt32 10002 : prependLength ["pong"])+        sendMany sock [+            encodeWord32 (encodeControlHeader CreatedNewConnection)+          , encodeWord32 10002+          ]+        sendMany sock (encodeWord32 10002 : prependLength ["pong"])          -- Close the socket         N.sClose sock@@ -179,10 +202,13 @@       putMVar clientAddr ourAddress        -- Connect to the server-      Right (sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr True False False Nothing Nothing+      Right (_, sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr True False False Nothing Nothing        -- Open a new connection-      sendMany sock [encodeInt32 CreatedNewConnection, encodeInt32 (10003 :: Int)]+      sendMany sock [+          encodeWord32 (encodeControlHeader CreatedNewConnection)+        , encodeWord32 10003+        ]        -- Close the socket without closing the connection explicitly       -- The server should receive an error event@@ -204,7 +230,7 @@     server :: MVar EndPointAddress -> MVar EndPointAddress -> MVar () -> IO ()     server serverAddr clientAddr serverDone = do       tlog "Server"-      Right transport <- createTransport "127.0.0.1" "0" defaultTCPParameters+      Right transport <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters       Right endpoint  <- newEndPoint transport       putMVar serverAddr (address endpoint)       theirAddr <- readMVar clientAddr@@ -260,53 +286,65 @@       tlog "Client"        -- Listen for incoming messages-      (clientPort, _) <- forkServer "127.0.0.1" "0" 5 True throwIO $ \sock -> do+      (clientPort, _) <- forkServer "127.0.0.1" "0" 5 True throwIO throwIO $ \socketFree (sock, _) -> do         -- Initial setup-        0 <- recvInt32 sock :: IO Int-        _ <- recvWithLength sock-        sendMany sock [encodeInt32 ConnectionRequestAccepted]+        0 <- recvWord32 sock+        _ <- recvWithLength maxBound sock+        sendMany sock [encodeWord32 (encodeConnectionRequestResponse ConnectionRequestAccepted)]          -- Server opens a logical connection-        CreatedNewConnection <- toEnum <$> (recvInt32 sock :: IO Int)-        1024 <- recvInt32 sock :: IO LightweightConnectionId+        Just CreatedNewConnection <- decodeControlHeader <$> recvWord32 sock+        1024 <- recvWord32 sock :: IO LightweightConnectionId          -- Server sends a message-        1024 <- recvInt32 sock :: IO Int-        ["ping"] <- recvWithLength sock+        1024 <- recvWord32 sock+        ["ping"] <- recvWithLength maxBound sock          -- Reply-        sendMany sock [encodeInt32 CreatedNewConnection, encodeInt32 (10002 :: Int)]-        sendMany sock (encodeInt32 (10002 :: Int) : prependLength ["pong"])+        sendMany sock [+            encodeWord32 (encodeControlHeader CreatedNewConnection)+          , encodeWord32 10002+          ]+        sendMany sock (encodeWord32 10002 : prependLength ["pong"])          -- Send a CloseSocket even though there are still connections *in both         -- directions*-        sendMany sock [encodeInt32 CloseSocket, encodeInt32 (1024 :: Int)]+        sendMany sock [+            encodeWord32 (encodeControlHeader CloseSocket)+          , encodeWord32 1024+          ]         N.sClose sock        let ourAddress = encodeEndPointAddress "127.0.0.1" clientPort 0       putMVar clientAddr ourAddress        -- Connect to the server-      Right (sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr True False False Nothing Nothing+      Right (_, sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr True False False Nothing Nothing        -- Open a new connection-      sendMany sock [encodeInt32 CreatedNewConnection, encodeInt32 (10003 :: Int)]+      sendMany sock [+          encodeWord32 (encodeControlHeader CreatedNewConnection)+        , encodeWord32 10003+        ]        -- Send a CloseSocket without sending a closeconnecton       -- The server should still receive a ConnectionClosed message-      sendMany sock [encodeInt32 CloseSocket, encodeInt32 (0 :: Int)]+      sendMany sock [+          encodeWord32 (encodeControlHeader CloseSocket)+        , encodeWord32 0+        ]       N.sClose sock  -- | Test the creation of a transport with an invalid address testInvalidAddress :: IO () testInvalidAddress = do-  Left _ <- createTransport "invalidHostName" "0" defaultTCPParameters+  Left _ <- createTransport "invalidHostName" "0" ((,) "invalidHostName") defaultTCPParameters   return ()  -- | Test connecting to invalid or non-existing endpoints testInvalidConnect :: IO () testInvalidConnect = do-  Right transport <- createTransport "127.0.0.1" "0" defaultTCPParameters+  Right transport <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters   Right endpoint  <- newEndPoint transport    -- Syntax error in the endpoint address@@ -337,7 +375,7 @@   clientDone <- newEmptyMVar   serverDone <- newEmptyMVar   connectionEstablished <- newEmptyMVar-  Right transport <- createTransport "127.0.0.1" "0" defaultTCPParameters+  Right transport <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters    -- Server   forkTry $ do@@ -375,32 +413,41 @@     theirAddress <- readMVar serverAddr      -- Connect to the server-    Right (sock, ConnectionRequestAccepted) <- socketToEndPoint ourAddress theirAddress True False False Nothing Nothing+    Right (_, sock, ConnectionRequestAccepted) <- socketToEndPoint ourAddress theirAddress True False False Nothing Nothing     putMVar connectionEstablished ()      -- Server connects to us, and then closes the connection-    CreatedNewConnection <- toEnum <$> (recvInt32 sock :: IO Int)-    1024 <- recvInt32 sock :: IO LightweightConnectionId+    Just CreatedNewConnection <- decodeControlHeader <$> recvWord32 sock+    1024 <- recvWord32 sock :: IO LightweightConnectionId -    CloseConnection <-  toEnum <$> (recvInt32 sock :: IO Int)-    1024 <- recvInt32 sock :: IO LightweightConnectionId+    Just CloseConnection <- decodeControlHeader <$> recvWord32 sock+    1024 <- recvWord32 sock :: IO LightweightConnectionId      -- Server will now send a CloseSocket request as its refcount reached 0     tlog "Waiting for CloseSocket request"-    CloseSocket <- toEnum <$> recvInt32 sock-    _ <- recvInt32 sock :: IO LightweightConnectionId+    Just CloseSocket <- decodeControlHeader <$> recvWord32 sock+    _ <- recvWord32 sock :: IO LightweightConnectionId      -- But we ignore it and request another connection in the other direction     tlog "Ignoring it, requesting another connection"-    sendMany sock [encodeInt32 CreatedNewConnection, encodeInt32 (1024 :: Int)]+    sendMany sock [+        encodeWord32 (encodeControlHeader CreatedNewConnection)+      , encodeWord32 1024+      ]      -- Close it again     tlog "Closing connection"-    sendMany sock [encodeInt32 CloseConnection, encodeInt32 (1024 :: Int)]+    sendMany sock [+        encodeWord32 (encodeControlHeader CloseConnection)+      , encodeWord32 1024+      ]      -- And close the connection completely     tlog "Closing socket"-    sendMany sock [encodeInt32 CloseSocket, encodeInt32 (1024 :: Int)]+    sendMany sock [+        encodeWord32 (encodeControlHeader CloseSocket)+      , encodeWord32 1024+      ]     N.sClose sock      putMVar clientDone ()@@ -418,7 +465,7 @@   clientDone <- newEmptyMVar   serverDone <- newEmptyMVar   connectionEstablished <- newEmptyMVar-  Right transport <- createTransport "127.0.0.1" "0" defaultTCPParameters+  Right transport <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters    -- Server   forkTry $ do@@ -452,20 +499,20 @@     theirAddress <- readMVar serverAddr      -- Connect to the server-    Right (sock, ConnectionRequestAccepted) <- socketToEndPoint ourAddress theirAddress True False False Nothing Nothing+    Right (_, sock, ConnectionRequestAccepted) <- socketToEndPoint ourAddress theirAddress True False False Nothing Nothing     putMVar connectionEstablished ()      -- Server connects to us, and then closes the connection-    CreatedNewConnection <- toEnum <$> (recvInt32 sock :: IO Int)-    1024 <- recvInt32 sock :: IO LightweightConnectionId+    Just CreatedNewConnection <- decodeControlHeader <$> recvWord32 sock+    1024 <- recvWord32 sock :: IO LightweightConnectionId -    CloseConnection <-  toEnum <$> (recvInt32 sock :: IO Int)-    1024 <- recvInt32 sock :: IO LightweightConnectionId+    Just CloseConnection <- decodeControlHeader <$> recvWord32 sock+    1024 <- recvWord32 sock :: IO LightweightConnectionId      -- Server will now send a CloseSocket request as its refcount reached 0     tlog "Waiting for CloseSocket request"-    CloseSocket <- toEnum <$> recvInt32 sock-    _ <- recvInt32 sock :: IO LightweightConnectionId+    Just CloseSocket <- decodeControlHeader <$> recvWord32 sock+    _ <- recvWord32 sock :: IO LightweightConnectionId      unblocked <- newMVar False @@ -473,7 +520,7 @@     -- responding to the CloseSocket request (in this case, we     -- respond by sending a ConnectionRequest)     forkTry $ do-      recvInt32 sock :: IO Int32+      recvWord32 sock       readMVar unblocked >>= guard       putMVar clientDone () @@ -482,7 +529,10 @@     tlog "Client ignores close socket and sends connection request"     tlog "This should unblock the server"     modifyMVar_ unblocked $ \_ -> return True-    sendMany sock [encodeInt32 CreatedNewConnection, encodeInt32 (1024 :: Int)]+    sendMany sock [+        encodeWord32 (encodeControlHeader CreatedNewConnection)+      , encodeWord32 1024+      ]    takeMVar clientDone   takeMVar serverDone@@ -495,13 +545,18 @@   serverAddr <- newEmptyMVar    forkTry $ do-    Right transport <- createTransport "127.0.0.1" "0" defaultTCPParameters+    Right transport <- createTransport "127.0.0.1" "0" ((,) "128.0.0.1") defaultTCPParameters     Right endpoint <- newEndPoint transport-    putMVar serverAddr (address endpoint)+    -- Since we're lying about the server's address, we have to manually+    -- construct the proper address. If we used its actual address, the clients+    -- would try to resolve "128.0.0.1" and then would fail due to invalid+    -- address.+    Just (_, port, epid) <- return $ decodeEndPointAddress (address endpoint)+    putMVar serverAddr $ encodeEndPointAddress "127.0.0.1" port epid    forkTry $ do-    -- We pick an address < 127.0.0.1 so that this is not rejected purely because of the "crossed" check-    let ourAddress = EndPointAddress "126.0.0.1"+    -- We pick an address < 128.0.0.1 so that this is not rejected purely because of the "crossed" check+    let ourAddress = encodeEndPointAddress "127.0.0.1" "1234" 0      -- We should only get a single 'Accepted' reply     gotAccepted <- newEmptyMVar@@ -512,14 +567,14 @@         -- immediately (depending on far the remote endpoint got with the initialization)         response <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr True False False Nothing Nothing         case response of-          Right (_, ConnectionRequestAccepted) ->+          Right (_, _, ConnectionRequestAccepted) ->             -- We don't close this socket because we want to keep this connection open             putMVar gotAccepted ()           -- We might get either Invalid or Crossed (the transport does not           -- maintain enough history to be able to tell)-          Right (sock, ConnectionRequestInvalid) ->+          Right (_, sock, ConnectionRequestInvalid) ->             N.sClose sock-          Right (sock, ConnectionRequestCrossed) ->+          Right (_, sock, ConnectionRequestCrossed) ->             N.sClose sock           Left _ ->             return ()@@ -534,11 +589,11 @@ -- | Test that we can create "many" transport instances testMany :: IO () testMany = do-  Right masterTransport <- createTransport "127.0.0.1" "0" defaultTCPParameters+  Right masterTransport <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters   Right masterEndPoint  <- newEndPoint masterTransport    replicateM_ 10 $ do-    mTransport <- createTransport "127.0.0.1" "0" defaultTCPParameters+    mTransport <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters     case mTransport of       Left ex -> do         putStrLn $ "IOException: " ++ show ex ++ "; errno = " ++ show (ioe_errno ex)@@ -555,7 +610,7 @@ -- | Test what happens when the transport breaks completely testBreakTransport :: IO () testBreakTransport = do-  Right (transport, internals) <- createTransportExposeInternals "127.0.0.1" "0" defaultTCPParameters+  Right (transport, internals) <- createTransportExposeInternals "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters   Right endpoint <- newEndPoint transport    killThread (transportThread internals) -- Uh oh@@ -564,6 +619,13 @@    return () +-- Used in testReconnect to block until a socket is closed. newtype is needed+-- for the Traceable instance.+newtype WaitSocketFree = WaitSocketFree (IO ())++instance Traceable WaitSocketFree where+  trace = const Nothing+ -- | Test that a second call to 'connect' might succeed even if the first -- failed. This is a TCP specific test rather than an endpoint specific test -- because we must manually create the endpoint address to match an endpoint we@@ -575,72 +637,110 @@ testReconnect = do   serverDone      <- newEmptyMVar   endpointCreated <- newEmptyMVar+  -- The server will put the 'socketFree' IO in here, so that the client can+  -- block until the server has closed the socket.+  socketClosed    <- newEmptyMVar    counter <- newMVar (0 :: Int)    -- Server-  (serverPort, _) <- forkServer "127.0.0.1" "0" 5 True throwIO $ \sock -> do+  (serverPort, _) <- forkServer "127.0.0.1" "0" 5 True throwIO throwIO $ \socketFree (sock, _) -> do     -- Accept the connection-    Right 0  <- tryIO $ (recvInt32 sock :: IO Int)-    Right _  <- tryIO $ recvWithLength sock+    Right 0  <- tryIO $ recvWord32 sock+    Right _  <- tryIO $ recvWithLength maxBound sock      -- The first time we close the socket before accepting the logical connection     count <- modifyMVar counter $ \i -> return (i + 1, i) +    -- Wait 100ms after the socket closes, to (hopefully) ensure that the client+    -- knows the connection is closed, and sending on that socket will therefore+    -- fail.+    putMVar socketClosed (WaitSocketFree (socketFree >> threadDelay 100000))+     when (count > 0) $ do-      Right () <- tryIO $ sendMany sock [encodeInt32 ConnectionRequestAccepted]+      -- The second, third, and fourth connections are accepted according to the+      -- protocol.+      -- On the second request, the socket then closes.+      Right () <- tryIO $ sendMany sock [+          encodeWord32 (encodeConnectionRequestResponse ConnectionRequestAccepted)+        ]       -- Client requests a logical connection       when (count > 1) $ do-        Right CreatedNewConnection <- tryIO $ toEnum <$> (recvInt32 sock :: IO Int)-        connId <- recvInt32 sock :: IO LightweightConnectionId-        return ()+        -- On the third and fourth requests, a new logical connection is+        -- accepted.+        -- On the third request the socket then closes.+        Right (Just CreatedNewConnection) <- tryIO $ decodeControlHeader <$> recvWord32 sock+        connId <- recvWord32 sock :: IO LightweightConnectionId          when (count > 2) $ do-          -- Client sends a message-          Right connId' <- tryIO $ (recvInt32 sock :: IO LightweightConnectionId)+          -- On the fourth request, a message is received and then the socket+          -- is closed.+          Right connId' <- tryIO $ (recvWord32 sock :: IO LightweightConnectionId)           True <- return $ connId == connId'-          Right ["ping"] <- tryIO $ recvWithLength sock+          Right ["ping"] <- tryIO $ recvWithLength maxBound sock           putMVar serverDone () -    Right () <- tryIO $ N.sClose sock     return ()    putMVar endpointCreated ()    -- Client   forkTry $ do-    Right transport <- createTransport "127.0.0.1" "0" defaultTCPParameters+    Right transport <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters     Right endpoint  <- newEndPoint transport     let theirAddr = encodeEndPointAddress "127.0.0.1" serverPort 0 -    -- The second attempt will fail because the server closes the socket before we can request a connection     takeMVar endpointCreated-    -- This might time out or not, depending on whether the server closes the-    -- socket before or after we can send the RequestConnectionId request-    resultConnect <- timeout 500000 $ connect endpoint theirAddr ReliableOrdered defaultConnectHints++    -- First attempt: fails because the server closes the socket without+    -- doing the handshake.+    resultConnect <- connect endpoint theirAddr ReliableOrdered defaultConnectHints     case resultConnect of-      Nothing -> return ()-      Just (Left (TransportError ConnectFailed _)) -> return ()-      Just (Left err) -> throwIO err-      Just (Right _) -> throwIO $ userError "testConnect: unexpected connect success"+      Left (TransportError ConnectFailed _) -> return ()+      Left err -> throwIO err+      Right _ -> throwIO $ userError "testConnect: unexpected connect success"+    WaitSocketFree wait <- takeMVar socketClosed+    wait -    resultConnect <- timeout 500000 $ connect endpoint theirAddr ReliableOrdered defaultConnectHints+    -- Second attempt: server accepts the connection but then closes the socket.+    -- We expect a failed connection if the socket is closed *before*+    -- CreatedNewConnection is sent, or a successful connection such that a+    -- subsequent send will fail in case CreatedNewConnection was sent before+    -- the close.+    resultConnect <- connect endpoint theirAddr ReliableOrdered defaultConnectHints+    -- We must be sure that the socket has closed before trying to send.+    WaitSocketFree wait <- takeMVar socketClosed+    wait     case resultConnect of-      Nothing -> return ()-      Just (Left (TransportError ConnectFailed _)) -> return ()-      Just (Left err) -> throwIO err-      Just (Right c) -> do-        ev <- send c ["foo"]+      Left (TransportError ConnectFailed _) -> return ()+      Left err -> throwIO err+      Right c -> do+        ev <- send c ["ping"]         case ev of           Left _ -> return ()           Right _ -> throwIO $ userError "testConnect: unexpected send success" -    -- The third attempt succeeds-    Right conn1 <- connect endpoint theirAddr ReliableOrdered defaultConnectHints+    -- In any case, since a heavyweight connection was made, we'll get a+    -- connection lost event.+    ErrorEvent (TransportError (EventConnectionLost _) _) <- receive endpoint -    -- But a send will fail because the server has closed the connection again-    threadDelay 100000-    Left (TransportError SendFailed _) <- send conn1 ["ping"]+    -- Third attempt: server accepts the heavyweight and the lightweight+    -- connection (CreatedNewConnection) but then closes the socket.+    -- The connection must succeed, but sending after the socket is closed+    -- must fail.+    resultConnect <- connect endpoint theirAddr ReliableOrdered defaultConnectHints+    -- Wait until close before trying to send.+    WaitSocketFree wait <- takeMVar socketClosed+    wait+    case resultConnect of+      Left err -> throwIO err+      Right c -> do+        ev <- send c ["ping"]+        case ev of+          Left (TransportError SendFailed _) -> return ()+          Left err -> throwIO err+          Right _ -> throwIO $ userError "testConnect: unexpected send success"+     ErrorEvent (TransportError (EventConnectionLost _) _) <- receive endpoint      -- But a subsequent call to connect should reestablish the connection@@ -648,6 +748,9 @@      -- Send should now succeed     Right () <- send conn2 ["ping"]++    WaitSocketFree wait <- takeMVar socketClosed+    wait     return ()    takeMVar serverDone@@ -662,28 +765,32 @@   serverGotPing <- newEmptyMVar    -- Server-  (serverPort, _) <- forkServer "127.0.0.1" "0" 5 True throwIO $ \sock -> do+  (serverPort, _) <- forkServer "127.0.0.1" "0" 5 True throwIO throwIO $ \socketFree (sock, _) -> do     -- We accept connections, but when an exception occurs we don't do     -- anything (in particular, we don't close the socket). This is important     -- because when we shutdown one direction of the socket a recv here will     -- fail, but we don't want to close that socket at that point (which     -- would shutdown the socket in the other direction)     void . (try :: IO () -> IO (Either SomeException ())) $ do-      0 <- recvInt32 sock :: IO Int-      _ <- recvWithLength sock-      () <- sendMany sock [encodeInt32 ConnectionRequestAccepted]+      0 <- recvWord32 sock+      _ <- recvWithLength maxBound sock+      () <- sendMany sock [encodeWord32 (encodeConnectionRequestResponse ConnectionRequestAccepted)] -      CreatedNewConnection <- toEnum <$> (recvInt32 sock :: IO Int)-      connId <- recvInt32 sock :: IO LightweightConnectionId+      Just CreatedNewConnection <- decodeControlHeader <$> recvWord32 sock+      connId <- recvWord32 sock :: IO LightweightConnectionId -      connId' <- recvInt32 sock :: IO LightweightConnectionId+      connId' <- recvWord32 sock :: IO LightweightConnectionId       True <- return $ connId == connId'-      ["ping"] <- recvWithLength sock+      ["ping"] <- recvWithLength maxBound sock       putMVar serverGotPing () +    -- Must read the clientDone MVar so that we don't close the socket+    -- (forkServer will close it once this action ends).+    readMVar clientDone+   -- Client   forkTry $ do-    Right (transport, internals) <- createTransportExposeInternals "127.0.0.1" "0" defaultTCPParameters+    Right (transport, internals) <- createTransportExposeInternals "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters     Right endpoint <- newEndPoint transport     let theirAddr = encodeEndPointAddress "127.0.0.1" serverPort 0 @@ -734,11 +841,11 @@      putMVar clientDone  () -  takeMVar clientDone+  readMVar clientDone  testInvalidCloseConnection :: IO () testInvalidCloseConnection = do-  Right (transport, internals) <- createTransportExposeInternals "127.0.0.1" "0" defaultTCPParameters+  Right (transport, internals) <- createTransportExposeInternals "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters   serverAddr <- newEmptyMVar   clientDone <- newEmptyMVar   serverDone <- newEmptyMVar@@ -767,7 +874,10 @@      -- Get a handle on the TCP connection and manually send an invalid CloseConnection request     sock <- socketBetween internals ourAddr theirAddr-    sendMany sock [encodeInt32 CloseConnection, encodeInt32 (12345 :: Int)]+    sendMany sock [+        encodeWord32 (encodeControlHeader CloseConnection)+      , encodeWord32 (12345 :: LightweightConnectionId)+      ]      putMVar clientDone () @@ -777,19 +887,157 @@ testUseRandomPort = do    testDone <- newEmptyMVar    forkTry $ do-     Right transport1 <- createTransport "127.0.0.1" "0" defaultTCPParameters+     Right transport1 <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters      Right ep1        <- newEndPoint transport1-     Right transport2 <- createTransport "127.0.0.1" "0" defaultTCPParameters+     -- Same as transport1, but is strict in the port.+     Right transport2 <- createTransport "127.0.0.1" "0" (\(!port) -> ("127.0.0.1", port)) defaultTCPParameters      Right ep2        <- newEndPoint transport2      Right conn1 <- connect ep2 (address ep1) ReliableOrdered defaultConnectHints      ConnectionOpened _ _ _ <- receive ep1      putMVar testDone ()    takeMVar testDone +-- | Verify that if a peer sends an address or data which exceeds the maximum+--   length, that peer's connection will be terminated, but other peers will+--   not be affected.+testMaxLength :: IO ()+testMaxLength = do++  Right serverTransport <- createTransport "127.0.0.1" "9998" ((,) "127.0.0.1") $ defaultTCPParameters {+      -- 17 bytes should fit every valid address at 127.0.0.1.+      -- Port is at most 5 bytes (65536) and id is a base-10 Word32 so+      -- at most 10 bytes. We'll have one client with a 5-byte port to push it+      -- over the chosen limit of 16+      tcpMaxAddressLength = 16+    , tcpMaxReceiveLength = 8+    }+  Right goodClientTransport <- createTransport "127.0.0.1" "9999" ((,) "127.0.0.1") defaultTCPParameters+  Right badClientTransport <- createTransport "127.0.0.1" "10000" ((,) "127.0.0.1") defaultTCPParameters++  serverAddress <- newEmptyMVar+  testDone <- newEmptyMVar+  goodClientConnected <- newEmptyMVar+  goodClientDone <- newEmptyMVar+  badClientDone <- newEmptyMVar++  forkTry $ do+    Right serverEp <- newEndPoint serverTransport+    putMVar serverAddress (address serverEp)+    readMVar badClientDone+    ConnectionOpened _ _ _ <- receive serverEp+    Received _ _ <- receive serverEp+    -- Will lose the connection when the good client sends 9 bytes.+    ErrorEvent (TransportError (EventConnectionLost _) _) <- receive serverEp+    readMVar goodClientDone+    putMVar testDone ()++  forkTry $ do+    Right badClientEp <- newEndPoint badClientTransport+    address <- readMVar serverAddress+    -- Wait until the good client connects, then try to connect. It'll fail,+    -- but the good client should still be OK.+    readMVar goodClientConnected+    Left (TransportError ConnectFailed _)+      <- connect badClientEp address ReliableOrdered defaultConnectHints+    closeEndPoint badClientEp+    putMVar badClientDone ()++  forkTry $ do+    Right goodClientEp <- newEndPoint goodClientTransport+    address <- readMVar serverAddress+    Right conn <- connect goodClientEp address ReliableOrdered defaultConnectHints+    putMVar goodClientConnected ()+    -- Wait until the bad client has tried and failed to connect before+    -- attempting a send, to ensure that its failure did not affect us.+    readMVar badClientDone+    Right () <- send conn ["00000000"]+    -- The send which breaches the limit does not appear to fail, but the+    -- (heavyweight) connection is now severed. We can reliably determine that+    -- by receiving.+    Right () <- send conn ["000000000"]+    ErrorEvent (TransportError (EventConnectionLost _) _) <- receive goodClientEp+    closeEndPoint goodClientEp+    putMVar goodClientDone ()++  readMVar testDone+  closeTransport badClientTransport+  closeTransport goodClientTransport+  closeTransport serverTransport++-- | Ensure that an end point closes up OK even if the peer disobeys the+--   protocol.+testCloseEndPoint :: IO ()+testCloseEndPoint = do++  serverAddress <- newEmptyMVar+  serverFinished <- newEmptyMVar++  -- A server which accepts one connection and then attempts to close the+  -- end point.+  forkTry $ do+    Right transport <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters+    Right ep <- newEndPoint transport+    putMVar serverAddress (address ep)+    ConnectionOpened _ _ _ <- receive ep+    Just () <- timeout 5000000 (closeEndPoint ep)+    putMVar serverFinished ()+    return ()++  -- A nefarious client which connects to the server then stops responding.+  forkTry $ do+    Just (hostName, serviceName, endPointId) <- decodeEndPointAddress <$> readMVar serverAddress+    addr:_ <- N.getAddrInfo (Just N.defaultHints) (Just hostName) (Just serviceName)+    sock <- N.socket (N.addrFamily addr) N.Stream N.defaultProtocol+    N.connect sock (N.addrAddress addr)+    let endPointAddress = "127.0.0.1:0:0"+        -- Version 0x00000000 handshake data.+        v0handshake = [+            encodeWord32 endPointId+          , encodeWord32 (fromIntegral (BS.length endPointAddress))+          , endPointAddress+          ]+        -- Version, and total length of the versioned handshake.+        handshake = [+            encodeWord32 0x00000000+          , encodeWord32 (fromIntegral (BS.length (BS.concat v0handshake)))+          ]+    sendMany sock $+         handshake+      ++ v0handshake+      ++ [ -- Create a lightweight connection.+           encodeWord32 (encodeControlHeader CreatedNewConnection)+         , encodeWord32 1024+         ]+    readMVar serverFinished+    N.close sock++  readMVar serverFinished++-- | Ensure that if the peer's claimed host doesn't match its actual host,+--   the connection is rejected (when tcpCheckPeerHost is enabled).+testCheckPeerHost :: IO ()+testCheckPeerHost = do++  let params = defaultTCPParameters { tcpCheckPeerHost = True }+  Right transport1 <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") params+  -- This transport claims 127.0.0.2 as its host, but connections from it to+  -- an EndPoint on transport1 will show 127.0.0.1 as the socket's source host.+  Right transport2 <- createTransport "127.0.0.1" "0" ((,) "127.0.0.2") defaultTCPParameters++  Right ep1 <- newEndPoint transport1+  Right ep2 <- newEndPoint transport2++  Left err <- connect ep2 (address ep1) ReliableOrdered defaultConnectHints++  TransportError ConnectFailed "setupRemoteEndPoint: Host mismatch 127.0.0.1"+    <- return err++  return ()+ main :: IO () main = do   tcpResult <- tryIO $ runTests-           [ ("Use random port"       , testUseRandomPort)+           [ ("Use random port",        testUseRandomPort)            , ("EarlyDisconnect",        testEarlyDisconnect)            , ("EarlyCloseSocket",       testEarlyCloseSocket)            , ("IgnoreCloseSocket",      testIgnoreCloseSocket)@@ -802,10 +1050,13 @@            , ("Reconnect",              testReconnect)            , ("UnidirectionalError",    testUnidirectionalError)            , ("InvalidCloseConnection", testInvalidCloseConnection)+           , ("MaxLength",              testMaxLength)+           , ("CloseEndPoint",          testCloseEndPoint)+           , ("CheckPeerHost",          testCheckPeerHost)            ]   -- Run the generic tests even if the TCP specific tests failed..   testTransport (either (Left . show) (Right) <$>-    createTransport "127.0.0.1" "0" defaultTCPParameters)+    createTransport "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters)   -- ..but if the generic tests pass, still fail if the specific tests did not   case tcpResult of     Left err -> throwIO err