fad3a08 · distributed-process

.gitignore+1

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

1111*.ps

1212*.swp

1313cabal-dev

14+cabal-dev*

1415.DS_Store

distributed-process.cabal+48−7

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

6464 Control.Distributed.Process.Closure,

6565 Control.Distributed.Process.Debug,

6666 Control.Distributed.Process.Internal.Types,

67+ Control.Distributed.Process.Management,

6768 Control.Distributed.Process.Node,

6869 Control.Distributed.Process.Serializable,

6970 Control.Distributed.Process.UnsafePrimitives

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

7778 Control.Distributed.Process.Internal.StrictMVar,

7879 Control.Distributed.Process.Internal.WeakTQueue,

7980 Control.Distributed.Process.Internal.StrictContainerAccessors,

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

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

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

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

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

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

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

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

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

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

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

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

8489 Extensions: RankNTypes,

8590 ScopedTypeVariables,

8691 FlexibleInstances,

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

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

164169 HS-Source-Dirs: tests

165170

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+

166202Test-Suite TestTracing

167203 Type: exitcode-stdio-1.0

168204 Main-Is: TestTracing.hs

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

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

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

182218 unix >= 2.5.0.0

183− Extensions: CPP,

219+ Extensions: RankNTypes,

184220 ScopedTypeVariables,

221+ FlexibleInstances,

222+ UndecidableInstances,

223+ ExistentialQuantification,

224+ GADTs,

225+ GeneralizedNewtypeDeriving,

185226 DeriveDataTypeable,

186− GeneralizedNewtypeDeriving

227+ CPP,

228+ BangPatterns

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

188230 HS-Source-Dirs: tests

189231

190−

191232Executable distributed-process-throughput

192233 if flag(benchmarks)

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

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

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

1717--

1818-- [Enabling Tracing]

1919--

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

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

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

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

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

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

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

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

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

29−-- published.

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

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

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

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

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

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

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

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

3028--

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

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

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

3735--

3836-- [Working with multiple tracer processes]

3937--

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

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

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

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

4442--

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

4745-- 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

4947-- previous tracer process.

5048--

5149-- [Built in tracers]

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

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

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

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

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

54+-- the 'traceEnable' function.

5755--

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

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

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

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

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

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

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

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

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

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

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

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

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

6563--

6664-- 1. @DISTRIBUTED_PROCESS_TRACE_FILE@

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

6866-- 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

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

71−-- disabled.

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

68+-- be selected.

7269--

7370-- 2. @DISTRIBUTED_PROCESS_TRACE_CONSOLE@

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

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

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

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

8380-- (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

85−-- the console or eventlog tracers instead.

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

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

8685--

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

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

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

9392module Control.Distributed.Process.Debug

9493 ( -- * Exported Data Types

9594 TraceArg(..)

96− , TraceEvent(..)

9795 , TraceFlags(..)

9896 , TraceSubject(..)

9997 -- * Configuring Tracing

100− , isTracingEnabled

10198 , enableTrace

10299 , enableTraceAsync

103100 , disableTrace

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

147144 , LocalProcess(..)

148145 , ProcessMonitorNotification(..)

149146 )

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

151151 ( TraceArg(..)

152− , TraceEvent(..)

153152 , TraceFlags(..)

154153 , TraceSubject(..)

155154 , defaultTraceFlags

156155 )

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

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

158157 ( systemLoggerTracer

159158 , logfileTracer

160159 , eventLogTracer

161160 )

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

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

163162 ( withRegisteredTracer

164163 , enableTrace

165164 , enableTraceAsync

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

167166 , setTraceFlags

168167 , setTraceFlagsAsync

169168 , getTraceFlags

170− , isTracingEnabled

171169 , traceOn

172170 , traceOff

173171 , traceOnly

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

175173 , traceLogFmt

176174 , traceMessage

177175 )

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

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

179177import Control.Distributed.Process.Node

180178

181179import Control.Exception (SomeException)

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

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

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

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

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

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

204202startTracer handler = do

205203 withRegisteredTracer $ \pid -> do

206204 node <- processNode <$> ask

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

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

213211-- in case of failure).

214212withTracer :: forall a.

215− (TraceEvent -> Process ())

213+ (MxEvent -> Process ())

216214 -> Process a

217215 -> Process (Either SomeException a)

218216withTracer handler proc = do

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

227225 Nothing -> return ()

228226 Just _ -> do

