git.iko.soy/distributed-process
on 9e7b360 help sign in
distributed-process/src/Control/Distributed/Process/Management/Agent.hs
133 lines 4.9 kB history raw
1{-# LANGUAGE RecordWildCards #-}
3module Control.Distributed.Process.Management.Agent where
5import Control.Applicative ((<$>))
6import Control.Concurrent.MVar (MVar, newEmptyMVar, putMVar, takeMVar)
7import Control.Concurrent.STM (atomically)
8import Control.Concurrent.STM.TChan
9 ( TChan
10 , newBroadcastTChanIO
11 , writeTChan
12 , dupTChan
13 )
14import Control.Distributed.Process.Internal.Primitives
15 ( receiveWait
16 , matchAny
17 , die
18 , catches
19 , Handler(..)
20 )
21import Control.Distributed.Process.Internal.CQueue
22 ( enqueueSTM
23 , CQueue
24 )
25import Control.Distributed.Process.Management.Types
26 ( Fork
27 )
28import Control.Distributed.Process.Management.Trace.Tracer
29 ( traceController
30 )
31import Control.Distributed.Process.Internal.Types
32 ( Process
33 , Message
34 , Tracer(..)
35 , LocalProcess(..)
36 , ProcessId
37 , forever'
38 )
39import Control.Exception (AsyncException(ThreadKilled), SomeException)
40import Control.Monad (void)
41import Control.Monad.IO.Class (liftIO)
42import Control.Monad.Reader (ask)
43import GHC.Weak (Weak, deRefWeak)
45--------------------------------------------------------------------------------
46-- Agent Controller Implementation --
47--------------------------------------------------------------------------------
49-- | A triple containing a configured tracer, weak pointer to the
50-- agent controller's mailbox (CQueue) and an expression used to
51-- instantiate new agents on the current node.
52type AgentConfig =
53 (Tracer, Weak (CQueue Message),
54 (((TChan Message, TChan Message) -> Process ()) -> IO ProcessId))
56-- | Starts a management agent for the current node. The agent process
57-- must not crash or be killed, so we generally avoid publishing its
58-- @ProcessId@ where possible.
59--
60-- Our process is also responsible for forwarding messages to the trace
61-- controller, since having two /special processes/ handled via the
62-- @LocalNode@ would be inelegant. We forward messages directly to the
63-- trace controller's message queue, just as the @MxEventBus@ that's
64-- set up on the @LocalNode@ forwards messages directly to us. This
65-- optimises the code path for tracing and avoids overloading the node
66-- node controller's internal control plane with additional routing, at the
67-- cost of a little more complexity and two cases where we break
68-- encapsulation.
69--
70mxAgentController :: Fork
71 -> MVar AgentConfig
72 -> Process ()
73mxAgentController forkProcess mv = do
74 trc <- liftIO $ startTracing forkProcess
75 sigbus <- liftIO $ newBroadcastTChanIO
76 weakQueue <- processWeakQ <$> ask
77 liftIO $ putMVar mv (trc, weakQueue, mxStartAgent forkProcess sigbus)
78 go sigbus trc
79 where
80 go bus tracer = forever' $ do
81 void $ receiveWait [
82 -- This is exactly what it appears to be: a "catch all" handler.
83 -- Since mxNotify can potentially pass an unevaluated thunk to
84 -- our mailbox, the dequeue (i.e., matchMessage) can fail and
85 -- crash this process, which we DO NOT want. Alternatively,
86 -- we handle IO exceptions here explicitly, since we don't want
87 -- this process to ever crash, and the assumption we therefore
88 -- make is thus:
89 --
90 -- 1. only ThreadKilled can tell this process to terminate
91 -- 2. all other exceptions are invalid and should be ignored
92 --
93 -- The outcome of course, is that /bad/ calls to mxNotify
94 -- (e.g., passing unevaluated thunks that will crash when
95 -- they're eventually forced) are thus silently ignored.
96 --
97 matchAny (\msg -> liftIO $ broadcast bus tracer msg)
98 ] `catches` [Handler (\ThreadKilled -> die "Killed"),
99 Handler (\(_ :: SomeException) -> return ())]
101 broadcast :: TChan Message -> Tracer -> Message -> IO ()
102 broadcast ch tr msg = do
103 tmQueue <- tracerQueue tr
104 atomicBroadcast ch tmQueue msg
106 tracerQueue :: Tracer -> IO (Maybe (CQueue Message))
107 tracerQueue (Tracer _ wQ) = deRefWeak wQ
109 atomicBroadcast :: TChan Message
110 -> Maybe (CQueue Message)
111 -> Message -> IO ()
112 atomicBroadcast ch Nothing msg = liftIO $ atomically $ writeTChan ch msg
113 atomicBroadcast ch (Just q) msg = do
114 -- liftIO $ putStrLn $ "broadcasting " ++ (show msg)
115 liftIO $ atomically $ enqueueSTM q msg >> writeTChan ch msg
117-- | Forks a new process in which an mxAgent is run.
118mxStartAgent :: Fork
119 -> TChan Message
120 -> ((TChan Message, TChan Message) -> Process ())
121 -> IO ProcessId
122mxStartAgent fork chan handler = do
123 chan' <- atomically (dupTChan chan)
124 let proc = handler (chan, chan')
125 fork proc
127startTracing :: Fork -> IO Tracer
128startTracing forkProcess = do
129 mv <- newEmptyMVar
130 pid <- forkProcess $ traceController mv
131 wQ <- liftIO $ takeMVar mv
132 return $ Tracer pid wQ