diff --git a/distributed-process.cabal b/distributed-process.cabal --- a/distributed-process.cabal +++ b/distributed-process.cabal @@ -1,5 +1,5 @@ Name: distributed-process -Version: 0.4.0.2 +Version: 0.4.1 Cabal-Version: >=1.8 Build-Type: Simple License: BSD3 @@ -39,7 +39,7 @@ Library Build-Depends: base >= 4.4 && < 5, - binary >= 0.5 && < 0.6, + binary >= 0.5 && < 0.7, network-transport >= 0.3 && < 0.4, stm >= 2.3 && < 2.5, transformers >= 0.2 && < 0.4, @@ -93,9 +93,11 @@ distributed-process, network-transport >= 0.3 && < 0.4, network-transport-tcp >= 0.3 && < 0.4, - binary >= 0.5 && < 0.6, - network >= 2.3 && < 2.5 - Other-modules: TestAuxiliary + binary >= 0.5 && < 0.7, + network >= 2.3 && < 2.5, + HUnit >= 1.2 && < 1.3, + test-framework >= 0.6 && < 0.7, + test-framework-hunit >= 0.2 && < 0.3 Extensions: CPP, ScopedTypeVariables, DeriveDataTypeable, @@ -114,8 +116,10 @@ network-transport >= 0.3 && < 0.4, network-transport-tcp >= 0.3 && < 0.4, bytestring >= 0.9 && < 0.11, - network >= 2.3 && < 2.5 - Other-modules: TestAuxiliary + network >= 2.3 && < 2.5, + HUnit >= 1.2 && < 1.3, + test-framework >= 0.6 && < 0.7, + test-framework-hunit >= 0.2 && < 0.3 Extensions: CPP, ScopedTypeVariables ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind @@ -127,7 +131,7 @@ distributed-process, network-transport-tcp >= 0.3 && < 0.4, bytestring >= 0.9 && < 0.11, - binary >= 0.5 && < 0.6 + binary >= 0.5 && < 0.7 else buildable: False Main-Is: benchmarks/Throughput.hs @@ -140,7 +144,7 @@ distributed-process, network-transport-tcp >= 0.3 && < 0.4, bytestring >= 0.9 && < 0.11, - binary >= 0.5 && < 0.6 + binary >= 0.5 && < 0.7 else buildable: False Main-Is: benchmarks/Latency.hs @@ -153,7 +157,7 @@ distributed-process, network-transport-tcp >= 0.3 && < 0.4, bytestring >= 0.9 && < 0.11, - binary >= 0.5 && < 0.6 + binary >= 0.5 && < 0.7 else buildable: False Main-Is: benchmarks/Channels.hs @@ -166,7 +170,7 @@ distributed-process, network-transport-tcp >= 0.3 && < 0.4, bytestring >= 0.9 && < 0.11, - binary >= 0.5 && < 0.6 + binary >= 0.5 && < 0.7 else buildable: False Main-Is: benchmarks/Spawns.hs diff --git a/src/Control/Distributed/Process.hs b/src/Control/Distributed/Process.hs --- a/src/Control/Distributed/Process.hs +++ b/src/Control/Distributed/Process.hs @@ -19,6 +19,7 @@ -- * Basic messaging , send , expect + , expectTimeout -- * Channels , ReceivePort , SendPort @@ -26,6 +27,7 @@ , newChan , sendChan , receiveChan + , receiveChanTimeout , mergePortsBiased , mergePortsRR -- * Advanced messaging @@ -42,6 +44,7 @@ , call , terminate , ProcessTerminationException(..) + , ProcessRegistrationException(..) , SpawnRef , getSelfPid , getSelfNode @@ -75,11 +78,13 @@ , say -- * Registry , register + , reregister , unregister , whereis , nsend - , registerRemote - , unregisterRemote + , registerRemoteAsync + , reregisterRemoteAsync + , unregisterRemoteAsync , whereisRemoteAsync , nsendRemote , WhereIsReply(..) @@ -91,7 +96,6 @@ , bracket_ , finally -- * Auxiliary API - , expectTimeout , spawnAsync , spawnSupervised , spawnLink @@ -134,6 +138,7 @@ , ProcessLinkException(..) , NodeLinkException(..) , PortLinkException(..) + , ProcessRegistrationException(..) , DiedReason(..) , SpawnRef(..) , DidSpawn(..) @@ -196,11 +201,13 @@ , say -- Registry , register + , reregister , unregister , whereis , nsend - , registerRemote - , unregisterRemote + , registerRemoteAsync + , reregisterRemoteAsync + , unregisterRemoteAsync , whereisRemoteAsync , nsendRemote -- Closures @@ -215,6 +222,7 @@ , finally -- Auxiliary API , expectTimeout + , receiveChanTimeout , spawnAsync -- Reconnecting , reconnect diff --git a/src/Control/Distributed/Process/Internal/Primitives.hs b/src/Control/Distributed/Process/Internal/Primitives.hs --- a/src/Control/Distributed/Process/Internal/Primitives.hs +++ b/src/Control/Distributed/Process/Internal/Primitives.hs @@ -35,11 +35,13 @@ , say -- * Registry , register + , reregister , unregister , whereis , nsend - , registerRemote - , unregisterRemote + , registerRemoteAsync + , reregisterRemoteAsync + , unregisterRemoteAsync , whereisRemoteAsync , nsendRemote -- * Closures @@ -54,6 +56,7 @@ , finally -- * Auxiliary API , expectTimeout + , receiveChanTimeout , spawnAsync , linkNode , linkPort @@ -74,6 +77,8 @@ import Data.Time.Clock (getCurrentTime) import Data.Time.Format (formatTime) import System.Locale (defaultTimeLocale) +import System.Timeout (timeout) +import Control.Monad (when) import Control.Monad.Reader (ask) import Control.Monad.IO.Class (MonadIO, liftIO) import Control.Applicative ((<$>)) @@ -87,6 +92,7 @@ import Control.Concurrent.Chan (writeChan) import Control.Concurrent.STM ( STM + , TVar , atomically , orElse , newTVar @@ -124,6 +130,8 @@ , DidUnlinkNode(..) , DidUnlinkPort(..) , WhereIsReply(..) + , RegisterReply(..) + , ProcessRegistrationException(..) , createMessage , runLocalProcess , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect) @@ -179,7 +187,7 @@ let sport = SendPort cid chan <- liftIO newTQueueIO chan' <- mkWeakTQueue chan $ finalizer (processState proc) lcid - let rport = ReceivePortSingle chan + let rport = ReceivePort $ readTQueue chan let tch = TypedChannel chan' return ( (channelCounter ^: (+ 1)) . (typedChannelWithId lcid ^= Just tch) @@ -188,7 +196,7 @@ ) where finalizer :: StrictMVar LocalProcessState -> LocalSendPortId -> IO () - finalizer processState lcid = modifyMVar_ processState $ + finalizer st lcid = modifyMVar_ st $ return . (typedChannelWithId lcid ^= Nothing) -- | Send a message on a typed channel @@ -204,34 +212,40 @@ -- | Wait for a message on a typed channel receiveChan :: Serializable a => ReceivePort a -> Process a receiveChan = liftIO . atomically . receiveSTM - where - receiveSTM :: ReceivePort a -> STM a - receiveSTM (ReceivePortSingle c) = - readTQueue c - receiveSTM (ReceivePortBiased ps) = - foldr1 orElse (map receiveSTM ps) - receiveSTM (ReceivePortRR psVar) = do - ps <- readTVar psVar - a <- foldr1 orElse (map receiveSTM ps) - writeTVar psVar (rotate ps) - return a - rotate :: [a] -> [a] - rotate [] = [] - rotate (x:xs) = xs ++ [x] +-- | Like 'receiveChan' but with a timeout. If the timeout is 0, do a +-- non-blocking check for a message. +receiveChanTimeout :: Serializable a => Int -> ReceivePort a -> Process (Maybe a) +receiveChanTimeout 0 ch = liftIO . atomically $ + (Just <$> receiveSTM ch) `orElse` return Nothing +receiveChanTimeout n ch = liftIO . timeout n . atomically $ + receiveSTM ch -- | Merge a list of typed channels. -- -- The result port is left-biased: if there are messages available on more -- than one port, the first available message is returned. mergePortsBiased :: Serializable a => [ReceivePort a] -> Process (ReceivePort a) -mergePortsBiased = return . ReceivePortBiased +mergePortsBiased = return . ReceivePort. foldr1 orElse . map receiveSTM -- | Like 'mergePortsBiased', but with a round-robin scheduler (rather than -- left-biased) mergePortsRR :: Serializable a => [ReceivePort a] -> Process (ReceivePort a) -mergePortsRR ps = liftIO . atomically $ ReceivePortRR <$> newTVar ps +mergePortsRR = \ps -> do + psVar <- liftIO . atomically $ newTVar (map receiveSTM ps) + return $ ReceivePort (rr psVar) + where + rotate :: [a] -> [a] + rotate [] = [] + rotate (x:xs) = xs ++ [x] + rr :: TVar [STM a] -> STM a + rr psVar = do + ps <- readTVar psVar + a <- foldr1 orElse ps + writeTVar psVar (rotate ps) + return a + -------------------------------------------------------------------------------- -- Advanced messaging -- -------------------------------------------------------------------------------- @@ -328,6 +342,25 @@ -- | Link to a remote process (asynchronous) -- +-- When process A links to process B (that is, process A calls +-- @link pidB@) then an asynchronous exception will be thrown to process A +-- when process B terminates (normally or abnormally), or when process A gets +-- disconnected from process B. Although it is /technically/ possible to catch +-- these exceptions, chances are if you find yourself trying to do so you should +-- probably be using 'monitor' rather than 'link'. In particular, code such as +-- +-- > link pidB -- Link to process B +-- > expect -- Wait for a message from process B +-- > unlink pidB -- Unlink again +-- +-- doesn't quite do what one might expect: if process B sends a message to +-- process A, and /subsequently terminates/, then process A might or might not +-- be terminated too, depending on whether the exception is thrown before or +-- after the 'unlink' (i.e., this code has a race condition). +-- +-- Linking is all-or-nothing: A is either linked to B, or it's not. A second +-- call to 'link' has no effect. +-- -- Note that 'link' provides unidirectional linking (see 'spawnSupervised'). -- Linking makes no distinction between normal and abnormal termination of -- the remote process. @@ -335,6 +368,17 @@ link = sendCtrlMsg Nothing . Link . ProcessIdentifier -- | Monitor another process (asynchronous) +-- +-- When process A monitors process B (that is, process A calls +-- @monitor pidB@) then process A will receive a 'ProcessMonitorNotification' +-- when process B terminates (normally or abnormally), or when process A gets +-- disconnected from process B. You receive this message like any other (using +-- 'expect'); the notification includes a reason ('DiedNormal', 'DiedException', +-- 'DiedDisconnect', etc.). +-- +-- Every call to 'monitor' returns a new monitor reference 'MonitorRef'; if +-- multiple monitors are set up, multiple notifications will be delivered +-- and monitors can be disabled individually using 'unmonitor'. monitor :: ProcessId -> Process MonitorRef monitor = monitor' . ProcessIdentifier @@ -429,7 +473,7 @@ -- | Like 'expect' but with a timeout expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a) -expectTimeout timeout = receiveTimeout timeout [match return] +expectTimeout n = receiveTimeout n [match return] -- | Asynchronous version of 'spawn' -- @@ -500,28 +544,68 @@ -------------------------------------------------------------------------------- -- | Register a process with the local registry (asynchronous). +-- This version will wait until a response is gotten from the +-- management process. The name must not already be registered. +-- The process need not be on this node. +-- A bad registration will result in a 'ProcessRegistrationException' -- -- The process to be registered does not have to be local itself. register :: String -> ProcessId -> Process () -register label pid = - sendCtrlMsg Nothing (Register label (Just pid)) +register = registerImpl False +-- | Like 'register', but will replace an existing registration. +-- The name must already be registered. +reregister :: String -> ProcessId -> Process () +reregister = registerImpl True + +registerImpl :: Bool -> String -> ProcessId -> Process () +registerImpl force label pid = do + mynid <- getSelfNode + sendCtrlMsg Nothing (Register label mynid (Just pid) force) + receiveWait [ matchIf (\(RegisterReply label' _) -> label == label') + (\(RegisterReply _ ok) -> handleRegistrationReply label ok) + ] + -- | Register a process with a remote registry (asynchronous). -- -- The process to be registered does not have to live on the same remote node. -registerRemote :: NodeId -> String -> ProcessId -> Process () -registerRemote nid label pid = - sendCtrlMsg (Just nid) (Register label (Just pid)) +-- Reply wil come in the form of a 'RegisterReply' message +-- +-- See comments in 'whereisRemoteAsync' +registerRemoteAsync :: NodeId -> String -> ProcessId -> Process () +registerRemoteAsync nid label pid = + sendCtrlMsg (Just nid) (Register label nid (Just pid) False) +reregisterRemoteAsync :: NodeId -> String -> ProcessId -> Process () +reregisterRemoteAsync nid label pid = + sendCtrlMsg (Just nid) (Register label nid (Just pid) True) + -- | Remove a process from the local registry (asynchronous). +-- This version will wait until a response is gotten from the +-- management process. The name must already be registered. unregister :: String -> Process () -unregister label = - sendCtrlMsg Nothing (Register label Nothing) +unregister label = do + mynid <- getSelfNode + sendCtrlMsg Nothing (Register label mynid Nothing False) + receiveWait [ matchIf (\(RegisterReply label' _) -> label == label') + (\(RegisterReply _ ok) -> handleRegistrationReply label ok) + ] +-- | Deal with the result from an attempted registration or unregistration +-- by throwing an exception if necessary +handleRegistrationReply :: String -> Bool -> Process () +handleRegistrationReply label ok = + when (not ok) $ + liftIO $ throwIO $ ProcessRegistrationException label + -- | Remove a process from a remote registry (asynchronous). -unregisterRemote :: NodeId -> String -> Process () -unregisterRemote nid label = - sendCtrlMsg (Just nid) (Register label Nothing) +-- +-- Reply wil come in the form of a 'RegisterReply' message +-- +-- See comments in 'whereisRemoteAsync' +unregisterRemoteAsync :: NodeId -> String -> Process () +unregisterRemoteAsync nid label = + sendCtrlMsg (Just nid) (Register label nid Nothing False) -- | Query the local process registry whereis :: String -> Process (Maybe ProcessId) diff --git a/src/Control/Distributed/Process/Internal/Types.hs b/src/Control/Distributed/Process/Internal/Types.hs --- a/src/Control/Distributed/Process/Internal/Types.hs +++ b/src/Control/Distributed/Process/Internal/Types.hs @@ -39,6 +39,7 @@ , ProcessLinkException(..) , NodeLinkException(..) , PortLinkException(..) + , ProcessRegistrationException(..) , DiedReason(..) , DidUnmonitor(..) , DidUnlinkProcess(..) @@ -47,6 +48,7 @@ , SpawnRef(..) , DidSpawn(..) , WhereIsReply(..) + , RegisterReply(..) -- * Node controller internal data types , NCMsg(..) , ProcessSignal(..) @@ -82,9 +84,9 @@ import Control.Exception (Exception) import Control.Concurrent (ThreadId) import Control.Concurrent.Chan (Chan) -import Control.Concurrent.STM (TVar) +import Control.Concurrent.STM (STM) import qualified Network.Transport as NT (EndPoint, EndPointAddress, Connection) -import Control.Applicative (Applicative, (<$>), (<*>)) +import Control.Applicative (Applicative, Alternative, (<$>), (<*>)) import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT) import Control.Monad.IO.Class (MonadIO) import Control.Distributed.Process.Serializable @@ -262,6 +264,13 @@ deriving (Typeable, Binary, Show, Eq, Ord) -- | The receive end of a typed channel (not serializable) +-- +-- Note that 'ReceivePort' implements 'Functor', 'Applicative', 'Alternative' +-- and 'Monad'. This is especially useful when merging receive ports. +newtype ReceivePort a = ReceivePort { receiveSTM :: STM a } + deriving (Typeable, Functor, Applicative, Alternative, Monad) + +{- data ReceivePort a = -- | A single receive port ReceivePortSingle (TQueue a) @@ -270,6 +279,7 @@ -- | A round-robin combination of receive ports | ReceivePortRR (TVar [ReceivePort a]) deriving Typeable +-} -------------------------------------------------------------------------------- -- Messages -- @@ -351,9 +361,17 @@ PortLinkException !SendPortId !DiedReason deriving (Typeable, Show) +-- | Exception thrown when a process attempts to register +-- a process under an already-registered name or to +-- unregister a name that hasn't been registered +data ProcessRegistrationException = + ProcessRegistrationException !String + deriving (Typeable, Show) + instance Exception ProcessLinkException instance Exception NodeLinkException instance Exception PortLinkException +instance Exception ProcessRegistrationException -- | Why did a process die? data DiedReason = @@ -398,6 +416,10 @@ data WhereIsReply = WhereIsReply String (Maybe ProcessId) deriving (Show, Typeable) +-- | (Asynchronous) reply from 'register' and 'unregister' +data RegisterReply = RegisterReply String Bool + deriving (Show, Typeable) + -------------------------------------------------------------------------------- -- Node controller internal data types -- -------------------------------------------------------------------------------- @@ -418,7 +440,7 @@ | Died Identifier !DiedReason | Spawn !(Closure (Process ())) !SpawnRef | WhereIs !String - | Register !String !(Maybe ProcessId) -- Use 'Nothing' to unregister + | Register !String !NodeId !(Maybe ProcessId) !Bool -- Use 'Nothing' to unregister, use True to force reregister | NamedSend !String !Message deriving Show @@ -462,7 +484,7 @@ put (Died who reason) = putWord8 4 >> put who >> put reason put (Spawn proc ref) = putWord8 5 >> put proc >> put ref put (WhereIs label) = putWord8 6 >> put label - put (Register label pid) = putWord8 7 >> put label >> put pid + put (Register label nid pid force) = putWord8 7 >> put label >> put nid >> put pid >> put force put (NamedSend label msg) = putWord8 8 >> put label >> put (messageToPayload msg) get = do header <- getWord8 @@ -474,7 +496,7 @@ 4 -> Died <$> get <*> get 5 -> Spawn <$> get <*> get 6 -> WhereIs <$> get - 7 -> Register <$> get <*> get + 7 -> Register <$> get <*> get <*> get <*> get 8 -> NamedSend <$> get <*> (payloadToMessage <$> get) _ -> fail "ProcessSignal.get: invalid" @@ -517,6 +539,10 @@ instance Binary WhereIsReply where put (WhereIsReply label mPid) = put label >> put mPid get = WhereIsReply <$> get <*> get + +instance Binary RegisterReply where + put (RegisterReply label ok) = put label >> put ok + get = RegisterReply <$> get <*> get -------------------------------------------------------------------------------- -- Accessors -- diff --git a/src/Control/Distributed/Process/Node.hs b/src/Control/Distributed/Process/Node.hs --- a/src/Control/Distributed/Process/Node.hs +++ b/src/Control/Distributed/Process/Node.hs @@ -24,6 +24,8 @@ import qualified Data.Map as Map ( empty , toList + , fromList + , filter , partitionWithKey , elems , filterWithKey @@ -31,7 +33,7 @@ import Data.Set (Set) import qualified Data.Set as Set (empty, insert, delete, member) import Data.Foldable (forM_) -import Data.Maybe (isJust) +import Data.Maybe (isJust, isNothing, catMaybes) import Data.Typeable (Typeable) import Control.Category ((>>>)) import Control.Applicative ((<$>)) @@ -118,6 +120,7 @@ , nodeOf , SendPortId(..) , typedChannelWithId + , RegisterReply(..) , WhereIsReply(..) , messageToPayload , payloadToMessage @@ -437,8 +440,9 @@ _links :: !(Map Identifier (Set ProcessId)) -- Mapping from remote processes to monitoring local processes , _monitors :: !(Map Identifier (Set (ProcessId, MonitorRef))) - -- Process registry - , _registry :: !(Map String ProcessId) + -- Process registry: names and where they live, mapped to the PIDs + , _registeredHere :: !(Map String ProcessId) + , _registeredOnNodes :: !(Map ProcessId [(NodeId,Int)]) } newtype NC a = NC { unNC :: StateT NCState (ReaderT LocalNode IO) a } @@ -447,7 +451,8 @@ initNCState :: NCState initNCState = NCState { _links = Map.empty , _monitors = Map.empty - , _registry = Map.empty + , _registeredHere = Map.empty + , _registeredOnNodes = Map.empty } -------------------------------------------------------------------------------- @@ -485,8 +490,8 @@ ncEffectDied ident reason NCMsg (ProcessIdentifier from) (Spawn proc ref) -> ncEffectSpawn from proc ref - NCMsg _from (Register label pid) -> - ncEffectRegister label pid + NCMsg (ProcessIdentifier from) (Register label atnode pid force) -> + ncEffectRegister from label atnode pid force NCMsg (ProcessIdentifier from) (WhereIs label) -> ncEffectWhereIs from label NCMsg from (NamedSend label msg') -> @@ -554,6 +559,8 @@ (affectedLinks, unaffectedLinks) <- gets (splitNotif ident . (^. links)) (affectedMons, unaffectedMons) <- gets (splitNotif ident . (^. monitors)) +-- _registry :: !(Map (String,NodeId) ProcessId) + let localOnly = case ident of NodeIdentifier _ -> True ; _ -> False forM_ (Map.toList affectedLinks) $ \(them, uss) -> @@ -564,10 +571,34 @@ forM_ (Map.toList affectedMons) $ \(them, refs) -> forM_ refs $ \(us, ref) -> when (localOnly <= isLocal node (ProcessIdentifier us)) $ - notifyDied us them reason (Just ref) + notifyDied us them reason (Just ref) modify' $ (links ^= unaffectedLinks) . (monitors ^= unaffectedMons) + modify' $ registeredHere ^: Map.filter (\pid -> not $ ident `impliesDeathOf` ProcessIdentifier pid) + + remaining <- fmap Map.toList (gets (^. registeredOnNodes)) >>= + mapM (\(pid,nidlist) -> + case ident `impliesDeathOf` ProcessIdentifier pid of + True -> + do forM_ nidlist $ \(nid,_) -> + when (not $ isLocal node (NodeIdentifier nid)) + (forwardNameDeath node nid) + return Nothing + False -> return $ Just (pid,nidlist) ) + modify' $ registeredOnNodes ^= (Map.fromList (catMaybes remaining)) + where + forwardNameDeath node nid = + liftIO $ sendBinary node + (NodeIdentifier $ localNodeId node) + (NodeIdentifier $ nid) + WithImplicitReconnect + NCMsg + { ctrlMsgSender = NodeIdentifier (localNodeId node) + , ctrlMsgSignal = Died ident reason + } + + -- [Unified: Table 13] ncEffectSpawn :: ProcessId -> Closure (Process ()) -> SpawnRef -> NC () ncEffectSpawn pid cProc ref = do @@ -587,18 +618,71 @@ -- Unified semantics does not explicitly describe how to implement 'register', -- but mentions it's "very similar to nsend" (Table 14) -ncEffectRegister :: String -> Maybe ProcessId -> NC () -ncEffectRegister label mPid = - modify' $ registryFor label ^= mPid - -- An acknowledgement is not necessary. If we want a synchronous register, - -- it suffices to send a whereis requiry immediately after the register - -- (that may not suffice if we do decide for unreliable messaging instead) +-- We send a response indicated if the operation is invalid +ncEffectRegister :: ProcessId -> String -> NodeId -> Maybe ProcessId -> Bool -> NC () +ncEffectRegister from label atnode mPid reregistration = do + node <- ask + currentVal <- gets (^. registeredHereFor label) + isOk <- + case mPid of + Nothing -> -- unregister request + return $ isJust currentVal + Just thepid -> -- register request + do isvalidlocal <- isValidLocalIdentifier (ProcessIdentifier thepid) + return $ (isNothing currentVal /= reregistration) && + (not (isLocal node (ProcessIdentifier thepid) ) || isvalidlocal ) + if isLocal node (NodeIdentifier atnode) + then do when (isOk) $ + do modify' $ registeredHereFor label ^= mPid + updateRemote node currentVal mPid + liftIO $ sendMessage node + (NodeIdentifier (localNodeId node)) + (ProcessIdentifier from) + WithImplicitReconnect + (RegisterReply label isOk) + else let operation = + case reregistration of + True -> flip decList + False -> flip incList + in case mPid of + Nothing -> return () + Just pid -> modify' $ registeredOnNodesFor pid ^: (maybeify $ operation atnode) + where updateRemote node (Just oldval) (Just newval) | processNodeId oldval /= processNodeId newval = + do forward node (processNodeId oldval) (Register label atnode (Just oldval) True) + forward node (processNodeId newval) (Register label atnode (Just newval) False) + updateRemote node Nothing (Just newval) = + forward node (processNodeId newval) (Register label atnode (Just newval) False) + updateRemote node (Just oldval) Nothing = + forward node (processNodeId oldval) (Register label atnode (Just oldval) True) + updateRemote _ _ _ = return () + maybeify f Nothing = unmaybeify $ f [] + maybeify f (Just x) = unmaybeify $ f x + unmaybeify [] = Nothing + unmaybeify x = Just x + incList [] tag = [(tag,1)] + incList ((atag,acount):xs) tag | tag==atag = (atag,acount+1) : xs + incList (x:xs) tag = x : incList xs tag + decList [] _ = [] + decList ((atag,1):xs) tag | atag == tag = xs + decList ((atag,n):xs) tag | atag == tag = (atag,n-1):xs + decList (x:xs) tag = x:decList xs tag + forward node to reg = + when (not $ isLocal node (NodeIdentifier to)) $ + liftIO $ sendBinary node + (ProcessIdentifier from) + (NodeIdentifier to) + WithImplicitReconnect + NCMsg + { ctrlMsgSender = ProcessIdentifier from + , ctrlMsgSignal = reg + } + -- Unified semantics does not explicitly describe 'whereis' ncEffectWhereIs :: ProcessId -> String -> NC () ncEffectWhereIs from label = do node <- ask - mPid <- gets (^. registryFor label) + mPid <- gets (^. registeredHereFor label) liftIO $ sendMessage node (NodeIdentifier (localNodeId node)) (ProcessIdentifier from) @@ -608,9 +692,9 @@ -- [Unified: Table 14] ncEffectNamedSend :: Identifier -> String -> Message -> NC () ncEffectNamedSend from label msg = do - mPid <- gets (^. registryFor label) + node <- ask + mPid <- gets (^. registeredHereFor label) -- If mPid is Nothing, we just ignore the named send (as per Table 14) - node <- ask forM_ mPid $ \pid -> liftIO $ sendPayload node from @@ -660,7 +744,7 @@ destNid (Monitor ref) = Just $ nodeOf (monitorRefIdent ref) destNid (Unmonitor ref) = Just $ nodeOf (monitorRefIdent ref) destNid (Spawn _ _) = Nothing -destNid (Register _ _) = Nothing +destNid (Register _ _ _ _) = Nothing destNid (WhereIs _) = Nothing destNid (NamedSend _ _) = Nothing -- We don't need to forward 'Died' signals; if monitoring/linking is setup, @@ -732,17 +816,23 @@ monitors :: Accessor NCState (Map Identifier (Set (ProcessId, MonitorRef))) monitors = accessor _monitors (\ms st -> st { _monitors = ms }) -registry :: Accessor NCState (Map String ProcessId) -registry = accessor _registry (\ry st -> st { _registry = ry }) +registeredHere :: Accessor NCState (Map String ProcessId) +registeredHere = accessor _registeredHere (\ry st -> st { _registeredHere = ry }) +registeredOnNodes :: Accessor NCState (Map ProcessId [(NodeId, Int)]) +registeredOnNodes = accessor _registeredOnNodes (\ry st -> st { _registeredOnNodes = ry }) + linksFor :: Identifier -> Accessor NCState (Set ProcessId) linksFor ident = links >>> DAC.mapDefault Set.empty ident monitorsFor :: Identifier -> Accessor NCState (Set (ProcessId, MonitorRef)) monitorsFor ident = monitors >>> DAC.mapDefault Set.empty ident -registryFor :: String -> Accessor NCState (Maybe ProcessId) -registryFor ident = registry >>> DAC.mapMaybe ident +registeredHereFor :: String -> Accessor NCState (Maybe ProcessId) +registeredHereFor ident = registeredHere >>> DAC.mapMaybe ident + +registeredOnNodesFor :: ProcessId -> Accessor NCState (Maybe [(NodeId,Int)]) +registeredOnNodesFor ident = registeredOnNodes >>> DAC.mapMaybe ident -- | @splitNotif ident@ splits a notifications map into those -- notifications that should trigger when 'ident' fails and those links that diff --git a/tests/TestAuxiliary.hs b/tests/TestAuxiliary.hs deleted file mode 100644 --- a/tests/TestAuxiliary.hs +++ /dev/null @@ -1,75 +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.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) - --- | 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) diff --git a/tests/TestCH.hs b/tests/TestCH.hs --- a/tests/TestCH.hs +++ b/tests/TestCH.hs @@ -4,10 +4,11 @@ import Prelude hiding (catch) #endif + import Data.Binary (Binary(..)) import Data.Typeable (Typeable) import Data.Foldable (forM_) -import Control.Concurrent (forkIO, threadDelay) +import Control.Concurrent (forkIO, threadDelay, myThreadId, throwTo, ThreadId) import Control.Concurrent.MVar ( MVar , newEmptyMVar @@ -16,8 +17,9 @@ , readMVar ) import Control.Monad (replicateM_, replicateM, forever) -import Control.Exception (throwIO) -import Control.Applicative ((<$>), (<*>)) +import Control.Exception (SomeException, throwIO) +import qualified Control.Exception as Ex (catch) +import Control.Applicative ((<$>), (<*>), pure, (<|>)) import qualified Network.Transport as NT (Transport, closeEndPoint) import Network.Socket (sClose) import Network.Transport.TCP @@ -29,17 +31,31 @@ import Control.Distributed.Process.Internal.Types ( NodeId(nodeAddress) , LocalNode(localEndPoint) + , RegisterReply(..) ) import Control.Distributed.Process.Node import Control.Distributed.Process.Serializable (Serializable) -import TestAuxiliary +import Test.HUnit (Assertion) +import Test.Framework (Test, defaultMain, testGroup) +import Test.Framework.Providers.HUnit (testCase) + newtype Ping = Ping ProcessId deriving (Typeable, Binary, Show) newtype Pong = Pong ProcessId deriving (Typeable, Binary, Show) +-------------------------------------------------------------------------------- +-- Supporting definitions -- +-------------------------------------------------------------------------------- + +-- | Like fork, but throw exceptions in the child thread to the parent +forkTry :: IO () -> IO ThreadId +forkTry p = do + tid <- myThreadId + forkIO $ Ex.catch p (\e -> throwTo tid (e :: SomeException)) + -- | The ping server from the paper ping :: Process () ping = do @@ -55,35 +71,32 @@ WhereIsReply _ mPid <- expect return mPid --- | Basic ping test -testPing :: NT.Transport -> IO () -testPing transport = do - serverAddr <- newEmptyMVar - clientDone <- newEmptyMVar - - -- Server - forkIO $ do - localNode <- newLocalNode transport initRemoteTable - addr <- forkProcess localNode ping - putMVar serverAddr addr - - -- Client - forkIO $ do - localNode <- newLocalNode transport initRemoteTable - pingServer <- readMVar serverAddr +data Add = Add ProcessId Double Double deriving (Typeable) +data Divide = Divide ProcessId Double Double deriving (Typeable) +data DivByZero = DivByZero deriving (Typeable) - let numPings = 10000 +instance Binary Add where + put (Add pid x y) = put pid >> put x >> put y + get = Add <$> get <*> get <*> get - runProcess localNode $ do - pid <- getSelfPid - replicateM_ numPings $ do - send pingServer (Pong pid) - Ping _ <- expect - return () +instance Binary Divide where + put (Divide pid x y) = put pid >> put x >> put y + get = Divide <$> get <*> get <*> get - putMVar clientDone () +instance Binary DivByZero where + put DivByZero = return () + get = return DivByZero - takeMVar clientDone +-- The math server from the paper +math :: Process () +math = do + receiveWait + [ match (\(Add pid x y) -> send pid (x + y)) + , matchIf (\(Divide _ _ y) -> y /= 0) + (\(Divide pid x y) -> send pid (x / y)) + , match (\(Divide pid _ _) -> send pid DivByZero) + ] + math -- | Monitor or link to a remote node monitorOrLink :: Bool -- ^ 'True' for monitor, 'False' for link @@ -125,9 +138,42 @@ liftIO $ putMVar done () ) +-------------------------------------------------------------------------------- +-- The tests proper -- +-------------------------------------------------------------------------------- +-- | Basic ping test +testPing :: NT.Transport -> Assertion +testPing transport = do + serverAddr <- newEmptyMVar + clientDone <- newEmptyMVar + + -- Server + forkIO $ do + localNode <- newLocalNode transport initRemoteTable + addr <- forkProcess localNode ping + putMVar serverAddr addr + + -- Client + forkIO $ do + localNode <- newLocalNode transport initRemoteTable + pingServer <- readMVar serverAddr + + let numPings = 10000 + + runProcess localNode $ do + pid <- getSelfPid + replicateM_ numPings $ do + send pingServer (Pong pid) + Ping _ <- expect + return () + + putMVar clientDone () + + takeMVar clientDone + -- | Monitor a process on an unreachable node -testMonitorUnreachable :: NT.Transport -> Bool -> Bool -> IO () +testMonitorUnreachable :: NT.Transport -> Bool -> Bool -> Assertion testMonitorUnreachable transport mOrL un = do deadProcess <- newEmptyMVar done <- newEmptyMVar @@ -147,7 +193,7 @@ takeMVar done -- | Monitor a process which terminates normally -testMonitorNormalTermination :: NT.Transport -> Bool -> Bool -> IO () +testMonitorNormalTermination :: NT.Transport -> Bool -> Bool -> Assertion testMonitorNormalTermination transport mOrL un = do monitorSetup <- newEmptyMVar monitoredProcess <- newEmptyMVar @@ -168,7 +214,7 @@ takeMVar done -- | Monitor a process which terminates abnormally -testMonitorAbnormalTermination :: NT.Transport -> Bool -> Bool -> IO () +testMonitorAbnormalTermination :: NT.Transport -> Bool -> Bool -> Assertion testMonitorAbnormalTermination transport mOrL un = do monitorSetup <- newEmptyMVar monitoredProcess <- newEmptyMVar @@ -192,7 +238,7 @@ takeMVar done -- | Monitor a local process that is already dead -testMonitorLocalDeadProcess :: NT.Transport -> Bool -> Bool -> IO () +testMonitorLocalDeadProcess :: NT.Transport -> Bool -> Bool -> Assertion testMonitorLocalDeadProcess transport mOrL un = do processDead <- newEmptyMVar processAddr <- newEmptyMVar @@ -212,7 +258,7 @@ takeMVar done -- | Monitor a remote process that is already dead -testMonitorRemoteDeadProcess :: NT.Transport -> Bool -> Bool -> IO () +testMonitorRemoteDeadProcess :: NT.Transport -> Bool -> Bool -> Assertion testMonitorRemoteDeadProcess transport mOrL un = do processDead <- newEmptyMVar processAddr <- newEmptyMVar @@ -233,7 +279,7 @@ takeMVar done -- | Monitor a process that becomes disconnected -testMonitorDisconnect :: NT.Transport -> Bool -> Bool -> IO () +testMonitorDisconnect :: NT.Transport -> Bool -> Bool -> Assertion testMonitorDisconnect transport mOrL un = do processAddr <- newEmptyMVar monitorSetup <- newEmptyMVar @@ -254,34 +300,8 @@ takeMVar done -data Add = Add ProcessId Double Double deriving (Typeable) -data Divide = Divide ProcessId Double Double deriving (Typeable) -data DivByZero = DivByZero deriving (Typeable) - -instance Binary Add where - put (Add pid x y) = put pid >> put x >> put y - get = Add <$> get <*> get <*> get - -instance Binary Divide where - put (Divide pid x y) = put pid >> put x >> put y - get = Divide <$> get <*> get <*> get - -instance Binary DivByZero where - put DivByZero = return () - get = return DivByZero - -math :: Process () -math = do - receiveWait - [ match (\(Add pid x y) -> send pid (x + y)) - , matchIf (\(Divide _ _ y) -> y /= 0) - (\(Divide pid x y) -> send pid (x / y)) - , match (\(Divide pid _ _) -> send pid DivByZero) - ] - math - -- | Test the math server (i.e., receiveWait) -testMath :: NT.Transport -> IO () +testMath :: NT.Transport -> Assertion testMath transport = do serverAddr <- newEmptyMVar clientDone <- newEmptyMVar @@ -312,7 +332,7 @@ -- | Send first message (i.e. connect) to an already terminated process -- (without monitoring); then send another message to a second process on -- the same remote node (we're checking that the remote node did not die) -testSendToTerminated :: NT.Transport -> IO () +testSendToTerminated :: NT.Transport -> Assertion testSendToTerminated transport = do serverAddr1 <- newEmptyMVar serverAddr2 <- newEmptyMVar @@ -342,7 +362,7 @@ takeMVar clientDone -- | Test (non-zero) timeout -testTimeout :: NT.Transport -> IO () +testTimeout :: NT.Transport -> Assertion testTimeout transport = do localNode <- newLocalNode transport initRemoteTable runProcess localNode $ do @@ -350,7 +370,7 @@ return () -- | Test zero timeout -testTimeout0 :: NT.Transport -> IO () +testTimeout0 :: NT.Transport -> Assertion testTimeout0 transport = do serverAddr <- newEmptyMVar clientDone <- newEmptyMVar @@ -384,7 +404,7 @@ takeMVar clientDone -- | Test typed channels -testTypedChannels :: NT.Transport -> IO () +testTypedChannels :: NT.Transport -> Assertion testTypedChannels transport = do serverChannel <- newEmptyMVar :: IO (MVar (SendPort (SendPort Bool, Int))) clientDone <- newEmptyMVar @@ -410,7 +430,7 @@ takeMVar clientDone -- | Test merging receive ports -testMergeChannels :: NT.Transport -> IO () +testMergeChannels :: NT.Transport -> Assertion testMergeChannels transport = do localNode <- newLocalNode transport initRemoteTable testFlat localNode True "aaabbbccc" @@ -424,29 +444,34 @@ where -- Single layer of merging testFlat :: LocalNode -> Bool -> String -> IO () - testFlat localNode biased expected = - runProcess localNode $ do + testFlat localNode biased expected = do + done <- newEmptyMVar + forkProcess localNode $ do rs <- mapM charChannel "abc" m <- mergePorts biased rs xs <- replicateM 9 $ receiveChan m True <- return $ xs == expected - return () + liftIO $ putMVar done () + takeMVar done -- Two layers of merging testNested :: LocalNode -> Bool -> Bool -> String -> IO () - testNested localNode biasedInner biasedOuter expected = - runProcess localNode $ do + testNested localNode biasedInner biasedOuter expected = do + done <- newEmptyMVar + forkProcess localNode $ do rss <- mapM (mapM charChannel) ["abc", "def", "ghi"] ms <- mapM (mergePorts biasedInner) rss m <- mergePorts biasedOuter ms xs <- replicateM (9 * 3) $ receiveChan m True <- return $ xs == expected - return () + liftIO $ putMVar done () + takeMVar done -- Test that if no messages are (immediately) available, the scheduler makes no difference testBlocked :: LocalNode -> Bool -> IO () testBlocked localNode biased = do vs <- replicateM 3 newEmptyMVar + done <- newEmptyMVar forkProcess localNode $ do [sa, sb, sc] <- liftIO $ mapM readMVar vs @@ -477,14 +502,16 @@ , (sa, 'a') ] - runProcess localNode $ do + forkProcess localNode $ do (ss, rs) <- unzip <$> replicateM 3 newChan liftIO $ mapM_ (uncurry putMVar) $ zip vs ss m <- mergePorts biased rs xs <- replicateM (6 * 3) $ receiveChan m True <- return $ xs == "abcacbbacbcacabcba" - return () + liftIO $ putMVar done () + takeMVar done + mergePorts :: Serializable a => Bool -> [ReceivePort a] -> Process (ReceivePort a) mergePorts True = mergePortsBiased mergePorts False = mergePortsRR @@ -496,7 +523,7 @@ liftIO $ threadDelay 10000 -- Make sure messages have been sent return rport -testTerminate :: NT.Transport -> IO () +testTerminate :: NT.Transport -> Assertion testTerminate transport = do localNode <- newLocalNode transport initRemoteTable @@ -510,7 +537,7 @@ True <- return $ ref == ref' && pid == pid' && ex == show ProcessTerminationException return () -testMonitorNode :: NT.Transport -> IO () +testMonitorNode :: NT.Transport -> Assertion testMonitorNode transport = do [node1, node2] <- replicateM 2 $ newLocalNode transport initRemoteTable @@ -522,7 +549,7 @@ True <- return $ ref == ref' && nid == localNodeId node1 return () -testMonitorChannel :: NT.Transport -> IO () +testMonitorChannel :: NT.Transport -> Assertion testMonitorChannel transport = do [node1, node2] <- replicateM 2 $ newLocalNode transport initRemoteTable gotNotification <- newEmptyMVar @@ -543,7 +570,7 @@ takeMVar gotNotification -testRegistry :: NT.Transport -> IO () +testRegistry :: NT.Transport -> Assertion testRegistry transport = do node <- newLocalNode transport initRemoteTable @@ -559,7 +586,7 @@ True <- return $ pingServer == pid' return () -testRemoteRegistry :: NT.Transport -> IO () +testRemoteRegistry :: NT.Transport -> Assertion testRemoteRegistry transport = do node1 <- newLocalNode transport initRemoteTable node2 <- newLocalNode transport initRemoteTable @@ -568,7 +595,11 @@ runProcess node2 $ do let nid1 = localNodeId node1 - registerRemote nid1 "ping" pingServer + registerRemoteAsync nid1 "ping" pingServer + receiveWait [ + matchIf (\(RegisterReply label' _) -> "ping" == label') + (\(RegisterReply _ _) -> return ()) ] + Just pid <- whereisRemote nid1 "ping" True <- return $ pingServer == pid us <- getSelfPid @@ -577,7 +608,7 @@ True <- return $ pingServer == pid' return () -testSpawnLocal :: NT.Transport -> IO () +testSpawnLocal :: NT.Transport -> Assertion testSpawnLocal transport = do node <- newLocalNode transport initRemoteTable @@ -595,7 +626,7 @@ send pid sport expect -testReconnect :: NT.Transport -> TransportInternals -> IO () +testReconnect :: NT.Transport -> TransportInternals -> Assertion testReconnect transport transportInternals = do [node1, node2] <- replicateM 2 $ newLocalNode transport initRemoteTable let nid1 = localNodeId node1 @@ -639,8 +670,14 @@ -} us <- getSelfPid - registerRemote nid1 "a" us >> liftIO (threadDelay 100000) -- registerRemote is asynchronous + registerRemoteAsync nid1 "a" us -- registerRemote is asynchronous + receiveWait [ + matchIf (\(RegisterReply label' _) -> "a" == label') + (\(RegisterReply _ _) -> return ()) ] + Just _ <- whereisRemote nid1 "a" + + -- Simulate network failure liftIO $ do sock <- socketBetween transportInternals (nodeAddress nid1) (nodeAddress nid2) @@ -648,14 +685,20 @@ threadDelay 10000 -- This will happen due to implicit reconnect - registerRemote nid1 "b" us + registerRemoteAsync nid1 "b" us + receiveWait [ + matchIf (\(RegisterReply label' _) -> "b" == label') + (\(RegisterReply _ _) -> return ()) ] -- Should happen - registerRemote nid1 "c" us + registerRemoteAsync nid1 "c" us + receiveWait [ + matchIf (\(RegisterReply label' _) -> "c" == label') + (\(RegisterReply _ _) -> return ()) ] -- Check - Just _ <- whereisRemote nid1 "a" - Just _ <- whereisRemote nid1 "b" + Nothing <- whereisRemote nid1 "a" -- this will fail because the name is removed when the node is disconnected + Just _ <- whereisRemote nid1 "b" -- this will suceed because the value is set after thereconnect Just _ <- whereisRemote nid1 "c" liftIO $ putMVar registerTestOk () @@ -664,7 +707,7 @@ -- | Test 'matchAny'. This repeats the 'testMath' but with a proxy server -- in between -testMatchAny :: NT.Transport -> IO () +testMatchAny :: NT.Transport -> Assertion testMatchAny transport = do proxyAddr <- newEmptyMVar clientDone <- newEmptyMVar @@ -695,48 +738,157 @@ takeMVar clientDone -main :: IO () -main = do - Right (transport, transportInternals) <- createTransportExposeInternals "127.0.0.1" "8080" defaultTCPParameters - runTests - [ ("Ping", testPing transport) - , ("Math", testMath transport) - , ("Timeout", testTimeout transport) - , ("Timeout0", testTimeout0 transport) - , ("SendToTerminated", testSendToTerminated transport) - , ("TypedChannnels", testTypedChannels transport) - , ("MergeChannels", testMergeChannels transport) - , ("Terminate", testTerminate transport) - , ("Registry", testRegistry transport) - , ("RemoteRegistry", testRemoteRegistry transport) - , ("SpawnLocal", testSpawnLocal transport) - , ("MatchAny", testMatchAny transport) +-- Test 'receiveChanTimeout' +testReceiveChanTimeout :: NT.Transport -> Assertion +testReceiveChanTimeout transport = do + done <- newEmptyMVar + sendPort <- newEmptyMVar + + forkTry $ do + localNode <- newLocalNode transport initRemoteTable + runProcess localNode $ do + -- Create a typed channel + (sp, rp) <- newChan :: Process (SendPort Bool, ReceivePort Bool) + liftIO $ putMVar sendPort sp + + -- Wait for a message with a delay. No message arrives, we should get Nothing after 1 second + Nothing <- receiveChanTimeout 1000000 rp + + -- Wait for a message with a delay again. Now a message arrives after 0.5 seconds + Just True <- receiveChanTimeout 1000000 rp + + -- Wait for a message with zero timeout: non-blocking check. No message is available, we get Nothing + Nothing <- receiveChanTimeout 0 rp + + -- Again, but now there is a message available + liftIO $ threadDelay 1000000 + Just False <- receiveChanTimeout 0 rp + + liftIO $ putMVar done () + + forkTry $ do + localNode <- newLocalNode transport initRemoteTable + runProcess localNode $ do + sp <- liftIO $ readMVar sendPort + + liftIO $ threadDelay 1500000 + sendChan sp True + + liftIO $ threadDelay 500000 + sendChan sp False + + takeMVar done + +-- | Test Functor, Applicative, Alternative and Monad instances for ReceiveChan +testReceiveChanFeatures :: NT.Transport -> Assertion +testReceiveChanFeatures transport = do + done <- newEmptyMVar + + forkTry $ do + localNode <- newLocalNode transport initRemoteTable + runProcess localNode $ do + (spInt, rpInt) <- newChan :: Process (SendPort Int, ReceivePort Int) + (spBool, rpBool) <- newChan :: Process (SendPort Bool, ReceivePort Bool) + + -- Test Functor instance + + sendChan spInt 2 + sendChan spBool False + + rp1 <- mergePortsBiased [even <$> rpInt, rpBool] + + True <- receiveChan rp1 + False <- receiveChan rp1 + + -- Test Applicative instance + + sendChan spInt 3 + sendChan spInt 4 + + let rp2 = pure (+) <*> rpInt <*> rpInt + + 7 <- receiveChan rp2 + + -- Test Alternative instance + + sendChan spInt 3 + sendChan spBool True + + let rp3 = (even <$> rpInt) <|> rpBool + + False <- receiveChan rp3 + True <- receiveChan rp3 + + -- Test Monad instance + + sendChan spBool True + sendChan spBool False + sendChan spInt 5 + + let rp4 :: ReceivePort Int + rp4 = do b <- rpBool + if b + then rpInt + else return 7 + + 5 <- receiveChan rp4 + 7 <- receiveChan rp4 + + liftIO $ putMVar done () + + takeMVar done + +tests :: (NT.Transport, TransportInternals) -> [Test] +tests (transport, transportInternals) = [ + testGroup "Basic features" [ + testCase "Ping" (testPing transport) + , testCase "Math" (testMath transport) + , testCase "Timeout" (testTimeout transport) + , testCase "Timeout0" (testTimeout0 transport) + , testCase "SendToTerminated" (testSendToTerminated transport) + , testCase "TypedChannnels" (testTypedChannels transport) + , testCase "MergeChannels" (testMergeChannels transport) + , testCase "Terminate" (testTerminate transport) + , testCase "Registry" (testRegistry transport) + , testCase "RemoteRegistry" (testRemoteRegistry transport) + , testCase "SpawnLocal" (testSpawnLocal transport) + , testCase "MatchAny" (testMatchAny transport) + , testCase "ReceiveChanTimeout" (testReceiveChanTimeout transport) + , testCase "ReceiveChanFeatures" (testReceiveChanFeatures transport) + ] + , testGroup "Monitoring and Linking" [ -- Monitoring processes -- -- The "missing" combinations in the list below don't make much sense, as -- we cannot guarantee that the monitor reply or link exception will not -- happen before the unmonitor or unlink - , ("MonitorUnreachable", testMonitorUnreachable transport True False) - , ("MonitorNormalTermination", testMonitorNormalTermination transport True False) - , ("MonitorAbnormalTermination", testMonitorAbnormalTermination transport True False) - , ("MonitorLocalDeadProcess", testMonitorLocalDeadProcess transport True False) - , ("MonitorRemoteDeadProcess", testMonitorRemoteDeadProcess transport True False) - , ("MonitorDisconnect", testMonitorDisconnect transport True False) - , ("LinkUnreachable", testMonitorUnreachable transport False False) - , ("LinkNormalTermination", testMonitorNormalTermination transport False False) - , ("LinkAbnormalTermination", testMonitorAbnormalTermination transport False False) - , ("LinkLocalDeadProcess", testMonitorLocalDeadProcess transport False False) - , ("LinkRemoteDeadProcess", testMonitorRemoteDeadProcess transport False False) - , ("LinkDisconnect", testMonitorDisconnect transport False False) - , ("UnmonitorNormalTermination", testMonitorNormalTermination transport True True) - , ("UnmonitorAbnormalTermination", testMonitorAbnormalTermination transport True True) - , ("UnmonitorDisconnect", testMonitorDisconnect transport True True) - , ("UnlinkNormalTermination", testMonitorNormalTermination transport False True) - , ("UnlinkAbnormalTermination", testMonitorAbnormalTermination transport False True) - , ("UnlinkDisconnect", testMonitorDisconnect transport False True) + testCase "MonitorUnreachable" (testMonitorUnreachable transport True False) + , testCase "MonitorNormalTermination" (testMonitorNormalTermination transport True False) + , testCase "MonitorAbnormalTermination" (testMonitorAbnormalTermination transport True False) + , testCase "MonitorLocalDeadProcess" (testMonitorLocalDeadProcess transport True False) + , testCase "MonitorRemoteDeadProcess" (testMonitorRemoteDeadProcess transport True False) + , testCase "MonitorDisconnect" (testMonitorDisconnect transport True False) + , testCase "LinkUnreachable" (testMonitorUnreachable transport False False) + , testCase "LinkNormalTermination" (testMonitorNormalTermination transport False False) + , testCase "LinkAbnormalTermination" (testMonitorAbnormalTermination transport False False) + , testCase "LinkLocalDeadProcess" (testMonitorLocalDeadProcess transport False False) + , testCase "LinkRemoteDeadProcess" (testMonitorRemoteDeadProcess transport False False) + , testCase "LinkDisconnect" (testMonitorDisconnect transport False False) + , testCase "UnmonitorNormalTermination" (testMonitorNormalTermination transport True True) + , testCase "UnmonitorAbnormalTermination" (testMonitorAbnormalTermination transport True True) + , testCase "UnmonitorDisconnect" (testMonitorDisconnect transport True True) + , testCase "UnlinkNormalTermination" (testMonitorNormalTermination transport False True) + , testCase "UnlinkAbnormalTermination" (testMonitorAbnormalTermination transport False True) + , testCase "UnlinkDisconnect" (testMonitorDisconnect transport False True) -- Monitoring nodes and channels - , ("MonitorNode", testMonitorNode transport) - , ("MonitorChannel", testMonitorChannel transport) + , testCase "MonitorNode" (testMonitorNode transport) + , testCase "MonitorChannel" (testMonitorChannel transport) -- Reconnect - , ("Reconnect", testReconnect transport transportInternals) + , testCase "Reconnect" (testReconnect transport transportInternals) ] + ] + +main :: IO () +main = do + Right transport <- createTransportExposeInternals "127.0.0.1" "8080" defaultTCPParameters + defaultMain (tests transport) diff --git a/tests/TestClosure.hs b/tests/TestClosure.hs --- a/tests/TestClosure.hs +++ b/tests/TestClosure.hs @@ -29,8 +29,15 @@ import Control.Distributed.Process.Node import Control.Distributed.Process.Internal.Types (NodeId(nodeAddress)) import Control.Distributed.Static (staticLabel, staticClosure) -import TestAuxiliary +import Test.HUnit (Assertion) +import Test.Framework (Test, defaultMain) +import Test.Framework.Providers.HUnit (testCase) + +-------------------------------------------------------------------------------- +-- Supporting definitions -- +-------------------------------------------------------------------------------- + quintuple :: a -> b -> c -> d -> e -> (a, b, c, d, e) quintuple a b c d e = (a, b, c, d, e) @@ -133,7 +140,18 @@ waitClosure :: Int -> Closure (Process ()) waitClosure = $(mkClosure 'wait) -testUnclosure :: Transport -> RemoteTable -> IO () +simulateNetworkFailure :: TransportInternals -> NodeId -> NodeId -> Process () +simulateNetworkFailure transportInternals fr to = liftIO $ do + threadDelay 10000 + sock <- socketBetween transportInternals (nodeAddress fr) (nodeAddress to) + sClose sock + threadDelay 10000 + +-------------------------------------------------------------------------------- +-- The tests proper -- +-------------------------------------------------------------------------------- + +testUnclosure :: Transport -> RemoteTable -> Assertion testUnclosure transport rtable = do node <- newLocalNode transport rtable done <- newEmptyMVar @@ -142,7 +160,7 @@ liftIO $ putMVar done () takeMVar done -testBind :: Transport -> RemoteTable -> IO () +testBind :: Transport -> RemoteTable -> Assertion testBind transport rtable = do node <- newLocalNode transport rtable done <- newEmptyMVar @@ -153,7 +171,7 @@ liftIO $ putMVar done () takeMVar done -testSendPureClosure :: Transport -> RemoteTable -> IO () +testSendPureClosure :: Transport -> RemoteTable -> Assertion testSendPureClosure transport rtable = do serverAddr <- newEmptyMVar serverDone <- newEmptyMVar @@ -174,7 +192,7 @@ takeMVar serverDone -testSendIOClosure :: Transport -> RemoteTable -> IO () +testSendIOClosure :: Transport -> RemoteTable -> Assertion testSendIOClosure transport rtable = do serverAddr <- newEmptyMVar serverDone <- newEmptyMVar @@ -198,7 +216,7 @@ takeMVar serverDone -testSendProcClosure :: Transport -> RemoteTable -> IO () +testSendProcClosure :: Transport -> RemoteTable -> Assertion testSendProcClosure transport rtable = do serverAddr <- newEmptyMVar clientDone <- newEmptyMVar @@ -222,7 +240,7 @@ takeMVar clientDone -testSpawn :: Transport -> RemoteTable -> IO () +testSpawn :: Transport -> RemoteTable -> Assertion testSpawn transport rtable = do serverNodeAddr <- newEmptyMVar clientDone <- newEmptyMVar @@ -243,7 +261,7 @@ takeMVar clientDone -testCall :: Transport -> RemoteTable -> IO () +testCall :: Transport -> RemoteTable -> Assertion testCall transport rtable = do serverNodeAddr <- newEmptyMVar clientDone <- newEmptyMVar @@ -261,7 +279,7 @@ takeMVar clientDone -testCallBind :: Transport -> RemoteTable -> IO () +testCallBind :: Transport -> RemoteTable -> Assertion testCallBind transport rtable = do serverNodeAddr <- newEmptyMVar clientDone <- newEmptyMVar @@ -279,7 +297,7 @@ takeMVar clientDone -testSeq :: Transport -> RemoteTable -> IO () +testSeq :: Transport -> RemoteTable -> Assertion testSeq transport rtable = do node <- newLocalNode transport rtable done <- newEmptyMVar @@ -297,7 +315,7 @@ -- child. The supervisor then throws an exception, the child dies because it -- was linked to the supervisor, and the third process notices that the child -- dies. -testSpawnSupervised :: Transport -> RemoteTable -> IO () +testSpawnSupervised :: Transport -> RemoteTable -> Assertion testSpawnSupervised transport rtable = do [node1, node2] <- replicateM 2 $ newLocalNode transport rtable [superPid, childPid] <- replicateM 2 $ newEmptyMVar @@ -326,7 +344,7 @@ supervisorDeath :: IOException supervisorDeath = userError "Supervisor died" -testSpawnInvalid :: Transport -> RemoteTable -> IO () +testSpawnInvalid :: Transport -> RemoteTable -> Assertion testSpawnInvalid transport rtable = do node <- newLocalNode transport rtable done <- newEmptyMVar @@ -338,7 +356,7 @@ liftIO $ putMVar done () takeMVar done -testClosureExpect :: Transport -> RemoteTable -> IO () +testClosureExpect :: Transport -> RemoteTable -> Assertion testClosureExpect transport rtable = do node <- newLocalNode transport rtable done <- newEmptyMVar @@ -351,7 +369,7 @@ liftIO $ putMVar done () takeMVar done -testSpawnChannel :: Transport -> RemoteTable -> IO () +testSpawnChannel :: Transport -> RemoteTable -> Assertion testSpawnChannel transport rtable = do done <- newEmptyMVar [node1, node2] <- replicateM 2 $ newLocalNode transport rtable @@ -368,7 +386,7 @@ takeMVar done -testTDict :: Transport -> RemoteTable -> IO () +testTDict :: Transport -> RemoteTable -> Assertion testTDict transport rtable = do done <- newEmptyMVar [node1, node2] <- replicateM 2 $ newLocalNode transport rtable @@ -377,14 +395,7 @@ liftIO $ putMVar done () takeMVar done -simulateNetworkFailure :: TransportInternals -> NodeId -> NodeId -> Process () -simulateNetworkFailure transportInternals fr to = liftIO $ do - threadDelay 10000 - sock <- socketBetween transportInternals (nodeAddress fr) (nodeAddress to) - sClose sock - threadDelay 10000 - -testFib :: Transport -> RemoteTable -> IO () +testFib :: Transport -> RemoteTable -> Assertion testFib transport rtable = do nodes <- replicateM 4 $ newLocalNode transport rtable done <- newEmptyMVar @@ -397,7 +408,7 @@ takeMVar done -testSpawnReconnect :: Transport -> RemoteTable -> TransportInternals -> IO () +testSpawnReconnect :: Transport -> RemoteTable -> TransportInternals -> Assertion testSpawnReconnect transport rtable transportInternals = do [node1, node2] <- replicateM 2 $ newLocalNode transport rtable let nid1 = localNodeId node1 @@ -426,7 +437,7 @@ -- | 'spawn' used to ave a race condition which would be triggered if the -- spawning process terminates immediately after spawning -testSpawnTerminate :: Transport -> RemoteTable -> IO () +testSpawnTerminate :: Transport -> RemoteTable -> Assertion testSpawnTerminate transport rtable = do slave <- newLocalNode transport rtable master <- newLocalNode transport rtable @@ -440,26 +451,30 @@ takeMVar masterDone +tests :: (Transport, TransportInternals) -> RemoteTable -> [Test] +tests (transport, transportInternals) rtable = [ + testCase "Unclosure" (testUnclosure transport rtable) + , testCase "Bind" (testBind transport rtable) + , testCase "SendPureClosure" (testSendPureClosure transport rtable) + , testCase "SendIOClosure" (testSendIOClosure transport rtable) + , testCase "SendProcClosure" (testSendProcClosure transport rtable) + , testCase "Spawn" (testSpawn transport rtable) + , testCase "Call" (testCall transport rtable) + , testCase "CallBind" (testCallBind transport rtable) + , testCase "Seq" (testSeq transport rtable) + , testCase "SpawnSupervised" (testSpawnSupervised transport rtable) + , testCase "SpawnInvalid" (testSpawnInvalid transport rtable) + , testCase "ClosureExpect" (testClosureExpect transport rtable) + , testCase "SpawnChannel" (testSpawnChannel transport rtable) + , testCase "TDict" (testTDict transport rtable) + , testCase "Fib" (testFib transport rtable) + , testCase "SpawnTerminate" (testSpawnTerminate transport rtable) + , testCase "SpawnReconnect" (testSpawnReconnect transport rtable transportInternals) + ] + + main :: IO () main = do - Right (transport, transportInternals) <- createTransportExposeInternals "127.0.0.1" "8080" defaultTCPParameters + Right transport <- createTransportExposeInternals "127.0.0.1" "8080" defaultTCPParameters let rtable = __remoteTable . __remoteTableDecl $ initRemoteTable - runTests - [ ("Unclosure", testUnclosure transport rtable) - , ("Bind", testBind transport rtable) - , ("SendPureClosure", testSendPureClosure transport rtable) - , ("SendIOClosure", testSendIOClosure transport rtable) - , ("SendProcClosure", testSendProcClosure transport rtable) - , ("Spawn", testSpawn transport rtable) - , ("Call", testCall transport rtable) - , ("CallBind", testCallBind transport rtable) - , ("Seq", testSeq transport rtable) - , ("SpawnSupervised", testSpawnSupervised transport rtable) - , ("SpawnInvalid", testSpawnInvalid transport rtable) - , ("ClosureExpect", testClosureExpect transport rtable) - , ("SpawnChannel", testSpawnChannel transport rtable) - , ("TDict", testTDict transport rtable) - , ("Fib", testFib transport rtable) - , ("SpawnTerminate", testSpawnTerminate transport rtable) - , ("SpawnReconnect", testSpawnReconnect transport rtable transportInternals) - ] + defaultMain (tests transport rtable)