git.iko.soy/distributed-process
on dcaeb3d help sign in
distributed-process/src/Control/Distributed/Process/Management/Table.hs
216 lines 6.7 kB history raw
1{-# LANGUAGE DeriveGeneric #-}
2{-# LANGUAGE RecordWildCards #-}
4module 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
17import 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 )
34import Control.Distributed.Process.Internal.Types
35 ( Process
36 , ProcessId
37 , ProcessMonitorNotification(..)
38 , SendPort
39 , ReceivePort
40 , Message
41 , unsafeCreateUnencodedMessage
42 )
43import Control.Distributed.Process.Management.Types
44 ( MxTableId(..)
45 , MxAgentId(..)
46 , MxAgentStart(..)
47 , Fork)
48import Control.Distributed.Process.Serializable (Serializable)
49import Control.Monad.IO.Class (liftIO)
50import Data.Accessor (Accessor, accessor, (^=), (^:))
51import Data.Binary (Binary)
52import Data.Map.Strict (Map)
53import qualified Data.Map.Strict as Map
54import Data.Typeable (Typeable)
56import GHC.Generics
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--
66data 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)
73instance Binary MxTableRequest where
75data MxTableState =
76 MxTableState { _name :: !String, _entries :: !(Map String Message) }
78type MxTables = Map MxAgentId ProcessId
80mxTableCoordinator :: String
81mxTableCoordinator = "mx.table.coordinator"
83delete :: MxTableId -> Process ()
84delete = sendReq Delete
86purge :: MxTableId -> Process ()
87purge = sendReq Purge
89clear :: String -> MxTableId -> Process ()
90clear k = sendReq (Clear k)
92set :: String -> Message -> MxTableId -> Process ()
93set k v = sendReq (Set k v)
95fetch :: forall a. (Serializable a)
96 => MxTableId
97 -> String
98 -> Process (Maybe a)
99fetch (MxForPid pid) key = get pid key
100fetch mxId@(MxForAgent _) key = do
101 (sp, rp) <- newChan :: Process (SendPort (Maybe Message),
102 ReceivePort (Maybe Message))
103 sendReq (Get key sp) mxId
104 receiveChan rp >>= maybe (return Nothing)
105 (unwrapMessage :: Message -> Process (Maybe a))
107-- [note: un-typed send port]
108-- Here, fetch uses a typed channel over a raw Message to obtain
109-- its result, so type checking is deferred until receipt and will
110-- be handled in the caller's thread. This is necessary because
111-- the server portion of the code knows nothing about the types
112-- involved, nor should it, since these tables can be used to
113-- store arbitrary serializable data.
115get :: forall a. (Serializable a)
116 => ProcessId
117 -> String
118 -> Process (Maybe a)
119get pid key = do
120 safeFetch pid key >>= maybe (return Nothing)
121 (unwrapMessage :: Message -> Process (Maybe a))
123safeFetch :: ProcessId -> String -> Process (Maybe Message)
124safeFetch pid key = do
125 (sp, rp) <- newChan
126 send pid $ Get key sp
127 withMonitor pid $ do
128 receiveWait [
129 matchChan rp return
130 , matchIf (\(ProcessMonitorNotification _ pid' _) -> pid' == pid)
131 (\_ -> return $ Just (unsafeCreateUnencodedMessage ()))
132 ]
134sendReq :: MxTableRequest -> MxTableId -> Process ()
135sendReq req tid = (resolve tid) req
137resolve :: Serializable a => MxTableId -> (a -> Process ())
138resolve (MxForAgent agent) = \msg -> nsend mxTableCoordinator (agent, msg)
139resolve (MxForPid pid) = \msg -> send pid msg
141startTableCoordinator :: Fork -> Process ()
142startTableCoordinator fork = run Map.empty
143 where
145 -- TODO: rethink this design. routing via the
146 -- coordinator will be a publishing bottle neck,
147 -- plus the code in 'fetch' really requires a
148 -- pid to send to, so that'll not work via an
149 -- intermediary anyway
151 run :: MxTables -> Process ()
152 run tables =
153 receiveWait [
154 match (\(MxAgentStart ch agent) ->
155 lookupAgent tables agent >>= \(p, t) -> do
156 sendChan ch p >> return t)
157 , match (\(agent, Get k sp) -> do
158 lookupAgent tables agent >>= \(p, t) -> do
159 safeFetch p k >>= sendChan sp >> return t)
160 , match $ handleRequest tables
161 ] >>= run
163 handleRequest :: MxTables
164 -> (MxAgentId, MxTableRequest)
165 -> Process MxTables
166 handleRequest tables' (agent, req) =
167 lookupAgent tables' agent >>= \(p, t) -> do send p req >> return t
169 lookupAgent :: MxTables -> MxAgentId -> Process (ProcessId, MxTables)
170 lookupAgent tables' agentId' =
171 case Map.lookup agentId' tables' of
172 Nothing -> launchNew agentId' tables'
173 Just p -> return (p, tables')
175 launchNew :: MxAgentId
176 -> MxTables
177 -> Process (ProcessId, MxTables)
178 launchNew mxId tblMap = do
179 let initState = MxTableState { _name = (agentId mxId)
180 , _entries = Map.empty
181 }
182 (pid, _) <- spawnSup $ tableHandler initState
183 return $ (pid, Map.insert mxId pid tblMap)
185 spawnSup proc = do
186 us <- getSelfPid
187 -- we need to use that passed in "fork", in order to
188 -- break an import cycle with Node.hs via the management
189 -- agent, management API and the tracing modules
190 them <- liftIO $ fork $ link us >> proc
191 -- them <- spawnLocal nid $ link us >> proc
192 ref <- monitor them
193 return (them, ref)
195tableHandler :: MxTableState -> Process ()
196tableHandler state = do
197 ns <- receiveWait [
198 match (\Delete -> return Nothing)
199 , match (\Purge -> return $ Just $ (entries ^= Map.empty) $ state)
200 , match (\(Clear k) -> return $ Just $ (entries ^: Map.delete k) $ state)
201 , match (\(Set k v) -> return $ Just $ (entries ^: Map.insert k v) state)
202 , match (\(Get k c) -> getEntry k c state >> return (Just state))
203 ]
204 case ns of
205 Nothing -> return ()
206 Just s' -> tableHandler s'
208getEntry :: String
209 -> SendPort (Maybe Message)
210 -> MxTableState
211 -> Process ()
212getEntry k m MxTableState{..} = sendChan m =<< return (Map.lookup k _entries)
214entries :: Accessor MxTableState (Map String Message)
215entries = accessor _entries (\ls st -> st { _entries = ls })