git.iko.soy/distributed-process
on 2ffa2bc help sign in
distributed-process/src/Control/Distributed/Process/Management.hs
281 lines 9.6 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 ( tryReadTChan
73 , writeTChan
74 )
75import Control.Distributed.Process.Internal.Primitives
76 ( newChan
77 , nsend
78 , receiveWait
79 , receiveTimeout
80 , matchChan
81 , matchAny
82 , unwrapMessage
83 , onException
84 , register
85 )
86import Control.Distributed.Process.Internal.Types
87 ( Process
88 , ProcessId
89 , Message
90 , LocalProcess(..)
91 , LocalNode(..)
92 , MxEventBus(..)
93 , unsafeCreateUnencodedMessage
94 )
95import Control.Distributed.Process.Management.Bus (publishEvent)
96import qualified Control.Distributed.Process.Management.Table as Table
97import Control.Distributed.Process.Management.Types
98 ( MxAgentId(..)
99 , MxAgent(..)
100 , MxAction(..)
101 , MxAgentState(..)
102 , MxAgentStart(..)
103 , MxSink
104 , MxEvent(..)
105 )
106import Control.Distributed.Process.Serializable (Serializable)
107import Control.Monad.IO.Class (liftIO)
108import Control.Monad.Reader (ask)
109import qualified Control.Monad.State as ST
110 ( MonadState
111 , StateT
112 , get
113 , modify
114 , lift
115 , runStateT
116 )
118-- | Publishes an arbitrary @Serializable@ message to the management event bus.
119-- Note that /no attempt is made to force the argument/, therefore it is very
120-- important that you do not pass unevaluated thunks that might crash the
121-- receiving process via this API, since /all/ registered agents will gain
122-- access to the data structure once it is broadcast by the agent controller.
123mxNotify :: (Serializable a) => a -> Process ()
124mxNotify msg = do
125 bus <- localEventBus . processNode <$> ask
126 liftIO $ publishEvent bus $ unsafeCreateUnencodedMessage msg
128-- | Publish an arbitrary @Message@ as a property in the management database.
129--
130-- For publishing @Serializable@ data, use 'mxSet' instead. For publishing
131-- properties efficiently within an agent, use 'mxAgentPublish' instead,
132-- since it will be more efficient in that context.
133mxPublish :: MxAgentId -> String -> Message -> Process ()
134mxPublish a k v = Table.set k v (Table.MxForAgent a)
136-- | Sets an arbitrary @Serializable@ datum against a key in the management
137-- database. Note that /no attempt is made to force the argument/, therefore
138-- it is very important that you do not pass unevaluated thunks that might
139-- crash some other, arbitrary process (or management agent!) that obtains
140-- and attempts to force the value later on.
141--
142mxSet :: Serializable a => MxAgentId -> String -> a -> Process ()
143mxSet mxId key msg = do
144 Table.set key (unsafeCreateUnencodedMessage msg) (Table.MxForAgent mxId)
146-- | Fetches a property from the management database for the given key.
147-- If the property is not set, or does not match the expected type when
148-- typechecked (at runtime), returns @Nothing@.
149mxGet :: Serializable a => MxAgentId -> String -> Process (Maybe a)
150mxGet = Table.fetch . Table.MxForAgent
152-- | Clears a property from the management database using the given key.
153-- If the key does not exist in the database, this is a noop.
154mxClear :: MxAgentId -> String -> Process ()
155mxClear mxId key = Table.clear key (Table.MxForAgent mxId)
157-- | Purges a table in the management database of all its stored properties.
158mxPurgeTable :: MxAgentId -> Process ()
159mxPurgeTable = Table.purge . Table.MxForAgent
161-- | Deletes a table from the management database.
162mxDropTable :: MxAgentId -> Process ()
163mxDropTable = Table.delete . Table.MxForAgent
165--------------------------------------------------------------------------------
166-- API for writing user defined management extensions (i.e., agents) --
167--------------------------------------------------------------------------------
169mxGetId :: MxAgent s MxAgentId
170mxGetId = ST.get >>= return . mxAgentId
172mxBroadcast :: (Serializable m) => m -> MxAgent s ()
173mxBroadcast msg = do
174 state <- ST.get
175 liftMX $ liftIO $ atomically $ do
176 writeTChan (mxBus state) (unsafeCreateUnencodedMessage msg)
178mxDeactivate :: forall s. String -> MxAgent s MxAction
179mxDeactivate = return . MxAgentDeactivate
181mxReady :: forall s. MxAgent s MxAction
182mxReady = return MxAgentReady
184liftMX :: Process a -> MxAgent s a
185liftMX p = MxAgent $ ST.lift p
187mxSetLocal :: s -> MxAgent s ()
188mxSetLocal s = ST.modify $ \st -> st { mxLocalState = s }
190mxGetLocal :: MxAgent s s
191mxGetLocal = ST.get >>= return . mxLocalState
193mxSink :: forall s m . (Serializable m)
194 => (m -> MxAgent s MxAction)
195 -> MxSink s
196mxSink act msg = do
197 msg' <- liftMX $ (unwrapMessage msg :: Process (Maybe m))
198 case msg' of
199 Nothing -> return Nothing
200 Just m -> act m >>= return . Just
202data MxPipeline s =
203 MxPipeline
204 {
205 current :: !(MxSink s)
206 , next :: !(MxPipeline s)
207 } | MxStop
209-- | Activates a new agent.
210--
211mxAgent :: MxAgentId -> s -> [MxSink s] -> Process ProcessId
212mxAgent mxId st hs = mxAgentWithFinalize mxId st hs $ return ()
214-- | Activates a new agent. This variant takes a /finalizer/ expression,
215-- that is run once the agent shuts down (even in case of failure/exceptions).
216-- The /finalizer/ expression runs in the mx monad - @MxAgent s ()@ - such
217-- that the agent's internal state remains accessible to the shutdown/cleanup
218-- code.
219--
220mxAgentWithFinalize :: MxAgentId
221 -> s
222 -> [MxSink s]
223 -> MxAgent s ()
224 -> Process ProcessId
225mxAgentWithFinalize mxId initState handlers dtor = do
226 node <- processNode <$> ask
227 pid <- liftIO $ mxNew (localEventBus node) $ start
228 register (agentId mxId) pid
229 return pid
230 where
231 start (sendTChan, recvTChan) = do
232 (sp, rp) <- newChan
233 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)
237 tablePid <- receiveWait [ matchChan rp (\(p :: ProcessId) -> return p) ]
238 -- liftIO $ putStrLn "starting agent listener..."
239 runAgent dtor handlers recvTChan $ MxAgentState mxId sendTChan tablePid initState
241 runAgent eh hs c s =
242 runAgentWithFinalizer eh hs c s `onException` runAgentFinalizer eh s
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
252 getNextMessage tch = do
253 inputs <- liftIO $ atomically $ tryReadTChan tch
254 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
262 runAgentFinalizer :: MxAgent s () -> MxAgentState s -> Process ()
263 runAgentFinalizer f s = ST.runStateT (unAgent f) s >>= return . fst
265 pipeline :: forall s . [MxSink s] -> MxPipeline s
266 pipeline [] = MxStop
267 pipeline (sink:sinks) = MxPipeline sink (pipeline sinks)
269 runPipeline :: forall s .
270 Message
271 -> MxAgentState s
272 -> MxPipeline s
273 -> Process (MxAction, MxAgentState s)
274 runPipeline _ state MxStop = return (MxAgentReady, state)
275 runPipeline msg state MxPipeline{..} = do
276 let act = current msg
277 (pass, state') <- ST.runStateT (unAgent act) state
278 case pass of
279 Nothing -> runPipeline msg state next
280 Just result -> return (result, state')