e2d0cf7 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Split the node management event bus up from the trace controller

e2d0cf7Tim Watson13 years ago

distributed-process.cabal+11−2

compare with master

unifiedsplit

@@ -64,10 +64,13 @@ Library

64 Control.Distributed.Process.Closure,

65 Control.Distributed.Process.Debug,

66 Control.Distributed.Process.Internal.Types,

67+ Control.Distributed.Process.Management,

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

67 Control.Distributed.Process.Node,

69 Control.Distributed.Process.Node,

68 Control.Distributed.Process.Serializable,

70 Control.Distributed.Process.Serializable,

69 Control.Distributed.Process.UnsafePrimitives

71 Control.Distributed.Process.UnsafePrimitives

70 Other-modules: Control.Distributed.Process.Internal.Primitives,

72 Other-modules: Control.Distributed.Process.Internal.Primitives,

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

71 Control.Distributed.Process.Internal.Spawn,

74 Control.Distributed.Process.Internal.Spawn,

72 Control.Distributed.Process.Internal.Primitives,

75 Control.Distributed.Process.Internal.Primitives,

73 Control.Distributed.Process.Internal.CQueue,

76 Control.Distributed.Process.Internal.CQueue,

@@ -180,10 +183,16 @@ Test-Suite TestTracing

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

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

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

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

182 unix >= 2.5.0.0

185 unix >= 2.5.0.0

183− Extensions: CPP,

186+ Extensions: RankNTypes,

184 ScopedTypeVariables,

187 ScopedTypeVariables,

188+ FlexibleInstances,

189+ UndecidableInstances,

190+ ExistentialQuantification,

191+ GADTs,

192+ GeneralizedNewtypeDeriving,

185 DeriveDataTypeable,

193 DeriveDataTypeable,

186− GeneralizedNewtypeDeriving

194+ CPP,

195+ BangPatterns

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

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

188 HS-Source-Dirs: tests

197 HS-Source-Dirs: tests

189

198

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

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

6 , MatchOn(..)

7 , newCQueue

8 , enqueue

9+ , enqueueSTM

9 , dequeue

10 , dequeue

10 , mkWeakCQueue

11 , mkWeakCQueue

11 ) where

12 ) where

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

52--

53--

53-- Enqueue is strict.

54-- Enqueue is strict.

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

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

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

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

57+

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

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

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

57data BlockSpec =

62data BlockSpec =

58 NonBlocking

63 NonBlocking

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

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

226 msg

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

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

229− liftIO $ traceEvent (localTracer node)

229+ liftIO $ traceEvent (localEventBus node)

230 (TraceEvSent them us (createUnencodedMessage msg))

231

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

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

410 (messageToPayload msg)

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

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

413− liftIO $ traceEvent (localTracer node)

413+ liftIO $ traceEvent (localEventBus node)

414 (TraceEvSent them us msg)

415

416

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

@@ -3,6 +3,8 @@

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

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

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

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

7+-- tracing infrastructure's messaging fabric.

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

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

7 ( Tracer

9 ( Tracer

8 , TraceEvent(..)

10 , TraceEvent(..)

@@ -59,21 +61,23 @@ import Control.Distributed.Process.Internal.Types

59 , Process

61 , Process

60 , ProcessId

62 , ProcessId

61 , LocalProcess(..)

63 , LocalProcess(..)

62− , LocalNode(localTracer)

64+ , LocalNode(localEventBus)

63 , SendPort

65 , SendPort

66+ , MxEventBus(..)

64 )

67 )

65import Control.Distributed.Process.Serializable

68import Control.Distributed.Process.Serializable

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

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

67import Control.Monad.Reader (ask)

70import Control.Monad.Reader (ask)

72+import Data.Maybe (isJust)

69import qualified Data.Set as Set (fromList)

73import qualified Data.Set as Set (fromList)

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

75--------------------------------------------------------------------------------

72-- Main API --

76-- Main API --

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

77--------------------------------------------------------------------------------

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

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

76−-- to a 'TraceSubject'.

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

77class Traceable a where

81class Traceable a where

78 uod :: [a] -> TraceSubject

82 uod :: [a] -> TraceSubject

@@ -98,10 +102,10 @@ traceOff = Nothing

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

102-- | Determine whether or not tracing has been enabled.

99isTracingEnabled :: Process Bool

103isTracingEnabled :: Process Bool

100isTracingEnabled = do

104isTracingEnabled = do

101− node <- processNode <$> ask

