@@ -1,17 +1,22 @@
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)
8−import Control.Distributed.Process.Internal.Types (ProcessExitException(..))
9+import Control.Distributed.Process.Internal.Types
10+ ( ProcessExitException(..)
11+ , unsafeCreateUnencodedMessage
12+ )
913import Control.Distributed.Process.Node
1014import qualified Control.Distributed.Process.UnsafePrimitives as Unsafe
1115 ( send
1216 , nsend
1317 , nsendRemote
1418 , usend
19+ , sendChan
1520 )
1621import Control.Distributed.Process.Management
1722 ( MxEvent(..)
@@ -32,8 +37,8 @@ import Control.Monad (void, unless)
3237import Control.Monad.Catch(finally, bracket, try)
3338import Control.Rematch (equalTo)
3439import Data.Binary
35−import Data.List (find, sort)
36−import Data.Maybe (isJust, fromJust, isNothing)
40+import Data.List (find, sort, intercalate)
41+import Data.Maybe (isJust, fromJust, isNothing, fromMaybe, catMaybes)
3742import Data.Typeable
3843import GHC.Generics hiding (from)
3944#if ! MIN_VERSION_base(4,6,0)
@@ -61,7 +66,7 @@ awaitExit pid =
6166 where
6267 withMonitorRef p = bracket (monitor p) unmonitor
6368
64−testAgentBroadcast :: TestResult () -> Process ()
69+testAgentBroadcast :: TestResult (Maybe ()) -> Process ()
6570testAgentBroadcast result = do
6671 (resultSP, resultRP) <- newChan :: Process (SendPort (), ReceivePort ())
6772
@@ -76,10 +81,10 @@ testAgentBroadcast result = do
7681 mxNotify ()
7782 -- Once the publisher has seen our message, it will broadcast the Publish
7883 -- and the consumer will see that and send the result to our typed channel.
79− stash result =<< receiveChan resultRP
84+ stash result =<< receiveChanTimeout 10000000 resultRP
8085
81− kill publisher "finished" >> awaitExit publisher
82− kill consumer "finished" >> awaitExit consumer
86+ kill publisher "finished"
87+ kill consumer "finished"
8388
8489testAgentDualInput :: TestResult (Maybe Int) -> Process ()
8590testAgentDualInput result = do
@@ -94,6 +99,8 @@ testAgentDualInput result = do
9499 else mxReady)
95100 ]
96101
102+ mRef <- monitor s
103+
97104 mxNotify (1 :: Int)
98105 nsend "sum-agent" (3 :: Int)
99106 mxNotify (2 :: Int)
@@ -101,7 +108,10 @@ testAgentDualInput result = do
101108 mxNotify (5 :: Int)
102109
103110 stash result =<< receiveChanTimeout 10000000 rp
104− awaitExit s
111+ died <- receiveTimeout 10000000 [
112+ matchIf (\(ProcessMonitorNotification r _ _) -> r == mRef) (const $ return True)
113+ ]
114+ died `shouldBe` equalTo (Just True)
105115
106116testAgentPrioritisation :: TestResult [String] -> Process ()
107117testAgentPrioritisation result = do
@@ -115,10 +125,10 @@ testAgentPrioritisation result = do
115125 (sp, rp) <- newChan
116126 s <- mxAgent (MxAgentId name) ["first"] [
117127 mxSink (\(s :: String) -> do
118− mxUpdateLocal ((s:))
128+ mxUpdateLocal (s:)
119129 st <- mxGetLocal
120130 case length st of
121− n | n == 5 -> do liftMX $ sendChan sp st
131+ n | n == 5 -> do liftMX $ sendChan sp (sort st)
122132 mxDeactivate "finished"
123133 _ -> mxReceive -- go to the mailbox
124134 )
@@ -129,7 +139,7 @@ testAgentPrioritisation result = do
129139 mxNotify "fourth"
130140 nsend name "fifth"
131141
132− stash result . sort =<< receiveChan rp
142+ stash result . fromMaybe [] =<< receiveChanTimeout 10000000 rp
133143 awaitExit s
134144
135145testAgentMailboxHandling :: TestResult (Maybe ()) -> Process ()
@@ -142,12 +152,19 @@ testAgentMailboxHandling result = do
142152 nsend "mailbox-agent" ()
143153
144154 stash result =<< receiveChanTimeout 1000000 rp
145− kill agent "finished" >> awaitExit agent
155+ kill agent "finished"
146156
147157testAgentEventHandling :: TestResult Bool -> Process ()
148158testAgentEventHandling result = do
149− 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"
150166
167+ let initState = [] :: [MxEvent]
151168 (rc, rs) <- newChan
152169 agentPid <- mxAgent (MxAgentId "lifecycle-listener-agent") initState [
153170 (mxSink $ \ev -> do
@@ -180,8 +197,7 @@ testAgentEventHandling result = do
180197 -- TODO: yes, this is racy, but we're at the mercy of the scheduler here...
181198 faff 2000000 rs
182199
183− _ <- monitor agentPid
184− (sp, rp) <- newChan
200+ (sp, rp) <- newChan :: Process (SendPort (), ReceivePort ())
185201
186202 pid <- spawnLocal $ expect >>= sendChan sp
187203
@@ -190,9 +206,11 @@ testAgentEventHandling result = do
190206 monitor pid
191207
192208 send pid ()
193− () <- receiveChan rp
209+ rct <- receiveChanTimeout 10000000 rp
210+ unless (isJust rct) $ die "No response on channel"
194211
195− receiveWait [ match (\(ProcessMonitorNotification _ _ _) -> return ()) ]
212+ pmn <- receiveTimeout 2000000 [ match (\ProcessMonitorNotification{} -> return ()) ]
213+ unless (isJust pmn) $ die "No monitor notification arrived"
196214
197215 -- TODO: yes, this is racy, but we're at the mercy of the scheduler here...
198216 faff 2000000 rs
@@ -201,17 +219,16 @@ testAgentEventHandling result = do
201219 mxNotify (MxSpawned pid, replyTo)
202220 mxNotify (MxProcessDied pid DiedNormal, replyTo)
203221
204− seenAlive <- receiveChan reply
205− seenDead <- receiveChan reply
222+ seenAlive <- receiveChanTimeout 10000000 reply
223+ seenDead <- receiveChanTimeout 10000000 reply
206224
207− stash result $ seenAlive && seenDead
208− kill agentPid "test-complete" >> awaitExit agentPid
225+ stash result $ foldr (&&) True $ catMaybes [seenAlive, seenDead]
226+ kill timer "test-complete"
227+ kill agentPid "test-complete"
209228 where
210229 faff delay port = do
211230 res <- receiveChanTimeout delay port
212− unless (isNothing res) $ do
213− pause delay
214− faff delay port
231+ unless (isNothing res) $ pause delay
215232
216233testMxRegEvents :: TestResult () -> Process ()
217234testMxRegEvents result = do
@@ -258,7 +275,6 @@ testMxRegEvents result = do
258275 reg3 `shouldBe` equalTo (Just (label, p1))
259276
260277 mapM_ (flip kill $ "test-complete") [agent, p1, p2]
261− awaitExit agent
262278
263279testMxRegMon :: LocalNode -> TestResult () -> Process ()
264280testMxRegMon remoteNode result = do
@@ -316,64 +332,99 @@ testMxRegMon remoteNode result = do
316332 evts `shouldContain` (Just (label1, p1))
317333 evts `shouldContain` (Just (label2, p1))
318334
319− kill agent "test-complete" >> awaitExit agent
335+ kill agent "test-complete"
320336
321337ensure :: Process () -> Process () -> Process ()
322338ensure = flip finally
323339
324−type SendTest = ProcessId -> ReceivePort MxEvent -> Process Bool
325−
326340testNSend :: (String -> () -> Process ())
327341 -> Maybe LocalNode
328− -> TestResult Bool
329342 -> Process ()
330−testNSend op n r = testMxSend n r $ \p1 sink -> do
343+testNSend op n = do
344+ us <- getSelfPid
331345 let delay = 5000000
332− let label = "testMxSend"
346+ let label = "testMxSend" ++ (show us)
333347 let isValid = isValidLabel n label
334348
335− register label p1
336− reg1 <- receiveChanTimeout delay sink
337− case reg1 of
338− Just (MxRegistered pd lb)
339− | pd == p1 && isValid lb -> return ()
340− _ -> 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
341356
342− op label ()
357+ op label ()
343358
344− us <- getSelfPid
345− sent <- receiveChanTimeout delay sink
346− case sent of
347− Just (MxSentToName lb by _)
348− | by == us && isValid lb -> return True
349− _ -> 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
350364
351365 where
352− isValidLabel n l1 l2
366+ isValidLabel nd l1 l2
353367 | l2 == l2 = True
354− | isJust n = l2 == l1 ++ "@" ++ show (localNodeId $ fromJust n)
368+ | isJust nd = l2 == l1 ++ "@" ++ show (localNodeId $ fromJust nd)
355369 | otherwise = False
356370
357−testSend :: (ProcessId -> () -> Process ()) -> Maybe LocalNode -> TestResult Bool -> Process ()
358−testSend op n r = testMxSend n r $ \p1 sink -> do
371+testSend :: (ProcessId -> () -> Process ())
372+ -> Maybe LocalNode
373+ -> 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+
359405 -- initiate a send
360− op p1 ()
406+ op ch' ()
361407
362408 -- verify the management event
363− us <- getSelfPid
364409 sent <- receiveChanTimeout 5000000 sink
365410 case sent of
366− Just (MxSent pidTo pidFrom _)
367− | pidTo == p1 && pidFrom == us -> return True
368− _ -> return False
411+ Just (MxSentToPort sId spId _)
412+ | sId == us && spId == sendPortId ch' -> return True
413+ _ -> return False
414+
415+type SendTest = ProcessId -> ReceivePort MxEvent -> Process Bool
369416
370−testMxSend :: Maybe LocalNode -> TestResult Bool -> SendTest -> Process ()
371−testMxSend mNode result test = do
417+testMxSend :: Maybe LocalNode -> String -> SendTest -> Process ()
418+testMxSend mNode label test = do
372419 us <- getSelfPid
420+ (sp, rp) <- newChan
373421 (chan, sink) <- newChan
374422 agent <- mxAgent (MxAgentId $ agentLabel us) () [
375423 mxSink $ \ev -> do
376424 case ev of
425+ m@(MxSentToPort _ cid _)
426+ | cid /= sendPortId chan
427+ && cid /= sendPortId sp -> liftMX $ sendChan chan m
377428 m@(MxSent _ fromPid _)
378429 | fromPid == us -> liftMX $ sendChan chan m
379430 m@(MxSentToName _ fromPid _)
@@ -384,29 +435,39 @@ testMxSend mNode result test = do
384435 mxReady
385436 ]
386437
387− (sp, rp) <- newChan
388438 case mNode of
389439 Nothing -> void $ spawnLocal (proc sp)
390440 Just remoteNode -> void $ liftIO $ forkProcess remoteNode $ proc sp
391441
392− p1 <- receiveChan rp
393− res <- try (test p1 sink)
394− case res of
395− Left (ProcessExitException _ m) -> (liftIO $ putStrLn $ "SomeException-" ++ show m) >> stash result False
396− 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+
397446 kill agent "bye"
398− 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+
399453
400454 where
401− label = "testMxSend"
402455 agentLabel s = "mx-unsafe-check-agent-" ++ show s
403− 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
404466
405467tests :: TestTransport -> IO [Test]
406468tests TestTransport{..} = do
407469 node1 <- newLocalNode testTransport initRemoteTable
408470 node2 <- newLocalNode testTransport initRemoteTable
409− let nid = localNodeId node2
410471 return [
411472 testGroup "MxAgents" [
412473 testCase "EventHandling"
@@ -416,7 +477,7 @@ tests TestTransport{..} = do
416477 , testCase "InterAgentBroadcast"
417478 (delayedAssertion
418479 "expected (), but no broadcast was received"
419− node1 () testAgentBroadcast)
480+ node1 (Just ()) testAgentBroadcast)
420481 , testCase "AgentMailboxHandling"
421482 (delayedAssertion
422483 "expected (Just ()), but no regular (mailbox) input was handled"
@@ -441,53 +502,41 @@ tests TestTransport{..} = do
441502 (delayedAssertion
442503 "expected process deaths to result in unregistration events"
443504 node1 () (testMxRegMon node2))
444− , testGroup "SentEvents" [
445− testGroup "RemoteTargets" [
446− testCase "Unsafe.nsend"
447− (delayedAssertion "expected mx events failed"
448− node1 True (testNSend Unsafe.nsend $ Just node2))
449− , testCase "Unsafe.nsendRemote"
450− (delayedAssertion "expected mx events failed"
451− node1 True (testNSend (Unsafe.nsendRemote nid) $ Just node2))
452− , testCase "Unsafe.send"
453− (delayedAssertion "expected mx events failed"
454− node1 True (testSend Unsafe.send $ Just node2))
455− , testCase "Unsafe.usend"
456− (delayedAssertion "expected mx events failed"
457− node1 True (testSend Unsafe.usend $ Just node2))
458− , testCase "nsend"
459− (delayedAssertion "expected mx events failed"
460− node1 True (testNSend nsend $ Just node2))
461− , testCase "nsendRemote"
462− (delayedAssertion "expected mx events failed"
463− node1 True (testNSend (nsendRemote nid) $ Just node2))
464− , testCase "send"
465− (delayedAssertion "expected mx events failed"
466− node1 True (testSend send $ Just node2))
467− , testCase "usend"
468− (delayedAssertion "expected mx events failed"
469− node1 True (testSend usend $ Just node2))
470− ]
471− , testGroup "LocalTargets" [
472− testCase "Unsafe.nsend"
473− (delayedAssertion "expected mx events failed"
474− node1 True (testNSend Unsafe.nsend Nothing))
475− , testCase "Unsafe.send"
476− (delayedAssertion "expected mx events failed"
477− node1 True (testSend Unsafe.send Nothing))
478− , testCase "Unsafe.usend"
479− (delayedAssertion "expected mx events failed"
480− node1 True (testSend Unsafe.usend Nothing))
481− , testCase "nsend"
482− (delayedAssertion "expected mx events failed"
483− node1 True (testNSend nsend Nothing))
484− , testCase "send"
485− (delayedAssertion "expected mx events failed"
486− node1 True (testSend send Nothing))
487− , testCase "usend"
488− (delayedAssertion "expected mx events failed"
489− node1 True (testSend usend 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+ ])
490528 ]
491− ]
492− ]
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)]
493542 ]