13377cf · distributed-process

git.iko.soy/distributed-process

Merge branch 'tracing' into development. Provides a new tracing API.

Configuration via environment variables is still supported. See the Control.Distributed.Process.Debug module for documentation.

commit13377cf3be017306f62241ef11b07a8f31414bef authorTim Watson <tim@rabbitmq.com> dateThu 9 May 2013, 17:38 (13 years ago) parent78192d1 3178dec

16 files changed+2015−380

compare with master

unifiedsplit

distributed-process.cabal+36−7

@@ -59,19 +59,24 @@ 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,

68+ Control.Distributed.Process.Internal.Spawn,

6669 Control.Distributed.Process.Internal.CQueue,

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

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

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

7071 Control.Distributed.Process.Internal.Messaging,

7172 Control.Distributed.Process.Internal.StrictList,

7273 Control.Distributed.Process.Internal.StrictMVar,

73− Control.Distributed.Process.Internal.WeakTQueue

74− Control.Distributed.Process.Internal.StrictContainerAccessors

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

75+ Control.Distributed.Process.Internal.StrictContainerAccessors,

76+ Control.Distributed.Process.Internal.Trace.Types,

77+ Control.Distributed.Process.Internal.Trace.Primitives,

78+ Control.Distributed.Process.Internal.Trace.Tracer

79+ Control.Distributed.Process.Internal.Trace.Remote

7580 Extensions: RankNTypes,

7681 ScopedTypeVariables,

7782 FlexibleInstances,

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

108113 ScopedTypeVariables,

109114 DeriveDataTypeable,

110115 GeneralizedNewtypeDeriving

111− ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind

116+ ghc-options: -Wall -threaded -debug -eventlog -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind

112117 HS-Source-Dirs: tests

113118

114119Test-Suite TestClosure

@@ -154,6 +159,30 @@ Test-Suite TestStats

154159 ghc-options: -Wall -debug -eventlog -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind

155160 HS-Source-Dirs: tests

156161

162+Test-Suite TestTracing

163+ Type: exitcode-stdio-1.0

164+ Main-Is: TestTracing.hs

165+ Build-Depends: base >= 4.4 && < 5,

166+ random >= 1.0 && < 1.1,

167+ ansi-terminal >= 0.5 && < 0.6,

168+ containers >= 0.4 && < 0.6,

169+ stm >= 2.4 && < 2.5,

170+ distributed-process,

171+ network-transport >= 0.3 && < 0.4,

172+ network-transport-tcp >= 0.3 && < 0.4,

173+ binary >= 0.5 && < 0.7,

174+ network >= 2.3 && < 2.5,

175+ HUnit >= 1.2 && < 1.3,

176+ test-framework >= 0.6 && < 0.9,

177+ test-framework-hunit >= 0.2.0 && < 0.4,

178+ unix >= 2.5.0.0

179+ Extensions: CPP,

180+ ScopedTypeVariables,

181+ DeriveDataTypeable,

182+ GeneralizedNewtypeDeriving

183+ ghc-options: -Wall -debug -eventlog -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind

184+ HS-Source-Dirs: tests

185+

157186

158187Executable distributed-process-throughput

159188 if flag(benchmarks)

src/Control/Distributed/Process.hs+24−129

@@ -2,7 +2,8 @@

33This is an implementation of Cloud Haskell, as described in

44/Towards Haskell in the Cloud/ by Jeff Epstein, Andrew Black, and Simon

5−Peyton Jones (<http://research.microsoft.com/en-us/um/people/simonpj/papers/parallel/>),

5+Peyton Jones (see

6+<http://research.microsoft.com/en-us/um/people/simonpj/papers/parallel/>),

67although some of the details are different. The precise message passing

78semantics are based on /A unified semantics for future Erlang/ by Hans

89Svensson, Lars-Åke Fredlund and Clara Benac Earle.

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

5253 , unwrapMessage

5354 , handleMessage

5455 , forward

56+ , delegate

57+ , relay

58+ , proxy

5559 -- * Process management

5660 , spawn

5761 , call

@@ -139,7 +143,6 @@ module Control.Distributed.Process

139143import Prelude hiding (catch)

140144#endif

141145

142−import Data.Typeable (Typeable)

143146import Control.Monad.IO.Class (liftIO)

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

145148import Control.Monad.Reader (ask)

@@ -149,8 +152,6 @@ import Control.Distributed.Static

149152 , closure

150153 , Static

151154 , RemoteTable

152− , closureCompose

153− , staticClosure

154155 )

155156import Control.Distributed.Process.Internal.Types

156157 ( NodeId(..)

@@ -174,21 +175,8 @@ import Control.Distributed.Process.Internal.Types

174175 , RegisterReply(..)

175176 , LocalProcess(processNode)

176177 , Message

177− , nullProcessId

178− )

179−import Control.Distributed.Process.Serializable (Serializable, SerializableDict)

180−import Control.Distributed.Process.Internal.Closure.BuiltIn

181− ( sdictSendPort

182− , sndStatic

183− , idCP

184− , seqCP

185− , bindCP

186− , splitCP

187− , cpLink

188− , cpSend

189− , cpNewChan

190− , cpDelay

191178 )

179+import Control.Distributed.Process.Serializable (Serializable)

192180import Control.Distributed.Process.Internal.Primitives

193181 ( -- Basic messaging

194182 send

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

215203 , unwrapMessage

216204 , handleMessage

217205 , forward

206+ , delegate

207+ , relay

208+ , proxy

218209 -- Process management

219210 , terminate

220211 , ProcessTerminationException(..)

@@ -274,6 +265,15 @@ import Control.Distributed.Process.Internal.Primitives

274265 , reconnectPort

275266 )

276267import Control.Distributed.Process.Node (forkProcess)

268+import Control.Distributed.Process.Internal.Spawn

269+ ( -- Spawning Processes/Channels

270+ spawn

271+ , spawnLink

272+ , spawnMonitor

273+ , spawnChannel

274+ , spawnSupervised

275+ , call

276+ )

277277

278278-- INTERNAL NOTES

279279--

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

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

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

287287-- return does not imply that the remote process received the message (much

288−-- less processed it)

288+-- less processed it). When 'send' targets a local process, it is dispatched

289+-- via the node controller however, in which cases it will /not/ block. This

290+-- isn't part of the semantics, so it should be ok. The ordering is maintained

291+-- because the ctrlChannel still has FIFO semantics with regards interactions

292+-- between two disparate forkIO threads.

289293--

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

291295-- sends messages m1, m2, m3 to process B, B will either arrive all three

@@ -334,116 +338,6 @@ import Control.Distributed.Process.Node (forkProcess)

334338-- [6] "Delivery of Messages", post on erlang-questions

335339-- http://erlang.org/pipermail/erlang-questions/2012-February/064767.html

336340

337−--------------------------------------------------------------------------------

338−-- Primitives are defined in a separate module; here we only define derived --

339−-- constructs --

340−--------------------------------------------------------------------------------

341−

342−-- | Spawn a process

343−--

344−-- For more information about 'Closure', see

345−-- "Control.Distributed.Process.Closure".

346−--

347−-- See also 'call'.

348−spawn :: NodeId -> Closure (Process ()) -> Process ProcessId

349−spawn nid proc = do

350− us <- getSelfPid

351− mRef <- monitorNode nid

352− sRef <- spawnAsync nid (cpDelay us proc)

353− receiveWait [

354− matchIf (\(DidSpawn ref _) -> ref == sRef) $ \(DidSpawn _ pid) -> do

355− unmonitor mRef

356− send pid ()

357− return pid

358− , matchIf (\(NodeMonitorNotification ref _ _) -> ref == mRef) $ \_ ->

359− return (nullProcessId nid)

360− ]

361−

362−-- | Spawn a process and link to it

363−--

364−-- Note that this is just the sequential composition of 'spawn' and 'link'.

365−-- (The "Unified" semantics that underlies Cloud Haskell does not even support

366−-- a synchronous link operation)

367−spawnLink :: NodeId -> Closure (Process ()) -> Process ProcessId

368−spawnLink nid proc = do

369− pid <- spawn nid proc

370− link pid

371− return pid

372−

373−-- | Like 'spawnLink', but monitor the spawned process

374−spawnMonitor :: NodeId -> Closure (Process ()) -> Process (ProcessId, MonitorRef)

375−spawnMonitor nid proc = do

376− pid <- spawn nid proc

377− ref <- monitor pid

378− return (pid, ref)

379−

380−-- | Run a process remotely and wait for it to reply

381−--

382−-- We monitor the remote process: if it dies before it can send a reply, we die

383−-- too.

384−--

385−-- For more information about 'Static', 'SerializableDict', and 'Closure', see

386−-- "Control.Distributed.Process.Closure".

387−--

388−-- See also 'spawn'.

389−call :: Serializable a

390− => Static (SerializableDict a)

391− -> NodeId

392− -> Closure (Process a)

393− -> Process a

394−call dict nid proc = do

395− us <- getSelfPid

396− (pid, mRef) <- spawnMonitor nid (proc `bindCP` cpSend dict us)

397− -- We are guaranteed to receive the reply before the monitor notification

398− -- (if a reply is sent at all)

399− -- NOTE: This might not be true if we switch to unreliable delivery.

400− mResult <- receiveWait

401− [ match (return . Right)

402− , matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mRef)

403− (\(ProcessMonitorNotification _ _ reason) -> return (Left reason))

404− ]

405− case mResult of

406− Right a -> do

407− -- Wait for the monitor message so that we the mailbox doesn't grow

408− receiveWait

409− [ matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mRef)

410− (\(ProcessMonitorNotification {}) -> return ())

411− ]

412− -- Clean up connection to pid

413− reconnect pid

414− return a

415− Left err ->

416− fail $ "call: remote process died: " ++ show err

417−

418−-- | Spawn a child process, have the child link to the parent and the parent

419−-- monitor the child

420−spawnSupervised :: NodeId

421− -> Closure (Process ())

422− -> Process (ProcessId, MonitorRef)

423−spawnSupervised nid proc = do

424− us <- getSelfPid

425− them <- spawn nid (cpLink us `seqCP` proc)

426− ref <- monitor them

427− return (them, ref)

428−

429−-- | Spawn a new process, supplying it with a new 'ReceivePort' and return

430−-- the corresponding 'SendPort'.

431−spawnChannel :: forall a. Typeable a => Static (SerializableDict a)

432− -> NodeId

433− -> Closure (ReceivePort a -> Process ())

434− -> Process (SendPort a)

435−spawnChannel dict nid proc = do

436− us <- getSelfPid

437− _ <- spawn nid (go us)

438− expect

439− where

440− go :: ProcessId -> Closure (Process ())

441− go pid = cpNewChan dict

442− `bindCP`

443− (cpSend (sdictSendPort dict) pid `splitCP` proc)

444− `bindCP`

445− (idCP `closureCompose` staticClosure sndStatic)

446−

447341--------------------------------------------------------------------------------

448342-- Local versions of spawn --

449343--------------------------------------------------------------------------------

@@ -470,3 +364,4 @@ spawnChannelLocal proc = do

470364 liftIO $ putMVar mvar sport

471365 proc rport

472366 takeMVar mvar

367+

src/Control/Distributed/Process/Closure.hs+2

@@ -173,6 +173,7 @@ module Control.Distributed.Process.Closure

173173 -- * CP versions of Cloud Haskell primitives

174174 , cpLink

175175 , cpUnlink

176+ , cpRelay

176177 , cpSend

177178 , cpExpect

178179 , cpNewChan

@@ -205,6 +206,7 @@ import Control.Distributed.Process.Internal.Closure.BuiltIn

205206 -- CP versions of Cloud Haskell primitives

206207 , cpLink

207208 , cpUnlink

209+ , cpRelay

208210 , cpSend

209211 , cpExpect

210212 , cpNewChan

src/Control/Distributed/Process/Debug.hsnew file+280

@@ -0,0 +1,280 @@

1+-----------------------------------------------------------------------------

2+-- |

3+-- Module : Control.Distributed.Process.Debug

4+-- Copyright : (c) Well-Typed / Tim Watson

5+-- License : BSD3 (see the file LICENSE)

6+--

7+-- Maintainer : Tim Watson <watson.timothy@gmail.com>

8+-- Stability : experimental

9+-- Portability : non-portable (requires concurrency)

10+--

11+-- [Tracing/Debugging Facilities]

12+--

13+-- Cloud Haskell provides a general purpose tracing mechanism, allowing a

14+-- user supplied /tracer process/ to receive messages when certain classes of

15+-- system events occur. It's possible to use this facility to aid in debugging

16+-- and/or perform other diagnostic tasks to a program at runtime.

17+--

18+-- [Enabling Tracing]

19+--

20+-- Tracing is disabled by default, because it carries a relatively small, but

21+-- tangible cost. To enable tracing, the environment variable

22+-- @DISTRIBUTED_PROCESS_TRACE_ENABLED@ must be set to some value (actual content

23+-- is ignored). When the environment variable /is/ set, the system will

24+-- generate /trace events/ in various circumstances - see 'TraceEvent' for a

25+-- list of all the published events. A user can additionally publish custom

26+-- trace events in the form of 'TraceEvLog' log messages or pass custom

27+-- (i.e., completely user defined) event data using the 'traceMessage' function.

28+-- If the environment variable is not set, no trace events will ever be

29+-- published.

