2a21c07 · distributed-process

git.iko.soy/distributed-process

Trace primitives and node controller integration

commit2a21c07fea96daeb56f85d064189b3869e48e525 authorTim Watson <tim@rabbitmq.com> dateWed 27 Feb 2013, 00:34 (14 years ago) parent78192d1

10 files changed+999−241

compare with master

unifiedsplit

distributed-process.cabal+10−7

@@ -59,19 +59,22 @@ Library

5959 rank1dynamic >= 0.1 && < 0.2,

6060 syb >= 0.3 && < 0.4

6161 Exposed-modules: Control.Distributed.Process,

62+ Control.Distributed.Process.Internal.Types,

6263 Control.Distributed.Process.Serializable,

6364 Control.Distributed.Process.Closure,

64− Control.Distributed.Process.Node,

65− Control.Distributed.Process.Internal.Primitives,

65+ Control.Distributed.Process.Debug,

66+ Control.Distributed.Process.Node

67+ Other-modules: Control.Distributed.Process.Internal.Primitives,

6668 Control.Distributed.Process.Internal.CQueue,

67− Control.Distributed.Process.Internal.Types,

68− Control.Distributed.Process.Internal.Trace,

6969 Control.Distributed.Process.Internal.Closure.BuiltIn,

7070 Control.Distributed.Process.Internal.Messaging,

7171 Control.Distributed.Process.Internal.StrictList,

7272 Control.Distributed.Process.Internal.StrictMVar,

73− Control.Distributed.Process.Internal.WeakTQueue

74− Control.Distributed.Process.Internal.StrictContainerAccessors

73+ Control.Distributed.Process.Internal.WeakTQueue,

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

7578 Extensions: RankNTypes,

7679 ScopedTypeVariables,

7780 FlexibleInstances,

@@ -108,7 +111,7 @@ Test-Suite TestCH

108111 ScopedTypeVariables,

109112 DeriveDataTypeable,

110113 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

112115 HS-Source-Dirs: tests

113116

114117Test-Suite TestClosure

src/Control/Distributed/Process.hs+11−1

@@ -52,6 +52,9 @@ module Control.Distributed.Process

5252 , unwrapMessage

5353 , handleMessage

5454 , forward

55+ , delegate

56+ , relay

57+ , proxy

5558 -- * Process management

5659 , spawn

5760 , call

@@ -215,6 +218,9 @@ import Control.Distributed.Process.Internal.Primitives

215218 , unwrapMessage

216219 , handleMessage

217220 , forward

221+ , delegate

222+ , relay

223+ , proxy

218224 -- Process management

219225 , terminate

220226 , ProcessTerminationException(..)

@@ -285,7 +291,11 @@ import Control.Distributed.Process.Node (forkProcess)

285291-- 2. 'send' may block (when the system TCP buffers are full, while we are

286292-- trying to establish a connection to the remote endpoint, etc.) but its

287293-- 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.

289299--

290300-- 3. Message delivery is reliable and ordered. That means that if process A

291301-- 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

2525 , matchMessage

2626 , matchMessageIf

2727 , wrapMessage

28+ , wrapMessageLocal

2829 , unwrapMessage

2930 , handleMessage

3031 , forward

32+ , delegate

33+ , relay

34+ , proxy

3135 -- * Process management

3236 , terminate

3337 , ProcessTerminationException(..)

@@ -88,8 +92,6 @@ module Control.Distributed.Process.Internal.Primitives

8892 -- * Reconnecting

8993 , reconnect

9094 , reconnectPort

91− -- * Tracing/Debugging

92− , trace

9395 ) where

9496

9597#if ! MIN_VERSION_base(4,6,0)

@@ -176,12 +178,16 @@ import Control.Distributed.Process.Internal.Messaging

176178 , sendPayload

177179 , disconnect

178180 )

179−import qualified Control.Distributed.Process.Internal.Trace as Trace

181+import Control.Distributed.Process.Internal.Trace.Primitives

182+ ( traceEvent

183+ , TraceEvent(..)

184+ )

180185import Control.Distributed.Process.Internal.WeakTQueue

181186 ( newTQueueIO

182187 , readTQueue

183188 , mkWeakTQueue

184189 )

190+

185191import Unsafe.Coerce

186192

187193--------------------------------------------------------------------------------

@@ -191,18 +197,24 @@ import Unsafe.Coerce

191197-- | Send a message

192198send :: Serializable a => ProcessId -> a -> Process ()

193199-- 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.

195201send them msg = do

196202 proc <- ask

197− let node = localNodeId (processNode proc)

203+ let us = processId proc

