1{-# LANGUAGE DeriveDataTypeable #-}
2{-# LANGUAGE DeriveGeneric #-}
3{-# LANGUAGE StandaloneDeriving #-}
4{-# LANGUAGE ExistentialQuantification #-}
5{-# LANGUAGE ScopedTypeVariables #-}
6{-# LANGUAGE RecordWildCards #-}
7{-# LANGUAGE PatternGuards #-}
8{-# LANGUAGE FlexibleInstances #-}
9{-# LANGUAGE UndecidableInstances #-}
11-----------------------------------------------------------------------------
13-- Module : Control.Distributed.Process.Management
14-- Copyright : (c) Well-Typed / Tim Watson
15-- License : BSD3 (see the file LICENSE)
17-- Maintainer : Tim Watson <watson.timothy@gmail.com>
18-- Stability : experimental
19-- Portability : non-portable (requires concurrency)
21-- [Management Extensions API]
23-- This module presents an API for creating /Management Agents/:
24-- special processes that are capable of receiving and responding to
25-- a node's internal system events. These /system events/ are delivered by
26-- the management event bus: An internal subsystem maintained for each
27-- running node, to which all agents are automatically subscribed.
29-- /Agents/ are defined in terms of /event sinks/, taking a particular
30-- @Serializable@ type and evaluating to an action in the 'MxAgent' monad in
31-- response. Each 'MxSink' evaluates to an 'MxAction' that specifies whether
32-- the agent should continue processing it's inputs or stop. If the type of a
33-- message cannot be matched to any of the agent's sinks, it will be discarded.
34-- A sink can also deliberately skip processing a message, deferring to the
35-- remaining handlers. This is the /only/ way that more than one event sink
36-- can handle the same data type, since otherwise the first /type match/ will
37-- /win/ every time a message arrives. See 'mxSkip' for details.
39-- Various events are published to the management event bus automatically,
40-- the full list of which can be found in the definition of the 'MxEvent' data
41-- type. Additionally, clients of the /Management API/ can publish arbitrary
42-- @Serializable@ data to the event bus using 'mxNotify'. All running agents
43-- receive all events (from the primary event bus to which they're subscribed).
45-- Agent processes are automatically registered on the local node, and can
46-- receive messages via their mailbox just like ordinary processes. Unlike
47-- ordinary @Process@ code however, it is unnecessary (though possible) for
48-- agents to use the base @expect@ and @receiveX@ primitives to do this, since
49-- the management infrastructure will continuously read from both the primary
50-- event bus /and/ the process' own mailbox. Messages are transparently passed
51-- to the agent's event sinks from both sources, so an agent need only concern
52-- itself with how to respond to its inputs.
54-- Some agents may wish to prioritise messages from their mailbox over traffic
55-- on the management event bus, or vice versa. The 'mxReceive' and
56-- 'mxReceiveChan' API calls do this for the mailbox and event bus,
57-- respectively. The /prioritisation/ these APIs offer is simply that the chosen
58-- data stream will be checked first. No blocking will occur if the chosen
59-- (prioritised) source is devoid of input messages, instead the agent handling
60-- code will revert to switching between the alternatives in /round-robin/ as
61-- usual. This switching (and prioritisation) will take place a limited number
62-- of times, after which - in order to avoid over-utilising system
63-- resources - the agent will revert to blocking on the event bus (and ignoring
66-- Prioritisation then, is a /hint/ about the preference of data source from
67-- which the next input should be chosen. No guarantee can be made that the
68-- chosen source will in fact be selected at runtime.
70-- [Management API Semantics]
72-- The management API provides /no guarantees whatsoever/, viz:
74-- * The ordering of messages delivered to the event bus.
76-- * The order in which agents will be executed.
78-- * Whether messages will be taken from the mailbox first, or the event bus.
80-- [Management Data API]
82-- Both management agents and clients of the API have access to a variety of
83-- data storage capabilities, to facilitate publishing and consuming useful
84-- system information. Agents maintain their own internal state privately (via a
85-- state transformer - see 'mxGetLocal' et al), however it is possible for
86-- agents to share additional data with each other (and the outside world)
87-- using /data tables/.
89-- Each agent is assigned its own data table, which acts as a shared map, where
90-- the keys are @String@s and the values are @Serializable@ datum of whatever
91-- type the agent or its clients stores.
93-- Because an agent's /data table/ stores its values in raw 'Message' format,
94-- it works effectively as an /un-typed dictionary/, into which data of varying
95-- types can be fed and later retrieved. The upside of this is that different
96-- keys can be mapped to various types without any additional work on the part
97-- of the developer. The downside is that the code reading these values must
98-- know in advance what type(s) to expect, and the API provides no additional
99-- support for handling that.
101-- Publishing is accomplished using the 'mxPublish' and 'mxSet' APIs, whilst
102-- querying and deletion are handled by 'mxGet', 'mxClear', 'mxPurgeTable' and
103-- 'mxDropTable' respectively.
105-- When a management agent terminates, their tables are left in memory despite
106-- termination, such that an agent may resume its role (by restarting) or have
107-- its 'MxAgentId' taken over by another subsequent agent, leaving the data
108-- originally captured in place.
112-- New agents are defined with 'mxAgent' and require a unique 'MxAgentId', an
113-- initial state - 'MxAgent' runs in a state transformer - and a list of the
114-- agent's event sinks. Each 'MxSink' is defined in terms of a specific
115-- @Serializable@ type, via the 'mxSink' function, binding the event handler
116-- expression to inputs of only that type.
118-- Apart from modifying its own local state, an agent can execute arbitrary
119-- @Process a@ code via lifting (see 'liftMX') and even publish its own messages
120-- back to the primary event bus (see 'mxBroadcast').
122-- Since messages are delivered to agents from both the management event bus and
123-- the agent processes mailbox, agents (i.e., event sinks) will generally have
124-- no idea as to their origin. An agent can, however, choose to prioritise the
125-- choice of input (source) each time one of its event sinks runs. The /standard/
126-- way for an event sink to indicate that the agent is ready for its next input
127-- is to evaluate 'mxReady'. When this happens, the management infrastructure
128-- will obtain data from the event bus and process' mailbox in a round robbin
129-- fashion, i.e., one after the other, changing each time.
133-- What follows is a grossly over-simplified example of a management agent that
134-- provides a basic name monitoring facility. Whenever a process name is
135-- registered or unregistered, clients are informed of the fact.
137-- > -- simple notification data type
139-- > data Registration = Reg { added :: Bool
140-- > , procId :: ProcessId
144-- > -- start a /name monitoring agent/
145-- > nameMonitorAgent = do
146-- > mxAgent (MxAgentId "name-monitor") Set.empty [
147-- > (mxSink $ \(pid :: ProcessId) -> do
148-- > mxUpdateState $ Set.insert pid
153-- > (MxRegistered p n) -> notify True n p
154-- > (MxUnRegistered p n) -> notify False n p
159-- > notify a n p = do
160-- > Foldable.mapM_ (liftMX . deliver (Reg a n p)) =<< mxGetLocal
163-- The client interface (for sending their pid) can take one of two forms:
165-- > monitorNames = getSelfPid >>= nsend "name-monitor"
166-- > monitorNames2 = getSelfPid >>= mxNotify
168-- For some real-world examples, see the distributed-process-platform package.
170-- [Performance, Stablity and Scalability]
172-- /Management Agents/ offer numerous advantages over regular processes:
173-- broadcast communication with them can have a lower latency, they offer
174-- simplified messgage (i.e., input type) handling and they have access to
175-- internal system information that would be otherwise unobtainable.
177-- Do not be tempted to implement everything (e.g., the kitchen sink) using the
178-- management API though. There are overheads associated with management agents
179-- which is why they're presented as tools for consuming low level system
180-- information, instead of as /application level/ development tools.
182-- Agents that rely heavily on a busy mailbox can cause the management event
183-- bus to backlog un-GC'ed data, leading to increased heap space. Producers that
184-- do not take care to avoid passing unevaluated thunks to the API can crash
185-- /all/ the agents in the system. Agents are not monitored or managed in any
186-- way, and those that crash will not be restarted.
188-- The management event bus can receive a great deal of traffic. Every time
189-- a message is sent and/or received, an event is passed to the agent controller
190-- and broadcast to all agents (plus the trace controller, if tracing is enabled
191-- for the node). This is already a significant overhead - though profiling and
192-- benchmarks have demonstrated that it does not adversely affect performance
193-- if few agents are installed. Agents will typically use more cycles than plain
194-- processes, since they perform additional work: selecting input data from both
195-- the event bus /and/ their own mailboxes, plus searching through the set of
196-- event sinks (for each agent) to determine the right handler for the event.
198-- Each management agent requires not only its own @Process@ (in which the agent
199-- code is run), but also a peer process that provides its /data table/. These
200-- data tables also have to be coordinated and manaaged on each agent's behalf.
202-- [Architecture Overview]
204-- The architecture of the management event bus is internal and subject to
205-- change without prior notice. The description that follows is provided for
206-- informational purposes only.
208-- When a node initially starts, two special, internal system processes are
209-- started to support the management infrastructure. The first, known as the
210-- /trace controller/, is responsible for consuming 'MxEvent's and forwarding
211-- them to the configured tracer - see "Control.Distributed.Process.Debug" for
212-- further details. The second is the /management agent controller/, and is the
213-- primary worker process underpinning the management infrastructure. All
214-- published management events are routed to this process, which places them
215-- onto a system wide /event bus/ and additionally passes them directly to the
216-- /trace controller/.
218-- There are several reasons for segregating the tracing and management control
219-- planes in this fashion. Tracing can be enabled or disabled by clients, whilst
220-- the management event bus cannot, since in addition to providing
221-- runtime instrumentation, its intended use-cases include node monitoring, peer
222-- discovery (via topology providing backends) and other essential system
223-- services that require knowledge of otherwise hidden system internals. Tracing
224-- is also subject to /trace flags/ that limit the specific 'MxEvent's delivered
225-- to trace clients - an overhead/complexity not shared by management agents.
226-- Finally, tracing and management agents are implemented using completely
227-- different signalling techniques - more on this later - which would introduce
228-- considerable complexity if the shared the same /event loop/.
230-- The management control plane is driven by a shared broadcast channel, which
231-- is written to by the agent controller and subscribed to by all agent
232-- processes. Agents are spawned as regular processes, whose primary
233-- implementation (i.e., /server loop/) is responsible for consuming
234-- messages from both the broadcast channel and their own mailbox. Once
235-- consumed, messages are applied to the agent's /event sinks/ until one
236-- matches the input, at which point it is applied and the loop continues.
237-- The implementation chooses from the event bus and the mailbox in a
238-- round-robin fashion, until a message is received. This polling activity would
239-- lead to management agents consuming considerable system resources if left
240-- unchecked, therefore the implementation will poll for a limitted number of
241-- retries, after which it will perform a blocking read on the event bus.
243-----------------------------------------------------------------------------
244module Control.Distributed.Process.Management
247 -- * Firing Arbitrary /Mx Events/
249 -- * Constructing Mx Agents
254 , mxAgentWithFinalize
277import Control.Applicative ((<$>))
278import Control.Concurrent.STM (atomically)
279import Control.Concurrent.STM.TChan
285import Control.Distributed.Process.Internal.Primitives
299import Control.Distributed.Process.Internal.Types
306 , unsafeCreateUnencodedMessage
308import Control.Distributed.Process.Management.Bus (publishEvent)
309import qualified Control.Distributed.Process.Management.Table as Table
310import Control.Distributed.Process.Management.Types
314 , ChannelSelector(..)
320import Control.Distributed.Process.Serializable (Serializable)
321import Control.Monad.IO.Class (liftIO)
322import Control.Monad.Reader (ask)
323import qualified Control.Monad.State as ST
332-- | Publishes an arbitrary @Serializable@ message to the management event bus.
333-- Note that /no attempt is made to force the argument/, therefore it is very
334-- important that you do not pass unevaluated thunks that might crash the
335-- receiving process via this API, since /all/ registered agents will gain
336-- access to the data structure once it is broadcast by the agent controller.
337mxNotify :: (Serializable a) => a -> Process ()
339 bus <- localEventBus . processNode <$> ask
340 liftIO $ publishEvent bus $ unsafeCreateUnencodedMessage msg
342-- | Publish an arbitrary @Message@ as a property in the management database.
344-- For publishing @Serializable@ data, use 'mxSet' instead.
346mxPublish :: MxAgentId -> String -> Message -> Process ()
347mxPublish a k v = Table.set k v (Table.MxForAgent a)
349-- | Sets an arbitrary @Serializable@ datum against a key in the management
350-- database. Note that /no attempt is made to force the argument/, therefore
351-- it is very important that you do not pass unevaluated thunks that might
352-- crash some other, arbitrary process (or management agent!) that obtains
353-- and attempts to force the value later on.
355mxSet :: Serializable a => MxAgentId -> String -> a -> Process ()
356mxSet mxId key msg = do
357 Table.set key (unsafeCreateUnencodedMessage msg) (Table.MxForAgent mxId)
359-- | Fetches a property from the management database for the given key.
360-- If the property is not set, or does not match the expected type when
361-- typechecked (at runtime), returns @Nothing@.
362mxGet :: Serializable a => MxAgentId -> String -> Process (Maybe a)
363mxGet = Table.fetch . Table.MxForAgent
365-- | Clears a property from the management database using the given key.
366-- If the key does not exist in the database, this is a noop.
367mxClear :: MxAgentId -> String -> Process ()
368mxClear mxId key = Table.clear key (Table.MxForAgent mxId)
370-- | Purges a table in the management database of all its stored properties.
371mxPurgeTable :: MxAgentId -> Process ()
372mxPurgeTable = Table.purge . Table.MxForAgent
374-- | Deletes a table from the management database.
375mxDropTable :: MxAgentId -> Process ()
376mxDropTable = Table.delete . Table.MxForAgent
378--------------------------------------------------------------------------------
379-- API for writing user defined management extensions (i.e., agents) --
380--------------------------------------------------------------------------------
382-- | Return the 'MxAgentId' for the currently executing agent.
384mxGetId :: MxAgent s MxAgentId
385mxGetId = ST.get >>= return . mxAgentId
387-- | The 'MxAgent' version of 'mxNotify'.
389mxBroadcast :: (Serializable m) => m -> MxAgent s ()
392 liftMX $ liftIO $ atomically $ do
393 writeTChan (mxBus state) (unsafeCreateUnencodedMessage msg)
395-- | Gracefully terminate an agent.
397mxDeactivate :: forall s. String -> MxAgent s MxAction
398mxDeactivate = return . MxAgentDeactivate
400-- | Continue executing (i.e., receiving and processing messages).
402mxReady :: forall s. MxAgent s MxAction
403mxReady = return MxAgentReady
405-- | Causes the currently executing /event sink/ to be skipped.
406-- The remaining declared event sinks will be evaluated to find
407-- a matching handler. Can be used to allow multiple event sinks
408-- to process data of the same type.
410mxSkip :: forall s. MxAgent s MxAction
411mxSkip = return MxAgentSkip
413-- | Continue exeucting, prioritising inputs from the process' own
414-- /mailbox/ ahead of data from the management event bus.
416mxReceive :: forall s. MxAgent s MxAction
417mxReceive = return $ MxAgentPrioritise Mailbox
419-- | Continue exeucting, prioritising inputs from the management event bus
420-- over the process' own /mailbox/.
422mxReceiveChan :: forall s. MxAgent s MxAction
423mxReceiveChan = return $ MxAgentPrioritise InputChan
425-- | Lift a @Process@ action.
427liftMX :: Process a -> MxAgent s a
428liftMX p = MxAgent $ ST.lift p
430-- | Set the agent's local state.
432mxSetLocal :: s -> MxAgent s ()
433mxSetLocal s = ST.modify $ \st -> st { mxLocalState = s }
435-- | Update the agent's local state.
437mxUpdateLocal :: (s -> s) -> MxAgent s ()
438mxUpdateLocal f = ST.modify $ \st -> st { mxLocalState = (f $ mxLocalState st) }
440-- | Fetch the agent's local state.
442mxGetLocal :: MxAgent s s
443mxGetLocal = ST.get >>= return . mxLocalState
445-- | Create an 'MxSink' from an expression taking a @Serializable@ type @m@,
446-- that yields an 'MxAction' in the 'MxAgent' monad.
448mxSink :: forall s m . (Serializable m)
449 => (m -> MxAgent s MxAction)
452 msg' <- liftMX $ (unwrapMessage msg :: Process (Maybe m))
454 Nothing -> return Nothing
458 MxAgentSkip -> return Nothing
461-- private ADT: a linked list of event sinks
465 current :: !(MxSink s)
466 , next :: !(MxPipeline s)
469-- | Activates a new agent.
471mxAgent :: MxAgentId -> s -> [MxSink s] -> Process ProcessId
472mxAgent mxId st hs = mxAgentWithFinalize mxId st hs $ return ()
474-- | Activates a new agent. This variant takes a /finalizer/ expression,
475-- that is run once the agent shuts down (even in case of failure/exceptions).
476-- The /finalizer/ expression runs in the mx monad - @MxAgent s ()@ - such
477-- that the agent's internal state remains accessible to the shutdown/cleanup
480mxAgentWithFinalize :: MxAgentId
485mxAgentWithFinalize mxId initState handlers dtor = do
486 let name = agentId mxId
487 existing <- whereis name
489 Just _ -> die "DuplicateAgentId" -- TODO: better error handling policy
491 node <- processNode <$> ask
492 pid <- liftIO $ mxNew (localEventBus node) $ start
496 start (sendTChan, recvTChan) = do
498 nsend Table.mxTableCoordinator (MxAgentStart sp mxId)
499 tablePid <- receiveWait [ matchChan rp (\(p :: ProcessId) -> return p) ]
500 let nState = MxAgentState mxId sendTChan tablePid initState
501 runAgent dtor handlers InputChan recvTChan nState
503 runAgent :: MxAgent s ()
509 runAgent eh hs cs c s =
510 runAgentWithFinalizer eh hs cs c s
511 `onException` runAgentFinalizer eh s
513 runAgentWithFinalizer :: MxAgent s ()
519 runAgentWithFinalizer eh' hs' cs' c' s' = do
520 msg <- getNextInput cs' c'
521 (action, state) <- runPipeline msg s' $ pipeline hs'
523 MxAgentReady -> runAgent eh' hs' InputChan c' state
524 MxAgentPrioritise priority -> runAgent eh' hs' priority c' state
525 MxAgentDeactivate _ -> runAgentFinalizer eh' state
526 MxAgentSkip -> error "IllegalState"
527-- MxAgentBecome h' -> runAgent h' c state
529 getNextInput sel chan =
530 let stmRead = atomically . readTChan
533 Mailbox -> [ matchAny return
534 , matchSTM (readTChan chan) return]
535 InputChan -> [ matchSTM (readTChan chan) return
537 in getNextInput' matches 100
539 -- when reading inputs, we generally want to maintain a degree of
540 -- fairness in choosing between the TChan and our mailbox, but to
541 -- ultimately favour the TChan overall. We do this by flipping
542 -- between the two (using the ChannelSelector) each time we call
543 -- getNextInput - it returns the opposite ChannelSelector to the
544 -- one which succeeded last time it was called.
546 -----------------------------------------------------------------------
548 -- This strategy works well, yet we wish to avoid blocking on one
549 -- input if the other is empty/busy, so we begin by reading both
550 -- sources conditionally - tryReadTChan for the event bus and
551 -- receiveTimeout for the mailbox. If polling (either source) does
552 -- not yield an input, we swap to the other, but we cannot continue
553 -- in this fashion ad infinitum, since that would waste considerable
554 -- system resoures. Instead, we switch ten times, after which (if no
555 -- data were obtained) we block on the event bus, since that is our
557 -----------------------------------------------------------------------
560 -- An agent can of course, choose to override which source should be
561 -- checked first. We consider this a /hint/ rather than a dictat.
562 -- When reading from the mailbox, we perform a non-blocking read.
563 -- We assume that most agents will prefer using mxReceiveChan to its
564 -- mailbox reading counterpart, this we expect the event bus to be
565 -- non-empty or the entire subsystem is likely doing very little work,
566 -- in which case we needn't worry too much about the overheads
567 -- described thus far.
569 -- TODO: use mersenne-random / (System.Random.MWC) to get a
570 -- uniform distribution of values between 10..2000
572 getNextInput' ms n = do
573 mIn <- receiveTimeout n ms
575 Nothing -> tryNextInput ms n
576 Just msg -> return msg
579 | n < 2000 = getNextInput' ms $ n * 2
580 | otherwise = getNextInput' ms 100
582 -- getNextInput' c' 0 = do
583 -- m <- liftIO $ atomically $ readTChan c'
584 -- return (m, Mailbox)
585 -- getNextInput' InputChan c' n = do
586 -- inputs <- liftIO $ atomically $ tryReadTChan c'
588 -- Nothing -> getNextInput' Mailbox c' (n - 1)
589 -- Just m -> return (m, Mailbox)
590 -- getNextInput' Mailbox c' n = do
591 -- m <- receiveTimeout 0 [ matchAny return ]
593 -- Nothing -> getNextInput' InputChan c' (n - 1)
594 -- Just msg -> return (msg, InputChan)
596 runAgentFinalizer :: MxAgent s () -> MxAgentState s -> Process ()
597 runAgentFinalizer f s = ST.runStateT (unAgent f) s >>= return . fst
599 pipeline :: forall s . [MxSink s] -> MxPipeline s
601 pipeline (sink:sinks) = MxPipeline sink (pipeline sinks)
603 runPipeline :: forall s .
607 -> Process (MxAction, MxAgentState s)
608 runPipeline _ state MxStop = return (MxAgentReady, state)
609 runPipeline msg state MxPipeline{..} = do
610 let act = current msg
611 (pass, state') <- ST.runStateT (unAgent act) state
613 Nothing -> runPipeline msg state next
614 Just result -> return (result, state')
keysesc to close
- j k
- move down, up
- enter
- open
- esc
- back, or up a level
- 1 2 3 4
- readme, files, commits, refs
- /
- filter
- y
- copy the clone command
- c
- compare the chosen commit with another (or shift-click both)
- ?
- show or hide this list