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