39fb200 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Additional support for remote tracing

39fb200Tim Watson14 years ago

distributed-process.cabal+2

compare with master

unifiedsplit

@@ -65,6 +65,7 @@ Library

65 Control.Distributed.Process.Debug,

66 Control.Distributed.Process.Node

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

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

68 Control.Distributed.Process.Internal.CQueue,

69 Control.Distributed.Process.Internal.CQueue,

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

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

70 Control.Distributed.Process.Internal.Messaging,

71 Control.Distributed.Process.Internal.Messaging,

@@ -75,6 +76,7 @@ Library

75 Control.Distributed.Process.Internal.Trace.Types,

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

76 Control.Distributed.Process.Internal.Trace.Primitives,

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

77 Control.Distributed.Process.Internal.Trace.Tracer

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

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

78 Extensions: RankNTypes,

80 Extensions: RankNTypes,

79 ScopedTypeVariables,

81 ScopedTypeVariables,

80 FlexibleInstances,

82 FlexibleInstances,

src/Control/Distributed/Process.hs+10−110

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

280 , reconnectPort

281 )

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

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

284+ ( -- Spawning Processes/Channels

285+ spawn

286+ , spawnLink

287+ , spawnMonitor

288+ , spawnChannel

289+ , spawnSupervised

290+ , call

291+ )

283

292

284-- INTERNAL NOTES

293-- INTERNAL NOTES

285--

294--

@@ -344,116 +353,6 @@ import Control.Distributed.Process.Node (forkProcess)

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

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

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

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

346

355

347−--------------------------------------------------------------------------------

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

349−-- constructs --

350−--------------------------------------------------------------------------------

351−

352−-- | Spawn a process

353−--

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

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

356−--

357−-- See also 'call'.

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

359−spawn nid proc = do

360− us <- getSelfPid

361− mRef <- monitorNode nid

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

363− receiveWait [

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

365− unmonitor mRef

366− send pid ()

367− return pid

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

369− return (nullProcessId nid)

370− ]

371−

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

373−--

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

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

376−-- a synchronous link operation)

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

378−spawnLink nid proc = do

379− pid <- spawn nid proc

380− link pid

381− return pid

382−

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

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

385−spawnMonitor nid proc = do

386− pid <- spawn nid proc

387− ref <- monitor pid

388− return (pid, ref)

389−

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

391−--

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

393−-- too.

394−--

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

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

397−--

398−-- See also 'spawn'.

399−call :: Serializable a

400− => Static (SerializableDict a)

401− -> NodeId

402− -> Closure (Process a)

403− -> Process a

404−call dict nid proc = do

405− us <- getSelfPid

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

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

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

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

410− mResult <- receiveWait

411− [ match (return . Right)

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

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

414− ]

415− case mResult of

416− Right a -> do

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

418− receiveWait

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

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

421− ]

422− -- Clean up connection to pid

423− reconnect pid

424− return a

425− Left err ->

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

427−

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

429−-- monitor the child

430−spawnSupervised :: NodeId

431− -> Closure (Process ())

432− -> Process (ProcessId, MonitorRef)

433−spawnSupervised nid proc = do

434− us <- getSelfPid

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

436− ref <- monitor them

437− return (them, ref)

438−

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

440−-- the corresponding 'SendPort'.

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

442− -> NodeId

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

444− -> Process (SendPort a)

445−spawnChannel dict nid proc = do

446− us <- getSelfPid

447− _ <- spawn nid (go us)

448− expect

449− where

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

451− go pid = cpNewChan dict

452− `bindCP`

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

454− `bindCP`

455− (idCP `closureCompose` staticClosure sndStatic)

456−

457--------------------------------------------------------------------------------

356--------------------------------------------------------------------------------

458-- Local versions of spawn --

357-- Local versions of spawn --

459--------------------------------------------------------------------------------

358--------------------------------------------------------------------------------

@@ -480,3 +379,4 @@ spawnChannelLocal proc = do

480 liftIO $ putMVar mvar sport

379 liftIO $ putMVar mvar sport

481 proc rport

380 proc rport

482 takeMVar mvar

381 takeMVar mvar

382+

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

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

173 -- * CP versions of Cloud Haskell primitives

174 , cpLink

175 , cpUnlink

176+ , cpRelay

176 , cpSend

177 , cpSend

177 , cpExpect

178 , cpExpect

178 , cpNewChan

179 , cpNewChan

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

205 -- CP versions of Cloud Haskell primitives

206 -- CP versions of Cloud Haskell primitives

206 , cpLink

207 , cpLink

207 , cpUnlink

208 , cpUnlink

209+ , cpRelay

208 , cpSend

210 , cpSend

209 , cpExpect

211 , cpExpect

210 , cpNewChan

212 , cpNewChan

src/Control/Distributed/Process/Debug.hs+45−131

@@ -10,24 +10,25 @@

10--

11-- [Tracing/Debugging Facilities]

12--

13−-- Cloud Haskell provides a general purpose tracing mechanism that allows a

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

14−-- user supplied /tracer process/ to receive messages when certain events

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

15−-- take place. It's possible to use this facility to debug a running system

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

16−-- and/or generate runtime monitoring data in a variety of formats.

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

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

21−-- real) cost at runtime. To enable any kind of tracing to work, the environment

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

