467a611 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Update comments to reflect what we're doing

467a611Tim Watson8 years ago

distributed-process-tests/src/Control/Distributed/Process/Tests/CH.hs+8

compare with master

unifiedsplit

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

39import Control.Distributed.Process.Node

40import Control.Distributed.Process.Debug

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

42+import Control.Distributed.Process.Tests.Internal.Utils (shouldBe)

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

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

44import Test.HUnit (Assertion, assertFailure, assertBool)

45import Test.HUnit (Assertion, assertFailure, assertBool)

@@ -1404,6 +1405,9 @@ testRegistryMonitoring TestTransport{..} = do

1404 w <- whereis "test" :: Process (Maybe ProcessId)

1405 w <- whereis "test" :: Process (Maybe ProcessId)

1405 forM_ w (const waitpoll)

1406 forM_ w (const waitpoll)

1406 in do register "test" remote1

1407 in do register "test" remote1

1408+ -- [race-condition] invalid test if we force exit before waitpoll runs

1409+ registered <- whereis "test"

1410+ registered `shouldBe` (equalTo (Just remote1))

1407 send remote1 ()

1411 send remote1 ()

1408 waitpoll

1412 waitpoll

1409 return ()

1413 return ()

@@ -1412,6 +1416,7 @@ testRegistryMonitoring TestTransport{..} = do

1412 -- Many labels. Test if all labels associated with process

1416 -- Many labels. Test if all labels associated with process

1413 -- are removed from registry when it dies.

1417 -- are removed from registry when it dies.

1414 remote2 <- testRemote remoteNode

1418 remote2 <- testRemote remoteNode

1419+ remoteX <- testRemote remoteNode

1415 runProcess localNode $

1420 runProcess localNode $

1416 let waitpoll = do

1421 let waitpoll = do

1417 w1 <- whereis "test-3" :: Process (Maybe ProcessId)

1422 w1 <- whereis "test-3" :: Process (Maybe ProcessId)

@@ -1419,8 +1424,11 @@ testRegistryMonitoring TestTransport{..} = do

1419 forM_ (w1 <|> w2) (const waitpoll)

1424 forM_ (w1 <|> w2) (const waitpoll)

1420 in do register "test-3" remote2

1425 in do register "test-3" remote2

1421 register "test-4" remote2

1426 register "test-4" remote2

1427+ register "test-X" remoteX

1422 send remote2 ()

1428 send remote2 ()

1423 waitpoll

1429 waitpoll

1430+ rX <- whereis "test-X"

1431+ rX `shouldBe` (equalTo (Just remoteX))

1424 return ()

1432 return ()

1425

1433

1426{- XXX: waiting including patch for nsend for remote process

1434{- XXX: waiting including patch for nsend for remote process

distributed-process-tests/src/Control/Distributed/Process/Tests/Mx.hs+63−5

@@ -176,14 +176,14 @@ testAgentEventHandling result = do

176

177 stash result $ seenAlive && seenDead

178

179−testMxMonitorEvents :: Process ()

179+testMxRegEvents :: Process ()

180−testMxMonitorEvents = do

180+testMxRegEvents = do

181

182 {- This test only deals with the local case, to ensure that we are being

183 notified in the expected order - the remote cases related to the

184 behaviour of the node controller are contained in the CH test suite. -}

185

186− let label = "testMxMonitorEvents"

186+ let label = "testMxRegEvents"

187 let agentLabel = "listener-agent"

188 let delay = 1000000

189 (regChan, regSink) <- newChan

@@ -223,9 +223,65 @@ testMxMonitorEvents = do

223

224 mapM_ (flip kill $ "test-complete") [agent, p1, p2]

225

226+testMxRegMon :: LocalNode -> Process ()

227+testMxRegMon remoteNode = do

228+

229+ -- ensure that when a registered process dies, we get a notification that

230+ -- it has been unregistered as well as seeing the name get removed

231+

232+ let label1 = "aaaaa"

233+ let label2 = "bbbbb"

234+ let isValid l = l ==label1 || l == label2

235+ let agentLabel = "listener-agent"

236+ let delay = 1000000

237+ (regChan, regSink) <- newChan

238+ (unRegChan, unRegSink) <- newChan

239+ agent <- mxAgent (MxAgentId agentLabel) () [

240+ mxSink $ \ev -> do

241+ case ev of

242+ MxRegistered pid label

243+ | isValid label -> liftMX $ sendChan regChan (label, pid)

244+ MxUnRegistered pid label

245+ | isValid label -> liftMX $ sendChan unRegChan (label, pid)

246+ _ -> return ()

247+ mxReady

248+ ]

249+

250+ (sp, rp) <- newChan

251+ liftIO $ forkProcess remoteNode $ do

252+ getSelfPid >>= sendChan sp

253+ expect :: Process ()

254+

255+ p1 <- receiveChan rp

256+

257+ register label1 p1

258+ reg1 <- receiveChanTimeout delay regSink

259+ reg1 `shouldBe` equalTo (Just (label1, p1))

260+

261+ register label2 p1

262+ reg2 <- receiveChanTimeout delay regSink

263+ reg2 `shouldBe` equalTo (Just (label2, p1))

264+

265+ n1 <- whereis label1

266+ n1 `shouldBe` equalTo (Just p1)

267+

268+ n2 <- whereis label2

269+ n2 `shouldBe` equalTo (Just p1)

270+

271+ kill p1 "goodbye"

272+

273+ unreg1 <- receiveChanTimeout delay unRegSink

274+ unreg2 <- receiveChanTimeout delay unRegSink

275+

276+ sort [unreg1, unreg2]

277+ `shouldBe` equalTo [Just (label1, p1), Just (label2, p1)]

278+

279+ kill agent "test-complete"

280+

226tests :: TestTransport -> IO [Test]

281tests :: TestTransport -> IO [Test]

227tests TestTransport{..} = do

282tests TestTransport{..} = do

228 node1 <- newLocalNode testTransport initRemoteTable

283 node1 <- newLocalNode testTransport initRemoteTable

284+ node2 <- newLocalNode testTransport initRemoteTable

229 return [

285 return [

230 testGroup "Mx Agents" [

286 testGroup "Mx Agents" [

231 testCase "Event Handling"

287 testCase "Event Handling"

@@ -252,7 +308,9 @@ tests TestTransport{..} = do

252 "fifth"]) testAgentPrioritisation)

308 "fifth"]) testAgentPrioritisation)

