f2f5fa5 · distributed-process

git.iko.soy/distributed-process

on master help sign in

TraceEvent => MxEvent

f2f5fa5Tim Watson13 years ago

src/Control/Distributed/Process/Debug.hs+13−13

compare with master

unifiedsplit

@@ -20,10 +20,10 @@

20-- Throughout the lifecycle of a local node, the distributed-process runtime

21-- generates /trace events/, describing internal runtime activities such as

22-- the spawning and death of processes, message sending, delivery and so on.

23−-- See the 'TraceEvent' type's documentation for a list of all the published

23+-- See the 'MxEvent' type's documentation for a list of all the published

24−-- event types.

24+-- event types, which correspond directly to the types of 'Management' event.

25-- Users can additionally publish custom trace events in the form of

26−-- 'TraceEvLog' log messages or pass custom (i.e., completely user defined)

26+-- 'MxLog' log messages or pass custom (i.e., completely user defined)

27-- event data using the 'traceMessage' function.

28--

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

@@ -92,7 +92,7 @@

92module Control.Distributed.Process.Debug

93 ( -- * Exported Data Types

94 TraceArg(..)

95− , TraceEvent(..)

95+ , MxEvent(..)

96 , TraceFlags(..)

97 , TraceSubject(..)

98 -- * Configuring Tracing

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

147 )

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

149 ( TraceArg(..)

150− , TraceEvent(..)

150+ , MxEvent(..)

151 , TraceFlags(..)

152 , TraceSubject(..)

153 , defaultTraceFlags

@@ -197,7 +197,7 @@ import Prelude hiding (catch)

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

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

199-- terminating. See 'withTracer' for a mechanism that handles that.

200−startTracer :: (TraceEvent -> Process ()) -> Process ProcessId

200+startTracer :: (MxEvent -> Process ()) -> Process ProcessId

201startTracer handler = do

202 withRegisteredTracer $ \pid -> do

203 node <- processNode <$> ask

@@ -209,7 +209,7 @@ startTracer handler = do

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

210-- in case of failure).

211withTracer :: forall a.

212− (TraceEvent -> Process ())

212+ (MxEvent -> Process ())

213 -> Process a

214 -> Process (Either SomeException a)

215withTracer handler proc = do

@@ -224,7 +224,7 @@ withTracer handler proc = do

224 Nothing -> return ()

225 Just _ -> do

226 ref <- monitor tracer

227− send tracer TraceEvDisable

227+ send tracer MxTraceDisable

228 receiveWait [

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

230 (\_ -> return ())

@@ -241,12 +241,12 @@ withFlags flags proc = do

241 finally (setTraceFlags flags >> try proc)

242 (setTraceFlags oldFlags)

243

244−traceProxy :: ProcessId -> (TraceEvent -> Process ()) -> Process ()

244+traceProxy :: ProcessId -> (MxEvent -> Process ()) -> Process ()

245traceProxy pid act = do

246− proxy pid $ \(ev :: TraceEvent) ->

246+ proxy pid $ \(ev :: MxEvent) ->

247 case ev of

248− (TraceEvTakeover _) -> return False

248+ (MxTraceTakeover _) -> return False

249− TraceEvDisable -> die "disabled"

249+ MxTraceDisable -> die "disabled"

250 _ -> act ev >> return True

251

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

@@ -272,5 +272,5 @@ stopTracer =

272 basePid <- whereis "tracer.initial"

273 case basePid == (Just pid) of

274 True -> return ()

275− False -> send pid TraceEvDisable

275+ False -> send pid MxTraceDisable

276

src/Control/Distributed/Process/Internal/Primitives.hs+3−3

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

193 )

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

195 ( traceEvent

196− , TraceEvent(..)

196+ , MxEvent(..)

197 )

198import Control.Distributed.Process.Internal.WeakTQueue