105+ processNode <$> ask >>= return . isActive . tracer . localEventBus

102− case (localTracer node) of

106+ where

107+ isActive InactiveTracer = False

103− (ActiveTracer _ _) -> return True

108+ isActive (ActiveTracer _ _) = True

104− _ -> return False

105

109

106-- | Enable tracing to the supplied process.

110-- | Enable tracing to the supplied process.

107enableTraceAsync :: ProcessId -> Process ()

111enableTraceAsync :: ProcessId -> Process ()

@@ -157,22 +161,28 @@ traceLogFmt d ls = withLocalTracer $ \t -> liftIO $ Tracer.traceLogFmt t d ls

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

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

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

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

159

163

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

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

161withLocalTracer act = do

165withLocalTracer act = do

162 node <- processNode <$> ask

166 node <- processNode <$> ask

163− act (localTracer node)

167+ act (localEventBus node)

164

168

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

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

166withLocalTracerSync act = do

170withLocalTracerSync act = do

167 (sp, rp) <- newChan

171 (sp, rp) <- newChan

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

172+ withLocalTracer $ \t -> do

173+ case (tracer t) of

174+ InactiveTracer -> error "TraceDisabled"

175+ _ -> liftIO $ (act t sp)

169 TraceOk <- receiveChan rp

176 TraceOk <- receiveChan rp

170 return ()

177 return ()

171

178

172withRegisteredTracer :: (ProcessId -> Process a) -> Process a

179withRegisteredTracer :: (ProcessId -> Process a) -> Process a

173withRegisteredTracer act = do

180withRegisteredTracer act = do

174 (sp, rp) <- newChan

181 (sp, rp) <- newChan

182+ withLocalTracer $ \t -> do

183+ case (tracer t) of

184+ InactiveTracer -> error "TraceDisabled"

175− withLocalTracer $ \t -> liftIO $ Tracer.getCurrentTraceClient t sp

185+ _ -> liftIO $ Tracer.getCurrentTraceClient t sp

176 currentTracer <- receiveChan rp

186 currentTracer <- receiveChan rp

177 case currentTracer of

187 case currentTracer of

