268bee3 · distributed-process

git.iko.soy/distributed-process

on master help sign in

[no-ci] TEMP: this commit is broken, to be ammended

268bee3Tim Watson10 years ago

distributed-process-tests/src/Control/Distributed/Process/Tests/Mx.hs+114−1

compare with master

unifiedsplit

@@ -5,10 +5,19 @@ import Control.Distributed.Process.Tests.Internal.Utils

5import Network.Transport.Test (TestTransport(..))

7import Control.Concurrent (threadDelay)

8+import Control.Concurrent.STM (atomically)

9+import Control.Concurrent.STM.TChan

10+ ( newBroadcastTChanIO

11+ , readTChan

12+ , writeTChan

13+ , dupTChan

14+ )

15+

8import Control.Distributed.Process

16import Control.Distributed.Process

9import Control.Distributed.Process.Node

17import Control.Distributed.Process.Node

10import Control.Distributed.Process.Management

18import Control.Distributed.Process.Management

11 ( MxEvent(..)

19 ( MxEvent(..)

20+ , Destination(..)

12 , MxAgentId(..)

21 , MxAgentId(..)

13 , mxAgent

22 , mxAgent

14 , mxSink

23 , mxSink

@@ -23,7 +32,7 @@ import Control.Distributed.Process.Management

23 , mxBroadcast

32 , mxBroadcast

24 , mxGetId

33 , mxGetId

25 )

34 )

26−import Control.Monad (void)

35+import Control.Monad (void, replicateM_, when)

27import Data.Binary

36import Data.Binary

28import Data.List (find, sort)

37import Data.List (find, sort)

29import Data.Maybe (isJust)

38import Data.Maybe (isJust)

@@ -175,6 +184,109 @@ testAgentEventHandling result = do

175

184

176 stash result $ seenAlive && seenDead

185 stash result $ seenAlive && seenDead

177

186

187+testAgentSendRecvHandling :: LocalNode -> IO ()

188+testAgentSendRecvHandling node = do

189+ runProcess node $ do

190+ testPid <- getSelfPid

191+ agent <- mxAgent (MxAgentId "sendrecv-listener-agent") initState [

192+ (mxSink $ \(p1, p2, sp) -> do

193+ mxSetLocal [p1, p2]

194+ liftMX $ sendChan sp ()

195+ mxReady

196+ ),

197+ (mxSink $ \ev -> do

198+ st <- mxGetLocal

199+ let possiblyNotifyTestProcess pid = when (pid `elem` st) (liftMX $ send testPid ev)

200+ let act = case ev of

201+ MxSent{..} -> possiblyNotifyTestProcess whichProcess

202+ MxReceived{..} -> possiblyNotifyTestProcess whichProcess

203+ _ -> return ()

204+ act >> mxReady)

205+ ]

206+

207+ mRef <- monitor agent

208+

209+ goSignal <- liftIO newBroadcastTChanIO

210+ p1Ready <- liftIO $ atomically (dupTChan goSignal)

211+ p2Ready <- liftIO $ atomically (dupTChan goSignal)

212+

213+ p1 <- spawnLocal $ do

214+ link testPid

215+ liftIO $ atomically (readTChan p1Ready)

216+

217+ -- first we expect a message via send

218+ () <- expect

219+

220+ -- then another, via nsend

221+ liftIO $ atomically (readTChan p1Ready)

222+ () <- expect

223+

224+ -- then another, via usend

225+ liftIO $ atomically (readTChan p1Ready)

226+ () <- expect

227+

228+ return ()

229+

230+ p2 <- spawnLocal $ do

231+ link testPid

232+ liftIO $ atomically (readTChan p2Ready)

233+

234+ send p1 ()

235+

236+ liftIO $ atomically (readTChan p2Ready)

237+ nsend "test.process.1" ()

238+

239+ liftIO $ atomically (readTChan p2Ready)

240+ usend p1 ()

241+ -- and using unsafe primitives...

242+

243+

244+ register testProcName p1

245+ (Just _) <- whereis testProcName

246+

247+ (agentGo, agentReady) <- newChan

248+ mxNotify (p1, p2, agentGo)

249+

250+ () <- receiveChan agentReady

251+

252+ replicateM_ 3 $ checkOnce p1 p2 goSignal

253+

254+ kill agent "finished"

255+ receiveWait [ matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mRef)

256+ (const $ return ()) ]

257+

258+ where

259+ checkOnce p1' p2' sig = do

