4957534 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Management Agent API Improvements

4957534Tim Watson13 years ago

distributed-process.cabal+1

compare with master

unifiedsplit

@@ -81,6 +81,7 @@ Library

81 Control.Distributed.Process.Internal.StrictContainerAccessors,

82 Control.Distributed.Process.Management.Bus,

83 Control.Distributed.Process.Management.Table,

84+ Control.Distributed.Process.Management.Types,

84 Control.Distributed.Process.Management.Trace.Primitives,

85 Control.Distributed.Process.Management.Trace.Primitives,

85 Control.Distributed.Process.Management.Trace.Remote,

86 Control.Distributed.Process.Management.Trace.Remote,

86 Control.Distributed.Process.Management.Trace.Types,

87 Control.Distributed.Process.Management.Trace.Types,

src/Control/Distributed/Process/Debug.hs+3−2

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

92module Control.Distributed.Process.Debug

93 ( -- * Exported Data Types

94 TraceArg(..)

95− , MxEvent(..)

96 , TraceFlags(..)

95 , TraceFlags(..)

97 , TraceSubject(..)

96 , TraceSubject(..)

98 -- * Configuring Tracing

97 -- * Configuring Tracing

@@ -145,9 +144,11 @@ import Control.Distributed.Process.Internal.Types

145 , LocalProcess(..)

144 , LocalProcess(..)

146 , ProcessMonitorNotification(..)

145 , ProcessMonitorNotification(..)

147 )

146 )

147+import Control.Distributed.Process.Management.Types

148+ ( MxEvent(..)

149+ )

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

150import Control.Distributed.Process.Management.Trace.Types

149 ( TraceArg(..)

151 ( TraceArg(..)

150− , MxEvent(..)

151 , TraceFlags(..)

152 , TraceFlags(..)

152 , TraceSubject(..)

153 , TraceSubject(..)

153 , defaultTraceFlags

154 , defaultTraceFlags

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

@@ -191,9 +191,14 @@ import Control.Distributed.Process.Internal.Messaging

191 , disconnect

192 , sendCtrlMsg

193 )

194+import qualified Control.Distributed.Process.Management.Bus as Mx

195+ ( publishEvent

196+ )

197+import Control.Distributed.Process.Management.Types

198+ ( MxEvent(..)

199+ )

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

200import Control.Distributed.Process.Management.Trace.Types

195 ( traceEvent

201 ( traceEvent

196− , MxEvent(..)

197 )

202 )

198import Control.Distributed.Process.Internal.WeakTQueue

203import Control.Distributed.Process.Internal.WeakTQueue

199 ( newTQueueIO

204 ( newTQueueIO

@@ -224,10 +229,11 @@ send them msg = do

224 (ProcessIdentifier them)

229 (ProcessIdentifier them)

225 NoImplicitReconnect

230 NoImplicitReconnect

226 msg

231 msg

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

232+ -- We do not fire the trace event until after sending completes, since

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

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

229− liftIO $ traceEvent (localEventBus node)

234+ liftIO $ do

230− (MxSent them us (createUnencodedMessage msg))

235+ let msg = createUnencodedMessage $ (MxSent them us (createUnencodedMessage msg))

236+ Mx.publishEvent (localEventBus node) msg

231

237

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

238-- | /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

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

@@ -410,8 +416,8 @@ forward msg them = do

410 (messageToPayload msg)

416 (messageToPayload msg)

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

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

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

413− liftIO $ traceEvent (localEventBus node)

419+ liftIO $ Mx.publishEvent (localEventBus node)

414− (MxSent them us msg)

420+ (createUnencodedMessage (MxSent them us msg))

415

421

416

422

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

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

src/Control/Distributed/Process/Internal/Types.hs+11−5

@@ -200,22 +200,26 @@ nullProcessId nid =

200-- Local nodes and processes --

201--------------------------------------------------------------------------------

202

203−-- | Provides access to the trace controller.

203+-- | Provides access to the trace controller

204data Tracer = Tracer

205 {

206+ -- | Process id for the currently active trace handler

206 tracerPid :: !ProcessId

207 tracerPid :: !ProcessId

208+ -- | Weak reference to the tracer controller's mailbox

207 , weakQ :: !(Weak (CQueue Message))

209 , weakQ :: !(Weak (CQueue Message))

208 }

210 }

209

211

210−-- | Required for the system management and tracing facilities.

212+-- | Local system management event bus state

211data MxEventBus =

213data MxEventBus =

212 MxEventBusInitialising

214 MxEventBusInitialising

213 | MxEventBus

215 | MxEventBus

214 {

216 {

217+ -- | Process id of the management agent controller process

215 agent :: !ProcessId

218 agent :: !ProcessId

216− , tracer :: !Tracer

219+ -- | Weak reference to the management agent controller's mailbox

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

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

221+ -- | API for adding management agents to a running node

218− , mxNew :: !((TChan Message -> Process ()) -> IO ProcessId)

222+ , mxNew :: !(((TChan Message, TChan Message) -> Process ()) -> IO ProcessId)

219-- , mxReg :: !(StrictMVar (Map MxAgentId ))

223-- , mxReg :: !(StrictMVar (Map MxAgentId ))

220 }

224 }

221

225

@@ -229,11 +233,13 @@ data LocalNode = LocalNode

229 , localState :: !(StrictMVar LocalNodeState)

233 , localState :: !(StrictMVar LocalNodeState)

230 -- | Channel for the node controller

234 -- | Channel for the node controller

231 , localCtrlChan :: !(Chan NCMsg)

235 , localCtrlChan :: !(Chan NCMsg)

232− -- | Current active system debug/trace log

236+ -- | Internal management event bus

233 , localEventBus :: !MxEventBus

237 , localEventBus :: !MxEventBus

234 -- | Runtime lookup table for supporting closures

238 -- | Runtime lookup table for supporting closures

235 -- TODO: this should be part of the CH state, not the local endpoint state

239 -- TODO: this should be part of the CH state, not the local endpoint state

236 , remoteTable :: !RemoteTable

240 , remoteTable :: !RemoteTable

241+ -- | Active local tracer process. This is our only non-strict field.

242+ , localTracer :: Tracer

237 }

243 }

238

244

239data ImplicitReconnect = WithImplicitReconnect | NoImplicitReconnect

245data ImplicitReconnect = WithImplicitReconnect | NoImplicitReconnect

src/Control/Distributed/Process/Management.hs+73−29

@@ -3,7 +3,6 @@

