@@ -5,7 +5,6 @@ module Control.Distributed.Process.Tests.Mx (tests) where
55
66import Control.Distributed.Process.Tests.Internal.Utils
77import Network.Transport.Test (TestTransport(..))
8−import Control.Exception (SomeException)
98import Control.Distributed.Process hiding (bracket, finally, try)
109import Control.Distributed.Process.Internal.Types
1110 ( ProcessExitException(..)
@@ -39,7 +38,7 @@ import Control.Monad.Catch(finally, bracket, try)
3938import Control.Rematch (equalTo)
4039import Data.Binary
4140import Data.List (find, sort, intercalate)
42−import Data.Maybe (isJust, fromJust, isNothing, fromMaybe)
41+import Data.Maybe (isJust, fromJust, isNothing, fromMaybe, catMaybes)
4342import Data.Typeable
4443import GHC.Generics hiding (from)
4544#if ! MIN_VERSION_base(4,6,0)
@@ -67,7 +66,7 @@ awaitExit pid =
6766 where
6867 withMonitorRef p = bracket (monitor p) unmonitor
6968
70−testAgentBroadcast :: TestResult () -> Process ()
69+testAgentBroadcast :: TestResult (Maybe ()) -> Process ()
7170testAgentBroadcast result = do
7271 (resultSP, resultRP) <- newChan :: Process (SendPort (), ReceivePort ())
7372
@@ -82,10 +81,10 @@ testAgentBroadcast result = do
8281 mxNotify ()
8382 -- Once the publisher has seen our message, it will broadcast the Publish
8483 -- and the consumer will see that and send the result to our typed channel.
85− stash result =<< receiveChan resultRP
84+ stash result =<< receiveChanTimeout 10000000 resultRP
8685
87− kill publisher "finished" >> awaitExit publisher
88− kill consumer "finished" >> awaitExit consumer
86+ kill publisher "finished"
87+ kill consumer "finished"
8988
9089testAgentDualInput :: TestResult (Maybe Int) -> Process ()
9190testAgentDualInput result = do
@@ -100,6 +99,8 @@ testAgentDualInput result = do
10099 else mxReady)
101100 ]
102101
102+ mRef <- monitor s
103+
103104 mxNotify (1 :: Int)
104105 nsend "sum-agent" (3 :: Int)
105106 mxNotify (2 :: Int)
@@ -107,7 +108,10 @@ testAgentDualInput result = do
107108 mxNotify (5 :: Int)
108109
109110 stash result =<< receiveChanTimeout 10000000 rp
110− awaitExit s
111+ died <- receiveTimeout 10000000 [
112+ matchIf (\(ProcessMonitorNotification r _ _) -> r == mRef) (const $ return True)
113+ ]
114+ died `shouldBe` equalTo (Just True)
111115
112116testAgentPrioritisation :: TestResult [String] -> Process ()
113117testAgentPrioritisation result = do
@@ -121,10 +125,10 @@ testAgentPrioritisation result = do
121125 (sp, rp) <- newChan
122126 s <- mxAgent (MxAgentId name) ["first"] [
123127 mxSink (\(s :: String) -> do
124− mxUpdateLocal ((s:))
128+ mxUpdateLocal (s:)
125129 st <- mxGetLocal
126130 case length st of
127− n | n == 5 -> do liftMX $ sendChan sp st
131+ n | n == 5 -> do liftMX $ sendChan sp (sort st)
128132 mxDeactivate "finished"
129133 _ -> mxReceive -- go to the mailbox
130134 )
@@ -135,7 +139,7 @@ testAgentPrioritisation result = do
135139 mxNotify "fourth"
136140 nsend name "fifth"
137141
138− stash result . sort =<< receiveChan rp
142+ stash result . fromMaybe [] =<< receiveChanTimeout 10000000 rp
139143 awaitExit s
140144
141145testAgentMailboxHandling :: TestResult (Maybe ()) -> Process ()
@@ -148,12 +152,19 @@ testAgentMailboxHandling result = do
148152 nsend "mailbox-agent" ()
149153
150154 stash result =<< receiveChanTimeout 1000000 rp
151− kill agent "finished" >> awaitExit agent
155+ kill agent "finished"
152156
153157testAgentEventHandling :: TestResult Bool -> Process ()
154158testAgentEventHandling result = do
155− let initState = [] :: [MxEvent]
159+ us <- getSelfPid
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...
164+ stash result False
165+ kill us "Test Timed Out"
156166
167+ let initState = [] :: [MxEvent]
157168 (rc, rs) <- newChan
158169 agentPid <- mxAgent (MxAgentId "lifecycle-listener-agent") initState [
159170 (mxSink $ \ev -> do
@@ -186,8 +197,7 @@ testAgentEventHandling result = do
186197 -- TODO: yes, this is racy, but we're at the mercy of the scheduler here...
187198 faff 2000000 rs
188199
189− _ <- monitor agentPid
190− (sp, rp) <- newChan
200+ (sp, rp) <- newChan :: Process (SendPort (), ReceivePort ())
191201
192202 pid <- spawnLocal $ expect >>= sendChan sp
193203
@@ -196,9 +206,11 @@ testAgentEventHandling result = do
196206 monitor pid
197207
198208 send pid ()
199− () <- receiveChan rp
209+ rct <- receiveChanTimeout 10000000 rp
210+ unless (isJust rct) $ die "No response on channel"
200211
201− receiveWait [ match (\(ProcessMonitorNotification _ _ _) -> return ()) ]
212+ pmn <- receiveTimeout 2000000 [ match (\ProcessMonitorNotification{} -> return ()) ]
213+ unless (isJust pmn) $ die "No monitor notification arrived"
202214
203215 -- TODO: yes, this is racy, but we're at the mercy of the scheduler here...
204216 faff 2000000 rs
@@ -207,17 +219,16 @@ testAgentEventHandling result = do
207219 mxNotify (MxSpawned pid, replyTo)
208220 mxNotify (MxProcessDied pid DiedNormal, replyTo)
209221
210− seenAlive <- receiveChan reply
211− seenDead <- receiveChan reply
222+ seenAlive <- receiveChanTimeout 10000000 reply
223+ seenDead <- receiveChanTimeout 10000000 reply
212224
213− stash result $ seenAlive && seenDead
214− kill agentPid "test-complete" >> awaitExit agentPid
225+ stash result $ foldr (&&) True $ catMaybes [seenAlive, seenDead]
226+ kill timer "test-complete"
227+ kill agentPid "test-complete"
215228 where
216229 faff delay port = do
217230 res <- receiveChanTimeout delay port
218− unless (isNothing res) $ do
219− pause delay
220− faff delay port
231+ unless (isNothing res) $ pause delay
221232
222233testMxRegEvents :: TestResult () -> Process ()
223234testMxRegEvents result = do
@@ -264,7 +275,6 @@ testMxRegEvents result = do
264275 reg3 `shouldBe` equalTo (Just (label, p1))
265276
266277 mapM_ (flip kill $ "test-complete") [agent, p1, p2]
267− awaitExit agent
268278
269279testMxRegMon :: LocalNode -> TestResult () -> Process ()
270280testMxRegMon remoteNode result = do
@@ -322,70 +332,70 @@ testMxRegMon remoteNode result = do
322332 evts `shouldContain` (Just (label1, p1))
323333 evts `shouldContain` (Just (label2, p1))
324334
325− kill agent "test-complete" >> awaitExit agent
335+ kill agent "test-complete"
326336
327337ensure :: Process () -> Process () -> Process ()
328338ensure = flip finally
329339
330340testNSend :: (String -> () -> Process ())
331341 -> Maybe LocalNode
332− -> TestResult Bool
333342 -> Process ()
334−testNSend op n r = testMxSend n r $ \p1 sink -> do
343+testNSend op n = do
344+ us <- getSelfPid
335345 let delay = 5000000
336− let label = "testMxSend"
346+ let label = "testMxSend" ++ (show us)
337347 let isValid = isValidLabel n label
338348
339− register label p1
340− reg1 <- receiveChanTimeout delay sink
341− case reg1 of
342− Just (MxRegistered pd lb)
343− | pd == p1 && isValid lb -> return ()
344− _ -> die $ "Reg-Failed: " ++ show reg1
349+ testMxSend n label $ \p1 sink -> do
350+ register label p1
351+ reg1 <- receiveChanTimeout delay sink
352+ case reg1 of
353+ Just (MxRegistered pd lb)
354+ | pd == p1 && isValid lb -> return ()
355+ _ -> die $ "Reg-Failed: " ++ show reg1
345356
346− op label ()
357+ op label ()
347358
348− us <- getSelfPid
349− sent <- receiveChanTimeout delay sink
350− case sent of
351− Just (MxSentToName lb by _)
352− | by == us && isValid lb -> return True
353− _ -> die $ "Send-Failed: " ++ show sent
359+ sent <- receiveChanTimeout delay sink
360+ case sent of
361+ Just (MxSentToName lb by _)
362+ | by == us && isValid lb -> return True
363+ _ -> die $ "Send-Failed: " ++ show sent
354364
355365 where
356− isValidLabel n l1 l2
366+ isValidLabel nd l1 l2
357367 | l2 == l2 = True
358− | isJust n = l2 == l1 ++ "@" ++ show (localNodeId $ fromJust n)
368+ | isJust nd = l2 == l1 ++ "@" ++ show (localNodeId $ fromJust nd)
359369 | otherwise = False
360370
361371testSend :: (ProcessId -> () -> Process ())
362372 -> Maybe LocalNode
363− -> TestResult Bool
364373 -> Process ()
365−testSend op n r = testMxSend n r $ \p1 sink -> do
366− -- initiate a send
367− op p1 ()
368−
369− -- verify the management event
374+testSend op n = do
370375 us <- getSelfPid
371− sent <- receiveChanTimeout 5000000 sink
372− case sent of
373− Just (MxSent pidTo pidFrom _)
374− | pidTo == p1 && pidFrom == us -> return True
375− _ -> return False
376+ let label = "testMxSend" ++ (show us)
377+ testMxSend n label $ \p1 sink -> do
378+ -- initiate a send
379+ op p1 ()
380+
381+ -- verify the management event
382+ sent <- receiveChanTimeout 5000000 sink
383+ case sent of
384+ Just (MxSent pidTo pidFrom _)
385+ | pidTo == p1 && pidFrom == us -> return True
386+ _ -> return False
376387
377388testChan :: (SendPort () -> () -> Process ())
378389 -> Maybe LocalNode
379− -> TestResult Bool
380390 -> Process ()
381−testChan op n r = testMxSend n r $ \p1 sink -> do
391+testChan op n = testMxSend n "" $ \p1 sink -> do
382392
383393 us <- getSelfPid
384394 send p1 us
385395
386− cleared <- receiveChan sink
396+ cleared <- receiveChanTimeout 2000000 sink
387397 case cleared of
388− (MxSent pidTo pidFrom _)
398+ Just (MxSent pidTo pidFrom _)
389399 | pidTo == p1 && pidFrom == us -> return ()
390400 _ -> die "Received uncleared Mx Event"
391401
@@ -404,8 +414,8 @@ testChan op n r = testMxSend n r $ \p1 sink -> do
404414
405415type SendTest = ProcessId -> ReceivePort MxEvent -> Process Bool
406416
407−testMxSend :: Maybe LocalNode -> TestResult Bool -> SendTest -> Process ()
408−testMxSend mNode result test = do
417+testMxSend :: Maybe LocalNode -> String -> SendTest -> Process ()
418+testMxSend mNode label test = do
409419 us <- getSelfPid
410420 (sp, rp) <- newChan
411421 (chan, sink) <- newChan
@@ -429,16 +439,19 @@ testMxSend mNode result test = do
429439 Nothing -> void $ spawnLocal (proc sp)
430440 Just remoteNode -> void $ liftIO $ forkProcess remoteNode $ proc sp
431441
432− p1 <- receiveChan rp
433− res <- try (test p1 sink)
434− case res of
435− Left (ProcessExitException _ m) -> (liftIO $ putStrLn $ "SomeException-" ++ show m) >> stash result False
436− Right tr -> stash result tr
442+ p1 <- receiveChanTimeout 2000000 rp
443+ unless (isJust p1) $ die "Timed out waiting for ProcessId"
444+ res <- try $ test (fromJust p1) sink
445+
437446 kill agent "bye"
438− kill p1 "bye"
447+ kill (fromJust p1) "bye"
448+
449+ case res of
450+ Left (ProcessExitException _ m) -> (liftIO $ putStrLn $ "SomeException-" ++ show m) >> die m
451+ Right tr -> tr `shouldBe` equalTo True
452+
439453
440454 where
441− label = "testMxSend"
442455 agentLabel s = "mx-unsafe-check-agent-" ++ show s
443456 proc sp' = getSelfPid >>= sendChan sp' >> go Nothing
444457
@@ -464,7 +477,7 @@ tests TestTransport{..} = do
464477 , testCase "InterAgentBroadcast"
465478 (delayedAssertion
466479 "expected (), but no broadcast was received"
467− node1 () testAgentBroadcast)
480+ node1 (Just ()) testAgentBroadcast)
468481 , testCase "AgentMailboxHandling"
469482 (delayedAssertion
470483 "expected (Just ()), but no regular (mailbox) input was handled"
@@ -489,12 +502,11 @@ tests TestTransport{..} = do
489502 (delayedAssertion
490503 "expected process deaths to result in unregistration events"
491504 node1 () (testMxRegMon node2))
492− -- , testGroup "SendEvents" $ buildTestCases node1 node2
505+ , testGroup "SendEvents" $ buildTestCases node1 node2
493506 ]
494507 ]
495508 where
496− buildTestCases n1 n2 =
497− let nid = localNodeId n2 in build n2 [
509+ buildTestCases n1 n2 = let nid = localNodeId n2 in build n1 n2 [
498510 ("NSend", [
499511 ("nsend", testNSend nsend)
500512 , ("Unsafe.nsend", testNSend Unsafe.nsend)
@@ -516,13 +528,13 @@ tests TestTransport{..} = do
516528 ]
517529
518530 build :: LocalNode
519− -> [(String, [(String, (Maybe LocalNode -> TestResult Bool -> Process ()))])]
531+ -> LocalNode
532+ -> [(String, [(String, (Maybe LocalNode -> Process ()))])]
520533 -> [Test]
521− build ln specs =
534+ build n ln specs =
522535 [ testGroup (intercalate "-" [groupName, caseSuffix]) [
523536 testCase (intercalate "-" [caseName, caseSuffix])
524− (delayedAssertion "expected mx events failed"
525− ln True (caseImpl caseNode))
537+ (runProcess n (caseImpl caseNode))
526538 | (caseName, caseImpl) <- groupCases
527539 ]
528540 | (groupName, groupCases) <- specs