199 ( newTQueueIO

@@ -227,7 +227,7 @@ send them msg = do

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

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

229 liftIO $ traceEvent (localEventBus node)

230− (TraceEvSent them us (createUnencodedMessage msg))

230+ (MxSent them us (createUnencodedMessage msg))

231

232-- | /Unsafe/ variant of 'send'. This function makes /no/ attempt to serialize

233-- and (in the case when the destination process resides on the same local

@@ -411,7 +411,7 @@ forward msg them = do

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

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

413 liftIO $ traceEvent (localEventBus node)

414− (TraceEvSent them us msg)

414+ (MxSent them us msg)

415

416

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

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

@@ -7,7 +7,7 @@

7-- tracing infrastructure's messaging fabric.

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

9 ( Tracer

10− , TraceEvent(..)

10+ , MxEvent(..)

11 -- * Sending Trace Data

12 , traceLog

13 , traceLogFmt

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

36 )

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

38 ( TraceArg(..)

39− , TraceEvent(..)

39+ , MxEvent(..)

40 , TraceFlags(..)

41 , TraceOk(..)

42 , TraceSubject(..)

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

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

32 , matchUnknown

33 )

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

35− ( TraceEvent(..)

35+ ( MxEvent(..)

36 , SetTrace(..)

37 , Addressable(..)

38 , TraceSubject(..)

@@ -135,7 +135,7 @@ systemLoggerTracer = do

135 let tr = sendTraceMsg node

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

137 where

138− sendTraceMsg :: LocalNode -> TraceEvent -> Process ()

138+ sendTraceMsg :: LocalNode -> MxEvent -> Process ()

139 sendTraceMsg node ev = do

140 now <- liftIO $ getCurrentTime

141 msg <- return $ (formatTime defaultTimeLocale "%c" now, buildTxt ev)

@@ -147,9 +147,9 @@ systemLoggerTracer = do

147 }

148 liftIO $ writeChan (localCtrlChan node) traceMsg

149

150− buildTxt :: TraceEvent -> String

150+ buildTxt :: MxEvent -> String

151− buildTxt (TraceEvLog msg) = msg

151+ buildTxt (MxLog msg) = msg

152− buildTxt ev = show ev

152+ buildTxt ev = show ev

153

154eventLogTracer :: Process ()

155eventLogTracer =

@@ -158,7 +158,7 @@ eventLogTracer =

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

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

160 where

161− writeTrace :: TraceEvent -> Process ()

161+ writeTrace :: MxEvent -> Process ()

162 writeTrace ev = liftIO $ traceEventIO (show ev)

163

164logfileTracer :: FilePath -> Process ()

@@ -172,14 +172,14 @@ logfileTracer p = do

172 logger h' = forever' $ do

173 receiveWait [

174 matchIf (\ev -> case ev of

175− TraceEvDisable -> True

175+ MxTraceDisable -> True

176− (TraceEvTakeover _) -> True

176+ (MxTraceTakeover _) -> True

177 _ -> False)

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

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

180 ]

181

182− writeTrace :: Handle -> TraceEvent -> Process ()

182+ writeTrace :: Handle -> MxEvent -> Process ()

183 writeTrace h ev = do

184 liftIO $ do

185 now <- getCurrentTime

@@ -200,23 +200,23 @@ traceController mv = do

200 where

201 traceLoop :: TracerState -> Process ()

202 traceLoop st = do

203− -- Trace events are forwarded to the trace target when tracing is enabled.

203+ -- Trace events are forwarded to the enabled trace target.

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

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

206 st' <- receiveWait [

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

208− -- We allow at most one trace client, which is a process id.

208+ -- We consider at most one trace client, which is a process.

209 -- Tracking multiple clients represents too high an overhead,

210− -- so we leave that kind of thing to our consumers to figure

210+ -- so we leave that kind of thing to our consumers (e.g., the

211− -- figure out.

211+ -- high level Debug client module) to figure out.

212 case set of

213 (TraceEnable pid) -> do

214 -- notify the previous tracer it has been replaced

215− sendTrace st (createUnencodedMessage (TraceEvTakeover pid))

215+ sendTrace st (createUnencodedMessage (MxTraceTakeover pid))

216 sendOk setResp

217 return st { client = (Just pid) }

218 TraceDisable -> do

219− sendTrace st (createUnencodedMessage TraceEvDisable)

219+ sendTrace st (createUnencodedMessage MxTraceDisable)

220 sendOk setResp

221 return st { client = Nothing })

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

@@ -225,7 +225,7 @@ traceController mv = do

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

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

227 , matchAny (\ev ->

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

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

229 ]

230 traceLoop st'

231

@@ -263,8 +263,8 @@ traceController mv = do

263applyTraceFlags :: TraceFlags -> TracerState -> Process TracerState

264applyTraceFlags flags' state = return state { flags = flags' }

265

266−handleTrace :: TracerState -> Message -> TraceEvent -> Process TracerState

266+handleTrace :: TracerState -> Message -> MxEvent -> Process TracerState

267−handleTrace st msg ev@(TraceEvRegistered p n) =

267+handleTrace st msg ev@(MxRegistered p n) =

268 let regNames' =

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

270 (Set.singleton n)

@@ -272,7 +272,7 @@ handleTrace st msg ev@(TraceEvRegistered p n) =

272 in do

273 traceEv ev msg (traceRegistered (flags st)) st

274 return st { regNames = regNames' }

275−handleTrace st msg ev@(TraceEvUnRegistered p n) =

275+handleTrace st msg ev@(MxUnRegistered p n) =

276 let f ns = case ns of

277 Nothing -> Nothing

278 Just ns' -> Just (Set.delete n ns')

@@ -280,29 +280,29 @@ handleTrace st msg ev@(TraceEvUnRegistered p n) =

280 in do

281 traceEv ev msg (traceUnregistered (flags st)) st

282 return st { regNames = regNames' }

283−handleTrace st msg ev@(TraceEvSpawned _) = do

283+handleTrace st msg ev@(MxSpawned _) = do

284 traceEv ev msg (traceSpawned (flags st)) st >> return st

285−handleTrace st msg ev@(TraceEvDied _ _) = do

285+handleTrace st msg ev@(MxProcessDied _ _) = do

286 traceEv ev msg (traceDied (flags st)) st >> return st

287−handleTrace st msg ev@(TraceEvSent _ _ _) =

287+handleTrace st msg ev@(MxSent _ _ _) =

288 traceEv ev msg (traceSend (flags st)) st >> return st

289−handleTrace st msg ev@(TraceEvReceived _ _) =

289+handleTrace st msg ev@(MxReceived _ _) =

290 traceEv ev msg (traceRecv (flags st)) st >> return st

291handleTrace st msg ev = do

292 case ev of

293− (TraceEvNodeDied _ _) ->

293+ (MxNodeDied _ _) ->

294 case (traceNodes (flags st)) of

295 True -> sendTrace st msg

296 False -> return ()

297− (TraceEvUser _) -> sendTrace st msg

297+ (MxUser _) -> sendTrace st msg

298− (TraceEvLog _) -> sendTrace st msg

298+ (MxLog _) -> sendTrace st msg

299 _ ->

300 case (traceConnections (flags st)) of

301 True -> sendTrace st msg

302 False -> return ()

303 return st

304

305−traceEv :: TraceEvent

305+traceEv :: MxEvent

306 -> Message

307 -> Maybe TraceSubject

308 -> TracerState

src/Control/Distributed/Process/Internal/Trace/Types.hs+55−52

@@ -1,7 +1,7 @@

1-- | Tracing/Debugging support - Types

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

3 ( MxEventBus(..)

4− , TraceEvent(..)

4+ , MxEvent(..)

5 , SetTrace(..)

6 , Addressable(..)

7 , TraceSubject(..)

@@ -47,6 +47,7 @@ import Data.Typeable

47import System.Mem.Weak (deRefWeak)

48import Network.Transport

49 ( ConnectionId

50+ , EndPointAddress

50 )

51 )

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

53--------------------------------------------------------------------------------

@@ -56,47 +57,49 @@ import Network.Transport

56data SetTrace = TraceEnable !ProcessId | TraceDisable

57data SetTrace = TraceEnable !ProcessId | TraceDisable

57 deriving (Typeable, Eq, Show)

58 deriving (Typeable, Eq, Show)

60+-- TODO: MOVE the whole Trace namespace into C.D.Process.Management

61+

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

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

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

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

61-- at most one can be registered per-node.

64-- at most one can be registered per-node.

62−data TraceEvent =

65+data MxEvent =

63− TraceEvSpawned ProcessId

66+ MxSpawned ProcessId

64 -- ^ fired whenever a local process is spawned

67 -- ^ fired whenever a local process is spawned

65− | TraceEvRegistered ProcessId String

68+ | MxRegistered ProcessId String

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

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

67− | TraceEvUnRegistered ProcessId String

70+ | MxUnRegistered ProcessId String

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

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

69− | TraceEvDied ProcessId DiedReason

72+ | MxProcessDied ProcessId DiedReason

70 -- ^ fired whenever a process dies

73 -- ^ fired whenever a process dies

71− | TraceEvNodeDied NodeId DiedReason

74+ | MxNodeDied NodeId DiedReason

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

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

73− | TraceEvSent ProcessId ProcessId Message

76+ | MxSent ProcessId ProcessId Message

74 -- ^ fired whenever a message is sent from a local process

77 -- ^ fired whenever a message is sent from a local process

75− | TraceEvReceived ProcessId Message

78+ | MxReceived ProcessId Message

76 -- ^ fired whenever a message is received by a local process

79 -- ^ fired whenever a message is received by a local process

77− | TraceEvConnected ConnectionId

80+ | MxConnected ConnectionId EndPointAddress

78 -- ^ fired when a network-transport connection is first established

81 -- ^ fired when a network-transport connection is first established

79− | TraceEvDisconnected ConnectionId

82+ | MxDisconnected ConnectionId EndPointAddress

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

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

81− | TraceEvUser Message

84+ | MxUser Message

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

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

83− | TraceEvLog String

86+ | MxLog String

84 -- ^ a /logging/ event - used for debugging purposes only

87 -- ^ a /logging/ event - used for debugging purposes only

85− | TraceEvTakeover ProcessId

88+ | MxTraceTakeover ProcessId

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

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

87− | TraceEvDisable

90+ | MxTraceDisable

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

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

89 deriving (Typeable, Show)

92 deriving (Typeable, Show)

91class Addressable a where

94class Addressable a where

92 resolveToPid :: a -> Maybe ProcessId

95 resolveToPid :: a -> Maybe ProcessId

94−instance Addressable TraceEvent where

97+instance Addressable MxEvent where

95− resolveToPid (TraceEvSpawned p) = Just p

98+ resolveToPid (MxSpawned p) = Just p

96− resolveToPid (TraceEvDied p _) = Just p

99+ resolveToPid (MxProcessDied p _) = Just p

97− resolveToPid (TraceEvSent _ p _) = Just p

100+ resolveToPid (MxSent _ p _) = Just p

98− resolveToPid (TraceEvReceived p _) = Just p

101+ resolveToPid (MxReceived p _) = Just p

99− resolveToPid _ = Nothing

102+ resolveToPid _ = Nothing

100

103

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

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

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

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

@@ -149,7 +152,7 @@ data TraceOk = TraceOk

149--------------------------------------------------------------------------------

152--------------------------------------------------------------------------------

150

153

151traceLog :: MxEventBus -> String -> IO ()

154traceLog :: MxEventBus -> String -> IO ()

152−traceLog tr s = publishEvent tr (unsafeCreateUnencodedMessage $ TraceEvLog s)

155+traceLog tr s = publishEvent tr (unsafeCreateUnencodedMessage $ MxLog s)

153

156

154traceLogFmt :: MxEventBus

157traceLogFmt :: MxEventBus

155 -> String

158 -> String

@@ -161,11 +164,11 @@ traceLogFmt t d ls =

161 toS (TraceStr s) = s

164 toS (TraceStr s) = s

162 toS (Trace a) = show a

165 toS (Trace a) = show a

163

166

164−traceEvent :: MxEventBus -> TraceEvent -> IO ()

167+traceEvent :: MxEventBus -> MxEvent -> IO ()

165traceEvent tr ev = publishEvent tr (unsafeCreateUnencodedMessage ev)

168traceEvent tr ev = publishEvent tr (unsafeCreateUnencodedMessage ev)

166

169

167traceMessage :: Serializable m => MxEventBus -> m -> IO ()

170traceMessage :: Serializable m => MxEventBus -> m -> IO ()

168−traceMessage tr msg = traceEvent tr (TraceEvUser (unsafeCreateUnencodedMessage msg))

171+traceMessage tr msg = traceEvent tr (MxUser (unsafeCreateUnencodedMessage msg))

169

172

170enableTrace :: MxEventBus -> ProcessId -> IO ()

173enableTrace :: MxEventBus -> ProcessId -> IO ()

171enableTrace t p =

174enableTrace t p =

@@ -223,38 +226,38 @@ instance Binary SetTrace where

223 2 -> return TraceDisable

226 2 -> return TraceDisable

224 _ -> error "SetTrace.get - invalid header"

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

225

228

226−instance Binary TraceEvent where

229+instance Binary MxEvent where

227− put (TraceEvSpawned pid) = putWord8 1 >> put pid

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

228− put (TraceEvRegistered pid str) = putWord8 2 >> put pid >> put str

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

229− put (TraceEvUnRegistered pid str) = putWord8 3 >> put pid >> put str

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

230− put (TraceEvDied pid res) = putWord8 4 >> put pid >> put res

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

231− put (TraceEvNodeDied nid res) = putWord8 5 >> put nid >> put res

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

232− put (TraceEvSent to from msg) = putWord8 6 >> put to >> put from >> put msg

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

233− put (TraceEvReceived pid msg) = putWord8 7 >> put pid >> put msg

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

234− put (TraceEvConnected cid) = putWord8 8 >> put cid

237+ put (MxConnected cid epid) = putWord8 8 >> put cid >> put epid

235− put (TraceEvDisconnected cid) = putWord8 9 >> put cid

238+ put (MxDisconnected cid epid) = putWord8 9 >> put cid >> put epid

236− put (TraceEvUser msg) = putWord8 10 >> put msg

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

237− put (TraceEvLog msg) = putWord8 11 >> put msg

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

238− put (TraceEvTakeover pid) = putWord8 12 >> put pid

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

239− put TraceEvDisable = putWord8 13

242+ put MxTraceDisable = putWord8 13

240

243

241 get = do

244 get = do

242 header <- getWord8

245 header <- getWord8

243 case header of

246 case header of

244− 1 -> TraceEvSpawned <$> get

247+ 1 -> MxSpawned <$> get

245− 2 -> TraceEvRegistered <$> get <*> get

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

246− 3 -> TraceEvUnRegistered <$> get <*> get

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

247− 4 -> TraceEvDied <$> get <*> get

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

248− 5 -> TraceEvNodeDied <$> get <*> get

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

249− 6 -> TraceEvSent <$> get <*> get <*> get

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

250− 7 -> TraceEvReceived <$> get <*> get

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

251− 8 -> TraceEvConnected <$> get

254+ 8 -> MxConnected <$> get <*> get

252− 9 -> TraceEvDisconnected <$> get

255+ 9 -> MxDisconnected <$> get <*> get

253− 10 -> TraceEvUser <$> get

256+ 10 -> MxUser <$> get

254− 11 -> TraceEvLog <$> get

257+ 11 -> MxLog <$> get

255− 12 -> TraceEvTakeover <$> get

258+ 12 -> MxTraceTakeover <$> get

256− 13 -> return TraceEvDisable

259+ 13 -> return MxTraceDisable

257− _ -> error "TraceEvent.get - invalid header"

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

258

261

259instance Binary TraceSubject where

262instance Binary TraceSubject where

260 put TraceAll = putWord8 1

263 put TraceAll = putWord8 1

src/Control/Distributed/Process/Internal/Types.hs+1

@@ -215,6 +215,7 @@ data MxEventBus =

215 , tracer :: !Tracer

216 , evbuss :: !(Weak (CQueue Message))

217 , mxNew :: !((Message -> Process ()) -> IO ProcessId)

218+-- , mxReg :: !(StrictMVar (Map MxAgentId ))

218 }

219 }

