2c9691b · distributed-process

git.iko.soy/distributed-process

on master help sign in

Prevent registered names from leaking

2c9691bTim Watson8 years ago

distributed-process-tests/src/Control/Distributed/Process/Tests/CH.hs+111−1

compare with master

unifiedsplit

@@ -37,9 +37,11 @@ import Control.Distributed.Process.Internal.Types

37 , createUnencodedMessage

38 )

39import Control.Distributed.Process.Node

40+import Control.Distributed.Process.Debug

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

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

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

42−import Test.HUnit (Assertion, assertFailure)

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

43import Test.Framework (Test, testGroup)

45import Test.Framework (Test, testGroup)

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

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

45import Control.Rematch hiding (match)

47import Control.Rematch hiding (match)

@@ -1369,6 +1371,113 @@ testExitRemote TestTransport{..} = do

1369 takeMVar supervisedDone

1371 takeMVar supervisedDone

1370 takeMVar supervisorDone

1372 takeMVar supervisorDone

1371

1373

1374+testRegistryMonitoring :: TestTransport -> Assertion

1375+testRegistryMonitoring TestTransport{..} = do

1376+ localNode <- newLocalNode testTransport initRemoteTable

1377+ remoteNode <- newLocalNode testTransport initRemoteTable

1378+ return ()

1379+

1380+ -- Local process. Test if local process will be removed from

1381+ -- registry when it dies.

1382+ box <- newEmptyMVar

1383+ runProcess localNode $ do

1384+ pid <- spawnLocal $ do

1385+ expect

1386+ register "test" pid

1387+ tpid <- whereis "test"

1388+ if tpid == Just pid

1389+ then do _ <- monitor pid

1390+ send pid ()

1391+ ProcessMonitorNotification{} <- expect

1392+ tpid1 <- whereis "test"

1393+ liftIO $ putMVar box (Nothing == tpid1)

1394+ else liftIO $ putMVar box False

1395+

1396+ takeMVar box >>= assertBool "expected local process to not be registered"

1397+ return ()

1398+

1399+ -- Remote process. Test if remote process entry is removed

1400+ -- from registry when process dies.

1401+ remote1 <- testRemote remoteNode

1402+ runProcess localNode $

1403+ let waitpoll = do

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

1405+ forM_ w (const waitpoll)

1406+ in do register "test" remote1

1407+ send remote1 ()

1408+ waitpoll

1409+ return ()

1410+ return ()

1411+

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

1413+ -- are removed from registry when it dies.

1414+ remote2 <- testRemote remoteNode

1415+ runProcess localNode $

1416+ let waitpoll = do

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

1418+ w2 <- whereis "test-4" :: Process (Maybe ProcessId)

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

1420+ in do register "test-3" remote2

1421+ register "test-4" remote2

1422+ send remote2 ()

1423+ waitpoll

1424+ return ()

1425+

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

1427+ remote3 <- testRemote remoteNode

1428+ remote4 <- testRemote remoteNode

1429+ -- test many labels

1430+ runProcess localNode $ do

1431+ register "test-3" remote3

1432+ reregister "test-3" remote4

1433+ send remote3 ()

1434+ liftIO $ threadDelay 50000 -- XXX: racy

1435+ monitor remote4

1436+ nsend "test-3" ()

1437+ ProcessMonitorNotification{} <- expect

1438+ return ()

1439+-}

1440+

1441+ -- Test registerRemoteAsync properties. Add a local process to

1442+ -- remote registry and checks that it is removed

1443+ -- when the process dies.

1444+ remote5 <- testRemote remoteNode

1445+ runProcess localNode $ do

1446+ registerRemoteAsync (localNodeId remoteNode) "test" remote5

1447+ receiveWait [

1448+ match (\(RegisterReply _ True _) -> return ())

1449+ ] >>= send remote5

1450+ let waitpoll = do

1451+ whereisRemoteAsync (localNodeId remoteNode) "test"

