git.iko.soy/distributed-process
on 7930062 help sign in
distributed-process/distributed-process-tests/src/Control/Distributed/Process/Tests/CH.hs
1872 lines 66 kB history raw
1module Control.Distributed.Process.Tests.CH (tests) where
3#if ! MIN_VERSION_base(4,6,0)
4import Prelude hiding (catch)
5#endif
7import Network.Transport.Test (TestTransport(..))
9import Data.Binary (Binary(..))
10import Data.Typeable (Typeable)
11import Data.Foldable (forM_)
12import Data.Function (fix)
13import Data.IORef
14 ( readIORef
15 , writeIORef
16 , newIORef
17 )
18import Control.Concurrent (forkIO, threadDelay, myThreadId, throwTo, ThreadId, yield)
19import Control.Concurrent.MVar
20 ( MVar
21 , newEmptyMVar
22 , putMVar
23 , takeMVar
24 , readMVar
25 )
26import Control.Monad (replicateM_, replicateM, forever, void, unless, join)
27import Control.Exception (SomeException, throwIO, ErrorCall(..))
28import Control.Monad.Catch (try, catch, finally, mask, onException)
29import Control.Applicative ((<|>))
30import qualified Network.Transport as NT (closeEndPoint, EndPointAddress)
31import Control.Distributed.Process hiding
32 ( try
33 , catch
34 , finally
35 , mask
36 , onException
37 )
38import Control.Distributed.Process.Internal.Types
39 ( LocalNode(localEndPoint)
40 , ProcessExitException(..)
41 , nullProcessId
42 , createUnencodedMessage
43 )
44import Control.Distributed.Process.Node
45import Control.Distributed.Process.Debug
46import Control.Distributed.Process.Management.Internal.Types
47import Control.Distributed.Process.Tests.Internal.Utils (shouldBe, pause)
48import Control.Distributed.Process.Serializable (Serializable)
49import Data.Maybe (isNothing)
50import Test.HUnit (Assertion, assertBool, assertFailure)
51import Test.Framework (Test, testGroup)
52import Test.Framework.Providers.HUnit (testCase)
53import Control.Rematch hiding (match, isNothing)
54import Control.Rematch.Run (Match(..))
56newtype Ping = Ping ProcessId
57 deriving (Typeable, Binary, Show)
59newtype Pong = Pong ProcessId
60 deriving (Typeable, Binary, Show)
62--------------------------------------------------------------------------------
63-- Supporting definitions --
64--------------------------------------------------------------------------------
66expectThat :: a -> Matcher a -> Assertion
67expectThat a matcher = case res of
68 MatchSuccess -> return ()
69 (MatchFailure msg) -> assertFailure msg
70 where res = runMatch matcher a
72-- | Like fork, but throw exceptions in the child thread to the parent
73forkTry :: IO () -> IO ThreadId
74forkTry p = do
75 tid <- myThreadId
76 forkIO $ catch p (\e -> throwTo tid (e :: SomeException))
78-- | The ping server from the paper
79ping :: Process ()
80ping = do
81 Pong partner <- expect
82 self <- getSelfPid
83 send partner (Ping self)
84 ping
86verifyClient :: String -> MVar Bool -> IO ()
87verifyClient s b = takeMVar b >>= assertBool s
89expectPing :: MVar Bool -> Process ()
90expectPing mv = expect >>= liftIO . putMVar mv . checkPing
91 where
92 checkPing (Ping _) = True
93 checkPing _ = False
95-- | Quick and dirty synchronous version of whereisRemoteAsync
96whereisRemote :: NodeId -> String -> Process (Maybe ProcessId)
97whereisRemote nid string = do
98 whereisRemoteAsync nid string
99 receiveWait [
100 match (\(WhereIsReply _ mPid) -> return mPid)
101 ]
103verifyWhereIsRemote :: NodeId -> String -> Process ProcessId
104verifyWhereIsRemote n s = whereisRemote n s >>= maybe (die "remote name not found") return
106waitForExit :: ProcessId -> Process DiedReason
107waitForExit pid = monitor pid >>= waitForDown
109waitForDown :: MonitorRef -> Process DiedReason
110waitForDown ref =
111 receiveWait [ matchIf (\(ProcessMonitorNotification ref' _ _) -> ref == ref')
112 (\(ProcessMonitorNotification _ _ dr) -> return dr) ]
114syncBreakConnection :: (NT.EndPointAddress -> NT.EndPointAddress -> IO ()) -> LocalNode -> LocalNode -> IO ()
115syncBreakConnection breakConnection nid0 nid1 = do
116 m <- newEmptyMVar
117 _ <- forkProcess nid1 $ getSelfPid >>= liftIO . putMVar m
118 runProcess nid0 $ do
119 them <- liftIO $ takeMVar m
120 pinger <- spawnLocal $ forever $ send them ()
121 _ <- monitorNode (localNodeId nid1)
122 liftIO $ breakConnection (nodeAddress $ localNodeId nid0)
123 (nodeAddress $ localNodeId nid1)
124 NodeMonitorNotification _ _ _ <- expect
125 kill pinger "finished"
126 return ()
128data Add = Add ProcessId Double Double deriving (Typeable)
129data Divide = Divide ProcessId Double Double deriving (Typeable)
130data DivByZero = DivByZero deriving (Typeable)
132instance Binary Add where
133 put (Add pid x y) = put pid >> put x >> put y
134 get = Add <$> get <*> get <*> get
136instance Binary Divide where
137 put (Divide pid x y) = put pid >> put x >> put y
138 get = Divide <$> get <*> get <*> get
140instance Binary DivByZero where
141 put DivByZero = return ()
142 get = return DivByZero
144-- The math server from the paper
145math :: Process ()
146math = do
147 receiveWait
148 [ match (\(Add pid x y) -> send pid (x + y))
149 , matchIf (\(Divide _ _ y) -> y /= 0)
150 (\(Divide pid x y) -> send pid (x / y))
151 , match (\(Divide pid _ _) -> send pid DivByZero)
152 ]
153 math
155-- | Monitor or link to a remote node
156monitorOrLink :: Bool -- ^ 'True' for monitor, 'False' for link
157 -> ProcessId -- ^ Process to monitor/link to
158 -> Maybe (MVar ()) -- ^ MVar to signal on once the monitor has been set up
159 -> Process (Maybe MonitorRef)
160monitorOrLink mOrL pid mSignal = do
161 result <- if mOrL then Just <$> monitor pid
162 else link pid >> return Nothing
163 -- Monitor is asynchronous, which usually does not matter but if we want a
164 -- *specific* signal then it does. Therefore we wait until the MonitorRef is
165 -- listed in the ProcessInfo and hope that this means the monitor has been set
166 -- up.
167 forM_ mSignal $ \signal -> do
168 self <- getSelfPid
169 spawnLocal $ do
170 let waitForMOrL = do
171 liftIO $ threadDelay 100000
172 mpinfo <- getProcessInfo pid
173 case mpinfo of
174 Nothing -> waitForMOrL
175 Just pinfo ->
176 if mOrL then
177 unless (result == lookup self (infoMonitors pinfo)) waitForMOrL
178 else
179 unless (elem self $ infoLinks pinfo) waitForMOrL
180 waitForMOrL
181 liftIO $ putMVar signal ()
182 return result
184monitorTestProcess :: ProcessId -- Process to monitor/link to
185 -> Bool -- 'True' for monitor, 'False' for link
186 -> Bool -- Should we unmonitor?
187 -> DiedReason -- Expected cause of death
188 -> Maybe (MVar ()) -- Signal for 'monitor set up'
189 -> MVar () -- Signal for successful termination
190 -> Process ()
191monitorTestProcess theirAddr mOrL un reason monitorSetup done =
192 catch (do mRef <- monitorOrLink mOrL theirAddr monitorSetup
193 case (un, mRef) of
194 (True, Nothing) -> do
195 unlink theirAddr
196 liftIO $ putMVar done ()
197 (True, Just ref) -> do
198 unmonitor ref
199 liftIO $ putMVar done ()
200 (False, ref) -> do
201 receiveWait [
202 match (\(ProcessMonitorNotification ref' pid reason') -> do
203 liftIO $ do
204 assertBool "Bad Monitor Signal"
205 (Just ref' == ref && pid == theirAddr &&
206 mOrL && reason == reason')
207 putMVar done ())
208 ]
209 )
210 (\(ProcessLinkException pid reason') -> do
211 (liftIO $ assertBool "link exception unmatched" $
212 pid == theirAddr && not mOrL && not un && reason == reason')
213 liftIO $ putMVar done ()
214 )
216--------------------------------------------------------------------------------
217-- The tests proper --
218--------------------------------------------------------------------------------
220-- | Basic ping test
221testPing :: TestTransport -> Assertion
222testPing TestTransport{..} = do
223 serverAddr <- newEmptyMVar
224 clientDone <- newEmptyMVar
226 -- Server
227 forkIO $ do
228 localNode <- newLocalNode testTransport initRemoteTable
229 addr <- forkProcess localNode ping
230 putMVar serverAddr addr
232 -- Client
233 forkIO $ do
234 localNode <- newLocalNode testTransport initRemoteTable
235 pingServer <- readMVar serverAddr
237 let numPings = 10000
239 runProcess localNode $ do
240 pid <- getSelfPid
241 replicateM_ numPings $ do
242 send pingServer (Pong pid)
243 p <- expect
244 case p of
245 Ping _ -> return ()
246 _ -> die "Unexpected message"
248 putMVar clientDone ()
250 takeMVar clientDone
252-- | Monitor a process on an unreachable node
253testMonitorUnreachable :: TestTransport -> Bool -> Bool -> Assertion
254testMonitorUnreachable TestTransport{..} mOrL un = do
255 deadProcess <- newEmptyMVar
256 done <- newEmptyMVar
258 forkIO $ do
259 localNode <- newLocalNode testTransport initRemoteTable
260 addr <- forkProcess localNode expect
261 closeLocalNode localNode
262 putMVar deadProcess addr
264 forkIO $ do
265 localNode <- newLocalNode testTransport initRemoteTable
266 theirAddr <- readMVar deadProcess
267 runProcess localNode $
268 monitorTestProcess theirAddr mOrL un DiedDisconnect Nothing done
270 takeMVar done
272-- | Monitor a process which terminates normally
273testMonitorNormalTermination :: TestTransport -> Bool -> Bool -> Assertion
274testMonitorNormalTermination TestTransport{..} mOrL un = do
275 monitorSetup <- newEmptyMVar
276 monitoredProcess <- newEmptyMVar
277 done <- newEmptyMVar
279 forkIO $ do
280 localNode <- newLocalNode testTransport initRemoteTable
281 addr <- forkProcess localNode $
282 liftIO $ readMVar monitorSetup
283 putMVar monitoredProcess addr
285 forkIO $ do
286 localNode <- newLocalNode testTransport initRemoteTable
287 theirAddr <- readMVar monitoredProcess
288 runProcess localNode $
289 monitorTestProcess theirAddr mOrL un DiedNormal (Just monitorSetup) done
291 takeMVar done
293-- | Monitor a process which terminates abnormally
294testMonitorAbnormalTermination :: TestTransport -> Bool -> Bool -> Assertion
295testMonitorAbnormalTermination TestTransport{..} mOrL un = do
296 monitorSetup <- newEmptyMVar
297 monitoredProcess <- newEmptyMVar
298 done <- newEmptyMVar
300 let err = userError "Abnormal termination"
302 forkIO $ do
303 localNode <- newLocalNode testTransport initRemoteTable
304 addr <- forkProcess localNode . liftIO $ do
305 readMVar monitorSetup
306 throwIO err
307 putMVar monitoredProcess addr
309 forkIO $ do
310 localNode <- newLocalNode testTransport initRemoteTable
311 theirAddr <- readMVar monitoredProcess
312 runProcess localNode $
313 monitorTestProcess theirAddr mOrL un (DiedException (show err)) (Just monitorSetup) done
315 takeMVar done
317-- | Monitor a local process that is already dead
318testMonitorLocalDeadProcess :: TestTransport -> Bool -> Bool -> Assertion
319testMonitorLocalDeadProcess TestTransport{..} mOrL un = do
320 processAddr <- newEmptyMVar
321 localNode <- newLocalNode testTransport initRemoteTable
322 done <- newEmptyMVar
324 forkIO $ do
325 addr <- forkProcess localNode $ return ()
326 putMVar processAddr addr
328 forkIO $ do
329 theirAddr <- readMVar processAddr
330 runProcess localNode $ do
331 monitor theirAddr
332 -- wait for the process to die
333 expect :: Process ProcessMonitorNotification
334 monitorTestProcess theirAddr mOrL un DiedUnknownId Nothing done
336 takeMVar done
338-- | Monitor a remote process that is already dead
339testMonitorRemoteDeadProcess :: TestTransport -> Bool -> Bool -> Assertion
340testMonitorRemoteDeadProcess TestTransport{..} mOrL un = do
341 processDead <- newEmptyMVar
342 processAddr <- newEmptyMVar
343 done <- newEmptyMVar
345 forkIO $ do
346 localNode <- newLocalNode testTransport initRemoteTable
347 addr <- forkProcess localNode . liftIO $ putMVar processDead ()
348 putMVar processAddr addr
350 forkIO $ do
351 localNode <- newLocalNode testTransport initRemoteTable
352 theirAddr <- readMVar processAddr
353 readMVar processDead
354 runProcess localNode $ do
355 monitorTestProcess theirAddr mOrL un DiedUnknownId Nothing done
357 takeMVar done
359-- | Monitor a process that becomes disconnected
360testMonitorDisconnect :: TestTransport -> Bool -> Bool -> Assertion
361testMonitorDisconnect TestTransport{..} mOrL un = do
362 processAddr <- newEmptyMVar
363 processAddr2 <- newEmptyMVar
364 monitorSetup <- newEmptyMVar
365 done <- newEmptyMVar
367 forkIO $ do
368 localNode <- newLocalNode testTransport initRemoteTable
369 addr <- forkProcess localNode $ expect
370 addr2 <- forkProcess localNode $ return ()
371 putMVar processAddr addr
372 readMVar monitorSetup
373 NT.closeEndPoint (localEndPoint localNode)
374 putMVar processAddr2 addr2
376 forkIO $ do
377 localNode <- newLocalNode testTransport initRemoteTable
378 theirAddr <- readMVar processAddr
379 forkProcess localNode $ do
380 lc <- liftIO $ readMVar processAddr2
381 send lc ()
382 runProcess localNode $ do
383 monitorTestProcess theirAddr mOrL un DiedDisconnect (Just monitorSetup) done
385 takeMVar done
387-- | Test the math server (i.e., receiveWait)
388testMath :: TestTransport -> Assertion
389testMath TestTransport{..} = do
390 serverAddr <- newEmptyMVar
391 clientDone <- newEmptyMVar
393 -- Server
394 forkIO $ do
395 localNode <- newLocalNode testTransport initRemoteTable
396 addr <- forkProcess localNode math
397 putMVar serverAddr addr
399 -- Client
400 forkIO $ do
401 localNode <- newLocalNode testTransport initRemoteTable
402 mathServer <- readMVar serverAddr
404 runProcess localNode $ do
405 pid <- getSelfPid
406 send mathServer (Add pid 1 2)
407 three <- expect :: Process Double
408 send mathServer (Divide pid 8 2)
409 four <- expect :: Process Double
410 send mathServer (Divide pid 8 0)
411 divByZ <- expect
412 liftIO $ putMVar clientDone (three, four, divByZ)
414 res <- takeMVar clientDone
415 case res of
416 (3, 4, DivByZero) -> return ()
417 _ -> error $ "Something went horribly wrong"
419-- | Send first message (i.e. connect) to an already terminated process
420-- (without monitoring); then send another message to a second process on
421-- the same remote node (we're checking that the remote node did not die)
422testSendToTerminated :: TestTransport -> Assertion
423testSendToTerminated TestTransport{..} = do
424 serverAddr1 <- newEmptyMVar
425 serverAddr2 <- newEmptyMVar
426 clientDone <- newEmptyMVar
428 forkIO $ do
429 terminated <- newEmptyMVar
430 localNode <- newLocalNode testTransport initRemoteTable
431 addr1 <- forkProcess localNode $ liftIO $ putMVar terminated ()
432 addr2 <- forkProcess localNode $ ping
433 readMVar terminated
434 putMVar serverAddr1 addr1
435 putMVar serverAddr2 addr2
437 forkIO $ do
438 localNode <- newLocalNode testTransport initRemoteTable
439 server1 <- readMVar serverAddr1
440 server2 <- readMVar serverAddr2
441 runProcess localNode $ do
442 pid <- getSelfPid
443 send server1 "Hi"
444 send server2 (Pong pid)
445 expectPing clientDone
447 verifyClient "Expected Ping from server" clientDone
449-- | Test (non-zero) timeout
450testTimeout :: TestTransport -> Assertion
451testTimeout TestTransport{..} = do
452 localNode <- newLocalNode testTransport initRemoteTable
453 done <- newEmptyMVar
455 runProcess localNode $ do
456 res <- receiveTimeout 1000000 [match (\Add{} -> return ())]
457 liftIO $ putMVar done $ res == Nothing
459 verifyClient "Expected receiveTimeout to timeout..." done
461-- | Test zero timeout
462testTimeout0 :: TestTransport -> Assertion
463testTimeout0 TestTransport{..} = do
464 serverAddr <- newEmptyMVar
465 clientDone <- newEmptyMVar
467 forkIO $ do
468 localNode <- newLocalNode testTransport initRemoteTable
469 addr <- forkProcess localNode $ do
470 -- Variation on the venerable ping server which uses a zero timeout
471 partner <- fix $ \loop ->
472 receiveTimeout 0 [match (\(Pong partner) -> return partner)]
473 >>= maybe (liftIO (threadDelay 100000) >> loop) return
474 self <- getSelfPid
475 send partner (Ping self)
476 putMVar serverAddr addr
478 forkIO $ do
479 localNode <- newLocalNode testTransport initRemoteTable
480 server <- readMVar serverAddr
481 runProcess localNode $ do
482 pid <- getSelfPid
483 -- Send a bunch of messages. A large number of messages that the server
484 -- is not interested in, and then a single message that it wants
485 replicateM_ 10000 $ send server "Irrelevant message"
486 send server (Pong pid)
487 expectPing clientDone
489 verifyClient "Expected Ping from server" clientDone
491-- | Test typed channels
492testTypedChannels :: TestTransport -> Assertion
493testTypedChannels TestTransport{..} = do
494 serverChannel <- newEmptyMVar :: IO (MVar (SendPort (SendPort Bool, Int)))
495 clientDone <- newEmptyMVar
497 forkIO $ do
498 localNode <- newLocalNode testTransport initRemoteTable
499 forkProcess localNode $ do
500 (serverSendPort, rport) <- newChan
501 liftIO $ putMVar serverChannel serverSendPort
502 (clientSendPort, i) <- receiveChan rport
503 sendChan clientSendPort (even i)
504 return ()
506 forkIO $ do
507 localNode <- newLocalNode testTransport initRemoteTable
508 serverSendPort <- readMVar serverChannel
509 runProcess localNode $ do
510 (clientSendPort, rport) <- newChan
511 sendChan serverSendPort (clientSendPort, 5)
512 ch <- receiveChan rport
513 liftIO $ putMVar clientDone $ ch == False
515 verifyClient "Expected channel to send 'False'" clientDone
517-- | Test merging receive ports
518testMergeChannels :: TestTransport -> Assertion
519testMergeChannels TestTransport{..} = do
520 localNode <- newLocalNode testTransport initRemoteTable
521 testFlat localNode True "aaabbbccc"
522 testFlat localNode False "abcabcabc"
523 testNested localNode True True "aaabbbcccdddeeefffggghhhiii"
524 testNested localNode True False "adgadgadgbehbehbehcficficfi"
525 testNested localNode False True "abcabcabcdefdefdefghighighi"
526 testNested localNode False False "adgbehcfiadgbehcfiadgbehcfi"
527 testBlocked localNode True
528 testBlocked localNode False
529 where
530 -- Single layer of merging
531 testFlat :: LocalNode -> Bool -> String -> IO ()
532 testFlat localNode biased expected = do
533 done <- newEmptyMVar
534 forkProcess localNode $ do
535 rs <- mapM charChannel "abc"
536 m <- mergePorts biased rs
537 xs <- replicateM 9 $ receiveChan m
538 liftIO $ putMVar done $ xs == expected
539 verifyClient "Expected single layer merge to match expected ordering" done
541 -- Two layers of merging
542 testNested :: LocalNode -> Bool -> Bool -> String -> IO ()
543 testNested localNode biasedInner biasedOuter expected = do
544 done <- newEmptyMVar
545 forkProcess localNode $ do
546 rss <- mapM (mapM charChannel) ["abc", "def", "ghi"]
547 ms <- mapM (mergePorts biasedInner) rss
548 m <- mergePorts biasedOuter ms
549 xs <- replicateM (9 * 3) $ receiveChan m
550 liftIO $ putMVar done $ xs == expected
551 verifyClient "Expected nested channels to match expeted ordering" done
553 -- Test that if no messages are (immediately) available, the scheduler makes no difference
554 testBlocked :: LocalNode -> Bool -> IO ()
555 testBlocked localNode biased = do
556 vs <- replicateM 3 newEmptyMVar
557 done <- newEmptyMVar
559 forkProcess localNode $ do
560 ss <- liftIO $ mapM readMVar vs
561 case ss of
562 [sa, sb, sc] ->
563 mapM_ ((>> liftIO (threadDelay 10000)) . uncurry sendChan)
564 [ -- a, b, c
565 (sa, 'a')
566 , (sb, 'b')
567 , (sc, 'c')
568 -- a, c, b
569 , (sa, 'a')
570 , (sc, 'c')
571 , (sb, 'b')
572 -- b, a, c
573 , (sb, 'b')
574 , (sa, 'a')
575 , (sc, 'c')
576 -- b, c, a
577 , (sb, 'b')
578 , (sc, 'c')
579 , (sa, 'a')
580 -- c, a, b
581 , (sc, 'c')
582 , (sa, 'a')
583 , (sb, 'b')
584 -- c, b, a
585 , (sc, 'c')
586 , (sb, 'b')
587 , (sa, 'a')
588 ]
589 _ -> die "Something went horribly wrong"
591 forkProcess localNode $ do
592 (ss, rs) <- unzip <$> replicateM 3 newChan
593 liftIO $ mapM_ (uncurry putMVar) $ zip vs ss
594 m <- mergePorts biased rs
595 xs <- replicateM (6 * 3) $ receiveChan m
596 liftIO $ putMVar done $ xs == "abcacbbacbcacabcba"
598 verifyClient "Expected merged ports to match expected ordering" done
600 mergePorts :: Serializable a => Bool -> [ReceivePort a] -> Process (ReceivePort a)
601 mergePorts True = mergePortsBiased
602 mergePorts False = mergePortsRR
604 charChannel :: Char -> Process (ReceivePort Char)
605 charChannel c = do
606 (sport, rport) <- newChan
607 replicateM_ 3 $ sendChan sport c
608 liftIO $ threadDelay 10000 -- Make sure messages have been sent
609 return rport
611testTerminate :: TestTransport -> Assertion
612testTerminate TestTransport{..} = do
613 localNode <- newLocalNode testTransport initRemoteTable
614 runProcess localNode $ do
615 e <- try terminate :: Process (Either ProcessTerminationException ())
616 if either show show e == show ProcessTerminationException
617 then return ()
618 else die "Unexpected result from terminate"
620testMonitorNode :: TestTransport -> Assertion
621testMonitorNode TestTransport{..} = do
622 [node1, node2] <- replicateM 2 $ newLocalNode testTransport initRemoteTable
623 done <- newEmptyMVar
625 closeLocalNode node1
627 runProcess node2 $ do
628 ref <- monitorNode (localNodeId node1)
629 receiveWait [
630 match (\(NodeMonitorNotification ref' nid DiedDisconnect) ->
631 return $ ref == ref' && nid == localNodeId node1)
632 ] >>= liftIO . putMVar done
634 verifyClient "Expected NodeMonitorNotification with matching ref & nodeId" done
636testMonitorLiveNode :: TestTransport -> Assertion
637testMonitorLiveNode TestTransport{..} = do
638 [node1, node2] <- replicateM 2 $ newLocalNode testTransport initRemoteTable
639 ready <- newEmptyMVar
640 readyr <- newEmptyMVar
641 done <- newEmptyMVar
643 p <- forkProcess node1 $ return ()
644 forkProcess node2 $ do
645 ref <- monitorNode (localNodeId node1)
646 liftIO $ putMVar ready ()
647 liftIO $ takeMVar readyr
648 send p ()
649 receiveWait [
650 match (\(NodeMonitorNotification ref' nid _) ->
651 (return $ ref == ref' && nid == localNodeId node1))
652 ] >>= liftIO . putMVar done
654 takeMVar ready
655 closeLocalNode node1
656 putMVar readyr ()
658 verifyClient "Expected NodeMonitorNotification for LIVE node" done
660testMonitorChannel :: TestTransport -> Assertion
661testMonitorChannel TestTransport{..} = do
662 [node1, node2] <- replicateM 2 $ newLocalNode testTransport initRemoteTable
663 gotNotification <- newEmptyMVar
665 pid <- forkProcess node1 $ do
666 sport <- expect :: Process (SendPort ())
667 ref <- monitorPort sport
668 receiveWait [
669 -- reason might be DiedUnknownId if the receive port is GCed before the
670 -- monitor is established (TODO: not sure that this is reasonable)
671 match (\(PortMonitorNotification ref' port' reason) ->
672 return $ ref' == ref && port' == sendPortId sport &&
673 (reason == DiedNormal || reason == DiedUnknownId))
674 ] >>= liftIO . putMVar gotNotification
676 runProcess node2 $ do
677 (sport, _) <- newChan :: Process (SendPort (), ReceivePort ())
678 send pid sport
679 liftIO $ threadDelay 100000
681 verifyClient "Expected PortMonitorNotification" gotNotification
683testRegistry :: TestTransport -> Assertion
684testRegistry TestTransport{..} = do
685 node <- newLocalNode testTransport initRemoteTable
686 done <- newEmptyMVar
688 pingServer <- forkProcess node ping
689 deadProcess <- forkProcess node (return ())
691 runProcess node $ do
692 register "ping" pingServer
693 whereis "ping" >>= liftIO . assertBool "Unexpected ping" . (== Just pingServer)
694 us <- getSelfPid
695 nsend "ping" (Pong us)
696 receiveWait [
697 matchIf (\(Ping pid') -> pingServer == pid') (const $ return ())
698 ]
699 checkRegException "dead" Nothing deadProcess
700 checkRegException "ping" (Just pingServer) deadProcess
701 try (unregister "dead") >>= checkReg "dead" Nothing
702 liftIO $ putMVar done ()
704 takeMVar done
706 where
707 checkRegException name pid dead =
708 try (register name dead) >>= checkReg name pid
710 checkReg name pid res =
711 case res of
712 Left (ProcessRegistrationException name pid) -> return ()
713 _ -> die $ "Unexpected Registration" ++ show res
715testRegistryRemoteProcess :: TestTransport -> Assertion
716testRegistryRemoteProcess TestTransport{..} = do
717 node1 <- newLocalNode testTransport initRemoteTable
718 node2 <- newLocalNode testTransport initRemoteTable
719 done <- newEmptyMVar
721 pingServer <- forkProcess node1 ping
723 runProcess node2 $ do
724 register "ping" pingServer
725 whereis "ping" >>= liftIO . assertBool "Unexpected ping" . (== Just pingServer)
726 us <- getSelfPid
727 nsend "ping" (Pong us)
728 receiveWait [
729 matchIf (\(Ping pid') -> pingServer == pid')
730 (const $ liftIO $ putMVar done ())
731 ]
733 takeMVar done
735testRemoteRegistry :: TestTransport -> Assertion
736testRemoteRegistry TestTransport{..} = do
737 node1 <- newLocalNode testTransport initRemoteTable
738 node2 <- newLocalNode testTransport initRemoteTable
739 done <- newEmptyMVar
741 pingServer <- forkProcess node1 ping
742 deadProcess <- forkProcess node1 (return ())
744 runProcess node2 $ do
745 let nid1 = localNodeId node1
746 registerRemoteAsync nid1 "ping" pingServer
747 receiveWait [
748 matchIf (\(RegisterReply label' _ (Just pid)) ->
749 "ping" == label' && pid == pingServer)
750 (\(RegisterReply _ _ _) -> return ()) ]
752 pid <- verifyWhereIsRemote nid1 "ping"
753 liftIO $ assertBool "Expected pindServer to match pid" $ pingServer == pid
754 us <- getSelfPid
755 nsendRemote nid1 "ping" (Pong us)
756 receiveWait [
757 match (\(Ping pid') -> return $ pingServer == pid')
758 ] >>= liftIO . assertBool "Expected Ping with ping server's ProcessId"
760 -- test that if process was not registered Nothing is returned
761 -- in owner field.
762 registerRemoteAsync nid1 "dead" deadProcess
763 receiveWait [
764 matchIf (\(RegisterReply label' False Nothing) -> "dead" == label')
765 (\(RegisterReply _ _ _) -> return ()) ]
766 registerRemoteAsync nid1 "ping" deadProcess
767 receiveWait [
768 matchIf (\(RegisterReply label' False (Just pid'')) ->
769 "ping" == label' && pid'' == pingServer)
770 (\(RegisterReply _ _ _) -> return ()) ]
771 unregisterRemoteAsync nid1 "dead"
772 receiveWait [
773 matchIf (\(RegisterReply label' False Nothing) ->
774 "dead" == label' && pid == pingServer)
775 (\(RegisterReply _ _ _) -> return ()) ]
776 liftIO $ putMVar done ()
778 takeMVar done
780testRemoteRegistryRemoteProcess :: TestTransport -> Assertion
781testRemoteRegistryRemoteProcess TestTransport{..} = do
782 node1 <- newLocalNode testTransport initRemoteTable
783 node2 <- newLocalNode testTransport initRemoteTable
784 done <- newEmptyMVar
786 pingServer <- forkProcess node2 ping
788 runProcess node2 $ do
789 let nid1 = localNodeId node1
790 registerRemoteAsync nid1 "ping" pingServer
791 receiveWait [
792 matchIf (\(RegisterReply label' _ _) -> "ping" == label')
793 (\(RegisterReply _ _ _) -> return ()) ]
794 pid <- verifyWhereIsRemote nid1 "ping"
795 liftIO $ assertBool "Expected pingServer to match remote name" $ pingServer == pid
796 us <- getSelfPid
797 nsendRemote nid1 "ping" (Pong us)
798 receiveWait [
799 match (\(Ping pid') -> return $ pingServer == pid')
800 ] >>= liftIO . putMVar done
802 verifyClient "Expected Ping with ping server's ProcessId" done
804testSpawnLocal :: TestTransport -> Assertion
805testSpawnLocal TestTransport{..} = do
806 node <- newLocalNode testTransport initRemoteTable
807 done <- newEmptyMVar
809 runProcess node $ do
810 us <- getSelfPid
812 pid <- spawnLocal $ do
813 sport <- expect
814 sendChan sport (1234 :: Int)
816 sport <- spawnChannelLocal $
817 \rport -> (receiveChan rport :: Process Int) >>= send us
819 send pid sport
820 expect >>= liftIO . putMVar done
822 res <- takeMVar done
823 assertBool "Expected 1234 :: Int" $ res == (1234 :: Int)
825testSpawnAsyncStrictness :: TestTransport -> Assertion
826testSpawnAsyncStrictness TestTransport{..} = do
827 node <- newLocalNode testTransport initRemoteTable
828 done <- newEmptyMVar
830 runProcess node $ do
831 here <-getSelfNode
833 ev <- try $ spawnAsync here (error "boom")
834 liftIO $ case ev of
835 Right _ -> putMVar done (error "Exception didn't fire")
836 Left (_::SomeException) -> putMVar done (return ())
838 join $ takeMVar done
840testReconnect :: TestTransport -> Assertion
841testReconnect TestTransport{..} = do
842 [node1, node2] <- replicateM 2 $ newLocalNode testTransport initRemoteTable
843 let nid1 = localNodeId node1
844 processA <- newEmptyMVar
845 [sendTestOk, registerTestOk] <- replicateM 2 newEmptyMVar
847 forkProcess node1 $ do
848 us <- getSelfPid
849 liftIO $ putMVar processA us
850 msg1 <- expect
851 msg2 <- expect
852 liftIO $ do
853 assertBool "messages did not match" $ msg1 == "message 1" && msg2 == "message 3"
854 putMVar sendTestOk ()
856 forkProcess node2 $ do
857 {-
858 - Make sure there is no implicit reconnect on normal message sending
859 -}
861 them <- liftIO $ readMVar processA
862 send them "message 1" >> liftIO (threadDelay 100000)
864 -- Simulate network failure
865 liftIO $ syncBreakConnection testBreakConnection node1 node2
868 -- Should not arrive
869 send them "message 2"
871 -- Should arrive
872 reconnect them
873 send them "message 3"
875 liftIO $ takeMVar sendTestOk
877 {-
878 - Test that there *is* implicit reconnect on node controller messages
879 -}
881 us <- getSelfPid
882 registerRemoteAsync nid1 "a" us -- registerRemote is asynchronous
883 receiveWait [
884 matchIf (\(RegisterReply label' _ _) -> "a" == label')
885 (\(RegisterReply _ _ _) -> return ()) ]
887 _ <- whereisRemote nid1 "a"
890 -- Simulate network failure
891 liftIO $ syncBreakConnection testBreakConnection node1 node2
893 -- This will happen due to implicit reconnect
894 registerRemoteAsync nid1 "b" us
895 receiveWait [
896 matchIf (\(RegisterReply label' _ _) -> "b" == label')
897 (\(RegisterReply _ _ _) -> return ()) ]
899 -- Should happen
900 registerRemoteAsync nid1 "c" us
901 receiveWait [
902 matchIf (\(RegisterReply label' _ _) -> "c" == label')
903 (\(RegisterReply _ _ _) -> return ()) ]
905 -- Check
906 mPid <- whereisRemote nid1 "a" -- this will fail because the name is removed when the node is disconnected
907 liftIO $ assertBool "Expected remote name to be lost" $ mPid == Nothing
908 verifyWhereIsRemote nid1 "b" -- this will suceed because the value is set after thereconnect
909 verifyWhereIsRemote nid1 "c"
911 liftIO $ putMVar registerTestOk ()
913 takeMVar registerTestOk
915-- | Tests that unreliable messages arrive sorted even when there are connection
916-- failures.
917testUSend :: (ProcessId -> Int -> Process ())
918 -> TestTransport -> Int -> Assertion
919testUSend usendPrim TestTransport{..} numMessages = do
920 [node1, node2] <- replicateM 2 $ newLocalNode testTransport initRemoteTable
921 let nid1 = localNodeId node1
922 nid2 = localNodeId node2
923 processA <- newEmptyMVar
924 usendTestOk <- newEmptyMVar
926 forkProcess node1 $ flip catch (\e -> liftIO $ print (e :: SomeException) ) $ do
927 us <- getSelfPid
928 liftIO $ putMVar processA us
929 them <- expect
930 send them ()
931 _ <- monitor them
932 let -- Collects messages from 'them' until the sender dies.
933 -- Disconnection notifications are ignored.
934 receiveMessages :: Process [Int]
935 receiveMessages = receiveWait
936 [ match $ \mn -> case mn of
937 ProcessMonitorNotification _ _ DiedDisconnect -> do
938 monitor them
939 receiveMessages
940 _ -> return []
941 , match $ \i -> fmap (i :) receiveMessages
942 ]
943 msgs <- receiveMessages
944 let -- Checks that the input list is sorted.
945 isSorted :: [Int] -> Bool
946 isSorted (x : xs@(y : _)) = x <= y && isSorted xs
947 isSorted _ = True
948 -- The list can't be null since there are no failures after sending
949 -- the last message.
950 liftIO $ putMVar usendTestOk $ isSorted msgs && not (null msgs)
952 forkProcess node2 $ do
953 them <- liftIO $ readMVar processA
954 getSelfPid >>= send them
955 expect :: Process ()
956 forM_ [1..numMessages] $ \i -> do
957 liftIO $ testBreakConnection (nodeAddress nid1) (nodeAddress nid2)
958 usendPrim them i
959 liftIO (threadDelay 30000)
961 res <- takeMVar usendTestOk
962 assertBool "Unexpected failure after sending last msg" res
964-- | Test 'matchAny'. This repeats the 'testMath' but with a proxy server
965-- in between
966testMatchAny :: TestTransport -> Assertion
967testMatchAny TestTransport{..} = do
968 proxyAddr <- newEmptyMVar
969 clientDone <- newEmptyMVar
971 -- Math server
972 forkIO $ do
973 localNode <- newLocalNode testTransport initRemoteTable
974 mathServer <- forkProcess localNode math
975 proxyServer <- forkProcess localNode $ forever $ do
976 msg <- receiveWait [ matchAny return ]
977 forward msg mathServer
978 putMVar proxyAddr proxyServer
980 -- Client
981 forkIO $ do
982 localNode <- newLocalNode testTransport initRemoteTable
983 mathServer <- readMVar proxyAddr
985 runProcess localNode $ do
986 pid <- getSelfPid
987 send mathServer (Add pid 1 2)
988 three <- expect :: Process Double
989 send mathServer (Divide pid 8 2)
990 four <- expect :: Process Double
991 send mathServer (Divide pid 8 0)
992 divByZ <- expect
993 liftIO $ putMVar clientDone (three, four, divByZ)
995 res <- takeMVar clientDone
996 case res of
997 (3, 4, DivByZero) -> return ()
998 _ -> error "Unexpected result"
1000-- | Test 'matchAny'. This repeats the 'testMath' but with a proxy server
1001-- in between, however we block 'Divide' requests ....
1002testMatchAnyHandle :: TestTransport -> Assertion
1003testMatchAnyHandle TestTransport{..} = do
1004 proxyAddr <- newEmptyMVar
1005 clientDone <- newEmptyMVar
1007 -- Math server
1008 forkIO $ do
1009 localNode <- newLocalNode testTransport initRemoteTable
1010 mathServer <- forkProcess localNode math
1011 proxyServer <- forkProcess localNode $ forever $ do
1012 receiveWait [
1013 matchAny (maybeForward mathServer)
1015 putMVar proxyAddr proxyServer
1017 -- Client
1018 forkIO $ do
1019 localNode <- newLocalNode testTransport initRemoteTable
1020 mathServer <- readMVar proxyAddr
1022 runProcess localNode $ do
1023 pid <- getSelfPid
1024 send mathServer (Add pid 1 2)
1025 three <- expect :: Process Double
1026 send mathServer (Divide pid 8 2)
1027 res <- (expectTimeout 100000) :: Process (Maybe Double)
1028 liftIO $ putMVar clientDone $ three == 3 && res == Nothing
1030 verifyClient "Expected Nothing (i.e. timeout)" clientDone
1032 where maybeForward :: ProcessId -> Message -> Process (Maybe ())
1033 maybeForward s msg =
1034 handleMessage msg (\m@(Add _ _ _) -> send s m)
1036testMatchAnyNoHandle :: TestTransport -> Assertion
1037testMatchAnyNoHandle TestTransport{..} = do
1038 addr <- newEmptyMVar
1039 clientDone <- newEmptyMVar
1040 serverDone <- newEmptyMVar
1042 -- Math server
1043 forkIO $ do
1044 localNode <- newLocalNode testTransport initRemoteTable
1045 server <- forkProcess localNode $ forever $ do
1046 receiveWait [
1047 matchAnyIf
1048 -- the condition has type `Add -> Bool`
1049 (\(Add _ _ _) -> True)
1050 -- the match `AbstractMessage -> Process ()` will succeed!
1051 (\m -> do
1052 -- `String -> Process ()` does *not* match the input types however
1053 r <- (handleMessage m (\(_ :: String) -> die "NONSENSE" ))
1054 case r of
1055 Nothing -> return ()
1056 Just _ -> die "NONSENSE")
1058 -- we *must* have removed the message from our mailbox though!!!
1059 res <- receiveTimeout 100000 [ match (\(Add _ _ _) -> return ()) ]
1060 liftIO $ do
1061 assertBool "Expected timeout!" $ res == Nothing
1062 putMVar serverDone ()
1063 putMVar addr server
1065 -- Client
1066 forkIO $ do
1067 localNode <- newLocalNode testTransport initRemoteTable
1068 server <- readMVar addr
1070 runProcess localNode $ do
1071 pid <- getSelfPid
1072 send server (Add pid 1 2)
1073 -- we only care about the client having sent a message, so we're done
1074 liftIO $ putMVar clientDone ()
1076 takeMVar clientDone
1077 takeMVar serverDone
1079-- | Test 'matchAnyIf'. We provide an /echo/ server, but it ignores requests
1080-- unless the text body @/= "bar"@ - this case should time out rather than
1081-- removing the message from the process mailbox.
1082testMatchAnyIf :: TestTransport -> Assertion
1083testMatchAnyIf TestTransport{..} = do
1084 echoAddr <- newEmptyMVar
1085 clientDone <- newEmptyMVar
1087 -- echo server
1088 forkIO $ do
1089 localNode <- newLocalNode testTransport initRemoteTable
1090 echoServer <- forkProcess localNode $ forever $ do
1091 receiveWait [
1092 matchAnyIf (\(_ :: ProcessId, (s :: String)) -> s /= "bar")
1093 tryHandleMessage
1095 putMVar echoAddr echoServer
1097 -- Client
1098 forkIO $ do
1099 localNode <- newLocalNode testTransport initRemoteTable
1100 server <- readMVar echoAddr
1102 runProcess localNode $ do
1103 pid <- getSelfPid
1104 send server (pid, "foo")
1105 foo <- expect
1106 -- provoking what would be the wrong ordering is informative here...
1108 send server (pid, "bar")
1109 bar <- (expectTimeout 100000) :: Process (Maybe String) -- was Double o_O !?
1111 send server (pid, "baz")
1112 baz <- expect
1114 liftIO $ putMVar clientDone (foo, bar, baz)
1116 res <- takeMVar clientDone
1117 let res' = res == ("foo", Nothing, "baz")
1118 assertBool "Expected timeout due to type mismatch" res'
1120 where tryHandleMessage :: Message -> Process (Maybe ())
1121 tryHandleMessage msg =
1122 handleMessage msg (\(pid :: ProcessId, (m :: String))
1123 -> do { send pid m; return () })
1125testMatchMessageWithUnwrap :: TestTransport -> Assertion
1126testMatchMessageWithUnwrap TestTransport{..} = do
1127 echoAddr <- newEmptyMVar
1128 clientDone <- newEmptyMVar
1130 -- echo server
1131 forkIO $ do
1132 localNode <- newLocalNode testTransport initRemoteTable
1133 echoServer <- forkProcess localNode $ forever $ do
1134 msg <- receiveWait [
1135 matchMessage (\(m :: Message) -> do
1136 return m)
1138 unwrapped <- unwrapMessage msg :: Process (Maybe (ProcessId, Message))
1139 case unwrapped of
1140 (Just (p, msg')) -> forward msg' p
1141 Nothing -> die "unable to unwrap the message"
1142 putMVar echoAddr echoServer
1144 -- Client
1145 forkIO $ do
1146 localNode <- newLocalNode testTransport initRemoteTable
1147 server <- readMVar echoAddr
1149 runProcess localNode $ do
1150 pid <- getSelfPid
1151 send server (pid, wrapMessage ("foo" :: String))
1152 foo <- expect
1153 send server (pid, wrapMessage ("baz" :: String))
1154 baz <- expect
1155 liftIO $ putMVar clientDone (foo, baz)
1157 res <- takeMVar clientDone
1158 assertBool "Unexpected unwrapped results" $ res == ("foo", "baz")
1160-- Test 'receiveChanTimeout'
1161testReceiveChanTimeout :: TestTransport -> Assertion
1162testReceiveChanTimeout TestTransport{..} = do
1163 done <- newEmptyMVar
1164 mvSender <- newEmptyMVar
1165 sendPort <- newEmptyMVar
1167 forkTry $ do
1168 localNode <- newLocalNode testTransport initRemoteTable
1169 runProcess localNode $ do
1170 -- Create a typed channel
1171 (sp, rp) <- newChan :: Process (SendPort Bool, ReceivePort Bool)
1172 liftIO $ putMVar sendPort sp
1174 -- Wait for a message with a delay. No message arrives, we should get
1175 -- Nothing after the delay.
1176 receiveChanTimeout 100000 rp >>= maybe (return ()) (const $ die "Expected Timeout")
1178 -- Let the sender know that it can send a message.
1179 liftIO $ putMVar mvSender ()
1181 -- Wait for a message with a delay again. Now a message arrives after
1182 -- 0.1 seconds
1183 res <- receiveChanTimeout 20000000 rp >>= maybe (die "Timeout") return
1184 liftIO $ assertBool "Expected result to be 'True'" res
1186 -- Wait for a message with zero timeout: non-blocking check. No message is
1187 -- available, we get Nothing
1188 receiveChanTimeout 0 rp >>= maybe (return ()) (const $ die "Expected Timeout")
1190 -- Let the sender know that it can send a message.
1191 liftIO $ putMVar mvSender ()
1193 -- Again, but now there is a message available
1194 fix $ \loop -> do
1195 liftIO $ threadDelay 100000
1196 mb <- receiveChanTimeout 0 rp
1197 case mb of
1198 Just b -> do liftIO $ assertBool "Unexpected Message" $ not b
1199 _ -> loop
1201 liftIO $ putMVar done ()
1203 forkTry $ do
1204 localNode <- newLocalNode testTransport initRemoteTable
1205 runProcess localNode $ do
1206 sp <- liftIO $ readMVar sendPort
1208 liftIO $ takeMVar mvSender
1209 liftIO $ threadDelay 100000
1210 sendChan sp True
1212 liftIO $ takeMVar mvSender
1213 sendChan sp False
1215 takeMVar done
1217-- | Test Functor, Applicative, Alternative and Monad instances for ReceiveChan
1218testReceiveChanFeatures :: TestTransport -> Assertion
1219testReceiveChanFeatures TestTransport{..} = do
1220 done <- newEmptyMVar
1222 forkTry $ do
1223 localNode <- newLocalNode testTransport initRemoteTable
1224 runProcess localNode $ do
1225 (spInt, rpInt) <- newChan :: Process (SendPort Int, ReceivePort Int)
1226 (spBool, rpBool) <- newChan :: Process (SendPort Bool, ReceivePort Bool)
1228 -- Test Functor instance
1230 sendChan spInt 2
1231 sendChan spBool False
1233 rp1 <- mergePortsBiased [even <$> rpInt, rpBool]
1235 receiveChan rp1 >>= liftIO . assertBool "Expected True"
1236 receiveChan rp1 >>= liftIO . assertBool "Expected False" . not
1238 -- Test Applicative instance
1240 sendChan spInt 3
1241 sendChan spInt 4
1243 let rp2 = pure (+) <*> rpInt <*> rpInt
1245 receiveChan rp2 >>= liftIO . assertBool "Expected 7" . (== 7)
1247 -- Test Alternative instance
1249 sendChan spInt 3
1250 sendChan spBool True
1252 let rp3 = (even <$> rpInt) <|> rpBool
1254 receiveChan rp3 >>= liftIO . assertBool "Expected False" . not
1255 receiveChan rp3 >>= liftIO . assertBool "Expected True"
1257 -- Test Monad instance
1259 sendChan spBool True
1260 sendChan spBool False
1261 sendChan spInt 5
1263 let rp4 :: ReceivePort Int
1264 rp4 = do b <- rpBool
1265 if b
1266 then rpInt
1267 else return 7
1269 receiveChan rp4 >>= liftIO . assertBool "Expected 5" . (== 5)
1270 receiveChan rp4 >>= liftIO . assertBool "Expected 7" . (== 7)
1272 liftIO $ putMVar done ()
1274 takeMVar done
1276testChanLifecycle :: TestTransport -> Assertion
1277testChanLifecycle TestTransport{..} = let delay = 3000000 in do
1278 result <- newEmptyMVar
1279 tchMV <- newEmptyMVar
1280 localNode <- newLocalNode testTransport initRemoteTable
1281 runProcess localNode $ do
1283 pid <- spawnLocal $ do tCh <- newChan :: Process (SendPort (), ReceivePort ())
1284 liftIO $ putMVar tchMV tCh
1285 expect :: Process ()
1286 let (sp, _) = tCh
1287 sendChan sp ()
1288 expect :: Process ()
1290 mRefPid <- monitor pid
1292 cPid <- spawnLocal $ do
1293 (sp', rp) <- liftIO $ takeMVar tchMV
1294 -- say "picked up our test channel"
1295 send pid ()
1296 -- say "told pid to continue"
1297 res <- receiveChanTimeout delay rp
1298 case res of
1299 Nothing -> say "initial chan () missing!" >> (liftIO $ putMVar result False)
1300 Just () -> do mr <- monitor pid
1301 pause 10000
1302 -- say "sending pid a second () will cause it to exit"
1303 send pid ()
1305 -- say "make sure we see a DOWN notification for pid having stopped"
1306 receiveWait [ match (\(_ :: ProcessMonitorNotification) -> return ()) ]
1308 -- now that pid has died, the send port should be useless...
1309 liftIO $ putMVar tchMV (sp', rp)
1311 -- let's verify that we do not see the message from our
1312 -- parent process on the channel, once pid has died...
1313 recv <- receiveChanTimeout delay rp
1314 -- say $ "finished waiting for second (), writing result" ++ (show recv)
1315 liftIO $ putMVar result $ isNothing recv
1317 mRefCPid <- monitor cPid
1319 receiveWait
1320 [ matchIf (\(ProcessMonitorNotification r _ _) -> r == mRefPid)
1321 (\_ -> return ())
1324 -- say "seen first pid die..."
1326 (sendPort, _) <- liftIO $ takeMVar tchMV
1327 sendChan sendPort ()
1328 -- say "sent () after owning pid died"
1330 -- let cPid know we've written to the channel...
1331 send cPid ()
1333 receiveWait
1334 [ matchIf (\(ProcessMonitorNotification r _ _) -> r == mRefCPid)
1335 (\_ -> return ())
1338 -- say "seen both pids die now..."
1340 -- and wait on the result back in IO land...
1341 testRes <- takeMVar result
1342 -- runProcess localNode $ say "got result..."
1343 assertBool "Expected sending on the channel to fail, but received data!" testRes
1346testKillLocal :: TestTransport -> Assertion
1347testKillLocal TestTransport{..} = do
1348 localNode <- newLocalNode testTransport initRemoteTable
1350 pid <- forkProcess localNode $ do
1351 liftIO $ threadDelay 1000000
1353 runProcess localNode $ do
1354 ref <- monitor pid
1355 us <- getSelfPid
1356 kill pid "TestKill"
1357 mn <- expect
1358 case mn of
1359 ProcessMonitorNotification ref' pid' (DiedException ex) ->
1360 case ref == ref' && pid == pid' && ex == "killed-by=" ++ show us ++ ",reason=TestKill" of
1361 True -> return ()
1362 False -> die "Invalid ProcessMonitorNotification received"
1363 _ -> die "Something went horribly wrong"
1365testKillRemote :: TestTransport -> Assertion
1366testKillRemote TestTransport{..} = do
1367 node1 <- newLocalNode testTransport initRemoteTable
1368 node2 <- newLocalNode testTransport initRemoteTable
1370 pid <- forkProcess node1 $ do
1371 liftIO $ threadDelay 1000000
1373 runProcess node2 $ do
1374 ref <- monitor pid
1375 us <- getSelfPid
1376 kill pid "TestKill"
1377 mn <- expect
1378 case mn of
1379 ProcessMonitorNotification ref' pid' (DiedException reason) ->
1380 case (ref == ref', pid == pid', reason == "killed-by=" ++ show us ++ ",reason=TestKill") of
1381 (True, True, True) -> return ()
1382 (a, b, c) -> do
1383 let a' = if a then "" else "Invalid ref"
1384 let b' = if b then "" else "Invalid pid"
1385 let c' = if c then "" else "Invalid message"
1386 die $ unwords [a', b', c']
1387 _ -> die "Received unexpected message"
1389testCatchesExit :: TestTransport -> Assertion
1390testCatchesExit TestTransport{..} = do
1391 localNode <- newLocalNode testTransport initRemoteTable
1392 done <- newEmptyMVar
1394 _ <- forkProcess localNode $ do
1395 (die ("foobar", 123 :: Int))
1396 `catchesExit` [
1397 (\_ m -> handleMessage m (\(_ :: String) -> return ()))
1398 , (\_ m -> handleMessage m (\(_ :: Maybe Int) -> return ()))
1399 , (\_ m -> handleMessage m (\(_ :: String, _ :: Int)
1400 -> (liftIO $ putMVar done ()) >> return ()))
1403 takeMVar done
1405testHandleMessageIf :: TestTransport -> Assertion
1406testHandleMessageIf TestTransport{..} = do
1407 localNode <- newLocalNode testTransport initRemoteTable
1408 done <- newEmptyMVar
1409 _ <- forkProcess localNode $ do
1410 self <- getSelfPid
1411 send self (5 :: Integer, 10 :: Integer)
1412 msg <- receiveWait [ matchMessage return ]
1413 handleMessageIf msg
1414 (\() -> True)
1415 (\() -> die "whoops") >>= maybe (return ())
1416 (const $ die "Expected Mismatch")
1417 handleMessageIf msg (\(x :: Integer, y :: Integer) -> x == 5 && y == 10)
1418 (\input -> liftIO $ putMVar done input)
1419 return ()
1421 result <- takeMVar done
1422 expectThat result $ equalTo (5, 10)
1424testCatches :: TestTransport -> Assertion
1425testCatches TestTransport{..} = do
1426 localNode <- newLocalNode testTransport initRemoteTable
1427 done <- newEmptyMVar
1429 _ <- forkProcess localNode $ do
1430 node <- getSelfNode
1431 (liftIO $ throwIO (ProcessLinkException (nullProcessId node) DiedNormal))
1432 `catches` [
1433 Handler (\(ProcessLinkException _ _) -> liftIO $ putMVar done ())
1436 takeMVar done
1438testMaskRestoreScope :: TestTransport -> Assertion
1439testMaskRestoreScope TestTransport{..} = do
1440 localNode <- newLocalNode testTransport initRemoteTable
1441 parentPid <- newEmptyMVar :: IO (MVar ProcessId)
1442 spawnedPid <- newEmptyMVar :: IO (MVar ProcessId)
1444 void $ runProcess localNode $ mask $ \unmask -> do
1445 getSelfPid >>= liftIO . putMVar parentPid
1446 void $ spawnLocal $ unmask (getSelfPid >>= liftIO . putMVar spawnedPid)
1448 parent <- liftIO $ takeMVar parentPid
1449 child <- liftIO $ takeMVar spawnedPid
1450 expectThat parent $ isNot $ equalTo child
1452testDie :: TestTransport -> Assertion
1453testDie TestTransport{..} = do
1454 localNode <- newLocalNode testTransport initRemoteTable
1455 done <- newEmptyMVar
1457 _ <- forkProcess localNode $ do
1458 (die ("foobar", 123 :: Int))
1459 `catchExit` \_from reason -> do
1460 -- TODO: should verify that 'from' has the right value
1461 let res = reason == ("foobar", 123 :: Int)
1462 liftIO $ putMVar done ()
1463 if res
1464 then return ()
1465 else die "Something went horribly wrong"
1467 takeMVar done
1469testPrettyExit :: TestTransport -> Assertion
1470testPrettyExit TestTransport{..} = do
1471 localNode <- newLocalNode testTransport initRemoteTable
1472 done <- newEmptyMVar
1474 _ <- forkProcess localNode $ do
1475 (die "timeout")
1476 `catch` \ex@(ProcessExitException from _) ->
1477 let expected = "exit-from=" ++ (show from)
1478 in do
1479 let res = (show ex) == expected
1480 liftIO $ putMVar done ()
1481 if res
1482 then return ()
1483 else die "Something went horribly wrong"
1485 takeMVar done
1487testExitLocal :: TestTransport -> Assertion
1488testExitLocal TestTransport{..} = do
1489 localNode <- newLocalNode testTransport initRemoteTable
1490 supervisedDone <- newEmptyMVar
1491 supervisorDone <- newEmptyMVar
1492 -- XXX: we guarantee that exception handler will be set up
1493 -- regardless if forkProcess preserve masking state or not.
1494 handlerSetUp <- newEmptyMVar
1496 pid <- forkProcess localNode $ do
1497 (liftIO (putMVar handlerSetUp ()) >> expect) `catchExit` \_from reason -> do
1498 -- TODO: should verify that 'from' has the right value
1499 res <- return $ reason == "TestExit"
1500 liftIO $ putMVar supervisedDone ()
1501 if res
1502 then return ()
1503 else die "Something went horribly wrong"
1505 runProcess localNode $ do
1506 liftIO $ takeMVar handlerSetUp
1507 ref <- monitor pid
1508 exit pid "TestExit"
1509 -- This time the client catches the exception, so it dies normally
1510 mn <- expect
1511 case mn of
1512 ProcessMonitorNotification ref' pid' DiedNormal -> do
1513 let res = ref == ref' && pid == pid'
1514 liftIO $ putMVar supervisorDone ()
1515 if res
1516 then return ()
1517 else die "Something went horribly wrong"
1518 _ -> die "Something went horribly wrong"
1520 takeMVar supervisedDone
1521 takeMVar supervisorDone
1523testExitRemote :: TestTransport -> Assertion
1524testExitRemote TestTransport{..} = do
1525 node1 <- newLocalNode testTransport initRemoteTable
1526 node2 <- newLocalNode testTransport initRemoteTable
1527 supervisedDone <- newEmptyMVar
1528 supervisorDone <- newEmptyMVar
1530 pid <- forkProcess node1 $ do
1531 (receiveWait [] :: Process ()) -- block forever
1532 `catchExit` \_from reason -> do
1533 -- TODO: should verify that 'from' has the right value
1534 res <- return $ reason == "TestExit"
1535 liftIO $ putMVar supervisedDone ()
1536 if res
1537 then return ()
1538 else die "Something went horribly wrong"
1540 runProcess node2 $ do
1541 ref <- monitor pid
1542 exit pid "TestExit"
1543 mn <- expect
1544 case mn of
1545 ProcessMonitorNotification ref' pid' DiedNormal -> do
1546 res' <- return $ ref == ref' && pid == pid'
1547 liftIO $ putMVar supervisorDone ()
1548 if res'
1549 then return ()
1550 else die "Something went horribly wrong"
1551 _ -> die "Something went horribly wrong"
1553 takeMVar supervisedDone
1554 takeMVar supervisorDone
1556testRegistryMonitoring :: TestTransport -> Assertion
1557testRegistryMonitoring TestTransport{..} = do
1558 node1 <- newLocalNode testTransport initRemoteTable
1559 node2 <- newLocalNode testTransport initRemoteTable
1560 waitH <- newEmptyMVar
1562 let nid = localNodeId node2
1563 pid <- forkProcess node1 $ do
1564 getSelfPid >>= runUntilRegistered nid
1565 liftIO $ takeMVar waitH
1567 runProcess node2 $ do
1568 register regName pid
1569 res <- whereis regName
1570 us <- getSelfPid
1571 liftIO $ do
1572 putMVar waitH ()
1573 assertBool "expected (Just pid)" $ res == (Just pid)
1575 -- This delay isn't essential!
1576 -- The test case passes perfectly fine without it (feel free to comment out
1577 -- and see), however waiting a few seconds here, makes it much more likely
1578 -- that in delayUntilMaybeUnregistered we will hit the match case right
1579 -- away, and thus not be subjected to a 20 second delay. The value of 4
1580 -- seconds appears to work optimally on osx and across several linux distros
1581 -- running in virtual machines (which is essentially what we do in CI)
1582 void $ receiveTimeout 4000000 [ matchAny return ]
1584 -- This delay doesn't serve much purpose in the happy path, however if some
1585 -- future patch breaks the cooperative behaviour of node controllers viz
1586 -- remote process registration and notification taking place via ncEffectDied,
1587 -- there would be the possibility of a race in the test case should we attempt
1588 -- to evaluate `whereis regName` on node2 right away. In case the name is still
1589 -- erroneously registered, observing the 20 second delay (or lack of), could at
1590 -- least give a hint that something is wrong, and we give up our time slice
1591 -- so that there's a higher change the registrations have been cleaned up
1592 -- in either case.
1593 runProcess node2 $ delayUntilMaybeUnregistered nid pid
1595 regHere <- newEmptyMVar
1596 runProcess node2 $ whereis regName >>= liftIO . putMVar regHere
1597 res <- takeMVar regHere
1598 assertBool "expected Nothing, but process still registered" (res == Nothing)
1600 where
1601 runUntilRegistered nid us = do
1602 whereisRemoteAsync nid regName
1603 receiveWait [
1604 matchIf (\(WhereIsReply n (Just p)) -> n == regName && p == us)
1605 (const $ return ())
1608 delayUntilMaybeUnregistered nid p = do
1609 whereisRemoteAsync nid regName
1610 receiveTimeout 20000000 {- 20 sec delay -} [
1611 matchIf (\(WhereIsReply n p) -> n == regName && p == Nothing)
1612 (const $ return ())
1614 return ()
1616 regName = "testRegisterRemote"
1618testUnsafeSend :: TestTransport -> Assertion
1619testUnsafeSend TestTransport{..} = do
1620 serverAddr <- newEmptyMVar
1621 clientDone <- newEmptyMVar
1623 localNode <- newLocalNode testTransport initRemoteTable
1624 void $ forkProcess localNode $ do
1625 self <- getSelfPid
1626 liftIO $ putMVar serverAddr self
1627 clientAddr <- expect
1628 unsafeSend clientAddr ()
1630 void $ forkProcess localNode $ do
1631 serverPid <- liftIO $ takeMVar serverAddr
1632 getSelfPid >>= unsafeSend serverPid
1633 expect >>= liftIO . putMVar clientDone
1635 takeMVar clientDone
1637testUnsafeUSend :: TestTransport -> Assertion
1638testUnsafeUSend TestTransport{..} = do
1639 serverAddr <- newEmptyMVar
1640 clientDone <- newEmptyMVar
1642 localNode <- newLocalNode testTransport initRemoteTable
1643 void $ forkProcess localNode $ do
1644 self <- getSelfPid
1645 liftIO $ putMVar serverAddr self
1646 clientAddr <- expect
1647 unsafeUSend clientAddr ()
1649 void $ forkProcess localNode $ do
1650 serverPid <- liftIO $ takeMVar serverAddr
1651 getSelfPid >>= unsafeUSend serverPid
1652 expect >>= liftIO . putMVar clientDone
1654 takeMVar clientDone
1656testUnsafeNSend :: TestTransport -> Assertion
1657testUnsafeNSend TestTransport{..} = do
1658 clientDone <- newEmptyMVar
1660 localNode <- newLocalNode testTransport initRemoteTable
1662 pid <- forkProcess localNode $ do
1663 expect >>= liftIO . putMVar clientDone
1665 void $ runProcess localNode $ do
1666 register "foobar" pid
1667 unsafeNSend "foobar" ()
1669 takeMVar clientDone
1671testUnsafeNSendRemote :: TestTransport -> Assertion
1672testUnsafeNSendRemote TestTransport{..} = do
1673 clientDone <- newEmptyMVar
1675 localNode1 <- newLocalNode testTransport initRemoteTable
1676 localNode2 <- newLocalNode testTransport initRemoteTable
1678 _ <- forkProcess localNode1 $ do
1679 getSelfPid >>= register "foobar"
1680 liftIO $ putMVar clientDone ()
1681 expect >>= liftIO . putMVar clientDone
1683 takeMVar clientDone
1684 void $ runProcess localNode2 $ do
1685 unsafeNSendRemote (localNodeId localNode1) "foobar" ()
1687 takeMVar clientDone
1689testUnsafeSendChan :: TestTransport -> Assertion
1690testUnsafeSendChan TestTransport{..} = do
1691 serverAddr <- newEmptyMVar
1692 clientDone <- newEmptyMVar
1694 localNode <- newLocalNode testTransport initRemoteTable
1695 void $ forkProcess localNode $ do
1696 self <- getSelfPid
1697 liftIO $ putMVar serverAddr self
1698 sp <- expect
1699 unsafeSendChan sp ()
1701 void $ forkProcess localNode $ do
1702 serverPid <- liftIO $ takeMVar serverAddr
1703 (sp, rp) <- newChan
1704 unsafeSend serverPid sp
1705 receiveChan rp :: Process ()
1706 liftIO $ putMVar clientDone ()
1708 takeMVar clientDone
1710testCallLocal :: TestTransport -> Assertion
1711testCallLocal TestTransport{..} = do
1712 node <- newLocalNode testTransport initRemoteTable
1714 -- Testing that (/=) <$> getSelfPid <*> callLocal getSelfPid.
1715 result <- newEmptyMVar
1716 runProcess node $ do
1717 r <- (/=) <$> getSelfPid <*> callLocal getSelfPid
1718 liftIO $ putMVar result r
1720 takeMVar result >>= assertBool "Expected 'True'"
1722 -- Testing that when callLocal is interrupted, the worker is interrupted.
1723 ibox <- newIORef False
1724 runProcess node $ do
1725 keeper <- getSelfPid
1726 spawnLocal $ do
1727 caller <- getSelfPid
1728 send keeper caller
1729 onException
1730 (callLocal $ do
1731 onException (do send keeper caller
1732 expect)
1733 (do liftIO $ writeIORef ibox True))
1734 (send keeper ())
1735 caller <- expect
1736 exit caller "test"
1737 expect :: Process ()
1739 readIORef ibox >>= assertBool "Expected 'True'"
1741 -- Testing that when the worker raises an exception, the exception is propagated to the parent.
1742 ibox2 <- newIORef False
1743 runProcess node $ do
1744 r <- try (callLocal $ error "e" >> return ())
1745 liftIO $ writeIORef ibox2 $ case r of
1746 Left (ErrorCall "e") -> True
1747 _ -> False
1749 readIORef ibox >>= assertBool "Expected 'True'"
1751 -- Test that caller waits for the worker in correct situation
1752 ibox3 <- newIORef False
1753 result3 <- newEmptyMVar
1754 runProcess node $ do
1755 keeper <- getSelfPid
1756 spawnLocal $ do
1757 callLocal $
1758 (do us <- getSelfPid
1759 send keeper us
1760 () <- expect
1761 liftIO yield)
1762 `finally` (liftIO $ writeIORef ibox3 True)
1763 liftIO $ putMVar result3 =<< readIORef ibox3
1764 worker <- expect
1765 send worker ()
1767 takeMVar result3 >>= assertBool "Expected 'True'"
1769 -- Test that caller waits for the worker in case when caller gets an exception
1770 ibox4 <- newIORef False
1771 result4 <- newEmptyMVar
1772 runProcess node $ do
1773 keeper <- getSelfPid
1774 spawnLocal $ do
1775 caller <- getSelfPid
1776 callLocal
1777 ((do send keeper caller
1778 expect)
1779 `finally` (liftIO $ writeIORef ibox4 True))
1780 `finally` (liftIO $ putMVar result4 =<< readIORef ibox4)
1781 caller <- expect
1782 exit caller "hi!"
1784 takeMVar result4 >>= assertBool "Expected 'True'"
1785 -- XXX: Testing that when mask_ $ callLocal p runs p in masked state.
1787tests :: TestTransport -> IO [Test]
1788tests testtrans = return [
1789 testGroup "Basic features" [
1790 testCase "Ping" (testPing testtrans)
1791 , testCase "Math" (testMath testtrans)
1792 , testCase "Timeout" (testTimeout testtrans)
1793 , testCase "Timeout0" (testTimeout0 testtrans)
1794 , testCase "SendToTerminated" (testSendToTerminated testtrans)
1795 , testCase "TypedChannnels" (testTypedChannels testtrans)
1796 , testCase "Terminate" (testTerminate testtrans)
1797 , testCase "RegistryRemoteProcess" (testRegistryRemoteProcess testtrans)
1798 , testCase "RemoteRegistry" (testRemoteRegistry testtrans)
1799 , testCase "RemoteRegistryRemoteProcess" (testRemoteRegistryRemoteProcess testtrans)
1800 , testCase "SpawnLocal" (testSpawnLocal testtrans)
1801 , testCase "SpawnAsyncStrictness" (testSpawnAsyncStrictness testtrans)
1802 , testCase "HandleMessageIf" (testHandleMessageIf testtrans)
1803 , testCase "MatchAny" (testMatchAny testtrans)
1804 , testCase "MatchAnyHandle" (testMatchAnyHandle testtrans)
1805 , testCase "MatchAnyNoHandle" (testMatchAnyNoHandle testtrans)
1806 , testCase "MatchAnyIf" (testMatchAnyIf testtrans)
1807 , testCase "MatchMessageUnwrap" (testMatchMessageWithUnwrap testtrans)
1808 , testCase "ReceiveChanTimeout" (testReceiveChanTimeout testtrans)
1809 , testCase "ReceiveChanFeatures" (testReceiveChanFeatures testtrans)
1810 , testCase "ChanLifecycle" (testChanLifecycle testtrans)
1811 , testCase "KillLocal" (testKillLocal testtrans)
1812 , testCase "KillRemote" (testKillRemote testtrans)
1813 , testCase "Die" (testDie testtrans)
1814 , testCase "PrettyExit" (testPrettyExit testtrans)
1815 , testCase "CatchesExit" (testCatchesExit testtrans)
1816 , testCase "Catches" (testCatches testtrans)
1817 , testCase "MaskRestoreScope" (testMaskRestoreScope testtrans)
1818 , testCase "ExitLocal" (testExitLocal testtrans)
1819 , testCase "ExitRemote" (testExitRemote testtrans)
1820 , testCase "RegistryMonitoring" (testRegistryMonitoring testtrans)
1821 , testCase "TextCallLocal" (testCallLocal testtrans)
1822 -- Unsafe Primitives
1823 , testCase "TestUnsafeSend" (testUnsafeSend testtrans)
1824 , testCase "TestUnsafeUSend" (testUnsafeUSend testtrans)
1825 , testCase "TestUnsafeNSend" (testUnsafeNSend testtrans)
1826 , testCase "TestUnsafeNSendRemote" (testUnsafeNSendRemote testtrans)
1827 , testCase "TestUnsafeSendChan" (testUnsafeSendChan testtrans)
1828 -- usend
1829 , testCase "USend" (testUSend usend testtrans 50)
1830 , testCase "UForward"
1831 (testUSend (\p m -> uforward (createUnencodedMessage m) p)
1832 testtrans 50
1835 , testGroup "Monitoring and Linking" [
1836 -- Monitoring processes
1837 --
1838 -- The "missing" combinations in the list below don't make much sense, as
1839 -- we cannot guarantee that the monitor reply or link exception will not
1840 -- happen before the unmonitor or unlink
1841 testCase "MonitorNormalTermination" (testMonitorNormalTermination testtrans True False)
1842 , testCase "MonitorAbnormalTermination" (testMonitorAbnormalTermination testtrans True False)
1843 , testCase "MonitorLocalDeadProcess" (testMonitorLocalDeadProcess testtrans True False)
1844 , testCase "MonitorRemoteDeadProcess" (testMonitorRemoteDeadProcess testtrans True False)
1845 , testCase "MonitorDisconnect" (testMonitorDisconnect testtrans True False)
1846 , testCase "LinkUnreachable" (testMonitorUnreachable testtrans False False)
1847 , testCase "LinkNormalTermination" (testMonitorNormalTermination testtrans False False)
1848 , testCase "LinkAbnormalTermination" (testMonitorAbnormalTermination testtrans False False)
1849 , testCase "LinkLocalDeadProcess" (testMonitorLocalDeadProcess testtrans False False)
1850 , testCase "LinkRemoteDeadProcess" (testMonitorRemoteDeadProcess testtrans False False)
1851 , testCase "LinkDisconnect" (testMonitorDisconnect testtrans False False)
1852 , testCase "UnmonitorNormalTermination" (testMonitorNormalTermination testtrans True True)
1853 , testCase "UnmonitorAbnormalTermination" (testMonitorAbnormalTermination testtrans True True)
1854 , testCase "UnmonitorDisconnect" (testMonitorDisconnect testtrans True True)
1855 , testCase "UnlinkNormalTermination" (testMonitorNormalTermination testtrans False True)
1856 , testCase "UnlinkAbnormalTermination" (testMonitorAbnormalTermination testtrans False True)
1857 , testCase "UnlinkDisconnect" (testMonitorDisconnect testtrans False True)
1858 -- Monitoring nodes and channels
1859 , testCase "MonitorNode" (testMonitorNode testtrans)
1860 , testCase "MonitorLiveNode" (testMonitorLiveNode testtrans)
1861 , testCase "MonitorChannel" (testMonitorChannel testtrans)
1862 -- Reconnect
1865 -- Tests that fail occasionally and should be revised
1866 , testGroup "Flaky" [
1867 testCase "Reconnect" (testReconnect testtrans)
1868 , testCase "Registry" (testRegistry testtrans)
1869 , testCase "MergeChannels" (testMergeChannels testtrans)
1870 , testCase "MonitorUnreachable" (testMonitorUnreachable testtrans True False)