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