distributed-process 0.4.0.2 → 0.4.1

raw patch · 8 files changed

+623/−319 lines, 8 filesdep +HUnitdep +test-frameworkdep +test-framework-hunitdep ~binaryPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependencies added: HUnit, test-framework, test-framework-hunit

Dependency ranges changed: binary

API changes (from Hackage documentation)

Files

• distributed-process.cabal +15/−11
• src/Control/Distributed/Process.hs +13/−5
• src/Control/Distributed/Process/Internal/Primitives.hs +115/−31
• src/Control/Distributed/Process/Internal/Types.hs +31/−5
• src/Control/Distributed/Process/Node.hs +111/−21
• tests/TestAuxiliary.hs +0/−75
• tests/TestCH.hs +278/−126
• tests/TestClosure.hs +60/−45

@@ -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

@@ -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

@@ -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)

@@ -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                                                                  --

@@ -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

@@ -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) 

@@ -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)

@@ -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)