diff --git a/network-transport-tcp.cabal b/network-transport-tcp.cabal --- a/network-transport-tcp.cabal +++ b/network-transport-tcp.cabal @@ -1,5 +1,5 @@ Name: network-transport-tcp -Version: 0.2.0.3 +Version: 0.3.0 Cabal-Version: >=1.8 Build-Type: Simple License: BSD3 @@ -10,8 +10,8 @@ Stability: experimental Homepage: http://github.com/haskell-distributed/distributed-process Bug-Reports: mailto:edsko@well-typed.com -Synopsis: TCP instantation of Network.Transport -Description: TCP instantation of Network.Transport +Synopsis: TCP instantiation of Network.Transport +Description: TCP instantiation of Network.Transport Tested-With: GHC==7.0.4 GHC==7.2.2 GHC==7.4.1 GHC==7.4.2 Category: Network @@ -22,11 +22,11 @@ Library Build-Depends: base >= 4.3 && < 5, - network-transport >= 0.2 && < 0.3, + network-transport >= 0.3 && < 0.4, data-accessor >= 0.2 && < 0.3, containers >= 0.4 && < 0.6, bytestring >= 0.9 && < 0.11, - network >= 2.3 && < 2.4 + network >= 2.3 && < 2.5 Exposed-modules: Network.Transport.TCP, Network.Transport.TCP.Internal Extensions: CPP @@ -37,23 +37,11 @@ Type: exitcode-stdio-1.0 Main-Is: TestTCP.hs Build-Depends: base >= 4.3 && < 5, - network-transport >= 0.2 && < 0.3, - data-accessor >= 0.2 && < 0.3, - containers >= 0.4 && < 0.6, - bytestring >= 0.9 && < 0.11, - network >= 2.3 && < 2.4, - random >= 1.0 && < 1.1, - ansi-terminal >= 0.5 && < 0.6, - mtl >= 2.0 && < 2.2 - Other-modules: TestAuxiliary, - TestTransport, - Traced - ghc-options: -Wall -fno-warn-unused-do-bind -threaded -rtsopts -with-rtsopts=-N - Extensions: ExistentialQuantification, - FlexibleInstances, - DeriveDataTypeable, - RankNTypes, - OverlappingInstances, - OverloadedStrings, - CPP - HS-Source-Dirs: tests src + network-transport-tests >= 0.1 && < 0.2, + network >= 2.3 && < 2.5, + network-transport >= 0.3 && < 0.4, + network-transport-tcp >= 0.3 && < 0.4 + ghc-options: -threaded -rtsopts -with-rtsopts=-N + HS-Source-Dirs: tests + Extensions: CPP, + OverloadedStrings diff --git a/src/Network/Transport/TCP.hs b/src/Network/Transport/TCP.hs --- a/src/Network/Transport/TCP.hs +++ b/src/Network/Transport/TCP.hs @@ -10,23 +10,26 @@ -- Applications that use the TCP transport should use -- 'Network.Socket.withSocketsDo' in their main function for Windows -- compatibility (see "Network.Socket"). -module Network.Transport.TCP ( -- * Main API - createTransport - , TCPParameters(..) - , defaultTCPParameters - -- * Internals (exposed for unit tests) - , createTransportExposeInternals - , TransportInternals(..) - , EndPointId - , encodeEndPointAddress - , decodeEndPointAddress - , ControlHeader(..) - , ConnectionRequestResponse(..) - , firstNonReservedConnectionId - , socketToEndPoint - -- * Design notes - -- $design - ) where +module Network.Transport.TCP + ( -- * Main API + createTransport + , TCPParameters(..) + , defaultTCPParameters + -- * Internals (exposed for unit tests) + , createTransportExposeInternals + , TransportInternals(..) + , EndPointId + , encodeEndPointAddress + , decodeEndPointAddress + , ControlHeader(..) + , ConnectionRequestResponse(..) + , firstNonReservedLightweightConnectionId + , firstNonReservedHeavyweightConnectionId + , socketToEndPoint + , LightweightConnectionId + -- * Design notes + -- $design + ) where import Prelude hiding ( mapM_ @@ -36,84 +39,86 @@ ) import Network.Transport -import Network.Transport.TCP.Internal ( forkServer - , recvWithLength - , recvInt32 - , tryCloseSocket - ) -import Network.Transport.Internal ( encodeInt32 - , decodeInt32 - , prependLength - , mapIOException - , tryIO - , tryToEnum - , void - , timeoutMaybe - , asyncWhenCancelled - ) -import qualified Network.Socket as N ( HostName - , ServiceName - , Socket - , getAddrInfo - , socket - , addrFamily - , addrAddress - , SocketType(Stream) - , defaultProtocol - , setSocketOption - , SocketOption(ReuseAddr) - , connect - , sOMAXCONN - , AddrInfo - ) +import Network.Transport.TCP.Internal + ( forkServer + , recvWithLength + , recvInt32 + , tryCloseSocket + ) +import Network.Transport.Internal + ( encodeInt32 + , prependLength + , mapIOException + , tryIO + , tryToEnum + , void + , timeoutMaybe + , asyncWhenCancelled + ) +import qualified Network.Socket as N + ( HostName + , ServiceName + , Socket + , getAddrInfo + , socket + , addrFamily + , addrAddress + , SocketType(Stream) + , defaultProtocol + , setSocketOption + , SocketOption(ReuseAddr) + , connect + , sOMAXCONN + , AddrInfo + ) import Network.Socket.ByteString (sendMany) import Control.Concurrent (forkIO, ThreadId, killThread, myThreadId) import Control.Concurrent.Chan (Chan, newChan, readChan, writeChan) -import Control.Concurrent.MVar ( MVar - , newMVar - , modifyMVar - , modifyMVar_ - , readMVar - , takeMVar - , putMVar - , newEmptyMVar - , withMVar - ) +import Control.Concurrent.MVar + ( MVar + , newMVar + , modifyMVar + , modifyMVar_ + , readMVar + , putMVar + , newEmptyMVar + , withMVar + ) import Control.Category ((>>>)) import Control.Applicative ((<$>)) -import Control.Monad (when, unless) -import Control.Exception ( IOException - , SomeException - , AsyncException - , handle - , throw - , throwIO - , try - , bracketOnError - , mask - , onException - , fromException - ) +import Control.Monad (when, unless, join) +import Control.Exception + ( IOException + , SomeException + , AsyncException + , handle + , throw + , throwIO + , try + , bracketOnError + , fromException + , catch + ) import Data.IORef (IORef, newIORef, writeIORef, readIORef) import Data.ByteString (ByteString) import qualified Data.ByteString as BS (concat) import qualified Data.ByteString.Char8 as BSC (pack, unpack) -import Data.Int (Int32) -import Data.IntMap (IntMap) -import qualified Data.IntMap as IntMap (empty) -import Data.IntSet (IntSet) -import qualified Data.IntSet as IntSet ( empty - , insert - , elems - , singleton - , null - , delete - , member - ) +import Data.Bits (shiftL, (.|.)) +import Data.Word (Word32) +import Data.Set (Set) +import qualified Data.Set as Set + ( empty + , insert + , elems + , singleton + , null + , delete + , member + ) import Data.Map (Map) import qualified Data.Map as Map (empty) import Data.Accessor (Accessor, accessor, (^.), (^=), (^:)) -import qualified Data.Accessor.Container as DAC (mapMaybe, intMapMaybe) +import qualified Data.Accessor.Container as DAC (mapMaybe) import Data.Foldable (forM_, mapM_) -- $design @@ -224,35 +229,35 @@ -- ValidRemoteEndPointState). data TCPTransport = TCPTransport - { transportHost :: N.HostName - , transportPort :: N.ServiceName - , transportState :: MVar TransportState - , transportParams :: TCPParameters + { transportHost :: !N.HostName + , transportPort :: !N.ServiceName + , transportState :: !(MVar TransportState) + , transportParams :: !TCPParameters } data TransportState = - TransportValid ValidTransportState + TransportValid !ValidTransportState | TransportClosed data ValidTransportState = ValidTransportState - { _localEndPoints :: Map EndPointAddress LocalEndPoint - , _nextEndPointId :: EndPointId + { _localEndPoints :: !(Map EndPointAddress LocalEndPoint) + , _nextEndPointId :: !EndPointId } data LocalEndPoint = LocalEndPoint - { localAddress :: EndPointAddress - , localChannel :: Chan Event - , localState :: MVar LocalEndPointState + { localAddress :: !EndPointAddress + , localChannel :: !(Chan Event) + , localState :: !(MVar LocalEndPointState) } data LocalEndPointState = - LocalEndPointValid ValidLocalEndPointState + LocalEndPointValid !ValidLocalEndPointState | LocalEndPointClosed data ValidLocalEndPointState = ValidLocalEndPointState - { _nextConnectionId :: !ConnectionId - , _localConnections :: Map EndPointAddress RemoteEndPoint - , _nextRemoteId :: !Int + { _nextConnOutId :: !LightweightConnectionId + , _nextConnInId :: !HeavyweightConnectionId + , _localConnections :: !(Map EndPointAddress RemoteEndPoint) } -- REMOTE ENDPOINTS @@ -272,14 +277,14 @@ -- \-------+----------+--------> Failed -- -- Init: There are two places where we create new remote endpoints: in --- requestConnectionTo (in response to an API 'connect' call) and in +-- createConnectionTo (in response to an API 'connect' call) and in -- handleConnectionRequest (when a remote node tries to connect to us). -- 'Init' carries an MVar () 'resolved' which concurrent threads can use to -- wait for the remote endpoint to finish initialization. We record who -- requested the connection (the local endpoint or the remote endpoint). -- -- Invalid: We put the remote endpoint in invalid state only during --- requestConnectionTo when we fail to connect. +-- createConnectionTo when we fail to connect. -- -- Valid: This is the "normal" state for a working remote endpoint. -- @@ -338,9 +343,10 @@ -- modifyRemoteState. data RemoteEndPoint = RemoteEndPoint - { remoteAddress :: EndPointAddress - , remoteState :: MVar RemoteState - , remoteId :: Int + { remoteAddress :: !EndPointAddress + , remoteState :: !(MVar RemoteState) + , remoteId :: !HeavyweightConnectionId + , remoteScheduled :: !(Chan (IO ())) } data RequestedBy = RequestedByUs | RequestedByThem @@ -348,48 +354,56 @@ data RemoteState = -- | Invalid remote endpoint (for example, invalid address) - RemoteEndPointInvalid (TransportError ConnectErrorCode) + RemoteEndPointInvalid !(TransportError ConnectErrorCode) -- | The remote endpoint is being initialized - | RemoteEndPointInit (MVar ()) RequestedBy + | RemoteEndPointInit !(MVar ()) !(MVar ()) !RequestedBy -- | "Normal" working endpoint - | RemoteEndPointValid ValidRemoteEndPointState + | RemoteEndPointValid !ValidRemoteEndPointState -- | The remote endpoint is being closed (garbage collected) - | RemoteEndPointClosing (MVar ()) ValidRemoteEndPointState + | RemoteEndPointClosing !(MVar ()) !ValidRemoteEndPointState -- | The remote endpoint has been closed (garbage collected) | RemoteEndPointClosed -- | The remote endpoint has failed, or has been forcefully shutdown -- using a closeTransport or closeEndPoint API call - | RemoteEndPointFailed IOException + | RemoteEndPointFailed !IOException +-- TODO: we might want to replace Set (here and elsewhere) by faster +-- containers +-- +-- TODO: we could get rid of 'remoteIncoming' (and maintain less state) if +-- we introduce a new event 'AllConnectionsClosed' data ValidRemoteEndPointState = ValidRemoteEndPointState { _remoteOutgoing :: !Int - , _remoteIncoming :: IntSet - , remoteSocket :: N.Socket - , sendOn :: [ByteString] -> IO () - , _pendingCtrlRequests :: IntMap (MVar (Either IOException [ByteString])) - , _nextCtrlRequestId :: !ControlRequestId + , _remoteIncoming :: !(Set LightweightConnectionId) + , _remoteMaxIncoming :: !LightweightConnectionId + , remoteSocket :: !N.Socket + , remoteSendLock :: !(MVar ()) } -- | Local identifier for an endpoint within this transport -type EndPointId = Int32 - --- | Control request ID --- --- Control requests are asynchronous; the request ID makes it possible to match --- requests and replies -type ControlRequestId = Int32 +type EndPointId = Word32 -- | Pair of local and a remote endpoint (for conciseness in signatures) -type EndPointPair = (LocalEndPoint, RemoteEndPoint) +type EndPointPair = (LocalEndPoint, RemoteEndPoint) +-- | Lightweight connection ID (sender allocated) +-- +-- A ConnectionId is the concentation of a 'HeavyweightConnectionId' and a +-- 'LightweightConnectionId'. +type LightweightConnectionId = Word32 + +-- | Heavyweight connection ID (recipient allocated) +-- +-- A ConnectionId is the concentation of a 'HeavyweightConnectionId' and a +-- 'LightweightConnectionId'. +type HeavyweightConnectionId = Word32 + -- | Control headers data ControlHeader = - -- | Request a new connection ID from the remote endpoint - RequestConnectionId + -- | Tell the remote endpoint that we created a new connection + CreatedNewConnection -- | Tell the remote endpoint we will no longer be using a connection | CloseConnection - -- | Respond to a control request /from/ the remote endpoint - | ControlResponse -- | Request to close the connection (see module description) | CloseSocket deriving (Enum, Bounded, Show) @@ -551,7 +565,7 @@ else do resetIfBroken ourEndPoint theirAddress (theirEndPoint, connId) <- - requestConnectionTo params ourEndPoint theirAddress hints + createConnectionTo params 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 @@ -561,60 +575,65 @@ } -- | Close a connection -apiClose :: EndPointPair -> ConnectionId -> IORef Bool -> IO () +apiClose :: EndPointPair -> LightweightConnectionId -> IORef Bool -> IO () apiClose (ourEndPoint, theirEndPoint) connId connAlive = void . tryIO . asyncWhenCancelled return $ do - modifyRemoteState_ (ourEndPoint, theirEndPoint) remoteStateIdentity - { caseValid = \vst -> do - alive <- readIORef connAlive - if alive - then do - writeIORef connAlive False - sendOn vst [encodeInt32 CloseConnection, encodeInt32 connId] - return ( RemoteEndPointValid - . (remoteOutgoing ^: (\x -> x - 1)) - $ vst - ) - else - return (RemoteEndPointValid vst) - } + mAct <- modifyMVar (remoteState theirEndPoint) $ \st -> case st of + RemoteEndPointValid vst -> do + alive <- readIORef connAlive + if alive + then do + writeIORef connAlive False + act <- schedule theirEndPoint $ + sendOn vst [encodeInt32 CloseConnection, encodeInt32 connId] + return ( RemoteEndPointValid + . (remoteOutgoing ^: (\x -> x - 1)) + $ vst + , Just act + ) + else + return (RemoteEndPointValid vst, Nothing) + _ -> + return (st, Nothing) + forM_ mAct $ runScheduledAction (ourEndPoint, theirEndPoint) closeIfUnused (ourEndPoint, theirEndPoint) + -- | Send data across a connection -apiSend :: EndPointPair -- ^ Local and remote endpoint - -> ConnectionId -- ^ Connection ID (supplied by remote endpoint) - -> IORef Bool -- ^ Is the connection still alive? - -> [ByteString] -- ^ Payload +apiSend :: EndPointPair -- ^ Local and remote endpoint + -> LightweightConnectionId -- ^ Connection ID + -> IORef Bool -- ^ Is the connection still alive? + -> [ByteString] -- ^ Payload -> IO (Either (TransportError SendErrorCode) ()) apiSend (ourEndPoint, theirEndPoint) connId connAlive payload = -- We don't need the overhead of asyncWhenCancelled here - try . mapIOException sendFailed $ - withRemoteState (ourEndPoint, theirEndPoint) RemoteStatePatternMatch - { caseInvalid = \_ -> - relyViolation (ourEndPoint, theirEndPoint) "apiSend" - , caseInit = \_ _ -> - relyViolation (ourEndPoint, theirEndPoint) "apiSend" - , caseValid = \vst -> do - alive <- readIORef connAlive - if alive - then sendOn vst (encodeInt32 connId : prependLength payload) - else throwIO $ TransportError SendClosed "Connection closed" - , caseClosing = \_ _ -> do - alive <- readIORef connAlive - if alive - then relyViolation (ourEndPoint, theirEndPoint) "apiSend" - else throwIO $ TransportError SendClosed "Connection closed" - , caseClosed = do - alive <- readIORef connAlive - if alive - then relyViolation (ourEndPoint, theirEndPoint) "apiSend" - else throwIO $ TransportError SendClosed "Connection closed" - , caseFailed = \err -> do - alive <- readIORef connAlive - if alive - then throwIO $ TransportError SendFailed (show err) - else throwIO $ TransportError SendClosed "Connection closed" - } + try . mapIOException sendFailed $ do + act <- withMVar (remoteState theirEndPoint) $ \st -> case st of + RemoteEndPointInvalid _ -> + relyViolation (ourEndPoint, theirEndPoint) "apiSend" + RemoteEndPointInit _ _ _ -> + relyViolation (ourEndPoint, theirEndPoint) "apiSend" + RemoteEndPointValid vst -> do + alive <- readIORef connAlive + if alive + then schedule theirEndPoint $ sendOn vst (encodeInt32 connId : prependLength payload) + else throwIO $ TransportError SendClosed "Connection closed" + RemoteEndPointClosing _ _ -> do + alive <- readIORef connAlive + if alive + then relyViolation (ourEndPoint, theirEndPoint) "apiSend" + else throwIO $ TransportError SendClosed "Connection closed" + RemoteEndPointClosed -> do + alive <- readIORef connAlive + if alive + then relyViolation (ourEndPoint, theirEndPoint) "apiSend" + else throwIO $ TransportError SendClosed "Connection closed" + RemoteEndPointFailed err -> do + alive <- readIORef connAlive + if alive + then throwIO $ TransportError SendFailed (show err) + else throwIO $ TransportError SendClosed "Connection closed" + runScheduledAction (ourEndPoint, theirEndPoint) act where sendFailed = TransportError SendFailed . show @@ -644,156 +663,30 @@ -- We make an attempt to close the connection nicely -- (by sending a CloseSocket first) let closed = RemoteEndPointFailed . userError $ "apiCloseEndPoint" - modifyMVar_ (remoteState theirEndPoint) $ \st -> + mAct <- modifyMVar (remoteState theirEndPoint) $ \st -> case st of RemoteEndPointInvalid _ -> - return st - RemoteEndPointInit resolved _ -> do + return (st, Nothing) + RemoteEndPointInit resolved _ _ -> do putMVar resolved () - return closed - RemoteEndPointValid conn -> do - tryIO $ sendOn conn [encodeInt32 CloseSocket] - tryCloseSocket (remoteSocket conn) - return closed - RemoteEndPointClosing resolved conn -> do + return (closed, Nothing) + RemoteEndPointValid vst -> do + act <- schedule theirEndPoint $ do + tryIO $ sendOn vst [ encodeInt32 CloseSocket + , encodeInt32 (vst ^. remoteMaxIncoming) + ] + tryCloseSocket (remoteSocket vst) + return (closed, Just act) + RemoteEndPointClosing resolved vst -> do putMVar resolved () - tryCloseSocket (remoteSocket conn) - return closed + act <- schedule theirEndPoint $ tryCloseSocket (remoteSocket vst) + return (closed, Just act) RemoteEndPointClosed -> - return st + return (st, Nothing) RemoteEndPointFailed err -> - return $ RemoteEndPointFailed err - --------------------------------------------------------------------------------- --- As soon as a remote connection fails, we want to put notify our endpoint -- --- and put it into a closed state. Since this may happen in many places, we -- --- provide some abstractions. -- --------------------------------------------------------------------------------- - -data RemoteStatePatternMatch a = RemoteStatePatternMatch - { caseInvalid :: TransportError ConnectErrorCode -> IO a - , caseInit :: MVar () -> RequestedBy -> IO a - , caseValid :: ValidRemoteEndPointState -> IO a - , caseClosing :: MVar () -> ValidRemoteEndPointState -> IO a - , caseClosed :: IO a - , caseFailed :: IOException -> IO a - } - -remoteStateIdentity :: RemoteStatePatternMatch RemoteState -remoteStateIdentity = - RemoteStatePatternMatch - { caseInvalid = return . RemoteEndPointInvalid - , caseInit = (return .) . RemoteEndPointInit - , caseValid = return . RemoteEndPointValid - , caseClosing = (return .) . RemoteEndPointClosing - , caseClosed = return RemoteEndPointClosed - , caseFailed = return . RemoteEndPointFailed - } - --- | Like modifyMVar, but if an I/O exception occurs don't restore the remote --- endpoint to its original value but close it instead -modifyRemoteState :: EndPointPair - -> RemoteStatePatternMatch (RemoteState, a) - -> IO a -modifyRemoteState (ourEndPoint, theirEndPoint) match = - mask $ \restore -> do - st <- takeMVar theirState - case st of - RemoteEndPointValid vst -> do - mResult <- try $ restore (caseValid match vst) - case mResult of - Right (st', a) -> do - putMVar theirState st' - return a - Left ex -> do - case fromException ex of - Just ioEx -> handleIOException ioEx vst - Nothing -> putMVar theirState st - throwIO ex - -- The other cases are less interesting, because unless the endpoint is - -- in Valid state we're not supposed to do any IO on it - RemoteEndPointInit resolved origin -> do - (st', a) <- onException (restore $ caseInit match resolved origin) - (putMVar theirState st) - putMVar theirState st' - return a - RemoteEndPointClosing resolved vst -> do - (st', a) <- onException (restore $ caseClosing match resolved vst) - (putMVar theirState st) - putMVar theirState st' - return a - RemoteEndPointInvalid err -> do - (st', a) <- onException (restore $ caseInvalid match err) - (putMVar theirState st) - putMVar theirState st' - return a - RemoteEndPointClosed -> do - (st', a) <- onException (restore $ caseClosed match) - (putMVar theirState st) - putMVar theirState st' - return a - RemoteEndPointFailed err -> do - (st', a) <- onException (restore $ caseFailed match err) - (putMVar theirState st) - putMVar theirState st' - return a - where - theirState :: MVar RemoteState - theirState = remoteState theirEndPoint - - handleIOException :: IOException -> ValidRemoteEndPointState -> IO () - handleIOException ex vst = do - tryCloseSocket (remoteSocket vst) - putMVar theirState (RemoteEndPointFailed ex) - let incoming = IntSet.elems $ vst ^. remoteIncoming - code = EventConnectionLost (Just $ remoteAddress theirEndPoint) incoming - err = TransportError code (show ex) - writeChan (localChannel ourEndPoint) $ ErrorEvent err - --- | Like 'modifyRemoteState' but without a return value -modifyRemoteState_ :: EndPointPair - -> RemoteStatePatternMatch RemoteState - -> IO () -modifyRemoteState_ (ourEndPoint, theirEndPoint) match = - modifyRemoteState (ourEndPoint, theirEndPoint) - RemoteStatePatternMatch - { caseInvalid = u . caseInvalid match - , caseInit = \resolved origin -> u $ caseInit match resolved origin - , caseValid = u . caseValid match - , caseClosing = \resolved vst -> u $ caseClosing match resolved vst - , caseClosed = u $ caseClosed match - , caseFailed = u . caseFailed match - } - where - u :: IO a -> IO (a, ()) - u p = p >>= \a -> return (a, ()) - --- | Like 'modifyRemoteState' but without the ability to change the state -withRemoteState :: EndPointPair - -> RemoteStatePatternMatch a - -> IO a -withRemoteState (ourEndPoint, theirEndPoint) match = - modifyRemoteState (ourEndPoint, theirEndPoint) - RemoteStatePatternMatch - { caseInvalid = \err -> do - a <- caseInvalid match err - return (RemoteEndPointInvalid err, a) - , caseInit = \resolved origin -> do - a <- caseInit match resolved origin - return (RemoteEndPointInit resolved origin, a) - , caseValid = \vst -> do - a <- caseValid match vst - return (RemoteEndPointValid vst, a) - , caseClosing = \resolved vst -> do - a <- caseClosing match resolved vst - return (RemoteEndPointClosing resolved vst, a) - , caseClosed = do - a <- caseClosed match - return (RemoteEndPointClosed, a) - , caseFailed = \err -> do - a <- caseFailed match err - return (RemoteEndPointFailed err, a) - } + return (RemoteEndPointFailed err, Nothing) + forM_ mAct $ runScheduledAction (ourEndPoint, theirEndPoint) + -------------------------------------------------------------------------------- -- Incoming requests -- @@ -843,13 +736,13 @@ tryCloseSocket sock return Nothing else do + sendLock <- newMVar () let vst = ValidRemoteEndPointState - { remoteSocket = sock - , _remoteOutgoing = 0 - , _remoteIncoming = IntSet.empty - , sendOn = sendMany sock - , _pendingCtrlRequests = IntMap.empty - , _nextCtrlRequestId = 0 + { remoteSocket = sock + , remoteSendLock = sendLock + , _remoteOutgoing = 0 + , _remoteIncoming = Set.empty + , _remoteMaxIncoming = 0 } sendMany sock [encodeInt32 ConnectionRequestAccepted] resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst) @@ -880,7 +773,7 @@ RemoteEndPointInvalid _ -> relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages (invalid)" - RemoteEndPointInit _ _ -> + RemoteEndPointInit _ _ _ -> relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages (init)" RemoteEndPointValid ep -> @@ -905,108 +798,76 @@ -- exception thrown by 'recv'. go :: N.Socket -> IO () go sock = do - connId <- recvInt32 sock - if connId >= firstNonReservedConnectionId + lcid <- recvInt32 sock :: IO LightweightConnectionId + if lcid >= firstNonReservedLightweightConnectionId then do - readMessage sock connId + readMessage sock lcid go sock else - case tryToEnum (fromIntegral connId) of - Just RequestConnectionId -> do - recvInt32 sock >>= createNewConnection + case tryToEnum (fromIntegral lcid) of + Just CreatedNewConnection -> do + recvInt32 sock >>= createdNewConnection go sock - Just ControlResponse -> do - recvInt32 sock >>= readControlResponse sock - go sock Just CloseConnection -> do recvInt32 sock >>= closeConnection go sock Just CloseSocket -> do - didClose <- closeSocket sock + didClose <- recvInt32 sock >>= closeSocket sock unless didClose $ go sock Nothing -> throwIO $ userError "Invalid control request" -- Create a new connection - createNewConnection :: ControlRequestId -> IO () - createNewConnection reqId = do - newId <- getNextConnectionId ourEndPoint + createdNewConnection :: LightweightConnectionId -> IO () + createdNewConnection lcid = do modifyMVar_ theirState $ \st -> do vst <- case st of RemoteEndPointInvalid _ -> relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:createNewConnection (invalid)" - RemoteEndPointInit _ _ -> + RemoteEndPointInit _ _ _ -> relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:createNewConnection (init)" RemoteEndPointValid vst -> - return (remoteIncoming ^: IntSet.insert newId $ vst) + return ( (remoteIncoming ^: Set.insert lcid) + $ (remoteMaxIncoming ^= lcid) + vst + ) RemoteEndPointClosing resolved vst -> do -- If the endpoint is in closing state that means we send a -- CloseSocket request to the remote endpoint. If the remote - -- endpoint replies with the request to create a new connection, it - -- either ignored our request or it sent the request before it got - -- ours. Either way, at this point we simply restore the endpoint - -- to RemoteEndPointValid + -- endpoint replies that it created a new connection, it either + -- ignored our request or it sent the request before it got ours. + -- Either way, at this point we simply restore the endpoint to + -- RemoteEndPointValid putMVar resolved () - return (remoteIncoming ^= IntSet.singleton newId $ vst) + return ( (remoteIncoming ^= Set.singleton lcid) + . (remoteMaxIncoming ^= lcid) + $ vst + ) RemoteEndPointFailed err -> throwIO err RemoteEndPointClosed -> relyViolation (ourEndPoint, theirEndPoint) "createNewConnection (closed)" - sendOn vst ( encodeInt32 ControlResponse - : encodeInt32 reqId - : prependLength [encodeInt32 newId] - ) return (RemoteEndPointValid vst) - writeChan ourChannel (ConnectionOpened newId ReliableOrdered theirAddr) - - -- Read a control response - readControlResponse :: N.Socket -> ControlRequestId -> IO () - readControlResponse sock reqId = do - response <- recvWithLength sock - mmvar <- modifyMVar theirState $ \st -> case st of - RemoteEndPointInvalid _ -> - relyViolation (ourEndPoint, theirEndPoint) - "readControlResponse (invalid)" - RemoteEndPointInit _ _ -> - relyViolation (ourEndPoint, theirEndPoint) - "readControlResponse (init)" - RemoteEndPointValid vst -> - return ( RemoteEndPointValid - . (pendingCtrlRequestsAt reqId ^= Nothing) - $ vst - , vst ^. pendingCtrlRequestsAt reqId - ) - RemoteEndPointClosing _ _ -> - throwIO $ userError "Invalid control response" - RemoteEndPointFailed err -> - throwIO err - RemoteEndPointClosed -> - relyViolation (ourEndPoint, theirEndPoint) - "readControlResponse (closed)" - case mmvar of - Nothing -> - throwIO $ userError "Invalid request ID" - Just mvar -> - putMVar mvar (Right response) + writeChan ourChannel (ConnectionOpened (connId lcid) ReliableOrdered theirAddr) -- Close a connection -- It is important that we verify that the connection is in fact open, -- because otherwise we should not decrement the reference count - closeConnection :: ConnectionId -> IO () - closeConnection cid = do + closeConnection :: LightweightConnectionId -> IO () + closeConnection lcid = do modifyMVar_ theirState $ \st -> case st of RemoteEndPointInvalid _ -> relyViolation (ourEndPoint, theirEndPoint) "closeConnection (invalid)" - RemoteEndPointInit _ _ -> + RemoteEndPointInit _ _ _ -> relyViolation (ourEndPoint, theirEndPoint) "closeConnection (init)" RemoteEndPointValid vst -> do - unless (IntSet.member cid (vst ^. remoteIncoming)) $ + unless (Set.member lcid (vst ^. remoteIncoming)) $ throwIO $ userError "Invalid CloseConnection" return ( RemoteEndPointValid - . (remoteIncoming ^: IntSet.delete cid) + . (remoteIncoming ^: Set.delete lcid) $ vst ) RemoteEndPointClosing _ _ -> @@ -1018,54 +879,79 @@ throwIO err RemoteEndPointClosed -> relyViolation (ourEndPoint, theirEndPoint) "closeConnection (closed)" - writeChan ourChannel (ConnectionClosed cid) - closeIfUnused (ourEndPoint, theirEndPoint) + writeChan ourChannel (ConnectionClosed (connId lcid)) -- Close the socket (if we don't have any outgoing connections) - closeSocket :: N.Socket -> IO Bool - closeSocket sock = - modifyMVar theirState $ \st -> + closeSocket :: N.Socket -> LightweightConnectionId -> IO Bool + closeSocket sock lastReceivedId = do + mAct <- modifyMVar theirState $ \st -> do + lastSentId <- getLastConnOutId ourEndPoint case st of RemoteEndPointInvalid _ -> relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:closeSocket (invalid)" - RemoteEndPointInit _ _ -> + RemoteEndPointInit _ _ _ -> relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:closeSocket (init)" RemoteEndPointValid vst -> do -- We regard a CloseSocket message as an (optimized) way for the -- remote endpoint to indicate that all its connections to us are -- now properly closed - forM_ (IntSet.elems $ vst ^. remoteIncoming) $ - writeChan ourChannel . ConnectionClosed - let vst' = remoteIncoming ^= IntSet.empty $ vst - -- Check if we agree that the connection should be closed - if vst' ^. remoteOutgoing == 0 - then do + forM_ (Set.elems $ vst ^. remoteIncoming) $ + writeChan ourChannel . 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 + then + return (RemoteEndPointValid vst', Nothing) + else do removeRemoteEndPoint (ourEndPoint, theirEndPoint) -- Attempt to reply (but don't insist) - tryIO $ sendOn vst' [encodeInt32 CloseSocket] - tryCloseSocket sock - return (RemoteEndPointClosed, True) - else - return (RemoteEndPointValid vst', False) - RemoteEndPointClosing resolved _ -> do - removeRemoteEndPoint (ourEndPoint, theirEndPoint) - tryCloseSocket sock - putMVar resolved () - return (RemoteEndPointClosed, True) + act <- schedule theirEndPoint $ do + tryIO $ sendOn vst' [ encodeInt32 CloseSocket + , encodeInt32 (vst ^. remoteMaxIncoming) + ] + tryCloseSocket sock + return (RemoteEndPointClosed, Just act) + RemoteEndPointClosing resolved vst -> + -- Like above, we need to check if there is a ConnectionCreated + -- message that we sent but that the remote endpoint has not yet + -- 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 + then + return (RemoteEndPointClosing resolved vst, Nothing) + else do + removeRemoteEndPoint (ourEndPoint, theirEndPoint) + act <- schedule theirEndPoint $ tryCloseSocket sock + putMVar resolved () + return (RemoteEndPointClosed, Just act) RemoteEndPointFailed err -> throwIO err RemoteEndPointClosed -> relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:closeSocket (closed)" + case mAct of + Nothing -> return False + Just act -> do + runScheduledAction (ourEndPoint, theirEndPoint) act + return True -- Read a message and output it on the endPoint's channel. By rights we -- should verify that the connection ID is valid, but this is unnecessary -- overhead - readMessage :: N.Socket -> ConnectionId -> IO () - readMessage sock connId = - recvWithLength sock >>= writeChan ourChannel . Received connId + readMessage :: N.Socket -> LightweightConnectionId -> IO () + readMessage sock lcid = + recvWithLength sock >>= writeChan ourChannel . Received (connId lcid) -- Arguments ourChannel = localChannel ourEndPoint @@ -1081,15 +967,12 @@ RemoteEndPointInvalid _ -> relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:prematureExit" - RemoteEndPointInit _ _ -> + RemoteEndPointInit _ _ _ -> relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:prematureExit" - RemoteEndPointValid vst -> do - let code = EventConnectionLost - (Just $ remoteAddress theirEndPoint) - (IntSet.elems $ vst ^. remoteIncoming) + RemoteEndPointValid _ -> do + let code = EventConnectionLost (remoteAddress theirEndPoint) writeChan ourChannel . ErrorEvent $ TransportError code (show err) - forM_ (vst ^. pendingCtrlRequests) $ flip putMVar (Left err) return (RemoteEndPointFailed err) RemoteEndPointClosing resolved _ -> do putMVar resolved () @@ -1100,6 +983,10 @@ RemoteEndPointFailed err' -> return (RemoteEndPointFailed err') + -- Construct a connection ID + connId :: LightweightConnectionId -> ConnectionId + connId = createConnectionId (remoteId theirEndPoint) + -------------------------------------------------------------------------------- -- Uninterruptable auxiliary functions -- -- -- @@ -1107,19 +994,18 @@ -- be killed. -------------------------------------------------------------------------------- --- | Request a connection to a remote endpoint +-- | Create a connection to a remote endpoint -- --- This will block until we get a connection ID from the remote endpoint; if --- the remote endpoint was in 'RemoteEndPointClosing' state then we will --- additionally block until that is resolved. +-- If the remote endpoint is in 'RemoteEndPointClosing' state then we will +-- block until that is resolved. -- -- May throw a TransportError ConnectErrorCode exception. -requestConnectionTo :: TCPParameters +createConnectionTo :: TCPParameters -> LocalEndPoint -> EndPointAddress -> ConnectHints - -> IO (RemoteEndPoint, ConnectionId) -requestConnectionTo params ourEndPoint theirAddress hints = go + -> IO (RemoteEndPoint, LightweightConnectionId) +createConnectionTo params ourEndPoint theirAddress hints = go where go = do (theirEndPoint, isNew) <- mapIOException connectFailed $ @@ -1131,9 +1017,26 @@ setupRemoteEndPoint params (ourEndPoint, theirEndPoint) hints go else do - reply <- mapIOException connectFailed $ - doRemoteRequest (ourEndPoint, theirEndPoint) RequestConnectionId - return (theirEndPoint, decodeInt32 . BS.concat $ reply) + -- 'findRemoteEndPoint' will have increased 'remoteOutgoing' + mapIOException connectFailed $ do + act <- withMVar (remoteState theirEndPoint) $ \st -> case st of + RemoteEndPointValid vst -> do + connId <- getNextConnOutId ourEndPoint + schedule theirEndPoint $ do + sendOn vst [encodeInt32 CreatedNewConnection, encodeInt32 connId] + return connId + -- Error cases + RemoteEndPointInvalid err -> + throwIO err + RemoteEndPointFailed err -> + throwIO err + -- Algorithmic errors + _ -> + relyViolation (ourEndPoint, theirEndPoint) "createConnectionTo" + -- TODO: deal with exception case? + connId <- runScheduledAction (ourEndPoint, theirEndPoint) act + return (theirEndPoint, connId) + connectFailed :: IOException -> TransportError ConnectErrorCode connectFailed = TransportError ConnectFailed . show @@ -1151,13 +1054,13 @@ (connectTimeout hints) didAccept <- case result of Right (sock, ConnectionRequestAccepted) -> do + sendLock <- newMVar () let vst = ValidRemoteEndPointState - { remoteSocket = sock - , _remoteOutgoing = 0 - , _remoteIncoming = IntSet.empty - , sendOn = sendMany sock - , _pendingCtrlRequests = IntMap.empty - , _nextCtrlRequestId = 0 + { remoteSocket = sock + , remoteSendLock = sendLock + , _remoteOutgoing = 0 + , _remoteIncoming = Set.empty + , _remoteMaxIncoming = 0 } resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst) return True @@ -1167,7 +1070,13 @@ tryCloseSocket sock return False Right (sock, ConnectionRequestCrossed) -> do - resolveInit (ourEndPoint, theirEndPoint) RemoteEndPointClosed + withMVar (remoteState theirEndPoint) $ \st -> case st of + RemoteEndPointInit _ crossed _ -> + putMVar crossed () + RemoteEndPointFailed ex -> + throwIO ex + _ -> + relyViolation (ourEndPoint, theirEndPoint) "setupRemoteEndPoint: Crossed" tryCloseSocket sock return False Left err -> do @@ -1180,53 +1089,25 @@ theirAddress = remoteAddress theirEndPoint invalidAddress = TransportError ConnectNotFound --- | Do a (blocking) remote request --- --- May throw IO (user) exception if the local or the remote endpoint is closed, --- if the send fails, or if the remote endpoint fails before it replies. -doRemoteRequest :: EndPointPair -> ControlHeader -> IO [ByteString] -doRemoteRequest (ourEndPoint, theirEndPoint) header = do - replyMVar <- newEmptyMVar - modifyRemoteState_ (ourEndPoint, theirEndPoint) RemoteStatePatternMatch - { caseValid = \vst -> do - let reqId = vst ^. nextCtrlRequestId - sendOn vst [encodeInt32 header, encodeInt32 reqId] - return ( RemoteEndPointValid - . (nextCtrlRequestId ^: (+ 1)) - . (pendingCtrlRequestsAt reqId ^= Just replyMVar) - $ vst - ) - -- Error cases - , caseInvalid = - throwIO - , caseInit = \_ _ -> - relyViolation (ourEndPoint, theirEndPoint) "doRemoteRequest (init)" - , caseClosing = \_ _ -> - relyViolation (ourEndPoint, theirEndPoint) "doRemoteRequest (closing)" - , caseClosed = - relyViolation (ourEndPoint, theirEndPoint) "doRemoteRequest (closed)" - , caseFailed = - throwIO - } - mReply <- takeMVar replyMVar - case mReply of - Left err -> throwIO err - Right reply -> return reply - -- | Send a CloseSocket request if the remote endpoint is unused closeIfUnused :: EndPointPair -> IO () -closeIfUnused (ourEndPoint, theirEndPoint) = - modifyRemoteState_ (ourEndPoint, theirEndPoint) remoteStateIdentity - { caseValid = \vst -> - if vst ^. remoteOutgoing == 0 && IntSet.null (vst ^. remoteIncoming) - then do - sendOn vst [encodeInt32 CloseSocket] - resolved <- newEmptyMVar - return $ RemoteEndPointClosing resolved vst - else - return $ RemoteEndPointValid vst - } - +closeIfUnused (ourEndPoint, theirEndPoint) = do + mAct <- modifyMVar (remoteState theirEndPoint) $ \st -> case st of + RemoteEndPointValid vst -> + if vst ^. remoteOutgoing == 0 && Set.null (vst ^. remoteIncoming) + then do + resolved <- newEmptyMVar + act <- schedule theirEndPoint $ + sendOn vst [ encodeInt32 CloseSocket + , encodeInt32 (vst ^. remoteMaxIncoming) + ] + return (RemoteEndPointClosing resolved vst, Just act) + else + return (RemoteEndPointValid vst, Nothing) + _ -> + return (st, Nothing) + forM_ mAct $ runScheduledAction (ourEndPoint, theirEndPoint) + -- | Reset a remote endpoint if it is in Invalid mode -- -- If the remote endpoint is currently in broken state, and @@ -1261,7 +1142,8 @@ -> IO Connection connectToSelf ourEndPoint = do connAlive <- newIORef True -- Protected by the local endpoint lock - connId <- mapIOException connectFailed $ getNextConnectionId ourEndPoint + lconnId <- mapIOException connectFailed $ getNextConnOutId ourEndPoint + let connId = createConnectionId heavyweightSelfConnectionId lconnId writeChan ourChan $ ConnectionOpened connId ReliableOrdered (localAddress ourEndPoint) return Connection @@ -1302,7 +1184,7 @@ resolveInit :: EndPointPair -> RemoteState -> IO () resolveInit (ourEndPoint, theirEndPoint) newState = modifyMVar_ (remoteState theirEndPoint) $ \st -> case st of - RemoteEndPointInit resolved _ -> do + RemoteEndPointInit resolved _ _ -> do putMVar resolved () case newState of RemoteEndPointClosed -> @@ -1315,21 +1197,33 @@ _ -> relyViolation (ourEndPoint, theirEndPoint) "resolveInit" --- | Get the next connection ID +-- | Get the next outgoing connection ID -- -- Throws an IO exception when the endpoint is closed. -getNextConnectionId :: LocalEndPoint -> IO ConnectionId -getNextConnectionId ourEndpoint = +getNextConnOutId :: LocalEndPoint -> IO LightweightConnectionId +getNextConnOutId ourEndpoint = modifyMVar (localState ourEndpoint) $ \st -> case st of LocalEndPointValid vst -> do - let connId = vst ^. nextConnectionId + let connId = vst ^. nextConnOutId return ( LocalEndPointValid - . (nextConnectionId ^= connId + 1) + . (nextConnOutId ^= connId + 1) $ vst , connId) LocalEndPointClosed -> throwIO $ userError "Local endpoint closed" +-- | The last outgoing connection ID we created, or zero if we never created any +getLastConnOutId :: LocalEndPoint -> IO LightweightConnectionId +getLastConnOutId ourEndPoint = + withMVar (localState ourEndPoint) $ \st -> case st of + LocalEndPointValid vst -> + let nextId = vst ^. nextConnOutId in + if nextId == firstNonReservedLightweightConnectionId + then return 0 + else return (nextId - 1) + LocalEndPointClosed -> + throwIO $ userError "Local endpoint closed" + -- | Create a new local endpoint -- -- May throw a TransportError NewEndPointErrorCode exception if the transport @@ -1338,9 +1232,9 @@ createLocalEndPoint transport = do chan <- newChan state <- newMVar . LocalEndPointValid $ ValidLocalEndPointState - { _nextConnectionId = firstNonReservedConnectionId - , _localConnections = Map.empty - , _nextRemoteId = 0 + { _nextConnOutId = firstNonReservedLightweightConnectionId + , _nextConnInId = firstNonReservedHeavyweightConnectionId + , _localConnections = Map.empty } modifyMVar (transportState transport) $ \st -> case st of TransportValid vst -> do @@ -1415,16 +1309,19 @@ Just theirEndPoint -> return (st, (theirEndPoint, False)) Nothing -> do - resolved <- newEmptyMVar - theirState <- newMVar (RemoteEndPointInit resolved findOrigin) + resolved <- newEmptyMVar + crossed <- newEmptyMVar + theirState <- newMVar (RemoteEndPointInit resolved crossed findOrigin) + scheduled <- newChan let theirEndPoint = RemoteEndPoint - { remoteAddress = theirAddress - , remoteState = theirState - , remoteId = vst ^. nextRemoteId + { remoteAddress = theirAddress + , remoteState = theirState + , remoteId = vst ^. nextConnInId + , remoteScheduled = scheduled } return ( LocalEndPointValid . (localConnectionTo theirAddress ^= Just theirEndPoint) - . (nextRemoteId ^: (+ 1)) + . (nextConnInId ^: (+ 1)) $ vst , (theirEndPoint, True) ) @@ -1454,7 +1351,7 @@ case snapshot of RemoteEndPointInvalid err -> throwIO err - RemoteEndPointInit resolved initOrigin -> + RemoteEndPointInit resolved crossed initOrigin -> case (findOrigin, initOrigin) of (RequestedByUs, RequestedByUs) -> readMVar resolved >> go @@ -1462,9 +1359,10 @@ readMVar resolved >> go (RequestedByThem, RequestedByUs) -> if ourAddress > theirAddress - then + then do -- Wait for the Crossed message - readMVar resolved >> go + readMVar crossed + return (theirEndPoint, True) else return (theirEndPoint, False) (RequestedByThem, RequestedByThem) -> @@ -1485,7 +1383,61 @@ ourState = localState ourEndPoint ourAddress = localAddress ourEndPoint +-- | Send a payload over a heavyweight connection (thread safe) +sendOn :: ValidRemoteEndPointState -> [ByteString] -> IO () +sendOn vst bs = withMVar (remoteSendLock vst) $ \() -> + sendMany (remoteSocket vst) bs + -------------------------------------------------------------------------------- +-- Scheduling actions -- +-------------------------------------------------------------------------------- + +-- | See 'schedule'/'runScheduledAction' +type Action a = MVar (Either SomeException a) + +-- | Schedule an action to be executed (see also 'runScheduledAction') +schedule :: RemoteEndPoint -> IO a -> IO (Action a) +schedule theirEndPoint act = do + mvar <- newEmptyMVar + writeChan (remoteScheduled theirEndPoint) $ + catch (act >>= putMVar mvar . Right) (putMVar mvar . Left) + return mvar + +-- | Run a scheduled action. Every call to 'schedule' should be paired with a +-- call to 'runScheduledAction' so that every scheduled action is run. Note +-- however that the there is no guarantee that in +-- +-- > do act <- schedule p +-- > runScheduledAction +-- +-- 'runScheduledAction' will run @p@ (it might run some other scheduled action). +-- However, it will then wait until @p@ is executed (by this call to +-- 'runScheduledAction' or by another). +runScheduledAction :: EndPointPair -> Action a -> IO a +runScheduledAction (ourEndPoint, theirEndPoint) mvar = do + join $ readChan (remoteScheduled theirEndPoint) + ma <- readMVar mvar + case ma of + Right a -> return a + Left e -> do + forM_ (fromException e) $ \ioe -> + modifyMVar_ (remoteState theirEndPoint) $ \st -> + case st of + RemoteEndPointValid vst -> handleIOException ioe vst + _ -> return (RemoteEndPointFailed ioe) + throwIO e + where + handleIOException :: IOException + -> ValidRemoteEndPointState + -> IO RemoteState + handleIOException ex vst = do + tryCloseSocket (remoteSocket vst) + let code = EventConnectionLost (remoteAddress theirEndPoint) + err = TransportError code (show ex) + writeChan (localChannel ourEndPoint) $ ErrorEvent err + return (RemoteEndPointFailed ex) + +-------------------------------------------------------------------------------- -- "Stateless" (MVar free) functions -- -------------------------------------------------------------------------------- @@ -1547,6 +1499,13 @@ _ -> Nothing +-- | Construct a ConnectionId +createConnectionId :: HeavyweightConnectionId + -> LightweightConnectionId + -> ConnectionId +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/ -- @@ -1589,7 +1548,7 @@ Nothing -> throwIO $ userError "Remote endpoint not found" Just ep -> return ep withMVar (remoteState theirEndPoint) $ \st -> case st of - RemoteEndPointInit _ _ -> + RemoteEndPointInit _ _ _ -> throwIO $ userError "Remote endpoint not yet initialized" RemoteEndPointValid vst -> return $ remoteSocket vst @@ -1607,9 +1566,17 @@ -------------------------------------------------------------------------------- -- | We reserve a bunch of connection IDs for control messages -firstNonReservedConnectionId :: ConnectionId -firstNonReservedConnectionId = 1024 +firstNonReservedLightweightConnectionId :: LightweightConnectionId +firstNonReservedLightweightConnectionId = 1024 +-- | Self-connection +heavyweightSelfConnectionId :: HeavyweightConnectionId +heavyweightSelfConnectionId = 0 + +-- | We reserve some connection IDs for special heavyweight connections +firstNonReservedHeavyweightConnectionId :: HeavyweightConnectionId +firstNonReservedHeavyweightConnectionId = 1 + -------------------------------------------------------------------------------- -- Accessor definitions -- -------------------------------------------------------------------------------- @@ -1620,35 +1587,28 @@ nextEndPointId :: Accessor ValidTransportState EndPointId nextEndPointId = accessor _nextEndPointId (\eid st -> st { _nextEndPointId = eid }) -nextConnectionId :: Accessor ValidLocalEndPointState ConnectionId -nextConnectionId = accessor _nextConnectionId (\cix st -> st { _nextConnectionId = cix }) +nextConnOutId :: Accessor ValidLocalEndPointState LightweightConnectionId +nextConnOutId = accessor _nextConnOutId (\cix st -> st { _nextConnOutId = cix }) localConnections :: Accessor ValidLocalEndPointState (Map EndPointAddress RemoteEndPoint) localConnections = accessor _localConnections (\es st -> st { _localConnections = es }) -nextRemoteId :: Accessor ValidLocalEndPointState Int -nextRemoteId = accessor _nextRemoteId (\rid st -> st { _nextRemoteId = rid }) +nextConnInId :: Accessor ValidLocalEndPointState HeavyweightConnectionId +nextConnInId = accessor _nextConnInId (\rid st -> st { _nextConnInId = rid }) remoteOutgoing :: Accessor ValidRemoteEndPointState Int remoteOutgoing = accessor _remoteOutgoing (\cs conn -> conn { _remoteOutgoing = cs }) -remoteIncoming :: Accessor ValidRemoteEndPointState IntSet +remoteIncoming :: Accessor ValidRemoteEndPointState (Set LightweightConnectionId) remoteIncoming = accessor _remoteIncoming (\cs conn -> conn { _remoteIncoming = cs }) -pendingCtrlRequests :: Accessor ValidRemoteEndPointState (IntMap (MVar (Either IOException [ByteString]))) -pendingCtrlRequests = accessor _pendingCtrlRequests (\rep st -> st { _pendingCtrlRequests = rep }) - -nextCtrlRequestId :: Accessor ValidRemoteEndPointState ControlRequestId -nextCtrlRequestId = accessor _nextCtrlRequestId (\cid st -> st { _nextCtrlRequestId = cid }) +remoteMaxIncoming :: Accessor ValidRemoteEndPointState LightweightConnectionId +remoteMaxIncoming = accessor _remoteMaxIncoming (\lcid st -> st { _remoteMaxIncoming = lcid }) localEndPointAt :: EndPointAddress -> Accessor ValidTransportState (Maybe LocalEndPoint) localEndPointAt addr = localEndPoints >>> DAC.mapMaybe addr -pendingCtrlRequestsAt :: ControlRequestId -> Accessor ValidRemoteEndPointState (Maybe (MVar (Either IOException [ByteString]))) -pendingCtrlRequestsAt ix = pendingCtrlRequests >>> DAC.intMapMaybe (fromIntegral ix) - -localConnectionTo :: EndPointAddress - -> Accessor ValidLocalEndPointState (Maybe RemoteEndPoint) +localConnectionTo :: EndPointAddress -> Accessor ValidLocalEndPointState (Maybe RemoteEndPoint) localConnectionTo addr = localConnections >>> DAC.mapMaybe addr ------------------------------------------------------------------------------- diff --git a/src/Network/Transport/TCP/Internal.hs b/src/Network/Transport/TCP/Internal.hs --- a/src/Network/Transport/TCP/Internal.hs +++ b/src/Network/Transport/TCP/Internal.hs @@ -1,33 +1,35 @@ -- | Utility functions for TCP sockets -module Network.Transport.TCP.Internal ( forkServer - , recvWithLength - , recvExact - , recvInt32 - , tryCloseSocket - ) where +module Network.Transport.TCP.Internal + ( forkServer + , recvWithLength + , recvExact + , recvInt32 + , tryCloseSocket + ) where #if ! MIN_VERSION_base(4,6,0) import Prelude hiding (catch) #endif import Network.Transport.Internal (decodeInt32, void, tryIO, forkIOWithUnmask) -import qualified Network.Socket as N ( HostName - , ServiceName - , Socket - , SocketType(Stream) - , SocketOption(ReuseAddr) - , getAddrInfo - , defaultHints - , socket - , bindSocket - , listen - , addrFamily - , addrAddress - , defaultProtocol - , setSocketOption - , accept - , sClose - ) +import qualified Network.Socket as N + ( HostName + , ServiceName + , Socket + , SocketType(Stream) + , SocketOption(ReuseAddr) + , getAddrInfo + , defaultHints + , socket + , bindSocket + , listen + , addrFamily + , addrAddress + , defaultProtocol + , setSocketOption + , accept + , sClose + ) import qualified Network.Socket.ByteString as NBS (recv) import Control.Concurrent (ThreadId) import Control.Monad (forever, when) @@ -36,6 +38,7 @@ import Data.ByteString (ByteString) import qualified Data.ByteString as BS (length, concat, null) import Data.Int (Int32) +import Data.ByteString.Lazy.Internal (smallChunkSize) -- | Start a server at the specified address. -- @@ -113,7 +116,7 @@ go :: [ByteString] -> Int32 -> IO [ByteString] go acc 0 = return (reverse acc) go acc l = do - bs <- NBS.recv sock (fromIntegral l `min` 4096) + 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)) diff --git a/tests/TestAuxiliary.hs b/tests/TestAuxiliary.hs deleted file mode 100644 --- a/tests/TestAuxiliary.hs +++ /dev/null @@ -1,111 +0,0 @@ -{-# OPTIONS_GHC -fno-warn-orphans #-} -module TestAuxiliary ( -- Running tests - runTest - , runTests - -- Writing tests - , forkTry - , trySome - , randomThreadDelay - ) where - -#if ! MIN_VERSION_base(4,6,0) -import Prelude hiding (catch) -#endif - -import Control.Concurrent (myThreadId, forkIO, ThreadId, throwTo, threadDelay) -import Control.Concurrent.Chan (Chan) -import Control.Monad (liftM2, unless) -import Control.Exception (SomeException, try, catch) -import System.Timeout (timeout) -import System.IO (stdout, hFlush) -import System.Console.ANSI ( SGR(SetColor, Reset) - , Color(Red, Green) - , ConsoleLayer(Foreground) - , ColorIntensity(Vivid) - , setSGR - ) -import System.Random (randomIO) -import Network.Transport -import Traced (Traceable(..), traceShow) - --- | Like fork, but throw exceptions in the child thread to the parent -forkTry :: IO () -> IO ThreadId -forkTry p = do - tid <- myThreadId - forkIO $ catch p (\e -> throwTo tid (e :: SomeException)) - --- | Like try, but specialized to SomeException -trySome :: IO a -> IO (Either SomeException a) -trySome = try - --- | Run the given test, catching timeouts and exceptions -runTest :: String -> IO () -> IO Bool -runTest description test = do - putStr $ "Running " ++ show description ++ ": " - hFlush stdout - done <- try . timeout 60000000 $ test -- 60 seconds - case done of - Left err -> failed $ "(exception: " ++ show (err :: SomeException) ++ ")" - Right Nothing -> failed $ "(timeout)" - Right (Just ()) -> ok - where - failed :: String -> IO Bool - failed err = do - setSGR [SetColor Foreground Vivid Red] - putStr "failed " - setSGR [Reset] - putStrLn err - return False - - ok :: IO Bool - ok = do - setSGR [SetColor Foreground Vivid Green] - putStrLn "ok" - setSGR [Reset] - return True - --- | Run a bunch of tests and throw an exception if any fails -runTests :: [(String, IO ())] -> IO () -runTests tests = do - success <- foldr (liftM2 (&&) . uncurry runTest) (return True) $ tests - unless success $ fail "Some tests failed" - --- | Random thread delay between 0 and the specified max -randomThreadDelay :: Int -> IO () -randomThreadDelay maxDelay = do - delay <- randomIO :: IO Int - threadDelay (delay `mod` maxDelay) - --------------------------------------------------------------------------------- --- traceShow instances -- --------------------------------------------------------------------------------- - -instance Traceable EndPoint where - trace = const Nothing - -instance Traceable Transport where - trace = const Nothing - -instance Traceable Connection where - trace = const Nothing - -instance Traceable Event where - trace = traceShow - -instance Show err => Traceable (TransportError err) where - trace = traceShow - -instance Traceable EndPointAddress where - trace = traceShow . endPointAddressToByteString - -instance Traceable SomeException where - trace = traceShow - -instance Traceable ThreadId where - trace = const Nothing - -instance Traceable (Chan a) where - trace = const Nothing - -instance Traceable Float where - trace = traceShow diff --git a/tests/TestTCP.hs b/tests/TestTCP.hs --- a/tests/TestTCP.hs +++ b/tests/TestTCP.hs @@ -1,4 +1,4 @@ -{-# LANGUAGE RebindableSyntax #-} +{-# LANGUAGE RebindableSyntax, TemplateHaskell #-} {-# OPTIONS_GHC -fno-warn-orphans #-} module Main where @@ -11,14 +11,13 @@ , catch #endif ) -import TestTransport (testTransport) -import TestAuxiliary (forkTry, runTests) import Network.Transport import Network.Transport.TCP ( createTransport , createTransportExposeInternals , TransportInternals(..) , encodeEndPointAddress , defaultTCPParameters + , LightweightConnectionId ) import Data.Int (Int32) import Control.Concurrent (threadDelay, killThread) @@ -30,6 +29,7 @@ , isEmptyMVar , newMVar , modifyMVar + , modifyMVar_ ) import Control.Monad (replicateM, guard, forM_, replicateM_, when) import Control.Applicative ((<$>)) @@ -54,10 +54,12 @@ ) import Network.Socket.ByteString (sendMany) import Data.String (fromString) -import Traced import GHC.IO.Exception (ioe_errno) import Foreign.C.Error (Errno(..), eADDRNOTAVAIL) import System.Timeout (timeout) +import Network.Transport.Tests (testTransport) +import Network.Transport.Tests.Auxiliary (forkTry, runTests) +import Network.Transport.Tests.Traced instance Traceable ControlHeader where trace = traceShow @@ -98,11 +100,11 @@ -- TEST 1: they connect to us, then drop the connection do - ConnectionOpened cid _ addr <- receive endpoint + ConnectionOpened _ _ addr <- receive endpoint True <- return $ addr == theirAddr - ErrorEvent (TransportError (EventConnectionLost (Just addr') [cid']) _) <- receive endpoint - True <- return $ addr' == theirAddr && cid' == cid + ErrorEvent (TransportError (EventConnectionLost addr') _) <- receive endpoint + True <- return $ addr' == theirAddr return () @@ -124,8 +126,8 @@ Received cid' ["pong"] <- receive endpoint True <- return $ cid == cid' - ErrorEvent (TransportError (EventConnectionLost (Just addr') [cid'']) _) <- receive endpoint - True <- return $ addr' == theirAddr && cid'' == cid + ErrorEvent (TransportError (EventConnectionLost addr') _) <- receive endpoint + True <- return $ addr' == theirAddr return () @@ -149,21 +151,17 @@ _ <- recvWithLength sock sendMany sock [encodeInt32 ConnectionRequestAccepted] - -- Server requests a logical connection - RequestConnectionId <- toEnum <$> (recvInt32 sock :: IO Int) - reqId <- recvInt32 sock :: IO Int - sendMany sock (encodeInt32 ControlResponse : encodeInt32 reqId : prependLength [encodeInt32 (10001 :: Int)]) + -- Server opens a logical connection + CreatedNewConnection <- toEnum <$> (recvInt32 sock :: IO Int) + 1024 <- recvInt32 sock :: IO LightweightConnectionId -- Server sends a message - 10001 <- recvInt32 sock :: IO Int + 1024 <- recvInt32 sock :: IO Int ["ping"] <- recvWithLength sock -- Reply - sendMany sock [encodeInt32 RequestConnectionId, encodeInt32 (10002 :: Int)] - ControlResponse <- toEnum <$> (recvInt32 sock :: IO Int) - 10002 <- recvInt32 sock :: IO Int - [cid] <- recvWithLength sock - sendMany sock (cid : prependLength ["pong"]) + sendMany sock [encodeInt32 CreatedNewConnection, encodeInt32 (10002 :: Int)] + sendMany sock (encodeInt32 10002 : prependLength ["pong"]) -- Close the socket N.sClose sock @@ -171,9 +169,8 @@ -- Connect to the server Right (sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr True Nothing - -- Request a new connection, but don't wait for the response - let reqId = 0 :: Int32 - sendMany sock [encodeInt32 RequestConnectionId, encodeInt32 reqId] + -- Open a new connection + sendMany sock [encodeInt32 CreatedNewConnection, encodeInt32 (10003 :: Int)] -- Close the socket without closing the connection explicitly -- The server should receive an error event @@ -235,7 +232,7 @@ ConnectionClosed cid'' <- receive endpoint True <- return $ cid'' == cid - ErrorEvent (TransportError (EventConnectionLost (Just addr') []) _) <- receive endpoint + ErrorEvent (TransportError (EventConnectionLost addr') _) <- receive endpoint True <- return $ addr' == theirAddr return () @@ -260,37 +257,32 @@ _ <- recvWithLength sock sendMany sock [encodeInt32 ConnectionRequestAccepted] - -- Server requests a logical connection - RequestConnectionId <- toEnum <$> (recvInt32 sock :: IO Int) - reqId <- recvInt32 sock :: IO Int - sendMany sock (encodeInt32 ControlResponse : encodeInt32 reqId : prependLength [encodeInt32 (10001 :: Int)]) + -- Server opens a logical connection + CreatedNewConnection <- toEnum <$> (recvInt32 sock :: IO Int) + 1024 <- recvInt32 sock :: IO LightweightConnectionId -- Server sends a message - 10001 <- recvInt32 sock :: IO Int + 1024 <- recvInt32 sock :: IO Int ["ping"] <- recvWithLength sock -- Reply - sendMany sock [encodeInt32 RequestConnectionId, encodeInt32 (10002 :: Int)] - ControlResponse <- toEnum <$> (recvInt32 sock :: IO Int) - 10002 <- recvInt32 sock :: IO Int - [cid] <- recvWithLength sock - sendMany sock (cid : prependLength ["pong"]) + sendMany sock [encodeInt32 CreatedNewConnection, encodeInt32 (10002 :: Int)] + sendMany sock (encodeInt32 (10002 :: Int) : prependLength ["pong"]) -- Send a CloseSocket even though there are still connections *in both -- directions* - sendMany sock [encodeInt32 CloseSocket] + sendMany sock [encodeInt32 CloseSocket, encodeInt32 (1024 :: Int)] N.sClose sock -- Connect to the server Right (sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr True Nothing - -- Request a new connection, but don't wait for the response - let reqId = 0 :: Int32 - sendMany sock [encodeInt32 RequestConnectionId, encodeInt32 reqId] + -- Open a new connection + sendMany sock [encodeInt32 CreatedNewConnection, encodeInt32 (10003 :: Int)] -- Send a CloseSocket without sending a closeconnecton -- The server should still receive a ConnectionClosed message - sendMany sock [encodeInt32 CloseSocket] + sendMany sock [encodeInt32 CloseSocket, encodeInt32 (0 :: Int)] N.sClose sock -- | Test the creation of a transport with an invalid address @@ -329,160 +321,161 @@ -- receiving an (already underway) CloseSocket request) testIgnoreCloseSocket :: IO N.ServiceName -> IO () testIgnoreCloseSocket nextPort = do - serverAddr <- newEmptyMVar - clientDone <- newEmptyMVar - Right transport <- nextPort >>= \port -> createTransport "127.0.0.1" port defaultTCPParameters - - forkTry $ server transport serverAddr - forkTry $ client transport serverAddr clientDone - - takeMVar clientDone + serverAddr <- newEmptyMVar + clientAddr <- newEmptyMVar + clientDone <- newEmptyMVar + serverDone <- newEmptyMVar + connectionEstablished <- newEmptyMVar + Right transport <- nextPort >>= \port -> createTransport "127.0.0.1" port defaultTCPParameters - where - server :: Transport -> MVar EndPointAddress -> IO () - server transport serverAddr = do - tlog "Server" - Right endpoint <- newEndPoint transport - putMVar serverAddr (address endpoint) + -- Server + forkTry $ do + tlog "Server" + Right endpoint <- newEndPoint transport + putMVar serverAddr (address endpoint) - -- Wait for the client to connect and disconnect - tlog "Waiting for ConnectionOpened" - ConnectionOpened _ _ _ <- receive endpoint - tlog "Waiting for ConnectionClosed" - ConnectionClosed _ <- receive endpoint + let ourAddress = address endpoint + theirAddress <- readMVar clientAddr - -- At this point the server will have sent a CloseSocket request to the - -- client, which however ignores it, instead it requests and closes - -- another connection - tlog "Waiting for ConnectionOpened" - ConnectionOpened _ _ _ <- receive endpoint - tlog "Waiting for ConnectionClosed" - ConnectionClosed _ <- receive endpoint + -- Wait for the client to set up the TCP connection to us + takeMVar connectionEstablished - - tlog "Server waiting.." + -- Connect then disconnect to the client + Right conn <- connect endpoint theirAddress ReliableOrdered defaultConnectHints + close conn - client :: Transport -> MVar EndPointAddress -> MVar () -> IO () - client transport serverAddr clientDone = do - tlog "Client" - Right endpoint <- newEndPoint transport - let ourAddress = address endpoint + -- At this point the server will have sent a CloseSocket request to the + -- client, which however ignores it, instead it requests and closes + -- another connection + tlog "Waiting for ConnectionOpened" + ConnectionOpened _ _ _ <- receive endpoint + tlog "Waiting for ConnectionClosed" + ConnectionClosed _ <- receive endpoint - -- Connect to the server - Right (sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr True Nothing + putMVar serverDone () - -- Request a new connection - tlog "Requesting connection" - let reqId = 0 :: Int32 - sendMany sock [encodeInt32 RequestConnectionId, encodeInt32 reqId] - response <- replicateM 4 $ recvInt32 sock :: IO [Int32] + -- Client + forkTry $ do + tlog "Client" + Right endpoint <- newEndPoint transport + putMVar clientAddr (address endpoint) - -- Close the connection again - tlog "Closing connection" - sendMany sock [encodeInt32 CloseConnection, encodeInt32 (response !! 3)] + let ourAddress = address endpoint + theirAddress <- readMVar serverAddr - -- Server will now send a CloseSocket request as its refcount reached 0 - tlog "Waiting for CloseSocket request" - CloseSocket <- toEnum <$> recvInt32 sock + -- Connect to the server + Right (sock, ConnectionRequestAccepted) <- socketToEndPoint ourAddress theirAddress True Nothing + putMVar connectionEstablished () - -- But we ignore it and request another connection - tlog "Ignoring it, requesting another connection" - let reqId' = 1 :: Int32 - sendMany sock [encodeInt32 RequestConnectionId, encodeInt32 reqId'] - response' <- replicateM 4 $ recvInt32 sock :: IO [Int32] + -- Server connects to us, and then closes the connection + CreatedNewConnection <- toEnum <$> (recvInt32 sock :: IO Int) + 1024 <- recvInt32 sock :: IO LightweightConnectionId - -- Close it again - tlog "Closing connection" - sendMany sock [encodeInt32 CloseConnection, encodeInt32 (response' !! 3)] + CloseConnection <- toEnum <$> (recvInt32 sock :: IO Int) + 1024 <- recvInt32 sock :: IO LightweightConnectionId - -- We now get a CloseSocket again, and this time we heed it - tlog "Waiting for second CloseSocket request" - CloseSocket <- toEnum <$> recvInt32 sock - - tlog "Closing socket" - sendMany sock [encodeInt32 CloseSocket] - N.sClose sock + -- 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 - putMVar clientDone () + -- 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)] + -- Close it again + tlog "Closing connection" + sendMany sock [encodeInt32 CloseConnection, encodeInt32 (1024 :: Int)] + + -- And close the connection completely + tlog "Closing socket" + sendMany sock [encodeInt32 CloseSocket, encodeInt32 (1024 :: Int)] + N.sClose sock + + putMVar clientDone () + + takeMVar clientDone + takeMVar serverDone + -- | Like 'testIgnoreSocket', but now the server requests a connection after the -- client closed their connection. In the meantime, the server will have sent a -- CloseSocket request to the client, and must block until the client responds. testBlockAfterCloseSocket :: IO N.ServiceName -> IO () testBlockAfterCloseSocket nextPort = do - serverAddr <- newEmptyMVar - clientAddr <- newEmptyMVar - clientDone <- newEmptyMVar - port <- nextPort - Right transport <- createTransport "127.0.0.1" port defaultTCPParameters - - forkTry $ server transport serverAddr clientAddr - forkTry $ client transport serverAddr clientAddr clientDone - - takeMVar clientDone + serverAddr <- newEmptyMVar + clientAddr <- newEmptyMVar + clientDone <- newEmptyMVar + serverDone <- newEmptyMVar + connectionEstablished <- newEmptyMVar + Right transport <- nextPort >>= \port -> createTransport "127.0.0.1" port defaultTCPParameters - where - server :: Transport -> MVar EndPointAddress -> MVar EndPointAddress -> IO () - server transport serverAddr clientAddr = do - tlog "Server" - Right endpoint <- newEndPoint transport - putMVar serverAddr (address endpoint) + -- Server + forkTry $ do + tlog "Server" + Right endpoint <- newEndPoint transport + putMVar serverAddr (address endpoint) - -- Wait for the client to connect and disconnect - tlog "Waiting for ConnectionOpened" - ConnectionOpened _ _ _ <- receive endpoint - tlog "Waiting for ConnectionClosed" - ConnectionClosed _ <- receive endpoint - - -- At this point the server will have sent a CloseSocket request to the - -- client, and must block until the client responds - tlog "Server waiting to connect to the client.." - Right _ <- readMVar clientAddr >>= \addr -> connect endpoint addr ReliableOrdered defaultConnectHints - - tlog "Server waiting.." + let ourAddress = address endpoint + theirAddress <- readMVar clientAddr - client :: Transport -> MVar EndPointAddress -> MVar EndPointAddress -> MVar () -> IO () - client transport serverAddr clientAddr clientDone = do - tlog "Client" - Right endpoint <- newEndPoint transport - putMVar clientAddr (address endpoint) - let ourAddress = address endpoint + -- Wait for the client to set up the TCP connection to us + takeMVar connectionEstablished - -- Connect to the server - Right (sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= \addr -> socketToEndPoint ourAddress addr True Nothing + -- Connect then disconnect to the client + Right conn <- connect endpoint theirAddress ReliableOrdered defaultConnectHints + close conn - -- Request a new connection - tlog "Requesting connection" - let reqId = 0 :: Int32 - sendMany sock [encodeInt32 RequestConnectionId, encodeInt32 reqId] - response <- replicateM 4 $ recvInt32 sock :: IO [Int32] + -- At this point the server will have sent a CloseSocket request to the + -- client, and must block until the client responds + Right conn <- connect endpoint theirAddress ReliableOrdered defaultConnectHints - -- Close the connection again - tlog "Closing connection" - sendMany sock [encodeInt32 CloseConnection, encodeInt32 (response !! 3)] + putMVar serverDone () - -- Server will now send a CloseSocket request as its refcount reached 0 - tlog "Waiting for CloseSocket request" - CloseSocket <- toEnum <$> recvInt32 sock + -- Client + forkTry $ do + tlog "Client" + Right endpoint <- newEndPoint transport + putMVar clientAddr (address endpoint) - unblocked <- newEmptyMVar + let ourAddress = address endpoint + theirAddress <- readMVar serverAddr - -- We should not hear from the server until we unblock him by - -- responding to the CloseSocket request (in this case, we - -- respond by sending a ConnectionRequest) - forkTry $ do - recvInt32 sock :: IO Int32 - isEmptyMVar unblocked >>= (guard . not) - putMVar clientDone () + -- Connect to the server + Right (sock, ConnectionRequestAccepted) <- socketToEndPoint ourAddress theirAddress True Nothing + putMVar connectionEstablished () - threadDelay 1000000 + -- Server connects to us, and then closes the connection + CreatedNewConnection <- toEnum <$> (recvInt32 sock :: IO Int) + 1024 <- recvInt32 sock :: IO LightweightConnectionId - tlog "Client ignores close socket and sends connection request" - tlog "This should unblock the server" - putMVar unblocked () - let reqId' = 1 :: Int32 - sendMany sock [encodeInt32 RequestConnectionId, encodeInt32 reqId'] + CloseConnection <- toEnum <$> (recvInt32 sock :: IO Int) + 1024 <- recvInt32 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 + + unblocked <- newMVar False + + -- We should not hear from the server until we unblock him by + -- responding to the CloseSocket request (in this case, we + -- respond by sending a ConnectionRequest) + forkTry $ do + recvInt32 sock :: IO Int32 + readMVar unblocked >>= guard + putMVar clientDone () + + threadDelay 1000000 + + 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)] + + takeMVar clientDone + takeMVar serverDone + -- | Test what happens when a remote endpoint sends a connection request to our -- transport for an endpoint it already has a connection to testUnnecessaryConnect :: IO N.ServiceName -> Int -> IO () @@ -592,16 +585,16 @@ when (count > 0) $ do -- Client requests a logical connection - Right RequestConnectionId <- tryIO $ toEnum <$> (recvInt32 sock :: IO Int) - Right reqId <- tryIO $ (recvInt32 sock :: IO Int) - Right () <- tryIO $ sendMany sock (encodeInt32 ControlResponse : encodeInt32 reqId : prependLength [encodeInt32 (10001 :: Int)]) + Right CreatedNewConnection <- tryIO $ toEnum <$> (recvInt32 sock :: IO Int) + connId <- recvInt32 sock :: IO LightweightConnectionId return () - when (count > 1) $ do - -- Client sends a message - Right 10001 <- tryIO $ (recvInt32 sock :: IO Int) - Right ["ping"] <- tryIO $ recvWithLength sock - putMVar serverDone () + when (count > 1) $ do + -- Client sends a message + Right connId' <- tryIO $ (recvInt32 sock :: IO LightweightConnectionId) + True <- return $ connId == connId' + Right ["ping"] <- tryIO $ recvWithLength sock + putMVar serverDone () Right () <- tryIO $ N.sClose sock return () @@ -615,7 +608,7 @@ let theirAddr = encodeEndPointAddress "127.0.0.1" serverPort 0 -- The first attempt will fail because no endpoint is yet set up - -- Left (TransportError ConnectNotFound _) <- connect endpoint theirAddr ReliableOrdered defaultConnectHints + Left (TransportError ConnectNotFound _) <- connect endpoint theirAddr ReliableOrdered defaultConnectHints putMVar firstAttempt () -- The second attempt will fail because the server closes the socket before we can request a connection @@ -633,8 +626,9 @@ Right conn1 <- connect endpoint theirAddr ReliableOrdered defaultConnectHints -- But a send will fail because the server has closed the connection again + threadDelay 100000 Left (TransportError SendFailed _) <- send conn1 ["ping"] - ErrorEvent (TransportError (EventConnectionLost _ []) _) <- receive endpoint + ErrorEvent (TransportError (EventConnectionLost _) _) <- receive endpoint -- But a subsequent call to connect should reestablish the connection Right conn2 <- connect endpoint theirAddr ReliableOrdered defaultConnectHints @@ -667,11 +661,11 @@ _ <- recvWithLength sock () <- sendMany sock [encodeInt32 ConnectionRequestAccepted] - RequestConnectionId <- toEnum <$> (recvInt32 sock :: IO Int) - reqId <- recvInt32 sock :: IO Int - sendMany sock (encodeInt32 ControlResponse : encodeInt32 reqId : prependLength [encodeInt32 (10001 :: Int)]) + CreatedNewConnection <- toEnum <$> (recvInt32 sock :: IO Int) + connId <- recvInt32 sock :: IO LightweightConnectionId - 10001 <- recvInt32 sock :: IO Int + connId' <- recvInt32 sock :: IO LightweightConnectionId + True <- return $ connId == connId' ["ping"] <- recvWithLength sock putMVar serverGotPing () @@ -695,7 +689,7 @@ -- But when we send we find the error Left (TransportError SendFailed _) <- send conn1 ["ping"] - ErrorEvent (TransportError (EventConnectionLost _ []) _) <- receive endpoint + ErrorEvent (TransportError (EventConnectionLost _) _) <- receive endpoint -- A call to connect should now re-establish the connection Right conn2 <- connect endpoint theirAddr ReliableOrdered defaultConnectHints @@ -709,7 +703,7 @@ -- We now find the error when we attempt to close the connection Nothing <- timeout 500000 $ receive endpoint close conn2 - ErrorEvent (TransportError (EventConnectionLost _ []) _) <- receive endpoint + ErrorEvent (TransportError (EventConnectionLost _) _) <- receive endpoint Right conn3 <- connect endpoint theirAddr ReliableOrdered defaultConnectHints send conn3 ["ping"] takeMVar serverGotPing @@ -721,7 +715,7 @@ -- Now we notice the problem when we try to connect Nothing <- timeout 500000 $ receive endpoint Left (TransportError ConnectFailed _) <- connect endpoint theirAddr ReliableOrdered defaultConnectHints - ErrorEvent (TransportError (EventConnectionLost _ []) _) <- receive endpoint + ErrorEvent (TransportError (EventConnectionLost _) _) <- receive endpoint Right conn4 <- connect endpoint theirAddr ReliableOrdered defaultConnectHints send conn4 ["ping"] takeMVar serverGotPing @@ -746,7 +740,7 @@ -- At this point the client sends an invalid request, so we terminate the -- connection - ErrorEvent (TransportError (EventConnectionLost _ [_]) _) <- receive endpoint + ErrorEvent (TransportError (EventConnectionLost _) _) <- receive endpoint putMVar serverDone () @@ -777,7 +771,7 @@ , ("EarlyCloseSocket", testEarlyCloseSocket nextPort) , ("IgnoreCloseSocket", testIgnoreCloseSocket nextPort) , ("BlockAfterCloseSocket", testBlockAfterCloseSocket nextPort) - , ("TestUnnecessaryConnect", testUnnecessaryConnect nextPort 10) + , ("UnnecessaryConnect", testUnnecessaryConnect nextPort 10) , ("InvalidAddress", testInvalidAddress nextPort) , ("InvalidConnect", testInvalidConnect nextPort) , ("Many", testMany nextPort) diff --git a/tests/TestTransport.hs b/tests/TestTransport.hs deleted file mode 100644 --- a/tests/TestTransport.hs +++ /dev/null @@ -1,964 +0,0 @@ -{-# LANGUAGE RebindableSyntax #-} -module TestTransport where - -import Prelude hiding - ( (>>=) - , return - , fail - , (>>) -#if ! MIN_VERSION_base(4,6,0) - , catch -#endif - ) -import TestAuxiliary (forkTry, runTests, trySome, randomThreadDelay) -import Control.Concurrent (forkIO, killThread, yield) -import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar, readMVar, tryTakeMVar, modifyMVar_, newMVar) -import Control.Exception (evaluate, throw, throwIO, bracket) -import Control.Monad (replicateM, replicateM_, when, guard, forM_, unless) -import Control.Monad.Error () -import Control.Applicative ((<$>)) -import Network.Transport -import Network.Transport.Internal (tlog, tryIO, timeoutMaybe) -import Network.Transport.Util (spawn) -import System.Random (randomIO) -import Data.ByteString (ByteString) -import Data.ByteString.Char8 (pack) -import Data.Map (Map) -import qualified Data.Map as Map (empty, insert, delete, findWithDefault, adjust, null, toList, map) -import Data.String (fromString) -import Data.List (permutations) -import Traced - --- | Server that echoes messages straight back to the origin endpoint. -echoServer :: EndPoint -> IO () -echoServer endpoint = do - go Map.empty - where - go :: Map ConnectionId Connection -> IO () - go cs = do - event <- receive endpoint - case event of - ConnectionOpened cid rel addr -> do - tlog $ "Opened new connection " ++ show cid - Right conn <- connect endpoint addr rel defaultConnectHints - go (Map.insert cid conn cs) - Received cid payload -> do - send (Map.findWithDefault (error $ "Received: Invalid cid " ++ show cid) cid cs) payload - go cs - ConnectionClosed cid -> do - tlog $ "Close connection " ++ show cid - close (Map.findWithDefault (error $ "ConnectionClosed: Invalid cid " ++ show cid) cid cs) - go (Map.delete cid cs) - ReceivedMulticast _ _ -> - -- Ignore - go cs - ErrorEvent _ -> - putStrLn $ "Echo server received error event: " ++ show event - EndPointClosed -> - return () - --- | Ping client used in a few tests -ping :: EndPoint -> EndPointAddress -> Int -> ByteString -> IO () -ping endpoint server numPings msg = do - -- Open connection to the server - tlog "Connect to echo server" - Right conn <- connect endpoint server ReliableOrdered defaultConnectHints - - -- Wait for the server to open reply connection - tlog "Wait for ConnectionOpened message" - ConnectionOpened cid _ _ <- receive endpoint - - -- Send pings and wait for reply - tlog "Send ping and wait for reply" - replicateM_ numPings $ do - send conn [msg] - Received cid' [reply] <- receive endpoint ; True <- return $ cid == cid' && reply == msg - return () - - -- Close the connection - tlog "Close the connection" - close conn - - -- Wait for the server to close its connection to us - tlog "Wait for ConnectionClosed message" - ConnectionClosed cid' <- receive endpoint ; True <- return $ cid == cid' - - -- Done - tlog "Ping client done" - --- | Basic ping test -testPingPong :: Transport -> Int -> IO () -testPingPong transport numPings = do - tlog "Starting ping pong test" - server <- spawn transport echoServer - result <- newEmptyMVar - - -- Client - forkTry $ do - tlog "Ping client" - Right endpoint <- newEndPoint transport - ping endpoint server numPings "ping" - putMVar result () - - takeMVar result - --- | Test that endpoints don't get confused -testEndPoints :: Transport -> Int -> IO () -testEndPoints transport numPings = do - server <- spawn transport echoServer - dones <- replicateM 2 newEmptyMVar - - forM_ (zip dones ['A'..]) $ \(done, name) -> forkTry $ do - let name' :: ByteString - name' = pack [name] - Right endpoint <- newEndPoint transport - tlog $ "Ping client " ++ show name' ++ ": " ++ show (address endpoint) - ping endpoint server numPings name' - putMVar done () - - forM_ dones takeMVar - --- Test that connections don't get confused -testConnections :: Transport -> Int -> IO () -testConnections transport numPings = do - server <- spawn transport echoServer - result <- newEmptyMVar - - -- Client - forkTry $ do - Right endpoint <- newEndPoint transport - - -- Open two connections to the server - Right conn1 <- connect endpoint server ReliableOrdered defaultConnectHints - ConnectionOpened serv1 _ _ <- receive endpoint - - Right conn2 <- connect endpoint server ReliableOrdered defaultConnectHints - ConnectionOpened serv2 _ _ <- receive endpoint - - -- One thread to send "pingA" on the first connection - forkTry $ replicateM_ numPings $ send conn1 ["pingA"] - - -- One thread to send "pingB" on the second connection - forkTry $ replicateM_ numPings $ send conn2 ["pingB"] - - -- Verify server responses - let verifyResponse 0 = putMVar result () - verifyResponse n = do - event <- receive endpoint - case event of - Received cid [payload] -> do - when (cid == serv1 && payload /= "pingA") $ error "Wrong message" - when (cid == serv2 && payload /= "pingB") $ error "Wrong message" - verifyResponse (n - 1) - _ -> - verifyResponse n - verifyResponse (2 * numPings) - - takeMVar result - --- | Test that closing one connection does not close the other -testCloseOneConnection :: Transport -> Int -> IO () -testCloseOneConnection transport numPings = do - server <- spawn transport echoServer - result <- newEmptyMVar - - -- Client - forkTry $ do - Right endpoint <- newEndPoint transport - - -- Open two connections to the server - Right conn1 <- connect endpoint server ReliableOrdered defaultConnectHints - ConnectionOpened serv1 _ _ <- receive endpoint - - Right conn2 <- connect endpoint server ReliableOrdered defaultConnectHints - ConnectionOpened serv2 _ _ <- receive endpoint - - -- One thread to send "pingA" on the first connection - forkTry $ do - replicateM_ numPings $ send conn1 ["pingA"] - close conn1 - - -- One thread to send "pingB" on the second connection - forkTry $ replicateM_ (numPings * 2) $ send conn2 ["pingB"] - - -- Verify server responses - let verifyResponse 0 = putMVar result () - verifyResponse n = do - event <- receive endpoint - case event of - Received cid [payload] -> do - when (cid == serv1 && payload /= "pingA") $ error "Wrong message" - when (cid == serv2 && payload /= "pingB") $ error "Wrong message" - verifyResponse (n - 1) - _ -> - verifyResponse n - verifyResponse (3 * numPings) - - takeMVar result - --- | Test that if A connects to B and B connects to A, B can still send to A after --- A closes its connection to B (for instance, in the TCP transport, the socket pair --- connecting A and B should not yet be closed). -testCloseOneDirection :: Transport -> Int -> IO () -testCloseOneDirection transport numPings = do - addrA <- newEmptyMVar - addrB <- newEmptyMVar - doneA <- newEmptyMVar - doneB <- newEmptyMVar - - -- A - forkTry $ do - tlog "A" - Right endpoint <- newEndPoint transport - tlog (show (address endpoint)) - putMVar addrA (address endpoint) - - -- Connect to B - tlog "Connect to B" - Right conn <- readMVar addrB >>= \addr -> connect endpoint addr ReliableOrdered defaultConnectHints - - -- Wait for B to connect to us - tlog "Wait for B" - ConnectionOpened cid _ _ <- receive endpoint - - -- Send pings to B - tlog "Send pings to B" - replicateM_ numPings $ send conn ["ping"] - - -- Close our connection to B - tlog "Close connection" - close conn - - -- Wait for B's pongs - tlog "Wait for pongs from B" - replicateM_ numPings $ do Received _ _ <- receive endpoint ; return () - - -- Wait for B to close it's connection to us - tlog "Wait for B to close connection" - ConnectionClosed cid' <- receive endpoint - guard (cid == cid') - - -- Done - tlog "Done" - putMVar doneA () - - -- B - forkTry $ do - tlog "B" - Right endpoint <- newEndPoint transport - tlog (show (address endpoint)) - putMVar addrB (address endpoint) - - -- Wait for A to connect - tlog "Wait for A to connect" - ConnectionOpened cid _ _ <- receive endpoint - - -- Connect to A - tlog "Connect to A" - Right conn <- readMVar addrA >>= \addr -> connect endpoint addr ReliableOrdered defaultConnectHints - - -- Wait for A's pings - tlog "Wait for pings from A" - replicateM_ numPings $ do Received _ _ <- receive endpoint ; return () - - -- Wait for A to close it's connection to us - tlog "Wait for A to close connection" - ConnectionClosed cid' <- receive endpoint - guard (cid == cid') - - -- Send pongs to A - tlog "Send pongs to A" - replicateM_ numPings $ send conn ["pong"] - - -- Close our connection to A - tlog "Close connection to A" - close conn - - -- Done - tlog "Done" - putMVar doneB () - - mapM_ takeMVar [doneA, doneB] - --- | Collect events and order them by connection ID -collect :: EndPoint -> Maybe Int -> Maybe Int -> IO [(ConnectionId, [[ByteString]])] -collect endPoint maxEvents timeout = go maxEvents Map.empty Map.empty - where - -- TODO: for more serious use of this function we'd need to make these arguments strict - go (Just 0) open closed = finish open closed - go n open closed = do - mEvent <- tryIO . timeoutMaybe timeout (userError "timeout") $ receive endPoint - case mEvent of - Left _ -> finish open closed - Right event -> do - let n' = (\x -> x - 1) <$> n - case event of - ConnectionOpened cid _ _ -> - go n' (Map.insert cid [] open) closed - ConnectionClosed cid -> - let list = Map.findWithDefault (error "Invalid ConnectionClosed") cid open in - go n' (Map.delete cid open) (Map.insert cid list closed) - Received cid msg -> - go n' (Map.adjust (msg :) cid open) closed - ReceivedMulticast _ _ -> - fail "Unexpected multicast" - ErrorEvent _ -> - fail "Unexpected error" - EndPointClosed -> - fail "Unexpected endpoint closure" - - finish open closed = - if Map.null open - then return . Map.toList . Map.map reverse $ closed - else fail $ "Open connections: " ++ show (map fst . Map.toList $ open) - --- | Open connection, close it, then reopen it --- (In the TCP transport this means the socket will be closed, then reopened) --- --- Note that B cannot expect to receive all of A's messages on the first connection --- before receiving the messages on the second connection. What might (and sometimes --- does) happen is that finishes sending all of its messages on the first connection --- (in the TCP transport, the first socket pair) while B is behind on reading _from_ --- this connection (socket pair) -- the messages are "in transit" on the network --- (these tests are done on localhost, so there are in some OS buffer). Then when --- A opens the second connection (socket pair) B will spawn a new thread for this --- connection, and hence might start interleaving messages from the first and second --- connection. --- --- This is correct behaviour, however: the transport API guarantees reliability and --- ordering _per connection_, but not _across_ connections. -testCloseReopen :: Transport -> Int -> IO () -testCloseReopen transport numPings = do - addrB <- newEmptyMVar - doneB <- newEmptyMVar - - let numRepeats = 2 :: Int - - -- A - forkTry $ do - Right endpoint <- newEndPoint transport - - forM_ [1 .. numRepeats] $ \i -> do - tlog "A connecting" - -- Connect to B - Right conn <- readMVar addrB >>= \addr -> connect endpoint addr ReliableOrdered defaultConnectHints - - tlog "A pinging" - -- Say hi - forM_ [1 .. numPings] $ \j -> send conn [pack $ "ping" ++ show i ++ "/" ++ show j] - - tlog "A closing" - -- Disconnect again - close conn - - tlog "A finishing" - - -- B - forkTry $ do - Right endpoint <- newEndPoint transport - putMVar addrB (address endpoint) - - eventss <- collect endpoint (Just (numRepeats * (numPings + 2))) Nothing - - forM_ (zip [1 .. numRepeats] eventss) $ \(i, (_, events)) -> do - forM_ (zip [1 .. numPings] events) $ \(j, event) -> do - guard (event == [pack $ "ping" ++ show i ++ "/" ++ show j]) - - putMVar doneB () - - takeMVar doneB - --- | Test lots of parallel connection attempts -testParallelConnects :: Transport -> Int -> IO () -testParallelConnects transport numPings = do - server <- spawn transport echoServer - done <- newEmptyMVar - - Right endpoint <- newEndPoint transport - - -- Spawn lots of clients - forM_ [1 .. numPings] $ \i -> forkTry $ do - Right conn <- connect endpoint server ReliableOrdered defaultConnectHints - send conn [pack $ "ping" ++ show i] - send conn [pack $ "ping" ++ show i] - close conn - - forkTry $ do - eventss <- collect endpoint (Just (numPings * 4)) Nothing - -- Check that no pings got sent to the wrong connection - forM_ eventss $ \(_, [[ping1], [ping2]]) -> - guard (ping1 == ping2) - putMVar done () - - takeMVar done - --- | Test that sending on a closed connection gives an error -testSendAfterClose :: Transport -> Int -> IO () -testSendAfterClose transport numRepeats = do - server <- spawn transport echoServer - clientDone <- newEmptyMVar - - forkTry $ do - Right endpoint <- newEndPoint transport - - -- We request two lightweight connections - replicateM numRepeats $ do - Right conn1 <- connect endpoint server ReliableOrdered defaultConnectHints - Right conn2 <- connect endpoint server ReliableOrdered defaultConnectHints - - -- Close the second, but leave the first open; then output on the second - -- connection (i.e., on a closed connection while there is still another - -- connection open) - close conn2 - Left (TransportError SendClosed _) <- send conn2 ["ping2"] - - -- Now close the first connection, and output on it (i.e., output while - -- there are no lightweight connection at all anymore) - close conn1 - Left (TransportError SendClosed _) <- send conn2 ["ping2"] - - return () - - putMVar clientDone () - - takeMVar clientDone - --- | Test that closing the same connection twice has no effect -testCloseTwice :: Transport -> Int -> IO () -testCloseTwice transport numRepeats = do - server <- spawn transport echoServer - clientDone <- newEmptyMVar - - forkTry $ do - Right endpoint <- newEndPoint transport - - replicateM numRepeats $ do - -- We request two lightweight connections - Right conn1 <- connect endpoint server ReliableOrdered defaultConnectHints - Right conn2 <- connect endpoint server ReliableOrdered defaultConnectHints - - -- Close the second one twice - close conn2 - close conn2 - - -- Then send a message on the first and close that twice too - send conn1 ["ping"] - close conn1 - - -- Verify expected response from the echo server - ConnectionOpened cid1 _ _ <- receive endpoint - ConnectionOpened cid2 _ _ <- receive endpoint - ConnectionClosed cid2' <- receive endpoint ; True <- return $ cid2' == cid2 - Received cid1' ["ping"] <- receive endpoint ; True <- return $ cid1' == cid1 - ConnectionClosed cid1'' <- receive endpoint ; True <- return $ cid1'' == cid1 - - return () - - putMVar clientDone () - - takeMVar clientDone - --- | Test that we can connect an endpoint to itself -testConnectToSelf :: Transport -> Int -> IO () -testConnectToSelf transport numPings = do - done <- newEmptyMVar - Right endpoint <- newEndPoint transport - - tlog "Creating self-connection" - Right conn <- connect endpoint (address endpoint) ReliableOrdered defaultConnectHints - - tlog "Talk to myself" - - -- One thread to write to the endpoint - forkTry $ do - tlog $ "writing" - - tlog $ "Sending ping" - replicateM_ numPings $ send conn ["ping"] - - tlog $ "Closing connection" - close conn - - -- And one thread to read - forkTry $ do - tlog $ "reading" - - tlog "Waiting for ConnectionOpened" - ConnectionOpened cid _ addr <- receive endpoint ; True <- return $ addr == address endpoint - - tlog "Waiting for Received" - replicateM_ numPings $ do - Received cid' ["ping"] <- receive endpoint ; True <- return $ cid == cid' - return () - - tlog "Waiting for ConnectionClosed" - ConnectionClosed cid' <- receive endpoint ; True <- return $ cid == cid' - - tlog "Done" - putMVar done () - - takeMVar done - --- | Test that we can connect an endpoint to itself multiple times -testConnectToSelfTwice :: Transport -> Int -> IO () -testConnectToSelfTwice transport numPings = do - done <- newEmptyMVar - Right endpoint <- newEndPoint transport - - tlog "Creating self-connection" - Right conn1 <- connect endpoint (address endpoint) ReliableOrdered defaultConnectHints - Right conn2 <- connect endpoint (address endpoint) ReliableOrdered defaultConnectHints - - tlog "Talk to myself" - - -- One thread to write to the endpoint using the first connection - forkTry $ do - tlog $ "writing" - - tlog $ "Sending ping" - replicateM_ numPings $ send conn1 ["pingA"] - - tlog $ "Closing connection" - close conn1 - - -- One thread to write to the endpoint using the second connection - forkTry $ do - tlog $ "writing" - - tlog $ "Sending ping" - replicateM_ numPings $ send conn2 ["pingB"] - - tlog $ "Closing connection" - close conn2 - - -- And one thread to read - forkTry $ do - tlog $ "reading" - - [(_, events1), (_, events2)] <- collect endpoint (Just (2 * (numPings + 2))) Nothing - True <- return $ events1 == replicate numPings ["pingA"] - True <- return $ events2 == replicate numPings ["pingB"] - - tlog "Done" - putMVar done () - - takeMVar done - --- | Test that we self-connections no longer work once we close our endpoint --- or our transport -testCloseSelf :: IO (Either String Transport) -> IO () -testCloseSelf newTransport = do - Right transport <- newTransport - Right endpoint1 <- newEndPoint transport - Right endpoint2 <- newEndPoint transport - Right conn1 <- connect endpoint1 (address endpoint1) ReliableOrdered defaultConnectHints - Right conn2 <- connect endpoint1 (address endpoint1) ReliableOrdered defaultConnectHints - Right conn3 <- connect endpoint2 (address endpoint2) ReliableOrdered defaultConnectHints - - -- Close the conneciton and try to send - close conn1 - Left (TransportError SendClosed _) <- send conn1 ["ping"] - - -- Close the first endpoint. We should not be able to use the first - -- connection anymore, or open more self connections, but the self connection - -- to the second endpoint should still be fine - closeEndPoint endpoint1 - Left (TransportError SendFailed _) <- send conn2 ["ping"] - Left (TransportError ConnectFailed _) <- connect endpoint1 (address endpoint1) ReliableOrdered defaultConnectHints - Right () <- send conn3 ["ping"] - - -- Close the transport; now the second should no longer work - closeTransport transport - Left (TransportError SendFailed _) <- send conn3 ["ping"] - Left (TransportError ConnectFailed _) <- connect endpoint2 (address endpoint2) ReliableOrdered defaultConnectHints - - return () - --- | Test various aspects of 'closeEndPoint' -testCloseEndPoint :: Transport -> Int -> IO () -testCloseEndPoint transport _ = do - serverDone <- newEmptyMVar - clientDone <- newEmptyMVar - clientAddr1 <- newEmptyMVar - clientAddr2 <- newEmptyMVar - serverAddr <- newEmptyMVar - - -- Server - forkTry $ do - Right endpoint <- newEndPoint transport - putMVar serverAddr (address endpoint) - - -- First test (see client) - do - theirAddr <- readMVar clientAddr1 - ConnectionOpened cid ReliableOrdered addr <- receive endpoint ; True <- return $ addr == theirAddr - ConnectionClosed cid' <- receive endpoint ; True <- return $ cid == cid' - return () - - -- Second test - do - theirAddr <- readMVar clientAddr2 - - ConnectionOpened cid ReliableOrdered addr <- receive endpoint ; True <- return $ addr == theirAddr - Received cid' ["ping"] <- receive endpoint ; True <- return $ cid == cid' - - Right conn <- connect endpoint theirAddr ReliableOrdered defaultConnectHints - send conn ["pong"] - - ConnectionClosed cid'' <- receive endpoint ; True <- return $ cid == cid'' - ErrorEvent (TransportError (EventConnectionLost (Just addr') []) _) <- receive endpoint ; True <- return $ addr' == theirAddr - - Left (TransportError SendFailed _) <- send conn ["pong2"] - - return () - - putMVar serverDone () - - -- Client - forkTry $ do - theirAddr <- readMVar serverAddr - - -- First test: close endpoint with one outgoing but no incoming connections - do - Right endpoint <- newEndPoint transport - putMVar clientAddr1 (address endpoint) - - -- Connect to the server, then close the endpoint without disconnecting explicitly - Right _ <- connect endpoint theirAddr ReliableOrdered defaultConnectHints - closeEndPoint endpoint - EndPointClosed <- receive endpoint - return () - - -- Second test: close endpoint with one outgoing and one incoming connection - do - Right endpoint <- newEndPoint transport - putMVar clientAddr2 (address endpoint) - - Right conn <- connect endpoint theirAddr ReliableOrdered defaultConnectHints - send conn ["ping"] - - -- Reply from the server - ConnectionOpened cid ReliableOrdered addr <- receive endpoint ; True <- return $ addr == theirAddr - Received cid' ["pong"] <- receive endpoint ; True <- return $ cid == cid' - - -- Close the endpoint - closeEndPoint endpoint - EndPointClosed <- receive endpoint - - -- Attempt to send should fail with connection closed - Left (TransportError SendFailed _) <- send conn ["ping2"] - - -- An attempt to close the already closed connection should just return - () <- close conn - - -- And so should an attempt to connect - Left (TransportError ConnectFailed _) <- connect endpoint theirAddr ReliableOrdered defaultConnectHints - - return () - - putMVar clientDone () - - mapM_ takeMVar [serverDone, clientDone] - --- Test closeTransport --- --- This tests many of the same things that testEndPoint does, and some more -testCloseTransport :: IO (Either String Transport) -> IO () -testCloseTransport newTransport = do - serverDone <- newEmptyMVar - clientDone <- newEmptyMVar - clientAddr1 <- newEmptyMVar - clientAddr2 <- newEmptyMVar - serverAddr <- newEmptyMVar - - -- Server - forkTry $ do - Right transport <- newTransport - Right endpoint <- newEndPoint transport - putMVar serverAddr (address endpoint) - - -- Client sets up first endpoint - theirAddr1 <- readMVar clientAddr1 - ConnectionOpened cid1 ReliableOrdered addr <- receive endpoint ; True <- return $ addr == theirAddr1 - - -- Client sets up second endpoint - theirAddr2 <- readMVar clientAddr2 - - ConnectionOpened cid2 ReliableOrdered addr' <- receive endpoint ; True <- return $ addr' == theirAddr2 - Received cid2' ["ping"] <- receive endpoint ; True <- return $ cid2' == cid2 - - Right conn <- connect endpoint theirAddr2 ReliableOrdered defaultConnectHints - send conn ["pong"] - - -- Client now closes down its transport. We should receive connection closed messages (we don't know the precise order, however) - evs <- replicateM 3 $ receive endpoint - let expected = [ ConnectionClosed cid1 - , ConnectionClosed cid2 - , ErrorEvent (TransportError (EventConnectionLost (Just theirAddr2) []) "") - ] - True <- return $ any (== expected) (permutations evs) - - -- An attempt to send to the endpoint should now fail - Left (TransportError SendFailed _) <- send conn ["pong2"] - - putMVar serverDone () - - -- Client - forkTry $ do - Right transport <- newTransport - theirAddr <- readMVar serverAddr - - -- Set up endpoint with one outgoing but no incoming connections - Right endpoint1 <- newEndPoint transport - putMVar clientAddr1 (address endpoint1) - - -- Connect to the server, then close the endpoint without disconnecting explicitly - Right _ <- connect endpoint1 theirAddr ReliableOrdered defaultConnectHints - - -- Set up an endpoint with one outgoing and out incoming connection - Right endpoint2 <- newEndPoint transport - putMVar clientAddr2 (address endpoint2) - - Right conn <- connect endpoint2 theirAddr ReliableOrdered defaultConnectHints - send conn ["ping"] - - -- Reply from the server - ConnectionOpened cid ReliableOrdered addr <- receive endpoint2 ; True <- return $ addr == theirAddr - Received cid' ["pong"] <- receive endpoint2 ; True <- return $ cid == cid' - - -- Now shut down the entire transport - closeTransport transport - - -- Both endpoints should report that they have been closed - EndPointClosed <- receive endpoint1 - EndPointClosed <- receive endpoint2 - - -- Attempt to send should fail with connection closed - Left (TransportError SendFailed _) <- send conn ["ping2"] - - -- An attempt to close the already closed connection should just return - () <- close conn - - -- And so should an attempt to connect on either endpoint - Left (TransportError ConnectFailed _) <- connect endpoint1 theirAddr ReliableOrdered defaultConnectHints - Left (TransportError ConnectFailed _) <- connect endpoint2 theirAddr ReliableOrdered defaultConnectHints - - -- And finally, so should an attempt to create a new endpoint - Left (TransportError NewEndPointFailed _) <- newEndPoint transport - - putMVar clientDone () - - mapM_ takeMVar [serverDone, clientDone] - --- | Remote node attempts to connect to a closed local endpoint -testConnectClosedEndPoint :: Transport -> IO () -testConnectClosedEndPoint transport = do - serverAddr <- newEmptyMVar - serverClosed <- newEmptyMVar - clientDone <- newEmptyMVar - - -- Server - forkTry $ do - Right endpoint <- newEndPoint transport - putMVar serverAddr (address endpoint) - - closeEndPoint endpoint - putMVar serverClosed () - - -- Client - forkTry $ do - Right endpoint <- newEndPoint transport - readMVar serverClosed - - Left (TransportError ConnectNotFound _) <- readMVar serverAddr >>= \addr -> connect endpoint addr ReliableOrdered defaultConnectHints - - putMVar clientDone () - - takeMVar clientDone - --- | We should receive an exception when doing a 'receive' after we have been --- notified that an endpoint has been closed -testExceptionOnReceive :: IO (Either String Transport) -> IO () -testExceptionOnReceive newTransport = do - Right transport <- newTransport - - -- Test one: when we close an endpoint specifically - Right endpoint1 <- newEndPoint transport - closeEndPoint endpoint1 - EndPointClosed <- receive endpoint1 - Left _ <- trySome (receive endpoint1 >>= evaluate) - - -- Test two: when we close the entire transport - Right endpoint2 <- newEndPoint transport - closeTransport transport - EndPointClosed <- receive endpoint2 - Left _ <- trySome (receive endpoint2 >>= evaluate) - - return () - --- | Test what happens when the argument to 'send' is an exceptional value -testSendException :: IO (Either String Transport) -> IO () -testSendException newTransport = do - Right transport <- newTransport - Right endpoint1 <- newEndPoint transport - Right endpoint2 <- newEndPoint transport - - -- Connect endpoint1 to endpoint2 - Right conn <- connect endpoint1 (address endpoint2) ReliableOrdered defaultConnectHints - ConnectionOpened _ _ _ <- receive endpoint2 - - -- Send an exceptional value - Left (TransportError SendFailed _) <- send conn (throw $ userError "uhoh") - - -- This will have been as a failure to send by endpoint1, which will - -- therefore have closed the socket. In turn this will have caused endpoint2 - -- to report that the connection was lost - ErrorEvent (TransportError (EventConnectionLost _ []) _) <- receive endpoint1 - ErrorEvent (TransportError (EventConnectionLost _ [_]) _) <- receive endpoint2 - - -- A new connection will re-establish the connection - Right conn2 <- connect endpoint1 (address endpoint2) ReliableOrdered defaultConnectHints - send conn2 ["ping"] - close conn2 - - ConnectionOpened _ _ _ <- receive endpoint2 - Received _ ["ping"] <- receive endpoint2 - ConnectionClosed _ <- receive endpoint2 - - return () - --- | If threads get killed while executing a 'connect', 'send', or 'close', this --- should not affect other threads. --- --- The intention of this test is to see what happens when a asynchronous --- exception happes _while executing a send_. This is exceedingly difficult to --- guarantee, however. Hence we run a large number of tests and insert random --- thread delays -- and even then it might not happen. Moreover, it will only --- happen when we run on multiple cores. -testKill :: IO (Either String Transport) -> Int -> IO () -testKill newTransport numThreads = do - Right transport1 <- newTransport - Right transport2 <- newTransport - Right endpoint1 <- newEndPoint transport1 - Right endpoint2 <- newEndPoint transport2 - - threads <- replicateM numThreads . forkIO $ do - randomThreadDelay 100 - bracket (connect endpoint1 (address endpoint2) ReliableOrdered defaultConnectHints) - -- Note that we should not insert a randomThreadDelay into the - -- exception handler itself as this means that the exception handler - -- could be interrupted and we might not close - (\(Right conn) -> close conn) - (\(Right conn) -> do randomThreadDelay 100 - Right () <- send conn ["ping"] - randomThreadDelay 100) - - numAlive <- newMVar (0 :: Int) - - -- Kill half of those threads - forkIO . forM_ threads $ \tid -> do - shouldKill <- randomIO - if shouldKill - then randomThreadDelay 600 >> killThread tid - else modifyMVar_ numAlive (return . (+ 1)) - - -- Since it is impossible to predict when the kill exactly happens, we don't - -- know how many connects were opened and how many pings were sent. But we - -- should not have any open connections (if we do, collect will throw an - -- error) and we should have at least the number of pings equal to the number - -- of threads we did *not* kill - eventss <- collect endpoint2 Nothing (Just 1000000) - let actualPings = sum . map (length . snd) $ eventss - expectedPings <- takeMVar numAlive - unless (actualPings >= expectedPings) $ - throwIO (userError "Missing pings") - --- print (actualPings, expectedPings) - - --- | Set up conditions with a high likelyhood of "crossing" (for transports --- that multiplex lightweight connections across heavyweight connections) -testCrossing :: Transport -> Int -> IO () -testCrossing transport numRepeats = do - [aAddr, bAddr] <- replicateM 2 newEmptyMVar - [aDone, bDone] <- replicateM 2 newEmptyMVar - [aTimeout, bTimeout] <- replicateM 2 newEmptyMVar - go <- newEmptyMVar - - let hints = defaultConnectHints { - connectTimeout = Just 5000000 - } - - -- A - forkTry $ do - Right endpoint <- newEndPoint transport - putMVar aAddr (address endpoint) - theirAddress <- readMVar bAddr - - replicateM_ numRepeats $ do - takeMVar go >> yield - -- Because we are creating lots of connections, it's possible that - -- connect times out (for instance, in the TCP transport, - -- Network.Socket.connect may time out). We shouldn't regard this as an - -- error in the Transport, though. - connectResult <- connect endpoint theirAddress ReliableOrdered hints - case connectResult of - Right conn -> close conn - Left (TransportError ConnectTimeout _) -> putMVar aTimeout () - Left (TransportError ConnectFailed _) -> readMVar bTimeout - Left err -> throwIO . userError $ "testCrossed: " ++ show err - putMVar aDone () - - -- B - forkTry $ do - Right endpoint <- newEndPoint transport - putMVar bAddr (address endpoint) - theirAddress <- readMVar aAddr - - replicateM_ numRepeats $ do - takeMVar go >> yield - connectResult <- connect endpoint theirAddress ReliableOrdered hints - case connectResult of - Right conn -> close conn - Left (TransportError ConnectTimeout _) -> putMVar bTimeout () - Left (TransportError ConnectFailed _) -> readMVar aTimeout - Left err -> throwIO . userError $ "testCrossed: " ++ show err - putMVar bDone () - - -- Driver - forM_ [1 .. numRepeats] $ \_i -> do - -- putStrLn $ "Round " ++ show _i - tryTakeMVar aTimeout - tryTakeMVar bTimeout - putMVar go () - putMVar go () - takeMVar aDone - takeMVar bDone - --- Transport tests -testTransport :: IO (Either String Transport) -> IO () -testTransport newTransport = do - Right transport <- newTransport - runTests - [ ("PingPong", testPingPong transport numPings) - , ("EndPoints", testEndPoints transport numPings) - , ("Connections", testConnections transport numPings) - , ("CloseOneConnection", testCloseOneConnection transport numPings) - , ("CloseOneDirection", testCloseOneDirection transport numPings) - , ("CloseReopen", testCloseReopen transport numPings) - , ("ParallelConnects", testParallelConnects transport numPings) - , ("SendAfterClose", testSendAfterClose transport 100) - , ("Crossing", testCrossing transport 10) - , ("CloseTwice", testCloseTwice transport 100) - , ("ConnectToSelf", testConnectToSelf transport numPings) - , ("ConnectToSelfTwice", testConnectToSelfTwice transport numPings) - , ("CloseSelf", testCloseSelf newTransport) - , ("CloseEndPoint", testCloseEndPoint transport numPings) - , ("CloseTransport", testCloseTransport newTransport) - , ("ConnectClosedEndPoint", testConnectClosedEndPoint transport) - , ("ExceptionOnReceive", testExceptionOnReceive newTransport) - , ("SendException", testSendException newTransport) - , ("Kill", testKill newTransport 1000) - ] - where - numPings = 10000 :: Int diff --git a/tests/Traced.hs b/tests/Traced.hs deleted file mode 100644 --- a/tests/Traced.hs +++ /dev/null @@ -1,199 +0,0 @@ --- | Add tracing to the IO monad (see examples). --- --- [Usage] --- --- > {-# LANGUAGE RebindableSyntax #-} --- > import Prelude hiding (catch, (>>=), (>>), return, fail) --- > import Traced --- --- [Example] --- --- > test1 :: IO Int --- > test1 = do --- > Left x <- return (Left 1 :: Either Int Int) --- > putStrLn "Hello world" --- > Right y <- return (Left 2 :: Either Int Int) --- > return (x + y) --- --- outputs --- --- > Hello world --- > *** Exception: user error (Pattern match failure in do expression at Traced.hs:187:3-9) --- > Trace: --- > 0 Left 2 --- > 1 Left 1 --- --- [Guards] --- --- Use the following idiom instead of using 'Control.Monad.guard': --- --- > test2 :: IO Int --- > test2 = do --- > Left x <- return (Left 1 :: Either Int Int) --- > True <- return (x == 3) --- > return x --- --- The advantage of this idiom is that it gives you line number information when the guard fails: --- --- > *Traced> test2 --- > *** Exception: user error (Pattern match failure in do expression at Traced.hs:193:3-6) --- > Trace: --- > 0 Left 1 -module Traced ( MonadS(..) - , return - , (>>=) - , (>>) - , fail - , ifThenElse - , Showable(..) - , Traceable(..) - , traceShow - ) where - -import Prelude hiding - ( (>>=) - , return - , fail - , (>>) -#if ! MIN_VERSION_base(4,6,0) - , catch -#endif - ) -import qualified Prelude -import Control.Exception (catches, Handler(..), SomeException, throwIO, Exception(..), IOException) -import Control.Applicative ((<$>)) -import Data.Typeable (Typeable) -import Data.Maybe (catMaybes) -import Data.ByteString (ByteString) -import Data.Int (Int32) -import Control.Concurrent.MVar (MVar) - --------------------------------------------------------------------------------- --- MonadS class -- --------------------------------------------------------------------------------- - --- | Like 'Monad' but bind is only defined for 'Trace'able instances -class MonadS m where - returnS :: a -> m a - bindS :: Traceable a => m a -> (a -> m b) -> m b - failS :: String -> m a - seqS :: m a -> m b -> m b - --- | Redefinition of 'Prelude.>>=' -(>>=) :: (MonadS m, Traceable a) => m a -> (a -> m b) -> m b -(>>=) = bindS - --- | Redefinition of 'Prelude.>>' -(>>) :: MonadS m => m a -> m b -> m b -(>>) = seqS - --- | Redefinition of 'Prelude.return' -return :: MonadS m => a -> m a -return = returnS - --- | Redefinition of 'Prelude.fail' -fail :: MonadS m => String -> m a -fail = failS - --------------------------------------------------------------------------------- --- Trace typeclass (for adding elements to a trace -- --------------------------------------------------------------------------------- - -data Showable = forall a. Show a => Showable a - -instance Show Showable where - show (Showable x) = show x - -mapShowable :: (forall a. Show a => a -> Showable) -> Showable -> Showable -mapShowable f (Showable x) = f x - -traceShow :: Show a => a -> Maybe Showable -traceShow = Just . Showable - -class Traceable a where - trace :: a -> Maybe Showable - -instance (Traceable a, Traceable b) => Traceable (Either a b) where - trace (Left x) = (mapShowable $ Showable . (Left :: forall c. c -> Either c ())) <$> trace x - trace (Right y) = (mapShowable $ Showable . (Right :: forall c. c -> Either () c)) <$> trace y - -instance (Traceable a, Traceable b) => Traceable (a, b) where - trace (x, y) = case (trace x, trace y) of - (Nothing, Nothing) -> Nothing - (Just t1, Nothing) -> traceShow t1 - (Nothing, Just t2) -> traceShow t2 - (Just t1, Just t2) -> traceShow (t1, t2) - -instance (Traceable a, Traceable b, Traceable c) => Traceable (a, b, c) where - trace (x, y, z) = case (trace x, trace y, trace z) of - (Nothing, Nothing, Nothing) -> Nothing - (Just t1, Nothing, Nothing) -> traceShow t1 - (Nothing, Just t2, Nothing) -> traceShow t2 - (Just t1, Just t2, Nothing) -> traceShow (t1, t2) - (Nothing, Nothing, Just t3) -> traceShow t3 - (Just t1, Nothing, Just t3) -> traceShow (t1, t3) - (Nothing, Just t2, Just t3) -> traceShow (t2, t3) - (Just t1, Just t2, Just t3) -> traceShow (t1, t2, t3) - -instance Traceable a => Traceable (Maybe a) where - trace Nothing = traceShow (Nothing :: Maybe ()) - trace (Just x) = mapShowable (Showable . Just) <$> trace x - -instance Traceable a => Traceable [a] where - trace = traceShow . catMaybes . map trace - -instance Traceable () where - trace = const Nothing - -instance Traceable Int where - trace = traceShow - -instance Traceable Int32 where - trace = traceShow - -instance Traceable Bool where - trace = const Nothing - -instance Traceable ByteString where - trace = traceShow - -instance Traceable (MVar a) where - trace = const Nothing - -instance Traceable [Char] where - trace = traceShow - -instance Traceable IOException where - trace = traceShow - --------------------------------------------------------------------------------- --- IO instance for MonadS -- --------------------------------------------------------------------------------- - -data TracedException = TracedException [String] SomeException - deriving Typeable - -instance Exception TracedException - --- | Add tracing to 'IO' (see examples) -instance MonadS IO where - returnS = Prelude.return - bindS = \x f -> x Prelude.>>= \a -> catches (f a) (traceHandlers a) - failS = Prelude.fail - seqS = (Prelude.>>) - -instance Show TracedException where - show (TracedException ts ex) = - show ex ++ "\nTrace:\n" ++ unlines (map (\(i, t) -> show i ++ "\t" ++ t) (zip ([0..] :: [Int]) (take 10 . reverse $ ts))) - -traceHandlers :: Traceable a => a -> [Handler b] -traceHandlers a = case trace a of - Nothing -> [ Handler $ \ex -> throwIO (ex :: SomeException) ] - Just t -> [ Handler $ \(TracedException ts ex) -> throwIO $ TracedException (show t : ts) ex - , Handler $ \ex -> throwIO $ TracedException [show t] (ex :: SomeException) - ] - --- | Definition of 'ifThenElse' for use with RebindableSyntax -ifThenElse :: Bool -> a -> a -> a -ifThenElse True x _ = x -ifThenElse False _ y = y