178 Nothing -> do { (Just p') <- whereis "tracer.initial"; act p' }

188 Nothing -> do { (Just p') <- whereis "tracer.initial"; act p' }

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

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

1-- | Tracing/Debugging support - Trace Implementation

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

3− ( -- * API for the Node Controller

3+ ( -- * API for the Management Agent

4 traceController

5 -- * Built in tracers

6 , defaultTracer

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

89 ( Weak

90 )

92−data TracerState = TracerST {

92+data TracerState =

93+ TracerST

94+ {

93 client :: !(Maybe ProcessId)

95 client :: !(Maybe ProcessId)

94 , flags :: !TraceFlags

96 , flags :: !TraceFlags

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

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

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

119checkEnv :: String -> Process String

121checkEnv :: String -> Process String

120checkEnv s = liftIO $ getEnv s

122checkEnv s = liftIO $ getEnv s

121

123

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

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

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

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

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

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

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

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

@@ -136,8 +138,7 @@ systemLoggerTracer = do

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

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

137 sendTraceMsg node ev = do

139 sendTraceMsg node ev = do

138 now <- liftIO $ getCurrentTime

140 now <- liftIO $ getCurrentTime

139− txt <- buildTxt ev

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

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

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

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

142 traceMsg <- return $ NCMsg {

143 traceMsg <- return $ NCMsg {

143 ctrlMsgSender = ProcessIdentifier (emptyPid)

144 ctrlMsgSender = ProcessIdentifier (emptyPid)

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

146 }

147 }

147 liftIO $ writeChan (localCtrlChan node) traceMsg

148 liftIO $ writeChan (localCtrlChan node) traceMsg

148

149

149− buildTxt :: TraceEvent -> Process String

150+ buildTxt :: TraceEvent -> String

150− buildTxt (TraceEvLog msg) = return msg

151+ buildTxt (TraceEvLog msg) = msg

151− buildTxt ev = return $ show ev

152+ buildTxt ev = show ev

152

153

153eventLogTracer :: Process ()

154eventLogTracer :: Process ()

154eventLogTracer =

155eventLogTracer =

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

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

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

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

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

247

248

248− -- quick and dirty fold-while

249 parseFlags :: String -> TraceFlags

250 parseFlags s = parseFlags' s defaultTraceFlags

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

@@ -260,22 +260,6 @@ traceController mv = do

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

261 | otherwise = parseFlags' xs parsedFlags

262

263−-- note [runtime tracer control]

264−--

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

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

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

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

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

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

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

272−-- to a single function call.

273−--

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

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

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

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

278−

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

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

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

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

281

265

@@ -334,8 +318,8 @@ traceEv ev msg (Just (TraceProcs pids)) st = do

334 True -> sendTrace st msg

318 True -> sendTrace st msg

335 False -> return ()

319 False -> return ()

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

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

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

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

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

322+ -- there are overlapping trace targets

339 node <- processNode <$> ask

323 node <- processNode <$> ask

340 let p = case resolveToPid ev of

324 let p = case resolveToPid ev of

341 Nothing -> (nullProcessId (localNodeId node))

325 Nothing -> (nullProcessId (localNodeId node))

src/Control/Distributed/Process/Internal/Trace/Types.hs+30−33

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

1-- | Tracing/Debugging support - Types

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

3− ( Tracer(..)

3+ ( MxEventBus(..)

4 , TraceEvent(..)

5 , SetTrace(..)

6 , Addressable(..)

@@ -26,13 +26,16 @@ module Control.Distributed.Process.Internal.Trace.Types

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

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

28import Control.Distributed.Process.Internal.Types

29− ( Tracer(..)

29+ ( MxEventBus(..)

30 , NodeId

31 , ProcessId

32 , DiedReason

33 , Message

34 , SendPort

35− , createUnencodedMessage

35+ , unsafeCreateUnencodedMessage

36+ )

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

38+ ( publishEvent

36 )

39 )

37import Control.Distributed.Process.Serializable

40import Control.Distributed.Process.Serializable

38import Data.Binary

41import Data.Binary

@@ -145,10 +148,10 @@ data TraceOk = TraceOk

145-- Internal/Common API --

148-- Internal/Common API --

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

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

147

150

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

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

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

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

150

153

151−traceLogFmt :: Tracer

154+traceLogFmt :: MxEventBus

152 -> String

155 -> String

153 -> [TraceArg]

156 -> [TraceArg]

154 -> IO ()

157 -> IO ()

@@ -158,43 +161,43 @@ traceLogFmt t d ls =

158 toS (TraceStr s) = s

161 toS (TraceStr s) = s

159 toS (Trace a) = show a

162 toS (Trace a) = show a

160

163

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

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

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

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

163

166

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

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

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

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

166

169

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

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

168enableTrace t p =

171enableTrace t p =

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

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

170− (TraceEnable p)))

173+ (TraceEnable p)))

171

174

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

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

173enableTraceSync t s p =

176enableTraceSync t s p =

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

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

175

178

176−disableTrace :: Tracer -> IO ()

179+disableTrace :: MxEventBus -> IO ()

177disableTrace t =

180disableTrace t =

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

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

179 TraceDisable))

182 TraceDisable))

180

183

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

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

182disableTraceSync t s =

185disableTraceSync t s =

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

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

184

187

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

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

186setTraceFlags t f =

189setTraceFlags t f =

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

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

188

191

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

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

190setTraceFlagsSync t s f =

193setTraceFlagsSync t s f =

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

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

192

195

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

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

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

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

195

198

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

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

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

200+getCurrentTraceClient t s = publishEvent t (unsafeCreateUnencodedMessage s)

198

201

199class Traceable a where

202class Traceable a where

200 uod :: [a] -> TraceSubject

203 uod :: [a] -> TraceSubject

@@ -205,12 +208,6 @@ instance Traceable ProcessId where

205instance Traceable String where

208instance Traceable String where

206 uod = TraceNames . Set.fromList

209 uod = TraceNames . Set.fromList

207

210

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

211--------------------------------------------------------------------------------

215-- Binary Instances --

212-- Binary Instances --

216--------------------------------------------------------------------------------

213--------------------------------------------------------------------------------

src/Control/Distributed/Process/Internal/Types.hs+19−16

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

17 -- * Local nodes and processes

18 , LocalNode(..)

19 , Tracer(..)

20+ , MxEventBus(..)

20 , LocalNodeState(..)

21 , LocalNodeState(..)

21 , LocalProcess(..)

22 , LocalProcess(..)

22 , LocalProcessState(..)

23 , LocalProcessState(..)

@@ -198,22 +199,24 @@ nullProcessId nid =

198-- Local nodes and processes --

199-- Local nodes and processes --

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