1452+ receiveWait [

1453+ match (\(WhereIsReply _ mr) -> forM_ mr (const waitpoll))

1454+ ]

1455+ waitpoll

1456+

1457+ -- Add remote process to remote registry and checks if

1458+ -- entry is removed then process is dead.

1459+ remote6 <- testRemote localNode

1460+ runProcess localNode $ do

1461+ registerRemoteAsync (localNodeId remoteNode) "test" remote6

1462+ receiveWait [

1463+ match (\(RegisterReply _ True _) -> return ())

1464+ ] >>= send remote6

1465+ let waitpoll = do

1466+ whereisRemoteAsync (localNodeId remoteNode) "test"

1467+ receiveWait [

1468+ match (\(WhereIsReply _ mr) -> forM_ mr (const waitpoll))

1469+ ]

1470+ waitpoll

1471+ where

1472+ testRemote node = do

1473+ -- test many labels

1474+ pidBox <- newEmptyMVar

1475+ forkProcess node $ do

1476+ us <- getSelfPid

1477+ liftIO $ putMVar pidBox us

1478+ expect :: Process ()

1479+ takeMVar pidBox

1480+

1372testUnsafeSend :: TestTransport -> Assertion

1481testUnsafeSend :: TestTransport -> Assertion

1373testUnsafeSend TestTransport{..} = do

1482testUnsafeSend TestTransport{..} = do

1374 serverAddr <- newEmptyMVar

1483 serverAddr <- newEmptyMVar

@@ -1579,6 +1688,7 @@ tests testtrans = return [

1579 , testCase "MaskRestoreScope" (testMaskRestoreScope testtrans)

1688 , testCase "MaskRestoreScope" (testMaskRestoreScope testtrans)

1580 , testCase "ExitLocal" (testExitLocal testtrans)

1689 , testCase "ExitLocal" (testExitLocal testtrans)

1581 , testCase "ExitRemote" (testExitRemote testtrans)

1690 , testCase "ExitRemote" (testExitRemote testtrans)

1691+ , testCase "TestRegistryMonitor" (testRegistryMonitoring testtrans)

1582 , testCase "TextCallLocal" (testCallLocal testtrans)

1692 , testCase "TextCallLocal" (testCallLocal testtrans)

1583 -- Unsafe Primitives

1693 -- Unsafe Primitives

1584 , testCase "TestUnsafeSend" (testUnsafeSend testtrans)

1694 , testCase "TestUnsafeSend" (testUnsafeSend testtrans)

distributed-process-tests/src/Control/Distributed/Process/Tests/Mx.hs+77−24

@@ -24,6 +24,7 @@ import Control.Distributed.Process.Management

24 , mxGetId

25 )

26import Control.Monad (void)

27+import Control.Rematch (equalTo)

27import Data.Binary

28import Data.Binary

28import Data.List (find, sort)

29import Data.List (find, sort)

29import Data.Maybe (isJust)

30import Data.Maybe (isJust)

@@ -175,31 +176,83 @@ testAgentEventHandling result = do

175

176

176 stash result $ seenAlive && seenDead

177 stash result $ seenAlive && seenDead

177

178

179+testMxMonitorEvents :: Process ()

180+testMxMonitorEvents = 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"

187+ let agentLabel = "listener-agent"

188+ let delay = 1000000

189+ (regChan, regSink) <- newChan

190+ (unRegChan, unRegSink) <- newChan

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

192+ mxSink $ \ev -> do

193+ case ev of

194+ MxRegistered pid label

195+ | label /= agentLabel -> liftMX $ sendChan regChan (label, pid)

196+ MxUnRegistered pid label

197+ | label /= agentLabel -> liftMX $ sendChan unRegChan (label, pid)

198+ _ -> return ()

199+ mxReady

200+ ]

201+

202+ p1 <- spawnLocal expect

203+ p2 <- spawnLocal expect

204+

205+ register label p1

206+ reg1 <- receiveChanTimeout delay regSink

207+ reg1 `shouldBe` equalTo (Just (label, p1))

208+

209+ unregister label

