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. If messages exist in one or more channels, they will be consumed as
62-- soon as they're available, priority is effectively a hint about which
63-- channel to consume from, should messages be available in both.
65-- Prioritisation then, is a /hint/ about the preference of data source from
66-- which the next input should be chosen. No guarantee can be made that the
67-- chosen source will in fact be selected at runtime.
69-- [Management API Semantics]
71-- The management API provides /no guarantees whatsoever/, viz:
73-- * The ordering of messages delivered to the event bus.
75-- * The order in which agents will be executed.
77-- * Whether messages will be taken from the mailbox first, or the event bus.
79-- Since the event bus uses STM broadcast channels to communicate with agents,
80-- no message written to the bus successfully can be lost.
82-- Agents can also receive messages via their mailboxes - these are subject to
83-- the same guarantees as all inter-process message sending.
85-- Messages dispatched on an STM broadcast channel (i.e., management event bus)
86-- are guaranteed to be delivered with the same FIFO ordering guarantees that
87-- exist between two communicating processes, such that communication from the
88-- node controller's threads (i.e., MxEvent's) will never be re-ordered, but
89-- messages dispatched to the event bus by other processes (including, but not
90-- limited to agents) are only guaranteed to be ordered between one sender and
93-- No guarantee exists for the ordering in which messages sent to an agent's
94-- mailbox will be delivered, vs messages dispatched via the event bus.
96-- Because of the above, there are no ordering guarantees for messages sent
97-- between agents, or for processes to agents, except for those that apply to
98-- messages sent between regular processes, since agents are
99-- implemented as such.
101-- The event bus is serial and single threaded. Anything that is published by
102-- the node controller will be seen in FIFO order. There are no ordering
103-- guarantees pertaining to entries published to the event bus by other
104-- processes or agents.
106-- It should not be possible to see, for example, an @MxReceived@ before the
107-- corresponding @MxSent@ event, since the places where we issue the @MxSent@
108-- write directly to the event bus (using STM) in the calling (green) thread,
109-- before dispatching instructions to the node controller to perform the
110-- necessary routing to deliver the message to a process (or registered name,
111-- or typed channel) locally or remotely.
113-- [Management Data API]
115-- Both management agents and clients of the API have access to a variety of
116-- data storage capabilities, to facilitate publishing and consuming useful
117-- system information. Agents maintain their own internal state privately (via a
118-- state transformer - see 'mxGetLocal' et al), however it is possible for
119-- agents to share additional data with each other (and the outside world)
120-- using whatever mechanism the user wishes, e.g., acidstate, or shared memory
125-- New agents are defined with 'mxAgent' and require a unique 'MxAgentId', an
126-- initial state - 'MxAgent' runs in a state transformer - and a list of the
127-- agent's event sinks. Each 'MxSink' is defined in terms of a specific
128-- @Serializable@ type, via the 'mxSink' function, binding the event handler
129-- expression to inputs of only that type.
131-- Apart from modifying its own local state, an agent can execute arbitrary
132-- @Process a@ code via lifting (see 'liftMX') and even publish its own messages
133-- back to the primary event bus (see 'mxBroadcast').
135-- Since messages are delivered to agents from both the management event bus and
136-- the agent processes mailbox, agents (i.e., event sinks) will generally have
137-- no idea as to their origin. An agent can, however, choose to prioritise the
138-- choice of input (source) each time one of its event sinks runs. The /standard/
139-- way for an event sink to indicate that the agent is ready for its next input
140-- is to evaluate 'mxReady'. When this happens, the management infrastructure
141-- will obtain data from the event bus and process' mailbox in a round robbin
142-- fashion, i.e., one after the other, changing each time.
146-- What follows is a grossly over-simplified example of a management agent that
147-- provides a basic name monitoring facility. Whenever a process name is
148-- registered or unregistered, clients are informed of the fact.
150-- > -- simple notification data type
152-- > data Registration = Reg { added :: Bool
153-- > , procId :: ProcessId
157-- > -- start a /name monitoring agent/
158-- > nameMonitorAgent = do
159-- > mxAgent (MxAgentId "name-monitor") Set.empty [
160-- > (mxSink $ \(pid :: ProcessId) -> do
161-- > mxUpdateState $ Set.insert pid
166-- > (MxRegistered p n) -> notify True n p
167-- > (MxUnRegistered p n) -> notify False n p
172-- > notify a n p = do
173-- > Foldable.mapM_ (liftMX . deliver (Reg a n p)) =<< mxGetLocal
176-- The client interface (for sending their pid) can take one of two forms:
178-- > monitorNames = getSelfPid >>= nsend "name-monitor"
179-- > monitorNames2 = getSelfPid >>= mxNotify
181-- [Performance, Stablity and Scalability]
183-- /Management Agents/ offer numerous advantages over regular processes:
184-- broadcast communication with them can have a lower latency, they offer
185-- simplified messgage (i.e., input type) handling and they have access to
186-- internal system information that would be otherwise unobtainable.
188-- Do not be tempted to implement everything (e.g., the kitchen sink) using the
189-- management API though. There are overheads associated with management agents
190-- which is why they're presented as tools for consuming low level system
191-- information, instead of as /application level/ development tools.
193-- Agents that rely heavily on a busy mailbox can cause the management event
194-- bus to backlog un-GC'ed data, leading to increased heap space. Producers that
195-- do not take care to avoid passing unevaluated thunks to the API can crash
196-- /all/ the agents in the system. Agents are not monitored or managed in any
197-- way, and those that crash will not be restarted.
199-- The management event bus can receive a great deal of traffic. Every time
200-- a message is sent and/or received, an event is passed to the agent controller
201-- and broadcast to all agents (plus the trace controller, if tracing is enabled
202-- for the node). This is already a significant overhead - though profiling and
203-- benchmarks have demonstrated that it does not adversely affect performance
204-- if few agents are installed. Agents will typically use more cycles than plain
205-- processes, since they perform additional work: selecting input data from both
206-- the event bus /and/ their own mailboxes, plus searching through the set of
207-- event sinks (for each agent) to determine the right handler for the event.
209-- [Architecture Overview]
211-- The architecture of the management event bus is internal and subject to
212-- change without prior notice. The description that follows is provided for
213-- informational purposes only.
215-- When a node initially starts, two special, internal system processes are
216-- started to support the management infrastructure. The first, known as the
217-- /trace controller/, is responsible for consuming 'MxEvent's and forwarding
218-- them to the configured tracer - see "Control.Distributed.Process.Debug" for
219-- further details. The second is the /management agent controller/, and is the
220-- primary worker process underpinning the management infrastructure. All
221-- published management events are routed to this process, which places them
222-- onto a system wide /event bus/ and additionally passes them directly to the
223-- /trace controller/.
225-- There are several reasons for segregating the tracing and management control
226-- planes in this fashion. Tracing can be enabled or disabled by clients, whilst
227-- the management event bus cannot, since in addition to providing
228-- runtime instrumentation, its intended use-cases include node monitoring, peer
229-- discovery (via topology providing backends) and other essential system
230-- services that require knowledge of otherwise hidden system internals. Tracing
231-- is also subject to /trace flags/ that limit the specific 'MxEvent's delivered
232-- to trace clients - an overhead/complexity not shared by management agents.
233-- Finally, tracing and management agents are implemented using completely
234-- different signalling techniques - more on this later - which would introduce
235-- considerable complexity if the shared the same /event loop/.
237-- The management control plane is driven by a shared broadcast channel, which
238-- is written to by the agent controller and subscribed to by all agent
239-- processes. Agents are spawned as regular processes, whose primary
240-- implementation (i.e., /server loop/) is responsible for consuming
241-- messages from both the broadcast channel and their own mailbox. Once
242-- consumed, messages are applied to the agent's /event sinks/ until one
243-- matches the input, at which point it is applied and the loop continues.
244-- The implementation chooses from the event bus and the mailbox in a
245-- round-robin fashion, until a message is received. This polling activity would
246-- lead to management agents consuming considerable system resources if left
247-- unchecked, therefore the implementation will poll for a limitted number of
248-- retries, after which it will perform a blocking read on the event bus.
250-----------------------------------------------------------------------------
251module Control.Distributed.Process.Management
254 -- * Firing Arbitrary /Mx Events/
256 -- * Constructing Mx Agents
261 , mxAgentWithFinalize
277import Control.Applicative
278import Control.Concurrent.STM (atomically)
279import Control.Concurrent.STM.TChan
284import Control.Distributed.Process.Internal.Primitives
293import Control.Distributed.Process.Internal.Types
300 , unsafeCreateUnencodedMessage
302import Control.Distributed.Process.Management.Internal.Bus (publishEvent)
303import Control.Distributed.Process.Management.Internal.Types
307 , ChannelSelector(..)
312import Control.Distributed.Process.Serializable (Serializable)
313import Control.Monad.IO.Class (liftIO)
314import Control.Monad.Reader (ask)
315import Control.Monad.Catch (onException)
316import qualified Control.Monad.State as ST
324-- | Publishes an arbitrary @Serializable@ message to the management event bus.
325-- Note that /no attempt is made to force the argument/, therefore it is very
326-- important that you do not pass unevaluated thunks that might crash the
327-- receiving process via this API, since /all/ registered agents will gain
328-- access to the data structure once it is broadcast by the agent controller.
329mxNotify :: (Serializable a) => a -> Process ()
331 bus <- localEventBus . processNode <$> ask
332 liftIO $ publishEvent bus $ unsafeCreateUnencodedMessage msg
334--------------------------------------------------------------------------------
335-- API for writing user defined management extensions (i.e., agents) --
336--------------------------------------------------------------------------------
338-- | Return the 'MxAgentId' for the currently executing agent.
340mxGetId :: MxAgent s MxAgentId
341mxGetId = ST.get >>= return . mxAgentId
343-- | The 'MxAgent' version of 'mxNotify'.
345mxBroadcast :: (Serializable m) => m -> MxAgent s ()
348 liftMX $ liftIO $ atomically $ do
349 writeTChan (mxBus state) (unsafeCreateUnencodedMessage msg)
351-- | Gracefully terminate an agent.
353mxDeactivate :: forall s. String -> MxAgent s MxAction
354mxDeactivate = return . MxAgentDeactivate
356-- | Continue executing (i.e., receiving and processing messages).
358mxReady :: forall s. MxAgent s MxAction
359mxReady = return MxAgentReady
361-- | Causes the currently executing /event sink/ to be skipped.
362-- The remaining declared event sinks will be evaluated to find
363-- a matching handler. Can be used to allow multiple event sinks
364-- to process data of the same type.
366mxSkip :: forall s. MxAgent s MxAction
367mxSkip = return MxAgentSkip
369-- | Continue exeucting, prioritising inputs from the process' own
370-- /mailbox/ ahead of data from the management event bus.
372mxReceive :: forall s. MxAgent s MxAction
373mxReceive = return $ MxAgentPrioritise Mailbox
375-- | Continue exeucting, prioritising inputs from the management event bus
376-- over the process' own /mailbox/.
378mxReceiveChan :: forall s. MxAgent s MxAction
379mxReceiveChan = return $ MxAgentPrioritise InputChan
381-- | Lift a @Process@ action.
383liftMX :: Process a -> MxAgent s a
384liftMX p = MxAgent $ ST.lift p
386-- | Set the agent's local state.
388mxSetLocal :: s -> MxAgent s ()
389mxSetLocal s = ST.modify $ \st -> st { mxLocalState = s }
391-- | Update the agent's local state.
393mxUpdateLocal :: (s -> s) -> MxAgent s ()
394mxUpdateLocal f = ST.modify $ \st -> st { mxLocalState = (f $ mxLocalState st) }
396-- | Fetch the agent's local state.
398mxGetLocal :: MxAgent s s
399mxGetLocal = ST.get >>= return . mxLocalState
401-- | Create an 'MxSink' from an expression taking a @Serializable@ type @m@,
402-- that yields an 'MxAction' in the 'MxAgent' monad.
404mxSink :: forall s m . (Serializable m)
405 => (m -> MxAgent s MxAction)
408 msg' <- liftMX $ (unwrapMessage msg :: Process (Maybe m))
410 Nothing -> return Nothing
414 MxAgentSkip -> return Nothing
417-- private ADT: a linked list of event sinks
421 current :: !(MxSink s)
422 , next :: !(MxPipeline s)
425-- | Activates a new agent.
427mxAgent :: MxAgentId -> s -> [MxSink s] -> Process ProcessId
428mxAgent mxId st hs = mxAgentWithFinalize mxId st hs $ return ()
430-- | Activates a new agent. This variant takes a /finalizer/ expression,
431-- that is run once the agent shuts down (even in case of failure/exceptions).
432-- The /finalizer/ expression runs in the mx monad - @MxAgent s ()@ - such
433-- that the agent's internal state remains accessible to the shutdown/cleanup
436mxAgentWithFinalize :: MxAgentId
441mxAgentWithFinalize mxId initState handlers dtor = do
442 let name = agentId mxId
443 existing <- whereis name
445 Just _ -> die "DuplicateAgentId" -- TODO: better error handling policy
447 node <- processNode <$> ask
448 pid <- liftIO $ mxNew (localEventBus node) $ start
452 start (sendTChan, recvTChan) = do
453 let nState = MxAgentState mxId sendTChan initState
454 runAgent dtor handlers InputChan recvTChan nState
456 runAgent :: MxAgent s ()
462 runAgent eh hs cs c s =
463 runAgentWithFinalizer eh hs cs c s
464 `onException` runAgentFinalizer eh s
466 runAgentWithFinalizer :: MxAgent s ()
472 runAgentWithFinalizer eh' hs' cs' c' s' = do
473 msg <- getNextInput cs' c'
474 (action, state) <- runPipeline msg s' $ pipeline hs'
476 MxAgentReady -> runAgent eh' hs' InputChan c' state
477 MxAgentPrioritise priority -> runAgent eh' hs' priority c' state
478 MxAgentDeactivate _ -> runAgentFinalizer eh' state
479 MxAgentSkip -> error "IllegalState"
480-- MxAgentBecome h' -> runAgent h' c state
482 getNextInput sel chan =
485 Mailbox -> [ matchAny return
486 , matchSTM (readTChan chan) return]
487 InputChan -> [ matchSTM (readTChan chan) return
489 in receiveWait matches
491 runAgentFinalizer :: MxAgent s () -> MxAgentState s -> Process ()
492 runAgentFinalizer f s = ST.runStateT (unAgent f) s >>= return . fst
494 pipeline :: forall s . [MxSink s] -> MxPipeline s
496 pipeline (sink:sinks) = MxPipeline sink (pipeline sinks)
498 runPipeline :: forall s .
502 -> Process (MxAction, MxAgentState s)
503 runPipeline _ state MxStop = return (MxAgentReady, state)
504 runPipeline msg state MxPipeline{..} = do
505 let act = current msg
506 (pass, state') <- ST.runStateT (unAgent act) state
508 Nothing -> runPipeline msg state next
509 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