3{-# LANGUAGE StandaloneDeriving #-}

4{-# LANGUAGE ExistentialQuantification #-}

5{-# LANGUAGE ScopedTypeVariables #-}

6−{-# LANGUAGE PatternGuards #-}

7{-# LANGUAGE RecordWildCards #-}

6{-# LANGUAGE RecordWildCards #-}

8{-# LANGUAGE FlexibleInstances #-}

7{-# LANGUAGE FlexibleInstances #-}

9{-# LANGUAGE UndecidableInstances #-}

8{-# LANGUAGE UndecidableInstances #-}

@@ -40,20 +39,24 @@

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

39-----------------------------------------------------------------------------

41module Control.Distributed.Process.Management

40module Control.Distributed.Process.Management

42 (

41 (

43− MxAction()

42+ MxEvent(..)

44− , MxAgentId

43+ -- * Firing Arbitrary /Mx Events/

45− -- * Firing /mx events/

46 , mxNotify

44 , mxNotify

47− -- * Constructing Agents

45+ -- * Constructing Mx Agents

46+ , MxAction()

47+ , MxAgentId

48+ , MxAgent()

48 , mxAgent

49 , mxAgent

49− , mxAgentDeactivate

50+ , MxSink

50− , mxAgentReady

51+ , mxSink

51− , mxAgentPublish

52+ , mxGetId

53+ , mxDeactivate

54+ , mxReady

55+ , mxBroadcast

52 , mxSetLocal

56 , mxSetLocal

53 , mxGetLocal

57 , mxGetLocal

54− -- * Lifting

58+ , mxLift

55− , liftP

56− -- * Management Data API

59+ -- * Mx Data API

57 , mxPublish

60 , mxPublish

58 , mxSet

61 , mxSet

59 , mxGet

62 , mxGet

@@ -63,12 +66,14 @@ import Control.Applicative ((<$>))

63import Control.Concurrent.STM (atomically)

66import Control.Concurrent.STM (atomically)

64import Control.Concurrent.STM.TChan

67import Control.Concurrent.STM.TChan

65 ( readTChan

68 ( readTChan

69+ , writeTChan

66 )

70 )

67import Control.Distributed.Process.Internal.Primitives

71import Control.Distributed.Process.Internal.Primitives

68 ( newChan

72 ( newChan

69 , nsend

73 , nsend

70 , receiveWait

74 , receiveWait

71 , matchChan

75 , matchChan

76+ , unwrapMessage

72 )

77 )

73import Control.Distributed.Process.Internal.Types

78import Control.Distributed.Process.Internal.Types

74 ( Process

79 ( Process

@@ -84,8 +89,11 @@ import qualified Control.Distributed.Process.Management.Table as Table

84import Control.Distributed.Process.Management.Types

89import Control.Distributed.Process.Management.Types

85 ( MxAgentId(..)

90 ( MxAgentId(..)

86 , MxAgent(..)

91 , MxAgent(..)

92+ , MxAction(..)

87 , MxAgentState(..)

93 , MxAgentState(..)

88 , MxAgentStart(..)

94 , MxAgentStart(..)

95+ , MxSink

96+ , MxEvent(..)

89 )

97 )

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

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

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

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

@@ -99,10 +107,6 @@ import qualified Control.Monad.State as ST

99 , runStateT

107 , runStateT

100 )

108 )

101

109

102−data MxAction =

103− MxAgentDeactivate !String

104− | MxAgentReady

105−

106-- | Publishes an arbitrary @Serializable@ message to the management event bus.

110-- | Publishes an arbitrary @Serializable@ message to the management event bus.

107-- Note that /no attempt is made to force the argument/, therefore it is very

111-- Note that /no attempt is made to force the argument/, therefore it is very

108-- important that you do not pass unevaluated thunks that might crash the

112-- important that you do not pass unevaluated thunks that might crash the

@@ -141,16 +145,23 @@ mxGet mxId = Table.fetch (Table.MxForAgent mxId)

141-- API for writing user defined management extensions (i.e., agents) --

145-- API for writing user defined management extensions (i.e., agents) --

142--------------------------------------------------------------------------------

146--------------------------------------------------------------------------------

143

147

144−mxAgentPublish = undefined

148+mxGetId :: MxAgent s MxAgentId

149+mxGetId = ST.get >>= return . mxAgentId

145

150

146−mxAgentDeactivate :: forall s. String -> MxAgent s MxAction

151+mxBroadcast :: (Serializable m) => m -> MxAgent s ()

147−mxAgentDeactivate = return . MxAgentDeactivate

152+mxBroadcast msg = do

153+ state <- ST.get

154+ mxLift $ liftIO $ atomically $ do

155+ writeTChan (mxBus state) (unsafeCreateUnencodedMessage msg)

148

156

149−mxAgentReady :: forall s. MxAgent s MxAction

157+mxDeactivate :: forall s. String -> MxAgent s MxAction

150−mxAgentReady = return MxAgentReady

158+mxDeactivate = return . MxAgentDeactivate

151

159

160+mxReady :: forall s. MxAgent s MxAction

161+mxReady = return MxAgentReady

162+

152−liftP :: Process a -> MxAgent s a

163+mxLift :: Process a -> MxAgent s a

153−liftP p = MxAgent $ ST.lift p

164+mxLift p = MxAgent $ ST.lift p

154

165

155mxSetLocal :: s -> MxAgent s ()

166mxSetLocal :: s -> MxAgent s ()

156mxSetLocal s = ST.modify $ \st -> st { mxLocalState = s }

167mxSetLocal s = ST.modify $ \st -> st { mxLocalState = s }

@@ -158,31 +169,64 @@ mxSetLocal s = ST.modify $ \st -> st { mxLocalState = s }

158mxGetLocal :: MxAgent s s

169mxGetLocal :: MxAgent s s

159mxGetLocal = ST.get >>= return . mxLocalState

170mxGetLocal = ST.get >>= return . mxLocalState

160

171

172+mxSink :: forall s m . (Serializable m)

173+ => (m -> MxAgent s MxAction)

174+ -> MxSink s

175+mxSink act msg = do

176+ msg' <- mxLift $ (unwrapMessage msg :: Process (Maybe m))

177+ case msg' of

178+ Nothing -> return Nothing

179+ Just m -> act m >>= return . Just

180+

181+data MxPipeline s =

182+ MxPipeline

183+ {

184+ current :: !(MxSink s)

185+ , next :: !(MxPipeline s)

186+ } | MxStop

187+

161-- | Activates a new agent.

188-- | Activates a new agent.

162--

189--

163mxAgent :: MxAgentId

190mxAgent :: MxAgentId

164 -> s

191 -> s

165− -> (Message -> MxAgent s MxAction)

192+ -> [MxSink s]

166 -> Process ProcessId

193 -> Process ProcessId

167−mxAgent mxId initState handler = do

194+mxAgent mxId initState handlers = do

168 node <- processNode <$> ask

195 node <- processNode <$> ask

169 pid <- liftIO $ mxNew (localEventBus node) $ start

196 pid <- liftIO $ mxNew (localEventBus node) $ start

170 return pid

197 return pid

171 where

198 where

172− start chan = do

199+ start (sendTChan, recvTChan) = do

173 (sp, rp) <- newChan

200 (sp, rp) <- newChan

174 nsend Table.mxTableCoordinator (MxAgentStart sp mxId)

201 nsend Table.mxTableCoordinator (MxAgentStart sp mxId)

175 tablePid <- receiveWait [ matchChan rp (\(p :: ProcessId) -> return p) ]

202 tablePid <- receiveWait [ matchChan rp (\(p :: ProcessId) -> return p) ]

176− runAgent handler chan $ MxAgentState mxId tablePid initState

203+ runAgent handlers recvTChan $ MxAgentState mxId sendTChan tablePid initState

177

204

178− runAgent h c s = do

205+ runAgent hs c s = do

179 msg <- (liftIO $ atomically $ readTChan c)

206 msg <- (liftIO $ atomically $ readTChan c)

180− (action, state) <- runAgentST s $ h msg

207+ (action, state) <- runPipeline msg s $ pipeline hs

181 case action of

208 case action of

182− MxAgentReady -> runAgent h c state

209+ MxAgentReady -> runAgent hs c state

183 MxAgentDeactivate _ -> {- TODO: log r -} return ()

210 MxAgentDeactivate _ -> {- TODO: log r -} return ()

184-- MxAgentBecome h' -> runAgent h' c state

211-- MxAgentBecome h' -> runAgent h' c state

185

212

213+ pipeline :: forall s . [MxSink s] -> MxPipeline s

214+ pipeline [] = MxStop

215+ pipeline (sink:sinks) = MxPipeline sink (pipeline sinks)

216+

217+ runPipeline :: forall s .

218+ Message

219+ -> MxAgentState s

220+ -> MxPipeline s

221+ -> Process (MxAction, MxAgentState s)

222+ runPipeline _ state MxStop = return (MxAgentReady, state)

223+ runPipeline msg state MxPipeline{..} = do

224+ let act = current msg

225+ (pass, state') <- ST.runStateT (unAgent act) state

226+ case pass of

227+ Nothing -> runPipeline msg state next

228+ Just result -> return (result, state')

229+

186 runAgentST :: MxAgentState s

230 runAgentST :: MxAgentState s

187 -> MxAgent s MxAction

231 -> MxAgent s MxAction

188 -> Process (MxAction, MxAgentState s)

232 -> Process (MxAction, MxAgentState s)

src/Control/Distributed/Process/Management/Agent.hs+12−7

@@ -46,16 +46,16 @@ import GHC.Weak (Weak, deRefWeak)

46-- Agent Controller Implementation --

47--------------------------------------------------------------------------------

49−

50-- | A triple containing a configured tracer, weak pointer to the

49-- | A triple containing a configured tracer, weak pointer to the

51-- agent controller's mailbox (CQueue) and an expression used to

50-- agent controller's mailbox (CQueue) and an expression used to

52-- instantiate new agents on the current node.

51-- instantiate new agents on the current node.

53type AgentConfig =

52type AgentConfig =

53+ (Tracer, Weak (CQueue Message),

54− (Tracer, Weak (CQueue Message), ((TChan Message -> Process ()) -> IO ProcessId))

54+ (((TChan Message, TChan Message) -> Process ()) -> IO ProcessId))

56-- | Starts a management agent for the current node. The agent process

57-- must not crash or be killed, so we generally avoid publishing its

58−-- 'ProcessId' where possible.

58+-- @ProcessId@ where possible.

59--

60-- Our process is also responsible for forwarding messages to the trace

61-- controller, since having two /special processes/ handled via the

@@ -76,9 +76,9 @@ mxAgentController forkProcess mv = do

76 sigbus <- liftIO $ newBroadcastTChanIO

77 weakQueue <- processWeakQ <$> ask

78 liftIO $ putMVar mv (trc, weakQueue, mxStartAgent forkProcess sigbus)

79− go sigbus trc forkProcess

79+ go sigbus trc

80 where

81− go bus tracer fork = forever' $ do

81+ go bus tracer = forever' $ do

82 void $ receiveWait [

83 -- This is exactly what it appears to be: a "catch all" handler.

84 -- Since mxNotify can potentially pass an unevaluated thunk to

@@ -115,8 +115,13 @@ mxAgentController forkProcess mv = do

115 liftIO $ atomically $ enqueueSTM q msg >> writeTChan ch msg

116

117-- | Forks a new process in which an mxAgent is run.

118−mxStartAgent :: Fork -> TChan Message -> (TChan Message -> Process ()) -> IO ProcessId

118+mxStartAgent :: Fork

119+ -> TChan Message

120+ -> ((TChan Message, TChan Message) -> Process ())

121+ -> IO ProcessId

119−mxStartAgent fork chan handler = (atomically (dupTChan chan)) >>= fork . handler

122+mxStartAgent fork chan handler = do

123+ chan' <- atomically (dupTChan chan)

124+ let proc = handler (chan, chan') in fork proc

120

125

121startTracing :: Fork -> IO Tracer

126startTracing :: Fork -> IO Tracer

122startTracing forkProcess = do

127startTracing forkProcess = do

src/Control/Distributed/Process/Management/Bus.hs+13−7

@@ -1,13 +1,16 @@

1-- | Interface to the management event bus.

2module Control.Distributed.Process.Management.Bus

3 ( publishEvent

4+ , enqueueEvent

4 ) where

5 ) where

6import Control.Distributed.Process.Internal.CQueue

7import Control.Distributed.Process.Internal.CQueue

8+ ( CQueue

7− ( enqueue

9+ , enqueue

8 )

10 )

9import Control.Distributed.Process.Internal.Types

11import Control.Distributed.Process.Internal.Types

12+ ( LocalNode(..)

10− ( MxEventBus(..)

13+ , MxEventBus(..)

11 , Message

14 , Message

12 , ProcessId

15 , ProcessId

13 , DiedReason

16 , DiedReason

@@ -19,11 +22,14 @@ import Network.Transport

19 ( ConnectionId

22 ( ConnectionId

20 , EndPointAddress

23 , EndPointAddress

21 )

24 )

22−import System.Mem.Weak (deRefWeak)

25+import System.Mem.Weak (Weak, deRefWeak)

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

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

25−publishEvent MxEventBusInitialising _ = return ()

28+publishEvent MxEventBusInitialising _ = return ()

26−publishEvent (MxEventBus _ _ wqRef _) msg = do

29+publishEvent (MxEventBus _ wqRef _) msg = enqueueEvent wqRef msg

30+

31+enqueueEvent :: Weak (CQueue Message) -> Message -> IO ()

32+enqueueEvent wqRef msg = do

27− mQueue <- deRefWeak wqRef

33+ mQueue <- deRefWeak wqRef

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

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

src/Control/Distributed/Process/Management/Table.hs+3−6

@@ -72,11 +72,8 @@ data MxTableRequest =

72 deriving (Typeable, Generic)

73instance Binary MxTableRequest where

75−data MxTableState = MxTableState

75+data MxTableState =

76− {

77− _name :: !String

78− , _entries :: !(Map String Message)

76+ MxTableState { _name :: !String, _entries :: !(Map String Message) }

79− }

81type MxTables = Map MxAgentId ProcessId

78type MxTables = Map MxAgentId ProcessId

@@ -187,7 +184,7 @@ startTableCoordinator fork = run Map.empty

187

184

188 spawnSup proc = do

185 spawnSup proc = do

189 us <- getSelfPid

186 us <- getSelfPid

190− -- we need to used that passed in "fork", in order to

187+ -- we need to use that passed in "fork", in order to

191 -- break an import cycle with Node.hs via the management

188 -- break an import cycle with Node.hs via the management

192 -- agent, management API and the tracing modules

189 -- agent, management API and the tracing modules

193 them <- liftIO $ fork $ link us >> proc

190 them <- liftIO $ fork $ link us >> proc

src/Control/Distributed/Process/Management/Trace/Primitives.hs+9−9

@@ -6,10 +6,8 @@

6-- This module is also used by the management agent, which relies on the

7-- tracing infrastructure's messaging fabric.

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

9− ( Tracer

10− , MxEvent(..)

11− -- * Sending Trace Data

9+ ( -- * Sending Trace Data

12− , traceLog

10+ traceLog

13 , traceLogFmt

11 , traceLogFmt

14 , traceMessage

12 , traceMessage

15 -- * Configuring A Tracer

13 -- * Configuring A Tracer

@@ -34,9 +32,11 @@ import Control.Distributed.Process.Internal.Primitives

34 , newChan

32 , newChan

35 , receiveChan

33 , receiveChan

36 )

34 )

35+import Control.Distributed.Process.Management.Types

36+ ( MxEvent(..)

37+ )

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

38import Control.Distributed.Process.Management.Trace.Types

38 ( TraceArg(..)

39 ( TraceArg(..)

39− , MxEvent(..)

40 , TraceFlags(..)

41 , TraceOk(..)

42 , TraceSubject(..)

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

60 , Process

61 , ProcessId

62 , LocalProcess(..)

63− , LocalNode(localEventBus)

63+ , LocalNode(localEventBus, localTracer)

64 , SendPort

65 , MxEventBus(..)

66 )

@@ -152,12 +152,12 @@ traceLogFmt d ls = withLocalTracer $ \t -> liftIO $ Tracer.traceLogFmt t d ls

152traceMessage :: Serializable m => m -> Process ()

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

154

155−withLocalTracer :: (MxEventBus -> Process ()) -> Process ()

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

156withLocalTracer act = do

157 node <- processNode <$> ask

158− act (localEventBus node)

158+ act (localTracer node)

159

160−withLocalTracerSync :: (MxEventBus -> SendPort TraceOk -> IO ()) -> Process ()

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

161withLocalTracerSync act = do

162 (sp, rp) <- newChan

163 withLocalTracer $ \t -> liftIO $ (act t sp)

src/Control/Distributed/Process/Management/Trace/Remote.hs+1−2

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

1module Control.Distributed.Process.Management.Trace.Remote

2− ( -- * Configuring A Tracer

2+ ( -- * Configuring A Remote Tracer

3 setTraceFlagsRemote

4 , startTraceRelay

5 -- * Remote Table

@@ -50,7 +50,6 @@ enableTraceRemote pid =

50startTraceRelay :: NodeId -> Process ProcessId

51startTraceRelay nodeId = do

52 withRegisteredTracer $ \pid ->

53− -- TODO: move cpEnableTraceRemote here?

54 spawn nodeId $ cpEnableTraceRemote pid

53 spawn nodeId $ cpEnableTraceRemote pid

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

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

src/Control/Distributed/Process/Management/Trace/Tracer.hs+4−2

@@ -31,10 +31,12 @@ import Control.Distributed.Process.Internal.Primitives

31 , handleMessage

32 , matchUnknown

33 )

34−import Control.Distributed.Process.Management.Trace.Types

34+import Control.Distributed.Process.Management.Types

35 ( MxEvent(..)

36− , SetTrace(..)

37 , Addressable(..)

36 , Addressable(..)

37+ )

38+import Control.Distributed.Process.Management.Trace.Types

39+ ( SetTrace(..)

38 , TraceSubject(..)

40 , TraceSubject(..)

39 , TraceFlags(..)

41 , TraceFlags(..)

40 , TraceOk(..)

42 , TraceOk(..)

src/Control/Distributed/Process/Management/Trace/Types.hs+47−153

@@ -1,9 +1,8 @@

1+{-# LANGUAGE DeriveGeneric #-}

1-- | Tracing/Debugging support - Types

3-- | Tracing/Debugging support - Types

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

4module Control.Distributed.Process.Management.Trace.Types

3− ( MxEventBus(..)

4− , MxEvent(..)

5− , SetTrace(..)

5+ ( SetTrace(..)

6− , Addressable(..)

7 , TraceSubject(..)

6 , TraceSubject(..)

8 , TraceFlags(..)

7 , TraceFlags(..)

9 , TraceArg(..)

8 , TraceArg(..)

@@ -27,6 +26,7 @@ import Control.Applicative ((<$>), (<*>))

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

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

28import Control.Distributed.Process.Internal.Types

27import Control.Distributed.Process.Internal.Types

29 ( MxEventBus(..)

28 ( MxEventBus(..)

29+ , Tracer(..)

30 , NodeId

31 , ProcessId

32 , DiedReason

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

35 , unsafeCreateUnencodedMessage

36 )

37import Control.Distributed.Process.Management.Bus

38− ( publishEvent

38+ ( enqueueEvent

39+ )

40+import Control.Distributed.Process.Management.Types

41+ ( MxEvent(..)

42+ , Addressable(..)

39 )

43 )

40import Control.Distributed.Process.Serializable

44import Control.Distributed.Process.Serializable

41import Data.Binary

45import Data.Binary

@@ -44,6 +48,7 @@ import Data.List (intersperse)

44import Data.Set (Set)

48import Data.Set (Set)

45import qualified Data.Set as Set (fromList)

49import qualified Data.Set as Set (fromList)

46import Data.Typeable

50import Data.Typeable

51+import GHC.Generics

47import System.Mem.Weak (deRefWeak)

52import System.Mem.Weak (deRefWeak)

48import Network.Transport

53import Network.Transport

49 ( ConnectionId

54 ( ConnectionId

@@ -55,51 +60,8 @@ import Network.Transport

55--------------------------------------------------------------------------------

60--------------------------------------------------------------------------------

57data SetTrace = TraceEnable !ProcessId | TraceDisable

62data SetTrace = TraceEnable !ProcessId | TraceDisable

58− deriving (Typeable, Eq, Show)

63+ deriving (Typeable, Generic, Eq, Show)

59−

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

61−

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

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

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

65−data MxEvent =

66− MxSpawned ProcessId

67− -- ^ fired whenever a local process is spawned

68− | MxRegistered ProcessId String

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

70− | MxUnRegistered ProcessId String

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

72− | MxProcessDied ProcessId DiedReason

73− -- ^ fired whenever a process dies

74− | MxNodeDied NodeId DiedReason

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

76− | MxSent ProcessId ProcessId Message

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

78− | MxReceived ProcessId Message

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

80− | MxConnected ConnectionId EndPointAddress

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

82− | MxDisconnected ConnectionId EndPointAddress

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

84− | MxUser Message

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

86− | MxLog String

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

88− | MxTraceTakeover ProcessId

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

90− | MxTraceDisable

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

92− deriving (Typeable, Show)

93−

94−class Addressable a where

95− resolveToPid :: a -> Maybe ProcessId

96−

97−instance Addressable MxEvent where

64+instance Binary SetTrace where

98− resolveToPid (MxSpawned p) = Just p

99− resolveToPid (MxProcessDied p _) = Just p

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

101− resolveToPid (MxReceived p _) = Just p

102− resolveToPid _ = Nothing

103

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

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

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

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

@@ -110,7 +72,8 @@ data TraceSubject =

110 TraceAll -- enable tracing for all running processes

72 TraceAll -- enable tracing for all running processes

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

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

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

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

113− deriving (Show)

75+ deriving (Typeable, Generic, Show)

76+instance Binary TraceSubject where

114

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

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

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

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

@@ -124,7 +87,8 @@ data TraceFlags = TraceFlags {

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

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

125 , traceNodes :: !Bool -- enable node status trace events

88 , traceNodes :: !Bool -- enable node status trace events

126 , traceConnections :: !Bool -- enable connection status trace events

89 , traceConnections :: !Bool -- enable connection status trace events

127− } deriving (Typeable, Show)

90+ } deriving (Typeable, Generic, Show)

91+instance Binary TraceFlags where

128

129defaultTraceFlags :: TraceFlags

93defaultTraceFlags :: TraceFlags

130defaultTraceFlags =

94defaultTraceFlags =

@@ -145,16 +109,17 @@ data TraceArg =

145

109

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

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

147data TraceOk = TraceOk

111data TraceOk = TraceOk

148− deriving (Typeable)

112+ deriving (Typeable, Generic)

113+instance Binary TraceOk where

149

114

150--------------------------------------------------------------------------------

115--------------------------------------------------------------------------------

151-- Internal/Common API --

116-- Internal/Common API --

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

117--------------------------------------------------------------------------------

153

118

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

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

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

120+traceLog (Tracer _ wqRef) s = enqueueEvent wqRef (unsafeCreateUnencodedMessage $ MxLog s)

156

121

157−traceLogFmt :: MxEventBus

122+traceLogFmt :: Tracer

158 -> String

123 -> String

159 -> [TraceArg]

124 -> [TraceArg]

160 -> IO ()

125 -> IO ()

@@ -164,43 +129,43 @@ traceLogFmt t d ls =

164 toS (TraceStr s) = s

129 toS (TraceStr s) = s

165 toS (Trace a) = show a

130 toS (Trace a) = show a

166

131

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

132+traceEvent :: Tracer -> MxEvent -> IO ()

168−traceEvent tr ev = publishEvent tr (unsafeCreateUnencodedMessage ev)

133+traceEvent (Tracer _ wqRef) ev = enqueueEvent wqRef (unsafeCreateUnencodedMessage ev)

169

134

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

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

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

136traceMessage tr msg = traceEvent tr (MxUser (unsafeCreateUnencodedMessage msg))

172

137

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

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

174−enableTrace t p =

139+enableTrace (Tracer _ wqRef) p =

175− publishEvent t (unsafeCreateUnencodedMessage ((Nothing :: Maybe (SendPort TraceOk)),

140+ enqueueEvent wqRef (unsafeCreateUnencodedMessage ((Nothing :: Maybe (SendPort TraceOk)),

176− (TraceEnable p)))

141+ (TraceEnable p)))

177

142

178−enableTraceSync :: MxEventBus -> SendPort TraceOk -> ProcessId -> IO ()

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

179−enableTraceSync t s p =

144+enableTraceSync (Tracer _ wqRef) s p =

180− publishEvent t (unsafeCreateUnencodedMessage (Just s, TraceEnable p))

145+ enqueueEvent wqRef (unsafeCreateUnencodedMessage (Just s, TraceEnable p))

181

146

182−disableTrace :: MxEventBus -> IO ()

147+disableTrace :: Tracer -> IO ()

183−disableTrace t =

148+disableTrace (Tracer _ wqRef) =

184− publishEvent t (unsafeCreateUnencodedMessage ((Nothing :: Maybe (SendPort TraceOk)),

149+ enqueueEvent wqRef (unsafeCreateUnencodedMessage ((Nothing :: Maybe (SendPort TraceOk)),

185 TraceDisable))

150 TraceDisable))

186

151

187−disableTraceSync :: MxEventBus -> SendPort TraceOk -> IO ()

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

188−disableTraceSync t s =

153+disableTraceSync (Tracer _ wqRef) s =

189− publishEvent t (unsafeCreateUnencodedMessage ((Just s), TraceDisable))

154+ enqueueEvent wqRef (unsafeCreateUnencodedMessage ((Just s), TraceDisable))

190

155

191−setTraceFlags :: MxEventBus -> TraceFlags -> IO ()

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

192−setTraceFlags t f =

157+setTraceFlags (Tracer _ wqRef) f =

193− publishEvent t (unsafeCreateUnencodedMessage ((Nothing :: Maybe (SendPort TraceOk)), f))

158+ enqueueEvent wqRef (unsafeCreateUnencodedMessage ((Nothing :: Maybe (SendPort TraceOk)), f))

194

159

195−setTraceFlagsSync :: MxEventBus -> SendPort TraceOk -> TraceFlags -> IO ()

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

196−setTraceFlagsSync t s f =

161+setTraceFlagsSync (Tracer _ wqRef) s f =

197− publishEvent t (unsafeCreateUnencodedMessage ((Just s), f))

162+ enqueueEvent wqRef (unsafeCreateUnencodedMessage ((Just s), f))

198

163

199−getTraceFlags :: MxEventBus -> SendPort TraceFlags -> IO ()

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

200−getTraceFlags t s = publishEvent t (unsafeCreateUnencodedMessage s)

165+getTraceFlags (Tracer _ wqRef) s = enqueueEvent wqRef (unsafeCreateUnencodedMessage s)

201

166

202−getCurrentTraceClient :: MxEventBus -> SendPort (Maybe ProcessId) -> IO ()

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

203−getCurrentTraceClient t s = publishEvent t (unsafeCreateUnencodedMessage s)

168+getCurrentTraceClient (Tracer _ wqRef) s = enqueueEvent wqRef (unsafeCreateUnencodedMessage s)

204

169

205class Traceable a where

170class Traceable a where

206 uod :: [a] -> TraceSubject

171 uod :: [a] -> TraceSubject

@@ -211,74 +176,3 @@ instance Traceable ProcessId where

211instance Traceable String where

176instance Traceable String where

212 uod = TraceNames . Set.fromList

177 uod = TraceNames . Set.fromList

213

178

214−--------------------------------------------------------------------------------

215−-- Binary Instances --

216−--------------------------------------------------------------------------------

217−

218−instance Binary SetTrace where

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

220− put TraceDisable = putWord8 2

221−

222− get = do

223− header <- getWord8

224− case header of

225− 1 -> TraceEnable <$> get

226− 2 -> return TraceDisable

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

228−

229−instance Binary MxEvent where

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

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

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

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

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

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

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

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

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

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

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

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

242− put MxTraceDisable = putWord8 13

243−

244− get = do

245− header <- getWord8

246− case header of

247− 1 -> MxSpawned <$> get

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

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

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

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

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

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

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

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

256− 10 -> MxUser <$> get

257− 11 -> MxLog <$> get

258− 12 -> MxTraceTakeover <$> get

259− 13 -> return MxTraceDisable

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

261−

262−instance Binary TraceSubject where

263− put TraceAll = putWord8 1

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

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

266−

267− get = do

268− header <- getWord8

269− case header of

270− 1 -> return TraceAll

271− 2 -> TraceProcs <$> get

272− 3 -> TraceNames <$> get

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

274−

275−instance Binary TraceFlags where

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

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

278− get =

279− TraceFlags <$> get <*> get <*> get <*> get <*> get <*> get <*> get <*> get

280−

281−instance Binary TraceOk where

282− put _ = return ()

283− get = return TraceOk

284−

src/Control/Distributed/Process/Management/Types.hs+108−3

@@ -4,16 +4,25 @@ module Control.Distributed.Process.Management.Types

4 , MxTableId(..)

5 , MxAgentState(..)

6 , MxAgent(..)

7+ , MxAction(..)

7 , MxAgentStart(..)

8 , MxAgentStart(..)

8 , Fork

9 , Fork

10+ , MxSink

11+ , MxEvent(..)

12+ , Addressable(..)

9 ) where

13 ) where

11−import Control.Applicative (Applicative)

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

16+import Control.Concurrent.STM

17+ ( TChan

18+ )

12import Control.Distributed.Process.Internal.Types

19import Control.Distributed.Process.Internal.Types

13 ( Process

20 ( Process

14 , ProcessId

21 , ProcessId

15 , Message

22 , Message

16 , SendPort

23 , SendPort

24+ , DiedReason

25+ , NodeId

17 )

26 )

18import Control.Monad.IO.Class (MonadIO)

27import Control.Monad.IO.Class (MonadIO)

19import qualified Control.Monad.State as ST

28import qualified Control.Monad.State as ST

@@ -23,10 +32,95 @@ import qualified Control.Monad.State as ST

23 , lift

32 , lift

24 , runStateT

33 , runStateT

25 )

34 )

26−import Data.Binary (Binary)

35+import Data.Binary

27import Data.Map.Strict (Map)

36import Data.Map.Strict (Map)

28import Data.Typeable (Typeable)

37import Data.Typeable (Typeable)

29import GHC.Generics

38import GHC.Generics

39+import Network.Transport

40+ ( ConnectionId

41+ , EndPointAddress

42+ )

43+

44+

45+-- TODO: consider factoring out MxEvent's tracing specific constructors

46+

47+-- | This is the /default/ management event, fired for various internal

48+-- events around the NT connection and Process lifecycle. All published

49+-- events that conform to this type, are eligible for tracing - i.e.,

50+-- they will be delivered to the trace controller.

51+--

52+data MxEvent =

53+ MxSpawned ProcessId

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

55+ | MxRegistered ProcessId String

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

57+ | MxUnRegistered ProcessId String

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

59+ | MxProcessDied ProcessId DiedReason

60+ -- ^ fired whenever a process dies

61+ | MxNodeDied NodeId DiedReason

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

63+ | MxSent ProcessId ProcessId Message

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

65+ | MxReceived ProcessId Message

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

67+ | MxConnected ConnectionId EndPointAddress

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

69+ | MxDisconnected ConnectionId EndPointAddress

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

71+ | MxUser Message

72+ -- ^ a user defined trace event

73+ | MxLog String

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

75+ | MxTraceTakeover ProcessId

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

77+ | MxTraceDisable

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

79+ deriving (Typeable, Generic, Show)

80+

81+instance Binary MxEvent where

82+

83+ -- put (MxSpawned pid) = putWord8 1 >> put pid

84+ -- put (MxRegistered pid str) = putWord8 2 >> put pid >> put str

85+ -- put (MxUnRegistered pid str) = putWord8 3 >> put pid >> put str

86+ -- put (MxProcessDied pid res) = putWord8 4 >> put pid >> put res

87+ -- put (MxNodeDied nid res) = putWord8 5 >> put nid >> put res

88+ -- put (MxSent to fro msg) = putWord8 6 >> put to >> put fro >> put msg

89+ -- put (MxReceived pid msg) = putWord8 7 >> put pid >> put msg

90+ -- put (MxConnected cid epid) = putWord8 8 >> put cid >> put epid

91+ -- put (MxDisconnected cid epid) = putWord8 9 >> put cid >> put epid

92+ -- put (MxUser msg) = putWord8 10 >> put msg

93+ -- put (MxLog msg) = putWord8 11 >> put msg

94+ -- put (MxTraceTakeover pid) = putWord8 12 >> put pid

95+ -- put MxTraceDisable = putWord8 13

96+

97+ -- get = do

98+ -- header <- getWord8

99+ -- case header of

100+ -- 1 -> MxSpawned <$> get

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

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

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

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

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

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

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

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

109+ -- 10 -> MxUser <$> get

110+ -- 11 -> MxLog <$> get

111+ -- 12 -> MxTraceTakeover <$> get

112+ -- 13 -> return MxTraceDisable

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

114+

115+class Addressable a where

116+ resolveToPid :: a -> Maybe ProcessId

117+

118+instance Addressable MxEvent where

119+ resolveToPid (MxSpawned p) = Just p

120+ resolveToPid (MxProcessDied p _) = Just p

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

122+ resolveToPid (MxReceived p _) = Just p

123+ resolveToPid _ = Nothing

124

31-- | Gross though it is, this synonym represents a function

125-- | Gross though it is, this synonym represents a function

32-- used to forking new processes, which has to be passed as a HOF

126-- used to forking new processes, which has to be passed as a HOF

@@ -41,15 +135,19 @@ newtype MxAgentId = MxAgentId { agentId :: String }

41data MxTableId =

135data MxTableId =

42 MxForAgent !MxAgentId

136 MxForAgent !MxAgentId

43 | MxForPid !ProcessId

137 | MxForPid !ProcessId

138+ deriving (Typeable, Generic)

139+instance Binary MxTableId where

140

45data MxAgentState s = MxAgentState

141data MxAgentState s = MxAgentState

46 {

142 {

47 mxAgentId :: !MxAgentId

143 mxAgentId :: !MxAgentId

48− -- , mxBus :: !(TChan Message)

144+ , mxBus :: !(TChan Message)

49 , mxSharedTable :: !ProcessId

145 , mxSharedTable :: !ProcessId

50 , mxLocalState :: !s

146 , mxLocalState :: !s

51 }

147 }

148

149+-- | Monad for management agents, encapsulting the

150+-- agent's state and private

53newtype MxAgent s a =

151newtype MxAgent s a =

54 MxAgent

152 MxAgent

55 {

153 {

@@ -70,3 +168,10 @@ data MxAgentStart = MxAgentStart

70 deriving (Typeable, Generic)

168 deriving (Typeable, Generic)

71instance Binary MxAgentStart where

169instance Binary MxAgentStart where

170

171+data MxAction =

172+ MxAgentDeactivate !String

173+ | MxAgentReady

174+

175+-- | Type of a management agent's event sink.

176+type MxSink s = Message -> MxAgent s (Maybe MxAction)

177+

src/Control/Distributed/Process/Node.hs+40−36

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

1+{-# LANGUAGE RecordWildCards #-}

1-- | Local nodes

3-- | Local nodes

2--

4--

3module Control.Distributed.Process.Node

5module Control.Distributed.Process.Node

@@ -145,22 +147,31 @@ import Control.Distributed.Process.Internal.Types

145 , WhereIsReply(..)

147 , WhereIsReply(..)

146 , payloadToMessage

148 , payloadToMessage

147 , createUnencodedMessage

149 , createUnencodedMessage

150+ , unsafeCreateUnencodedMessage

148 , runLocalProcess

151 , runLocalProcess

149 , firstNonReservedProcessId

152 , firstNonReservedProcessId

150 , ImplicitReconnect(WithImplicitReconnect)

153 , ImplicitReconnect(WithImplicitReconnect)

151 )

154 )

155+import Control.Distributed.Process.Management.Agent

156+ ( mxAgentController

157+ )

158+import qualified Control.Distributed.Process.Management.Bus as Mx

159+ ( publishEvent

160+ )

152import qualified Control.Distributed.Process.Management.Trace.Remote as Trace

161import qualified Control.Distributed.Process.Management.Trace.Remote as Trace

153 ( remoteTable

162 ( remoteTable

154 )

163 )

164+import Control.Distributed.Process.Management.Trace.Tracer

165+ ( defaultTracer

166+ )

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

167import Control.Distributed.Process.Management.Trace.Types

156 ( TraceArg(..)

168 ( TraceArg(..)

157− , MxEvent(..)

158 , traceEvent

169 , traceEvent

159 , traceLogFmt

170 , traceLogFmt

160 , enableTrace

171 , enableTrace

161 )

172 )

162−import Control.Distributed.Process.Management.Trace.Tracer

173+import Control.Distributed.Process.Management.Types

163− ( defaultTracer

174+ ( MxEvent(..)

164 )

175 )

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

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

166import Control.Distributed.Process.Internal.Messaging

177import Control.Distributed.Process.Internal.Messaging

@@ -185,9 +196,6 @@ import qualified Control.Distributed.Process.Internal.StrictContainerAccessors a

185 ( mapMaybe

196 ( mapMaybe

186 , mapDefault

197 , mapDefault

187 )

198 )

188−import Control.Distributed.Process.Management.Agent

189− ( mxAgentController

190− )

191import Unsafe.Coerce

199import Unsafe.Coerce

192

200

193--------------------------------------------------------------------------------

201--------------------------------------------------------------------------------

@@ -251,18 +259,17 @@ startMxAgent node = do

251 mv <- MVar.newEmptyMVar

259 mv <- MVar.newEmptyMVar

252 pid <- fork $ mxAgentController fork mv

260 pid <- fork $ mxAgentController fork mv

253 (tracer', wqRef, mxNew') <- MVar.takeMVar mv

261 (tracer', wqRef, mxNew') <- MVar.takeMVar mv

254− return node { localEventBus = (MxEventBus pid tracer' wqRef mxNew') }

262+ return $ node { localEventBus = (MxEventBus pid wqRef mxNew')

263+ , localTracer = tracer'

264+ }

255

265

256startDefaultTracer :: LocalNode -> IO ()

266startDefaultTracer :: LocalNode -> IO ()

257startDefaultTracer node' = do

267startDefaultTracer node' = do

258− let t = localEventBus node'

268+ let (Tracer pid _) = localTracer node'

259− case t of

260− MxEventBus _ (Tracer pid _) _ _ -> do

261− runProcess node' $ register "trace.controller" pid

269+ runProcess node' $ register "trace.controller" pid

262− pid' <- forkProcess node' defaultTracer

270+ pid' <- forkProcess node' defaultTracer

263− enableTrace (localEventBus node') pid'

271+ enableTrace (localTracer node') pid'

264− runProcess node' $ register "tracer.initial" pid'

272+ runProcess node' $ register "tracer.initial" pid'

265− _ -> return ()

266

273

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

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

268startServiceProcesses :: LocalNode -> IO ()

275startServiceProcesses :: LocalNode -> IO ()

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

353 return (tid', lproc)

360 return (tid', lproc)

354

361

355 -- see note [tracer/forkProcess races]

362 -- see note [tracer/forkProcess races]

356− trace node (MxSpawned pid)

363+ mxNotify node (MxSpawned pid)

357

364

358 if lpidCounter lpid == maxBound

365 if lpidCounter lpid == maxBound

359 then do

366 then do

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

435 NT.ConnectionOpened cid rel theirAddr ->

442 NT.ConnectionOpened cid rel theirAddr ->

436 if rel == NT.ReliableOrdered

443 if rel == NT.ReliableOrdered

437 then

444 then

438− trace node (MxConnected cid theirAddr)

445+ mxNotify node (MxConnected cid theirAddr)

439 >> go (

446 >> go (

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

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

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

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

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

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

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

452 let msg = payloadToMessage payload

459 let msg = payloadToMessage payload

453 enqueue queue msg -- 'enqueue' is strict

460 enqueue queue msg -- 'enqueue' is strict

454− trace node (MxReceived pid msg)

461+ mxNotify node (MxReceived pid msg)

455 go st

462 go st

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

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

457 mChan <- deRefWeak chan'

464 mChan <- deRefWeak chan'

@@ -495,12 +502,13 @@ handleIncomingMessages node = go initConnectionState

495 else invalidRequest cid st

502 else invalidRequest cid st

496 Nothing ->

503 Nothing ->

497 invalidRequest cid st

504 invalidRequest cid st

498− NT.ConnectionClosed cid ->

505+ NT.ConnectionClosed cid -> do

506+ putStrLn "connection closed..."

499 case st ^. incomingAt cid of

507 case st ^. incomingAt cid of

500 Nothing ->

508 Nothing ->

501 invalidRequest cid st

509 invalidRequest cid st

502 Just (src, _) -> do

510 Just (src, _) -> do

503− trace node (MxDisconnected cid src)

511+ mxNotify node (MxDisconnected cid src)

504 go ( (incomingAt cid ^= Nothing)

512 go ( (incomingAt cid ^= Nothing)

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

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

506 $ st

514 $ st

@@ -596,25 +604,21 @@ instance Show ProcessKillException where

596

604

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

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

598traceNotifyDied node ident reason =

606traceNotifyDied node ident reason =

599− -- TODO: sendPortDied notifications

607+ -- TODO: SendPortDied notifications

600− liftIO $ withLocalTracer node $ \t ->

601− case ident of

608+ case ident of

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

609+ (NodeIdentifier nid) -> liftIO $ mxNotify node (MxNodeDied nid reason)

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

610+ (ProcessIdentifier pid) -> liftIO $ mxNotify node (MxProcessDied pid reason)

604− _ -> return ()

611+ _ -> return ()

605

612

606traceEventFmtIO :: LocalNode

613traceEventFmtIO :: LocalNode

607 -> String

614 -> String

608 -> [TraceArg]

615 -> [TraceArg]

609 -> IO ()

616 -> IO ()

610−traceEventFmtIO node fmt args =

617+traceEventFmtIO LocalNode{..} fmt args = traceLogFmt localTracer fmt args

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

612−

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

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

615

618

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

619+mxNotify :: LocalNode -> MxEvent -> IO ()

617−withLocalTracer node act = act (localEventBus node)

620+mxNotify LocalNode{..} ev =

621+ Mx.publishEvent localEventBus $ unsafeCreateUnencodedMessage ev

618

622

619--------------------------------------------------------------------------------

623--------------------------------------------------------------------------------

620-- Core functionality --

624-- Core functionality --

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

813 do modify' $ registeredHereFor label ^= mPid

817 do modify' $ registeredHereFor label ^= mPid

814 updateRemote node currentVal mPid

818 updateRemote node currentVal mPid

815 case mPid of

819 case mPid of

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

820+ (Just p) -> liftIO $ mxNotify node (MxRegistered p label)

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

821+ Nothing -> liftIO $ mxNotify node (MxUnRegistered (fromJust currentVal) label)

818 liftIO $ sendMessage node

822 liftIO $ sendMessage node

819 (NodeIdentifier (localNodeId node))

823 (NodeIdentifier (localNodeId node))

820 (ProcessIdentifier from)

824 (ProcessIdentifier from)

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

882ncEffectLocalSend node to msg =

886ncEffectLocalSend node to msg =

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

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

884 enqueue (processQueue p) msg

888 enqueue (processQueue p) msg

885− trace node (MxReceived to msg)

889+ mxNotify node (MxReceived to msg)

886

890

887-- [Issue #DP-20]

891-- [Issue #DP-20]

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

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

tests/TestTracing.hs+3

@@ -12,6 +12,9 @@ import Control.Concurrent.MVar

12 )

13import Control.Distributed.Process

14import Control.Distributed.Process.Debug

15+import Control.Distributed.Process.Management

16+ ( MxEvent(..)

17+ )

15import Control.Distributed.Process.Node

18import Control.Distributed.Process.Node

16 ( forkProcess

19 ( forkProcess

17 , newLocalNode

20 , newLocalNode