@@ -1,222 +0,0 @@
1−{-# LANGUAGE DeriveDataTypeable #-}
2−{-# LANGUAGE ScopedTypeVariables #-}
3−{-# LANGUAGE RankNTypes #-}
4−{-# LANGUAGE DeriveGeneric #-}
5−{-# LANGUAGE RecordWildCards #-}
6−
7−module 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
19−
20−import 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− )
37−import Control.Distributed.Process.Internal.Types
38− ( Process
39− , ProcessId
40− , ProcessMonitorNotification(..)
41− , SendPort
42− , ReceivePort
43− , Message
44− , unsafeCreateUnencodedMessage
45− )
46−import Control.Distributed.Process.Management.Internal.Types
47− ( MxTableId(..)
48− , MxAgentId(..)
49− , MxAgentStart(..)
50− , Fork)
51−import Control.Distributed.Process.Serializable (Serializable)
52−import Control.Monad.IO.Class (liftIO)
53−import Data.Accessor (Accessor, accessor, (^=), (^:))
54−import Data.Binary (Binary)
55−import Data.Map (Map)
56−import qualified Data.Map as Map
57−import Data.Typeable (Typeable)
58−
59−import GHC.Generics
60−
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−--
68−
69−data 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)
76−instance Binary MxTableRequest where
77−
78−data MxTableState = MxTableState { _name :: !String
79− , _entries :: !(Map String Message)
80− }
81−
82−type MxTables = Map MxAgentId ProcessId
83−
84−mxTableCoordinator :: String
85−mxTableCoordinator = "mx.table.coordinator"
86−
87−delete :: MxTableId -> Process ()
88−delete = sendReq Delete
89−
90−purge :: MxTableId -> Process ()
91−purge = sendReq Purge
92−
93−clear :: String -> MxTableId -> Process ()
94−clear k = sendReq (Clear k)
95−
96−set :: String -> Message -> MxTableId -> Process ()
97−set k v = sendReq (Set k v)
98−
99−fetch :: forall a. (Serializable a)
100− => MxTableId
101− -> String
102− -> Process (Maybe a)
103−fetch (MxForPid pid) key = get pid key
104−fetch 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))
110−
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.
118−
119−get :: forall a. (Serializable a)
120− => ProcessId
121− -> String
122− -> Process (Maybe a)
123−get pid key = do
124− safeFetch pid key >>= maybe (return Nothing)
125− (unwrapMessage :: Message -> Process (Maybe a))
126−
127−safeFetch :: ProcessId -> String -> Process (Maybe Message)
128−safeFetch 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− ]
137−
138−sendReq :: MxTableRequest -> MxTableId -> Process ()
139−sendReq req tid = (resolve tid) req
140−
141−resolve :: Serializable a => MxTableId -> (a -> Process ())
142−resolve (MxForAgent agent) = \msg -> nsend mxTableCoordinator (agent, msg)
143−resolve (MxForPid pid) = \msg -> send pid msg
144−
145−startTableCoordinator :: Fork -> Process ()
146−startTableCoordinator 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
166−
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
172−
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')
178−
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)
188−
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)
197−
198−tableHandler :: MxTableState -> Process ()
199−tableHandler 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)
213−
214−getEntry :: String
215− -> SendPort (Maybe Message)
216− -> MxTableState
217− -> Process ()
218−getEntry k m MxTableState{..} = do
219− sendChan m =<< return (Map.lookup k _entries)
220−
221−entries :: Accessor MxTableState (Map String Message)
222−entries = accessor _entries (\ls st -> st { _entries = ls })