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