git.iko.soy/distributed-process
on 6366fa4 help sign in
distributed-process/src/Control/Distributed/Process/Management/Table.hs
220 lines 7.1 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 = MxTableState { _name :: !String
76 , _entries :: !(Map String Message)
77 }
79type MxTables = Map MxAgentId ProcessId
81mxTableCoordinator :: String
82mxTableCoordinator = "mx.table.coordinator"
84delete :: MxTableId -> Process ()
85delete = sendReq Delete
87purge :: MxTableId -> Process ()
88purge = sendReq Purge
90clear :: String -> MxTableId -> Process ()
91clear k = sendReq (Clear k)
93set :: String -> Message -> MxTableId -> Process ()
94set k v = sendReq (Set k v)
96fetch :: forall a. (Serializable a)
97 => MxTableId
98 -> String
99 -> Process (Maybe a)
100fetch (MxForPid pid) key = get pid key
101fetch mxId@(MxForAgent _) key = do
102 (sp, rp) <- newChan :: Process (SendPort (Maybe Message),
103 ReceivePort (Maybe Message))
104 sendReq (Get key sp) mxId
105 receiveChan rp >>= maybe (return Nothing)
106 (unwrapMessage :: Message -> Process (Maybe a))
108-- [note: un-typed send port]
109-- Here, fetch uses a typed channel over a raw Message to obtain
110-- its result, so type checking is deferred until receipt and will
111-- be handled in the caller's thread. This is necessary because
112-- the server portion of the code knows nothing about the types
113-- involved, nor should it, since these tables can be used to
114-- store arbitrary serializable data.
116get :: forall a. (Serializable a)
117 => ProcessId
118 -> String
119 -> Process (Maybe a)
120get pid key = do
121 safeFetch pid key >>= maybe (return Nothing)
122 (unwrapMessage :: Message -> Process (Maybe a))
124safeFetch :: ProcessId -> String -> Process (Maybe Message)
125safeFetch pid key = do
126 (sp, rp) <- newChan
127 send pid $ Get key sp
128 withMonitor pid $ do
129 receiveWait [
130 matchChan rp return
131 , matchIf (\(ProcessMonitorNotification _ pid' _) -> pid' == pid)
132 (\_ -> return $ Just (unsafeCreateUnencodedMessage ()))
133 ]
135sendReq :: MxTableRequest -> MxTableId -> Process ()
136sendReq req tid = (resolve tid) req
138resolve :: Serializable a => MxTableId -> (a -> Process ())
139resolve (MxForAgent agent) = \msg -> nsend mxTableCoordinator (agent, msg)
140resolve (MxForPid pid) = \msg -> send pid msg
142startTableCoordinator :: Fork -> Process ()
143startTableCoordinator fork = run Map.empty
144 where
145 run :: MxTables -> Process ()
146 run tables =
147 receiveWait [
148 -- note that this state change can race with MxAgentStart requests
149 match (\(ProcessMonitorNotification _ pid _) -> do
150 return $ Map.filter (/= pid) tables)
151 , match (\(MxAgentStart ch agent) -> do
152 lookupAgent tables agent >>= \(p, t) -> do
153 sendChan ch p >> return t)
154 , match (\req@(agent, tReq :: MxTableRequest) -> do
155 case tReq of
156 Get k sp -> do
157 lookupAgent tables agent >>= \(p, t) -> do
158 safeFetch p k >>= sendChan sp >> return t
159 _ -> do
160 handleRequest tables req)
161 , matchAny (\_ -> return tables) -- unrecognised messages are dropped
162 ] >>= run
164 handleRequest :: MxTables
165 -> (MxAgentId, MxTableRequest)
166 -> Process MxTables
167 handleRequest tables' (agent, req) = do
168 lookupAgent tables' agent >>= \(p, t) -> send p req >> return t
170 lookupAgent :: MxTables -> MxAgentId -> Process (ProcessId, MxTables)
171 lookupAgent tables' agentId' = do
172 case Map.lookup agentId' tables' of
173 Nothing -> launchNew agentId' tables'
174 Just p -> return (p, tables')
176 launchNew :: MxAgentId
177 -> MxTables
178 -> Process (ProcessId, MxTables)
179 launchNew mxId tblMap = do
180 let initState = MxTableState { _name = (agentId mxId)
181 , _entries = Map.empty
182 }
183 (pid, _) <- spawnSup $ tableHandler initState
184 return $ (pid, Map.insert mxId pid tblMap)
186 spawnSup proc = do
187 us <- getSelfPid
188 -- we need to use that passed in "fork", in order to
189 -- break an import cycle with Node.hs courtesy of the
190 -- management agent, API and tracing modules
191 them <- liftIO $ fork $ link us >> proc
192 ref <- monitor them
193 return (them, ref)
195tableHandler :: MxTableState -> Process ()
196tableHandler state = do
197 ns <- receiveWait [
198 match (handleTableRequest state)
199 , matchAny (\_ -> return (Just state))
200 ]
201 case ns of
202 Nothing -> return ()
203 Just s' -> tableHandler s'
204 where
205 handleTableRequest _ Delete = return Nothing
206 handleTableRequest st Purge = return $ Just $ (entries ^= Map.empty) $ st
207 handleTableRequest st (Clear k) = return $ Just $ (entries ^: Map.delete k) $ st
208 handleTableRequest st (Set k v) = return $ Just $ (entries ^: Map.insert k v) st
209 handleTableRequest st (Get k c) = getEntry k c st >> return (Just st)
211getEntry :: String
212 -> SendPort (Maybe Message)
213 -> MxTableState
214 -> Process ()
215getEntry k m MxTableState{..} = do
216 sendChan m =<< return (Map.lookup k _entries)
218entries :: Accessor MxTableState (Map String Message)
219entries = accessor _entries (\ls st -> st { _entries = ls })