fad3a08 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Merge branch 'mx' into development

fad3a08Tim Watson13 years ago

.gitignore+1

compare with master

unifiedsplit

@@ -11,4 +11,5 @@ dist/

11*.ps

12*.swp

13cabal-dev

14+cabal-dev*

14.DS_Store

15.DS_Store

distributed-process.cabal+48−7

@@ -64,6 +64,7 @@ Library

64 Control.Distributed.Process.Closure,

65 Control.Distributed.Process.Debug,

66 Control.Distributed.Process.Internal.Types,

67+ Control.Distributed.Process.Management,

67 Control.Distributed.Process.Node,

68 Control.Distributed.Process.Node,

68 Control.Distributed.Process.Serializable,

69 Control.Distributed.Process.Serializable,

69 Control.Distributed.Process.UnsafePrimitives

70 Control.Distributed.Process.UnsafePrimitives

@@ -77,10 +78,14 @@ Library

77 Control.Distributed.Process.Internal.StrictMVar,

78 Control.Distributed.Process.Internal.StrictMVar,

78 Control.Distributed.Process.Internal.WeakTQueue,

79 Control.Distributed.Process.Internal.WeakTQueue,

79 Control.Distributed.Process.Internal.StrictContainerAccessors,

80 Control.Distributed.Process.Internal.StrictContainerAccessors,

80− Control.Distributed.Process.Internal.Trace.Types,

81+ Control.Distributed.Process.Management.Agent,

82+ Control.Distributed.Process.Management.Bus,

83+ Control.Distributed.Process.Management.Table,

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

81− Control.Distributed.Process.Internal.Trace.Primitives,

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

82− Control.Distributed.Process.Internal.Trace.Tracer,

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

83− Control.Distributed.Process.Internal.Trace.Remote

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

88+ Control.Distributed.Process.Management.Trace.Tracer

84 Extensions: RankNTypes,

89 Extensions: RankNTypes,

85 ScopedTypeVariables,

90 ScopedTypeVariables,

86 FlexibleInstances,

91 FlexibleInstances,

@@ -163,6 +168,37 @@ Test-Suite TestStats

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

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

164 HS-Source-Dirs: tests

169 HS-Source-Dirs: tests

165

170

171+

172+Test-Suite TestMx

173+ Type: exitcode-stdio-1.0

174+ Main-Is: TestMx.hs

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

176+ random >= 1.0 && < 1.1,

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

178+ containers >= 0.4 && < 0.6,

179+ stm >= 2.4 && < 2.5,

180+ distributed-process,

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

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

183+ binary >= 0.5 && < 0.7,

184+ network >= 2.3 && < 2.5,

185+ HUnit >= 1.2 && < 1.3,

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

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

188+ unix >= 2.5.0.0

189+ Extensions: RankNTypes,

190+ ScopedTypeVariables,

191+ FlexibleInstances,

192+ UndecidableInstances,

193+ ExistentialQuantification,

194+ GADTs,

195+ GeneralizedNewtypeDeriving,

196+ DeriveDataTypeable,

197+ CPP,

198+ BangPatterns

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

200+ HS-Source-Dirs: tests

201+

166Test-Suite TestTracing

202Test-Suite TestTracing

167 Type: exitcode-stdio-1.0

203 Type: exitcode-stdio-1.0

168 Main-Is: TestTracing.hs

204 Main-Is: TestTracing.hs

@@ -180,14 +216,19 @@ Test-Suite TestTracing

180 test-framework >= 0.6 && < 0.9,

216 test-framework >= 0.6 && < 0.9,

181 test-framework-hunit >= 0.2.0 && < 0.4,

217 test-framework-hunit >= 0.2.0 && < 0.4,

182 unix >= 2.5.0.0

218 unix >= 2.5.0.0

183− Extensions: CPP,

219+ Extensions: RankNTypes,

184 ScopedTypeVariables,

220 ScopedTypeVariables,

221+ FlexibleInstances,

222+ UndecidableInstances,

223+ ExistentialQuantification,

224+ GADTs,

225+ GeneralizedNewtypeDeriving,

185 DeriveDataTypeable,

226 DeriveDataTypeable,

186− GeneralizedNewtypeDeriving

227+ CPP,

228+ BangPatterns

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

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

188 HS-Source-Dirs: tests

230 HS-Source-Dirs: tests

189

231

190−

191Executable distributed-process-throughput

232Executable distributed-process-throughput

192 if flag(benchmarks)

233 if flag(benchmarks)

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

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

src/Control/Distributed/Process/Debug.hs+40−43

@@ -17,16 +17,14 @@

17--

18-- [Enabling Tracing]

19--

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

29−-- published.

30--

28--

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

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

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

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

@@ -37,15 +35,15 @@

37--

35--

38-- [Working with multiple tracer processes]

36-- [Working with multiple tracer processes]

39--

37--

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

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

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

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

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

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

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

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

44--

42--

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

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

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

44+-- supplied handler and also forwards trace events to the original tracer.

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

45-- The corresponding 'stopTracer' function terminates tracer processes in

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

46+-- reverse of the order in which they were started, and re-registers the

49-- previous tracer process.

47-- previous tracer process.

50--

48--

51-- [Built in tracers]

49-- [Built in tracers]

@@ -53,22 +51,21 @@

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

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

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

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

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

53-- can be configured using environment variables, or specified manually using

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

54+-- the 'traceEnable' function.

57--

55--

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

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

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

57+-- variables are checked to determine which default tracer process is selected.

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

58+-- If none of these variables is set, a no-op tracer process is installed,

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

59+-- which effectively ignores all trace messages. Note that in this case,

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

60+-- trace events are still generated and passed through the system.

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

61+-- Only one default tracer will be chosen - the first that contains a (valid)

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

62+-- value. These environment variables, in the order they're examined, are:

65--

63--

66-- 1. @DISTRIBUTED_PROCESS_TRACE_FILE@

64-- 1. @DISTRIBUTED_PROCESS_TRACE_FILE@

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

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

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

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

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

67+-- path is invalid, or the file is unavailable for writing, this tracer will not

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

68+-- be selected.

71−-- disabled.

72--

69--

73-- 2. @DISTRIBUTED_PROCESS_TRACE_CONSOLE@

70-- 2. @DISTRIBUTED_PROCESS_TRACE_CONSOLE@

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

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

@@ -81,8 +78,10 @@

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

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

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

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

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

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

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

81+-- will prevent slaves from tracing to the file. They will need to fall back to

85−-- the console or eventlog tracers instead.

82+-- the console or eventlog tracers instead, which can be accomplished by setting

83+-- one of these environment variables /as well/, since the latter will only be

84+-- selected on slaves (when the file tracer selection fails).

86--

85--

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

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

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

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

@@ -93,11 +92,9 @@

93module Control.Distributed.Process.Debug

92module Control.Distributed.Process.Debug

94 ( -- * Exported Data Types

93 ( -- * Exported Data Types

95 TraceArg(..)

94 TraceArg(..)

96− , TraceEvent(..)

97 , TraceFlags(..)

95 , TraceFlags(..)

98 , TraceSubject(..)

96 , TraceSubject(..)

99 -- * Configuring Tracing

97 -- * Configuring Tracing

100− , isTracingEnabled

101 , enableTrace

98 , enableTrace

102 , enableTraceAsync

99 , enableTraceAsync

103 , disableTrace

100 , disableTrace

@@ -147,19 +144,21 @@ import Control.Distributed.Process.Internal.Types

147 , LocalProcess(..)

144 , LocalProcess(..)

148 , ProcessMonitorNotification(..)

145 , ProcessMonitorNotification(..)

149 )

146 )

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

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

148+ ( MxEvent(..)

149+ )

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

151 ( TraceArg(..)

152− , TraceEvent(..)

153 , TraceFlags(..)

152 , TraceFlags(..)

154 , TraceSubject(..)

153 , TraceSubject(..)

155 , defaultTraceFlags

154 , defaultTraceFlags

156 )

155 )

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

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

158 ( systemLoggerTracer

157 ( systemLoggerTracer

159 , logfileTracer

158 , logfileTracer

160 , eventLogTracer

159 , eventLogTracer

161 )

160 )

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

161+import Control.Distributed.Process.Management.Trace.Primitives

163 ( withRegisteredTracer

162 ( withRegisteredTracer

164 , enableTrace

163 , enableTrace

165 , enableTraceAsync

164 , enableTraceAsync

@@ -167,7 +166,6 @@ import Control.Distributed.Process.Internal.Trace.Primitives

167 , setTraceFlags

166 , setTraceFlags

168 , setTraceFlagsAsync

167 , setTraceFlagsAsync

169 , getTraceFlags

168 , getTraceFlags

170− , isTracingEnabled

171 , traceOn

169 , traceOn

172 , traceOff

170 , traceOff

173 , traceOnly

171 , traceOnly

@@ -175,7 +173,7 @@ import Control.Distributed.Process.Internal.Trace.Primitives

175 , traceLogFmt

173 , traceLogFmt

176 , traceMessage

174 , traceMessage

177 )

175 )

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

176+import qualified Control.Distributed.Process.Management.Trace.Remote as Remote

179import Control.Distributed.Process.Node

177import Control.Distributed.Process.Node

180

178

181import Control.Exception (SomeException)

179import Control.Exception (SomeException)

@@ -200,7 +198,7 @@ import Prelude hiding (catch)

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

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

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

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

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

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

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

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

204startTracer handler = do

202startTracer handler = do

205 withRegisteredTracer $ \pid -> do

203 withRegisteredTracer $ \pid -> do

206 node <- processNode <$> ask

204 node <- processNode <$> ask

@@ -212,7 +210,7 @@ startTracer handler = do

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

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

213-- in case of failure).

211-- in case of failure).

214withTracer :: forall a.

212withTracer :: forall a.

215− (TraceEvent -> Process ())

213+ (MxEvent -> Process ())

216 -> Process a

214 -> Process a

217 -> Process (Either SomeException a)

215 -> Process (Either SomeException a)

218withTracer handler proc = do

216withTracer handler proc = do

@@ -227,7 +225,7 @@ withTracer handler proc = do

227 Nothing -> return ()

225 Nothing -> return ()

228 Just _ -> do

226 Just _ -> do

229 ref <- monitor tracer

227 ref <- monitor tracer

230− send tracer TraceEvDisable

228+ send tracer MxTraceDisable

231 receiveWait [

229 receiveWait [

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

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

233 (\_ -> return ())

231 (\_ -> return ())

@@ -244,12 +242,12 @@ withFlags flags proc = do

244 finally (setTraceFlags flags >> try proc)

242 finally (setTraceFlags flags >> try proc)

245 (setTraceFlags oldFlags)

243 (setTraceFlags oldFlags)

246

244

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

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

248traceProxy pid act = do

246traceProxy pid act = do

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

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

250 case ev of

248 case ev of

251− (TraceEvTakeover _) -> return False

249+ (MxTraceTakeover _) -> return False

252− TraceEvDisable -> die "disabled"

250+ MxTraceDisable -> die "disabled"

253 _ -> act ev >> return True

251 _ -> act ev >> return True

254

252

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

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

@@ -264,8 +262,7 @@ traceProxy pid act = do

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

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

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

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

266--

264--

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

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

265+-- If the last tracer to have been registered was not started

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

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

270stopTracer :: Process ()

267stopTracer :: Process ()

271stopTracer =

268stopTracer =

@@ -276,5 +273,5 @@ stopTracer =

276 basePid <- whereis "tracer.initial"

273 basePid <- whereis "tracer.initial"

277 case basePid == (Just pid) of

274 case basePid == (Just pid) of

278 True -> return ()

275 True -> return ()

279− False -> send pid TraceEvDisable

276+ False -> send pid MxTraceDisable

280

277

src/Control/Distributed/Process/Internal/CQueue.hs+6−1

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

6 , MatchOn(..)

7 , newCQueue

8 , enqueue

9+ , enqueueSTM

9 , dequeue

10 , dequeue

10 , mkWeakCQueue

11 , mkWeakCQueue

11 ) where

12 ) where

@@ -52,7 +53,11 @@ newCQueue = CQueue <$> newMVar Nil <*> atomically newTChan

52--

53--

53-- Enqueue is strict.

54-- Enqueue is strict.

54enqueue :: CQueue a -> a -> IO ()

55enqueue :: CQueue a -> a -> IO ()

55−enqueue (CQueue _arrived incoming) !a = atomically $ writeTChan incoming a

56+enqueue (CQueue _arrived incoming) !a = atomically $ writeTChan incoming a

57+

58+-- | Variant of enqueue for use in the STM monad.

59+enqueueSTM :: CQueue a -> a -> STM ()

60+enqueueSTM (CQueue _arrived incoming) !a = writeTChan incoming a

57data BlockSpec =

62data BlockSpec =

58 NonBlocking

63 NonBlocking

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

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

191 , disconnect

192 , sendCtrlMsg

193 )

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

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

195+ ( MxEvent(..)

196+ )

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

195 ( traceEvent

198 ( traceEvent

196− , TraceEvent(..)

197 )

199 )

198import Control.Distributed.Process.Internal.WeakTQueue

200import Control.Distributed.Process.Internal.WeakTQueue