260+ -- tell both the sending and receiving actors to take a step

261+ liftIO $ atomically $ writeTChan sig ()

262+

263+ -- ensure we saw the send first...

264+ receiveWait [

265+ matchIf (\(ev :: MxEvent) ->

266+ case ev of

267+ (MxSent p' d' _) -> p' == p2' && destProc d' p1'

268+ -- we should never get any other kind of MxEvent here...

269+ _ -> error "unexpected MxEvent"

270+ ) (const $ return ()) ]

271+

272+ -- and the receive after...

273+ receiveWait [

274+ matchIf (\(ev :: MxEvent) ->

275+ case ev of

276+ (MxReceived p' _) -> p' == p1'

277+ -- we should never get any other kind of MxEvent here...

278+ _ -> error "unexpected MxEvent"

279+ ) (const $ return ()) ]

280+

281+ initState :: [ProcessId]

282+ initState = []

283+

284+ testProcName = "test.process.1"

285+

286+ destProc (ProcId p) p' = p == p'

287+ destProc (ProcName n) _ = n == testProcName

288+ destProc _ _ = False

289+

178tests :: TestTransport -> IO [Test]

290tests :: TestTransport -> IO [Test]

179tests TestTransport{..} = do

291tests TestTransport{..} = do

180 node1 <- newLocalNode testTransport initRemoteTable

292 node1 <- newLocalNode testTransport initRemoteTable

@@ -184,6 +296,7 @@ tests TestTransport{..} = do

184 (delayedAssertion

296 (delayedAssertion

185 "expected True, but events where not as expected"

297 "expected True, but events where not as expected"

186 node1 True testAgentEventHandling)

298 node1 True testAgentEventHandling)

299+-- , testCase "Send & Receive" (testAgentSendRecvHandling node1)

187 , testCase "Inter-Agent Broadcast"

300 , testCase "Inter-Agent Broadcast"

188 (delayedAssertion

301 (delayedAssertion

189 "expected (), but no broadcast was received"

302 "expected (), but no broadcast was received"

distributed-process-tests/src/Control/Distributed/Process/Tests/Tracing.hs+26−14

@@ -1,4 +1,3 @@

1−{-# LANGUAGE DeriveDataTypeable #-}

2{-# OPTIONS_GHC -fno-warn-orphans #-}

1{-# OPTIONS_GHC -fno-warn-orphans #-}

3module Control.Distributed.Process.Tests.Tracing (tests) where

2module Control.Distributed.Process.Tests.Tracing (tests) where

@@ -19,6 +18,7 @@ import Control.Distributed.Process.Node

19import Control.Distributed.Process.Debug

18import Control.Distributed.Process.Debug

20import Control.Distributed.Process.Management

19import Control.Distributed.Process.Management

21 ( MxEvent(..)

20 ( MxEvent(..)

21+ , Destination(..)

22 )

23import qualified Control.Exception as IO (bracket)

24import Data.List (isPrefixOf, isSuffixOf)

@@ -124,10 +124,11 @@ testTraceSending result = do

124 withTracer

125 (\ev ->

126 case ev of

127− (MxSent to from msg) -> do

127+ MxSent{..} -> do

128+ let to = procId whereTo

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

129+ (Just s) <- unwrapMessage message :: Process (Maybe String)

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

130+ stash res (to == pid && whichProcess == self && s == "hello there")

130− stash res (to == pid && from == self)

131+ stash res (to == pid && whichProcess == self)

131 _ ->

132 _ ->

132 return ()) $ do

133 return ()) $ do

133 send pid "hello there"

134 send pid "hello there"

@@ -314,24 +315,29 @@ testSystemLoggerMsg t action interestingMessage =

314 runProcess n $ do

315 runProcess n $ do

315 self <- getSelfPid

316 self <- getSelfPid

316 reregister "trace.logger" self

317 reregister "trace.logger" self

318+

317 a <- action

319 a <- action

318 let interestingMessage' (_ :: String, msg) = interestingMessage a msg

320 let interestingMessage' (_ :: String, msg) = interestingMessage a msg

319 -- Wait for the trace message.

321 -- Wait for the trace message.

320 receiveWait [ matchIf interestingMessage' $ const $ return () ]

322 receiveWait [ matchIf interestingMessage' $ const $ return () ]

321 -- Only one interesting message should arrive.

323 -- Only one interesting message should arrive.

322 Nothing <- receiveTimeout 100000

324 Nothing <- receiveTimeout 100000

323− [ matchIf interestingMessage' $ const $ return () ]

325+ [ matchIf interestingMessage' $ const $ return () ]

326+ -- return ()

327+ _ <- expectTimeout 5000000 :: Process (Maybe ())

324 return ()

328 return ()

325

329

330+ closeLocalNode n

326

331

327-- | Tests that one and only one trace message is produced when a message is

332-- | Tests that one and only one trace message is produced when a message is

328-- received.

333-- received.

329testSystemLoggerMxReceive :: TestTransport -> IO ()

334testSystemLoggerMxReceive :: TestTransport -> IO ()

330testSystemLoggerMxReceive t = testSystemLoggerMsg t

335testSystemLoggerMxReceive t = testSystemLoggerMsg t

331− (getSelfPid >>= flip send ())

336+ -- this is never going to work because we are trace.logger dumbass

337+ ((spawnLocal expect) >>= \pid -> send pid () >> return pid)

332− (\_ msg -> "MxReceived" `isPrefixOf` msg

338+ (\pid msg -> ("MxReceived" `isPrefixOf` msg)

333 -- discard traces of internal messages

339 -- discard traces of internal messages

334− && not (":: RegisterReply" `isSuffixOf` msg)

340+ && not (":: RegisterReply}" `isSuffixOf` msg)

335 )

341 )

336

342

337-- | Tests that one and only one trace message is produced when a message is

343-- | Tests that one and only one trace message is produced when a message is

@@ -339,33 +345,39 @@ testSystemLoggerMxReceive t = testSystemLoggerMsg t

339testSystemLoggerMxSent :: TestTransport -> IO ()

345testSystemLoggerMxSent :: TestTransport -> IO ()

340testSystemLoggerMxSent t = testSystemLoggerMsg t

346testSystemLoggerMxSent t = testSystemLoggerMsg t

341 (getSelfPid >>= flip send ())

347 (getSelfPid >>= flip send ())

342− (const $ isPrefixOf "MxSent")

348+ (\pid msg -> (("MxSent {whichProcess = " ++ show pid) `isPrefixOf` msg)

349+ -- discard traces of internal messages

350+ && not (":: RegisterReply}" `isSuffixOf` msg)

351+ )

343

352

344-- | Tests that one and only one trace message is produced when a process dies.

353-- | Tests that one and only one trace message is produced when a process dies.

345testSystemLoggerMxProcessDied :: TestTransport -> IO ()

354testSystemLoggerMxProcessDied :: TestTransport -> IO ()

346testSystemLoggerMxProcessDied t = testSystemLoggerMsg t

355testSystemLoggerMxProcessDied t = testSystemLoggerMsg t

347 (spawnLocal $ return ())

356 (spawnLocal $ return ())

348− (\pid -> isPrefixOf $ "MxProcessDied " ++ show pid)

357+ (\pid -> (== "MxProcessDied {whichProcess = " ++

358+ show pid ++ ", why = DiedNormal}"))

349

359

350-- | Tests that one and only one trace message appears when a process spawns.

360-- | Tests that one and only one trace message appears when a process spawns.

351testSystemLoggerMxSpawned :: TestTransport -> IO ()

361testSystemLoggerMxSpawned :: TestTransport -> IO ()

352testSystemLoggerMxSpawned t = testSystemLoggerMsg t

362testSystemLoggerMxSpawned t = testSystemLoggerMsg t

353 (spawnLocal $ return ())

363 (spawnLocal $ return ())

354− (\pid -> isPrefixOf $ "MxSpawned " ++ show pid)

364+ (\pid -> isPrefixOf $ "MxSpawned {whichProcess = " ++ show pid)

355

365

356-- | Tests that one and only one trace message appears when a process is

366-- | Tests that one and only one trace message appears when a process is

357-- registered.

367-- registered.

358testSystemLoggerMxRegistered :: TestTransport -> IO ()

368testSystemLoggerMxRegistered :: TestTransport -> IO ()

359testSystemLoggerMxRegistered t = testSystemLoggerMsg t

369testSystemLoggerMxRegistered t = testSystemLoggerMsg t

360 (getSelfPid >>= register "a" >> getSelfPid)

370 (getSelfPid >>= register "a" >> getSelfPid)

361− (\self -> isPrefixOf $ "MxRegistered " ++ show self ++ " " ++ show "a")

371+ (\pid -> (== "MxRegistered {whichProcess = " ++ show pid ++

372+ ", whichName = " ++ show "a" ++ "}"))

362

373

363-- | Tests that one and only one trace message appears when a process is

374-- | Tests that one and only one trace message appears when a process is

364-- unregistered.

375-- unregistered.

365testSystemLoggerMxUnRegistered :: TestTransport -> IO ()

376testSystemLoggerMxUnRegistered :: TestTransport -> IO ()

366testSystemLoggerMxUnRegistered t = testSystemLoggerMsg t

377testSystemLoggerMxUnRegistered t = testSystemLoggerMsg t

367 (getSelfPid >>= register "a" >> unregister "a" >> getSelfPid)

378 (getSelfPid >>= register "a" >> unregister "a" >> getSelfPid)

368− (\self -> isPrefixOf $ "MxUnRegistered " ++ show self ++ " " ++ show "a")

379+ (\pid -> (== "MxUnRegistered {whichProcess = " ++

380+ show pid ++ ", whichName = " ++ show "a" ++ "}"))

369

381

370tests :: TestTransport -> IO [Test]

382tests :: TestTransport -> IO [Test]

371tests testtrans@TestTransport{..} = do

383tests testtrans@TestTransport{..} = do

src/Control/Distributed/Process/Internal/Primitives.hs+71−21

@@ -226,6 +226,7 @@ import Control.Distributed.Process.Internal.Messaging

226 )

227import Control.Distributed.Process.Management.Internal.Types

228 ( MxEvent(..)

229+ , Destination(..)

229 )

230 )

230import Control.Distributed.Process.Management.Internal.Trace.Types

231import Control.Distributed.Process.Management.Internal.Trace.Types

231 ( traceEvent

232 ( traceEvent

@@ -256,14 +257,16 @@ send them msg = do

256 case destNode == nodeId of

257 case destNode == nodeId of

257 True -> sendLocal them msg

258 True -> sendLocal them msg

258 False -> liftIO $ sendMessage (processNode proc)

259 False -> liftIO $ sendMessage (processNode proc)

259− (ProcessIdentifier (processId proc))

260+ (ProcessIdentifier (processId proc))

260− (ProcessIdentifier them)

261+ (ProcessIdentifier them)

261− NoImplicitReconnect

262+ NoImplicitReconnect

262− msg

263+ msg

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

264+ -- see note [trace MxSent]

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

265 liftIO $ traceEvent (localEventBus node)

266+ MxSent { whichProcess = us

267+ , whereTo = ProcId them

266− (MxSent them us (createUnencodedMessage msg))

268+ , message = (createUnencodedMessage msg)

269+ }

267

270

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

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

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

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

@@ -283,12 +286,22 @@ unsafeSend = Unsafe.send

283--

286--

284usend :: Serializable a => ProcessId -> a -> Process ()

287usend :: Serializable a => ProcessId -> a -> Process ()

285usend them msg = do

288usend them msg = do

286− here <- getSelfNode

289+ proc <- ask

290+ let us = processId proc

291+ node = processNode proc

292+ nodeId = localNodeId node

287− let there = processNodeId them

293+ let destNodeId = processNodeId them

288− if here == there

294+ if nodeId == destNodeId

289 then sendLocal them msg

295 then sendLocal them msg

296+ else do -- see note [trace MxSent]

297+ let evBus = localEventBus node

298+ let msg' = createMessage msg

290− else sendCtrlMsg (Just there) $ UnreliableSend (processLocalId them)

299+ sendCtrlMsg (Just destNodeId) $ UnreliableSend (processLocalId them) msg'

291− (createMessage msg)

300+ liftIO $ traceEvent evBus $ MxSent { whichProcess = us

301+ , whereTo = ProcId them

302+ , message = msg'

303+ }

304+

292

305

293-- | /Unsafe/ variant of 'usend'. This function makes /no/ attempt to serialize

306-- | /Unsafe/ variant of 'usend'. This function makes /no/ attempt to serialize

294-- the message when the destination process resides on the same local

307-- the message when the destination process resides on the same local

@@ -487,10 +500,12 @@ forward msg them = do

487 (ProcessIdentifier them)

500 (ProcessIdentifier them)

488 NoImplicitReconnect

501 NoImplicitReconnect

489 (messageToPayload msg)

502 (messageToPayload msg)

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

503+

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

492 liftIO $ traceEvent (localEventBus node)

504 liftIO $ traceEvent (localEventBus node)

493− (MxSent them us msg)

505+ MxSent { whichProcess = us

506+ , whereTo = ProcId them

507+ , message = msg

508+ }

494

509

495-- | Forward a raw 'Message' to the given 'ProcessId'.

510-- | Forward a raw 'Message' to the given 'ProcessId'.

496--

511--

@@ -511,7 +526,10 @@ uforward msg them = do

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

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

512 -- In the remote case, 'sendCtrlMsg' can block in the networking stack.

527 -- In the remote case, 'sendCtrlMsg' can block in the networking stack.

513 liftIO $ traceEvent (localEventBus node)

528 liftIO $ traceEvent (localEventBus node)

514− (MxSent them us msg)

529+ MxSent { whichProcess = us

530+ , whereTo = ProcId them

531+ , message = msg

532+ }

515

533

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

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

517-- 'Serializable' - like the datum they contain - but also note, deserialising

535-- 'Serializable' - like the datum they contain - but also note, deserialising

@@ -1220,8 +1238,20 @@ whereisRemoteAsync nid label = do

1220

1238

1221-- | Named send to a process in the local registry (asynchronous)

1239-- | Named send to a process in the local registry (asynchronous)

1222nsend :: Serializable a => String -> a -> Process ()

1240nsend :: Serializable a => String -> a -> Process ()

1223−nsend label msg =

1241+nsend label msg = do

1242+ proc <- ask

1243+ let us = processId proc

1244+ node = processNode proc

1245+ msg' = createMessage msg -- NB: using UnencodedMessage here causes chaos... :/

1246+

1224− sendCtrlMsg Nothing (NamedSend label (createUnencodedMessage msg))

1247+ sendCtrlMsg Nothing (NamedSend label msg')

1248+

1249+ -- see note [trace MxSent]

1250+ liftIO $ traceEvent (localEventBus node)

1251+ MxSent { whichProcess = us

1252+ , whereTo = ProcName label

1253+ , message = msg'

1254+ }

1225

1255

1226-- | Named send to a process in the local registry (asynchronous).

1256-- | Named send to a process in the local registry (asynchronous).

1227-- This function makes /no/ attempt to serialize and (in the case when the

1257-- This function makes /no/ attempt to serialize and (in the case when the

@@ -1233,9 +1263,19 @@ unsafeNSend = Unsafe.nsend

1233-- | Named send to a process in a remote registry (asynchronous)

1263-- | Named send to a process in a remote registry (asynchronous)

1234nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()

1264nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()

1235nsendRemote nid label msg = do

1265nsendRemote nid label msg = do

1266+ proc <- ask

1267+ let us = processId proc

1268+ node = processNode proc

1236 here <- getSelfNode

1269 here <- getSelfNode

1237 if here == nid then nsend label msg

1270 if here == nid then nsend label msg

1271+ else do let msg' = createMessage msg

1238− else sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))

1272+ sendCtrlMsg (Just nid) (NamedSend label msg')

1273+ -- see note [trace MxSent]

1274+ liftIO $ traceEvent (localEventBus node)

1275+ MxSent { whichProcess = us

1276+ , whereTo = ProcName label

1277+ , message = msg'

1278+ }

1239

1279

1240-- | Named send to a process in a remote registry (asynchronous)

1280-- | Named send to a process in a remote registry (asynchronous)

1241-- This function makes /no/ attempt to serialize and (in the case when the

1281-- This function makes /no/ attempt to serialize and (in the case when the

@@ -1306,8 +1346,18 @@ reconnectPort them = do

1306--------------------------------------------------------------------------------

1346--------------------------------------------------------------------------------

1307

1347

1308sendLocal :: (Serializable a) => ProcessId -> a -> Process ()

1348sendLocal :: (Serializable a) => ProcessId -> a -> Process ()

1309−sendLocal to msg =

1349+sendLocal to msg = do

1350+ proc <- ask

1351+ let us = processId proc

1352+ node = processNode proc

1353+ msg' = createUnencodedMessage msg

1354+

1310− sendCtrlMsg Nothing $ LocalSend to (createUnencodedMessage msg)

1355+ sendCtrlMsg Nothing $ LocalSend to msg'

1356+ liftIO $ traceEvent (localEventBus node)

1357+ MxSent { whichProcess = us

1358+ , whereTo = ProcId to

1359+ , message = msg'

1360+ }

1311

1361

1312sendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()

1362sendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()

1313sendChanLocal spId msg =

1363sendChanLocal spId msg =

src/Control/Distributed/Process/Management.hs+2

@@ -219,6 +219,7 @@

219module Control.Distributed.Process.Management

220 (

221 MxEvent(..)

222+ , Destination(..)

222 -- * Firing Arbitrary /Mx Events/

223 -- * Firing Arbitrary /Mx Events/

223 , mxNotify

224 , mxNotify

224 -- * Constructing Mx Agents

225 -- * Constructing Mx Agents

@@ -276,6 +277,7 @@ import Control.Distributed.Process.Management.Internal.Types

276 , MxAgentState(..)

277 , MxAgentState(..)

277 , MxSink

278 , MxSink

278 , MxEvent(..)

279 , MxEvent(..)

280+ , Destination(..)

279 )

281 )

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

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

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

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

src/Control/Distributed/Process/Management/Internal/Agent.hs+2−2

@@ -57,8 +57,8 @@ type AgentConfig =

57 (Tracer, Weak (CQueue Message),

58 (((TChan Message, TChan Message) -> Process ()) -> IO ProcessId))

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

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

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

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

62-- @ProcessId@ where possible.

63--

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

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

@@ -167,7 +167,10 @@ withRegisteredTracer act = do

167 (sp, rp) <- newChan

168 withLocalTracer $ \t -> liftIO $ Tracer.getCurrentTraceClient t sp

169 currentTracer <- receiveChan rp

170+ -- TODO: there is an obvious race here, whereby tracer.initial could

171+ -- die or be replaced (via reregister) between the call to whereis

172+ -- and the expression acting on the pid. Maybe pass the trace client

173+ -- instead???

170 case currentTracer of

174 case currentTracer of

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

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

172 (Just p) -> act p

176 (Just p) -> act p

173−

src/Control/Distributed/Process/Management/Internal/Trace/Tracer.hs+29−15

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

1{-# LANGUAGE CPP #-}

2{-# LANGUAGE ScopedTypeVariables #-}

3+{-# LANGUAGE RecordWildCards #-}

4+{-# LANGUAGE PatternGuards #-}

3-- | Tracing/Debugging support - Trace Implementation

5-- | Tracing/Debugging support - Trace Implementation

4module Control.Distributed.Process.Management.Internal.Trace.Tracer

6module Control.Distributed.Process.Management.Internal.Trace.Tracer

5 ( -- * API for the Management Agent

7 ( -- * API for the Management Agent

@@ -142,21 +144,28 @@ nullTracer =

142

144

143systemLoggerTracer :: Process ()

145systemLoggerTracer :: Process ()

144systemLoggerTracer = do

146systemLoggerTracer = do

145− node <- processNode <$> ask

147+ proc <- ask

146− let tr = sendTraceLog node

148+ let pid = processId proc

149+ node = processNode proc

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

150+ forever' $ receiveWait [ matchAny (\m -> handleMessage m (sendTraceLog pid node)) ]

148 where

151 where

149− sendTraceLog :: LocalNode -> MxEvent -> Process ()

152+ sendTraceLog :: ProcessId -> LocalNode -> MxEvent -> Process ()

150− sendTraceLog node ev = do

153+ sendTraceLog pid' node' ev

154+ -- because process is the trace client, the trace coordinator cannot

155+ -- filter out messages pertaining to the trace.logger, so we do so here

156+ | MxReceived{..} <- ev

157+ , whichProcess == pid' = return ()

158+ | otherwise = do

151− now <- liftIO $ getCurrentTime

159+ now <- liftIO $ getCurrentTime

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

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

153− emptyPid <- return $ (nullProcessId (localNodeId node))

161+ emptyPid <- return $ nullPid node'

154− traceMsg <- return $ NCMsg {

155− ctrlMsgSender = ProcessIdentifier (emptyPid)

162+ traceMsg <- return $ NCMsg { ctrlMsgSender = ProcessIdentifier (emptyPid)

156− , ctrlMsgSignal = (NamedSend "trace.logger"

163+ , ctrlMsgSignal = (NamedSend "trace.logger"

157− (createUnencodedMessage msg))

164+ (createUnencodedMessage msg))

158− }

165+ }

159− liftIO $ writeChan (localCtrlChan node) traceMsg

166+ liftIO $ writeChan (localCtrlChan node') traceMsg

167+

168+ nullPid node'' = nullProcessId (localNodeId node'')

160

169

161 buildTxt :: MxEvent -> String

170 buildTxt :: MxEvent -> String

162 buildTxt (MxLog msg) = msg

171 buildTxt (MxLog msg) = msg

@@ -235,7 +244,8 @@ traceController mv = do

235 sendOk confResp >> applyTraceFlags flags' st)

244 sendOk confResp >> applyTraceFlags flags' st)

236 , match (\chGetFlags -> sendChan chGetFlags (flags st) >> return st)

245 , match (\chGetFlags -> sendChan chGetFlags (flags st) >> return st)

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

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

247+ -- we do not forward messages to the client that pertain to itself

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

248+ -- also, we de-queue incoming events even if we don't process them

239 , matchAny (\ev ->

249 , matchAny (\ev ->

240 handleMessage ev (handleTrace st ev) >>= return . fromMaybe st)

250 handleMessage ev (handleTrace st ev) >>= return . fromMaybe st)

241 ]

251 ]

@@ -332,6 +342,10 @@ traceEv ev msg (Just (TraceProcs pids)) st = do

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

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

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

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

334 -- there are overlapping trace targets

344 -- there are overlapping trace targets

345+ -- NB: this will now be broken for instances of Destination that cannot be

346+ -- resolved, thus we need to re-examine this branch of the tracing code

347+ -- to understand what the impact of that will actually be...

348+

335 node <- processNode <$> ask

349 node <- processNode <$> ask

336 let p = case resolveToPid ev of

350 let p = case resolveToPid ev of

337 Nothing -> (nullProcessId (localNodeId node))

351 Nothing -> (nullProcessId (localNodeId node))

src/Control/Distributed/Process/Management/Internal/Types.hs+56−15

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

11 , Fork

12 , MxSink

13 , MxEvent(..)

14+ , Destination(..)

14 , Addressable(..)

15 , Addressable(..)

15 ) where

16 ) where

@@ -18,10 +19,12 @@ import Control.Applicative (Applicative)

18import Control.Concurrent.STM

19import Control.Concurrent.STM

19 ( TChan

20 ( TChan

20 )

21 )

22+import Control.DeepSeq (NFData(..))

21import Control.Distributed.Process.Internal.Types

23import Control.Distributed.Process.Internal.Types

22 ( Process

24 ( Process

23 , ProcessId

25 , ProcessId

24 , Message

26 , Message

27+ , SendPort

25 , DiedReason

28 , DiedReason

26 , NodeId

29 , NodeId

27 )

30 )

@@ -38,35 +41,68 @@ import Network.Transport

38 , EndPointAddress

41 , EndPointAddress

39 )

42 )

44+-- | A simple means of mapping send events to their destinations

45+data Destination =

46+ ProcId { procId :: !ProcessId }

47+ | ProcName { procName :: !String }

48+ | RemoteProcName { procName :: !String

49+ , procNode :: !NodeId }

50+ | UserDefined { userData :: !Message }

51+ deriving (Typeable, Generic, Show)

52+instance Binary Destination where

53+instance NFData Destination where

54+ rnf (ProcId p) = rnf p `seq` ()

55+ rnf (ProcName s) = rnf s `seq` ()

56+ rnf (RemoteProcName s n) = rnf s `seq` rnf n `seq` ()

57+ rnf (UserDefined m) = rnf m `seq` ()

58+

59+

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

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

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

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

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

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

44-- they will be delivered to the trace controller.

63-- they will be delivered to the trace controller.

45--

64--

46data MxEvent =

65data MxEvent =

47− MxSpawned ProcessId

66+ MxSpawned { whichProcess :: ProcessId }

48 -- ^ fired whenever a local process is spawned

67 -- ^ fired whenever a local process is spawned

49− | MxRegistered ProcessId String

68+ | MxRegistered { whichProcess :: ProcessId

69+ , whichName :: String

70+ }

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

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

51− | MxUnRegistered ProcessId String

72+ | MxUnRegistered { whichProcess :: ProcessId

73+ , whichName :: String

74+ }

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

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

53− | MxProcessDied ProcessId DiedReason

76+ | MxProcessDied { whichProcess :: ProcessId

77+ , why :: DiedReason

78+ }

54 -- ^ fired whenever a process dies

79 -- ^ fired whenever a process dies

55− | MxNodeDied NodeId DiedReason

80+ | MxNodeDied { whichNode :: NodeId

81+ , why :: DiedReason

82+ }

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

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

57− | MxSent ProcessId ProcessId Message

84+ | MxSent { whichProcess :: ProcessId

85+ , whereTo :: Destination

86+ , message :: Message

87+ }

58 -- ^ fired whenever a message is sent from a local process

88 -- ^ fired whenever a message is sent from a local process

59− | MxReceived ProcessId Message

89+ | MxReceived { whichProcess :: ProcessId

90+ , message :: Message

91+ }

60 -- ^ fired whenever a message is received by a local process

92 -- ^ fired whenever a message is received by a local process

61− | MxConnected ConnectionId EndPointAddress

93+ | MxConnected { whichConnection :: ConnectionId

94+ , whichEndoint :: EndPointAddress

95+ }

62 -- ^ fired when a network-transport connection is first established

96 -- ^ fired when a network-transport connection is first established

63− | MxDisconnected ConnectionId EndPointAddress

97+ | MxDisconnected { whichConnection :: ConnectionId

98+ , whichEndpoint :: EndPointAddress

99+ }

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

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

65− | MxUser Message

101+ | MxUser { message :: Message }

66 -- ^ a user defined trace event

102 -- ^ a user defined trace event

67− | MxLog String

103+ | MxLog { text :: String }

68 -- ^ a /logging/ event - used for debugging purposes only

104 -- ^ a /logging/ event - used for debugging purposes only

69− | MxTraceTakeover ProcessId

105+ | MxTraceTakeover { whichProcess :: ProcessId }

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

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

71 | MxTraceDisable

107 | MxTraceDisable

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

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

@@ -74,15 +110,20 @@ data MxEvent =

110

75instance Binary MxEvent where

111instance Binary MxEvent where

112

77−-- | The class of things that we might be able to resolve to

113+-- The class of things that we might be able to resolve to

78−-- a @ProcessId@ (or not).

114+-- a @ProcessId@ (or not). Note we don't want to export this!

79class Addressable a where

115class Addressable a where

80 resolveToPid :: a -> Maybe ProcessId

116 resolveToPid :: a -> Maybe ProcessId

117

118+instance Addressable Destination where

119+ resolveToPid dest

120+ | (ProcId pid) <- dest = Just pid

121+ | otherwise = Nothing

122+

82instance Addressable MxEvent where

123instance Addressable MxEvent where

83 resolveToPid (MxSpawned p) = Just p

124 resolveToPid (MxSpawned p) = Just p

84 resolveToPid (MxProcessDied p _) = Just p

125 resolveToPid (MxProcessDied p _) = Just p

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

126+ resolveToPid (MxSent _ p _) = resolveToPid p

86 resolveToPid (MxReceived p _) = Just p

127 resolveToPid (MxReceived p _) = Just p

87 resolveToPid _ = Nothing

128 resolveToPid _ = Nothing

129

src/Control/Distributed/Process/Node.hs+1

@@ -550,6 +550,7 @@ handleIncomingMessages node = go initConnectionState

550 forM_ mChan $ \chan -> atomically $

551 -- We make sure the message is fully decoded when it is enqueued

552 writeTQueue chan $! decode (BSL.fromChunks payload)

553+ -- trace node (MxReceived pid msg)

553 go st

554 go st

554 Just (_, ToNode) -> do

555 Just (_, ToNode) -> do

555 let ctrlMsg = decode . BSL.fromChunks $ payload

556 let ctrlMsg = decode . BSL.fromChunks $ payload

stack-ghc-8.0.1.yamlnew file+18

@@ -0,0 +1,18 @@

1+# Specifies the GHC version and set of packages available

2+resolver: lts-7.14

4+packages:

5+- '.'

6+- distributed-process-tests/

8+extra-deps:

9+- rank1dynamic-0.3.3.0

10+- distributed-static-0.3.5.0

11+- network-transport-0.4.4.0

12+- network-transport-tcp-0.5.1

13+- network-transport-inmemory-0.5.1

14+- rematch-0.2.0.0

15+

16+flags:

17+ distributed-process-tests:

18+ tcp: true

stack.yamlmode 100644 → 120000+1−17

@@ -1,17 +0,0 @@

1−resolver: nightly-2016-06-09

2−

3−packages:

4−- '.'

5−- distributed-process-tests/

6−

7−extra-deps:

8−- rank1dynamic-0.3.3.0

9−- distributed-static-0.3.5.0

10−- network-transport-0.4.4.0

11−- network-transport-tcp-0.5.1

12−- network-transport-inmemory-0.5.1

13−- rematch-0.2.0.0

14−

15−flags:

16− distributed-process-tests:

17− tcp: true

@@ -0,0 +1 @@

1+stack-ghc-8.0.1.yaml