219

220

220-- | Local nodes

221-- | Local nodes

src/Control/Distributed/Process/Management.hs+11−2

@@ -20,7 +20,7 @@

20--

21-- /Management Extensions/ for Cloud Haskell.

22--

23−-- This module provides a standard API for monitoring and/or responding to

23+-- This module provides a standard API for receiving and/or responding to

24-- system events in a Cloud Haskell node.

25--

26-- [Architecture Overview]

@@ -37,7 +37,6 @@

37-- 3. @Remote@ - provides an API for remote management and administration

38--

39--

40−--

41-----------------------------------------------------------------------------

40-----------------------------------------------------------------------------

42module Control.Distributed.Process.Management where

41module Control.Distributed.Process.Management where

@@ -73,3 +72,13 @@ mxAgent handler = do

73 -- registerAgent pid

72 -- registerAgent pid

74 return pid

73 return pid

75+mxPublish = undefined

76+

77+mxSet = undefined

78+

79+mxGet = undefined

80+

81+mxIncrement = undefined

82+

83+mxDecrement = undefined

84+

src/Control/Distributed/Process/Management/Bus.hs+31

@@ -9,10 +9,41 @@ import Control.Distributed.Process.Internal.CQueue

9import Control.Distributed.Process.Internal.Types

10 ( MxEventBus(..)

11 , Message

12+ , ProcessId

13+ , DiedReason

14+ , NodeId

12 )

