003e522 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Publish MxUnregistered when a registered name dies

003e522Tim Watson8 years ago

.travis.yml+11−6

compare with master

unifiedsplit

@@ -6,11 +6,11 @@ matrix:

6 include:

7 - env: ARGS="--stack-yaml stack-ghc-7.10.3.yaml"

8 addons: {apt: {packages: [libgmp-dev]}}

9−

10− - env: ARGS="--stack-yaml stack-ghc-8.0.2.yaml"

9+ - env: ARGS="--stack-yaml stack-ghc-8.2.2.yaml"

11 addons: {apt: {packages: [libgmp-dev]}}

10 addons: {apt: {packages: [libgmp-dev]}}

12−

11+ - env: ARGS="--stack-yaml stack-ghc-8.4.4.yaml"

13− - env: ARGS=

12+ addons: {apt: {packages: [libgmp-dev]}}

13+ - env: ARGS="--stack-yaml stack-nightly.yaml --resolver nightly"

14 addons: {apt: {packages: [libgmp-dev]}}

16cache:

@@ -34,5 +34,10 @@ script:

34- stack ${ARGS} test $ARG='--plain -t "!Flaky"' ${TEST_PACKAGE}TestCHInTCP

35- stack ${ARGS} test $ARG='--plain -t "!SpawnReconnect"' ${TEST_PACKAGE}TestClosure

36- stack ${ARGS} test $ARG='--plain' ${TEST_PACKAGE}TestStats

37−- stack ${ARGS} test $ARG='--plain' ${TEST_PACKAGE}TestMx

37+- stack ${ARGS} test $ARG='--plain' ${TEST_PACKAGE}TestMxInMemory

38+- stack ${ARGS} test $ARG='--plain' ${TEST_PACKAGE}TestMxInTCP

38−- stack ${ARGS} test $ARG='--plain' ${TEST_PACKAGE}TestTracing

39+- stack ${ARGS} test $ARG='--plain' ${TEST_PACKAGE}TestTracingInMemory

40+

41+notifications:

42+ slack:

43+ secure: LbqbNjK1KiHNiQo/i/Za4vkMD9l9mhmN+PWYFBQkNvGFWffSRayRKFrTqt4znV2p7h4dY1XZFoC7T5RemBVQLq2ppZAUWxkITeu1OUlnmVLDfJLeVYGTR/fn90nP6Y5ITE7T3A07nKmWaRbKpIgFBAOjzgqaM0csscDx3z0WW18=

distributed-process-tests/distributed-process-tests.cabal+16−3

@@ -114,8 +114,7 @@ Test-Suite TestStats

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

115 HS-Source-Dirs: tests

116

117−

118−Test-Suite TestMx

117+Test-Suite TestMxInMemory

119 Type: exitcode-stdio-1.0

118 Type: exitcode-stdio-1.0

120 Main-Is: runInMemory.hs

119 Main-Is: runInMemory.hs

121 CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Mx

120 CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Mx

@@ -129,7 +128,7 @@ Test-Suite TestMx

129 ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind

128 ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind

130 HS-Source-Dirs: tests

129 HS-Source-Dirs: tests

131

130

132−Test-Suite TestTracing

131+Test-Suite TestTracingInMemory

133 Type: exitcode-stdio-1.0

132 Type: exitcode-stdio-1.0

134 Main-Is: runInMemory.hs

133 Main-Is: runInMemory.hs

135 CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Tracing

134 CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Tracing

@@ -142,3 +141,17 @@ Test-Suite TestTracing

142 Extensions: CPP

141 Extensions: CPP

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

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

144 HS-Source-Dirs: tests

143 HS-Source-Dirs: tests

144+

145+Test-Suite TestMxInTCP

146+ Type: exitcode-stdio-1.0

147+ Main-Is: runInMemory.hs

148+ CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Mx

149+ Build-Depends: base >= 4.4 && < 5,

150+ distributed-process-tests,

151+ network >= 2.3 && < 2.9,

152+ network-transport >= 0.4.1.0 && < 0.6,

