git.iko.soy/distributed-process
on 42c71eb help sign in
distributed-process/distributed-process-tests/src/Control/Distributed/Process/Tests/Closure.hs
548 lines 19 kB history raw
1{-# LANGUAGE TemplateHaskell, KindSignatures #-}
2module Control.Distributed.Process.Tests.Closure (tests) where
4import Network.Transport.Test (TestTransport(..))
6import Data.ByteString.Lazy (empty)
7import Data.IORef
8import Data.Typeable (Typeable)
9import Data.Maybe
10import Control.Monad (join, replicateM, forever, replicateM_, void, when, unless)
11import Control.Exception (IOException, throw)
12import Control.Concurrent (forkIO, threadDelay)
13import Control.Concurrent.MVar
14 ( MVar
15 , newEmptyMVar
16 , readMVar
17 , takeMVar
18 , putMVar
19 , modifyMVar_
20 , newMVar
21 )
22import Control.Applicative ((<$>))
23import System.Random (randomIO)
24import Control.Distributed.Process
25import Control.Distributed.Process.Closure
26import Control.Distributed.Process.Node
27import Control.Distributed.Process.Internal.Types
28 ( NodeId(nodeAddress)
29 , createMessage
30 , messageToPayload
31 )
32import Control.Distributed.Static (staticLabel, staticClosure)
33import qualified Network.Transport as NT
35import Test.HUnit (Assertion)
36import Test.Framework (Test)
37import Test.Framework.Providers.HUnit (testCase)
39--------------------------------------------------------------------------------
40-- Supporting definitions --
41--------------------------------------------------------------------------------
43quintuple :: a -> b -> c -> d -> e -> (a, b, c, d, e)
44quintuple a b c d e = (a, b, c, d, e)
46sdictInt :: SerializableDict Int
47sdictInt = SerializableDict
49factorial :: Int -> Process Int
50factorial 0 = return 1
51factorial n = (n *) <$> factorial (n - 1)
53addInt :: Int -> Int -> Int
54addInt x y = x + y
56putInt :: Int -> MVar Int -> IO ()
57putInt = flip putMVar
59sendPid :: ProcessId -> Process ()
60sendPid toPid = do
61 fromPid <- getSelfPid
62 send toPid fromPid
64wait :: Int -> Process ()
65wait = liftIO . threadDelay
67expectUnit :: Process ()
68expectUnit = expect
70isPrime :: Integer -> Process Bool
71isPrime n = return . (n `elem`) . takeWhile (<= n) . sieve $ [2..]
72 where
73 sieve :: [Integer] -> [Integer]
74 sieve (p : xs) = p : sieve [x | x <- xs, x `mod` p > 0]
75 sieve [] = error "Uh oh -- we've run out of primes"
77-- | First argument indicates empty closure environment
78typedPingServer :: () -> ReceivePort (SendPort ()) -> Process ()
79typedPingServer () rport = forever $ do
80 sport <- receiveChan rport
81 sendChan sport ()
83signal :: ProcessId -> Process ()
84signal pid = send pid ()
86remotable [ 'factorial
87 , 'addInt
88 , 'putInt
89 , 'sendPid
90 , 'sdictInt
91 , 'wait
92 , 'expectUnit
93 , 'typedPingServer
94 , 'isPrime
95 , 'quintuple
96 , 'signal
97 ]
99randomElement :: [a] -> IO a
100randomElement xs = do
101 ix <- randomIO
102 return (xs !! (ix `mod` length xs))
104remotableDecl [
105 [d| dfib :: ([NodeId], SendPort Integer, Integer) -> Process () ;
106 dfib (_, reply, 0) = sendChan reply 0
107 dfib (_, reply, 1) = sendChan reply 1
108 dfib (nids, reply, n) = do
109 nid1 <- liftIO $ randomElement nids
110 nid2 <- liftIO $ randomElement nids
111 (sport, rport) <- newChan
112 spawn nid1 $ $(mkClosure 'dfib) (nids, sport, n - 2)
113 spawn nid2 $ $(mkClosure 'dfib) (nids, sport, n - 1)
114 n1 <- receiveChan rport
115 n2 <- receiveChan rport
116 sendChan reply $ n1 + n2
117 |]
118 ]
120-- Just try creating a static polymorphic value
121staticQuintuple :: (Typeable a, Typeable b, Typeable c, Typeable d, Typeable e)
122 => Static (a -> b -> c -> d -> e -> (a, b, c, d, e))
123staticQuintuple = $(mkStatic 'quintuple)
125factorialClosure :: Int -> Closure (Process Int)
126factorialClosure = $(mkClosure 'factorial)
128addIntClosure :: Int -> Closure (Int -> Int)
129addIntClosure = $(mkClosure 'addInt)
131putIntClosure :: Int -> Closure (MVar Int -> IO ())
132putIntClosure = $(mkClosure 'putInt)
134sendPidClosure :: ProcessId -> Closure (Process ())
135sendPidClosure = $(mkClosure 'sendPid)
137sendFac :: Int -> ProcessId -> Closure (Process ())
138sendFac n pid = factorialClosure n `bindCP` cpSend $(mkStatic 'sdictInt) pid
140factorialOf :: Closure (Int -> Process Int)
141factorialOf = staticClosure $(mkStatic 'factorial)
143factorial' :: Int -> Closure (Process Int)
144factorial' n = returnCP $(mkStatic 'sdictInt) n `bindCP` factorialOf
146waitClosure :: Int -> Closure (Process ())
147waitClosure = $(mkClosure 'wait)
149simulateNetworkFailure :: TestTransport -> LocalNode -> LocalNode -> Process ()
150simulateNetworkFailure TestTransport{..} from to = liftIO $ do
151 m <- newEmptyMVar
152 _ <- forkProcess to $ getSelfPid >>= liftIO . putMVar m
153 runProcess from $ do
154 them <- liftIO $ takeMVar m
155 pinger <- spawnLocal $ forever $ send them ()
156 _ <- monitorNode (localNodeId to)
157 liftIO $ testBreakConnection (nodeAddress $ localNodeId from)
158 (nodeAddress $ localNodeId to)
159 NodeMonitorNotification _ _ _ <- expect
160 kill pinger "finished"
161 return ()
163--------------------------------------------------------------------------------
164-- The tests proper --
165--------------------------------------------------------------------------------
167testUnclosure :: TestTransport -> RemoteTable -> Assertion
168testUnclosure TestTransport{..} rtable = do
169 node <- newLocalNode testTransport rtable
170 done <- newEmptyMVar
171 forkProcess node $ do
172 120 <- join . unClosure $ factorialClosure 5
173 liftIO $ putMVar done ()
174 takeMVar done
176testBind :: TestTransport -> RemoteTable -> Assertion
177testBind TestTransport{..} rtable = do
178 node <- newLocalNode testTransport rtable
179 done <- newEmptyMVar
180 runProcess node $ do
181 us <- getSelfPid
182 join . unClosure $ sendFac 6 us
183 (720 :: Int) <- expect
184 liftIO $ putMVar done ()
185 takeMVar done
187testSendPureClosure :: TestTransport -> RemoteTable -> Assertion
188testSendPureClosure TestTransport{..} rtable = do
189 serverAddr <- newEmptyMVar
190 serverDone <- newEmptyMVar
192 forkIO $ do
193 node <- newLocalNode testTransport rtable
194 addr <- forkProcess node $ do
195 cl <- expect
196 fn <- unClosure cl :: Process (Int -> Int)
197 13 <- return $ fn 6
198 liftIO $ putMVar serverDone ()
199 putMVar serverAddr addr
201 forkIO $ do
202 node <- newLocalNode testTransport rtable
203 theirAddr <- readMVar serverAddr
204 runProcess node $ send theirAddr (addIntClosure 7)
206 takeMVar serverDone
208testSendIOClosure :: TestTransport -> RemoteTable -> Assertion
209testSendIOClosure TestTransport{..} rtable = do
210 serverAddr <- newEmptyMVar
211 serverDone <- newEmptyMVar
213 forkIO $ do
214 node <- newLocalNode testTransport rtable
215 addr <- forkProcess node $ do
216 cl <- expect
217 io <- unClosure cl :: Process (MVar Int -> IO ())
218 liftIO $ do
219 someMVar <- newEmptyMVar
220 io someMVar
221 5 <- readMVar someMVar
222 putMVar serverDone ()
223 putMVar serverAddr addr
225 forkIO $ do
226 node <- newLocalNode testTransport rtable
227 theirAddr <- readMVar serverAddr
228 runProcess node $ send theirAddr (putIntClosure 5)
230 takeMVar serverDone
232testSendProcClosure :: TestTransport -> RemoteTable -> Assertion
233testSendProcClosure TestTransport{..} rtable = do
234 serverAddr <- newEmptyMVar
235 clientDone <- newEmptyMVar
237 forkIO $ do
238 node <- newLocalNode testTransport rtable
239 addr <- forkProcess node $ do
240 cl <- expect
241 pr <- unClosure cl :: Process (Int -> Process ())
242 pr 5
243 putMVar serverAddr addr
245 forkIO $ do
246 node <- newLocalNode testTransport rtable
247 theirAddr <- readMVar serverAddr
248 runProcess node $ do
249 pid <- getSelfPid
250 send theirAddr (cpSend $(mkStatic 'sdictInt) pid)
251 5 <- expect :: Process Int
252 liftIO $ putMVar clientDone ()
254 takeMVar clientDone
256testSpawn :: TestTransport -> RemoteTable -> Assertion
257testSpawn TestTransport{..} rtable = do
258 serverNodeAddr <- newEmptyMVar
259 clientDone <- newEmptyMVar
261 forkIO $ do
262 node <- newLocalNode testTransport rtable
263 putMVar serverNodeAddr (localNodeId node)
265 forkIO $ do
266 node <- newLocalNode testTransport rtable
267 nid <- readMVar serverNodeAddr
268 runProcess node $ do
269 pid <- getSelfPid
270 pid' <- spawn nid (sendPidClosure pid)
271 pid'' <- expect
272 True <- return $ pid' == pid''
273 liftIO $ putMVar clientDone ()
275 takeMVar clientDone
277-- | Tests that spawn executes the supplied closure even if the caller dies
278-- immediately after calling spawn.
279--
280-- This situation is of interest because the implementation of spawn has the
281-- remote peer monitor the caller. See DP-99.
282--
283-- The condition is tested by using a transport which refuses to send to the
284-- remote peer the message that it is waiting to stop monitoring the caller,
285-- namely @()@.
286--
287testSpawnRace :: TestTransport -> RemoteTable -> Assertion
288testSpawnRace TestTransport{..} rtable = do
289 node1 <- newLocalNode (wrapTransport testTransport) rtable
290 node2 <- newLocalNode testTransport rtable
292 runProcess node1 $ do
293 pid <- getSelfPid
294 spawnLocal $ spawn (localNodeId node2) (sendPidClosure pid) >>= send pid
295 pid' <- expect :: Process ProcessId
296 pid'' <- expect :: Process ProcessId
297 True <- return $ pid' == pid''
298 return ()
300 where
302 wrapTransport (NT.Transport ne ct) = NT.Transport (fmap (fmap wrapEP) ne) ct
304 wrapEP :: NT.EndPoint -> NT.EndPoint
305 wrapEP e =
306 e { NT.connect = \x y z -> do
307 healthy <- newIORef True
308 fmap (fmap $ wrapConnection healthy) $ NT.connect e x y z
309 }
311 wrapConnection :: IORef Bool -> NT.Connection -> NT.Connection
312 wrapConnection healthy (NT.Connection s closeC) =
313 flip NT.Connection closeC $ \msg -> do
314 when (msg == messageToPayload (createMessage ())) $ do
315 writeIORef healthy False
316 isHealthy <- readIORef healthy
317 if isHealthy then s msg
318 else return $ Left $ NT.TransportError NT.SendFailed ""
320testCall :: TestTransport -> RemoteTable -> Assertion
321testCall TestTransport{..} rtable = do
322 serverNodeAddr <- newEmptyMVar
323 clientDone <- newEmptyMVar
325 forkIO $ do
326 node <- newLocalNode testTransport rtable
327 putMVar serverNodeAddr (localNodeId node)
329 forkIO $ do
330 node <- newLocalNode testTransport rtable
331 nid <- readMVar serverNodeAddr
332 runProcess node $ do
333 (120 :: Int) <- call $(mkStatic 'sdictInt) nid (factorialClosure 5)
334 liftIO $ putMVar clientDone ()
336 takeMVar clientDone
338testCallBind :: TestTransport -> RemoteTable -> Assertion
339testCallBind TestTransport{..} rtable = do
340 serverNodeAddr <- newEmptyMVar
341 clientDone <- newEmptyMVar
343 forkIO $ do
344 node <- newLocalNode testTransport rtable
345 putMVar serverNodeAddr (localNodeId node)
347 forkIO $ do
348 node <- newLocalNode testTransport rtable
349 nid <- readMVar serverNodeAddr
350 runProcess node $ do
351 (120 :: Int) <- call $(mkStatic 'sdictInt) nid (factorial' 5)
352 liftIO $ putMVar clientDone ()
354 takeMVar clientDone
356testSeq :: TestTransport -> RemoteTable -> Assertion
357testSeq TestTransport{..} rtable = do
358 node <- newLocalNode testTransport rtable
359 done <- newEmptyMVar
360 runProcess node $ do
361 us <- getSelfPid
362 join . unClosure $ sendFac 5 us `seqCP` sendFac 6 us
363 120 :: Int <- expect
364 720 :: Int <- expect
365 liftIO $ putMVar done ()
366 takeMVar done
368-- Test 'spawnSupervised'
369--
370-- Set up a supervisor, spawn a child, then have a third process monitor the
371-- child. The supervisor then throws an exception, the child dies because it
372-- was linked to the supervisor, and the third process notices that the child
373-- dies.
374testSpawnSupervised :: TestTransport -> RemoteTable -> Assertion
375testSpawnSupervised TestTransport{..} rtable = do
376 [node1, node2] <- replicateM 2 $ newLocalNode testTransport rtable
377 [superPid, childPid] <- replicateM 2 $ newEmptyMVar
378 thirdProcessDone <- newEmptyMVar
379 linkUp <- newEmptyMVar
381 forkProcess node1 $ do
382 us <- getSelfPid
383 liftIO $ putMVar superPid us
384 (child, _ref) <- spawnSupervised (localNodeId node2)
385 (sendPidClosure us `seqCP` $(mkStaticClosure 'expectUnit))
386 _ <- expect :: Process ProcessId
388 liftIO $ do putMVar childPid child
389 -- Give the child a chance to link to us
390 takeMVar linkUp
391 throw supervisorDeath
393 forkProcess node2 $ do
394 [super, child] <- liftIO $ mapM readMVar [superPid, childPid]
395 ref <- monitor child
396 self <- getSelfPid
397 let waitForMOrL = do
398 liftIO $ threadDelay 10000
399 mpinfo <- getProcessInfo child
400 case mpinfo of
401 Nothing -> waitForMOrL
402 Just pinfo ->
403 unless (isJust $ lookup self (infoMonitors pinfo)) waitForMOrL
404 waitForMOrL
405 liftIO $ putMVar linkUp ()
406 -- because monitor message was sent before message to process
407 -- we hope that it will be processed before
408 ProcessMonitorNotification ref' pid' (DiedException e) <- expect
409 True <- return $ ref' == ref
410 && pid' == child
411 && e == show (ProcessLinkException super (DiedException (show supervisorDeath)))
412 liftIO $ putMVar thirdProcessDone ()
414 takeMVar thirdProcessDone
415 where
416 supervisorDeath :: IOException
417 supervisorDeath = userError "Supervisor died"
419testSpawnInvalid :: TestTransport -> RemoteTable -> Assertion
420testSpawnInvalid TestTransport{..} rtable = do
421 node <- newLocalNode testTransport rtable
422 done <- newEmptyMVar
423 forkProcess node $ do
424 (pid, ref) <- spawnMonitor (localNodeId node) (closure (staticLabel "ThisDoesNotExist") empty)
425 ProcessMonitorNotification ref' pid' _reason <- expect
426 -- Depending on the exact interleaving, reason might be NoProc or the exception thrown by the absence of the static closure
427 True <- return $ ref' == ref && pid == pid'
428 liftIO $ putMVar done ()
429 takeMVar done
431testClosureExpect :: TestTransport -> RemoteTable -> Assertion
432testClosureExpect TestTransport{..} rtable = do
433 node <- newLocalNode testTransport rtable
434 done <- newEmptyMVar
435 runProcess node $ do
436 nodeId <- getSelfNode
437 us <- getSelfPid
438 them <- spawn nodeId $ cpExpect $(mkStatic 'sdictInt) `bindCP` cpSend $(mkStatic 'sdictInt) us
439 send them (1234 :: Int)
440 (1234 :: Int) <- expect
441 liftIO $ putMVar done ()
442 takeMVar done
444testSpawnChannel :: TestTransport -> RemoteTable -> Assertion
445testSpawnChannel TestTransport{..} rtable = do
446 done <- newEmptyMVar
447 [node1, node2] <- replicateM 2 $ newLocalNode testTransport rtable
449 forkProcess node1 $ do
450 pingServer <- spawnChannel
451 (sdictSendPort sdictUnit)
452 (localNodeId node2)
453 ($(mkClosure 'typedPingServer) ())
454 (sendReply, receiveReply) <- newChan
455 sendChan pingServer sendReply
456 receiveChan receiveReply
457 liftIO $ putMVar done ()
459 takeMVar done
461testTDict :: TestTransport -> RemoteTable -> Assertion
462testTDict TestTransport{..} rtable = do
463 done <- newEmptyMVar
464 [node1, node2] <- replicateM 2 $ newLocalNode testTransport rtable
465 forkProcess node1 $ do
466 True <- call $(functionTDict 'isPrime) (localNodeId node2) ($(mkClosure 'isPrime) (79 :: Integer))
467 liftIO $ putMVar done ()
468 takeMVar done
470testFib :: TestTransport -> RemoteTable -> Assertion
471testFib TestTransport{..} rtable = do
472 nodes <- replicateM 4 $ newLocalNode testTransport rtable
473 done <- newEmptyMVar
475 forkProcess (head nodes) $ do
476 (sport, rport) <- newChan
477 spawnLocal $ dfib (map localNodeId nodes, sport, 10)
478 55 <- receiveChan rport :: Process Integer
479 liftIO $ putMVar done ()
481 takeMVar done
483testSpawnReconnect :: TestTransport -> RemoteTable -> Assertion
484testSpawnReconnect testtrans@TestTransport{..} rtable = do
485 [node1, node2] <- replicateM 2 $ newLocalNode testTransport rtable
486 let nid1 = localNodeId node1
487 nid2 = localNodeId node2
488 done <- newEmptyMVar
489 iv <- newMVar (0 :: Int)
491 incr <- forkProcess node1 $ forever $ do
492 () <- expect
493 liftIO $ modifyMVar_ iv (return . (+ 1))
495 forkProcess node2 $ do
496 _pid1 <- spawn nid1 ($(mkClosure 'signal) incr)
497 simulateNetworkFailure testtrans node2 node1
498 _pid2 <- spawn nid1 ($(mkClosure 'signal) incr)
499 _pid3 <- spawn nid1 ($(mkClosure 'signal) incr)
501 liftIO $ threadDelay 100000
503 count <- liftIO $ takeMVar iv
504 True <- return $ count == 2 || count == 3 -- It depends on which message we get first in 'spawn'
506 liftIO $ putMVar done ()
508 takeMVar done
510-- | 'spawn' used to ave a race condition which would be triggered if the
511-- spawning process terminates immediately after spawning
512testSpawnTerminate :: TestTransport -> RemoteTable -> Assertion
513testSpawnTerminate TestTransport{..} rtable = do
514 slave <- newLocalNode testTransport rtable
515 master <- newLocalNode testTransport rtable
516 masterDone <- newEmptyMVar
518 runProcess master $ do
519 us <- getSelfPid
520 replicateM_ 1000 . spawnLocal . void . spawn (localNodeId slave) $ $(mkClosure 'signal) us
521 replicateM_ 1000 $ (expect :: Process ())
522 liftIO $ putMVar masterDone ()
524 takeMVar masterDone
526tests :: TestTransport -> IO [Test]
527tests testtrans = do
528 let rtable = __remoteTable . __remoteTableDecl $ initRemoteTable
529 return
530 [ testCase "Unclosure" (testUnclosure testtrans rtable)
531 , testCase "Bind" (testBind testtrans rtable)
532 , testCase "SendPureClosure" (testSendPureClosure testtrans rtable)
533 , testCase "SendIOClosure" (testSendIOClosure testtrans rtable)
534 , testCase "SendProcClosure" (testSendProcClosure testtrans rtable)
535 , testCase "Spawn" (testSpawn testtrans rtable)
536 , testCase "SpawnRace" (testSpawnRace testtrans rtable)
537 , testCase "Call" (testCall testtrans rtable)
538 , testCase "CallBind" (testCallBind testtrans rtable)
539 , testCase "Seq" (testSeq testtrans rtable)
540 , testCase "SpawnSupervised" (testSpawnSupervised testtrans rtable)
541 , testCase "SpawnInvalid" (testSpawnInvalid testtrans rtable)
542 , testCase "ClosureExpect" (testClosureExpect testtrans rtable)
543 , testCase "SpawnChannel" (testSpawnChannel testtrans rtable)
544 , testCase "TDict" (testTDict testtrans rtable)
545 , testCase "Fib" (testFib testtrans rtable)
546 , testCase "SpawnTerminate" (testSpawnTerminate testtrans rtable)
547 , testCase "SpawnReconnect" (testSpawnReconnect testtrans rtable)
548 ]