b3fa53c · distributed-process

git.iko.soy/distributed-process

Begin to tease out an API for scoped tracing

commitb3fa53c87142296710e9640458568d95ef5dfcec authorTim Watson <tim@rabbitmq.com> dateFri 22 Mar 2013, 21:27 (14 years ago) parent1520aa5

5 files changed+214−31

compare with master

unifiedsplit

src/Control/Distributed/Process/Debug.hs+49−19 src/Control/Distributed/Process/Internal/Primitives.hs+4−2 src/Control/Distributed/Process/Internal/Trace/Primitives.hs+29 src/Control/Distributed/Process/Internal/Trace/Tracer.hs+10−6 tests/TestTracing.hs+122−4

src/Control/Distributed/Process/Debug.hs+49−19

@@ -94,9 +94,14 @@ module Control.Distributed.Process.Debug

9494 , disableTrace

9595 , disableTraceAsync

9696 , withTracer

97+ , withFlags

98+ , getTraceFlags

9799 , setTraceFlags

98100 , setTraceFlagsAsync

99101 , defaultTraceFlags

102+ , Tracer.traceOn

103+ , Tracer.traceOnly

104+ , Tracer.traceOff

100105 -- * Debugging

101106 , startTracer

102107 , stopTracer

@@ -128,6 +133,7 @@ import Control.Distributed.Process.Internal.Primitives

128133 , finally

129134 , try

130135 , monitor

136+ , say

131137 )

132138import Control.Distributed.Process.Internal.Types

133139 ( Tracer(..)

@@ -161,6 +167,10 @@ import qualified Control.Distributed.Process.Internal.Trace.Primitives as Tracer

161167 , disableTraceSync

162168 , setTraceFlags

163169 , setTraceFlagsSync

170+ , getTraceFlags

171+ , traceOn

172+ , traceOff

173+ , traceOnly

164174 )

165175import Control.Distributed.Process.Node

166176import Control.Distributed.Process.Serializable

@@ -189,7 +199,6 @@ import Prelude hiding (catch)

189199-- reregister the original (prior) tracer pid before terminating.

190200startTracer :: (TraceEvent -> Process ()) -> Process ProcessId

191201startTracer handler = do

192− (sp, rp) <- newChan

193202 withRegisteredTracer $ \pid -> do

194203 node <- processNode <$> ask

195204 newPid <- liftIO $ forkProcess node $ do

@@ -200,9 +209,8 @@ startTracer handler = do

200209 -- close succession, due to scheduling

201210 finally (traceProxy pid handler)

202211 (resetTracer pid)

203− enableTrace newPid

204− sendChan sp newPid

205− receiveChan rp

212+ enableTrace newPid -- invokes sync + registration

213+ return newPid

206214

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

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

@@ -212,22 +220,38 @@ withTracer :: forall a.

212220 -> Process a

213221 -> Process (Either SomeException a)

214222withTracer handler proc = do

223+ say "starting new tracer"

215224 tracer <- startTracer handler

216− finally (try proc)

225+ finally (try ((say "evaluating tracer proc...") >> proc))

217226 (stopTracing tracer)

218227 where

219228 stopTracing :: ProcessId -> Process ()

220229 stopTracing tracer = do

230+ say "stopping tracer"

221231 ref <- monitor tracer

222232 stopTracer

223233 receiveWait [

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

225235 (\_ -> return ())

226236 ]

237+ say "stopped..."

238+

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

240+-- flags are restored immediately afterwards.

241+withFlags :: forall a.

242+ TraceFlags

243+ -> Process a

244+ -> Process (Either SomeException a)

245+withFlags flags proc = do

246+ (sp, rp) <- newChan

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

248+ oldFlags <- receiveChan rp

249+ finally (setTraceFlags flags >> try proc)

250+ (setTraceFlags oldFlags >> say "exit withFlags")

227251

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

229253traceProxy pid act = do

230− getSelfPid >>= reregister "tracer" -- registration is synchronous

254+ say "tracing has started"

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

232256 case ev of

