@@ -5,6 +5,7 @@ module Control.Distributed.Process.Tests.Mx (tests) where
6import Control.Distributed.Process.Tests.Internal.Utils
6import Control.Distributed.Process.Tests.Internal.Utils
7import Network.Transport.Test (TestTransport(..))
7import Network.Transport.Test (TestTransport(..))
8+import Control.Exception (SomeException)
8import Control.Distributed.Process hiding (bracket, finally, try)
9import Control.Distributed.Process hiding (bracket, finally, try)
9import Control.Distributed.Process.Internal.Types
10import Control.Distributed.Process.Internal.Types
10 ( ProcessExitException(..)
11 ( ProcessExitException(..)
@@ -38,7 +39,7 @@ import Control.Monad.Catch(finally, bracket, try)
38import Control.Rematch (equalTo)
39import Control.Rematch (equalTo)
39import Data.Binary
40import Data.Binary
40import Data.List (find, sort, intercalate)
41import Data.List (find, sort, intercalate)
41−import Data.Maybe (isJust, fromJust, isNothing, fromMaybe, catMaybes)
42+import Data.Maybe (isJust, fromJust, isNothing, fromMaybe)
42import Data.Typeable
43import Data.Typeable
43import GHC.Generics hiding (from)
44import GHC.Generics hiding (from)
44#if ! MIN_VERSION_base(4,6,0)
45#if ! MIN_VERSION_base(4,6,0)
@@ -66,7 +67,7 @@ awaitExit pid =
67 withMonitorRef p = bracket (monitor p) unmonitor
68 withMonitorRef p = bracket (monitor p) unmonitor
69−testAgentBroadcast :: TestResult (Maybe ()) -> Process ()
70+testAgentBroadcast :: TestResult () -> Process ()
70testAgentBroadcast result = do
71testAgentBroadcast result = do
71 (resultSP, resultRP) <- newChan :: Process (SendPort (), ReceivePort ())
72 (resultSP, resultRP) <- newChan :: Process (SendPort (), ReceivePort ())
@@ -81,10 +82,10 @@ testAgentBroadcast result = do
81 mxNotify ()
82 mxNotify ()
82 -- Once the publisher has seen our message, it will broadcast the Publish
83 -- Once the publisher has seen our message, it will broadcast the Publish
83 -- and the consumer will see that and send the result to our typed channel.
84 -- and the consumer will see that and send the result to our typed channel.
84− stash result =<< receiveChanTimeout 10000000 resultRP
85+ stash result =<< receiveChan resultRP
86− kill publisher "finished"
87+ kill publisher "finished" >> awaitExit publisher
87− kill consumer "finished"
88+ kill consumer "finished" >> awaitExit consumer
89testAgentDualInput :: TestResult (Maybe Int) -> Process ()
90testAgentDualInput :: TestResult (Maybe Int) -> Process ()
90testAgentDualInput result = do
91testAgentDualInput result = do
@@ -99,8 +100,6 @@ testAgentDualInput result = do
99 else mxReady)
100 else mxReady)
104 mxNotify (1 :: Int)
103 mxNotify (1 :: Int)
105 nsend "sum-agent" (3 :: Int)
104 nsend "sum-agent" (3 :: Int)
106 mxNotify (2 :: Int)
105 mxNotify (2 :: Int)
@@ -108,10 +107,7 @@ testAgentDualInput result = do
108 mxNotify (5 :: Int)
107 mxNotify (5 :: Int)
110 stash result =<< receiveChanTimeout 10000000 rp
109 stash result =<< receiveChanTimeout 10000000 rp
111− died <- receiveTimeout 10000000 [
110+ awaitExit s
112− matchIf (\(ProcessMonitorNotification r _ _) -> r == mRef) (const $ return True)
114− died `shouldBe` equalTo (Just True)
116testAgentPrioritisation :: TestResult [String] -> Process ()
112testAgentPrioritisation :: TestResult [String] -> Process ()
117testAgentPrioritisation result = do
113testAgentPrioritisation result = do
@@ -125,10 +121,10 @@ testAgentPrioritisation result = do
125 (sp, rp) <- newChan
121 (sp, rp) <- newChan
126 s <- mxAgent (MxAgentId name) ["first"] [
122 s <- mxAgent (MxAgentId name) ["first"] [
127 mxSink (\(s :: String) -> do
123 mxSink (\(s :: String) -> do
128− mxUpdateLocal (s:)
124+ mxUpdateLocal ((s:))
129 st <- mxGetLocal
125 st <- mxGetLocal
130 case length st of
126 case length st of
131− n | n == 5 -> do liftMX $ sendChan sp (sort st)
127+ n | n == 5 -> do liftMX $ sendChan sp st
132 mxDeactivate "finished"
128 mxDeactivate "finished"
133 _ -> mxReceive -- go to the mailbox
129 _ -> mxReceive -- go to the mailbox
@@ -139,7 +135,7 @@ testAgentPrioritisation result = do
139 mxNotify "fourth"
135 mxNotify "fourth"
140 nsend name "fifth"
136 nsend name "fifth"
142− stash result . fromMaybe [] =<< receiveChanTimeout 10000000 rp
138+ stash result . sort =<< receiveChan rp
143 awaitExit s
139 awaitExit s
145testAgentMailboxHandling :: TestResult (Maybe ()) -> Process ()
141testAgentMailboxHandling :: TestResult (Maybe ()) -> Process ()
@@ -152,19 +148,12 @@ testAgentMailboxHandling result = do
152 nsend "mailbox-agent" ()
148 nsend "mailbox-agent" ()
154 stash result =<< receiveChanTimeout 1000000 rp
150 stash result =<< receiveChanTimeout 1000000 rp
155− kill agent "finished"
151+ kill agent "finished" >> awaitExit agent
157testAgentEventHandling :: TestResult Bool -> Process ()
153testAgentEventHandling :: TestResult Bool -> Process ()
158testAgentEventHandling result = do
154testAgentEventHandling result = do
160− -- because this test is a bit racy, let's ensure it can't run indefinitely
161− timer <- spawnLocal $ do
162− pause (1000000 * 60 * 5)
163− -- okay we've waited 5 mins, let's kill the test off if it's stuck...
165− kill us "Test Timed Out"
167 let initState = [] :: [MxEvent]
155 let initState = [] :: [MxEvent]
168 (rc, rs) <- newChan
157 (rc, rs) <- newChan
169 agentPid <- mxAgent (MxAgentId "lifecycle-listener-agent") initState [
158 agentPid <- mxAgent (MxAgentId "lifecycle-listener-agent") initState [
170 (mxSink $ \ev -> do
159 (mxSink $ \ev -> do
@@ -197,7 +186,8 @@ testAgentEventHandling result = do
197 -- TODO: yes, this is racy, but we're at the mercy of the scheduler here...
186 -- TODO: yes, this is racy, but we're at the mercy of the scheduler here...
198 faff 2000000 rs
187 faff 2000000 rs
189+ _ <- monitor agentPid
200− (sp, rp) <- newChan :: Process (SendPort (), ReceivePort ())
190+ (sp, rp) <- newChan
202 pid <- spawnLocal $ expect >>= sendChan sp
192 pid <- spawnLocal $ expect >>= sendChan sp
@@ -206,11 +196,9 @@ testAgentEventHandling result = do
206 monitor pid
196 monitor pid
208 send pid ()
198 send pid ()
209− rct <- receiveChanTimeout 10000000 rp
199+ () <- receiveChan rp
210− unless (isJust rct) $ die "No response on channel"
212− pmn <- receiveTimeout 2000000 [ match (\ProcessMonitorNotification{} -> return ()) ]
201+ receiveWait [ match (\(ProcessMonitorNotification _ _ _) -> return ()) ]
213− unless (isJust pmn) $ die "No monitor notification arrived"
215 -- TODO: yes, this is racy, but we're at the mercy of the scheduler here...
203 -- TODO: yes, this is racy, but we're at the mercy of the scheduler here...
216 faff 2000000 rs
204 faff 2000000 rs
@@ -219,16 +207,17 @@ testAgentEventHandling result = do
219 mxNotify (MxSpawned pid, replyTo)
207 mxNotify (MxSpawned pid, replyTo)
220 mxNotify (MxProcessDied pid DiedNormal, replyTo)
208 mxNotify (MxProcessDied pid DiedNormal, replyTo)
222− seenAlive <- receiveChanTimeout 10000000 reply
210+ seenAlive <- receiveChan reply
223− seenDead <- receiveChanTimeout 10000000 reply
211+ seenDead <- receiveChan reply
225− stash result $ foldr (&&) True $ catMaybes [seenAlive, seenDead]
213+ stash result $ seenAlive && seenDead
226− kill timer "test-complete"
214+ kill agentPid "test-complete" >> awaitExit agentPid
227− kill agentPid "test-complete"
229 faff delay port = do
216 faff delay port = do
230 res <- receiveChanTimeout delay port
217 res <- receiveChanTimeout delay port
231− unless (isNothing res) $ pause delay
218+ unless (isNothing res) $ do
233testMxRegEvents :: TestResult () -> Process ()
222testMxRegEvents :: TestResult () -> Process ()
234testMxRegEvents result = do
223testMxRegEvents result = do
@@ -275,6 +264,7 @@ testMxRegEvents result = do
275 reg3 `shouldBe` equalTo (Just (label, p1))
264 reg3 `shouldBe` equalTo (Just (label, p1))
277 mapM_ (flip kill $ "test-complete") [agent, p1, p2]
266 mapM_ (flip kill $ "test-complete") [agent, p1, p2]
279testMxRegMon :: LocalNode -> TestResult () -> Process ()
269testMxRegMon :: LocalNode -> TestResult () -> Process ()
280testMxRegMon remoteNode result = do
270testMxRegMon remoteNode result = do
@@ -332,70 +322,70 @@ testMxRegMon remoteNode result = do
332 evts `shouldContain` (Just (label1, p1))
322 evts `shouldContain` (Just (label1, p1))
333 evts `shouldContain` (Just (label2, p1))
323 evts `shouldContain` (Just (label2, p1))
335− kill agent "test-complete"
325+ kill agent "test-complete" >> awaitExit agent
337ensure :: Process () -> Process () -> Process ()
327ensure :: Process () -> Process () -> Process ()
338ensure = flip finally
328ensure = flip finally
340testNSend :: (String -> () -> Process ())
330testNSend :: (String -> () -> Process ())
341 -> Maybe LocalNode
331 -> Maybe LocalNode
342 -> Process ()
333 -> Process ()
343−testNSend op n = do
334+testNSend op n r = testMxSend n r $ \p1 sink -> do
345 let delay = 5000000
335 let delay = 5000000
346− let label = "testMxSend" ++ (show us)
336+ let label = "testMxSend"
347 let isValid = isValidLabel n label
337 let isValid = isValidLabel n label
349− testMxSend n label $ \p1 sink -> do
350− register label p1
339+ register label p1
351− reg1 <- receiveChanTimeout delay sink
340+ reg1 <- receiveChanTimeout delay sink
352− case reg1 of
341+ case reg1 of
353− Just (MxRegistered pd lb)
342+ Just (MxRegistered pd lb)
354− | pd == p1 && isValid lb -> return ()
343+ | pd == p1 && isValid lb -> return ()
355− _ -> die $ "Reg-Failed: " ++ show reg1
344+ _ -> die $ "Reg-Failed: " ++ show reg1
357− op label ()
346+ op label ()
359− sent <- receiveChanTimeout delay sink
349+ sent <- receiveChanTimeout delay sink
360− case sent of
350+ case sent of
361− Just (MxSentToName lb by _)
351+ Just (MxSentToName lb by _)
362− | by == us && isValid lb -> return True
352+ | by == us && isValid lb -> return True
363− _ -> die $ "Send-Failed: " ++ show sent
353+ _ -> die $ "Send-Failed: " ++ show sent
366− isValidLabel nd l1 l2
356+ isValidLabel n l1 l2
367 | l2 == l2 = True
357 | l2 == l2 = True
368− | isJust nd = l2 == l1 ++ "@" ++ show (localNodeId $ fromJust nd)
358+ | isJust n = l2 == l1 ++ "@" ++ show (localNodeId $ fromJust n)
369 | otherwise = False
359 | otherwise = False
371testSend :: (ProcessId -> () -> Process ())
361testSend :: (ProcessId -> () -> Process ())
372 -> Maybe LocalNode
362 -> Maybe LocalNode
373 -> Process ()
364 -> Process ()
374−testSend op n = do
365+testSend op n r = testMxSend n r $ \p1 sink -> do
369+ -- verify the management event
375 us <- getSelfPid
370 us <- getSelfPid
376− let label = "testMxSend" ++ (show us)
377− testMxSend n label $ \p1 sink -> do
381− -- verify the management event
382− sent <- receiveChanTimeout 5000000 sink
371+ sent <- receiveChanTimeout 5000000 sink
383− case sent of
372+ case sent of
384− Just (MxSent pidTo pidFrom _)
373+ Just (MxSent pidTo pidFrom _)
385− | pidTo == p1 && pidFrom == us -> return True
374+ | pidTo == p1 && pidFrom == us -> return True
386− _ -> return False
375+ _ -> return False
388testChan :: (SendPort () -> () -> Process ())
377testChan :: (SendPort () -> () -> Process ())
389 -> Maybe LocalNode
378 -> Maybe LocalNode
390 -> Process ()
380 -> Process ()
391−testChan op n = testMxSend n "" $ \p1 sink -> do
381+testChan op n r = testMxSend n r $ \p1 sink -> do
393 us <- getSelfPid
383 us <- getSelfPid
394 send p1 us
384 send p1 us
396− cleared <- receiveChanTimeout 2000000 sink
386+ cleared <- receiveChan sink
397 case cleared of
387 case cleared of
398− Just (MxSent pidTo pidFrom _)
388+ (MxSent pidTo pidFrom _)
399 | pidTo == p1 && pidFrom == us -> return ()
389 | pidTo == p1 && pidFrom == us -> return ()
400 _ -> die "Received uncleared Mx Event"
390 _ -> die "Received uncleared Mx Event"
@@ -414,8 +404,8 @@ testChan op n = testMxSend n "" $ \p1 sink -> do
415type SendTest = ProcessId -> ReceivePort MxEvent -> Process Bool
405type SendTest = ProcessId -> ReceivePort MxEvent -> Process Bool
417−testMxSend :: Maybe LocalNode -> String -> SendTest -> Process ()
407+testMxSend :: Maybe LocalNode -> TestResult Bool -> SendTest -> Process ()
418−testMxSend mNode label test = do
408+testMxSend mNode result test = do
419 us <- getSelfPid
409 us <- getSelfPid
420 (sp, rp) <- newChan
410 (sp, rp) <- newChan
421 (chan, sink) <- newChan
411 (chan, sink) <- newChan
@@ -439,19 +429,16 @@ testMxSend mNode label test = do
439 Nothing -> void $ spawnLocal (proc sp)
429 Nothing -> void $ spawnLocal (proc sp)
440 Just remoteNode -> void $ liftIO $ forkProcess remoteNode $ proc sp
430 Just remoteNode -> void $ liftIO $ forkProcess remoteNode $ proc sp
442− p1 <- receiveChanTimeout 2000000 rp
432+ p1 <- receiveChan rp
443− unless (isJust p1) $ die "Timed out waiting for ProcessId"
444− res <- try $ test (fromJust p1) sink
433+ res <- try (test p1 sink)
447− kill (fromJust p1) "bye"
449 case res of
434 case res of
450− Left (ProcessExitException _ m) -> (liftIO $ putStrLn $ "SomeException-" ++ show m) >> die m
435+ Left (ProcessExitException _ m) -> (liftIO $ putStrLn $ "SomeException-" ++ show m) >> stash result False
451− Right tr -> tr `shouldBe` equalTo True
436+ Right tr -> stash result tr
452−
437+ kill agent "bye"
441+ label = "testMxSend"
455 agentLabel s = "mx-unsafe-check-agent-" ++ show s
442 agentLabel s = "mx-unsafe-check-agent-" ++ show s
456 proc sp' = getSelfPid >>= sendChan sp' >> go Nothing
443 proc sp' = getSelfPid >>= sendChan sp' >> go Nothing
@@ -477,7 +464,7 @@ tests TestTransport{..} = do
477 , testCase "InterAgentBroadcast"
464 , testCase "InterAgentBroadcast"
478 (delayedAssertion
465 (delayedAssertion
479 "expected (), but no broadcast was received"
466 "expected (), but no broadcast was received"
480− node1 (Just ()) testAgentBroadcast)
467+ node1 () testAgentBroadcast)
481 , testCase "AgentMailboxHandling"
468 , testCase "AgentMailboxHandling"
482 (delayedAssertion
469 (delayedAssertion
483 "expected (Just ()), but no regular (mailbox) input was handled"
470 "expected (Just ()), but no regular (mailbox) input was handled"
@@ -502,11 +489,12 @@ tests TestTransport{..} = do
502 (delayedAssertion
489 (delayedAssertion
503 "expected process deaths to result in unregistration events"
490 "expected process deaths to result in unregistration events"
504 node1 () (testMxRegMon node2))
491 node1 () (testMxRegMon node2))
505− , testGroup "SendEvents" $ buildTestCases node1 node2
492+ -- , testGroup "SendEvents" $ buildTestCases node1 node2
496+ buildTestCases n1 n2 =
509− buildTestCases n1 n2 = let nid = localNodeId n2 in build n1 n2 [
497+ let nid = localNodeId n2 in build n2 [
510 ("NSend", [
498 ("NSend", [
511 ("nsend", testNSend nsend)
499 ("nsend", testNSend nsend)
512 , ("Unsafe.nsend", testNSend Unsafe.nsend)
500 , ("Unsafe.nsend", testNSend Unsafe.nsend)
@@ -528,13 +516,13 @@ tests TestTransport{..} = do
530 build :: LocalNode
518 build :: LocalNode
532− -> [(String, [(String, (Maybe LocalNode -> Process ()))])]
519+ -> [(String, [(String, (Maybe LocalNode -> TestResult Bool -> Process ()))])]
533 -> [Test]
520 -> [Test]
534− build n ln specs =
521+ build ln specs =
535 [ testGroup (intercalate "-" [groupName, caseSuffix]) [
522 [ testGroup (intercalate "-" [groupName, caseSuffix]) [
536 testCase (intercalate "-" [caseName, caseSuffix])
523 testCase (intercalate "-" [caseName, caseSuffix])
524+ (delayedAssertion "expected mx events failed"
537− (runProcess n (caseImpl caseNode))
525+ ln True (caseImpl caseNode))
538 | (caseName, caseImpl) <- groupCases
526 | (caseName, caseImpl) <- groupCases
540 | (groupName, groupCases) <- specs
528 | (groupName, groupCases) <- specs