git.iko.soy/distributed-process
on dd2ddb5 help sign in
distributed-process/src/Control/Distributed/Process/Management/Agent.hs
127 lines 4.8 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--------------------------------------------------------------------------------
50-- | A triple containing a configured tracer, weak pointer to the
51-- agent controller's mailbox (CQueue) and an expression used to
52-- instantiate new agents on the current node.
53type AgentConfig =
54 (Tracer, Weak (CQueue 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 -- node <- processNode <$> ask
75 trc <- liftIO $ startTracing forkProcess
76 sigbus <- liftIO $ newBroadcastTChanIO
77 weakQueue <- processWeakQ <$> ask
78 liftIO $ putMVar mv (trc, weakQueue, mxStartAgent forkProcess sigbus)
79 go sigbus trc forkProcess
80 where
81 go bus tracer fork = forever' $ do
82 void $ receiveWait [
83 -- This is exactly what it appears to be: a "catch all" handler.
84 -- Since mxNotify can potentially pass an unevaluated thunk to
85 -- our mailbox, the dequeue (i.e., matchMessage) can fail and
86 -- crash this process, which we DO NOT want. Alternatively,
87 -- we handle IO exceptions here explicitly, since we don't want
88 -- this process to ever crash, and the assumption we therefore
89 -- make is thus:
90 --
91 -- 1. only ThreadKilled can tell this process to terminate
92 -- 2. all other exceptions are invalid and should be ignored
93 --
94 -- The outcome of course, is that /bad/ calls to mxNotify
95 -- (e.g., passing unevaluated thunks that will crash when
96 -- they're eventually forced) are thus silently ignored.
97 --
98 matchAny (\msg -> liftIO $ broadcast bus tracer msg)
99 ] `catches` [Handler (\ThreadKilled -> die "Killed"),
100 Handler (\(_ :: SomeException) -> return ())]
102 broadcast :: TChan Message -> Tracer -> Message -> IO ()
103 broadcast ch tr msg = do
104 tmQueue <- tracerQueue tr
105 atomicBroadcast ch tmQueue msg
107 tracerQueue :: Tracer -> IO (Maybe (CQueue Message))
108 tracerQueue (Tracer _ wQ) = deRefWeak wQ
110 atomicBroadcast :: TChan Message
111 -> Maybe (CQueue Message)
112 -> Message -> IO ()
113 atomicBroadcast ch Nothing msg = liftIO $ atomically $ writeTChan ch msg
114 atomicBroadcast ch (Just q) msg = do
115 liftIO $ atomically $ enqueueSTM q msg >> writeTChan ch msg
117-- | Forks a new process in which an mxAgent is run.
118mxStartAgent :: Fork -> TChan Message -> (TChan Message -> Process ()) -> IO ProcessId
119mxStartAgent fork chan handler = (atomically (dupTChan chan)) >>= fork . handler
121startTracing :: Fork -> IO Tracer
122startTracing forkProcess = do
123 mv <- newEmptyMVar
124 pid <- forkProcess $ traceController mv
125 wQ <- liftIO $ takeMVar mv
126 return $ Tracer pid wQ