233257 (TraceEvTakeover _) -> die "takeover"

@@ -235,7 +259,9 @@ traceProxy pid act = do

235259 _ -> act ev >> return True

236260

237261resetTracer :: ProcessId -> Process ()

238−resetTracer pid = enableTrace pid

262+resetTracer pid = do

263+ say $ "resetting trace to " ++ (show pid)

264+ enableTrace pid

239265

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

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

@@ -252,19 +278,17 @@ resetTracer pid = enableTrace pid

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

253279-- have no effect.

254280stopTracer :: Process ()

255−stopTracer = withLocalTracer $ \t ->

256− case t of

257− InactiveTracer -> return ()

258− (ActiveTracer _ _) -> withRegisteredTracer $ \pid -> do

259− -- we need to avoid killing the initial (base) tracer, as

260− -- nothing we rely on having exactly 1 registered tracer

261− -- process at all times!

262− basePid <- whereis "tracer.initial"

263− case basePid == (Just pid) of

264− True -> return ()

265− False -> send pid TraceEvDisable

281+stopTracer =

282+ withRegisteredTracer $ \pid -> do

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

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

285+ -- process at all times!

286+ basePid <- whereis "tracer.initial"

287+ case basePid == (Just pid) of

288+ True -> return ()

289+ False -> send pid TraceEvDisable

266290

267−withRegisteredTracer :: (ProcessId -> Process ()) -> Process ()

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

268292withRegisteredTracer act = do

269293 currentTracer <- whereis "tracer"

270294 case currentTracer of

@@ -301,6 +325,12 @@ disableTrace :: Process ()

301325disableTrace =

302326 withLocalTracerSync $ \t sp -> Tracer.disableTraceSync t sp

303327

328+getTraceFlags :: Process TraceFlags

329+getTraceFlags = do

330+ (sp, rp) <- newChan

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

332+ receiveChan rp

333+

304334-- | Set the given flags for the current tracer.

305335setTraceFlagsAsync :: TraceFlags -> Process ()

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

src/Control/Distributed/Process/Internal/Primitives.hs+4−2

@@ -525,10 +525,12 @@ proxy :: Serializable a => ProcessId -> (a -> Process Bool) -> Process ()

525525proxy pid proc = do

526526 receiveWait [

527527 matchAny (\m -> do

528+ say "handling proxied message..."

528529 next <- handleMessage m proc

529530 case next of

530− Just True -> forward m pid

531− _ -> return ())

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

532+ Just False -> say "ignoring explicitly ignored msg"

533+ Nothing -> say "ignoring unroutable msg")

532534 ]

533535 proxy pid proc

534536

src/Control/Distributed/Process/Internal/Trace/Primitives.hs+29

@@ -15,8 +15,12 @@ module Control.Distributed.Process.Internal.Trace.Primitives

1515 , enableTraceSync

1616 , disableTrace

1717 , disableTraceSync

18+ , getTraceFlags

1819 , setTraceFlags

1920 , setTraceFlagsSync

21+ , traceOnly

22+ , traceOn

23+ , traceOff

2024 ) where

2125

2226import Control.Distributed.Process.Internal.CQueue (enqueue)

@@ -26,6 +30,7 @@ import Control.Distributed.Process.Internal.Trace.Types

2630 , SetTrace(..)

2731 , TraceFlags(..)

2832 , TraceOk(..)

33+ , TraceSubject(..)

2934 , defaultTraceFlags

3035 )

3136import Control.Distributed.Process.Internal.Types

@@ -38,6 +43,7 @@ import Control.Distributed.Process.Internal.Types

3843import Control.Distributed.Process.Serializable

3944import Data.Foldable (forM_)

4045import Data.List (intersperse)

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

4147import System.Mem.Weak (deRefWeak)

4248

4349--------------------------------------------------------------------------------

@@ -89,6 +95,29 @@ setTraceFlagsSync :: Tracer -> SendPort TraceOk -> TraceFlags -> IO ()

8995setTraceFlagsSync t s f =

9096 traceIt t (createUnencodedMessage ((Just s), f))