210+ unreg1 <- receiveChanTimeout delay unRegSink

211+ unreg1 `shouldBe` equalTo (Just (label, p1))

212+

213+ register label p2

214+ reg2 <- receiveChanTimeout delay regSink

215+ reg2 `shouldBe` equalTo (Just (label, p2))

216+

217+ reregister label p1

218+ unreg2 <- receiveChanTimeout delay unRegSink

219+ unreg2 `shouldBe` equalTo (Just (label, p2))

220+

221+ reg3 <- receiveChanTimeout delay regSink

222+ reg3 `shouldBe` equalTo (Just (label, p1))

223+

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

225+

178tests :: TestTransport -> IO [Test]

226tests :: TestTransport -> IO [Test]

179tests TestTransport{..} = do

227tests TestTransport{..} = do

180 node1 <- newLocalNode testTransport initRemoteTable

228 node1 <- newLocalNode testTransport initRemoteTable

181 return [

229 return [

182− testGroup "Mx Agents" [

230+ testGroup "Mx Agents" [

183− testCase "Event Handling"

231+ testCase "Event Handling"

184− (delayedAssertion

232+ (delayedAssertion

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

233+ "expected True, but events where not as expected"

186− node1 True testAgentEventHandling)

234+ node1 True testAgentEventHandling)

187− , testCase "Inter-Agent Broadcast"

235+ , testCase "Inter-Agent Broadcast"

188− (delayedAssertion

236+ (delayedAssertion

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

237+ "expected (), but no broadcast was received"

190− node1 () testAgentBroadcast)

238+ node1 () testAgentBroadcast)

191− , testCase "Agent Mailbox Handling"

239+ , testCase "Agent Mailbox Handling"

192− (delayedAssertion

240+ (delayedAssertion

193− "expected (Just ()), but no regular (mailbox) input was handled"

241+ "expected (Just ()), but no regular (mailbox) input was handled"

194− node1 (Just ()) testAgentMailboxHandling)

242+ node1 (Just ()) testAgentMailboxHandling)

195− , testCase "Agent Dual Input Handling"

243+ , testCase "Agent Dual Input Handling"

196− (delayedAssertion

244+ (delayedAssertion

197− "expected sum = 15, but the result was Nothing"

245+ "expected sum = 15, but the result was Nothing"

198− node1 (Just 15 :: Maybe Int) testAgentDualInput)

246+ node1 (Just 15 :: Maybe Int) testAgentDualInput)

199− , testCase "Agent Input Prioritisation"

247+ , testCase "Agent Input Prioritisation"

200− (delayedAssertion

248+ (delayedAssertion

201− "expected [first, second, third, fourth, fifth], but result diverged"

249+ "expected [first, second, third, fourth, fifth], but result diverged"

202− node1 (sort ["first", "second",

250+ node1 (sort ["first", "second",

203− "third", "fourth",

251+ "third", "fourth",

204− "fifth"]) testAgentPrioritisation)

252+ "fifth"]) testAgentPrioritisation)

253+ ]

254+ , testGroup "Mx Events" [

255+ testCase "Monitor Events"

256+ (runProcess node1 testMxMonitorEvents)

205− ]]

257+ ]

258+ ]

src/Control/Distributed/Process/Internal/Primitives.hs+1

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

76 , unlink

77 , monitor

78 , unmonitor

79+ , unmonitorAsync

79 , withMonitor

80 , withMonitor

80 , withMonitor_

81 , withMonitor_

81 -- * Logging

82 -- * Logging

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

@@ -30,6 +30,7 @@ import qualified Control.Monad.State as ST

30 ( MonadState

31 , StateT

32 )

33+import Control.Monad.Fix (MonadFix)

33import Data.Binary

34import Data.Binary

34import Data.Typeable (Typeable)

35import Data.Typeable (Typeable)

35import GHC.Generics

36import GHC.Generics

@@ -112,6 +113,7 @@ newtype MxAgent s a =