22−-- variable @DISTRIBUTED_PROCESS_TRACE_ENABLED@ must be set (the contents are

22+-- @DISTRIBUTED_PROCESS_TRACE_ENABLED@ must be set to some value (the contents

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

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

24−-- a /trace events/ in various circumstances - see 'TraceEvent' for a list of

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

25−-- all the published events. A user can additionally publish custom trace events

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

26−-- in the form of 'TraceEvLog' log messages or pass custom (i.e., completely

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

27−-- user defined) event data using the 'traceMessage' function. If the

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

28−-- environment variable is not set, no trace events will ever be published.

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

29+-- published.

29--

30--

30−-- All published trace events are forwarded to a /tracer process/, which can be

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

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

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

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

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

33-- or the GHC eventlog.

34-- or the GHC eventlog.

@@ -100,16 +101,15 @@ module Control.Distributed.Process.Debug

100 , enableTrace

101 , enableTrace

101 , enableTraceAsync

102 , enableTraceAsync

102 , disableTrace

103 , disableTrace

103− , disableTraceAsync

104 , withTracer

105 , withFlags

106 , getTraceFlags

107 , setTraceFlags

108 , setTraceFlagsAsync

109 , defaultTraceFlags

110− , Tracer.traceOn

110+ , traceOn

111− , Tracer.traceOnly

111+ , traceOnly

112− , Tracer.traceOff

112+ , traceOff

113 -- * Debugging

114 , startTracer

115 , stopTracer

@@ -117,6 +117,10 @@ module Control.Distributed.Process.Debug

117 , traceLog

118 , traceLogFmt

119 , traceMessage

120+ -- * Working with remote nodes

121+ , Remote.remoteTable

122+ , Remote.startTraceRelay

123+ , Remote.setTraceFlagsRemote

120 -- * Built in tracers

124 -- * Built in tracers

121 , systemLoggerTracer

125 , systemLoggerTracer

122 , logfileTracer

126 , logfileTracer

@@ -129,12 +133,9 @@ import Control.Distributed.Process.Internal.Primitives

129 ( proxy

133 ( proxy

130 , finally

134 , finally

131 , die

135 , die

132− , getSelfPid

133− , reregister

134 , whereis

136 , whereis

135 , send

137 , send

136 , newChan

138 , newChan

137− , sendChan

138 , receiveChan

139 , receiveChan

139 , receiveWait

140 , receiveWait

140 , matchIf

141 , matchIf

@@ -143,6 +144,9 @@ import Control.Distributed.Process.Internal.Primitives

143 , monitor

144 , monitor

144 , say

145 , say

145 )

146 )

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

148+ ( spawn

149+ )

146import Control.Distributed.Process.Internal.Types

150import Control.Distributed.Process.Internal.Types

147 ( Tracer(..)

151 ( Tracer(..)

148 , LocalNode(..)

152 , LocalNode(..)

@@ -151,6 +155,7 @@ import Control.Distributed.Process.Internal.Types

151 , LocalProcess(..)

155 , LocalProcess(..)

152 , SendPort(..)

156 , SendPort(..)

153 , ProcessMonitorNotification(..)

157 , ProcessMonitorNotification(..)

158+ , NodeId(..)

154 )

159 )

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

160import Control.Distributed.Process.Internal.Trace.Types

156 ( TraceArg(..)

161 ( TraceArg(..)

@@ -165,22 +170,23 @@ import Control.Distributed.Process.Internal.Trace.Tracer

165 , logfileTracer

170 , logfileTracer

166 , eventLogTracer

171 , eventLogTracer

167 )

172 )

168−import qualified Control.Distributed.Process.Internal.Trace.Primitives as Tracer

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

169− ( traceLog

174+ ( withRegisteredTracer

170− , traceLogFmt

171− , traceMessage

172 , enableTrace

175 , enableTrace

173− , enableTraceSync

176+ , enableTraceAsync

174 , disableTrace

177 , disableTrace

175− , disableTraceSync

176 , setTraceFlags

178 , setTraceFlags

177− , setTraceFlagsSync

179+ , setTraceFlagsAsync

178 , getTraceFlags

180 , getTraceFlags

179− , getCurrentTraceClient

181+ , isTracingEnabled

180 , traceOn

182 , traceOn

181 , traceOff

183 , traceOff

182 , traceOnly

184 , traceOnly

185+ , traceLog

186+ , traceLogFmt

187+ , traceMessage

183 )

188 )

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

184import Control.Distributed.Process.Node

190import Control.Distributed.Process.Node

185import Control.Distributed.Process.Serializable

191import Control.Distributed.Process.Serializable

186

192

@@ -222,7 +228,6 @@ withTracer :: forall a.

222 -> Process a

228 -> Process a

223 -> Process (Either SomeException a)

229 -> Process (Either SomeException a)

224withTracer handler proc = do

230withTracer handler proc = do

225− say "starting new tracer"

226 previous <- whereis "tracer"

231 previous <- whereis "tracer"

227 tracer <- startTracer handler

232 tracer <- startTracer handler

228 finally (try proc)

233 finally (try proc)

@@ -234,42 +239,36 @@ withTracer handler proc = do

234 Nothing -> return ()

239 Nothing -> return ()

235 Just p -> do

240 Just p -> do

236 ref <- monitor tracer

241 ref <- monitor tracer

237− say "resetting to previous tracer" >> enableTrace p

238− say "stopping tracer"

239 send tracer TraceEvDisable

242 send tracer TraceEvDisable

240− say $ "waiting for EXIT from tracer " ++ (show tracer)

241 receiveWait [

243 receiveWait [

242 matchIf (\(ProcessMonitorNotification ref' _ _) -> ref == ref')

244 matchIf (\(ProcessMonitorNotification ref' _ _) -> ref == ref')

243− (\_ -> say "seen EXIT ok")

245+ (\_ -> return ())

244 ]

246 ]

245

247

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

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

247−-- flags are restored immediately afterwards.

249+-- trace flags are restored immediately afterwards.

248withFlags :: forall a.

250withFlags :: forall a.

249 TraceFlags

251 TraceFlags

250 -> Process a

252 -> Process a

251 -> Process (Either SomeException a)

253 -> Process (Either SomeException a)

252withFlags flags proc = do

254withFlags flags proc = do

253− (sp, rp) <- newChan

255+ oldFlags <- getTraceFlags

254− withLocalTracer $ \t -> liftIO $ Tracer.getTraceFlags t sp

255− oldFlags <- receiveChan rp

256 finally (setTraceFlags flags >> try proc)

257− (setTraceFlags oldFlags >> say "exit withFlags")

257+ (setTraceFlags oldFlags)

258

259traceProxy :: ProcessId -> (TraceEvent -> Process ()) -> Process ()

260traceProxy pid act = do

261− say "tracing has started"

262 proxy pid $ \(ev :: TraceEvent) ->

261 proxy pid $ \(ev :: TraceEvent) ->

263 case ev of

262 case ev of

264− (TraceEvTakeover p) -> say ((show p) ++ " has taken over...") >> return False

263+ (TraceEvTakeover p) -> return False

265− TraceEvDisable -> say "dying - disabled..." >> die "disabled"

264+ TraceEvDisable -> die "disabled"

266− _ -> say ("act on " ++ (show ev)) >> act ev >> return True

265+ _ -> act ev >> return True

267

266

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

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

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

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

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

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

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

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

272−-- function will stop the tracers in the reverse order in which they were

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

273-- started.

272-- started.

274--

273--

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

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

@@ -278,10 +277,8 @@ traceProxy pid act = do

278-- subsequent calls to this function will have no effect.

277-- subsequent calls to this function will have no effect.

279--

278--

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

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

281−-- have no effect. If the last tracer to have been registered (or the process

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

282−-- registered with the label @"tracer"@ at the time the stop instruction

283−-- is received) was not started with 'startTracer' then the behaviour of

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

284−-- this function is /undefined/.

285stopTracer :: Process ()

282stopTracer :: Process ()

286stopTracer =

283stopTracer =

287 withRegisteredTracer $ \pid -> do

284 withRegisteredTracer $ \pid -> do

@@ -293,86 +290,3 @@ stopTracer =

293 True -> return ()

290 True -> return ()

294 False -> send pid TraceEvDisable

291 False -> send pid TraceEvDisable

295

292

296−withRegisteredTracer :: (ProcessId -> Process a) -> Process a

297−withRegisteredTracer act = do

298− (sp, rp) <- newChan

299− withLocalTracer $ \t -> liftIO $ Tracer.getCurrentTraceClient t sp

300− currentTracer <- receiveChan rp

301− case currentTracer of

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

303− (Just p) -> act p

304−

305−-- | Determine whether or not tracing has been enabled.

306−isTracingEnabled :: Process Bool

307−isTracingEnabled = do

308− node <- processNode <$> ask

309− case (localTracer node) of

310− (ActiveTracer _ _) -> return True

311− _ -> return False

312−

313−-- | Enable tracing to the supplied process.

314−enableTraceAsync :: ProcessId -> Process ()

315−enableTraceAsync pid = withLocalTracer $ \t -> liftIO $ Tracer.enableTrace t pid

316−

317−-- TODO: refactor _Sync versions of trace configuration functions...

318−

319−-- | Enable tracing to the supplied process and wait for a @TraceOk@

320−-- response from the trace coordinator process.

321−enableTrace :: ProcessId -> Process ()

322−enableTrace pid =

323− withLocalTracerSync $ \t sp -> Tracer.enableTraceSync t sp pid

324−

325−-- | Disable the currently configured trace.

326−disableTraceAsync :: Process ()

327−disableTraceAsync = withLocalTracer $ \t -> liftIO $ Tracer.disableTrace t

328−

329−-- | Disable the currently configured trace and wait for a @TraceOk@

330−-- response from the trace coordinator process.

331−disableTrace :: Process ()

332−disableTrace =

333− withLocalTracerSync $ \t sp -> Tracer.disableTraceSync t sp

334−

335−getTraceFlags :: Process TraceFlags

336−getTraceFlags = do

337− (sp, rp) <- newChan

338− withLocalTracer $ \t -> liftIO $ Tracer.getTraceFlags t sp

339− receiveChan rp

340−

341−-- | Set the given flags for the current tracer.

342−setTraceFlagsAsync :: TraceFlags -> Process ()

343−setTraceFlagsAsync f = withLocalTracer $ \t -> liftIO $ Tracer.setTraceFlags t f

344−

345−-- | Set the given flags for the current tracer and wait for a @TraceOk@

346−-- response from the trace coordinator process.

347−setTraceFlags :: TraceFlags -> Process ()

348−setTraceFlags f =

349− withLocalTracerSync $ \t sp -> Tracer.setTraceFlagsSync t sp f

350−

351−withLocalTracerSync :: (Tracer -> SendPort TraceOk -> IO ()) -> Process ()

352−withLocalTracerSync act = do

353− (sp, rp) <- newChan

354− withLocalTracer $ \t -> liftIO $ (act t sp)

355− TraceOk <- receiveChan rp

356− return ()

357−

358−-- | Send a log message to the internal tracing facility. If tracing is

359−-- enabled, this will create a custom trace log event.

360−--

361−traceLog :: String -> Process ()

362−traceLog s = withLocalTracer $ \t -> liftIO $ Tracer.traceLog t s

363−

364−-- | Send a log message to the internal tracing facility, using the given

365−-- list of printable 'TraceArg's interspersed with the preceding delimiter.

366−--

367−traceLogFmt :: String -> [TraceArg] -> Process ()

368−traceLogFmt d ls = withLocalTracer $ \t -> liftIO $ Tracer.traceLogFmt t d ls

369−

370−-- | Send an arbitrary 'Message' to the tracer process.

371−traceMessage :: Serializable m => m -> Process ()

372−traceMessage msg = withLocalTracer $ \t -> liftIO $ Tracer.traceMessage t msg

373−

374−withLocalTracer :: (Tracer -> Process ()) -> Process ()

375−withLocalTracer act = do

376− node <- processNode <$> ask

377− act (localTracer node)

378−

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

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

16 , bindCP

17 , seqCP

18 -- * CP versions of Cloud Haskell primitives

19+ , decodeProcessIdStatic

19 , cpLink

20 , cpLink

20 , cpUnlink

21 , cpUnlink

22+ , cpRelay

21 , cpSend

23 , cpSend

22 , cpExpect

24 , cpExpect

23 , cpNewChan

25 , cpNewChan

24 -- * Support for some CH operations

26 -- * Support for some CH operations

25 , cpDelay

27 , cpDelay

28+ , cpEnableTraceRemote

26 ) where

29 ) where

28import Data.ByteString.Lazy (ByteString)

31import Data.ByteString.Lazy (ByteString)

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

56import Control.Distributed.Process.Internal.Primitives

59import Control.Distributed.Process.Internal.Primitives

57 ( link

60 ( link

58 , unlink

61 , unlink

62+ , relay

59 , send

63 , send

60 , expect

64 , expect

61 , newChan

65 , newChan

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

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

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

83 . registerStatic "$link" (toDynamic link)

87 . registerStatic "$link" (toDynamic link)

84 . registerStatic "$unlink" (toDynamic unlink)

88 . registerStatic "$unlink" (toDynamic unlink)

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

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

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

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

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

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

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

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

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

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

247 newChanDictStatic = staticLabel "$newChanDict"

252 newChanDictStatic = staticLabel "$newChanDict"

248

253

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

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

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

257+ where

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

259+ relayStatic = staticLabel "$relay"

260+

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

262+

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

264+cpEnableTraceRemote =

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

266+ where

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

268+ enableTraceStatic = staticLabel "$enableTraceRemote"

269+

249--------------------------------------------------------------------------------

270--------------------------------------------------------------------------------

250-- Support for spawn --

271-- Support for spawn --

251--------------------------------------------------------------------------------

272--------------------------------------------------------------------------------

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

275

296

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

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

277 delayStatic = staticLabel "$delay"

298 delayStatic = staticLabel "$delay"

299+

300+

301+

src/Control/Distributed/Process/Internal/Primitives.hs+20−7

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

1+{-# LANGUAGE BangPatterns #-}

1-- | Cloud Haskell primitives

3-- | Cloud Haskell primitives

2--

4--

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

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

@@ -131,7 +133,19 @@ import Control.Distributed.Process.Internal.CQueue

131 )

133 )

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

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

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

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

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

136+import Control.Distributed.Static

137+ ( RemoteTable

138+ , registerStatic

139+ , Static

140+ , staticLabel

141+ , staticApply

142+ , Closure

143+ , closure

144+ , closureApplyStatic

145+ , closureApply

146+ , staticCompose

147+ , staticClosure

148+ )

135import Data.Rank1Typeable (Typeable)

149import Data.Rank1Typeable (Typeable)

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

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

137import Control.Distributed.Process.Internal.Types

151import Control.Distributed.Process.Internal.Types

@@ -178,7 +192,7 @@ import Control.Distributed.Process.Internal.Messaging

178 , sendPayload

192 , sendPayload

179 , disconnect

193 , disconnect

180 )

194 )

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

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

182 ( traceEvent

196 ( traceEvent

183 , TraceEvent(..)

197 , TraceEvent(..)

184 )

198 )

@@ -516,7 +530,7 @@ delegate pid p = do

516

530

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

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

518relay :: ProcessId -> Process ()

532relay :: ProcessId -> Process ()

519−relay pid = receiveWait [ matchAny (\m -> forward m pid) ] >> relay pid

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

520

534

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

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

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

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

@@ -526,12 +540,11 @@ proxy :: Serializable a => ProcessId -> (a -> Process Bool) -> Process ()

526proxy pid proc = do

540proxy pid proc = do

527 receiveWait [

541 receiveWait [

528 matchAny (\m -> do

542 matchAny (\m -> do

529− say "handling proxied message..."

530 next <- handleMessage m proc

543 next <- handleMessage m proc

531 case next of

544 case next of

532− Just True -> say ("forwarding msg to " ++ (show pid)) >> forward m pid

545+ Just True -> forward m pid

533− Just False -> say "ignoring explicitly ignored msg"

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

534− Nothing -> say "ignoring unroutable msg")

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

535 ]

548 ]

536 proxy pid proc

549 proxy pid proc

537

550

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

@@ -0,0 +1,168 @@

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.Monad.IO.Class (liftIO)

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

13+import Control.Monad.Reader (ask)

14+import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar)

15+import Control.Distributed.Static

16+ ( Static

17+ , Closure

18+ , closureCompose

19+ , staticClosure

20+ )

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

22+ ( NodeId(..)

23+ , ProcessId(..)

24+ , Process(..)

25+ , MonitorRef(..)

26+ , ProcessMonitorNotification(..)

27+ , NodeMonitorNotification(..)

28+ , DidSpawn(..)

29+ , SendPort(..)

30+ , ReceivePort(..)

31+ , LocalProcess(processNode)

32+ , nullProcessId

33+ )

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

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

36+ ( sdictSendPort

37+ , sndStatic

38+ , idCP

39+ , seqCP

40+ , bindCP

41+ , splitCP

42+ , cpLink

43+ , cpSend

44+ , cpNewChan

45+ , cpDelay

46+ )

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

48+ ( -- Basic messaging

49+ send

50+ , expect

51+ , receiveWait

52+ , match

53+ , matchIf

54+ , link

55+ , getSelfPid

56+ , monitor

57+ , monitorNode

58+ , unmonitor

59+ , spawnAsync

60+ , reconnect

61+ , newChan

62+ )

63+

64+-- | Spawn a process

65+--

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

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

68+--

69+-- See also 'call'.

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

71+spawn nid proc = do

72+ us <- getSelfPid

73+ mRef <- monitorNode nid

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

75+ receiveWait [

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

77+ unmonitor mRef

78+ send pid ()

79+ return pid

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

81+ return (nullProcessId nid)

82+ ]

83+

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

85+--

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

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

88+-- a synchronous link operation)

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

