@@ -1,12 +1,23 @@
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−
7−import Control.Concurrent (threadDelay)
8−import Control.Distributed.Process
8+import Control.Distributed.Process hiding (bracket, finally, try)
9+import Control.Distributed.Process.Internal.Types
10+ ( ProcessExitException(..)
11+ , unsafeCreateUnencodedMessage
12+ )
913import Control.Distributed.Process.Node
14+import qualified Control.Distributed.Process.UnsafePrimitives as Unsafe
15+ ( send
16+ , nsend
17+ , nsendRemote
18+ , usend
19+ , sendChan
20+ )
1021import Control.Distributed.Process.Management
1122 ( MxEvent(..)
1223 , MxAgentId(..)
@@ -21,14 +32,15 @@ import Control.Distributed.Process.Management
2132 , mxUpdateLocal
2233 , mxNotify
2334 , mxBroadcast
24− , mxGetId
2535 )
26−import Control.Monad (void)
36+import Control.Monad (void, unless)
37+import Control.Monad.Catch(finally, bracket, try)
38+import Control.Rematch (equalTo)
2739import Data.Binary
28−import Data.List (find, sort)
29−import Data.Maybe (isJust)
40+import Data.List (find, sort, intercalate)
41+import Data.Maybe (isJust, fromJust, isNothing, fromMaybe, catMaybes)
3042import Data.Typeable
31−import GHC.Generics
43+import GHC.Generics hiding (from)
3244#if ! MIN_VERSION_base(4,6,0)
3345import Prelude hiding (catch, log)
3446#endif
@@ -44,7 +56,17 @@ data Publish = Publish
4456
4557instance Binary Publish where
4658
47−testAgentBroadcast :: TestResult () -> Process ()
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 ()
4870testAgentBroadcast result = do
4971 (resultSP, resultRP) <- newChan :: Process (SendPort (), ReceivePort ())
5072
@@ -59,7 +81,7 @@ testAgentBroadcast result = do
5981 mxNotify ()
6082 -- Once the publisher has seen our message, it will broadcast the Publish
6183 -- and the consumer will see that and send the result to our typed channel.
62− stash result =<< receiveChan resultRP
84+ stash result =<< receiveChanTimeout 10000000 resultRP
6385
6486 kill publisher "finished"
6587 kill consumer "finished"
@@ -67,7 +89,7 @@ testAgentBroadcast result = do
6789testAgentDualInput :: TestResult (Maybe Int) -> Process ()
6890testAgentDualInput result = do
6991 (sp, rp) <- newChan
70− _ <- mxAgent (MxAgentId "sum-agent") (0 :: Int) [
92+ s <- mxAgent (MxAgentId "sum-agent") (0 :: Int) [
7193 mxSink $ (\(i :: Int) -> do
7294 mxSetLocal . (+i) =<< mxGetLocal
7395 i' <- mxGetLocal
@@ -77,6 +99,8 @@ testAgentDualInput result = do
7799 else mxReady)
78100 ]
79101
102+ mRef <- monitor s
103+
80104 mxNotify (1 :: Int)
81105 nsend "sum-agent" (3 :: Int)
82106 mxNotify (2 :: Int)
@@ -84,6 +108,10 @@ testAgentDualInput result = do
84108 mxNotify (5 :: Int)
85109
86110 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)
87115
88116testAgentPrioritisation :: TestResult [String] -> Process ()
89117testAgentPrioritisation result = do
@@ -95,12 +123,12 @@ testAgentPrioritisation result = do
95123
96124 let name = "prioritising-agent"
97125 (sp, rp) <- newChan
98− void $ mxAgent (MxAgentId name) ["first"] [
99− mxSink (\(s :: String) -> do
100− mxUpdateLocal ((s:))
126+ s <- mxAgent (MxAgentId name) ["first"] [
127+ mxSink (\(s :: String) -> do
128+ mxUpdateLocal (s:)
101129 st <- mxGetLocal
102130 case length st of
103− n | n == 5 -> do liftMX $ sendChan sp st
131+ n | n == 5 -> do liftMX $ sendChan sp (sort st)
104132 mxDeactivate "finished"
105133 _ -> mxReceive -- go to the mailbox
106134 )
@@ -111,23 +139,33 @@ testAgentPrioritisation result = do
111139 mxNotify "fourth"
112140 nsend name "fifth"
113141
114− stash result . sort =<< receiveChan rp
142+ stash result . fromMaybe [] =<< receiveChanTimeout 10000000 rp
143+ awaitExit s
115144
116145testAgentMailboxHandling :: TestResult (Maybe ()) -> Process ()
117146testAgentMailboxHandling result = do
118147 (sp, rp) <- newChan
119− agent <- mxAgent (MxAgentId "listener-agent") () [
148+ agent <- mxAgent (MxAgentId "mailbox-agent") () [
120149 mxSink $ \() -> (liftMX $ sendChan sp ()) >> mxReady
121150 ]
122151
123− nsend "listener-agent" ()
152+ nsend "mailbox-agent" ()
124153
125154 stash result =<< receiveChanTimeout 1000000 rp
126155 kill agent "finished"
127156
128157testAgentEventHandling :: TestResult Bool -> Process ()
129158testAgentEventHandling 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+
130167 let initState = [] :: [MxEvent]
168+ (rc, rs) <- newChan
131169 agentPid <- mxAgent (MxAgentId "lifecycle-listener-agent") initState [
132170 (mxSink $ \ev -> do
133171 st <- mxGetLocal
@@ -136,7 +174,7 @@ testAgentEventHandling result = do
136174 (MxSpawned _) -> mxSetLocal (ev:st)
137175 (MxProcessDied _ _) -> mxSetLocal (ev:st)
138176 _ -> return ()
139− act >> mxReady),
177+ act >> (liftMX $ sendChan rc ()) >> mxReady),
140178 (mxSink $ \(ev, sp :: SendPort Bool) -> do
141179 st <- mxGetLocal
142180 let found =
@@ -156,50 +194,349 @@ testAgentEventHandling result = do
156194 mxReady)
157195 ]
158196
159− _ <- monitor agentPid
160− (sp, rp) <- newChan
161− pid <- spawnLocal $ sendChan sp ()
162− () <- receiveChan rp
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
163203
164204 -- By waiting for a monitor notification, we have a
165− -- higher probably that the agent has seen
205+ -- higher probably that the agent has seen the spawn and died events
166206 monitor pid
167− receiveWait [ match (\(ProcessMonitorNotification _ _ _) -> return ()) ]
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
168217
169218 (replyTo, reply) <- newChan :: Process (SendPort Bool, ReceivePort Bool)
170219 mxNotify (MxSpawned pid, replyTo)
171220 mxNotify (MxProcessDied pid DiedNormal, replyTo)
172221
173− seenAlive <- receiveChan reply
174− seenDead <- receiveChan reply
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"
175324
176− stash result $ seenAlive && seenDead
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
177466
178467tests :: TestTransport -> IO [Test]
179468tests TestTransport{..} = do
180469 node1 <- newLocalNode testTransport initRemoteTable
470+ node2 <- newLocalNode testTransport initRemoteTable
181471 return [
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− ]]
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+ ]