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