229227 ref <- monitor tracer

230− send tracer TraceEvDisable

228+ send tracer MxTraceDisable

231229 receiveWait [

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

233231 (\_ -> return ())

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

244242 finally (setTraceFlags flags >> try proc)

245243 (setTraceFlags oldFlags)

246244

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

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

248246traceProxy pid act = do

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

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

250248 case ev of

251− (TraceEvTakeover _) -> return False

252− TraceEvDisable -> die "disabled"

249+ (MxTraceTakeover _) -> return False

250+ MxTraceDisable -> die "disabled"

253251 _ -> act ev >> return True

254252

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

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

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

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

266264--

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

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

270267stopTracer :: Process ()

271268stopTracer =

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

276273 basePid <- whereis "tracer.initial"

277274 case basePid == (Just pid) of

278275 True -> return ()

279− False -> send pid TraceEvDisable

276+ False -> send pid MxTraceDisable

280277

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

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

66 , MatchOn(..)

77 , newCQueue

88 , enqueue

9+ , enqueueSTM

910 , dequeue

1011 , mkWeakCQueue

1112 ) where

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

5253--

5354-- Enqueue is strict.

5455enqueue :: 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

5661

5762data BlockSpec =

5863 NonBlocking

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

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

191191 , disconnect

192192 , sendCtrlMsg

193193 )

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

195198 ( traceEvent

196− , TraceEvent(..)

197199 )

198200import Control.Distributed.Process.Internal.WeakTQueue

199201 ( newTQueueIO

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

226228 msg

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

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

229− liftIO $ traceEvent (localTracer node)

230− (TraceEvSent them us (createUnencodedMessage msg))

231+ liftIO $ traceEvent (localEventBus node)

232+ (MxSent them us (createUnencodedMessage msg))

231233

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

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

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

410412 (messageToPayload msg)

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

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

413− liftIO $ traceEvent (localTracer node)

414− (TraceEvSent them us msg)

415+ liftIO $ traceEvent (localEventBus node)

416+ (MxSent them us msg)

415417

416418

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

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

551553 ]

552554 delegate pid p

553555

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.

555557relay :: ProcessId -> Process ()

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

557559

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

1717 -- * Local nodes and processes

1818 , LocalNode(..)

1919 , Tracer(..)

20+ , MxEventBus(..)

2021 , LocalNodeState(..)

2122 , LocalProcess(..)

2223 , LocalProcessState(..)

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

9899import Control.Concurrent (ThreadId)

99100import Control.Concurrent.Chan (Chan)

100101import Control.Concurrent.STM (STM)

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

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

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

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

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

198200-- Local nodes and processes --

199201--------------------------------------------------------------------------------

200202

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

202−data Tracer =

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

204− | InactiveTracer

205−

206−-- note [tracer longevity]

207−--

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

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

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

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

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

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

214−--

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

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

203+-- | Provides access to the trace controller

204+data Tracer = Tracer

205+ {

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

207+ tracerPid :: !ProcessId

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

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

210+ }

211+

212+-- | Local system management event bus state

213+data MxEventBus =

214+ MxEventBusInitialising

215+ | MxEventBus

216+ {

217+ -- | Process id of the management agent controller 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+ }

217227

218228-- | Local nodes

219229data LocalNode = LocalNode

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

225235 , localState :: !(StrictMVar LocalNodeState)

226236 -- | Channel for the node controller

227237 , localCtrlChan :: !(Chan NCMsg)

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

229− , localTracer :: !Tracer

238+ -- | Internal management event bus

239+ , localEventBus :: !MxEventBus

230240 -- | Runtime lookup table for supporting closures

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

232242 , 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 #-}

9+

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 #-}

2+

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

4+

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

5+

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 #-}

3+

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

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

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

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

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

