@@ -1,222 +0,0 @@
1−{-# LANGUAGE DeriveDataTypeable #-}
2−{-# LANGUAGE ScopedTypeVariables #-}
3−{-# LANGUAGE RankNTypes #-}
4−{-# LANGUAGE DeriveGeneric #-}
5−{-# LANGUAGE RecordWildCards #-}
7−module Control.Distributed.Process.Management.Internal.Table
11− , startTableCoordinator
20−import Control.Distributed.Process.Internal.Primitives
37−import Control.Distributed.Process.Internal.Types
40− , ProcessMonitorNotification(..)
44− , unsafeCreateUnencodedMessage
46−import Control.Distributed.Process.Management.Internal.Types
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)
56−import qualified Data.Map as Map
57−import Data.Typeable (Typeable)
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
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
78−data MxTableState = MxTableState { _name :: !String
79− , _entries :: !(Map String Message)
82−type MxTables = Map MxAgentId ProcessId
84−mxTableCoordinator :: String
85−mxTableCoordinator = "mx.table.coordinator"
87−delete :: MxTableId -> Process ()
88−delete = sendReq Delete
90−purge :: MxTableId -> Process ()
93−clear :: String -> MxTableId -> Process ()
94−clear k = sendReq (Clear k)
96−set :: String -> Message -> MxTableId -> Process ()
97−set k v = sendReq (Set k v)
99−fetch :: forall a. (Serializable a)
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))
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.
119−get :: forall a. (Serializable a)
122− -> Process (Maybe a)
124− safeFetch pid key >>= maybe (return Nothing)
125− (unwrapMessage :: Message -> Process (Maybe a))
127−safeFetch :: ProcessId -> String -> Process (Maybe Message)
128−safeFetch pid key = do
130− send pid $ Get key sp
131− withMonitor pid $ do
134− , matchIf (\(ProcessMonitorNotification _ pid' _) -> pid' == pid)
135− (\_ -> return $ Just (unsafeCreateUnencodedMessage ()))
138−sendReq :: MxTableRequest -> MxTableId -> Process ()
139−sendReq req tid = (resolve tid) req
141−resolve :: Serializable a => MxTableId -> (a -> Process ())
142−resolve (MxForAgent agent) = \msg -> nsend mxTableCoordinator (agent, msg)
143−resolve (MxForPid pid) = \msg -> send pid msg
145−startTableCoordinator :: Fork -> Process ()
146−startTableCoordinator fork = run Map.empty
148− run :: MxTables -> Process ()
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
160− lookupAgent tables agent >>= \(p, t) -> do
161− safeFetch p k >>= sendChan sp >> return t
163− handleRequest tables req)
164− , matchAny (\_ -> return tables) -- unrecognised messages are dropped
167− handleRequest :: MxTables
168− -> (MxAgentId, MxTableRequest)
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
181− -> Process (ProcessId, MxTables)
182− launchNew mxId tblMap = do
183− let initState = MxTableState { _name = (agentId mxId)
184− , _entries = Map.empty
186− (pid, _) <- spawnSup $ tableHandler initState
187− return $ (pid, mxId `seq` pid `seq` Map.insert mxId pid tblMap)
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
198−tableHandler :: MxTableState -> Process ()
199−tableHandler state = do
201− match (handleTableRequest state)
202− , matchAny (\_ -> return (Just state))
205− Nothing -> return ()
206− Just s' -> tableHandler s'
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)
215− -> SendPort (Maybe Message)
218−getEntry k m MxTableState{..} = do
219− sendChan m =<< return (Map.lookup k _entries)
221−entries :: Accessor MxTableState (Map String Message)
222−entries = accessor _entries (\ls st -> st { _entries = ls })