153+ network-transport-inmemory >= 0.5,

154+ test-framework >= 0.6 && < 0.9

155+ Extensions: CPP

156+ ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind

157+ HS-Source-Dirs: tests

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

@@ -37,6 +37,9 @@ 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

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

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

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

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

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

@@ -1476,6 +1479,68 @@ testExitRemote TestTransport{..} = do

1476 takeMVar supervisedDone

1479 takeMVar supervisedDone

1477 takeMVar supervisorDone

1480 takeMVar supervisorDone

1478

1481

1482+testRegistryMonitoring :: TestTransport -> Assertion

1483+testRegistryMonitoring TestTransport{..} = do

1484+ node1 <- newLocalNode testTransport initRemoteTable

1485+ node2 <- newLocalNode testTransport initRemoteTable

1486+ waitH <- newEmptyMVar

1487+

1488+ let nid = localNodeId node2

1489+ pid <- forkProcess node1 $ do

1490+ getSelfPid >>= runUntilRegistered nid

1491+ liftIO $ takeMVar waitH

1492+

1493+ runProcess node2 $ do

1494+ register regName pid

1495+ res <- whereis regName

1496+ us <- getSelfPid

1497+ liftIO $ do

1498+ putMVar waitH ()

1499+ assertBool "expected (Just pid)" $ res == (Just pid)

1500+

1501+ -- This delay isn't essential!

1502+ -- The test case passes perfectly fine without it (feel free to comment out

1503+ -- and see), however waiting a few seconds here, makes it much more likely

1504+ -- that in delayUntilMaybeUnregistered we will hit the match case right

1505+ -- away, and thus not be subjected to a 20 second delay. The value of 4

1506+ -- seconds appears to work optimally on osx and across several linux distros

1507+ -- running in virtual machines (which is essentially what we do in CI)

1508+ void $ receiveTimeout 4000000 [ matchAny return ]

1509+

1510+ -- This delay doesn't serve much purpose in the happy path, however if some

1511+ -- future patch breaks the cooperative behaviour of node controllers viz

1512+ -- remote process registration and notification taking place via ncEffectDied,

1513+ -- there would be the possibility of a race in the test case should we attempt

1514+ -- to evaluate `whereis regName` on node2 right away. In case the name is still

1515+ -- erroneously registered, observing the 20 second delay (or lack of), could at

1516+ -- least give a hint that something is wrong, and we give up our time slice

1517+ -- so that there's a higher change the registrations have been cleaned up

1518+ -- in either case.

1519+ runProcess node2 $ delayUntilMaybeUnregistered nid pid

1520+

1521+ regHere <- newEmptyMVar

1522+ runProcess node2 $ whereis regName >>= liftIO . putMVar regHere

1523+ res <- takeMVar regHere

1524+ assertBool "expected Nothing, but process still registered" (res == Nothing)

1525+

1526+ where

1527+ runUntilRegistered nid us = do

1528+ whereisRemoteAsync nid regName

1529+ receiveWait [

1530+ matchIf (\(WhereIsReply n (Just p)) -> n == regName && p == us)

1531+ (const $ return ())

1532+ ]

1533+

1534+ delayUntilMaybeUnregistered nid p = do

1535+ whereisRemoteAsync nid regName

1536+ receiveTimeout 20000000 {- 20 sec delay -} [

1537+ matchIf (\(WhereIsReply n p) -> n == regName && p == Nothing)

1538+ (const $ return ())

1539+ ]

1540+ return ()

1541+

1542+ regName = "testRegisterRemote"

1543+

1479testUnsafeSend :: TestTransport -> Assertion

1544testUnsafeSend :: TestTransport -> Assertion

1480testUnsafeSend TestTransport{..} = do

1545testUnsafeSend TestTransport{..} = do

1481 serverAddr <- newEmptyMVar

1546 serverAddr <- newEmptyMVar