30+--

31+-- All published traces are forwarded to a /tracer process/, which can be

32+-- specified (and changed) at runtime using 'traceEnable'. Some pre-defined

33+-- tracer processes are provided for conveniently printing to stderr, a log file

34+-- or the GHC eventlog.

35+--

36+-- If a tracer process crashes, no attempt is made to restart it.

37+--

38+-- [Working with multiple tracer processes]

39+--

40+-- The system tracing facility only ever writes to a single tracer process. This

41+-- invariant insulates the tracer controller and ensures a fast path for

42+-- handling all trace events. /This/ module provides facilities for layering

43+-- trace handlers using Cloud Haskell's built-in delegation primitives.

44+--

45+-- The 'startTracer' function wraps the registered @tracer@ process with the

46+-- supplied handler and also forwarding trace events to the original tracer.

47+-- The corresponding 'stopTracer' function terminates tracer processes in

48+-- reverse of the order in which they were started, and re-establishes the

49+-- previous tracer process.

50+--

51+-- [Built in tracers]

52+--

53+-- The built in tracers provide a simple /logging/ facility that writes trace

54+-- events out to either a log file, @stderr@ or the GHC eventlog. These tracers

55+-- can be configured using environment variables, or specified manually using

56+-- the 'traceEnable' function. The base tracer process cannot be

57+--

58+-- When a new local node is started, the contents of several environment

59+-- variables are checked to determine how the default tracer process

60+-- will handle trace events. If none of these variables is set, then the

61+-- trace events will be effectively ignored, although they will still be

62+-- generated and passed through the system. Only one configuration will be

63+-- chosen - the first that contains a (valid) value. These environment

64+-- variables, in the order they're checked, are:

65+--

66+-- 1. @DISTRIBUTED_PROCESS_TRACE_FILE@

67+-- This is checked for a valid file path. If it exists and the file can be

68+-- opened for writing, all trace output will be directed thence. If the supplied

69+-- path is invalid, or the file is unavailable for writing - e.g., because

70+-- another node has already started tracing to it - then this tracer will be

71+-- disabled.

72+--

73+-- 2. @DISTRIBUTED_PROCESS_TRACE_CONSOLE@

74+-- This is checked for /any/ non-empty value. If set, then all trace output will

75+-- be directed to the system logger process.

76+--

77+-- 3. @DISTRIBUTED_PROCESS_TRACE_EVENTLOG@

78+-- This is checked for /any/ non-empty value. If set, all internal traces are

79+-- written to the GHC eventlog.

80+--

81+-- Users of the /simplelocalnet/ Cloud Haskell backend should also note that

82+-- because the trace file option only supports trace output from a single node

83+-- (so as to avoid interleaving), a file trace configured for the master node

84+-- will prevent slaves from tracing to the file and will need to fall back to

85+-- the console or eventlog tracers instead.

86+--

87+-- Support for writing to the eventlog requires specific intervention to work,

88+-- without which, written traces are silently dropped/ignored and no output will

89+-- be generated. The GHC eventlog documentation provides information about

90+-- enabling, viewing and working with event traces at

91+-- <http://hackage.haskell.org/trac/ghc/wiki/EventLog>.

92+--

93+module Control.Distributed.Process.Debug

94+ ( -- * Exported Data Types

95+ TraceArg(..)

96+ , TraceEvent(..)

97+ , TraceFlags(..)

98+ , TraceSubject(..)

99+ -- * Configuring Tracing

100+ , isTracingEnabled

101+ , enableTrace

102+ , enableTraceAsync

103+ , disableTrace

104+ , withTracer

105+ , withFlags

106+ , getTraceFlags

107+ , setTraceFlags

108+ , setTraceFlagsAsync

109+ , defaultTraceFlags

110+ , traceOn

111+ , traceOnly

112+ , traceOff

113+ -- * Debugging

114+ , startTracer

115+ , stopTracer

116+ -- * Sending Custom Trace Data

117+ , traceLog

118+ , traceLogFmt

119+ , traceMessage

120+ -- * Working with remote nodes

121+ , Remote.remoteTable

122+ , Remote.startTraceRelay

123+ , Remote.setTraceFlagsRemote

124+ -- * Built in tracers

125+ , systemLoggerTracer

126+ , logfileTracer

127+ , eventLogTracer

128+ )

129+ where

130+

131+import Control.Applicative ((<$>))

132+import Control.Distributed.Process.Internal.Primitives

133+ ( proxy

134+ , finally

135+ , die

136+ , whereis

137+ , send

138+ , receiveWait

139+ , matchIf

140+ , finally

141+ , try

142+ , monitor

143+ )

144+import Control.Distributed.Process.Internal.Types

145+ ( ProcessId

146+ , Process

147+ , LocalProcess(..)

148+ , ProcessMonitorNotification(..)

149+ )

150+import Control.Distributed.Process.Internal.Trace.Types

151+ ( TraceArg(..)

152+ , TraceEvent(..)

153+ , TraceFlags(..)

154+ , TraceSubject(..)

155+ , defaultTraceFlags

156+ )

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

158+ ( systemLoggerTracer

159+ , logfileTracer

160+ , eventLogTracer

161+ )

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

163+ ( withRegisteredTracer

164+ , enableTrace

165+ , enableTraceAsync

166+ , disableTrace

167+ , setTraceFlags

168+ , setTraceFlagsAsync

169+ , getTraceFlags

170+ , isTracingEnabled

171+ , traceOn

172+ , traceOff

173+ , traceOnly

174+ , traceLog

175+ , traceLogFmt

176+ , traceMessage

177+ )

178+import qualified Control.Distributed.Process.Internal.Trace.Remote as Remote

179+import Control.Distributed.Process.Node

180+

181+import Control.Exception (SomeException)

182+

183+import Control.Monad.IO.Class (liftIO)

184+import Control.Monad.Reader (ask)

185+

186+import Data.Binary()

187+

188+#if ! MIN_VERSION_base(4,6,0)

189+import Prelude hiding (catch)

190+#endif

191+

192+--------------------------------------------------------------------------------

193+-- Debugging/Tracing API --

194+--------------------------------------------------------------------------------

195+

196+-- | Starts a new tracer, using the supplied trace function.

197+-- Only one tracer can be registered at a time, however /this/ function overlays

198+-- the registered tracer with the supplied handler, allowing the user to layer

199+-- multiple tracers on top of one another, with trace events forwarded down

200+-- through all the layers in turn. Once the top layer is stopped, the user

201+-- is responsible for re-registering the original (prior) tracer pid before

202+-- terminating. See 'withTracer' for a mechanism that handles that.

203+startTracer :: (TraceEvent -> Process ()) -> Process ProcessId

204+startTracer handler = do

205+ withRegisteredTracer $ \pid -> do

206+ node <- processNode <$> ask

207+ newPid <- liftIO $ forkProcess node $ traceProxy pid handler

208+ enableTrace newPid -- invokes sync + registration

209+ return newPid

210+

211+-- | Evaluate @proc@ with tracing enabled via @handler@, and immediately

212+-- disable tracing thereafter, before giving the result (or exception

213+-- in case of failure).

214+withTracer :: forall a.

215+ (TraceEvent -> Process ())

216+ -> Process a

217+ -> Process (Either SomeException a)

218+withTracer handler proc = do

219+ previous <- whereis "tracer"

220+ tracer <- startTracer handler

221+ finally (try proc)

222+ (stopTracing tracer previous)

223+ where

224+ stopTracing :: ProcessId -> Maybe ProcessId -> Process ()

225+ stopTracing tracer previousTracer = do

226+ case previousTracer of

227+ Nothing -> return ()

228+ Just _ -> do

229+ ref <- monitor tracer

230+ send tracer TraceEvDisable

231+ receiveWait [

232+ matchIf (\(ProcessMonitorNotification ref' _ _) -> ref == ref')

233+ (\_ -> return ())

234+ ]

235+

236+-- | Evaluate @proc@ with the supplied flags enabled. Any previously set

237+-- trace flags are restored immediately afterwards.

238+withFlags :: forall a.

239+ TraceFlags

240+ -> Process a

241+ -> Process (Either SomeException a)

242+withFlags flags proc = do

243+ oldFlags <- getTraceFlags

244+ finally (setTraceFlags flags >> try proc)

245+ (setTraceFlags oldFlags)

246+

247+traceProxy :: ProcessId -> (TraceEvent -> Process ()) -> Process ()

248+traceProxy pid act = do

249+ proxy pid $ \(ev :: TraceEvent) ->

250+ case ev of

251+ (TraceEvTakeover _) -> return False

252+ TraceEvDisable -> die "disabled"

253+ _ -> act ev >> return True

254+

255+-- | Stops a user supplied tracer started with 'startTracer'.

256+-- Note that only one tracer process can be active at any given time.

257+-- This process will stop the last process started with 'startTracer'.

258+-- If 'startTracer' is called multiple times, successive calls to this

259+-- function will stop the tracers in the reverse order which they were

260+-- started.

261+--

262+-- This function will never stop the system tracer (i.e., the tracer

263+-- initially started when the node is created), therefore once all user

264+-- supplied tracers (i.e., processes started via 'startTracer') have exited,

265+-- subsequent calls to this function will have no effect.

266+--

267+-- If tracing support is disabled for the node, this function will also

268+-- have no effect. If the last tracer to have been registered was not started

269+-- with 'startTracer' then the behaviour of this function is /undefined/.

270+stopTracer :: Process ()

271+stopTracer =

272+ withRegisteredTracer $ \pid -> do

273+ -- we need to avoid killing the initial (base) tracer, as

274+ -- nothing we rely on having exactly 1 registered tracer

275+ -- process at all times.

276+ basePid <- whereis "tracer.initial"

277+ case basePid == (Just pid) of

278+ True -> return ()

279+ False -> send pid TraceEvDisable

280+

src/Control/Distributed/Process/Internal/Closure/BuiltIn.hs+24

@@ -16,13 +16,16 @@ module Control.Distributed.Process.Internal.Closure.BuiltIn

1616 , bindCP

1717 , seqCP

1818 -- * CP versions of Cloud Haskell primitives

19+ , decodeProcessIdStatic

1920 , cpLink

2021 , cpUnlink

22+ , cpRelay

2123 , cpSend

2224 , cpExpect

2325 , cpNewChan

2426 -- * Support for some CH operations

2527 , cpDelay

28+ , cpEnableTraceRemote

2629 ) where

2730

2831import Data.ByteString.Lazy (ByteString)

@@ -56,6 +59,7 @@ import Control.Distributed.Process.Internal.Types

5659import Control.Distributed.Process.Internal.Primitives

5760 ( link

5861 , unlink

62+ , relay

5963 , send

6064 , expect

6165 , newChan

@@ -82,6 +86,7 @@ remoteTable =

8286 . registerStatic "$decodeProcessId" (toDynamic (decode :: ByteString -> ProcessId))

8387 . registerStatic "$link" (toDynamic link)

8488 . registerStatic "$unlink" (toDynamic unlink)

89+ . registerStatic "$relay" (toDynamic relay)

8590 . registerStatic "$sendDict" (toDynamic (sendDict :: SerializableDict ANY -> ProcessId -> ANY -> Process ()))

8691 . registerStatic "$expectDict" (toDynamic (expectDict :: SerializableDict ANY -> Process ANY))

8792 . registerStatic "$newChanDict" (toDynamic (newChanDict :: SerializableDict ANY -> Process (SendPort ANY, ReceivePort ANY)))

@@ -246,6 +251,22 @@ cpNewChan dict = staticClosure (newChanDictStatic `staticApply` dict)

246251 => Static (SerializableDict a -> Process (SendPort a, ReceivePort a))

247252 newChanDictStatic = staticLabel "$newChanDict"

248253

254+-- | 'CP' version of 'relay'

255+cpRelay :: ProcessId -> Closure (Process ())

256+cpRelay = closure (relayStatic `staticCompose` decodeProcessIdStatic) . encode

257+ where

258+ relayStatic :: Static (ProcessId -> Process ())

259+ relayStatic = staticLabel "$relay"

260+

261+-- TODO: move cpEnableTraceRemote into Trace/Primitives.hs

262+

263+cpEnableTraceRemote :: ProcessId -> Closure (Process ())

264+cpEnableTraceRemote =

265+ closure (enableTraceStatic `staticCompose` decodeProcessIdStatic) . encode

266+ where

267+ enableTraceStatic :: Static (ProcessId -> Process ())

268+ enableTraceStatic = staticLabel "$enableTraceRemote"

269+

249270--------------------------------------------------------------------------------

250271-- Support for spawn --

251272--------------------------------------------------------------------------------

@@ -275,3 +296,6 @@ cpDelay = closureApply . cpDelay'

275296

276297 delayStatic :: Static (ProcessId -> Process () -> Process ())

277298 delayStatic = staticLabel "$delay"

299+

300+

301+

src/Control/Distributed/Process/Internal/Primitives.hs+76−62

@@ -1,3 +1,5 @@

1+{-# LANGUAGE BangPatterns #-}

13-- | Cloud Haskell primitives

24--

35-- We define these in a separate module so that we don't have to rely on

@@ -28,6 +30,9 @@ module Control.Distributed.Process.Internal.Primitives

2830 , unwrapMessage

2931 , handleMessage

3032 , forward

33+ , delegate

34+ , relay

35+ , proxy

3136 -- * Process management

3237 , terminate

3338 , ProcessTerminationException(..)

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

8893 -- * Reconnecting

8994 , reconnect

9095 , reconnectPort

91− -- * Tracing/Debugging

92− , trace

9396 ) where

9497

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

@@ -129,7 +132,10 @@ import Control.Distributed.Process.Internal.CQueue

129132 )

