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-- [Management Data API]
56-- Both management agents and clients of the API have access to a variety of
57-- data storage capabilities, to facilitate publishing and consuming information
58-- obtained by agents. Each agent is assigned its own data table, which acts
59-- as a shared map, where the keys are @String@s and the values are
60-- @Serializable@ datum of whatever type the agent or its clients stores.
62-- Publishing is accomplished using the 'mxPublish' and 'mxSet' APIs, whilst
63-- querying and deletion are handled by 'mxGet', 'mxClear', 'mxPurgeTable' and
64-- 'mxDropTable' respectively.
66-- When management agents terminate, their tables are left in memory despite the
67-- termination, such that an agent may resume its role (by restarting) or have
68-- its 'MxAgentId' taken over by another subsequent agent, leaving the data
69-- originally captured in place.
73-- New agents are defined with 'mxAgent' and require a unique 'MxAgentId', an
74-- initial state - 'MxAgent' runs in a state transformer - and a list of the
75-- agent's event sinks. Each 'MxSink' is defined in terms of a specific
76-- @Serializable@ type, via the 'mxSink' function, binding the event handler
77-- expression to inputs of only that type.
79-- Apart from modifying its own local state, an agent can execute arbitrary
80-- @Process a@ code via lifting (see 'liftMX') and even publish its own messages
81-- back to the primary event bus (see 'mxBroadcast').
83-- Since messages are delivered to agents from both the management event bus and
84-- the agent processes mailbox, agents (i.e., event sinks) will generally have
85-- no idea as to their origin. An agent can, however, choose to prioritise the
86-- choice of input (source) each time one of its event sinks runs. The /standard/
87-- way for an event sink to indicate that the agent is ready for its next input
88-- is to evaluate 'mxReady'. When this happens, the management infrastructure
89-- will obtain data from the event bus and process' mailbox in a round robbin
90-- fashion, i.e., one after the other, changing each time.
92-- [Architecture Overview]
94-- /Management Extensions/ can be used by infrastructure or application code,
95-- to provide access to runtime instrumentation data, expose control planes
96-- and/or provide remote access to external clients.
98-- The management infrastructure is broken down into three components, which
99-- roughly correspond to the modules in which they're implemented:
101-- 1. @Agent@ - registration and definition of /Management Agents/
102-- 2. @Table@ - provides an API for publishing /instrumentation data/
103-- 3. @Trace@ - provides an API for tracing/debugging capabilities
105-- The architecture of the management event bus is internal and subject to
106-- change without prior notice. The description that follows is provided for
107-- informational purposes only.
109-- When a node initially starts, two special, internal system processes are
110-- started to support the management infrastructure. The first, known as the
111-- /trace controller/, is responsible for consuming 'MxEvent's and forwarding
112-- them to the configured tracer - see "Control.Distributed.Process.Debug" for
113-- further details. The second is the /management agent controller/, and is the
114-- primary worker process underpinning the management infrastructure. All
115-- published management events are routed to this process, which places them
116-- onto a system wide /event bus/ and additionally passes them directly to the
117-- /trace controller/.
119-- There are several reasons for segregating the tracing and management control
120-- planes in this fashion. Tracing can be enabled or disabled by clients, whilst
121-- the management event bus cannot, since in addition to providing
122-- runtime instrumentation, its intended use-cases include node monitoring, peer
123-- discovery (via topology providing backends) and other essential system
124-- services that require knowledge of otherwise hidden system internals. Tracing
125-- is also subject to /trace flags/ that limit the specific 'MxEvent's delivered
126-- to trace clients - an overhead/complexity not shared by management agents.
127-- Finally, tracing and management agents are implemented using completely
128-- different signalling techniques - more on this later - which would introduce
129-- considerable complexity if the shared the same /event loop/.
131-- The management control plane is driven by a shared broadcast channel, which
132-- is written to by the agent controller and subscribed to by all agent
133-- processes. Agents are spawned as regular processes, whose primary
134-- implementation (i.e., their "server loop") is responsible for consuming
135-- messages from both the broadcast channel and their own mailbox. Once
136-- consumed, messages are applied to the agent's /event sinks/ until one
137-- matches the input, at which point it is applied and the loop continues.
138-- The implementation chooses from the event bus and the mailbox in a
139-- round-robin fashion, until a message is received. This polling activity would
140-- lead to management agents consuming considerable system resources if left
141-- unchecked, therefore the implementation will poll for a limitted number of
142-- retries, after which it will perform a blocking read on the event bus.
144-----------------------------------------------------------------------------
145module Control.Distributed.Process.Management
148 -- * Firing Arbitrary /Mx Events/
150 -- * Constructing Mx Agents
155 , mxAgentWithFinalize
178import Control.Applicative ((<$>))
179import Control.Concurrent.STM (atomically)
180import Control.Concurrent.STM.TChan
186import Control.Distributed.Process.Internal.Primitives
199import Control.Distributed.Process.Internal.Types
206 , unsafeCreateUnencodedMessage
208import Control.Distributed.Process.Management.Bus (publishEvent)
209import qualified Control.Distributed.Process.Management.Table as Table
210import Control.Distributed.Process.Management.Types
214 , ChannelSelector(..)
220import Control.Distributed.Process.Serializable (Serializable)
221import Control.Monad.IO.Class (liftIO)
222import Control.Monad.Reader (ask)
223import qualified Control.Monad.State as ST
232-- | Publishes an arbitrary @Serializable@ message to the management event bus.
233-- Note that /no attempt is made to force the argument/, therefore it is very
234-- important that you do not pass unevaluated thunks that might crash the
235-- receiving process via this API, since /all/ registered agents will gain
236-- access to the data structure once it is broadcast by the agent controller.
237mxNotify :: (Serializable a) => a -> Process ()
239 bus <- localEventBus . processNode <$> ask
240 liftIO $ publishEvent bus $ unsafeCreateUnencodedMessage msg
242-- | Publish an arbitrary @Message@ as a property in the management database.
244-- For publishing @Serializable@ data, use 'mxSet' instead. For publishing
245-- properties efficiently within an agent, use 'mxAgentPublish' instead,
246-- since it will be more efficient in that context.
247mxPublish :: MxAgentId -> String -> Message -> Process ()
248mxPublish a k v = Table.set k v (Table.MxForAgent a)
250-- | Sets an arbitrary @Serializable@ datum against a key in the management
251-- database. Note that /no attempt is made to force the argument/, therefore
252-- it is very important that you do not pass unevaluated thunks that might
253-- crash some other, arbitrary process (or management agent!) that obtains
254-- and attempts to force the value later on.
256mxSet :: Serializable a => MxAgentId -> String -> a -> Process ()
257mxSet mxId key msg = do
258 Table.set key (unsafeCreateUnencodedMessage msg) (Table.MxForAgent mxId)
260-- | Fetches a property from the management database for the given key.
261-- If the property is not set, or does not match the expected type when
262-- typechecked (at runtime), returns @Nothing@.
263mxGet :: Serializable a => MxAgentId -> String -> Process (Maybe a)
264mxGet = Table.fetch . Table.MxForAgent
266-- | Clears a property from the management database using the given key.
267-- If the key does not exist in the database, this is a noop.
268mxClear :: MxAgentId -> String -> Process ()
269mxClear mxId key = Table.clear key (Table.MxForAgent mxId)
271-- | Purges a table in the management database of all its stored properties.
272mxPurgeTable :: MxAgentId -> Process ()
273mxPurgeTable = Table.purge . Table.MxForAgent
275-- | Deletes a table from the management database.
276mxDropTable :: MxAgentId -> Process ()
277mxDropTable = Table.delete . Table.MxForAgent
279--------------------------------------------------------------------------------
280-- API for writing user defined management extensions (i.e., agents) --
281--------------------------------------------------------------------------------
283-- | Return the 'MxAgentId' for the currently executing agent.
285mxGetId :: MxAgent s MxAgentId
286mxGetId = ST.get >>= return . mxAgentId
288-- | The 'MxAgent' version of 'mxNotify'.
290mxBroadcast :: (Serializable m) => m -> MxAgent s ()
293 liftMX $ liftIO $ atomically $ do
294 writeTChan (mxBus state) (unsafeCreateUnencodedMessage msg)
296-- | Gracefully terminate an agent.
298mxDeactivate :: forall s. String -> MxAgent s MxAction
299mxDeactivate = return . MxAgentDeactivate
301-- | Continue executing (i.e., receiving and processing messages).
303mxReady :: forall s. MxAgent s MxAction
304mxReady = return MxAgentReady
306-- | Causes the currently executing /event sink/ to be skipped.
307-- The remaining declared event sinks will be evaluated to find
308-- a matching handler. Can be used to allow multiple event sinks
309-- to process data of the same type.
311mxSkip :: forall s. MxAgent s MxAction
312mxSkip = return MxAgentSkip
314-- | Continue exeucting, prioritising inputs from the process' own
315-- /mailbox/ ahead of data from the management event bus.
317mxReceive :: forall s. MxAgent s MxAction
318mxReceive = return $ MxAgentPrioritise Mailbox
320-- | Continue exeucting, prioritising inputs from the management event bus
321-- over the process' own /mailbox/.
323mxReceiveChan :: forall s. MxAgent s MxAction
324mxReceiveChan = return $ MxAgentPrioritise InputChan
326-- | Lift a @Process@ action.
328liftMX :: Process a -> MxAgent s a
329liftMX p = MxAgent $ ST.lift p
331-- | Set the agent's local state.
333mxSetLocal :: s -> MxAgent s ()
334mxSetLocal s = ST.modify $ \st -> st { mxLocalState = s }
336-- | Update the agent's local state.
338mxUpdateLocal :: (s -> s) -> MxAgent s ()
339mxUpdateLocal f = ST.modify $ \st -> st { mxLocalState = (f $ mxLocalState st) }
341-- | Fetch the agent's local state.
343mxGetLocal :: MxAgent s s
344mxGetLocal = ST.get >>= return . mxLocalState
346-- | Create an 'MxSink' from an expression taking a @Serializable@ type @m@,
347-- that yields an 'MxAction' in the 'MxAgent' monad.
349mxSink :: forall s m . (Serializable m)
350 => (m -> MxAgent s MxAction)
353 msg' <- liftMX $ (unwrapMessage msg :: Process (Maybe m))
355 Nothing -> return Nothing
359 MxAgentSkip -> return Nothing
362-- private ADT: a linked list of event sinks
366 current :: !(MxSink s)
367 , next :: !(MxPipeline s)
370-- | Activates a new agent.
372mxAgent :: MxAgentId -> s -> [MxSink s] -> Process ProcessId
373mxAgent mxId st hs = mxAgentWithFinalize mxId st hs $ return ()
375-- | Activates a new agent. This variant takes a /finalizer/ expression,
376-- that is run once the agent shuts down (even in case of failure/exceptions).
377-- The /finalizer/ expression runs in the mx monad - @MxAgent s ()@ - such
378-- that the agent's internal state remains accessible to the shutdown/cleanup
381mxAgentWithFinalize :: MxAgentId
386mxAgentWithFinalize mxId initState handlers dtor = do
387 let name = agentId mxId
388 existing <- whereis name
390 Just _ -> die "DuplicateAgentId" -- TODO: better error handling policy
392 node <- processNode <$> ask
393 pid <- liftIO $ mxNew (localEventBus node) $ start
397 start (sendTChan, recvTChan) = do
399 nsend Table.mxTableCoordinator (MxAgentStart sp mxId)
400 tablePid <- receiveWait [ matchChan rp (\(p :: ProcessId) -> return p) ]
401 let nState = MxAgentState mxId sendTChan tablePid initState
402 runAgent dtor handlers InputChan recvTChan nState
404 runAgent :: MxAgent s ()
410 runAgent eh hs r c s =
411 runAgentWithFinalizer eh hs r c s
412 `onException` runAgentFinalizer eh s
414 runAgentWithFinalizer :: MxAgent s ()
420 runAgentWithFinalizer eh' hs' cs' c' s' = do
421 (msg, selector') <- getNextInput cs' c'
422 (action, state) <- runPipeline msg s' $ pipeline hs'
424 MxAgentReady -> runAgent eh' hs' selector' c' state
425 MxAgentPrioritise priority -> runAgent eh' hs' priority c' state
426 MxAgentDeactivate _ -> runAgentFinalizer eh' state
427 MxAgentSkip -> error "IllegalState"
428-- MxAgentBecome h' -> runAgent h' c state
430 getNextInput sel chan = getNextInput' sel chan (10 :: Int)
432 -- when reading inputs, we generally want to maintain a degree of
433 -- fairness in choosing between the TChan and our mailbox, but to
434 -- ultimately favour the TChan overall. We do this by flipping
435 -- between the two (using the ChannelSelector) each time we call
436 -- getNextInput - it returns the opposite ChannelSelector to the
437 -- one which succeeded last time it was called.
439 -- This strategy works well, yet we wish to avoid blocking on one
440 -- input if the other is empty/busy, so we begin by reading both
441 -- sources conditionally - tryReadTChan for the event bus and
442 -- receiveTimeout for the mailbox. If polling (either source) does
443 -- not yield an input, we swap to the other, but we cannot continue
444 -- in this fashion ad infinitum, since that would waste considerable
445 -- system resoures. Instead, we switch ten times, after which (if no
446 -- data were obtained) we block on the event bus, since that is our
449 -- An agent can of course, choose to override which source should be
450 -- checked first. We consider this a /hint/ rather than a dictat.
451 -- When reading from the mailbox, we perform a non-blocking read.
452 -- We assume that most agents will prefer using mxReceiveChan to its
453 -- mailbox reading counterpart, this we expect the event bus to be
454 -- non-empty or the entire subsystem is likely doing very little work,
455 -- in which case we needn't worry too much about the overheads
456 -- described thus far.
458 getNextInput' InputChan c' 0 = do
459 m <- liftIO $ atomically $ readTChan c'
461 getNextInput' InputChan c' n = do
462 inputs <- liftIO $ atomically $ tryReadTChan c'
464 Nothing -> getNextInput' Mailbox c' (n - 1)
465 Just m -> return (m, Mailbox)
466 getNextInput' Mailbox c' n = do
467 m <- receiveTimeout 0 [ matchAny return ]
469 Nothing -> getNextInput' InputChan c' (n - 1)
470 Just msg -> return (msg, InputChan)
472 runAgentFinalizer :: MxAgent s () -> MxAgentState s -> Process ()
473 runAgentFinalizer f s = ST.runStateT (unAgent f) s >>= return . fst
475 pipeline :: forall s . [MxSink s] -> MxPipeline s
477 pipeline (sink:sinks) = MxPipeline sink (pipeline sinks)
479 runPipeline :: forall s .
483 -> Process (MxAction, MxAgentState s)
484 runPipeline _ state MxStop = return (MxAgentReady, state)
485 runPipeline msg state MxPipeline{..} = do
486 let act = current msg
487 (pass, state') <- ST.runStateT (unAgent act) state
489 Nothing -> runPipeline msg state next
490 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