204+ node = processNode proc

205+ nodeId = localNodeId node

198206 destNode = (processNodeId them) in do

199− case destNode == node of

207+ case destNode == nodeId of

200208 True -> sendLocal them msg

201209 False -> liftIO $ sendMessage (processNode proc)

202210 (ProcessIdentifier (processId proc))

203211 (ProcessIdentifier them)

204212 NoImplicitReconnect

205213 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))

206218

207219-- | Wait for a message of a specific type

208220expect :: forall a. Serializable a => Process a

@@ -358,15 +370,21 @@ matchMessageIf c p = Match $ MatchMsg $ \msg ->

358370forward :: Message -> ProcessId -> Process ()

359371forward msg them = do

360372 proc <- ask

361− let node = localNodeId (processNode proc)

373+ let node = processNode proc

374+ us = processId proc

375+ nid = localNodeId node

362376 destNode = (processNodeId them) in do

363− case destNode == node of

377+ case destNode == nid of

364378 True -> sendCtrlMsg Nothing (LocalSend them msg)

365379 False -> liftIO $ sendPayload (processNode proc)

366380 (ProcessIdentifier (processId proc))

367381 (ProcessIdentifier them)

368382 NoImplicitReconnect

369383 (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))

370388

371389

372390-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are

@@ -386,6 +404,12 @@ forward msg them = do

386404wrapMessage :: Serializable a => a -> Message

387405wrapMessage = createMessage

388406

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+

389413-- | Attempt to unwrap a raw 'Message'.

390414-- If the type of the decoded message payload matches the expected type, the

391415-- value will be returned with @Just@, otherwise @Nothing@ indicates the types

@@ -480,6 +504,34 @@ succeeds) that unsafeCoerce will not fail.

480504matchUnknown :: Process b -> Match b

481505matchUnknown p = Match $ MatchMsg (const (Just p))

482506

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+

483535--------------------------------------------------------------------------------

484536-- Process management --

485537--------------------------------------------------------------------------------

@@ -750,7 +802,7 @@ finally a sequel = bracket_ (return ()) sequel a

750802data Handler a = forall e . Exception e => Handler (e -> Process a)

751803

752804instance Functor Handler where

753− fmap f (Handler h) = Handler (fmap f . h)

805+ fmap f (Handler h) = Handler (fmap f . h)

754806

755807-- | Lift 'Control.Exception.catches'

756808catches :: Process a -> [Handler a] -> Process a

@@ -991,56 +1043,13 @@ reconnectPort them = do

9911043 node <- processNode <$> ask

9921044 liftIO $ disconnect node (ProcessIdentifier us) (SendPortIdentifier (sendPortId them))

9931045

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−

10371046--------------------------------------------------------------------------------

10381047-- Auxiliary functions --

10391048--------------------------------------------------------------------------------

10401049

10411050sendLocal :: (Serializable a) => ProcessId -> a -> Process ()

1042−sendLocal pid msg =

1043− sendCtrlMsg Nothing $ LocalSend pid (createUnencodedMessage msg)

1051+sendLocal to msg =

1052+ sendCtrlMsg Nothing $ LocalSend to (createUnencodedMessage msg)

10441053

10451054sendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()

10461055sendChanLocal 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)

9393import Control.Concurrent (ThreadId)

9494import Control.Concurrent.Chan (Chan)

9595import Control.Concurrent.STM (STM)

96−import qualified Control.Concurrent.STM as STM (TQueue)

9796import qualified Network.Transport as NT (EndPoint, EndPointAddress, Connection)

9897import Control.Applicative (Applicative, Alternative, (<$>), (<*>))

9998import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)

@@ -112,7 +111,6 @@ import Control.Distributed.Process.Internal.StrictMVar (StrictMVar)

112111import Control.Distributed.Process.Internal.WeakTQueue (TQueue)

113112import Control.Distributed.Static (RemoteTable, Closure)

114113import qualified Control.Distributed.Process.Internal.StrictContainerAccessors as DAC (mapMaybe)

115−import System.IO (Handle)

116114

117115--------------------------------------------------------------------------------

118116-- Node and process identifiers --

@@ -184,10 +182,20 @@ nullProcessId nid =

184182

185183-- | Required for system tracing in the node controller

186184data Tracer =

187− LogFileTracer !ThreadId !(STM.TQueue String) !Handle

188− | EventLogTracer !(String -> IO ())

189− | LocalNodeTracer !LocalNode

190− | InactiveTracer -- NB: never used, this is required to initialize LocalNode

185+ ActiveTracer !ProcessId !(Weak (CQueue Message)) -- note [tracer longevity]