253 ]

309 ]

254 , testGroup "Mx Events" [

310 , testGroup "Mx Events" [

255− testCase "Monitor Events"

311+ testCase "Name Registration Events"

312+ (runProcess node1 testMxRegEvents)

313+ , testCase "Post Death Name UnRegistration Events"

256− (runProcess node1 testMxMonitorEvents)

314+ (runProcess node1 (testMxRegMon node2))

257 ]

315 ]

258 ]

316 ]

src/Control/Distributed/Process/Node.hs+21−22

@@ -34,7 +34,7 @@ import qualified Data.Map as Map

34 , filter

35 , insert

36 , lookup

37− , insertLookupWithKey

37+ , partition

38 , partitionWithKey

39 , elems

40 , size

@@ -328,28 +328,23 @@ startDefaultTracer node' = do

328registryMonitorAgentId :: MxAgentId

329registryMonitorAgentId = MxAgentId "service.registry.monitoring"

330

331−-- note [registry monitoring agent]

331+{- note [registry monitoring agent]

332−-- this agent listens for 'MxRegistered' and 'MxUnRegistered' events and tracks

332+ This agent listens for 'MxRegistered' and 'MxUnRegistered' events and tracks

333−-- all remote 'ProcessId's that are stored in the registry.

333+ all labels for remote 'ProcessId's that are stored in the registry.

334−--

334+

335−-- When a remote process is registered, the agent starts monitoring it until it

335+ When a remote process is registered, the agent starts monitoring it until it

336−-- is unregistered or the monitor notification arrives.

336+ is unregistered. We can safely ignore ProcessMonitorNotification's, since the

337−--

337+ node controller is obligated to issue an MxUnRegistered event for any labels

338−-- The agent keeps the amount of labels associated with each registered remote

338+ connected to a dead process. Thus, our only responsibility here is to ensure

339−-- process. This is necessary so the process is unmonitored only when it has

339+ that we set up monitors for locally registered processes, and tear them down

340−-- been unregistered from all of the labels.

340+ once unregistration occurs.

341−--

341+ -}

342

343registryMonitorAgent :: Process ProcessId

344registryMonitorAgent = do

345 nid <- getSelfNode

346− -- For each process the map associates the 'MonitorRef' used to monitor it and

347− -- the amount of labels associated with it.

348 mxAgent registryMonitorAgentId (Map.empty :: Map String MonitorRef)

346 mxAgent registryMonitorAgentId (Map.empty :: Map String MonitorRef)

349− [ mxSink $ \(ProcessMonitorNotification mref _ _) -> do

350− mxUpdateLocal $ Map.filter (== mref)

351− mxReady

352− , mxSink $ \ev -> do

347+ [ mxSink $ \ev -> do

353 case ev of

348 case ev of

354 MxRegistered pid label

349 MxRegistered pid label

355 | processNodeId pid /= nid -> do

350 | processNodeId pid /= nid -> do

@@ -363,9 +358,9 @@ registryMonitorAgent = do

363 mxUpdateLocal (Map.delete label)

358 mxUpdateLocal (Map.delete label)

364 _ -> return ()

359 _ -> return ()

365 mxReady

360 mxReady

366− -- remove async answers from mailbox

361+ -- [note: no need to remove async answers from mailbox]

367− , mxSink $ \RegisterReply{} -> mxReady

362+ -- The framework simply discards any input you don't have a handler for.

368− , mxSink $ \DidUnmonitor{} -> mxReady

363+ -- See Management.hs `runAgent` for details.

369 ]

364 ]

370

365

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

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

@@ -961,7 +956,11 @@ ncEffectDied ident reason = do

961

956

962 modify' $ (links ^= unaffectedLinks') . (monitors ^= unaffectedMons')

957 modify' $ (links ^= unaffectedLinks') . (monitors ^= unaffectedMons')

963

958

959+ -- we now consider all labels for this identifier unregistered

964− modify' $ registeredHere ^: Map.filter (\pid -> not $ ident `impliesDeathOf` ProcessIdentifier pid)

960+ let toDrop pid = not $ ident `impliesDeathOf` ProcessIdentifier pid

961+ (keepNames, dropNames) <- Map.partition toDrop <$> gets (^. registeredHere)

962+ mapM_ (\(p, l) -> liftIO $ trace node (MxUnRegistered l p)) (Map.toList dropNames)

963+ modify' $ registeredHere ^= keepNames

965

964

966 remaining <- fmap Map.toList (gets (^. registeredOnNodes)) >>=

965 remaining <- fmap Map.toList (gets (^. registeredOnNodes)) >>=

967 mapM (\(pid,nidlist) ->

966 mapM (\(pid,nidlist) ->