90+spawnLink nid proc = do

91+ pid <- spawn nid proc

92+ link pid

93+ return pid

94+

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

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

97+spawnMonitor nid proc = do

98+ pid <- spawn nid proc

99+ ref <- monitor pid

100+ return (pid, ref)

101+

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

103+--

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

105+-- too.

106+--

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

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

109+--

110+-- See also 'spawn'.

111+call :: Serializable a

112+ => Static (SerializableDict a)

113+ -> NodeId

114+ -> Closure (Process a)

115+ -> Process a

116+call dict nid proc = do

117+ us <- getSelfPid

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

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

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

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

122+ mResult <- receiveWait

123+ [ match (return . Right)

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

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

126+ ]

127+ case mResult of

128+ Right a -> do

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

130+ receiveWait

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

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

133+ ]

134+ -- Clean up connection to pid

135+ reconnect pid

136+ return a

137+ Left err ->

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

139+

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

141+-- monitor the child

142+spawnSupervised :: NodeId

143+ -> Closure (Process ())

144+ -> Process (ProcessId, MonitorRef)

145+spawnSupervised nid proc = do

146+ us <- getSelfPid

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

148+ ref <- monitor them

149+ return (them, ref)

150+

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

152+-- the corresponding 'SendPort'.

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

154+ -> NodeId

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