199 ( newTQueueIO

201 ( newTQueueIO

@@ -226,8 +228,8 @@ send them msg = do

226 msg

228 msg

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

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

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

229− liftIO $ traceEvent (localTracer node)

231+ liftIO $ traceEvent (localEventBus node)

230− (TraceEvSent them us (createUnencodedMessage msg))

232+ (MxSent them us (createUnencodedMessage msg))

231

233

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

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

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

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

410 (messageToPayload msg)

412 (messageToPayload msg)

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

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

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

413− liftIO $ traceEvent (localTracer node)

415+ liftIO $ traceEvent (localEventBus node)

414− (TraceEvSent them us msg)

416+ (MxSent them us msg)

415

417

416

418

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

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

@@ -551,7 +553,7 @@ delegate pid p = do

551 ]

553 ]

552 delegate pid p

554 delegate pid p

553

555

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

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

555relay :: ProcessId -> Process ()

557relay :: ProcessId -> Process ()

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

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

557

559

src/Control/Distributed/Process/Internal/Trace/Types.hsdeleted−284

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

1−-- | Tracing/Debugging support - Types

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

3− ( Tracer(..)

4− , TraceEvent(..)

5− , SetTrace(..)

6− , Addressable(..)

7− , TraceSubject(..)

8− , TraceFlags(..)

9− , TraceArg(..)

10− , TraceOk(..)

11− , traceLog

12− , traceLogFmt

13− , traceEvent

14− , traceMessage

15− , defaultTraceFlags

16− , enableTrace

17− , enableTraceSync

18− , disableTrace

19− , disableTraceSync

20− , getTraceFlags

21− , setTraceFlags

22− , setTraceFlagsSync

23− , getCurrentTraceClient

24− ) where

25−

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

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

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

29− ( Tracer(..)

30− , NodeId

31− , ProcessId

32− , DiedReason

33− , Message

34− , SendPort

35− , createUnencodedMessage

36− )

37−import Control.Distributed.Process.Serializable

38−import Data.Binary

39−import Data.Foldable (forM_)

40−import Data.List (intersperse)

41−import Data.Set (Set)

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

43−import Data.Typeable

44−import System.Mem.Weak (deRefWeak)

45−import Network.Transport

46− ( ConnectionId

47− )

48−

49−--------------------------------------------------------------------------------

50−-- Types --

51−--------------------------------------------------------------------------------

52−

53−data SetTrace = TraceEnable !ProcessId | TraceDisable

54− deriving (Typeable, Eq, Show)

55−

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

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

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

59−data TraceEvent =

60− TraceEvSpawned ProcessId

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

62− | TraceEvRegistered ProcessId String

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

64− | TraceEvUnRegistered ProcessId String

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

66− | TraceEvDied ProcessId DiedReason

67− -- ^ fired whenever a process dies

68− | TraceEvNodeDied NodeId DiedReason

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

70− | TraceEvSent ProcessId ProcessId Message

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

72− | TraceEvReceived ProcessId Message

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

74− | TraceEvConnected ConnectionId

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

76− | TraceEvDisconnected ConnectionId

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

78− | TraceEvUser Message

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

80− | TraceEvLog String

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

82− | TraceEvTakeover ProcessId

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

84− | TraceEvDisable

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

86− deriving (Typeable, Show)

87−

88−class Addressable a where

89− resolveToPid :: a -> Maybe ProcessId

90−

91−instance Addressable TraceEvent where

92− resolveToPid (TraceEvSpawned p) = Just p

93− resolveToPid (TraceEvDied p _) = Just p

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

95− resolveToPid (TraceEvReceived p _) = Just p

96− resolveToPid _ = Nothing

97−

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

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

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

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

102−--

103−data TraceSubject =

104− TraceAll -- enable tracing for all running processes

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

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

107− deriving (Show)

108−

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

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

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

112−data TraceFlags = TraceFlags {

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

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

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

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

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

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

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

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

121− } deriving (Typeable, Show)

122−

123−defaultTraceFlags :: TraceFlags

124−defaultTraceFlags =

125− TraceFlags {

126− traceSpawned = Nothing

127− , traceDied = Nothing

128− , traceRegistered = Nothing

129− , traceUnregistered = Nothing

130− , traceSend = Nothing

131− , traceRecv = Nothing

132− , traceNodes = False

133− , traceConnections = False

134− }

135−

136−data TraceArg =

137− TraceStr String

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

139−

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

141−data TraceOk = TraceOk

142− deriving (Typeable)

143−

144−--------------------------------------------------------------------------------

145−-- Internal/Common API --

146−--------------------------------------------------------------------------------

147−

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

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

150−

151−traceLogFmt :: Tracer

152− -> String

153− -> [TraceArg]

154− -> IO ()

155−traceLogFmt t d ls =

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

157− where toS :: TraceArg -> String

158− toS (TraceStr s) = s

159− toS (Trace a) = show a

160−

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

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

163−

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

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

166−

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

168−enableTrace t p =

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

170− (TraceEnable p)))

171−

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

173−enableTraceSync t s p =

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

175−

176−disableTrace :: Tracer -> IO ()

177−disableTrace t =

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

179− TraceDisable))

180−

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

182−disableTraceSync t s =

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

184−

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

186−setTraceFlags t f =

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

188−

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

190−setTraceFlagsSync t s f =

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

192−

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

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

195−

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

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

198−

199−class Traceable a where

200− uod :: [a] -> TraceSubject

201−

202−instance Traceable ProcessId where

203− uod = TraceProcs . Set.fromList

204−

205−instance Traceable String where

206− uod = TraceNames . Set.fromList

207−

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

209−traceIt InactiveTracer _ = return ()

210−traceIt (ActiveTracer _ wqRef) msg = do

211− mQueue <- deRefWeak wqRef

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

213−

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

215−-- Binary Instances --

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

217−

218−instance Binary SetTrace where

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

220− put TraceDisable = putWord8 2

221−

222− get = do

223− header <- getWord8

224− case header of

225− 1 -> TraceEnable <$> get

226− 2 -> return TraceDisable

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

228−

229−instance Binary TraceEvent where

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

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

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

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

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

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

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

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

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

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

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

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

242− put TraceEvDisable = putWord8 13

243−

244− get = do

245− header <- getWord8

246− case header of

247− 1 -> TraceEvSpawned <$> get

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

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

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

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

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

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

254− 8 -> TraceEvConnected <$> get

255− 9 -> TraceEvDisconnected <$> get

256− 10 -> TraceEvUser <$> get

257− 11 -> TraceEvLog <$> get

258− 12 -> TraceEvTakeover <$> get

259− 13 -> return TraceEvDisable

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

261−

262−instance Binary TraceSubject where

263− put TraceAll = putWord8 1

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

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

266−

267− get = do

268− header <- getWord8

269− case header of

270− 1 -> return TraceAll

271− 2 -> TraceProcs <$> get

272− 3 -> TraceNames <$> get

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

274−

275−instance Binary TraceFlags where

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

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

278− get =

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

280−

281−instance Binary TraceOk where

282− put _ = return ()

283− get = return TraceOk

284−

src/Control/Distributed/Process/Internal/Types.hs+28−18

@@ -17,6 +17,7 @@ module Control.Distributed.Process.Internal.Types

17 -- * Local nodes and processes

18 , LocalNode(..)

19 , Tracer(..)

20+ , MxEventBus(..)

20 , LocalNodeState(..)

21 , LocalNodeState(..)

21 , LocalProcess(..)

22 , LocalProcess(..)

22 , LocalProcessState(..)

23 , LocalProcessState(..)

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

98import Control.Concurrent (ThreadId)

99import Control.Concurrent (ThreadId)

99import Control.Concurrent.Chan (Chan)

100import Control.Concurrent.Chan (Chan)

100import Control.Concurrent.STM (STM)

101import Control.Concurrent.STM (STM)

102+import Control.Concurrent.STM.TChan (TChan)

101import qualified Network.Transport as NT (EndPoint, EndPointAddress, Connection)

103import qualified Network.Transport as NT (EndPoint, EndPointAddress, Connection)

102import Control.Applicative (Applicative, Alternative, (<$>), (<*>))

104import Control.Applicative (Applicative, Alternative, (<$>), (<*>))

103import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)

105import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)

@@ -198,22 +200,30 @@ nullProcessId nid =

198-- Local nodes and processes --

200-- Local nodes and processes --

199--------------------------------------------------------------------------------

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

200

202

201−-- | Required for system tracing in the node controller

203+-- | Provides access to the trace controller

202−data Tracer =

204+data Tracer = Tracer

203− ActiveTracer !ProcessId !(Weak (CQueue Message)) -- note [tracer longevity]

205+ {

204− | InactiveTracer

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

205−

207+ tracerPid :: !ProcessId

206−-- note [tracer longevity]

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

207−--

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

208−-- If the trace controller process crashes, we make no attempt to restart it

210+ }

209−-- or handle the failure. Tracing is done on a 'best effort' basis, because

211+

210−-- to make stronger guarantees (e.g., handling failures) would introduce

212+-- | Local system management event bus state

211−-- significant complexity and adversely affect performance. As such, we access

213+data MxEventBus =

212−-- the process' mailbox directly via a weak reference, and calls to a dead

214+ MxEventBusInitialising

213−-- process have no effect just as with normal process/send interactions.

215+ | MxEventBus

214−--

216+ {

215−-- Also note that the ProcessId stored in an active tracer is that of the first

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

216−-- (initially registered) system tracer process.

218+ agent :: !ProcessId

219+ -- | Configuration for the local trace controller

220+ , tracer :: !Tracer

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

222+ , evbuss :: !(Weak (CQueue Message))

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

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

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

226+ }

217

227

218-- | Local nodes

228-- | Local nodes

219data LocalNode = LocalNode

229data LocalNode = LocalNode

@@ -225,8 +235,8 @@ data LocalNode = LocalNode

225 , localState :: !(StrictMVar LocalNodeState)

235 , localState :: !(StrictMVar LocalNodeState)

226 -- | Channel for the node controller

236 -- | Channel for the node controller

227 , localCtrlChan :: !(Chan NCMsg)

237 , localCtrlChan :: !(Chan NCMsg)

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

238+ -- | Internal management event bus

229− , localTracer :: !Tracer

239+ , localEventBus :: !MxEventBus

230 -- | Runtime lookup table for supporting closures

240 -- | Runtime lookup table for supporting closures

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

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

232 , remoteTable :: !RemoteTable

242 , remoteTable :: !RemoteTable

src/Control/Distributed/Process/Management.hsnew file+249

@@ -0,0 +1,249 @@

