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