distributed-process-extras 0.1.1 → 0.2.0

raw patch · 3 files changed

+214/−18 lines, 3 filesdep ~ansi-terminaldep ~distributed-processdep ~distributed-process-testsPVP ok

version bump matches the API change (PVP)

Dependency ranges changed: ansi-terminal, distributed-process, distributed-process-tests, network

API changes (from Hackage documentation)

Files

• distributed-process-extras.cabal +18/−17
• src/Control/Distributed/Process/Extras/Monitoring.hs +139/−0
• tests/TestPrimitives.hs +57/−1

@@ -1,5 +1,5 @@ name:           distributed-process-extras-version:        0.1.1+version:        0.2.0 cabal-version:  >=1.8 build-type:     Simple license:        BSD3@@ -30,7 +30,7 @@   build-depends:                    base >= 4.4 && < 5,                    data-accessor >= 0.2.2.3,-                   distributed-process >= 0.5.2 && < 0.6,+                   distributed-process >= 0.5.3 && < 0.6,                    binary >= 0.6.3.0 && < 0.8,                    deepseq >= 1.3.0.1 && < 1.4,                    mtl,@@ -41,7 +41,7 @@                    stm >= 2.4 && < 2.5,                    time > 1.4 && < 1.5,                    transformers-  if impl(ghc <= 7.5) +  if impl(ghc <= 7.5)     Build-Depends:   template-haskell == 2.7.0.0,                      derive == 2.5.5,                      uniplate == 1.6.12,@@ -52,6 +52,7 @@   exposed-modules:                    Control.Distributed.Process.Extras,                    Control.Distributed.Process.Extras.Call,+                   Control.Distributed.Process.Extras.Monitoring,                    Control.Distributed.Process.Extras.SystemLog,                    Control.Distributed.Process.Extras.Time,                    Control.Distributed.Process.Extras.Timer,@@ -69,10 +70,10 @@   x-uses-tf:       true   build-depends:                    base >= 4.4 && < 5,-                   ansi-terminal >= 0.5 && < 0.6,-                   distributed-process >= 0.5.2 && < 0.6,+                   ansi-terminal >= 0.5 && < 0.7,+                   distributed-process >= 0.5.3 && < 0.6,                    distributed-process-extras,-                   distributed-process-tests >= 0.4.1 && < 0.5,+                   distributed-process-tests >= 0.4.3 && < 0.5,                    HUnit >= 1.2 && < 2,                    test-framework >= 0.6 && < 0.9,                    test-framework-hunit,@@ -92,17 +93,17 @@   x-uses-tf:       true   build-depends:                    base >= 4.4 && < 5,-                   ansi-terminal >= 0.5 && < 0.6,-                   distributed-process >= 0.5.2 && < 0.6,+                   ansi-terminal >= 0.5 && < 0.7,+                   distributed-process >= 0.5.3 && < 0.6,                    distributed-process-extras,-                   distributed-process-tests >= 0.4.1 && < 0.5,+                   distributed-process-tests >= 0.4.3 && < 0.5,                    network-transport >= 0.4 && < 0.5,                    mtl,                    containers >= 0.4 && < 0.6,                    network-transport-tcp >= 0.4 && < 0.5,                    binary >= 0.6.3.0 && < 0.8,                    deepseq >= 1.3.0.1 && < 1.4,-                   network >= 2.3 && < 2.6,+                   network >= 2.3 && < 2.7,                    HUnit >= 1.2 && < 2,                    stm >= 2.3 && < 2.5,                    test-framework >= 0.6 && < 0.9,@@ -120,11 +121,11 @@   x-uses-tf:       true   build-depends:                    base >= 4.4 && < 5,-                   ansi-terminal >= 0.5 && < 0.6,+                   ansi-terminal >= 0.5 && < 0.7,                    deepseq >= 1.3.0.1 && < 1.4,-                   distributed-process >= 0.5.2 && < 0.6,+                   distributed-process >= 0.5.3 && < 0.6,                    distributed-process-extras,-                   distributed-process-tests >= 0.4.1 && < 0.5,+                   distributed-process-tests >= 0.4.3 && < 0.5,                    network-transport >= 0.4 && < 0.5,                    network-transport-tcp >= 0.4 && < 0.5,                    HUnit >= 1.2 && < 2,@@ -146,13 +147,13 @@ --  x-uses-tf:       true   build-depends:                    base >= 4.4 && < 5,-                   ansi-terminal >= 0.5 && < 0.6,+                   ansi-terminal >= 0.5 && < 0.7,                    containers,                    hashable,                    unordered-containers >= 0.2.3.0 && < 0.3,-                   distributed-process >= 0.5.2 && < 0.6,+                   distributed-process >= 0.5.3 && < 0.6,                    distributed-process-extras,-                   distributed-process-tests >= 0.4.1 && < 0.5,+                   distributed-process-tests >= 0.4.3 && < 0.5,                    distributed-static,                    bytestring,                    data-accessor,@@ -162,7 +163,7 @@                    mtl,                    network-transport-tcp >= 0.4 && < 0.5,                    binary >= 0.6.3.0 && < 0.8,-                   network >= 2.3 && < 2.6,+                   network >= 2.3 && < 2.7,                    HUnit >= 1.2 && < 2,                    stm >= 2.3 && < 2.5,                    time > 1.4 && < 1.5,