130133import Control.Distributed.Process.Serializable (Serializable, fingerprint)

131134import Data.Accessor ((^.), (^:), (^=))

132−import Control.Distributed.Static (Closure, Static)

135+import Control.Distributed.Static

136+ ( Static

137+ , Closure

138+ )

133139import Data.Rank1Typeable (Typeable)

134140import qualified Control.Distributed.Static as Static (unstatic, unclosure)

135141import Control.Distributed.Process.Internal.Types

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

176182 , sendPayload

177183 , disconnect

178184 )

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

185+import Control.Distributed.Process.Internal.Trace.Types

186+ ( traceEvent

187+ , TraceEvent(..)

188+ )

180189import Control.Distributed.Process.Internal.WeakTQueue

181190 ( newTQueueIO

182191 , readTQueue

183192 , mkWeakTQueue

184193 )

194+

185195import Unsafe.Coerce

186196

187197--------------------------------------------------------------------------------

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

191201-- | Send a message

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

193203-- This requires a lookup on every send. If we want to avoid that we need to

194−-- modify serializable to allow for stateful (IO) deserialization

204+-- modify serializable to allow for stateful (IO) deserialization.

195205send them msg = do

196206 proc <- ask

197− let node = localNodeId (processNode proc)

207+ let us = processId proc

208+ node = processNode proc

209+ nodeId = localNodeId node

198210 destNode = (processNodeId them) in do

199− case destNode == node of

211+ case destNode == nodeId of

200212 True -> sendLocal them msg

201213 False -> liftIO $ sendMessage (processNode proc)

202214 (ProcessIdentifier (processId proc))

203215 (ProcessIdentifier them)

204216 NoImplicitReconnect

205217 msg

218+ -- We do not fire the trace event until after the sending is complete;

219+ -- In the remote case, 'sendMessage' can block in the networking stack.

220+ liftIO $ traceEvent (localTracer node)

221+ (TraceEvSent them us (createUnencodedMessage msg))

206222

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

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

@@ -358,33 +374,44 @@ matchMessageIf c p = Match $ MatchMsg $ \msg ->

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

359375forward msg them = do

360376 proc <- ask

361− let node = localNodeId (processNode proc)

377+ let node = processNode proc

378+ us = processId proc

379+ nid = localNodeId node

362380 destNode = (processNodeId them) in do

363− case destNode == node of

381+ case destNode == nid of

364382 True -> sendCtrlMsg Nothing (LocalSend them msg)

365383 False -> liftIO $ sendPayload (processNode proc)

366384 (ProcessIdentifier (processId proc))

367385 (ProcessIdentifier them)

368386 NoImplicitReconnect

369387 (messageToPayload msg)

388+ -- We do not fire the trace event until after the sending is complete;

389+ -- In the remote case, 'sendMessage' can block in the networking stack.

390+ liftIO $ traceEvent (localTracer node)

391+ (TraceEvSent them us (createUnencodedMessage msg))

370392

371393

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

373−-- 'Serializable' - like the datum they contain - but remember that deserializing

395+-- 'Serializable' - like the datum they contain - but also note, deserialising

374396-- a 'Message' will yield a 'Message', not the type within it! To obtain the

375397-- wrapped datum, use 'unwrapMessage' or 'handleMessage' with a specific type.

376398--

377399-- @

378400-- do

379−-- self <- getSelfPid

380−-- send self (wrapMessage "blah")

381−-- Nothing <- expectTimeout 1000000 :: Process (Maybe String)

382−-- (Just m) <- expectTimeout 1000000 :: Process (Maybe Message)

383−-- (Just "blah") <- unwrapMessage m :: Process (Maybe String)

401+-- self <- getSelfPid

402+-- send self (wrapMessage "blah")

403+-- Nothing <- expectTimeout 1000000 :: Process (Maybe String)

404+-- (Just m) <- expectTimeout 1000000 :: Process (Maybe Message)

405+-- (Just "blah") <- unwrapMessage m :: Process (Maybe String)

384406-- @

385407--

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

387−wrapMessage = createMessage

409+wrapMessage = createUnencodedMessage -- see [note Serializable UnencodedMessage]

410+

411+-- [note Serializable UnencodedMessage]

412+-- if we attempt to serialise an UnencodedMessage, it will be converted to an

413+-- EncodedMessage first, so we're ok here. See 'messageToPayload' in

414+-- Primitives.hs for the gory details

388415

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

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

@@ -480,6 +507,36 @@ succeeds) that unsafeCoerce will not fail.

480507matchUnknown :: Process b -> Match b

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

482509

510+-- | Receives messages and forwards them to @pid@ if @p msg == True@.

511+delegate :: ProcessId -> (Message -> Bool) -> Process ()

512+delegate pid p = do

513+ receiveWait [

514+ matchAny (\m -> case (p m) of

515+ True -> forward m pid

516+ False -> return ())

517+ ]

518+ delegate pid p

519+

520+-- | A straight relay that simply forwards all messages to the supplied pid.

521+relay :: ProcessId -> Process ()

522+relay !pid = receiveWait [ matchAny (\m -> forward m pid) ] >> relay pid

523+

524+-- | Proxies @pid@ and forwards messages whenever @proc@ evaluates to @True@.

525+-- Unlike 'delegate' the predicate @proc@ runs in the 'Process' monad, allowing

526+-- for richer proxy behaviour. If @proc@ returns @False@ or the /runtime type check/

527+-- fails, no action is taken and the proxy process will continue running.

528+proxy :: Serializable a => ProcessId -> (a -> Process Bool) -> Process ()

529+proxy pid proc = do

530+ receiveWait [

531+ matchAny (\m -> do

532+ next <- handleMessage m proc

533+ case next of

534+ Just True -> forward m pid

535+ Just False -> return () -- explicitly ignored

536+ Nothing -> return ()) -- un-routable

537+ ]

538+ proxy pid proc

539+

483540--------------------------------------------------------------------------------

484541-- Process management --

485542--------------------------------------------------------------------------------

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

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

751808

752809instance Functor Handler where

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

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

754811

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

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

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

9911048 node <- processNode <$> ask

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

9931050

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−

10371051--------------------------------------------------------------------------------

10381052-- Auxiliary functions --

10391053--------------------------------------------------------------------------------

10401054

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

1042−sendLocal pid msg =

1043− sendCtrlMsg Nothing $ LocalSend pid (createUnencodedMessage msg)

1056+sendLocal to msg =

1057+ sendCtrlMsg Nothing $ LocalSend to (createUnencodedMessage msg)

10441058

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

10461060sendChanLocal spId msg =

src/Control/Distributed/Process/Internal/Spawn.hsnew file+162

@@ -0,0 +1,162 @@

1+module Control.Distributed.Process.Internal.Spawn

2+ ( spawn

3+ , spawnLink

4+ , spawnMonitor

5+ , call

6+ , spawnSupervised

7+ , spawnChannel

8+ ) where

10+import Data.Typeable (Typeable)

11+import Control.Distributed.Static

12+ ( Static

13+ , Closure

14+ , closureCompose

15+ , staticClosure

16+ )

17+import Control.Distributed.Process.Internal.Types

18+ ( NodeId(..)

19+ , ProcessId(..)

20+ , Process(..)

21+ , MonitorRef(..)

22+ , ProcessMonitorNotification(..)

23+ , NodeMonitorNotification(..)

24+ , DidSpawn(..)

25+ , SendPort(..)

26+ , ReceivePort(..)

27+ , nullProcessId

28+ )

29+import Control.Distributed.Process.Serializable (Serializable, SerializableDict)

30+import Control.Distributed.Process.Internal.Closure.BuiltIn

31+ ( sdictSendPort

32+ , sndStatic

33+ , idCP

34+ , seqCP

35+ , bindCP

36+ , splitCP

37+ , cpLink

38+ , cpSend

39+ , cpNewChan

40+ , cpDelay

41+ )

42+import Control.Distributed.Process.Internal.Primitives

43+ ( -- Basic messaging

44+ send

45+ , expect

46+ , receiveWait

47+ , match

48+ , matchIf

49+ , link

50+ , getSelfPid

51+ , monitor

52+ , monitorNode

53+ , unmonitor

54+ , spawnAsync

55+ , reconnect

56+ )

57+

58+-- | Spawn a process

59+--

60+-- For more information about 'Closure', see

61+-- "Control.Distributed.Process.Closure".

62+--

63+-- See also 'call'.

64+spawn :: NodeId -> Closure (Process ()) -> Process ProcessId

65+spawn nid proc = do

66+ us <- getSelfPid

67+ mRef <- monitorNode nid

68+ sRef <- spawnAsync nid (cpDelay us proc)

69+ receiveWait [

70+ matchIf (\(DidSpawn ref _) -> ref == sRef) $ \(DidSpawn _ pid) -> do

71+ unmonitor mRef

72+ send pid ()

73+ return pid

74+ , matchIf (\(NodeMonitorNotification ref _ _) -> ref == mRef) $ \_ ->

75+ return (nullProcessId nid)

76+ ]

77+

78+-- | Spawn a process and link to it

79+--

80+-- Note that this is just the sequential composition of 'spawn' and 'link'.

81+-- (The "Unified" semantics that underlies Cloud Haskell does not even support

82+-- a synchronous link operation)

83+spawnLink :: NodeId -> Closure (Process ()) -> Process ProcessId

84+spawnLink nid proc = do

85+ pid <- spawn nid proc

86+ link pid

87+ return pid

88+

89+-- | Like 'spawnLink', but monitor the spawned process

90+spawnMonitor :: NodeId -> Closure (Process ()) -> Process (ProcessId, MonitorRef)

91+spawnMonitor nid proc = do

92+ pid <- spawn nid proc

93+ ref <- monitor pid

94+ return (pid, ref)

95+

96+-- | Run a process remotely and wait for it to reply

97+--

98+-- We monitor the remote process: if it dies before it can send a reply, we die

99+-- too.

100+--

101+-- For more information about 'Static', 'SerializableDict', and 'Closure', see

102+-- "Control.Distributed.Process.Closure".

103+--

104+-- See also 'spawn'.

105+call :: Serializable a

106+ => Static (SerializableDict a)

107+ -> NodeId

108+ -> Closure (Process a)

109+ -> Process a

110+call dict nid proc = do

111+ us <- getSelfPid

112+ (pid, mRef) <- spawnMonitor nid (proc `bindCP` cpSend dict us)

113+ -- We are guaranteed to receive the reply before the monitor notification

114+ -- (if a reply is sent at all)

115+ -- NOTE: This might not be true if we switch to unreliable delivery.

116+ mResult <- receiveWait

117+ [ match (return . Right)

118+ , matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mRef)

119+ (\(ProcessMonitorNotification _ _ reason) -> return (Left reason))

120+ ]

121+ case mResult of

122+ Right a -> do

123+ -- Wait for the monitor message so that we the mailbox doesn't grow

124+ receiveWait

125+ [ matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mRef)

126+ (\(ProcessMonitorNotification {}) -> return ())

127+ ]

128+ -- Clean up connection to pid

129+ reconnect pid

130+ return a

131+ Left err ->

132+ fail $ "call: remote process died: " ++ show err

133+

134+-- | Spawn a child process, have the child link to the parent and the parent

135+-- monitor the child

136+spawnSupervised :: NodeId

137+ -> Closure (Process ())

138+ -> Process (ProcessId, MonitorRef)

139+spawnSupervised nid proc = do

140+ us <- getSelfPid

141+ them <- spawn nid (cpLink us `seqCP` proc)

142+ ref <- monitor them

143+ return (them, ref)

144+

145+-- | Spawn a new process, supplying it with a new 'ReceivePort' and return

146+-- the corresponding 'SendPort'.

147+spawnChannel :: forall a. Typeable a => Static (SerializableDict a)

148+ -> NodeId

149+ -> Closure (ReceivePort a -> Process ())

150+ -> Process (SendPort a)

151+spawnChannel dict nid proc = do

152+ us <- getSelfPid

153+ _ <- spawn nid (go us)

154+ expect

155+ where

156+ go :: ProcessId -> Closure (Process ())

157+ go pid = cpNewChan dict

158+ `bindCP`

159+ (cpSend (sdictSendPort dict) pid `splitCP` proc)

160+ `bindCP`

161+ (idCP `closureCompose` staticClosure sndStatic)

162+

src/Control/Distributed/Process/Internal/Trace.hsdeleted−122

@@ -1,122 +0,0 @@

1−-- | Simple (internal) system logging/tracing support.

2−module Control.Distributed.Process.Internal.Trace

3− ( Tracer

4− , TraceArg(..)

5− , trace

6− , traceFormat

7− , startTracing

8− , stopTracer

9− ) where

10−

11−import Control.Concurrent (forkIO)

12−import Control.Concurrent.Chan (writeChan)

13−import Control.Concurrent.STM

14− ( TQueue

15− , newTQueueIO

16− , readTQueue

17− , writeTQueue

18− , atomically

19− )

20−import Control.Distributed.Process.Internal.Types

