git.iko.soy/distributed-process
on 2ffa2bc help sign in
distributed-process/src/Control/Distributed/Process/Management/Table.hs
219 lines 7.0 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
144 run :: MxTables -> Process ()
145 run tables =
146 receiveWait [
147 -- note that this state change can race with MxAgentStart requests
148 match (\(ProcessMonitorNotification _ pid _) -> do
149 return $ Map.filter (/= pid) tables)
150 , match (\(MxAgentStart ch agent) -> do
151 lookupAgent tables agent >>= \(p, t) -> do
152 sendChan ch p >> return t)
153 , match (\req@(agent, tReq :: MxTableRequest) -> do
154 case tReq of
155 Get k sp -> do
156 lookupAgent tables agent >>= \(p, t) -> do
157 safeFetch p k >>= sendChan sp >> return t
158 _ -> do
159 handleRequest tables req)
160 , matchAny (\_ -> return tables) -- unrecognised messages are dropped
161 ] >>= run
163 handleRequest :: MxTables
164 -> (MxAgentId, MxTableRequest)
165 -> Process MxTables
166 handleRequest tables' (agent, req) = do
167 lookupAgent tables' agent >>= \(p, t) -> send p req >> return t
169 lookupAgent :: MxTables -> MxAgentId -> Process (ProcessId, MxTables)
170 lookupAgent tables' agentId' = do
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 ref <- monitor them
192 return (them, ref)
194tableHandler :: MxTableState -> Process ()
195tableHandler state = do
196 ns <- receiveWait [
197 match (handleTableRequest state)
198 , matchAny (\_ -> return (Just state))
199 ]
200 case ns of
201 Nothing -> return ()
202 Just s' -> tableHandler s'
203 where
204 handleTableRequest _ Delete = return Nothing
205 handleTableRequest st Purge = return $ Just $ (entries ^= Map.empty) $ st
206 handleTableRequest st (Clear k) = return $ Just $ (entries ^: Map.delete k) $ st
207 handleTableRequest st (Set k v) = return $ Just $ (entries ^: Map.insert k v) st
208 handleTableRequest st (Get k c) = getEntry k c st >> return (Just st)
210getEntry :: String
211 -> SendPort (Maybe Message)
212 -> MxTableState
213 -> Process ()
214getEntry k m MxTableState{..} = do
215 sendChan m =<< return (Map.lookup k _entries)
217entries :: Accessor MxTableState (Map String Message)
218entries = accessor _entries (\ls st -> st { _entries = ls })