@@ -0,0 +1,139 @@+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE DeriveGeneric      #-}++-----------------------------------------------------------------------------+-- |+-- Module      :  Control.Distributed.Process.Extras.Monitoring+-- Copyright   :  (c) Tim Watson 2013 - 2014+-- License     :  BSD3 (see the file LICENSE)+--+-- Maintainer  :  Tim Watson <watson.timothy@gmail.com>+-- Stability   :  experimental+-- Portability :  non-portable (requires concurrency)+--+-- This module provides a primitive node monitoring capability, implemented as+-- a /distributed-process Management Agent/. Once the 'nodeMonitor' agent is+-- started, calling 'monitorNodes' will ensure that whenever the local node+-- detects a new network-transport connection (from another cloud haskell node),+-- the caller will receive a 'NodeUp' message in its mailbox. If a node+-- disconnects, a corollary 'NodeDown' message will be delivered as well.+--+-----------------------------------------------------------------------------++module Control.Distributed.Process.Extras.Monitoring+  (+    NodeUp(..)+  , NodeDown(..)+  , nodeMonitorAgentId+  , nodeMonitor+  , monitorNodes+  , unmonitorNodes+  ) where++import Control.DeepSeq (NFData)+import Control.Distributed.Process  -- NB: requires NodeId(..) to be exported!+import Control.Distributed.Process.Management+  ( MxEvent(MxConnected, MxDisconnected)+  , MxAgentId(..)+  , mxAgent+  , mxSink+  , mxReady+  , liftMX+  , mxGetLocal+  , mxSetLocal+  , mxNotify+  )+import Control.Distributed.Process.Extras (deliver)+import Data.Binary+import qualified Data.Foldable as Foldable+import Data.HashSet (HashSet)+import qualified Data.HashSet as Set++import Data.Typeable (Typeable)+import GHC.Generics++data Register = Register !ProcessId+  deriving (Typeable, Generic)+instance Binary Register where+instance NFData Register where++data UnRegister = UnRegister !ProcessId+  deriving (Typeable, Generic)+instance Binary UnRegister where+instance NFData UnRegister where++-- | Sent to subscribing processes when a connection+-- (from a remote node) is detected.+--+data NodeUp = NodeUp !NodeId+  deriving (Typeable, Generic, Show)+instance Binary NodeUp where+instance NFData NodeUp where++-- | Sent to subscribing processes when a dis-connection+-- (from a remote node) is detected.+--+data NodeDown = NodeDown !NodeId+  deriving (Typeable, Generic, Show)+instance Binary NodeDown where+instance NFData NodeDown where++-- | The @MxAgentId@ for the node monitoring agent.+nodeMonitorAgentId :: MxAgentId+nodeMonitorAgentId = MxAgentId "service.monitoring.nodes"++-- | Start monitoring node connection/disconnection events. When a+-- connection event occurs, the calling process will receive a message+-- @NodeUp NodeId@ in its mailbox. When a disconnect occurs, the+-- corollary @NodeDown NodeId@ message will be delivered instead.+--+-- No guaranatee is made about the timeliness of the delivery, nor can+-- the receiver expect that the node (for which it is being notified)+-- is still up/connected or down/disconnected at the point when it receives+-- a message from the node monitoring agent.+--+monitorNodes :: Process ()+monitorNodes = do+  us <- getSelfPid+  mxNotify $ Register us++-- | Stop monitoring node connection/disconnection events. This does not+-- flush the caller's mailbox, nor does it guarantee that any/all node+-- up/down notifications will have been delivered before it is evaluated.+--+unmonitorNodes :: Process ()+unmonitorNodes = do+  us <- getSelfPid+  mxNotify $ UnRegister us++-- | Starts the node monitoring agent. No call to @monitorNodes@ and+-- @unmonitorNodes@ will have any effect unless the agent is already+-- running. Note that we make /no guarantees what-so-ever/ about the+-- timeliness or ordering semantics of node monitoring notifications.+--+nodeMonitor :: Process ProcessId+nodeMonitor = do+  mxAgent nodeMonitorAgentId initState [+        (mxSink $ \(Register pid) -> do+            mxSetLocal . Set.insert pid =<< mxGetLocal+            mxReady)+      , (mxSink $ \(UnRegister pid) -> do+            mxSetLocal . Set.delete pid =<< mxGetLocal+            mxReady)+      , (mxSink $ \ev -> do+            let act =+                  case ev of+                    (MxConnected    _ ep) -> notify $ nodeUp ep+                    (MxDisconnected _ ep) -> notify $ nodeDown ep+                    _                     -> return ()+            act >> mxReady)+    ]+  where+    initState :: HashSet ProcessId+    initState = Set.empty++    notify msg = Foldable.mapM_ (liftMX . deliver msg) =<< mxGetLocal++    nodeUp = NodeUp . NodeId+    nodeDown = NodeDown . NodeId+

