@@ -5,6 +5,7 @@ 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)
89import Control.Distributed.Process hiding (bracket, finally, try)
910import Control.Distributed.Process.Internal.Types
1011 ( ProcessExitException(..)
@@ -16,7 +17,6 @@ import qualified Control.Distributed.Process.UnsafePrimitives as Unsafe
1617 , nsend
1718 , nsendRemote
1819 , usend
19− , sendChan
2020 )
2121import Control.Distributed.Process.Management
2222 ( MxEvent(..)
@@ -38,7 +38,7 @@ import Control.Monad.Catch(finally, bracket, try)
3838import Control.Rematch (equalTo)
3939import Data.Binary
4040import Data.List (find, sort, intercalate)
41−import Data.Maybe (isJust, fromJust, isNothing, fromMaybe, catMaybes)
41+import Data.Maybe (isJust, fromJust, isNothing)
4242import Data.Typeable
4343import GHC.Generics hiding (from)
4444#if ! MIN_VERSION_base(4,6,0)
@@ -66,7 +66,7 @@ awaitExit pid =
6666 where
6767 withMonitorRef p = bracket (monitor p) unmonitor
6868
69−testAgentBroadcast :: TestResult (Maybe ()) -> Process ()
69+testAgentBroadcast :: TestResult () -> Process ()
7070testAgentBroadcast result = do
7171 (resultSP, resultRP) <- newChan :: Process (SendPort (), ReceivePort ())
7272
@@ -81,10 +81,10 @@ testAgentBroadcast result = do
8181 mxNotify ()
8282 -- Once the publisher has seen our message, it will broadcast the Publish
8383 -- and the consumer will see that and send the result to our typed channel.
84− stash result =<< receiveChanTimeout 10000000 resultRP
84+ stash result =<< receiveChan resultRP
8585
86− kill publisher "finished"
87− kill consumer "finished"
86+ kill publisher "finished" >> awaitExit publisher
87+ kill consumer "finished" >> awaitExit consumer
8888
8989testAgentDualInput :: TestResult (Maybe Int) -> Process ()
9090testAgentDualInput result = do
@@ -99,8 +99,6 @@ testAgentDualInput result = do
9999 else mxReady)
100100 ]
101101
102− mRef <- monitor s
103−
104102 mxNotify (1 :: Int)
105103 nsend "sum-agent" (3 :: Int)
106104 mxNotify (2 :: Int)
@@ -108,10 +106,7 @@ testAgentDualInput result = do
108106 mxNotify (5 :: Int)
109107
110108 stash result =<< receiveChanTimeout 10000000 rp
111− died <- receiveTimeout 10000000 [
112− matchIf (\(ProcessMonitorNotification r _ _) -> r == mRef) (const $ return True)
113− ]
114− died `shouldBe` equalTo (Just True)
109+ awaitExit s
115110
116111testAgentPrioritisation :: TestResult [String] -> Process ()
117112testAgentPrioritisation result = do
@@ -125,10 +120,10 @@ testAgentPrioritisation result = do
125120 (sp, rp) <- newChan
126121 s <- mxAgent (MxAgentId name) ["first"] [
127122 mxSink (\(s :: String) -> do
128− mxUpdateLocal (s:)
123+ mxUpdateLocal ((s:))
129124 st <- mxGetLocal
130125 case length st of
131− n | n == 5 -> do liftMX $ sendChan sp (sort st)
126+ n | n == 5 -> do liftMX $ sendChan sp st
132127 mxDeactivate "finished"
133128 _ -> mxReceive -- go to the mailbox
134129 )
@@ -139,7 +134,7 @@ testAgentPrioritisation result = do
139134 mxNotify "fourth"
140135 nsend name "fifth"
141136
142− stash result . fromMaybe [] =<< receiveChanTimeout 10000000 rp
137+ stash result . sort =<< receiveChan rp
143138 awaitExit s
144139
145140testAgentMailboxHandling :: TestResult (Maybe ()) -> Process ()
@@ -152,19 +147,12 @@ testAgentMailboxHandling result = do
152147 nsend "mailbox-agent" ()
153148
154149 stash result =<< receiveChanTimeout 1000000 rp
155− kill agent "finished"
150+ kill agent "finished" >> awaitExit agent
156151
157152testAgentEventHandling :: TestResult Bool -> Process ()
158153testAgentEventHandling result = do
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"
166−
167154 let initState = [] :: [MxEvent]
155+
168156 (rc, rs) <- newChan
169157 agentPid <- mxAgent (MxAgentId "lifecycle-listener-agent") initState [
170158 (mxSink $ \ev -> do
@@ -197,7 +185,8 @@ testAgentEventHandling result = do
197185 -- TODO: yes, this is racy, but we're at the mercy of the scheduler here...
198186 faff 2000000 rs
199187
200− (sp, rp) <- newChan :: Process (SendPort (), ReceivePort ())
188+ _ <- monitor agentPid
189+ (sp, rp) <- newChan
201190
202191 pid <- spawnLocal $ expect >>= sendChan sp
203192
@@ -206,11 +195,9 @@ testAgentEventHandling result = do
206195 monitor pid
207196
208197 send pid ()
209− rct <- receiveChanTimeout 10000000 rp
210− unless (isJust rct) $ die "No response on channel"
198+ () <- receiveChan rp
211199
212− pmn <- receiveTimeout 2000000 [ match (\ProcessMonitorNotification{} -> return ()) ]
213− unless (isJust pmn) $ die "No monitor notification arrived"
200+ receiveWait [ match (\(ProcessMonitorNotification _ _ _) -> return ()) ]
214201
215202 -- TODO: yes, this is racy, but we're at the mercy of the scheduler here...
216203 faff 2000000 rs
@@ -219,16 +206,17 @@ testAgentEventHandling result = do
219206 mxNotify (MxSpawned pid, replyTo)
220207 mxNotify (MxProcessDied pid DiedNormal, replyTo)
221208
222− seenAlive <- receiveChanTimeout 10000000 reply
223− seenDead <- receiveChanTimeout 10000000 reply
209+ seenAlive <- receiveChan reply
210+ seenDead <- receiveChan reply
224211
225− stash result $ foldr (&&) True $ catMaybes [seenAlive, seenDead]
226− kill timer "test-complete"
227− kill agentPid "test-complete"
212+ stash result $ seenAlive && seenDead
213+ kill agentPid "test-complete" >> awaitExit agentPid
228214 where
229215 faff delay port = do
230216 res <- receiveChanTimeout delay port
231− unless (isNothing res) $ pause delay
217+ unless (isNothing res) $ do
218+ pause delay
219+ faff delay port
232220
233221testMxRegEvents :: TestResult () -> Process ()
234222testMxRegEvents result = do
@@ -275,6 +263,7 @@ testMxRegEvents result = do
275263 reg3 `shouldBe` equalTo (Just (label, p1))
276264
277265 mapM_ (flip kill $ "test-complete") [agent, p1, p2]
266+ awaitExit agent
278267
279268testMxRegMon :: LocalNode -> TestResult () -> Process ()
280269testMxRegMon remoteNode result = do
@@ -332,99 +321,67 @@ testMxRegMon remoteNode result = do
332321 evts `shouldContain` (Just (label1, p1))
333322 evts `shouldContain` (Just (label2, p1))
334323
335− kill agent "test-complete"
324+ kill agent "test-complete" >> awaitExit agent
336325
337326ensure :: Process () -> Process () -> Process ()
338327ensure = flip finally
339328
340329testNSend :: (String -> () -> Process ())
341330 -> Maybe LocalNode
331+ -> TestResult Bool
342332 -> Process ()
343−testNSend op n = do
344− us <- getSelfPid
333+testNSend op n r = testMxSend n r $ \p1 sink -> do
345334 let delay = 5000000
346− let label = "testMxSend" ++ (show us)
335+ let label = "testMxSend"
347336 let isValid = isValidLabel n label
348337
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
338+ register label p1
339+ reg1 <- receiveChanTimeout delay sink
340+ case reg1 of
341+ Just (MxRegistered pd lb)
342+ | pd == p1 && isValid lb -> return ()
343+ _ -> die $ "Reg-Failed: " ++ show reg1
356344
357− op label ()
345+ op label ()
358346
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
347+ us <- getSelfPid
348+ sent <- receiveChanTimeout delay sink
349+ case sent of
350+ Just (MxSentToName lb by _)
351+ | by == us && isValid lb -> return True
352+ _ -> die $ "Send-Failed: " ++ show sent
364353
365354 where
366− isValidLabel nd l1 l2
355+ isValidLabel n l1 l2
367356 | l2 == l2 = True
368− | isJust nd = l2 == l1 ++ "@" ++ show (localNodeId $ fromJust nd)
357+ | isJust n = l2 == l1 ++ "@" ++ show (localNodeId $ fromJust n)
369358 | otherwise = False
370359
371360testSend :: (ProcessId -> () -> Process ())
372361 -> Maybe LocalNode
362+ -> TestResult Bool
373363 -> Process ()
374−testSend op n = do
375− us <- getSelfPid
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
387−
388−testChan :: (SendPort () -> () -> Process ())
389− -> Maybe LocalNode
390− -> Process ()
391−testChan op n = testMxSend n "" $ \p1 sink -> do
392−
393− us <- getSelfPid
394− send p1 us
395−
396− cleared <- receiveChanTimeout 2000000 sink
397− case cleared of
398− Just (MxSent pidTo pidFrom _)
399− | pidTo == p1 && pidFrom == us -> return ()
400− _ -> die "Received uncleared Mx Event"
401−
402− chan <- receiveTimeout 5000000 [ match return ]
403− let ch' = fromMaybe (error "No reply chan received") chan
404−
364+testSend op n r = testMxSend n r $ \p1 sink -> do
405365 -- initiate a send
406− op ch' ()
366+ op p1 ()
407367
408368 -- verify the management event
369+ us <- getSelfPid
409370 sent <- receiveChanTimeout 5000000 sink
410371 case sent of
411− Just (MxSentToPort sId spId _)
412− | sId == us && spId == sendPortId ch' -> return True
413− _ -> return False
372+ Just (MxSent pidTo pidFrom _)
373+ | pidTo == p1 && pidFrom == us -> return True
374+ _ -> return False
414375
415376type SendTest = ProcessId -> ReceivePort MxEvent -> Process Bool
416377
417−testMxSend :: Maybe LocalNode -> String -> SendTest -> Process ()
418−testMxSend mNode label test = do
378+testMxSend :: Maybe LocalNode -> TestResult Bool -> SendTest -> Process ()
379+testMxSend mNode result test = do
419380 us <- getSelfPid
420− (sp, rp) <- newChan
421381 (chan, sink) <- newChan
422382 agent <- mxAgent (MxAgentId $ agentLabel us) () [
423383 mxSink $ \ev -> do
424384 case ev of
425− m@(MxSentToPort _ cid _)
426− | cid /= sendPortId chan
427− && cid /= sendPortId sp -> liftMX $ sendChan chan m
428385 m@(MxSent _ fromPid _)
429386 | fromPid == us -> liftMX $ sendChan chan m
430387 m@(MxSentToName _ fromPid _)
@@ -435,34 +392,23 @@ testMxSend mNode label test = do
435392 mxReady
436393 ]
437394
395+ (sp, rp) <- newChan
438396 case mNode of
439397 Nothing -> void $ spawnLocal (proc sp)
440398 Just remoteNode -> void $ liftIO $ forkProcess remoteNode $ proc sp
441399
442− p1 <- receiveChanTimeout 2000000 rp
443− unless (isJust p1) $ die "Timed out waiting for ProcessId"
444− res <- try $ test (fromJust p1) sink
445−
446− kill agent "bye"
447− kill (fromJust p1) "bye"
448−
400+ p1 <- receiveChan rp
401+ res <- try (test p1 sink)
449402 case res of
450− Left (ProcessExitException _ m) -> (liftIO $ putStrLn $ "SomeException-" ++ show m) >> die m
451− Right tr -> tr `shouldBe` equalTo True
452−
403+ Left (ProcessExitException _ m) -> (liftIO $ putStrLn $ "SomeException-" ++ show m) >> stash result False
404+ Right tr -> stash result tr
405+ kill agent "bye"
406+ kill p1 "bye"
453407
454408 where
409+ label = "testMxSend"
455410 agentLabel s = "mx-unsafe-check-agent-" ++ show s
456− proc sp' = getSelfPid >>= sendChan sp' >> go Nothing
457−
458− go :: Maybe (ReceivePort ()) -> Process ()
459− go Nothing = receiveTimeout 5000000 [ match replyChannel ] >>= go
460− go c@(Just c') = receiveWait [ matchChan c' return ] >> go c
461−
462− replyChannel p' = do
463− (s, r) <- newChan
464− send p' s
465− return r
411+ proc sp' = getSelfPid >>= sendChan sp' >> expect :: Process ()
466412
467413tests :: TestTransport -> IO [Test]
468414tests TestTransport{..} = do
@@ -477,7 +423,7 @@ tests TestTransport{..} = do
477423 , testCase "InterAgentBroadcast"
478424 (delayedAssertion
479425 "expected (), but no broadcast was received"
480− node1 (Just ()) testAgentBroadcast)
426+ node1 () testAgentBroadcast)
481427 , testCase "AgentMailboxHandling"
482428 (delayedAssertion
483429 "expected (Just ()), but no regular (mailbox) input was handled"
@@ -506,20 +452,19 @@ tests TestTransport{..} = do
506452 ]
507453 ]
508454 where
509− buildTestCases n1 n2 = let nid = localNodeId n2 in build n1 n2 [
455+ buildTestCases n1 n2 =
456+ let nid = localNodeId n2 in build n2 [
510457 ("NSend", [
511− ("nsend", testNSend nsend)
512− , ("Unsafe.nsend", testNSend Unsafe.nsend)
513− , ("nsendRemote", testNSend (nsendRemote nid))
458+ ("Unsafe.nsend", testNSend Unsafe.nsend)
514459 , ("Unsafe.nsendRemote", testNSend (Unsafe.nsendRemote nid))
460+ , ("nsend", testNSend nsend)
461+ , ("nsendRemote", testNSend (nsendRemote nid))
515462 ])
516463 , ("Send", [
517− ("send", testSend send)
518− , ("Unsafe.send", testSend Unsafe.send)
519− , ("usend", testSend usend)
520− , ("Unsafe.usend", testSend Unsafe.usend)
521− , ("sendChan", testChan sendChan)
522− , ("Unsafe.sendChan", testChan Unsafe.sendChan)
464+ ("Unsafe.send", testSend Unsafe.send)
465+ , ("Unsafe.usend", testSend Unsafe.usend)
466+ , ("send", testSend send)
467+ , ("usend", testSend usend)
523468 ])
524469 , ("Forward", [
525470 ("forward", testSend (\p m -> forward (unsafeCreateUnencodedMessage m) p))
@@ -528,13 +473,13 @@ tests TestTransport{..} = do
528473 ]
529474
530475 build :: LocalNode
531− -> LocalNode
532− -> [(String, [(String, (Maybe LocalNode -> Process ()))])]
476+ -> [(String, [(String, (Maybe LocalNode -> TestResult Bool -> Process ()))])]
533477 -> [Test]
534− build n ln specs =
478+ build ln specs =
535479 [ testGroup (intercalate "-" [groupName, caseSuffix]) [
536480 testCase (intercalate "-" [caseName, caseSuffix])
537− (runProcess n (caseImpl caseNode))
481+ (delayedAssertion "expected mx events failed"
482+ ln True (caseImpl caseNode))
538483 | (caseName, caseImpl) <- groupCases
539484 ]
540485 | (groupName, groupCases) <- specs