156+ -> Process (SendPort a)

157+spawnChannel dict nid proc = do

158+ us <- getSelfPid

159+ _ <- spawn nid (go us)

160+ expect

161+ where

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

163+ go pid = cpNewChan dict

164+ `bindCP`

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

166+ `bindCP`

167+ (idCP `closureCompose` staticClosure sndStatic)

168+

src/Control/Distributed/Process/Internal/Trace/Primitives.hs+144−68

@@ -1,30 +1,52 @@

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

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

4module Control.Distributed.Process.Internal.Trace.Primitives

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

5 ( Tracer

7 ( Tracer

6 , TraceEvent(..)

8 , TraceEvent(..)

7 -- * Sending Trace Data

9 -- * Sending Trace Data

8 , traceLog

10 , traceLog

9 , traceLogFmt

11 , traceLogFmt

10− , traceEvent

11 , traceMessage

12 , traceMessage

12 -- * Configuring A Tracer

13 -- * Configuring A Tracer

14+ , isTracingEnabled

13 , defaultTraceFlags

15 , defaultTraceFlags

14 , enableTrace

16 , enableTrace

15− , enableTraceSync

17+ , enableTraceAsync

16 , disableTrace

18 , disableTrace

17− , disableTraceSync

19+ , disableTraceAsync

18 , getTraceFlags

20 , getTraceFlags

19 , setTraceFlags

21 , setTraceFlags

20− , setTraceFlagsSync

22+ , setTraceFlagsAsync

21− , getCurrentTraceClient

22 , traceOnly

23 , traceOnly

23 , traceOn

24 , traceOn

24 , traceOff

25 , traceOff

26+ , withLocalTracer

27+ , withRegisteredTracer

25 ) where

28 ) where

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

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

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