112 } deriving ( Functor

113 } deriving ( Functor

113 , Monad

114 , Monad

114 , MonadIO

115 , MonadIO

116+ , MonadFix

115 , ST.MonadState (MxAgentState s)

117 , ST.MonadState (MxAgentState s)

116 , Typeable

118 , Typeable

117 , Applicative

119 , Applicative

src/Control/Distributed/Process/Node.hs+61−3

@@ -28,9 +28,13 @@ import Data.Binary (decode)

28import Data.Map (Map)

29import qualified Data.Map as Map

30 ( empty

31+ , delete

31 , toList

32 , toList

32 , fromList

33 , fromList

33 , filter

34 , filter

35+ , insert

36+ , lookup

37+ , insertLookupWithKey

34 , partitionWithKey

38 , partitionWithKey

35 , elems

39 , elems

36 , size

40 , size

@@ -183,6 +187,17 @@ import Control.Distributed.Process.Internal.Types

183import Control.Distributed.Process.Management.Internal.Agent

187import Control.Distributed.Process.Management.Internal.Agent

184 ( mxAgentController

188 ( mxAgentController

185 )

189 )

190+import Control.Distributed.Process.Management

191+ ( mxAgent

192+ , mxSink

193+ , liftMX

194+ , mxUpdateLocal

195+ , mxGetLocal

196+ , mxSetLocal

197+ , mxReady

198+ , MxEvent(..)

199+ , MxAgentId(..)

200+ )

186import qualified Control.Distributed.Process.Management.Internal.Trace.Remote as Trace

201import qualified Control.Distributed.Process.Management.Internal.Trace.Remote as Trace

187 ( remoteTable

202 ( remoteTable

188 )

203 )

@@ -195,9 +210,6 @@ import Control.Distributed.Process.Management.Internal.Trace.Types

195 , traceLogFmt

210 , traceLogFmt

196 , enableTrace

211 , enableTrace

197 )

212 )

198−import Control.Distributed.Process.Management.Internal.Types

199− ( MxEvent(..)

200− )

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

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

202import Control.Distributed.Process.Internal.Messaging

214import Control.Distributed.Process.Internal.Messaging

203 ( sendBinary

215 ( sendBinary

@@ -210,6 +222,9 @@ import Control.Distributed.Process.Internal.Primitives

210 , match

222 , match

211 , sendChan

223 , sendChan

212 , unwrapMessage

224 , unwrapMessage

225+ , monitor

226+ , unmonitorAsync

227+ , getSelfNode

213 , SayMessage(..)

228 , SayMessage(..)

214 )

229 )

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

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

@@ -310,6 +325,49 @@ startDefaultTracer node' = do

310

325

311-- TODO: we need a better mechanism for defining and registering services

326-- TODO: we need a better mechanism for defining and registering services

312

327

328+registryMonitorAgentId :: MxAgentId

329+registryMonitorAgentId = MxAgentId "service.registry.monitoring"

330+

331+-- note [registry monitoring agent]

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

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

334+--

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

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

337+--

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

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

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

341+--

342+

343+registryMonitorAgent :: Process ProcessId

344+registryMonitorAgent = 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)

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

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

351+ mxReady

352+ , mxSink $ \ev -> do

353+ case ev of

354+ MxRegistered pid label

355+ | processNodeId pid /= nid -> do

356+ mref <- liftMX $ monitor pid

357+ mxUpdateLocal (Map.insert label mref)

358+ MxUnRegistered pid label

359+ | processNodeId pid /= nid -> do

360+ mrefs <- mxGetLocal

361+ forM_ (label `Map.lookup` mrefs) $ \mref -> do

362+ liftMX $ unmonitorAsync mref

363+ mxUpdateLocal (Map.delete label)

364+ _ -> return ()

365+ mxReady

366+ -- remove async answers from mailbox

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

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

369+ ]

370+

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

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

314startServiceProcesses :: LocalNode -> IO ()

372startServiceProcesses :: LocalNode -> IO ()

315startServiceProcesses node = do

373startServiceProcesses node = do