diff --git a/distributed-process-extras.cabal b/distributed-process-extras.cabal --- a/distributed-process-extras.cabal +++ b/distributed-process-extras.cabal @@ -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, diff --git a/src/Control/Distributed/Process/Extras/Monitoring.hs b/src/Control/Distributed/Process/Extras/Monitoring.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Extras/Monitoring.hs @@ -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 +-- 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 + diff --git a/tests/TestPrimitives.hs b/tests/TestPrimitives.hs --- a/tests/TestPrimitives.hs +++ b/tests/TestPrimitives.hs @@ -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)) ] ]