git.iko.soy/distributed-process
on fix-spawnChan help sign in
distributed-process/tests/TestClosure.hs
480 lines 16 kB history raw
1{-# LANGUAGE TemplateHaskell #-}
2module Main where
4import Data.ByteString.Lazy (empty)
5import Data.Typeable (Typeable)
6import Control.Monad (join, replicateM, forever, replicateM_, void)
7import Control.Exception (IOException, throw)
8import Control.Concurrent (forkIO, threadDelay)
9import Control.Concurrent.MVar
10 ( MVar
11 , newEmptyMVar
12 , readMVar
13 , takeMVar
14 , putMVar
15 , modifyMVar_
16 , newMVar
17 )
18import Control.Applicative ((<$>))
19import System.Random (randomIO)
20import Network.Transport (Transport)
21import Network.Transport.TCP
22 ( createTransportExposeInternals
23 , defaultTCPParameters
24 , TransportInternals(socketBetween)
25 )
26import Network.Socket (sClose)
27import Control.Distributed.Process
28import Control.Distributed.Process.Closure
29import Control.Distributed.Process.Node
30import Control.Distributed.Process.Internal.Types (NodeId(nodeAddress))
31import Control.Distributed.Static (staticLabel, staticClosure)
33import Test.HUnit (Assertion)
34import Test.Framework (Test, defaultMain)
35import Test.Framework.Providers.HUnit (testCase)
37--------------------------------------------------------------------------------
38-- Supporting definitions --
39--------------------------------------------------------------------------------
41quintuple :: a -> b -> c -> d -> e -> (a, b, c, d, e)
42quintuple a b c d e = (a, b, c, d, e)
44sdictInt :: SerializableDict Int
45sdictInt = SerializableDict
47factorial :: Int -> Process Int
48factorial 0 = return 1
49factorial n = (n *) <$> factorial (n - 1)
51addInt :: Int -> Int -> Int
52addInt x y = x + y
54putInt :: Int -> MVar Int -> IO ()
55putInt = flip putMVar
57sendPid :: ProcessId -> Process ()
58sendPid toPid = do
59 fromPid <- getSelfPid
60 send toPid fromPid
62wait :: Int -> Process ()
63wait = liftIO . threadDelay
65isPrime :: Integer -> Process Bool
66isPrime n = return . (n `elem`) . takeWhile (<= n) . sieve $ [2..]
67 where
68 sieve :: [Integer] -> [Integer]
69 sieve (p : xs) = p : sieve [x | x <- xs, x `mod` p > 0]
70 sieve [] = error "Uh oh -- we've run out of primes"
72-- | First argument indicates empty closure environment
73typedPingServer :: () -> ReceivePort (SendPort ()) -> Process ()
74typedPingServer () rport = forever $ do
75 sport <- receiveChan rport
76 sendChan sport ()
78signal :: ProcessId -> Process ()
79signal pid = send pid ()
81remotable [ 'factorial
82 , 'addInt
83 , 'putInt
84 , 'sendPid
85 , 'sdictInt
86 , 'wait
87 , 'typedPingServer
88 , 'isPrime
89 , 'quintuple
90 , 'signal
91 ]
93randomElement :: [a] -> IO a
94randomElement xs = do
95 ix <- randomIO
96 return (xs !! (ix `mod` length xs))
98remotableDecl [
99 [d| dfib :: ([NodeId], SendPort Integer, Integer) -> Process () ;
100 dfib (_, reply, 0) = sendChan reply 0
101 dfib (_, reply, 1) = sendChan reply 1
102 dfib (nids, reply, n) = do
103 nid1 <- liftIO $ randomElement nids
104 nid2 <- liftIO $ randomElement nids
105 (sport, rport) <- newChan
106 spawn nid1 $ $(mkClosure 'dfib) (nids, sport, n - 2)
107 spawn nid2 $ $(mkClosure 'dfib) (nids, sport, n - 1)
108 n1 <- receiveChan rport
109 n2 <- receiveChan rport
110 sendChan reply $ n1 + n2
111 |]
112 ]
114-- Just try creating a static polymorphic value
115staticQuintuple :: (Typeable a, Typeable b, Typeable c, Typeable d, Typeable e)
116 => Static (a -> b -> c -> d -> e -> (a, b, c, d, e))
117staticQuintuple = $(mkStatic 'quintuple)
119factorialClosure :: Int -> Closure (Process Int)
120factorialClosure = $(mkClosure 'factorial)
122addIntClosure :: Int -> Closure (Int -> Int)
123addIntClosure = $(mkClosure 'addInt)
125putIntClosure :: Int -> Closure (MVar Int -> IO ())
126putIntClosure = $(mkClosure 'putInt)
128sendPidClosure :: ProcessId -> Closure (Process ())
129sendPidClosure = $(mkClosure 'sendPid)
131sendFac :: Int -> ProcessId -> Closure (Process ())
132sendFac n pid = factorialClosure n `bindCP` cpSend $(mkStatic 'sdictInt) pid
134factorialOf :: Closure (Int -> Process Int)
135factorialOf = staticClosure $(mkStatic 'factorial)
137factorial' :: Int -> Closure (Process Int)
138factorial' n = returnCP $(mkStatic 'sdictInt) n `bindCP` factorialOf
140waitClosure :: Int -> Closure (Process ())
141waitClosure = $(mkClosure 'wait)
143simulateNetworkFailure :: TransportInternals -> NodeId -> NodeId -> Process ()
144simulateNetworkFailure transportInternals fr to = liftIO $ do
145 threadDelay 10000
146 sock <- socketBetween transportInternals (nodeAddress fr) (nodeAddress to)
147 sClose sock
148 threadDelay 10000
150--------------------------------------------------------------------------------
151-- The tests proper --
152--------------------------------------------------------------------------------
154testUnclosure :: Transport -> RemoteTable -> Assertion
155testUnclosure transport rtable = do
156 node <- newLocalNode transport rtable
157 done <- newEmptyMVar
158 forkProcess node $ do
159 120 <- join . unClosure $ factorialClosure 5
160 liftIO $ putMVar done ()
161 takeMVar done
163testBind :: Transport -> RemoteTable -> Assertion
164testBind transport rtable = do
165 node <- newLocalNode transport rtable
166 done <- newEmptyMVar
167 runProcess node $ do
168 us <- getSelfPid
169 join . unClosure $ sendFac 6 us
170 (720 :: Int) <- expect
171 liftIO $ putMVar done ()
172 takeMVar done
174testSendPureClosure :: Transport -> RemoteTable -> Assertion
175testSendPureClosure transport rtable = do
176 serverAddr <- newEmptyMVar
177 serverDone <- newEmptyMVar
179 forkIO $ do
180 node <- newLocalNode transport rtable
181 addr <- forkProcess node $ do
182 cl <- expect
183 fn <- unClosure cl :: Process (Int -> Int)
184 13 <- return $ fn 6
185 liftIO $ putMVar serverDone ()
186 putMVar serverAddr addr
188 forkIO $ do
189 node <- newLocalNode transport rtable
190 theirAddr <- readMVar serverAddr
191 runProcess node $ send theirAddr (addIntClosure 7)
193 takeMVar serverDone
195testSendIOClosure :: Transport -> RemoteTable -> Assertion
196testSendIOClosure transport rtable = do
197 serverAddr <- newEmptyMVar
198 serverDone <- newEmptyMVar
200 forkIO $ do
201 node <- newLocalNode transport rtable
202 addr <- forkProcess node $ do
203 cl <- expect
204 io <- unClosure cl :: Process (MVar Int -> IO ())
205 liftIO $ do
206 someMVar <- newEmptyMVar
207 io someMVar
208 5 <- readMVar someMVar
209 putMVar serverDone ()
210 putMVar serverAddr addr
212 forkIO $ do
213 node <- newLocalNode transport rtable
214 theirAddr <- readMVar serverAddr
215 runProcess node $ send theirAddr (putIntClosure 5)
217 takeMVar serverDone
219testSendProcClosure :: Transport -> RemoteTable -> Assertion
220testSendProcClosure transport rtable = do
221 serverAddr <- newEmptyMVar
222 clientDone <- newEmptyMVar
224 forkIO $ do
225 node <- newLocalNode transport rtable
226 addr <- forkProcess node $ do
227 cl <- expect
228 pr <- unClosure cl :: Process (Int -> Process ())
229 pr 5
230 putMVar serverAddr addr
232 forkIO $ do
233 node <- newLocalNode transport rtable
234 theirAddr <- readMVar serverAddr
235 runProcess node $ do
236 pid <- getSelfPid
237 send theirAddr (cpSend $(mkStatic 'sdictInt) pid)
238 5 <- expect :: Process Int
239 liftIO $ putMVar clientDone ()
241 takeMVar clientDone
243testSpawn :: Transport -> RemoteTable -> Assertion
244testSpawn transport rtable = do
245 serverNodeAddr <- newEmptyMVar
246 clientDone <- newEmptyMVar
248 forkIO $ do
249 node <- newLocalNode transport rtable
250 putMVar serverNodeAddr (localNodeId node)
252 forkIO $ do
253 node <- newLocalNode transport rtable
254 nid <- readMVar serverNodeAddr
255 runProcess node $ do
256 pid <- getSelfPid
257 pid' <- spawn nid (sendPidClosure pid)
258 pid'' <- expect
259 True <- return $ pid' == pid''
260 liftIO $ putMVar clientDone ()
262 takeMVar clientDone
264testCall :: Transport -> RemoteTable -> Assertion
265testCall transport rtable = do
266 serverNodeAddr <- newEmptyMVar
267 clientDone <- newEmptyMVar
269 forkIO $ do
270 node <- newLocalNode transport rtable
271 putMVar serverNodeAddr (localNodeId node)
273 forkIO $ do
274 node <- newLocalNode transport rtable
275 nid <- readMVar serverNodeAddr
276 runProcess node $ do
277 (120 :: Int) <- call $(mkStatic 'sdictInt) nid (factorialClosure 5)
278 liftIO $ putMVar clientDone ()
280 takeMVar clientDone
282testCallBind :: Transport -> RemoteTable -> Assertion
283testCallBind transport rtable = do
284 serverNodeAddr <- newEmptyMVar
285 clientDone <- newEmptyMVar
287 forkIO $ do
288 node <- newLocalNode transport rtable
289 putMVar serverNodeAddr (localNodeId node)
291 forkIO $ do
292 node <- newLocalNode transport rtable
293 nid <- readMVar serverNodeAddr
294 runProcess node $ do
295 (120 :: Int) <- call $(mkStatic 'sdictInt) nid (factorial' 5)
296 liftIO $ putMVar clientDone ()
298 takeMVar clientDone
300testSeq :: Transport -> RemoteTable -> Assertion
301testSeq transport rtable = do
302 node <- newLocalNode transport rtable
303 done <- newEmptyMVar
304 runProcess node $ do
305 us <- getSelfPid
306 join . unClosure $ sendFac 5 us `seqCP` sendFac 6 us
307 120 :: Int <- expect
308 720 :: Int <- expect
309 liftIO $ putMVar done ()
310 takeMVar done
312-- Test 'spawnSupervised'
313--
314-- Set up a supervisor, spawn a child, then have a third process monitor the
315-- child. The supervisor then throws an exception, the child dies because it
316-- was linked to the supervisor, and the third process notices that the child
317-- dies.
318testSpawnSupervised :: Transport -> RemoteTable -> Assertion
319testSpawnSupervised transport rtable = do
320 [node1, node2] <- replicateM 2 $ newLocalNode transport rtable
321 [superPid, childPid] <- replicateM 2 $ newEmptyMVar
322 thirdProcessDone <- newEmptyMVar
324 forkProcess node1 $ do
325 us <- getSelfPid
326 liftIO $ putMVar superPid us
327 (child, _ref) <- spawnSupervised (localNodeId node2) (waitClosure 1000000)
328 liftIO $ do
329 putMVar childPid child
330 threadDelay 500000 -- Give the child a chance to link to us
331 throw supervisorDeath
333 forkProcess node2 $ do
334 [super, child] <- liftIO $ mapM readMVar [superPid, childPid]
335 ref <- monitor child
336 ProcessMonitorNotification ref' pid' (DiedException e) <- expect
337 True <- return $ ref' == ref
338 && pid' == child
339 && e == show (ProcessLinkException super (DiedException (show supervisorDeath)))
340 liftIO $ putMVar thirdProcessDone ()
342 takeMVar thirdProcessDone
343 where
344 supervisorDeath :: IOException
345 supervisorDeath = userError "Supervisor died"
347testSpawnInvalid :: Transport -> RemoteTable -> Assertion
348testSpawnInvalid transport rtable = do
349 node <- newLocalNode transport rtable
350 done <- newEmptyMVar
351 forkProcess node $ do
352 (pid, ref) <- spawnMonitor (localNodeId node) (closure (staticLabel "ThisDoesNotExist") empty)
353 ProcessMonitorNotification ref' pid' _reason <- expect
354 -- Depending on the exact interleaving, reason might be NoProc or the exception thrown by the absence of the static closure
355 True <- return $ ref' == ref && pid == pid'
356 liftIO $ putMVar done ()
357 takeMVar done
359testClosureExpect :: Transport -> RemoteTable -> Assertion
360testClosureExpect transport rtable = do
361 node <- newLocalNode transport rtable
362 done <- newEmptyMVar
363 runProcess node $ do
364 nodeId <- getSelfNode
365 us <- getSelfPid
366 them <- spawn nodeId $ cpExpect $(mkStatic 'sdictInt) `bindCP` cpSend $(mkStatic 'sdictInt) us
367 send them (1234 :: Int)
368 (1234 :: Int) <- expect
369 liftIO $ putMVar done ()
370 takeMVar done
372testSpawnChannel :: Transport -> RemoteTable -> Assertion
373testSpawnChannel transport rtable = do
374 done <- newEmptyMVar
375 [node1, node2] <- replicateM 2 $ newLocalNode transport rtable
377 forkProcess node1 $ do
378 pingServer <- spawnChannel
379 (sdictSendPort sdictUnit)
380 (localNodeId node2)
381 ($(mkClosure 'typedPingServer) ())
382 (sendReply, receiveReply) <- newChan
383 sendChan pingServer sendReply
384 receiveChan receiveReply
385 liftIO $ putMVar done ()
387 takeMVar done
389testTDict :: Transport -> RemoteTable -> Assertion
390testTDict transport rtable = do
391 done <- newEmptyMVar
392 [node1, node2] <- replicateM 2 $ newLocalNode transport rtable
393 forkProcess node1 $ do
394 True <- call $(functionTDict 'isPrime) (localNodeId node2) ($(mkClosure 'isPrime) (79 :: Integer))
395 liftIO $ putMVar done ()
396 takeMVar done
398testFib :: Transport -> RemoteTable -> Assertion
399testFib transport rtable = do
400 nodes <- replicateM 4 $ newLocalNode transport rtable
401 done <- newEmptyMVar
403 forkProcess (head nodes) $ do
404 (sport, rport) <- newChan
405 spawnLocal $ dfib (map localNodeId nodes, sport, 10)
406 55 <- receiveChan rport :: Process Integer
407 liftIO $ putMVar done ()
409 takeMVar done
411testSpawnReconnect :: Transport -> RemoteTable -> TransportInternals -> Assertion
412testSpawnReconnect transport rtable transportInternals = do
413 [node1, node2] <- replicateM 2 $ newLocalNode transport rtable
414 let nid1 = localNodeId node1
415 nid2 = localNodeId node2
416 done <- newEmptyMVar
417 iv <- newMVar (0 :: Int)
419 incr <- forkProcess node1 $ forever $ do
420 () <- expect
421 liftIO $ modifyMVar_ iv (return . (+ 1))
423 forkProcess node2 $ do
424 _pid1 <- spawn nid1 ($(mkClosure 'signal) incr)
425 simulateNetworkFailure transportInternals nid2 nid1
426 _pid2 <- spawn nid1 ($(mkClosure 'signal) incr)
427 _pid3 <- spawn nid1 ($(mkClosure 'signal) incr)
429 liftIO $ threadDelay 100000
431 count <- liftIO $ takeMVar iv
432 True <- return $ count == 2 || count == 3 -- It depends on which message we get first in 'spawn'
434 liftIO $ putMVar done ()
436 takeMVar done
438-- | 'spawn' used to ave a race condition which would be triggered if the
439-- spawning process terminates immediately after spawning
440testSpawnTerminate :: Transport -> RemoteTable -> Assertion
441testSpawnTerminate transport rtable = do
442 slave <- newLocalNode transport rtable
443 master <- newLocalNode transport rtable
444 masterDone <- newEmptyMVar
446 runProcess master $ do
447 us <- getSelfPid
448 replicateM_ 1000 . spawnLocal . void . spawn (localNodeId slave) $ $(mkClosure 'signal) us
449 replicateM_ 1000 $ (expect :: Process ())
450 liftIO $ putMVar masterDone ()
452 takeMVar masterDone
454tests :: (Transport, TransportInternals) -> RemoteTable -> [Test]
455tests (transport, transportInternals) rtable = [
456 testCase "Unclosure" (testUnclosure transport rtable)
457 , testCase "Bind" (testBind transport rtable)
458 , testCase "SendPureClosure" (testSendPureClosure transport rtable)
459 , testCase "SendIOClosure" (testSendIOClosure transport rtable)
460 , testCase "SendProcClosure" (testSendProcClosure transport rtable)
461 , testCase "Spawn" (testSpawn transport rtable)
462 , testCase "Call" (testCall transport rtable)
463 , testCase "CallBind" (testCallBind transport rtable)
464 , testCase "Seq" (testSeq transport rtable)
465 , testCase "SpawnSupervised" (testSpawnSupervised transport rtable)
466 , testCase "SpawnInvalid" (testSpawnInvalid transport rtable)
467 , testCase "ClosureExpect" (testClosureExpect transport rtable)
468 , testCase "SpawnChannel" (testSpawnChannel transport rtable)
469 , testCase "TDict" (testTDict transport rtable)
470 , testCase "Fib" (testFib transport rtable)
471 , testCase "SpawnTerminate" (testSpawnTerminate transport rtable)
472 , testCase "SpawnReconnect" (testSpawnReconnect transport rtable transportInternals)
473 ]
476main :: IO ()
477main = do
478 Right transport <- createTransportExposeInternals "127.0.0.1" "8080" defaultTCPParameters
479 let rtable = __remoteTable . __remoteTableDecl $ initRemoteTable
480 defaultMain (tests transport rtable)