git.iko.soy/distributed-process
on e52ec08 help sign in
distributed-process/src/Control/Distributed/Process/Management.hs
267 lines 9.2 kB history raw
1{-# LANGUAGE DeriveDataTypeable #-}
2{-# LANGUAGE DeriveGeneric #-}
3{-# LANGUAGE StandaloneDeriving #-}
4{-# LANGUAGE ExistentialQuantification #-}
5{-# LANGUAGE ScopedTypeVariables #-}
6{-# LANGUAGE RecordWildCards #-}
7{-# LANGUAGE FlexibleInstances #-}
8{-# LANGUAGE UndecidableInstances #-}
10-----------------------------------------------------------------------------
11-- |
12-- Module : Control.Distributed.Process.Management
13-- Copyright : (c) Well-Typed / Tim Watson
14-- License : BSD3 (see the file LICENSE)
15--
16-- Maintainer : Tim Watson <watson.timothy@gmail.com>
17-- Stability : experimental
18-- Portability : non-portable (requires concurrency)
19--
20-- /Management Extensions/ for Cloud Haskell.
21--
22-- This module provides a standard API for receiving and/or responding to
23-- system events in a Cloud Haskell node.
24--
25-- [Architecture Overview]
26--
27-- /Management Extensions/ can be used by infrastructure or application code
28-- to provide access to runtime instrumentation data, expose control planes
29-- and/or provide remote access to external clients.
30--
31-- The management infrastructure is broken down into three components, which
32-- roughly correspond to the modules it exposes:
33--
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
37--
38--
39-----------------------------------------------------------------------------
40module Control.Distributed.Process.Management
41 (
42 MxEvent(..)
43 -- * Firing Arbitrary /Mx Events/
44 , mxNotify
45 -- * Constructing Mx Agents
46 , MxAction()
47 , MxAgentId(..)
48 , MxAgent()
49 , mxAgent
50 , mxAgentWithFinalize
51 , MxSink
52 , mxSink
53 , mxGetId
54 , mxDeactivate
55 , mxReady
56 , mxBroadcast
57 , mxSetLocal
58 , mxGetLocal
59 , liftMX
60 -- * Mx Data API
61 , mxPublish
62 , mxSet
63 , mxGet
64 , mxClear
65 , mxPurgeTable
66 , mxDropTable
67 ) where
69import Control.Applicative ((<$>))
70import Control.Concurrent.STM (atomically)
71import Control.Concurrent.STM.TChan
72 ( readTChan
73 , writeTChan
74 )
75import Control.Distributed.Process.Internal.Primitives
76 ( newChan
77 , nsend
78 , receiveWait
79 , matchChan
80 , unwrapMessage
81 , onException
82 )
83import Control.Distributed.Process.Internal.Types
84 ( Process
85 , ProcessId
86 , Message
87 , LocalProcess(..)
88 , LocalNode(..)
89 , MxEventBus(..)
90 , unsafeCreateUnencodedMessage
91 )
92import Control.Distributed.Process.Management.Bus (publishEvent)
93import qualified Control.Distributed.Process.Management.Table as Table
94import Control.Distributed.Process.Management.Types
95 ( MxAgentId(..)
96 , MxAgent(..)
97 , MxAction(..)
98 , MxAgentState(..)
99 , MxAgentStart(..)
100 , MxSink
101 , MxEvent(..)
102 )
103import Control.Distributed.Process.Serializable (Serializable)
104import Control.Monad.IO.Class (liftIO)
105import Control.Monad.Reader (ask)
106import qualified Control.Monad.State as ST
107 ( MonadState
108 , StateT
109 , get
110 , modify
111 , lift
112 , runStateT
113 )
115-- | Publishes an arbitrary @Serializable@ message to the management event bus.
116-- Note that /no attempt is made to force the argument/, therefore it is very
117-- important that you do not pass unevaluated thunks that might crash the
118-- receiving process via this API, since /all/ registered agents will gain
119-- access to the data structure once it is broadcast by the agent controller.
120mxNotify :: (Serializable a) => a -> Process ()
121mxNotify msg = do
122 bus <- localEventBus . processNode <$> ask
123 liftIO $ publishEvent bus $ unsafeCreateUnencodedMessage msg
125-- | Publish an arbitrary @Message@ as a property in the management database.
126--
127-- For publishing @Serializable@ data, use 'mxSet' instead. For publishing
128-- properties efficiently within an agent, use 'mxAgentPublish' instead,
129-- since it will be more efficient in that context.
130mxPublish :: MxAgentId -> String -> Message -> Process ()
131mxPublish a k v = Table.set k v (Table.MxForAgent a)
133-- | Sets an arbitrary @Serializable@ datum against a key in the management
134-- database. Note that /no attempt is made to force the argument/, therefore
135-- it is very important that you do not pass unevaluated thunks that might
136-- crash some other, arbitrary process (or management agent!) that obtains
137-- and attempts to force the value later on.
138--
139mxSet :: Serializable a => MxAgentId -> String -> a -> Process ()
140mxSet mxId key msg = do
141 Table.set key (unsafeCreateUnencodedMessage msg) (Table.MxForAgent mxId)
143-- | Fetches a property from the management database for the given key.
144-- If the property is not set, or does not match the expected type when
145-- typechecked (at runtime), returns @Nothing@.
146mxGet :: Serializable a => MxAgentId -> String -> Process (Maybe a)
147mxGet = Table.fetch . Table.MxForAgent
149-- | Clears a property from the management database using the given key.
150-- If the key does not exist in the database, this is a noop.
151mxClear :: MxAgentId -> String -> Process ()
152mxClear mxId key = Table.clear key (Table.MxForAgent mxId)
154-- | Purges a table in the management database of all its stored properties.
155mxPurgeTable :: MxAgentId -> Process ()
156mxPurgeTable = Table.purge . Table.MxForAgent
158-- | Deletes a table from the management database.
159mxDropTable :: MxAgentId -> Process ()
160mxDropTable = Table.delete . Table.MxForAgent
162--------------------------------------------------------------------------------
163-- API for writing user defined management extensions (i.e., agents) --
164--------------------------------------------------------------------------------
166mxGetId :: MxAgent s MxAgentId
167mxGetId = ST.get >>= return . mxAgentId
169mxBroadcast :: (Serializable m) => m -> MxAgent s ()
170mxBroadcast msg = do
171 state <- ST.get
172 liftMX $ liftIO $ atomically $ do
173 writeTChan (mxBus state) (unsafeCreateUnencodedMessage msg)
175mxDeactivate :: forall s. String -> MxAgent s MxAction
176mxDeactivate = return . MxAgentDeactivate
178mxReady :: forall s. MxAgent s MxAction
179mxReady = return MxAgentReady
181liftMX :: Process a -> MxAgent s a
182liftMX p = MxAgent $ ST.lift p
184mxSetLocal :: s -> MxAgent s ()
185mxSetLocal s = ST.modify $ \st -> st { mxLocalState = s }
187mxGetLocal :: MxAgent s s
188mxGetLocal = ST.get >>= return . mxLocalState
190mxSink :: forall s m . (Serializable m)
191 => (m -> MxAgent s MxAction)
192 -> MxSink s
193mxSink act msg = do
194 msg' <- liftMX $ (unwrapMessage msg :: Process (Maybe m))
195 case msg' of
196 Nothing -> return Nothing
197 Just m -> act m >>= return . Just
199data MxPipeline s =
200 MxPipeline
201 {
202 current :: !(MxSink s)
203 , next :: !(MxPipeline s)
204 } | MxStop
206-- | Activates a new agent.
207--
208mxAgent :: MxAgentId -> s -> [MxSink s] -> Process ProcessId
209mxAgent mxId st hs = mxAgentWithFinalize mxId st hs $ return ()
211-- | Activates a new agent. This variant takes a /finalizer/ expression,
212-- that is run once the agent shuts down (even in case of failure/exceptions).
213-- The /finalizer/ expression runs in the mx monad - @MxAgent s ()@ - such
214-- that the agent's internal state remains accessible to the shutdown/cleanup
215-- code.
216--
217mxAgentWithFinalize :: MxAgentId
218 -> s
219 -> [MxSink s]
220 -> MxAgent s ()
221 -> Process ProcessId
222mxAgentWithFinalize mxId initState handlers dtor = do
223 node <- processNode <$> ask
224 pid <- liftIO $ mxNew (localEventBus node) $ start
225 return pid
226 where
227 start (sendTChan, recvTChan) = do
228 (sp, rp) <- newChan
229 nsend Table.mxTableCoordinator (MxAgentStart sp mxId)
230 -- liftIO $ putStrLn $ "waiting on table coordinator " ++ Table.mxTableCoordinator
231 -- p <- whereis Table.mxTableCoordinator
232 -- liftIO $ putStrLn $ "registration == " ++ (show p)
233 tablePid <- receiveWait [ matchChan rp (\(p :: ProcessId) -> return p) ]
234 -- liftIO $ putStrLn "starting agent listener..."
235 runAgent dtor handlers recvTChan $ MxAgentState mxId sendTChan tablePid initState
237 runAgent eh hs c s =
238 runAgentWithFinalizer eh hs c s `onException` runAgentFinalizer eh s
240 runAgentWithFinalizer eh' hs' c' s' = do
241 msg <- (liftIO $ atomically $ readTChan c')
242 (action, state) <- runPipeline msg s' $ pipeline hs'
243 case action of
244 MxAgentReady -> runAgent eh' hs' c' state
245 MxAgentDeactivate _ -> runAgentFinalizer eh' state
246-- MxAgentBecome h' -> runAgent h' c state
248 runAgentFinalizer :: MxAgent s () -> MxAgentState s -> Process ()
249 runAgentFinalizer f s = ST.runStateT (unAgent f) s >>= return . fst
251 pipeline :: forall s . [MxSink s] -> MxPipeline s
252 pipeline [] = MxStop
253 pipeline (sink:sinks) = MxPipeline sink (pipeline sinks)
255 runPipeline :: forall s .
256 Message
257 -> MxAgentState s
258 -> MxPipeline s
259 -> Process (MxAction, MxAgentState s)
260 runPipeline _ state MxStop = return (MxAgentReady, state)
261 runPipeline msg state MxPipeline{..} = do
262 let act = current msg
263 (pass, state') <- ST.runStateT (unAgent act) state
264 case pass of
265 Nothing -> runPipeline msg state next
266 Just result -> return (result, state')