@@ -4,6 +4,7 @@
44{-# LANGUAGE ExistentialQuantification #-}
55{-# LANGUAGE ScopedTypeVariables #-}
66{-# LANGUAGE RecordWildCards #-}
7+{-# LANGUAGE PatternGuards #-}
78{-# LANGUAGE FlexibleInstances #-}
89{-# LANGUAGE UndecidableInstances #-}
910
@@ -17,24 +18,128 @@
1718-- Stability : experimental
1819-- Portability : non-portable (requires concurrency)
1920--
20−-- /Management Extensions/ for Cloud Haskell.
21+-- [Management Extensions API]
2122--
22−-- This module provides a standard API for receiving and/or responding to
23−-- system events in a Cloud Haskell node.
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.
28+--
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.
38+--
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).
44+--
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.
53+--
54+-- [Management Data API]
55+--
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.
61+--
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.
65+--
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.
70+--
71+-- [Defining Agents]
72+--
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.
78+--
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').
82+--
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.
2491--
2592-- [Architecture Overview]
2693--
27−-- /Management Extensions/ can be used by infrastructure or application code
94+-- /Management Extensions/ can be used by infrastructure or application code,
2895-- to provide access to runtime instrumentation data, expose control planes
2996-- and/or provide remote access to external clients.
3097--
3198-- The management infrastructure is broken down into three components, which
32−-- roughly correspond to the modules it exposes:
99+-- roughly correspond to the modules in which they're implemented:
100+--
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
104+--
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.
108+--
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/.
33118--
34−-- 1. @Agent@ - registration and management of /Management Agents/
35−-- 2. @Instrumentation@ - provides an API for publishing /instrumentation data/
36−-- 3. @Remote@ - provides an API for remote management and administration
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/.
37130--
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.
38143--
39144-----------------------------------------------------------------------------
40145module Control.Distributed.Process.Management
@@ -48,14 +153,18 @@ module Control.Distributed.Process.Management
48153 , MxAgent()
49154 , mxAgent
50155 , mxAgentWithFinalize
51− , MxSink
156+ , MxSink()
52157 , mxSink
53158 , mxGetId
54159 , mxDeactivate
55160 , mxReady
161+ , mxSkip
162+ , mxReceive
163+ , mxReceiveChan
56164 , mxBroadcast
57165 , mxSetLocal
58166 , mxGetLocal
167+ , mxUpdateLocal
59168 , liftMX
60169 -- * Mx Data API
61170 , mxPublish
@@ -70,7 +179,9 @@ import Control.Applicative ((<$>))
70179import Control.Concurrent.STM (atomically)
71180import Control.Concurrent.STM.TChan
72181 ( tryReadTChan
182+ , readTChan
73183 , writeTChan
184+ , TChan
74185 )
75186import Control.Distributed.Process.Internal.Primitives
76187 ( newChan
@@ -82,6 +193,8 @@ import Control.Distributed.Process.Internal.Primitives
82193 , unwrapMessage
83194 , onException
84195 , register
196+ , whereis
197+ , die
85198 )
86199import Control.Distributed.Process.Internal.Types
87200 ( Process
@@ -98,6 +211,7 @@ import Control.Distributed.Process.Management.Types
98211 ( MxAgentId(..)
99212 , MxAgent(..)
100213 , MxAction(..)
214+ , ChannelSelector(..)
101215 , MxAgentState(..)
102216 , MxAgentStart(..)
103217 , MxSink
@@ -166,30 +280,72 @@ mxDropTable = Table.delete . Table.MxForAgent
166280-- API for writing user defined management extensions (i.e., agents) --
167281--------------------------------------------------------------------------------
168282
283+-- | Return the 'MxAgentId' for the currently executing agent.
284+--
169285mxGetId :: MxAgent s MxAgentId
170286mxGetId = ST.get >>= return . mxAgentId
171287
288+-- | The 'MxAgent' version of 'mxNotify'.
289+--
172290mxBroadcast :: (Serializable m) => m -> MxAgent s ()
173291mxBroadcast msg = do
174292 state <- ST.get
175293 liftMX $ liftIO $ atomically $ do
176294 writeTChan (mxBus state) (unsafeCreateUnencodedMessage msg)
177295
296+-- | Gracefully terminate an agent.
297+--
178298mxDeactivate :: forall s. String -> MxAgent s MxAction
179299mxDeactivate = return . MxAgentDeactivate
180300
301+-- | Continue executing (i.e., receiving and processing messages).
302+--
181303mxReady :: forall s. MxAgent s MxAction
182304mxReady = return MxAgentReady
183305
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.
310+--
311+mxSkip :: forall s. MxAgent s MxAction
312+mxSkip = return MxAgentSkip
313+
314+-- | Continue exeucting, prioritising inputs from the process' own
315+-- /mailbox/ ahead of data from the management event bus.
316+--
317+mxReceive :: forall s. MxAgent s MxAction
318+mxReceive = return $ MxAgentPrioritise Mailbox
319+
320+-- | Continue exeucting, prioritising inputs from the management event bus
321+-- over the process' own /mailbox/.
322+--
323+mxReceiveChan :: forall s. MxAgent s MxAction
324+mxReceiveChan = return $ MxAgentPrioritise InputChan
325+
326+-- | Lift a @Process@ action.
327+--
184328liftMX :: Process a -> MxAgent s a
185329liftMX p = MxAgent $ ST.lift p
186330
331+-- | Set the agent's local state.
332+--
187333mxSetLocal :: s -> MxAgent s ()
188334mxSetLocal s = ST.modify $ \st -> st { mxLocalState = s }
189335
336+-- | Update the agent's local state.
337+--
338+mxUpdateLocal :: (s -> s) -> MxAgent s ()
339+mxUpdateLocal f = ST.modify $ \st -> st { mxLocalState = (f $ mxLocalState st) }
340+
341+-- | Fetch the agent's local state.
342+--
190343mxGetLocal :: MxAgent s s
191344mxGetLocal = ST.get >>= return . mxLocalState
192345
346+-- | Create an 'MxSink' from an expression taking a @Serializable@ type @m@,
347+-- that yields an 'MxAction' in the 'MxAgent' monad.
348+--
193349mxSink :: forall s m . (Serializable m)
194350 => (m -> MxAgent s MxAction)
195351 -> MxSink s
@@ -197,8 +353,13 @@ mxSink act msg = do
197353 msg' <- liftMX $ (unwrapMessage msg :: Process (Maybe m))
198354 case msg' of
199355 Nothing -> return Nothing
200− Just m -> act m >>= return . Just
356+ Just m -> do
357+ r <- act m
358+ case r of
359+ MxAgentSkip -> return Nothing
360+ _ -> return $ Just r
201361
362+-- private ADT: a linked list of event sinks
202363data MxPipeline s =
203364 MxPipeline
204365 {
@@ -223,41 +384,90 @@ mxAgentWithFinalize :: MxAgentId
223384 -> MxAgent s ()
224385 -> Process ProcessId
225386mxAgentWithFinalize mxId initState handlers dtor = do
226− node <- processNode <$> ask
227− pid <- liftIO $ mxNew (localEventBus node) $ start
228− register (agentId mxId) pid
229− return pid
387+ let name = agentId mxId
388+ existing <- whereis name
389+ case existing of
390+ Just _ -> die "DuplicateAgentId" -- TODO: better error handling policy
391+ Nothing -> do
392+ node <- processNode <$> ask
393+ pid <- liftIO $ mxNew (localEventBus node) $ start
394+ register name pid
395+ return pid
230396 where
231397 start (sendTChan, recvTChan) = do
232398 (sp, rp) <- newChan
233399 nsend Table.mxTableCoordinator (MxAgentStart sp mxId)
234− -- liftIO $ putStrLn $ "waiting on table coordinator " ++ Table.mxTableCoordinator
235− -- p <- whereis Table.mxTableCoordinator
236− -- liftIO $ putStrLn $ "registration == " ++ (show p)
237400 tablePid <- receiveWait [ matchChan rp (\(p :: ProcessId) -> return p) ]
238− -- liftIO $ putStrLn "starting agent listener..."
239− runAgent dtor handlers recvTChan $ MxAgentState mxId sendTChan tablePid initState
240−
241− runAgent eh hs c s =
242− runAgentWithFinalizer eh hs c s `onException` runAgentFinalizer eh s
243−
244− runAgentWithFinalizer eh' hs' c' s' = do
245− msg <- getNextMessage c'
246− (action, state) <- runPipeline msg s' $ pipeline hs'
247− case action of
248− MxAgentReady -> runAgent eh' hs' c' state
249− MxAgentDeactivate _ -> runAgentFinalizer eh' state
250−-- MxAgentBecome h' -> runAgent h' c state
251−
252− getNextMessage tch = do
253− inputs <- liftIO $ atomically $ tryReadTChan tch
401+ let nState = MxAgentState mxId sendTChan tablePid initState
402+ runAgent dtor handlers InputChan recvTChan nState
403+
404+ runAgent :: MxAgent s ()
405+ -> [MxSink s]
406+ -> ChannelSelector
407+ -> TChan Message
408+ -> MxAgentState s
409+ -> Process ()
410+ runAgent eh hs r c s =
411+ runAgentWithFinalizer eh hs r c s
412+ `onException` runAgentFinalizer eh s
413+
414+ runAgentWithFinalizer :: MxAgent s ()
415+ -> [MxSink s]
416+ -> ChannelSelector
417+ -> TChan Message
418+ -> MxAgentState s
419+ -> Process ()
420+ runAgentWithFinalizer eh' hs' cs' c' s' = do
421+ (msg, selector') <- getNextInput cs' c'
422+ (action, state) <- runPipeline msg s' $ pipeline hs'
423+ case action of
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
429+
430+ getNextInput sel chan = getNextInput' sel chan (10 :: Int)
431+
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.
438+ --
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
447+ -- main priority.
448+ --
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.
457+
458+ getNextInput' InputChan c' 0 = do
459+ m <- liftIO $ atomically $ readTChan c'
460+ return (m, Mailbox)
461+ getNextInput' InputChan c' n = do
462+ inputs <- liftIO $ atomically $ tryReadTChan c'
254463 case inputs of
255− Nothing -> do
256− m <- receiveTimeout 0 [ matchAny return ]
257− case m of
258− Nothing -> getNextMessage tch
259− Just msg -> return msg
260− Just m -> return m
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 ]
468+ case m of
469+ Nothing -> getNextInput' InputChan c' (n - 1)
470+ Just msg -> return (msg, InputChan)
261471
262472 runAgentFinalizer :: MxAgent s () -> MxAgentState s -> Process ()
263473 runAgentFinalizer f s = ST.runStateT (unAgent f) s >>= return . fst