@@ -106,7 +106,7 @@ import Control.Distributed.Process.Internal.Types
106106 , LocalProcessId(..)
107107 , ProcessId(..)
108108 , LocalNode(..)
109− , Tracer(..)
109+ , MxEventBus(..)
110110 , LocalNodeState(..)
111111 , LocalProcess(..)
112112 , LocalProcessState(..)
@@ -160,8 +160,7 @@ import Control.Distributed.Process.Internal.Trace.Types
160160 , enableTrace
161161 )
162162import Control.Distributed.Process.Internal.Trace.Tracer
163− ( traceController
164− , defaultTracer
163+ ( defaultTracer
165164 )
166165import Control.Distributed.Process.Serializable (Serializable)
167166import Control.Distributed.Process.Internal.Messaging
@@ -179,13 +178,16 @@ import Control.Distributed.Process.Internal.Primitives
179178 , catch
180179 , unwrapMessage
181180 )
182−import Control.Distributed.Process.Internal.Types (SendPort)
181+import Control.Distributed.Process.Internal.Types (SendPort, Tracer(..))
183182import qualified Control.Distributed.Process.Internal.Closure.BuiltIn as BuiltIn (remoteTable)
184183import Control.Distributed.Process.Internal.WeakTQueue (TQueue, writeTQueue)
185184import qualified Control.Distributed.Process.Internal.StrictContainerAccessors as DAC
186185 ( mapMaybe
187186 , mapDefault
188187 )
188+import Control.Distributed.Process.Management.Agent
189+ ( mxAgentController
190+ )
189191import System.Environment (getEnv)
190192import Unsafe.Coerce
191193
@@ -222,10 +224,10 @@ createBareLocalNode endPoint rtable = do
222224 , localEndPoint = endPoint
223225 , localState = state
224226 , localCtrlChan = ctrlChan
225− , localTracer = InactiveTracer
227+ , localEventBus = MxEventBusInitialising
226228 , remoteTable = rtable
227229 }
228− tracedNode <- startTracing node
230+ tracedNode <- startMxAgent node
229231
230232 -- Once the NC terminates, the endpoint isn't much use,
231233 void $ forkIO $ Exception.finally (runNodeController tracedNode)
@@ -243,38 +245,25 @@ createBareLocalNode endPoint rtable = do
243245 , ctrlMsgSignal = SigShutdown
244246 }
245247
246−startTracing :: LocalNode -> IO LocalNode
247−startTracing node =
248− -- TODO: use tryGetEnv once we drop support for GHC 7.2.x
249− Exception.catch
250− (getEnv "DISTRIBUTED_PROCESS_TRACE_ENABLED" >> procTracer node)
251− (\(_ :: IOError) -> return node)
252− where procTracer :: LocalNode -> IO LocalNode
253− procTracer n = do
254− -- see note [tracer/forkProcess races]
255− mv <- MVar.newEmptyMVar
256− pid <- forkProcess n (traceController mv)
257− wqRef <- MVar.takeMVar mv
258− return n { localTracer = (ActiveTracer pid wqRef) }
259−
260−startDefaultTracer :: LocalNode -> IO (Maybe ProcessId)
261−startDefaultTracer node =
262− Exception.catch
263− (getEnv "DISTRIBUTED_PROCESS_TRACE_ENABLED" >> go node)
264− (\(_ :: IOError) -> return Nothing)
265− where go :: LocalNode -> IO (Maybe ProcessId)
266− go node' = do
267− registerTraceController node'
268− pid <- forkProcess node' defaultTracer
269− enableTrace (localTracer node') pid
270− return (Just pid)
271−
272− registerTraceController :: LocalNode -> IO ()
273− registerTraceController n =
274− let t = localTracer n in
275− case t of
276− InactiveTracer -> return ()
277− (ActiveTracer pid _) -> runProcess n $ register "trace.controller" pid
248+startMxAgent :: LocalNode -> IO LocalNode
249+startMxAgent node = do
250+ -- see note [tracer/forkProcess races]
251+ let fork = forkProcess node
252+ mv <- MVar.newEmptyMVar
253+ pid <- fork $ mxAgentController fork mv
254+ (tracer, wqRef, mxNew) <- MVar.takeMVar mv
255+ return node { localEventBus = (MxEventBus pid tracer wqRef mxNew) }
256+
257+startDefaultTracer :: LocalNode -> IO ()
258+startDefaultTracer node' = do
259+ let t = localEventBus node'
260+ case t of
261+ MxEventBus _ (ActiveTracer pid _) _ _ -> do
262+ runProcess node' $ register "trace.controller" pid
263+ pid' <- forkProcess node' defaultTracer
264+ enableTrace (localEventBus node') pid'
265+ runProcess node' $ register "tracer.initial" pid'
266+ _ -> return ()
278267
279268-- | Start and register the service processes on a node
280269startServiceProcesses :: LocalNode -> IO ()
@@ -283,11 +272,7 @@ startServiceProcesses node = do
283272 -- the default tracer first, even though it might @nsend@ to the logger
284273 -- before /that/ process has started - this is a totally harmless race
285274 -- however, so we deliberably ignore it
286− mPid <- startDefaultTracer node
287− case mPid of
288− Nothing -> return ()
289− Just pid -> runProcess node $ register "tracer.initial" pid
290−
275+ startDefaultTracer node
291276 logger <- forkProcess node loop
292277 runProcess node $ register "logger" logger
293278 where
@@ -401,7 +386,7 @@ forkProcess node proc = modifyMVar (localState node) startProcess
401386--
402387-- Our startTracing function uses forkProcess to start the trace controller
403388-- process, and of course forkProcess attempts to call traceEvent once the
404−-- process has started. This is harmless, as the localTracer is not updated
389+-- process has started. This is harmless, as the localEventBus is not updated
405390-- until /after/ the initial forkProcess completes, so the first call to
406391-- traceEvent behaves as if tracing were disabled (i.e., it is ignored).
407392--
@@ -607,7 +592,7 @@ instance Show ProcessKillException where
607592-- Tracing/Debugging --
608593--------------------------------------------------------------------------------
609594
610−-- [Issue #104]
595+-- [Issue #104 / DP-13]
611596
612597traceNotifyDied :: LocalNode -> Identifier -> DiedReason -> NC ()
613598traceNotifyDied node ident reason =
@@ -628,8 +613,8 @@ traceEventFmtIO node fmt args =
628613trace :: LocalNode -> TraceEvent -> IO ()
629614trace node ev = withLocalTracer node $ \t -> traceEvent t ev
630615
631−withLocalTracer :: LocalNode -> (Tracer -> IO ()) -> IO ()
632−withLocalTracer node act = act (localTracer node)
616+withLocalTracer :: LocalNode -> (MxEventBus -> IO ()) -> IO ()
617+withLocalTracer node act = act (localEventBus node)
633618
634619--------------------------------------------------------------------------------
635620-- Core functionality --