ad1c63b · distributed-process

git.iko.soy/distributed-process

on master help sign in

Back out of the split between tracer and local mx event bus

ad1c63bTim Watson13 years ago

src/Control/Distributed/Process/Internal/Primitives.hs+6−10

compare with master

unifiedsplit

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

191 , disconnect

192 , sendCtrlMsg

193 )

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

195− ( publishEvent

196− )

197import Control.Distributed.Process.Management.Types

194import Control.Distributed.Process.Management.Types

198 ( MxEvent(..)

195 ( MxEvent(..)

199 )

196 )

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

229 (ProcessIdentifier them)

226 (ProcessIdentifier them)

230 NoImplicitReconnect

227 NoImplicitReconnect

231 msg

228 msg

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

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

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

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

234− liftIO $ do

231+ liftIO $ traceEvent (localEventBus node)

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

232+ (MxSent them us (createUnencodedMessage msg))

236− Mx.publishEvent (localEventBus node) msg

237

233

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

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

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

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

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

416 (messageToPayload msg)

412 (messageToPayload msg)

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

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

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

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

419− liftIO $ Mx.publishEvent (localEventBus node)

415+ liftIO $ traceEvent (localEventBus node)

420− (createUnencodedMessage (MxSent them us msg))

416+ (MxSent them us msg)

421

417

422

418

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

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

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

@@ -216,6 +216,8 @@ data MxEventBus =

216 {

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

218 agent :: !ProcessId

219+ -- | Configuration for the local trace controller

220+ , tracer :: !Tracer

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

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

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

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

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

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

@@ -238,8 +240,6 @@ data LocalNode = LocalNode

238 -- | Runtime lookup table for supporting closures

240 -- | Runtime lookup table for supporting closures

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

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

240 , remoteTable :: !RemoteTable

242 , remoteTable :: !RemoteTable

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

242− , localTracer :: Tracer

243 }

244

245data ImplicitReconnect = WithImplicitReconnect | NoImplicitReconnect

src/Control/Distributed/Process/Management/Bus.hs+5−9

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

1-- | Interface to the management event bus.

2module Control.Distributed.Process.Management.Bus

3 ( publishEvent

4− , enqueueEvent

5 ) where

4 ) where

7import Control.Distributed.Process.Internal.CQueue

6import Control.Distributed.Process.Internal.CQueue

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

22 ( ConnectionId

21 ( ConnectionId

23 , EndPointAddress

22 , EndPointAddress

24 )

23 )

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

24+import System.Mem.Weak (deRefWeak)

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

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

28−publishEvent MxEventBusInitialising _ = return ()

27+publishEvent MxEventBusInitialising _ = return ()

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

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

30−

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

32−enqueueEvent wqRef msg = do

33− mQueue <- deRefWeak wqRef

29+ mQueue <- deRefWeak wqRef

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

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

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

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

60 , Process

61 , ProcessId

62 , LocalProcess(..)

63− , LocalNode(localEventBus, localTracer)

63+ , LocalNode(localEventBus)

64 , SendPort

65 , MxEventBus(..)

66 )

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

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

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

154

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

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

156withLocalTracer act = do

157 node <- processNode <$> ask

158− act (localTracer node)

158+ act (localEventBus node)

159

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

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

161withLocalTracerSync act = do

162 (sp, rp) <- newChan

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

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

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

35 , unsafeCreateUnencodedMessage

36 )

37import Control.Distributed.Process.Management.Bus

38− ( enqueueEvent

38+ ( publishEvent

39 )

40import Control.Distributed.Process.Management.Types

41 ( MxEvent(..)

@@ -116,10 +116,10 @@ instance Binary TraceOk where

116-- Internal/Common API --

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

118

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

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

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

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

121

122−traceLogFmt :: Tracer

122+traceLogFmt :: MxEventBus

123 -> String

124 -> [TraceArg]

125 -> IO ()

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

129 toS (TraceStr s) = s

130 toS (Trace a) = show a

131

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

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

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

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

134

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

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

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

137

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

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

139−enableTrace (Tracer _ wqRef) p =

139+enableTrace t p =

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

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

141− (TraceEnable p)))

141+ (TraceEnable p)))

142

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

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

144−enableTraceSync (Tracer _ wqRef) s p =

144+enableTraceSync t s p =

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

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

146

147−disableTrace :: Tracer -> IO ()

147+disableTrace :: MxEventBus -> IO ()

148−disableTrace (Tracer _ wqRef) =

148+disableTrace t =

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

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

150 TraceDisable))

151

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

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

153−disableTraceSync (Tracer _ wqRef) s =

153+disableTraceSync t s =

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

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

155

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

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

157−setTraceFlags (Tracer _ wqRef) f =

157+setTraceFlags t f =

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

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

159

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

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

161−setTraceFlagsSync (Tracer _ wqRef) s f =