32+ ( cpEnableTraceRemote

33+ )

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

35+ ( proxy

36+ , finally

37+ , die

38+ , whereis

39+ , send

40+ , newChan

41+ , receiveChan

42+ , receiveWait

43+ , matchIf

44+ , finally

45+ , try

46+ , monitor

47+ , getSelfPid

48+ , relay

49+ )

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

50import Control.Distributed.Process.Internal.Trace.Types

29 ( TraceArg(..)

51 ( TraceArg(..)

30 , TraceEvent(..)

52 , TraceEvent(..)

@@ -34,76 +56,50 @@ import Control.Distributed.Process.Internal.Trace.Types

34 , TraceSubject(..)

56 , TraceSubject(..)

35 , defaultTraceFlags

57 , defaultTraceFlags

36 )

58 )

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

60+ ( traceLog

61+ , traceLogFmt

62+ , traceMessage

63+ , enableTrace

64+ , enableTraceSync

65+ , disableTrace

66+ , disableTraceSync

67+ , setTraceFlags

68+ , setTraceFlagsSync

69+ , getTraceFlags

70+ , getCurrentTraceClient

71+ )

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

73+ ( spawn

74+ )

37import Control.Distributed.Process.Internal.Types

75import Control.Distributed.Process.Internal.Types

38 ( Tracer(..)

76 ( Tracer(..)

77+ , Process

39 , ProcessId

78 , ProcessId

79+ , LocalProcess(..)

80+ , LocalNode(localTracer)

40 , Message

81 , Message

41 , SendPort

82 , SendPort

83+ , NodeId

42 , createUnencodedMessage

84 , createUnencodedMessage

43 )

85 )

44import Control.Distributed.Process.Serializable

86import Control.Distributed.Process.Serializable

45−import Data.Foldable (forM_)

87+import Control.Distributed.Static

46−import Data.List (intersperse)

88+ ( RemoteTable

89+ , registerStatic

90+ )

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

92+import Control.Monad.Reader (ask)

93+

94+import Data.Rank1Dynamic (toDynamic)

47import qualified Data.Set as Set (fromList)

95import qualified Data.Set as Set (fromList)

48−import System.Mem.Weak (deRefWeak)

50--------------------------------------------------------------------------------

97--------------------------------------------------------------------------------

51-- Main API --

98-- Main API --

52--------------------------------------------------------------------------------

99--------------------------------------------------------------------------------

100

54−traceLog :: Tracer -> String -> IO ()

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

55−traceLog tr s = traceMessage tr (TraceEvLog s)

102+-- to a 'TraceSubject'.

56−

57−traceLogFmt :: Tracer

58− -> String

59− -> [TraceArg]

60− -> IO ()

