network-transport-tcp 0.2.0.3 → 0.3.0
raw patch · 7 files changed
+686/−2015 lines, 7 filesdep +network-transport-tcpdep +network-transport-testsdep −ansi-terminaldep −mtldep −randomdep ~networkdep ~network-transportPVP ok
version bump matches the API change (PVP)
Dependencies added: network-transport-tcp, network-transport-tests
Dependencies removed: ansi-terminal, mtl, random
Dependency ranges changed: network, network-transport
API changes (from Hackage documentation)
- Network.Transport.TCP: ControlResponse :: ControlHeader
- Network.Transport.TCP: RequestConnectionId :: ControlHeader
- Network.Transport.TCP: firstNonReservedConnectionId :: ConnectionId
+ Network.Transport.TCP: CreatedNewConnection :: ControlHeader
+ Network.Transport.TCP: firstNonReservedHeavyweightConnectionId :: HeavyweightConnectionId
+ Network.Transport.TCP: firstNonReservedLightweightConnectionId :: LightweightConnectionId
+ Network.Transport.TCP: type LightweightConnectionId = Word32
- Network.Transport.TCP: type EndPointId = Int32
+ Network.Transport.TCP: type EndPointId = Word32
Files
- network-transport-tcp.cabal +13/−25
- src/Network/Transport/TCP.hs +472/−512
- src/Network/Transport/TCP/Internal.hs +27/−24
- tests/TestAuxiliary.hs +0/−111
- tests/TestTCP.hs +174/−180
- tests/TestTransport.hs +0/−964
- tests/Traced.hs +0/−199
network-transport-tcp.cabal view
@@ -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
src/Network/Transport/TCP.hs view
@@ -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 -------------------------------------------------------------------------------
src/Network/Transport/TCP/Internal.hs view
@@ -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))
− tests/TestAuxiliary.hs
@@ -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
tests/TestTCP.hs view
@@ -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)
− tests/TestTransport.hs
@@ -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
− tests/Traced.hs
@@ -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