git.iko.soy/distributed-process
on 391cb93 help sign in
distributed-process/distributed-process-tests/src/Control/Distributed/Process/Tests/Closure.hs
581 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 [super, child] <- liftIO $ mapM readMVar [superPid, childPid]
416 ref <- monitor child
417 self <- getSelfPid
418 let waitForMOrL = do
419 liftIO $ threadDelay 10000
420 mpinfo <- getProcessInfo child
421 case mpinfo of
422 Nothing -> waitForMOrL
423 Just pinfo ->
424 unless (isJust $ lookup self (infoMonitors pinfo)) waitForMOrL
425 waitForMOrL
426 liftIO $ putMVar linkUp ()
427 -- because monitor message was sent before message to process
428 -- we hope that it will be processed before
429 res <- expect
430 case res of
431 (ProcessMonitorNotification ref' pid' (DiedException e)) ->
432 if (ref' == ref && pid' == child &&
433 e == show (ProcessLinkException super
434 (DiedException (show supervisorDeath))))
435 then liftIO $ putMVar thirdProcessDone ()
436 else error "Something went horribly wrong"
437 _ -> error "Something went horribly wrong"
438 takeMVar thirdProcessDone
439 where
440 supervisorDeath :: IOException
441 supervisorDeath = userError "Supervisor died"
443testSpawnInvalid :: TestTransport -> RemoteTable -> Assertion
444testSpawnInvalid TestTransport{..} rtable = do
445 node <- newLocalNode testTransport rtable
446 done <- newEmptyMVar
447 forkProcess node $ do
448 (pid, ref) <- spawnMonitor (localNodeId node) (closure (staticLabel "ThisDoesNotExist") empty)
449 ProcessMonitorNotification ref' pid' _reason <- expect
450 -- Depending on the exact interleaving, reason might be NoProc or the exception thrown by the absence of the static closure
451 res <- return $ ref' == ref && pid == pid'
452 if res == True
453 then liftIO $ putMVar done ()
454 else error "Something went horribly wrong"
455 takeMVar done
457testClosureExpect :: TestTransport -> RemoteTable -> Assertion
458testClosureExpect TestTransport{..} rtable = do
459 node <- newLocalNode testTransport rtable
460 done <- newEmptyMVar
461 runProcess node $ do
462 nodeId <- getSelfNode
463 us <- getSelfPid
464 them <- spawn nodeId $ cpExpect $(mkStatic 'sdictInt) `bindCP` cpSend $(mkStatic 'sdictInt) us
465 send them (1234 :: Int)
466 (res :: Int) <- expect
467 if res == 1234
468 then liftIO $ putMVar done ()
469 else error "Something went horribly wrong"
470 takeMVar done
472testSpawnChannel :: TestTransport -> RemoteTable -> Assertion
473testSpawnChannel TestTransport{..} rtable = do
474 done <- newEmptyMVar
475 [node1, node2] <- replicateM 2 $ newLocalNode testTransport rtable
477 forkProcess node1 $ do
478 pingServer <- spawnChannel
479 (sdictSendPort sdictUnit)
480 (localNodeId node2)
481 ($(mkClosure 'typedPingServer) ())
482 (sendReply, receiveReply) <- newChan
483 sendChan pingServer sendReply
484 receiveChan receiveReply
485 liftIO $ putMVar done ()
487 takeMVar done
489testTDict :: TestTransport -> RemoteTable -> Assertion
490testTDict TestTransport{..} rtable = do
491 done <- newEmptyMVar
492 [node1, node2] <- replicateM 2 $ newLocalNode testTransport rtable
493 forkProcess node1 $ do
494 res <- call $(functionTDict 'isPrime) (localNodeId node2) ($(mkClosure 'isPrime) (79 :: Integer))
495 if res == True
496 then liftIO $ putMVar done ()
497 else error "Something went horribly wrong..."
498 takeMVar done
500testFib :: TestTransport -> RemoteTable -> Assertion
501testFib TestTransport{..} rtable = do
502 nodes <- replicateM 4 $ newLocalNode testTransport rtable
503 done <- newEmptyMVar
505 forkProcess (head nodes) $ do
506 (sport, rport) <- newChan
507 spawnLocal $ dfib (map localNodeId nodes, sport, 10)
508 55 <- receiveChan rport :: Process Integer
509 liftIO $ putMVar done ()
511 takeMVar done
513testSpawnReconnect :: TestTransport -> RemoteTable -> Assertion
514testSpawnReconnect testtrans@TestTransport{..} rtable = do
515 [node1, node2] <- replicateM 2 $ newLocalNode testTransport rtable
516 let nid1 = localNodeId node1
517 nid2 = localNodeId node2
518 done <- newEmptyMVar
519 iv <- newMVar (0 :: Int)
521 incr <- forkProcess node1 $ forever $ do
522 () <- expect
523 liftIO $ modifyMVar_ iv (return . (+ 1))
525 forkProcess node2 $ do
526 _pid1 <- spawn nid1 ($(mkClosure 'signal) incr)
527 simulateNetworkFailure testtrans node2 node1
528 _pid2 <- spawn nid1 ($(mkClosure 'signal) incr)
529 _pid3 <- spawn nid1 ($(mkClosure 'signal) incr)
531 liftIO $ threadDelay 100000
533 count <- liftIO $ takeMVar iv
534 res <- return $ count == 2 || count == 3 -- It depends on which message we get first in 'spawn'
536 liftIO $ putMVar done ()
537 if res /= True
538 then error "Something went horribly wrong"
539 else return ()
541 takeMVar done
543-- | 'spawn' used to ave a race condition which would be triggered if the
544-- spawning process terminates immediately after spawning
545testSpawnTerminate :: TestTransport -> RemoteTable -> Assertion
546testSpawnTerminate TestTransport{..} rtable = do
547 slave <- newLocalNode testTransport rtable
548 master <- newLocalNode testTransport rtable
549 masterDone <- newEmptyMVar
551 runProcess master $ do
552 us <- getSelfPid
553 replicateM_ 1000 . spawnLocal . void . spawn (localNodeId slave) $ $(mkClosure 'signal) us
554 replicateM_ 1000 $ (expect :: Process ())
555 liftIO $ putMVar masterDone ()
557 takeMVar masterDone
559tests :: TestTransport -> IO [Test]
560tests testtrans = do
561 let rtable = __remoteTable . __remoteTableDecl $ initRemoteTable
562 return
563 [ testCase "Unclosure" (testUnclosure testtrans rtable)
564 , testCase "Bind" (testBind testtrans rtable)
565 , testCase "SendPureClosure" (testSendPureClosure testtrans rtable)
566 , testCase "SendIOClosure" (testSendIOClosure testtrans rtable)
567 , testCase "SendProcClosure" (testSendProcClosure testtrans rtable)
568 , testCase "Spawn" (testSpawn testtrans rtable)
569 , testCase "SpawnRace" (testSpawnRace testtrans rtable)
570 , testCase "Call" (testCall testtrans rtable)
571 , testCase "CallBind" (testCallBind testtrans rtable)
572 , testCase "Seq" (testSeq testtrans rtable)
573 , testCase "SpawnSupervised" (testSpawnSupervised testtrans rtable)
574 , testCase "SpawnInvalid" (testSpawnInvalid testtrans rtable)
575 , testCase "ClosureExpect" (testClosureExpect testtrans rtable)
576 , testCase "SpawnChannel" (testSpawnChannel testtrans rtable)
577 , testCase "TDict" (testTDict testtrans rtable)
578 , testCase "Fib" (testFib testtrans rtable)
579 , testCase "SpawnTerminate" (testSpawnTerminate testtrans rtable)
580 , testCase "SpawnReconnect" (testSpawnReconnect testtrans rtable)
581 ]