@@ -1677,6 +1742,7 @@ tests testtrans = return [

1677 , testCase "MaskRestoreScope" (testMaskRestoreScope testtrans)

1742 , testCase "MaskRestoreScope" (testMaskRestoreScope testtrans)

1678 , testCase "ExitLocal" (testExitLocal testtrans)

1743 , testCase "ExitLocal" (testExitLocal testtrans)

1679 , testCase "ExitRemote" (testExitRemote testtrans)

1744 , testCase "ExitRemote" (testExitRemote testtrans)

1745+ , testCase "RegistryMonitoring" (testRegistryMonitoring testtrans)

1680 , testCase "TextCallLocal" (testCallLocal testtrans)

1746 , testCase "TextCallLocal" (testCallLocal testtrans)

1681 -- Unsafe Primitives

1747 -- Unsafe Primitives

1682 , testCase "TestUnsafeSend" (testUnsafeSend testtrans)

1748 , testCase "TestUnsafeSend" (testUnsafeSend testtrans)

distributed-process-tests/src/Control/Distributed/Process/Tests/Internal/Utils.hs+11−3

@@ -20,6 +20,7 @@ module Control.Distributed.Process.Tests.Internal.Utils

20 -- ping !

21 , Ping(Ping)

22 , ping

23+ , pause

23 , shouldBe

24 , shouldBe

24 , shouldMatch

25 , shouldMatch

25 , shouldContain

26 , shouldContain

@@ -78,10 +79,10 @@ import Control.Distributed.Process.Node

78import Control.Distributed.Process.Serializable()

79import Control.Distributed.Process.Serializable()

80import Control.Exception (AsyncException(ThreadKilled), SomeException)

81import Control.Exception (AsyncException(ThreadKilled), SomeException)

81−import Control.Monad (forever)

82+import Control.Monad (forever, void)

82import Control.Monad.STM (atomically)

83import Control.Monad.STM (atomically)

83import Control.Rematch hiding (match)

84import Control.Rematch hiding (match)

84−import Control.Rematch.Run

85+import Control.Rematch.Run

85import Data.Binary

86import Data.Binary

86import Data.Typeable (Typeable)

87import Data.Typeable (Typeable)

@@ -107,10 +108,18 @@ data TestProcessControl = Stop | Go | Report ProcessId

107

108

108instance Binary TestProcessControl where

109instance Binary TestProcessControl where

109

110

111+data Private = Private

112+ deriving (Typeable, Generic)

113+instance Binary Private where

114+

110-- | Does exactly what it says on the tin, doing so in the @Process@ monad.

115-- | Does exactly what it says on the tin, doing so in the @Process@ monad.

111noop :: Process ()

116noop :: Process ()

112noop = return ()

117noop = return ()

113

118

119+pause :: Int -> Process ()

120+pause delay =

121+ void $ receiveTimeout delay [ match (\Private -> return ()) ]

122+

114synchronisedAssertion :: Eq a

123synchronisedAssertion :: Eq a

115 => String

124 => String

116 -> LocalNode

125 -> LocalNode

@@ -229,4 +238,3 @@ tryForkProcess :: LocalNode -> Process () -> IO ProcessId

229tryForkProcess node p = do

238tryForkProcess node p = do

230 tid <- liftIO myThreadId

239 tid <- liftIO myThreadId

231 forkProcess node $ catch p (\e -> liftIO $ throwTo tid (e::SomeException))

240 forkProcess node $ catch p (\e -> liftIO $ throwTo tid (e::SomeException))

232−

distributed-process-tests/src/Control/Distributed/Process/Tests/Mx.hs+190−38

@@ -23,10 +23,11 @@ import Control.Distributed.Process.Management

23 , mxBroadcast

24 , mxGetId

25 )

26−import Control.Monad (void)

26+import Control.Monad (void, unless)

27+import Control.Rematch (equalTo)

27import Data.Binary

28import Data.Binary

28import Data.List (find, sort)

29import Data.List (find, sort)

29−import Data.Maybe (isJust)

30+import Data.Maybe (isJust, isNothing)

30import Data.Typeable

31import Data.Typeable

31import GHC.Generics

32import GHC.Generics

32#if ! MIN_VERSION_base(4,6,0)

33#if ! MIN_VERSION_base(4,6,0)

