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