git.iko.soy/distributed-process
on 4957534 help sign in
distributed-process/src/Control/Distributed/Process/Management.hs
234 lines 7.7 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 , MxSink
51 , mxSink
52 , mxGetId
53 , mxDeactivate
54 , mxReady
55 , mxBroadcast
56 , mxSetLocal
57 , mxGetLocal
58 , mxLift
59 -- * Mx Data API
60 , mxPublish
61 , mxSet
62 , mxGet
63 ) where
65import Control.Applicative ((<$>))
66import Control.Concurrent.STM (atomically)
67import Control.Concurrent.STM.TChan
68 ( readTChan
69 , writeTChan
70 )
71import Control.Distributed.Process.Internal.Primitives
72 ( newChan
73 , nsend
74 , receiveWait
75 , matchChan
76 , unwrapMessage
77 )
78import Control.Distributed.Process.Internal.Types
79 ( Process
80 , ProcessId
81 , Message
82 , LocalProcess(..)
83 , LocalNode(..)
84 , MxEventBus(..)
85 , unsafeCreateUnencodedMessage
86 )
87import Control.Distributed.Process.Management.Bus (publishEvent)
88import qualified Control.Distributed.Process.Management.Table as Table
89import Control.Distributed.Process.Management.Types
90 ( MxAgentId(..)
91 , MxAgent(..)
92 , MxAction(..)
93 , MxAgentState(..)
94 , MxAgentStart(..)
95 , MxSink
96 , MxEvent(..)
97 )
98import Control.Distributed.Process.Serializable (Serializable)
99import Control.Monad.IO.Class (liftIO)
100import Control.Monad.Reader (ask)
101import qualified Control.Monad.State as ST
102 ( MonadState
103 , StateT
104 , get
105 , modify
106 , lift
107 , runStateT
108 )
110-- | Publishes an arbitrary @Serializable@ message to the management event bus.
111-- Note that /no attempt is made to force the argument/, therefore it is very
112-- important that you do not pass unevaluated thunks that might crash the
113-- receiving process via this API, since /all/ registered agents will gain
114-- access to the data structure once it is broadcast by the agent controller.
115mxNotify :: (Serializable a) => a -> Process ()
116mxNotify msg = do
117 bus <- localEventBus . processNode <$> ask
118 liftIO $ publishEvent bus $ unsafeCreateUnencodedMessage msg
120-- | Publish an arbitrary @Message@ as a property in the management database.
121--
122-- For publishing @Serializable@ data, use 'mxSet' instead. For publishing
123-- properties efficiently within an agent, use 'mxAgentPublish' instead,
124-- since it will be more efficient in that context.
125mxPublish :: MxAgentId -> String -> Message -> Process ()
126mxPublish a k v = Table.set k v (Table.MxForAgent a)
128-- | Sets an arbitrary @Serializable@ datum against a key in the management
129-- database. Note that /no attempt is made to force the argument/, therefore
130-- it is very important that you do not pass unevaluated thunks that might
131-- crash some other, arbitrary process (or management agent!) that obtains
132-- and attempts to force the value later on.
133--
134mxSet :: Serializable a => MxAgentId -> String -> a -> Process ()
135mxSet mxId key msg =
136 Table.set key (unsafeCreateUnencodedMessage msg) (Table.MxForAgent mxId)
138-- | Fetches a property from the management database for the given key.
139-- If the property is not set, or does not match the expected type when
140-- typechecked (at runtime), returns @Nothing@.
141mxGet :: Serializable a => MxAgentId -> String -> Process (Maybe a)
142mxGet mxId = Table.fetch (Table.MxForAgent mxId)
144--------------------------------------------------------------------------------
145-- API for writing user defined management extensions (i.e., agents) --
146--------------------------------------------------------------------------------
148mxGetId :: MxAgent s MxAgentId
149mxGetId = ST.get >>= return . mxAgentId
151mxBroadcast :: (Serializable m) => m -> MxAgent s ()
152mxBroadcast msg = do
153 state <- ST.get
154 mxLift $ liftIO $ atomically $ do
155 writeTChan (mxBus state) (unsafeCreateUnencodedMessage msg)
157mxDeactivate :: forall s. String -> MxAgent s MxAction
158mxDeactivate = return . MxAgentDeactivate
160mxReady :: forall s. MxAgent s MxAction
161mxReady = return MxAgentReady
163mxLift :: Process a -> MxAgent s a
164mxLift p = MxAgent $ ST.lift p
166mxSetLocal :: s -> MxAgent s ()
167mxSetLocal s = ST.modify $ \st -> st { mxLocalState = s }
169mxGetLocal :: MxAgent s s
170mxGetLocal = ST.get >>= return . mxLocalState
172mxSink :: forall s m . (Serializable m)
173 => (m -> MxAgent s MxAction)
174 -> MxSink s
175mxSink act msg = do
176 msg' <- mxLift $ (unwrapMessage msg :: Process (Maybe m))
177 case msg' of
178 Nothing -> return Nothing
179 Just m -> act m >>= return . Just
181data MxPipeline s =
182 MxPipeline
183 {
184 current :: !(MxSink s)
185 , next :: !(MxPipeline s)
186 } | MxStop
188-- | Activates a new agent.
189--
190mxAgent :: MxAgentId
191 -> s
192 -> [MxSink s]
193 -> Process ProcessId
194mxAgent mxId initState handlers = do
195 node <- processNode <$> ask
196 pid <- liftIO $ mxNew (localEventBus node) $ start
197 return pid
198 where
199 start (sendTChan, recvTChan) = do
200 (sp, rp) <- newChan
201 nsend Table.mxTableCoordinator (MxAgentStart sp mxId)
202 tablePid <- receiveWait [ matchChan rp (\(p :: ProcessId) -> return p) ]
203 runAgent handlers recvTChan $ MxAgentState mxId sendTChan tablePid initState
205 runAgent hs c s = do
206 msg <- (liftIO $ atomically $ readTChan c)
207 (action, state) <- runPipeline msg s $ pipeline hs
208 case action of
209 MxAgentReady -> runAgent hs c state
210 MxAgentDeactivate _ -> {- TODO: log r -} return ()
211-- MxAgentBecome h' -> runAgent h' c state
213 pipeline :: forall s . [MxSink s] -> MxPipeline s
214 pipeline [] = MxStop
215 pipeline (sink:sinks) = MxPipeline sink (pipeline sinks)
217 runPipeline :: forall s .
218 Message
219 -> MxAgentState s
220 -> MxPipeline s
221 -> Process (MxAction, MxAgentState s)
222 runPipeline _ state MxStop = return (MxAgentReady, state)
223 runPipeline msg state MxPipeline{..} = do
224 let act = current msg
225 (pass, state') <- ST.runStateT (unAgent act) state
226 case pass of
227 Nothing -> runPipeline msg state next
228 Just result -> return (result, state')
230 runAgentST :: MxAgentState s
231 -> MxAgent s MxAction
232 -> Process (MxAction, MxAgentState s)
233 runAgentST state proc = ST.runStateT (unAgent proc) state