@@ -44,6 +45,16 @@ data Publish = Publish

45instance Binary Publish where

46instance Binary Publish where

48+awaitExit :: ProcessId -> Process ()

49+awaitExit pid =

50+ withMonitorRef pid $ \ref -> do

51+ receiveWait

52+ [ matchIf (\(ProcessMonitorNotification r _ _) -> r == ref)

53+ (\_ -> return ())

54+ ]

55+ where

56+ withMonitorRef pid = bracket (monitor pid) unmonitor

57+

47testAgentBroadcast :: TestResult () -> Process ()

58testAgentBroadcast :: TestResult () -> Process ()

48testAgentBroadcast result = do

59testAgentBroadcast result = do

49 (resultSP, resultRP) <- newChan :: Process (SendPort (), ReceivePort ())

60 (resultSP, resultRP) <- newChan :: Process (SendPort (), ReceivePort ())

@@ -61,13 +72,13 @@ testAgentBroadcast result = do

61 -- and the consumer will see that and send the result to our typed channel.

72 -- and the consumer will see that and send the result to our typed channel.

62 stash result =<< receiveChan resultRP

73 stash result =<< receiveChan resultRP

64− kill publisher "finished"

75+ kill publisher "finished" >> awaitExit publisher

65− kill consumer "finished"

76+ kill consumer "finished" >> awaitExit consumer

67testAgentDualInput :: TestResult (Maybe Int) -> Process ()

78testAgentDualInput :: TestResult (Maybe Int) -> Process ()

68testAgentDualInput result = do

79testAgentDualInput result = do

69 (sp, rp) <- newChan

80 (sp, rp) <- newChan

