@@ -0,0 +1,219 @@
1+{-# LANGUAGE DeriveGeneric #-}
2+{-# LANGUAGE RecordWildCards #-}
4+module Control.Distributed.Process.Management.Table
8+ , startTableCoordinator
17+import Control.Distributed.Process.Internal.Primitives
34+import Control.Distributed.Process.Internal.Types
37+ , ProcessMonitorNotification(..)
41+ , unsafeCreateUnencodedMessage
43+import Control.Distributed.Process.Management.Types
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)
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
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
75+data MxTableState = MxTableState
78+ , _entries :: !(Map String Message)
81+type MxTables = Map MxAgentId ProcessId
83+mxTableCoordinator :: String
84+mxTableCoordinator = "mx.table.coordinator"
86+delete :: MxTableId -> Process ()
87+delete = sendReq Delete
89+purge :: MxTableId -> Process ()
92+clear :: String -> MxTableId -> Process ()
93+clear k = sendReq (Clear k)
95+set :: String -> Message -> MxTableId -> Process ()
96+set k v = sendReq (Set k v)
98+fetch :: forall a. (Serializable a)
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))
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.
118+get :: forall a. (Serializable a)
121+ -> Process (Maybe a)
123+ safeFetch pid key >>= maybe (return Nothing)
124+ (unwrapMessage :: Message -> Process (Maybe a))
126+safeFetch :: ProcessId -> String -> Process (Maybe Message)
127+safeFetch pid key = do
129+ send pid $ Get key sp
130+ withMonitor pid $ do
133+ , matchIf (\(ProcessMonitorNotification _ pid' _) -> pid' == pid)
134+ (\_ -> return $ Just (unsafeCreateUnencodedMessage ()))
137+sendReq :: MxTableRequest -> MxTableId -> Process ()
138+sendReq req tid = (resolve tid) req
140+resolve :: Serializable a => MxTableId -> (a -> Process ())
141+resolve (MxForAgent agent) = \msg -> nsend mxTableCoordinator (agent, msg)
142+resolve (MxForPid pid) = \msg -> send pid msg
144+startTableCoordinator :: Fork -> Process ()
145+startTableCoordinator fork = run Map.empty
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
154+ run :: MxTables -> Process ()
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
166+ handleRequest :: MxTables
167+ -> (MxAgentId, MxTableRequest)
169+ handleRequest tables' (agent, req) =
170+ lookupAgent tables' agent >>= \(p, t) -> do send p req >> return t
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')
178+ launchNew :: MxAgentId
180+ -> Process (ProcessId, MxTables)
181+ launchNew mxId tblMap = do
182+ let initState = MxTableState { _name = (agentId mxId)
183+ , _entries = Map.empty
185+ (pid, _) <- spawnSup $ tableHandler initState
186+ return $ (pid, Map.insert mxId pid tblMap)
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
198+tableHandler :: MxTableState -> Process ()
199+tableHandler state = do
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))
208+ Nothing -> return ()
209+ Just s' -> tableHandler s'
212+ -> SendPort (Maybe Message)
215+getEntry k m MxTableState{..} = sendChan m =<< return (Map.lookup k _entries)
217+entries :: Accessor MxTableState (Map String Message)
218+entries = accessor _entries (\ls st -> st { _entries = ls })