git.iko.soy/distributed-process
on 7930062 help sign in
distributed-process/distributed-process-tests/src/Control/Distributed/Process/Tests/Closure.hs
589 lines 20 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 i <- join . unClosure $ factorialClosure 5
173 liftIO $ putMVar done ()
174 if i == 720
175 then return ()
176 else error "Something went horribly wrong"
177 takeMVar done
179testBind :: TestTransport -> RemoteTable -> Assertion
180testBind TestTransport{..} rtable = do
181 node <- newLocalNode testTransport rtable
182 done <- newEmptyMVar
183 runProcess node $ do
184 us <- getSelfPid
185 join . unClosure $ sendFac 6 us
186 (i :: Int) <- expect
187 liftIO $ putMVar done ()
188 if i == 720
189 then return ()
190 else error "Something went horribly wrong"
191 takeMVar done
193testSendPureClosure :: TestTransport -> RemoteTable -> Assertion
194testSendPureClosure TestTransport{..} rtable = do
195 serverAddr <- newEmptyMVar
196 serverDone <- newEmptyMVar
198 forkIO $ do
199 node <- newLocalNode testTransport rtable
200 addr <- forkProcess node $ do
201 cl <- expect
202 fn <- unClosure cl :: Process (Int -> Int)
203 (_ :: Int) <- return $ fn 6
204 liftIO $ putMVar serverDone ()
205 putMVar serverAddr addr
207 forkIO $ do
208 node <- newLocalNode testTransport rtable
209 theirAddr <- readMVar serverAddr
210 runProcess node $ send theirAddr (addIntClosure 7)
212 takeMVar serverDone
214testSendIOClosure :: TestTransport -> RemoteTable -> Assertion
215testSendIOClosure TestTransport{..} rtable = do
216 serverAddr <- newEmptyMVar
217 serverDone <- newEmptyMVar
219 forkIO $ do
220 node <- newLocalNode testTransport rtable
221 addr <- forkProcess node $ do
222 cl <- expect
223 io <- unClosure cl :: Process (MVar Int -> IO ())
224 liftIO $ do
225 someMVar <- newEmptyMVar
226 io someMVar
227 i <- readMVar someMVar
228 putMVar serverDone ()
229 if i == 5
230 then return ()
231 else error "Something went horribly wrong"
232 putMVar serverAddr addr
234 forkIO $ do
235 node <- newLocalNode testTransport rtable
236 theirAddr <- readMVar serverAddr
237 runProcess node $ send theirAddr (putIntClosure 5)
239 takeMVar serverDone
241testSendProcClosure :: TestTransport -> RemoteTable -> Assertion
242testSendProcClosure TestTransport{..} rtable = do
243 serverAddr <- newEmptyMVar
244 clientDone <- newEmptyMVar
246 forkIO $ do
247 node <- newLocalNode testTransport rtable
248 addr <- forkProcess node $ do
249 cl <- expect
250 pr <- unClosure cl :: Process (Int -> Process ())
251 pr 5
252 putMVar serverAddr addr
254 forkIO $ do
255 node <- newLocalNode testTransport rtable
256 theirAddr <- readMVar serverAddr
257 runProcess node $ do
258 pid <- getSelfPid
259 send theirAddr (cpSend $(mkStatic 'sdictInt) pid)
260 i <- expect :: Process Int
261 if i == 5
262 then liftIO $ putMVar clientDone ()
263 else error "Something went horribly wrong"
265 takeMVar clientDone
267testSpawn :: TestTransport -> RemoteTable -> Assertion
268testSpawn TestTransport{..} rtable = do
269 serverNodeAddr <- newEmptyMVar
270 clientDone <- newEmptyMVar
272 forkIO $ do
273 node <- newLocalNode testTransport rtable
274 putMVar serverNodeAddr (localNodeId node)
276 forkIO $ do
277 node <- newLocalNode testTransport rtable
278 nid <- readMVar serverNodeAddr
279 runProcess node $ do
280 pid <- getSelfPid
281 pid' <- spawn nid (sendPidClosure pid)
282 pid'' <- expect
283 if pid' == pid''
284 then liftIO $ putMVar clientDone ()
285 else error "Something went horribly wrong"
287 takeMVar clientDone
289-- | Tests that spawn executes the supplied closure even if the caller dies
290-- immediately after calling spawn.
291--
292-- This situation is of interest because the implementation of spawn has the
293-- remote peer monitor the caller. See DP-99.
294--
295-- The condition is tested by using a transport which refuses to send to the
296-- remote peer the message that it is waiting to stop monitoring the caller,
297-- namely @()@.
298--
299testSpawnRace :: TestTransport -> RemoteTable -> Assertion
300testSpawnRace TestTransport{..} rtable = do
301 node1 <- newLocalNode (wrapTransport testTransport) rtable
302 node2 <- newLocalNode testTransport rtable
304 runProcess node1 $ do
305 pid <- getSelfPid
306 spawnLocal $ spawn (localNodeId node2) (sendPidClosure pid) >>= send pid
307 pid' <- expect :: Process ProcessId
308 pid'' <- expect :: Process ProcessId
309 if pid' == pid''
310 then return ()
311 else error "Something went horribly wrong"
313 where
315 wrapTransport (NT.Transport ne ct) = NT.Transport (fmap (fmap wrapEP) ne) ct
317 wrapEP :: NT.EndPoint -> NT.EndPoint
318 wrapEP e =
319 e { NT.connect = \x y z -> do
320 healthy <- newIORef True
321 fmap (fmap $ wrapConnection healthy e x) $ NT.connect e x y z
322 }
324 wrapConnection :: IORef Bool -> NT.EndPoint -> NT.EndPointAddress
325 -> NT.Connection -> NT.Connection
326 wrapConnection healthy e remoteAddr (NT.Connection s closeC) =
327 flip NT.Connection closeC $ \msg -> do
328 when (msg == messageToPayload (createMessage ())) $ do
329 writeIORef healthy False
330 testBreakConnection (NT.address e) remoteAddr
331 isHealthy <- readIORef healthy
332 if isHealthy then s msg
333 else return $ Left $ NT.TransportError NT.SendFailed ""
335testCall :: TestTransport -> RemoteTable -> Assertion
336testCall TestTransport{..} rtable = do
337 serverNodeAddr <- newEmptyMVar
338 clientDone <- newEmptyMVar
340 forkIO $ do
341 node <- newLocalNode testTransport rtable
342 putMVar serverNodeAddr (localNodeId node)
344 forkIO $ do
345 node <- newLocalNode testTransport rtable
346 nid <- readMVar serverNodeAddr
347 runProcess node $ do
348 (a :: Int) <- call $(mkStatic 'sdictInt) nid (factorialClosure 5)
349 if a == 120
350 then liftIO $ putMVar clientDone ()
351 else error "something went horribly wrong"
353 takeMVar clientDone
355testCallBind :: TestTransport -> RemoteTable -> Assertion
356testCallBind TestTransport{..} rtable = do
357 serverNodeAddr <- newEmptyMVar
358 clientDone <- newEmptyMVar
360 forkIO $ do
361 node <- newLocalNode testTransport rtable
362 putMVar serverNodeAddr (localNodeId node)
364 forkIO $ do
365 node <- newLocalNode testTransport rtable
366 nid <- readMVar serverNodeAddr
367 runProcess node $ do
368 (a :: Int) <- call $(mkStatic 'sdictInt) nid (factorial' 5)
369 if a == 120
370 then liftIO $ putMVar clientDone ()
371 else error "Something went horribly wrong"
373 takeMVar clientDone
375testSeq :: TestTransport -> RemoteTable -> Assertion
376testSeq TestTransport{..} rtable = do
377 node <- newLocalNode testTransport rtable
378 done <- newEmptyMVar
379 runProcess node $ do
380 us <- getSelfPid
381 join . unClosure $ sendFac 5 us `seqCP` sendFac 6 us
382 a :: Int <- expect
383 b :: Int <- expect
384 if a == 120 && b == 720
385 then liftIO $ putMVar done ()
386 else error "Something went horribly wrong"
387 takeMVar done
389-- Test 'spawnSupervised'
390--
391-- Set up a supervisor, spawn a child, then have a third process monitor the
392-- child. The supervisor then throws an exception, the child dies because it
393-- was linked to the supervisor, and the third process notices that the child
394-- dies.
395testSpawnSupervised :: TestTransport -> RemoteTable -> Assertion
396testSpawnSupervised TestTransport{..} rtable = do
397 [node1, node2] <- replicateM 2 $ newLocalNode testTransport rtable
398 [superPid, childPid] <- replicateM 2 $ newEmptyMVar
399 thirdProcessDone <- newEmptyMVar
400 linkUp <- newEmptyMVar
402 forkProcess node1 $ do
403 us <- getSelfPid
404 liftIO $ putMVar superPid us
405 (child, _ref) <- spawnSupervised (localNodeId node2)
406 (sendPidClosure us `seqCP` $(mkStaticClosure 'expectUnit))
407 _ <- expect :: Process ProcessId
409 liftIO $ do putMVar childPid child
410 -- Give the child a chance to link to us
411 takeMVar linkUp
412 throw supervisorDeath
414 forkProcess node2 $ do
415 res <- liftIO $ mapM readMVar [superPid, childPid]
416 case res of
417 [super, child] -> do
418 ref <- monitor child
419 self <- getSelfPid
420 let waitForMOrL = do
421 liftIO $ threadDelay 10000
422 mpinfo <- getProcessInfo child
423 case mpinfo of
424 Nothing -> waitForMOrL
425 Just pinfo ->
426 unless (isJust $ lookup self (infoMonitors pinfo)) waitForMOrL
427 waitForMOrL
428 liftIO $ putMVar linkUp ()
429 -- because monitor message was sent before message to process
430 -- we hope that it will be processed before
431 res' <- expect
432 case res' of
433 (ProcessMonitorNotification ref' pid' (DiedException e)) ->
434 if (ref' == ref && pid' == child &&
435 e == show (ProcessLinkException super
436 (DiedException (show supervisorDeath))))
437 then liftIO $ putMVar thirdProcessDone ()
438 else error "Something went horribly wrong"
439 _ -> error "Something went horribly wrong"
441 _ -> die $ "Something went horribly wrong"
443 takeMVar thirdProcessDone
444 where
445 supervisorDeath :: IOException
446 supervisorDeath = userError "Supervisor died"
448testSpawnInvalid :: TestTransport -> RemoteTable -> Assertion
449testSpawnInvalid TestTransport{..} rtable = do
450 node <- newLocalNode testTransport rtable
451 done <- newEmptyMVar
452 forkProcess node $ do
453 (pid, ref) <- spawnMonitor (localNodeId node) (closure (staticLabel "ThisDoesNotExist") empty)
454 ProcessMonitorNotification ref' pid' _reason <- expect
455 -- Depending on the exact interleaving, reason might be NoProc or the exception thrown by the absence of the static closure
456 res <- return $ ref' == ref && pid == pid'
457 if res == True
458 then liftIO $ putMVar done ()
459 else error "Something went horribly wrong"
460 takeMVar done
462testClosureExpect :: TestTransport -> RemoteTable -> Assertion
463testClosureExpect TestTransport{..} rtable = do
464 node <- newLocalNode testTransport rtable
465 done <- newEmptyMVar
466 runProcess node $ do
467 nodeId <- getSelfNode
468 us <- getSelfPid
469 them <- spawn nodeId $ cpExpect $(mkStatic 'sdictInt) `bindCP` cpSend $(mkStatic 'sdictInt) us
470 send them (1234 :: Int)
471 (res :: Int) <- expect
472 if res == 1234
473 then liftIO $ putMVar done ()
474 else error "Something went horribly wrong"
475 takeMVar done
477testSpawnChannel :: TestTransport -> RemoteTable -> Assertion
478testSpawnChannel TestTransport{..} rtable = do
479 done <- newEmptyMVar
480 [node1, node2] <- replicateM 2 $ newLocalNode testTransport rtable
482 forkProcess node1 $ do
483 pingServer <- spawnChannel
484 (sdictSendPort sdictUnit)
485 (localNodeId node2)
486 ($(mkClosure 'typedPingServer) ())
487 (sendReply, receiveReply) <- newChan
488 sendChan pingServer sendReply
489 receiveChan receiveReply
490 liftIO $ putMVar done ()
492 takeMVar done
494testTDict :: TestTransport -> RemoteTable -> Assertion
495testTDict TestTransport{..} rtable = do
496 done <- newEmptyMVar
497 [node1, node2] <- replicateM 2 $ newLocalNode testTransport rtable
498 forkProcess node1 $ do
499 res <- call $(functionTDict 'isPrime) (localNodeId node2) ($(mkClosure 'isPrime) (79 :: Integer))
500 if res == True
501 then liftIO $ putMVar done ()
502 else error "Something went horribly wrong..."
503 takeMVar done
505testFib :: TestTransport -> RemoteTable -> Assertion
506testFib TestTransport{..} rtable = do
507 nodes <- replicateM 4 $ newLocalNode testTransport rtable
508 done <- newEmptyMVar
510 forkProcess (head nodes) $ do
511 (sport, rport) <- newChan
512 spawnLocal $ dfib (map localNodeId nodes, sport, 10)
513 ff <- receiveChan rport :: Process Integer
514 liftIO $ putMVar done ()
515 if ff /= 55
516 then die $ "Something went horribly wrong"
517 else return ()
519 takeMVar done
521testSpawnReconnect :: TestTransport -> RemoteTable -> Assertion
522testSpawnReconnect testtrans@TestTransport{..} rtable = do
523 [node1, node2] <- replicateM 2 $ newLocalNode testTransport rtable
524 let nid1 = localNodeId node1
525 -- nid2 = localNodeId node2
526 done <- newEmptyMVar
527 iv <- newMVar (0 :: Int)
529 incr <- forkProcess node1 $ forever $ do
530 () <- expect
531 liftIO $ modifyMVar_ iv (return . (+ 1))
533 forkProcess node2 $ do
534 _pid1 <- spawn nid1 ($(mkClosure 'signal) incr)
535 simulateNetworkFailure testtrans node2 node1
536 _pid2 <- spawn nid1 ($(mkClosure 'signal) incr)
537 _pid3 <- spawn nid1 ($(mkClosure 'signal) incr)
539 liftIO $ threadDelay 100000
541 count <- liftIO $ takeMVar iv
542 res <- return $ count == 2 || count == 3 -- It depends on which message we get first in 'spawn'
544 liftIO $ putMVar done ()
545 if res /= True
546 then error "Something went horribly wrong"
547 else return ()
549 takeMVar done
551-- | 'spawn' used to ave a race condition which would be triggered if the
552-- spawning process terminates immediately after spawning
553testSpawnTerminate :: TestTransport -> RemoteTable -> Assertion
554testSpawnTerminate TestTransport{..} rtable = do
555 slave <- newLocalNode testTransport rtable
556 master <- newLocalNode testTransport rtable
557 masterDone <- newEmptyMVar
559 runProcess master $ do
560 us <- getSelfPid
561 replicateM_ 1000 . spawnLocal . void . spawn (localNodeId slave) $ $(mkClosure 'signal) us
562 replicateM_ 1000 $ (expect :: Process ())
563 liftIO $ putMVar masterDone ()
565 takeMVar masterDone
567tests :: TestTransport -> IO [Test]
568tests testtrans = do
569 let rtable = __remoteTable . __remoteTableDecl $ initRemoteTable
570 return
571 [ testCase "Unclosure" (testUnclosure testtrans rtable)
572 , testCase "Bind" (testBind testtrans rtable)
573 , testCase "SendPureClosure" (testSendPureClosure testtrans rtable)
574 , testCase "SendIOClosure" (testSendIOClosure testtrans rtable)
575 , testCase "SendProcClosure" (testSendProcClosure testtrans rtable)
576 , testCase "Spawn" (testSpawn testtrans rtable)
577 , testCase "SpawnRace" (testSpawnRace testtrans rtable)
578 , testCase "Call" (testCall testtrans rtable)
579 , testCase "CallBind" (testCallBind testtrans rtable)
580 , testCase "Seq" (testSeq testtrans rtable)
581 , testCase "SpawnSupervised" (testSpawnSupervised testtrans rtable)
582 , testCase "SpawnInvalid" (testSpawnInvalid testtrans rtable)
583 , testCase "ClosureExpect" (testClosureExpect testtrans rtable)
584 , testCase "SpawnChannel" (testSpawnChannel testtrans rtable)
585 , testCase "TDict" (testTDict testtrans rtable)
586 , testCase "Fib" (testFib testtrans rtable)
587 , testCase "SpawnTerminate" (testSpawnTerminate testtrans rtable)
588 , testCase "SpawnReconnect" (testSpawnReconnect testtrans rtable)
589 ]