70− _ <- mxAgent (MxAgentId "sum-agent") (0 :: Int) [

81+ s <- mxAgent (MxAgentId "sum-agent") (0 :: Int) [

71 mxSink $ (\(i :: Int) -> do

82 mxSink $ (\(i :: Int) -> do

72 mxSetLocal . (+i) =<< mxGetLocal

83 mxSetLocal . (+i) =<< mxGetLocal

73 i' <- mxGetLocal

84 i' <- mxGetLocal

@@ -84,6 +95,7 @@ testAgentDualInput result = do

84 mxNotify (5 :: Int)

95 mxNotify (5 :: Int)

86 stash result =<< receiveChanTimeout 10000000 rp

97 stash result =<< receiveChanTimeout 10000000 rp

98+ awaitExit s

88testAgentPrioritisation :: TestResult [String] -> Process ()

100testAgentPrioritisation :: TestResult [String] -> Process ()

89testAgentPrioritisation result = do

101testAgentPrioritisation result = do

@@ -95,8 +107,8 @@ testAgentPrioritisation result = do

107

96 let name = "prioritising-agent"

108 let name = "prioritising-agent"

97 (sp, rp) <- newChan

109 (sp, rp) <- newChan

98− void $ mxAgent (MxAgentId name) ["first"] [

110+ s <- mxAgent (MxAgentId name) ["first"] [

99− mxSink (\(s :: String) -> do

111+ mxSink (\(s :: String) -> do

100 mxUpdateLocal ((s:))

112 mxUpdateLocal ((s:))

101 st <- mxGetLocal

113 st <- mxGetLocal

102 case length st of

114 case length st of

@@ -112,22 +124,25 @@ testAgentPrioritisation result = do

112 nsend name "fifth"

124 nsend name "fifth"

113

125

114 stash result . sort =<< receiveChan rp

126 stash result . sort =<< receiveChan rp

127+ awaitExit s

115

128

116testAgentMailboxHandling :: TestResult (Maybe ()) -> Process ()

129testAgentMailboxHandling :: TestResult (Maybe ()) -> Process ()

117testAgentMailboxHandling result = do

130testAgentMailboxHandling result = do

118 (sp, rp) <- newChan

131 (sp, rp) <- newChan

119− agent <- mxAgent (MxAgentId "listener-agent") () [

132+ agent <- mxAgent (MxAgentId "mailbox-agent") () [

120 mxSink $ \() -> (liftMX $ sendChan sp ()) >> mxReady

133 mxSink $ \() -> (liftMX $ sendChan sp ()) >> mxReady

121 ]

134 ]

122

135

123− nsend "listener-agent" ()

136+ nsend "mailbox-agent" ()

124

137

125 stash result =<< receiveChanTimeout 1000000 rp

138 stash result =<< receiveChanTimeout 1000000 rp

126− kill agent "finished"

139+ kill agent "finished" >> awaitExit agent

127

140

128testAgentEventHandling :: TestResult Bool -> Process ()

141testAgentEventHandling :: TestResult Bool -> Process ()

129testAgentEventHandling result = do

142testAgentEventHandling result = do

130 let initState = [] :: [MxEvent]

143 let initState = [] :: [MxEvent]

144+

145+ (rc, rs) <- newChan

131 agentPid <- mxAgent (MxAgentId "lifecycle-listener-agent") initState [

146 agentPid <- mxAgent (MxAgentId "lifecycle-listener-agent") initState [

132 (mxSink $ \ev -> do

147 (mxSink $ \ev -> do

133 st <- mxGetLocal

148 st <- mxGetLocal

@@ -136,7 +151,7 @@ testAgentEventHandling result = do

136 (MxSpawned _) -> mxSetLocal (ev:st)

151 (MxSpawned _) -> mxSetLocal (ev:st)

137 (MxProcessDied _ _) -> mxSetLocal (ev:st)

152 (MxProcessDied _ _) -> mxSetLocal (ev:st)

138 _ -> return ()

153 _ -> return ()

139− act >> mxReady),

154+ act >> (liftMX $ sendChan rc ()) >> mxReady),

140 (mxSink $ \(ev, sp :: SendPort Bool) -> do

155 (mxSink $ \(ev, sp :: SendPort Bool) -> do

141 st <- mxGetLocal

156 st <- mxGetLocal

142 let found =

157 let found =

@@ -156,16 +171,26 @@ testAgentEventHandling result = do

156 mxReady)

171 mxReady)

157 ]

172 ]

158

173

174+ -- TODO: yes, this is racy, but we're at the mercy of the scheduler here...

175+ faff 2000000 rs

176+

159 _ <- monitor agentPid

177 _ <- monitor agentPid

160 (sp, rp) <- newChan

178 (sp, rp) <- newChan

179+

161− pid <- spawnLocal $ sendChan sp ()

180+ pid <- spawnLocal $ expect >>= sendChan sp

162− () <- receiveChan rp

163

181

164 -- By waiting for a monitor notification, we have a

182 -- By waiting for a monitor notification, we have a

165− -- higher probably that the agent has seen

183+ -- higher probably that the agent has seen the spawn and died events

166 monitor pid

184 monitor pid

185+

186+ send pid ()

187+ () <- receiveChan rp

188+

167 receiveWait [ match (\(ProcessMonitorNotification _ _ _) -> return ()) ]

189 receiveWait [ match (\(ProcessMonitorNotification _ _ _) -> return ()) ]

168

190

191+ -- TODO: yes, this is racy, but we're at the mercy of the scheduler here...

192+ faff 2000000 rs

193+

169 (replyTo, reply) <- newChan :: Process (SendPort Bool, ReceivePort Bool)

194 (replyTo, reply) <- newChan :: Process (SendPort Bool, ReceivePort Bool)

170 mxNotify (MxSpawned pid, replyTo)

195 mxNotify (MxSpawned pid, replyTo)

171 mxNotify (MxProcessDied pid DiedNormal, replyTo)

196 mxNotify (MxProcessDied pid DiedNormal, replyTo)

@@ -174,32 +199,159 @@ testAgentEventHandling result = do

174 seenDead <- receiveChan reply

199 seenDead <- receiveChan reply

175

200

176 stash result $ seenAlive && seenDead

201 stash result $ seenAlive && seenDead

202+ kill agentPid "test-complete" >> awaitExit agentPid

203+ where

204+ faff delay port = do

205+ res <- receiveChanTimeout delay port

206+ unless (isNothing res) $ do

207+ pause delay

208+ faff delay port

209+

210+testMxRegEvents :: TestResult () -> Process ()

211+testMxRegEvents result = do

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

213+ notified in the expected order - the remote cases related to the

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

215+ ensure (stash result ()) $ do

216+ let label = "testMxRegEvents"

217+ let agentLabel = "mxRegEvents-agent"

218+ let delay = 1000000

219+ (regChan, regSink) <- newChan

220+ (unRegChan, unRegSink) <- newChan

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

222+ mxSink $ \ev -> do

223+ case ev of

224+ MxRegistered pid label'

225+ | label' == label -> liftMX $ sendChan regChan (label', pid)

226+ MxUnRegistered pid label'

227+ | label' == label -> liftMX $ sendChan unRegChan (label', pid)

228+ _ -> return ()

229+ mxReady

230+ ]

231+

232+ p1 <- spawnLocal expect

233+ p2 <- spawnLocal expect

234+

235+ register label p1

236+ reg1 <- receiveChanTimeout delay regSink

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

238+

239+ unregister label

240+ unreg1 <- receiveChanTimeout delay unRegSink

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

242+

243+ register label p2

244+ reg2 <- receiveChanTimeout delay regSink

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

246+

247+ reregister label p1

248+ unreg2 <- receiveChanTimeout delay unRegSink

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

250+

251+ reg3 <- receiveChanTimeout delay regSink

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

253+

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

255+ awaitExit agent

256+

257+testMxRegMon :: LocalNode -> TestResult () -> Process ()

258+testMxRegMon remoteNode result = do

259+ ensure (stash result ()) $ do

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

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

262+ let label1 = "aaaaa"

263+ let label2 = "bbbbb"

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

265+ let agentLabel = "mxRegMon-agent"

266+ let delay = 1000000

267+ (regChan, regSink) <- newChan

268+ (unRegChan, unRegSink) <- newChan

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

270+ mxSink $ \ev -> do

271+ case ev of

272+ MxRegistered pid label

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

274+ MxUnRegistered pid label

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

276+ _ -> return ()

277+ mxReady

278+ ]

279+

280+ (sp, rp) <- newChan

281+ liftIO $ forkProcess remoteNode $ do

282+ getSelfPid >>= sendChan sp

283+ expect :: Process ()

284+

285+ p1 <- receiveChan rp

286+

287+ register label1 p1

288+ reg1 <- receiveChanTimeout delay regSink

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

290+

291+ register label2 p1

292+ reg2 <- receiveChanTimeout delay regSink

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

294+

295+ n1 <- whereis label1

296+ n1 `shouldBe` equalTo (Just p1)

297+

298+ n2 <- whereis label2

299+ n2 `shouldBe` equalTo (Just p1)

300+

301+ kill p1 "goodbye"

302+

303+ unreg1 <- receiveChanTimeout delay unRegSink

304+ unreg2 <- receiveChanTimeout delay unRegSink

305+

306+ let evts = [unreg1, unreg2]

307+ -- we can't rely on the order of the values in the node controller's

308+ -- map (it's either racy to do so, or no such guarantee exists for Data.Map),

309+ -- so we simply verify that we received the un-registration events we expect

310+ evts `shouldContain` (Just (label1, p1))

311+ evts `shouldContain` (Just (label2, p1))

312+

313+ kill agent "test-complete" >> awaitExit agent

314+

315+ensure :: Process () -> Process () -> Process ()

316+ensure = flip finally

177

317

178tests :: TestTransport -> IO [Test]

318tests :: TestTransport -> IO [Test]

179tests TestTransport{..} = do

319tests TestTransport{..} = do

180 node1 <- newLocalNode testTransport initRemoteTable

320 node1 <- newLocalNode testTransport initRemoteTable

321+ node2 <- newLocalNode testTransport initRemoteTable

181 return [

322 return [

182− testGroup "Mx Agents" [

323+ testGroup "Mx Agents" [

183− testCase "Event Handling"

324+ testCase "Event Handling"

184− (delayedAssertion

325+ (delayedAssertion

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

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

186− node1 True testAgentEventHandling)

327+ node1 True testAgentEventHandling)

187− , testCase "Inter-Agent Broadcast"

328+ , testCase "Inter-Agent Broadcast"

188− (delayedAssertion

329+ (delayedAssertion

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

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

190− node1 () testAgentBroadcast)

331+ node1 () testAgentBroadcast)

191− , testCase "Agent Mailbox Handling"

332+ , testCase "Agent Mailbox Handling"

192− (delayedAssertion

333+ (delayedAssertion

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

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

194− node1 (Just ()) testAgentMailboxHandling)

335+ node1 (Just ()) testAgentMailboxHandling)

195− , testCase "Agent Dual Input Handling"

336+ , testCase "Agent Dual Input Handling"

196− (delayedAssertion

337+ (delayedAssertion

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

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

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

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

199− , testCase "Agent Input Prioritisation"

340+ , testCase "Agent Input Prioritisation"

200− (delayedAssertion

341+ (delayedAssertion

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

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

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

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

203− "third", "fourth",

344+ "third", "fourth",

204− "fifth"]) testAgentPrioritisation)

345+ "fifth"]) testAgentPrioritisation)

346+ ]

347+ , testGroup "Mx Events" [

348+ testCase "Name Registration Events"

349+ (delayedAssertion

350+ "expected registration events to map to the correct ProcessId"

351+ node1 () testMxRegEvents)

352+ , testCase "Post Death Name UnRegistration Events"

353+ (delayedAssertion

354+ "expected process deaths to result in unregistration events"

355+ node1 () (testMxRegMon node2))

205− ]]

356+ ]

357+ ]

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+23−9

@@ -28,9 +28,12 @@ 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+ , insert

35+ , lookup

36+ , partition

34 , partitionWithKey

37 , partitionWithKey

35 , elems

38 , elems

36 , size

39 , size

@@ -183,6 +186,9 @@ import Control.Distributed.Process.Internal.Types

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

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

184 ( mxAgentController

187 ( mxAgentController

185 )

188 )

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

190+ ( MxEvent(..)

191+ )

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

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

187 ( remoteTable

193 ( remoteTable

188 )

194 )

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

195 , traceLogFmt

201 , traceLogFmt

196 , enableTrace

202 , enableTrace

197 )

203 )

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

199− ( MxEvent(..)

200− )

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

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

202import Control.Distributed.Process.Internal.Messaging

205import Control.Distributed.Process.Internal.Messaging

203 ( sendBinary

206 ( sendBinary

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

210 , match

213 , match

211 , sendChan

214 , sendChan

212 , unwrapMessage

215 , unwrapMessage

216+ , monitor

217+ , unmonitorAsync

218+ , getSelfNode

213 , SayMessage(..)

219 , SayMessage(..)

214 )

220 )

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

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

@@ -903,7 +909,11 @@ ncEffectDied ident reason = do

903

909

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

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

905

911

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

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

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

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

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

916+ modify' $ registeredHere ^= keepNames

907

917

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

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

909 mapM (\(pid,nidlist) ->

919 mapM (\(pid,nidlist) ->

@@ -955,7 +965,11 @@ ncEffectRegister from label atnode mPid reregistration = do

955 do modify' $ registeredHereFor label ^= mPid

965 do modify' $ registeredHereFor label ^= mPid

956 updateRemote node currentVal mPid

966 updateRemote node currentVal mPid

957 case mPid of

967 case mPid of

968+ (Just p) -> do

969+ if reregistration

970+ then liftIO $ trace node (MxUnRegistered (fromJust currentVal) label)

971+ else return ()

958− (Just p) -> liftIO $ trace node (MxRegistered p label)

972+ liftIO $ trace node (MxRegistered p label)

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

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

960 newVal <- gets (^. registeredHereFor label)

974 newVal <- gets (^. registeredHereFor label)

961 ncSendToProcess from $ unsafeCreateUnencodedMessage $

975 ncSendToProcess from $ unsafeCreateUnencodedMessage $

@@ -1070,7 +1084,7 @@ ncEffectGetInfo from pid =

1070 case mProc of

1084 case mProc of

1071 Nothing -> dispatch (isLocal node (ProcessIdentifier from))

1085 Nothing -> dispatch (isLocal node (ProcessIdentifier from))

1072 from (ProcessInfoNone DiedUnknownId)

1086 from (ProcessInfoNone DiedUnknownId)

1073− Just proc -> do

1087+ Just proc -> do

1074 itsLinks <- Set.map fst . BiMultiMap.lookupBy1st them <$>

1088 itsLinks <- Set.map fst . BiMultiMap.lookupBy1st them <$>

1075 gets (^. links)

1089 gets (^. links)

1076 itsMons <- BiMultiMap.lookupBy1st them <$> gets (^. monitors)

1090 itsMons <- BiMultiMap.lookupBy1st them <$> gets (^. monitors)

@@ -1084,8 +1098,8 @@ ncEffectGetInfo from pid =

1084 infoNode = (processNodeId pid)

1098 infoNode = (processNodeId pid)

1085 , infoRegisteredNames = reg

1099 , infoRegisteredNames = reg

1086 , infoMessageQueueLength = size

1100 , infoMessageQueueLength = size

1087− , infoMonitors = Set.toList itsMons

1101+ , infoMonitors = Set.toList itsMons

1088− , infoLinks = Set.toList itsLinks

1102+ , infoLinks = Set.toList itsLinks

1089 }

1103 }

1090 where dispatch :: (Serializable a)

1104 where dispatch :: (Serializable a)

1091 => Bool

1105 => Bool

stack-ghc-8.2.2.yamlnew file+20

@@ -0,0 +1,20 @@

1+resolver: lts-11.22

3+packages:

4+- '.'

5+- distributed-process-tests/

6+- location:

7+ git: https://github.com/haskell-distributed/network-transport-tcp.git

8+ commit: d87d6f55697a94a1fbf211ff9c1bb769a1e129cd

9+ extra-dep: true

10+

11+extra-deps:

12+- rank1dynamic-0.4.0

13+- distributed-static-0.3.8

14+- network-transport-0.5.2

15+- network-transport-inmemory-0.5.2

16+- rematch-0.2.0.0

17+

18+flags:

19+ distributed-process-tests:

20+ tcp: true

stack-ghc-8.4.4.yamlnew file+20

@@ -0,0 +1,20 @@

1+resolver: lts-12.18

3+packages:

4+- '.'

5+- distributed-process-tests/

6+- location:

7+ git: https://github.com/haskell-distributed/network-transport-tcp.git

8+ commit: d87d6f55697a94a1fbf211ff9c1bb769a1e129cd

9+ extra-dep: true

10+

11+extra-deps:

12+- rank1dynamic-0.4.0

13+- distributed-static-0.3.8

14+- network-transport-0.5.2

15+- network-transport-inmemory-0.5.2

16+- rematch-0.2.0.0

17+

18+flags:

19+ distributed-process-tests:

20+ tcp: true

stack-nightly.yaml+3−3

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

1−resolver: nightly-2018-11-10

1+resolver: nightly

3packages:

4- '.'

@@ -9,11 +9,11 @@ packages:

9 extra-dep: true

10- location:

11 git: https://github.com/haskell-distributed/network-transport-tcp.git

12− commit: d87d6f55697a94a1fbf211ff9c1bb769a1e129cd

12+ commit: d58250e2388b86a1865ac4139373cbdc106e72e1

13 extra-dep: true

14- location:

15 git: https://github.com/haskell-distributed/network-transport-inmemory.git

16− commit: b7a10ffc07a77f4fe6affbf8abe535b11933a01a

16+ commit: 0ce97924f15a8c1570b2355151834305c9bd2a28

17 extra-dep: true

19extra-deps:

stack.yaml+1−1

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

1−resolver: nightly-2017-08-20

1+resolver: lts-12.18 # Use GHC 8.4.4

3packages:

4- '.'