git.iko.soy/distributed-process
on master help sign in
Trace primitives and node controller integration
2a21c07Tim Watson
distributed-process.cabal+10−7
compare with master
unifiedsplit
@@ -59,19 +59,22 @@ 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,
66 Control.Distributed.Process.Internal.CQueue,
68 Control.Distributed.Process.Internal.CQueue,
67− Control.Distributed.Process.Internal.Types,
68− Control.Distributed.Process.Internal.Trace,
69 Control.Distributed.Process.Internal.Closure.BuiltIn,
69 Control.Distributed.Process.Internal.Closure.BuiltIn,
70 Control.Distributed.Process.Internal.Messaging,
70 Control.Distributed.Process.Internal.Messaging,
71 Control.Distributed.Process.Internal.StrictList,
71 Control.Distributed.Process.Internal.StrictList,
72 Control.Distributed.Process.Internal.StrictMVar,
72 Control.Distributed.Process.Internal.StrictMVar,
73− Control.Distributed.Process.Internal.WeakTQueue
73+ Control.Distributed.Process.Internal.WeakTQueue,
74− Control.Distributed.Process.Internal.StrictContainerAccessors
74+ Control.Distributed.Process.Internal.StrictContainerAccessors,
75+ Control.Distributed.Process.Internal.Trace.Types,
76+ Control.Distributed.Process.Internal.Trace.Primitives,
77+ Control.Distributed.Process.Internal.Trace.Tracer
75 Extensions: RankNTypes,
78 Extensions: RankNTypes,
76 ScopedTypeVariables,
79 ScopedTypeVariables,
77 FlexibleInstances,
80 FlexibleInstances,
@@ -108,7 +111,7 @@ Test-Suite TestCH
108 ScopedTypeVariables,
111 ScopedTypeVariables,
109 DeriveDataTypeable,
112 DeriveDataTypeable,
110 GeneralizedNewtypeDeriving
113 GeneralizedNewtypeDeriving
111− ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind
114+ ghc-options: -Wall -threaded -debug -eventlog -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind
112 HS-Source-Dirs: tests
115 HS-Source-Dirs: tests
113
116
114Test-Suite TestClosure
117Test-Suite TestClosure
src/Control/Distributed/Process.hs+11−1
@@ -52,6 +52,9 @@ module Control.Distributed.Process
52 , unwrapMessage
52 , unwrapMessage
53 , handleMessage
53 , handleMessage
54 , forward
54 , forward
55+ , delegate
56+ , relay
57+ , proxy
55 -- * Process management
58 -- * Process management
56 , spawn
59 , spawn
57 , call
60 , call
@@ -215,6 +218,9 @@ import Control.Distributed.Process.Internal.Primitives
215 , unwrapMessage
218 , unwrapMessage
216 , handleMessage
219 , handleMessage
217 , forward
220 , forward
221+ , delegate
222+ , relay
223+ , proxy
218 -- Process management
224 -- Process management
219 , terminate
225 , terminate
220 , ProcessTerminationException(..)
226 , ProcessTerminationException(..)
@@ -285,7 +291,11 @@ import Control.Distributed.Process.Node (forkProcess)
285-- 2. 'send' may block (when the system TCP buffers are full, while we are
291-- 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
292-- 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
293-- return does not imply that the remote process received the message (much
288−-- less processed it)
294+-- less processed it). When 'send' targets a local process, it is dispatched
295+-- via the node controller however, in which cases it will /not/ block. This
296+-- isn't part of the semantics, so it should be ok. The ordering is maintained
297+-- because the ctrlChannel still has FIFO semantics with regards interactions
298+-- between two disparate forkIO threads.
289--
299--
290-- 3. Message delivery is reliable and ordered. That means that if process A
300-- 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
301-- sends messages m1, m2, m3 to process B, B will either arrive all three
src/Control/Distributed/Process/Debug.hsnew file+248
@@ -0,0 +1,248 @@
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 that allows a
14+-- user supplied /tracer process/ to receive messages when certain events
15+-- take place. It's possible to use this facility to debug a running system
16+-- and/or generate runtime monitoring data in a variety of formats.
17+--
18+-- [Enabling Tracing]
19+--
20+-- Tracing is disabled by default, because it carries a (relatively small, but
21+-- real) cost at runtime. To enable any kind of tracing to work, the environment
22+-- variable @DISTRIBUTED_PROCESS_TRACE_ENABLED@ must be set (the contents are
23+-- ignored). When the environment variable /is/ set, the system will generate
24+-- a /trace events/ in various circumstances - see 'TraceEvent' for a list of
25+-- all the published events. A user can additionally publish custom trace events
26+-- in the form of 'TraceEvLog' log messages or pass custom (i.e., completely
27+-- user defined) event data using the 'traceMessage' function. If the
28+-- environment variable is not set, no trace events will ever be published.
29+--
30+-- All published trace events are forwarded to a /tracer process/, which can be
31+-- specified (and changed) at runtime using 'traceEnable'. Some pre-defined
32+-- tracer processes are provided for conveniently printing to stderr, a log file
33+-- or the GHC eventlog.
34+--
35+-- If a tracer process crashes, no attempt is made to restart it.
36+--
37+-- [Working with multiple tracer processes]
38+--
39+-- The system tracing facility only ever writes to a single tracer process. This
40+-- invariant insulates the tracer controller and ensures a fast path for
41+-- handling all trace events. /This/ module provides facilities for layering
42+-- trace handlers using Cloud Haskell's built-in delegation primitives.
43+--
44+-- The 'startTracer' function wraps the registered @tracer@ process with the
45+-- supplied handler, forwarding trace events to the original tracer. The
46+-- corresponding 'stopTracer' function terminates tracer processes in reverse
47+-- of the order in which they were started.
48+--
49+-- [Built in tracers]
50+--
51+-- The built in tracers provide a simple /logging/ facility that writes trace
52+-- events out to either a log file, @stderr@ or the GHC eventlog. These tracers
53+-- can be configured using environment variables, or specified manually using
54+-- the 'traceEnable' function. The base tracer process cannot be
55+--
56+-- When a new local node is started, the contents of the environment variable
57+-- @DISTRIBUTED_PROCESS_TRACE_FILE@ are checked for a valid file path. If this
58+-- exists and the file can be opened for writing, all trace output will be
59+-- directed thence. If the environment variable is empty, the path invalid, or
60+-- the file is unavailable for writing - e.g., because another node has already
61+-- started tracing to it - then the @DISTRIBUTED_PROCESS_TRACE_CONSOLE@
62+-- environment variable is checked for /any/ non-empty value. If this is set,
63+-- then all trace output will be directed to the system logger process. If
64+-- neither variable provides a valid trace configuration, all internal traces
65+-- are written to "Debug.Trace.traceEventIO", which writes to the GHC eventlog.
66+--
67+-- Users of the /simplelocalnet/ Cloud Haskell backend should also note that
68+-- because the trace file option only supports trace output from a single node
69+-- (so as to avoid interleaving), a file trace configured for the master node
70+-- will prevent slaves from tracing to the file and they will fall back to using
71+-- the console or eventlog tracers instead.
72+--
73+-- Just as with the "Debug.Trace" module, this is a debugging/tracing facility
74+-- for use in development, and probably not best used in a production setting -
75+-- which is why the default behaviour is to trace to the GHC eventlog. For a
76+-- general purpose logging facility, you should consider 'say'.
77+--
78+-- Support for writing to the eventlog requires specific intervention to work,
79+-- without which traces are silently dropped/ignored and no output will be
80+-- generated. The GHC eventlog documentation provides information about enabling,
81+-- viewing and working with event traces at
82+-- <http://hackage.haskell.org/trac/ghc/wiki/EventLog>.
83+--
84+module Control.Distributed.Process.Debug
85+ ( -- * Exported Data Types
86+ TraceArg(..)
87+ , TraceEvent(..)
88+ , TraceFlags(..)
89+ , TraceSubject(..)
90+ -- * Configuring Tracing
91+ , enableTrace
92+ , disableTrace
93+ , setTraceFlags
94+ , defaultTraceFlags
95+ -- * Debugging
96+ , startTracer
97+ , stopTracer
98+ -- * Sending Trace Data
99+ , traceLog
100+ , traceLogFmt
101+ , traceMessage
102+ -- * Built in tracers
103+ , systemLoggerTracer
104+ , logfileTracer
105+ , eventLogTracer
106+ )
107+ where
108+
109+import Control.Applicative ((<$>))
110+import Control.Distributed.Process.Internal.Primitives
111+ ( proxy
112+ , finally
113+ , die
114+ , getSelfPid
115+ , reregister
116+ , register
117+ , whereis
118+ , send
119+ )
120+import Control.Distributed.Process.Internal.Types
121+ ( Tracer(..)
122+ , LocalNode(..)
123+ , ProcessId
124+ , Process
125+ , LocalProcess(..)
126+ )
127+import Control.Distributed.Process.Internal.Trace.Types
128+ ( TraceArg(..)
129+ , TraceEvent(..)
130+ , TraceFlags(..)
131+ , TraceSubject(..)
132+ , defaultTraceFlags
133+ )
134+import Control.Distributed.Process.Internal.Trace.Tracer
135+ ( systemLoggerTracer
136+ , logfileTracer
137+ , eventLogTracer
138+ )
139+import qualified Control.Distributed.Process.Internal.Trace.Primitives as Tracer
140+ ( traceLog
141+ , traceLogFmt
142+ , traceMessage
143+ , enableTrace
144+ , disableTrace
145+ , setTraceFlags
146+ )
147+import Control.Distributed.Process.Node
148+import Control.Distributed.Process.Serializable
149+
150+import Control.Monad.IO.Class (liftIO)
151+import Control.Monad.Reader (ask)
152+
153+import Data.Binary()
154+
155+--------------------------------------------------------------------------------
156+-- Debugging/Tracing API --
157+--------------------------------------------------------------------------------
158+
159+-- | Starts a new tracer, using the supplied trace function.
160+-- Only one tracer can be registered at a time, however /this/ function overlays
161+-- the registered tracer with the supplied handler, allowing the user to layer
162+-- multiple tracers on top of one another, with trace events forwarded down
163+-- through all the layers in turn. Once the top layer is stopped, it will
164+-- reregister the original tracer pid before terminating. If no tracer is
165+-- currently registered, this function will have no effect.
166+startTracer :: (TraceEvent -> Process ()) -> Process ()
167+startTracer handler = withRegisteredTracer $ \pid -> do
168+ node <- processNode <$> ask
169+ newPid <- liftIO $ forkProcess node $ do
170+ finally (mkProxy pid handler)
171+ (reregister "tracer" pid)
172+ enableTrace newPid
173+ where
174+ mkProxy :: ProcessId -> (TraceEvent -> Process ()) -> Process ()
175+ mkProxy pid act = do
176+ getSelfPid >>= register "tracer"
177+ proxy pid $ \(ev :: TraceEvent) ->
178+ case ev of
179+ (TraceEvTakeover _) -> die "finished"
180+ TraceEvDisable -> die "stop"
181+ _ -> act ev >> return True
182+
183+-- | Stops a user supplied tracer started with 'startTracer'.
184+-- Note that only one tracer process can be active at any given time.
185+-- This process will stop the last process started with 'startTracer'.
186+-- If 'startTracer' is called multiple times, successive calls to this
187+-- function will stop the tracers in the reverse order in which they were
188+-- started.
189+--
190+-- This function will never stop the system tracer (i.e., the tracer
191+-- initially started when the node is created), therefore once all user
192+-- supplied tracers (i.e., processes started via 'startTracer') have exited,
193+-- subsequent calls to this function have no effect.
194+--
195+-- If tracing is support disabled for the node, this function will also
196+-- have no effect.
197+stopTracer :: Process ()
198+stopTracer = withLocalTracer $ \t ->
199+ case t of
200+ InactiveTracer -> return ()
201+ (ActiveTracer _ _) -> withRegisteredTracer $ \pid -> do
202+ -- we need to avoid killing the initial (base) tracer, as
203+ -- nothing we rely on having exactly 1 registered tracer
204+ -- process at all times!
205+ basePid <- whereis "tracer.initial"
206+ case basePid == (Just pid) of
207+ True -> return ()
208+ False -> send pid TraceEvDisable
209+
210+withRegisteredTracer :: (ProcessId -> Process ()) -> Process ()
211+withRegisteredTracer act = do
212+ t <- whereis "tracer"
213+ case t of
214+ Nothing -> return ()
215+ Just pid -> act pid
216+
217+-- | Enable tracing to the supplied process.
218+enableTrace :: ProcessId -> Process ()
219+enableTrace pid = withLocalTracer $ \t -> liftIO $ Tracer.enableTrace t pid
220+
221+-- | Disable the currently configured trace.
222+disableTrace :: Process ()
223+disableTrace = withLocalTracer $ \t -> liftIO $ Tracer.disableTrace t
224+
225+-- | Set the given flags for the current tracer.
226+setTraceFlags :: TraceFlags -> Process ()
227+setTraceFlags f = withLocalTracer $ \t -> liftIO $ Tracer.setTraceFlags t f
228+
229+-- | Send a log message to the internal tracing facility. If tracing is
230+-- enabled, this will create a custom trace log event.
231+--
232+traceLog :: String -> Process ()
233+traceLog s = withLocalTracer $ \t -> liftIO $ Tracer.traceLog t s
234+
235+-- | Send a log message to the internal tracing facility, using the given
236+-- list of printable 'TraceArg's interspersed with the preceding delimiter.
237+--
238+traceLogFmt :: String -> [TraceArg] -> Process ()
239+traceLogFmt d ls = withLocalTracer $ \t -> liftIO $ Tracer.traceLogFmt t d ls
240+
241+-- | Send an arbitrary 'Message' to the tracer process.
242+traceMessage :: Serializable m => m -> Process ()
243+traceMessage msg = withLocalTracer $ \t -> liftIO $ Tracer.traceMessage t msg
244+
245+withLocalTracer :: (Tracer -> Process ()) -> Process ()
246+withLocalTracer act = do
247+ node <- processNode <$> ask
248+ act (localTracer node)
src/Control/Distributed/Process/Internal/Primitives.hs+63−54
@@ -25,9 +25,13 @@ module Control.Distributed.Process.Internal.Primitives
25 , matchMessage
25 , matchMessage
26 , matchMessageIf
26 , matchMessageIf
27 , wrapMessage
27 , wrapMessage
28+ , wrapMessageLocal
28 , unwrapMessage
29 , unwrapMessage
29 , handleMessage
30 , handleMessage
30 , forward
31 , forward
32+ , delegate
33+ , relay
34+ , proxy
31 -- * Process management
35 -- * Process management
32 , terminate
36 , terminate
33 , ProcessTerminationException(..)
37 , ProcessTerminationException(..)
@@ -88,8 +92,6 @@ module Control.Distributed.Process.Internal.Primitives
88 -- * Reconnecting
92 -- * Reconnecting
89 , reconnect
93 , reconnect
90 , reconnectPort
94 , reconnectPort
91− -- * Tracing/Debugging
92− , trace
93 ) where
95 ) where
94
96
95#if ! MIN_VERSION_base(4,6,0)
97#if ! MIN_VERSION_base(4,6,0)
@@ -176,12 +178,16 @@ import Control.Distributed.Process.Internal.Messaging
176 , sendPayload
178 , sendPayload
177 , disconnect
179 , disconnect
178 )
180 )
179−import qualified Control.Distributed.Process.Internal.Trace as Trace
181+import Control.Distributed.Process.Internal.Trace.Primitives
182+ ( traceEvent
183+ , TraceEvent(..)
184+ )
180import Control.Distributed.Process.Internal.WeakTQueue
185import Control.Distributed.Process.Internal.WeakTQueue
181 ( newTQueueIO
186 ( newTQueueIO
182 , readTQueue
187 , readTQueue
183 , mkWeakTQueue
188 , mkWeakTQueue
184 )
189 )
190+
185import Unsafe.Coerce
191import Unsafe.Coerce
186
192
187--------------------------------------------------------------------------------
193--------------------------------------------------------------------------------
@@ -191,18 +197,24 @@ import Unsafe.Coerce
191-- | Send a message
197-- | Send a message
192send :: Serializable a => ProcessId -> a -> Process ()
198send :: Serializable a => ProcessId -> a -> Process ()
193-- This requires a lookup on every send. If we want to avoid that we need to
199-- 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
200+-- modify serializable to allow for stateful (IO) deserialization.
195send them msg = do
201send them msg = do
196 proc <- ask
202 proc <- ask
203+ let us = processId proc
204+ node = processNode proc
197− let node = localNodeId (processNode proc)
205+ nodeId = localNodeId node
198 destNode = (processNodeId them) in do
206 destNode = (processNodeId them) in do
199− case destNode == node of
207+ case destNode == nodeId of
200 True -> sendLocal them msg
208 True -> sendLocal them msg
201 False -> liftIO $ sendMessage (processNode proc)
209 False -> liftIO $ sendMessage (processNode proc)
202 (ProcessIdentifier (processId proc))
210 (ProcessIdentifier (processId proc))
203 (ProcessIdentifier them)
211 (ProcessIdentifier them)
204 NoImplicitReconnect
212 NoImplicitReconnect
205 msg
213 msg
214+ -- We do not fire the trace event until after the sending is complete;
215+ -- In the remote case, 'sendMessage' can block in the networking stack.
216+ liftIO $ traceEvent (localTracer node)
217+ (TraceEvSent them us (createUnencodedMessage msg))
206
218
207-- | Wait for a message of a specific type
219-- | Wait for a message of a specific type
208expect :: forall a. Serializable a => Process a
220expect :: forall a. Serializable a => Process a
@@ -358,15 +370,21 @@ matchMessageIf c p = Match $ MatchMsg $ \msg ->
358forward :: Message -> ProcessId -> Process ()
370forward :: Message -> ProcessId -> Process ()
359forward msg them = do
371forward msg them = do
360 proc <- ask
372 proc <- ask
361− let node = localNodeId (processNode proc)
373+ let node = processNode proc
374+ us = processId proc
375+ nid = localNodeId node
362 destNode = (processNodeId them) in do
376 destNode = (processNodeId them) in do
363− case destNode == node of
377+ case destNode == nid of
364 True -> sendCtrlMsg Nothing (LocalSend them msg)
378 True -> sendCtrlMsg Nothing (LocalSend them msg)
365 False -> liftIO $ sendPayload (processNode proc)
379 False -> liftIO $ sendPayload (processNode proc)
366 (ProcessIdentifier (processId proc))
380 (ProcessIdentifier (processId proc))
367 (ProcessIdentifier them)
381 (ProcessIdentifier them)
368 NoImplicitReconnect
382 NoImplicitReconnect
369 (messageToPayload msg)
383 (messageToPayload msg)
384+ -- We do not fire the trace event until after the sending is complete;
385+ -- In the remote case, 'sendMessage' can block in the networking stack.
386+ liftIO $ traceEvent (localTracer node)
387+ (TraceEvSent them us (createUnencodedMessage msg))
370
388
371
389
372-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are
390-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are
@@ -386,6 +404,12 @@ forward msg them = do
386wrapMessage :: Serializable a => a -> Message
404wrapMessage :: Serializable a => a -> Message
387wrapMessage = createMessage
405wrapMessage = createMessage
388
406
407+-- | Like 'wrapMessage', but produces a 'Message' that is only usable in the
408+-- local node. This is primarily useful when consuming the tracing and debugging
409+-- APIs in "Control.Distributed.Process.Debug".
410+wrapMessageLocal :: Serializable a => a -> Message
411+wrapMessageLocal = createUnencodedMessage
412+
389-- | Attempt to unwrap a raw 'Message'.
413-- | Attempt to unwrap a raw 'Message'.
390-- If the type of the decoded message payload matches the expected type, the
414-- If the type of the decoded message payload matches the expected type, the
391-- value will be returned with @Just@, otherwise @Nothing@ indicates the types
415-- value will be returned with @Just@, otherwise @Nothing@ indicates the types
@@ -480,6 +504,34 @@ succeeds) that unsafeCoerce will not fail.
480matchUnknown :: Process b -> Match b
504matchUnknown :: Process b -> Match b
481matchUnknown p = Match $ MatchMsg (const (Just p))
505matchUnknown p = Match $ MatchMsg (const (Just p))
482
506
507+-- | Receives messages and forwards them to @pid@ if @p msg == True@.
508+delegate :: ProcessId -> (Message -> Bool) -> Process ()
509+delegate pid p = do
510+ receiveWait [
511+ matchAny (\m -> case (p m) of
512+ True -> forward m pid
513+ False -> return ())
514+ ]
515+ delegate pid p
516+
517+-- | A straight relay that simply forwards all messages to the supplied pid.
518+relay :: ProcessId -> Process ()
519+relay pid = receiveWait [ matchAny (\m -> forward m pid) ] >> relay pid
520+
521+-- | Proxies @pid@ and forwards messages whenever @proc@ evaluates to @True@.
522+-- Unlike 'delegate' the predicate @proc@ runs in the 'Process' monad, allowing
523+-- for richer proxy behaviour.
524+proxy :: Serializable a => ProcessId -> (a -> Process Bool) -> Process ()
525+proxy pid proc = do
526+ receiveWait [
527+ matchAny (\m -> do
528+ next <- handleMessage m proc
529+ case next of
530+ Just True -> forward m pid
531+ _ -> return ())
532+ ]
533+ proxy pid proc
534+
483--------------------------------------------------------------------------------
535--------------------------------------------------------------------------------
484-- Process management --
536-- Process management --
485--------------------------------------------------------------------------------
537--------------------------------------------------------------------------------
@@ -750,7 +802,7 @@ finally a sequel = bracket_ (return ()) sequel a
750data Handler a = forall e . Exception e => Handler (e -> Process a)
802data Handler a = forall e . Exception e => Handler (e -> Process a)
751
803
752instance Functor Handler where
804instance Functor Handler where
753− fmap f (Handler h) = Handler (fmap f . h)
805+ fmap f (Handler h) = Handler (fmap f . h)
754
806
755-- | Lift 'Control.Exception.catches'
807-- | Lift 'Control.Exception.catches'
756catches :: Process a -> [Handler a] -> Process a
808catches :: Process a -> [Handler a] -> Process a
@@ -991,56 +1043,13 @@ reconnectPort them = do
991 node <- processNode <$> ask
1043 node <- processNode <$> ask
992 liftIO $ disconnect node (ProcessIdentifier us) (SendPortIdentifier (sendPortId them))
1044 liftIO $ disconnect node (ProcessIdentifier us) (SendPortIdentifier (sendPortId them))
993
1045
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--------------------------------------------------------------------------------
1046--------------------------------------------------------------------------------
1038-- Auxiliary functions --
1047-- Auxiliary functions --
1039--------------------------------------------------------------------------------
1048--------------------------------------------------------------------------------
1040
1049
1041sendLocal :: (Serializable a) => ProcessId -> a -> Process ()
1050sendLocal :: (Serializable a) => ProcessId -> a -> Process ()
1042−sendLocal pid msg =
1051+sendLocal to msg =
1043− sendCtrlMsg Nothing $ LocalSend pid (createUnencodedMessage msg)
1052+ sendCtrlMsg Nothing $ LocalSend to (createUnencodedMessage msg)
1044
1053
1045sendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()
1054sendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()
1046sendChanLocal spId msg =
1055sendChanLocal spId msg =
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+75
@@ -0,0 +1,75 @@
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
4+module Control.Distributed.Process.Internal.Trace.Primitives
5+ ( Tracer
6+ , TraceEvent(..)
7+ -- * Sending Trace Data
8+ , traceLog
9+ , traceLogFmt
10+ , traceEvent
11+ , traceMessage
12+ -- * Configuring A Tracer
13+ , defaultTraceFlags
14+ , enableTrace
15+ , disableTrace
16+ , setTraceFlags
17+ ) where
18+
19+import Control.Distributed.Process.Internal.CQueue (enqueue)
20+import Control.Distributed.Process.Internal.Trace.Types
21+ ( TraceArg(..)
22+ , TraceEvent(..)
23+ , SetTrace(..)
24+ , TraceFlags(..)
25+ , defaultTraceFlags
26+ )
27+import Control.Distributed.Process.Internal.Types
28+ ( Tracer(..)
29+ , ProcessId
30+ , Message
31+ , createUnencodedMessage
32+ )
33+import Control.Distributed.Process.Serializable
34+import Data.Foldable (forM_)
35+import Data.List (intersperse)
36+import System.Mem.Weak (deRefWeak)
37+
38+--------------------------------------------------------------------------------
39+-- Main API --
40+--------------------------------------------------------------------------------
41+
42+traceLog :: Tracer -> String -> IO ()
43+traceLog tr s = traceMessage tr (TraceEvLog s)
44+
45+traceLogFmt :: Tracer
46+ -> String
47+ -> [TraceArg]
48+ -> IO ()
49+traceLogFmt t d ls =
50+ traceLog t $ concat (intersperse d (map toS ls))
51+ where toS :: TraceArg -> String
52+ toS (TraceStr s) = s
53+ toS (Trace a) = show a
54+
55+traceEvent :: Tracer -> TraceEvent -> IO ()
56+traceEvent tr ev = traceIt tr (createUnencodedMessage ev)
57+
58+traceMessage :: Serializable m => Tracer -> m -> IO ()
59+traceMessage tr msg = traceEvent tr (TraceEvUser (createUnencodedMessage msg))
60+
61+enableTrace :: Tracer -> ProcessId -> IO ()
62+enableTrace t p = traceIt t (createUnencodedMessage (TraceEnable p))
63+
64+disableTrace :: Tracer -> IO ()
65+disableTrace t = traceIt t (createUnencodedMessage TraceDisable)
66+
67+setTraceFlags :: Tracer -> TraceFlags -> IO ()
68+setTraceFlags t f = traceIt t (createUnencodedMessage f)
69+
70+traceIt :: Tracer -> Message -> IO ()
71+traceIt InactiveTracer _ = return ()
72+traceIt (ActiveTracer _ wqRef) msg = do
73+ mQueue <- deRefWeak wqRef
74+ forM_ mQueue $ \queue -> enqueue queue msg
75+
src/Control/Distributed/Process/Internal/Trace/Tracer.hsnew file+288
@@ -0,0 +1,288 @@
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+ , receiveWait
24+ , forward
25+ , match
26+ , matchAny
27+ , matchAnyIf
28+ , handleMessage
29+ )
30+import Control.Distributed.Process.Internal.Trace.Types
31+ ( TraceEvent(..)
32+ , SetTrace(..)
33+ , Addressable(..)
34+ , TraceSubject(..)
35+ , TraceFlags(..)
36+ , defaultTraceFlags
37+ )
38+import Control.Distributed.Process.Internal.Types
39+ ( LocalNode(..)
40+ , NCMsg(..)
41+ , ProcessId
42+ , Process
43+ , LocalProcess(..)
44+ , Identifier(..)
45+ , ProcessSignal(NamedSend)
46+ , Message
47+ , forever'
48+ , nullProcessId
49+ , createUnencodedMessage
50+ )
51+import Control.Exception
52+ ( SomeException
53+ )
54+
55+import Control.Monad.IO.Class (liftIO)
56+import Control.Monad.Reader (ask)
57+
58+import Data.Set (Set)
59+import qualified Data.Set as Set
60+import Data.Map (Map)
61+import qualified Data.Map as Map
62+
63+import Data.Maybe (fromMaybe)
64+import Data.Time.Clock (getCurrentTime)
65+import Data.Time.Format (formatTime)
66+import Debug.Trace (traceEventIO)
67+
68+import Prelude hiding (catch)
69+
70+import System.Environment (getEnv)
71+import System.IO
72+ ( Handle
73+ , IOMode(AppendMode)
74+ , BufferMode(..)
75+ , openFile
76+ , hClose
77+ , hPutStrLn
78+ , hSetBuffering
79+ )
80+import System.Locale (defaultTimeLocale)
81+import System.Mem.Weak
82+ ( Weak
83+ )
84+
85+data TracerState = TracerST {
86+ sendTrace :: !(Message -> Process ())
87+ , flags :: !TraceFlags
88+ , regNames :: !(Map ProcessId (Set String))
89+ }
90+
91+--------------------------------------------------------------------------------
92+-- Trace Handlers --
93+--------------------------------------------------------------------------------
94+
95+defaultTracer :: Process ()
96+defaultTracer =
97+ catch (checkEnv "DISTRIBUTED_PROCESS_TRACE_FILE" >>= logfileTracer)
98+ (\(_ :: IOError) -> defaultTracerAux)
99+
100+defaultTracerAux :: Process ()
101+defaultTracerAux =
102+ catch (checkEnv "DISTRIBUTED_PROCESS_TRACE_CONSOLE" >> systemLoggerTracer)
103+ (\(_ :: IOError) -> eventLogTracer)
104+
105+checkEnv :: String -> Process String
106+checkEnv s = liftIO $ getEnv s
107+
108+systemLoggerTracer :: Process ()
109+systemLoggerTracer = do
110+ node <- processNode <$> ask
111+ let tr = sendTraceMsg node
112+ forever' $ receiveWait [ matchAny (\m -> handleMessage m tr) ]
113+ where
114+ sendTraceMsg :: LocalNode -> TraceEvent -> Process ()
115+ sendTraceMsg node ev = do
116+ now <- liftIO $ getCurrentTime
117+ msg <- return $ (formatTime defaultTimeLocale "%c" now, (show ev))
118+ emptyPid <- return $ (nullProcessId (localNodeId node))
119+ traceMsg <- return $ NCMsg {
120+ ctrlMsgSender = ProcessIdentifier (emptyPid)
121+ , ctrlMsgSignal = (NamedSend "logger"
122+ (createUnencodedMessage msg))
123+ }
124+ liftIO $ writeChan (localCtrlChan node) traceMsg
125+
126+eventLogTracer :: Process ()
127+eventLogTracer = do
128+ liftIO $ traceEventIO "starting event log tracer"
129+ -- NB: when the GHC event log supports tracing arbitrary (ish) data, we will
130+ -- almost certainly use *that* facility independently of whether or not there
131+ -- is a tracer process installed. This is just a stop gap until then.
132+ forever' $ receiveWait [ matchAny (\m -> handleMessage m writeTrace) ]
133+ where
134+ writeTrace :: TraceEvent -> Process ()
135+ writeTrace ev = liftIO $ traceEventIO (show ev)
136+
137+logfileTracer :: FilePath -> Process ()
138+logfileTracer p = do
139+ -- TODO: error handling if the handle cannot be opened
140+ h <- liftIO $ openFile p AppendMode
141+ liftIO $ hSetBuffering h LineBuffering
142+ logger h `catch` (\(_ :: SomeException) -> liftIO $ hClose h)
143+ where
144+ logger :: Handle -> Process ()
145+ logger h' = forever' $ do
146+ receiveWait [
147+ matchAnyIf (\ev -> case ev of
148+ TraceEvDisable -> True
149+ (TraceEvTakeover _) -> True
150+ _ -> False)
151+ (\_ -> (liftIO $ hClose h') >> error "trace stopped")
152+ , matchAny (\ev -> handleMessage ev (writeTrace h'))
153+ ]
154+
155+ writeTrace :: Handle -> TraceEvent -> Process ()
156+ writeTrace h ev = do
157+ liftIO $ do
158+ now <- getCurrentTime
159+ hPutStrLn h $ (formatTime defaultTimeLocale "%c - " now) ++ (show ev)
160+
161+--------------------------------------------------------------------------------
162+-- Tracer Implementation --
163+--------------------------------------------------------------------------------
164+
165+traceController :: MVar ((Weak (CQueue Message))) -> Process ()
166+traceController mv = do
167+ -- This breach of encapsulation is deliberate: we don't want to tie up
168+ -- the node's internal control channel if we can possibly help it!
169+ weakQueue <- processWeakQ <$> ask
170+ liftIO $ putMVar mv weakQueue
171+ traceLoop initialState
172+ where
173+ traceLoop :: TracerState -> Process ()
174+ traceLoop st = do
175+ -- Trace events are forwarded to the trace target when tracing is enabled.
176+ -- At some point in the future, we're going to start writing these custom
177+ -- events to the ghc eventlog, at which point this design might change.
178+ st' <- receiveWait [
179+ -- we notify the previous tracer process that it has been replaced
180+ match (\(set :: SetTrace) ->
181+ -- We allow at most one trace target, which is a process id.
182+ case set of
183+ (TraceEnable pid) -> do
184+ sendTrace st (createUnencodedMessage (TraceEvTakeover pid))
185+ return st { sendTrace = (mkSender pid) }
186+ TraceDisable -> do
187+ sendTrace st (createUnencodedMessage TraceEvDisable)
188+ return st { sendTrace = (\_ -> return ()) })
189+ , match (\flags' -> applyTraceFlags flags' st)
190+ -- we dequeue incoming messages even if we don't process them
191+ , matchAny (\ev ->
192+ handleMessage ev (handleTrace st ev) >>= return . fromMaybe st)
193+ ]
194+ traceLoop st'
195+
196+ mkSender :: ProcessId -> (Message -> Process ())
197+ mkSender pid = (flip forward) pid
198+
199+ initialState :: TracerState
200+ initialState =
201+ TracerST
202+ { sendTrace = (\_ -> return ())
203+ , flags = defaultTraceFlags
204+ , regNames = Map.empty
205+ }
206+
207+-- node [runtime tracer control]
208+--
209+-- The trace mechanism is designed to put as little stress on the system, and
210+-- in particular the node controller, as possible. The LocalNode's @tracer@
211+-- field is therefore immutable, since we don't want to be blocking the node
212+-- controller when writing trace information out. The runtime cost of enabling
213+-- tracing is therefore, the cost of enqueue for all trace-able operations at
214+-- all times. If tracing is not explicitly enabled, the trace record stored in
215+-- the local node state is matched and ignored in API calls, reducing the cost
216+-- to a single function call.
217+--
218+-- To /disable/ tracing facilities in a runtime system that has tracing enabled
219+-- then, is to instruct the tracer process not to forward any trace event data
220+-- and it is not possible to remove the runtime overhead (of enqueue in the
221+-- caller and dequeue + noop in the tracer process) once the node is started.
222+
223+applyTraceFlags :: TraceFlags -> TracerState -> Process TracerState
224+applyTraceFlags flags' state = return state { flags = flags' }
225+
226+handleTrace :: TracerState -> Message -> TraceEvent -> Process TracerState
227+handleTrace st _ (TraceEvRegistered p n) =
228+ let regNames' =
229+ Map.insertWith (\_ ns -> Set.insert n ns) p
230+ (Set.singleton n)
231+ (regNames st)
232+ in return st { regNames = regNames' }
233+handleTrace st _ (TraceEvUnRegistered p n) =
234+ let f ns = case ns of
235+ Nothing -> Nothing
236+ Just ns' -> Just (Set.delete n ns')
237+ regNames' = Map.alter f p (regNames st)
238+ in return st { regNames = regNames' }
239+handleTrace st msg ev@(TraceEvSpawned _) =
240+ traceEv ev msg (traceSpawned (flags st)) st >> return st
241+handleTrace st msg ev@(TraceEvDied _ _) =
242+ traceEv ev msg (traceDied (flags st)) st >> return st
243+handleTrace st msg ev@(TraceEvSent _ _ _) =
244+ traceEv ev msg (traceSend (flags st)) st >> return st
245+handleTrace st msg ev@(TraceEvReceived _ _) =
246+ traceEv ev msg (traceRecv (flags st)) st >> return st
247+handleTrace st msg ev = do
248+ case ev of
249+ (TraceEvNodeDied _ _) ->
250+ case (traceNodes (flags st)) of
251+ True -> (sendTrace st) msg
252+ False -> return ()
253+ (TraceEvUser _) ->
254+ (sendTrace st) msg
255+ _ ->
256+ case (traceConnections (flags st)) of
257+ True -> (sendTrace st) msg
258+ False -> return ()
259+ return st
260+
261+traceEv :: TraceEvent
262+ -> Message
263+ -> Maybe TraceSubject
264+ -> TracerState
265+ -> Process ()
266+traceEv _ _ Nothing _ = return ()
267+traceEv _ msg (Just TraceAll) st = (sendTrace st) msg
268+traceEv ev msg (Just (TraceProcs pids)) st = do
269+ node <- processNode <$> ask
270+ let p = case resolveToPid ev of
271+ Nothing -> (nullProcessId (localNodeId node))
272+ Just pid -> pid
273+ case (Set.member p pids) of
274+ True -> (sendTrace st) msg
275+ False -> return ()
276+traceEv ev msg (Just (TraceNames names)) st = do
277+ -- TODO: if we have recorded regnames for p, then we
278+ -- forward the trace iif there are overlapping trace targets
279+ node <- processNode <$> ask
280+ let p = case resolveToPid ev of
281+ Nothing -> (nullProcessId (localNodeId node))
282+ Just pid -> pid
283+ case (Map.lookup p (regNames st)) of
284+ Nothing -> return ()
285+ Just ns -> if (Set.null (Set.intersection ns names))
286+ then return ()
287+ else (sendTrace st) msg
288+
src/Control/Distributed/Process/Internal/Trace/Types.hsnew file+178
@@ -0,0 +1,178 @@
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+ , defaultTraceFlags
10+ , TraceArg(..)
11+ ) where
12+
13+import Control.Applicative ((<$>), (<*>))
14+import Control.Distributed.Process.Internal.Types
15+ ( Tracer(..)
16+ , NodeId
17+ , ProcessId
18+ , DiedReason
19+ , Message
20+ )
21+import Data.Binary
22+import Data.Set (Set)
23+import Data.Typeable
24+import Network.Transport
25+ ( ConnectionId
26+ )
27+
28+--------------------------------------------------------------------------------
29+-- Types --
30+--------------------------------------------------------------------------------
31+
32+data SetTrace = TraceEnable !ProcessId | TraceDisable
33+ deriving (Typeable, Eq, Show)
34+
35+-- | An event that is fired when something /interesting/ happens.
36+-- Trace events are forwarded to a trace listener process, of which
37+-- at most one can be registered per-node.
38+data TraceEvent =
39+ TraceEvSpawned ProcessId
40+ -- ^ fired whenever a local process is spawned
41+ | TraceEvRegistered ProcessId String
42+ -- ^ fired whenever a process/name is registered (locally)
43+ | TraceEvUnRegistered ProcessId String
44+ -- ^ fired whenever a process/name is unregistered (locally)
45+ | TraceEvDied ProcessId DiedReason
46+ -- ^ fired whenever a process dies
47+ | TraceEvNodeDied NodeId DiedReason
48+ -- ^ fired whenever a node /dies/ (i.e., the connection is broken/disconnected)
49+ | TraceEvSent ProcessId ProcessId Message
50+ -- ^ fired whenever a message is sent from a local process
51+ | TraceEvReceived ProcessId Message
52+ -- ^ fired whenever a message is received by a local process
53+ | TraceEvConnected ConnectionId
54+ -- ^ fired when a network-transport connection is first established
55+ | TraceEvDisconnected ConnectionId
56+ -- ^ fired when a network-transport connection is broken/disconnected
57+ | TraceEvUser Message
58+ -- ^ a user defined trace event (see 'traceMessage')
59+ | TraceEvLog String
60+ -- ^ a /logging/ event - used for debugging purposes only
61+ | TraceEvTakeover ProcessId
62+ -- ^ notifies a trace listener that all subsequent traces will be sent to /pid/
63+ | TraceEvDisable
64+ -- ^ notifies a trace listener that it has been disabled/removed
65+ deriving (Typeable, Show)
66+
67+class Addressable a where
68+ resolveToPid :: a -> Maybe ProcessId
69+
70+instance Addressable TraceEvent where
71+ resolveToPid (TraceEvSpawned p) = Just p
72+ resolveToPid (TraceEvDied p _) = Just p
73+ resolveToPid (TraceEvSent _ p _) = Just p
74+ resolveToPid (TraceEvReceived p _) = Just p
75+ resolveToPid _ = Nothing
76+
77+-- | Defines which processes will be traced by a given 'TraceFlag',
78+-- either by name, or @ProcessId@. Choosing @TraceAll@ is /by far/
79+-- the most efficient approach, as the tracer process therefore
80+-- avoids deciding whether or not a trace event is viable.
81+--
82+data TraceSubject =
83+ TraceAll -- enable tracing for all running processes
84+ | TraceProcs !(Set ProcessId) -- enable tracing for a set of processes
85+ | TraceNames !(Set String) -- enable tracing for a set of named/registered processes
86+
87+-- | Defines /what/ will be traced. Flags that control tracing of
88+-- @Process@ events, take a 'TraceSubject' controlling which processes
89+-- should generate trace events in the target process.
90+data TraceFlags = TraceFlags {
91+ traceSpawned :: !(Maybe TraceSubject) -- filter process spawned tracing
92+ , traceDied :: !(Maybe TraceSubject) -- filter process died tracing
93+ , traceSend :: !(Maybe TraceSubject) -- filter process/message tracing by sender
94+ , traceRecv :: !(Maybe TraceSubject) -- filter process/message tracing by receiver
95+ , traceNodes :: !Bool -- enable node status trace events
96+ , traceConnections :: !Bool -- enable connection status trace events
97+ } deriving (Typeable)
98+
99+defaultTraceFlags :: TraceFlags
100+defaultTraceFlags =
101+ TraceFlags {
102+ traceSpawned = Nothing
103+ , traceDied = Nothing
104+ , traceSend = Nothing
105+ , traceRecv = Nothing
106+ , traceNodes = False
107+ , traceConnections = False
108+ }
109+
110+data TraceArg =
111+ TraceStr String
112+ | forall a. (Show a) => Trace a
113+
114+--------------------------------------------------------------------------------
115+-- Binary Instances --
116+--------------------------------------------------------------------------------
117+
118+instance Binary SetTrace where
119+ put (TraceEnable pid) = putWord8 1 >> put pid
120+ put TraceDisable = putWord8 2
121+
122+ get = do
123+ header <- getWord8
124+ case header of
125+ 1 -> TraceEnable <$> get
126+ 2 -> return TraceDisable
127+ _ -> error "SetTrace.get - invalid header"
128+
129+instance Binary TraceEvent where
130+ put (TraceEvSpawned pid) = putWord8 1 >> put pid
131+ put (TraceEvRegistered pid str) = putWord8 2 >> put pid >> put str
132+ put (TraceEvUnRegistered pid str) = putWord8 3 >> put pid >> put str
133+ put (TraceEvDied pid res) = putWord8 4 >> put pid >> put res
134+ put (TraceEvNodeDied nid res) = putWord8 5 >> put nid >> put res
135+ put (TraceEvSent to from msg) = putWord8 6 >> put to >> put from >> put msg
136+ put (TraceEvReceived pid msg) = putWord8 7 >> put pid >> put msg
137+ put (TraceEvConnected cid) = putWord8 8 >> put cid
138+ put (TraceEvDisconnected cid) = putWord8 9 >> put cid
139+ put (TraceEvUser msg) = putWord8 10 >> put msg
140+ put (TraceEvLog msg) = putWord8 11 >> put msg
141+ put (TraceEvTakeover pid) = putWord8 12 >> put pid
142+ put TraceEvDisable = putWord8 13
143+
144+ get = do
145+ header <- getWord8
146+ case header of
147+ 1 -> TraceEvSpawned <$> get
148+ 2 -> TraceEvRegistered <$> get <*> get
149+ 3 -> TraceEvUnRegistered <$> get <*> get
150+ 4 -> TraceEvDied <$> get <*> get
151+ 5 -> TraceEvNodeDied <$> get <*> get
152+ 6 -> TraceEvSent <$> get <*> get <*> get
153+ 7 -> TraceEvReceived <$> get <*> get
154+ 8 -> TraceEvConnected <$> get
155+ 9 -> TraceEvDisconnected <$> get
156+ 10 -> TraceEvUser <$> get
157+ 11 -> TraceEvLog <$> get
158+ 12 -> TraceEvTakeover <$> get
159+ 13 -> return TraceEvDisable
160+ _ -> error "TraceEvent.get - invalid header"
161+
162+instance Binary TraceSubject where
163+ put TraceAll = putWord8 1
164+ put (TraceProcs pids) = putWord8 2 >> put pids
165+ put (TraceNames names) = putWord8 3 >> put names
166+
167+ get = do
168+ header <- getWord8
169+ case header of
170+ 1 -> return TraceAll
171+ 2 -> TraceProcs <$> get
172+ 3 -> TraceNames <$> get
173+ _ -> error "TraceSubject.get - invalid header"
174+
175+instance Binary TraceFlags where
176+ put (TraceFlags s d m r n c) = put s >> put d >> put m >> put r >> put n >> put c
177+ get = TraceFlags <$> get <*> get <*> get <*> get <*> get <*> get
178+
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+111−50
@@ -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,17 @@ import Control.Distributed.Process.Internal.Types
148 , firstNonReservedProcessId
149 , firstNonReservedProcessId
149 , ImplicitReconnect(WithImplicitReconnect)
150 , ImplicitReconnect(WithImplicitReconnect)
150 )
151 )
151−import Control.Distributed.Process.Internal.Trace
152+import Control.Distributed.Process.Internal.Trace.Primitives
153+ ( traceLogFmt
154+ , traceEvent
155+ )
156+import Control.Distributed.Process.Internal.Trace.Types
152 ( TraceArg(..)
157 ( TraceArg(..)
153− , traceFormat
158+ , TraceEvent(..)
154− , startTracing
159+ )
155− , stopTracer
160+import Control.Distributed.Process.Internal.Trace.Tracer
161+ ( traceController
162+ , defaultTracer
156 )
163 )
157import Control.Distributed.Process.Serializable (Serializable)
164import Control.Distributed.Process.Serializable (Serializable)
158import Control.Distributed.Process.Internal.Messaging
165import Control.Distributed.Process.Internal.Messaging
@@ -175,7 +182,7 @@ import qualified Control.Distributed.Process.Internal.StrictContainerAccessors a
175 ( mapMaybe
182 ( mapMaybe
176 , mapDefault
183 , mapDefault
177 )
184 )
178−
185+import System.Environment (getEnv)
179import Unsafe.Coerce
186import Unsafe.Coerce
180
187
181--------------------------------------------------------------------------------
188--------------------------------------------------------------------------------
@@ -232,10 +239,41 @@ createBareLocalNode endPoint rtable = do
232 , ctrlMsgSignal = SigShutdown
239 , ctrlMsgSignal = SigShutdown
233 }
240 }
234
241
242+startTracing :: LocalNode -> IO LocalNode
243+startTracing node =
244+ Exception.catch
245+ (getEnv "DISTRIBUTED_PROCESS_TRACE_ENABLED" >> procTracer node)
246+ (\(_ :: IOError) -> return node)
247+ where procTracer :: LocalNode -> IO LocalNode
248+ procTracer n = do
249+ -- see note [tracer/forkProcess races]
250+ mv <- MVar.newEmptyMVar
251+ pid <- forkProcess n (traceController mv)
252+ wqRef <- MVar.takeMVar mv
253+ return n { localTracer = (ActiveTracer pid wqRef) }
254+
255+startDefaultTracer :: LocalNode -> IO (Maybe ProcessId)
256+startDefaultTracer node =
257+ Exception.catch
258+ (getEnv "DISTRIBUTED_PROCESS_TRACE_ENABLED" >> go node)
259+ (\(_ :: IOError) -> return Nothing)
260+ where go :: LocalNode -> IO (Maybe ProcessId)
261+ go node' = do
262+ pid <- forkProcess node' defaultTracer
263+ return (Just pid)
264+
235-- | Start and register the service processes on a node
265-- | Start and register the service processes on a node
236−-- (for now, this is only the logger)
237startServiceProcesses :: LocalNode -> IO ()
266startServiceProcesses :: LocalNode -> IO ()
238startServiceProcesses node = do
267startServiceProcesses node = do
268+ -- tracing /spawns/ relies on the tracer being enabled, but we start
269+ -- the default tracer first, even though it might @nsend@ to the logger
270+ -- before /that/ process has started - this is a totally harmless race
271+ mPid <- startDefaultTracer node
272+ case mPid of
273+ Nothing -> return ()
274+ Just pid ->
275+ runProcess node (register "tracer" pid >> register "tracer.initial" pid)
276+
239 logger <- forkProcess node loop
277 logger <- forkProcess node loop
240 runProcess node $ register "logger" logger
278 runProcess node $ register "logger" logger
241 where
279 where
@@ -257,6 +295,7 @@ startServiceProcesses node = do
257-- TODO: for now we just close the associated endpoint
295-- TODO: for now we just close the associated endpoint
258closeLocalNode :: LocalNode -> IO ()
296closeLocalNode :: LocalNode -> IO ()
259closeLocalNode node =
297closeLocalNode node =
298+ -- TODO: close all our processes, surely!?
260 NT.closeEndPoint (localEndPoint node)
299 NT.closeEndPoint (localEndPoint node)
261
300
262-- | Run a process on a local node and wait for it to finish
301-- | Run a process on a local node and wait for it to finish
@@ -306,6 +345,10 @@ forkProcess node proc = modifyMVar (localState node) startProcess
306 }
345 }
307 return (tid', lproc)
346 return (tid', lproc)
308
347
348+
349+ -- see note [tracer/forkProcess races]
350+ trace node (TraceEvSpawned pid)
351+
309 if lpidCounter lpid == maxBound
352 if lpidCounter lpid == maxBound
310 then do
353 then do
311 newUnique <- randomIO
354 newUnique <- randomIO
@@ -331,6 +374,15 @@ forkProcess node proc = modifyMVar (localState node) startProcess
331 . (localConnections ^= unaffected)
374 . (localConnections ^= unaffected)
332 $ st
375 $ st
333
376
377+-- note [tracer/forkProcess races]
378+--
379+-- Our startTracing function uses forkProcess to start the trace controller
380+-- process, and of course forkProcess attempts to call traceEvent once the
381+-- process has started. This is harmless, as the localTracer is not updated
382+-- until /after/ the initial forkProcess completes, so the first call to
383+-- traceEvent behaves as if tracing were disabled (i.e., it is ignored).
384+--
385+
334--------------------------------------------------------------------------------
386--------------------------------------------------------------------------------
335-- Handle incoming messages --
387-- Handle incoming messages --
336--------------------------------------------------------------------------------
388--------------------------------------------------------------------------------
@@ -339,7 +391,7 @@ type IncomingConnection = (NT.EndPointAddress, IncomingTarget)
339
391
340data IncomingTarget =
392data IncomingTarget =
341 Uninit
393 Uninit
342− | ToProc (Weak (CQueue Message))
394+ | ToProc ProcessId (Weak (CQueue Message))
343 | ToChan TypedChannel
395 | ToChan TypedChannel
344 | ToNode
396 | ToNode
345
397
@@ -374,20 +426,24 @@ handleIncomingMessages node = go initConnectionState
374 case event of
426 case event of
375 NT.ConnectionOpened cid rel theirAddr ->
427 NT.ConnectionOpened cid rel theirAddr ->
376 if rel == NT.ReliableOrdered
428 if rel == NT.ReliableOrdered
429+ then
430+ trace node (TraceEvConnected cid)
431+ >> go (
377− then go ( (incomingAt cid ^= Just (theirAddr, Uninit))
432+ (incomingAt cid ^= Just (theirAddr, Uninit))
378 . (incomingFrom theirAddr ^: Set.insert cid)
433 . (incomingFrom theirAddr ^: Set.insert cid)
379 $ st
434 $ st
380 )
435 )
381 else invalidRequest cid st
436 else invalidRequest cid st
382 NT.Received cid payload ->
437 NT.Received cid payload ->
383 case st ^. incomingAt cid of
438 case st ^. incomingAt cid of
384− Just (_, ToProc weakQueue) -> do
439+ Just (_, ToProc pid weakQueue) -> do
385 mQueue <- deRefWeak weakQueue
440 mQueue <- deRefWeak weakQueue
386 forM_ mQueue $ \queue -> do
441 forM_ mQueue $ \queue -> do
387 -- TODO: if we find that the queue is Nothing, should we remove
442 -- TODO: if we find that the queue is Nothing, should we remove
388 -- it from the NC state? (and same for channels, below)
443 -- it from the NC state? (and same for channels, below)
389 let msg = payloadToMessage payload
444 let msg = payloadToMessage payload
390 enqueue queue msg -- 'enqueue' is strict
445 enqueue queue msg -- 'enqueue' is strict
446+ trace node (TraceEvReceived pid msg)
391 go st
447 go st
392 Just (_, ToChan (TypedChannel chan')) -> do
448 Just (_, ToChan (TypedChannel chan')) -> do
393 mChan <- deRefWeak chan'
449 mChan <- deRefWeak chan'
@@ -408,7 +464,7 @@ handleIncomingMessages node = go initConnectionState
408 mProc <- withMVar state $ return . (^. localProcessWithId lpid)
464 mProc <- withMVar state $ return . (^. localProcessWithId lpid)
409 case mProc of
465 case mProc of
410 Just proc ->
466 Just proc ->
411− go (incomingAt cid ^= Just (src, ToProc (processWeakQ proc)) $ st)
467+ go (incomingAt cid ^= Just (src, ToProc pid (processWeakQ proc)) $ st)
412 Nothing ->
468 Nothing ->
413 invalidRequest cid st
469 invalidRequest cid st
414 SendPortIdentifier chId -> do
470 SendPortIdentifier chId -> do
@@ -435,7 +491,8 @@ handleIncomingMessages node = go initConnectionState
435 case st ^. incomingAt cid of
491 case st ^. incomingAt cid of
436 Nothing ->
492 Nothing ->
437 invalidRequest cid st
493 invalidRequest cid st
438− Just (src, _) ->
494+ Just (src, _) -> do
495+ trace node (TraceEvDisconnected cid)
439 go ( (incomingAt cid ^= Nothing)
496 go ( (incomingAt cid ^= Nothing)
440 . (incomingFrom src ^: Set.delete cid)
497 . (incomingFrom src ^: Set.delete cid)
441 $ st
498 $ st
@@ -454,18 +511,15 @@ handleIncomingMessages node = go initConnectionState
454 $ st
511 $ st
455 )
512 )
456 NT.ErrorEvent (NT.TransportError NT.EventEndPointFailed str) ->
513 NT.ErrorEvent (NT.TransportError NT.EventEndPointFailed str) ->
457− fatal $ "Cloud Haskell fatal error: end point failed: " ++ str
514+ fail $ "Cloud Haskell fatal error: end point failed: " ++ str
458 NT.ErrorEvent (NT.TransportError NT.EventTransportFailed str) ->
515 NT.ErrorEvent (NT.TransportError NT.EventTransportFailed str) ->
459− fatal $ "Cloud Haskell fatal error: transport failed: " ++ str
516+ fail $ "Cloud Haskell fatal error: transport failed: " ++ str
460 NT.EndPointClosed ->
517 NT.EndPointClosed ->
461− stopTracer (localTracer node) >> return ()
518+ return ()
462 NT.ReceivedMulticast _ _ ->
519 NT.ReceivedMulticast _ _ ->
463 -- If we received a multicast message, something went horribly wrong
520 -- If we received a multicast message, something went horribly wrong
464 -- and we just give up
521 -- and we just give up
465− fatal "Cloud Haskell fatal error: received unexpected multicast"
522+ fail "Cloud Haskell fatal error: received unexpected multicast"
466−
467− fatal :: String -> IO ()
468− fatal msg = stopTracer (localTracer node) >> fail msg
469
523
470 invalidRequest :: NT.ConnectionId -> ConnectionState -> IO ()
524 invalidRequest :: NT.ConnectionId -> ConnectionState -> IO ()
471 invalidRequest cid st = do
525 invalidRequest cid st = do
@@ -473,7 +527,7 @@ handleIncomingMessages node = go initConnectionState
473 -- node. That is, we should report the remote node as having died, and we
527 -- node. That is, we should report the remote node as having died, and we
474 -- should close incoming connections (this requires a Transport layer
528 -- should close incoming connections (this requires a Transport layer
475 -- extension).
529 -- extension).
476− traceEventFmtIO node "" [(TraceStr "[network] invalid request: "),
530+ traceEventFmtIO node "" [(TraceStr " [network] invalid request: "),
477 (Trace cid)]
531 (Trace cid)]
478 go ( incomingAt cid ^= Nothing
532 go ( incomingAt cid ^= Nothing
479 $ st
533 $ st
@@ -534,23 +588,22 @@ instance Show ProcessKillException where
534
588
535traceNotifyDied :: LocalNode -> Identifier -> DiedReason -> NC ()
589traceNotifyDied :: LocalNode -> Identifier -> DiedReason -> NC ()
536traceNotifyDied node ident reason =
590traceNotifyDied node ident reason =
591+ -- TODO: sendPortDied notifications
592+ liftIO $ withLocalTracer node $ \t ->
537− case reason of
593+ case ident of
594+ (NodeIdentifier nid) -> traceEvent t (TraceEvNodeDied nid reason)
595+ (ProcessIdentifier pid) -> traceEvent t (TraceEvDied pid reason)
538− DiedNormal -> return ()
596+ _ -> 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
597
548traceEventFmtIO :: LocalNode
598traceEventFmtIO :: LocalNode
549 -> String
599 -> String
550 -> [TraceArg]
600 -> [TraceArg]
551 -> IO ()
601 -> IO ()
552traceEventFmtIO node fmt args =
602traceEventFmtIO node fmt args =
553− withLocalTracer node $ \t -> traceFormat t fmt args
603+ withLocalTracer node $ \t -> traceLogFmt t fmt args
604+
605+trace :: LocalNode -> TraceEvent -> IO ()
606+trace node ev = withLocalTracer node $ \t -> traceEvent t ev
554
607
555withLocalTracer :: LocalNode -> (Tracer -> IO ()) -> IO ()
608withLocalTracer :: LocalNode -> (Tracer -> IO ()) -> IO ()
556withLocalTracer node act = act (localTracer node)
609withLocalTracer node act = act (localTracer node)
@@ -597,7 +650,7 @@ nodeController = do
597 NCMsg _ (NamedSend label msg') ->
650 NCMsg _ (NamedSend label msg') ->
598 ncEffectNamedSend label msg'
651 ncEffectNamedSend label msg'
599 NCMsg _ (LocalSend to msg') ->
652 NCMsg _ (LocalSend to msg') ->
600− ncEffectLocalSend to msg'
653+ ncEffectLocalSend node to msg'
601 NCMsg _ (LocalPortSend to msg') ->
654 NCMsg _ (LocalPortSend to msg') ->
602 ncEffectLocalPortSend to msg'
655 ncEffectLocalPortSend to msg'
603 NCMsg (ProcessIdentifier from) (Kill to reason) ->
656 NCMsg (ProcessIdentifier from) (Kill to reason) ->
@@ -720,6 +773,7 @@ ncEffectSpawn pid cProc ref = do
720 mProc <- unClosure cProc
773 mProc <- unClosure cProc
721 -- If the closure does not exist, we spawn a process that throws an exception
774 -- 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
775 -- This allows the remote node to find out what's happening
776+ -- TODO:
723 let proc = case mProc of
777 let proc = case mProc of
724 Left err -> fail $ "Error: Could not resolve closure: " ++ err
778 Left err -> fail $ "Error: Could not resolve closure: " ++ err
725 Right p -> p
779 Right p -> p
@@ -750,6 +804,9 @@ ncEffectRegister from label atnode mPid reregistration = do
750 then do when (isOk) $
804 then do when (isOk) $
751 do modify' $ registeredHereFor label ^= mPid
805 do modify' $ registeredHereFor label ^= mPid
752 updateRemote node currentVal mPid
806 updateRemote node currentVal mPid
807+ case mPid of
808+ (Just p) -> liftIO $ trace node (TraceEvRegistered p label)
809+ Nothing -> liftIO $ trace node (TraceEvUnRegistered (fromJust currentVal) label)
753 liftIO $ sendMessage node
810 liftIO $ sendMessage node
754 (NodeIdentifier (localNodeId node))
811 (NodeIdentifier (localNodeId node))
755 (ProcessIdentifier from)
812 (ProcessIdentifier from)
@@ -772,6 +829,7 @@ ncEffectRegister from label atnode mPid reregistration = do
772 updateRemote _ _ _ = return ()
829 updateRemote _ _ _ = return ()
773 maybeify f Nothing = unmaybeify $ f []
830 maybeify f Nothing = unmaybeify $ f []
774 maybeify f (Just x) = unmaybeify $ f x
831 maybeify f (Just x) = unmaybeify $ f x
832+
775 unmaybeify [] = Nothing
833 unmaybeify [] = Nothing
776 unmaybeify x = Just x
834 unmaybeify x = Just x
777 incList [] tag = [(tag,1)]
835 incList [] tag = [(tag,1)]
@@ -812,16 +870,19 @@ ncEffectNamedSend label msg = do
812 forM_ mPid $ \pid -> postMessage pid msg
870 forM_ mPid $ \pid -> postMessage pid msg
813
871
814-- [Issue #DP-20]
872-- [Issue #DP-20]
815−ncEffectLocalSend :: ProcessId -> Message -> NC ()
873+ncEffectLocalSend :: LocalNode -> ProcessId -> Message -> NC ()
816−ncEffectLocalSend = postMessage
874+ncEffectLocalSend node to msg =
875+ liftIO $ withLocalProc node to $ \p -> do
876+ enqueue (processQueue p) msg
877+ trace node (TraceEvReceived to msg)
817
878
818-- [Issue #DP-20]
879-- [Issue #DP-20]
819ncEffectLocalPortSend :: SendPortId -> Message -> NC ()
880ncEffectLocalPortSend :: SendPortId -> Message -> NC ()
820−ncEffectLocalPortSend from msg =
881+ncEffectLocalPortSend from msg = do
882+ node <- ask
821 let pid = sendPortProcessId from
883 let pid = sendPortProcessId from
822 cid = sendPortLocalId from
884 cid = sendPortLocalId from
823− in do
824− withLocalProc pid $ \proc -> do
885+ liftIO $ withLocalProc node pid $ \proc -> do
825 mChan <- withMVar (processState proc) $ return . (^. typedChannelWithId cid)
886 mChan <- withMVar (processState proc) $ return . (^. typedChannelWithId cid)
826 case mChan of
887 case mChan of
827 -- in the unlikely event we know nothing about this channel id,
888 -- in the unlikely event we know nothing about this channel id,
@@ -995,21 +1056,21 @@ postAsMessage pid = postMessage pid . createUnencodedMessage
995
1056
996postMessage :: ProcessId -> Message -> NC ()
1057postMessage :: ProcessId -> Message -> NC ()
997postMessage pid msg = do
1058postMessage pid msg = do
1059+ node <- ask
998− withLocalProc pid $ \p -> enqueue (processQueue p) msg
1060+ liftIO $ withLocalProc node pid $ \p -> enqueue (processQueue p) msg
999
1061
1000throwException :: Exception e => ProcessId -> e -> NC ()
1062throwException :: Exception e => ProcessId -> e -> NC ()
1001−throwException pid e = withLocalProc pid $ \p ->
1063+throwException pid e = do
1002− throwTo (processThread p) e
1003−
1004−withLocalProc :: ProcessId -> (LocalProcess -> IO ()) -> NC ()
1005−withLocalProc pid p = do
1006 node <- ask
1064 node <- ask
1007− liftIO $ do
1065+ liftIO $ withLocalProc node pid $ \p -> throwTo (processThread p) e
1066+
1067+withLocalProc :: LocalNode -> ProcessId -> (LocalProcess -> IO ()) -> IO ()
1068+withLocalProc node pid p =
1008− -- By [Unified: table 6, rule missing_process] messages to dead processes
1069+ -- By [Unified: table 6, rule missing_process] messages to dead processes
1009− -- can silently be dropped
1070+ -- can silently be dropped
1010− let lpid = processLocalId pid
1071+ let lpid = processLocalId pid in do
1011− mProc <- withMVar (localState node) $ return . (^. localProcessWithId lpid)
1072+ mProc <- withMVar (localState node) $ return . (^. localProcessWithId lpid)
1012− forM_ mProc p
1073+ forM_ mProc p
1013
1074
1014--------------------------------------------------------------------------------
1075--------------------------------------------------------------------------------
1015-- Accessors --
1076-- Accessors --