7− ( Tracer

8− , TraceEvent(..)

9− -- * Sending Trace Data

10− , traceLog

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

7+-- tracing infrastructure's messaging fabric.

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

9+ ( -- * Sending Trace Data

10+ traceLog

1111 , traceLogFmt

1212 , traceMessage

1313 -- * Configuring A Tracer

14− , isTracingEnabled

1514 , defaultTraceFlags

1615 , enableTrace

1716 , enableTraceAsync

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

3332 , newChan

3433 , receiveChan

3534 )

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

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

3736 ( TraceArg(..)

38− , TraceEvent(..)

3937 , TraceFlags(..)

4038 , TraceOk(..)

4139 , TraceSubject(..)

4240 , defaultTraceFlags

4341 )

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

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

4543 ( traceLog

4644 , traceLogFmt

4745 , traceMessage

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

5553 , getCurrentTraceClient

5654 )

5755import Control.Distributed.Process.Internal.Types

58− ( Tracer(..)

59− , Process

56+ ( Process

6057 , ProcessId

6158 , LocalProcess(..)

62− , LocalNode(localTracer)

59+ , LocalNode(localEventBus)

6360 , SendPort

61+ , MxEventBus(..)

6462 )

6563import Control.Distributed.Process.Serializable

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

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

7270-- Main API --

7371--------------------------------------------------------------------------------

7472

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

76−-- to a 'TraceSubject'.

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

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

7775class Traceable a where

7876 uod :: [a] -> TraceSubject

7977

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

9593traceOff :: Maybe TraceSubject

9694traceOff = Nothing

9795

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−

10696-- | Enable tracing to the supplied process.

10797enableTraceAsync :: ProcessId -> Process ()

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

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

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

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

159149

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

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

161151withLocalTracer act = do

162152 node <- processNode <$> ask

163− act (localTracer node)

153+ act (localEventBus node)

164154

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

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

166156withLocalTracerSync act = do

167157 (sp, rp) <- newChan

168158 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

2− ( -- * Configuring A Tracer

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

2+ ( -- * Configuring A Remote Tracer

33 setTraceFlagsRemote

44 , startTraceRelay

55 -- * Remote Table

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

1414 , relay

1515 , nsendRemote

1616 )

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

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

1818 ( TraceFlags(..)

1919 , TraceOk(..)

2020 )

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

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

2222 ( withRegisteredTracer

2323 , enableTrace

2424 )

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

5050startTraceRelay :: NodeId -> Process ProcessId

5151startTraceRelay nodeId = do

5252 withRegisteredTracer $ \pid ->

53− -- TODO: move cpEnableTraceRemote here?

5453 spawn nodeId $ cpEnableTraceRemote pid

5554

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

11-- | Tracing/Debugging support - Trace Implementation

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

3− ( -- * API for the Node Controller

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

3+ ( -- * API for the Management Agent

44 traceController

55 -- * Built in tracers

66 , defaultTracer

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

3131 , handleMessage

3232 , matchUnknown

3333 )

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

35− ( TraceEvent(..)

36− , SetTrace(..)

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

35+ ( MxEvent(..)

3736 , Addressable(..)

37+ )

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

39+ ( SetTrace(..)

3840 , TraceSubject(..)

3941 , TraceFlags(..)

4042 , TraceOk(..)

4143 , defaultTraceFlags

4244 )

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

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

4446 ( traceOn )

4547import Control.Distributed.Process.Internal.Types

4648 ( LocalNode(..)

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

8991 ( Weak

9092 )

9193

92−data TracerState = TracerST {

94+data TracerState =

95+ TracerST

96+ {

9397 client :: !(Maybe ProcessId)

9498 , flags :: !TraceFlags

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

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

119123checkEnv :: String -> Process String

120124checkEnv s = liftIO $ getEnv s

121125

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

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

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

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

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

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

130134systemLoggerTracer :: Process ()

131135systemLoggerTracer = do

132136 node <- processNode <$> ask

133− let tr = sendTraceMsg node

137+ let tr = sendTraceLog node

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

135139 where

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

137− sendTraceMsg node ev = do

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

141+ sendTraceLog node ev = do

138142 now <- liftIO $ getCurrentTime

139− txt <- buildTxt ev

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

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

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

142145 traceMsg <- return $ NCMsg {

143146 ctrlMsgSender = ProcessIdentifier (emptyPid)

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

146149 }

147150 liftIO $ writeChan (localCtrlChan node) traceMsg

148151

149− buildTxt :: TraceEvent -> Process String

150− buildTxt (TraceEvLog msg) = return msg

151− buildTxt ev = return $ show ev

152+ buildTxt :: MxEvent -> String

153+ buildTxt (MxLog msg) = msg

154+ buildTxt ev = show ev

152155

153156eventLogTracer :: Process ()

154157eventLogTracer =

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

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

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

159162 where

160− writeTrace :: TraceEvent -> Process ()

163+ writeTrace :: MxEvent -> Process ()

161164 writeTrace ev = liftIO $ traceEventIO (show ev)

162165

163166logfileTracer :: FilePath -> Process ()

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

171174 logger h' = forever' $ do

172175 receiveWait [

173176 matchIf (\ev -> case ev of

174− TraceEvDisable -> True

175− (TraceEvTakeover _) -> True

177+ MxTraceDisable -> True

178+ (MxTraceTakeover _) -> True

176179 _ -> False)

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

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

179182 ]

180183

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

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

182185 writeTrace h ev = do

183186 liftIO $ do

184187 now <- getCurrentTime

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

190193

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

192195traceController mv = do

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

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

196+ -- See the documentation for mxAgentController for a

197+ -- commentary that explains this breach of encapsulation

195198 weakQueue <- processWeakQ <$> ask

196199 liftIO $ putMVar mv weakQueue

197200 initState <- initialState

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

199202 where

200203 traceLoop :: TracerState -> Process ()

201204 traceLoop st = do

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

205+ let client' = client st

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

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

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

205209 st' <- receiveWait [

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

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

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

210− -- figure out.

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

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

211215 case set of

212216 (TraceEnable pid) -> do

213217 -- notify the previous tracer it has been replaced

214− sendTrace st (createUnencodedMessage (TraceEvTakeover pid))

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

215219 sendOk setResp

216220 return st { client = (Just pid) }

217221 TraceDisable -> do

218− sendTrace st (createUnencodedMessage TraceEvDisable)

222+ sendTraceMsg client' (createUnencodedMessage MxTraceDisable)

219223 sendOk setResp

220224 return st { client = Nothing })

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

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

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

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

226230 , matchAny (\ev ->

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

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

228232 ]

229233 traceLoop st'

230234

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

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

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

247251

248− -- quick and dirty fold-while

249252 parseFlags :: String -> TraceFlags

250253 parseFlags s = parseFlags' s defaultTraceFlags

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

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

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

261264 | otherwise = parseFlags' xs parsedFlags

262265

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−

279266applyTraceFlags :: TraceFlags -> TracerState -> Process TracerState

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

281268

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

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

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

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

284271 let regNames' =

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

286273 (Set.singleton n)

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

288275 in do

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

290277 return st { regNames = regNames' }

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

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

292279 let f ns = case ns of

293280 Nothing -> Nothing

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

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

296283 in do

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

298285 return st { regNames = regNames' }

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

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

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

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

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

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

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

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

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

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

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

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

307294handleTrace st msg ev = do

308295 case ev of

309− (TraceEvNodeDied _ _) ->

296+ (MxNodeDied _ _) ->

310297 case (traceNodes (flags st)) of

311− True -> sendTrace st msg

298+ True -> sendTrace st ev msg

312299 False -> return ()

313− (TraceEvUser _) -> sendTrace st msg

314− (TraceEvLog _) -> sendTrace st msg

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

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

315302 _ ->

316303 case (traceConnections (flags st)) of

317− True -> sendTrace st msg

304+ True -> sendTrace st ev msg

318305 False -> return ()

319306 return st

320307

321−traceEv :: TraceEvent

308+traceEv :: MxEvent

322309 -> Message

323310 -> Maybe TraceSubject

324311 -> TracerState

325312 -> Process ()

326313traceEv _ _ Nothing _ = return ()

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

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

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

329316 node <- processNode <$> ask

330317 let p = case resolveToPid ev of

331318 Nothing -> (nullProcessId (localNodeId node))

332319 Just pid -> pid

333320 case (Set.member p pids) of

334− True -> sendTrace st msg

321+ True -> sendTrace st ev msg

335322 False -> return ()

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

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

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

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

325+ -- there are overlapping trace targets

339326 node <- processNode <$> ask

340327 let p = case resolveToPid ev of

341328 Nothing -> (nullProcessId (localNodeId node))

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

344331 Nothing -> return ()

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

346333 then return ()

347− else sendTrace st msg

334+ else sendTrace st ev msg

348335

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

350−sendTrace st msg =

351− let pid = (client st) in do

352− case pid of

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

354− Nothing -> return ()

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

337+sendTrace st ev msg = do

338+ let c = client st

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

340+ then return ()

341+ else sendTraceMsg c msg

355342

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

344+sendTraceMsg Nothing _ = return ()

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

356346

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

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

1+{-# LANGUAGE DeriveGeneric #-}

2+

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 #-}

2+

13-- | Local nodes

24--

35module Control.Distributed.Process.Node

@@ -56,7 +58,7 @@ import Control.Exception

5658 , Exception

5759 , throwTo

5860 )

59−import qualified Control.Exception as Exception (Handler(..), catch, catches, finally)

61+import qualified Control.Exception as Exception (Handler(..), catches, finally)

6062import Control.Concurrent (forkIO, myThreadId)

6163import Control.Distributed.Process.Internal.StrictMVar

6264 ( newMVar

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

106108 , LocalProcessId(..)

107109 , ProcessId(..)

108110 , LocalNode(..)

109− , Tracer(..)

111+ , MxEventBus(..)

110112 , LocalNodeState(..)

111113 , LocalProcess(..)

112114 , LocalProcessState(..)

@@ -149,19 +151,27 @@ import Control.Distributed.Process.Internal.Types

149151 , firstNonReservedProcessId

150152 , ImplicitReconnect(WithImplicitReconnect)

151153 )

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

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

155+ ( mxAgentController

156+ )

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

153162 ( remoteTable

154163 )

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

156168 ( TraceArg(..)

157− , TraceEvent(..)

158169 , traceEvent

159170 , traceLogFmt

160171 , enableTrace

161172 )

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

163− ( traceController

164− , defaultTracer

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

174+ ( MxEvent(..)

165175 )

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

167177import Control.Distributed.Process.Internal.Messaging

@@ -179,14 +189,13 @@ import Control.Distributed.Process.Internal.Primitives

179189 , catch

180190 , unwrapMessage

181191 )

182−import Control.Distributed.Process.Internal.Types (SendPort)

192+import Control.Distributed.Process.Internal.Types (SendPort, Tracer(..))

183193import qualified Control.Distributed.Process.Internal.Closure.BuiltIn as BuiltIn (remoteTable)

184194import Control.Distributed.Process.Internal.WeakTQueue (TQueue, writeTQueue)

185195import qualified Control.Distributed.Process.Internal.StrictContainerAccessors as DAC

186196 ( mapMaybe

187197 , mapDefault

188198 )

189−import System.Environment (getEnv)

190199import Unsafe.Coerce

191200

192201--------------------------------------------------------------------------------

@@ -222,10 +231,10 @@ createBareLocalNode endPoint rtable = do

222231 , localEndPoint = endPoint

223232 , localState = state

224233 , localCtrlChan = ctrlChan

225− , localTracer = InactiveTracer

234+ , localEventBus = MxEventBusInitialising

226235 , remoteTable = rtable

227236 }

228− tracedNode <- startTracing node

237+ tracedNode <- startMxAgent node

229238

230239 -- Once the NC terminates, the endpoint isn't much use,

231240 void $ forkIO $ Exception.finally (runNodeController tracedNode)

@@ -243,38 +252,27 @@ createBareLocalNode endPoint rtable = do

243252 , ctrlMsgSignal = SigShutdown

244253 }

245254

246−startTracing :: LocalNode -> IO LocalNode

247−startTracing node =

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]

255− mv <- MVar.newEmptyMVar

256− pid <- forkProcess n (traceController mv)

257− wqRef <- MVar.takeMVar mv

258− return n { localTracer = (ActiveTracer pid wqRef) }

259−

260−startDefaultTracer :: LocalNode -> IO (Maybe ProcessId)

261−startDefaultTracer node =

262− Exception.catch

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

276− InactiveTracer -> return ()

277− (ActiveTracer pid _) -> runProcess n $ register "trace.controller" pid

255+startMxAgent :: LocalNode -> IO LocalNode

256+startMxAgent node = do

257+ -- see note [tracer/forkProcess races]

258+ let fork = forkProcess node

259+ mv <- MVar.newEmptyMVar

260+ pid <- fork $ mxAgentController fork mv

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

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

263+

264+startDefaultTracer :: LocalNode -> IO ()

265+startDefaultTracer node' = do

266+ let t = localEventBus node'

267+ case t of

268+ MxEventBus _ (Tracer pid _) _ _ -> do

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

278276

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

280278startServiceProcesses :: LocalNode -> IO ()

@@ -283,26 +281,26 @@ startServiceProcesses node = do

283281 -- the default tracer first, even though it might @nsend@ to the logger

284282 -- before /that/ process has started - this is a totally harmless race

285283 -- however, so we deliberably ignore it

286− mPid <- startDefaultTracer node

287− case mPid of

288− Nothing -> return ()

289− Just pid -> runProcess node $ register "tracer.initial" pid

290−

284+ startDefaultTracer node

285+ tableCoordinatorPid <- fork $ Table.startTableCoordinator fork

286+ runProcess node $ register Table.mxTableCoordinator tableCoordinatorPid

291287 logger <- forkProcess node loop

292288 runProcess node $ register "logger" logger

293289 where

294− loop = do

295− receiveWait

296− [ match $ \((time, pid, string) ::(String, ProcessId, String)) -> do

297− liftIO . hPutStrLn stderr $ time ++ " " ++ show pid ++ ": " ++ string

298− loop

299− , match $ \((time, string) :: (String, String)) -> do

300− -- this is a 'trace' message from the local node tracer

301− liftIO . hPutStrLn stderr $ time ++ " [trace] " ++ string

302− loop

303− , match $ \(ch :: SendPort ()) -> -- a shutdown request

304− sendChan ch ()

305− ]

290+ fork = forkProcess node

291+

292+ loop = do

293+ receiveWait

294+ [ match $ \((time, pid, string) ::(String, ProcessId, String)) -> do

295+ liftIO . hPutStrLn stderr $ time ++ " " ++ show pid ++ ": " ++ string

296+ loop

297+ , match $ \((time, string) :: (String, String)) -> do

298+ -- this is a 'trace' message from the local node tracer

299+ liftIO . hPutStrLn stderr $ time ++ " [trace] " ++ string

300+ loop

301+ , match $ \(ch :: SendPort ()) -> -- a shutdown request

302+ sendChan ch ()

303+ ]

306304

307305-- | Force-close a local node

308306--

@@ -368,12 +366,13 @@ forkProcess node proc = modifyMVar (localState node) startProcess

368366 }

369367 return (tid', lproc)

370368

371−

372369 -- see note [tracer/forkProcess races]

373− trace node (TraceEvSpawned pid)

370+ trace node (MxSpawned pid)

374371

375372 if lpidCounter lpid == maxBound

376373 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!?

377376 newUnique <- randomIO

378377 return ( (localProcessWithId lpid ^= Just lproc)

379378 . (localPidCounter ^= firstNonReservedProcessId)

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

401400--

402401-- Our startTracing function uses forkProcess to start the trace controller

403402-- 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

405404-- until /after/ the initial forkProcess completes, so the first call to

406405-- traceEvent behaves as if tracing were disabled (i.e., it is ignored).

407406--

@@ -450,7 +449,7 @@ handleIncomingMessages node = go initConnectionState

450449 NT.ConnectionOpened cid rel theirAddr ->

451450 if rel == NT.ReliableOrdered

452451 then

453− trace node (TraceEvConnected cid)

452+ trace node (MxConnected cid theirAddr)

454453 >> go (

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

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

@@ -466,7 +465,7 @@ handleIncomingMessages node = go initConnectionState

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

467466 let msg = payloadToMessage payload

468467 enqueue queue msg -- 'enqueue' is strict

469− trace node (TraceEvReceived pid msg)

468+ trace node (MxReceived pid msg)

470469 go st

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

472471 mChan <- deRefWeak chan'

@@ -515,7 +514,7 @@ handleIncomingMessages node = go initConnectionState

515514 Nothing ->

516515 invalidRequest cid st

517516 Just (src, _) -> do

518− trace node (TraceEvDisconnected cid)

517+ trace node (MxDisconnected cid src)

519518 go ( (incomingAt cid ^= Nothing)

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

521520 $ st

@@ -607,15 +606,15 @@ instance Show ProcessKillException where

607606-- Tracing/Debugging --

608607--------------------------------------------------------------------------------

609608

610−-- [Issue #104]

609+-- [Issue #104 / DP-13]

611610

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

613612traceNotifyDied node ident reason =

614613 -- TODO: sendPortDied notifications

615614 liftIO $ withLocalTracer node $ \t ->

616615 case ident of

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

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

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

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

619618 _ -> return ()

620619

621620traceEventFmtIO :: LocalNode

@@ -625,11 +624,11 @@ traceEventFmtIO :: LocalNode

625624traceEventFmtIO node fmt args =

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

627626

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

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

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

630629

631−withLocalTracer :: LocalNode -> (Tracer -> IO ()) -> IO ()

632−withLocalTracer node act = act (localTracer node)

630+withLocalTracer :: LocalNode -> (MxEventBus -> IO ()) -> IO ()

631+withLocalTracer node act = act (localEventBus node)

633632

634633--------------------------------------------------------------------------------

635634-- Core functionality --

@@ -796,7 +795,7 @@ ncEffectSpawn pid cProc ref = do

796795 mProc <- unClosure cProc

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

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

799− -- TODO:

798+ -- TODO:

800799 let proc = case mProc of

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

802801 Right p -> p

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

828827 do modify' $ registeredHereFor label ^= mPid

829828 updateRemote node currentVal mPid

830829 case mPid of

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

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

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

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

833832 liftIO $ sendMessage node

834833 (NodeIdentifier (localNodeId node))

835834 (ProcessIdentifier from)

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

897896ncEffectLocalSend node to msg =

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

899898 enqueue (processQueue p) msg

900− trace node (TraceEvReceived to msg)

899+ trace node (MxReceived to msg)

901900

902901-- [Issue #DP-20]

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

tests/TestMx.hsnew file+213

@@ -0,0 +1,213 @@

1+{-# LANGUAGE DeriveGeneric #-}

2+module Main where

3+

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

1111import Control.Distributed.Process

1212import Control.Distributed.Process.Node

1313 ( forkProcess

14− , newLocalNode

15− , initRemoteTable

16− , closeLocalNode

1714 , LocalNode)

1815import Data.Binary()

1916import Data.Typeable()

20−import Network.Transport.TCP

17+import qualified Network.Transport as NT

18+

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

2220import Prelude hiding (catch)

2321#endif

22+

2423import Test.Framework

2524 ( Test

26− , defaultMain

2725 , testGroup

2826 )

2927import Test.HUnit (Assertion)

30−import Test.HUnit.Base (assertBool)

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

32−

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−------

29+import TestUtils

5530

5631testLocalDeadProcessInfo :: TestResult (Maybe ProcessInfo) -> Process ()

5732testLocalDeadProcessInfo result = do

@@ -149,16 +124,10 @@ tests node1 = do

149124 (testRemoteLiveProcessInfo node1)

150125 ] ]

151126

152−mkNode :: String -> IO LocalNode

153−mkNode port = do

154− Right (transport1, _) <- createTransportExposeInternals

155− "127.0.0.1" port defaultTCPParameters

156− newLocalNode transport1 initRemoteTable

127+statsTests :: NT.Transport -> IO [Test]

128+statsTests _ = do

129+ mkNode "8080" >>= tests >>= return

157130

158131main :: IO ()

159−main = do

160− node1 <- mkNode "8081"

161− testData <- tests node1

162− defaultMain testData

163− closeLocalNode node1

164− return ()

132+main = testMain $ statsTests

133+

tests/TestTracing.hs+38−95

@@ -12,63 +12,25 @@ import Control.Concurrent.MVar

1212 )

1313import Control.Distributed.Process

1414import Control.Distributed.Process.Debug

15+import Control.Distributed.Process.Management

16+ ( MxEvent(..)

17+ )

1518import Control.Distributed.Process.Node

1619 ( forkProcess

17− , newLocalNode

18− , initRemoteTable

1920 , closeLocalNode

2021 , LocalNode)

21−import Control.Exception as Ex (catch)

22−import Network.Transport.TCP

22+import qualified Network.Transport as NT

23+

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

2425import Prelude hiding (catch, log)

2526#endif

27+

2628import Test.Framework

2729 ( Test

28− , defaultMain

2930 , testGroup

3031 )

31−import Test.HUnit (Assertion)

32−import Test.HUnit.Base (assertBool)

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

34−

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−------

33+import TestUtils

7234

7335testSpawnTracing :: TestResult Bool -> Process ()

7436testSpawnTracing result = do

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

8143 evDied <- liftIO $ newEmptyMVar

8244 tracer <- startTracer $ \ev -> do

8345 case ev of

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

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

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

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

8648 _ -> return ()

8749

8850 (sp, rp) <- newChan

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

11577 withTracer

11678 (\ev ->

11779 case ev of

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

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

11981 _ -> return ()) $ do

12082 (sp, rp) <- newChan

12183 send pid sp

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

13799 withTracer

138100 (\ev ->

139101 case ev of

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

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

141103 _ -> return ()) $ do

142104 (sp, rp) <- newChan

143105 send pid sp

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

155117 withTracer

156118 (\ev ->

157119 case ev of

158− (TraceEvSent to from msg) -> do

120+ (MxSent to from msg) -> do

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

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

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

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

180142 withTracer

181143 (\ev ->

182144 case ev of

183− TraceEvRegistered p s ->

145+ MxRegistered p s ->

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

185147 _ ->

186148 return ()) $ do

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

207169 withTracer

208170 (\ev ->

209171 case ev of

210− TraceEvUnRegistered p n -> do

172+ MxUnRegistered p n -> do

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

212174 send pid ()

213175 _ ->

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

241203 withTracer

242204 (\ev ->

243205 case ev of

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

245− _ -> return ())

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

207+ _ -> return ())

246208 ( do {

247209 recv <- liftIO $ newEmptyMVar

248210 ; withTracer

249211 (\ev' ->

250212 case ev' of

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

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

252214 _ -> return ())

253215 ( do {

254216 user <- liftIO $ newEmptyMVar

255217 ; withTracer

256218 (\ev'' ->

257219 case ev'' of

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

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

259221 _ -> return ())

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

261223 ; liftIO $ takeMVar recv

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

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

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

278240 -- 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

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

281243 pid <- liftIO $ forkProcess node2 $ do

282244 logRelay <- spawnLocal $ relay log'

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

293255 withTracer

294256 (\ev ->

295257 case ev of

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

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

297259 _ -> return ()) $ do

298260 relayPid <- startTraceRelay nid

299261 liftIO $ threadDelay 1000000

@@ -328,65 +290,46 @@ tests node1 = do

328290 -- various tracers will race with one another and

329291 -- we'll get garbage results (or worse, deadlocks)

330292 lock <- liftIO $ newMVar ()

331− enabled <- checkTraceEnabled

332− case enabled of

333− True ->

334− return [

335− testGroup "Tracing" [

293+ return [

294+ testGroup "Tracing" [

336295 testCase "Spawn Tracing"

337− (delayedAssertion

296+ (synchronisedAssertion

338297 "expected dead process-info to be ProcessInfoNone"

339298 node1 True testSpawnTracing lock)

340299 , testCase "Recv Tracing (Explicit Pid)"

341− (delayedAssertion

300+ (synchronisedAssertion

342301 "expected a recv trace for the supplied pid"

343302 node1 True testTraceRecvExplicitPid lock)

344303 , testCase "Recv Tracing (Named Pid)"

345− (delayedAssertion

304+ (synchronisedAssertion

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

347306 node1 True testTraceRecvNamedPid lock)

348307 , testCase "Trace Send(er)"

349− (delayedAssertion

308+ (synchronisedAssertion

350309 "expected a 'send' trace with the requisite fields set"

351310 node1 True testTraceSending lock)

352311 , testCase "Trace Registration"

353− (delayedAssertion

312+ (synchronisedAssertion

354313 "expected a 'registered' trace"

355314 node1 True testTraceRegistration lock)

356315 , testCase "Trace Unregistration"

357− (delayedAssertion

316+ (synchronisedAssertion

358317 "expected an 'unregistered' trace"

359318 node1 True testTraceUnRegistration lock)

360319 , testCase "Trace Layering"

361− (delayedAssertion

362− "expected blah"

363− node1 () testTraceLayering lock)

320+ (synchronisedAssertion

321+ "expected blah"

322+ node1 () testTraceLayering lock)

364323 , testCase "Remote Trace Relay"

365− (delayedAssertion

366− "expected blah"

367− node1 True testRemoteTraceRelay lock)

324+ (synchronisedAssertion

325+ "expected blah"

326+ node1 True testRemoteTraceRelay lock)

368327 ] ]

369− False ->

370− return []

371328

372−checkTraceEnabled :: IO Bool

373−checkTraceEnabled =

374− Ex.catch (getEnv "DISTRIBUTED_PROCESS_TRACE_ENABLED" >> return True)

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

329+timerTests :: NT.Transport -> IO [Test]

330+timerTests _ = do

331+ mkNode "8080" >>= tests >>= return

383332

384333main :: IO ()

385−main = do

386− setEnv "DISTRIBUTED_PROCESS_TRACE_ENABLED" "true" True

387− node1 <- mkNode "8081"

388− testData <- tests node1

389− defaultMain testData

390− closeLocalNode node1

391− return ()

334+main = testMain $ timerTests

392335

tests/TestUtils.hsnew file+75

@@ -0,0 +1,75 @@

1+-- | General testing support. Just @import TestUtils@ in your test module.

2+module TestUtils where

3+

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+