61−traceLogFmt t d ls =

62− traceLog t $ concat (intersperse d (map toS ls))

63− where toS :: TraceArg -> String

64− toS (TraceStr s) = s

65− toS (Trace a) = show a

66−

67−traceEvent :: Tracer -> TraceEvent -> IO ()

68−traceEvent tr ev = traceIt tr (createUnencodedMessage ev)

69−

70−traceMessage :: Serializable m => Tracer -> m -> IO ()

71−traceMessage tr msg = traceEvent tr (TraceEvUser (createUnencodedMessage msg))

72−

73−enableTrace :: Tracer -> ProcessId -> IO ()

74−enableTrace t p =

75− traceIt t (createUnencodedMessage ((Nothing :: Maybe (SendPort TraceOk)),

76− (TraceEnable p)))

77−

78−enableTraceSync :: Tracer -> SendPort TraceOk -> ProcessId -> IO ()

79−enableTraceSync t s p =

80− traceIt t (createUnencodedMessage ((Just s), (TraceEnable p)))

81−

82−disableTrace :: Tracer -> IO ()

83−disableTrace t =

84− traceIt t (createUnencodedMessage ((Nothing :: Maybe (SendPort TraceOk)),

85− TraceDisable))

86−

87−disableTraceSync :: Tracer -> SendPort TraceOk -> IO ()

88−disableTraceSync t s =

89− traceIt t (createUnencodedMessage ((Just s), TraceDisable))

90−

91−setTraceFlags :: Tracer -> TraceFlags -> IO ()

92−setTraceFlags t f =

93− traceIt t (createUnencodedMessage ((Nothing :: Maybe (SendPort TraceOk)), f))

94−

95−setTraceFlagsSync :: Tracer -> SendPort TraceOk -> TraceFlags -> IO ()

96−setTraceFlagsSync t s f =

97− traceIt t (createUnencodedMessage ((Just s), f))

98−

99−getTraceFlags :: Tracer -> SendPort TraceFlags -> IO ()

100−getTraceFlags t s = traceIt t (createUnencodedMessage s)

101−

102−getCurrentTraceClient :: Tracer -> SendPort (Maybe ProcessId) -> IO ()

103−getCurrentTraceClient t s = traceIt t (createUnencodedMessage s)

104−

105−-- aux API and utilities

106−

107class Traceable a where

103class Traceable a where

108 uod :: [a] -> TraceSubject

104 uod :: [a] -> TraceSubject

109

105

@@ -113,18 +109,98 @@ instance Traceable ProcessId where

113instance Traceable String where

109instance Traceable String where

114 uod = TraceNames . Set.fromList

110 uod = TraceNames . Set.fromList

115

111

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

116traceOnly :: Traceable a => [a] -> Maybe TraceSubject

113traceOnly :: Traceable a => [a] -> Maybe TraceSubject

117traceOnly = Just . uod

114traceOnly = Just . uod

118

115

116+-- | Trace all targets.

119traceOn :: Maybe TraceSubject

117traceOn :: Maybe TraceSubject

120traceOn = Just TraceAll

118traceOn = Just TraceAll

121

119

120+-- | Trace no targets.

122traceOff :: Maybe TraceSubject

121traceOff :: Maybe TraceSubject

123traceOff = Nothing

122traceOff = Nothing

124

123

125−traceIt :: Tracer -> Message -> IO ()

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

126−traceIt InactiveTracer _ = return ()

125+isTracingEnabled :: Process Bool

127−traceIt (ActiveTracer _ wqRef) msg = do

126+isTracingEnabled = do

128− mQueue <- deRefWeak wqRef

127+ node <- processNode <$> ask

129− forM_ mQueue $ \queue -> enqueue queue msg

128+ case (localTracer node) of

129+ (ActiveTracer _ _) -> return True

130+ _ -> return False

131+

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

133+enableTraceAsync :: ProcessId -> Process ()

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

135+

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

137+

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

139+-- response from the trace coordinator process.

140+enableTrace :: ProcessId -> Process ()

141+enableTrace pid =

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

143+

144+-- | Disable the currently configured trace.

145+disableTraceAsync :: Process ()

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

147+

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

149+-- response from the trace coordinator process.

150+disableTrace :: Process ()

151+disableTrace =

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

153+

154+getTraceFlags :: Process TraceFlags

155+getTraceFlags = do

156+ (sp, rp) <- newChan

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

158+ receiveChan rp

159+

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

161+setTraceFlagsAsync :: TraceFlags -> Process ()

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

163+

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

165+-- response from the trace coordinator process.

166+setTraceFlags :: TraceFlags -> Process ()

167+setTraceFlags f =

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

169+

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

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

172+--

173+traceLog :: String -> Process ()

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

175+

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

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

178+--

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

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

181+

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

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

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

185+

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

187+withLocalTracer act = do

188+ node <- processNode <$> ask

189+ act (localTracer node)

190+

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

192+withLocalTracerSync act = do

193+ (sp, rp) <- newChan

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

195+ TraceOk <- receiveChan rp

196+ return ()

197+

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

199+withRegisteredTracer act = do

200+ (sp, rp) <- newChan

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

202+ currentTracer <- receiveChan rp

203+ case currentTracer of

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

205+ (Just p) -> act p

130

206

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

@@ -0,0 +1,80 @@

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.Applicative ((<$>))

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

11+ ( cpEnableTraceRemote

12+ )

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

14+ ( proxy

15+ , finally

16+ , die

17+ , whereis

18+ , send

19+ , newChan

20+ , receiveChan

21+ , receiveWait

22+ , matchIf

23+ , finally

24+ , try

25+ , monitor

26+ , getSelfPid

27+ , relay

28+ , nsendRemote

29+ , say

30+ )

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

32+ ( TraceFlags(..)

33+ , TraceOk(..)

34+ , TraceSubject(..)

35+ , defaultTraceFlags

36+ )

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

