git.iko.soy/distributed-process
on 1d099cc help sign in
distributed-process/distributed-process-tests/src/Control/Distributed/Process/Tests/Closure.hs
550 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 e x) $ NT.connect e x y z
309 }
311 wrapConnection :: IORef Bool -> NT.EndPoint -> NT.EndPointAddress
312 -> NT.Connection -> NT.Connection
313 wrapConnection healthy e remoteAddr (NT.Connection s closeC) =
314 flip NT.Connection closeC $ \msg -> do
315 when (msg == messageToPayload (createMessage ())) $ do
316 writeIORef healthy False
317 testBreakConnection (NT.address e) remoteAddr
318 isHealthy <- readIORef healthy
319 if isHealthy then s msg
320 else return $ Left $ NT.TransportError NT.SendFailed ""
322testCall :: TestTransport -> RemoteTable -> Assertion
323testCall TestTransport{..} rtable = do
324 serverNodeAddr <- newEmptyMVar
325 clientDone <- newEmptyMVar
327 forkIO $ do
328 node <- newLocalNode testTransport rtable
329 putMVar serverNodeAddr (localNodeId node)
331 forkIO $ do
332 node <- newLocalNode testTransport rtable
333 nid <- readMVar serverNodeAddr
334 runProcess node $ do
335 (120 :: Int) <- call $(mkStatic 'sdictInt) nid (factorialClosure 5)
336 liftIO $ putMVar clientDone ()
338 takeMVar clientDone
340testCallBind :: TestTransport -> RemoteTable -> Assertion
341testCallBind TestTransport{..} rtable = do
342 serverNodeAddr <- newEmptyMVar
343 clientDone <- newEmptyMVar
345 forkIO $ do
346 node <- newLocalNode testTransport rtable
347 putMVar serverNodeAddr (localNodeId node)
349 forkIO $ do
350 node <- newLocalNode testTransport rtable
351 nid <- readMVar serverNodeAddr
352 runProcess node $ do
353 (120 :: Int) <- call $(mkStatic 'sdictInt) nid (factorial' 5)
354 liftIO $ putMVar clientDone ()
356 takeMVar clientDone
358testSeq :: TestTransport -> RemoteTable -> Assertion
359testSeq TestTransport{..} rtable = do
360 node <- newLocalNode testTransport rtable
361 done <- newEmptyMVar
362 runProcess node $ do
363 us <- getSelfPid
364 join . unClosure $ sendFac 5 us `seqCP` sendFac 6 us
365 120 :: Int <- expect
366 720 :: Int <- expect
367 liftIO $ putMVar done ()
368 takeMVar done
370-- Test 'spawnSupervised'
371--
372-- Set up a supervisor, spawn a child, then have a third process monitor the
373-- child. The supervisor then throws an exception, the child dies because it
374-- was linked to the supervisor, and the third process notices that the child
375-- dies.
376testSpawnSupervised :: TestTransport -> RemoteTable -> Assertion
377testSpawnSupervised TestTransport{..} rtable = do
378 [node1, node2] <- replicateM 2 $ newLocalNode testTransport rtable
379 [superPid, childPid] <- replicateM 2 $ newEmptyMVar
380 thirdProcessDone <- newEmptyMVar
381 linkUp <- newEmptyMVar
383 forkProcess node1 $ do
384 us <- getSelfPid
385 liftIO $ putMVar superPid us
386 (child, _ref) <- spawnSupervised (localNodeId node2)
387 (sendPidClosure us `seqCP` $(mkStaticClosure 'expectUnit))
388 _ <- expect :: Process ProcessId
390 liftIO $ do putMVar childPid child
391 -- Give the child a chance to link to us
392 takeMVar linkUp
393 throw supervisorDeath
395 forkProcess node2 $ do
396 [super, child] <- liftIO $ mapM readMVar [superPid, childPid]
397 ref <- monitor child
398 self <- getSelfPid
399 let waitForMOrL = do
400 liftIO $ threadDelay 10000
401 mpinfo <- getProcessInfo child
402 case mpinfo of
403 Nothing -> waitForMOrL
404 Just pinfo ->
405 unless (isJust $ lookup self (infoMonitors pinfo)) waitForMOrL
406 waitForMOrL
407 liftIO $ putMVar linkUp ()
408 -- because monitor message was sent before message to process
409 -- we hope that it will be processed before
410 ProcessMonitorNotification ref' pid' (DiedException e) <- expect
411 True <- return $ ref' == ref
412 && pid' == child
413 && e == show (ProcessLinkException super (DiedException (show supervisorDeath)))
414 liftIO $ putMVar thirdProcessDone ()
416 takeMVar thirdProcessDone
417 where
418 supervisorDeath :: IOException
419 supervisorDeath = userError "Supervisor died"
421testSpawnInvalid :: TestTransport -> RemoteTable -> Assertion
422testSpawnInvalid TestTransport{..} rtable = do
423 node <- newLocalNode testTransport rtable
424 done <- newEmptyMVar
425 forkProcess node $ do
426 (pid, ref) <- spawnMonitor (localNodeId node) (closure (staticLabel "ThisDoesNotExist") empty)
427 ProcessMonitorNotification ref' pid' _reason <- expect
428 -- Depending on the exact interleaving, reason might be NoProc or the exception thrown by the absence of the static closure
429 True <- return $ ref' == ref && pid == pid'
430 liftIO $ putMVar done ()
431 takeMVar done
433testClosureExpect :: TestTransport -> RemoteTable -> Assertion
434testClosureExpect TestTransport{..} rtable = do
435 node <- newLocalNode testTransport rtable
436 done <- newEmptyMVar
437 runProcess node $ do
438 nodeId <- getSelfNode
439 us <- getSelfPid
440 them <- spawn nodeId $ cpExpect $(mkStatic 'sdictInt) `bindCP` cpSend $(mkStatic 'sdictInt) us
441 send them (1234 :: Int)
442 (1234 :: Int) <- expect
443 liftIO $ putMVar done ()
444 takeMVar done
446testSpawnChannel :: TestTransport -> RemoteTable -> Assertion
447testSpawnChannel TestTransport{..} rtable = do
448 done <- newEmptyMVar
449 [node1, node2] <- replicateM 2 $ newLocalNode testTransport rtable
451 forkProcess node1 $ do
452 pingServer <- spawnChannel
453 (sdictSendPort sdictUnit)
454 (localNodeId node2)
455 ($(mkClosure 'typedPingServer) ())
456 (sendReply, receiveReply) <- newChan
457 sendChan pingServer sendReply
458 receiveChan receiveReply
459 liftIO $ putMVar done ()
461 takeMVar done
463testTDict :: TestTransport -> RemoteTable -> Assertion
464testTDict TestTransport{..} rtable = do
465 done <- newEmptyMVar
466 [node1, node2] <- replicateM 2 $ newLocalNode testTransport rtable
467 forkProcess node1 $ do
468 True <- call $(functionTDict 'isPrime) (localNodeId node2) ($(mkClosure 'isPrime) (79 :: Integer))
469 liftIO $ putMVar done ()
470 takeMVar done
472testFib :: TestTransport -> RemoteTable -> Assertion
473testFib TestTransport{..} rtable = do
474 nodes <- replicateM 4 $ newLocalNode testTransport rtable
475 done <- newEmptyMVar
477 forkProcess (head nodes) $ do
478 (sport, rport) <- newChan
479 spawnLocal $ dfib (map localNodeId nodes, sport, 10)
480 55 <- receiveChan rport :: Process Integer
481 liftIO $ putMVar done ()
483 takeMVar done
485testSpawnReconnect :: TestTransport -> RemoteTable -> Assertion
486testSpawnReconnect testtrans@TestTransport{..} rtable = do
487 [node1, node2] <- replicateM 2 $ newLocalNode testTransport rtable
488 let nid1 = localNodeId node1
489 nid2 = localNodeId node2
490 done <- newEmptyMVar
491 iv <- newMVar (0 :: Int)
493 incr <- forkProcess node1 $ forever $ do
494 () <- expect
495 liftIO $ modifyMVar_ iv (return . (+ 1))
497 forkProcess node2 $ do
498 _pid1 <- spawn nid1 ($(mkClosure 'signal) incr)
499 simulateNetworkFailure testtrans node2 node1
500 _pid2 <- spawn nid1 ($(mkClosure 'signal) incr)
501 _pid3 <- spawn nid1 ($(mkClosure 'signal) incr)
503 liftIO $ threadDelay 100000
505 count <- liftIO $ takeMVar iv
506 True <- return $ count == 2 || count == 3 -- It depends on which message we get first in 'spawn'
508 liftIO $ putMVar done ()
510 takeMVar done
512-- | 'spawn' used to ave a race condition which would be triggered if the
513-- spawning process terminates immediately after spawning
514testSpawnTerminate :: TestTransport -> RemoteTable -> Assertion
515testSpawnTerminate TestTransport{..} rtable = do
516 slave <- newLocalNode testTransport rtable
517 master <- newLocalNode testTransport rtable
518 masterDone <- newEmptyMVar
520 runProcess master $ do
521 us <- getSelfPid
522 replicateM_ 1000 . spawnLocal . void . spawn (localNodeId slave) $ $(mkClosure 'signal) us
523 replicateM_ 1000 $ (expect :: Process ())
524 liftIO $ putMVar masterDone ()
526 takeMVar masterDone
528tests :: TestTransport -> IO [Test]
529tests testtrans = do
530 let rtable = __remoteTable . __remoteTableDecl $ initRemoteTable
531 return
532 [ testCase "Unclosure" (testUnclosure testtrans rtable)
533 , testCase "Bind" (testBind testtrans rtable)
534 , testCase "SendPureClosure" (testSendPureClosure testtrans rtable)
535 , testCase "SendIOClosure" (testSendIOClosure testtrans rtable)
536 , testCase "SendProcClosure" (testSendProcClosure testtrans rtable)
537 , testCase "Spawn" (testSpawn testtrans rtable)
538 , testCase "SpawnRace" (testSpawnRace testtrans rtable)
539 , testCase "Call" (testCall testtrans rtable)
540 , testCase "CallBind" (testCallBind testtrans rtable)
541 , testCase "Seq" (testSeq testtrans rtable)
542 , testCase "SpawnSupervised" (testSpawnSupervised testtrans rtable)
543 , testCase "SpawnInvalid" (testSpawnInvalid testtrans rtable)
544 , testCase "ClosureExpect" (testClosureExpect testtrans rtable)
545 , testCase "SpawnChannel" (testSpawnChannel testtrans rtable)
546 , testCase "TDict" (testTDict testtrans rtable)
547 , testCase "Fib" (testFib testtrans rtable)
548 , testCase "SpawnTerminate" (testSpawnTerminate testtrans rtable)
549 , testCase "SpawnReconnect" (testSpawnReconnect testtrans rtable)
550 ]