9197

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

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

100+

101+-- aux API and utilities

102+

103+class Traceable a where

104+ uod :: [a] -> TraceSubject

105+

106+instance Traceable ProcessId where

107+ uod = TraceProcs . Set.fromList

108+

109+instance Traceable String where

110+ uod = TraceNames . Set.fromList

111+

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

113+traceOnly = Just . uod

114+

115+traceOn :: Maybe TraceSubject

116+traceOn = Just TraceAll

117+

118+traceOff :: Maybe TraceSubject

119+traceOff = Nothing

120+

92121traceIt :: Tracer -> Message -> IO ()

93122traceIt InactiveTracer _ = return ()

94123traceIt (ActiveTracer _ wqRef) msg = do

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

@@ -27,7 +27,7 @@ import Control.Distributed.Process.Internal.Primitives

2727 , matchAny

2828 , matchAnyIf

2929 , handleMessage

30− , reregister

30+ , reregister, say

3131 )

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

3333 ( TraceEvent(..)

@@ -190,12 +190,14 @@ traceController mv = do

190190 return st { sendTrace = (mkSender pid) }

191191 TraceDisable -> do

192192 sendTrace st (createUnencodedMessage TraceEvDisable)

193− sendOk setResp >> return st { sendTrace = (\_ -> return ()) })

193+ sendOk setResp

194+ return st { sendTrace = (\_ -> return ()) })

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

195− sendOk confResp >> applyTraceFlags flags' st) -- setTraceFlags

196+ sendOk confResp >> applyTraceFlags flags' st)

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

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

197199 , matchAny (\ev ->

198− handleMessage ev (handleTrace st ev) >>= return . fromMaybe st)

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

199201 ]

200202 traceLoop st'

201203

@@ -204,7 +206,9 @@ traceController mv = do

204206 sendOk (Just sp) = sendChan sp TraceOk

205207

206208 mkSender :: ProcessId -> (Message -> Process ())

207− mkSender pid = (flip forward) pid

209+ mkSender pid = \m -> do

210+ say ("sending " ++ (show m) ++ " to " ++ (show pid))

211+ (flip forward) pid m

208212

209213 initialState :: TracerState

210214 initialState =

@@ -260,7 +264,7 @@ handleTrace st msg ev = do

260264 case (traceNodes (flags st)) of

261265 True -> (sendTrace st) msg

262266 False -> return ()

263− (TraceEvUser _) ->

267+ (TraceEvUser _) -> do

264268 (sendTrace st) msg

265269 _ ->

266270 case (traceConnections (flags st)) of

tests/TestTracing.hs+122−4

@@ -82,10 +82,115 @@ testSpawnTracing result = do

8282 (tracedDead, tracedReason) <- liftIO $ takeMVar evDied

8383

8484 stopTracer

85+ setTraceFlags defaultTraceFlags

8586 stash result (tracedAlive == pid &&

8687 tracedDead == pid &&

8788 tracedReason == DiedNormal)

8889

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

91+testTraceRecvExplicitPid result = do

92+ say "testTraceRecvExplicitPid"

93+ pid <- spawnLocal $ do

94+ self <- getSelfPid

95+ expect >>= (flip sendChan) self

96+ withFlags defaultTraceFlags {

97+ traceRecv = traceOnly [pid]

98+ } $ do

99+ withTracer

100+ (\ev ->

101+ case ev of

102+ (TraceEvReceived pid' _) -> stash result (pid == pid')

103+ _ -> return ()) $ do

104+ (sp, rp) <- newChan

105+ send pid sp

106+ p <- receiveChan rp

107+ res <- liftIO $ takeMVar result

108+ stash result (res && (p == pid))

109+ say "Explicit done!!!"

110+ return ()

111+

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

113+testTraceRecvNamedPid result = do

114+ say "testTraceRecvNamedPid"

115+ pid <- spawnLocal $ do

116+ self <- getSelfPid

117+ register "foobar" self

118+ expect >>= (flip sendChan) self