38+ ( withRegisteredTracer

39+ , enableTrace

40+ )

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

42+ ( spawn

43+ )

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

45+ ( Process

46+ , ProcessId

47+ , SendPort

48+ , NodeId

49+ )

50+import Control.Distributed.Process.Serializable

51+import Control.Distributed.Static

52+ ( RemoteTable

53+ , registerStatic

54+ )

55+import Data.Rank1Dynamic (toDynamic)

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

57+

58+-- | Remote Table.

59+remoteTable :: RemoteTable -> RemoteTable

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

61+

62+enableTraceRemote :: ProcessId -> Process ()

63+enableTraceRemote pid =

64+ getSelfPid >>= enableTrace >> relay pid

65+

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

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

68+startTraceRelay :: NodeId -> Process ProcessId

69+startTraceRelay nodeId = do

70+ withRegisteredTracer $ \pid ->

71+ -- TODO: move cpEnableTraceRemote here?

72+ spawn nodeId $ cpEnableTraceRemote pid

73+

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

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

76+setTraceFlagsRemote flags node = do

77+ nsendRemote node

78+ "trace.controller"

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

80+

src/Control/Distributed/Process/Internal/Trace/Tracer.hs+6−5

@@ -20,6 +20,7 @@ import Control.Distributed.Process.Internal.CQueue

20 )

21import Control.Distributed.Process.Internal.Primitives

22 ( catch

23+ , die

23 , receiveWait

24 , receiveWait

24 , forward

25 , forward

25 , sendChan

26 , sendChan

@@ -29,7 +30,9 @@ import Control.Distributed.Process.Internal.Primitives

29 , handleMessage

30 , handleMessage

30 , matchUnknown

31 , matchUnknown

31 , whereis

32 , whereis

33+ , getSelfPid

34+ , register

32− , reregister, say, unwrapMessage

35+ , say, unwrapMessage

33 )

36 )

34import Control.Distributed.Process.Internal.Trace.Types

37import Control.Distributed.Process.Internal.Trace.Types

35 ( TraceEvent(..)

38 ( TraceEvent(..)

@@ -156,6 +159,7 @@ eventLogTracer = do

156logfileTracer :: FilePath -> Process ()

159logfileTracer :: FilePath -> Process ()

157logfileTracer p = do

160logfileTracer p = do

158 -- TODO: error handling if the handle cannot be opened

161 -- TODO: error handling if the handle cannot be opened

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

159 h <- liftIO $ openFile p AppendMode

163 h <- liftIO $ openFile p AppendMode

160 liftIO $ hSetBuffering h LineBuffering

164 liftIO $ hSetBuffering h LineBuffering

161 logger h `catch` (\(_ :: SomeException) -> liftIO $ hClose h)

165 logger h `catch` (\(_ :: SomeException) -> liftIO $ hClose h)

@@ -167,7 +171,7 @@ logfileTracer p = do

167 TraceEvDisable -> True

171 TraceEvDisable -> True

168 (TraceEvTakeover _) -> True

172 (TraceEvTakeover _) -> True

169 _ -> False)

173 _ -> False)

170− (\_ -> (liftIO $ hClose h') >> error "trace stopped")

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

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

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

172 ]

176 ]

173

177

@@ -207,13 +211,11 @@ traceController mv = do

207 (TraceEnable pid) -> do

211 (TraceEnable pid) -> do

208 -- notify the previous tracer it has been replaced

212 -- notify the previous tracer it has been replaced

209 sendTrace st (createUnencodedMessage (TraceEvTakeover pid))

213 sendTrace st (createUnencodedMessage (TraceEvTakeover pid))

210− say $ "registering " ++ (show pid) ++ " as new tracer"

211 sendOk setResp

214 sendOk setResp

212 return st { client = (Just pid) }

215 return st { client = (Just pid) }

213 TraceDisable -> do

216 TraceDisable -> do

214 sendTrace st (createUnencodedMessage TraceEvDisable)

217 sendTrace st (createUnencodedMessage TraceEvDisable)

215 sendOk setResp

218 sendOk setResp

216− say "setting sendTrace to noop"

217 return st { client = Nothing })

219 return st { client = Nothing })