200--------------------------------------------------------------------------------

200

201

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

202data Tracer =

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

204− | InactiveTracer

203+ InactiveTracer

205−

204+ | ActiveTracer

206−-- note [tracer longevity]

205+ {

207−--

206+ tracerPid :: !ProcessId

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

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

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

208+ }

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

209+

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

210+-- | Required for the system management and tracing facilities

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

211+data MxEventBus =

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

212+ MxEventBusInitialising

214−--

213+ | MxEventBus

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

214+ {

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

215+ agent :: !ProcessId

216+ , tracer :: !Tracer

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

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

219+ }

217

220

218-- | Local nodes

221-- | Local nodes

219data LocalNode = LocalNode

222data LocalNode = LocalNode

@@ -226,7 +229,7 @@ data LocalNode = LocalNode

226 -- | Channel for the node controller

229 -- | Channel for the node controller

227 , localCtrlChan :: !(Chan NCMsg)

230 , localCtrlChan :: !(Chan NCMsg)

228 -- | Current active system debug/trace log

231 -- | Current active system debug/trace log

229− , localTracer :: !Tracer

232+ , localEventBus :: !MxEventBus

230 -- | Runtime lookup table for supporting closures

233 -- | Runtime lookup table for supporting closures

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

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

232 , remoteTable :: !RemoteTable

235 , remoteTable :: !RemoteTable

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

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

