git.iko.soy/distributed-process
on 0d2e609 help sign in
distributed-process/src/Control/Distributed/Process/Management.hs
190 lines 6.5 kB history raw
1{-# LANGUAGE DeriveDataTypeable #-}
2{-# LANGUAGE DeriveGeneric #-}
3{-# LANGUAGE StandaloneDeriving #-}
4{-# LANGUAGE ExistentialQuantification #-}
5{-# LANGUAGE ScopedTypeVariables #-}
6{-# LANGUAGE PatternGuards #-}
7{-# LANGUAGE RecordWildCards #-}
8{-# LANGUAGE FlexibleInstances #-}
9{-# LANGUAGE UndecidableInstances #-}
11-----------------------------------------------------------------------------
12-- |
13-- Module : Control.Distributed.Process.Management
14-- Copyright : (c) Well-Typed / Tim Watson
15-- License : BSD3 (see the file LICENSE)
16--
17-- Maintainer : Tim Watson <watson.timothy@gmail.com>
18-- Stability : experimental
19-- Portability : non-portable (requires concurrency)
20--
21-- /Management Extensions/ for Cloud Haskell.
22--
23-- This module provides a standard API for receiving and/or responding to
24-- system events in a Cloud Haskell node.
25--
26-- [Architecture Overview]
27--
28-- /Management Extensions/ can be used by infrastructure or application code
29-- to provide access to runtime instrumentation data, expose control planes
30-- and/or provide remote access to external clients.
31--
32-- The management infrastructure is broken down into three components, which
33-- roughly correspond to the modules it exposes:
34--
35-- 1. @Agent@ - registration and management of /Management Agents/
36-- 2. @Instrumentation@ - provides an API for publishing /instrumentation data/
37-- 3. @Remote@ - provides an API for remote management and administration
38--
39--
40-----------------------------------------------------------------------------
41module Control.Distributed.Process.Management
42 (
43 MxAction()
44 , MxAgentId
45 -- * Firing /mx events/
46 , mxNotify
47 -- * Constructing Agents
48 , mxAgent
49 , mxAgentDeactivate
50 , mxAgentReady
51 , mxAgentPublish
52 , mxSetLocal
53 , mxGetLocal
54 -- * Lifting
55 , liftP
56 -- * Management Data API
57 , mxPublish
58 , mxSet
59 , mxGet
60 ) where
62import Control.Applicative ((<$>))
63import Control.Concurrent.STM (atomically)
64import Control.Concurrent.STM.TChan
65 ( readTChan
66 )
67import Control.Distributed.Process.Internal.Primitives
68 ( newChan
69 , nsend
70 , receiveWait
71 , matchChan
72 )
73import Control.Distributed.Process.Internal.Types
74 ( Process
75 , ProcessId
76 , Message
77 , LocalProcess(..)
78 , LocalNode(..)
79 , MxEventBus(..)
80 , unsafeCreateUnencodedMessage
81 )
82import Control.Distributed.Process.Management.Bus (publishEvent)
83import qualified Control.Distributed.Process.Management.Table as Table
84import Control.Distributed.Process.Management.Types
85 ( MxAgentId(..)
86 , MxAgent(..)
87 , MxAgentState(..)
88 , MxAgentStart(..)
89 )
90import Control.Distributed.Process.Serializable (Serializable)
91import Control.Monad.IO.Class (liftIO)
92import Control.Monad.Reader (ask)
93import qualified Control.Monad.State as ST
94 ( MonadState
95 , StateT
96 , get
97 , modify
98 , lift
99 , runStateT
100 )
102data MxAction =
103 MxAgentDeactivate !String
104 | MxAgentReady
106-- | Publishes an arbitrary @Serializable@ message to the management event bus.
107-- Note that /no attempt is made to force the argument/, therefore it is very
108-- important that you do not pass unevaluated thunks that might crash the
109-- receiving process via this API, since /all/ registered agents will gain
110-- access to the data structure once it is broadcast by the agent controller.
111mxNotify :: (Serializable a) => a -> Process ()
112mxNotify msg = do
113 bus <- localEventBus . processNode <$> ask
114 liftIO $ publishEvent bus $ unsafeCreateUnencodedMessage msg
116-- | Publish an arbitrary @Message@ as a property in the management database.
117--
118-- For publishing @Serializable@ data, use 'mxSet' instead. For publishing
119-- properties efficiently within an agent, use 'mxAgentPublish' instead,
120-- since it will be more efficient in that context.
121mxPublish :: MxAgentId -> String -> Message -> Process ()
122mxPublish a k v = Table.set k v (Table.MxForAgent a)
124-- | Sets an arbitrary @Serializable@ datum against a key in the management
125-- database. Note that /no attempt is made to force the argument/, therefore
126-- it is very important that you do not pass unevaluated thunks that might
127-- crash some other, arbitrary process (or management agent!) that obtains
128-- and attempts to force the value later on.
129--
130mxSet :: Serializable a => MxAgentId -> String -> a -> Process ()
131mxSet mxId key msg =
132 Table.set key (unsafeCreateUnencodedMessage msg) (Table.MxForAgent mxId)
134-- | Fetches a property from the management database for the given key.
135-- If the property is not set, or does not match the expected type when
136-- typechecked (at runtime), returns @Nothing@.
137mxGet :: Serializable a => MxAgentId -> String -> Process (Maybe a)
138mxGet mxId = Table.fetch (Table.MxForAgent mxId)
140--------------------------------------------------------------------------------
141-- API for writing user defined management extensions (i.e., agents) --
142--------------------------------------------------------------------------------
144mxAgentPublish = undefined
146mxAgentDeactivate :: forall s. String -> MxAgent s MxAction
147mxAgentDeactivate = return . MxAgentDeactivate
149mxAgentReady :: forall s. MxAgent s MxAction
150mxAgentReady = return MxAgentReady
152liftP :: Process a -> MxAgent s a
153liftP p = MxAgent $ ST.lift p
155mxSetLocal :: s -> MxAgent s ()
156mxSetLocal s = ST.modify $ \st -> st { mxLocalState = s }
158mxGetLocal :: MxAgent s s
159mxGetLocal = ST.get >>= return . mxLocalState
161-- | Activates a new agent.
162--
163mxAgent :: MxAgentId
164 -> s
165 -> (Message -> MxAgent s MxAction)
166 -> Process ProcessId
167mxAgent mxId initState handler = do
168 node <- processNode <$> ask
169 pid <- liftIO $ mxNew (localEventBus node) $ start
170 return pid
171 where
172 start chan = do
173 (sp, rp) <- newChan
174 nsend Table.mxTableCoordinator (MxAgentStart sp mxId)
175 tablePid <- receiveWait [ matchChan rp (\(p :: ProcessId) -> return p) ]
176 runAgent handler chan $ MxAgentState mxId tablePid initState
178 runAgent h c s = do
179 msg <- (liftIO $ atomically $ readTChan c)
180 (action, state) <- runAgentST s $ h msg
181 case action of
182 MxAgentReady -> runAgent h c state
183 MxAgentDeactivate _ -> {- TODO: log r -} return ()
184-- MxAgentBecome h' -> runAgent h' c state
186 runAgentST :: MxAgentState s
187 -> MxAgent s MxAction
188 -> Process (MxAction, MxAgentState s)
189 runAgentST state proc = ST.runStateT (unAgent proc) state