218 , match (\(confResp, flags') ->

220 , match (\(confResp, flags') ->

219 sendOk confResp >> applyTraceFlags flags' st)

221 sendOk confResp >> applyTraceFlags flags' st)

@@ -327,7 +329,6 @@ sendTrace :: TracerState -> Message -> Process ()

327sendTrace st msg =

329sendTrace st msg =

328 let pid = (client st) in do

330 let pid = (client st) in do

329 (Just ev) <- unwrapMessage msg :: Process (Maybe TraceEvent)

331 (Just ev) <- unwrapMessage msg :: Process (Maybe TraceEvent)

330− say ("sending " ++ (show ev) ++ " to " ++ (show pid))

331 case pid of

332 case pid of

332 Just p -> (flip forward) p msg

333 Just p -> (flip forward) p msg

333 Nothing -> return ()

334 Nothing -> return ()

src/Control/Distributed/Process/Internal/Trace/Types.hs+100

@@ -9,19 +9,40 @@ module Control.Distributed.Process.Internal.Trace.Types

9 , defaultTraceFlags

10 , TraceArg(..)

11 , TraceOk(..)

12+ , traceLog

13+ , traceLogFmt

14+ , traceEvent

15+ , traceMessage

16+ , defaultTraceFlags

17+ , enableTrace

18+ , enableTraceSync

19+ , disableTrace

20+ , disableTraceSync

21+ , getTraceFlags

22+ , setTraceFlags

23+ , setTraceFlagsSync

24+ , getCurrentTraceClient

12 ) where

25 ) where

14import Control.Applicative ((<$>), (<*>))

27import Control.Applicative ((<$>), (<*>))

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

15import Control.Distributed.Process.Internal.Types

29import Control.Distributed.Process.Internal.Types

16 ( Tracer(..)

30 ( Tracer(..)

17 , NodeId

31 , NodeId

18 , ProcessId

32 , ProcessId

19 , DiedReason

33 , DiedReason

20 , Message

34 , Message

35+ , SendPort

36+ , createUnencodedMessage

21 )

37 )

38+import Control.Distributed.Process.Serializable

22import Data.Binary

39import Data.Binary

40+import Data.Foldable (forM_)

41+import Data.List (intersperse)

23import Data.Set (Set)

42import Data.Set (Set)

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

24import Data.Typeable

44import Data.Typeable

45+import System.Mem.Weak (deRefWeak)

25import Network.Transport

46import Network.Transport

26 ( ConnectionId

47 ( ConnectionId

27 )

48 )

@@ -120,6 +141,85 @@ data TraceArg =

120data TraceOk = TraceOk

141data TraceOk = TraceOk

121 deriving (Typeable)

142 deriving (Typeable)

122

143

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

145+-- Internal/Common API --

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

147+

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

149+traceLog tr s = traceMessage tr (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+traceOnly :: Traceable a => [a] -> Maybe TraceSubject

209+traceOnly = Just . uod

210+

211+traceOn :: Maybe TraceSubject

212+traceOn = Just TraceAll

213+

214+traceOff :: Maybe TraceSubject

215+traceOff = Nothing

216+

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

218+traceIt InactiveTracer _ = return ()

219+traceIt (ActiveTracer _ wqRef) msg = do

220+ mQueue <- deRefWeak wqRef

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

222+

123--------------------------------------------------------------------------------

223--------------------------------------------------------------------------------

124-- Binary Instances --

224-- Binary Instances --

125--------------------------------------------------------------------------------

225--------------------------------------------------------------------------------

src/Control/Distributed/Process/Node.hs+14−4

@@ -149,13 +149,14 @@ import Control.Distributed.Process.Internal.Types

149 , firstNonReservedProcessId

150 , ImplicitReconnect(WithImplicitReconnect)

151 )

152−import Control.Distributed.Process.Internal.Trace.Primitives

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

153− ( traceLogFmt

153+ ( remoteTable

154− , traceEvent

155 )

154 )

156import Control.Distributed.Process.Internal.Trace.Types

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

157 ( TraceArg(..)

156 ( TraceArg(..)

158 , TraceEvent(..)

157 , TraceEvent(..)

158+ , traceEvent

159+ , traceLogFmt

159 )

160 )

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

161import Control.Distributed.Process.Internal.Trace.Tracer

161 ( traceController

162 ( traceController

@@ -190,7 +191,7 @@ import Unsafe.Coerce

190--------------------------------------------------------------------------------

191--------------------------------------------------------------------------------

191

192

192initRemoteTable :: RemoteTable

193initRemoteTable :: RemoteTable

193−initRemoteTable = BuiltIn.remoteTable Static.initRemoteTable

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

194

195

195-- | Initialize a new local node.

196-- | Initialize a new local node.

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

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

@@ -241,6 +242,7 @@ createBareLocalNode endPoint rtable = do

241

242

242startTracing :: LocalNode -> IO LocalNode

243startTracing :: LocalNode -> IO LocalNode

243startTracing node =

244startTracing node =

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

244 Exception.catch

246 Exception.catch

245 (getEnv "DISTRIBUTED_PROCESS_TRACE_ENABLED" >> procTracer node)

247 (getEnv "DISTRIBUTED_PROCESS_TRACE_ENABLED" >> procTracer node)

246 (\(_ :: IOError) -> return node)

248 (\(_ :: IOError) -> return node)

@@ -259,9 +261,17 @@ startDefaultTracer node =

259 (\(_ :: IOError) -> return Nothing)

261 (\(_ :: IOError) -> return Nothing)

260 where go :: LocalNode -> IO (Maybe ProcessId)

262 where go :: LocalNode -> IO (Maybe ProcessId)

261 go node' = do

263 go node' = do

264+ registerTraceController node'

262 pid <- forkProcess node' defaultTracer

265 pid <- forkProcess node' defaultTracer

263 return (Just pid)

266 return (Just pid)

264

267

268+ registerTraceController :: LocalNode -> IO ()

269+ registerTraceController n =

270+ let t = localTracer n in

271+ case t of

272+ InactiveTracer -> return ()

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

274+

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

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

266startServiceProcesses :: LocalNode -> IO ()

276startServiceProcesses :: LocalNode -> IO ()

267startServiceProcesses node = do

277startServiceProcesses node = do

tests/TestTracing.hs+63

@@ -2,6 +2,7 @@

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

3module Main where

5+import Control.Concurrent (threadDelay)

5import Control.Concurrent.MVar

6import Control.Concurrent.MVar

6 ( MVar

7 ( MVar

7 , newEmptyMVar

8 , newEmptyMVar

@@ -259,6 +260,64 @@ testTraceLayering result = do

259 })

260 })

260 liftIO $ putMVar result' ()

261 liftIO $ putMVar result' ()

261

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+

262tests :: LocalNode -> IO [Test]

321tests :: LocalNode -> IO [Test]

263tests node1 = do

322tests node1 = do

264 -- if we execute the test cases in parallel, the

323 -- if we execute the test cases in parallel, the

@@ -298,6 +357,10 @@ tests node1 = do

298 (delayedAssertion

357 (delayedAssertion

299 "expected blah"

358 "expected blah"

300 node1 () testTraceLayering lock)

359 node1 () testTraceLayering lock)

360+ , testCase "Remote Trace Relay"

361+ (delayedAssertion

362+ "expected blah"

363+ node1 True testRemoteTraceRelay lock)

301 ] ]

364 ] ]

302 False ->

365 False ->

303 return []

366 return []