15 )

13import Data.Foldable (forM_)

16import Data.Foldable (forM_)

17+import Data.Typeable (Typeable)

18+import Network.Transport

19+ ( ConnectionId

20+ , EndPointAddress

21+ )

14import System.Mem.Weak (deRefWeak)

22import System.Mem.Weak (deRefWeak)

24+data MxEvent =

25+ MxSpawned ProcessId

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

27+ | MxRegistered ProcessId String

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

29+ | MxUnRegistered ProcessId String

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

31+ | MxProcessDied ProcessId DiedReason

32+ -- ^ fired whenever a process dies

33+ | MxNodeDied NodeId DiedReason

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

35+ | MxSent ProcessId ProcessId Message

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

37+ | MxReceived ProcessId Message

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

39+ | MxConnected ConnectionId EndPointAddress

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

41+ | MxDisconnected ConnectionId

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

43+ | MxUser Message

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

45+ deriving (Typeable, Show)

46+

16publishEvent :: MxEventBus -> Message -> IO ()

47publishEvent :: MxEventBus -> Message -> IO ()

17publishEvent MxEventBusInitialising _ = return ()

48publishEvent MxEventBusInitialising _ = return ()

18publishEvent (MxEventBus _ _ wqRef _) msg = do

49publishEvent (MxEventBus _ _ wqRef _) msg = do

