git.iko.soy/distributed-process
on 28beaa2 help sign in
distributed-process/src/Control/Distributed/Process/Management/Agent.hs
135 lines 5.1 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 , readTChan
12 , writeTChan
13 , dupTChan
14 )
15import Control.Distributed.Process.Internal.Primitives
16 ( receiveWait
17 , matchAny
18 , die
19 , catches
20 , Handler(..)
21 )
22import Control.Distributed.Process.Internal.CQueue
23 ( enqueueSTM
24 , CQueue
25 )
26import Control.Distributed.Process.Internal.Trace.Tracer
27 ( traceController
28 )
29import Control.Distributed.Process.Internal.Types
30 ( Process
31 , SendPort
32 , Message
33 , MxEventBus(..)
34 , Tracer(..)
35 , LocalNode(..)
36 , LocalProcess(..)
37 , ProcessId
38 , forever'
39 )
40import Control.Exception (AsyncException(ThreadKilled), SomeException)
41import qualified Control.Exception as Ex (catch, finally)
42import Control.Monad (void)
43import Control.Monad.IO.Class (liftIO)
44import Control.Monad.Reader (ask)
45import GHC.Weak (Weak(Weak), deRefWeak)
46import System.Environment (getEnv)
48-- | Gross though it is, this synonym represents a function
49-- used to forking new processes, which has to be passed as a HOF
50-- when calling mxAgentController, since there's no other way to
51-- avoid a circular dependency with Node.hs
52type Fork = (Process () -> IO ProcessId)
54-- | A triple containing a configured tracer, weak pointer to the
55-- agent controller's mailbox (CQueue) and an expression used to
56-- instantiate new agents on the current node.
57type AgentConfig =
58 (Tracer, Weak (CQueue Message), ((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 node <- processNode <$> ask
79 trc <- liftIO $ startTracing forkProcess
80 sigbus <- liftIO $ newBroadcastTChanIO
81 weakQueue <- processWeakQ <$> ask
82 liftIO $ putMVar mv (trc, weakQueue, mxAgent forkProcess sigbus)
83 go sigbus trc forkProcess
84 where
85 go bus tracer fork = 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 (\msg -> liftIO $ broadcast bus tracer msg)
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 $ atomically $ enqueueSTM q msg >> writeTChan ch msg
121-- | Forks a new process in which an mxAgent is run.
122mxAgent :: Fork -> TChan Message -> (Message -> Process ()) -> IO ProcessId
123mxAgent fork chan handler = (atomically (dupTChan chan)) >>= fork . run handler
124 where
125 run :: (Message -> Process ()) -> TChan Message -> Process ()
126 run handler' chan' = do
127 (liftIO $ atomically $ readTChan chan') >>= handler' >> run handler' chan'
129startTracing :: Fork -> IO Tracer
130startTracing forkProcess = do
131 mv <- newEmptyMVar
132 pid <- forkProcess $ traceController mv
133 wQ <- liftIO $ takeMVar mv
134 return $ Tracer pid wQ