@@ -1,9 +1,10 @@
1−{-# LANGUAGE DeriveGeneric #-}
1+{-# LANGUAGE DeriveGeneric #-}
2+{-# LANGUAGE ParallelListComp #-}
3+
24module Control.Distributed.Process.Tests.Mx (tests) where
35
46import Control.Distributed.Process.Tests.Internal.Utils
57import Network.Transport.Test (TestTransport(..))
6−import Control.Exception (SomeException)
78import Control.Distributed.Process hiding (bracket, finally, try)
89import Control.Distributed.Process.Internal.Types
910 ( ProcessExitException(..)
@@ -15,6 +16,7 @@ import qualified Control.Distributed.Process.UnsafePrimitives as Unsafe
1516 , nsend
1617 , nsendRemote
1718 , usend
19+ , sendChan
1820 )
1921import Control.Distributed.Process.Management
2022 ( MxEvent(..)
@@ -35,8 +37,8 @@ import Control.Monad (void, unless)
3537import Control.Monad.Catch(finally, bracket, try)
3638import Control.Rematch (equalTo)
3739import Data.Binary
38−import Data.List (find, sort)
39−import Data.Maybe (isJust, fromJust, isNothing)
40+import Data.List (find, sort, intercalate)
41+import Data.Maybe (isJust, fromJust, isNothing, fromMaybe, catMaybes)
4042import Data.Typeable
4143import GHC.Generics hiding (from)
4244#if ! MIN_VERSION_base(4,6,0)
@@ -64,7 +66,7 @@ awaitExit pid =
6466 where
6567 withMonitorRef p = bracket (monitor p) unmonitor
6668
67−testAgentBroadcast :: TestResult () -> Process ()
69+testAgentBroadcast :: TestResult (Maybe ()) -> Process ()
6870testAgentBroadcast result = do
6971 (resultSP, resultRP) <- newChan :: Process (SendPort (), ReceivePort ())
7072
@@ -79,10 +81,10 @@ testAgentBroadcast result = do
7981 mxNotify ()
8082 -- Once the publisher has seen our message, it will broadcast the Publish
8183 -- and the consumer will see that and send the result to our typed channel.
82− stash result =<< receiveChan resultRP
84+ stash result =<< receiveChanTimeout 10000000 resultRP
8385
84− kill publisher "finished" >> awaitExit publisher
85− kill consumer "finished" >> awaitExit consumer
86+ kill publisher "finished"
87+ kill consumer "finished"
8688
8789testAgentDualInput :: TestResult (Maybe Int) -> Process ()
8890testAgentDualInput result = do
@@ -97,6 +99,8 @@ testAgentDualInput result = do
9799 else mxReady)
98100 ]
99101
102+ mRef <- monitor s
103+
100104 mxNotify (1 :: Int)
101105 nsend "sum-agent" (3 :: Int)
102106 mxNotify (2 :: Int)
@@ -104,7 +108,10 @@ testAgentDualInput result = do
104108 mxNotify (5 :: Int)
105109
106110 stash result =<< receiveChanTimeout 10000000 rp
107− awaitExit s
111+ died <- receiveTimeout 10000000 [
112+ matchIf (\(ProcessMonitorNotification r _ _) -> r == mRef) (const $ return True)
113+ ]
114+ died `shouldBe` equalTo (Just True)
108115
109116testAgentPrioritisation :: TestResult [String] -> Process ()
110117testAgentPrioritisation result = do
@@ -118,10 +125,10 @@ testAgentPrioritisation result = do
118125 (sp, rp) <- newChan
119126 s <- mxAgent (MxAgentId name) ["first"] [
120127 mxSink (\(s :: String) -> do
121− mxUpdateLocal ((s:))
128+ mxUpdateLocal (s:)
122129 st <- mxGetLocal
123130 case length st of
124− n | n == 5 -> do liftMX $ sendChan sp st
131+ n | n == 5 -> do liftMX $ sendChan sp (sort st)
125132 mxDeactivate "finished"
126133 _ -> mxReceive -- go to the mailbox
127134 )
@@ -132,7 +139,7 @@ testAgentPrioritisation result = do
132139 mxNotify "fourth"
133140 nsend name "fifth"
134141
135− stash result . sort =<< receiveChan rp
142+ stash result . fromMaybe [] =<< receiveChanTimeout 10000000 rp
136143 awaitExit s
137144
138145testAgentMailboxHandling :: TestResult (Maybe ()) -> Process ()
@@ -145,12 +152,19 @@ testAgentMailboxHandling result = do
145152 nsend "mailbox-agent" ()
146153
147154 stash result =<< receiveChanTimeout 1000000 rp
148− kill agent "finished" >> awaitExit agent
155+ kill agent "finished"
149156
150157testAgentEventHandling :: TestResult Bool -> Process ()
151158testAgentEventHandling result = do
152− 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"
153166
167+ let initState = [] :: [MxEvent]
154168 (rc, rs) <- newChan
155169 agentPid <- mxAgent (MxAgentId "lifecycle-listener-agent") initState [
156170 (mxSink $ \ev -> do
@@ -183,8 +197,7 @@ testAgentEventHandling result = do
183197 -- TODO: yes, this is racy, but we're at the mercy of the scheduler here...
184198 faff 2000000 rs
185199
186− _ <- monitor agentPid
187− (sp, rp) <- newChan
200+ (sp, rp) <- newChan :: Process (SendPort (), ReceivePort ())
188201
189202 pid <- spawnLocal $ expect >>= sendChan sp
190203
@@ -193,9 +206,11 @@ testAgentEventHandling result = do
193206 monitor pid
194207
195208 send pid ()
196− () <- receiveChan rp
209+ rct <- receiveChanTimeout 10000000 rp
210+ unless (isJust rct) $ die "No response on channel"
197211
198− receiveWait [ match (\(ProcessMonitorNotification _ _ _) -> return ()) ]
212+ pmn <- receiveTimeout 2000000 [ match (\ProcessMonitorNotification{} -> return ()) ]
213+ unless (isJust pmn) $ die "No monitor notification arrived"
199214
200215 -- TODO: yes, this is racy, but we're at the mercy of the scheduler here...
201216 faff 2000000 rs
@@ -204,17 +219,16 @@ testAgentEventHandling result = do
204219 mxNotify (MxSpawned pid, replyTo)
205220 mxNotify (MxProcessDied pid DiedNormal, replyTo)
206221
207− seenAlive <- receiveChan reply
208− seenDead <- receiveChan reply
222+ seenAlive <- receiveChanTimeout 10000000 reply
223+ seenDead <- receiveChanTimeout 10000000 reply
209224
210− stash result $ seenAlive && seenDead
211− kill agentPid "test-complete" >> awaitExit agentPid
225+ stash result $ foldr (&&) True $ catMaybes [seenAlive, seenDead]
226+ kill timer "test-complete"
227+ kill agentPid "test-complete"
212228 where
213229 faff delay port = do
214230 res <- receiveChanTimeout delay port
215− unless (isNothing res) $ do
216− pause delay
217− faff delay port
231+ unless (isNothing res) $ pause delay
218232
219233testMxRegEvents :: TestResult () -> Process ()
220234testMxRegEvents result = do
@@ -261,7 +275,6 @@ testMxRegEvents result = do
261275 reg3 `shouldBe` equalTo (Just (label, p1))
262276
263277 mapM_ (flip kill $ "test-complete") [agent, p1, p2]
264− awaitExit agent
265278
266279testMxRegMon :: LocalNode -> TestResult () -> Process ()
267280testMxRegMon remoteNode result = do
@@ -319,67 +332,99 @@ testMxRegMon remoteNode result = do
319332 evts `shouldContain` (Just (label1, p1))
320333 evts `shouldContain` (Just (label2, p1))
321334
322− kill agent "test-complete" >> awaitExit agent
335+ kill agent "test-complete"
323336
324337ensure :: Process () -> Process () -> Process ()
325338ensure = flip finally
326339
327340testNSend :: (String -> () -> Process ())
328341 -> Maybe LocalNode
329− -> TestResult Bool
330342 -> Process ()
331−testNSend op n r = testMxSend n r $ \p1 sink -> do
343+testNSend op n = do
344+ us <- getSelfPid
332345 let delay = 5000000
333− let label = "testMxSend"
346+ let label = "testMxSend" ++ (show us)
334347 let isValid = isValidLabel n label
335348
336− register label p1
337− reg1 <- receiveChanTimeout delay sink
338− case reg1 of
339− Just (MxRegistered pd lb)
340− | pd == p1 && isValid lb -> return ()
341− _ -> 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
342356
343− op label ()
357+ op label ()
344358
345− us <- getSelfPid
346− sent <- receiveChanTimeout delay sink
347− case sent of
348− Just (MxSentToName lb by _)
349− | by == us && isValid lb -> return True
350− _ -> 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
351364
352365 where
353− isValidLabel n l1 l2
366+ isValidLabel nd l1 l2
354367 | l2 == l2 = True
355− | isJust n = l2 == l1 ++ "@" ++ show (localNodeId $ fromJust n)
368+ | isJust nd = l2 == l1 ++ "@" ++ show (localNodeId $ fromJust nd)
356369 | otherwise = False
357370
358371testSend :: (ProcessId -> () -> Process ())
359372 -> Maybe LocalNode
360− -> TestResult Bool
361373 -> Process ()
362−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+
363405 -- initiate a send
364− op p1 ()
406+ op ch' ()
365407
366408 -- verify the management event
367− us <- getSelfPid
368409 sent <- receiveChanTimeout 5000000 sink
369410 case sent of
370− Just (MxSent pidTo pidFrom _)
371− | pidTo == p1 && pidFrom == us -> return True
372− _ -> return False
411+ Just (MxSentToPort sId spId _)
412+ | sId == us && spId == sendPortId ch' -> return True
413+ _ -> return False
373414
374415type SendTest = ProcessId -> ReceivePort MxEvent -> Process Bool
375416
376−testMxSend :: Maybe LocalNode -> TestResult Bool -> SendTest -> Process ()
377−testMxSend mNode result test = do
417+testMxSend :: Maybe LocalNode -> String -> SendTest -> Process ()
418+testMxSend mNode label test = do
378419 us <- getSelfPid
420+ (sp, rp) <- newChan
379421 (chan, sink) <- newChan
380422 agent <- mxAgent (MxAgentId $ agentLabel us) () [
381423 mxSink $ \ev -> do
382424 case ev of
425+ m@(MxSentToPort _ cid _)
426+ | cid /= sendPortId chan
427+ && cid /= sendPortId sp -> liftMX $ sendChan chan m
383428 m@(MxSent _ fromPid _)
384429 | fromPid == us -> liftMX $ sendChan chan m
385430 m@(MxSentToName _ fromPid _)
@@ -390,29 +435,39 @@ testMxSend mNode result test = do
390435 mxReady
391436 ]
392437
393− (sp, rp) <- newChan
394438 case mNode of
395439 Nothing -> void $ spawnLocal (proc sp)
396440 Just remoteNode -> void $ liftIO $ forkProcess remoteNode $ proc sp
397441
398− p1 <- receiveChan rp
399− res <- try (test p1 sink)
400− case res of
401− Left (ProcessExitException _ m) -> (liftIO $ putStrLn $ "SomeException-" ++ show m) >> stash result False
402− 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+
403446 kill agent "bye"
404− 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+
405453
406454 where
407− label = "testMxSend"
408455 agentLabel s = "mx-unsafe-check-agent-" ++ show s
409− 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
410466
411467tests :: TestTransport -> IO [Test]
412468tests TestTransport{..} = do
413469 node1 <- newLocalNode testTransport initRemoteTable
414470 node2 <- newLocalNode testTransport initRemoteTable
415− let nid = localNodeId node2
416471 return [
417472 testGroup "MxAgents" [
418473 testCase "EventHandling"
@@ -422,7 +477,7 @@ tests TestTransport{..} = do
422477 , testCase "InterAgentBroadcast"
423478 (delayedAssertion
424479 "expected (), but no broadcast was received"
425− node1 () testAgentBroadcast)
480+ node1 (Just ()) testAgentBroadcast)
426481 , testCase "AgentMailboxHandling"
427482 (delayedAssertion
428483 "expected (Just ()), but no regular (mailbox) input was handled"
@@ -447,65 +502,41 @@ tests TestTransport{..} = do
447502 (delayedAssertion
448503 "expected process deaths to result in unregistration events"
449504 node1 () (testMxRegMon node2))
450− , testGroup "SentEvents" [
451− testGroup "RemoteTargets" [
452− testCase "Unsafe.nsend"
453− (delayedAssertion "expected mx events failed"
454− node1 True (testNSend Unsafe.nsend $ Just node2))
455− , testCase "Unsafe.nsendRemote"
456− (delayedAssertion "expected mx events failed"
457− node1 True (testNSend (Unsafe.nsendRemote nid) $ Just node2))
458− , testCase "Unsafe.send"
459− (delayedAssertion "expected mx events failed"
460− node1 True (testSend Unsafe.send $ Just node2))
461− , testCase "Unsafe.usend"
462− (delayedAssertion "expected mx events failed"
463− node1 True (testSend Unsafe.usend $ Just node2))
464− , testCase "nsend"
465− (delayedAssertion "expected mx events failed"
466− node1 True (testNSend nsend $ Just node2))
467− , testCase "nsendRemote"
468− (delayedAssertion "expected mx events failed"
469− node1 True (testNSend (nsendRemote nid) $ Just node2))
470− , testCase "send"
471− (delayedAssertion "expected mx events failed"
472− node1 True (testSend send $ Just node2))
473− , testCase "usend"
474− (delayedAssertion "expected mx events failed"
475− node1 True (testSend usend $ Just node2))
476− , testCase "forward"
477− (delayedAssertion "expected mx events failed"
478− node1 True (testSend (\p m -> forward (unsafeCreateUnencodedMessage m) p) $ Just node2))
479− , testCase "uforward"
480− (delayedAssertion "expected mx events failed"
481− node1 True (testSend (\p m -> uforward (unsafeCreateUnencodedMessage m) p) $ Just node2))
482− ]
483− , testGroup "LocalTargets" [
484− testCase "Unsafe.nsend"
485− (delayedAssertion "expected mx events failed"
486− node1 True (testNSend Unsafe.nsend Nothing))
487− , testCase "Unsafe.send"
488− (delayedAssertion "expected mx events failed"
489− node1 True (testSend Unsafe.send Nothing))
490− , testCase "Unsafe.usend"
491− (delayedAssertion "expected mx events failed"
492− node1 True (testSend Unsafe.usend Nothing))
493− , testCase "nsend"
494− (delayedAssertion "expected mx events failed"
495− node1 True (testNSend nsend Nothing))
496− , testCase "send"
497− (delayedAssertion "expected mx events failed"
498− node1 True (testSend send Nothing))
499− , testCase "usend"
500− (delayedAssertion "expected mx events failed"
501− node1 True (testSend usend Nothing))
502− , testCase "forward"
503− (delayedAssertion "expected mx events failed"
504− node1 True (testSend (\p m -> forward (unsafeCreateUnencodedMessage m) p) Nothing))
505− , testCase "uforward"
506− (delayedAssertion "expected mx events failed"
507− node1 True (testSend (\p m -> uforward (unsafeCreateUnencodedMessage m) p) Nothing))
505+ , testGroup "SendEvents" $ buildTestCases node1 node2
506+ ]
507+ ]
508+ where
509+ buildTestCases n1 n2 = let nid = localNodeId n2 in build n1 n2 [
510+ ("NSend", [
511+ ("nsend", testNSend nsend)
512+ , ("Unsafe.nsend", testNSend Unsafe.nsend)
513+ , ("nsendRemote", testNSend (nsendRemote nid))
514+ , ("Unsafe.nsendRemote", testNSend (Unsafe.nsendRemote nid))
515+ ])
516+ , ("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)
523+ ])
524+ , ("Forward", [
525+ ("forward", testSend (\p m -> forward (unsafeCreateUnencodedMessage m) p))
526+ , ("uforward", testSend (\p m -> uforward (unsafeCreateUnencodedMessage m) p))
527+ ])
508528 ]
509− ]
510− ]
529+
530+ build :: LocalNode
531+ -> LocalNode
532+ -> [(String, [(String, (Maybe LocalNode -> Process ()))])]
533+ -> [Test]
534+ build n ln specs =
535+ [ testGroup (intercalate "-" [groupName, caseSuffix]) [
536+ testCase (intercalate "-" [caseName, caseSuffix])
537+ (runProcess n (caseImpl caseNode))
538+ | (caseName, caseImpl) <- groupCases
539+ ]
540+ | (groupName, groupCases) <- specs
541+ , (caseSuffix, caseNode) <- [("RemotePid", Just ln), ("LocalPid", Nothing)]
511542 ]