git.iko.soy/distributed-process
on fe544e5 help sign in
distributed-process/distributed-process-tests/src/Control/Distributed/Process/Tests/Mx.hs
286 lines 9.3 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 , mxGet
26 , mxSet
27 , mxClear
28 , mxPurgeTable
29 , mxDropTable
30 )
31import Control.Monad (void)
32import Data.Binary
33import Data.List (find, sort)
34import Data.Maybe (isJust)
35import Data.Typeable
36import GHC.Generics
37#if ! MIN_VERSION_base(4,6,0)
38import Prelude hiding (catch, log)
39#endif
41import Test.Framework
42 ( Test
43 , testGroup
44 )
45import Test.Framework.Providers.HUnit (testCase)
47data Publish = Publish
48 deriving (Typeable, Generic, Eq)
50instance Binary Publish where
52testAgentBroadcast :: TestResult () -> Process ()
53testAgentBroadcast result = do
54 (resultSP, resultRP) <- newChan :: Process (SendPort (), ReceivePort ())
56 publisher <- mxAgent (MxAgentId "publisher-agent") () [
57 mxSink $ \() -> mxBroadcast Publish >> mxReady
58 ]
60 consumer <- mxAgent (MxAgentId "consumer-agent") () [
61 mxSink $ \Publish -> (liftMX $ sendChan resultSP ()) >> mxReady
62 ]
64 mxNotify ()
65 -- Once the publisher has seen our message, it will broadcast the Publish
66 -- and the consumer will see that and send the result to our typed channel.
67 stash result =<< receiveChan resultRP
69 kill publisher "finished"
70 kill consumer "finished"
72testAgentDualInput :: TestResult (Maybe Int) -> Process ()
73testAgentDualInput result = do
74 (sp, rp) <- newChan
75 _ <- mxAgent (MxAgentId "sum-agent") (0 :: Int) [
76 mxSink $ (\(i :: Int) -> do
77 mxSetLocal . (+i) =<< mxGetLocal
78 i' <- mxGetLocal
79 if i' == 15
80 then do mxGetLocal >>= liftMX . sendChan sp
81 mxDeactivate "finished"
82 else mxReady)
83 ]
85 mxNotify (1 :: Int)
86 nsend "sum-agent" (3 :: Int)
87 mxNotify (2 :: Int)
88 nsend "sum-agent" (4 :: Int)
89 mxNotify (5 :: Int)
91 stash result =<< receiveChanTimeout 10000000 rp
93testAgentPrioritisation :: TestResult [String] -> Process ()
94testAgentPrioritisation result = do
96 -- TODO: this isn't really testing how we /prioritise/ one source
97 -- over another at all, but I've not yet figured out the right way
98 -- to do so, since we're at the whim of the scheduler with regards
99 -- the timeliness of nsend versus mxNotify anyway.
101 let name = "prioritising-agent"
102 (sp, rp) <- newChan
103 void $ mxAgent (MxAgentId name) ["first"] [
104 mxSink (\(s :: String) -> do
105 mxUpdateLocal ((s:))
106 st <- mxGetLocal
107 case length st of
108 n | n == 5 -> do liftMX $ sendChan sp st
109 mxDeactivate "finished"
110 _ -> mxReceive -- go to the mailbox
111 )
112 ]
114 nsend name "second"
115 mxNotify "third"
116 mxNotify "fourth"
117 nsend name "fifth"
119 stash result . sort =<< receiveChan rp
121testAgentMailboxHandling :: TestResult (Maybe ()) -> Process ()
122testAgentMailboxHandling result = do
123 (sp, rp) <- newChan
124 agent <- mxAgent (MxAgentId "listener-agent") () [
125 mxSink $ \() -> (liftMX $ sendChan sp ()) >> mxReady
126 ]
128 nsend "listener-agent" ()
130 stash result =<< receiveChanTimeout 1000000 rp
131 kill agent "finished"
133testAgentEventHandling :: TestResult Bool -> Process ()
134testAgentEventHandling result = do
135 let initState = [] :: [MxEvent]
136 agentPid <- mxAgent (MxAgentId "lifecycle-listener-agent") initState [
137 (mxSink $ \ev -> do
138 st <- mxGetLocal
139 let act =
140 case ev of
141 (MxSpawned _) -> mxSetLocal (ev:st)
142 (MxProcessDied _ _) -> mxSetLocal (ev:st)
143 _ -> return ()
144 act >> mxReady),
145 (mxSink $ \(ev, sp :: SendPort Bool) -> do
146 st <- mxGetLocal
147 let found =
148 case ev of
149 MxSpawned p ->
150 isJust $ find (\ev' ->
151 case ev' of
152 (MxSpawned p') -> p' == p
153 _ -> False) st
154 MxProcessDied p r ->
155 isJust $ find (\ev' ->
156 case ev' of
157 (MxProcessDied p' r') -> p' == p && r == r'
158 _ -> False) st
159 _ -> False
160 liftMX $ sendChan sp found
161 mxReady)
162 ]
164 _ <- monitor agentPid
165 (sp, rp) <- newChan
166 pid <- spawnLocal $ sendChan sp ()
167 () <- receiveChan rp
169 -- By waiting for a monitor notification, we have a
170 -- higher probably that the agent has seen
171 monitor pid
172 receiveWait [ match (\(ProcessMonitorNotification _ _ _) -> return ()) ]
174 (replyTo, reply) <- newChan :: Process (SendPort Bool, ReceivePort Bool)
175 mxNotify (MxSpawned pid, replyTo)
176 mxNotify (MxProcessDied pid DiedNormal, replyTo)
178 seenAlive <- receiveChan reply
179 seenDead <- receiveChan reply
181 stash result $ seenAlive && seenDead
183testAgentPublication :: TestResult (Maybe Publish) -> Process ()
184testAgentPublication result = do
185 (syncChan, rp) <- newChan
186 let ourId = MxAgentId "publication-agent"
187 agentPid <- mxAgent ourId () [
188 mxSink $ \() -> do
189 selfId <- mxGetId
190 liftMX $ mxSet selfId "publish" Publish >> sendChan syncChan ()
191 mxReady
192 ]
194 mxNotify ()
195 () <- receiveChan rp
197 stash result =<< mxGet ourId "publish"
199 mref <- monitor agentPid
200 kill agentPid "finished"
201 receiveWait [
202 matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mref)
203 ((\_ -> return ()))
204 ]
206testAgentTableClear :: Int -> TestResult (Maybe Int, Maybe Int) -> Process ()
207testAgentTableClear val result =
208 let tId = (MxAgentId "agent-1")
209 tKey = "key-1" in do
210 mxSet tId tKey val
211 get1 <- mxGet tId tKey
212 mxClear tId tKey
213 get2 <- mxGet tId tKey
214 stash result (get1, get2)
216testAgentTablePurge :: TestResult (Maybe Int) -> Process ()
217testAgentTablePurge result =
218 let tId = MxAgentId "agent-2"
219 tKey = "key-2" in do
220 mxSet tId tKey (12345 :: Int)
221 mxPurgeTable tId
222 stash result =<< mxGet tId tKey
224testAgentTableDelete :: Int
225 -> TestResult (Maybe Int, Maybe Int, Maybe Int)
226 -> Process ()
227testAgentTableDelete val result =
228 let tId = (MxAgentId "agent-3")
229 tKey = "key-3" in do
230 mxSet tId tKey val
231 get1 <- mxGet tId tKey
232 mxDropTable tId
233 get2 <- mxGet tId tKey
234 mxSet tId tKey val
235 get3 <- mxGet tId tKey
236 stash result (get1, get2, get3)
238tests :: TestTransport -> IO [Test]
239tests TestTransport{..} = do
240 node1 <- newLocalNode testTransport initRemoteTable
241 return [
242 testGroup "Mx Agents" [
243 testCase "Event Handling"
244 (delayedAssertion
245 "expected True, but events where not as expected"
246 node1 True testAgentEventHandling)
247 , testCase "Inter-Agent Broadcast"
248 (delayedAssertion
249 "expected (), but no broadcast was received"
250 node1 () testAgentBroadcast)
251 , testCase "Agent Mailbox Handling"
252 (delayedAssertion
253 "expected (Just ()), but no regular (mailbox) input was handled"
254 node1 (Just ()) testAgentMailboxHandling)
255 , testCase "Agent Dual Input Handling"
256 (delayedAssertion
257 "expected sum = 15, but the result was Nothing"
258 node1 (Just 15 :: Maybe Int) testAgentDualInput)
259 , testCase "Agent Input Prioritisation"
260 (delayedAssertion
261 "expected [first, second, third, fourth, fifth], but result diverged"
262 node1 (sort ["first", "second",
263 "third", "fourth",
264 "fifth"]) testAgentPrioritisation)
265 ],
266 testGroup "Mx Global Properties" [
267 testCase "Global Property Publication"
268 (delayedAssertion
269 "expected (Just Publish), but no table entry was found"
270 node1 (Just Publish) testAgentPublication)
271 , testCase "Clearing Global Properties"
272 (delayedAssertion
273 "expected (Just 1024, Nothing): invalid table entry found!"
274 node1 (Just 1024, Nothing) (testAgentTableClear 1024))
275 , testCase "Purging Global Tables"
276 (delayedAssertion
277 "expected Nothing, but a table entry was found"
278 node1 Nothing testAgentTablePurge)
279 , testCase "Deleting and (Re)Creating Global Tables"
280 (delayedAssertion
281 "expected (Just 15, Nothing, Just 15): invalid table entry found!"
282 node1 (Just 15, Nothing, Just 15) (testAgentTableDelete 15))
283 -- Wait for other tests to finish.
284 , testCase "Wait" $
285 threadDelay 100000
286 ]]