21− ( Tracer(..)

22− , LocalNode(..)

23− , NCMsg(..)

24− , Identifier(ProcessIdentifier)

25− , ProcessSignal(NamedSend)

26− , forever'

27− , nullProcessId

28− , createMessage

29− )

30−import Control.Exception

31− ( catch

32− , throwTo

33− , SomeException

34− , AsyncException(ThreadKilled)

35− )

36−import Data.List (intersperse)

37−import Data.Time.Clock (getCurrentTime)

38−import Data.Time.Format (formatTime)

39−import Debug.Trace (traceEventIO)

40−

41−import Prelude hiding (catch)

42−

43−import System.Environment (getEnv)

44−import System.IO

45− ( Handle

46− , IOMode(AppendMode)

47− , BufferMode(..)

48− , openFile

49− , hClose

50− , hPutStrLn

51− , hSetBuffering

52− )

53−import System.Locale (defaultTimeLocale)

54−

55−data TraceArg =

56− TraceStr String

57− | forall a. (Show a) => Trace a

58−

59−startTracing :: LocalNode -> IO LocalNode

60−startTracing node = do

61− tracer <- defaultTracer node

62− return node { localTracer = tracer }

63−

64−defaultTracer :: LocalNode -> IO Tracer

65−defaultTracer node = do

66− catch (getEnv "DISTRIBUTED_PROCESS_TRACE_FILE" >>= logfileTracer)

67− (\(_ :: IOError) -> defaultTracerAux node)

68−

69−defaultTracerAux :: LocalNode -> IO Tracer

70−defaultTracerAux node = do

71− catch (getEnv "DISTRIBUTED_PROCESS_TRACE_CONSOLE" >> procTracer node)

72− (\(_ :: IOError) -> return (EventLogTracer traceEventIO))

73− where procTracer :: LocalNode -> IO Tracer

74− procTracer n = return $ (LocalNodeTracer n)

75−

76−logfileTracer :: FilePath -> IO Tracer

77−logfileTracer p = do

78− q <- newTQueueIO

79− h <- openFile p AppendMode

80− hSetBuffering h LineBuffering

81− tid <- forkIO $ logger h q `catch` (\(_ :: SomeException) ->

82− hClose h >> return ())

83− return $ LogFileTracer tid q h

84− where logger :: Handle -> TQueue String -> IO ()

85− logger h q' = forever' $ do

86− msg <- atomically $ readTQueue q'

87− now <- getCurrentTime

88− hPutStrLn h $ msg ++ (formatTime defaultTimeLocale " - %c" now)

89−

