network-transport-tcp 0.6.1 → 0.7.0

raw patch · 5 files changed

+293/−156 lines, 5 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• ChangeLog +1/−4
• network-transport-tcp.cabal +1/−1
• src/Network/Transport/TCP.hs +192/−120
• src/Network/Transport/TCP/Internal.hs +15/−1
• tests/TestTCP.hs +84/−30

@@ -4,11 +4,8 @@  2019-12-31 FacundoDominguez <facundo.dominguez@tweag.io> 0.7.0 -* Added support for unaddressable endpoints. (#61)--2019-12-31 FacundoDominguez <facundo.dominguez@tweag.io> 0.6.1- * Relax dependency bounds to build with ghc-8.6.5+* Added support for unaddressable endpoints. (#61) * apiSend RELY violation is removed for closed remote endpoints (#70) * The server no longer needs crash if accept throws an exception. * Check peer-reported host against socket host (#54)

@@ -1,5 +1,5 @@ Name:          network-transport-tcp-Version:       0.6.1+Version:       0.7.0 Cabal-Version: >=1.10 Build-Type:    Simple License:       BSD3

@@ -19,6 +19,9 @@ module Network.Transport.TCP   ( -- * Main API     createTransport+  , TCPAddr(..)+  , defaultTCPAddr+  , TCPAddrInfo(..)   , TCPParameters(..)   , defaultTCPParameters     -- * Internals (exposed for unit tests)@@ -65,6 +68,7 @@   , encodeEndPointAddress   , decodeEndPointAddress   , currentProtocolVersion+  , randomEndPointAddress   ) import Network.Transport.Internal   ( prependLength@@ -150,7 +154,7 @@   ) import Data.IORef (IORef, newIORef, writeIORef, readIORef, writeIORef) import Data.ByteString (ByteString)-import qualified Data.ByteString as BS (concat)+import qualified Data.ByteString as BS (concat, length, null) import qualified Data.ByteString.Char8 as BSC (pack, unpack) import Data.Bits (shiftL, (.|.)) import Data.Maybe (isJust, isNothing, fromJust)@@ -172,7 +176,6 @@ import qualified Data.Accessor.Container as DAC (mapMaybe) import Data.Foldable (forM_, mapM_) import qualified System.Timeout (timeout)-import qualified Data.ByteString as BS (length)  -- $design --@@ -285,11 +288,21 @@ --   about the state (ValidTransportState, ValidLocalEndPointState, --   ValidRemoteEndPointState). -data TCPTransport = TCPTransport+-- | Information about the network addresses of a transport: the external+-- host/port as well as the bound host/port, which are not necessarily the+-- same.+data TransportAddrInfo = TransportAddrInfo   { transportHost     :: !N.HostName   , transportPort     :: !N.ServiceName   , transportBindHost :: !N.HostName   , transportBindPort :: !N.ServiceName+  }++data TCPTransport = TCPTransport+  { transportAddrInfo :: !(Maybe TransportAddrInfo)+    -- ^ This is 'Nothing' in case the transport is not addressable from the+    -- network: peers cannot connect to it unless it has a connection to the+    -- peer.   , transportState    :: !(MVar TransportState)   , transportParams   :: !TCPParameters   }@@ -299,16 +312,17 @@   | TransportClosed  data ValidTransportState = ValidTransportState-  { _localEndPoints :: !(Map EndPointAddress LocalEndPoint)+  { _localEndPoints :: !(Map EndPointId LocalEndPoint)   , _nextEndPointId :: !EndPointId   }  data LocalEndPoint = LocalEndPoint-  { localAddress :: !EndPointAddress-  , localState   :: !(MVar LocalEndPointState)+  { localAddress    :: !EndPointAddress+  , localEndPointId :: !EndPointId+  , localState      :: !(MVar LocalEndPointState)     -- | A 'QDisc' is held here rather than on the 'ValidLocalEndPointState'     --   because even closed 'LocalEndPoint's can have queued input data.-  , localQueue   :: !(QDisc Event)+  , localQueue      :: !(QDisc Event)   }  data LocalEndPointState =@@ -474,6 +488,28 @@ -- 'LightweightConnectionId'. type HeavyweightConnectionId = Word32 +-- | A transport which is addressable from the network must give a host/port+-- on which to bind/listen, and determine its external address (host/port) from+-- the actual port (which may not be known, in case 0 is used for the bind+-- port).+data TCPAddrInfo = TCPAddrInfo {+    tcpBindHost :: N.HostName+  , tcpBindPort :: N.ServiceName+  , tcpExternalAddress :: N.ServiceName -> (N.HostName, N.ServiceName)+  }++-- | Addressability of a transport. If your transport cannot be connected+-- to, for instance because it runs behind NAT, use Unaddressable.+data TCPAddr = Addressable TCPAddrInfo | Unaddressable++-- | The bind and external host/port are the same.+defaultTCPAddr :: N.HostName -> N.ServiceName -> TCPAddr+defaultTCPAddr host port = Addressable $ TCPAddrInfo {+    tcpBindHost = host+  , tcpBindPort = port+  , tcpExternalAddress = (,) host+  }+ -- | Parameters for setting up the TCP transport data TCPParameters = TCPParameters {     -- | Backlog for 'listen'.@@ -528,7 +564,7 @@ -- | Internal functionality we expose for unit testing data TransportInternals = TransportInternals   { -- | The ID of the thread that listens for new incoming connections-    transportThread     :: ThreadId+    transportThread     :: Maybe ThreadId     -- | A variant of newEndPoint in which the QDisc determined by the     -- transport's TCPParameters can be optionally overridden.   , newEndPointInternal :: (forall t . Maybe (QDisc t))@@ -544,74 +580,79 @@ --------------------------------------------------------------------------------  -- | Create a TCP transport-createTransport :: N.HostName    -- ^ Bind host name.-                -> N.ServiceName -- ^ Bind port.-                -> (N.ServiceName -> (N.HostName, N.ServiceName))-                   -- ^ External address host name and port, computed from the-                   --   actual bind port.-                -> TCPParameters-                -> IO (Either IOException Transport)-createTransport bindHost bindPort mkExternal params =-  either Left (Right . fst) <$>-    createTransportExposeInternals bindHost bindPort mkExternal params+createTransport+  :: TCPAddr+  -> TCPParameters+  -> IO (Either IOException Transport)+createTransport addr params =+  either Left (Right . fst) <$> createTransportExposeInternals addr params  -- | You should probably not use this function (used for unit testing only) createTransportExposeInternals-  :: N.HostName-  -> N.ServiceName-  -> (N.ServiceName -> (N.HostName, N.ServiceName))+  :: TCPAddr   -> TCPParameters   -> IO (Either IOException (Transport, TransportInternals))-createTransportExposeInternals bindHost bindPort mkExternal params = do+createTransportExposeInternals addr params = do     state <- newMVar . TransportValid $ ValidTransportState       { _localEndPoints = Map.empty       , _nextEndPointId = 0       }-    tryIO $ mdo-       when ( isJust (tcpUserTimeout params) &&-              not (N.isSupportedSocketOption N.UserTimeout)-            ) $-         throwIO $ userError $ "Network.Transport.TCP.createTransport: " ++-                               "the parameter tcpUserTimeout is unsupported " ++-                               "in this system."-       -- We don't know for sure the actual port 'forkServer' binded until it-       -- completes (see description of 'forkServer'), yet we need the port to-       -- construct a transport. So we tie a recursive knot.-       (port', result) <- do-         let (externalHost, externalPort) = mkExternal port'-         let transport = TCPTransport { transportState    = state-                                      , transportHost     = externalHost-                                      , transportPort     = externalPort-                                      , transportBindHost = bindHost-                                      , transportBindPort = port'-                                      , transportParams   = params-                                      }-         bracketOnError (forkServer-                             bindHost-                             bindPort-                             (tcpBacklog params)-                             (tcpReuseServerAddr params)-                             (errorHandler transport)-                             (terminationHandler transport)-                             (handleConnectionRequest transport))-                      (\(_port', tid) -> killThread tid)-                      (\(port'', tid) -> (port'',) <$> mkTransport transport tid)-       return result+    case addr of++      Unaddressable ->+        let transport = TCPTransport { transportState    = state+                                     , transportAddrInfo = Nothing+                                     , transportParams   = params+                                     }+        in  fmap Right (mkTransport transport Nothing)++      Addressable (TCPAddrInfo bindHost bindPort mkExternal) -> tryIO $ mdo+        when ( isJust (tcpUserTimeout params) &&+               not (N.isSupportedSocketOption N.UserTimeout)+             ) $+          throwIO $ userError $ "Network.Transport.TCP.createTransport: " +++                                "the parameter tcpUserTimeout is unsupported " +++                                "in this system."+        -- We don't know for sure the actual port 'forkServer' binded until it+        -- completes (see description of 'forkServer'), yet we need the port to+        -- construct a transport. So we tie a recursive knot.+        (port', result) <- do+          let (externalHost, externalPort) = mkExternal port'+          let addrInfo = TransportAddrInfo { transportHost     = externalHost+                                           , transportPort     = externalPort+                                           , transportBindHost = bindHost+                                           , transportBindPort = port'+                                           }+          let transport = TCPTransport { transportState    = state+                                       , transportAddrInfo = Just addrInfo+                                       , transportParams   = params+                                       }+          bracketOnError (forkServer+                              bindHost+                              bindPort+                              (tcpBacklog params)+                              (tcpReuseServerAddr params)+                              (errorHandler transport)+                              (terminationHandler transport)+                              (handleConnectionRequest transport))+                       (\(_port', tid) -> killThread tid)+                       (\(port'', tid) -> (port'',) <$> mkTransport transport (Just tid))+        return result   where     mkTransport :: TCPTransport-                -> ThreadId+                -> Maybe ThreadId                 -> IO (Transport, TransportInternals)-    mkTransport transport tid = do+    mkTransport transport mtid = do       return         ( Transport             { newEndPoint = do                 qdisc <- tcpNewQDisc params                 apiNewEndPoint transport qdisc             , closeTransport = let evs = [ EndPointClosed ]-                               in apiCloseTransport transport (Just tid) evs+                               in apiCloseTransport transport mtid evs             }         , TransportInternals-            { transportThread     = tid+            { transportThread     = mtid             , socketBetween       = internalSocketBetween transport             , newEndPointInternal = \mqdisc -> case mqdisc of                 Just qdisc -> apiNewEndPoint transport qdisc@@ -675,7 +716,7 @@     return EndPoint       { receive       = qdiscDequeue (localQueue ourEndPoint)       , address       = localAddress ourEndPoint-      , connect       = apiConnect (transportParams transport) ourEndPoint+      , connect       = apiConnect transport ourEndPoint       , closeEndPoint = let evs = [ EndPointClosed ]                         in  apiCloseEndPoint transport evs ourEndPoint       , newMulticastGroup     = return . Left $ newMulticastGroupError@@ -757,20 +798,20 @@     }  -- | Connnect to an endpoint-apiConnect :: TCPParameters    -- ^ Parameters+apiConnect :: TCPTransport            -> LocalEndPoint    -- ^ Local end point            -> EndPointAddress  -- ^ Remote address            -> Reliability      -- ^ Reliability (ignored)            -> ConnectHints     -- ^ Hints            -> IO (Either (TransportError ConnectErrorCode) Connection)-apiConnect params ourEndPoint theirAddress _reliability hints =+apiConnect transport ourEndPoint theirAddress _reliability hints =   try . asyncWhenCancelled close $     if localAddress ourEndPoint == theirAddress       then connectToSelf ourEndPoint       else do         resetIfBroken ourEndPoint theirAddress         (theirEndPoint, connId) <--          createConnectionTo params ourEndPoint theirAddress hints+          createConnectionTo transport ourEndPoint theirAddress hints         -- connAlive can be an IORef rather than an MVar because it is protected         -- by the remoteState MVar. We don't need the overhead of locking twice.         connAlive <- newIORef True@@ -778,6 +819,8 @@           { send  = apiSend  (ourEndPoint, theirEndPoint) connId connAlive           , close = apiClose (ourEndPoint, theirEndPoint) connId connAlive           }+  where+  params = transportParams transport  -- | Close a connection apiClose :: EndPointPair -> LightweightConnectionId -> IORef Bool -> IO ()@@ -983,37 +1026,45 @@       (numericHost, resolvedHost, actualPort) <-         resolveSockAddr sockAddr >>=           maybe (throwIO (userError "handleConnectionRequest: invalid socket address")) return--      (ourEndPointId, theirAddress) <- do+      -- The peer must send our identifier and their address promptly, if a+      -- timeout is set.+      (ourEndPointId, theirAddress, mTheirHost) <- do         ourEndPointId <- recvWord32 sock         let maxAddressLength = tcpMaxAddressLength $ transportParams transport-        theirAddress <- EndPointAddress . BS.concat <$>-          recvWithLength maxAddressLength sock-        return (ourEndPointId, theirAddress)-      let ourAddress = encodeEndPointAddress (transportHost transport)-                                             (transportPort transport)-                                             ourEndPointId-      (theirHost, _, _)-        <- maybe (throwIO (userError "handleConnectionRequest: peer gave malformed address"))-                 return-                 (decodeEndPointAddress theirAddress)+        mTheirAddress <- BS.concat <$> recvWithLength maxAddressLength sock+        -- Sending a length = 0 address means unaddressable.+        if BS.null mTheirAddress+        then do+          theirAddress <- randomEndPointAddress+          return (ourEndPointId, theirAddress, Nothing)+        else do+          let theirAddress = EndPointAddress mTheirAddress+          (theirHost, _, _)+            <- maybe (throwIO (userError "handleConnectionRequest: peer gave malformed address"))+                     return+                     (decodeEndPointAddress theirAddress)+          return (ourEndPointId, theirAddress, Just theirHost)       let checkPeerHost = tcpCheckPeerHost (transportParams transport)-      if checkPeerHost && (theirHost /= resolvedHost) && (theirHost /= numericHost)+      continue <- case (mTheirHost, checkPeerHost) of+        (Just theirHost, True) -> do+          -- If the OS-determined host doesn't match the host that the peer gave us,+          -- then we have no choice but to reject the connection. It's because we+          -- use the EndPointAddress to key the remote end points (localConnections)+          -- and we don't want to allow a peer to deny service to other peers by+          -- claiming to have their host and port.+          if theirHost == numericHost || theirHost == resolvedHost+          then return True+          else do+            sendMany sock $+                encodeWord32 (encodeConnectionRequestResponse ConnectionRequestHostMismatch)+              : (prependLength [BSC.pack theirHost] ++ prependLength [BSC.pack numericHost] ++ prependLength [BSC.pack resolvedHost])+            return False+        _ -> return True+      if continue       then do-        -- If the OS-determined host doesn't match the host that the peer gave us,-        -- then we have no choice but to reject the connection. It's because we-        -- use the EndPointAddress to key the remote end points (localConnections)-        -- and we don't want to allow a peer to deny service to other peers by-        -- claiming to have their host and port.-        sendMany sock $-            encodeWord32 (encodeConnectionRequestResponse ConnectionRequestHostMismatch)-          : (prependLength [BSC.pack theirHost] ++ prependLength [BSC.pack numericHost] ++ prependLength [BSC.pack resolvedHost])--        return Nothing-      else do         ourEndPoint <- withMVar (transportState transport) $ \st -> case st of           TransportValid vst ->-            case vst ^. localEndPointAt ourAddress of+            case vst ^. localEndPointAt ourEndPointId of               Nothing -> do                 sendMany sock [encodeWord32 (encodeConnectionRequestResponse ConnectionRequestInvalid)]                 throwIO $ userError "handleConnectionRequest: Invalid endpoint"@@ -1022,11 +1073,13 @@           TransportClosed ->             throwIO $ userError "Transport closed"         return (Just (go ourEndPoint theirAddress))+      else return Nothing        where        go :: LocalEndPoint -> EndPointAddress -> IO ()       go ourEndPoint theirAddress = handle handleException $ do+         resetIfBroken ourEndPoint theirAddress         (theirEndPoint, isNew) <-           findRemoteEndPoint ourEndPoint theirAddress RequestedByThem Nothing@@ -1425,12 +1478,13 @@ -- block until that is resolved. -- -- May throw a TransportError ConnectErrorCode exception.-createConnectionTo :: TCPParameters-                    -> LocalEndPoint-                    -> EndPointAddress-                    -> ConnectHints-                    -> IO (RemoteEndPoint, LightweightConnectionId)-createConnectionTo params ourEndPoint theirAddress hints = do+createConnectionTo+  :: TCPTransport+  -> LocalEndPoint+  -> EndPointAddress+  -> ConnectHints+  -> IO (RemoteEndPoint, LightweightConnectionId)+createConnectionTo transport ourEndPoint theirAddress hints = do     -- @timer@ is an IO action that completes when the timeout expires.     timer <- case connTimeout of               Just t -> do@@ -1442,6 +1496,7 @@    where +    params = transportParams transport     connTimeout = connectTimeout hints `mplus` transportConnectTimeout params      -- The second argument indicates the response obtained to the last@@ -1462,7 +1517,7 @@       if isNew         then do           mr' <- handle (absorbAllExceptions Nothing) $-            setupRemoteEndPoint params (ourEndPoint, theirEndPoint) connTimeout+            setupRemoteEndPoint transport (ourEndPoint, theirEndPoint) connTimeout           go timer (fmap ((,) theirEndPoint) mr')         else do           -- 'findRemoteEndPoint' will have increased 'remoteOutgoing'@@ -1502,10 +1557,14 @@       return a  -- | Set up a remote endpoint-setupRemoteEndPoint :: TCPParameters -> EndPointPair -> Maybe Int-                    -> IO (Maybe ConnectionRequestResponse)-setupRemoteEndPoint params (ourEndPoint, theirEndPoint) connTimeout = do-    result <- socketToEndPoint ourAddress+setupRemoteEndPoint+  :: TCPTransport+  -> EndPointPair+  -> Maybe Int+  -> IO (Maybe ConnectionRequestResponse)+setupRemoteEndPoint transport (ourEndPoint, theirEndPoint) connTimeout = do+    let mOurAddress = const ourAddress <$> transportAddrInfo transport+    result <- socketToEndPoint mOurAddress                                theirAddress                                (tcpReuseClientAddr params)                                (tcpNoDelay params)@@ -1587,6 +1646,7 @@       (tryCloseSocket sock `finally` putMVar socketClosed ())     return $ either (const Nothing) (Just . (\(_,_,x) -> x)) result   where+    params          = transportParams transport     ourAddress      = localAddress ourEndPoint     theirAddress    = remoteAddress theirEndPoint     invalidAddress  = TransportError ConnectNotFound@@ -1736,15 +1796,19 @@     modifyMVar (transportState transport) $ \st -> case st of       TransportValid vst -> do         let ix   = vst ^. nextEndPointId-        let addr = encodeEndPointAddress (transportHost transport)-                                         (transportPort transport)-                                         ix-        let localEndPoint = LocalEndPoint { localAddress = addr-                                          , localQueue   = qdisc-                                          , localState   = state+        addr <- case transportAddrInfo transport of+          Nothing -> randomEndPointAddress+          Just addrInfo -> return $+            encodeEndPointAddress (transportHost addrInfo)+                                  (transportPort addrInfo)+                                  ix+        let localEndPoint = LocalEndPoint { localAddress    = addr+                                          , localEndPointId = ix+                                          , localQueue      = qdisc+                                          , localState      = state                                           }         return ( TransportValid-               . (localEndPointAt addr ^= Just localEndPoint)+               . (localEndPointAt ix ^= Just localEndPoint)                . (nextEndPointId ^= ix + 1)                $ vst                , localEndPoint@@ -1785,7 +1849,7 @@   modifyMVar_ (transportState transport) $ \st -> case st of     TransportValid vst ->       return ( TransportValid-             . (localEndPointAt (localAddress ourEndPoint) ^= Nothing)+             . (localEndPointAt (localEndPointId ourEndPoint) ^= Nothing)              $ vst              )     TransportClosed ->@@ -1994,16 +2058,16 @@ -- responsible for eventually closing the socket and filling the MVar (which -- is empty). The MVar must be filled immediately after, and never before, -- the socket is closed.-socketToEndPoint :: EndPointAddress -- ^ Our address-                 -> EndPointAddress -- ^ Their address-                 -> Bool            -- ^ Use SO_REUSEADDR?-                 -> Bool            -- ^ Use TCP_NODELAY-                 -> Bool            -- ^ Use TCP_KEEPALIVE-                 -> Maybe Int       -- ^ Maybe TCP_USER_TIMEOUT-                 -> Maybe Int       -- ^ Timeout for connect+socketToEndPoint :: Maybe EndPointAddress -- ^ Our address+                 -> EndPointAddress       -- ^ Their address+                 -> Bool                  -- ^ Use SO_REUSEADDR?+                 -> Bool                  -- ^ Use TCP_NODELAY+                 -> Bool                  -- ^ Use TCP_KEEPALIVE+                 -> Maybe Int             -- ^ Maybe TCP_USER_TIMEOUT+                 -> Maybe Int             -- ^ Timeout for connect                  -> IO (Either (TransportError ConnectErrorCode)                                (MVar (), N.Socket, ConnectionRequestResponse))-socketToEndPoint (EndPointAddress ourAddress) theirAddress reuseAddr noDelay keepAlive+socketToEndPoint mOurAddress theirAddress reuseAddr noDelay keepAlive                  mUserTimeout timeout =   try $ do     (host, port, theirEndPointId) <- case decodeEndPointAddress theirAddress of@@ -2024,11 +2088,15 @@         mapIOException invalidAddress $           N.connect sock (N.addrAddress addr)         mapIOException failed $ do-          sendMany sock $-              -- The version.-              encodeWord32 currentProtocolVersion-              -- The V0 handshake data with the length prepended.-            : prependLength (encodeWord32 theirEndPointId : prependLength [ourAddress])+          case mOurAddress of+            Just (EndPointAddress ourAddress) ->+              sendMany sock $+                  encodeWord32 currentProtocolVersion+                : prependLength (encodeWord32 theirEndPointId : prependLength [ourAddress])+            Nothing ->+              sendMany sock $+                  encodeWord32 currentProtocolVersion+                : prependLength ([encodeWord32 theirEndPointId, encodeWord32 0])           recvWord32 sock       case decodeConnectionRequestResponse response of         Nothing -> throwIO (failed . userError $ "Unexpected response")@@ -2064,11 +2132,14 @@                       -> EndPointAddress -- ^ Remote endpoint                       -> IO N.Socket internalSocketBetween transport ourAddress theirAddress = do+  ourEndPointId <- case decodeEndPointAddress ourAddress of+    Just (_, _, eid) -> return eid+    _ -> throwIO $ userError "Malformed local EndPointAddress"   ourEndPoint <- withMVar (transportState transport) $ \st -> case st of       TransportClosed ->         throwIO $ userError "Transport closed"       TransportValid vst ->-        case vst ^. localEndPointAt ourAddress of+        case vst ^. localEndPointAt ourEndPointId of           Nothing -> throwIO $ userError "Local endpoint not found"           Just ep -> return ep   theirEndPoint <- withMVar (localState ourEndPoint) $ \st -> case st of@@ -2091,6 +2162,7 @@       throwIO err     RemoteEndPointFailed err ->       throwIO err+  where  -------------------------------------------------------------------------------- -- Constants                                                                  --@@ -2112,7 +2184,7 @@ -- Accessor definitions                                                       -- -------------------------------------------------------------------------------- -localEndPoints :: Accessor ValidTransportState (Map EndPointAddress LocalEndPoint)+localEndPoints :: Accessor ValidTransportState (Map EndPointId LocalEndPoint) localEndPoints = accessor _localEndPoints (\es st -> st { _localEndPoints = es })  nextEndPointId :: Accessor ValidTransportState EndPointId@@ -2139,7 +2211,7 @@ remoteNextConnOutId :: Accessor ValidRemoteEndPointState LightweightConnectionId remoteNextConnOutId = accessor _remoteNextConnOutId (\cix st -> st { _remoteNextConnOutId = cix }) -localEndPointAt :: EndPointAddress -> Accessor ValidTransportState (Maybe LocalEndPoint)+localEndPointAt :: EndPointId -> Accessor ValidTransportState (Maybe LocalEndPoint) localEndPointAt addr = localEndPoints >>> DAC.mapMaybe addr  localConnectionTo :: EndPointAddress -> Accessor ValidLocalEndPointState (Maybe RemoteEndPoint)

@@ -17,6 +17,7 @@   , EndPointId   , encodeEndPointAddress   , decodeEndPointAddress+  , randomEndPointAddress   , ProtocolVersion   , currentProtocolVersion   ) where@@ -70,7 +71,7 @@ import qualified Network.Socket.ByteString as NBS (recv) #endif -import Data.Word (Word32)+import Data.Word (Word32, Word64)  import Control.Monad (forever, when) import Control.Exception (SomeException, catch, bracketOnError, throwIO, mask_)@@ -98,7 +99,12 @@ import Data.ByteString (ByteString) import qualified Data.ByteString as BS (length, concat, null) import Data.ByteString.Lazy.Internal (smallChunkSize)+import Data.ByteString.Lazy (toStrict) import qualified Data.ByteString.Char8 as BSC (unpack, pack)+import Data.ByteString.Lazy.Builder (word64BE, toLazyByteString)+import Data.Monoid ((<>))+import qualified Data.UUID as UUID+import qualified Data.UUID.V4 as UUID  -- | Local identifier for an endpoint within this transport type EndPointId = Word32@@ -176,6 +182,14 @@   ConnectionRequestInvalid            -> 0x00000001   ConnectionRequestCrossed            -> 0x00000002   ConnectionRequestHostMismatch       -> 0x00000003++-- | Generate an EndPointAddress which does not encode a host/port/endpointid.+-- Such addresses are used for unreachable endpoints, and for ephemeral+-- addresses when such endpoints establish new heavyweight connections.+randomEndPointAddress :: IO EndPointAddress+randomEndPointAddress = do+  uuid <- UUID.nextRandom+  return $ EndPointAddress (UUID.toASCIIBytes uuid)  -- | Start a server at the specified address. --

@@ -19,6 +19,9 @@                              , TCPParameters(..)                              , defaultTCPParameters                              , LightweightConnectionId+                             , TCPAddrInfo(..)+                             , TCPAddr(..)+                             , defaultTCPAddr                              ) import Control.Concurrent (threadDelay, killThread) import Control.Concurrent.MVar ( MVar@@ -125,7 +128,7 @@     server :: MVar EndPointAddress -> MVar EndPointAddress -> MVar () -> IO ()     server serverAddr clientAddr serverDone = do       tlog "Server"-      Right transport <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters+      Right transport <- createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters       Right endpoint  <- newEndPoint transport       putMVar serverAddr (address endpoint)       theirAddr <- readMVar clientAddr@@ -202,7 +205,7 @@       putMVar clientAddr ourAddress        -- Connect to the server-      Right (_, sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr True False False Nothing Nothing+      Right (_, sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= \addr -> socketToEndPoint (Just ourAddress) addr True False False Nothing Nothing        -- Open a new connection       sendMany sock [@@ -230,7 +233,7 @@     server :: MVar EndPointAddress -> MVar EndPointAddress -> MVar () -> IO ()     server serverAddr clientAddr serverDone = do       tlog "Server"-      Right transport <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters+      Right transport <- createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters       Right endpoint  <- newEndPoint transport       putMVar serverAddr (address endpoint)       theirAddr <- readMVar clientAddr@@ -319,7 +322,7 @@       putMVar clientAddr ourAddress        -- Connect to the server-      Right (_, sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr True False False Nothing Nothing+      Right (_, sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= \addr -> socketToEndPoint (Just ourAddress) addr True False False Nothing Nothing        -- Open a new connection       sendMany sock [@@ -338,13 +341,13 @@ -- | Test the creation of a transport with an invalid address testInvalidAddress :: IO () testInvalidAddress = do-  Left _ <- createTransport "invalidHostName" "0" ((,) "invalidHostName") defaultTCPParameters+  Left _ <- createTransport (defaultTCPAddr "invalidHostName" "0") defaultTCPParameters   return ()  -- | Test connecting to invalid or non-existing endpoints testInvalidConnect :: IO () testInvalidConnect = do-  Right transport <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters+  Right transport <- createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters   Right endpoint  <- newEndPoint transport    -- Syntax error in the endpoint address@@ -375,7 +378,7 @@   clientDone <- newEmptyMVar   serverDone <- newEmptyMVar   connectionEstablished <- newEmptyMVar-  Right transport <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters+  Right transport <- createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters    -- Server   forkTry $ do@@ -413,7 +416,7 @@     theirAddress <- readMVar serverAddr      -- Connect to the server-    Right (_, sock, ConnectionRequestAccepted) <- socketToEndPoint ourAddress theirAddress True False False Nothing Nothing+    Right (_, sock, ConnectionRequestAccepted) <- socketToEndPoint (Just ourAddress) theirAddress True False False Nothing Nothing     putMVar connectionEstablished ()      -- Server connects to us, and then closes the connection@@ -465,7 +468,7 @@   clientDone <- newEmptyMVar   serverDone <- newEmptyMVar   connectionEstablished <- newEmptyMVar-  Right transport <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters+  Right transport <- createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters    -- Server   forkTry $ do@@ -499,7 +502,7 @@     theirAddress <- readMVar serverAddr      -- Connect to the server-    Right (_, sock, ConnectionRequestAccepted) <- socketToEndPoint ourAddress theirAddress True False False Nothing Nothing+    Right (_, sock, ConnectionRequestAccepted) <- socketToEndPoint (Just ourAddress) theirAddress True False False Nothing Nothing     putMVar connectionEstablished ()      -- Server connects to us, and then closes the connection@@ -545,7 +548,7 @@   serverAddr <- newEmptyMVar    forkTry $ do-    Right transport <- createTransport "127.0.0.1" "0" ((,) "128.0.0.1") defaultTCPParameters+    Right transport <- createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters     Right endpoint <- newEndPoint transport     -- Since we're lying about the server's address, we have to manually     -- construct the proper address. If we used its actual address, the clients@@ -565,7 +568,7 @@       forkTry $ do         -- It is possible that the remote endpoint just rejects the request by closing the socket         -- immediately (depending on far the remote endpoint got with the initialization)-        response <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr True False False Nothing Nothing+        response <- readMVar serverAddr >>= \addr -> socketToEndPoint (Just ourAddress) addr True False False Nothing Nothing         case response of           Right (_, _, ConnectionRequestAccepted) ->             -- We don't close this socket because we want to keep this connection open@@ -589,11 +592,11 @@ -- | Test that we can create "many" transport instances testMany :: IO () testMany = do-  Right masterTransport <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters+  Right masterTransport <- createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters   Right masterEndPoint  <- newEndPoint masterTransport    replicateM_ 10 $ do-    mTransport <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters+    mTransport <- createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters     case mTransport of       Left ex -> do         putStrLn $ "IOException: " ++ show ex ++ "; errno = " ++ show (ioe_errno ex)@@ -610,10 +613,11 @@ -- | Test what happens when the transport breaks completely testBreakTransport :: IO () testBreakTransport = do-  Right (transport, internals) <- createTransportExposeInternals "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters+  Right (transport, internals) <- createTransportExposeInternals (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters   Right endpoint <- newEndPoint transport -  killThread (transportThread internals) -- Uh oh+  let Just tid = transportThread internals+  killThread tid -- Uh oh    ErrorEvent (TransportError EventTransportFailed _) <- receive endpoint @@ -686,7 +690,7 @@    -- Client   forkTry $ do-    Right transport <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters+    Right transport <- createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters     Right endpoint  <- newEndPoint transport     let theirAddr = encodeEndPointAddress "127.0.0.1" serverPort 0 @@ -790,7 +794,7 @@    -- Client   forkTry $ do-    Right (transport, internals) <- createTransportExposeInternals "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters+    Right (transport, internals) <- createTransportExposeInternals (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters     Right endpoint <- newEndPoint transport     let theirAddr = encodeEndPointAddress "127.0.0.1" serverPort 0 @@ -845,7 +849,7 @@  testInvalidCloseConnection :: IO () testInvalidCloseConnection = do-  Right (transport, internals) <- createTransportExposeInternals "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters+  Right (transport, internals) <- createTransportExposeInternals (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters   serverAddr <- newEmptyMVar   clientDone <- newEmptyMVar   serverDone <- newEmptyMVar@@ -887,10 +891,10 @@ testUseRandomPort = do    testDone <- newEmptyMVar    forkTry $ do-     Right transport1 <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters+     Right transport1 <- createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters      Right ep1        <- newEndPoint transport1      -- Same as transport1, but is strict in the port.-     Right transport2 <- createTransport "127.0.0.1" "0" (\(!port) -> ("127.0.0.1", port)) defaultTCPParameters+     Right transport2 <- createTransport (Addressable (TCPAddrInfo "127.0.0.1" "0" (\(!port) -> ("127.0.0.1", port)))) defaultTCPParameters      Right ep2        <- newEndPoint transport2      Right conn1 <- connect ep2 (address ep1) ReliableOrdered defaultConnectHints      ConnectionOpened _ _ _ <- receive ep1@@ -903,7 +907,7 @@ testMaxLength :: IO () testMaxLength = do -  Right serverTransport <- createTransport "127.0.0.1" "9998" ((,) "127.0.0.1") $ defaultTCPParameters {+  Right serverTransport <- createTransport (defaultTCPAddr "127.0.0.1" "9998") $ defaultTCPParameters {       -- 17 bytes should fit every valid address at 127.0.0.1.       -- Port is at most 5 bytes (65536) and id is a base-10 Word32 so       -- at most 10 bytes. We'll have one client with a 5-byte port to push it@@ -911,8 +915,8 @@       tcpMaxAddressLength = 16     , tcpMaxReceiveLength = 8     }-  Right goodClientTransport <- createTransport "127.0.0.1" "9999" ((,) "127.0.0.1") defaultTCPParameters-  Right badClientTransport <- createTransport "127.0.0.1" "10000" ((,) "127.0.0.1") defaultTCPParameters+  Right goodClientTransport <- createTransport (defaultTCPAddr "127.0.0.1" "9999") defaultTCPParameters+  Right badClientTransport <- createTransport (defaultTCPAddr "127.0.0.1" "10000") defaultTCPParameters    serverAddress <- newEmptyMVar   testDone <- newEmptyMVar@@ -975,7 +979,7 @@   -- A server which accepts one connection and then attempts to close the   -- end point.   forkTry $ do-    Right transport <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters+    Right transport <- createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters     Right ep <- newEndPoint transport     putMVar serverAddress (address ep)     ConnectionOpened _ _ _ <- receive ep@@ -1019,10 +1023,10 @@ testCheckPeerHostReject = do    let params = defaultTCPParameters { tcpCheckPeerHost = True }-  Right transport1 <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") params+  Right transport1 <- createTransport (defaultTCPAddr "127.0.0.1" "0") params   -- This transport claims 127.0.0.2 as its host, but connections from it to   -- an EndPoint on transport1 will show 127.0.0.1 as the socket's source host.-  Right transport2 <- createTransport "127.0.0.1" "0" ((,) "127.0.0.2") defaultTCPParameters+  Right transport2 <- createTransport (Addressable (TCPAddrInfo "127.0.0.1" "0" ((,) "127.0.0.2"))) defaultTCPParameters    Right ep1 <- newEndPoint transport1   Right ep2 <- newEndPoint transport2@@ -1040,9 +1044,9 @@ testCheckPeerHostResolve = do    let params = defaultTCPParameters { tcpCheckPeerHost = True }-  Right transport1 <- createTransport "127.0.0.1" "0" ((,) "127.0.0.1") params+  Right transport1 <- createTransport (defaultTCPAddr "127.0.0.1" "0") params   -- EndPoints on this transport have addresses with "localhost" host part.-  Right transport2 <- createTransport "127.0.0.1" "0" ((,) "localhost") defaultTCPParameters+  Right transport2 <- createTransport (Addressable (TCPAddrInfo "127.0.0.1" "0" ((,) "localhost"))) defaultTCPParameters    Right ep1 <- newEndPoint transport1   Right ep2 <- newEndPoint transport2@@ -1053,6 +1057,54 @@    return () +-- | Test that an unreachable EndPoint can use its own address to connect+-- to itself.+testUnreachableSelfConnect :: IO ()+testUnreachableSelfConnect = do+  Right transport <- createTransport Unaddressable defaultTCPParameters+  Right ep <- newEndPoint transport+  Right conn <- connect ep (address ep) ReliableOrdered defaultConnectHints+  ConnectionOpened connid ReliableOrdered _ <- receive ep+  Right () <- send conn ["ping"]+  Received connid' bytes <- receive ep+  _ <- close conn+  ConnectionClosed connid'' <- receive ep+  closeEndPoint ep+  closeTransport transport++-- | Test that+--+-- 1. Connecting to an unreachable EndPoint's address gives ConnectFailed+-- 2. An unreachable EndPoint can successfully connect to a reachable EndPoint+-- 3. The address given in the ConnectionOpened event at the reachable EndPoint+--    can be used to connect to the unreachable EndPoint, so long as there is+--    at least one lightweight connection open between the two.+testUnreachableConnect :: IO ()+testUnreachableConnect = do+  Right rtransport <- createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters+  Right utransport <- createTransport Unaddressable defaultTCPParameters+  Right rep <- newEndPoint rtransport+  Right uep <- newEndPoint utransport+  -- Reachable endpoint connects to the unreachable endpoint, but it fails.+  -- NB ConnectNotFound isn't the error; that would mean the address makes+  -- sense but the host could not be found.+  Left (TransportError ConnectFailed _) <- connect rep (address uep) ReliableOrdered defaultConnectHints+  -- Unreachable endpoint connects to the reachable endpoint.+  Right conn <- connect uep (address rep) ReliableOrdered defaultConnectHints+  -- Reachable endpoint now has an address at which it can connect to the+  -- unreachable+  ConnectionOpened _ _ addr <- receive rep+  Right conn' <- connect rep addr ReliableOrdered defaultConnectHints+  ConnectionOpened _ _ addr' <- receive uep+  close conn+  ConnectionClosed _ <- receive rep+  close conn'+  ConnectionClosed _ <- receive uep+  closeEndPoint rep+  closeEndPoint uep+  closeTransport rtransport+  closeTransport utransport+ main :: IO () main = do   tcpResult <- tryIO $ runTests@@ -1073,10 +1125,12 @@            , ("CloseEndPoint",          testCloseEndPoint)            , ("CheckPeerHostReject",    testCheckPeerHostReject)            , ("CheckPeerHostResolve",   testCheckPeerHostResolve)+           , ("UnreachableSelfConnect", testUnreachableSelfConnect)+           , ("UnreachableConnect",     testUnreachableConnect)            ]   -- Run the generic tests even if the TCP specific tests failed..   testTransport (either (Left . show) (Right) <$>-    createTransport "127.0.0.1" "0" ((,) "127.0.0.1") defaultTCPParameters)+    createTransport (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters)   -- ..but if the generic tests pass, still fail if the specific tests did not   case tcpResult of     Left err -> throwIO err