161+setTraceFlagsSync t s f =

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

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

163

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

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

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

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

166

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

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

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

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

169

170class Traceable a where

171 uod :: [a] -> TraceSubject

src/Control/Distributed/Process/Node.hs+30−26

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

259 mv <- MVar.newEmptyMVar

260 pid <- fork $ mxAgentController fork mv

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

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

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

263− , localTracer = tracer'

264− }

265

263

266startDefaultTracer :: LocalNode -> IO ()

264startDefaultTracer :: LocalNode -> IO ()

267startDefaultTracer node' = do

265startDefaultTracer node' = do

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

266+ let t = localEventBus node'

267+ case t of

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

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

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

270− pid' <- forkProcess node' defaultTracer

270+ pid' <- forkProcess node' defaultTracer

271− enableTrace (localTracer node') pid'

271+ enableTrace (localEventBus node') pid'

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

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

273+ _ -> return ()

273

274

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

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

275startServiceProcesses :: LocalNode -> IO ()

276startServiceProcesses :: LocalNode -> IO ()

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

360 return (tid', lproc)

361 return (tid', lproc)

361

362

362 -- see note [tracer/forkProcess races]

363 -- see note [tracer/forkProcess races]

363− mxNotify node (MxSpawned pid)

364+ trace node (MxSpawned pid)

364

365

365 if lpidCounter lpid == maxBound

366 if lpidCounter lpid == maxBound

366 then do

367 then do

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

442 NT.ConnectionOpened cid rel theirAddr ->

443 NT.ConnectionOpened cid rel theirAddr ->

443 if rel == NT.ReliableOrdered

444 if rel == NT.ReliableOrdered

444 then

445 then

445− mxNotify node (MxConnected cid theirAddr)

446+ trace node (MxConnected cid theirAddr)

446 >> go (

447 >> go (

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

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

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

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

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

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

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

459 let msg = payloadToMessage payload

460 let msg = payloadToMessage payload

460 enqueue queue msg -- 'enqueue' is strict

461 enqueue queue msg -- 'enqueue' is strict

461− mxNotify node (MxReceived pid msg)

462+ trace node (MxReceived pid msg)

462 go st

463 go st

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

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

464 mChan <- deRefWeak chan'

465 mChan <- deRefWeak chan'

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

502 else invalidRequest cid st

503 else invalidRequest cid st

503 Nothing ->

504 Nothing ->

504 invalidRequest cid st

505 invalidRequest cid st

505− NT.ConnectionClosed cid -> do

506+ NT.ConnectionClosed cid ->

506− putStrLn "connection closed..."

507 case st ^. incomingAt cid of

508 Nothing ->

509 invalidRequest cid st

510 Just (src, _) -> do

511− mxNotify node (MxDisconnected cid src)

511+ trace node (MxDisconnected cid src)

512 go ( (incomingAt cid ^= Nothing)

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

514 $ st

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

604

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

606traceNotifyDied node ident reason =

607− -- TODO: SendPortDied notifications

607+ -- TODO: sendPortDied notifications

608+ liftIO $ withLocalTracer node $ \t ->

608− case ident of

609+ case ident of

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

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

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

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

611− _ -> return ()

612+ _ -> return ()

612

613

613traceEventFmtIO :: LocalNode

614traceEventFmtIO :: LocalNode

614 -> String

615 -> String

615 -> [TraceArg]

616 -> [TraceArg]

616 -> IO ()

617 -> IO ()

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

618+traceEventFmtIO node fmt args =

619+ withLocalTracer node $ \t -> traceLogFmt t fmt args

620+

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

622+trace node ev = withLocalTracer node $ \t -> traceEvent t ev

618

623

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

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

620−mxNotify LocalNode{..} ev =

625+withLocalTracer node act = act (localEventBus node)

621− Mx.publishEvent localEventBus $ unsafeCreateUnencodedMessage ev

622

626

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

627--------------------------------------------------------------------------------

624-- Core functionality --

628-- Core functionality --

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

817 do modify' $ registeredHereFor label ^= mPid

821 do modify' $ registeredHereFor label ^= mPid

818 updateRemote node currentVal mPid

822 updateRemote node currentVal mPid

819 case mPid of

823 case mPid of

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

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

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

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

822 liftIO $ sendMessage node

826 liftIO $ sendMessage node

823 (NodeIdentifier (localNodeId node))

827 (NodeIdentifier (localNodeId node))

824 (ProcessIdentifier from)

828 (ProcessIdentifier from)

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

886ncEffectLocalSend node to msg =

890ncEffectLocalSend node to msg =

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

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

888 enqueue (processQueue p) msg

892 enqueue (processQueue p) msg

889− mxNotify node (MxReceived to msg)

893+ trace node (MxReceived to msg)

890

894

891-- [Issue #DP-20]

895-- [Issue #DP-20]

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

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