@@ -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(..)
@@ -17,6 +16,7 @@ import qualified Control.Distributed.Process.UnsafePrimitives as Unsafe
1716 , nsend
1817 , nsendRemote
1918 , 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)
41+import Data.Maybe (isJust, fromJust, isNothing, fromMaybe, catMaybes)
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 () -> Process ()
69+testAgentBroadcast :: TestResult (Maybe ()) -> 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 =<< receiveChan resultRP
84+ stash result =<< receiveChanTimeout 10000000 resultRP
8585
86− kill publisher "finished" >> awaitExit publisher
87− kill consumer "finished" >> awaitExit consumer
86+ kill publisher "finished"
87+ kill consumer "finished"
8888
8989testAgentDualInput :: TestResult (Maybe Int) -> Process ()
9090testAgentDualInput result = do
@@ -99,6 +99,8 @@ testAgentDualInput result = do
9999 else mxReady)
100100 ]
101101
102+ mRef <- monitor s
103+
102104 mxNotify (1 :: Int)
103105 nsend "sum-agent" (3 :: Int)
104106 mxNotify (2 :: Int)
@@ -106,7 +108,10 @@ testAgentDualInput result = do
106108 mxNotify (5 :: Int)
107109
108110 stash result =<< receiveChanTimeout 10000000 rp
109− awaitExit s
111+ died <- receiveTimeout 10000000 [
112+ matchIf (\(ProcessMonitorNotification r _ _) -> r == mRef) (const $ return True)
113+ ]
114+ died `shouldBe` equalTo (Just True)
110115
111116testAgentPrioritisation :: TestResult [String] -> Process ()
112117testAgentPrioritisation result = do
@@ -120,10 +125,10 @@ testAgentPrioritisation result = do
120125 (sp, rp) <- newChan
121126 s <- mxAgent (MxAgentId name) ["first"] [
122127 mxSink (\(s :: String) -> do
123− mxUpdateLocal ((s:))
128+ mxUpdateLocal (s:)
124129 st <- mxGetLocal
125130 case length st of
126− n | n == 5 -> do liftMX $ sendChan sp st
131+ n | n == 5 -> do liftMX $ sendChan sp (sort st)
127132 mxDeactivate "finished"
128133 _ -> mxReceive -- go to the mailbox
129134 )
@@ -134,7 +139,7 @@ testAgentPrioritisation result = do
134139 mxNotify "fourth"
135140 nsend name "fifth"
136141
137− stash result . sort =<< receiveChan rp
142+ stash result . fromMaybe [] =<< receiveChanTimeout 10000000 rp
138143 awaitExit s
139144
140145testAgentMailboxHandling :: TestResult (Maybe ()) -> Process ()
@@ -147,12 +152,19 @@ testAgentMailboxHandling result = do
147152 nsend "mailbox-agent" ()
148153
149154 stash result =<< receiveChanTimeout 1000000 rp
150− kill agent "finished" >> awaitExit agent
155+ kill agent "finished"
151156
152157testAgentEventHandling :: TestResult Bool -> Process ()
153158testAgentEventHandling result = do
154− 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"
155166
167+ let initState = [] :: [MxEvent]
156168 (rc, rs) <- newChan
157169 agentPid <- mxAgent (MxAgentId "lifecycle-listener-agent") initState [
158170 (mxSink $ \ev -> do
@@ -185,8 +197,7 @@ testAgentEventHandling result = do
185197 -- TODO: yes, this is racy, but we're at the mercy of the scheduler here...
186198 faff 2000000 rs
187199
188− _ <- monitor agentPid
189− (sp, rp) <- newChan
200+ (sp, rp) <- newChan :: Process (SendPort (), ReceivePort ())
190201
191202 pid <- spawnLocal $ expect >>= sendChan sp
192203
@@ -195,9 +206,11 @@ testAgentEventHandling result = do
195206 monitor pid
196207
197208 send pid ()
198− () <- receiveChan rp
209+ rct <- receiveChanTimeout 10000000 rp
210+ unless (isJust rct) $ die "No response on channel"
199211
200− receiveWait [ match (\(ProcessMonitorNotification _ _ _) -> return ()) ]
212+ pmn <- receiveTimeout 2000000 [ match (\ProcessMonitorNotification{} -> return ()) ]
213+ unless (isJust pmn) $ die "No monitor notification arrived"
201214
202215 -- TODO: yes, this is racy, but we're at the mercy of the scheduler here...
203216 faff 2000000 rs
@@ -206,17 +219,16 @@ testAgentEventHandling result = do
206219 mxNotify (MxSpawned pid, replyTo)
207220 mxNotify (MxProcessDied pid DiedNormal, replyTo)
208221
209− seenAlive <- receiveChan reply
210− seenDead <- receiveChan reply
222+ seenAlive <- receiveChanTimeout 10000000 reply
223+ seenDead <- receiveChanTimeout 10000000 reply
211224
212− stash result $ seenAlive && seenDead
213− kill agentPid "test-complete" >> awaitExit agentPid
225+ stash result $ foldr (&&) True $ catMaybes [seenAlive, seenDead]
226+ kill timer "test-complete"
227+ kill agentPid "test-complete"
214228 where
215229 faff delay port = do
216230 res <- receiveChanTimeout delay port
217− unless (isNothing res) $ do
218− pause delay
219− faff delay port
231+ unless (isNothing res) $ pause delay
220232
221233testMxRegEvents :: TestResult () -> Process ()
222234testMxRegEvents result = do
@@ -263,7 +275,6 @@ testMxRegEvents result = do
263275 reg3 `shouldBe` equalTo (Just (label, p1))
264276
265277 mapM_ (flip kill $ "test-complete") [agent, p1, p2]
266− awaitExit agent
267278
268279testMxRegMon :: LocalNode -> TestResult () -> Process ()
269280testMxRegMon remoteNode result = do
@@ -321,67 +332,99 @@ testMxRegMon remoteNode result = do
321332 evts `shouldContain` (Just (label1, p1))
322333 evts `shouldContain` (Just (label2, p1))
323334
324− kill agent "test-complete" >> awaitExit agent
335+ kill agent "test-complete"
325336
326337ensure :: Process () -> Process () -> Process ()
327338ensure = flip finally
328339
329340testNSend :: (String -> () -> Process ())
330341 -> Maybe LocalNode
331− -> TestResult Bool
332342 -> Process ()
333−testNSend op n r = testMxSend n r $ \p1 sink -> do
343+testNSend op n = do
344+ us <- getSelfPid
334345 let delay = 5000000
335− let label = "testMxSend"
346+ let label = "testMxSend" ++ (show us)
336347 let isValid = isValidLabel n label
337348
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
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
344356
345− op label ()
357+ op label ()
346358
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
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
353364
354365 where
355− isValidLabel n l1 l2
366+ isValidLabel nd l1 l2
356367 | l2 == l2 = True
357− | isJust n = l2 == l1 ++ "@" ++ show (localNodeId $ fromJust n)
368+ | isJust nd = l2 == l1 ++ "@" ++ show (localNodeId $ fromJust nd)
358369 | otherwise = False
359370
360371testSend :: (ProcessId -> () -> Process ())
361372 -> Maybe LocalNode
362− -> TestResult Bool
363373 -> Process ()
364−testSend op n r = testMxSend n r $ \p1 sink -> do
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+
365405 -- initiate a send
366− op p1 ()
406+ op ch' ()
367407
368408 -- verify the management event
369− us <- getSelfPid
370409 sent <- receiveChanTimeout 5000000 sink
371410 case sent of
372− Just (MxSent pidTo pidFrom _)
373− | pidTo == p1 && pidFrom == us -> return True
374− _ -> return False
411+ Just (MxSentToPort sId spId _)
412+ | sId == us && spId == sendPortId ch' -> return True
413+ _ -> return False
375414
376415type SendTest = ProcessId -> ReceivePort MxEvent -> Process Bool
377416
378−testMxSend :: Maybe LocalNode -> TestResult Bool -> SendTest -> Process ()
379−testMxSend mNode result test = do
417+testMxSend :: Maybe LocalNode -> String -> SendTest -> Process ()
418+testMxSend mNode label test = do
380419 us <- getSelfPid
420+ (sp, rp) <- newChan
381421 (chan, sink) <- newChan
382422 agent <- mxAgent (MxAgentId $ agentLabel us) () [
383423 mxSink $ \ev -> do
384424 case ev of
425+ m@(MxSentToPort _ cid _)
426+ | cid /= sendPortId chan
427+ && cid /= sendPortId sp -> liftMX $ sendChan chan m
385428 m@(MxSent _ fromPid _)
386429 | fromPid == us -> liftMX $ sendChan chan m
387430 m@(MxSentToName _ fromPid _)
@@ -392,23 +435,34 @@ testMxSend mNode result test = do
392435 mxReady
393436 ]
394437
395− (sp, rp) <- newChan
396438 case mNode of
397439 Nothing -> void $ spawnLocal (proc sp)
398440 Just remoteNode -> void $ liftIO $ forkProcess remoteNode $ proc sp
399441
400− p1 <- receiveChan rp
401− res <- try (test p1 sink)
402− case res of
403− Left (ProcessExitException _ m) -> (liftIO $ putStrLn $ "SomeException-" ++ show m) >> stash result False
404− 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+
405446 kill agent "bye"
406− 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+
407453
408454 where
409− label = "testMxSend"
410455 agentLabel s = "mx-unsafe-check-agent-" ++ show s
411− proc sp' = getSelfPid >>= sendChan sp' >> expect :: Process ()
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
412466
413467tests :: TestTransport -> IO [Test]
414468tests TestTransport{..} = do
@@ -423,7 +477,7 @@ tests TestTransport{..} = do
423477 , testCase "InterAgentBroadcast"
424478 (delayedAssertion
425479 "expected (), but no broadcast was received"
426− node1 () testAgentBroadcast)
480+ node1 (Just ()) testAgentBroadcast)
427481 , testCase "AgentMailboxHandling"
428482 (delayedAssertion
429483 "expected (Just ()), but no regular (mailbox) input was handled"
@@ -452,19 +506,20 @@ tests TestTransport{..} = do
452506 ]
453507 ]
454508 where
455− buildTestCases n1 n2 =
456− let nid = localNodeId n2 in build n2 [
509+ buildTestCases n1 n2 = let nid = localNodeId n2 in build n1 n2 [
457510 ("NSend", [
458− ("Unsafe.nsend", testNSend Unsafe.nsend)
459− , ("Unsafe.nsendRemote", testNSend (Unsafe.nsendRemote nid))
460− , ("nsend", testNSend nsend)
511+ ("nsend", testNSend nsend)
512+ , ("Unsafe.nsend", testNSend Unsafe.nsend)
461513 , ("nsendRemote", testNSend (nsendRemote nid))
514+ , ("Unsafe.nsendRemote", testNSend (Unsafe.nsendRemote nid))
462515 ])
463516 , ("Send", [
464− ("Unsafe.send", testSend Unsafe.send)
465− , ("Unsafe.usend", testSend Unsafe.usend)
466− , ("send", testSend send)
467− , ("usend", testSend usend)
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)
468523 ])
469524 , ("Forward", [
470525 ("forward", testSend (\p m -> forward (unsafeCreateUnencodedMessage m) p))
@@ -473,13 +528,13 @@ tests TestTransport{..} = do
473528 ]
474529
475530 build :: LocalNode
476− -> [(String, [(String, (Maybe LocalNode -> TestResult Bool -> Process ()))])]
531+ -> LocalNode
532+ -> [(String, [(String, (Maybe LocalNode -> Process ()))])]
477533 -> [Test]
478− build ln specs =
534+ build n ln specs =
479535 [ testGroup (intercalate "-" [groupName, caseSuffix]) [
480536 testCase (intercalate "-" [caseName, caseSuffix])
481− (delayedAssertion "expected mx events failed"
482− ln True (caseImpl caseNode))
537+ (runProcess n (caseImpl caseNode))
483538 | (caseName, caseImpl) <- groupCases
484539 ]
485540 | (groupName, groupCases) <- specs