@@ -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)
@@ -50,7 +49,6 @@ import Test.Framework
5049 ( Test
5150 , testGroup
5251 )
53−import Test.HUnit (Assertion)
5452import Test.Framework.Providers.HUnit (testCase)
5553
5654data Publish = Publish
@@ -68,7 +66,7 @@ awaitExit pid =
6866 where
6967 withMonitorRef p = bracket (monitor p) unmonitor
7068
71−testAgentBroadcast :: TestResult () -> Process ()
69+testAgentBroadcast :: TestResult (Maybe ()) -> Process ()
7270testAgentBroadcast result = do
7371 (resultSP, resultRP) <- newChan :: Process (SendPort (), ReceivePort ())
7472
@@ -83,10 +81,10 @@ testAgentBroadcast result = do
8381 mxNotify ()
8482 -- Once the publisher has seen our message, it will broadcast the Publish
8583 -- and the consumer will see that and send the result to our typed channel.
86− stash result =<< receiveChan resultRP
84+ stash result =<< receiveChanTimeout 10000000 resultRP
8785
88− kill publisher "finished" >> awaitExit publisher
89− kill consumer "finished" >> awaitExit consumer
86+ kill publisher "finished"
87+ kill consumer "finished"
9088
9189testAgentDualInput :: TestResult (Maybe Int) -> Process ()
9290testAgentDualInput result = do
@@ -101,6 +99,8 @@ testAgentDualInput result = do
10199 else mxReady)
102100 ]
103101
102+ mRef <- monitor s
103+
104104 mxNotify (1 :: Int)
105105 nsend "sum-agent" (3 :: Int)
106106 mxNotify (2 :: Int)
@@ -108,7 +108,10 @@ testAgentDualInput result = do
108108 mxNotify (5 :: Int)
109109
110110 stash result =<< receiveChanTimeout 10000000 rp
111− awaitExit s
111+ died <- receiveTimeout 10000000 [
112+ matchIf (\(ProcessMonitorNotification r _ _) -> r == mRef) (const $ return True)
113+ ]
114+ died `shouldBe` equalTo (Just True)
112115
113116testAgentPrioritisation :: TestResult [String] -> Process ()
114117testAgentPrioritisation result = do
@@ -122,10 +125,10 @@ testAgentPrioritisation result = do
122125 (sp, rp) <- newChan
123126 s <- mxAgent (MxAgentId name) ["first"] [
124127 mxSink (\(s :: String) -> do
125− mxUpdateLocal ((s:))
128+ mxUpdateLocal (s:)
126129 st <- mxGetLocal
127130 case length st of
128− n | n == 5 -> do liftMX $ sendChan sp st
131+ n | n == 5 -> do liftMX $ sendChan sp (sort st)
129132 mxDeactivate "finished"
130133 _ -> mxReceive -- go to the mailbox
131134 )
@@ -136,7 +139,7 @@ testAgentPrioritisation result = do
136139 mxNotify "fourth"
137140 nsend name "fifth"
138141
139− stash result . sort =<< receiveChan rp
142+ stash result . fromMaybe [] =<< receiveChanTimeout 10000000 rp
140143 awaitExit s
141144
142145testAgentMailboxHandling :: TestResult (Maybe ()) -> Process ()
@@ -149,12 +152,19 @@ testAgentMailboxHandling result = do
149152 nsend "mailbox-agent" ()
150153
151154 stash result =<< receiveChanTimeout 1000000 rp
152− kill agent "finished" >> awaitExit agent
155+ kill agent "finished"
153156
154157testAgentEventHandling :: TestResult Bool -> Process ()
155158testAgentEventHandling result = do
156− 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"
157166
167+ let initState = [] :: [MxEvent]
158168 (rc, rs) <- newChan
159169 agentPid <- mxAgent (MxAgentId "lifecycle-listener-agent") initState [
160170 (mxSink $ \ev -> do
@@ -187,8 +197,7 @@ testAgentEventHandling result = do
187197 -- TODO: yes, this is racy, but we're at the mercy of the scheduler here...
188198 faff 2000000 rs
189199
190− _ <- monitor agentPid
191− (sp, rp) <- newChan
200+ (sp, rp) <- newChan :: Process (SendPort (), ReceivePort ())
192201
193202 pid <- spawnLocal $ expect >>= sendChan sp
194203
@@ -197,9 +206,11 @@ testAgentEventHandling result = do
197206 monitor pid
198207
199208 send pid ()
200− () <- receiveChan rp
209+ rct <- receiveChanTimeout 10000000 rp
210+ unless (isJust rct) $ die "No response on channel"
201211
202− receiveWait [ match (\(ProcessMonitorNotification _ _ _) -> return ()) ]
212+ pmn <- receiveTimeout 2000000 [ match (\ProcessMonitorNotification{} -> return ()) ]
213+ unless (isJust pmn) $ die "No monitor notification arrived"
203214
204215 -- TODO: yes, this is racy, but we're at the mercy of the scheduler here...
205216 faff 2000000 rs
@@ -208,17 +219,16 @@ testAgentEventHandling result = do
208219 mxNotify (MxSpawned pid, replyTo)
209220 mxNotify (MxProcessDied pid DiedNormal, replyTo)
210221
211− seenAlive <- receiveChan reply
212− seenDead <- receiveChan reply
222+ seenAlive <- receiveChanTimeout 10000000 reply
223+ seenDead <- receiveChanTimeout 10000000 reply
213224
214− stash result $ seenAlive && seenDead
215− kill agentPid "test-complete" >> awaitExit agentPid
225+ stash result $ foldr (&&) True $ catMaybes [seenAlive, seenDead]
226+ kill timer "test-complete"
227+ kill agentPid "test-complete"
216228 where
217229 faff delay port = do
218230 res <- receiveChanTimeout delay port
219− unless (isNothing res) $ do
220− pause delay
221− faff delay port
231+ unless (isNothing res) $ pause delay
222232
223233testMxRegEvents :: TestResult () -> Process ()
224234testMxRegEvents result = do
@@ -265,7 +275,6 @@ testMxRegEvents result = do
265275 reg3 `shouldBe` equalTo (Just (label, p1))
266276
267277 mapM_ (flip kill $ "test-complete") [agent, p1, p2]
268− awaitExit agent
269278
270279testMxRegMon :: LocalNode -> TestResult () -> Process ()
271280testMxRegMon remoteNode result = do
@@ -323,7 +332,7 @@ testMxRegMon remoteNode result = do
323332 evts `shouldContain` (Just (label1, p1))
324333 evts `shouldContain` (Just (label2, p1))
325334
326− kill agent "test-complete" >> awaitExit agent
335+ kill agent "test-complete"
327336
328337ensure :: Process () -> Process () -> Process ()
329338ensure = flip finally
@@ -331,59 +340,62 @@ ensure = flip finally
331340testNSend :: (String -> () -> Process ())
332341 -> Maybe LocalNode
333342 -> Process ()
334−testNSend op n = testMxSend n $ \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
363373 -> Process ()
364−testSend op n = testMxSend n $ \p1 sink -> do
365− -- initiate a send
366− op p1 ()
367−
368− -- verify the management event
374+testSend op n = do
369375 us <- getSelfPid
370− sent <- receiveChanTimeout 5000000 sink
371− case sent of
372− Just (MxSent pidTo pidFrom _)
373− | pidTo == p1 && pidFrom == us -> return True
374− _ -> 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
375387
376388testChan :: (SendPort () -> () -> Process ())
377389 -> Maybe LocalNode
378390 -> Process ()
379−testChan op n = testMxSend n $ \p1 sink -> do
391+testChan op n = testMxSend n "" $ \p1 sink -> do
380392
381393 us <- getSelfPid
382394 send p1 us
383395
384− cleared <- receiveChan sink
396+ cleared <- receiveChanTimeout 2000000 sink
385397 case cleared of
386− (MxSent pidTo pidFrom _)
398+ Just (MxSent pidTo pidFrom _)
387399 | pidTo == p1 && pidFrom == us -> return ()
388400 _ -> die "Received uncleared Mx Event"
389401
@@ -402,8 +414,8 @@ testChan op n = testMxSend n $ \p1 sink -> do
402414
403415type SendTest = ProcessId -> ReceivePort MxEvent -> Process Bool
404416
405−testMxSend :: Maybe LocalNode -> SendTest -> Process ()
406−testMxSend mNode test = do
417+testMxSend :: Maybe LocalNode -> String -> SendTest -> Process ()
418+testMxSend mNode label test = do
407419 us <- getSelfPid
408420 (sp, rp) <- newChan
409421 (chan, sink) <- newChan
@@ -427,11 +439,12 @@ testMxSend mNode test = do
427439 Nothing -> void $ spawnLocal (proc sp)
428440 Just remoteNode -> void $ liftIO $ forkProcess remoteNode $ proc sp
429441
430− p1 <- receiveChan rp
431− res <- try (test p1 sink)
442+ p1 <- receiveChanTimeout 2000000 rp
443+ unless (isJust p1) $ die "Timed out waiting for ProcessId"
444+ res <- try $ test (fromJust p1) sink
432445
433446 kill agent "bye"
434− kill p1 "bye"
447+ kill (fromJust p1) "bye"
435448
436449 case res of
437450 Left (ProcessExitException _ m) -> (liftIO $ putStrLn $ "SomeException-" ++ show m) >> die m
@@ -439,7 +452,6 @@ testMxSend mNode test = do
439452
440453
441454 where
442− label = "testMxSend"
443455 agentLabel s = "mx-unsafe-check-agent-" ++ show s
444456 proc sp' = getSelfPid >>= sendChan sp' >> go Nothing
445457
@@ -453,7 +465,7 @@ testMxSend mNode test = do
453465 return r
454466
455467tests :: TestTransport -> IO [Test]
456−tests tt@TestTransport{..} = do
468+tests TestTransport{..} = do
457469 node1 <- newLocalNode testTransport initRemoteTable
458470 node2 <- newLocalNode testTransport initRemoteTable
459471 return [
@@ -465,7 +477,7 @@ tests tt@TestTransport{..} = do
465477 , testCase "InterAgentBroadcast"
466478 (delayedAssertion
467479 "expected (), but no broadcast was received"
468− node1 () testAgentBroadcast)
480+ node1 (Just ()) testAgentBroadcast)
469481 , testCase "AgentMailboxHandling"
470482 (delayedAssertion
471483 "expected (Just ()), but no regular (mailbox) input was handled"
@@ -490,11 +502,11 @@ tests tt@TestTransport{..} = do
490502 (delayedAssertion
491503 "expected process deaths to result in unregistration events"
492504 node1 () (testMxRegMon node2))
493− , testGroup "SendEvents" $ buildTestCases tt node2
505+ , testGroup "SendEvents" $ buildTestCases node1 node2
494506 ]
495507 ]
496508 where
497− buildTestCases tt' n2 = let nid = localNodeId n2 in build tt' 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)
@@ -515,15 +527,14 @@ tests tt@TestTransport{..} = do
515527 ])
516528 ]
517529
518− build :: TestTransport
530+ build :: LocalNode
519531 -> LocalNode
520532 -> [(String, [(String, (Maybe LocalNode -> Process ()))])]
521533 -> [Test]
522− build TestTransport{..} ln specs =
534+ build n ln specs =
523535 [ testGroup (intercalate "-" [groupName, caseSuffix]) [
524536 testCase (intercalate "-" [caseName, caseSuffix])
525− (newLocalNode testTransport initRemoteTable >>= \n ->
526− tryRunProcess n (caseImpl caseNode))
537+ (runProcess n (caseImpl caseNode))
527538 | (caseName, caseImpl) <- groupCases
528539 ]
529540 | (groupName, groupCases) <- specs