@@ -1,23 +1,12 @@
1−{-# LANGUAGE DeriveGeneric #-}
2−{-# LANGUAGE ParallelListComp #-}
3−
1+{-# LANGUAGE DeriveGeneric #-}
42module Control.Distributed.Process.Tests.Mx (tests) where
53
64import Control.Distributed.Process.Tests.Internal.Utils
75import Network.Transport.Test (TestTransport(..))
8−import Control.Distributed.Process hiding (bracket, finally, try)
9−import Control.Distributed.Process.Internal.Types
10− ( ProcessExitException(..)
11− , unsafeCreateUnencodedMessage
12− )
6+
7+import Control.Concurrent (threadDelay)
8+import Control.Distributed.Process
139import Control.Distributed.Process.Node
14−import qualified Control.Distributed.Process.UnsafePrimitives as Unsafe
15− ( send
16− , nsend
17− , nsendRemote
18− , usend
19− , sendChan
20− )
2110import Control.Distributed.Process.Management
2211 ( MxEvent(..)
2312 , MxAgentId(..)
@@ -32,15 +21,14 @@ import Control.Distributed.Process.Management
3221 , mxUpdateLocal
3322 , mxNotify
3423 , mxBroadcast
24+ , mxGetId
3525 )
36−import Control.Monad (void, unless)
37−import Control.Monad.Catch(finally, bracket, try)
38−import Control.Rematch (equalTo)
26+import Control.Monad (void)
3927import Data.Binary
40−import Data.List (find, sort, intercalate)
41−import Data.Maybe (isJust, fromJust, isNothing, fromMaybe, catMaybes)
28+import Data.List (find, sort)
29+import Data.Maybe (isJust)
4230import Data.Typeable
43−import GHC.Generics hiding (from)
31+import GHC.Generics
4432#if ! MIN_VERSION_base(4,6,0)
4533import Prelude hiding (catch, log)
4634#endif
@@ -56,17 +44,7 @@ data Publish = Publish
5644
5745instance Binary Publish where
5846
59−awaitExit :: ProcessId -> Process ()
60−awaitExit pid =
61− withMonitorRef pid $ \ref -> do
62− receiveWait
63− [ matchIf (\(ProcessMonitorNotification r _ _) -> r == ref)
64− (\_ -> return ())
65− ]
66− where
67− withMonitorRef p = bracket (monitor p) unmonitor
68−
69−testAgentBroadcast :: TestResult (Maybe ()) -> Process ()
47+testAgentBroadcast :: TestResult () -> Process ()
7048testAgentBroadcast result = do
7149 (resultSP, resultRP) <- newChan :: Process (SendPort (), ReceivePort ())
7250
@@ -81,7 +59,7 @@ testAgentBroadcast result = do
8159 mxNotify ()
8260 -- Once the publisher has seen our message, it will broadcast the Publish
8361 -- and the consumer will see that and send the result to our typed channel.
84− stash result =<< receiveChanTimeout 10000000 resultRP
62+ stash result =<< receiveChan resultRP
8563
8664 kill publisher "finished"
8765 kill consumer "finished"
@@ -89,7 +67,7 @@ testAgentBroadcast result = do
8967testAgentDualInput :: TestResult (Maybe Int) -> Process ()
9068testAgentDualInput result = do
9169 (sp, rp) <- newChan
92− s <- mxAgent (MxAgentId "sum-agent") (0 :: Int) [
70+ _ <- mxAgent (MxAgentId "sum-agent") (0 :: Int) [
9371 mxSink $ (\(i :: Int) -> do
9472 mxSetLocal . (+i) =<< mxGetLocal
9573 i' <- mxGetLocal
@@ -99,8 +77,6 @@ testAgentDualInput result = do
9977 else mxReady)
10078 ]
10179
102− mRef <- monitor s
103−
10480 mxNotify (1 :: Int)
10581 nsend "sum-agent" (3 :: Int)
10682 mxNotify (2 :: Int)
@@ -108,10 +84,6 @@ testAgentDualInput result = do
10884 mxNotify (5 :: Int)
10985
11086 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)
11587
11688testAgentPrioritisation :: TestResult [String] -> Process ()
11789testAgentPrioritisation result = do
@@ -123,12 +95,12 @@ testAgentPrioritisation result = do
12395
12496 let name = "prioritising-agent"
12597 (sp, rp) <- newChan
126− s <- mxAgent (MxAgentId name) ["first"] [
127− mxSink (\(s :: String) -> do
128− mxUpdateLocal (s:)
98+ void $ mxAgent (MxAgentId name) ["first"] [
99+ mxSink (\(s :: String) -> do
100+ mxUpdateLocal ((s:))
129101 st <- mxGetLocal
130102 case length st of
131− n | n == 5 -> do liftMX $ sendChan sp (sort st)
103+ n | n == 5 -> do liftMX $ sendChan sp st
132104 mxDeactivate "finished"
133105 _ -> mxReceive -- go to the mailbox
134106 )
@@ -139,33 +111,23 @@ testAgentPrioritisation result = do
139111 mxNotify "fourth"
140112 nsend name "fifth"
141113
142− stash result . fromMaybe [] =<< receiveChanTimeout 10000000 rp
143− awaitExit s
114+ stash result . sort =<< receiveChan rp
144115
145116testAgentMailboxHandling :: TestResult (Maybe ()) -> Process ()
146117testAgentMailboxHandling result = do
147118 (sp, rp) <- newChan
148− agent <- mxAgent (MxAgentId "mailbox-agent") () [
119+ agent <- mxAgent (MxAgentId "listener-agent") () [
149120 mxSink $ \() -> (liftMX $ sendChan sp ()) >> mxReady
150121 ]
151122
152− nsend "mailbox-agent" ()
123+ nsend "listener-agent" ()
153124
154125 stash result =<< receiveChanTimeout 1000000 rp
155126 kill agent "finished"
156127
157128testAgentEventHandling :: TestResult Bool -> Process ()
158129testAgentEventHandling 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−
167130 let initState = [] :: [MxEvent]
168− (rc, rs) <- newChan
169131 agentPid <- mxAgent (MxAgentId "lifecycle-listener-agent") initState [
170132 (mxSink $ \ev -> do
171133 st <- mxGetLocal
@@ -174,7 +136,7 @@ testAgentEventHandling result = do
174136 (MxSpawned _) -> mxSetLocal (ev:st)
175137 (MxProcessDied _ _) -> mxSetLocal (ev:st)
176138 _ -> return ()
177− act >> (liftMX $ sendChan rc ()) >> mxReady),
139+ act >> mxReady),
178140 (mxSink $ \(ev, sp :: SendPort Bool) -> do
179141 st <- mxGetLocal
180142 let found =
@@ -194,349 +156,50 @@ testAgentEventHandling result = do
194156 mxReady)
195157 ]
196158
197− -- TODO: yes, this is racy, but we're at the mercy of the scheduler here...
198− faff 2000000 rs
199−
200− (sp, rp) <- newChan :: Process (SendPort (), ReceivePort ())
201−
202− pid <- spawnLocal $ expect >>= sendChan sp
159+ _ <- monitor agentPid
160+ (sp, rp) <- newChan
161+ pid <- spawnLocal $ sendChan sp ()
162+ () <- receiveChan rp
203163
204164 -- By waiting for a monitor notification, we have a
205− -- higher probably that the agent has seen the spawn and died events
165+ -- higher probably that the agent has seen
206166 monitor pid
207−
208− send pid ()
209− rct <- receiveChanTimeout 10000000 rp
210− unless (isJust rct) $ die "No response on channel"
211−
212− pmn <- receiveTimeout 2000000 [ match (\ProcessMonitorNotification{} -> return ()) ]
213− unless (isJust pmn) $ die "No monitor notification arrived"
214−
215− -- TODO: yes, this is racy, but we're at the mercy of the scheduler here...
216− faff 2000000 rs
167+ receiveWait [ match (\(ProcessMonitorNotification _ _ _) -> return ()) ]
217168
218169 (replyTo, reply) <- newChan :: Process (SendPort Bool, ReceivePort Bool)
219170 mxNotify (MxSpawned pid, replyTo)
220171 mxNotify (MxProcessDied pid DiedNormal, replyTo)
221172
222− seenAlive <- receiveChanTimeout 10000000 reply
223− seenDead <- receiveChanTimeout 10000000 reply
224−
225− stash result $ foldr (&&) True $ catMaybes [seenAlive, seenDead]
226− kill timer "test-complete"
227− kill agentPid "test-complete"
228− where
229− faff delay port = do
230− res <- receiveChanTimeout delay port
231− unless (isNothing res) $ pause delay
232−
233−testMxRegEvents :: TestResult () -> Process ()
234−testMxRegEvents result = do
235− {- This test only deals with the local case, to ensure that we are being
236− notified in the expected order - the remote cases related to the
237− behaviour of the node controller are contained in the CH test suite. -}
238− ensure (stash result ()) $ do
239− let label = "testMxRegEvents"
240− let agentLabel = "mxRegEvents-agent"
241− let delay = 1000000
242− (regChan, regSink) <- newChan
243− (unRegChan, unRegSink) <- newChan
244− agent <- mxAgent (MxAgentId agentLabel) () [
245− mxSink $ \ev -> do
246− case ev of
247− MxRegistered pid label'
248− | label' == label -> liftMX $ sendChan regChan (label', pid)
249− MxUnRegistered pid label'
250− | label' == label -> liftMX $ sendChan unRegChan (label', pid)
251− _ -> return ()
252− mxReady
253− ]
254−
255− p1 <- spawnLocal expect
256− p2 <- spawnLocal expect
257−
258− register label p1
259− reg1 <- receiveChanTimeout delay regSink
260− reg1 `shouldBe` equalTo (Just (label, p1))
261−
262− unregister label
263− unreg1 <- receiveChanTimeout delay unRegSink
264− unreg1 `shouldBe` equalTo (Just (label, p1))
265−
266− register label p2
267− reg2 <- receiveChanTimeout delay regSink
268− reg2 `shouldBe` equalTo (Just (label, p2))
269−
270− reregister label p1
271− unreg2 <- receiveChanTimeout delay unRegSink
272− unreg2 `shouldBe` equalTo (Just (label, p2))
273−
274− reg3 <- receiveChanTimeout delay regSink
275− reg3 `shouldBe` equalTo (Just (label, p1))
276−
277− mapM_ (flip kill $ "test-complete") [agent, p1, p2]
278−
279−testMxRegMon :: LocalNode -> TestResult () -> Process ()
280−testMxRegMon remoteNode result = do
281− ensure (stash result ()) $ do
282− -- ensure that when a registered process dies, we get a notification that
283− -- it has been unregistered as well as seeing the name get removed
284− let label1 = "aaaaa"
285− let label2 = "bbbbb"
286− let isValid l = l == label1 || l == label2
287− let agentLabel = "mxRegMon-agent"
288− let delay = 1000000
289− (regChan, regSink) <- newChan
290− (unRegChan, unRegSink) <- newChan
291− agent <- mxAgent (MxAgentId agentLabel) () [
292− mxSink $ \ev -> do
293− case ev of
294− MxRegistered pid label
295− | isValid label -> liftMX $ sendChan regChan (label, pid)
296− MxUnRegistered pid label
297− | isValid label -> liftMX $ sendChan unRegChan (label, pid)
298− _ -> return ()
299− mxReady
300− ]
301−
302− (sp, rp) <- newChan
303− liftIO $ forkProcess remoteNode $ do
304− getSelfPid >>= sendChan sp
305− expect :: Process ()
306−
307− p1 <- receiveChan rp
308−
309− register label1 p1
310− reg1 <- receiveChanTimeout delay regSink
311− reg1 `shouldBe` equalTo (Just (label1, p1))
312−
313− register label2 p1
314− reg2 <- receiveChanTimeout delay regSink
315− reg2 `shouldBe` equalTo (Just (label2, p1))
316−
317− n1 <- whereis label1
318− n1 `shouldBe` equalTo (Just p1)
319−
320− n2 <- whereis label2
321− n2 `shouldBe` equalTo (Just p1)
322−
323− kill p1 "goodbye"
173+ seenAlive <- receiveChan reply
174+ seenDead <- receiveChan reply
324175
325− unreg1 <- receiveChanTimeout delay unRegSink
326− unreg2 <- receiveChanTimeout delay unRegSink
327−
328− let evts = [unreg1, unreg2]
329− -- we can't rely on the order of the values in the node controller's
330− -- map (it's either racy to do so, or no such guarantee exists for Data.Map),
331− -- so we simply verify that we received the un-registration events we expect
332− evts `shouldContain` (Just (label1, p1))
333− evts `shouldContain` (Just (label2, p1))
334−
335− kill agent "test-complete"
336−
337−ensure :: Process () -> Process () -> Process ()
338−ensure = flip finally
339−
340−testNSend :: (String -> () -> Process ())
341− -> Maybe LocalNode
342− -> Process ()
343−testNSend op n = do
344− us <- getSelfPid
345− let delay = 5000000
346− let label = "testMxSend" ++ (show us)
347− let isValid = isValidLabel n label
348−
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
356−
357− op label ()
358−
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
364−
365− where
366− isValidLabel nd l1 l2
367− | l2 == l2 = True
368− | isJust nd = l2 == l1 ++ "@" ++ show (localNodeId $ fromJust nd)
369− | otherwise = False
370−
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−
405− -- initiate a send
406− op ch' ()
407−
408− -- verify the management event
409− sent <- receiveChanTimeout 5000000 sink
410− case sent of
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
416−
417−testMxSend :: Maybe LocalNode -> String -> SendTest -> Process ()
418−testMxSend mNode label test = do
419− us <- getSelfPid
420− (sp, rp) <- newChan
421− (chan, sink) <- newChan
422− agent <- mxAgent (MxAgentId $ agentLabel us) () [
423− mxSink $ \ev -> do
424− case ev of
425− m@(MxSentToPort _ cid _)
426− | cid /= sendPortId chan
427− && cid /= sendPortId sp -> liftMX $ sendChan chan m
428− m@(MxSent _ fromPid _)
429− | fromPid == us -> liftMX $ sendChan chan m
430− m@(MxSentToName _ fromPid _)
431− | fromPid == us -> liftMX $ sendChan chan m
432− m@(MxRegistered _ name)
433− | name == label -> liftMX $ sendChan chan m
434− _ -> return ()
435− mxReady
436− ]
437−
438− case mNode of
439− Nothing -> void $ spawnLocal (proc sp)
440− Just remoteNode -> void $ liftIO $ forkProcess remoteNode $ proc sp
441−
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−
449− case res of
450− Left (ProcessExitException _ m) -> (liftIO $ putStrLn $ "SomeException-" ++ show m) >> die m
451− Right tr -> tr `shouldBe` equalTo True
452−
453−
454− where
455− 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
176+ stash result $ seenAlive && seenDead
466177
467178tests :: TestTransport -> IO [Test]
468179tests TestTransport{..} = do
469180 node1 <- newLocalNode testTransport initRemoteTable
470− node2 <- newLocalNode testTransport initRemoteTable
471181 return [
472− testGroup "MxAgents" [
473− testCase "EventHandling"
474− (delayedAssertion
475− "expected True, but events where not as expected"
476− node1 True testAgentEventHandling)
477− , testCase "InterAgentBroadcast"
478− (delayedAssertion
479− "expected (), but no broadcast was received"
480− node1 (Just ()) testAgentBroadcast)
481− , testCase "AgentMailboxHandling"
482− (delayedAssertion
483− "expected (Just ()), but no regular (mailbox) input was handled"
484− node1 (Just ()) testAgentMailboxHandling)
485− , testCase "AgentDualInputHandling"
486− (delayedAssertion
487− "expected sum = 15, but the result was Nothing"
488− node1 (Just 15 :: Maybe Int) testAgentDualInput)
489− , testCase "AgentInputPrioritisation"
490− (delayedAssertion
491− "expected [first, second, third, fourth, fifth], but result diverged"
492− node1 (sort ["first", "second",
493− "third", "fourth",
494− "fifth"]) testAgentPrioritisation)
495− ]
496− , testGroup "MxEvents" [
497− testCase "NameRegistrationEvents"
498− (delayedAssertion
499− "expected registration events to map to the correct ProcessId"
500− node1 () testMxRegEvents)
501− , testCase "PostDeathNameUnRegistrationEvents"
502− (delayedAssertion
503− "expected process deaths to result in unregistration events"
504− node1 () (testMxRegMon node2))
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− ])
528− ]
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)]
542− ]
182+ testGroup "Mx Agents" [
183+ testCase "Event Handling"
184+ (delayedAssertion
185+ "expected True, but events where not as expected"
186+ node1 True testAgentEventHandling)
187+ , testCase "Inter-Agent Broadcast"
188+ (delayedAssertion
189+ "expected (), but no broadcast was received"
190+ node1 () testAgentBroadcast)
191+ , testCase "Agent Mailbox Handling"
192+ (delayedAssertion
193+ "expected (Just ()), but no regular (mailbox) input was handled"
194+ node1 (Just ()) testAgentMailboxHandling)
195+ , testCase "Agent Dual Input Handling"
196+ (delayedAssertion
197+ "expected sum = 15, but the result was Nothing"
198+ node1 (Just 15 :: Maybe Int) testAgentDualInput)
199+ , testCase "Agent Input Prioritisation"
200+ (delayedAssertion
201+ "expected [first, second, third, fourth, fifth], but result diverged"
202+ node1 (sort ["first", "second",
203+ "third", "fourth",
204+ "fifth"]) testAgentPrioritisation)
205+ ]]