119+ withFlags defaultTraceFlags {

120+ traceRecv = traceOnly ["foobar"]

121+ } $ do

122+ withTracer

123+ (\ev ->

124+ case ev of

125+ (TraceEvReceived pid' _) -> stash result (pid == pid')

126+ _ -> return ()) $ do

127+ (sp, rp) <- newChan

128+ send pid sp

129+ p <- receiveChan rp

130+ res <- liftIO $ takeMVar result

131+ stash result (res && (p == pid))

132+ return ()

133+

134+data TraceWhat = User | Die | Rec

135+ deriving (Show)

136+

137+testTraceLayering :: TestResult () -> Process ()

138+testTraceLayering result = do

139+ pid <- spawnLocal $ do

140+ getSelfPid >>= register "foobar"

141+ say "registered - waiting for 'hello'"

142+ "hello" <- expect

143+ say "received 'hello' - waiting for () go"

144+ p <- expect

145+ say "pid received - sending trace msg"

146+ traceMessage ("traceMsg", 123 :: Int)

147+ say "sending reply"

148+ send p ()

149+ () <- expect

150+ return ()

151+ withFlags defaultTraceFlags {

152+ traceDied = traceOnly ["foobar"]

153+ , traceRecv = traceOnly [pid]

154+ } $ doTest pid result

155+ return ()

156+ where

157+ doTest :: ProcessId -> MVar () -> Process ()

158+ doTest pid result' = do

159+ go Die $ do

160+ go Rec $ do

161+ say "sending 'hello'"

162+ send pid "hello"

163+ go User $ do

164+ getSelfPid >>= send pid

165+ () <- expect

166+ say "received reply, returning to rec block"

167+ return ()

168+ liftIO $ putMVar result' ()

169+

170+ go :: TraceWhat -> Process () -> Process ()

171+ go what proc = do {

172+ say ("tracing " ++ (show what))

173+ ; mv <- liftIO $ newEmptyMVar

174+ ; r <- (withTracer

175+ (\ev ->

176+ case (matchTest what ev) of

177+ True -> do

178+ say ("matched " ++ (show ev))

179+ liftIO $ putMVar mv ()

180+ False -> say ("ignoring " ++ (show ev))) $ do

181+ say ("in block for " ++ (show what))

182+ proc)

183+ ; case r of

184+ Left e -> say "whoops"

185+ Right _ -> say "ok!" >> (liftIO $ takeMVar mv)

186+ }

187+

188+ matchTest :: TraceWhat -> TraceEvent -> Bool

189+ matchTest User (TraceEvUser _) = True

190+ matchTest Die (TraceEvDied _ _) = True

191+ matchTest Rec (TraceEvReceived _ _) = True

192+ matchTest _ _ = False

193+

89194tests :: LocalNode -> IO [Test]

90195tests node1 = do

91196 enabled <- checkTraceEnabled

@@ -94,10 +199,22 @@ tests node1 = do

94199 return [

95200 testGroup "Process Info" [

96201 testCase "Test Spawn Tracing"

97− (delayedAssertion

98− "expected dead process-info to be ProcessInfoNone"

99− node1 True testSpawnTracing)

100− ] ]

202+ (delayedAssertion

203+ "expected dead process-info to be ProcessInfoNone"

204+ node1 True testSpawnTracing)

205+ , testCase "Test Recv Tracing (Explicit Pid)"

206+ (delayedAssertion

207+ "expected a recv trace for the supplied pid"

208+ node1 True testTraceRecvExplicitPid)

209+ , testCase "Test Recv Tracing (Named Pid)"

210+ (delayedAssertion

211+ "expected a recv trace for the process registered as 'foobar'"

212+ node1 True testTraceRecvNamedPid)

213+ , testCase "Test Trace Layering"

214+ (delayedAssertion

215+ "expected blah"

216+ node1 () testTraceLayering)

217+ ] ]

101218 False ->

102219 return []

103220

@@ -121,3 +238,4 @@ main = do

121238 defaultMain testData

122239 closeLocalNode node1

123240 return ()

241+