git.iko.soy/distributed-process
on 7930062 help sign in
distributed-process/distributed-process-tests/src/Control/Distributed/Process/Tests/Mx.hs
355 lines 12 kB history raw
1{-# LANGUAGE DeriveGeneric #-}
2module Control.Distributed.Process.Tests.Mx (tests) where
4import Control.Distributed.Process.Tests.Internal.Utils
5import Network.Transport.Test (TestTransport(..))
7import Control.Distributed.Process
8import Control.Distributed.Process.Node
9import Control.Distributed.Process.Management
10 ( MxEvent(..)
11 , MxAgentId(..)
12 , mxAgent
13 , mxSink
14 , mxReady
15 , mxReceive
16 , mxDeactivate
17 , liftMX
18 , mxGetLocal
19 , mxSetLocal
20 , mxUpdateLocal
21 , mxNotify
22 , mxBroadcast
23 )
24import Control.Monad (void, unless)
25import Control.Rematch (equalTo)
26import Data.Binary
27import Data.List (find, sort)
28import Data.Maybe (isJust, isNothing)
29import Data.Typeable
30import GHC.Generics
31#if ! MIN_VERSION_base(4,6,0)
32import Prelude hiding (catch, log)
33#endif
35import Test.Framework
36 ( Test
37 , testGroup
38 )
39import Test.Framework.Providers.HUnit (testCase)
41data Publish = Publish
42 deriving (Typeable, Generic, Eq)
44instance Binary Publish where
46awaitExit :: ProcessId -> Process ()
47awaitExit pid =
48 withMonitorRef pid $ \ref -> do
49 receiveWait
50 [ matchIf (\(ProcessMonitorNotification r _ _) -> r == ref)
51 (\_ -> return ())
52 ]
53 where
54 withMonitorRef pid = bracket (monitor pid) unmonitor
56testAgentBroadcast :: TestResult () -> Process ()
57testAgentBroadcast result = do
58 (resultSP, resultRP) <- newChan :: Process (SendPort (), ReceivePort ())
60 publisher <- mxAgent (MxAgentId "publisher-agent") () [
61 mxSink $ \() -> mxBroadcast Publish >> mxReady
62 ]
64 consumer <- mxAgent (MxAgentId "consumer-agent") () [
65 mxSink $ \Publish -> (liftMX $ sendChan resultSP ()) >> mxReady
66 ]
68 mxNotify ()
69 -- Once the publisher has seen our message, it will broadcast the Publish
70 -- and the consumer will see that and send the result to our typed channel.
71 stash result =<< receiveChan resultRP
73 kill publisher "finished" >> awaitExit publisher
74 kill consumer "finished" >> awaitExit consumer
76testAgentDualInput :: TestResult (Maybe Int) -> Process ()
77testAgentDualInput result = do
78 (sp, rp) <- newChan
79 s <- mxAgent (MxAgentId "sum-agent") (0 :: Int) [
80 mxSink $ (\(i :: Int) -> do
81 mxSetLocal . (+i) =<< mxGetLocal
82 i' <- mxGetLocal
83 if i' == 15
84 then do mxGetLocal >>= liftMX . sendChan sp
85 mxDeactivate "finished"
86 else mxReady)
87 ]
89 mxNotify (1 :: Int)
90 nsend "sum-agent" (3 :: Int)
91 mxNotify (2 :: Int)
92 nsend "sum-agent" (4 :: Int)
93 mxNotify (5 :: Int)
95 stash result =<< receiveChanTimeout 10000000 rp
96 awaitExit s
98testAgentPrioritisation :: TestResult [String] -> Process ()
99testAgentPrioritisation result = do
101 -- TODO: this isn't really testing how we /prioritise/ one source
102 -- over another at all, but I've not yet figured out the right way
103 -- to do so, since we're at the whim of the scheduler with regards
104 -- the timeliness of nsend versus mxNotify anyway.
106 let name = "prioritising-agent"
107 (sp, rp) <- newChan
108 s <- mxAgent (MxAgentId name) ["first"] [
109 mxSink (\(s :: String) -> do
110 mxUpdateLocal ((s:))
111 st <- mxGetLocal
112 case length st of
113 n | n == 5 -> do liftMX $ sendChan sp st
114 mxDeactivate "finished"
115 _ -> mxReceive -- go to the mailbox
116 )
117 ]
119 nsend name "second"
120 mxNotify "third"
121 mxNotify "fourth"
122 nsend name "fifth"
124 stash result . sort =<< receiveChan rp
125 awaitExit s
127testAgentMailboxHandling :: TestResult (Maybe ()) -> Process ()
128testAgentMailboxHandling result = do
129 (sp, rp) <- newChan
130 agent <- mxAgent (MxAgentId "mailbox-agent") () [
131 mxSink $ \() -> (liftMX $ sendChan sp ()) >> mxReady
132 ]
134 nsend "mailbox-agent" ()
136 stash result =<< receiveChanTimeout 1000000 rp
137 kill agent "finished" >> awaitExit agent
139testAgentEventHandling :: TestResult Bool -> Process ()
140testAgentEventHandling result = do
141 let initState = [] :: [MxEvent]
143 (rc, rs) <- newChan
144 agentPid <- mxAgent (MxAgentId "lifecycle-listener-agent") initState [
145 (mxSink $ \ev -> do
146 st <- mxGetLocal
147 let act =
148 case ev of
149 (MxSpawned _) -> mxSetLocal (ev:st)
150 (MxProcessDied _ _) -> mxSetLocal (ev:st)
151 _ -> return ()
152 act >> (liftMX $ sendChan rc ()) >> mxReady),
153 (mxSink $ \(ev, sp :: SendPort Bool) -> do
154 st <- mxGetLocal
155 let found =
156 case ev of
157 MxSpawned p ->
158 isJust $ find (\ev' ->
159 case ev' of
160 (MxSpawned p') -> p' == p
161 _ -> False) st
162 MxProcessDied p r ->
163 isJust $ find (\ev' ->
164 case ev' of
165 (MxProcessDied p' r') -> p' == p && r == r'
166 _ -> False) st
167 _ -> False
168 liftMX $ sendChan sp found
169 mxReady)
170 ]
172 -- TODO: yes, this is racy, but we're at the mercy of the scheduler here...
173 faff 2000000 rs
175 _ <- monitor agentPid
176 (sp, rp) <- newChan
178 pid <- spawnLocal $ expect >>= sendChan sp
180 -- By waiting for a monitor notification, we have a
181 -- higher probably that the agent has seen the spawn and died events
182 monitor pid
184 send pid ()
185 () <- receiveChan rp
187 receiveWait [ match (\(ProcessMonitorNotification _ _ _) -> return ()) ]
189 -- TODO: yes, this is racy, but we're at the mercy of the scheduler here...
190 faff 2000000 rs
192 (replyTo, reply) <- newChan :: Process (SendPort Bool, ReceivePort Bool)
193 mxNotify (MxSpawned pid, replyTo)
194 mxNotify (MxProcessDied pid DiedNormal, replyTo)
196 seenAlive <- receiveChan reply
197 seenDead <- receiveChan reply
199 stash result $ seenAlive && seenDead
200 kill agentPid "test-complete" >> awaitExit agentPid
201 where
202 faff delay port = do
203 res <- receiveChanTimeout delay port
204 unless (isNothing res) $ do
205 pause delay
206 faff delay port
208testMxRegEvents :: TestResult () -> Process ()
209testMxRegEvents result = do
210 {- This test only deals with the local case, to ensure that we are being
211 notified in the expected order - the remote cases related to the
212 behaviour of the node controller are contained in the CH test suite. -}
213 ensure (stash result ()) $ do
214 let label = "testMxRegEvents"
215 let agentLabel = "mxRegEvents-agent"
216 let delay = 1000000
217 (regChan, regSink) <- newChan
218 (unRegChan, unRegSink) <- newChan
219 agent <- mxAgent (MxAgentId agentLabel) () [
220 mxSink $ \ev -> do
221 case ev of
222 MxRegistered pid label'
223 | label' == label -> liftMX $ sendChan regChan (label', pid)
224 MxUnRegistered pid label'
225 | label' == label -> liftMX $ sendChan unRegChan (label', pid)
226 _ -> return ()
227 mxReady
228 ]
230 p1 <- spawnLocal expect
231 p2 <- spawnLocal expect
233 register label p1
234 reg1 <- receiveChanTimeout delay regSink
235 reg1 `shouldBe` equalTo (Just (label, p1))
237 unregister label
238 unreg1 <- receiveChanTimeout delay unRegSink
239 unreg1 `shouldBe` equalTo (Just (label, p1))
241 register label p2
242 reg2 <- receiveChanTimeout delay regSink
243 reg2 `shouldBe` equalTo (Just (label, p2))
245 reregister label p1
246 unreg2 <- receiveChanTimeout delay unRegSink
247 unreg2 `shouldBe` equalTo (Just (label, p2))
249 reg3 <- receiveChanTimeout delay regSink
250 reg3 `shouldBe` equalTo (Just (label, p1))
252 mapM_ (flip kill $ "test-complete") [agent, p1, p2]
253 awaitExit agent
255testMxRegMon :: LocalNode -> TestResult () -> Process ()
256testMxRegMon remoteNode result = do
257 ensure (stash result ()) $ do
258 -- ensure that when a registered process dies, we get a notification that
259 -- it has been unregistered as well as seeing the name get removed
260 let label1 = "aaaaa"
261 let label2 = "bbbbb"
262 let isValid l = l == label1 || l == label2
263 let agentLabel = "mxRegMon-agent"
264 let delay = 1000000
265 (regChan, regSink) <- newChan
266 (unRegChan, unRegSink) <- newChan
267 agent <- mxAgent (MxAgentId agentLabel) () [
268 mxSink $ \ev -> do
269 case ev of
270 MxRegistered pid label
271 | isValid label -> liftMX $ sendChan regChan (label, pid)
272 MxUnRegistered pid label
273 | isValid label -> liftMX $ sendChan unRegChan (label, pid)
274 _ -> return ()
275 mxReady
276 ]
278 (sp, rp) <- newChan
279 liftIO $ forkProcess remoteNode $ do
280 getSelfPid >>= sendChan sp
281 expect :: Process ()
283 p1 <- receiveChan rp
285 register label1 p1
286 reg1 <- receiveChanTimeout delay regSink
287 reg1 `shouldBe` equalTo (Just (label1, p1))
289 register label2 p1
290 reg2 <- receiveChanTimeout delay regSink
291 reg2 `shouldBe` equalTo (Just (label2, p1))
293 n1 <- whereis label1
294 n1 `shouldBe` equalTo (Just p1)
296 n2 <- whereis label2
297 n2 `shouldBe` equalTo (Just p1)
299 kill p1 "goodbye"
301 unreg1 <- receiveChanTimeout delay unRegSink
302 unreg2 <- receiveChanTimeout delay unRegSink
304 let evts = [unreg1, unreg2]
305 -- we can't rely on the order of the values in the node controller's
306 -- map (it's either racy to do so, or no such guarantee exists for Data.Map),
307 -- so we simply verify that we received the un-registration events we expect
308 evts `shouldContain` (Just (label1, p1))
309 evts `shouldContain` (Just (label2, p1))
311 kill agent "test-complete" >> awaitExit agent
313ensure :: Process () -> Process () -> Process ()
314ensure = flip finally
316tests :: TestTransport -> IO [Test]
317tests TestTransport{..} = do
318 node1 <- newLocalNode testTransport initRemoteTable
319 node2 <- newLocalNode testTransport initRemoteTable
320 return [
321 testGroup "Mx Agents" [
322 testCase "Event Handling"
323 (delayedAssertion
324 "expected True, but events where not as expected"
325 node1 True testAgentEventHandling)
326 , testCase "Inter-Agent Broadcast"
327 (delayedAssertion
328 "expected (), but no broadcast was received"
329 node1 () testAgentBroadcast)
330 , testCase "Agent Mailbox Handling"
331 (delayedAssertion
332 "expected (Just ()), but no regular (mailbox) input was handled"
333 node1 (Just ()) testAgentMailboxHandling)
334 , testCase "Agent Dual Input Handling"
335 (delayedAssertion
336 "expected sum = 15, but the result was Nothing"
337 node1 (Just 15 :: Maybe Int) testAgentDualInput)
338 , testCase "Agent Input Prioritisation"
339 (delayedAssertion
340 "expected [first, second, third, fourth, fifth], but result diverged"
341 node1 (sort ["first", "second",
342 "third", "fourth",
343 "fifth"]) testAgentPrioritisation)
344 ]
345 , testGroup "Mx Events" [
346 testCase "Name Registration Events"
347 (delayedAssertion
348 "expected registration events to map to the correct ProcessId"
349 node1 () testMxRegEvents)
350 , testCase "Post Death Name UnRegistration Events"
351 (delayedAssertion
352 "expected process deaths to result in unregistration events"
353 node1 () (testMxRegMon node2))
354 ]
355 ]