186+ | InactiveTracer

187+

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.

191199

192200-- | Local nodes

193201data LocalNode = LocalNode

@@ -553,7 +561,7 @@ instance Binary ProcessSignal where

553561 put (NamedSend label msg) = putWord8 8 >> put label >> put (messageToPayload msg)

554562 put (Kill pid reason) = putWord8 9 >> put pid >> put reason

555563 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)

557565 put (LocalPortSend sid msg) = putWord8 12 >> put sid >> put (messageToPayload msg)

558566 put (GetInfo about) = putWord8 30 >> put about

559567 put (SigShutdown) = putWord8 31

src/Control/Distributed/Process/Node.hs+111−50

@@ -40,7 +40,7 @@ import qualified Data.Set as Set

4040 , toList

4141 )

4242import Data.Foldable (forM_)

43−import Data.Maybe (isJust, isNothing, catMaybes)

43+import Data.Maybe (isJust, fromJust, isNothing, catMaybes)

4444import Data.Typeable (Typeable)

4545import Control.Category ((>>>))

4646import Control.Applicative ((<$>))

@@ -68,6 +68,7 @@ import Control.Distributed.Process.Internal.StrictMVar

6868 , takeMVar

6969 )

7070import Control.Concurrent.Chan (newChan, writeChan, readChan)

71+import qualified Control.Concurrent.MVar as MVar (newEmptyMVar, takeMVar)

7172import Control.Concurrent.STM

7273 ( atomically

7374 )

@@ -105,7 +106,7 @@ import Control.Distributed.Process.Internal.Types

105106 , LocalProcessId(..)

106107 , ProcessId(..)

107108 , LocalNode(..)

108− , Tracer(InactiveTracer)

109+ , Tracer(..)

109110 , LocalNodeState(..)

110111 , LocalProcess(..)

111112 , LocalProcessState(..)

@@ -148,11 +149,17 @@ import Control.Distributed.Process.Internal.Types

148149 , firstNonReservedProcessId

149150 , ImplicitReconnect(WithImplicitReconnect)

150151 )

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

152157 ( TraceArg(..)

153− , traceFormat

154− , startTracing

155− , stopTracer

158+ , TraceEvent(..)

159+ )

160+import Control.Distributed.Process.Internal.Trace.Tracer

161+ ( traceController

162+ , defaultTracer

156163 )

157164import Control.Distributed.Process.Serializable (Serializable)

158165import Control.Distributed.Process.Internal.Messaging

@@ -175,7 +182,7 @@ import qualified Control.Distributed.Process.Internal.StrictContainerAccessors a

175182 ( mapMaybe

176183 , mapDefault

177184 )

178−

185+import System.Environment (getEnv)

179186import Unsafe.Coerce

180187

181188--------------------------------------------------------------------------------

@@ -232,10 +239,41 @@ createBareLocalNode endPoint rtable = do

232239 , ctrlMsgSignal = SigShutdown

233240 }

234241

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+

235265-- | Start and register the service processes on a node

236−-- (for now, this is only the logger)

237266startServiceProcesses :: LocalNode -> IO ()

238267startServiceProcesses 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+

239277 logger <- forkProcess node loop

240278 runProcess node $ register "logger" logger

241279 where

@@ -257,6 +295,7 @@ startServiceProcesses node = do

257295-- TODO: for now we just close the associated endpoint

258296closeLocalNode :: LocalNode -> IO ()

259297closeLocalNode node =

298+ -- TODO: close all our processes, surely!?

260299 NT.closeEndPoint (localEndPoint node)

261300

262301-- | Run a process on a local node and wait for it to finish

@@ -306,6 +345,10 @@ forkProcess node proc = modifyMVar (localState node) startProcess

306345 }