1+{-# LANGUAGE DeriveDataTypeable #-}

2+{-# LANGUAGE DeriveGeneric #-}

3+{-# LANGUAGE StandaloneDeriving #-}

4+{-# LANGUAGE ExistentialQuantification #-}

5+{-# LANGUAGE ScopedTypeVariables #-}

6+{-# LANGUAGE RecordWildCards #-}

7+{-# LANGUAGE FlexibleInstances #-}

8+{-# LANGUAGE UndecidableInstances #-}

10+-----------------------------------------------------------------------------

11+-- |

12+-- Module : Control.Distributed.Process.Management

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

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

15+--

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

17+-- Stability : experimental

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

19+--

20+-- /Management Extensions/ for Cloud Haskell.

21+--

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

23+-- system events in a Cloud Haskell node.

24+--

25+-- [Architecture Overview]

26+--

27+-- /Management Extensions/ can be used by infrastructure or application code

28+-- to provide access to runtime instrumentation data, expose control planes

29+-- and/or provide remote access to external clients.

30+--

31+-- The management infrastructure is broken down into three components, which

32+-- roughly correspond to the modules it exposes:

33+--

34+-- 1. @Agent@ - registration and management of /Management Agents/

35+-- 2. @Instrumentation@ - provides an API for publishing /instrumentation data/

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

37+--

38+--

39+-----------------------------------------------------------------------------

40+module Control.Distributed.Process.Management

41+ (

42+ MxEvent(..)

43+ -- * Firing Arbitrary /Mx Events/

44+ , mxNotify

45+ -- * Constructing Mx Agents

46+ , MxAction()

47+ , MxAgentId(..)

48+ , MxAgent()

49+ , mxAgent

50+ , MxSink

51+ , mxSink

52+ , mxGetId

53+ , mxDeactivate

54+ , mxReady

55+ , mxBroadcast

56+ , mxSetLocal

57+ , mxGetLocal

58+ , liftMX

59+ -- * Mx Data API

60+ , mxPublish

61+ , mxSet

62+ , mxGet

63+ , mxClear

64+ , mxPurgeTable

65+ , mxDropTable

66+ ) where

67+

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

69+import Control.Concurrent.STM (atomically)

70+import Control.Concurrent.STM.TChan

71+ ( readTChan

72+ , writeTChan

73+ )

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

75+ ( newChan

76+ , nsend

77+ , receiveWait

78+ , matchChan

79+ , unwrapMessage

80+ )

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

82+ ( Process

83+ , ProcessId

84+ , Message

85+ , LocalProcess(..)

86+ , LocalNode(..)

87+ , MxEventBus(..)

88+ , unsafeCreateUnencodedMessage

89+ )

90+import Control.Distributed.Process.Management.Bus (publishEvent)

91+import qualified Control.Distributed.Process.Management.Table as Table

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

93+ ( MxAgentId(..)

94+ , MxAgent(..)

95+ , MxAction(..)

96+ , MxAgentState(..)

97+ , MxAgentStart(..)

98+ , MxSink

99+ , MxEvent(..)

100+ )

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

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

103+import Control.Monad.Reader (ask)

104+import qualified Control.Monad.State as ST

105+ ( MonadState

106+ , StateT

107+ , get

108+ , modify

109+ , lift

110+ , runStateT

111+ )

112+

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

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

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

116+-- receiving process via this API, since /all/ registered agents will gain

117+-- access to the data structure once it is broadcast by the agent controller.

118+mxNotify :: (Serializable a) => a -> Process ()

119+mxNotify msg = do

120+ bus <- localEventBus . processNode <$> ask

121+ liftIO $ publishEvent bus $ unsafeCreateUnencodedMessage msg

122+

123+-- | Publish an arbitrary @Message@ as a property in the management database.

124+--

125+-- For publishing @Serializable@ data, use 'mxSet' instead. For publishing

126+-- properties efficiently within an agent, use 'mxAgentPublish' instead,

127+-- since it will be more efficient in that context.

128+mxPublish :: MxAgentId -> String -> Message -> Process ()

129+mxPublish a k v = Table.set k v (Table.MxForAgent a)

130+

131+-- | Sets an arbitrary @Serializable@ datum against a key in the management

132+-- database. Note that /no attempt is made to force the argument/, therefore

133+-- it is very important that you do not pass unevaluated thunks that might

134+-- crash some other, arbitrary process (or management agent!) that obtains

135+-- and attempts to force the value later on.

136+--

137+mxSet :: Serializable a => MxAgentId -> String -> a -> Process ()

138+mxSet mxId key msg = do

139+ Table.set key (unsafeCreateUnencodedMessage msg) (Table.MxForAgent mxId)

140+

141+-- | Fetches a property from the management database for the given key.

142+-- If the property is not set, or does not match the expected type when

143+-- typechecked (at runtime), returns @Nothing@.

144+mxGet :: Serializable a => MxAgentId -> String -> Process (Maybe a)

145+mxGet = Table.fetch . Table.MxForAgent

146+

147+-- | Clears a property from the management database using the given key.

148+-- If the key does not exist in the database, this is a noop.

149+mxClear :: MxAgentId -> String -> Process ()

150+mxClear mxId key = Table.clear key (Table.MxForAgent mxId)

151+

152+-- | Purges a table in the management database of all its stored properties.

153+mxPurgeTable :: MxAgentId -> Process ()

154+mxPurgeTable = Table.purge . Table.MxForAgent

155+

156+-- | Deletes a table from the management database.

157+mxDropTable :: MxAgentId -> Process ()

158+mxDropTable = Table.delete . Table.MxForAgent

159+

160+--------------------------------------------------------------------------------

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

162+--------------------------------------------------------------------------------

163+

164+mxGetId :: MxAgent s MxAgentId

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

166+

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

168+mxBroadcast msg = do

169+ state <- ST.get

170+ liftMX $ liftIO $ atomically $ do

171+ writeTChan (mxBus state) (unsafeCreateUnencodedMessage msg)

172+

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

174+mxDeactivate = return . MxAgentDeactivate

175+

176+mxReady :: forall s. MxAgent s MxAction

177+mxReady = return MxAgentReady

178+

179+liftMX :: Process a -> MxAgent s a

180+liftMX p = MxAgent $ ST.lift p

181+

182+mxSetLocal :: s -> MxAgent s ()

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

184+

185+mxGetLocal :: MxAgent s s

186+mxGetLocal = ST.get >>= return . mxLocalState

187+

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

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

190+ -> MxSink s

191+mxSink act msg = do

192+ msg' <- liftMX $ (unwrapMessage msg :: Process (Maybe m))

193+ case msg' of

194+ Nothing -> return Nothing

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

196+

197+data MxPipeline s =

198+ MxPipeline

199+ {

200+ current :: !(MxSink s)

201+ , next :: !(MxPipeline s)

202+ } | MxStop

203+

204+-- | Activates a new agent.

205+--

206+mxAgent :: MxAgentId

207+ -> s

208+ -> [MxSink s]

209+ -> Process ProcessId

210+mxAgent mxId initState handlers = do

211+ node <- processNode <$> ask

212+ pid <- liftIO $ mxNew (localEventBus node) $ start

213+ return pid

214+ where

215+ start (sendTChan, recvTChan) = do

216+ (sp, rp) <- newChan

217+ nsend Table.mxTableCoordinator (MxAgentStart sp mxId)

218+ -- liftIO $ putStrLn $ "waiting on table coordinator " ++ Table.mxTableCoordinator

219+ -- p <- whereis Table.mxTableCoordinator

220+ -- liftIO $ putStrLn $ "registration == " ++ (show p)

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

222+ -- liftIO $ putStrLn "starting agent listener..."

223+ runAgent handlers recvTChan $ MxAgentState mxId sendTChan tablePid initState

224+

225+ runAgent hs c s = do

226+ msg <- (liftIO $ atomically $ readTChan c)

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

228+ case action of

229+ MxAgentReady -> runAgent hs c state

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

231+-- MxAgentBecome h' -> runAgent h' c state

232+

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

234+ pipeline [] = MxStop

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

236+

237+ runPipeline :: forall s .

238+ Message

239+ -> MxAgentState s

240+ -> MxPipeline s

241+ -> Process (MxAction, MxAgentState s)

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

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

244+ let act = current msg

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

246+ case pass of

247+ Nothing -> runPipeline msg state next

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

249+

src/Control/Distributed/Process/Management/Agent.hsnew file+133

@@ -0,0 +1,133 @@

1+{-# LANGUAGE RecordWildCards #-}

3+module Control.Distributed.Process.Management.Agent where

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

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

7+import Control.Concurrent.STM (atomically)

8+import Control.Concurrent.STM.TChan

9+ ( TChan

10+ , newBroadcastTChanIO

11+ , writeTChan

12+ , dupTChan

13+ )

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

15+ ( receiveWait

16+ , matchAny

17+ , die

18+ , catches

19+ , Handler(..)

20+ )

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

22+ ( enqueueSTM

23+ , CQueue

24+ )

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

26+ ( Fork

27+ )

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

29+ ( traceController

30+ )

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

32+ ( Process

33+ , Message

34+ , Tracer(..)

35+ , LocalProcess(..)

36+ , ProcessId

37+ , forever'

38+ )

39+import Control.Exception (AsyncException(ThreadKilled), SomeException)

40+import Control.Monad (void)

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

42+import Control.Monad.Reader (ask)

43+import GHC.Weak (Weak, deRefWeak)

44+

45+--------------------------------------------------------------------------------

46+-- Agent Controller Implementation --

47+--------------------------------------------------------------------------------

48+

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

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

51+-- instantiate new agents on the current node.

52+type AgentConfig =

53+ (Tracer, Weak (CQueue Message),

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

55+

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.

59+--

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

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

62+-- @LocalNode@ would be inelegant. We forward messages directly to the

63+-- trace controller's message queue, just as the @MxEventBus@ that's

64+-- set up on the @LocalNode@ forwards messages directly to us. This

65+-- optimises the code path for tracing and avoids overloading the node

66+-- node controller's internal control plane with additional routing, at the

67+-- cost of a little more complexity and two cases where we break

68+-- encapsulation.

69+--

70+mxAgentController :: Fork

71+ -> MVar AgentConfig

72+ -> Process ()

73+mxAgentController forkProcess mv = do

74+ trc <- liftIO $ startTracing forkProcess

75+ sigbus <- liftIO $ newBroadcastTChanIO

76+ weakQueue <- processWeakQ <$> ask

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

78+ go sigbus trc

79+ where

80+ go bus tracer = forever' $ do

81+ void $ receiveWait [

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

83+ -- Since mxNotify can potentially pass an unevaluated thunk to

84+ -- our mailbox, the dequeue (i.e., matchMessage) can fail and

85+ -- crash this process, which we DO NOT want. Alternatively,

86+ -- we handle IO exceptions here explicitly, since we don't want

87+ -- this process to ever crash, and the assumption we therefore

88+ -- make is thus:

89+ --

90+ -- 1. only ThreadKilled can tell this process to terminate

91+ -- 2. all other exceptions are invalid and should be ignored

92+ --

93+ -- The outcome of course, is that /bad/ calls to mxNotify

94+ -- (e.g., passing unevaluated thunks that will crash when

95+ -- they're eventually forced) are thus silently ignored.

96+ --

97+ matchAny (\msg -> liftIO $ broadcast bus tracer msg)

98+ ] `catches` [Handler (\ThreadKilled -> die "Killed"),

99+ Handler (\(_ :: SomeException) -> return ())]

100+

101+ broadcast :: TChan Message -> Tracer -> Message -> IO ()

102+ broadcast ch tr msg = do

103+ tmQueue <- tracerQueue tr

104+ atomicBroadcast ch tmQueue msg

105+

106+ tracerQueue :: Tracer -> IO (Maybe (CQueue Message))

107+ tracerQueue (Tracer _ wQ) = deRefWeak wQ

108+

109+ atomicBroadcast :: TChan Message

110+ -> Maybe (CQueue Message)

111+ -> Message -> IO ()

112+ atomicBroadcast ch Nothing msg = liftIO $ atomically $ writeTChan ch msg

113+ atomicBroadcast ch (Just q) msg = do

114+ -- liftIO $ putStrLn $ "broadcasting " ++ (show msg)

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

116+

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

118+mxStartAgent :: Fork

119+ -> TChan Message

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

121+ -> IO ProcessId

122+mxStartAgent fork chan handler = do

123+ chan' <- atomically (dupTChan chan)

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

125+ fork proc

126+

127+startTracing :: Fork -> IO Tracer

128+startTracing forkProcess = do

129+ mv <- newEmptyMVar

130+ pid <- forkProcess $ traceController mv

131+ wQ <- liftIO $ takeMVar mv

132+ return $ Tracer pid wQ

133+

src/Control/Distributed/Process/Management/Bus.hsnew file+21

@@ -0,0 +1,21 @@

1+-- | Interface to the management event bus.

2+module Control.Distributed.Process.Management.Bus

3+ ( publishEvent

4+ ) where

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

7+ ( enqueue

8+ )

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

10+ ( MxEventBus(..)

11+ , Message

12+ )

13+import Data.Foldable (forM_)

14+import System.Mem.Weak (deRefWeak)

15+

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

17+publishEvent MxEventBusInitialising _ = return ()

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

19+ mQueue <- deRefWeak wqRef

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

21+

src/Control/Distributed/Process/Management/Table.hsnew file+219

@@ -0,0 +1,219 @@

1+{-# LANGUAGE DeriveGeneric #-}

2+{-# LANGUAGE RecordWildCards #-}

4+module Control.Distributed.Process.Management.Table

5+ ( MxTableRequest(..)

6+ , MxTableId(..)

7+ , mxTableCoordinator

8+ , startTableCoordinator

9+ , delete

10+ , purge

11+ , clear

12+ , set

13+ , get

14+ , fetch

15+ ) where

16+

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

18+ ( receiveWait

19+ , receiveChan

20+ , match

21+ , matchAny

22+ , matchIf

23+ , matchChan

24+ , send

25+ , nsend

26+ , sendChan

27+ , getSelfPid

28+ , link

29+ , monitor

30+ , unwrapMessage

31+ , newChan

32+ , withMonitor

33+ )

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

35+ ( Process

36+ , ProcessId

37+ , ProcessMonitorNotification(..)

38+ , SendPort

39+ , ReceivePort

40+ , Message

41+ , unsafeCreateUnencodedMessage

42+ )

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

44+ ( MxTableId(..)

45+ , MxAgentId(..)

46+ , MxAgentStart(..)

47+ , Fork)

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

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

50+import Data.Accessor (Accessor, accessor, (^=), (^:))

51+import Data.Binary (Binary)

52+import Data.Map.Strict (Map)

53+import qualified Data.Map.Strict as Map

54+import Data.Typeable (Typeable)

55+

56+import GHC.Generics

57+

58+-- An extremely lightweight shared Map implementation, for use

59+-- by /management agents/ and their cohorts. Each agent is assigned

60+-- a table, into which any serializable @Message@ can be inserted.

61+-- Data are inserted, removed and searched for via their key, which

62+-- is a string. Tables can be purged, values can be set, fetched or

63+-- cleared/removed.

64+--

65+

66+data MxTableRequest =

67+ Delete

68+ | Purge

69+ | Clear !String

70+ | Set !String !Message

71+ | Get !String !(SendPort (Maybe Message)) -- see [note: un-typed send port]

72+ deriving (Typeable, Generic)

73+instance Binary MxTableRequest where

74+

75+data MxTableState =

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

77+

78+type MxTables = Map MxAgentId ProcessId

79+

80+mxTableCoordinator :: String

81+mxTableCoordinator = "mx.table.coordinator"

82+

83+delete :: MxTableId -> Process ()

84+delete = sendReq Delete

85+

86+purge :: MxTableId -> Process ()

87+purge = sendReq Purge

88+

89+clear :: String -> MxTableId -> Process ()

90+clear k = sendReq (Clear k)

91+

92+set :: String -> Message -> MxTableId -> Process ()

93+set k v = sendReq (Set k v)

94+

95+fetch :: forall a. (Serializable a)

96+ => MxTableId

97+ -> String

98+ -> Process (Maybe a)

99+fetch (MxForPid pid) key = get pid key

100+fetch mxId@(MxForAgent _) key = do

101+ (sp, rp) <- newChan :: Process (SendPort (Maybe Message),

102+ ReceivePort (Maybe Message))

103+ sendReq (Get key sp) mxId

104+ receiveChan rp >>= maybe (return Nothing)

105+ (unwrapMessage :: Message -> Process (Maybe a))

106+

107+-- [note: un-typed send port]

108+-- Here, fetch uses a typed channel over a raw Message to obtain

109+-- its result, so type checking is deferred until receipt and will

110+-- be handled in the caller's thread. This is necessary because

111+-- the server portion of the code knows nothing about the types

112+-- involved, nor should it, since these tables can be used to

113+-- store arbitrary serializable data.

114+

115+get :: forall a. (Serializable a)

116+ => ProcessId

117+ -> String

118+ -> Process (Maybe a)

119+get pid key = do

120+ safeFetch pid key >>= maybe (return Nothing)

121+ (unwrapMessage :: Message -> Process (Maybe a))

122+

123+safeFetch :: ProcessId -> String -> Process (Maybe Message)

124+safeFetch pid key = do

125+ (sp, rp) <- newChan

126+ send pid $ Get key sp

127+ withMonitor pid $ do

128+ receiveWait [

129+ matchChan rp return

130+ , matchIf (\(ProcessMonitorNotification _ pid' _) -> pid' == pid)

131+ (\_ -> return $ Just (unsafeCreateUnencodedMessage ()))

132+ ]

133+

134+sendReq :: MxTableRequest -> MxTableId -> Process ()

135+sendReq req tid = (resolve tid) req

136+

137+resolve :: Serializable a => MxTableId -> (a -> Process ())

138+resolve (MxForAgent agent) = \msg -> nsend mxTableCoordinator (agent, msg)

139+resolve (MxForPid pid) = \msg -> send pid msg

140+

141+startTableCoordinator :: Fork -> Process ()

142+startTableCoordinator fork = run Map.empty

143+ where

144+ run :: MxTables -> Process ()

145+ run tables =

146+ receiveWait [

147+ -- note that this state change can race with MxAgentStart requests

148+ match (\(ProcessMonitorNotification _ pid _) -> do

149+ return $ Map.filter (/= pid) tables)

150+ , match (\(MxAgentStart ch agent) -> do

151+ lookupAgent tables agent >>= \(p, t) -> do

152+ sendChan ch p >> return t)

153+ , match (\req@(agent, tReq :: MxTableRequest) -> do

154+ case tReq of

155+ Get k sp -> do

156+ lookupAgent tables agent >>= \(p, t) -> do

157+ safeFetch p k >>= sendChan sp >> return t

158+ _ -> do

159+ handleRequest tables req)

160+ , matchAny (\_ -> return tables) -- unrecognised messages are dropped

161+ ] >>= run

162+

163+ handleRequest :: MxTables

164+ -> (MxAgentId, MxTableRequest)

165+ -> Process MxTables

166+ handleRequest tables' (agent, req) = do

167+ lookupAgent tables' agent >>= \(p, t) -> send p req >> return t

168+

169+ lookupAgent :: MxTables -> MxAgentId -> Process (ProcessId, MxTables)

170+ lookupAgent tables' agentId' = do

171+ case Map.lookup agentId' tables' of

172+ Nothing -> launchNew agentId' tables'

173+ Just p -> return (p, tables')

174+

175+ launchNew :: MxAgentId

176+ -> MxTables

177+ -> Process (ProcessId, MxTables)

178+ launchNew mxId tblMap = do

179+ let initState = MxTableState { _name = (agentId mxId)

180+ , _entries = Map.empty

181+ }

182+ (pid, _) <- spawnSup $ tableHandler initState

183+ return $ (pid, Map.insert mxId pid tblMap)

184+

185+ spawnSup proc = do

186+ us <- getSelfPid

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

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

189+ -- agent, management API and the tracing modules

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

191+ ref <- monitor them

192+ return (them, ref)

193+

194+tableHandler :: MxTableState -> Process ()

195+tableHandler state = do

196+ ns <- receiveWait [

197+ match (handleTableRequest state)

198+ , matchAny (\_ -> return (Just state))

199+ ]

200+ case ns of

201+ Nothing -> return ()

202+ Just s' -> tableHandler s'

203+ where

204+ handleTableRequest _ Delete = return Nothing

205+ handleTableRequest st Purge = return $ Just $ (entries ^= Map.empty) $ st

206+ handleTableRequest st (Clear k) = return $ Just $ (entries ^: Map.delete k) $ st

207+ handleTableRequest st (Set k v) = return $ Just $ (entries ^: Map.insert k v) st

208+ handleTableRequest st (Get k c) = getEntry k c st >> return (Just st)

209+

210+getEntry :: String

211+ -> SendPort (Maybe Message)

212+ -> MxTableState

213+ -> Process ()

214+getEntry k m MxTableState{..} = do

215+ sendChan m =<< return (Map.lookup k _entries)

216+

217+entries :: Accessor MxTableState (Map String Message)

218+entries = accessor _entries (\ls st -> st { _entries = ls })

219+

src/Control/Distributed/Process/Internal/Trace/Primitives.hs → src/Control/Distributed/Process/Management/Trace/Primitives.hsrenamed+15−25

@@ -3,15 +3,14 @@

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

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

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

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

7+-- tracing infrastructure's messaging fabric.

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

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

7− ( Tracer

8− , TraceEvent(..)

9− -- * Sending Trace Data

9+ ( -- * Sending Trace Data

10− , traceLog

10+ traceLog

11 , traceLogFmt

12 , traceMessage

13 -- * Configuring A Tracer

14− , isTracingEnabled

15 , defaultTraceFlags

14 , defaultTraceFlags

16 , enableTrace

15 , enableTrace

17 , enableTraceAsync

16 , enableTraceAsync

@@ -33,15 +32,14 @@ import Control.Distributed.Process.Internal.Primitives

33 , newChan

32 , newChan

34 , receiveChan

33 , receiveChan

35 )

34 )

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

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

37 ( TraceArg(..)

36 ( TraceArg(..)

38− , TraceEvent(..)

39 , TraceFlags(..)

37 , TraceFlags(..)

40 , TraceOk(..)

38 , TraceOk(..)

41 , TraceSubject(..)

39 , TraceSubject(..)

42 , defaultTraceFlags

40 , defaultTraceFlags

43 )

41 )

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

42+import qualified Control.Distributed.Process.Management.Trace.Types as Tracer

45 ( traceLog

43 ( traceLog

46 , traceLogFmt

44 , traceLogFmt

47 , traceMessage

45 , traceMessage

@@ -55,12 +53,12 @@ import qualified Control.Distributed.Process.Internal.Trace.Types as Tracer

55 , getCurrentTraceClient

53 , getCurrentTraceClient

56 )

54 )

57import Control.Distributed.Process.Internal.Types

55import Control.Distributed.Process.Internal.Types

58− ( Tracer(..)

59− , Process

56+ ( Process

60 , ProcessId

57 , ProcessId

61 , LocalProcess(..)

58 , LocalProcess(..)

62− , LocalNode(localTracer)

59+ , LocalNode(localEventBus)

63 , SendPort

60 , SendPort

61+ , MxEventBus(..)

64 )

62 )

65import Control.Distributed.Process.Serializable

63import Control.Distributed.Process.Serializable

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

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

@@ -72,8 +70,8 @@ import qualified Data.Set as Set (fromList)

72-- Main API --

70-- Main API --

73--------------------------------------------------------------------------------

71--------------------------------------------------------------------------------

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

73+-- | Converts a list of identifiers (that can be

76−-- to a 'TraceSubject'.

74+-- mapped to process ids), to a 'TraceSubject'.

77class Traceable a where

75class Traceable a where

78 uod :: [a] -> TraceSubject

76 uod :: [a] -> TraceSubject

@@ -95,14 +93,6 @@ traceOn = Just TraceAll

95traceOff :: Maybe TraceSubject

93traceOff :: Maybe TraceSubject

96traceOff = Nothing

94traceOff = Nothing

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

99−isTracingEnabled :: Process Bool

100−isTracingEnabled = do

101− node <- processNode <$> ask

102− case (localTracer node) of

103− (ActiveTracer _ _) -> return True

104− _ -> return False

105−

106-- | Enable tracing to the supplied process.

96-- | Enable tracing to the supplied process.

107enableTraceAsync :: ProcessId -> Process ()

97enableTraceAsync :: ProcessId -> Process ()

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

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

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

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

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

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

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

159

149

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

150+withLocalTracer :: (MxEventBus -> Process ()) -> Process ()

161withLocalTracer act = do

151withLocalTracer act = do

162 node <- processNode <$> ask

152 node <- processNode <$> ask

163− act (localTracer node)

153+ act (localEventBus node)

164

154

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

155+withLocalTracerSync :: (MxEventBus -> SendPort TraceOk -> IO ()) -> Process ()

166withLocalTracerSync act = do

156withLocalTracerSync act = do

167 (sp, rp) <- newChan

157 (sp, rp) <- newChan

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

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

src/Control/Distributed/Process/Internal/Trace/Remote.hs → src/Control/Distributed/Process/Management/Trace/Remote.hsrenamed+4−5

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

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

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

2− ( -- * Configuring A Tracer

2+ ( -- * Configuring A Remote Tracer

3 setTraceFlagsRemote

4 , startTraceRelay

5 -- * Remote Table

@@ -14,11 +14,11 @@ import Control.Distributed.Process.Internal.Primitives

14 , relay

15 , nsendRemote

16 )

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

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

18 ( TraceFlags(..)

19 , TraceOk(..)

20 )

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

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

22 ( withRegisteredTracer

23 , enableTrace

24 )

@@ -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/Internal/Trace/Tracer.hs → src/Control/Distributed/Process/Management/Trace/Tracer.hsrenamed+60−70

@@ -1,6 +1,6 @@

1-- | Tracing/Debugging support - Trace Implementation

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

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

3− ( -- * API for the Node Controller

3+ ( -- * API for the Management Agent

4 traceController

5 -- * Built in tracers

6 , defaultTracer

@@ -31,16 +31,18 @@ import Control.Distributed.Process.Internal.Primitives

31 , handleMessage

32 , matchUnknown

33 )

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

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

35− ( TraceEvent(..)

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

41 , defaultTraceFlags

43 , defaultTraceFlags

42 )

44 )

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

45+import Control.Distributed.Process.Management.Trace.Primitives

44 ( traceOn )

46 ( traceOn )

45import Control.Distributed.Process.Internal.Types

47import Control.Distributed.Process.Internal.Types

46 ( LocalNode(..)

48 ( LocalNode(..)

@@ -89,7 +91,9 @@ import System.Mem.Weak

89 ( Weak

91 ( Weak

90 )

92 )

92−data TracerState = TracerST {

94+data TracerState =

95+ TracerST

96+ {

93 client :: !(Maybe ProcessId)

97 client :: !(Maybe ProcessId)

94 , flags :: !TraceFlags

98 , flags :: !TraceFlags

95 , regNames :: !(Map ProcessId (Set String))

99 , regNames :: !(Map ProcessId (Set String))

@@ -119,7 +123,7 @@ defaultEventLogTracer =

119checkEnv :: String -> Process String

123checkEnv :: String -> Process String

120checkEnv s = liftIO $ getEnv s

124checkEnv s = liftIO $ getEnv s

121

125

122−-- This trace client is (intentionally) useful - it simply provides

126+-- This trace client is (intentionally) a noop - it simply provides

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

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

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

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

125-- it. This is the default trace client.

129-- it. This is the default trace client.

@@ -130,14 +134,13 @@ nullTracer =

130systemLoggerTracer :: Process ()

134systemLoggerTracer :: Process ()

131systemLoggerTracer = do

135systemLoggerTracer = do

132 node <- processNode <$> ask

136 node <- processNode <$> ask

133− let tr = sendTraceMsg node

137+ let tr = sendTraceLog node

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

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

135 where

139 where

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

140+ sendTraceLog :: LocalNode -> MxEvent -> Process ()

137− sendTraceMsg node ev = do

141+ sendTraceLog node ev = do

138 now <- liftIO $ getCurrentTime

142 now <- liftIO $ getCurrentTime

139− txt <- buildTxt ev

140− msg <- return $ (formatTime defaultTimeLocale "%c" now, txt)

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

141 emptyPid <- return $ (nullProcessId (localNodeId node))

144 emptyPid <- return $ (nullProcessId (localNodeId node))

142 traceMsg <- return $ NCMsg {

145 traceMsg <- return $ NCMsg {

143 ctrlMsgSender = ProcessIdentifier (emptyPid)

146 ctrlMsgSender = ProcessIdentifier (emptyPid)

@@ -146,9 +149,9 @@ systemLoggerTracer = do

146 }

149 }

147 liftIO $ writeChan (localCtrlChan node) traceMsg

150 liftIO $ writeChan (localCtrlChan node) traceMsg

148

151

149− buildTxt :: TraceEvent -> Process String

152+ buildTxt :: MxEvent -> String

150− buildTxt (TraceEvLog msg) = return msg

153+ buildTxt (MxLog msg) = msg

151− buildTxt ev = return $ show ev

154+ buildTxt ev = show ev

152

155

153eventLogTracer :: Process ()

156eventLogTracer :: Process ()

154eventLogTracer =

157eventLogTracer =

@@ -157,7 +160,7 @@ eventLogTracer =

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

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

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

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

159 where

162 where

160− writeTrace :: TraceEvent -> Process ()

163+ writeTrace :: MxEvent -> Process ()

161 writeTrace ev = liftIO $ traceEventIO (show ev)

164 writeTrace ev = liftIO $ traceEventIO (show ev)

162

165

163logfileTracer :: FilePath -> Process ()

166logfileTracer :: FilePath -> Process ()

@@ -171,14 +174,14 @@ logfileTracer p = do

171 logger h' = forever' $ do

174 logger h' = forever' $ do

172 receiveWait [

175 receiveWait [

173 matchIf (\ev -> case ev of

176 matchIf (\ev -> case ev of

174− TraceEvDisable -> True

177+ MxTraceDisable -> True

175− (TraceEvTakeover _) -> True

178+ (MxTraceTakeover _) -> True

176 _ -> False)

179 _ -> False)

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

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

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

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

179 ]

182 ]

180

183

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

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

182 writeTrace h ev = do

185 writeTrace h ev = do

183 liftIO $ do

186 liftIO $ do

184 now <- getCurrentTime

187 now <- getCurrentTime

@@ -190,8 +193,8 @@ logfileTracer p = do

190

193

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

194traceController :: MVar ((Weak (CQueue Message))) -> Process ()

192traceController mv = do

195traceController mv = do

196+ -- See the documentation for mxAgentController for a

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

197+ -- commentary that explains this breach of encapsulation

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

195 weakQueue <- processWeakQ <$> ask

198 weakQueue <- processWeakQ <$> ask

196 liftIO $ putMVar mv weakQueue

199 liftIO $ putMVar mv weakQueue

197 initState <- initialState

200 initState <- initialState

@@ -199,23 +202,24 @@ traceController mv = do

199 where

202 where

200 traceLoop :: TracerState -> Process ()

203 traceLoop :: TracerState -> Process ()

201 traceLoop st = do

204 traceLoop st = do

205+ let client' = client st

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

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

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

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

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

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

205 st' <- receiveWait [

209 st' <- receiveWait [

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

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

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

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

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

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

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

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

210− -- figure out.

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

211 case set of

215 case set of

212 (TraceEnable pid) -> do

216 (TraceEnable pid) -> do

213 -- notify the previous tracer it has been replaced

217 -- notify the previous tracer it has been replaced

214− sendTrace st (createUnencodedMessage (TraceEvTakeover pid))

218+ sendTraceMsg client' (createUnencodedMessage (MxTraceTakeover pid))

215 sendOk setResp

219 sendOk setResp

216 return st { client = (Just pid) }

220 return st { client = (Just pid) }

217 TraceDisable -> do

221 TraceDisable -> do

218− sendTrace st (createUnencodedMessage TraceEvDisable)

222+ sendTraceMsg client' (createUnencodedMessage MxTraceDisable)

219 sendOk setResp

223 sendOk setResp

220 return st { client = Nothing })

224 return st { client = Nothing })

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

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

@@ -224,7 +228,7 @@ traceController mv = do

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

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

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

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

226 , matchAny (\ev ->

230 , matchAny (\ev ->

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

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

228 ]

232 ]

229 traceLoop st'

233 traceLoop st'

230

234

@@ -245,7 +249,6 @@ traceController mv = do

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

249 catch (checkEnv "DISTRIBUTED_PROCESS_TRACE_FLAGS" >>= return . parseFlags)

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

250 (\(_ :: IOError) -> return defaultTraceFlags)

247

251

248− -- quick and dirty fold-while

249 parseFlags :: String -> TraceFlags

252 parseFlags :: String -> TraceFlags

250 parseFlags s = parseFlags' s defaultTraceFlags

253 parseFlags s = parseFlags' s defaultTraceFlags

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

254 where parseFlags' :: String -> TraceFlags -> TraceFlags

@@ -260,27 +263,11 @@ traceController mv = do

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

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

261 | otherwise = parseFlags' xs parsedFlags

264 | otherwise = parseFlags' xs parsedFlags

262

265

263−-- note [runtime tracer control]

264−--

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

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

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

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

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

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

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

272−-- to a single function call.

273−--

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

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

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

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

278−

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

266applyTraceFlags :: TraceFlags -> TracerState -> Process TracerState

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

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

281

268

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

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

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

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

284 let regNames' =

271 let regNames' =

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

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

286 (Set.singleton n)

273 (Set.singleton n)

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

288 in do

275 in do

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

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

290 return st { regNames = regNames' }

277 return st { regNames = regNames' }

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

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

292 let f ns = case ns of

279 let f ns = case ns of

293 Nothing -> Nothing

280 Nothing -> Nothing

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

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

@@ -296,46 +283,46 @@ handleTrace st msg ev@(TraceEvUnRegistered p n) =

296 in do

283 in do

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

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

298 return st { regNames = regNames' }

285 return st { regNames = regNames' }

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

307handleTrace st msg ev = do

294handleTrace st msg ev = do

308 case ev of

295 case ev of

309− (TraceEvNodeDied _ _) ->

296+ (MxNodeDied _ _) ->

310 case (traceNodes (flags st)) of

297 case (traceNodes (flags st)) of

311− True -> sendTrace st msg

298+ True -> sendTrace st ev msg

312 False -> return ()

299 False -> return ()

313− (TraceEvUser _) -> sendTrace st msg

300+ (MxUser _) -> sendTrace st ev msg

314− (TraceEvLog _) -> sendTrace st msg

301+ (MxLog _) -> sendTrace st ev msg

315 _ ->

302 _ ->

316 case (traceConnections (flags st)) of

303 case (traceConnections (flags st)) of

317− True -> sendTrace st msg

304+ True -> sendTrace st ev msg

318 False -> return ()

305 False -> return ()

319 return st

306 return st

320

307

321−traceEv :: TraceEvent

308+traceEv :: MxEvent

322 -> Message

309 -> Message

323 -> Maybe TraceSubject

310 -> Maybe TraceSubject

324 -> TracerState

311 -> TracerState

325 -> Process ()

312 -> Process ()

326traceEv _ _ Nothing _ = return ()

313traceEv _ _ Nothing _ = return ()

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

314+traceEv ev msg (Just TraceAll) st = sendTrace st ev msg

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

315traceEv ev msg (Just (TraceProcs pids)) st = do

329 node <- processNode <$> ask

316 node <- processNode <$> ask

330 let p = case resolveToPid ev of

317 let p = case resolveToPid ev of

331 Nothing -> (nullProcessId (localNodeId node))

318 Nothing -> (nullProcessId (localNodeId node))

332 Just pid -> pid

319 Just pid -> pid

333 case (Set.member p pids) of

320 case (Set.member p pids) of

334− True -> sendTrace st msg

321+ True -> sendTrace st ev msg

335 False -> return ()

322 False -> return ()

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

323traceEv ev msg (Just (TraceNames names)) st = do

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

324+ -- if we have recorded regnames for p, then we forward the trace iif

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

325+ -- there are overlapping trace targets

339 node <- processNode <$> ask

326 node <- processNode <$> ask

340 let p = case resolveToPid ev of

327 let p = case resolveToPid ev of

341 Nothing -> (nullProcessId (localNodeId node))

328 Nothing -> (nullProcessId (localNodeId node))

@@ -344,13 +331,16 @@ traceEv ev msg (Just (TraceNames names)) st = do

344 Nothing -> return ()

331 Nothing -> return ()

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

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

346 then return ()

333 then return ()

347− else sendTrace st msg

334+ else sendTrace st ev msg

348

335

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

336+sendTrace :: TracerState -> MxEvent -> Message -> Process ()

350−sendTrace st msg =

337+sendTrace st ev msg = do

351− let pid = (client st) in do

338+ let c = client st

352− case pid of

339+ if c == (resolveToPid ev) -- we do not send the tracer events about itself...

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

354− Nothing -> return ()

340+ then return ()

341+ else sendTraceMsg c msg

355

342

343+sendTraceMsg :: Maybe ProcessId -> Message -> Process ()

344+sendTraceMsg Nothing _ = return ()

345+sendTraceMsg (Just p) msg = (flip forward) p msg

356

346

src/Control/Distributed/Process/Management/Trace/Types.hsnew file+165

@@ -0,0 +1,165 @@

1+{-# LANGUAGE DeriveGeneric #-}

3+-- | Tracing/Debugging support - Types

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

5+ ( SetTrace(..)

6+ , TraceSubject(..)

7+ , TraceFlags(..)

8+ , TraceArg(..)

9+ , TraceOk(..)

10+ , traceLog

11+ , traceLogFmt

12+ , traceEvent

13+ , traceMessage

14+ , defaultTraceFlags

15+ , enableTrace

16+ , enableTraceSync

17+ , disableTrace

18+ , disableTraceSync

19+ , getTraceFlags

20+ , setTraceFlags

21+ , setTraceFlagsSync

22+ , getCurrentTraceClient

23+ ) where

24+

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

26+ ( MxEventBus(..)

27+ , ProcessId

28+ , SendPort

29+ , unsafeCreateUnencodedMessage

30+ )

31+import Control.Distributed.Process.Management.Bus

32+ ( publishEvent

33+ )

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

35+ ( MxEvent(..)

36+ )

37+import Control.Distributed.Process.Serializable

38+import Data.Binary

39+import Data.List (intersperse)

40+import Data.Set (Set)

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

42+import Data.Typeable

43+import GHC.Generics

44+

45+--------------------------------------------------------------------------------

46+-- Types --

47+--------------------------------------------------------------------------------

48+

49+data SetTrace = TraceEnable !ProcessId | TraceDisable

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

51+instance Binary SetTrace where

52+

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

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

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

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

57+--

58+data TraceSubject =

59+ TraceAll -- enable tracing for all running processes

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

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

62+ deriving (Typeable, Generic, Show)

63+instance Binary TraceSubject where

64+

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

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

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

68+data TraceFlags = TraceFlags {

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

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

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

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

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

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

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

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

77+ } deriving (Typeable, Generic, Show)

78+instance Binary TraceFlags where

79+

80+defaultTraceFlags :: TraceFlags

81+defaultTraceFlags =

82+ TraceFlags {

83+ traceSpawned = Nothing

84+ , traceDied = Nothing

85+ , traceRegistered = Nothing

86+ , traceUnregistered = Nothing

87+ , traceSend = Nothing

88+ , traceRecv = Nothing

89+ , traceNodes = False

90+ , traceConnections = False

91+ }

92+

93+data TraceArg =

94+ TraceStr String

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

96+

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

98+data TraceOk = TraceOk

99+ deriving (Typeable, Generic)

100+instance Binary TraceOk where

101+

102+--------------------------------------------------------------------------------

103+-- Internal/Common API --

104+--------------------------------------------------------------------------------

105+

106+traceLog :: MxEventBus -> String -> IO ()

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

108+

109+traceLogFmt :: MxEventBus

110+ -> String

111+ -> [TraceArg]

112+ -> IO ()

113+traceLogFmt t d ls =

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

115+ where toS :: TraceArg -> String

116+ toS (TraceStr s) = s

117+ toS (Trace a) = show a

118+

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

120+traceEvent tr ev = publishEvent tr (unsafeCreateUnencodedMessage ev)

121+

122+traceMessage :: Serializable m => MxEventBus -> m -> IO ()

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

124+

125+enableTrace :: MxEventBus -> ProcessId -> IO ()

126+enableTrace t p =

127+ publishEvent t (unsafeCreateUnencodedMessage ((Nothing :: Maybe (SendPort TraceOk)),

128+ (TraceEnable p)))

129+

130+enableTraceSync :: MxEventBus -> SendPort TraceOk -> ProcessId -> IO ()

131+enableTraceSync t s p =

132+ publishEvent t (unsafeCreateUnencodedMessage (Just s, TraceEnable p))

133+

134+disableTrace :: MxEventBus -> IO ()

135+disableTrace t =

136+ publishEvent t (unsafeCreateUnencodedMessage ((Nothing :: Maybe (SendPort TraceOk)),

137+ TraceDisable))

138+

139+disableTraceSync :: MxEventBus -> SendPort TraceOk -> IO ()

140+disableTraceSync t s =

141+ publishEvent t (unsafeCreateUnencodedMessage ((Just s), TraceDisable))

142+

143+setTraceFlags :: MxEventBus -> TraceFlags -> IO ()

144+setTraceFlags t f =

145+ publishEvent t (unsafeCreateUnencodedMessage ((Nothing :: Maybe (SendPort TraceOk)), f))

146+

147+setTraceFlagsSync :: MxEventBus -> SendPort TraceOk -> TraceFlags -> IO ()

148+setTraceFlagsSync t s f =

149+ publishEvent t (unsafeCreateUnencodedMessage ((Just s), f))

150+

151+getTraceFlags :: MxEventBus -> SendPort TraceFlags -> IO ()

152+getTraceFlags t s = publishEvent t (unsafeCreateUnencodedMessage s)

153+

154+getCurrentTraceClient :: MxEventBus -> SendPort (Maybe ProcessId) -> IO ()

155+getCurrentTraceClient t s = publishEvent t (unsafeCreateUnencodedMessage s)

156+

157+class Traceable a where

158+ uod :: [a] -> TraceSubject

159+

160+instance Traceable ProcessId where

161+ uod = TraceProcs . Set.fromList

162+

163+instance Traceable String where

164+ uod = TraceNames . Set.fromList

165+

src/Control/Distributed/Process/Management/Types.hsnew file+172

@@ -0,0 +1,172 @@

1+{-# LANGUAGE DeriveGeneric #-}

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

3+ ( MxAgentId(..)

4+ , MxTableId(..)

5+ , MxAgentState(..)

6+ , MxAgent(..)

7+ , MxAction(..)

8+ , MxAgentStart(..)

9+ , Fork

10+ , MxSink

11+ , MxEvent(..)

12+ , Addressable(..)

13+ ) where

14+

15+import Control.Applicative (Applicative)

16+import Control.Concurrent.STM

17+ ( TChan

18+ )

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

20+ ( Process

21+ , ProcessId

22+ , Message

23+ , SendPort

24+ , DiedReason

25+ , NodeId

26+ )

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

28+import qualified Control.Monad.State as ST

29+ ( MonadState

30+ , StateT

31+ )

32+import Data.Binary

33+import Data.Typeable (Typeable)

34+import GHC.Generics

35+import Network.Transport

36+ ( ConnectionId

37+ , EndPointAddress

38+ )

39+

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

41+

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

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

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

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

46+--

47+data MxEvent =

48+ MxSpawned ProcessId

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

50+ | MxRegistered ProcessId String

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

52+ | MxUnRegistered ProcessId String

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

54+ | MxProcessDied ProcessId DiedReason

55+ -- ^ fired whenever a process dies

56+ | MxNodeDied NodeId DiedReason

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

58+ | MxSent ProcessId ProcessId Message

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

60+ | MxReceived ProcessId Message

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

62+ | MxConnected ConnectionId EndPointAddress

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

64+ | MxDisconnected ConnectionId EndPointAddress

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

66+ | MxUser Message

67+ -- ^ a user defined trace event

68+ | MxLog String

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

70+ | MxTraceTakeover ProcessId

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

72+ | MxTraceDisable

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

74+ deriving (Typeable, Generic, Show)

75+

76+instance Binary MxEvent where

77+

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

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

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

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

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

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

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

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

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

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

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

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

90+ -- put MxTraceDisable = putWord8 13

91+

92+ -- get = do

93+ -- header <- getWord8

94+ -- case header of

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

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

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

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

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

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

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

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

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

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

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

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

107+ -- 13 -> return MxTraceDisable

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

109+

110+class Addressable a where

111+ resolveToPid :: a -> Maybe ProcessId

112+

113+instance Addressable MxEvent where

114+ resolveToPid (MxSpawned p) = Just p

115+ resolveToPid (MxProcessDied p _) = Just p

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

117+ resolveToPid (MxReceived p _) = Just p

118+ resolveToPid _ = Nothing

119+

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

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

122+-- when calling mxAgentController, since there's no other way to

123+-- avoid a circular dependency with Node.hs

124+type Fork = (Process () -> IO ProcessId)

125+

126+-- | A newtype wrapper for an agent id (which is a string).

127+newtype MxAgentId = MxAgentId { agentId :: String }

128+ deriving (Typeable, Binary, Eq, Ord)

129+

130+data MxTableId =

131+ MxForAgent !MxAgentId

132+ | MxForPid !ProcessId

133+ deriving (Typeable, Generic)

134+instance Binary MxTableId where

135+

136+data MxAgentState s = MxAgentState

137+ {

138+ mxAgentId :: !MxAgentId

139+ , mxBus :: !(TChan Message)

140+ , mxSharedTable :: !ProcessId

141+ , mxLocalState :: !s

142+ }

143+

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

145+-- agent's state and private

146+newtype MxAgent s a =

147+ MxAgent

148+ {

149+ unAgent :: ST.StateT (MxAgentState s) Process a

150+ } deriving ( Functor

151+ , Monad

152+ , MonadIO

153+ , ST.MonadState (MxAgentState s)

154+ , Typeable

155+ , Applicative

156+ )

157+

158+data MxAgentStart = MxAgentStart

159+ {

160+ mxAgentTableChan :: SendPort ProcessId

161+ , mxAgentIdStart :: MxAgentId

162+ }

163+ deriving (Typeable, Generic)

164+instance Binary MxAgentStart where

165+

166+data MxAction =

167+ MxAgentDeactivate !String

168+ | MxAgentReady

169+

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

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

172+

src/Control/Distributed/Process/Node.hs+76−77

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

@@ -56,7 +58,7 @@ import Control.Exception

56 , Exception

58 , Exception

57 , throwTo

59 , throwTo

58 )

60 )

59−import qualified Control.Exception as Exception (Handler(..), catch, catches, finally)

61+import qualified Control.Exception as Exception (Handler(..), catches, finally)

60import Control.Concurrent (forkIO, myThreadId)

62import Control.Concurrent (forkIO, myThreadId)

61import Control.Distributed.Process.Internal.StrictMVar

63import Control.Distributed.Process.Internal.StrictMVar

62 ( newMVar

64 ( newMVar

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

106 , LocalProcessId(..)

108 , LocalProcessId(..)

107 , ProcessId(..)

109 , ProcessId(..)

108 , LocalNode(..)

110 , LocalNode(..)

109− , Tracer(..)

111+ , MxEventBus(..)

110 , LocalNodeState(..)

112 , LocalNodeState(..)

111 , LocalProcess(..)

113 , LocalProcess(..)

112 , LocalProcessState(..)

114 , LocalProcessState(..)

@@ -149,19 +151,27 @@ import Control.Distributed.Process.Internal.Types

149 , firstNonReservedProcessId

151 , firstNonReservedProcessId

150 , ImplicitReconnect(WithImplicitReconnect)

152 , ImplicitReconnect(WithImplicitReconnect)

151 )

153 )

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

155+ ( mxAgentController

156+ )

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

157+import qualified Control.Distributed.Process.Management.Table as Table

158+ ( mxTableCoordinator

159+ , startTableCoordinator

160+ )

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

153 ( remoteTable

162 ( remoteTable

154 )

163 )

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

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

165+ ( defaultTracer

166+ )

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

156 ( TraceArg(..)

168 ( TraceArg(..)

157− , TraceEvent(..)

158 , traceEvent

169 , traceEvent

159 , traceLogFmt

170 , traceLogFmt

160 , enableTrace

171 , enableTrace

161 )

172 )

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

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

163− ( traceController

174+ ( MxEvent(..)

164− , defaultTracer

165 )

175 )

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

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

167import Control.Distributed.Process.Internal.Messaging

177import Control.Distributed.Process.Internal.Messaging

@@ -179,14 +189,13 @@ import Control.Distributed.Process.Internal.Primitives

179 , catch

189 , catch

180 , unwrapMessage

190 , unwrapMessage

181 )

191 )

182−import Control.Distributed.Process.Internal.Types (SendPort)

192+import Control.Distributed.Process.Internal.Types (SendPort, Tracer(..))

183import qualified Control.Distributed.Process.Internal.Closure.BuiltIn as BuiltIn (remoteTable)

193import qualified Control.Distributed.Process.Internal.Closure.BuiltIn as BuiltIn (remoteTable)

184import Control.Distributed.Process.Internal.WeakTQueue (TQueue, writeTQueue)

194import Control.Distributed.Process.Internal.WeakTQueue (TQueue, writeTQueue)

185import qualified Control.Distributed.Process.Internal.StrictContainerAccessors as DAC

195import qualified Control.Distributed.Process.Internal.StrictContainerAccessors as DAC

186 ( mapMaybe

196 ( mapMaybe

187 , mapDefault

197 , mapDefault

188 )

198 )

189−import System.Environment (getEnv)

190import Unsafe.Coerce

199import Unsafe.Coerce

191

200

192--------------------------------------------------------------------------------

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

@@ -222,10 +231,10 @@ createBareLocalNode endPoint rtable = do

222 , localEndPoint = endPoint

231 , localEndPoint = endPoint

223 , localState = state

232 , localState = state

224 , localCtrlChan = ctrlChan

233 , localCtrlChan = ctrlChan

225− , localTracer = InactiveTracer

234+ , localEventBus = MxEventBusInitialising

226 , remoteTable = rtable

235 , remoteTable = rtable

227 }

236 }

228− tracedNode <- startTracing node

237+ tracedNode <- startMxAgent node

229

238

230 -- Once the NC terminates, the endpoint isn't much use,

239 -- Once the NC terminates, the endpoint isn't much use,

231 void $ forkIO $ Exception.finally (runNodeController tracedNode)

240 void $ forkIO $ Exception.finally (runNodeController tracedNode)

@@ -243,38 +252,27 @@ createBareLocalNode endPoint rtable = do

243 , ctrlMsgSignal = SigShutdown

252 , ctrlMsgSignal = SigShutdown

244 }

253 }

245

254

246−startTracing :: LocalNode -> IO LocalNode

255+startMxAgent :: LocalNode -> IO LocalNode

247−startTracing node =

256+startMxAgent node = do

248− -- TODO: use tryGetEnv once we drop support for GHC 7.2.x

249− Exception.catch

250− (getEnv "DISTRIBUTED_PROCESS_TRACE_ENABLED" >> procTracer node)

251− (\(_ :: IOError) -> return node)

252− where procTracer :: LocalNode -> IO LocalNode

253− procTracer n = do

254− -- see note [tracer/forkProcess races]

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

258+ let fork = forkProcess node

255− mv <- MVar.newEmptyMVar

259+ mv <- MVar.newEmptyMVar

256− pid <- forkProcess n (traceController mv)

260+ pid <- fork $ mxAgentController fork mv

257− wqRef <- MVar.takeMVar mv

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

258− return n { localTracer = (ActiveTracer pid wqRef) }

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

259−

263+

260−startDefaultTracer :: LocalNode -> IO (Maybe ProcessId)

264+startDefaultTracer :: LocalNode -> IO ()

261−startDefaultTracer node =

265+startDefaultTracer node' = do

262− Exception.catch

266+ let t = localEventBus node'

263− (getEnv "DISTRIBUTED_PROCESS_TRACE_ENABLED" >> go node)

264− (\(_ :: IOError) -> return Nothing)

265− where go :: LocalNode -> IO (Maybe ProcessId)

266− go node' = do

267− registerTraceController node'

268− pid <- forkProcess node' defaultTracer

269− enableTrace (localTracer node') pid

270− return (Just pid)

271−

272− registerTraceController :: LocalNode -> IO ()

273− registerTraceController n =

274− let t = localTracer n in

275− case t of

267+ case t of

276− InactiveTracer -> return ()

268+ MxEventBus _ (Tracer pid _) _ _ -> do

277− (ActiveTracer pid _) -> runProcess n $ register "trace.controller" pid

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

270+ pid' <- forkProcess node' defaultTracer

271+ enableTrace (localEventBus node') pid'

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

273+ _ -> return ()

274+

275+-- TODO: we need a better mechanism for defining and registering services

278

276

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

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

280startServiceProcesses :: LocalNode -> IO ()

278startServiceProcesses :: LocalNode -> IO ()

@@ -283,26 +281,26 @@ startServiceProcesses node = do

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

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

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

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

285 -- however, so we deliberably ignore it

283 -- however, so we deliberably ignore it

286− mPid <- startDefaultTracer node

284+ startDefaultTracer node

287− case mPid of

285+ tableCoordinatorPid <- fork $ Table.startTableCoordinator fork

288− Nothing -> return ()

289− Just pid -> runProcess node $ register "tracer.initial" pid

286+ runProcess node $ register Table.mxTableCoordinator tableCoordinatorPid

290−

291 logger <- forkProcess node loop

287 logger <- forkProcess node loop

292 runProcess node $ register "logger" logger

288 runProcess node $ register "logger" logger

293 where

289 where

290+ fork = forkProcess node

291+

294− loop = do

292+ loop = do

295− receiveWait

293+ receiveWait

296− [ match $ \((time, pid, string) ::(String, ProcessId, String)) -> do

294+ [ match $ \((time, pid, string) ::(String, ProcessId, String)) -> do

297− liftIO . hPutStrLn stderr $ time ++ " " ++ show pid ++ ": " ++ string

295+ liftIO . hPutStrLn stderr $ time ++ " " ++ show pid ++ ": " ++ string

298− loop

296+ loop

299− , match $ \((time, string) :: (String, String)) -> do

297+ , match $ \((time, string) :: (String, String)) -> do

300− -- this is a 'trace' message from the local node tracer

298+ -- this is a 'trace' message from the local node tracer

301− liftIO . hPutStrLn stderr $ time ++ " [trace] " ++ string

299+ liftIO . hPutStrLn stderr $ time ++ " [trace] " ++ string

302− loop

300+ loop

303− , match $ \(ch :: SendPort ()) -> -- a shutdown request

301+ , match $ \(ch :: SendPort ()) -> -- a shutdown request

304− sendChan ch ()

302+ sendChan ch ()

305− ]

303+ ]

306

304

307-- | Force-close a local node

305-- | Force-close a local node

308--

306--

@@ -368,12 +366,13 @@ forkProcess node proc = modifyMVar (localState node) startProcess

368 }

366 }

369 return (tid', lproc)

367 return (tid', lproc)

370

368

371−

372 -- see note [tracer/forkProcess races]

369 -- see note [tracer/forkProcess races]

373− trace node (TraceEvSpawned pid)

370+ trace node (MxSpawned pid)

374

371

375 if lpidCounter lpid == maxBound

372 if lpidCounter lpid == maxBound

376 then do

373 then do

374+ -- TODO: this doesn't look right at all - how do we know

375+ -- that newUnique represents a process id that is available!?

377 newUnique <- randomIO

376 newUnique <- randomIO

378 return ( (localProcessWithId lpid ^= Just lproc)

377 return ( (localProcessWithId lpid ^= Just lproc)

379 . (localPidCounter ^= firstNonReservedProcessId)

378 . (localPidCounter ^= firstNonReservedProcessId)

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

401--

400--

402-- Our startTracing function uses forkProcess to start the trace controller

401-- Our startTracing function uses forkProcess to start the trace controller

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

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

404−-- process has started. This is harmless, as the localTracer is not updated

403+-- process has started. This is harmless, as the localEventBus is not updated

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

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

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

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

407--

406--

@@ -450,7 +449,7 @@ handleIncomingMessages node = go initConnectionState

450 NT.ConnectionOpened cid rel theirAddr ->

449 NT.ConnectionOpened cid rel theirAddr ->

451 if rel == NT.ReliableOrdered

450 if rel == NT.ReliableOrdered

452 then

451 then

453− trace node (TraceEvConnected cid)

452+ trace node (MxConnected cid theirAddr)

454 >> go (

453 >> go (

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

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

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

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

@@ -466,7 +465,7 @@ handleIncomingMessages node = go initConnectionState

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

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

467 let msg = payloadToMessage payload

466 let msg = payloadToMessage payload

468 enqueue queue msg -- 'enqueue' is strict

467 enqueue queue msg -- 'enqueue' is strict

469− trace node (TraceEvReceived pid msg)

468+ trace node (MxReceived pid msg)

470 go st

469 go st

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

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

472 mChan <- deRefWeak chan'

471 mChan <- deRefWeak chan'

@@ -515,7 +514,7 @@ handleIncomingMessages node = go initConnectionState

515 Nothing ->

514 Nothing ->

516 invalidRequest cid st

515 invalidRequest cid st

517 Just (src, _) -> do

516 Just (src, _) -> do

518− trace node (TraceEvDisconnected cid)

517+ trace node (MxDisconnected cid src)

519 go ( (incomingAt cid ^= Nothing)

518 go ( (incomingAt cid ^= Nothing)

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

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

521 $ st

520 $ st

@@ -607,15 +606,15 @@ instance Show ProcessKillException where

607-- Tracing/Debugging --

606-- Tracing/Debugging --

608--------------------------------------------------------------------------------

607--------------------------------------------------------------------------------

609

608

610−-- [Issue #104]

609+-- [Issue #104 / DP-13]

611

610

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

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

613traceNotifyDied node ident reason =

612traceNotifyDied node ident reason =

614 -- TODO: sendPortDied notifications

613 -- TODO: sendPortDied notifications

615 liftIO $ withLocalTracer node $ \t ->

614 liftIO $ withLocalTracer node $ \t ->

616 case ident of

615 case ident of

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

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

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

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

619 _ -> return ()

618 _ -> return ()

620

619

621traceEventFmtIO :: LocalNode

620traceEventFmtIO :: LocalNode

@@ -625,11 +624,11 @@ traceEventFmtIO :: LocalNode

625traceEventFmtIO node fmt args =

624traceEventFmtIO node fmt args =

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

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

627

626

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

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

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

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

630

629

631−withLocalTracer :: LocalNode -> (Tracer -> IO ()) -> IO ()

630+withLocalTracer :: LocalNode -> (MxEventBus -> IO ()) -> IO ()

632−withLocalTracer node act = act (localTracer node)

631+withLocalTracer node act = act (localEventBus node)

633

632

634--------------------------------------------------------------------------------

633--------------------------------------------------------------------------------

635-- Core functionality --

634-- Core functionality --

@@ -796,7 +795,7 @@ ncEffectSpawn pid cProc ref = do

796 mProc <- unClosure cProc

795 mProc <- unClosure cProc

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

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

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

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

799− -- TODO:

798+ -- TODO:

800 let proc = case mProc of

799 let proc = case mProc of

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

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

802 Right p -> p

801 Right p -> p

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

828 do modify' $ registeredHereFor label ^= mPid

827 do modify' $ registeredHereFor label ^= mPid

829 updateRemote node currentVal mPid

828 updateRemote node currentVal mPid

830 case mPid of

829 case mPid of

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

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

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

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

833 liftIO $ sendMessage node

832 liftIO $ sendMessage node

834 (NodeIdentifier (localNodeId node))

833 (NodeIdentifier (localNodeId node))

835 (ProcessIdentifier from)

834 (ProcessIdentifier from)

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

897ncEffectLocalSend node to msg =

896ncEffectLocalSend node to msg =

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

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

899 enqueue (processQueue p) msg

898 enqueue (processQueue p) msg

900− trace node (TraceEvReceived to msg)

899+ trace node (MxReceived to msg)

901

900

902-- [Issue #DP-20]

901-- [Issue #DP-20]

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

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

tests/TestMx.hsnew file+213

@@ -0,0 +1,213 @@

1+{-# LANGUAGE DeriveGeneric #-}

2+module Main where

4+import Control.Concurrent (threadDelay)

5+import Control.Distributed.Process

6+import Control.Distributed.Process.Management

7+ ( MxEvent(..)

8+ , MxAgentId(..)

9+ , mxAgent

10+ , mxSink

11+ , mxReady

12+ , liftMX

13+ , mxGetLocal

14+ , mxSetLocal

15+ , mxNotify

16+ , mxBroadcast

17+ , mxGetId

18+ , mxGet

19+ , mxSet

20+ , mxClear

21+ , mxPurgeTable

22+ , mxDropTable

23+ )

24+import Control.Distributed.Process.Node

25+ ( LocalNode

26+ )

27+import Data.Binary

28+import Data.List (find)

29+import Data.Maybe (isJust)

30+import Data.Typeable

31+import GHC.Generics

32+import qualified Network.Transport as NT

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

34+import Prelude hiding (catch, log)

35+#endif

36+

37+import Test.Framework

38+ ( Test

39+ , testGroup

40+ )

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

42+import TestUtils

43+

44+data Publish = Publish

45+ deriving (Typeable, Generic, Eq)

46+

47+instance Binary Publish where

48+

49+testAgentBroadcast :: TestResult (Maybe ()) -> Process ()

50+testAgentBroadcast result = do

51+ (resultSP, resultRP) <- newChan :: Process (SendPort (), ReceivePort ())

52+

53+ publisher <- mxAgent (MxAgentId "publisher-agent") () [

54+ mxSink $ \() -> mxBroadcast Publish >> mxReady

55+ ]

56+

57+ consumer <- mxAgent (MxAgentId "consumer-agent") () [

58+ mxSink $ \Publish -> (liftMX $ sendChan resultSP ()) >> mxReady

59+ ]

60+

61+ mxNotify ()

62+ -- once the publisher has seen our message, it will broadcast the Publish

63+ stash result =<< receiveChanTimeout 1000000 resultRP

64+

65+ kill publisher "finished"

66+ kill consumer "finished"

67+

68+testAgentEventHandling :: TestResult Bool -> Process ()

69+testAgentEventHandling result = do

70+ let initState = [] :: [MxEvent]

71+ agentPid <- mxAgent (MxAgentId "lifecycle-listener-agent") initState [

72+ (mxSink $ \ev -> do

73+ st <- mxGetLocal

74+ let act =

75+ case ev of

76+ (MxSpawned _) -> mxSetLocal (ev:st)

77+ (MxProcessDied _ _) -> mxSetLocal (ev:st)

78+ _ -> return ()

79+ act >> mxReady),

80+ (mxSink $ \(ev, sp :: SendPort Bool) -> do

81+ st <- mxGetLocal

82+ let found =

83+ case ev of

84+ MxSpawned p ->

85+ isJust $ find (\ev' ->

86+ case ev' of

87+ (MxSpawned p') -> p' == p

88+ _ -> False) st

89+ MxProcessDied p r ->

90+ isJust $ find (\ev' ->

91+ case ev' of

92+ (MxProcessDied p' r') -> p' == p && r == r'

93+ _ -> False) st

94+ _ -> False

95+ liftMX $ sendChan sp found

96+ mxReady)

97+ ]

98+

99+ _ <- monitor agentPid

100+ (sp, rp) <- newChan

101+ pid <- spawnLocal $ sendChan sp ()

102+ () <- receiveChan rp

103+

104+ -- By waiting for a monitor notification, we have a

105+ -- higher probably that the agent has seen

106+ monitor pid

107+ receiveWait [ match (\(ProcessMonitorNotification _ _ _) -> return ()) ]

108+

109+ (replyTo, reply) <- newChan :: Process (SendPort Bool, ReceivePort Bool)

110+ mxNotify (MxSpawned pid, replyTo)

111+ mxNotify (MxProcessDied pid DiedNormal, replyTo)

112+

113+ seenAlive <- receiveChan reply

114+ seenDead <- receiveChan reply

115+

116+ stash result $ seenAlive && seenDead

117+

118+testAgentPublication :: TestResult (Maybe Publish) -> Process ()

119+testAgentPublication result = do

120+ (syncChan, rp) <- newChan

121+ let ourId = MxAgentId "publication-agent"

122+ agentPid <- mxAgent ourId () [

123+ mxSink $ \() -> do

124+ selfId <- mxGetId

125+ liftMX $ mxSet selfId "publish" Publish >> sendChan syncChan ()

126+ mxReady

127+ ]

128+

129+ mxNotify ()

130+ () <- receiveChan rp

131+

132+ stash result =<< mxGet ourId "publish"

133+

134+ mref <- monitor agentPid

135+ kill agentPid "finished"

136+ receiveWait [

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

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

139+ ]

140+

141+testAgentTableClear :: Int -> TestResult (Maybe Int, Maybe Int) -> Process ()

142+testAgentTableClear val result =

143+ let tId = (MxAgentId "agent-1")

144+ tKey = "key-1" in do

145+ mxSet tId tKey val

146+ get1 <- mxGet tId tKey

147+ mxClear tId tKey

148+ get2 <- mxGet tId tKey

149+ stash result (get1, get2)

150+

151+testAgentTablePurge :: TestResult (Maybe Int) -> Process ()

152+testAgentTablePurge result =

153+ let tId = MxAgentId "agent-2"

154+ tKey = "key-2" in do

155+ mxSet tId tKey (12345 :: Int)

156+ mxPurgeTable tId

157+ stash result =<< mxGet tId tKey

158+

159+testAgentTableDelete :: Int

160+ -> TestResult (Maybe Int, Maybe Int, Maybe Int)

161+ -> Process ()

162+testAgentTableDelete val result =

163+ let tId = (MxAgentId "agent-3")

164+ tKey = "key-3" in do

165+ mxSet tId tKey val

166+ get1 <- mxGet tId tKey

167+ mxDropTable tId

168+ get2 <- mxGet tId tKey

169+ mxSet tId tKey val

170+ get3 <- mxGet tId tKey

171+ stash result (get1, get2, get3)

172+

173+tests :: LocalNode -> IO [Test]

174+tests node1 = do

175+ return [

176+ testGroup "Mx Agents" [

177+ testCase "Event Handling"

178+ (delayedAssertion

179+ "expected True, but events where not as expected"

180+ node1 True testAgentEventHandling)

181+ , testCase "Inter-Agent Broadcast"

182+ (delayedAssertion

183+ "expected (Just ()), but no broadcast was received"

184+ node1 (Just ()) testAgentBroadcast)

185+ ],

186+ testGroup "Mx Global Properties" [

187+ testCase "Global Property Publication"

188+ (delayedAssertion

189+ "expected (Just Publish), but no table entry was found"

190+ node1 (Just Publish) testAgentPublication)

191+ , testCase "Clearing Global Properties"

192+ (delayedAssertion

193+ "expected (Just 1024, Nothing): invalid table entry found!"

194+ node1 (Just 1024, Nothing) (testAgentTableClear 1024))

195+ , testCase "Purging Global Tables"

196+ (delayedAssertion

197+ "expected Nothing, but a table entry was found"

198+ node1 Nothing testAgentTablePurge)

199+ , testCase "Deleting and (Re)Creating Global Tables"

200+ (delayedAssertion

201+ "expected (Just 15, Nothing, Just 15): invalid table entry found!"

202+ node1 (Just 15, Nothing, Just 15) (testAgentTableDelete 15))

203+ ]]

204+

205+statsTests :: NT.Transport -> IO [Test]

206+statsTests _ = do

207+ mkNode "8080" >>= tests >>= return

208+

209+main :: IO ()

210+main = do

211+ testMain $ statsTests

212+ threadDelay 100000

213+

tests/TestStats.hs+9−40

@@ -11,47 +11,22 @@ import Control.Concurrent.MVar

11import Control.Distributed.Process

12import Control.Distributed.Process.Node

13 ( forkProcess

14− , newLocalNode

15− , initRemoteTable

16− , closeLocalNode

17 , LocalNode)

14 , LocalNode)

18import Data.Binary()

15import Data.Binary()

19import Data.Typeable()

16import Data.Typeable()

20−import Network.Transport.TCP

17+import qualified Network.Transport as NT

18+

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

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

22import Prelude hiding (catch)

20import Prelude hiding (catch)

23#endif

21#endif

22+

24import Test.Framework

23import Test.Framework

25 ( Test

24 ( Test

26− , defaultMain

27 , testGroup

25 , testGroup

28 )

26 )

29import Test.HUnit (Assertion)

27import Test.HUnit (Assertion)

30−import Test.HUnit.Base (assertBool)

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

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

32−

29+import TestUtils

33−-- these utilities have been cribbed from distributed-process-platform

34−-- we should really find a way to share them...

35−

36−-- | A mutable cell containing a test result.

37−type TestResult a = MVar a

38−

39−delayedAssertion :: (Eq a) => String -> LocalNode -> a ->

40− (TestResult a -> Process ()) -> Assertion

41−delayedAssertion note localNode expected testProc = do

42− result <- newEmptyMVar

43− _ <- forkProcess localNode $ testProc result

44− assertComplete note result expected

45−

46−assertComplete :: (Eq a) => String -> MVar a -> a -> IO ()

47−assertComplete msg mv a = do

48− b <- takeMVar mv

49− assertBool msg (a == b)

50−

51−stash :: TestResult a -> a -> Process ()

52−stash mvar x = liftIO $ putMVar mvar x

53−

54−------

56testLocalDeadProcessInfo :: TestResult (Maybe ProcessInfo) -> Process ()

31testLocalDeadProcessInfo :: TestResult (Maybe ProcessInfo) -> Process ()

57testLocalDeadProcessInfo result = do

32testLocalDeadProcessInfo result = do

@@ -149,16 +124,10 @@ tests node1 = do

149 (testRemoteLiveProcessInfo node1)

124 (testRemoteLiveProcessInfo node1)

150 ] ]

125 ] ]

151

126

152−mkNode :: String -> IO LocalNode

127+statsTests :: NT.Transport -> IO [Test]

153−mkNode port = do

128+statsTests _ = do

154− Right (transport1, _) <- createTransportExposeInternals

129+ mkNode "8080" >>= tests >>= return

155− "127.0.0.1" port defaultTCPParameters

156− newLocalNode transport1 initRemoteTable

157

130

158main :: IO ()

131main :: IO ()

159−main = do

132+main = testMain $ statsTests

160− node1 <- mkNode "8081"

133+

161− testData <- tests node1

162− defaultMain testData

163− closeLocalNode node1

164− return ()

tests/TestTracing.hs+38−95

@@ -12,63 +12,25 @@ 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

18− , initRemoteTable

19 , closeLocalNode

20 , closeLocalNode

20 , LocalNode)

21 , LocalNode)

21−import Control.Exception as Ex (catch)

22−import Network.Transport.TCP

22+import qualified Network.Transport as NT

23+

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

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

24import Prelude hiding (catch, log)

25import Prelude hiding (catch, log)

25#endif

26#endif

27+

26import Test.Framework

28import Test.Framework

27 ( Test

29 ( Test

28− , defaultMain

29 , testGroup

30 , testGroup

30 )

31 )

31−import Test.HUnit (Assertion)

32−import Test.HUnit.Base (assertBool)

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

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

34−

33+import TestUtils

35−import System.Environment (getEnv)

36−import System.Posix.Env

37− ( setEnv

38− )

39−

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

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

42−

43−-- | A mutable cell containing a test result.

44−type TestResult a = MVar a

45−

46−delayedAssertion :: (Eq a)

47− => String

48− -> LocalNode

49− -> a

50− -> (TestResult a -> Process ())

51− -> MVar ()

52− -> Assertion

53−delayedAssertion note localNode expected testProc lock = do

54− result <- newEmptyMVar

55− _ <- forkProcess localNode $ do

56− acquire lock

57− finally (testProc result)

58− (release lock)

59− assertComplete note result expected

60− where acquire lock' = liftIO $ takeMVar lock'

61− release lock' = liftIO $ putMVar lock' ()

62−

63−assertComplete :: (Eq a) => String -> MVar a -> a -> IO ()

64−assertComplete msg mv a = do

65− b <- takeMVar mv

66− assertBool msg (a == b)

67−

68−stash :: TestResult a -> a -> Process ()

69−stash mvar x = liftIO $ putMVar mvar x

70−

71−------

73testSpawnTracing :: TestResult Bool -> Process ()

35testSpawnTracing :: TestResult Bool -> Process ()

74testSpawnTracing result = do

36testSpawnTracing result = do

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

81 evDied <- liftIO $ newEmptyMVar

43 evDied <- liftIO $ newEmptyMVar

82 tracer <- startTracer $ \ev -> do

44 tracer <- startTracer $ \ev -> do

83 case ev of

45 case ev of

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

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

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

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

86 _ -> return ()

48 _ -> return ()

88 (sp, rp) <- newChan

50 (sp, rp) <- newChan

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

115 withTracer

77 withTracer

116 (\ev ->

78 (\ev ->

117 case ev of

79 case ev of

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

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

119 _ -> return ()) $ do

81 _ -> return ()) $ do

120 (sp, rp) <- newChan

82 (sp, rp) <- newChan

121 send pid sp

83 send pid sp

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

137 withTracer

99 withTracer

138 (\ev ->

100 (\ev ->

139 case ev of

101 case ev of

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

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

141 _ -> return ()) $ do

103 _ -> return ()) $ do

142 (sp, rp) <- newChan

104 (sp, rp) <- newChan

143 send pid sp

105 send pid sp

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

155 withTracer

117 withTracer

156 (\ev ->

118 (\ev ->

157 case ev of

119 case ev of

158− (TraceEvSent to from msg) -> do

120+ (MxSent to from msg) -> do

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

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

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

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

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

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

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

180 withTracer

142 withTracer

181 (\ev ->

143 (\ev ->

182 case ev of

144 case ev of

183− TraceEvRegistered p s ->

145+ MxRegistered p s ->

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

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

185 _ ->

147 _ ->

186 return ()) $ do

148 return ()) $ do

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

207 withTracer

169 withTracer

208 (\ev ->

170 (\ev ->

209 case ev of

171 case ev of

210− TraceEvUnRegistered p n -> do

172+ MxUnRegistered p n -> do

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

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

212 send pid ()

174 send pid ()

213 _ ->

175 _ ->

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

241 withTracer

203 withTracer

242 (\ev ->

204 (\ev ->

243 case ev of

205 case ev of

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

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

245− _ -> return ())

207+ _ -> return ())

246 ( do {

208 ( do {

247 recv <- liftIO $ newEmptyMVar

209 recv <- liftIO $ newEmptyMVar

248 ; withTracer

210 ; withTracer

249 (\ev' ->

211 (\ev' ->

250 case ev' of

212 case ev' of

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

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

252 _ -> return ())

214 _ -> return ())

253 ( do {

215 ( do {

254 user <- liftIO $ newEmptyMVar

216 user <- liftIO $ newEmptyMVar

255 ; withTracer

217 ; withTracer

256 (\ev'' ->

218 (\ev'' ->

257 case ev'' of

219 case ev'' of

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

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

259 _ -> return ())

221 _ -> return ())

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

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

261 ; liftIO $ takeMVar recv

223 ; liftIO $ takeMVar recv

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

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

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

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

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

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

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

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

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

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

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

281 pid <- liftIO $ forkProcess node2 $ do

243 pid <- liftIO $ forkProcess node2 $ do

282 logRelay <- spawnLocal $ relay log'

244 logRelay <- spawnLocal $ relay log'

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

293 withTracer

255 withTracer

294 (\ev ->

256 (\ev ->

295 case ev of

257 case ev of

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

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

297 _ -> return ()) $ do

259 _ -> return ()) $ do

298 relayPid <- startTraceRelay nid

260 relayPid <- startTraceRelay nid

299 liftIO $ threadDelay 1000000

261 liftIO $ threadDelay 1000000

@@ -328,65 +290,46 @@ tests node1 = do

328 -- various tracers will race with one another and

290 -- various tracers will race with one another and

329 -- we'll get garbage results (or worse, deadlocks)

291 -- we'll get garbage results (or worse, deadlocks)

330 lock <- liftIO $ newMVar ()

292 lock <- liftIO $ newMVar ()

331− enabled <- checkTraceEnabled

293+ return [

332− case enabled of

333− True ->

334− return [

335− testGroup "Tracing" [

294+ testGroup "Tracing" [

336 testCase "Spawn Tracing"

295 testCase "Spawn Tracing"

337− (delayedAssertion

296+ (synchronisedAssertion

338 "expected dead process-info to be ProcessInfoNone"

297 "expected dead process-info to be ProcessInfoNone"

339 node1 True testSpawnTracing lock)

298 node1 True testSpawnTracing lock)

340 , testCase "Recv Tracing (Explicit Pid)"

299 , testCase "Recv Tracing (Explicit Pid)"

341− (delayedAssertion

300+ (synchronisedAssertion

342 "expected a recv trace for the supplied pid"

301 "expected a recv trace for the supplied pid"

343 node1 True testTraceRecvExplicitPid lock)

302 node1 True testTraceRecvExplicitPid lock)

344 , testCase "Recv Tracing (Named Pid)"

303 , testCase "Recv Tracing (Named Pid)"

345− (delayedAssertion

304+ (synchronisedAssertion

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

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

347 node1 True testTraceRecvNamedPid lock)

306 node1 True testTraceRecvNamedPid lock)

348 , testCase "Trace Send(er)"

307 , testCase "Trace Send(er)"

349− (delayedAssertion

308+ (synchronisedAssertion

350 "expected a 'send' trace with the requisite fields set"

309 "expected a 'send' trace with the requisite fields set"

351 node1 True testTraceSending lock)

310 node1 True testTraceSending lock)

352 , testCase "Trace Registration"

311 , testCase "Trace Registration"

353− (delayedAssertion

312+ (synchronisedAssertion

354 "expected a 'registered' trace"

313 "expected a 'registered' trace"

355 node1 True testTraceRegistration lock)

314 node1 True testTraceRegistration lock)

356 , testCase "Trace Unregistration"

315 , testCase "Trace Unregistration"

357− (delayedAssertion

316+ (synchronisedAssertion

358 "expected an 'unregistered' trace"

317 "expected an 'unregistered' trace"

359 node1 True testTraceUnRegistration lock)

318 node1 True testTraceUnRegistration lock)

360 , testCase "Trace Layering"

319 , testCase "Trace Layering"

361− (delayedAssertion

320+ (synchronisedAssertion

362− "expected blah"

321+ "expected blah"

363− node1 () testTraceLayering lock)

322+ node1 () testTraceLayering lock)

364 , testCase "Remote Trace Relay"

323 , testCase "Remote Trace Relay"

365− (delayedAssertion

324+ (synchronisedAssertion

366− "expected blah"

325+ "expected blah"

367− node1 True testRemoteTraceRelay lock)

326+ node1 True testRemoteTraceRelay lock)

368 ] ]

327 ] ]

369− False ->

370− return []

371

328

372−checkTraceEnabled :: IO Bool

329+timerTests :: NT.Transport -> IO [Test]

373−checkTraceEnabled =

330+timerTests _ = do

374− Ex.catch (getEnv "DISTRIBUTED_PROCESS_TRACE_ENABLED" >> return True)

331+ mkNode "8080" >>= tests >>= return

375− (\(_ :: IOError) -> return False)

376−

377−

378−mkNode :: String -> IO LocalNode

379−mkNode port = do

380− Right (transport1, _) <- createTransportExposeInternals

381− "127.0.0.1" port defaultTCPParameters

382− newLocalNode transport1 initRemoteTable

383

332

384main :: IO ()

333main :: IO ()

385−main = do

334+main = testMain $ timerTests

386− setEnv "DISTRIBUTED_PROCESS_TRACE_ENABLED" "true" True

387− node1 <- mkNode "8081"

388− testData <- tests node1

389− defaultMain testData

390− closeLocalNode node1

391− return ()

392

335

tests/TestUtils.hsnew file+75

@@ -0,0 +1,75 @@

1+-- | General testing support. Just @import TestUtils@ in your test module.

2+module TestUtils where

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

5+import Prelude hiding (catch)

6+#endif

7+import Control.Concurrent.MVar

8+ ( MVar

9+ , newEmptyMVar

10+ , takeMVar

11+ , putMVar

12+ )

13+

14+import Control.Distributed.Process

15+import Control.Distributed.Process.Node

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

17+import Test.HUnit (Assertion)

18+import Test.HUnit.Base (assertBool)

19+import Test.Framework (Test, defaultMain)

20+

21+import Network.Transport.TCP

22+import qualified Network.Transport as NT

23+

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

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

26+

27+mkNode :: String -> IO LocalNode

28+mkNode port = do

29+ Right (transport1, _) <- createTransportExposeInternals

30+ "127.0.0.1" port defaultTCPParameters

31+ newLocalNode transport1 initRemoteTable

32+

33+-- | Given a @builder@ function, make and run a test suite on a single transport

34+testMain :: (NT.Transport -> IO [Test]) -> IO ()

35+testMain builder = do

36+ Right (transport, _) <- createTransportExposeInternals

37+ "127.0.0.1" "10001" defaultTCPParameters

38+ testData <- builder transport

39+ defaultMain testData

40+

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

42+type TestResult a = MVar a

43+

44+synchronisedAssertion :: (Eq a)

45+ => String

46+ -> LocalNode

47+ -> a

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

49+ -> MVar ()

50+ -> Assertion

51+synchronisedAssertion note localNode expected testProc lock = do

52+ result <- newEmptyMVar

53+ _ <- forkProcess localNode $ do

54+ acquire lock

55+ finally (testProc result)

56+ (release lock)

57+ assertComplete note result expected

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

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

60+

61+delayedAssertion :: (Eq a) => String -> LocalNode -> a ->

62+ (TestResult a -> Process ()) -> Assertion

63+delayedAssertion note localNode expected testProc = do

64+ result <- newEmptyMVar

65+ _ <- forkProcess localNode $ testProc result

66+ assertComplete note result expected

67+

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

69+assertComplete msg mv a = do

70+ b <- takeMVar mv

71+ assertBool msg (a == b)

72+

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

74+stash mvar x = liftIO $ putMVar mvar x

75+