git.iko.soy/distributed-process
on 6dfd5b1 help sign in
distributed-process/distributed-process-tests/src/Control/Distributed/Process/Tests/Mx.hs
205 lines 6.7 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.Concurrent (threadDelay)
8import Control.Distributed.Process
9import Control.Distributed.Process.Node
10import Control.Distributed.Process.Management
11 ( MxEvent(..)
12 , MxAgentId(..)
13 , mxAgent
14 , mxSink
15 , mxReady
16 , mxReceive
17 , mxDeactivate
18 , liftMX
19 , mxGetLocal
20 , mxSetLocal
21 , mxUpdateLocal
22 , mxNotify
23 , mxBroadcast
24 , mxGetId
25 )
26import Control.Monad (void)
27import Data.Binary
28import Data.List (find, sort)
29import Data.Maybe (isJust)
30import Data.Typeable
31import GHC.Generics
32#if ! MIN_VERSION_base(4,6,0)
33import Prelude hiding (catch, log)
34#endif
36import Test.Framework
37 ( Test
38 , testGroup
39 )
40import Test.Framework.Providers.HUnit (testCase)
42data Publish = Publish
43 deriving (Typeable, Generic, Eq)
45instance Binary Publish where
47testAgentBroadcast :: TestResult () -> Process ()
48testAgentBroadcast result = do
49 (resultSP, resultRP) <- newChan :: Process (SendPort (), ReceivePort ())
51 publisher <- mxAgent (MxAgentId "publisher-agent") () [
52 mxSink $ \() -> mxBroadcast Publish >> mxReady
53 ]
55 consumer <- mxAgent (MxAgentId "consumer-agent") () [
56 mxSink $ \Publish -> (liftMX $ sendChan resultSP ()) >> mxReady
57 ]
59 mxNotify ()
60 -- Once the publisher has seen our message, it will broadcast the Publish
61 -- and the consumer will see that and send the result to our typed channel.
62 stash result =<< receiveChan resultRP
64 kill publisher "finished"
65 kill consumer "finished"
67testAgentDualInput :: TestResult (Maybe Int) -> Process ()
68testAgentDualInput result = do
69 (sp, rp) <- newChan
70 _ <- mxAgent (MxAgentId "sum-agent") (0 :: Int) [
71 mxSink $ (\(i :: Int) -> do
72 mxSetLocal . (+i) =<< mxGetLocal
73 i' <- mxGetLocal
74 if i' == 15
75 then do mxGetLocal >>= liftMX . sendChan sp
76 mxDeactivate "finished"
77 else mxReady)
78 ]
80 mxNotify (1 :: Int)
81 nsend "sum-agent" (3 :: Int)
82 mxNotify (2 :: Int)
83 nsend "sum-agent" (4 :: Int)
84 mxNotify (5 :: Int)
86 stash result =<< receiveChanTimeout 10000000 rp
88testAgentPrioritisation :: TestResult [String] -> Process ()
89testAgentPrioritisation result = do
91 -- TODO: this isn't really testing how we /prioritise/ one source
92 -- over another at all, but I've not yet figured out the right way
93 -- to do so, since we're at the whim of the scheduler with regards
94 -- the timeliness of nsend versus mxNotify anyway.
96 let name = "prioritising-agent"
97 (sp, rp) <- newChan
98 void $ mxAgent (MxAgentId name) ["first"] [
99 mxSink (\(s :: String) -> do
100 mxUpdateLocal ((s:))
101 st <- mxGetLocal
102 case length st of
103 n | n == 5 -> do liftMX $ sendChan sp st
104 mxDeactivate "finished"
105 _ -> mxReceive -- go to the mailbox
106 )
107 ]
109 nsend name "second"
110 mxNotify "third"
111 mxNotify "fourth"
112 nsend name "fifth"
114 stash result . sort =<< receiveChan rp
116testAgentMailboxHandling :: TestResult (Maybe ()) -> Process ()
117testAgentMailboxHandling result = do
118 (sp, rp) <- newChan
119 agent <- mxAgent (MxAgentId "listener-agent") () [
120 mxSink $ \() -> (liftMX $ sendChan sp ()) >> mxReady
121 ]
123 nsend "listener-agent" ()
125 stash result =<< receiveChanTimeout 1000000 rp
126 kill agent "finished"
128testAgentEventHandling :: TestResult Bool -> Process ()
129testAgentEventHandling result = do
130 let initState = [] :: [MxEvent]
131 agentPid <- mxAgent (MxAgentId "lifecycle-listener-agent") initState [
132 (mxSink $ \ev -> do
133 st <- mxGetLocal
134 let act =
135 case ev of
136 (MxSpawned _) -> mxSetLocal (ev:st)
137 (MxProcessDied _ _) -> mxSetLocal (ev:st)
138 _ -> return ()
139 act >> mxReady),
140 (mxSink $ \(ev, sp :: SendPort Bool) -> do
141 st <- mxGetLocal
142 let found =
143 case ev of
144 MxSpawned p ->
145 isJust $ find (\ev' ->
146 case ev' of
147 (MxSpawned p') -> p' == p
148 _ -> False) st
149 MxProcessDied p r ->
150 isJust $ find (\ev' ->
151 case ev' of
152 (MxProcessDied p' r') -> p' == p && r == r'
153 _ -> False) st
154 _ -> False
155 liftMX $ sendChan sp found
156 mxReady)
157 ]
159 _ <- monitor agentPid
160 (sp, rp) <- newChan
161 pid <- spawnLocal $ sendChan sp ()
162 () <- receiveChan rp
164 -- By waiting for a monitor notification, we have a
165 -- higher probably that the agent has seen
166 monitor pid
167 receiveWait [ match (\(ProcessMonitorNotification _ _ _) -> return ()) ]
169 (replyTo, reply) <- newChan :: Process (SendPort Bool, ReceivePort Bool)
170 mxNotify (MxSpawned pid, replyTo)
171 mxNotify (MxProcessDied pid DiedNormal, replyTo)
173 seenAlive <- receiveChan reply
174 seenDead <- receiveChan reply
176 stash result $ seenAlive && seenDead
178tests :: TestTransport -> IO [Test]
179tests TestTransport{..} = do
180 node1 <- newLocalNode testTransport initRemoteTable
181 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 ]]