@@ -0,0 +1,219 @@
1+{-# LANGUAGE DeriveGeneric #-}
2+{-# LANGUAGE RecordWildCards #-}
3+
4+module Control.Distributed.Process.Management.Table
5+ ( MxTableRequest(..)
6+ , MxTableId(..)
7+ , mxTableCoordinator
8+ , startTableCoordinator
9+ , delete
10+ , purge
11+ , clear
12+ , set
13+ , get
14+ , fetch
15+ ) where
16+
17+import Control.Distributed.Process.Internal.Primitives
18+ ( receiveWait
19+ , receiveChan
20+ , match
21+ , matchAny
22+ , matchIf
23+ , matchChan
24+ , send
25+ , nsend
26+ , sendChan
27+ , getSelfPid
28+ , link
29+ , monitor
30+ , unwrapMessage
31+ , newChan
32+ , withMonitor
33+ )
34+import Control.Distributed.Process.Internal.Types
35+ ( Process
36+ , ProcessId
37+ , ProcessMonitorNotification(..)
38+ , SendPort
39+ , ReceivePort
40+ , Message
41+ , unsafeCreateUnencodedMessage
42+ )
43+import Control.Distributed.Process.Management.Types
44+ ( MxTableId(..)
45+ , MxAgentId(..)
46+ , MxAgentStart(..)
47+ , Fork)
48+import Control.Distributed.Process.Serializable (Serializable)
49+import Control.Monad.IO.Class (liftIO)
50+import Data.Accessor (Accessor, accessor, (^=), (^:))
51+import Data.Binary (Binary)
52+import Data.Map.Strict (Map)
53+import qualified Data.Map.Strict as Map
54+import Data.Typeable (Typeable)
55+
56+import GHC.Generics
57+
58+-- An extremely lightweight shared Map implementation, for use
59+-- by /management agents/ and their cohorts. Each agent is assigned
60+-- a table, into which any serializable @Message@ can be inserted.
61+-- Data are inserted, removed and searched for via their key, which
62+-- is a string. Tables can be purged, values can be set, fetched or
63+-- cleared/removed.
64+--
65+
66+data MxTableRequest =
67+ Delete
68+ | Purge
69+ | Clear !String
70+ | Set !String !Message
71+ | Get !String !(SendPort (Maybe Message)) -- see [note: un-typed send port]
72+ deriving (Typeable, Generic)
73+instance Binary MxTableRequest where
74+
75+data MxTableState = MxTableState
76+ {
77+ _name :: !String
78+ , _entries :: !(Map String Message)
79+ }
80+
81+type MxTables = Map MxAgentId ProcessId
82+
83+mxTableCoordinator :: String
84+mxTableCoordinator = "mx.table.coordinator"
85+
86+delete :: MxTableId -> Process ()
87+delete = sendReq Delete
88+
89+purge :: MxTableId -> Process ()
90+purge = sendReq Purge
91+
92+clear :: String -> MxTableId -> Process ()
93+clear k = sendReq (Clear k)
94+
95+set :: String -> Message -> MxTableId -> Process ()
96+set k v = sendReq (Set k v)
97+
98+fetch :: forall a. (Serializable a)
99+ => MxTableId
100+ -> String
101+ -> Process (Maybe a)
102+fetch (MxForPid pid) key = get pid key
103+fetch mxId@(MxForAgent _) key = do
104+ (sp, rp) <- newChan :: Process (SendPort (Maybe Message),
105+ ReceivePort (Maybe Message))
106+ sendReq (Get key sp) mxId
107+ receiveChan rp >>= maybe (return Nothing)
108+ (unwrapMessage :: Message -> Process (Maybe a))
109+
110+-- [note: un-typed send port]
111+-- Here, fetch uses a typed channel over a raw Message to obtain
112+-- its result, so type checking is deferred until receipt and will
113+-- be handled in the caller's thread. This is necessary because
114+-- the server portion of the code knows nothing about the types
115+-- involved, nor should it, since these tables can be used to
116+-- store arbitrary serializable data.
117+
118+get :: forall a. (Serializable a)
119+ => ProcessId
120+ -> String
121+ -> Process (Maybe a)
122+get pid key = do
123+ safeFetch pid key >>= maybe (return Nothing)
124+ (unwrapMessage :: Message -> Process (Maybe a))
125+
126+safeFetch :: ProcessId -> String -> Process (Maybe Message)
127+safeFetch pid key = do
128+ (sp, rp) <- newChan
129+ send pid $ Get key sp
130+ withMonitor pid $ do
131+ receiveWait [
132+ matchChan rp return
133+ , matchIf (\(ProcessMonitorNotification _ pid' _) -> pid' == pid)
134+ (\_ -> return $ Just (unsafeCreateUnencodedMessage ()))
135+ ]
136+
137+sendReq :: MxTableRequest -> MxTableId -> Process ()
138+sendReq req tid = (resolve tid) req
139+
140+resolve :: Serializable a => MxTableId -> (a -> Process ())
141+resolve (MxForAgent agent) = \msg -> nsend mxTableCoordinator (agent, msg)
142+resolve (MxForPid pid) = \msg -> send pid msg
143+
144+startTableCoordinator :: Fork -> Process ()
145+startTableCoordinator fork = run Map.empty
146+ where
147+
148+ -- TODO: rethink this design. routing via the
149+ -- coordinator will be a publishing bottle neck,
150+ -- plus the code in 'fetch' really requires a
151+ -- pid to send to, so that'll not work via an
152+ -- intermediary anyway
153+
154+ run :: MxTables -> Process ()
155+ run tables =
156+ receiveWait [
157+ match (\(MxAgentStart ch agent) ->
158+ lookupAgent tables agent >>= \(p, t) -> do
159+ sendChan ch p >> return t)
160+ , match (\(agent, Get k sp) -> do
161+ lookupAgent tables agent >>= \(p, t) -> do
162+ safeFetch p k >>= sendChan sp >> return t)
163+ , match $ handleRequest tables
164+ ] >>= run
165+
166+ handleRequest :: MxTables
167+ -> (MxAgentId, MxTableRequest)
168+ -> Process MxTables
169+ handleRequest tables' (agent, req) =
170+ lookupAgent tables' agent >>= \(p, t) -> do send p req >> return t
171+
172+ lookupAgent :: MxTables -> MxAgentId -> Process (ProcessId, MxTables)
173+ lookupAgent tables' agentId' =
174+ case Map.lookup agentId' tables' of
175+ Nothing -> launchNew agentId' tables'
176+ Just p -> return (p, tables')
177+
178+ launchNew :: MxAgentId
179+ -> MxTables
180+ -> Process (ProcessId, MxTables)
181+ launchNew mxId tblMap = do
182+ let initState = MxTableState { _name = (agentId mxId)
183+ , _entries = Map.empty
184+ }
185+ (pid, _) <- spawnSup $ tableHandler initState
186+ return $ (pid, Map.insert mxId pid tblMap)
187+
188+ spawnSup proc = do
189+ us <- getSelfPid
190+ -- we need to used that passed in "fork", in order to
191+ -- break an import cycle with Node.hs via the management
192+ -- agent, management API and the tracing modules
193+ them <- liftIO $ fork $ link us >> proc
194+ -- them <- spawnLocal nid $ link us >> proc
195+ ref <- monitor them
196+ return (them, ref)
197+
198+tableHandler :: MxTableState -> Process ()
199+tableHandler state = do
200+ ns <- receiveWait [
201+ match (\Delete -> return Nothing)
202+ , match (\Purge -> return $ Just $ (entries ^= Map.empty) $ state)
203+ , match (\(Clear k) -> return $ Just $ (entries ^: Map.delete k) $ state)
204+ , match (\(Set k v) -> return $ Just $ (entries ^: Map.insert k v) state)
205+ , match (\(Get k c) -> getEntry k c state >> return (Just state))
206+ ]
207+ case ns of
208+ Nothing -> return ()
209+ Just s' -> tableHandler s'
210+
211+getEntry :: String
212+ -> SendPort (Maybe Message)
213+ -> MxTableState
214+ -> Process ()
215+getEntry k m MxTableState{..} = sendChan m =<< return (Map.lookup k _entries)
216+
217+entries :: Accessor MxTableState (Map String Message)
218+entries = accessor _entries (\ls st -> st { _entries = ls })
219+