@@ -23,10 +23,11 @@ import Control.Distributed.Process.Management
2323 , mxBroadcast
2424 , mxGetId
2525 )
26−import Control.Monad (void)
26+import Control.Monad (void, unless)
27+import Control.Rematch (equalTo)
2728import Data.Binary
2829import Data.List (find, sort)
29−import Data.Maybe (isJust)
30+import Data.Maybe (isJust, isNothing)
3031import Data.Typeable
3132import GHC.Generics
3233#if ! MIN_VERSION_base(4,6,0)
@@ -44,6 +45,16 @@ data Publish = Publish
4445
4546instance Binary Publish where
4647
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+
4758testAgentBroadcast :: TestResult () -> Process ()
4859testAgentBroadcast result = do
4960 (resultSP, resultRP) <- newChan :: Process (SendPort (), ReceivePort ())
@@ -61,13 +72,13 @@ testAgentBroadcast result = do
6172 -- and the consumer will see that and send the result to our typed channel.
6273 stash result =<< receiveChan resultRP
6374
64− kill publisher "finished"
65− kill consumer "finished"
75+ kill publisher "finished" >> awaitExit publisher
76+ kill consumer "finished" >> awaitExit consumer
6677
6778testAgentDualInput :: TestResult (Maybe Int) -> Process ()
6879testAgentDualInput result = do
6980 (sp, rp) <- newChan
70− _ <- mxAgent (MxAgentId "sum-agent") (0 :: Int) [
81+ s <- mxAgent (MxAgentId "sum-agent") (0 :: Int) [
7182 mxSink $ (\(i :: Int) -> do
7283 mxSetLocal . (+i) =<< mxGetLocal
7384 i' <- mxGetLocal
@@ -84,6 +95,7 @@ testAgentDualInput result = do
8495 mxNotify (5 :: Int)
8596
8697 stash result =<< receiveChanTimeout 10000000 rp
98+ awaitExit s
8799
88100testAgentPrioritisation :: TestResult [String] -> Process ()
89101testAgentPrioritisation result = do
@@ -95,8 +107,8 @@ testAgentPrioritisation result = do
95107
96108 let name = "prioritising-agent"
97109 (sp, rp) <- newChan
98− void $ mxAgent (MxAgentId name) ["first"] [
99− mxSink (\(s :: String) -> do
110+ s <- mxAgent (MxAgentId name) ["first"] [
111+ mxSink (\(s :: String) -> do
100112 mxUpdateLocal ((s:))
101113 st <- mxGetLocal
102114 case length st of
@@ -112,22 +124,25 @@ testAgentPrioritisation result = do
112124 nsend name "fifth"
113125
114126 stash result . sort =<< receiveChan rp
127+ awaitExit s
115128
116129testAgentMailboxHandling :: TestResult (Maybe ()) -> Process ()
117130testAgentMailboxHandling result = do
118131 (sp, rp) <- newChan
119− agent <- mxAgent (MxAgentId "listener-agent") () [
132+ agent <- mxAgent (MxAgentId "mailbox-agent") () [
120133 mxSink $ \() -> (liftMX $ sendChan sp ()) >> mxReady
121134 ]
122135
123− nsend "listener-agent" ()
136+ nsend "mailbox-agent" ()
124137
125138 stash result =<< receiveChanTimeout 1000000 rp
126− kill agent "finished"
139+ kill agent "finished" >> awaitExit agent
127140
128141testAgentEventHandling :: TestResult Bool -> Process ()
129142testAgentEventHandling result = do
130143 let initState = [] :: [MxEvent]
144+
145+ (rc, rs) <- newChan
131146 agentPid <- mxAgent (MxAgentId "lifecycle-listener-agent") initState [
132147 (mxSink $ \ev -> do
133148 st <- mxGetLocal
@@ -136,7 +151,7 @@ testAgentEventHandling result = do
136151 (MxSpawned _) -> mxSetLocal (ev:st)
137152 (MxProcessDied _ _) -> mxSetLocal (ev:st)
138153 _ -> return ()
139− act >> mxReady),
154+ act >> (liftMX $ sendChan rc ()) >> mxReady),
140155 (mxSink $ \(ev, sp :: SendPort Bool) -> do
141156 st <- mxGetLocal
142157 let found =
@@ -156,16 +171,26 @@ testAgentEventHandling result = do
156171 mxReady)
157172 ]
158173
174+ -- TODO: yes, this is racy, but we're at the mercy of the scheduler here...
175+ faff 2000000 rs
176+
159177 _ <- monitor agentPid
160178 (sp, rp) <- newChan
161− pid <- spawnLocal $ sendChan sp ()
162− () <- receiveChan rp
179+
180+ pid <- spawnLocal $ expect >>= sendChan sp
163181
164182 -- 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
166184 monitor pid
185+
186+ send pid ()
187+ () <- receiveChan rp
188+
167189 receiveWait [ match (\(ProcessMonitorNotification _ _ _) -> return ()) ]
168190
191+ -- TODO: yes, this is racy, but we're at the mercy of the scheduler here...
192+ faff 2000000 rs
193+
169194 (replyTo, reply) <- newChan :: Process (SendPort Bool, ReceivePort Bool)
170195 mxNotify (MxSpawned pid, replyTo)
171196 mxNotify (MxProcessDied pid DiedNormal, replyTo)
@@ -174,32 +199,159 @@ testAgentEventHandling result = do
174199 seenDead <- receiveChan reply
175200
176201 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
177317
178318tests :: TestTransport -> IO [Test]
179319tests TestTransport{..} = do
180320 node1 <- newLocalNode testTransport initRemoteTable
321+ node2 <- newLocalNode testTransport initRemoteTable
181322 return [
182− testGroup "Mx Agents" [
183− testCase "Event Handling"
184− (delayedAssertion
185− "expected True, but events where not as expected"
186− node1 True testAgentEventHandling)
187− , testCase "Inter-Agent Broadcast"
188− (delayedAssertion
189− "expected (), but no broadcast was received"
190− node1 () testAgentBroadcast)
191− , testCase "Agent Mailbox Handling"
192− (delayedAssertion
193− "expected (Just ()), but no regular (mailbox) input was handled"
194− node1 (Just ()) testAgentMailboxHandling)
195− , testCase "Agent Dual Input Handling"
196− (delayedAssertion
197− "expected sum = 15, but the result was Nothing"
198− node1 (Just 15 :: Maybe Int) testAgentDualInput)
199− , testCase "Agent Input Prioritisation"
200− (delayedAssertion
201− "expected [first, second, third, fourth, fifth], but result diverged"
202− node1 (sort ["first", "second",
203− "third", "fourth",
204− "fifth"]) testAgentPrioritisation)
205− ]]
323+ testGroup "Mx Agents" [
324+ testCase "Event Handling"
325+ (delayedAssertion
326+ "expected True, but events where not as expected"
327+ node1 True testAgentEventHandling)
328+ , testCase "Inter-Agent Broadcast"
329+ (delayedAssertion
330+ "expected (), but no broadcast was received"
331+ node1 () testAgentBroadcast)
332+ , testCase "Agent Mailbox Handling"
333+ (delayedAssertion
334+ "expected (Just ()), but no regular (mailbox) input was handled"
335+ node1 (Just ()) testAgentMailboxHandling)
336+ , testCase "Agent Dual Input Handling"
337+ (delayedAssertion
338+ "expected sum = 15, but the result was Nothing"
339+ node1 (Just 15 :: Maybe Int) testAgentDualInput)
340+ , testCase "Agent Input Prioritisation"
341+ (delayedAssertion
342+ "expected [first, second, third, fourth, fifth], but result diverged"
343+ node1 (sort ["first", "second",
344+ "third", "fourth",
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))
356+ ]
357+ ]