src/Control/Distributed/Process/Node.hs+12−12

@@ -154,7 +154,7 @@ import qualified Control.Distributed.Process.Internal.Trace.Remote as Trace

154 )

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

156 ( TraceArg(..)

157− , TraceEvent(..)

157+ , MxEvent(..)

158 , traceEvent

159 , traceLogFmt

160 , enableTrace

@@ -355,7 +355,7 @@ forkProcess node proc = modifyMVar (localState node) startProcess

355

356

357 -- see note [tracer/forkProcess races]

358− trace node (TraceEvSpawned pid)

358+ trace node (MxSpawned pid)

359

360 if lpidCounter lpid == maxBound

361 then do

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

435 NT.ConnectionOpened cid rel theirAddr ->

436 if rel == NT.ReliableOrdered

437 then

438− trace node (TraceEvConnected cid)

438+ trace node (MxConnected cid theirAddr)

439 >> go (

440 (incomingAt cid ^= Just (theirAddr, Uninit))

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

@@ -451,7 +451,7 @@ handleIncomingMessages node = go initConnectionState

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

452 let msg = payloadToMessage payload

453 enqueue queue msg -- 'enqueue' is strict

454− trace node (TraceEvReceived pid msg)

454+ trace node (MxReceived pid msg)

455 go st

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

457 mChan <- deRefWeak chan'

@@ -500,7 +500,7 @@ handleIncomingMessages node = go initConnectionState

500 Nothing ->

501 invalidRequest cid st

502 Just (src, _) -> do

503− trace node (TraceEvDisconnected cid)

503+ trace node (MxDisconnected cid src)

504 go ( (incomingAt cid ^= Nothing)

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

506 $ st

@@ -599,8 +599,8 @@ traceNotifyDied node ident reason =

599 -- TODO: sendPortDied notifications

600 liftIO $ withLocalTracer node $ \t ->

601 case ident of

602− (NodeIdentifier nid) -> traceEvent t (TraceEvNodeDied nid reason)

602+ (NodeIdentifier nid) -> traceEvent t (MxNodeDied nid reason)

603− (ProcessIdentifier pid) -> traceEvent t (TraceEvDied pid reason)

603+ (ProcessIdentifier pid) -> traceEvent t (MxProcessDied pid reason)

604 _ -> return ()

605

606traceEventFmtIO :: LocalNode

@@ -610,7 +610,7 @@ traceEventFmtIO :: LocalNode

610traceEventFmtIO node fmt args =

611 withLocalTracer node $ \t -> traceLogFmt t fmt args

612

613−trace :: LocalNode -> TraceEvent -> IO ()

613+trace :: LocalNode -> MxEvent -> IO ()

614trace node ev = withLocalTracer node $ \t -> traceEvent t ev

615

616withLocalTracer :: LocalNode -> (MxEventBus -> IO ()) -> IO ()

@@ -781,7 +781,7 @@ ncEffectSpawn pid cProc ref = do

781 mProc <- unClosure cProc

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

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

784− -- TODO:

784+ -- TODO:

785 let proc = case mProc of

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

787 Right p -> p

@@ -813,8 +813,8 @@ ncEffectRegister from label atnode mPid reregistration = do

813 do modify' $ registeredHereFor label ^= mPid

814 updateRemote node currentVal mPid

815 case mPid of

816− (Just p) -> liftIO $ trace node (TraceEvRegistered p label)

816+ (Just p) -> liftIO $ trace node (MxRegistered p label)

817− Nothing -> liftIO $ trace node (TraceEvUnRegistered (fromJust currentVal) label)

817+ Nothing -> liftIO $ trace node (MxUnRegistered (fromJust currentVal) label)

818 liftIO $ sendMessage node

819 (NodeIdentifier (localNodeId node))

820 (ProcessIdentifier from)

@@ -882,7 +882,7 @@ ncEffectLocalSend :: LocalNode -> ProcessId -> Message -> NC ()

882ncEffectLocalSend node to msg =

883 liftIO $ withLocalProc node to $ \p -> do

884 enqueue (processQueue p) msg

885− trace node (TraceEvReceived to msg)

885+ trace node (MxReceived to msg)

886

887-- [Issue #DP-20]

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

tests/TestTracing.hs+13−19

@@ -18,7 +18,6 @@ import Control.Distributed.Process.Node

18 , initRemoteTable

19 , closeLocalNode

20 , LocalNode)

21−import Control.Exception as Ex (catch)

22import Network.Transport.TCP

21import Network.Transport.TCP

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

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

24import Prelude hiding (catch, log)

23import Prelude hiding (catch, log)

@@ -32,11 +31,6 @@ import Test.HUnit (Assertion)

32import Test.HUnit.Base (assertBool)

31import Test.HUnit.Base (assertBool)

33import Test.Framework.Providers.HUnit (testCase)

32import Test.Framework.Providers.HUnit (testCase)

35−import System.Environment (getEnv)

36−import System.Posix.Env

37− ( setEnv

38− )

39−

40-- these utilities have been cribbed from distributed-process-platform

34-- these utilities have been cribbed from distributed-process-platform

41-- we should really find a way to share them...

35-- we should really find a way to share them...

@@ -81,8 +75,8 @@ testSpawnTracing result = do

81 evDied <- liftIO $ newEmptyMVar

75 evDied <- liftIO $ newEmptyMVar

82 tracer <- startTracer $ \ev -> do

76 tracer <- startTracer $ \ev -> do

83 case ev of

77 case ev of

84− (TraceEvSpawned p) -> liftIO $ putMVar evSpawned p

78+ (MxSpawned p) -> liftIO $ putMVar evSpawned p

85− (TraceEvDied p r) -> liftIO $ putMVar evDied (p, r)

79+ (MxProcessDied p r) -> liftIO $ putMVar evDied (p, r)

86 _ -> return ()

80 _ -> return ()

88 (sp, rp) <- newChan

82 (sp, rp) <- newChan

@@ -115,7 +109,7 @@ testTraceRecvExplicitPid result = do

115 withTracer

109 withTracer

116 (\ev ->

110 (\ev ->

117 case ev of

111 case ev of

118− (TraceEvReceived pid' _) -> stash res (pid == pid')

112+ (MxReceived pid' _) -> stash res (pid == pid')

119 _ -> return ()) $ do

113 _ -> return ()) $ do

120 (sp, rp) <- newChan

114 (sp, rp) <- newChan

121 send pid sp

115 send pid sp

@@ -137,7 +131,7 @@ testTraceRecvNamedPid result = do

137 withTracer

131 withTracer

138 (\ev ->

132 (\ev ->

139 case ev of

133 case ev of

140− (TraceEvReceived pid' _) -> stash res (pid == pid')

134+ (MxReceived pid' _) -> stash res (pid == pid')

141 _ -> return ()) $ do

135 _ -> return ()) $ do

142 (sp, rp) <- newChan

136 (sp, rp) <- newChan

143 send pid sp

137 send pid sp

@@ -155,7 +149,7 @@ testTraceSending result = do

155 withTracer

149 withTracer

156 (\ev ->

150 (\ev ->

157 case ev of

151 case ev of

158− (TraceEvSent to from msg) -> do

152+ (MxSent to from msg) -> do

159 (Just s) <- unwrapMessage msg :: Process (Maybe String)

153 (Just s) <- unwrapMessage msg :: Process (Maybe String)

160 stash res (to == pid && from == self && s == "hello there")

154 stash res (to == pid && from == self && s == "hello there")

161 stash res (to == pid && from == self)

155 stash res (to == pid && from == self)

@@ -180,7 +174,7 @@ testTraceRegistration result = do

180 withTracer

174 withTracer

181 (\ev ->

175 (\ev ->

182 case ev of

176 case ev of

183− TraceEvRegistered p s ->

177+ MxRegistered p s ->

184 stash res (p == pid && s == "foobar")

178 stash res (p == pid && s == "foobar")

185 _ ->

179 _ ->

186 return ()) $ do

180 return ()) $ do

@@ -207,7 +201,7 @@ testTraceUnRegistration result = do

207 withTracer

201 withTracer

208 (\ev ->

202 (\ev ->

209 case ev of

203 case ev of

210− TraceEvUnRegistered p n -> do

204+ MxUnRegistered p n -> do

211 stash res (p == pid && n == "foobar")

205 stash res (p == pid && n == "foobar")

212 send pid ()

206 send pid ()

213 _ ->

207 _ ->

@@ -241,21 +235,21 @@ testTraceLayering result = do

241 withTracer

235 withTracer

242 (\ev ->

236 (\ev ->

243 case ev of

237 case ev of

244− TraceEvDied _ _ -> liftIO $ putMVar died ()

238+ MxProcessDied _ _ -> liftIO $ putMVar died ()

245− _ -> return ())

239+ _ -> return ())

246 ( do {

240 ( do {

247 recv <- liftIO $ newEmptyMVar

241 recv <- liftIO $ newEmptyMVar

248 ; withTracer

242 ; withTracer

249 (\ev' ->

243 (\ev' ->

250 case ev' of

244 case ev' of

251− TraceEvReceived _ _ -> liftIO $ putMVar recv ()

245+ MxReceived _ _ -> liftIO $ putMVar recv ()

252 _ -> return ())

246 _ -> return ())

253 ( do {

247 ( do {

254 user <- liftIO $ newEmptyMVar

248 user <- liftIO $ newEmptyMVar

255 ; withTracer

249 ; withTracer

256 (\ev'' ->

250 (\ev'' ->

257 case ev'' of

251 case ev'' of

258− TraceEvUser _ -> liftIO $ putMVar user ()

252+ MxUser _ -> liftIO $ putMVar user ()

259 _ -> return ())

253 _ -> return ())

260 (send pid () >> (liftIO $ takeMVar user))

254 (send pid () >> (liftIO $ takeMVar user))

261 ; liftIO $ takeMVar recv

255 ; liftIO $ takeMVar recv

@@ -276,7 +270,7 @@ testRemoteTraceRelay result =

276 -- garbage on stderr for the duration of the test run.

270 -- garbage on stderr for the duration of the test run.

277 -- Here we set up that relay, and then wait for a signal

271 -- Here we set up that relay, and then wait for a signal

278 -- that the tracer (on node1) has seen the expected

272 -- that the tracer (on node1) has seen the expected

279− -- TraceEvSpawned message, at which point we're finished

273+ -- MxSpawned message, at which point we're finished

280 (Just log') <- whereis "logger"

274 (Just log') <- whereis "logger"

281 pid <- liftIO $ forkProcess node2 $ do

275 pid <- liftIO $ forkProcess node2 $ do

282 logRelay <- spawnLocal $ relay log'

276 logRelay <- spawnLocal $ relay log'

@@ -293,7 +287,7 @@ testRemoteTraceRelay result =

293 withTracer

287 withTracer

294 (\ev ->

288 (\ev ->

295 case ev of

289 case ev of

296− TraceEvSpawned p -> stash observedPid p >> send pid ()

290+ MxSpawned p -> stash observedPid p >> send pid ()

297 _ -> return ()) $ do

291 _ -> return ()) $ do

298 relayPid <- startTraceRelay nid

292 relayPid <- startTraceRelay nid

299 liftIO $ threadDelay 1000000

293 liftIO $ threadDelay 1000000