@@ -12,7 +12,7 @@ import Control.Distributed.Process.Extras hiding (__remoteTable, monitor, send) import qualified Control.Distributed.Process.Extras (__remoteTable) import Control.Distributed.Process.Extras.Call--- import Control.Distributed.Process.Extras.Service.Monitoring+import Control.Distributed.Process.Extras.Monitoring import Control.Distributed.Process.Extras.Time import Control.Monad (void) import Control.Rematch hiding (match)@@ -88,6 +88,56 @@     (Just _)               -> stash result $ Just False     Nothing                -> stash result Nothing +testMonitorNodeDeath :: NT.Transport -> TestResult () -> Process ()+testMonitorNodeDeath transport result = do+    void $ nodeMonitor >> monitorNodes   -- start node monitoring++    nid1 <- getSelfNode+    nid2 <- liftIO $ newEmptyMVar+    nid3 <- liftIO $ newEmptyMVar++    node2 <- liftIO $ newLocalNode transport initRemoteTable+    node3 <- liftIO $ newLocalNode transport initRemoteTable++    -- sending to (nodeId, "ignored") is a short cut to force a connection+    liftIO $ tryForkProcess node2 $ ensureNodeRunning nid2 (nid1, "ignored")+    liftIO $ tryForkProcess node3 $ ensureNodeRunning nid3 (nid1, "ignored")++    NodeUp _ <- expect+    NodeUp _ <- expect++    void $ liftIO $ closeLocalNode node2+    void $ liftIO $ closeLocalNode node3++    NodeDown n1 <- expect+    NodeDown n2 <- expect++    mn1 <- liftIO $ takeMVar nid2+    mn2 <- liftIO $ takeMVar nid3++    [mn1, mn2] `shouldContain` n1+    [mn1, mn2] `shouldContain` n2++    nid4 <- liftIO $ newEmptyMVar+    node4 <- liftIO $ newLocalNode transport initRemoteTable+    void $ liftIO $ runProcess node4 $ do+      us <- getSelfNode+      liftIO $ putMVar nid4 us+      monitorNode nid1 >> return ()++    mn3 <- liftIO $ takeMVar nid4+    NodeUp n3 <- expect+    mn3 `shouldBe` (equalTo n3)++    liftIO $ closeLocalNode node4+    stash result ()++    where+      ensureNodeRunning mvar nid = do+        us <- getSelfNode+        liftIO $ putMVar mvar us+        sendTo nid "connected"+ myRemoteTable :: RemoteTable myRemoteTable = Control.Distributed.Process.Extras.__remoteTable initRemoteTable @@ -145,6 +195,12 @@       ],     testGroup "Call/RPC" [         testCase "multicallTest" (multicallTest transport)+      ],+    testGroup "Node Monitoring" [+        testCase "Death Notifications"+          (delayedAssertion+           "subscribers should both have received NodeDown twice"+           localNode () (testMonitorNodeDeath transport))       ]   ]