git.iko.soy/distributed-process
on master help sign in
Merge branch 'tracing' into development. Provides a new tracing API.
13377cfTim Watson
distributed-process.cabal+36−7
compare with master
unifiedsplit
@@ -59,19 +59,24 @@ Library
59 rank1dynamic >= 0.1 && < 0.2,
59 rank1dynamic >= 0.1 && < 0.2,
60 syb >= 0.3 && < 0.4
60 syb >= 0.3 && < 0.4
61 Exposed-modules: Control.Distributed.Process,
61 Exposed-modules: Control.Distributed.Process,
62+ Control.Distributed.Process.Internal.Types,
62 Control.Distributed.Process.Serializable,
63 Control.Distributed.Process.Serializable,
63 Control.Distributed.Process.Closure,
64 Control.Distributed.Process.Closure,
64− Control.Distributed.Process.Node,
65+ Control.Distributed.Process.Debug,
66+ Control.Distributed.Process.Node
65− Control.Distributed.Process.Internal.Primitives,
67+ Other-modules: Control.Distributed.Process.Internal.Primitives,
68+ Control.Distributed.Process.Internal.Spawn,
66 Control.Distributed.Process.Internal.CQueue,
69 Control.Distributed.Process.Internal.CQueue,
67− Control.Distributed.Process.Internal.Types,
68− Control.Distributed.Process.Internal.Trace,
69 Control.Distributed.Process.Internal.Closure.BuiltIn,
70 Control.Distributed.Process.Internal.Closure.BuiltIn,
70 Control.Distributed.Process.Internal.Messaging,
71 Control.Distributed.Process.Internal.Messaging,
71 Control.Distributed.Process.Internal.StrictList,
72 Control.Distributed.Process.Internal.StrictList,
72 Control.Distributed.Process.Internal.StrictMVar,
73 Control.Distributed.Process.Internal.StrictMVar,
73− Control.Distributed.Process.Internal.WeakTQueue
74+ Control.Distributed.Process.Internal.WeakTQueue,
74− Control.Distributed.Process.Internal.StrictContainerAccessors
75+ Control.Distributed.Process.Internal.StrictContainerAccessors,
76+ Control.Distributed.Process.Internal.Trace.Types,
77+ Control.Distributed.Process.Internal.Trace.Primitives,
78+ Control.Distributed.Process.Internal.Trace.Tracer
79+ Control.Distributed.Process.Internal.Trace.Remote
75 Extensions: RankNTypes,
80 Extensions: RankNTypes,
76 ScopedTypeVariables,
81 ScopedTypeVariables,
77 FlexibleInstances,
82 FlexibleInstances,
@@ -108,7 +113,7 @@ Test-Suite TestCH
108 ScopedTypeVariables,
113 ScopedTypeVariables,
109 DeriveDataTypeable,
114 DeriveDataTypeable,
110 GeneralizedNewtypeDeriving
115 GeneralizedNewtypeDeriving
111− ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind
116+ ghc-options: -Wall -threaded -debug -eventlog -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind
112 HS-Source-Dirs: tests
117 HS-Source-Dirs: tests
113
118
114Test-Suite TestClosure
119Test-Suite TestClosure
@@ -154,6 +159,30 @@ Test-Suite TestStats
154 ghc-options: -Wall -debug -eventlog -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind
159 ghc-options: -Wall -debug -eventlog -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind
155 HS-Source-Dirs: tests
160 HS-Source-Dirs: tests
156
161
162+Test-Suite TestTracing
163+ Type: exitcode-stdio-1.0
164+ Main-Is: TestTracing.hs
165+ Build-Depends: base >= 4.4 && < 5,
166+ random >= 1.0 && < 1.1,
167+ ansi-terminal >= 0.5 && < 0.6,
168+ containers >= 0.4 && < 0.6,
169+ stm >= 2.4 && < 2.5,
170+ distributed-process,
171+ network-transport >= 0.3 && < 0.4,
172+ network-transport-tcp >= 0.3 && < 0.4,
173+ binary >= 0.5 && < 0.7,
174+ network >= 2.3 && < 2.5,
175+ HUnit >= 1.2 && < 1.3,
176+ test-framework >= 0.6 && < 0.9,
177+ test-framework-hunit >= 0.2.0 && < 0.4,
178+ unix >= 2.5.0.0
179+ Extensions: CPP,
180+ ScopedTypeVariables,
181+ DeriveDataTypeable,
182+ GeneralizedNewtypeDeriving
183+ ghc-options: -Wall -debug -eventlog -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind
184+ HS-Source-Dirs: tests
185+
157
186
158Executable distributed-process-throughput
187Executable distributed-process-throughput
159 if flag(benchmarks)
188 if flag(benchmarks)
src/Control/Distributed/Process.hs+24−129
@@ -2,7 +2,8 @@
2
2
3This is an implementation of Cloud Haskell, as described in
3This is an implementation of Cloud Haskell, as described in
4/Towards Haskell in the Cloud/ by Jeff Epstein, Andrew Black, and Simon
4/Towards Haskell in the Cloud/ by Jeff Epstein, Andrew Black, and Simon
5+Peyton Jones (see
5−Peyton Jones (<http://research.microsoft.com/en-us/um/people/simonpj/papers/parallel/>),
6+<http://research.microsoft.com/en-us/um/people/simonpj/papers/parallel/>),
6although some of the details are different. The precise message passing
7although some of the details are different. The precise message passing
7semantics are based on /A unified semantics for future Erlang/ by Hans
8semantics are based on /A unified semantics for future Erlang/ by Hans
8Svensson, Lars-Åke Fredlund and Clara Benac Earle.
9Svensson, Lars-Åke Fredlund and Clara Benac Earle.
@@ -52,6 +53,9 @@ module Control.Distributed.Process
52 , unwrapMessage
53 , unwrapMessage
53 , handleMessage
54 , handleMessage
54 , forward
55 , forward
56+ , delegate
57+ , relay
58+ , proxy
55 -- * Process management
59 -- * Process management
56 , spawn
60 , spawn
57 , call
61 , call
@@ -139,7 +143,6 @@ module Control.Distributed.Process
139import Prelude hiding (catch)
143import Prelude hiding (catch)
140#endif
144#endif
141
145
142−import Data.Typeable (Typeable)
143import Control.Monad.IO.Class (liftIO)
146import Control.Monad.IO.Class (liftIO)
144import Control.Applicative ((<$>))
147import Control.Applicative ((<$>))
145import Control.Monad.Reader (ask)
148import Control.Monad.Reader (ask)
@@ -149,8 +152,6 @@ import Control.Distributed.Static
149 , closure
152 , closure
150 , Static
153 , Static
151 , RemoteTable
154 , RemoteTable
152− , closureCompose
153− , staticClosure
154 )
155 )
155import Control.Distributed.Process.Internal.Types
156import Control.Distributed.Process.Internal.Types
156 ( NodeId(..)
157 ( NodeId(..)
@@ -174,21 +175,8 @@ import Control.Distributed.Process.Internal.Types
174 , RegisterReply(..)
175 , RegisterReply(..)
175 , LocalProcess(processNode)
176 , LocalProcess(processNode)
176 , Message
177 , Message
177− , nullProcessId
178− )
179−import Control.Distributed.Process.Serializable (Serializable, SerializableDict)
180−import Control.Distributed.Process.Internal.Closure.BuiltIn
181− ( sdictSendPort
182− , sndStatic
183− , idCP
184− , seqCP
185− , bindCP
186− , splitCP
187− , cpLink
188− , cpSend
189− , cpNewChan
190− , cpDelay
191 )
178 )
179+import Control.Distributed.Process.Serializable (Serializable)
192import Control.Distributed.Process.Internal.Primitives
180import Control.Distributed.Process.Internal.Primitives
193 ( -- Basic messaging
181 ( -- Basic messaging
194 send
182 send
@@ -215,6 +203,9 @@ import Control.Distributed.Process.Internal.Primitives
215 , unwrapMessage
203 , unwrapMessage
216 , handleMessage
204 , handleMessage
217 , forward
205 , forward
206+ , delegate
207+ , relay
208+ , proxy
218 -- Process management
209 -- Process management
219 , terminate
210 , terminate
220 , ProcessTerminationException(..)
211 , ProcessTerminationException(..)
@@ -274,6 +265,15 @@ import Control.Distributed.Process.Internal.Primitives
274 , reconnectPort
265 , reconnectPort
275 )
266 )
276import Control.Distributed.Process.Node (forkProcess)
267import Control.Distributed.Process.Node (forkProcess)
268+import Control.Distributed.Process.Internal.Spawn
269+ ( -- Spawning Processes/Channels
270+ spawn
271+ , spawnLink
272+ , spawnMonitor
273+ , spawnChannel
274+ , spawnSupervised
275+ , call
276+ )
277
277
278-- INTERNAL NOTES
278-- INTERNAL NOTES
279--
279--
@@ -285,7 +285,11 @@ import Control.Distributed.Process.Node (forkProcess)
285-- 2. 'send' may block (when the system TCP buffers are full, while we are
285-- 2. 'send' may block (when the system TCP buffers are full, while we are
286-- trying to establish a connection to the remote endpoint, etc.) but its
286-- trying to establish a connection to the remote endpoint, etc.) but its
287-- return does not imply that the remote process received the message (much
287-- return does not imply that the remote process received the message (much
288−-- less processed it)
288+-- less processed it). When 'send' targets a local process, it is dispatched
289+-- via the node controller however, in which cases it will /not/ block. This
290+-- isn't part of the semantics, so it should be ok. The ordering is maintained
291+-- because the ctrlChannel still has FIFO semantics with regards interactions
292+-- between two disparate forkIO threads.
289--
293--
290-- 3. Message delivery is reliable and ordered. That means that if process A
294-- 3. Message delivery is reliable and ordered. That means that if process A
291-- sends messages m1, m2, m3 to process B, B will either arrive all three
295-- sends messages m1, m2, m3 to process B, B will either arrive all three
@@ -334,116 +338,6 @@ import Control.Distributed.Process.Node (forkProcess)
334-- [6] "Delivery of Messages", post on erlang-questions
338-- [6] "Delivery of Messages", post on erlang-questions
335-- http://erlang.org/pipermail/erlang-questions/2012-February/064767.html
339-- http://erlang.org/pipermail/erlang-questions/2012-February/064767.html
336
340
337−--------------------------------------------------------------------------------
338−-- Primitives are defined in a separate module; here we only define derived --
339−-- constructs --
340−--------------------------------------------------------------------------------
341−
342−-- | Spawn a process
343−--
344−-- For more information about 'Closure', see
345−-- "Control.Distributed.Process.Closure".
346−--
347−-- See also 'call'.
348−spawn :: NodeId -> Closure (Process ()) -> Process ProcessId
349−spawn nid proc = do
350− us <- getSelfPid
351− mRef <- monitorNode nid
352− sRef <- spawnAsync nid (cpDelay us proc)
353− receiveWait [
354− matchIf (\(DidSpawn ref _) -> ref == sRef) $ \(DidSpawn _ pid) -> do
355− unmonitor mRef
356− send pid ()
357− return pid
358− , matchIf (\(NodeMonitorNotification ref _ _) -> ref == mRef) $ \_ ->
359− return (nullProcessId nid)
360− ]
361−
362−-- | Spawn a process and link to it
363−--
364−-- Note that this is just the sequential composition of 'spawn' and 'link'.
365−-- (The "Unified" semantics that underlies Cloud Haskell does not even support
366−-- a synchronous link operation)
367−spawnLink :: NodeId -> Closure (Process ()) -> Process ProcessId
368−spawnLink nid proc = do
369− pid <- spawn nid proc
370− link pid
371− return pid
372−
373−-- | Like 'spawnLink', but monitor the spawned process
374−spawnMonitor :: NodeId -> Closure (Process ()) -> Process (ProcessId, MonitorRef)
375−spawnMonitor nid proc = do
376− pid <- spawn nid proc
377− ref <- monitor pid
378− return (pid, ref)
379−
380−-- | Run a process remotely and wait for it to reply
381−--
382−-- We monitor the remote process: if it dies before it can send a reply, we die
383−-- too.
384−--
385−-- For more information about 'Static', 'SerializableDict', and 'Closure', see
386−-- "Control.Distributed.Process.Closure".
387−--
388−-- See also 'spawn'.
389−call :: Serializable a
390− => Static (SerializableDict a)
391− -> NodeId
392− -> Closure (Process a)
393− -> Process a
394−call dict nid proc = do
395− us <- getSelfPid
396− (pid, mRef) <- spawnMonitor nid (proc `bindCP` cpSend dict us)
397− -- We are guaranteed to receive the reply before the monitor notification
398− -- (if a reply is sent at all)
399− -- NOTE: This might not be true if we switch to unreliable delivery.
400− mResult <- receiveWait
401− [ match (return . Right)
402− , matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mRef)
403− (\(ProcessMonitorNotification _ _ reason) -> return (Left reason))
404− ]
405− case mResult of
406− Right a -> do
407− -- Wait for the monitor message so that we the mailbox doesn't grow
408− receiveWait
409− [ matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mRef)
410− (\(ProcessMonitorNotification {}) -> return ())
411− ]
412− -- Clean up connection to pid
413− reconnect pid
414− return a
415− Left err ->
416− fail $ "call: remote process died: " ++ show err
417−
418−-- | Spawn a child process, have the child link to the parent and the parent
419−-- monitor the child
420−spawnSupervised :: NodeId
421− -> Closure (Process ())
422− -> Process (ProcessId, MonitorRef)
423−spawnSupervised nid proc = do
424− us <- getSelfPid
425− them <- spawn nid (cpLink us `seqCP` proc)
426− ref <- monitor them
427− return (them, ref)
428−
429−-- | Spawn a new process, supplying it with a new 'ReceivePort' and return
430−-- the corresponding 'SendPort'.
431−spawnChannel :: forall a. Typeable a => Static (SerializableDict a)
432− -> NodeId
433− -> Closure (ReceivePort a -> Process ())
434− -> Process (SendPort a)
435−spawnChannel dict nid proc = do
436− us <- getSelfPid
437− _ <- spawn nid (go us)
438− expect
439− where
440− go :: ProcessId -> Closure (Process ())
441− go pid = cpNewChan dict
442− `bindCP`
443− (cpSend (sdictSendPort dict) pid `splitCP` proc)
444− `bindCP`
445− (idCP `closureCompose` staticClosure sndStatic)
446−
447--------------------------------------------------------------------------------
341--------------------------------------------------------------------------------
448-- Local versions of spawn --
342-- Local versions of spawn --
449--------------------------------------------------------------------------------
343--------------------------------------------------------------------------------
@@ -470,3 +364,4 @@ spawnChannelLocal proc = do
470 liftIO $ putMVar mvar sport
364 liftIO $ putMVar mvar sport
471 proc rport
365 proc rport
472 takeMVar mvar
366 takeMVar mvar
367+
src/Control/Distributed/Process/Closure.hs+2
@@ -173,6 +173,7 @@ module Control.Distributed.Process.Closure
173 -- * CP versions of Cloud Haskell primitives
173 -- * CP versions of Cloud Haskell primitives
174 , cpLink
174 , cpLink
175 , cpUnlink
175 , cpUnlink
176+ , cpRelay
176 , cpSend
177 , cpSend
177 , cpExpect
178 , cpExpect
178 , cpNewChan
179 , cpNewChan
@@ -205,6 +206,7 @@ import Control.Distributed.Process.Internal.Closure.BuiltIn
205 -- CP versions of Cloud Haskell primitives
206 -- CP versions of Cloud Haskell primitives
206 , cpLink
207 , cpLink
207 , cpUnlink
208 , cpUnlink
209+ , cpRelay
208 , cpSend
210 , cpSend
209 , cpExpect
211 , cpExpect
210 , cpNewChan
212 , cpNewChan
src/Control/Distributed/Process/Debug.hsnew file+280
@@ -0,0 +1,280 @@
1+-----------------------------------------------------------------------------
2+-- |
3+-- Module : Control.Distributed.Process.Debug
4+-- Copyright : (c) Well-Typed / Tim Watson
5+-- License : BSD3 (see the file LICENSE)
6+--
7+-- Maintainer : Tim Watson <watson.timothy@gmail.com>
8+-- Stability : experimental
9+-- Portability : non-portable (requires concurrency)
10+--
11+-- [Tracing/Debugging Facilities]
12+--
13+-- Cloud Haskell provides a general purpose tracing mechanism, allowing a
14+-- user supplied /tracer process/ to receive messages when certain classes of
15+-- system events occur. It's possible to use this facility to aid in debugging
16+-- and/or perform other diagnostic tasks to a program at runtime.
17+--
18+-- [Enabling Tracing]
19+--
20+-- Tracing is disabled by default, because it carries a relatively small, but
21+-- tangible cost. To enable tracing, the environment variable
22+-- @DISTRIBUTED_PROCESS_TRACE_ENABLED@ must be set to some value (actual content
23+-- is ignored). When the environment variable /is/ set, the system will
24+-- generate /trace events/ in various circumstances - see 'TraceEvent' for a
25+-- list of all the published events. A user can additionally publish custom
26+-- trace events in the form of 'TraceEvLog' log messages or pass custom
27+-- (i.e., completely user defined) event data using the 'traceMessage' function.
28+-- If the environment variable is not set, no trace events will ever be
29+-- published.
30+--
31+-- All published traces are forwarded to a /tracer process/, which can be
32+-- specified (and changed) at runtime using 'traceEnable'. Some pre-defined
33+-- tracer processes are provided for conveniently printing to stderr, a log file
34+-- or the GHC eventlog.
35+--
36+-- If a tracer process crashes, no attempt is made to restart it.
37+--
38+-- [Working with multiple tracer processes]
39+--
40+-- The system tracing facility only ever writes to a single tracer process. This
41+-- invariant insulates the tracer controller and ensures a fast path for
42+-- handling all trace events. /This/ module provides facilities for layering
43+-- trace handlers using Cloud Haskell's built-in delegation primitives.
44+--
45+-- The 'startTracer' function wraps the registered @tracer@ process with the
46+-- supplied handler and also forwarding trace events to the original tracer.
47+-- The corresponding 'stopTracer' function terminates tracer processes in
48+-- reverse of the order in which they were started, and re-establishes the
49+-- previous tracer process.
50+--
51+-- [Built in tracers]
52+--
53+-- The built in tracers provide a simple /logging/ facility that writes trace
54+-- events out to either a log file, @stderr@ or the GHC eventlog. These tracers
55+-- can be configured using environment variables, or specified manually using
56+-- the 'traceEnable' function. The base tracer process cannot be
57+--
58+-- When a new local node is started, the contents of several environment
59+-- variables are checked to determine how the default tracer process
60+-- will handle trace events. If none of these variables is set, then the
61+-- trace events will be effectively ignored, although they will still be
62+-- generated and passed through the system. Only one configuration will be
63+-- chosen - the first that contains a (valid) value. These environment
64+-- variables, in the order they're checked, are:
65+--
66+-- 1. @DISTRIBUTED_PROCESS_TRACE_FILE@
67+-- This is checked for a valid file path. If it exists and the file can be
68+-- opened for writing, all trace output will be directed thence. If the supplied
69+-- path is invalid, or the file is unavailable for writing - e.g., because
70+-- another node has already started tracing to it - then this tracer will be
71+-- disabled.
72+--
73+-- 2. @DISTRIBUTED_PROCESS_TRACE_CONSOLE@
74+-- This is checked for /any/ non-empty value. If set, then all trace output will
75+-- be directed to the system logger process.
76+--
77+-- 3. @DISTRIBUTED_PROCESS_TRACE_EVENTLOG@
78+-- This is checked for /any/ non-empty value. If set, all internal traces are
79+-- written to the GHC eventlog.
80+--
81+-- Users of the /simplelocalnet/ Cloud Haskell backend should also note that
82+-- because the trace file option only supports trace output from a single node
83+-- (so as to avoid interleaving), a file trace configured for the master node
84+-- will prevent slaves from tracing to the file and will need to fall back to
85+-- the console or eventlog tracers instead.
86+--
87+-- Support for writing to the eventlog requires specific intervention to work,
88+-- without which, written traces are silently dropped/ignored and no output will
89+-- be generated. The GHC eventlog documentation provides information about
90+-- enabling, viewing and working with event traces at
91+-- <http://hackage.haskell.org/trac/ghc/wiki/EventLog>.
92+--
93+module Control.Distributed.Process.Debug
94+ ( -- * Exported Data Types
95+ TraceArg(..)
96+ , TraceEvent(..)
97+ , TraceFlags(..)
98+ , TraceSubject(..)
99+ -- * Configuring Tracing
100+ , isTracingEnabled
101+ , enableTrace
102+ , enableTraceAsync
103+ , disableTrace
104+ , withTracer
105+ , withFlags
106+ , getTraceFlags
107+ , setTraceFlags
108+ , setTraceFlagsAsync
109+ , defaultTraceFlags
110+ , traceOn
111+ , traceOnly
112+ , traceOff
113+ -- * Debugging
114+ , startTracer
115+ , stopTracer
116+ -- * Sending Custom Trace Data
117+ , traceLog
118+ , traceLogFmt
119+ , traceMessage
120+ -- * Working with remote nodes
121+ , Remote.remoteTable
122+ , Remote.startTraceRelay
123+ , Remote.setTraceFlagsRemote
124+ -- * Built in tracers
125+ , systemLoggerTracer
126+ , logfileTracer
127+ , eventLogTracer
128+ )
129+ where
130+
131+import Control.Applicative ((<$>))
132+import Control.Distributed.Process.Internal.Primitives
133+ ( proxy
134+ , finally
135+ , die
136+ , whereis
137+ , send
138+ , receiveWait
139+ , matchIf
140+ , finally
141+ , try
142+ , monitor
143+ )
144+import Control.Distributed.Process.Internal.Types
145+ ( ProcessId
146+ , Process
147+ , LocalProcess(..)
148+ , ProcessMonitorNotification(..)
149+ )
150+import Control.Distributed.Process.Internal.Trace.Types
151+ ( TraceArg(..)
152+ , TraceEvent(..)
153+ , TraceFlags(..)
154+ , TraceSubject(..)
155+ , defaultTraceFlags
156+ )
157+import Control.Distributed.Process.Internal.Trace.Tracer
158+ ( systemLoggerTracer
159+ , logfileTracer
160+ , eventLogTracer
161+ )
162+import Control.Distributed.Process.Internal.Trace.Primitives
163+ ( withRegisteredTracer
164+ , enableTrace
165+ , enableTraceAsync
166+ , disableTrace
167+ , setTraceFlags
168+ , setTraceFlagsAsync
169+ , getTraceFlags
170+ , isTracingEnabled
171+ , traceOn
172+ , traceOff
173+ , traceOnly
174+ , traceLog
175+ , traceLogFmt
176+ , traceMessage
177+ )
178+import qualified Control.Distributed.Process.Internal.Trace.Remote as Remote
179+import Control.Distributed.Process.Node
180+
181+import Control.Exception (SomeException)
182+
183+import Control.Monad.IO.Class (liftIO)
184+import Control.Monad.Reader (ask)
185+
186+import Data.Binary()
187+
188+#if ! MIN_VERSION_base(4,6,0)
189+import Prelude hiding (catch)
190+#endif
191+
192+--------------------------------------------------------------------------------
193+-- Debugging/Tracing API --
194+--------------------------------------------------------------------------------
195+
196+-- | Starts a new tracer, using the supplied trace function.
197+-- Only one tracer can be registered at a time, however /this/ function overlays
198+-- the registered tracer with the supplied handler, allowing the user to layer
199+-- multiple tracers on top of one another, with trace events forwarded down
200+-- through all the layers in turn. Once the top layer is stopped, the user
201+-- is responsible for re-registering the original (prior) tracer pid before
202+-- terminating. See 'withTracer' for a mechanism that handles that.
203+startTracer :: (TraceEvent -> Process ()) -> Process ProcessId
204+startTracer handler = do
205+ withRegisteredTracer $ \pid -> do
206+ node <- processNode <$> ask
207+ newPid <- liftIO $ forkProcess node $ traceProxy pid handler
208+ enableTrace newPid -- invokes sync + registration
209+ return newPid
210+
211+-- | Evaluate @proc@ with tracing enabled via @handler@, and immediately
212+-- disable tracing thereafter, before giving the result (or exception
213+-- in case of failure).
214+withTracer :: forall a.
215+ (TraceEvent -> Process ())
216+ -> Process a
217+ -> Process (Either SomeException a)
218+withTracer handler proc = do
219+ previous <- whereis "tracer"
220+ tracer <- startTracer handler
221+ finally (try proc)
222+ (stopTracing tracer previous)
223+ where
224+ stopTracing :: ProcessId -> Maybe ProcessId -> Process ()
225+ stopTracing tracer previousTracer = do
226+ case previousTracer of
227+ Nothing -> return ()
228+ Just _ -> do
229+ ref <- monitor tracer
230+ send tracer TraceEvDisable
231+ receiveWait [
232+ matchIf (\(ProcessMonitorNotification ref' _ _) -> ref == ref')
233+ (\_ -> return ())
234+ ]
235+
236+-- | Evaluate @proc@ with the supplied flags enabled. Any previously set
237+-- trace flags are restored immediately afterwards.
238+withFlags :: forall a.
239+ TraceFlags
240+ -> Process a
241+ -> Process (Either SomeException a)
242+withFlags flags proc = do
243+ oldFlags <- getTraceFlags
244+ finally (setTraceFlags flags >> try proc)
245+ (setTraceFlags oldFlags)
246+
247+traceProxy :: ProcessId -> (TraceEvent -> Process ()) -> Process ()
248+traceProxy pid act = do
249+ proxy pid $ \(ev :: TraceEvent) ->
250+ case ev of
251+ (TraceEvTakeover _) -> return False
252+ TraceEvDisable -> die "disabled"
253+ _ -> act ev >> return True
254+
255+-- | Stops a user supplied tracer started with 'startTracer'.
256+-- Note that only one tracer process can be active at any given time.
257+-- This process will stop the last process started with 'startTracer'.
258+-- If 'startTracer' is called multiple times, successive calls to this
259+-- function will stop the tracers in the reverse order which they were
260+-- started.
261+--
262+-- This function will never stop the system tracer (i.e., the tracer
263+-- initially started when the node is created), therefore once all user
264+-- supplied tracers (i.e., processes started via 'startTracer') have exited,
265+-- subsequent calls to this function will have no effect.
266+--
267+-- If tracing support is disabled for the node, this function will also
268+-- have no effect. If the last tracer to have been registered was not started
269+-- with 'startTracer' then the behaviour of this function is /undefined/.
270+stopTracer :: Process ()
271+stopTracer =
272+ withRegisteredTracer $ \pid -> do
273+ -- we need to avoid killing the initial (base) tracer, as
274+ -- nothing we rely on having exactly 1 registered tracer
275+ -- process at all times.
276+ basePid <- whereis "tracer.initial"
277+ case basePid == (Just pid) of
278+ True -> return ()
279+ False -> send pid TraceEvDisable
280+
src/Control/Distributed/Process/Internal/Closure/BuiltIn.hs+24
@@ -16,13 +16,16 @@ module Control.Distributed.Process.Internal.Closure.BuiltIn
16 , bindCP
16 , bindCP
17 , seqCP
17 , seqCP
18 -- * CP versions of Cloud Haskell primitives
18 -- * CP versions of Cloud Haskell primitives
19+ , decodeProcessIdStatic
19 , cpLink
20 , cpLink
20 , cpUnlink
21 , cpUnlink
22+ , cpRelay
21 , cpSend
23 , cpSend
22 , cpExpect
24 , cpExpect
23 , cpNewChan
25 , cpNewChan
24 -- * Support for some CH operations
26 -- * Support for some CH operations
25 , cpDelay
27 , cpDelay
28+ , cpEnableTraceRemote
26 ) where
29 ) where
27
30
28import Data.ByteString.Lazy (ByteString)
31import Data.ByteString.Lazy (ByteString)
@@ -56,6 +59,7 @@ import Control.Distributed.Process.Internal.Types
56import Control.Distributed.Process.Internal.Primitives
59import Control.Distributed.Process.Internal.Primitives
57 ( link
60 ( link
58 , unlink
61 , unlink
62+ , relay
59 , send
63 , send
60 , expect
64 , expect
61 , newChan
65 , newChan
@@ -82,6 +86,7 @@ remoteTable =
82 . registerStatic "$decodeProcessId" (toDynamic (decode :: ByteString -> ProcessId))
86 . registerStatic "$decodeProcessId" (toDynamic (decode :: ByteString -> ProcessId))
83 . registerStatic "$link" (toDynamic link)
87 . registerStatic "$link" (toDynamic link)
84 . registerStatic "$unlink" (toDynamic unlink)
88 . registerStatic "$unlink" (toDynamic unlink)
89+ . registerStatic "$relay" (toDynamic relay)
85 . registerStatic "$sendDict" (toDynamic (sendDict :: SerializableDict ANY -> ProcessId -> ANY -> Process ()))
90 . registerStatic "$sendDict" (toDynamic (sendDict :: SerializableDict ANY -> ProcessId -> ANY -> Process ()))
86 . registerStatic "$expectDict" (toDynamic (expectDict :: SerializableDict ANY -> Process ANY))
91 . registerStatic "$expectDict" (toDynamic (expectDict :: SerializableDict ANY -> Process ANY))
87 . registerStatic "$newChanDict" (toDynamic (newChanDict :: SerializableDict ANY -> Process (SendPort ANY, ReceivePort ANY)))
92 . registerStatic "$newChanDict" (toDynamic (newChanDict :: SerializableDict ANY -> Process (SendPort ANY, ReceivePort ANY)))
@@ -246,6 +251,22 @@ cpNewChan dict = staticClosure (newChanDictStatic `staticApply` dict)
246 => Static (SerializableDict a -> Process (SendPort a, ReceivePort a))
251 => Static (SerializableDict a -> Process (SendPort a, ReceivePort a))
247 newChanDictStatic = staticLabel "$newChanDict"
252 newChanDictStatic = staticLabel "$newChanDict"
248
253
254+-- | 'CP' version of 'relay'
255+cpRelay :: ProcessId -> Closure (Process ())
256+cpRelay = closure (relayStatic `staticCompose` decodeProcessIdStatic) . encode
257+ where
258+ relayStatic :: Static (ProcessId -> Process ())
259+ relayStatic = staticLabel "$relay"
260+
261+-- TODO: move cpEnableTraceRemote into Trace/Primitives.hs
262+
263+cpEnableTraceRemote :: ProcessId -> Closure (Process ())
264+cpEnableTraceRemote =
265+ closure (enableTraceStatic `staticCompose` decodeProcessIdStatic) . encode
266+ where
267+ enableTraceStatic :: Static (ProcessId -> Process ())
268+ enableTraceStatic = staticLabel "$enableTraceRemote"
269+
249--------------------------------------------------------------------------------
270--------------------------------------------------------------------------------
250-- Support for spawn --
271-- Support for spawn --
251--------------------------------------------------------------------------------
272--------------------------------------------------------------------------------
@@ -275,3 +296,6 @@ cpDelay = closureApply . cpDelay'
275
296
276 delayStatic :: Static (ProcessId -> Process () -> Process ())
297 delayStatic :: Static (ProcessId -> Process () -> Process ())
277 delayStatic = staticLabel "$delay"
298 delayStatic = staticLabel "$delay"
299+
300+
301+
src/Control/Distributed/Process/Internal/Primitives.hs+76−62
@@ -1,3 +1,5 @@
1+{-# LANGUAGE BangPatterns #-}
2+
1-- | Cloud Haskell primitives
3-- | Cloud Haskell primitives
2--
4--
3-- We define these in a separate module so that we don't have to rely on
5-- We define these in a separate module so that we don't have to rely on
@@ -28,6 +30,9 @@ module Control.Distributed.Process.Internal.Primitives
28 , unwrapMessage
30 , unwrapMessage
29 , handleMessage
31 , handleMessage
30 , forward
32 , forward
33+ , delegate
34+ , relay
35+ , proxy
31 -- * Process management
36 -- * Process management
32 , terminate
37 , terminate
33 , ProcessTerminationException(..)
38 , ProcessTerminationException(..)
@@ -88,8 +93,6 @@ module Control.Distributed.Process.Internal.Primitives
88 -- * Reconnecting
93 -- * Reconnecting
89 , reconnect
94 , reconnect
90 , reconnectPort
95 , reconnectPort
91− -- * Tracing/Debugging
92− , trace
93 ) where
96 ) where
94
97
95#if ! MIN_VERSION_base(4,6,0)
98#if ! MIN_VERSION_base(4,6,0)
@@ -129,7 +132,10 @@ import Control.Distributed.Process.Internal.CQueue
129 )
132 )
130import Control.Distributed.Process.Serializable (Serializable, fingerprint)
133import Control.Distributed.Process.Serializable (Serializable, fingerprint)
131import Data.Accessor ((^.), (^:), (^=))
134import Data.Accessor ((^.), (^:), (^=))
132−import Control.Distributed.Static (Closure, Static)
135+import Control.Distributed.Static
136+ ( Static
137+ , Closure
138+ )
133import Data.Rank1Typeable (Typeable)
139import Data.Rank1Typeable (Typeable)
134import qualified Control.Distributed.Static as Static (unstatic, unclosure)
140import qualified Control.Distributed.Static as Static (unstatic, unclosure)
135import Control.Distributed.Process.Internal.Types
141import Control.Distributed.Process.Internal.Types
@@ -176,12 +182,16 @@ import Control.Distributed.Process.Internal.Messaging
176 , sendPayload
182 , sendPayload
177 , disconnect
183 , disconnect
178 )
184 )
179−import qualified Control.Distributed.Process.Internal.Trace as Trace
185+import Control.Distributed.Process.Internal.Trace.Types
186+ ( traceEvent
187+ , TraceEvent(..)
188+ )
180import Control.Distributed.Process.Internal.WeakTQueue
189import Control.Distributed.Process.Internal.WeakTQueue
181 ( newTQueueIO
190 ( newTQueueIO
182 , readTQueue
191 , readTQueue
183 , mkWeakTQueue
192 , mkWeakTQueue
184 )
193 )
194+
185import Unsafe.Coerce
195import Unsafe.Coerce
186
196
187--------------------------------------------------------------------------------
197--------------------------------------------------------------------------------
@@ -191,18 +201,24 @@ import Unsafe.Coerce
191-- | Send a message
201-- | Send a message
192send :: Serializable a => ProcessId -> a -> Process ()
202send :: Serializable a => ProcessId -> a -> Process ()
193-- This requires a lookup on every send. If we want to avoid that we need to
203-- This requires a lookup on every send. If we want to avoid that we need to
194−-- modify serializable to allow for stateful (IO) deserialization
204+-- modify serializable to allow for stateful (IO) deserialization.
195send them msg = do
205send them msg = do
196 proc <- ask
206 proc <- ask
207+ let us = processId proc
208+ node = processNode proc
197− let node = localNodeId (processNode proc)
209+ nodeId = localNodeId node
198 destNode = (processNodeId them) in do
210 destNode = (processNodeId them) in do
199− case destNode == node of
211+ case destNode == nodeId of
200 True -> sendLocal them msg
212 True -> sendLocal them msg
201 False -> liftIO $ sendMessage (processNode proc)
213 False -> liftIO $ sendMessage (processNode proc)
202 (ProcessIdentifier (processId proc))
214 (ProcessIdentifier (processId proc))
203 (ProcessIdentifier them)
215 (ProcessIdentifier them)
204 NoImplicitReconnect
216 NoImplicitReconnect
205 msg
217 msg
218+ -- We do not fire the trace event until after the sending is complete;
219+ -- In the remote case, 'sendMessage' can block in the networking stack.
220+ liftIO $ traceEvent (localTracer node)
221+ (TraceEvSent them us (createUnencodedMessage msg))
206
222
207-- | Wait for a message of a specific type
223-- | Wait for a message of a specific type
208expect :: forall a. Serializable a => Process a
224expect :: forall a. Serializable a => Process a
@@ -358,33 +374,44 @@ matchMessageIf c p = Match $ MatchMsg $ \msg ->
358forward :: Message -> ProcessId -> Process ()
374forward :: Message -> ProcessId -> Process ()
359forward msg them = do
375forward msg them = do
360 proc <- ask
376 proc <- ask
361− let node = localNodeId (processNode proc)
377+ let node = processNode proc
378+ us = processId proc
379+ nid = localNodeId node
362 destNode = (processNodeId them) in do
380 destNode = (processNodeId them) in do
363− case destNode == node of
381+ case destNode == nid of
364 True -> sendCtrlMsg Nothing (LocalSend them msg)
382 True -> sendCtrlMsg Nothing (LocalSend them msg)
365 False -> liftIO $ sendPayload (processNode proc)
383 False -> liftIO $ sendPayload (processNode proc)
366 (ProcessIdentifier (processId proc))
384 (ProcessIdentifier (processId proc))
367 (ProcessIdentifier them)
385 (ProcessIdentifier them)
368 NoImplicitReconnect
386 NoImplicitReconnect
369 (messageToPayload msg)
387 (messageToPayload msg)
388+ -- We do not fire the trace event until after the sending is complete;
389+ -- In the remote case, 'sendMessage' can block in the networking stack.
390+ liftIO $ traceEvent (localTracer node)
391+ (TraceEvSent them us (createUnencodedMessage msg))
370
392
371
393
372-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are
394-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are
373−-- 'Serializable' - like the datum they contain - but remember that deserializing
395+-- 'Serializable' - like the datum they contain - but also note, deserialising
374-- a 'Message' will yield a 'Message', not the type within it! To obtain the
396-- a 'Message' will yield a 'Message', not the type within it! To obtain the
375-- wrapped datum, use 'unwrapMessage' or 'handleMessage' with a specific type.
397-- wrapped datum, use 'unwrapMessage' or 'handleMessage' with a specific type.
376--
398--
377-- @
399-- @
378-- do
400-- do
379−-- self <- getSelfPid
401+-- self <- getSelfPid
380−-- send self (wrapMessage "blah")
402+-- send self (wrapMessage "blah")
381−-- Nothing <- expectTimeout 1000000 :: Process (Maybe String)
403+-- Nothing <- expectTimeout 1000000 :: Process (Maybe String)
382−-- (Just m) <- expectTimeout 1000000 :: Process (Maybe Message)
404+-- (Just m) <- expectTimeout 1000000 :: Process (Maybe Message)
383−-- (Just "blah") <- unwrapMessage m :: Process (Maybe String)
405+-- (Just "blah") <- unwrapMessage m :: Process (Maybe String)
384-- @
406-- @
385--
407--
386wrapMessage :: Serializable a => a -> Message
408wrapMessage :: Serializable a => a -> Message
387−wrapMessage = createMessage
409+wrapMessage = createUnencodedMessage -- see [note Serializable UnencodedMessage]
410+
411+-- [note Serializable UnencodedMessage]
412+-- if we attempt to serialise an UnencodedMessage, it will be converted to an
413+-- EncodedMessage first, so we're ok here. See 'messageToPayload' in
414+-- Primitives.hs for the gory details
388
415
389-- | Attempt to unwrap a raw 'Message'.
416-- | Attempt to unwrap a raw 'Message'.
390-- If the type of the decoded message payload matches the expected type, the
417-- If the type of the decoded message payload matches the expected type, the
@@ -480,6 +507,36 @@ succeeds) that unsafeCoerce will not fail.
480matchUnknown :: Process b -> Match b
507matchUnknown :: Process b -> Match b
481matchUnknown p = Match $ MatchMsg (const (Just p))
508matchUnknown p = Match $ MatchMsg (const (Just p))
482
509
510+-- | Receives messages and forwards them to @pid@ if @p msg == True@.
511+delegate :: ProcessId -> (Message -> Bool) -> Process ()
512+delegate pid p = do
513+ receiveWait [
514+ matchAny (\m -> case (p m) of
515+ True -> forward m pid
516+ False -> return ())
517+ ]
518+ delegate pid p
519+
520+-- | A straight relay that simply forwards all messages to the supplied pid.
521+relay :: ProcessId -> Process ()
522+relay !pid = receiveWait [ matchAny (\m -> forward m pid) ] >> relay pid
523+
524+-- | Proxies @pid@ and forwards messages whenever @proc@ evaluates to @True@.
525+-- Unlike 'delegate' the predicate @proc@ runs in the 'Process' monad, allowing
526+-- for richer proxy behaviour. If @proc@ returns @False@ or the /runtime type check/
527+-- fails, no action is taken and the proxy process will continue running.
528+proxy :: Serializable a => ProcessId -> (a -> Process Bool) -> Process ()
529+proxy pid proc = do
530+ receiveWait [
531+ matchAny (\m -> do
532+ next <- handleMessage m proc
533+ case next of
534+ Just True -> forward m pid
535+ Just False -> return () -- explicitly ignored
536+ Nothing -> return ()) -- un-routable
537+ ]
538+ proxy pid proc
539+
483--------------------------------------------------------------------------------
540--------------------------------------------------------------------------------
484-- Process management --
541-- Process management --
485--------------------------------------------------------------------------------
542--------------------------------------------------------------------------------
@@ -750,7 +807,7 @@ finally a sequel = bracket_ (return ()) sequel a
750data Handler a = forall e . Exception e => Handler (e -> Process a)
807data Handler a = forall e . Exception e => Handler (e -> Process a)
751
808
752instance Functor Handler where
809instance Functor Handler where
753− fmap f (Handler h) = Handler (fmap f . h)
810+ fmap f (Handler h) = Handler (fmap f . h)
754
811
755-- | Lift 'Control.Exception.catches'
812-- | Lift 'Control.Exception.catches'
756catches :: Process a -> [Handler a] -> Process a
813catches :: Process a -> [Handler a] -> Process a
@@ -991,56 +1048,13 @@ reconnectPort them = do
991 node <- processNode <$> ask
1048 node <- processNode <$> ask
992 liftIO $ disconnect node (ProcessIdentifier us) (SendPortIdentifier (sendPortId them))
1049 liftIO $ disconnect node (ProcessIdentifier us) (SendPortIdentifier (sendPortId them))
993
1050
994−--------------------------------------------------------------------------------
995−-- Debugging/Tracing --
996−--------------------------------------------------------------------------------
997−
998−-- | Send a message to the internal (system) trace facility. If tracing is
999−-- enabled, this will create a custom trace event. Note that several Cloud Haskell
1000−-- sub-systems also generate trace events for informational/debugging purposes,
1001−-- thus traces generated this way will not be the only output seen.
1002−--
1003−-- Just as with the "Debug.Trace" module, this is a debugging/tracing facility
1004−-- for use in development, and should not be used in a production setting -
1005−-- which is why the default behaviour is to trace to the GHC eventlog. For a
1006−-- general purpose logging facility, you should consider 'say'.
1007−--
1008−-- Trace events can be written to the GHC event log, a text file, or to the
1009−-- standard system logger process (see 'say'). The default behaviour for writing
1010−-- to the eventlog requires specific intervention to work, without which traces
1011−-- are silently dropped/ignored and no output will be generated.
1012−-- The GHC eventlog documentation provides information about enabling, viewing
1013−-- and working with event traces: <http://hackage.haskell.org/trac/ghc/wiki/EventLog>.
1014−--
1015−-- When a new local node is started, the contents of the environment variable
1016−-- @DISTRIBUTED_PROCESS_TRACE_FILE@ are checked for a valid file path. If this
1017−-- exists and the file can be opened for writing, all trace output will be directed
1018−-- thence. If the environment variable is empty, the path invalid, or the file
1019−-- unavailable for writing - e.g., because another node has already started
1020−-- tracing to it - then the @DISTRIBUTED_PROCESS_TRACE_CONSOLE@ environment
1021−-- variable is checked for /any/ non-empty value. If this is set, then all trace
1022−-- output will be directed to the system logger process. If neither evironment
1023−-- variable provides a valid trace configuration, all internal traces are written
1024−-- to "Debug.Trace.traceEventIO", which writes to the GHC eventlog.
1025−--
1026−-- Users of the /simplelocalnet/ Cloud Haskell backend should also note that
1027−-- because the trace file option only supports trace output from a single node
1028−-- (so as to avoid interleaving), a file trace configured for the master node will
1029−-- prevent slaves from tracing to the file and they will fall back to using the
1030−-- console/'say' or eventlog instead.
1031−--
1032−trace :: String -> Process ()
1033−trace s = do
1034− node <- processNode <$> ask
1035− liftIO $ Trace.trace (localTracer node) s
1036−
1037--------------------------------------------------------------------------------
1051--------------------------------------------------------------------------------
1038-- Auxiliary functions --
1052-- Auxiliary functions --
1039--------------------------------------------------------------------------------
1053--------------------------------------------------------------------------------
1040
1054
1041sendLocal :: (Serializable a) => ProcessId -> a -> Process ()
1055sendLocal :: (Serializable a) => ProcessId -> a -> Process ()
1042−sendLocal pid msg =
1056+sendLocal to msg =
1043− sendCtrlMsg Nothing $ LocalSend pid (createUnencodedMessage msg)
1057+ sendCtrlMsg Nothing $ LocalSend to (createUnencodedMessage msg)
1044
1058
1045sendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()
1059sendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()
1046sendChanLocal spId msg =
1060sendChanLocal spId msg =
src/Control/Distributed/Process/Internal/Spawn.hsnew file+162
@@ -0,0 +1,162 @@
1+module Control.Distributed.Process.Internal.Spawn
2+ ( spawn
3+ , spawnLink
4+ , spawnMonitor
5+ , call
6+ , spawnSupervised
7+ , spawnChannel
8+ ) where
9+
10+import Data.Typeable (Typeable)
11+import Control.Distributed.Static
12+ ( Static
13+ , Closure
14+ , closureCompose
15+ , staticClosure
16+ )
17+import Control.Distributed.Process.Internal.Types
18+ ( NodeId(..)
19+ , ProcessId(..)
20+ , Process(..)
21+ , MonitorRef(..)
22+ , ProcessMonitorNotification(..)
23+ , NodeMonitorNotification(..)
24+ , DidSpawn(..)
25+ , SendPort(..)
26+ , ReceivePort(..)
27+ , nullProcessId
28+ )
29+import Control.Distributed.Process.Serializable (Serializable, SerializableDict)
30+import Control.Distributed.Process.Internal.Closure.BuiltIn
31+ ( sdictSendPort
32+ , sndStatic
33+ , idCP
34+ , seqCP
35+ , bindCP
36+ , splitCP
37+ , cpLink
38+ , cpSend
39+ , cpNewChan
40+ , cpDelay
41+ )
42+import Control.Distributed.Process.Internal.Primitives
43+ ( -- Basic messaging
44+ send
45+ , expect
46+ , receiveWait
47+ , match
48+ , matchIf
49+ , link
50+ , getSelfPid
51+ , monitor
52+ , monitorNode
53+ , unmonitor
54+ , spawnAsync
55+ , reconnect
56+ )
57+
58+-- | Spawn a process
59+--
60+-- For more information about 'Closure', see
61+-- "Control.Distributed.Process.Closure".
62+--
63+-- See also 'call'.
64+spawn :: NodeId -> Closure (Process ()) -> Process ProcessId
65+spawn nid proc = do
66+ us <- getSelfPid
67+ mRef <- monitorNode nid
68+ sRef <- spawnAsync nid (cpDelay us proc)
69+ receiveWait [
70+ matchIf (\(DidSpawn ref _) -> ref == sRef) $ \(DidSpawn _ pid) -> do
71+ unmonitor mRef
72+ send pid ()
73+ return pid
74+ , matchIf (\(NodeMonitorNotification ref _ _) -> ref == mRef) $ \_ ->
75+ return (nullProcessId nid)
76+ ]
77+
78+-- | Spawn a process and link to it
79+--
80+-- Note that this is just the sequential composition of 'spawn' and 'link'.
81+-- (The "Unified" semantics that underlies Cloud Haskell does not even support
82+-- a synchronous link operation)
83+spawnLink :: NodeId -> Closure (Process ()) -> Process ProcessId
84+spawnLink nid proc = do
85+ pid <- spawn nid proc
86+ link pid
87+ return pid
88+
89+-- | Like 'spawnLink', but monitor the spawned process
90+spawnMonitor :: NodeId -> Closure (Process ()) -> Process (ProcessId, MonitorRef)
91+spawnMonitor nid proc = do
92+ pid <- spawn nid proc
93+ ref <- monitor pid
94+ return (pid, ref)
95+
96+-- | Run a process remotely and wait for it to reply
97+--
98+-- We monitor the remote process: if it dies before it can send a reply, we die
99+-- too.
100+--
101+-- For more information about 'Static', 'SerializableDict', and 'Closure', see
102+-- "Control.Distributed.Process.Closure".
103+--
104+-- See also 'spawn'.
105+call :: Serializable a
106+ => Static (SerializableDict a)
107+ -> NodeId
108+ -> Closure (Process a)
109+ -> Process a
110+call dict nid proc = do
111+ us <- getSelfPid
112+ (pid, mRef) <- spawnMonitor nid (proc `bindCP` cpSend dict us)
113+ -- We are guaranteed to receive the reply before the monitor notification
114+ -- (if a reply is sent at all)
115+ -- NOTE: This might not be true if we switch to unreliable delivery.
116+ mResult <- receiveWait
117+ [ match (return . Right)
118+ , matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mRef)
119+ (\(ProcessMonitorNotification _ _ reason) -> return (Left reason))
120+ ]
121+ case mResult of
122+ Right a -> do
123+ -- Wait for the monitor message so that we the mailbox doesn't grow
124+ receiveWait
125+ [ matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mRef)
126+ (\(ProcessMonitorNotification {}) -> return ())
127+ ]
128+ -- Clean up connection to pid
129+ reconnect pid
130+ return a
131+ Left err ->
132+ fail $ "call: remote process died: " ++ show err
133+
134+-- | Spawn a child process, have the child link to the parent and the parent
135+-- monitor the child
136+spawnSupervised :: NodeId
137+ -> Closure (Process ())
138+ -> Process (ProcessId, MonitorRef)
139+spawnSupervised nid proc = do
140+ us <- getSelfPid
141+ them <- spawn nid (cpLink us `seqCP` proc)
142+ ref <- monitor them
143+ return (them, ref)
144+
145+-- | Spawn a new process, supplying it with a new 'ReceivePort' and return
146+-- the corresponding 'SendPort'.
147+spawnChannel :: forall a. Typeable a => Static (SerializableDict a)
148+ -> NodeId
149+ -> Closure (ReceivePort a -> Process ())
150+ -> Process (SendPort a)
151+spawnChannel dict nid proc = do
152+ us <- getSelfPid
153+ _ <- spawn nid (go us)
154+ expect
155+ where
156+ go :: ProcessId -> Closure (Process ())
157+ go pid = cpNewChan dict
158+ `bindCP`
159+ (cpSend (sdictSendPort dict) pid `splitCP` proc)
160+ `bindCP`
161+ (idCP `closureCompose` staticClosure sndStatic)
162+
src/Control/Distributed/Process/Internal/Trace.hsdeleted−122
@@ -1,122 +0,0 @@
1−-- | Simple (internal) system logging/tracing support.
2−module Control.Distributed.Process.Internal.Trace
3− ( Tracer
4− , TraceArg(..)
5− , trace
6− , traceFormat
7− , startTracing
8− , stopTracer
9− ) where
10−
11−import Control.Concurrent (forkIO)
12−import Control.Concurrent.Chan (writeChan)
13−import Control.Concurrent.STM
14− ( TQueue
15− , newTQueueIO
16− , readTQueue
17− , writeTQueue
18− , atomically
19− )
20−import Control.Distributed.Process.Internal.Types
21− ( Tracer(..)
22− , LocalNode(..)
23− , NCMsg(..)
24− , Identifier(ProcessIdentifier)
25− , ProcessSignal(NamedSend)
26− , forever'
27− , nullProcessId
28− , createMessage
29− )
30−import Control.Exception
31− ( catch
32− , throwTo
33− , SomeException
34− , AsyncException(ThreadKilled)
35− )
36−import Data.List (intersperse)
37−import Data.Time.Clock (getCurrentTime)
38−import Data.Time.Format (formatTime)
39−import Debug.Trace (traceEventIO)
40−
41−import Prelude hiding (catch)
42−
43−import System.Environment (getEnv)
44−import System.IO
45− ( Handle
46− , IOMode(AppendMode)
47− , BufferMode(..)
48− , openFile
49− , hClose
50− , hPutStrLn
51− , hSetBuffering
52− )
53−import System.Locale (defaultTimeLocale)
54−
55−data TraceArg =
56− TraceStr String
57− | forall a. (Show a) => Trace a
58−
59−startTracing :: LocalNode -> IO LocalNode
60−startTracing node = do
61− tracer <- defaultTracer node
62− return node { localTracer = tracer }
63−
64−defaultTracer :: LocalNode -> IO Tracer
65−defaultTracer node = do
66− catch (getEnv "DISTRIBUTED_PROCESS_TRACE_FILE" >>= logfileTracer)
67− (\(_ :: IOError) -> defaultTracerAux node)
68−
69−defaultTracerAux :: LocalNode -> IO Tracer
70−defaultTracerAux node = do
71− catch (getEnv "DISTRIBUTED_PROCESS_TRACE_CONSOLE" >> procTracer node)
72− (\(_ :: IOError) -> return (EventLogTracer traceEventIO))
73− where procTracer :: LocalNode -> IO Tracer
74− procTracer n = return $ (LocalNodeTracer n)
75−
76−logfileTracer :: FilePath -> IO Tracer
77−logfileTracer p = do
78− q <- newTQueueIO
79− h <- openFile p AppendMode
80− hSetBuffering h LineBuffering
81− tid <- forkIO $ logger h q `catch` (\(_ :: SomeException) ->
82− hClose h >> return ())
83− return $ LogFileTracer tid q h
84− where logger :: Handle -> TQueue String -> IO ()
85− logger h q' = forever' $ do
86− msg <- atomically $ readTQueue q'
87− now <- getCurrentTime
88− hPutStrLn h $ msg ++ (formatTime defaultTimeLocale " - %c" now)
89−
90−-- TODO: compatibility layer for GHC/base versions (e.g., where's killThread?)
91−
92−stopTracer :: Tracer -> IO () -- overzealous but harmless duplication of hClose
93−stopTracer (LogFileTracer tid _ h) = throwTo tid ThreadKilled >> hClose h
94−stopTracer _ = return ()
95−
96−trace :: Tracer -> String -> IO ()
97−trace (LogFileTracer _ q _) msg = atomically $ writeTQueue q msg
98−trace (LocalNodeTracer n) msg = sendTraceMsg n msg
99−trace (EventLogTracer t) msg = t msg
100−trace InactiveTracer _ = return ()
101−
102−traceFormat :: Tracer
103− -> String
104− -> [TraceArg]
105− -> IO ()
106−traceFormat t d ls =
107− trace t $ concat (intersperse d (map toS ls))
108− where toS :: TraceArg -> String
109− toS (TraceStr s) = s
110− toS (Trace a) = show a
111−
112−sendTraceMsg :: LocalNode -> String -> IO ()
113−sendTraceMsg node string = do
114− now <- getCurrentTime
115− msg <- return $ (formatTime defaultTimeLocale "%c" now, string)
116− emptyPid <- return $ (nullProcessId (localNodeId node))
117− traceMsg <- return $ NCMsg {
118− ctrlMsgSender = ProcessIdentifier (emptyPid)
119− , ctrlMsgSignal = (NamedSend "logger" (createMessage msg))
120− }
121− writeChan (localCtrlChan node) traceMsg
122−
src/Control/Distributed/Process/Internal/Trace/Primitives.hsnew file+180
@@ -0,0 +1,180 @@
1+-- | Keeps the tracing API calls separate from the Tracer implementation,
2+-- which allows us to avoid a nasty import cycle between tracing and
3+-- the messaging primitives that rely on it, and also between the node
4+-- controller (which requires access to the tracing related elements of
5+-- our RemoteTable) and the Debug module, which requires @forkProcess@.
6+module Control.Distributed.Process.Internal.Trace.Primitives
7+ ( Tracer
8+ , TraceEvent(..)
9+ -- * Sending Trace Data
10+ , traceLog
11+ , traceLogFmt
12+ , traceMessage
13+ -- * Configuring A Tracer
14+ , isTracingEnabled
15+ , defaultTraceFlags
16+ , enableTrace
17+ , enableTraceAsync
18+ , disableTrace
19+ , disableTraceAsync
20+ , getTraceFlags
21+ , setTraceFlags
22+ , setTraceFlagsAsync
23+ , traceOnly
24+ , traceOn
25+ , traceOff
26+ , withLocalTracer
27+ , withRegisteredTracer
28+ ) where
29+
30+import Control.Applicative ((<$>))
31+import Control.Distributed.Process.Internal.Primitives
32+ ( whereis
33+ , newChan
34+ , receiveChan
35+ )
36+import Control.Distributed.Process.Internal.Trace.Types
37+ ( TraceArg(..)
38+ , TraceEvent(..)
39+ , TraceFlags(..)
40+ , TraceOk(..)
41+ , TraceSubject(..)
42+ , defaultTraceFlags
43+ )
44+import qualified Control.Distributed.Process.Internal.Trace.Types as Tracer
45+ ( traceLog
46+ , traceLogFmt
47+ , traceMessage
48+ , enableTrace
49+ , enableTraceSync
50+ , disableTrace
51+ , disableTraceSync
52+ , setTraceFlags
53+ , setTraceFlagsSync
54+ , getTraceFlags
55+ , getCurrentTraceClient
56+ )
57+import Control.Distributed.Process.Internal.Types
58+ ( Tracer(..)
59+ , Process
60+ , ProcessId
61+ , LocalProcess(..)
62+ , LocalNode(localTracer)
63+ , SendPort
64+ )
65+import Control.Distributed.Process.Serializable
66+import Control.Monad.IO.Class (liftIO)
67+import Control.Monad.Reader (ask)
68+
69+import qualified Data.Set as Set (fromList)
70+
71+--------------------------------------------------------------------------------
72+-- Main API --
73+--------------------------------------------------------------------------------
74+
75+-- | Maps a list of identifiers that can be mapped to process ids,
76+-- to a 'TraceSubject'.
77+class Traceable a where
78+ uod :: [a] -> TraceSubject
79+
80+instance Traceable ProcessId where
81+ uod = TraceProcs . Set.fromList
82+
83+instance Traceable String where
84+ uod = TraceNames . Set.fromList
85+
86+-- | Turn tracing for for a subset of trace targets.
87+traceOnly :: Traceable a => [a] -> Maybe TraceSubject
88+traceOnly = Just . uod
89+
90+-- | Trace all targets.
91+traceOn :: Maybe TraceSubject
92+traceOn = Just TraceAll
93+
94+-- | Trace no targets.
95+traceOff :: Maybe TraceSubject
96+traceOff = Nothing
97+
98+-- | Determine whether or not tracing has been enabled.
99+isTracingEnabled :: Process Bool
100+isTracingEnabled = do
101+ node <- processNode <$> ask
102+ case (localTracer node) of
103+ (ActiveTracer _ _) -> return True
104+ _ -> return False
105+
106+-- | Enable tracing to the supplied process.
107+enableTraceAsync :: ProcessId -> Process ()
108+enableTraceAsync pid = withLocalTracer $ \t -> liftIO $ Tracer.enableTrace t pid
109+
110+-- TODO: refactor _Sync versions of trace configuration functions...
111+
112+-- | Enable tracing to the supplied process and wait for a @TraceOk@
113+-- response from the trace coordinator process.
114+enableTrace :: ProcessId -> Process ()
115+enableTrace pid =
116+ withLocalTracerSync $ \t sp -> Tracer.enableTraceSync t sp pid
117+
118+-- | Disable the currently configured trace.
119+disableTraceAsync :: Process ()
120+disableTraceAsync = withLocalTracer $ \t -> liftIO $ Tracer.disableTrace t
121+
122+-- | Disable the currently configured trace and wait for a @TraceOk@
123+-- response from the trace coordinator process.
124+disableTrace :: Process ()
125+disableTrace =
126+ withLocalTracerSync $ \t sp -> Tracer.disableTraceSync t sp
127+
128+getTraceFlags :: Process TraceFlags
129+getTraceFlags = do
130+ (sp, rp) <- newChan
131+ withLocalTracer $ \t -> liftIO $ Tracer.getTraceFlags t sp
132+ receiveChan rp
133+
134+-- | Set the given flags for the current tracer.
135+setTraceFlagsAsync :: TraceFlags -> Process ()
136+setTraceFlagsAsync f = withLocalTracer $ \t -> liftIO $ Tracer.setTraceFlags t f
137+
138+-- | Set the given flags for the current tracer and wait for a @TraceOk@
139+-- response from the trace coordinator process.
140+setTraceFlags :: TraceFlags -> Process ()
141+setTraceFlags f =
142+ withLocalTracerSync $ \t sp -> Tracer.setTraceFlagsSync t sp f
143+
144+-- | Send a log message to the internal tracing facility. If tracing is
145+-- enabled, this will create a custom trace log event.
146+--
147+traceLog :: String -> Process ()
148+traceLog s = withLocalTracer $ \t -> liftIO $ Tracer.traceLog t s
149+
150+-- | Send a log message to the internal tracing facility, using the given
151+-- list of printable 'TraceArg's interspersed with the preceding delimiter.
152+--
153+traceLogFmt :: String -> [TraceArg] -> Process ()
154+traceLogFmt d ls = withLocalTracer $ \t -> liftIO $ Tracer.traceLogFmt t d ls
155+
156+-- | Send an arbitrary 'Message' to the tracer process.
157+traceMessage :: Serializable m => m -> Process ()
158+traceMessage msg = withLocalTracer $ \t -> liftIO $ Tracer.traceMessage t msg
159+
160+withLocalTracer :: (Tracer -> Process ()) -> Process ()
161+withLocalTracer act = do
162+ node <- processNode <$> ask
163+ act (localTracer node)
164+
165+withLocalTracerSync :: (Tracer -> SendPort TraceOk -> IO ()) -> Process ()
166+withLocalTracerSync act = do
167+ (sp, rp) <- newChan
168+ withLocalTracer $ \t -> liftIO $ (act t sp)
169+ TraceOk <- receiveChan rp
170+ return ()
171+
172+withRegisteredTracer :: (ProcessId -> Process a) -> Process a
173+withRegisteredTracer act = do
174+ (sp, rp) <- newChan
175+ withLocalTracer $ \t -> liftIO $ Tracer.getCurrentTraceClient t sp
176+ currentTracer <- receiveChan rp
177+ case currentTracer of
178+ Nothing -> do { (Just p') <- whereis "tracer.initial"; act p' }
179+ (Just p) -> act p
180+
src/Control/Distributed/Process/Internal/Trace/Remote.hsnew file+62
@@ -0,0 +1,62 @@
1+module Control.Distributed.Process.Internal.Trace.Remote
2+ ( -- * Configuring A Tracer
3+ setTraceFlagsRemote
4+ , startTraceRelay
5+ -- * Remote Table
6+ , remoteTable
7+ ) where
8+
9+import Control.Distributed.Process.Internal.Closure.BuiltIn
10+ ( cpEnableTraceRemote
11+ )
12+import Control.Distributed.Process.Internal.Primitives
13+ ( getSelfPid
14+ , relay
15+ , nsendRemote
16+ )
17+import Control.Distributed.Process.Internal.Trace.Types
18+ ( TraceFlags(..)
19+ , TraceOk(..)
20+ )
21+import Control.Distributed.Process.Internal.Trace.Primitives
22+ ( withRegisteredTracer
23+ , enableTrace
24+ )
25+import Control.Distributed.Process.Internal.Spawn
26+ ( spawn
27+ )
28+import Control.Distributed.Process.Internal.Types
29+ ( Process
30+ , ProcessId
31+ , SendPort
32+ , NodeId
33+ )
34+import Control.Distributed.Static
35+ ( RemoteTable
36+ , registerStatic
37+ )
38+import Data.Rank1Dynamic (toDynamic)
39+
40+-- | Remote Table.
41+remoteTable :: RemoteTable -> RemoteTable
42+remoteTable = registerStatic "$enableTraceRemote" (toDynamic enableTraceRemote)
43+
44+enableTraceRemote :: ProcessId -> Process ()
45+enableTraceRemote pid =
46+ getSelfPid >>= enableTrace >> relay pid
47+
48+-- | Starts a /trace relay/ process on the remote node, which forwards all trace
49+-- events to the registered tracer on /this/ (the calling process') node.
50+startTraceRelay :: NodeId -> Process ProcessId
51+startTraceRelay nodeId = do
52+ withRegisteredTracer $ \pid ->
53+ -- TODO: move cpEnableTraceRemote here?
54+ spawn nodeId $ cpEnableTraceRemote pid
55+
56+-- | Set the given flags for a remote node (asynchronous).
57+setTraceFlagsRemote :: TraceFlags -> NodeId -> Process ()
58+setTraceFlagsRemote flags node = do
59+ nsendRemote node
60+ "trace.controller"
61+ ((Nothing :: Maybe (SendPort TraceOk)), flags)
62+
src/Control/Distributed/Process/Internal/Trace/Tracer.hsnew file+356
@@ -0,0 +1,356 @@
1+-- | Tracing/Debugging support - Trace Implementation
2+module Control.Distributed.Process.Internal.Trace.Tracer
3+ ( -- * API for the Node Controller
4+ traceController
5+ -- * Built in tracers
6+ , defaultTracer
7+ , systemLoggerTracer
8+ , logfileTracer
9+ , eventLogTracer
10+ ) where
11+
12+import Control.Applicative ((<$>))
13+import Control.Concurrent.Chan (writeChan)
14+import Control.Concurrent.MVar
15+ ( MVar
16+ , putMVar
17+ )
18+import Control.Distributed.Process.Internal.CQueue
19+ ( CQueue
20+ )
21+import Control.Distributed.Process.Internal.Primitives
22+ ( catch
23+ , finally
24+ , die
25+ , receiveWait
26+ , forward
27+ , sendChan
28+ , match
29+ , matchAny
30+ , matchIf
31+ , handleMessage
32+ , unwrapMessage
33+ , matchUnknown
34+ )
35+import Control.Distributed.Process.Internal.Trace.Types
36+ ( TraceEvent(..)
37+ , SetTrace(..)
38+ , Addressable(..)
39+ , TraceSubject(..)
40+ , TraceFlags(..)
41+ , TraceOk(..)
42+ , defaultTraceFlags
43+ )
44+import Control.Distributed.Process.Internal.Trace.Primitives
45+ ( traceOn )
46+import Control.Distributed.Process.Internal.Types
47+ ( LocalNode(..)
48+ , NCMsg(..)
49+ , ProcessId
50+ , Process
51+ , LocalProcess(..)
52+ , Identifier(..)
53+ , ProcessSignal(NamedSend)
54+ , Message
55+ , SendPort
56+ , forever'
57+ , nullProcessId
58+ , createUnencodedMessage
59+ )
60+
61+import Control.Monad.IO.Class (liftIO)
62+import Control.Monad.Reader (ask)
63+
64+import Data.Set (Set)
65+import qualified Data.Set as Set
66+import Data.Map (Map)
67+import qualified Data.Map as Map
68+
69+import Data.Maybe (fromMaybe)
70+import Data.Time.Clock (getCurrentTime)
71+import Data.Time.Format (formatTime)
72+import Debug.Trace (traceEventIO)
73+
74+import Prelude hiding (catch)
75+
76+import System.Environment (getEnv)
77+import System.IO
78+ ( Handle
79+ , IOMode(AppendMode)
80+ , BufferMode(..)
81+ , openFile
82+ , hClose
83+ , hPutStrLn
84+ , hSetBuffering
85+ )
86+import System.Locale (defaultTimeLocale)
87+import System.Mem.Weak
88+ ( Weak
89+ )
90+
91+data TracerState = TracerST {
92+ client :: !(Maybe ProcessId)
93+ , flags :: !TraceFlags
94+ , regNames :: !(Map ProcessId (Set String))
95+ }
96+
97+--------------------------------------------------------------------------------
98+-- Trace Handlers --
99+--------------------------------------------------------------------------------
100+
101+defaultTracer :: Process ()
102+defaultTracer =
103+ catch (checkEnv "DISTRIBUTED_PROCESS_TRACE_FILE" >>= logfileTracer)
104+ (\(_ :: IOError) -> defaultTracerAux)
105+
106+defaultTracerAux :: Process ()
107+defaultTracerAux =
108+ catch (checkEnv "DISTRIBUTED_PROCESS_TRACE_CONSOLE" >> systemLoggerTracer)
109+ (\(_ :: IOError) -> defaultEventLogTracer)
110+
111+-- TODO: it would be /nice/ if we had some way of checking the runtime
112+-- options to see if +RTS -v (or similar) has been given...
113+defaultEventLogTracer :: Process ()
114+defaultEventLogTracer =
115+ catch (checkEnv "DISTRIBUTED_PROCESS_TRACE_EVENTLOG" >> eventLogTracer)
116+ (\(_ :: IOError) -> nullTracer)
117+
118+checkEnv :: String -> Process String
119+checkEnv s = liftIO $ getEnv s
120+
121+-- This trace client is (intentionally) useful - it simply provides
122+-- an intial client for the trace controller to talk to, until some
123+-- other (hopefully more useful) client is installed over the top of
124+-- it. This is the default trace client.
125+nullTracer :: Process ()
126+nullTracer =
127+ forever' $ receiveWait [ matchUnknown (return ()) ]
128+
129+systemLoggerTracer :: Process ()
130+systemLoggerTracer = do
131+ node <- processNode <$> ask
132+ let tr = sendTraceMsg node
133+ forever' $ receiveWait [ matchAny (\m -> handleMessage m tr) ]
134+ where
135+ sendTraceMsg :: LocalNode -> TraceEvent -> Process ()
136+ sendTraceMsg node ev = do
137+ now <- liftIO $ getCurrentTime
138+ txt <- buildTxt ev
139+ msg <- return $ (formatTime defaultTimeLocale "%c" now, txt)
140+ emptyPid <- return $ (nullProcessId (localNodeId node))
141+ traceMsg <- return $ NCMsg {
142+ ctrlMsgSender = ProcessIdentifier (emptyPid)
143+ , ctrlMsgSignal = (NamedSend "logger"
144+ (createUnencodedMessage msg))
145+ }
146+ liftIO $ writeChan (localCtrlChan node) traceMsg
147+
148+ buildTxt :: TraceEvent -> Process String
149+ buildTxt (TraceEvLog msg) = return msg
150+ buildTxt ev = return $ show ev
151+
152+eventLogTracer :: Process ()
153+eventLogTracer =
154+ -- NB: when the GHC event log supports tracing arbitrary (ish) data, we will
155+ -- almost certainly use *that* facility independently of whether or not there
156+ -- is a tracer process installed. This is just a stop gap until then.
157+ forever' $ receiveWait [ matchAny (\m -> handleMessage m writeTrace) ]
158+ where
159+ writeTrace :: TraceEvent -> Process ()
160+ writeTrace ev = liftIO $ traceEventIO (show ev)
161+
162+logfileTracer :: FilePath -> Process ()
163+logfileTracer p = do
164+ -- TODO: error handling if the handle cannot be opened
165+ liftIO $ putStrLn $ "writing to " ++ (show p)
166+ h <- liftIO $ openFile p AppendMode
167+ liftIO $ hSetBuffering h LineBuffering
168+ logger h `finally` (liftIO $ hClose h)
169+ where
170+ logger :: Handle -> Process ()
171+ logger h' = forever' $ do
172+ receiveWait [
173+ matchIf (\ev -> case ev of
174+ TraceEvDisable -> True
175+ (TraceEvTakeover _) -> True
176+ _ -> False)
177+ (\_ -> (liftIO $ hClose h') >> die "trace stopped")
178+ , matchAny (\ev -> handleMessage ev (writeTrace h'))
179+ ]
180+
181+ writeTrace :: Handle -> TraceEvent -> Process ()
182+ writeTrace h ev = do
183+ liftIO $ do
184+ now <- getCurrentTime
185+ hPutStrLn h $ (formatTime defaultTimeLocale "%c - " now) ++ (show ev)
186+
187+--------------------------------------------------------------------------------
188+-- Tracer Implementation --
189+--------------------------------------------------------------------------------
190+
191+traceController :: MVar ((Weak (CQueue Message))) -> Process ()
192+traceController mv = do
193+ -- This breach of encapsulation is deliberate: we don't want to tie up
194+ -- the node's internal control channel if we can possibly help it!
195+ weakQueue <- processWeakQ <$> ask
196+ liftIO $ putMVar mv weakQueue
197+ initState <- initialState
198+ traceLoop initState { client = Nothing }
199+ where
200+ traceLoop :: TracerState -> Process ()
201+ traceLoop st = do
202+ -- Trace events are forwarded to the trace target when tracing is enabled.
203+ -- At some point in the future, we're going to start writing these custom
204+ -- events to the ghc eventlog, at which point this design might change.
205+ st' <- receiveWait [
206+ match (\(setResp, set :: SetTrace) -> do
207+ -- We allow at most one trace client, which is a process id.
208+ -- Tracking multiple clients represents too high an overhead,
209+ -- so we leave that kind of thing to our consumers to figure
210+ -- figure out.
211+ case set of
212+ (TraceEnable pid) -> do
213+ -- notify the previous tracer it has been replaced
214+ sendTrace st (createUnencodedMessage (TraceEvTakeover pid))
215+ sendOk setResp
216+ return st { client = (Just pid) }
217+ TraceDisable -> do
218+ sendTrace st (createUnencodedMessage TraceEvDisable)
219+ sendOk setResp
220+ return st { client = Nothing })
221+ , match (\(confResp, flags') ->
222+ sendOk confResp >> applyTraceFlags flags' st)
223+ , match (\chGetFlags -> sendChan chGetFlags (flags st) >> return st)
224+ , match (\chGetCurrent -> sendChan chGetCurrent (client st) >> return st)
225+ -- we dequeue incoming events even if we don't process them
226+ , matchAny (\ev ->
227+ handleMessage ev (handleTrace st ev) >>= return . fromMaybe st)
228+ ]
229+ traceLoop st'
230+
231+ sendOk :: Maybe (SendPort TraceOk) -> Process ()
232+ sendOk Nothing = return ()
233+ sendOk (Just sp) = sendChan sp TraceOk
234+
235+ initialState :: Process TracerState
236+ initialState = do
237+ flags' <- checkEnvFlags
238+ return $ TracerST { client = Nothing
239+ , flags = flags'
240+ , regNames = Map.empty
241+ }
242+
243+ checkEnvFlags :: Process TraceFlags
244+ checkEnvFlags =
245+ catch (checkEnv "DISTRIBUTED_PROCESS_TRACE_FLAGS" >>= return . parseFlags)
246+ (\(_ :: IOError) -> return defaultTraceFlags)
247+
248+ -- quick and dirty fold-while
249+ parseFlags :: String -> TraceFlags
250+ parseFlags s = parseFlags' s defaultTraceFlags
251+ where parseFlags' :: String -> TraceFlags -> TraceFlags
252+ parseFlags' [] parsedFlags = parsedFlags
253+ parseFlags' (x:xs) parsedFlags
254+ | x == 'p' = parseFlags' xs parsedFlags { traceSpawned = traceOn }
255+ | x == 'n' = parseFlags' xs parsedFlags { traceRegistered = traceOn }
256+ | x == 'u' = parseFlags' xs parsedFlags { traceUnregistered = traceOn }
257+ | x == 'd' = parseFlags' xs parsedFlags { traceDied = traceOn }
258+ | x == 's' = parseFlags' xs parsedFlags { traceSend = traceOn }
259+ | x == 'r' = parseFlags' xs parsedFlags { traceRecv = traceOn }
260+ | x == 'l' = parseFlags' xs parsedFlags { traceNodes = True }
261+ | otherwise = parseFlags' xs parsedFlags
262+
263+-- note [runtime tracer control]
264+--
265+-- The trace mechanism is designed to put as little stress on the system, and
266+-- in particular the node controller, as possible. The LocalNode's @tracer@
267+-- field is therefore immutable, since we don't want to be blocking the node
268+-- controller when writing trace information out. The runtime cost of enabling
269+-- tracing is therefore the cost of enqueue for all trace-able operations at
270+-- all times. If tracing is not explicitly enabled, the trace record stored in
271+-- the local node state is matched and ignored in API calls, reducing the cost
272+-- to a single function call.
273+--
274+-- To /disable/ tracing facilities in a runtime system that has tracing enabled
275+-- then, is to instruct the tracer process not to forward any trace event data
276+-- and it is not possible to remove the runtime overhead (of enqueue in the
277+-- caller and dequeue + noop in the tracer process) once the node is started.
278+
279+applyTraceFlags :: TraceFlags -> TracerState -> Process TracerState
280+applyTraceFlags flags' state = return state { flags = flags' }
281+
282+handleTrace :: TracerState -> Message -> TraceEvent -> Process TracerState
283+handleTrace st msg ev@(TraceEvRegistered p n) =
284+ let regNames' =
285+ Map.insertWith (\_ ns -> Set.insert n ns) p
286+ (Set.singleton n)
287+ (regNames st)
288+ in do
289+ traceEv ev msg (traceRegistered (flags st)) st
290+ return st { regNames = regNames' }
291+handleTrace st msg ev@(TraceEvUnRegistered p n) =
292+ let f ns = case ns of
293+ Nothing -> Nothing
294+ Just ns' -> Just (Set.delete n ns')
295+ regNames' = Map.alter f p (regNames st)
296+ in do
297+ traceEv ev msg (traceUnregistered (flags st)) st
298+ return st { regNames = regNames' }
299+handleTrace st msg ev@(TraceEvSpawned _) = do
300+ traceEv ev msg (traceSpawned (flags st)) st >> return st
301+handleTrace st msg ev@(TraceEvDied _ _) = do
302+ traceEv ev msg (traceDied (flags st)) st >> return st
303+handleTrace st msg ev@(TraceEvSent _ _ _) =
304+ traceEv ev msg (traceSend (flags st)) st >> return st
305+handleTrace st msg ev@(TraceEvReceived _ _) =
306+ traceEv ev msg (traceRecv (flags st)) st >> return st
307+handleTrace st msg ev = do
308+ case ev of
309+ (TraceEvNodeDied _ _) ->
310+ case (traceNodes (flags st)) of
311+ True -> sendTrace st msg
312+ False -> return ()
313+ (TraceEvUser _) -> sendTrace st msg
314+ (TraceEvLog _) -> sendTrace st msg
315+ _ ->
316+ case (traceConnections (flags st)) of
317+ True -> sendTrace st msg
318+ False -> return ()
319+ return st
320+
321+traceEv :: TraceEvent
322+ -> Message
323+ -> Maybe TraceSubject
324+ -> TracerState
325+ -> Process ()
326+traceEv _ _ Nothing _ = return ()
327+traceEv _ msg (Just TraceAll) st = sendTrace st msg
328+traceEv ev msg (Just (TraceProcs pids)) st = do
329+ node <- processNode <$> ask
330+ let p = case resolveToPid ev of
331+ Nothing -> (nullProcessId (localNodeId node))
332+ Just pid -> pid
333+ case (Set.member p pids) of
334+ True -> sendTrace st msg
335+ False -> return ()
336+traceEv ev msg (Just (TraceNames names)) st = do
337+ -- TODO: if we have recorded regnames for p, then we
338+ -- forward the trace iif there are overlapping trace targets
339+ node <- processNode <$> ask
340+ let p = case resolveToPid ev of
341+ Nothing -> (nullProcessId (localNodeId node))
342+ Just pid -> pid
343+ case (Map.lookup p (regNames st)) of
344+ Nothing -> return ()
345+ Just ns -> if (Set.null (Set.intersection ns names))
346+ then return ()
347+ else sendTrace st msg
348+
349+sendTrace :: TracerState -> Message -> Process ()
350+sendTrace st msg =
351+ let pid = (client st) in do
352+ case pid of
353+ Just p -> (flip forward) p msg
354+ Nothing -> return ()
355+
356+
src/Control/Distributed/Process/Internal/Trace/Types.hsnew file+284
@@ -0,0 +1,284 @@
1+-- | Tracing/Debugging support - Types
2+module Control.Distributed.Process.Internal.Trace.Types
3+ ( Tracer(..)
4+ , TraceEvent(..)
5+ , SetTrace(..)
6+ , Addressable(..)
7+ , TraceSubject(..)
8+ , TraceFlags(..)
9+ , TraceArg(..)
10+ , TraceOk(..)
11+ , traceLog
12+ , traceLogFmt
13+ , traceEvent
14+ , traceMessage
15+ , defaultTraceFlags
16+ , enableTrace
17+ , enableTraceSync
18+ , disableTrace
19+ , disableTraceSync
20+ , getTraceFlags
21+ , setTraceFlags
22+ , setTraceFlagsSync
23+ , getCurrentTraceClient
24+ ) where
25+
26+import Control.Applicative ((<$>), (<*>))
27+import Control.Distributed.Process.Internal.CQueue (enqueue)
28+import Control.Distributed.Process.Internal.Types
29+ ( Tracer(..)
30+ , NodeId
31+ , ProcessId
32+ , DiedReason
33+ , Message
34+ , SendPort
35+ , createUnencodedMessage
36+ )
37+import Control.Distributed.Process.Serializable
38+import Data.Binary
39+import Data.Foldable (forM_)
40+import Data.List (intersperse)
41+import Data.Set (Set)
42+import qualified Data.Set as Set (fromList)
43+import Data.Typeable
44+import System.Mem.Weak (deRefWeak)
45+import Network.Transport
46+ ( ConnectionId
47+ )
48+
49+--------------------------------------------------------------------------------
50+-- Types --
51+--------------------------------------------------------------------------------
52+
53+data SetTrace = TraceEnable !ProcessId | TraceDisable
54+ deriving (Typeable, Eq, Show)
55+
56+-- | An event that is fired when something /interesting/ happens.
57+-- Trace events are forwarded to a trace listener process, of which
58+-- at most one can be registered per-node.
59+data TraceEvent =
60+ TraceEvSpawned ProcessId
61+ -- ^ fired whenever a local process is spawned
62+ | TraceEvRegistered ProcessId String
63+ -- ^ fired whenever a process/name is registered (locally)
64+ | TraceEvUnRegistered ProcessId String
65+ -- ^ fired whenever a process/name is unregistered (locally)
66+ | TraceEvDied ProcessId DiedReason
67+ -- ^ fired whenever a process dies
68+ | TraceEvNodeDied NodeId DiedReason
69+ -- ^ fired whenever a node /dies/ (i.e., the connection is broken/disconnected)
70+ | TraceEvSent ProcessId ProcessId Message
71+ -- ^ fired whenever a message is sent from a local process
72+ | TraceEvReceived ProcessId Message
73+ -- ^ fired whenever a message is received by a local process
74+ | TraceEvConnected ConnectionId
75+ -- ^ fired when a network-transport connection is first established
76+ | TraceEvDisconnected ConnectionId
77+ -- ^ fired when a network-transport connection is broken/disconnected
78+ | TraceEvUser Message
79+ -- ^ a user defined trace event (see 'traceMessage')
80+ | TraceEvLog String
81+ -- ^ a /logging/ event - used for debugging purposes only
82+ | TraceEvTakeover ProcessId
83+ -- ^ notifies a trace listener that all subsequent traces will be sent to /pid/
84+ | TraceEvDisable
85+ -- ^ notifies a trace listener that it has been disabled/removed
86+ deriving (Typeable, Show)
87+
88+class Addressable a where
89+ resolveToPid :: a -> Maybe ProcessId
90+
91+instance Addressable TraceEvent where
92+ resolveToPid (TraceEvSpawned p) = Just p
93+ resolveToPid (TraceEvDied p _) = Just p
94+ resolveToPid (TraceEvSent _ p _) = Just p
95+ resolveToPid (TraceEvReceived p _) = Just p
96+ resolveToPid _ = Nothing
97+
98+-- | Defines which processes will be traced by a given 'TraceFlag',
99+-- either by name, or @ProcessId@. Choosing @TraceAll@ is /by far/
100+-- the most efficient approach, as the tracer process therefore
101+-- avoids deciding whether or not a trace event is viable.
102+--
103+data TraceSubject =
104+ TraceAll -- enable tracing for all running processes
105+ | TraceProcs !(Set ProcessId) -- enable tracing for a set of processes
106+ | TraceNames !(Set String) -- enable tracing for a set of named/registered processes
107+ deriving (Show)
108+
109+-- | Defines /what/ will be traced. Flags that control tracing of
110+-- @Process@ events, take a 'TraceSubject' controlling which processes
111+-- should generate trace events in the target process.
112+data TraceFlags = TraceFlags {
113+ traceSpawned :: !(Maybe TraceSubject) -- filter process spawned tracing
114+ , traceDied :: !(Maybe TraceSubject) -- filter process died tracing
115+ , traceRegistered :: !(Maybe TraceSubject) -- filter process registration tracing
116+ , traceUnregistered :: !(Maybe TraceSubject) -- filter process un-registration
117+ , traceSend :: !(Maybe TraceSubject) -- filter process/message tracing by sender
118+ , traceRecv :: !(Maybe TraceSubject) -- filter process/message tracing by receiver
119+ , traceNodes :: !Bool -- enable node status trace events
120+ , traceConnections :: !Bool -- enable connection status trace events
121+ } deriving (Typeable, Show)
122+
123+defaultTraceFlags :: TraceFlags
124+defaultTraceFlags =
125+ TraceFlags {
126+ traceSpawned = Nothing
127+ , traceDied = Nothing
128+ , traceRegistered = Nothing
129+ , traceUnregistered = Nothing
130+ , traceSend = Nothing
131+ , traceRecv = Nothing
132+ , traceNodes = False
133+ , traceConnections = False
134+ }
135+
136+data TraceArg =
137+ TraceStr String
138+ | forall a. (Show a) => Trace a
139+
140+-- | A generic 'ok' response from the trace coordinator.
141+data TraceOk = TraceOk
142+ deriving (Typeable)
143+
144+--------------------------------------------------------------------------------
145+-- Internal/Common API --
146+--------------------------------------------------------------------------------
147+
148+traceLog :: Tracer -> String -> IO ()
149+traceLog tr s = traceIt tr (createUnencodedMessage $ TraceEvLog s)
150+
151+traceLogFmt :: Tracer
152+ -> String
153+ -> [TraceArg]
154+ -> IO ()
155+traceLogFmt t d ls =
156+ traceLog t $ concat (intersperse d (map toS ls))
157+ where toS :: TraceArg -> String
158+ toS (TraceStr s) = s
159+ toS (Trace a) = show a
160+
161+traceEvent :: Tracer -> TraceEvent -> IO ()
162+traceEvent tr ev = traceIt tr (createUnencodedMessage ev)
163+
164+traceMessage :: Serializable m => Tracer -> m -> IO ()
165+traceMessage tr msg = traceEvent tr (TraceEvUser (createUnencodedMessage msg))
166+
167+enableTrace :: Tracer -> ProcessId -> IO ()
168+enableTrace t p =
169+ traceIt t (createUnencodedMessage ((Nothing :: Maybe (SendPort TraceOk)),
170+ (TraceEnable p)))
171+
172+enableTraceSync :: Tracer -> SendPort TraceOk -> ProcessId -> IO ()
173+enableTraceSync t s p =
174+ traceIt t (createUnencodedMessage ((Just s), (TraceEnable p)))
175+
176+disableTrace :: Tracer -> IO ()
177+disableTrace t =
178+ traceIt t (createUnencodedMessage ((Nothing :: Maybe (SendPort TraceOk)),
179+ TraceDisable))
180+
181+disableTraceSync :: Tracer -> SendPort TraceOk -> IO ()
182+disableTraceSync t s =
183+ traceIt t (createUnencodedMessage ((Just s), TraceDisable))
184+
185+setTraceFlags :: Tracer -> TraceFlags -> IO ()
186+setTraceFlags t f =
187+ traceIt t (createUnencodedMessage ((Nothing :: Maybe (SendPort TraceOk)), f))
188+
189+setTraceFlagsSync :: Tracer -> SendPort TraceOk -> TraceFlags -> IO ()
190+setTraceFlagsSync t s f =
191+ traceIt t (createUnencodedMessage ((Just s), f))
192+
193+getTraceFlags :: Tracer -> SendPort TraceFlags -> IO ()
194+getTraceFlags t s = traceIt t (createUnencodedMessage s)
195+
196+getCurrentTraceClient :: Tracer -> SendPort (Maybe ProcessId) -> IO ()
197+getCurrentTraceClient t s = traceIt t (createUnencodedMessage s)
198+
199+class Traceable a where
200+ uod :: [a] -> TraceSubject
201+
202+instance Traceable ProcessId where
203+ uod = TraceProcs . Set.fromList
204+
205+instance Traceable String where
206+ uod = TraceNames . Set.fromList
207+
208+traceIt :: Tracer -> Message -> IO ()
209+traceIt InactiveTracer _ = return ()
210+traceIt (ActiveTracer _ wqRef) msg = do
211+ mQueue <- deRefWeak wqRef
212+ forM_ mQueue $ \queue -> enqueue queue msg
213+
214+--------------------------------------------------------------------------------
215+-- Binary Instances --
216+--------------------------------------------------------------------------------
217+
218+instance Binary SetTrace where
219+ put (TraceEnable pid) = putWord8 1 >> put pid
220+ put TraceDisable = putWord8 2
221+
222+ get = do
223+ header <- getWord8
224+ case header of
225+ 1 -> TraceEnable <$> get
226+ 2 -> return TraceDisable
227+ _ -> error "SetTrace.get - invalid header"
228+
229+instance Binary TraceEvent where
230+ put (TraceEvSpawned pid) = putWord8 1 >> put pid
231+ put (TraceEvRegistered pid str) = putWord8 2 >> put pid >> put str
232+ put (TraceEvUnRegistered pid str) = putWord8 3 >> put pid >> put str
233+ put (TraceEvDied pid res) = putWord8 4 >> put pid >> put res
234+ put (TraceEvNodeDied nid res) = putWord8 5 >> put nid >> put res
235+ put (TraceEvSent to from msg) = putWord8 6 >> put to >> put from >> put msg
236+ put (TraceEvReceived pid msg) = putWord8 7 >> put pid >> put msg
237+ put (TraceEvConnected cid) = putWord8 8 >> put cid
238+ put (TraceEvDisconnected cid) = putWord8 9 >> put cid
239+ put (TraceEvUser msg) = putWord8 10 >> put msg
240+ put (TraceEvLog msg) = putWord8 11 >> put msg
241+ put (TraceEvTakeover pid) = putWord8 12 >> put pid
242+ put TraceEvDisable = putWord8 13
243+
244+ get = do
245+ header <- getWord8
246+ case header of
247+ 1 -> TraceEvSpawned <$> get
248+ 2 -> TraceEvRegistered <$> get <*> get
249+ 3 -> TraceEvUnRegistered <$> get <*> get
250+ 4 -> TraceEvDied <$> get <*> get
251+ 5 -> TraceEvNodeDied <$> get <*> get
252+ 6 -> TraceEvSent <$> get <*> get <*> get
253+ 7 -> TraceEvReceived <$> get <*> get
254+ 8 -> TraceEvConnected <$> get
255+ 9 -> TraceEvDisconnected <$> get
256+ 10 -> TraceEvUser <$> get
257+ 11 -> TraceEvLog <$> get
258+ 12 -> TraceEvTakeover <$> get
259+ 13 -> return TraceEvDisable
260+ _ -> error "TraceEvent.get - invalid header"
261+
262+instance Binary TraceSubject where
263+ put TraceAll = putWord8 1
264+ put (TraceProcs pids) = putWord8 2 >> put pids
265+ put (TraceNames names) = putWord8 3 >> put names
266+
267+ get = do
268+ header <- getWord8
269+ case header of
270+ 1 -> return TraceAll
271+ 2 -> TraceProcs <$> get
272+ 3 -> TraceNames <$> get
273+ _ -> error "TraceSubject.get - invalid header"
274+
275+instance Binary TraceFlags where
276+ put (TraceFlags s d g u m r n c) =
277+ put s >> put d >> put g >> put u >> put m >> put r >> put n >> put c
278+ get =
279+ TraceFlags <$> get <*> get <*> get <*> get <*> get <*> get <*> get <*> get
280+
281+instance Binary TraceOk where
282+ put _ = return ()
283+ get = return TraceOk
284+
src/Control/Distributed/Process/Internal/Types.hs+15−7
@@ -93,7 +93,6 @@ import Control.Exception (Exception)
93import Control.Concurrent (ThreadId)
93import Control.Concurrent (ThreadId)
94import Control.Concurrent.Chan (Chan)
94import Control.Concurrent.Chan (Chan)
95import Control.Concurrent.STM (STM)
95import Control.Concurrent.STM (STM)
96−import qualified Control.Concurrent.STM as STM (TQueue)
97import qualified Network.Transport as NT (EndPoint, EndPointAddress, Connection)
96import qualified Network.Transport as NT (EndPoint, EndPointAddress, Connection)
98import Control.Applicative (Applicative, Alternative, (<$>), (<*>))
97import Control.Applicative (Applicative, Alternative, (<$>), (<*>))
99import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)
98import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)
@@ -112,7 +111,6 @@ import Control.Distributed.Process.Internal.StrictMVar (StrictMVar)
112import Control.Distributed.Process.Internal.WeakTQueue (TQueue)
111import Control.Distributed.Process.Internal.WeakTQueue (TQueue)
113import Control.Distributed.Static (RemoteTable, Closure)
112import Control.Distributed.Static (RemoteTable, Closure)
114import qualified Control.Distributed.Process.Internal.StrictContainerAccessors as DAC (mapMaybe)
113import qualified Control.Distributed.Process.Internal.StrictContainerAccessors as DAC (mapMaybe)
115−import System.IO (Handle)
116
114
117--------------------------------------------------------------------------------
115--------------------------------------------------------------------------------
118-- Node and process identifiers --
116-- Node and process identifiers --
@@ -184,10 +182,20 @@ nullProcessId nid =
184
182
185-- | Required for system tracing in the node controller
183-- | Required for system tracing in the node controller
186data Tracer =
184data Tracer =
187− LogFileTracer !ThreadId !(STM.TQueue String) !Handle
185+ ActiveTracer !ProcessId !(Weak (CQueue Message)) -- note [tracer longevity]
188− | EventLogTracer !(String -> IO ())
186+ | InactiveTracer
189− | LocalNodeTracer !LocalNode
187+
190− | InactiveTracer -- NB: never used, this is required to initialize LocalNode
188+-- note [tracer longevity]
189+--
190+-- If the trace controller process crashes, we make no attempt to restart it
191+-- or handle the failure. Tracing is done on a 'best effort' basis, because
192+-- to make stronger guarantees (e.g., handling failures) would introduce
193+-- significant complexity and adversely affect performance. As such, we access
194+-- the process' mailbox directly via a weak reference, and calls to a dead
195+-- process have no effect just as with normal process/send interactions.
196+--
197+-- Also note that the ProcessId stored in an active tracer is that of the first
198+-- (initially registered) system tracer process.
191
199
192-- | Local nodes
200-- | Local nodes
193data LocalNode = LocalNode
201data LocalNode = LocalNode
@@ -553,7 +561,7 @@ instance Binary ProcessSignal where
553 put (NamedSend label msg) = putWord8 8 >> put label >> put (messageToPayload msg)
561 put (NamedSend label msg) = putWord8 8 >> put label >> put (messageToPayload msg)
554 put (Kill pid reason) = putWord8 9 >> put pid >> put reason
562 put (Kill pid reason) = putWord8 9 >> put pid >> put reason
555 put (Exit pid reason) = putWord8 10 >> put pid >> put (messageToPayload reason)
563 put (Exit pid reason) = putWord8 10 >> put pid >> put (messageToPayload reason)
556− put (LocalSend pid msg) = putWord8 11 >> put pid >> put (messageToPayload msg)
564+ put (LocalSend to msg) = putWord8 11 >> put to >> put (messageToPayload msg)
557 put (LocalPortSend sid msg) = putWord8 12 >> put sid >> put (messageToPayload msg)
565 put (LocalPortSend sid msg) = putWord8 12 >> put sid >> put (messageToPayload msg)
558 put (GetInfo about) = putWord8 30 >> put about
566 put (GetInfo about) = putWord8 30 >> put about
559 put (SigShutdown) = putWord8 31
567 put (SigShutdown) = putWord8 31
src/Control/Distributed/Process/Node.hs+124−51
@@ -40,7 +40,7 @@ import qualified Data.Set as Set
40 , toList
40 , toList
41 )
41 )
42import Data.Foldable (forM_)
42import Data.Foldable (forM_)
43−import Data.Maybe (isJust, isNothing, catMaybes)
43+import Data.Maybe (isJust, fromJust, isNothing, catMaybes)
44import Data.Typeable (Typeable)
44import Data.Typeable (Typeable)
45import Control.Category ((>>>))
45import Control.Category ((>>>))
46import Control.Applicative ((<$>))
46import Control.Applicative ((<$>))
@@ -68,6 +68,7 @@ import Control.Distributed.Process.Internal.StrictMVar
68 , takeMVar
68 , takeMVar
69 )
69 )
70import Control.Concurrent.Chan (newChan, writeChan, readChan)
70import Control.Concurrent.Chan (newChan, writeChan, readChan)
71+import qualified Control.Concurrent.MVar as MVar (newEmptyMVar, takeMVar)
71import Control.Concurrent.STM
72import Control.Concurrent.STM
72 ( atomically
73 ( atomically
73 )
74 )
@@ -105,7 +106,7 @@ import Control.Distributed.Process.Internal.Types
105 , LocalProcessId(..)
106 , LocalProcessId(..)
106 , ProcessId(..)
107 , ProcessId(..)
107 , LocalNode(..)
108 , LocalNode(..)
108− , Tracer(InactiveTracer)
109+ , Tracer(..)
109 , LocalNodeState(..)
110 , LocalNodeState(..)
110 , LocalProcess(..)
111 , LocalProcess(..)
111 , LocalProcessState(..)
112 , LocalProcessState(..)
@@ -148,11 +149,19 @@ import Control.Distributed.Process.Internal.Types
148 , firstNonReservedProcessId
149 , firstNonReservedProcessId
149 , ImplicitReconnect(WithImplicitReconnect)
150 , ImplicitReconnect(WithImplicitReconnect)
150 )
151 )
152+import qualified Control.Distributed.Process.Internal.Trace.Remote as Trace
153+ ( remoteTable
154+ )
151−import Control.Distributed.Process.Internal.Trace
155+import Control.Distributed.Process.Internal.Trace.Types
152 ( TraceArg(..)
156 ( TraceArg(..)
153− , traceFormat
157+ , TraceEvent(..)
154− , startTracing
158+ , traceEvent
155− , stopTracer
159+ , traceLogFmt
160+ , enableTrace
161+ )
162+import Control.Distributed.Process.Internal.Trace.Tracer
163+ ( traceController
164+ , defaultTracer
156 )
165 )
157import Control.Distributed.Process.Serializable (Serializable)
166import Control.Distributed.Process.Serializable (Serializable)
158import Control.Distributed.Process.Internal.Messaging
167import Control.Distributed.Process.Internal.Messaging
@@ -175,7 +184,7 @@ import qualified Control.Distributed.Process.Internal.StrictContainerAccessors a
175 ( mapMaybe
184 ( mapMaybe
176 , mapDefault
185 , mapDefault
177 )
186 )
178−
187+import System.Environment (getEnv)
179import Unsafe.Coerce
188import Unsafe.Coerce
180
189
181--------------------------------------------------------------------------------
190--------------------------------------------------------------------------------
@@ -183,7 +192,7 @@ import Unsafe.Coerce
183--------------------------------------------------------------------------------
192--------------------------------------------------------------------------------
184
193
185initRemoteTable :: RemoteTable
194initRemoteTable :: RemoteTable
186−initRemoteTable = BuiltIn.remoteTable Static.initRemoteTable
195+initRemoteTable = Trace.remoteTable $ BuiltIn.remoteTable Static.initRemoteTable
187
196
188-- | Initialize a new local node.
197-- | Initialize a new local node.
189newLocalNode :: NT.Transport -> RemoteTable -> IO LocalNode
198newLocalNode :: NT.Transport -> RemoteTable -> IO LocalNode
@@ -232,10 +241,51 @@ createBareLocalNode endPoint rtable = do
232 , ctrlMsgSignal = SigShutdown
241 , ctrlMsgSignal = SigShutdown
233 }
242 }
234
243
244+startTracing :: LocalNode -> IO LocalNode
245+startTracing node =
246+ -- TODO: use tryGetEnv once we drop support for GHC 7.2.x
247+ Exception.catch
248+ (getEnv "DISTRIBUTED_PROCESS_TRACE_ENABLED" >> procTracer node)
249+ (\(_ :: IOError) -> return node)
250+ where procTracer :: LocalNode -> IO LocalNode
251+ procTracer n = do
252+ -- see note [tracer/forkProcess races]
253+ mv <- MVar.newEmptyMVar
254+ pid <- forkProcess n (traceController mv)
255+ wqRef <- MVar.takeMVar mv
256+ return n { localTracer = (ActiveTracer pid wqRef) }
257+
258+startDefaultTracer :: LocalNode -> IO (Maybe ProcessId)
259+startDefaultTracer node =
260+ Exception.catch
261+ (getEnv "DISTRIBUTED_PROCESS_TRACE_ENABLED" >> go node)
262+ (\(_ :: IOError) -> return Nothing)
263+ where go :: LocalNode -> IO (Maybe ProcessId)
264+ go node' = do
265+ registerTraceController node'
266+ pid <- forkProcess node' defaultTracer
267+ enableTrace (localTracer node') pid
268+ return (Just pid)
269+
270+ registerTraceController :: LocalNode -> IO ()
271+ registerTraceController n =
272+ let t = localTracer n in
273+ case t of
274+ InactiveTracer -> return ()
275+ (ActiveTracer pid _) -> runProcess n $ register "trace.controller" pid
276+
235-- | Start and register the service processes on a node
277-- | Start and register the service processes on a node
236−-- (for now, this is only the logger)
237startServiceProcesses :: LocalNode -> IO ()
278startServiceProcesses :: LocalNode -> IO ()
238startServiceProcesses node = do
279startServiceProcesses node = do
280+ -- tracing /spawns/ relies on the tracer being enabled, but we start
281+ -- the default tracer first, even though it might @nsend@ to the logger
282+ -- before /that/ process has started - this is a totally harmless race
283+ -- however, so we deliberably ignore it
284+ mPid <- startDefaultTracer node
285+ case mPid of
286+ Nothing -> return ()
287+ Just pid -> runProcess node $ register "tracer.initial" pid
288+
239 logger <- forkProcess node loop
289 logger <- forkProcess node loop
240 runProcess node $ register "logger" logger
290 runProcess node $ register "logger" logger
241 where
291 where
@@ -257,6 +307,7 @@ startServiceProcesses node = do
257-- TODO: for now we just close the associated endpoint
307-- TODO: for now we just close the associated endpoint
258closeLocalNode :: LocalNode -> IO ()
308closeLocalNode :: LocalNode -> IO ()
259closeLocalNode node =
309closeLocalNode node =
310+ -- TODO: close all our processes, surely!?
260 NT.closeEndPoint (localEndPoint node)
311 NT.closeEndPoint (localEndPoint node)
261
312
262-- | Run a process on a local node and wait for it to finish
313-- | Run a process on a local node and wait for it to finish
@@ -306,6 +357,10 @@ forkProcess node proc = modifyMVar (localState node) startProcess
306 }
357 }
307 return (tid', lproc)
358 return (tid', lproc)
308
359
360+
361+ -- see note [tracer/forkProcess races]
362+ trace node (TraceEvSpawned pid)
363+
309 if lpidCounter lpid == maxBound
364 if lpidCounter lpid == maxBound
310 then do
365 then do
311 newUnique <- randomIO
366 newUnique <- randomIO
@@ -331,6 +386,15 @@ forkProcess node proc = modifyMVar (localState node) startProcess
331 . (localConnections ^= unaffected)
386 . (localConnections ^= unaffected)
332 $ st
387 $ st
333
388
389+-- note [tracer/forkProcess races]
390+--
391+-- Our startTracing function uses forkProcess to start the trace controller
392+-- process, and of course forkProcess attempts to call traceEvent once the
393+-- process has started. This is harmless, as the localTracer is not updated
394+-- until /after/ the initial forkProcess completes, so the first call to
395+-- traceEvent behaves as if tracing were disabled (i.e., it is ignored).
396+--
397+
334--------------------------------------------------------------------------------
398--------------------------------------------------------------------------------
335-- Handle incoming messages --
399-- Handle incoming messages --
336--------------------------------------------------------------------------------
400--------------------------------------------------------------------------------
@@ -339,7 +403,7 @@ type IncomingConnection = (NT.EndPointAddress, IncomingTarget)
339
403
340data IncomingTarget =
404data IncomingTarget =
341 Uninit
405 Uninit
342− | ToProc (Weak (CQueue Message))
406+ | ToProc ProcessId (Weak (CQueue Message))
343 | ToChan TypedChannel
407 | ToChan TypedChannel
344 | ToNode
408 | ToNode
345
409
@@ -374,20 +438,24 @@ handleIncomingMessages node = go initConnectionState
374 case event of
438 case event of
375 NT.ConnectionOpened cid rel theirAddr ->
439 NT.ConnectionOpened cid rel theirAddr ->
376 if rel == NT.ReliableOrdered
440 if rel == NT.ReliableOrdered
441+ then
442+ trace node (TraceEvConnected cid)
443+ >> go (
377− then go ( (incomingAt cid ^= Just (theirAddr, Uninit))
444+ (incomingAt cid ^= Just (theirAddr, Uninit))
378 . (incomingFrom theirAddr ^: Set.insert cid)
445 . (incomingFrom theirAddr ^: Set.insert cid)
379 $ st
446 $ st
380 )
447 )
381 else invalidRequest cid st
448 else invalidRequest cid st
382 NT.Received cid payload ->
449 NT.Received cid payload ->
383 case st ^. incomingAt cid of
450 case st ^. incomingAt cid of
384− Just (_, ToProc weakQueue) -> do
451+ Just (_, ToProc pid weakQueue) -> do
385 mQueue <- deRefWeak weakQueue
452 mQueue <- deRefWeak weakQueue
386 forM_ mQueue $ \queue -> do
453 forM_ mQueue $ \queue -> do
387 -- TODO: if we find that the queue is Nothing, should we remove
454 -- TODO: if we find that the queue is Nothing, should we remove
388 -- it from the NC state? (and same for channels, below)
455 -- it from the NC state? (and same for channels, below)
389 let msg = payloadToMessage payload
456 let msg = payloadToMessage payload
390 enqueue queue msg -- 'enqueue' is strict
457 enqueue queue msg -- 'enqueue' is strict
458+ trace node (TraceEvReceived pid msg)
391 go st
459 go st
392 Just (_, ToChan (TypedChannel chan')) -> do
460 Just (_, ToChan (TypedChannel chan')) -> do
393 mChan <- deRefWeak chan'
461 mChan <- deRefWeak chan'
@@ -408,7 +476,7 @@ handleIncomingMessages node = go initConnectionState
408 mProc <- withMVar state $ return . (^. localProcessWithId lpid)
476 mProc <- withMVar state $ return . (^. localProcessWithId lpid)
409 case mProc of
477 case mProc of
410 Just proc ->
478 Just proc ->
411− go (incomingAt cid ^= Just (src, ToProc (processWeakQ proc)) $ st)
479+ go (incomingAt cid ^= Just (src, ToProc pid (processWeakQ proc)) $ st)
412 Nothing ->
480 Nothing ->
413 invalidRequest cid st
481 invalidRequest cid st
414 SendPortIdentifier chId -> do
482 SendPortIdentifier chId -> do
@@ -435,7 +503,8 @@ handleIncomingMessages node = go initConnectionState
435 case st ^. incomingAt cid of
503 case st ^. incomingAt cid of
436 Nothing ->
504 Nothing ->
437 invalidRequest cid st
505 invalidRequest cid st
438− Just (src, _) ->
506+ Just (src, _) -> do
507+ trace node (TraceEvDisconnected cid)
439 go ( (incomingAt cid ^= Nothing)
508 go ( (incomingAt cid ^= Nothing)
440 . (incomingFrom src ^: Set.delete cid)
509 . (incomingFrom src ^: Set.delete cid)
441 $ st
510 $ st
@@ -454,18 +523,15 @@ handleIncomingMessages node = go initConnectionState
454 $ st
523 $ st
455 )
524 )
456 NT.ErrorEvent (NT.TransportError NT.EventEndPointFailed str) ->
525 NT.ErrorEvent (NT.TransportError NT.EventEndPointFailed str) ->
457− fatal $ "Cloud Haskell fatal error: end point failed: " ++ str
526+ fail $ "Cloud Haskell fatal error: end point failed: " ++ str
458 NT.ErrorEvent (NT.TransportError NT.EventTransportFailed str) ->
527 NT.ErrorEvent (NT.TransportError NT.EventTransportFailed str) ->
459− fatal $ "Cloud Haskell fatal error: transport failed: " ++ str
528+ fail $ "Cloud Haskell fatal error: transport failed: " ++ str
460 NT.EndPointClosed ->
529 NT.EndPointClosed ->
461− stopTracer (localTracer node) >> return ()
530+ return ()
462 NT.ReceivedMulticast _ _ ->
531 NT.ReceivedMulticast _ _ ->
463 -- If we received a multicast message, something went horribly wrong
532 -- If we received a multicast message, something went horribly wrong
464 -- and we just give up
533 -- and we just give up
465− fatal "Cloud Haskell fatal error: received unexpected multicast"
534+ fail "Cloud Haskell fatal error: received unexpected multicast"
466−
467− fatal :: String -> IO ()
468− fatal msg = stopTracer (localTracer node) >> fail msg
469
535
470 invalidRequest :: NT.ConnectionId -> ConnectionState -> IO ()
536 invalidRequest :: NT.ConnectionId -> ConnectionState -> IO ()
471 invalidRequest cid st = do
537 invalidRequest cid st = do
@@ -473,7 +539,7 @@ handleIncomingMessages node = go initConnectionState
473 -- node. That is, we should report the remote node as having died, and we
539 -- node. That is, we should report the remote node as having died, and we
474 -- should close incoming connections (this requires a Transport layer
540 -- should close incoming connections (this requires a Transport layer
475 -- extension).
541 -- extension).
476− traceEventFmtIO node "" [(TraceStr "[network] invalid request: "),
542+ traceEventFmtIO node "" [(TraceStr " [network] invalid request: "),
477 (Trace cid)]
543 (Trace cid)]
478 go ( incomingAt cid ^= Nothing
544 go ( incomingAt cid ^= Nothing
479 $ st
545 $ st
@@ -534,23 +600,22 @@ instance Show ProcessKillException where
534
600
535traceNotifyDied :: LocalNode -> Identifier -> DiedReason -> NC ()
601traceNotifyDied :: LocalNode -> Identifier -> DiedReason -> NC ()
536traceNotifyDied node ident reason =
602traceNotifyDied node ident reason =
603+ -- TODO: sendPortDied notifications
604+ liftIO $ withLocalTracer node $ \t ->
537− case reason of
605+ case ident of
606+ (NodeIdentifier nid) -> traceEvent t (TraceEvNodeDied nid reason)
607+ (ProcessIdentifier pid) -> traceEvent t (TraceEvDied pid reason)
538− DiedNormal -> return ()
608+ _ -> return ()
539− _ -> traceNcEventFmt node " "
540− [(TraceStr "[node-controller]"),
541− (Trace ident),
542− (Trace reason)]
543−
544−traceNcEventFmt :: LocalNode -> String -> [TraceArg] -> NC ()
545−traceNcEventFmt node fmt args =
546− liftIO $ traceEventFmtIO node fmt args
547
609
548traceEventFmtIO :: LocalNode
610traceEventFmtIO :: LocalNode
549 -> String
611 -> String
550 -> [TraceArg]
612 -> [TraceArg]
551 -> IO ()
613 -> IO ()
552traceEventFmtIO node fmt args =
614traceEventFmtIO node fmt args =
553− withLocalTracer node $ \t -> traceFormat t fmt args
615+ withLocalTracer node $ \t -> traceLogFmt t fmt args
616+
617+trace :: LocalNode -> TraceEvent -> IO ()
618+trace node ev = withLocalTracer node $ \t -> traceEvent t ev
554
619
555withLocalTracer :: LocalNode -> (Tracer -> IO ()) -> IO ()
620withLocalTracer :: LocalNode -> (Tracer -> IO ()) -> IO ()
556withLocalTracer node act = act (localTracer node)
621withLocalTracer node act = act (localTracer node)
@@ -597,7 +662,7 @@ nodeController = do
597 NCMsg _ (NamedSend label msg') ->
662 NCMsg _ (NamedSend label msg') ->
598 ncEffectNamedSend label msg'
663 ncEffectNamedSend label msg'
599 NCMsg _ (LocalSend to msg') ->
664 NCMsg _ (LocalSend to msg') ->
600− ncEffectLocalSend to msg'
665+ ncEffectLocalSend node to msg'
601 NCMsg _ (LocalPortSend to msg') ->
666 NCMsg _ (LocalPortSend to msg') ->
602 ncEffectLocalPortSend to msg'
667 ncEffectLocalPortSend to msg'
603 NCMsg (ProcessIdentifier from) (Kill to reason) ->
668 NCMsg (ProcessIdentifier from) (Kill to reason) ->
@@ -720,6 +785,7 @@ ncEffectSpawn pid cProc ref = do
720 mProc <- unClosure cProc
785 mProc <- unClosure cProc
721 -- If the closure does not exist, we spawn a process that throws an exception
786 -- If the closure does not exist, we spawn a process that throws an exception
722 -- This allows the remote node to find out what's happening
787 -- This allows the remote node to find out what's happening
788+ -- TODO:
723 let proc = case mProc of
789 let proc = case mProc of
724 Left err -> fail $ "Error: Could not resolve closure: " ++ err
790 Left err -> fail $ "Error: Could not resolve closure: " ++ err
725 Right p -> p
791 Right p -> p
@@ -750,6 +816,9 @@ ncEffectRegister from label atnode mPid reregistration = do
750 then do when (isOk) $
816 then do when (isOk) $
751 do modify' $ registeredHereFor label ^= mPid
817 do modify' $ registeredHereFor label ^= mPid
752 updateRemote node currentVal mPid
818 updateRemote node currentVal mPid
819+ case mPid of
820+ (Just p) -> liftIO $ trace node (TraceEvRegistered p label)
821+ Nothing -> liftIO $ trace node (TraceEvUnRegistered (fromJust currentVal) label)
753 liftIO $ sendMessage node
822 liftIO $ sendMessage node
754 (NodeIdentifier (localNodeId node))
823 (NodeIdentifier (localNodeId node))
755 (ProcessIdentifier from)
824 (ProcessIdentifier from)
@@ -772,6 +841,7 @@ ncEffectRegister from label atnode mPid reregistration = do
772 updateRemote _ _ _ = return ()
841 updateRemote _ _ _ = return ()
773 maybeify f Nothing = unmaybeify $ f []
842 maybeify f Nothing = unmaybeify $ f []
774 maybeify f (Just x) = unmaybeify $ f x
843 maybeify f (Just x) = unmaybeify $ f x
844+
775 unmaybeify [] = Nothing
845 unmaybeify [] = Nothing
776 unmaybeify x = Just x
846 unmaybeify x = Just x
777 incList [] tag = [(tag,1)]
847 incList [] tag = [(tag,1)]
@@ -812,16 +882,19 @@ ncEffectNamedSend label msg = do
812 forM_ mPid $ \pid -> postMessage pid msg
882 forM_ mPid $ \pid -> postMessage pid msg
813
883
814-- [Issue #DP-20]
884-- [Issue #DP-20]
815−ncEffectLocalSend :: ProcessId -> Message -> NC ()
885+ncEffectLocalSend :: LocalNode -> ProcessId -> Message -> NC ()
816−ncEffectLocalSend = postMessage
886+ncEffectLocalSend node to msg =
887+ liftIO $ withLocalProc node to $ \p -> do
888+ enqueue (processQueue p) msg
889+ trace node (TraceEvReceived to msg)
817
890
818-- [Issue #DP-20]
891-- [Issue #DP-20]
819ncEffectLocalPortSend :: SendPortId -> Message -> NC ()
892ncEffectLocalPortSend :: SendPortId -> Message -> NC ()
820−ncEffectLocalPortSend from msg =
893+ncEffectLocalPortSend from msg = do
894+ node <- ask
821 let pid = sendPortProcessId from
895 let pid = sendPortProcessId from
822 cid = sendPortLocalId from
896 cid = sendPortLocalId from
823− in do
824− withLocalProc pid $ \proc -> do
897+ liftIO $ withLocalProc node pid $ \proc -> do
825 mChan <- withMVar (processState proc) $ return . (^. typedChannelWithId cid)
898 mChan <- withMVar (processState proc) $ return . (^. typedChannelWithId cid)
826 case mChan of
899 case mChan of
827 -- in the unlikely event we know nothing about this channel id,
900 -- in the unlikely event we know nothing about this channel id,
@@ -995,21 +1068,21 @@ postAsMessage pid = postMessage pid . createUnencodedMessage
995
1068
996postMessage :: ProcessId -> Message -> NC ()
1069postMessage :: ProcessId -> Message -> NC ()
997postMessage pid msg = do
1070postMessage pid msg = do
1071+ node <- ask
998− withLocalProc pid $ \p -> enqueue (processQueue p) msg
1072+ liftIO $ withLocalProc node pid $ \p -> enqueue (processQueue p) msg
999
1073
1000throwException :: Exception e => ProcessId -> e -> NC ()
1074throwException :: Exception e => ProcessId -> e -> NC ()
1001−throwException pid e = withLocalProc pid $ \p ->
1075+throwException pid e = do
1002− throwTo (processThread p) e
1003−
1004−withLocalProc :: ProcessId -> (LocalProcess -> IO ()) -> NC ()
1005−withLocalProc pid p = do
1006 node <- ask
1076 node <- ask
1007− liftIO $ do
1077+ liftIO $ withLocalProc node pid $ \p -> throwTo (processThread p) e
1078+
1079+withLocalProc :: LocalNode -> ProcessId -> (LocalProcess -> IO ()) -> IO ()
1080+withLocalProc node pid p =
1008− -- By [Unified: table 6, rule missing_process] messages to dead processes
1081+ -- By [Unified: table 6, rule missing_process] messages to dead processes
1009− -- can silently be dropped
1082+ -- can silently be dropped
1010− let lpid = processLocalId pid
1083+ let lpid = processLocalId pid in do
1011− mProc <- withMVar (localState node) $ return . (^. localProcessWithId lpid)
1084+ mProc <- withMVar (localState node) $ return . (^. localProcessWithId lpid)
1012− forM_ mProc p
1085+ forM_ mProc p
1013
1086
1014--------------------------------------------------------------------------------
1087--------------------------------------------------------------------------------
1015-- Accessors --
1088-- Accessors --
tests/TestStats.hs+2−2
@@ -111,7 +111,7 @@ testRemoteLiveProcessInfo node1 = do
111 "pong" <- expect
111 "pong" <- expect
112 pInfo <- getProcessInfo serverPid
112 pInfo <- getProcessInfo serverPid
113 stash result $ pInfo /= Nothing
113 stash result $ pInfo /= Nothing
114− where
114+ where
115 launchRemote :: MVar ProcessId -> IO ()
115 launchRemote :: MVar ProcessId -> IO ()
116 launchRemote locMV = do
116 launchRemote locMV = do
117 node2 <- liftIO $ mkNode "8082"
117 node2 <- liftIO $ mkNode "8082"
@@ -119,7 +119,7 @@ testRemoteLiveProcessInfo node1 = do
119 self <- getSelfPid
119 self <- getSelfPid
120 liftIO $ putMVar locMV self
120 liftIO $ putMVar locMV self
121 _ <- receiveWait [
121 _ <- receiveWait [
122− match (\(pid, "ping") -> send pid "pong")
122+ match (\(pid, "ping") -> send pid "pong")
123 ]
123 ]
124 "stop" <- expect
124 "stop" <- expect
125 return ()
125 return ()
tests/TestTracing.hsnew file+388
@@ -0,0 +1,388 @@
1+{-# LANGUAGE DeriveDataTypeable #-}
2+{-# OPTIONS_GHC -fno-warn-orphans #-}
3+module Main where
4+
5+import Control.Concurrent (threadDelay)
6+import Control.Concurrent.MVar
7+ ( MVar
8+ , newEmptyMVar
9+ , newMVar
10+ , putMVar
11+ , takeMVar
12+ )
13+import Control.Distributed.Process
14+import Control.Distributed.Process.Debug
15+import Control.Distributed.Process.Node
16+ ( forkProcess
17+ , newLocalNode
18+ , initRemoteTable
19+ , closeLocalNode
20+ , LocalNode)
21+import Control.Exception as Ex (catch)
22+import Network.Transport.TCP
23+import Prelude hiding (catch, log)
24+import Test.Framework
25+ ( Test
26+ , defaultMain
27+ , testGroup
28+ )
29+import Test.HUnit (Assertion)
30+import Test.HUnit.Base (assertBool)
31+import Test.Framework.Providers.HUnit (testCase)
32+
33+import System.Environment (getEnv)
34+import System.Posix.Env
35+ ( setEnv
36+ )
37+
38+-- these utilities have been cribbed from distributed-process-platform
39+-- we should really find a way to share them...
40+
41+-- | A mutable cell containing a test result.
42+type TestResult a = MVar a
43+
44+delayedAssertion :: (Eq a)
45+ => String
46+ -> LocalNode
47+ -> a
48+ -> (TestResult a -> Process ())
49+ -> MVar ()
50+ -> Assertion
51+delayedAssertion note localNode expected testProc lock = do
52+ result <- newEmptyMVar
53+ _ <- forkProcess localNode $ do
54+ acquire lock
55+ finally (testProc result)
56+ (release lock)
57+ assertComplete note result expected
58+ where acquire lock' = liftIO $ takeMVar lock'
59+ release lock' = liftIO $ putMVar lock' ()
60+
61+assertComplete :: (Eq a) => String -> MVar a -> a -> IO ()
62+assertComplete msg mv a = do
63+ b <- takeMVar mv
64+ assertBool msg (a == b)
65+
66+stash :: TestResult a -> a -> Process ()
67+stash mvar x = liftIO $ putMVar mvar x
68+
69+------
70+
71+testSpawnTracing :: TestResult Bool -> Process ()
72+testSpawnTracing result = do
73+ setTraceFlags defaultTraceFlags {
74+ traceSpawned = (Just TraceAll)
75+ , traceDied = (Just TraceAll)
76+ }
77+
78+ evSpawned <- liftIO $ newEmptyMVar
79+ evDied <- liftIO $ newEmptyMVar
80+ tracer <- startTracer $ \ev -> do
81+ case ev of
82+ (TraceEvSpawned p) -> liftIO $ putMVar evSpawned p
83+ (TraceEvDied p r) -> liftIO $ putMVar evDied (p, r)
84+ _ -> return ()
85+
86+ (sp, rp) <- newChan
87+ pid <- spawnLocal $ sendChan sp ()
88+ () <- receiveChan rp
89+
90+ tracedAlive <- liftIO $ takeMVar evSpawned
91+ (tracedDead, tracedReason) <- liftIO $ takeMVar evDied
92+
93+ mref <- monitor tracer
94+ stopTracer -- this is asynchronous, so we need to wait...
95+ receiveWait [
96+ matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mref)
97+ ((\_ -> return ()))
98+ ]
99+ setTraceFlags defaultTraceFlags
100+ stash result (tracedAlive == pid &&
101+ tracedDead == pid &&
102+ tracedReason == DiedNormal)
103+
104+testTraceRecvExplicitPid :: TestResult Bool -> Process ()
105+testTraceRecvExplicitPid result = do
106+ res <- liftIO $ newEmptyMVar
107+ pid <- spawnLocal $ do
108+ self <- getSelfPid
109+ expect >>= (flip sendChan) self
110+ withFlags defaultTraceFlags {
111+ traceRecv = traceOnly [pid]
112+ } $ do
113+ withTracer
114+ (\ev ->
115+ case ev of
116+ (TraceEvReceived pid' _) -> stash res (pid == pid')
117+ _ -> return ()) $ do
118+ (sp, rp) <- newChan
119+ send pid sp
120+ p <- receiveChan rp
121+ res' <- liftIO $ takeMVar res
122+ stash result (res' && (p == pid))
123+ return ()
124+
125+testTraceRecvNamedPid :: TestResult Bool -> Process ()
126+testTraceRecvNamedPid result = do
127+ res <- liftIO $ newEmptyMVar
128+ pid <- spawnLocal $ do
129+ self <- getSelfPid
130+ register "foobar" self
131+ expect >>= (flip sendChan) self
132+ withFlags defaultTraceFlags {
133+ traceRecv = traceOnly ["foobar"]
134+ } $ do
135+ withTracer
136+ (\ev ->
137+ case ev of
138+ (TraceEvReceived pid' _) -> stash res (pid == pid')
139+ _ -> return ()) $ do
140+ (sp, rp) <- newChan
141+ send pid sp
142+ p <- receiveChan rp
143+ res' <- liftIO $ takeMVar res
144+ stash result (res' && (p == pid))
145+ return ()
146+
147+testTraceSending :: TestResult Bool -> Process ()
148+testTraceSending result = do
149+ pid <- spawnLocal $ (expect :: Process String) >> return ()
150+ self <- getSelfPid
151+ res <- liftIO $ newEmptyMVar
152+ withFlags defaultTraceFlags { traceSend = traceOn } $ do
153+ withTracer
154+ (\ev ->
155+ case ev of
156+ (TraceEvSent to from _) ->
157+ stash res (to == pid && from == self)
158+ _ ->
159+ return ()) $ do
160+ send pid "hello there"
161+ res' <- liftIO $ takeMVar res
162+ stash result res'
163+
164+testTraceRegistration :: TestResult Bool -> Process ()
165+testTraceRegistration result = do
166+ (sp, rp) <- newChan
167+ pid <- spawnLocal $ do
168+ self <- getSelfPid
169+ () <- expect
170+ register "foobar" self
171+ sendChan sp ()
172+ () <- expect
173+ return ()
174+ res <- liftIO $ newEmptyMVar
175+ withFlags defaultTraceFlags { traceRegistered = traceOn } $ do
176+ withTracer
177+ (\ev ->
178+ case ev of
179+ TraceEvRegistered p s ->
180+ stash res (p == pid && s == "foobar")
181+ _ ->
182+ return ()) $ do
183+ _ <- monitor pid
184+ send pid ()
185+ () <- receiveChan rp
186+ send pid ()
187+ receiveWait [
188+ match (\(ProcessMonitorNotification _ _ _) -> return ())
189+ ]
190+ res' <- liftIO $ takeMVar res
191+ stash result res'
192+
193+testTraceUnRegistration :: TestResult Bool -> Process ()
194+testTraceUnRegistration result = do
195+ pid <- spawnLocal $ do
196+ () <- expect
197+ unregister "foobar"
198+ () <- expect
199+ return ()
200+ register "foobar" pid
201+ res <- liftIO $ newEmptyMVar
202+ withFlags defaultTraceFlags { traceUnregistered = traceOn } $ do
203+ withTracer
204+ (\ev ->
205+ case ev of
206+ TraceEvUnRegistered p n -> do
207+ stash res (p == pid && n == "foobar")
208+ send pid ()
209+ _ ->
210+ return ()) $ do
211+ mref <- monitor pid
212+ send pid ()
213+ receiveWait [
214+ matchIf (\(ProcessMonitorNotification mref' _ _) -> mref == mref')
215+ (\_ -> return ())
216+ ]
217+ res' <- liftIO $ takeMVar res
218+ stash result res'
219+
220+testTraceLayering :: TestResult () -> Process ()
221+testTraceLayering result = do
222+ pid <- spawnLocal $ do
223+ getSelfPid >>= register "foobar"
224+ () <- expect
225+ traceMessage ("traceMsg", 123 :: Int)
226+ return ()
227+ withFlags defaultTraceFlags {
228+ traceDied = traceOnly [pid]
229+ , traceRecv = traceOnly ["foobar"]
230+ } $ doTest pid result
231+ return ()
232+ where
233+ doTest :: ProcessId -> MVar () -> Process ()
234+ doTest pid result' = do
235+ -- TODO: this is pretty gross, even for a test case
236+ died <- liftIO $ newEmptyMVar
237+ withTracer
238+ (\ev ->
239+ case ev of
240+ TraceEvDied _ _ -> liftIO $ putMVar died ()
241+ _ -> return ())
242+ ( do {
243+ recv <- liftIO $ newEmptyMVar
244+ ; withTracer
245+ (\ev' ->
246+ case ev' of
247+ TraceEvReceived _ _ -> liftIO $ putMVar recv ()
248+ _ -> return ())
249+ ( do {
250+ user <- liftIO $ newEmptyMVar
251+ ; withTracer
252+ (\ev'' ->
253+ case ev'' of
254+ TraceEvUser _ -> liftIO $ putMVar user ()
255+ _ -> return ())
256+ (send pid () >> (liftIO $ takeMVar user))
257+ ; liftIO $ takeMVar recv
258+ })
259+ ; liftIO $ takeMVar died
260+ })
261+ liftIO $ putMVar result' ()
262+
263+testRemoteTraceRelay :: TestResult Bool -> Process ()
264+testRemoteTraceRelay result =
265+ let flags = defaultTraceFlags { traceSpawned = traceOn }
266+ in do
267+ node2 <- liftIO $ mkNode "8082"
268+ mvNid <- liftIO $ newEmptyMVar
269+
270+ -- As well as needing node2's NodeId, we want to
271+ -- redirect all its logs back here, to avoid generating
272+ -- garbage on stderr for the duration of the test run.
273+ -- Here we set up that relay, and then wait for a signal
274+ -- that the tracer (on node1) has seen the expected
275+ -- TraceEvSpawned message, at which point we're finished
276+ (Just log) <- whereis "logger"
277+ pid <- liftIO $ forkProcess node2 $ do
278+ logRelay <- spawnLocal $ relay log
279+ reregister "logger" logRelay
280+ getSelfNode >>= stash mvNid >> (expect :: Process ())
281+
282+ nid <- liftIO $ takeMVar mvNid
283+ mref <- monitor pid
284+ observedPid <- liftIO $ newEmptyMVar
285+ spawnedPid <- liftIO $ newEmptyMVar
286+ setTraceFlagsRemote flags nid
287+
288+ withFlags defaultTraceFlags { traceSpawned = traceOn } $ do
289+ withTracer
290+ (\ev ->
291+ case ev of
292+ TraceEvSpawned p -> stash observedPid p >> send pid ()
293+ _ -> return ()) $ do
294+ relayPid <- startTraceRelay nid
295+ liftIO $ threadDelay 1000000
296+ p <- liftIO $ forkProcess node2 $ do
297+ expectTimeout 1000000 :: Process (Maybe ())
298+ return ()
299+ stash spawnedPid p
300+
301+ -- Now we wait for (the outer) pid to exit. This won't happen until
302+ -- our tracer has seen the trace event for `p' and send `p' the
303+ -- message its waiting for prior to exiting
304+ receiveWait [
305+ matchIf (\(ProcessMonitorNotification mref' _ _) -> mref == mref')
306+ (\_ -> return ())
307+ ]
308+
309+ relayRef <- monitor relayPid
310+ kill relayPid "stop"
311+ receiveWait [
312+ matchIf (\(ProcessMonitorNotification rref' _ _) -> rref' == relayRef)
313+ (\_ -> return ())
314+ ]
315+ observed <- liftIO $ takeMVar observedPid
316+ expected <- liftIO $ takeMVar spawnedPid
317+ stash result (observed == expected)
318+ -- and just to be polite...
319+ liftIO $ closeLocalNode node2
320+
321+tests :: LocalNode -> IO [Test]
322+tests node1 = do
323+ -- if we execute the test cases in parallel, the
324+ -- various tracers will race with one another and
325+ -- we'll get garbage results (or worse, deadlocks)
326+ lock <- liftIO $ newMVar ()
327+ enabled <- checkTraceEnabled
328+ case enabled of
329+ True ->
330+ return [
331+ testGroup "Tracing" [
332+ testCase "Spawn Tracing"
333+ (delayedAssertion
334+ "expected dead process-info to be ProcessInfoNone"
335+ node1 True testSpawnTracing lock)
336+ , testCase "Recv Tracing (Explicit Pid)"
337+ (delayedAssertion
338+ "expected a recv trace for the supplied pid"
339+ node1 True testTraceRecvExplicitPid lock)
340+ , testCase "Recv Tracing (Named Pid)"
341+ (delayedAssertion
342+ "expected a recv trace for the process registered as 'foobar'"
343+ node1 True testTraceRecvNamedPid lock)
344+ , testCase "Trace Send(er)"
345+ (delayedAssertion
346+ "expected a 'send' trace with the requisite fields set"
347+ node1 True testTraceSending lock)
348+ , testCase "Trace Registration"
349+ (delayedAssertion
350+ "expected a 'registered' trace"
351+ node1 True testTraceRegistration lock)
352+ , testCase "Trace Unregistration"
353+ (delayedAssertion
354+ "expected an 'unregistered' trace"
355+ node1 True testTraceUnRegistration lock)
356+ , testCase "Trace Layering"
357+ (delayedAssertion
358+ "expected blah"
359+ node1 () testTraceLayering lock)
360+ , testCase "Remote Trace Relay"
361+ (delayedAssertion
362+ "expected blah"
363+ node1 True testRemoteTraceRelay lock)
364+ ] ]
365+ False ->
366+ return []
367+
368+checkTraceEnabled :: IO Bool
369+checkTraceEnabled =
370+ Ex.catch (getEnv "DISTRIBUTED_PROCESS_TRACE_ENABLED" >> return True)
371+ (\(_ :: IOError) -> return False)
372+
373+
374+mkNode :: String -> IO LocalNode
375+mkNode port = do
376+ Right (transport1, _) <- createTransportExposeInternals
377+ "127.0.0.1" port defaultTCPParameters
378+ newLocalNode transport1 initRemoteTable
379+
380+main :: IO ()
381+main = do
382+ setEnv "DISTRIBUTED_PROCESS_TRACE_ENABLED" "true" True
383+ node1 <- mkNode "8081"
384+ testData <- tests node1
385+ defaultMain testData
386+ closeLocalNode node1
387+ return ()
388+