1+{-# LANGUAGE DeriveDataTypeable #-}

2+{-# LANGUAGE DeriveGeneric #-}

3+{-# LANGUAGE StandaloneDeriving #-}

4+{-# LANGUAGE ExistentialQuantification #-}

5+{-# LANGUAGE ScopedTypeVariables #-}

6+{-# LANGUAGE PatternGuards #-}

7+{-# LANGUAGE RecordWildCards #-}

8+{-# LANGUAGE FlexibleInstances #-}

9+{-# LANGUAGE UndecidableInstances #-}

10+

11+-----------------------------------------------------------------------------

12+-- |

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

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

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

16+--

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

18+-- Stability : experimental

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

20+--

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

22+--

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

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

25+--

26+-- [Architecture Overview]

27+--

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

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

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

31+--

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

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

34+--

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

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

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

38+--

39+--

40+--

41+-----------------------------------------------------------------------------

42+module Control.Distributed.Process.Management where

43+

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

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

46+ ( Process

47+ , ProcessId

48+ , Message

49+ , LocalProcess(..)

50+ , LocalNode(..)

51+ , MxEventBus(..)

52+ , unsafeCreateUnencodedMessage

53+ )

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

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

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

57+import Control.Monad.Reader (ask)

58+

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

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

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

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

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

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

65+mxNotify msg = do

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

67+ liftIO $ publishEvent bus $ unsafeCreateUnencodedMessage msg

68+

69+mxAgent :: (Message -> Process ()) -> Process ProcessId

70+mxAgent handler = do

71+ node <- processNode <$> ask

72+ pid <- liftIO $ mxNew (localEventBus node) handler

73+ -- registerAgent pid

74+ return pid

75+

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

@@ -0,0 +1,141 @@

1+{-# LANGUAGE RecordWildCards #-}

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

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

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

7+import Control.Concurrent.STM (atomically)

8+import Control.Concurrent.STM.TChan

9+ ( TChan

10+ , newBroadcastTChanIO

11+ , readTChan

12+ , writeTChan

13+ , dupTChan

14+ )

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

16+ ( receiveWait

17+ , matchAny

18+ , die

19+ , catches

20+ , Handler(..)

21+ )

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

23+ ( enqueueSTM

24+ , CQueue

25+ )

26+import Control.Distributed.Process.Internal.Trace.Tracer

27+ ( traceController

28+ )

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

30+ ( Process

31+ , SendPort

32+ , Message

33+ , MxEventBus(..)

34+ , Tracer(..)

35+ , LocalNode(..)

36+ , LocalProcess(..)

37+ , ProcessId

38+ , forever'

39+ )

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

41+import qualified Control.Exception as Ex (catch, finally)

42+import Control.Monad (void)

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

44+import Control.Monad.Reader (ask)

45+import GHC.Weak (Weak(Weak), deRefWeak)

46+import System.Environment (getEnv)

47+

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

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

50+-- when calling mxAgentController, since there's no other way to

51+-- avoid a circular dependency with Node.hs

52+type Fork = (Process () -> IO ProcessId)

53+

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

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

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

57+type AgentConfig =

58+ (Tracer, Weak (CQueue Message), ((Message -> Process ()) -> IO ProcessId))

59+

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

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

62+-- 'ProcessId' where possible.

63+--

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

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

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

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

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

69+-- also optimises the code path for tracing and avoids overloading the

70+-- node controller with additional routing, at the cost of a little

71+-- more complexity for us here.

72+--

73+mxAgentController :: Fork

74+ -> MVar AgentConfig

75+ -> Process ()

76+mxAgentController forkProcess mv = do

77+ node <- processNode <$> ask

78+ trc <- liftIO $ startTracing forkProcess node

79+ sigbus <- liftIO $ newBroadcastTChanIO

80+ weakQueue <- processWeakQ <$> ask

81+ liftIO $ putMVar mv (trc, weakQueue, mxAgent forkProcess sigbus)

82+ go sigbus trc forkProcess

83+ where

84+ go bus tracer fork = forever' $ do

85+ void $ receiveWait [

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

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

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

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

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

91+ -- this process to every crash, and the assumption we therefore

92+ -- make is thus:

93+ --

94+ -- 1. only ThreadKilled can tell this thread to terminate

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

96+ --

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

98+ -- be silently ignored.

99+ --

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

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

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

103+

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

105+ broadcast ch tr msg = do

106+ tmQueue <- tracerQueue tr

107+ atomicBroadcast ch tmQueue msg

108+

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

110+ tracerQueue InactiveTracer = return Nothing

111+ tracerQueue (ActiveTracer _ wQ) = deRefWeak wQ

112+

113+ atomicBroadcast :: TChan Message

114+ -> Maybe (CQueue Message)

115+ -> Message -> IO ()

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

117+ atomicBroadcast ch (Just q) msg = do

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

119+

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

121+mxAgent :: Fork -> TChan Message -> (Message -> Process ()) -> IO ProcessId

122+mxAgent fork chan handler = (atomically (dupTChan chan)) >>= fork . run handler

123+ where

124+ run :: (Message -> Process ()) -> TChan Message -> Process ()

125+ run handler' chan' = do

126+ (liftIO $ atomically $ readTChan chan') >>= handler' >> run handler' chan'

127+

128+startTracing :: Fork -> LocalNode -> IO Tracer

129+startTracing forkProcess node =

130+ -- TODO: use tryGetEnv once we drop support for GHC 7.2.x

131+ Ex.catch

132+ (getEnv "DISTRIBUTED_PROCESS_TRACE_ENABLED" >> startTracing' forkProcess node)

133+ (\(_ :: IOError) -> return InactiveTracer)

134+ where

135+ startTracing' :: Fork -> LocalNode -> IO Tracer

136+ startTracing' fork' n = do

137+ mv <- newEmptyMVar

138+ pid <- fork' $ traceController mv

139+ wQ <- liftIO $ takeMVar mv

140+ return $ ActiveTracer pid wQ

141+

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

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

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

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

3+ ( publishEvent

4+ ) where

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

7+ ( enqueue

8+ )

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

10+ ( MxEventBus(..)

11+ , Message

12+ )

13+import Data.Foldable (forM_)

14+import System.Mem.Weak (deRefWeak)

15+

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

17+publishEvent MxEventBusInitialising _ = return ()

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

19+ mQueue <- deRefWeak wqRef

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

21+

src/Control/Distributed/Process/Node.hs+32−47

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

106 , LocalProcessId(..)

107 , ProcessId(..)

108 , LocalNode(..)

109− , Tracer(..)

109+ , MxEventBus(..)

110 , LocalNodeState(..)

111 , LocalProcess(..)

112 , LocalProcessState(..)

@@ -160,8 +160,7 @@ import Control.Distributed.Process.Internal.Trace.Types

160 , enableTrace

161 )

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

163− ( traceController

164− , defaultTracer

163+ ( defaultTracer

165 )

164 )

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

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

167import Control.Distributed.Process.Internal.Messaging

166import Control.Distributed.Process.Internal.Messaging

@@ -179,13 +178,16 @@ import Control.Distributed.Process.Internal.Primitives

179 , catch

178 , catch

180 , unwrapMessage

179 , unwrapMessage

181 )

180 )

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

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

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

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

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

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

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

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

186 ( mapMaybe

185 ( mapMaybe

187 , mapDefault

186 , mapDefault

188 )

187 )

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

189+ ( mxAgentController

190+ )

189import System.Environment (getEnv)

191import System.Environment (getEnv)

190import Unsafe.Coerce

192import Unsafe.Coerce

191

193

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

222 , localEndPoint = endPoint

224 , localEndPoint = endPoint

223 , localState = state

225 , localState = state

224 , localCtrlChan = ctrlChan

226 , localCtrlChan = ctrlChan

225− , localTracer = InactiveTracer

227+ , localEventBus = MxEventBusInitialising

226 , remoteTable = rtable

228 , remoteTable = rtable

227 }

229 }

228− tracedNode <- startTracing node

230+ tracedNode <- startMxAgent node

229

231

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

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

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

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

@@ -243,38 +245,25 @@ createBareLocalNode endPoint rtable = do

243 , ctrlMsgSignal = SigShutdown

245 , ctrlMsgSignal = SigShutdown

244 }

246 }

245

247

246−startTracing :: LocalNode -> IO LocalNode

248+startMxAgent :: LocalNode -> IO LocalNode

247−startTracing node =

249+startMxAgent node = do

248− -- TODO: use tryGetEnv once we drop support for GHC 7.2.x

249− Exception.catch

250− (getEnv "DISTRIBUTED_PROCESS_TRACE_ENABLED" >> procTracer node)

251− (\(_ :: IOError) -> return node)

252− where procTracer :: LocalNode -> IO LocalNode

253− procTracer n = do

254− -- see note [tracer/forkProcess races]

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

251+ let fork = forkProcess node

255− mv <- MVar.newEmptyMVar

252+ mv <- MVar.newEmptyMVar

256− pid <- forkProcess n (traceController mv)

253+ pid <- fork $ mxAgentController fork mv

257− wqRef <- MVar.takeMVar mv

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

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

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

259−

256+

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

257+startDefaultTracer :: LocalNode -> IO ()

261−startDefaultTracer node =

258+startDefaultTracer node' = do

262− Exception.catch

259+ let t = localEventBus node'

263− (getEnv "DISTRIBUTED_PROCESS_TRACE_ENABLED" >> go node)

264− (\(_ :: IOError) -> return Nothing)

265− where go :: LocalNode -> IO (Maybe ProcessId)

266− go node' = do

267− registerTraceController node'

268− pid <- forkProcess node' defaultTracer

269− enableTrace (localTracer node') pid

270− return (Just pid)

271−

272− registerTraceController :: LocalNode -> IO ()

273− registerTraceController n =

274− let t = localTracer n in

275− case t of

260+ case t of

276− InactiveTracer -> return ()

261+ MxEventBus _ (ActiveTracer pid _) _ _ -> do

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

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

263+ pid' <- forkProcess node' defaultTracer

264+ enableTrace (localEventBus node') pid'

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

266+ _ -> return ()

278

267

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

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

280startServiceProcesses :: LocalNode -> IO ()

269startServiceProcesses :: LocalNode -> IO ()

@@ -283,11 +272,7 @@ startServiceProcesses node = do

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

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

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

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

285 -- however, so we deliberably ignore it

274 -- however, so we deliberably ignore it

286− mPid <- startDefaultTracer node

275+ startDefaultTracer node

287− case mPid of

288− Nothing -> return ()

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

290−

291 logger <- forkProcess node loop

276 logger <- forkProcess node loop

292 runProcess node $ register "logger" logger

277 runProcess node $ register "logger" logger

293 where

278 where

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

401--

386--

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

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

403-- process, and of course forkProcess attempts to call traceEvent once the

388-- process, and of course forkProcess attempts to call traceEvent once the

404−-- process has started. This is harmless, as the localTracer is not updated

389+-- process has started. This is harmless, as the localEventBus is not updated

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

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

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

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

407--

392--

@@ -607,7 +592,7 @@ instance Show ProcessKillException where

607-- Tracing/Debugging --

592-- Tracing/Debugging --

608--------------------------------------------------------------------------------

593--------------------------------------------------------------------------------

609

594

610−-- [Issue #104]

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

611

596

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

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

613traceNotifyDied node ident reason =

598traceNotifyDied node ident reason =

@@ -628,8 +613,8 @@ traceEventFmtIO node fmt args =

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

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

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

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

630

615

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

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

632−withLocalTracer node act = act (localTracer node)

617+withLocalTracer node act = act (localEventBus node)

633

618

634--------------------------------------------------------------------------------

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

635-- Core functionality --

620-- Core functionality --