307346 return (tid', lproc)

308347

348+

349+ -- see note [tracer/forkProcess races]

350+ trace node (TraceEvSpawned pid)

351+

309352 if lpidCounter lpid == maxBound

310353 then do

311354 newUnique <- randomIO

@@ -331,6 +374,15 @@ forkProcess node proc = modifyMVar (localState node) startProcess

331374 . (localConnections ^= unaffected)

332375 $ st

333376

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+

334386--------------------------------------------------------------------------------

335387-- Handle incoming messages --

336388--------------------------------------------------------------------------------

@@ -339,7 +391,7 @@ type IncomingConnection = (NT.EndPointAddress, IncomingTarget)

339391

340392data IncomingTarget =

341393 Uninit

342− | ToProc (Weak (CQueue Message))

394+ | ToProc ProcessId (Weak (CQueue Message))

343395 | ToChan TypedChannel

344396 | ToNode

345397

@@ -374,20 +426,24 @@ handleIncomingMessages node = go initConnectionState

374426 case event of

375427 NT.ConnectionOpened cid rel theirAddr ->

376428 if rel == NT.ReliableOrdered

377− then go ( (incomingAt cid ^= Just (theirAddr, Uninit))

429+ then

430+ trace node (TraceEvConnected cid)

431+ >> go (

432+ (incomingAt cid ^= Just (theirAddr, Uninit))

378433 . (incomingFrom theirAddr ^: Set.insert cid)

379434 $ st

380435 )

381436 else invalidRequest cid st

382437 NT.Received cid payload ->

383438 case st ^. incomingAt cid of

384− Just (_, ToProc weakQueue) -> do

439+ Just (_, ToProc pid weakQueue) -> do

385440 mQueue <- deRefWeak weakQueue

386441 forM_ mQueue $ \queue -> do

387442 -- TODO: if we find that the queue is Nothing, should we remove

388443 -- it from the NC state? (and same for channels, below)

389444 let msg = payloadToMessage payload

390445 enqueue queue msg -- 'enqueue' is strict

446+ trace node (TraceEvReceived pid msg)

391447 go st

392448 Just (_, ToChan (TypedChannel chan')) -> do

393449 mChan <- deRefWeak chan'

@@ -408,7 +464,7 @@ handleIncomingMessages node = go initConnectionState

408464 mProc <- withMVar state $ return . (^. localProcessWithId lpid)

409465 case mProc of

410466 Just proc ->

411− go (incomingAt cid ^= Just (src, ToProc (processWeakQ proc)) $ st)

467+ go (incomingAt cid ^= Just (src, ToProc pid (processWeakQ proc)) $ st)

412468 Nothing ->

413469 invalidRequest cid st

414470 SendPortIdentifier chId -> do

@@ -435,7 +491,8 @@ handleIncomingMessages node = go initConnectionState

435491 case st ^. incomingAt cid of

436492 Nothing ->

437493 invalidRequest cid st

438− Just (src, _) ->

494+ Just (src, _) -> do

495+ trace node (TraceEvDisconnected cid)

439496 go ( (incomingAt cid ^= Nothing)

440497 . (incomingFrom src ^: Set.delete cid)

441498 $ st

@@ -454,18 +511,15 @@ handleIncomingMessages node = go initConnectionState

454511 $ st

455512 )

456513 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

458515 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

460517 NT.EndPointClosed ->

461− stopTracer (localTracer node) >> return ()

518+ return ()

462519 NT.ReceivedMulticast _ _ ->

463520 -- If we received a multicast message, something went horribly wrong

464521 -- and we just give up

465− fatal "Cloud Haskell fatal error: received unexpected multicast"

466−

467− fatal :: String -> IO ()

468− fatal msg = stopTracer (localTracer node) >> fail msg

522+ fail "Cloud Haskell fatal error: received unexpected multicast"

469523

470524 invalidRequest :: NT.ConnectionId -> ConnectionState -> IO ()

471525 invalidRequest cid st = do

@@ -473,7 +527,7 @@ handleIncomingMessages node = go initConnectionState

473527 -- node. That is, we should report the remote node as having died, and we

474528 -- should close incoming connections (this requires a Transport layer

475529 -- extension).

476− traceEventFmtIO node "" [(TraceStr "[network] invalid request: "),

530+ traceEventFmtIO node "" [(TraceStr " [network] invalid request: "),

477531 (Trace cid)]

478532 go ( incomingAt cid ^= Nothing

479533 $ st

@@ -534,23 +588,22 @@ instance Show ProcessKillException where

534588

535589traceNotifyDied :: LocalNode -> Identifier -> DiedReason -> NC ()

536590traceNotifyDied node ident reason =

537− case reason of

538− DiedNormal -> 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

591+ -- TODO: sendPortDied notifications

592+ liftIO $ withLocalTracer node $ \t ->

593+ case ident of

594+ (NodeIdentifier nid) -> traceEvent t (TraceEvNodeDied nid reason)

595+ (ProcessIdentifier pid) -> traceEvent t (TraceEvDied pid reason)

596+ _ -> return ()

547597

548598traceEventFmtIO :: LocalNode

549599 -> String

550600 -> [TraceArg]

551601 -> IO ()

552602traceEventFmtIO 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

554607

555608withLocalTracer :: LocalNode -> (Tracer -> IO ()) -> IO ()

556609withLocalTracer node act = act (localTracer node)

@@ -597,7 +650,7 @@ nodeController = do

597650 NCMsg _ (NamedSend label msg') ->

598651 ncEffectNamedSend label msg'

599652 NCMsg _ (LocalSend to msg') ->

600− ncEffectLocalSend to msg'

653+ ncEffectLocalSend node to msg'

601654 NCMsg _ (LocalPortSend to msg') ->

602655 ncEffectLocalPortSend to msg'

603656 NCMsg (ProcessIdentifier from) (Kill to reason) ->

@@ -720,6 +773,7 @@ ncEffectSpawn pid cProc ref = do

720773 mProc <- unClosure cProc

721774 -- If the closure does not exist, we spawn a process that throws an exception

722775 -- This allows the remote node to find out what's happening

776+ -- TODO:

723777 let proc = case mProc of

724778 Left err -> fail $ "Error: Could not resolve closure: " ++ err

725779 Right p -> p

@@ -750,6 +804,9 @@ ncEffectRegister from label atnode mPid reregistration = do

750804 then do when (isOk) $

751805 do modify' $ registeredHereFor label ^= mPid

752806 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)

753810 liftIO $ sendMessage node

754811 (NodeIdentifier (localNodeId node))

755812 (ProcessIdentifier from)

@@ -772,6 +829,7 @@ ncEffectRegister from label atnode mPid reregistration = do

772829 updateRemote _ _ _ = return ()

773830 maybeify f Nothing = unmaybeify $ f []

774831 maybeify f (Just x) = unmaybeify $ f x

832+

775833 unmaybeify [] = Nothing

776834 unmaybeify x = Just x

777835 incList [] tag = [(tag,1)]

@@ -812,16 +870,19 @@ ncEffectNamedSend label msg = do

812870 forM_ mPid $ \pid -> postMessage pid msg

813871

814872-- [Issue #DP-20]

815−ncEffectLocalSend :: ProcessId -> Message -> NC ()

816−ncEffectLocalSend = postMessage

873+ncEffectLocalSend :: LocalNode -> ProcessId -> Message -> NC ()

874+ncEffectLocalSend node to msg =

875+ liftIO $ withLocalProc node to $ \p -> do

876+ enqueue (processQueue p) msg

877+ trace node (TraceEvReceived to msg)

817878

818879-- [Issue #DP-20]

819880ncEffectLocalPortSend :: SendPortId -> Message -> NC ()

820−ncEffectLocalPortSend from msg =

881+ncEffectLocalPortSend from msg = do

882+ node <- ask

821883 let pid = sendPortProcessId from

822884 cid = sendPortLocalId from

823− in do

824− withLocalProc pid $ \proc -> do

885+ liftIO $ withLocalProc node pid $ \proc -> do

825886 mChan <- withMVar (processState proc) $ return . (^. typedChannelWithId cid)

826887 case mChan of

827888 -- in the unlikely event we know nothing about this channel id,

@@ -995,21 +1056,21 @@ postAsMessage pid = postMessage pid . createUnencodedMessage

9951056

9961057postMessage :: ProcessId -> Message -> NC ()

9971058postMessage pid msg = do

998− withLocalProc pid $ \p -> enqueue (processQueue p) msg

1059+ node <- ask

1060+ liftIO $ withLocalProc node pid $ \p -> enqueue (processQueue p) msg

9991061

10001062throwException :: Exception e => ProcessId -> e -> NC ()

1001−throwException pid e = withLocalProc pid $ \p ->

1002− throwTo (processThread p) e

1003−

1004−withLocalProc :: ProcessId -> (LocalProcess -> IO ()) -> NC ()

1005−withLocalProc pid p = do

1063+throwException pid e = do

10061064 node <- ask

1007− liftIO $ do

1008− -- By [Unified: table 6, rule missing_process] messages to dead processes

1009− -- can silently be dropped

1010− let lpid = processLocalId pid

1011− mProc <- withMVar (localState node) $ return . (^. localProcessWithId lpid)

1012− forM_ mProc p

1065+ liftIO $ withLocalProc node pid $ \p -> throwTo (processThread p) e

1066+

1067+withLocalProc :: LocalNode -> ProcessId -> (LocalProcess -> IO ()) -> IO ()

1068+withLocalProc node pid p =

1069+ -- By [Unified: table 6, rule missing_process] messages to dead processes

1070+ -- can silently be dropped

1071+ let lpid = processLocalId pid in do

1072+ mProc <- withMVar (localState node) $ return . (^. localProcessWithId lpid)

1073+ forM_ mProc p

10131074

10141075--------------------------------------------------------------------------------

10151076-- Accessors --