90−-- TODO: compatibility layer for GHC/base versions (e.g., where's killThread?)

91−

92−stopTracer :: Tracer -> IO () -- overzealous but harmless duplication of hClose

93−stopTracer (LogFileTracer tid _ h) = throwTo tid ThreadKilled >> hClose h

94−stopTracer _ = return ()

95−

96−trace :: Tracer -> String -> IO ()

97−trace (LogFileTracer _ q _) msg = atomically $ writeTQueue q msg

98−trace (LocalNodeTracer n) msg = sendTraceMsg n msg

99−trace (EventLogTracer t) msg = t msg

100−trace InactiveTracer _ = return ()

101−

102−traceFormat :: Tracer

103− -> String

104− -> [TraceArg]

105− -> IO ()

106−traceFormat t d ls =

107− trace t $ concat (intersperse d (map toS ls))

108− where toS :: TraceArg -> String

109− toS (TraceStr s) = s

110− toS (Trace a) = show a

111−

112−sendTraceMsg :: LocalNode -> String -> IO ()

113−sendTraceMsg node string = do

114− now <- getCurrentTime

115− msg <- return $ (formatTime defaultTimeLocale "%c" now, string)

116− emptyPid <- return $ (nullProcessId (localNodeId node))

117− traceMsg <- return $ NCMsg {

118− ctrlMsgSender = ProcessIdentifier (emptyPid)

119− , ctrlMsgSignal = (NamedSend "logger" (createMessage msg))

120− }

121− writeChan (localCtrlChan node) traceMsg

122−

src/Control/Distributed/Process/Internal/Trace/Primitives.hsnew file+180

@@ -0,0 +1,180 @@

1+-- | Keeps the tracing API calls separate from the Tracer implementation,

2+-- which allows us to avoid a nasty import cycle between tracing and

3+-- the messaging primitives that rely on it, and also between the node

4+-- controller (which requires access to the tracing related elements of

5+-- our RemoteTable) and the Debug module, which requires @forkProcess@.

6+module Control.Distributed.Process.Internal.Trace.Primitives

7+ ( Tracer

8+ , TraceEvent(..)

9+ -- * Sending Trace Data

10+ , traceLog

11+ , traceLogFmt

12+ , traceMessage

13+ -- * Configuring A Tracer

14+ , isTracingEnabled

15+ , defaultTraceFlags

16+ , enableTrace

17+ , enableTraceAsync

18+ , disableTrace

19+ , disableTraceAsync

20+ , getTraceFlags

21+ , setTraceFlags

22+ , setTraceFlagsAsync

23+ , traceOnly

24+ , traceOn

25+ , traceOff

26+ , withLocalTracer

27+ , withRegisteredTracer

28+ ) where

29+

30+import Control.Applicative ((<$>))

31+import Control.Distributed.Process.Internal.Primitives

32+ ( whereis

33+ , newChan

34+ , receiveChan

35+ )

36+import Control.Distributed.Process.Internal.Trace.Types

37+ ( TraceArg(..)

38+ , TraceEvent(..)

39+ , TraceFlags(..)

40+ , TraceOk(..)

41+ , TraceSubject(..)

42+ , defaultTraceFlags

43+ )

44+import qualified Control.Distributed.Process.Internal.Trace.Types as Tracer

45+ ( traceLog

46+ , traceLogFmt

47+ , traceMessage

48+ , enableTrace

49+ , enableTraceSync

50+ , disableTrace

51+ , disableTraceSync

52+ , setTraceFlags

53+ , setTraceFlagsSync

54+ , getTraceFlags

55+ , getCurrentTraceClient

56+ )

57+import Control.Distributed.Process.Internal.Types

58+ ( Tracer(..)

59+ , Process

60+ , ProcessId

61+ , LocalProcess(..)

62+ , LocalNode(localTracer)

63+ , SendPort

64+ )

65+import Control.Distributed.Process.Serializable

66+import Control.Monad.IO.Class (liftIO)

67+import Control.Monad.Reader (ask)

68+

69+import qualified Data.Set as Set (fromList)

70+

71+--------------------------------------------------------------------------------

72+-- Main API --

73+--------------------------------------------------------------------------------

74+

75+-- | Maps a list of identifiers that can be mapped to process ids,

76+-- to a 'TraceSubject'.

77+class Traceable a where

78+ uod :: [a] -> TraceSubject

79+

80+instance Traceable ProcessId where

81+ uod = TraceProcs . Set.fromList

82+

83+instance Traceable String where

84+ uod = TraceNames . Set.fromList

85+

86+-- | Turn tracing for for a subset of trace targets.

87+traceOnly :: Traceable a => [a] -> Maybe TraceSubject

88+traceOnly = Just . uod

89+

90+-- | Trace all targets.

91+traceOn :: Maybe TraceSubject

92+traceOn = Just TraceAll

93+

94+-- | Trace no targets.

95+traceOff :: Maybe TraceSubject

96+traceOff = Nothing

97+

98+-- | Determine whether or not tracing has been enabled.

99+isTracingEnabled :: Process Bool

100+isTracingEnabled = do

101+ node <- processNode <$> ask

102+ case (localTracer node) of

103+ (ActiveTracer _ _) -> return True

104+ _ -> return False

105+

106+-- | Enable tracing to the supplied process.

107+enableTraceAsync :: ProcessId -> Process ()

108+enableTraceAsync pid = withLocalTracer $ \t -> liftIO $ Tracer.enableTrace t pid

109+

110+-- TODO: refactor _Sync versions of trace configuration functions...

111+

112+-- | Enable tracing to the supplied process and wait for a @TraceOk@

113+-- response from the trace coordinator process.

114+enableTrace :: ProcessId -> Process ()

115+enableTrace pid =

116+ withLocalTracerSync $ \t sp -> Tracer.enableTraceSync t sp pid

117+

118+-- | Disable the currently configured trace.

119+disableTraceAsync :: Process ()

120+disableTraceAsync = withLocalTracer $ \t -> liftIO $ Tracer.disableTrace t

121+

122+-- | Disable the currently configured trace and wait for a @TraceOk@

123+-- response from the trace coordinator process.

124+disableTrace :: Process ()

125+disableTrace =

126+ withLocalTracerSync $ \t sp -> Tracer.disableTraceSync t sp

127+

128+getTraceFlags :: Process TraceFlags

129+getTraceFlags = do

130+ (sp, rp) <- newChan

131+ withLocalTracer $ \t -> liftIO $ Tracer.getTraceFlags t sp

132+ receiveChan rp

133+

134+-- | Set the given flags for the current tracer.

135+setTraceFlagsAsync :: TraceFlags -> Process ()

136+setTraceFlagsAsync f = withLocalTracer $ \t -> liftIO $ Tracer.setTraceFlags t f

137+

138+-- | Set the given flags for the current tracer and wait for a @TraceOk@

139+-- response from the trace coordinator process.

140+setTraceFlags :: TraceFlags -> Process ()

141+setTraceFlags f =

142+ withLocalTracerSync $ \t sp -> Tracer.setTraceFlagsSync t sp f

143+

144+-- | Send a log message to the internal tracing facility. If tracing is

145+-- enabled, this will create a custom trace log event.

146+--

147+traceLog :: String -> Process ()

148+traceLog s = withLocalTracer $ \t -> liftIO $ Tracer.traceLog t s

149+

150+-- | Send a log message to the internal tracing facility, using the given

151+-- list of printable 'TraceArg's interspersed with the preceding delimiter.

152+--

153+traceLogFmt :: String -> [TraceArg] -> Process ()

154+traceLogFmt d ls = withLocalTracer $ \t -> liftIO $ Tracer.traceLogFmt t d ls

155+

156+-- | Send an arbitrary 'Message' to the tracer process.

157+traceMessage :: Serializable m => m -> Process ()

158+traceMessage msg = withLocalTracer $ \t -> liftIO $ Tracer.traceMessage t msg

159+

160+withLocalTracer :: (Tracer -> Process ()) -> Process ()

161+withLocalTracer act = do

162+ node <- processNode <$> ask

163+ act (localTracer node)

164+

165+withLocalTracerSync :: (Tracer -> SendPort TraceOk -> IO ()) -> Process ()

166+withLocalTracerSync act = do

167+ (sp, rp) <- newChan

168+ withLocalTracer $ \t -> liftIO $ (act t sp)

169+ TraceOk <- receiveChan rp

170+ return ()

171+

172+withRegisteredTracer :: (ProcessId -> Process a) -> Process a

173+withRegisteredTracer act = do

174+ (sp, rp) <- newChan

175+ withLocalTracer $ \t -> liftIO $ Tracer.getCurrentTraceClient t sp

176+ currentTracer <- receiveChan rp

177+ case currentTracer of

178+ Nothing -> do { (Just p') <- whereis "tracer.initial"; act p' }

179+ (Just p) -> act p

180+

src/Control/Distributed/Process/Internal/Trace/Remote.hsnew file+62

@@ -0,0 +1,62 @@

1+module Control.Distributed.Process.Internal.Trace.Remote

2+ ( -- * Configuring A Tracer

3+ setTraceFlagsRemote

4+ , startTraceRelay

5+ -- * Remote Table

6+ , remoteTable

7+ ) where

9+import Control.Distributed.Process.Internal.Closure.BuiltIn

10+ ( cpEnableTraceRemote

11+ )

12+import Control.Distributed.Process.Internal.Primitives

13+ ( getSelfPid

14+ , relay

15+ , nsendRemote

16+ )

17+import Control.Distributed.Process.Internal.Trace.Types

18+ ( TraceFlags(..)

19+ , TraceOk(..)

20+ )

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

22+ ( withRegisteredTracer

23+ , enableTrace

24+ )

25+import Control.Distributed.Process.Internal.Spawn

26+ ( spawn

27+ )

28+import Control.Distributed.Process.Internal.Types

29+ ( Process

30+ , ProcessId

31+ , SendPort

32+ , NodeId

33+ )

34+import Control.Distributed.Static

35+ ( RemoteTable

36+ , registerStatic

37+ )

38+import Data.Rank1Dynamic (toDynamic)

39+

40+-- | Remote Table.

41+remoteTable :: RemoteTable -> RemoteTable

42+remoteTable = registerStatic "$enableTraceRemote" (toDynamic enableTraceRemote)

43+

44+enableTraceRemote :: ProcessId -> Process ()

45+enableTraceRemote pid =

46+ getSelfPid >>= enableTrace >> relay pid

47+

48+-- | Starts a /trace relay/ process on the remote node, which forwards all trace

49+-- events to the registered tracer on /this/ (the calling process') node.

50+startTraceRelay :: NodeId -> Process ProcessId

51+startTraceRelay nodeId = do

52+ withRegisteredTracer $ \pid ->

53+ -- TODO: move cpEnableTraceRemote here?

54+ spawn nodeId $ cpEnableTraceRemote pid

55+

56+-- | Set the given flags for a remote node (asynchronous).

57+setTraceFlagsRemote :: TraceFlags -> NodeId -> Process ()

58+setTraceFlagsRemote flags node = do

59+ nsendRemote node

60+ "trace.controller"

61+ ((Nothing :: Maybe (SendPort TraceOk)), flags)

62+

src/Control/Distributed/Process/Internal/Trace/Tracer.hsnew file+356

@@ -0,0 +1,356 @@

1+-- | Tracing/Debugging support - Trace Implementation

2+module Control.Distributed.Process.Internal.Trace.Tracer

3+ ( -- * API for the Node Controller

4+ traceController

5+ -- * Built in tracers

6+ , defaultTracer

7+ , systemLoggerTracer

8+ , logfileTracer

9+ , eventLogTracer

10+ ) where

11+

12+import Control.Applicative ((<$>))

13+import Control.Concurrent.Chan (writeChan)

14+import Control.Concurrent.MVar

15+ ( MVar

16+ , putMVar

17+ )

18+import Control.Distributed.Process.Internal.CQueue

19+ ( CQueue

20+ )

21+import Control.Distributed.Process.Internal.Primitives

22+ ( catch

23+ , finally

24+ , die

25+ , receiveWait

26+ , forward

27+ , sendChan

28+ , match

29+ , matchAny

30+ , matchIf

31+ , handleMessage

32+ , unwrapMessage

33+ , matchUnknown

34+ )

35+import Control.Distributed.Process.Internal.Trace.Types

36+ ( TraceEvent(..)

37+ , SetTrace(..)

38+ , Addressable(..)

39+ , TraceSubject(..)

40+ , TraceFlags(..)

41+ , TraceOk(..)

42+ , defaultTraceFlags

43+ )

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

45+ ( traceOn )

46+import Control.Distributed.Process.Internal.Types

47+ ( LocalNode(..)

48+ , NCMsg(..)

49+ , ProcessId

50+ , Process

51+ , LocalProcess(..)

52+ , Identifier(..)

53+ , ProcessSignal(NamedSend)

54+ , Message

55+ , SendPort

56+ , forever'

57+ , nullProcessId

58+ , createUnencodedMessage

59+ )

60+

61+import Control.Monad.IO.Class (liftIO)

62+import Control.Monad.Reader (ask)

63+

64+import Data.Set (Set)

65+import qualified Data.Set as Set

66+import Data.Map (Map)

67+import qualified Data.Map as Map

68+

69+import Data.Maybe (fromMaybe)

70+import Data.Time.Clock (getCurrentTime)

71+import Data.Time.Format (formatTime)

72+import Debug.Trace (traceEventIO)

73+

74+import Prelude hiding (catch)

75+

76+import System.Environment (getEnv)

77+import System.IO

78+ ( Handle

79+ , IOMode(AppendMode)

80+ , BufferMode(..)

81+ , openFile

82+ , hClose

83+ , hPutStrLn

84+ , hSetBuffering

85+ )

86+import System.Locale (defaultTimeLocale)

87+import System.Mem.Weak

88+ ( Weak

89+ )

90+

91+data TracerState = TracerST {

92+ client :: !(Maybe ProcessId)

93+ , flags :: !TraceFlags

94+ , regNames :: !(Map ProcessId (Set String))

95+ }

96+

97+--------------------------------------------------------------------------------

98+-- Trace Handlers --

99+--------------------------------------------------------------------------------

100+

101+defaultTracer :: Process ()

102+defaultTracer =

103+ catch (checkEnv "DISTRIBUTED_PROCESS_TRACE_FILE" >>= logfileTracer)

104+ (\(_ :: IOError) -> defaultTracerAux)

105+

106+defaultTracerAux :: Process ()

107+defaultTracerAux =

108+ catch (checkEnv "DISTRIBUTED_PROCESS_TRACE_CONSOLE" >> systemLoggerTracer)

109+ (\(_ :: IOError) -> defaultEventLogTracer)

110+

111+-- TODO: it would be /nice/ if we had some way of checking the runtime

112+-- options to see if +RTS -v (or similar) has been given...

113+defaultEventLogTracer :: Process ()

114+defaultEventLogTracer =

115+ catch (checkEnv "DISTRIBUTED_PROCESS_TRACE_EVENTLOG" >> eventLogTracer)

116+ (\(_ :: IOError) -> nullTracer)

117+

118+checkEnv :: String -> Process String

119+checkEnv s = liftIO $ getEnv s

120+

121+-- This trace client is (intentionally) useful - it simply provides

122+-- an intial client for the trace controller to talk to, until some

123+-- other (hopefully more useful) client is installed over the top of

124+-- it. This is the default trace client.

125+nullTracer :: Process ()

126+nullTracer =

127+ forever' $ receiveWait [ matchUnknown (return ()) ]

128+

129+systemLoggerTracer :: Process ()

130+systemLoggerTracer = do

131+ node <- processNode <$> ask

132+ let tr = sendTraceMsg node

133+ forever' $ receiveWait [ matchAny (\m -> handleMessage m tr) ]

134+ where

135+ sendTraceMsg :: LocalNode -> TraceEvent -> Process ()

136+ sendTraceMsg node ev = do

137+ now <- liftIO $ getCurrentTime

138+ txt <- buildTxt ev

139+ msg <- return $ (formatTime defaultTimeLocale "%c" now, txt)

140+ emptyPid <- return $ (nullProcessId (localNodeId node))

141+ traceMsg <- return $ NCMsg {

142+ ctrlMsgSender = ProcessIdentifier (emptyPid)

143+ , ctrlMsgSignal = (NamedSend "logger"

144+ (createUnencodedMessage msg))

145+ }

146+ liftIO $ writeChan (localCtrlChan node) traceMsg

147+

148+ buildTxt :: TraceEvent -> Process String

149+ buildTxt (TraceEvLog msg) = return msg

150+ buildTxt ev = return $ show ev

151+

152+eventLogTracer :: Process ()

153+eventLogTracer =

154+ -- NB: when the GHC event log supports tracing arbitrary (ish) data, we will

155+ -- almost certainly use *that* facility independently of whether or not there

156+ -- is a tracer process installed. This is just a stop gap until then.

157+ forever' $ receiveWait [ matchAny (\m -> handleMessage m writeTrace) ]

158+ where

159+ writeTrace :: TraceEvent -> Process ()

160+ writeTrace ev = liftIO $ traceEventIO (show ev)

161+

162+logfileTracer :: FilePath -> Process ()

163+logfileTracer p = do

164+ -- TODO: error handling if the handle cannot be opened

165+ liftIO $ putStrLn $ "writing to " ++ (show p)

166+ h <- liftIO $ openFile p AppendMode

167+ liftIO $ hSetBuffering h LineBuffering

168+ logger h `finally` (liftIO $ hClose h)

169+ where

170+ logger :: Handle -> Process ()

171+ logger h' = forever' $ do

172+ receiveWait [

173+ matchIf (\ev -> case ev of

174+ TraceEvDisable -> True

175+ (TraceEvTakeover _) -> True

176+ _ -> False)

177+ (\_ -> (liftIO $ hClose h') >> die "trace stopped")

178+ , matchAny (\ev -> handleMessage ev (writeTrace h'))

179+ ]

180+

181+ writeTrace :: Handle -> TraceEvent -> Process ()

182+ writeTrace h ev = do

183+ liftIO $ do

184+ now <- getCurrentTime

185+ hPutStrLn h $ (formatTime defaultTimeLocale "%c - " now) ++ (show ev)

186+

187+--------------------------------------------------------------------------------

188+-- Tracer Implementation --

189+--------------------------------------------------------------------------------

190+

191+traceController :: MVar ((Weak (CQueue Message))) -> Process ()

192+traceController mv = do

193+ -- This breach of encapsulation is deliberate: we don't want to tie up

194+ -- the node's internal control channel if we can possibly help it!

195+ weakQueue <- processWeakQ <$> ask

196+ liftIO $ putMVar mv weakQueue

197+ initState <- initialState

198+ traceLoop initState { client = Nothing }

199+ where

200+ traceLoop :: TracerState -> Process ()

201+ traceLoop st = do

202+ -- Trace events are forwarded to the trace target when tracing is enabled.

203+ -- At some point in the future, we're going to start writing these custom

204+ -- events to the ghc eventlog, at which point this design might change.

205+ st' <- receiveWait [

206+ match (\(setResp, set :: SetTrace) -> do

207+ -- We allow at most one trace client, which is a process id.

208+ -- Tracking multiple clients represents too high an overhead,

209+ -- so we leave that kind of thing to our consumers to figure

210+ -- figure out.

211+ case set of

212+ (TraceEnable pid) -> do

213+ -- notify the previous tracer it has been replaced

214+ sendTrace st (createUnencodedMessage (TraceEvTakeover pid))

215+ sendOk setResp

216+ return st { client = (Just pid) }

217+ TraceDisable -> do

218+ sendTrace st (createUnencodedMessage TraceEvDisable)

219+ sendOk setResp

220+ return st { client = Nothing })

221+ , match (\(confResp, flags') ->

222+ sendOk confResp >> applyTraceFlags flags' st)

223+ , match (\chGetFlags -> sendChan chGetFlags (flags st) >> return st)

224+ , match (\chGetCurrent -> sendChan chGetCurrent (client st) >> return st)

225+ -- we dequeue incoming events even if we don't process them

226+ , matchAny (\ev ->

227+ handleMessage ev (handleTrace st ev) >>= return . fromMaybe st)

228+ ]

229+ traceLoop st'

230+

231+ sendOk :: Maybe (SendPort TraceOk) -> Process ()

232+ sendOk Nothing = return ()

233+ sendOk (Just sp) = sendChan sp TraceOk

234+

235+ initialState :: Process TracerState

236+ initialState = do

237+ flags' <- checkEnvFlags

238+ return $ TracerST { client = Nothing

239+ , flags = flags'

240+ , regNames = Map.empty

241+ }

242+

243+ checkEnvFlags :: Process TraceFlags

244+ checkEnvFlags =

245+ catch (checkEnv "DISTRIBUTED_PROCESS_TRACE_FLAGS" >>= return . parseFlags)

246+ (\(_ :: IOError) -> return defaultTraceFlags)

247+

248+ -- quick and dirty fold-while

249+ parseFlags :: String -> TraceFlags

250+ parseFlags s = parseFlags' s defaultTraceFlags

251+ where parseFlags' :: String -> TraceFlags -> TraceFlags

252+ parseFlags' [] parsedFlags = parsedFlags

253+ parseFlags' (x:xs) parsedFlags

254+ | x == 'p' = parseFlags' xs parsedFlags { traceSpawned = traceOn }

255+ | x == 'n' = parseFlags' xs parsedFlags { traceRegistered = traceOn }

256+ | x == 'u' = parseFlags' xs parsedFlags { traceUnregistered = traceOn }

257+ | x == 'd' = parseFlags' xs parsedFlags { traceDied = traceOn }

258+ | x == 's' = parseFlags' xs parsedFlags { traceSend = traceOn }

259+ | x == 'r' = parseFlags' xs parsedFlags { traceRecv = traceOn }

260+ | x == 'l' = parseFlags' xs parsedFlags { traceNodes = True }

261+ | otherwise = parseFlags' xs parsedFlags

262+

263+-- note [runtime tracer control]

264+--

265+-- The trace mechanism is designed to put as little stress on the system, and

266+-- in particular the node controller, as possible. The LocalNode's @tracer@

267+-- field is therefore immutable, since we don't want to be blocking the node

268+-- controller when writing trace information out. The runtime cost of enabling

269+-- tracing is therefore the cost of enqueue for all trace-able operations at

270+-- all times. If tracing is not explicitly enabled, the trace record stored in

271+-- the local node state is matched and ignored in API calls, reducing the cost

272+-- to a single function call.

273+--

274+-- To /disable/ tracing facilities in a runtime system that has tracing enabled

275+-- then, is to instruct the tracer process not to forward any trace event data

276+-- and it is not possible to remove the runtime overhead (of enqueue in the

277+-- caller and dequeue + noop in the tracer process) once the node is started.

278+

279+applyTraceFlags :: TraceFlags -> TracerState -> Process TracerState

280+applyTraceFlags flags' state = return state { flags = flags' }

281+

282+handleTrace :: TracerState -> Message -> TraceEvent -> Process TracerState

283+handleTrace st msg ev@(TraceEvRegistered p n) =

284+ let regNames' =

285+ Map.insertWith (\_ ns -> Set.insert n ns) p

286+ (Set.singleton n)

287+ (regNames st)

288+ in do

289+ traceEv ev msg (traceRegistered (flags st)) st

290+ return st { regNames = regNames' }

291+handleTrace st msg ev@(TraceEvUnRegistered p n) =

292+ let f ns = case ns of

293+ Nothing -> Nothing

294+ Just ns' -> Just (Set.delete n ns')

295+ regNames' = Map.alter f p (regNames st)

296+ in do

297+ traceEv ev msg (traceUnregistered (flags st)) st

298+ return st { regNames = regNames' }

299+handleTrace st msg ev@(TraceEvSpawned _) = do

300+ traceEv ev msg (traceSpawned (flags st)) st >> return st

301+handleTrace st msg ev@(TraceEvDied _ _) = do

302+ traceEv ev msg (traceDied (flags st)) st >> return st

303+handleTrace st msg ev@(TraceEvSent _ _ _) =

304+ traceEv ev msg (traceSend (flags st)) st >> return st

305+handleTrace st msg ev@(TraceEvReceived _ _) =

306+ traceEv ev msg (traceRecv (flags st)) st >> return st

307+handleTrace st msg ev = do

308+ case ev of

309+ (TraceEvNodeDied _ _) ->

310+ case (traceNodes (flags st)) of

311+ True -> sendTrace st msg

312+ False -> return ()

313+ (TraceEvUser _) -> sendTrace st msg

314+ (TraceEvLog _) -> sendTrace st msg

315+ _ ->

316+ case (traceConnections (flags st)) of

317+ True -> sendTrace st msg

318+ False -> return ()

319+ return st

320+

321+traceEv :: TraceEvent

322+ -> Message

323+ -> Maybe TraceSubject

324+ -> TracerState

325+ -> Process ()

326+traceEv _ _ Nothing _ = return ()

327+traceEv _ msg (Just TraceAll) st = sendTrace st msg

328+traceEv ev msg (Just (TraceProcs pids)) st = do

329+ node <- processNode <$> ask

330+ let p = case resolveToPid ev of

331+ Nothing -> (nullProcessId (localNodeId node))

332+ Just pid -> pid

333+ case (Set.member p pids) of

334+ True -> sendTrace st msg

335+ False -> return ()

336+traceEv ev msg (Just (TraceNames names)) st = do

337+ -- TODO: if we have recorded regnames for p, then we

338+ -- forward the trace iif there are overlapping trace targets

339+ node <- processNode <$> ask

340+ let p = case resolveToPid ev of

341+ Nothing -> (nullProcessId (localNodeId node))

342+ Just pid -> pid

343+ case (Map.lookup p (regNames st)) of

344+ Nothing -> return ()

345+ Just ns -> if (Set.null (Set.intersection ns names))

346+ then return ()

347+ else sendTrace st msg

348+

349+sendTrace :: TracerState -> Message -> Process ()

350+sendTrace st msg =

351+ let pid = (client st) in do

352+ case pid of

353+ Just p -> (flip forward) p msg

354+ Nothing -> return ()

355+

356+

src/Control/Distributed/Process/Internal/Trace/Types.hsnew file+284

@@ -0,0 +1,284 @@

1+-- | Tracing/Debugging support - Types

2+module Control.Distributed.Process.Internal.Trace.Types

3+ ( Tracer(..)

4+ , TraceEvent(..)

5+ , SetTrace(..)

6+ , Addressable(..)

7+ , TraceSubject(..)

8+ , TraceFlags(..)

9+ , TraceArg(..)

10+ , TraceOk(..)

11+ , traceLog

12+ , traceLogFmt

13+ , traceEvent

14+ , traceMessage

15+ , defaultTraceFlags

16+ , enableTrace

17+ , enableTraceSync

18+ , disableTrace

19+ , disableTraceSync

20+ , getTraceFlags

21+ , setTraceFlags

22+ , setTraceFlagsSync

23+ , getCurrentTraceClient

24+ ) where

25+

26+import Control.Applicative ((<$>), (<*>))

27+import Control.Distributed.Process.Internal.CQueue (enqueue)

28+import Control.Distributed.Process.Internal.Types

29+ ( Tracer(..)

30+ , NodeId

31+ , ProcessId

32+ , DiedReason

33+ , Message

34+ , SendPort

35+ , createUnencodedMessage

36+ )

37+import Control.Distributed.Process.Serializable

38+import Data.Binary

39+import Data.Foldable (forM_)

40+import Data.List (intersperse)

41+import Data.Set (Set)

42+import qualified Data.Set as Set (fromList)

43+import Data.Typeable

44+import System.Mem.Weak (deRefWeak)

45+import Network.Transport

46+ ( ConnectionId

47+ )

48+

49+--------------------------------------------------------------------------------

50+-- Types --

51+--------------------------------------------------------------------------------

52+

53+data SetTrace = TraceEnable !ProcessId | TraceDisable

54+ deriving (Typeable, Eq, Show)

55+

56+-- | An event that is fired when something /interesting/ happens.

57+-- Trace events are forwarded to a trace listener process, of which

58+-- at most one can be registered per-node.

59+data TraceEvent =

60+ TraceEvSpawned ProcessId

61+ -- ^ fired whenever a local process is spawned

62+ | TraceEvRegistered ProcessId String

63+ -- ^ fired whenever a process/name is registered (locally)

64+ | TraceEvUnRegistered ProcessId String

65+ -- ^ fired whenever a process/name is unregistered (locally)

66+ | TraceEvDied ProcessId DiedReason

67+ -- ^ fired whenever a process dies

68+ | TraceEvNodeDied NodeId DiedReason

69+ -- ^ fired whenever a node /dies/ (i.e., the connection is broken/disconnected)

70+ | TraceEvSent ProcessId ProcessId Message

71+ -- ^ fired whenever a message is sent from a local process

72+ | TraceEvReceived ProcessId Message

73+ -- ^ fired whenever a message is received by a local process

74+ | TraceEvConnected ConnectionId

75+ -- ^ fired when a network-transport connection is first established

76+ | TraceEvDisconnected ConnectionId

77+ -- ^ fired when a network-transport connection is broken/disconnected

78+ | TraceEvUser Message

79+ -- ^ a user defined trace event (see 'traceMessage')

80+ | TraceEvLog String

81+ -- ^ a /logging/ event - used for debugging purposes only

82+ | TraceEvTakeover ProcessId

83+ -- ^ notifies a trace listener that all subsequent traces will be sent to /pid/

84+ | TraceEvDisable

85+ -- ^ notifies a trace listener that it has been disabled/removed

86+ deriving (Typeable, Show)

87+

88+class Addressable a where

89+ resolveToPid :: a -> Maybe ProcessId

90+

91+instance Addressable TraceEvent where

92+ resolveToPid (TraceEvSpawned p) = Just p

93+ resolveToPid (TraceEvDied p _) = Just p

94+ resolveToPid (TraceEvSent _ p _) = Just p

95+ resolveToPid (TraceEvReceived p _) = Just p

96+ resolveToPid _ = Nothing

97+

98+-- | Defines which processes will be traced by a given 'TraceFlag',

99+-- either by name, or @ProcessId@. Choosing @TraceAll@ is /by far/

100+-- the most efficient approach, as the tracer process therefore

101+-- avoids deciding whether or not a trace event is viable.

102+--

103+data TraceSubject =

104+ TraceAll -- enable tracing for all running processes

105+ | TraceProcs !(Set ProcessId) -- enable tracing for a set of processes

106+ | TraceNames !(Set String) -- enable tracing for a set of named/registered processes

107+ deriving (Show)

108+

109+-- | Defines /what/ will be traced. Flags that control tracing of

110+-- @Process@ events, take a 'TraceSubject' controlling which processes

111+-- should generate trace events in the target process.

112+data TraceFlags = TraceFlags {

113+ traceSpawned :: !(Maybe TraceSubject) -- filter process spawned tracing

114+ , traceDied :: !(Maybe TraceSubject) -- filter process died tracing

115+ , traceRegistered :: !(Maybe TraceSubject) -- filter process registration tracing

116+ , traceUnregistered :: !(Maybe TraceSubject) -- filter process un-registration

117+ , traceSend :: !(Maybe TraceSubject) -- filter process/message tracing by sender

118+ , traceRecv :: !(Maybe TraceSubject) -- filter process/message tracing by receiver

119+ , traceNodes :: !Bool -- enable node status trace events

120+ , traceConnections :: !Bool -- enable connection status trace events

121+ } deriving (Typeable, Show)

122+

123+defaultTraceFlags :: TraceFlags

124+defaultTraceFlags =

125+ TraceFlags {

126+ traceSpawned = Nothing

127+ , traceDied = Nothing

128+ , traceRegistered = Nothing

129+ , traceUnregistered = Nothing

130+ , traceSend = Nothing

131+ , traceRecv = Nothing

132+ , traceNodes = False

133+ , traceConnections = False

134+ }

135+

136+data TraceArg =

137+ TraceStr String

138+ | forall a. (Show a) => Trace a

139+

140+-- | A generic 'ok' response from the trace coordinator.

141+data TraceOk = TraceOk

142+ deriving (Typeable)

143+

144+--------------------------------------------------------------------------------

145+-- Internal/Common API --

146+--------------------------------------------------------------------------------

147+

148+traceLog :: Tracer -> String -> IO ()

149+traceLog tr s = traceIt tr (createUnencodedMessage $ TraceEvLog s)

150+

151+traceLogFmt :: Tracer

152+ -> String

153+ -> [TraceArg]

154+ -> IO ()

155+traceLogFmt t d ls =

156+ traceLog t $ concat (intersperse d (map toS ls))

157+ where toS :: TraceArg -> String

158+ toS (TraceStr s) = s

159+ toS (Trace a) = show a

160+

161+traceEvent :: Tracer -> TraceEvent -> IO ()

162+traceEvent tr ev = traceIt tr (createUnencodedMessage ev)

163+

164+traceMessage :: Serializable m => Tracer -> m -> IO ()

165+traceMessage tr msg = traceEvent tr (TraceEvUser (createUnencodedMessage msg))

166+

167+enableTrace :: Tracer -> ProcessId -> IO ()

168+enableTrace t p =

169+ traceIt t (createUnencodedMessage ((Nothing :: Maybe (SendPort TraceOk)),

170+ (TraceEnable p)))

171+

172+enableTraceSync :: Tracer -> SendPort TraceOk -> ProcessId -> IO ()

173+enableTraceSync t s p =

174+ traceIt t (createUnencodedMessage ((Just s), (TraceEnable p)))

175+

176+disableTrace :: Tracer -> IO ()

177+disableTrace t =

178+ traceIt t (createUnencodedMessage ((Nothing :: Maybe (SendPort TraceOk)),

179+ TraceDisable))

180+

181+disableTraceSync :: Tracer -> SendPort TraceOk -> IO ()

182+disableTraceSync t s =

183+ traceIt t (createUnencodedMessage ((Just s), TraceDisable))

184+

185+setTraceFlags :: Tracer -> TraceFlags -> IO ()

186+setTraceFlags t f =

187+ traceIt t (createUnencodedMessage ((Nothing :: Maybe (SendPort TraceOk)), f))

188+

189+setTraceFlagsSync :: Tracer -> SendPort TraceOk -> TraceFlags -> IO ()

190+setTraceFlagsSync t s f =

191+ traceIt t (createUnencodedMessage ((Just s), f))

192+

193+getTraceFlags :: Tracer -> SendPort TraceFlags -> IO ()

194+getTraceFlags t s = traceIt t (createUnencodedMessage s)

195+

196+getCurrentTraceClient :: Tracer -> SendPort (Maybe ProcessId) -> IO ()

197+getCurrentTraceClient t s = traceIt t (createUnencodedMessage s)

198+

199+class Traceable a where

200+ uod :: [a] -> TraceSubject

201+

202+instance Traceable ProcessId where

203+ uod = TraceProcs . Set.fromList

204+

205+instance Traceable String where

206+ uod = TraceNames . Set.fromList

207+

208+traceIt :: Tracer -> Message -> IO ()

209+traceIt InactiveTracer _ = return ()

210+traceIt (ActiveTracer _ wqRef) msg = do

211+ mQueue <- deRefWeak wqRef

212+ forM_ mQueue $ \queue -> enqueue queue msg

213+

214+--------------------------------------------------------------------------------

215+-- Binary Instances --

216+--------------------------------------------------------------------------------

217+

218+instance Binary SetTrace where

219+ put (TraceEnable pid) = putWord8 1 >> put pid

220+ put TraceDisable = putWord8 2

221+

222+ get = do

223+ header <- getWord8

224+ case header of

225+ 1 -> TraceEnable <$> get

226+ 2 -> return TraceDisable

227+ _ -> error "SetTrace.get - invalid header"

228+

229+instance Binary TraceEvent where

230+ put (TraceEvSpawned pid) = putWord8 1 >> put pid

231+ put (TraceEvRegistered pid str) = putWord8 2 >> put pid >> put str

232+ put (TraceEvUnRegistered pid str) = putWord8 3 >> put pid >> put str

233+ put (TraceEvDied pid res) = putWord8 4 >> put pid >> put res

234+ put (TraceEvNodeDied nid res) = putWord8 5 >> put nid >> put res

235+ put (TraceEvSent to from msg) = putWord8 6 >> put to >> put from >> put msg

236+ put (TraceEvReceived pid msg) = putWord8 7 >> put pid >> put msg

237+ put (TraceEvConnected cid) = putWord8 8 >> put cid

238+ put (TraceEvDisconnected cid) = putWord8 9 >> put cid

239+ put (TraceEvUser msg) = putWord8 10 >> put msg

240+ put (TraceEvLog msg) = putWord8 11 >> put msg

241+ put (TraceEvTakeover pid) = putWord8 12 >> put pid

242+ put TraceEvDisable = putWord8 13

243+

244+ get = do

245+ header <- getWord8

246+ case header of

247+ 1 -> TraceEvSpawned <$> get

248+ 2 -> TraceEvRegistered <$> get <*> get

249+ 3 -> TraceEvUnRegistered <$> get <*> get

250+ 4 -> TraceEvDied <$> get <*> get

251+ 5 -> TraceEvNodeDied <$> get <*> get

252+ 6 -> TraceEvSent <$> get <*> get <*> get

253+ 7 -> TraceEvReceived <$> get <*> get

254+ 8 -> TraceEvConnected <$> get

255+ 9 -> TraceEvDisconnected <$> get

256+ 10 -> TraceEvUser <$> get

257+ 11 -> TraceEvLog <$> get

258+ 12 -> TraceEvTakeover <$> get

259+ 13 -> return TraceEvDisable

260+ _ -> error "TraceEvent.get - invalid header"

261+

262+instance Binary TraceSubject where

263+ put TraceAll = putWord8 1

264+ put (TraceProcs pids) = putWord8 2 >> put pids

265+ put (TraceNames names) = putWord8 3 >> put names

266+

267+ get = do

268+ header <- getWord8

269+ case header of

270+ 1 -> return TraceAll

271+ 2 -> TraceProcs <$> get

272+ 3 -> TraceNames <$> get

273+ _ -> error "TraceSubject.get - invalid header"

274+

275+instance Binary TraceFlags where

276+ put (TraceFlags s d g u m r n c) =

277+ put s >> put d >> put g >> put u >> put m >> put r >> put n >> put c

278+ get =

279+ TraceFlags <$> get <*> get <*> get <*> get <*> get <*> get <*> get <*> get

280+

281+instance Binary TraceOk where

282+ put _ = return ()

283+ get = return TraceOk

284+

src/Control/Distributed/Process/Internal/Types.hs+15−7

@@ -93,7 +93,6 @@ import Control.Exception (Exception)

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+124−51

@@ -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,19 @@ import Control.Distributed.Process.Internal.Types

148149 , firstNonReservedProcessId

149150 , ImplicitReconnect(WithImplicitReconnect)

150151 )

151−import Control.Distributed.Process.Internal.Trace

152+import qualified Control.Distributed.Process.Internal.Trace.Remote as Trace

153+ ( remoteTable

154+ )

155+import Control.Distributed.Process.Internal.Trace.Types

152156 ( TraceArg(..)

153− , traceFormat

154− , startTracing

155− , stopTracer

157+ , TraceEvent(..)

158+ , traceEvent

159+ , traceLogFmt

160+ , enableTrace

161+ )

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

163+ ( traceController

164+ , defaultTracer

156165 )

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

158167import Control.Distributed.Process.Internal.Messaging

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

175184 ( mapMaybe

176185 , mapDefault

177186 )

178−

187+import System.Environment (getEnv)

179188import Unsafe.Coerce

180189

181190--------------------------------------------------------------------------------

@@ -183,7 +192,7 @@ import Unsafe.Coerce

183192--------------------------------------------------------------------------------

184193

185194initRemoteTable :: RemoteTable

186−initRemoteTable = BuiltIn.remoteTable Static.initRemoteTable

195+initRemoteTable = Trace.remoteTable $ BuiltIn.remoteTable Static.initRemoteTable

187196

188197-- | Initialize a new local node.

189198newLocalNode :: NT.Transport -> RemoteTable -> IO LocalNode

@@ -232,10 +241,51 @@ createBareLocalNode endPoint rtable = do

232241 , ctrlMsgSignal = SigShutdown

233242 }

234243

244+startTracing :: LocalNode -> IO LocalNode

245+startTracing node =

246+ -- TODO: use tryGetEnv once we drop support for GHC 7.2.x

247+ Exception.catch

248+ (getEnv "DISTRIBUTED_PROCESS_TRACE_ENABLED" >> procTracer node)

249+ (\(_ :: IOError) -> return node)

250+ where procTracer :: LocalNode -> IO LocalNode

251+ procTracer n = do

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

253+ mv <- MVar.newEmptyMVar

254+ pid <- forkProcess n (traceController mv)

255+ wqRef <- MVar.takeMVar mv

256+ return n { localTracer = (ActiveTracer pid wqRef) }

257+

258+startDefaultTracer :: LocalNode -> IO (Maybe ProcessId)

259+startDefaultTracer node =

260+ Exception.catch

261+ (getEnv "DISTRIBUTED_PROCESS_TRACE_ENABLED" >> go node)

262+ (\(_ :: IOError) -> return Nothing)

263+ where go :: LocalNode -> IO (Maybe ProcessId)

264+ go node' = do

265+ registerTraceController node'

266+ pid <- forkProcess node' defaultTracer

267+ enableTrace (localTracer node') pid

268+ return (Just pid)

269+

270+ registerTraceController :: LocalNode -> IO ()

271+ registerTraceController n =

272+ let t = localTracer n in

273+ case t of

274+ InactiveTracer -> return ()

275+ (ActiveTracer pid _) -> runProcess n $ register "trace.controller" pid

276+

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

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

237278startServiceProcesses :: LocalNode -> IO ()

238279startServiceProcesses node = do

280+ -- tracing /spawns/ relies on the tracer being enabled, but we start

281+ -- the default tracer first, even though it might @nsend@ to the logger

282+ -- before /that/ process has started - this is a totally harmless race

283+ -- however, so we deliberably ignore it

284+ mPid <- startDefaultTracer node

285+ case mPid of

286+ Nothing -> return ()

287+ Just pid -> runProcess node $ register "tracer.initial" pid

288+

239289 logger <- forkProcess node loop

240290 runProcess node $ register "logger" logger

241291 where

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

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

258308closeLocalNode :: LocalNode -> IO ()

259309closeLocalNode node =

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

260311 NT.closeEndPoint (localEndPoint node)

261312

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

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

306357 }

307358 return (tid', lproc)

308359

360+

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

362+ trace node (TraceEvSpawned pid)

363+

309364 if lpidCounter lpid == maxBound

310365 then do

311366 newUnique <- randomIO

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

331386 . (localConnections ^= unaffected)

332387 $ st

333388

389+-- note [tracer/forkProcess races]

390+--

391+-- Our startTracing function uses forkProcess to start the trace controller

392+-- process, and of course forkProcess attempts to call traceEvent once the

393+-- process has started. This is harmless, as the localTracer is not updated

394+-- until /after/ the initial forkProcess completes, so the first call to

395+-- traceEvent behaves as if tracing were disabled (i.e., it is ignored).

396+--

397+

334398--------------------------------------------------------------------------------

335399-- Handle incoming messages --

336400--------------------------------------------------------------------------------

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

339403

340404data IncomingTarget =

341405 Uninit

342− | ToProc (Weak (CQueue Message))

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

343407 | ToChan TypedChannel

344408 | ToNode

345409

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

374438 case event of

375439 NT.ConnectionOpened cid rel theirAddr ->

376440 if rel == NT.ReliableOrdered

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

441+ then

442+ trace node (TraceEvConnected cid)

443+ >> go (

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

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

379446 $ st

380447 )

381448 else invalidRequest cid st

382449 NT.Received cid payload ->

383450 case st ^. incomingAt cid of

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

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

385452 mQueue <- deRefWeak weakQueue

386453 forM_ mQueue $ \queue -> do

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

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

389456 let msg = payloadToMessage payload

390457 enqueue queue msg -- 'enqueue' is strict

458+ trace node (TraceEvReceived pid msg)

391459 go st

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

393461 mChan <- deRefWeak chan'

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

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

409477 case mProc of

410478 Just proc ->

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

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

412480 Nothing ->

413481 invalidRequest cid st

414482 SendPortIdentifier chId -> do

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

435503 case st ^. incomingAt cid of

436504 Nothing ->

437505 invalidRequest cid st

438− Just (src, _) ->

506+ Just (src, _) -> do

507+ trace node (TraceEvDisconnected cid)

439508 go ( (incomingAt cid ^= Nothing)

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

441510 $ st

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

454523 $ st

455524 )

456525 NT.ErrorEvent (NT.TransportError NT.EventEndPointFailed str) ->

457− fatal $ "Cloud Haskell fatal error: end point failed: " ++ str

526+ fail $ "Cloud Haskell fatal error: end point failed: " ++ str

458527 NT.ErrorEvent (NT.TransportError NT.EventTransportFailed str) ->

459− fatal $ "Cloud Haskell fatal error: transport failed: " ++ str

528+ fail $ "Cloud Haskell fatal error: transport failed: " ++ str

460529 NT.EndPointClosed ->

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

530+ return ()

462531 NT.ReceivedMulticast _ _ ->

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

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

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

469535

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

471537 invalidRequest cid st = do

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

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

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

475541 -- extension).

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

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

477543 (Trace cid)]

478544 go ( incomingAt cid ^= Nothing

479545 $ st

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

534600

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

536602traceNotifyDied 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

603+ -- TODO: sendPortDied notifications

604+ liftIO $ withLocalTracer node $ \t ->

605+ case ident of

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

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

608+ _ -> return ()

547609

548610traceEventFmtIO :: LocalNode

549611 -> String

550612 -> [TraceArg]

551613 -> IO ()

552614traceEventFmtIO node fmt args =

553− withLocalTracer node $ \t -> traceFormat t fmt args

615+ withLocalTracer node $ \t -> traceLogFmt t fmt args

616+

617+trace :: LocalNode -> TraceEvent -> IO ()

618+trace node ev = withLocalTracer node $ \t -> traceEvent t ev

554619

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

556621withLocalTracer node act = act (localTracer node)

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

597662 NCMsg _ (NamedSend label msg') ->

598663 ncEffectNamedSend label msg'

599664 NCMsg _ (LocalSend to msg') ->

600− ncEffectLocalSend to msg'

665+ ncEffectLocalSend node to msg'

601666 NCMsg _ (LocalPortSend to msg') ->

602667 ncEffectLocalPortSend to msg'

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

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

720785 mProc <- unClosure cProc

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

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

788+ -- TODO:

723789 let proc = case mProc of

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

725791 Right p -> p

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

750816 then do when (isOk) $

751817 do modify' $ registeredHereFor label ^= mPid

752818 updateRemote node currentVal mPid

819+ case mPid of

820+ (Just p) -> liftIO $ trace node (TraceEvRegistered p label)

821+ Nothing -> liftIO $ trace node (TraceEvUnRegistered (fromJust currentVal) label)

753822 liftIO $ sendMessage node

754823 (NodeIdentifier (localNodeId node))

755824 (ProcessIdentifier from)

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

772841 updateRemote _ _ _ = return ()

773842 maybeify f Nothing = unmaybeify $ f []

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

844+

775845 unmaybeify [] = Nothing

776846 unmaybeify x = Just x

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

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

812882 forM_ mPid $ \pid -> postMessage pid msg

813883

814884-- [Issue #DP-20]

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

816−ncEffectLocalSend = postMessage

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

886+ncEffectLocalSend node to msg =

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

888+ enqueue (processQueue p) msg

889+ trace node (TraceEvReceived to msg)

817890

818891-- [Issue #DP-20]

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

820−ncEffectLocalPortSend from msg =

893+ncEffectLocalPortSend from msg = do

894+ node <- ask

821895 let pid = sendPortProcessId from

822896 cid = sendPortLocalId from

823− in do

824− withLocalProc pid $ \proc -> do

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

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

826899 case mChan of

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

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

9951068

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

9971070postMessage pid msg = do

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

1071+ node <- ask

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

9991073

10001074throwException :: 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

1075+throwException pid e = do

10061076 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

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

1078+

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

1080+withLocalProc node pid p =

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

1082+ -- can silently be dropped

1083+ let lpid = processLocalId pid in do

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

1085+ forM_ mProc p

10131086

10141087--------------------------------------------------------------------------------

10151088-- Accessors --

tests/TestStats.hs+2−2

@@ -111,7 +111,7 @@ testRemoteLiveProcessInfo node1 = do

111111 "pong" <- expect

112112 pInfo <- getProcessInfo serverPid

113113 stash result $ pInfo /= Nothing

114− where

114+ where

115115 launchRemote :: MVar ProcessId -> IO ()

116116 launchRemote locMV = do

117117 node2 <- liftIO $ mkNode "8082"

@@ -119,7 +119,7 @@ testRemoteLiveProcessInfo node1 = do

119119 self <- getSelfPid

120120 liftIO $ putMVar locMV self

121121 _ <- receiveWait [

122− match (\(pid, "ping") -> send pid "pong")

122+ match (\(pid, "ping") -> send pid "pong")

123123 ]

124124 "stop" <- expect

125125 return ()

tests/TestTracing.hsnew file+388

@@ -0,0 +1,388 @@

1+{-# LANGUAGE DeriveDataTypeable #-}

2+{-# OPTIONS_GHC -fno-warn-orphans #-}

3+module Main where

5+import Control.Concurrent (threadDelay)

6+import Control.Concurrent.MVar

7+ ( MVar

8+ , newEmptyMVar

9+ , newMVar

10+ , putMVar

11+ , takeMVar

12+ )

13+import Control.Distributed.Process

14+import Control.Distributed.Process.Debug

15+import Control.Distributed.Process.Node

16+ ( forkProcess

17+ , newLocalNode

18+ , initRemoteTable

19+ , closeLocalNode

20+ , LocalNode)

21+import Control.Exception as Ex (catch)

22+import Network.Transport.TCP

23+import Prelude hiding (catch, log)

24+import Test.Framework

25+ ( Test

26+ , defaultMain

27+ , testGroup

28+ )

29+import Test.HUnit (Assertion)

30+import Test.HUnit.Base (assertBool)

31+import Test.Framework.Providers.HUnit (testCase)

32+

33+import System.Environment (getEnv)

34+import System.Posix.Env

35+ ( setEnv

36+ )

37+

38+-- these utilities have been cribbed from distributed-process-platform

39+-- we should really find a way to share them...

40+

41+-- | A mutable cell containing a test result.

42+type TestResult a = MVar a

43+

44+delayedAssertion :: (Eq a)

45+ => String

46+ -> LocalNode

47+ -> a

48+ -> (TestResult a -> Process ())

49+ -> MVar ()

50+ -> Assertion

51+delayedAssertion note localNode expected testProc lock = do

52+ result <- newEmptyMVar

53+ _ <- forkProcess localNode $ do

54+ acquire lock

55+ finally (testProc result)

56+ (release lock)

57+ assertComplete note result expected

58+ where acquire lock' = liftIO $ takeMVar lock'

59+ release lock' = liftIO $ putMVar lock' ()

60+

61+assertComplete :: (Eq a) => String -> MVar a -> a -> IO ()

62+assertComplete msg mv a = do

63+ b <- takeMVar mv

64+ assertBool msg (a == b)

65+

66+stash :: TestResult a -> a -> Process ()

67+stash mvar x = liftIO $ putMVar mvar x

68+

69+------

70+

71+testSpawnTracing :: TestResult Bool -> Process ()

72+testSpawnTracing result = do

73+ setTraceFlags defaultTraceFlags {

74+ traceSpawned = (Just TraceAll)

75+ , traceDied = (Just TraceAll)

76+ }

77+

78+ evSpawned <- liftIO $ newEmptyMVar

79+ evDied <- liftIO $ newEmptyMVar

80+ tracer <- startTracer $ \ev -> do

81+ case ev of

82+ (TraceEvSpawned p) -> liftIO $ putMVar evSpawned p

83+ (TraceEvDied p r) -> liftIO $ putMVar evDied (p, r)

84+ _ -> return ()

85+

86+ (sp, rp) <- newChan

87+ pid <- spawnLocal $ sendChan sp ()

88+ () <- receiveChan rp

89+

90+ tracedAlive <- liftIO $ takeMVar evSpawned

91+ (tracedDead, tracedReason) <- liftIO $ takeMVar evDied

92+

93+ mref <- monitor tracer

94+ stopTracer -- this is asynchronous, so we need to wait...

95+ receiveWait [

96+ matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mref)

97+ ((\_ -> return ()))

98+ ]

99+ setTraceFlags defaultTraceFlags

100+ stash result (tracedAlive == pid &&

101+ tracedDead == pid &&

102+ tracedReason == DiedNormal)

103+

104+testTraceRecvExplicitPid :: TestResult Bool -> Process ()

105+testTraceRecvExplicitPid result = do

106+ res <- liftIO $ newEmptyMVar

107+ pid <- spawnLocal $ do

108+ self <- getSelfPid

109+ expect >>= (flip sendChan) self

110+ withFlags defaultTraceFlags {

111+ traceRecv = traceOnly [pid]

112+ } $ do

113+ withTracer

114+ (\ev ->

115+ case ev of

116+ (TraceEvReceived pid' _) -> stash res (pid == pid')

117+ _ -> return ()) $ do

118+ (sp, rp) <- newChan

119+ send pid sp

120+ p <- receiveChan rp

121+ res' <- liftIO $ takeMVar res

122+ stash result (res' && (p == pid))

123+ return ()

124+

125+testTraceRecvNamedPid :: TestResult Bool -> Process ()

126+testTraceRecvNamedPid result = do

127+ res <- liftIO $ newEmptyMVar

128+ pid <- spawnLocal $ do

129+ self <- getSelfPid

130+ register "foobar" self

131+ expect >>= (flip sendChan) self

132+ withFlags defaultTraceFlags {

133+ traceRecv = traceOnly ["foobar"]

134+ } $ do

135+ withTracer

136+ (\ev ->

137+ case ev of

138+ (TraceEvReceived pid' _) -> stash res (pid == pid')

139+ _ -> return ()) $ do

140+ (sp, rp) <- newChan

141+ send pid sp

142+ p <- receiveChan rp

143+ res' <- liftIO $ takeMVar res

144+ stash result (res' && (p == pid))

145+ return ()

146+

147+testTraceSending :: TestResult Bool -> Process ()

148+testTraceSending result = do

149+ pid <- spawnLocal $ (expect :: Process String) >> return ()

150+ self <- getSelfPid

151+ res <- liftIO $ newEmptyMVar

152+ withFlags defaultTraceFlags { traceSend = traceOn } $ do

153+ withTracer

154+ (\ev ->

155+ case ev of

156+ (TraceEvSent to from _) ->

157+ stash res (to == pid && from == self)

158+ _ ->

159+ return ()) $ do

160+ send pid "hello there"

161+ res' <- liftIO $ takeMVar res

162+ stash result res'

163+

164+testTraceRegistration :: TestResult Bool -> Process ()

165+testTraceRegistration result = do

166+ (sp, rp) <- newChan

167+ pid <- spawnLocal $ do

168+ self <- getSelfPid

169+ () <- expect

170+ register "foobar" self

171+ sendChan sp ()

172+ () <- expect

173+ return ()

174+ res <- liftIO $ newEmptyMVar

175+ withFlags defaultTraceFlags { traceRegistered = traceOn } $ do

176+ withTracer

177+ (\ev ->

178+ case ev of

179+ TraceEvRegistered p s ->

180+ stash res (p == pid && s == "foobar")

181+ _ ->

182+ return ()) $ do

183+ _ <- monitor pid

184+ send pid ()

185+ () <- receiveChan rp

186+ send pid ()

187+ receiveWait [

188+ match (\(ProcessMonitorNotification _ _ _) -> return ())

189+ ]

190+ res' <- liftIO $ takeMVar res

191+ stash result res'

192+

193+testTraceUnRegistration :: TestResult Bool -> Process ()

194+testTraceUnRegistration result = do

195+ pid <- spawnLocal $ do

196+ () <- expect

197+ unregister "foobar"

198+ () <- expect

199+ return ()

200+ register "foobar" pid

201+ res <- liftIO $ newEmptyMVar

202+ withFlags defaultTraceFlags { traceUnregistered = traceOn } $ do

203+ withTracer

204+ (\ev ->

205+ case ev of

206+ TraceEvUnRegistered p n -> do

207+ stash res (p == pid && n == "foobar")

208+ send pid ()

209+ _ ->

210+ return ()) $ do

211+ mref <- monitor pid

212+ send pid ()

213+ receiveWait [

214+ matchIf (\(ProcessMonitorNotification mref' _ _) -> mref == mref')

215+ (\_ -> return ())

216+ ]

217+ res' <- liftIO $ takeMVar res

218+ stash result res'

219+

220+testTraceLayering :: TestResult () -> Process ()

221+testTraceLayering result = do

222+ pid <- spawnLocal $ do

223+ getSelfPid >>= register "foobar"

224+ () <- expect

225+ traceMessage ("traceMsg", 123 :: Int)

226+ return ()

227+ withFlags defaultTraceFlags {

228+ traceDied = traceOnly [pid]

229+ , traceRecv = traceOnly ["foobar"]

230+ } $ doTest pid result

231+ return ()

232+ where

233+ doTest :: ProcessId -> MVar () -> Process ()

234+ doTest pid result' = do

235+ -- TODO: this is pretty gross, even for a test case

236+ died <- liftIO $ newEmptyMVar

237+ withTracer

238+ (\ev ->

239+ case ev of

240+ TraceEvDied _ _ -> liftIO $ putMVar died ()

241+ _ -> return ())

242+ ( do {

243+ recv <- liftIO $ newEmptyMVar

244+ ; withTracer

245+ (\ev' ->

246+ case ev' of

247+ TraceEvReceived _ _ -> liftIO $ putMVar recv ()

248+ _ -> return ())

249+ ( do {

250+ user <- liftIO $ newEmptyMVar

251+ ; withTracer

252+ (\ev'' ->

253+ case ev'' of

254+ TraceEvUser _ -> liftIO $ putMVar user ()

255+ _ -> return ())

256+ (send pid () >> (liftIO $ takeMVar user))

257+ ; liftIO $ takeMVar recv

258+ })

259+ ; liftIO $ takeMVar died

260+ })

261+ liftIO $ putMVar result' ()

262+

263+testRemoteTraceRelay :: TestResult Bool -> Process ()

264+testRemoteTraceRelay result =

265+ let flags = defaultTraceFlags { traceSpawned = traceOn }

266+ in do

267+ node2 <- liftIO $ mkNode "8082"

268+ mvNid <- liftIO $ newEmptyMVar

269+

270+ -- As well as needing node2's NodeId, we want to

271+ -- redirect all its logs back here, to avoid generating

272+ -- garbage on stderr for the duration of the test run.

273+ -- Here we set up that relay, and then wait for a signal

274+ -- that the tracer (on node1) has seen the expected

275+ -- TraceEvSpawned message, at which point we're finished

276+ (Just log) <- whereis "logger"

277+ pid <- liftIO $ forkProcess node2 $ do

278+ logRelay <- spawnLocal $ relay log

279+ reregister "logger" logRelay

280+ getSelfNode >>= stash mvNid >> (expect :: Process ())

281+

282+ nid <- liftIO $ takeMVar mvNid

283+ mref <- monitor pid

284+ observedPid <- liftIO $ newEmptyMVar

285+ spawnedPid <- liftIO $ newEmptyMVar

286+ setTraceFlagsRemote flags nid

287+

288+ withFlags defaultTraceFlags { traceSpawned = traceOn } $ do

289+ withTracer

290+ (\ev ->

291+ case ev of

292+ TraceEvSpawned p -> stash observedPid p >> send pid ()

293+ _ -> return ()) $ do

294+ relayPid <- startTraceRelay nid

295+ liftIO $ threadDelay 1000000

296+ p <- liftIO $ forkProcess node2 $ do

297+ expectTimeout 1000000 :: Process (Maybe ())

298+ return ()

299+ stash spawnedPid p

300+

301+ -- Now we wait for (the outer) pid to exit. This won't happen until

302+ -- our tracer has seen the trace event for `p' and send `p' the

303+ -- message its waiting for prior to exiting

304+ receiveWait [

305+ matchIf (\(ProcessMonitorNotification mref' _ _) -> mref == mref')

306+ (\_ -> return ())

307+ ]

308+

309+ relayRef <- monitor relayPid

310+ kill relayPid "stop"

311+ receiveWait [

312+ matchIf (\(ProcessMonitorNotification rref' _ _) -> rref' == relayRef)

313+ (\_ -> return ())

314+ ]

315+ observed <- liftIO $ takeMVar observedPid

316+ expected <- liftIO $ takeMVar spawnedPid

317+ stash result (observed == expected)

318+ -- and just to be polite...

319+ liftIO $ closeLocalNode node2

320+

321+tests :: LocalNode -> IO [Test]

322+tests node1 = do

323+ -- if we execute the test cases in parallel, the

324+ -- various tracers will race with one another and

325+ -- we'll get garbage results (or worse, deadlocks)

326+ lock <- liftIO $ newMVar ()

327+ enabled <- checkTraceEnabled

328+ case enabled of

329+ True ->

330+ return [

331+ testGroup "Tracing" [

332+ testCase "Spawn Tracing"

333+ (delayedAssertion

334+ "expected dead process-info to be ProcessInfoNone"

335+ node1 True testSpawnTracing lock)

336+ , testCase "Recv Tracing (Explicit Pid)"

337+ (delayedAssertion

338+ "expected a recv trace for the supplied pid"

339+ node1 True testTraceRecvExplicitPid lock)

340+ , testCase "Recv Tracing (Named Pid)"

341+ (delayedAssertion

342+ "expected a recv trace for the process registered as 'foobar'"

343+ node1 True testTraceRecvNamedPid lock)

344+ , testCase "Trace Send(er)"

345+ (delayedAssertion

346+ "expected a 'send' trace with the requisite fields set"

347+ node1 True testTraceSending lock)

348+ , testCase "Trace Registration"

349+ (delayedAssertion

350+ "expected a 'registered' trace"

351+ node1 True testTraceRegistration lock)

352+ , testCase "Trace Unregistration"

353+ (delayedAssertion

354+ "expected an 'unregistered' trace"

355+ node1 True testTraceUnRegistration lock)

356+ , testCase "Trace Layering"

357+ (delayedAssertion

358+ "expected blah"

359+ node1 () testTraceLayering lock)

360+ , testCase "Remote Trace Relay"

361+ (delayedAssertion

362+ "expected blah"

363+ node1 True testRemoteTraceRelay lock)

364+ ] ]

365+ False ->

366+ return []

367+

368+checkTraceEnabled :: IO Bool

369+checkTraceEnabled =

370+ Ex.catch (getEnv "DISTRIBUTED_PROCESS_TRACE_ENABLED" >> return True)

371+ (\(_ :: IOError) -> return False)

372+

373+

374+mkNode :: String -> IO LocalNode

375+mkNode port = do

376+ Right (transport1, _) <- createTransportExposeInternals

377+ "127.0.0.1" port defaultTCPParameters

378+ newLocalNode transport1 initRemoteTable

379+

380+main :: IO ()

381+main = do

382+ setEnv "DISTRIBUTED_PROCESS_TRACE_ENABLED" "true" True

383+ node1 <- mkNode "8081"

384+ testData <- tests node1

385+ defaultMain testData

386+ closeLocalNode node1

387+ return ()

388+