@@ -0,0 +1,1602 @@
1+module Control.Distributed.Process.Tests.CH (tests) where
2+
3+#if ! MIN_VERSION_base(4,6,0)
4+import Prelude hiding (catch)
5+#endif
6+
7+import Network.Transport.Test (TestTransport(..))
8+
9+import Data.Binary (Binary(..))
10+import Data.Typeable (Typeable)
11+import Data.Foldable (forM_)
12+import Data.IORef
13+ ( readIORef
14+ , writeIORef
15+ , newIORef
16+ )
17+import Control.Concurrent (forkIO, threadDelay, myThreadId, throwTo, ThreadId, yield)
18+import Control.Concurrent.MVar
19+ ( MVar
20+ , newEmptyMVar
21+ , putMVar
22+ , takeMVar
23+ , readMVar
24+ )
25+import Control.Monad (replicateM_, replicateM, forever, void, unless, join)
26+import Control.Exception (SomeException, throwIO, ErrorCall(..))
27+import qualified Control.Exception as Ex (catch)
28+import Control.Applicative ((<$>), (<*>), pure, (<|>))
29+import qualified Network.Transport as NT (closeEndPoint, EndPointAddress)
30+import Control.Distributed.Process
31+import Control.Distributed.Process.Internal.Types
32+ ( NodeId(nodeAddress)
33+ , LocalNode(localEndPoint)
34+ , ProcessExitException(..)
35+ , nullProcessId
36+ , createUnencodedMessage
37+ )
38+import Control.Distributed.Process.Node
39+import Control.Distributed.Process.Serializable (Serializable)
40+
41+import Test.HUnit (Assertion, assertFailure)
42+import Test.Framework (Test, testGroup)
43+import Test.Framework.Providers.HUnit (testCase)
44+import Control.Rematch hiding (match)
45+import Control.Rematch.Run (Match(..))
46+
47+newtype Ping = Ping ProcessId
48+ deriving (Typeable, Binary, Show)
49+
50+newtype Pong = Pong ProcessId
51+ deriving (Typeable, Binary, Show)
52+
53+--------------------------------------------------------------------------------
54+-- Supporting definitions --
55+--------------------------------------------------------------------------------
56+
57+expectThat :: a -> Matcher a -> Assertion
58+expectThat a matcher = case res of
59+ MatchSuccess -> return ()
60+ (MatchFailure msg) -> assertFailure msg
61+ where res = runMatch matcher a
62+
63+-- | Like fork, but throw exceptions in the child thread to the parent
64+forkTry :: IO () -> IO ThreadId
65+forkTry p = do
66+ tid <- myThreadId
67+ forkIO $ Ex.catch p (\e -> throwTo tid (e :: SomeException))
68+
69+-- | The ping server from the paper
70+ping :: Process ()
71+ping = do
72+ Pong partner <- expect
73+ self <- getSelfPid
74+ send partner (Ping self)
75+ ping
76+
77+-- | Quick and dirty synchronous version of whereisRemoteAsync
78+whereisRemote :: NodeId -> String -> Process (Maybe ProcessId)
79+whereisRemote nid string = do
80+ whereisRemoteAsync nid string
81+ WhereIsReply _ mPid <- expect
82+ return mPid
83+
84+data Add = Add ProcessId Double Double deriving (Typeable)
85+data Divide = Divide ProcessId Double Double deriving (Typeable)
86+data DivByZero = DivByZero deriving (Typeable)
87+
88+instance Binary Add where
89+ put (Add pid x y) = put pid >> put x >> put y
90+ get = Add <$> get <*> get <*> get
91+
92+instance Binary Divide where
93+ put (Divide pid x y) = put pid >> put x >> put y
94+ get = Divide <$> get <*> get <*> get
95+
96+instance Binary DivByZero where
97+ put DivByZero = return ()
98+ get = return DivByZero
99+
100+-- The math server from the paper
101+math :: Process ()
102+math = do
103+ receiveWait
104+ [ match (\(Add pid x y) -> send pid (x + y))
105+ , matchIf (\(Divide _ _ y) -> y /= 0)
106+ (\(Divide pid x y) -> send pid (x / y))
107+ , match (\(Divide pid _ _) -> send pid DivByZero)
108+ ]
109+ math
110+
111+-- | Monitor or link to a remote node
112+monitorOrLink :: Bool -- ^ 'True' for monitor, 'False' for link
113+ -> ProcessId -- ^ Process to monitor/link to
114+ -> Maybe (MVar ()) -- ^ MVar to signal on once the monitor has been set up
115+ -> Process (Maybe MonitorRef)
116+monitorOrLink mOrL pid mSignal = do
117+ result <- if mOrL then Just <$> monitor pid
118+ else link pid >> return Nothing
119+ -- Monitor is asynchronous, which usually does not matter but if we want a
120+ -- *specific* signal then it does. Therefore we wait until the MonitorRef is
121+ -- listed in the ProcessInfo and hope that this means the monitor has been set
122+ -- up.
123+ forM_ mSignal $ \signal -> do
124+ self <- getSelfPid
125+ spawnLocal $ do
126+ let waitForMOrL = do
127+ liftIO $ threadDelay 100000
128+ mpinfo <- getProcessInfo pid
129+ case mpinfo of
130+ Nothing -> waitForMOrL
131+ Just pinfo ->
132+ if mOrL then
133+ unless (result == lookup self (infoMonitors pinfo)) waitForMOrL
134+ else
135+ unless (elem self $ infoLinks pinfo) waitForMOrL
136+ waitForMOrL
137+ liftIO $ putMVar signal ()
138+ return result
139+
140+monitorTestProcess :: ProcessId -- Process to monitor/link to
141+ -> Bool -- 'True' for monitor, 'False' for link
142+ -> Bool -- Should we unmonitor?
143+ -> DiedReason -- Expected cause of death
144+ -> Maybe (MVar ()) -- Signal for 'monitor set up'
145+ -> MVar () -- Signal for successful termination
146+ -> Process ()
147+monitorTestProcess theirAddr mOrL un reason monitorSetup done =
148+ catch (do mRef <- monitorOrLink mOrL theirAddr monitorSetup
149+ case (un, mRef) of
150+ (True, Nothing) -> do
151+ unlink theirAddr
152+ liftIO $ putMVar done ()
153+ (True, Just ref) -> do
154+ unmonitor ref
155+ liftIO $ putMVar done ()
156+ (False, ref) -> do
157+ ProcessMonitorNotification ref' pid reason' <- expect
158+ True <- return $ Just ref' == ref && pid == theirAddr && mOrL && reason == reason'
159+ liftIO $ putMVar done ()
160+ )
161+ (\(ProcessLinkException pid reason') -> do
162+ True <- return $ pid == theirAddr && not mOrL && not un && reason == reason'
163+ liftIO $ putMVar done ()
164+ )
165+
166+--------------------------------------------------------------------------------
167+-- The tests proper --
168+--------------------------------------------------------------------------------
169+
170+-- | Basic ping test
171+testPing :: TestTransport -> Assertion
172+testPing TestTransport{..} = do
173+ serverAddr <- newEmptyMVar
174+ clientDone <- newEmptyMVar
175+
176+ -- Server
177+ forkIO $ do
178+ localNode <- newLocalNode testTransport initRemoteTable
179+ addr <- forkProcess localNode ping
180+ putMVar serverAddr addr
181+
182+ -- Client
183+ forkIO $ do
184+ localNode <- newLocalNode testTransport initRemoteTable
185+ pingServer <- readMVar serverAddr
186+
187+ let numPings = 10000
188+
189+ runProcess localNode $ do
190+ pid <- getSelfPid
191+ replicateM_ numPings $ do
192+ send pingServer (Pong pid)
193+ Ping _ <- expect
194+ return ()
195+
196+ putMVar clientDone ()
197+
198+ takeMVar clientDone
199+
200+-- | Monitor a process on an unreachable node
201+testMonitorUnreachable :: TestTransport -> Bool -> Bool -> Assertion
202+testMonitorUnreachable TestTransport{..} mOrL un = do
203+ deadProcess <- newEmptyMVar
204+ done <- newEmptyMVar
205+
206+ forkIO $ do
207+ localNode <- newLocalNode testTransport initRemoteTable
208+ addr <- forkProcess localNode expect
209+ closeLocalNode localNode
210+ putMVar deadProcess addr
211+
212+ forkIO $ do
213+ localNode <- newLocalNode testTransport initRemoteTable
214+ theirAddr <- readMVar deadProcess
215+ runProcess localNode $
216+ monitorTestProcess theirAddr mOrL un DiedDisconnect Nothing done
217+
218+ takeMVar done
219+
220+-- | Monitor a process which terminates normally
221+testMonitorNormalTermination :: TestTransport -> Bool -> Bool -> Assertion
222+testMonitorNormalTermination TestTransport{..} mOrL un = do
223+ monitorSetup <- newEmptyMVar
224+ monitoredProcess <- newEmptyMVar
225+ done <- newEmptyMVar
226+
227+ forkIO $ do
228+ localNode <- newLocalNode testTransport initRemoteTable
229+ addr <- forkProcess localNode $
230+ liftIO $ readMVar monitorSetup
231+ putMVar monitoredProcess addr
232+
233+ forkIO $ do
234+ localNode <- newLocalNode testTransport initRemoteTable
235+ theirAddr <- readMVar monitoredProcess
236+ runProcess localNode $
237+ monitorTestProcess theirAddr mOrL un DiedNormal (Just monitorSetup) done
238+
239+ takeMVar done
240+
241+-- | Monitor a process which terminates abnormally
242+testMonitorAbnormalTermination :: TestTransport -> Bool -> Bool -> Assertion
243+testMonitorAbnormalTermination TestTransport{..} mOrL un = do
244+ monitorSetup <- newEmptyMVar
245+ monitoredProcess <- newEmptyMVar
246+ done <- newEmptyMVar
247+
248+ let err = userError "Abnormal termination"
249+
250+ forkIO $ do
251+ localNode <- newLocalNode testTransport initRemoteTable
252+ addr <- forkProcess localNode . liftIO $ do
253+ readMVar monitorSetup
254+ throwIO err
255+ putMVar monitoredProcess addr
256+
257+ forkIO $ do
258+ localNode <- newLocalNode testTransport initRemoteTable
259+ theirAddr <- readMVar monitoredProcess
260+ runProcess localNode $
261+ monitorTestProcess theirAddr mOrL un (DiedException (show err)) (Just monitorSetup) done
262+
263+ takeMVar done
264+
265+-- | Monitor a local process that is already dead
266+testMonitorLocalDeadProcess :: TestTransport -> Bool -> Bool -> Assertion
267+testMonitorLocalDeadProcess TestTransport{..} mOrL un = do
268+ processAddr <- newEmptyMVar
269+ localNode <- newLocalNode testTransport initRemoteTable
270+ done <- newEmptyMVar
271+
272+ forkIO $ do
273+ addr <- forkProcess localNode $ return ()
274+ putMVar processAddr addr
275+
276+ forkIO $ do
277+ theirAddr <- readMVar processAddr
278+ runProcess localNode $ do
279+ monitor theirAddr
280+ -- wait for the process to die
281+ ProcessMonitorNotification _ _ _ <- expect
282+ monitorTestProcess theirAddr mOrL un DiedUnknownId Nothing done
283+
284+ takeMVar done
285+
286+-- | Monitor a remote process that is already dead
287+testMonitorRemoteDeadProcess :: TestTransport -> Bool -> Bool -> Assertion
288+testMonitorRemoteDeadProcess TestTransport{..} mOrL un = do
289+ processDead <- newEmptyMVar
290+ processAddr <- newEmptyMVar
291+ done <- newEmptyMVar
292+
293+ forkIO $ do
294+ localNode <- newLocalNode testTransport initRemoteTable
295+ addr <- forkProcess localNode . liftIO $ putMVar processDead ()
296+ putMVar processAddr addr
297+
298+ forkIO $ do
299+ localNode <- newLocalNode testTransport initRemoteTable
300+ theirAddr <- readMVar processAddr
301+ readMVar processDead
302+ runProcess localNode $ do
303+ monitorTestProcess theirAddr mOrL un DiedUnknownId Nothing done
304+
305+ takeMVar done
306+
307+-- | Monitor a process that becomes disconnected
308+testMonitorDisconnect :: TestTransport -> Bool -> Bool -> Assertion
309+testMonitorDisconnect TestTransport{..} mOrL un = do
310+ processAddr <- newEmptyMVar
311+ processAddr2 <- newEmptyMVar
312+ monitorSetup <- newEmptyMVar
313+ done <- newEmptyMVar
314+
315+ forkIO $ do
316+ localNode <- newLocalNode testTransport initRemoteTable
317+ addr <- forkProcess localNode $ expect
318+ addr2 <- forkProcess localNode $ return ()
319+ putMVar processAddr addr
320+ readMVar monitorSetup
321+ NT.closeEndPoint (localEndPoint localNode)
322+ putMVar processAddr2 addr2
323+
324+ forkIO $ do
325+ localNode <- newLocalNode testTransport initRemoteTable
326+ theirAddr <- readMVar processAddr
327+ forkProcess localNode $ do
328+ lc <- liftIO $ readMVar processAddr2
329+ send lc ()
330+ runProcess localNode $ do
331+ monitorTestProcess theirAddr mOrL un DiedDisconnect (Just monitorSetup) done
332+
333+ takeMVar done
334+
335+-- | Test the math server (i.e., receiveWait)
336+testMath :: TestTransport -> Assertion
337+testMath TestTransport{..} = do
338+ serverAddr <- newEmptyMVar
339+ clientDone <- newEmptyMVar
340+
341+ -- Server
342+ forkIO $ do
343+ localNode <- newLocalNode testTransport initRemoteTable
344+ addr <- forkProcess localNode math
345+ putMVar serverAddr addr
346+
347+ -- Client
348+ forkIO $ do
349+ localNode <- newLocalNode testTransport initRemoteTable
350+ mathServer <- readMVar serverAddr
351+
352+ runProcess localNode $ do
353+ pid <- getSelfPid
354+ send mathServer (Add pid 1 2)
355+ 3 <- expect :: Process Double
356+ send mathServer (Divide pid 8 2)
357+ 4 <- expect :: Process Double
358+ send mathServer (Divide pid 8 0)
359+ DivByZero <- expect
360+ liftIO $ putMVar clientDone ()
361+
362+ takeMVar clientDone
363+
364+-- | Send first message (i.e. connect) to an already terminated process
365+-- (without monitoring); then send another message to a second process on
366+-- the same remote node (we're checking that the remote node did not die)
367+testSendToTerminated :: TestTransport -> Assertion
368+testSendToTerminated TestTransport{..} = do
369+ serverAddr1 <- newEmptyMVar
370+ serverAddr2 <- newEmptyMVar
371+ clientDone <- newEmptyMVar
372+
373+ forkIO $ do
374+ terminated <- newEmptyMVar
375+ localNode <- newLocalNode testTransport initRemoteTable
376+ addr1 <- forkProcess localNode $ liftIO $ putMVar terminated ()
377+ addr2 <- forkProcess localNode $ ping
378+ readMVar terminated
379+ putMVar serverAddr1 addr1
380+ putMVar serverAddr2 addr2
381+
382+ forkIO $ do
383+ localNode <- newLocalNode testTransport initRemoteTable
384+ server1 <- readMVar serverAddr1
385+ server2 <- readMVar serverAddr2
386+ runProcess localNode $ do
387+ pid <- getSelfPid
388+ send server1 "Hi"
389+ send server2 (Pong pid)
390+ Ping pid' <- expect
391+ True <- return $ pid' == server2
392+ liftIO $ putMVar clientDone ()
393+
394+ takeMVar clientDone
395+
396+-- | Test (non-zero) timeout
397+testTimeout :: TestTransport -> Assertion
398+testTimeout TestTransport{..} = do
399+ localNode <- newLocalNode testTransport initRemoteTable
400+ done <- newEmptyMVar
401+
402+ runProcess localNode $ do
403+ Nothing <- receiveTimeout 1000000 [match (\(Add _ _ _) -> return ())]
404+ liftIO $ putMVar done ()
405+
406+ takeMVar done
407+
408+-- | Test zero timeout
409+testTimeout0 :: TestTransport -> Assertion
410+testTimeout0 TestTransport{..} = do
411+ serverAddr <- newEmptyMVar
412+ clientDone <- newEmptyMVar
413+ messagesSent <- newEmptyMVar
414+
415+ forkIO $ do
416+ localNode <- newLocalNode testTransport initRemoteTable
417+ addr <- forkProcess localNode $ do
418+ liftIO $ readMVar messagesSent >> threadDelay 1000000
419+ -- Variation on the venerable ping server which uses a zero timeout
420+ -- Since we wait for all messages to be sent before doing this receive,
421+ -- we should nevertheless find the right message immediately
422+ Just partner <- receiveTimeout 0 [match (\(Pong partner) -> return partner)]
423+ self <- getSelfPid
424+ send partner (Ping self)
425+ putMVar serverAddr addr
426+
427+ forkIO $ do
428+ localNode <- newLocalNode testTransport initRemoteTable
429+ server <- readMVar serverAddr
430+ runProcess localNode $ do
431+ pid <- getSelfPid
432+ -- Send a bunch of messages. A large number of messages that the server
433+ -- is not interested in, and then a single message that it wants
434+ replicateM_ 10000 $ send server "Irrelevant message"
435+ send server (Pong pid)
436+ liftIO $ putMVar messagesSent ()
437+ Ping _ <- expect
438+ liftIO $ putMVar clientDone ()
439+
440+ takeMVar clientDone
441+
442+-- | Test typed channels
443+testTypedChannels :: TestTransport -> Assertion
444+testTypedChannels TestTransport{..} = do
445+ serverChannel <- newEmptyMVar :: IO (MVar (SendPort (SendPort Bool, Int)))
446+ clientDone <- newEmptyMVar
447+
448+ forkIO $ do
449+ localNode <- newLocalNode testTransport initRemoteTable
450+ forkProcess localNode $ do
451+ (serverSendPort, rport) <- newChan
452+ liftIO $ putMVar serverChannel serverSendPort
453+ (clientSendPort, i) <- receiveChan rport
454+ sendChan clientSendPort (even i)
455+ return ()
456+
457+ forkIO $ do
458+ localNode <- newLocalNode testTransport initRemoteTable
459+ serverSendPort <- readMVar serverChannel
460+ runProcess localNode $ do
461+ (clientSendPort, rport) <- newChan
462+ sendChan serverSendPort (clientSendPort, 5)
463+ False <- receiveChan rport
464+ liftIO $ putMVar clientDone ()
465+
466+ takeMVar clientDone
467+
468+-- | Test merging receive ports
469+testMergeChannels :: TestTransport -> Assertion
470+testMergeChannels TestTransport{..} = do
471+ localNode <- newLocalNode testTransport initRemoteTable
472+ testFlat localNode True "aaabbbccc"
473+ testFlat localNode False "abcabcabc"
474+ testNested localNode True True "aaabbbcccdddeeefffggghhhiii"
475+ testNested localNode True False "adgadgadgbehbehbehcficficfi"
476+ testNested localNode False True "abcabcabcdefdefdefghighighi"
477+ testNested localNode False False "adgbehcfiadgbehcfiadgbehcfi"
478+ testBlocked localNode True
479+ testBlocked localNode False
480+ where
481+ -- Single layer of merging
482+ testFlat :: LocalNode -> Bool -> String -> IO ()
483+ testFlat localNode biased expected = do
484+ done <- newEmptyMVar
485+ forkProcess localNode $ do
486+ rs <- mapM charChannel "abc"
487+ m <- mergePorts biased rs
488+ xs <- replicateM 9 $ receiveChan m
489+ True <- return $ xs == expected
490+ liftIO $ putMVar done ()
491+ takeMVar done
492+
493+ -- Two layers of merging
494+ testNested :: LocalNode -> Bool -> Bool -> String -> IO ()
495+ testNested localNode biasedInner biasedOuter expected = do
496+ done <- newEmptyMVar
497+ forkProcess localNode $ do
498+ rss <- mapM (mapM charChannel) ["abc", "def", "ghi"]
499+ ms <- mapM (mergePorts biasedInner) rss
500+ m <- mergePorts biasedOuter ms
501+ xs <- replicateM (9 * 3) $ receiveChan m
502+ True <- return $ xs == expected
503+ liftIO $ putMVar done ()
504+ takeMVar done
505+
506+ -- Test that if no messages are (immediately) available, the scheduler makes no difference
507+ testBlocked :: LocalNode -> Bool -> IO ()
508+ testBlocked localNode biased = do
509+ vs <- replicateM 3 newEmptyMVar
510+ done <- newEmptyMVar
511+
512+ forkProcess localNode $ do
513+ [sa, sb, sc] <- liftIO $ mapM readMVar vs
514+ mapM_ ((>> liftIO (threadDelay 10000)) . uncurry sendChan)
515+ [ -- a, b, c
516+ (sa, 'a')
517+ , (sb, 'b')
518+ , (sc, 'c')
519+ -- a, c, b
520+ , (sa, 'a')
521+ , (sc, 'c')
522+ , (sb, 'b')
523+ -- b, a, c
524+ , (sb, 'b')
525+ , (sa, 'a')
526+ , (sc, 'c')
527+ -- b, c, a
528+ , (sb, 'b')
529+ , (sc, 'c')
530+ , (sa, 'a')
531+ -- c, a, b
532+ , (sc, 'c')
533+ , (sa, 'a')
534+ , (sb, 'b')
535+ -- c, b, a
536+ , (sc, 'c')
537+ , (sb, 'b')
538+ , (sa, 'a')
539+ ]
540+
541+ forkProcess localNode $ do
542+ (ss, rs) <- unzip <$> replicateM 3 newChan
543+ liftIO $ mapM_ (uncurry putMVar) $ zip vs ss
544+ m <- mergePorts biased rs
545+ xs <- replicateM (6 * 3) $ receiveChan m
546+ True <- return $ xs == "abcacbbacbcacabcba"
547+ liftIO $ putMVar done ()
548+
549+ takeMVar done
550+
551+ mergePorts :: Serializable a => Bool -> [ReceivePort a] -> Process (ReceivePort a)
552+ mergePorts True = mergePortsBiased
553+ mergePorts False = mergePortsRR
554+
555+ charChannel :: Char -> Process (ReceivePort Char)
556+ charChannel c = do
557+ (sport, rport) <- newChan
558+ replicateM_ 3 $ sendChan sport c
559+ liftIO $ threadDelay 10000 -- Make sure messages have been sent
560+ return rport
561+
562+testTerminate :: TestTransport -> Assertion
563+testTerminate TestTransport{..} = do
564+ localNode <- newLocalNode testTransport initRemoteTable
565+ done <- newEmptyMVar
566+
567+ pid <- forkProcess localNode $ do
568+ liftIO $ threadDelay 100000
569+ terminate
570+
571+ runProcess localNode $ do
572+ ref <- monitor pid
573+ ProcessMonitorNotification ref' pid' (DiedException ex) <- expect
574+ True <- return $ ref == ref' && pid == pid' && ex == show ProcessTerminationException
575+ liftIO $ putMVar done ()
576+
577+ takeMVar done
578+
579+testMonitorNode :: TestTransport -> Assertion
580+testMonitorNode TestTransport{..} = do
581+ [node1, node2] <- replicateM 2 $ newLocalNode testTransport initRemoteTable
582+ done <- newEmptyMVar
583+
584+ closeLocalNode node1
585+
586+ runProcess node2 $ do
587+ ref <- monitorNode (localNodeId node1)
588+ NodeMonitorNotification ref' nid DiedDisconnect <- expect
589+ True <- return $ ref == ref' && nid == localNodeId node1
590+ liftIO $ putMVar done ()
591+
592+ takeMVar done
593+
594+testMonitorLiveNode :: TestTransport -> Assertion
595+testMonitorLiveNode TestTransport{..} = do
596+ [node1, node2] <- replicateM 2 $ newLocalNode testTransport initRemoteTable
597+ ready <- newEmptyMVar
598+ readyr <- newEmptyMVar
599+ done <- newEmptyMVar
600+
601+ p <- forkProcess node1 $ return ()
602+ forkProcess node2 $ do
603+ ref <- monitorNode (localNodeId node1)
604+ liftIO $ putMVar ready ()
605+ liftIO $ takeMVar readyr
606+ send p ()
607+ NodeMonitorNotification ref' nid _ <- expect
608+ True <- return $ ref == ref' && nid == localNodeId node1
609+ liftIO $ putMVar done ()
610+
611+ takeMVar ready
612+ closeLocalNode node1
613+ putMVar readyr ()
614+
615+ takeMVar done
616+
617+testMonitorChannel :: TestTransport -> Assertion
618+testMonitorChannel TestTransport{..} = do
619+ [node1, node2] <- replicateM 2 $ newLocalNode testTransport initRemoteTable
620+ gotNotification <- newEmptyMVar
621+
622+ pid <- forkProcess node1 $ do
623+ sport <- expect :: Process (SendPort ())
624+ ref <- monitorPort sport
625+ PortMonitorNotification ref' port' reason <- expect
626+ -- reason might be DiedUnknownId if the receive port is GCed before the
627+ -- monitor is established (TODO: not sure that this is reasonable)
628+ return $ ref' == ref && port' == sendPortId sport && (reason == DiedNormal || reason == DiedUnknownId)
629+ liftIO $ putMVar gotNotification ()
630+
631+ runProcess node2 $ do
632+ (sport, _) <- newChan :: Process (SendPort (), ReceivePort ())
633+ send pid sport
634+ liftIO $ threadDelay 100000
635+
636+ takeMVar gotNotification
637+
638+testRegistry :: TestTransport -> Assertion
639+testRegistry TestTransport{..} = do
640+ node <- newLocalNode testTransport initRemoteTable
641+ done <- newEmptyMVar
642+
643+ pingServer <- forkProcess node ping
644+ deadProcess <- forkProcess node (return ())
645+
646+ runProcess node $ do
647+ register "ping" pingServer
648+ Just pid <- whereis "ping"
649+ True <- return $ pingServer == pid
650+ us <- getSelfPid
651+ nsend "ping" (Pong us)
652+ Ping pid' <- expect
653+ True <- return $ pingServer == pid'
654+ Left (ProcessRegistrationException "dead" Nothing) <- try $ register "dead" deadProcess
655+ Left (ProcessRegistrationException "ping" (Just x)) <- try $ register "ping" deadProcess
656+ True <- return $ x == pingServer
657+ Left (ProcessRegistrationException "dead" Nothing) <- try $ unregister "dead"
658+ liftIO $ putMVar done ()
659+
660+ takeMVar done
661+
662+testRegistryRemoteProcess :: TestTransport -> Assertion
663+testRegistryRemoteProcess TestTransport{..} = do
664+ node1 <- newLocalNode testTransport initRemoteTable
665+ node2 <- newLocalNode testTransport initRemoteTable
666+ done <- newEmptyMVar
667+
668+ pingServer <- forkProcess node1 ping
669+
670+ runProcess node2 $ do
671+ register "ping" pingServer
672+ Just pid <- whereis "ping"
673+ True <- return $ pingServer == pid
674+ us <- getSelfPid
675+ nsend "ping" (Pong us)
676+ Ping pid' <- expect
677+ True <- return $ pingServer == pid'
678+ liftIO $ putMVar done ()
679+
680+ takeMVar done
681+
682+testRemoteRegistry :: TestTransport -> Assertion
683+testRemoteRegistry TestTransport{..} = do
684+ node1 <- newLocalNode testTransport initRemoteTable
685+ node2 <- newLocalNode testTransport initRemoteTable
686+ done <- newEmptyMVar
687+
688+ pingServer <- forkProcess node1 ping
689+ deadProcess <- forkProcess node1 (return ())
690+
691+ runProcess node2 $ do
692+ let nid1 = localNodeId node1
693+ registerRemoteAsync nid1 "ping" pingServer
694+ receiveWait [
695+ matchIf (\(RegisterReply label' _ (Just pid)) ->
696+ "ping" == label' && pid == pingServer)
697+ (\(RegisterReply _ _ _) -> return ()) ]
698+
699+ Just pid <- whereisRemote nid1 "ping"
700+ True <- return $ pingServer == pid
701+ us <- getSelfPid
702+ nsendRemote nid1 "ping" (Pong us)
703+ Ping pid' <- expect
704+ True <- return $ pingServer == pid'
705+
706+ -- test that if process was not registered Nothing is returned
707+ -- in owner field.
708+ registerRemoteAsync nid1 "dead" deadProcess
709+ receiveWait [
710+ matchIf (\(RegisterReply label' False Nothing) -> "dead" == label')
711+ (\(RegisterReply _ _ _) -> return ()) ]
712+ registerRemoteAsync nid1 "ping" deadProcess
713+ receiveWait [
714+ matchIf (\(RegisterReply label' False (Just pid)) ->
715+ "ping" == label' && pid == pingServer)
716+ (\(RegisterReply _ _ _) -> return ()) ]
717+ unregisterRemoteAsync nid1 "dead"
718+ receiveWait [
719+ matchIf (\(RegisterReply label' False Nothing) ->
720+ "dead" == label' && pid == pingServer)
721+ (\(RegisterReply _ _ _) -> return ()) ]
722+ liftIO $ putMVar done ()
723+
724+ takeMVar done
725+
726+testRemoteRegistryRemoteProcess :: TestTransport -> Assertion
727+testRemoteRegistryRemoteProcess TestTransport{..} = do
728+ node1 <- newLocalNode testTransport initRemoteTable
729+ node2 <- newLocalNode testTransport initRemoteTable
730+ done <- newEmptyMVar
731+
732+ pingServer <- forkProcess node2 ping
733+
734+ runProcess node2 $ do
735+ let nid1 = localNodeId node1
736+ registerRemoteAsync nid1 "ping" pingServer
737+ receiveWait [
738+ matchIf (\(RegisterReply label' _ _) -> "ping" == label')
739+ (\(RegisterReply _ _ _) -> return ()) ]
740+ Just pid <- whereisRemote nid1 "ping"
741+ True <- return $ pingServer == pid
742+ us <- getSelfPid
743+ nsendRemote nid1 "ping" (Pong us)
744+ Ping pid' <- expect
745+ True <- return $ pingServer == pid'
746+ liftIO $ putMVar done ()
747+
748+ takeMVar done
749+
750+testSpawnLocal :: TestTransport -> Assertion
751+testSpawnLocal TestTransport{..} = do
752+ node <- newLocalNode testTransport initRemoteTable
753+ done <- newEmptyMVar
754+
755+ runProcess node $ do
756+ us <- getSelfPid
757+
758+ pid <- spawnLocal $ do
759+ sport <- expect
760+ sendChan sport (1234 :: Int)
761+
762+ sport <- spawnChannelLocal $ \rport -> do
763+ (1234 :: Int) <- receiveChan rport
764+ send us ()
765+
766+ send pid sport
767+ () <- expect
768+ liftIO $ putMVar done ()
769+
770+ takeMVar done
771+
772+testSpawnAsyncStrictness :: TestTransport -> Assertion
773+testSpawnAsyncStrictness TestTransport{..} = do
774+ node <- newLocalNode testTransport initRemoteTable
775+ done <- newEmptyMVar
776+
777+ runProcess node $ do
778+ here <-getSelfNode
779+
780+ ev <- try $ spawnAsync here (error "boom")
781+ liftIO $ case ev of
782+ Right _ -> putMVar done (error "Exception didn't fire")
783+ Left (_::SomeException) -> putMVar done (return ())
784+
785+ join $ takeMVar done
786+
787+testReconnect :: TestTransport -> Assertion
788+testReconnect TestTransport{..} = do
789+ [node1, node2] <- replicateM 2 $ newLocalNode testTransport initRemoteTable
790+ let nid1 = localNodeId node1
791+ processA <- newEmptyMVar
792+ [sendTestOk, registerTestOk] <- replicateM 2 newEmptyMVar
793+
794+ forkProcess node1 $ do
795+ us <- getSelfPid
796+ liftIO $ putMVar processA us
797+ msg1 <- expect
798+ msg2 <- expect
799+ True <- return $ msg1 == "message 1" && msg2 == "message 3"
800+ liftIO $ putMVar sendTestOk ()
801+
802+ forkProcess node2 $ do
803+ {-
804+ - Make sure there is no implicit reconnect on normal message sending
805+ -}
806+
807+ them <- liftIO $ readMVar processA
808+ send them "message 1" >> liftIO (threadDelay 100000)
809+
810+ -- Simulate network failure
811+ liftIO $ syncBreakConnection testBreakConnection node1 node2
812+
813+
814+ -- Should not arrive
815+ send them "message 2"
816+
817+ -- Should arrive
818+ reconnect them
819+ send them "message 3"
820+
821+ liftIO $ takeMVar sendTestOk
822+
823+ {-
824+ - Test that there *is* implicit reconnect on node controller messages
825+ -}
826+
827+ us <- getSelfPid
828+ registerRemoteAsync nid1 "a" us -- registerRemote is asynchronous
829+ receiveWait [
830+ matchIf (\(RegisterReply label' _ _) -> "a" == label')
831+ (\(RegisterReply _ _ _) -> return ()) ]
832+
833+ Just _ <- whereisRemote nid1 "a"
834+
835+
836+ -- Simulate network failure
837+ liftIO $ syncBreakConnection testBreakConnection node1 node2
838+
839+ -- This will happen due to implicit reconnect
840+ registerRemoteAsync nid1 "b" us
841+ receiveWait [
842+ matchIf (\(RegisterReply label' _ _) -> "b" == label')
843+ (\(RegisterReply _ _ _) -> return ()) ]
844+
845+ -- Should happen
846+ registerRemoteAsync nid1 "c" us
847+ receiveWait [
848+ matchIf (\(RegisterReply label' _ _) -> "c" == label')
849+ (\(RegisterReply _ _ _) -> return ()) ]
850+
851+ -- Check
852+ Nothing <- whereisRemote nid1 "a" -- this will fail because the name is removed when the node is disconnected
853+ Just _ <- whereisRemote nid1 "b" -- this will suceed because the value is set after thereconnect
854+ Just _ <- whereisRemote nid1 "c"
855+
856+ liftIO $ putMVar registerTestOk ()
857+
858+ takeMVar registerTestOk
859+
860+-- | Tests that unreliable messages arrive sorted even when there are connection
861+-- failures.
862+testUSend :: (ProcessId -> Int -> Process ())
863+ -> TestTransport -> Int -> Assertion
864+testUSend usendPrim TestTransport{..} numMessages = do
865+ [node1, node2] <- replicateM 2 $ newLocalNode testTransport initRemoteTable
866+ let nid1 = localNodeId node1
867+ nid2 = localNodeId node2
868+ processA <- newEmptyMVar
869+ usendTestOk <- newEmptyMVar
870+
871+ forkProcess node1 $ flip catch (\e -> liftIO $ print (e :: SomeException) ) $ do
872+ us <- getSelfPid
873+ liftIO $ putMVar processA us
874+ them <- expect
875+ send them ()
876+ _ <- monitor them
877+ let -- Collects messages from 'them' until the sender dies.
878+ -- Disconnection notifications are ignored.
879+ receiveMessages :: Process [Int]
880+ receiveMessages = receiveWait
881+ [ match $ \mn -> case mn of
882+ ProcessMonitorNotification _ _ DiedDisconnect -> do
883+ monitor them
884+ receiveMessages
885+ _ -> return []
886+ , match $ \i -> fmap (i :) receiveMessages
887+ ]
888+ msgs <- receiveMessages
889+ let -- Checks that the input list is sorted.
890+ isSorted :: [Int] -> Bool
891+ isSorted (x : xs@(y : _)) = x <= y && isSorted xs
892+ isSorted _ = True
893+ -- The list can't be null since there are no failures after sending
894+ -- the last message.
895+ True <- return $ isSorted msgs && not (null msgs)
896+ liftIO $ putMVar usendTestOk ()
897+
898+ forkProcess node2 $ do
899+ them <- liftIO $ readMVar processA
900+ getSelfPid >>= send them
901+ () <- expect
902+ forM_ [1..numMessages] $ \i -> do
903+ liftIO $ testBreakConnection (nodeAddress nid1) (nodeAddress nid2)
904+ usendPrim them i
905+ liftIO (threadDelay 30000)
906+
907+ takeMVar usendTestOk
908+
909+-- | Test 'matchAny'. This repeats the 'testMath' but with a proxy server
910+-- in between
911+testMatchAny :: TestTransport -> Assertion
912+testMatchAny TestTransport{..} = do
913+ proxyAddr <- newEmptyMVar
914+ clientDone <- newEmptyMVar
915+
916+ -- Math server
917+ forkIO $ do
918+ localNode <- newLocalNode testTransport initRemoteTable
919+ mathServer <- forkProcess localNode math
920+ proxyServer <- forkProcess localNode $ forever $ do
921+ msg <- receiveWait [ matchAny return ]
922+ forward msg mathServer
923+ putMVar proxyAddr proxyServer
924+
925+ -- Client
926+ forkIO $ do
927+ localNode <- newLocalNode testTransport initRemoteTable
928+ mathServer <- readMVar proxyAddr
929+
930+ runProcess localNode $ do
931+ pid <- getSelfPid
932+ send mathServer (Add pid 1 2)
933+ 3 <- expect :: Process Double
934+ send mathServer (Divide pid 8 2)
935+ 4 <- expect :: Process Double
936+ send mathServer (Divide pid 8 0)
937+ DivByZero <- expect
938+ liftIO $ putMVar clientDone ()
939+
940+ takeMVar clientDone
941+
942+-- | Test 'matchAny'. This repeats the 'testMath' but with a proxy server
943+-- in between, however we block 'Divide' requests ....
944+testMatchAnyHandle :: TestTransport -> Assertion
945+testMatchAnyHandle TestTransport{..} = do
946+ proxyAddr <- newEmptyMVar
947+ clientDone <- newEmptyMVar
948+
949+ -- Math server
950+ forkIO $ do
951+ localNode <- newLocalNode testTransport initRemoteTable
952+ mathServer <- forkProcess localNode math
953+ proxyServer <- forkProcess localNode $ forever $ do
954+ receiveWait [
955+ matchAny (maybeForward mathServer)
956+ ]
957+ putMVar proxyAddr proxyServer
958+
959+ -- Client
960+ forkIO $ do
961+ localNode <- newLocalNode testTransport initRemoteTable
962+ mathServer <- readMVar proxyAddr
963+
964+ runProcess localNode $ do
965+ pid <- getSelfPid
966+ send mathServer (Add pid 1 2)
967+ 3 <- expect :: Process Double
968+ send mathServer (Divide pid 8 2)
969+ Nothing <- (expectTimeout 100000) :: Process (Maybe Double)
970+ liftIO $ putMVar clientDone ()
971+
972+ takeMVar clientDone
973+ where maybeForward :: ProcessId -> Message -> Process (Maybe ())
974+ maybeForward s msg =
975+ handleMessage msg (\m@(Add _ _ _) -> send s m)
976+
977+testMatchAnyNoHandle :: TestTransport -> Assertion
978+testMatchAnyNoHandle TestTransport{..} = do
979+ addr <- newEmptyMVar
980+ clientDone <- newEmptyMVar
981+ serverDone <- newEmptyMVar
982+
983+ -- Math server
984+ forkIO $ do
985+ localNode <- newLocalNode testTransport initRemoteTable
986+ server <- forkProcess localNode $ forever $ do
987+ receiveWait [
988+ matchAnyIf
989+ -- the condition has type `Add -> Bool`
990+ (\(Add _ _ _) -> True)
991+ -- the match `AbstractMessage -> Process ()` will succeed!
992+ (\m -> do
993+ -- `String -> Process ()` does *not* match the input types however
994+ r <- (handleMessage m (\(_ :: String) -> die "NONSENSE" ))
995+ case r of
996+ Nothing -> return ()
997+ Just _ -> die "NONSENSE")
998+ ]
999+ -- we *must* have removed the message from our mailbox though!!!
1000+ Nothing <- receiveTimeout 100000 [ match (\(Add _ _ _) -> return ()) ]
1001+ liftIO $ putMVar serverDone ()
1002+ putMVar addr server
1003+
1004+ -- Client
1005+ forkIO $ do
1006+ localNode <- newLocalNode testTransport initRemoteTable
1007+ server <- readMVar addr
1008+
1009+ runProcess localNode $ do
1010+ pid <- getSelfPid
1011+ send server (Add pid 1 2)
1012+ -- we only care about the client having sent a message, so we're done
1013+ liftIO $ putMVar clientDone ()
1014+
1015+ takeMVar clientDone
1016+ takeMVar serverDone
1017+
1018+-- | Test 'matchAnyIf'. We provide an /echo/ server, but it ignores requests
1019+-- unless the text body @/= "bar"@ - this case should time out rather than
1020+-- removing the message from the process mailbox.
1021+testMatchAnyIf :: TestTransport -> Assertion
1022+testMatchAnyIf TestTransport{..} = do
1023+ echoAddr <- newEmptyMVar
1024+ clientDone <- newEmptyMVar
1025+
1026+ -- echo server
1027+ forkIO $ do
1028+ localNode <- newLocalNode testTransport initRemoteTable
1029+ echoServer <- forkProcess localNode $ forever $ do
1030+ receiveWait [
1031+ matchAnyIf (\(_ :: ProcessId, (s :: String)) -> s /= "bar")
1032+ tryHandleMessage
1033+ ]
1034+ putMVar echoAddr echoServer
1035+
1036+ -- Client
1037+ forkIO $ do
1038+ localNode <- newLocalNode testTransport initRemoteTable
1039+ server <- readMVar echoAddr
1040+
1041+ runProcess localNode $ do
1042+ pid <- getSelfPid
1043+ send server (pid, "foo")
1044+ "foo" <- expect
1045+ send server (pid, "baz")
1046+ "baz" <- expect
1047+ send server (pid, "bar")
1048+ Nothing <- (expectTimeout 100000) :: Process (Maybe Double)
1049+ liftIO $ putMVar clientDone ()
1050+
1051+ takeMVar clientDone
1052+ where tryHandleMessage :: Message -> Process (Maybe ())
1053+ tryHandleMessage msg =
1054+ handleMessage msg (\(pid :: ProcessId, (m :: String))
1055+ -> do { send pid m; return () })
1056+
1057+testMatchMessageWithUnwrap :: TestTransport -> Assertion
1058+testMatchMessageWithUnwrap TestTransport{..} = do
1059+ echoAddr <- newEmptyMVar
1060+ clientDone <- newEmptyMVar
1061+
1062+ -- echo server
1063+ forkIO $ do
1064+ localNode <- newLocalNode testTransport initRemoteTable
1065+ echoServer <- forkProcess localNode $ forever $ do
1066+ msg <- receiveWait [
1067+ matchMessage (\(m :: Message) -> do
1068+ return m)
1069+ ]
1070+ unwrapped <- unwrapMessage msg :: Process (Maybe (ProcessId, Message))
1071+ case unwrapped of
1072+ (Just (p, msg')) -> forward msg' p
1073+ Nothing -> die "unable to unwrap the message"
1074+ putMVar echoAddr echoServer
1075+
1076+ -- Client
1077+ forkIO $ do
1078+ localNode <- newLocalNode testTransport initRemoteTable
1079+ server <- readMVar echoAddr
1080+
1081+ runProcess localNode $ do
1082+ pid <- getSelfPid
1083+ send server (pid, wrapMessage ("foo" :: String))
1084+ "foo" <- expect
1085+ send server (pid, wrapMessage ("baz" :: String))
1086+ "baz" <- expect
1087+ liftIO $ putMVar clientDone ()
1088+
1089+ takeMVar clientDone
1090+
1091+-- Test 'receiveChanTimeout'
1092+testReceiveChanTimeout :: TestTransport -> Assertion
1093+testReceiveChanTimeout TestTransport{..} = do
1094+ done <- newEmptyMVar
1095+ sendPort <- newEmptyMVar
1096+
1097+ forkTry $ do
1098+ localNode <- newLocalNode testTransport initRemoteTable
1099+ runProcess localNode $ do
1100+ -- Create a typed channel
1101+ (sp, rp) <- newChan :: Process (SendPort Bool, ReceivePort Bool)
1102+ liftIO $ putMVar sendPort sp
1103+
1104+ -- Wait for a message with a delay. No message arrives, we should get Nothing after 1 second
1105+ Nothing <- receiveChanTimeout 1000000 rp
1106+
1107+ -- Wait for a message with a delay again. Now a message arrives after 0.5 seconds
1108+ Just True <- receiveChanTimeout 1000000 rp
1109+
1110+ -- Wait for a message with zero timeout: non-blocking check. No message is available, we get Nothing
1111+ Nothing <- receiveChanTimeout 0 rp
1112+
1113+ -- Again, but now there is a message available
1114+ liftIO $ threadDelay 1000000
1115+ Just False <- receiveChanTimeout 0 rp
1116+
1117+ liftIO $ putMVar done ()
1118+
1119+ forkTry $ do
1120+ localNode <- newLocalNode testTransport initRemoteTable
1121+ runProcess localNode $ do
1122+ sp <- liftIO $ readMVar sendPort
1123+
1124+ liftIO $ threadDelay 1500000
1125+ sendChan sp True
1126+
1127+ liftIO $ threadDelay 500000
1128+ sendChan sp False
1129+
1130+ takeMVar done
1131+
1132+-- | Test Functor, Applicative, Alternative and Monad instances for ReceiveChan
1133+testReceiveChanFeatures :: TestTransport -> Assertion
1134+testReceiveChanFeatures TestTransport{..} = do
1135+ done <- newEmptyMVar
1136+
1137+ forkTry $ do
1138+ localNode <- newLocalNode testTransport initRemoteTable
1139+ runProcess localNode $ do
1140+ (spInt, rpInt) <- newChan :: Process (SendPort Int, ReceivePort Int)
1141+ (spBool, rpBool) <- newChan :: Process (SendPort Bool, ReceivePort Bool)
1142+
1143+ -- Test Functor instance
1144+
1145+ sendChan spInt 2
1146+ sendChan spBool False
1147+
1148+ rp1 <- mergePortsBiased [even <$> rpInt, rpBool]
1149+
1150+ True <- receiveChan rp1
1151+ False <- receiveChan rp1
1152+
1153+ -- Test Applicative instance
1154+
1155+ sendChan spInt 3
1156+ sendChan spInt 4
1157+
1158+ let rp2 = pure (+) <*> rpInt <*> rpInt
1159+
1160+ 7 <- receiveChan rp2
1161+
1162+ -- Test Alternative instance
1163+
1164+ sendChan spInt 3
1165+ sendChan spBool True
1166+
1167+ let rp3 = (even <$> rpInt) <|> rpBool
1168+
1169+ False <- receiveChan rp3
1170+ True <- receiveChan rp3
1171+
1172+ -- Test Monad instance
1173+
1174+ sendChan spBool True
1175+ sendChan spBool False
1176+ sendChan spInt 5
1177+
1178+ let rp4 :: ReceivePort Int
1179+ rp4 = do b <- rpBool
1180+ if b
1181+ then rpInt
1182+ else return 7
1183+
1184+ 5 <- receiveChan rp4
1185+ 7 <- receiveChan rp4
1186+
1187+ liftIO $ putMVar done ()
1188+
1189+ takeMVar done
1190+
1191+testKillLocal :: TestTransport -> Assertion
1192+testKillLocal TestTransport{..} = do
1193+ localNode <- newLocalNode testTransport initRemoteTable
1194+ done <- newEmptyMVar
1195+
1196+ pid <- forkProcess localNode $ do
1197+ liftIO $ threadDelay 1000000
1198+
1199+ runProcess localNode $ do
1200+ ref <- monitor pid
1201+ us <- getSelfPid
1202+ kill pid "TestKill"
1203+ ProcessMonitorNotification ref' pid' (DiedException ex) <- expect
1204+ True <- return $ ref == ref' && pid == pid' && ex == "killed-by=" ++ show us ++ ",reason=TestKill"
1205+ liftIO $ putMVar done ()
1206+
1207+ takeMVar done
1208+
1209+testKillRemote :: TestTransport -> Assertion
1210+testKillRemote TestTransport{..} = do
1211+ node1 <- newLocalNode testTransport initRemoteTable
1212+ node2 <- newLocalNode testTransport initRemoteTable
1213+ done <- newEmptyMVar
1214+
1215+ pid <- forkProcess node1 $ do
1216+ liftIO $ threadDelay 1000000
1217+
1218+ runProcess node2 $ do
1219+ ref <- monitor pid
1220+ us <- getSelfPid
1221+ kill pid "TestKill"
1222+ ProcessMonitorNotification ref' pid' (DiedException reason) <- expect
1223+ True <- return $ ref == ref' && pid == pid' && reason == "killed-by=" ++ show us ++ ",reason=TestKill"
1224+ liftIO $ putMVar done ()
1225+
1226+ takeMVar done
1227+
1228+testCatchesExit :: TestTransport -> Assertion
1229+testCatchesExit TestTransport{..} = do
1230+ localNode <- newLocalNode testTransport initRemoteTable
1231+ done <- newEmptyMVar
1232+
1233+ _ <- forkProcess localNode $ do
1234+ (die ("foobar", 123 :: Int))
1235+ `catchesExit` [
1236+ (\_ m -> handleMessage m (\(_ :: String) -> return ()))
1237+ , (\_ m -> handleMessage m (\(_ :: Maybe Int) -> return ()))
1238+ , (\_ m -> handleMessage m (\(_ :: String, _ :: Int)
1239+ -> (liftIO $ putMVar done ()) >> return ()))
1240+ ]
1241+
1242+ takeMVar done
1243+
1244+testHandleMessageIf :: TestTransport -> Assertion
1245+testHandleMessageIf TestTransport{..} = do
1246+ localNode <- newLocalNode testTransport initRemoteTable
1247+ done <- newEmptyMVar
1248+ _ <- forkProcess localNode $ do
1249+ self <- getSelfPid
1250+ send self (5 :: Integer, 10 :: Integer)
1251+ msg <- receiveWait [ matchMessage return ]
1252+ Nothing <- handleMessageIf msg (\() -> True) (\() -> die $ "whoops")
1253+ handleMessageIf msg (\(x :: Integer, y :: Integer) -> x == 5 && y == 10)
1254+ (\input -> liftIO $ putMVar done input)
1255+ return ()
1256+
1257+ result <- takeMVar done
1258+ expectThat result $ equalTo (5, 10)
1259+
1260+testCatches :: TestTransport -> Assertion
1261+testCatches TestTransport{..} = do
1262+ localNode <- newLocalNode testTransport initRemoteTable
1263+ done <- newEmptyMVar
1264+
1265+ _ <- forkProcess localNode $ do
1266+ node <- getSelfNode
1267+ (liftIO $ throwIO (ProcessLinkException (nullProcessId node) DiedNormal))
1268+ `catches` [
1269+ Handler (\(ProcessLinkException _ _) -> liftIO $ putMVar done ())
1270+ ]
1271+
1272+ takeMVar done
1273+
1274+testMaskRestoreScope :: TestTransport -> Assertion
1275+testMaskRestoreScope TestTransport{..} = do
1276+ localNode <- newLocalNode testTransport initRemoteTable
1277+ parentPid <- newEmptyMVar :: IO (MVar ProcessId)
1278+ spawnedPid <- newEmptyMVar :: IO (MVar ProcessId)
1279+
1280+ void $ runProcess localNode $ mask $ \unmask -> do
1281+ getSelfPid >>= liftIO . putMVar parentPid
1282+ void $ spawnLocal $ unmask (getSelfPid >>= liftIO . putMVar spawnedPid)
1283+
1284+ parent <- liftIO $ takeMVar parentPid
1285+ child <- liftIO $ takeMVar spawnedPid
1286+ expectThat parent $ isNot $ equalTo child
1287+
1288+testDie :: TestTransport -> Assertion
1289+testDie TestTransport{..} = do
1290+ localNode <- newLocalNode testTransport initRemoteTable
1291+ done <- newEmptyMVar
1292+
1293+ _ <- forkProcess localNode $ do
1294+ (die ("foobar", 123 :: Int))
1295+ `catchExit` \_from reason -> do
1296+ -- TODO: should verify that 'from' has the right value
1297+ True <- return $ reason == ("foobar", 123 :: Int)
1298+ liftIO $ putMVar done ()
1299+
1300+ takeMVar done
1301+
1302+testPrettyExit :: TestTransport -> Assertion
1303+testPrettyExit TestTransport{..} = do
1304+ localNode <- newLocalNode testTransport initRemoteTable
1305+ done <- newEmptyMVar
1306+
1307+ _ <- forkProcess localNode $ do
1308+ (die "timeout")
1309+ `catch` \ex@(ProcessExitException from _) ->
1310+ let expected = "exit-from=" ++ (show from)
1311+ in do
1312+ True <- return $ (show ex) == expected
1313+ liftIO $ putMVar done ()
1314+
1315+ takeMVar done
1316+
1317+testExitLocal :: TestTransport -> Assertion
1318+testExitLocal TestTransport{..} = do
1319+ localNode <- newLocalNode testTransport initRemoteTable
1320+ supervisedDone <- newEmptyMVar
1321+ supervisorDone <- newEmptyMVar
1322+ -- XXX: we guarantee that exception handler will be set up
1323+ -- regardless if forkProcess preserve masking state or not.
1324+ handlerSetUp <- newEmptyMVar
1325+
1326+ pid <- forkProcess localNode $ do
1327+ (liftIO (putMVar handlerSetUp ()) >> expect) `catchExit` \_from reason -> do
1328+ -- TODO: should verify that 'from' has the right value
1329+ True <- return $ reason == "TestExit"
1330+ liftIO $ putMVar supervisedDone ()
1331+
1332+ runProcess localNode $ do
1333+ liftIO $ takeMVar handlerSetUp
1334+ ref <- monitor pid
1335+ exit pid "TestExit"
1336+ -- This time the client catches the exception, so it dies normally
1337+ ProcessMonitorNotification ref' pid' DiedNormal <- expect
1338+ True <- return $ ref == ref' && pid == pid'
1339+ liftIO $ putMVar supervisorDone ()
1340+
1341+ takeMVar supervisedDone
1342+ takeMVar supervisorDone
1343+
1344+testExitRemote :: TestTransport -> Assertion
1345+testExitRemote TestTransport{..} = do
1346+ node1 <- newLocalNode testTransport initRemoteTable
1347+ node2 <- newLocalNode testTransport initRemoteTable
1348+ supervisedDone <- newEmptyMVar
1349+ supervisorDone <- newEmptyMVar
1350+
1351+ pid <- forkProcess node1 $ do
1352+ (receiveWait [] :: Process ()) -- block forever
1353+ `catchExit` \_from reason -> do
1354+ -- TODO: should verify that 'from' has the right value
1355+ True <- return $ reason == "TestExit"
1356+ liftIO $ putMVar supervisedDone ()
1357+
1358+ runProcess node2 $ do
1359+ ref <- monitor pid
1360+ exit pid "TestExit"
1361+ ProcessMonitorNotification ref' pid' DiedNormal <- expect
1362+ True <- return $ ref == ref' && pid == pid'
1363+ liftIO $ putMVar supervisorDone ()
1364+
1365+ takeMVar supervisedDone
1366+ takeMVar supervisorDone
1367+
1368+testUnsafeSend :: TestTransport -> Assertion
1369+testUnsafeSend TestTransport{..} = do
1370+ serverAddr <- newEmptyMVar
1371+ clientDone <- newEmptyMVar
1372+
1373+ localNode <- newLocalNode testTransport initRemoteTable
1374+ void $ forkProcess localNode $ do
1375+ self <- getSelfPid
1376+ liftIO $ putMVar serverAddr self
1377+ clientAddr <- expect
1378+ unsafeSend clientAddr ()
1379+
1380+ void $ forkProcess localNode $ do
1381+ serverPid <- liftIO $ takeMVar serverAddr
1382+ getSelfPid >>= unsafeSend serverPid
1383+ () <- expect
1384+ liftIO $ putMVar clientDone ()
1385+
1386+ takeMVar clientDone
1387+
1388+testUnsafeNSend :: TestTransport -> Assertion
1389+testUnsafeNSend TestTransport{..} = do
1390+ clientDone <- newEmptyMVar
1391+
1392+ localNode <- newLocalNode testTransport initRemoteTable
1393+
1394+ pid <- forkProcess localNode $ do
1395+ () <- expect
1396+ liftIO $ putMVar clientDone ()
1397+
1398+ void $ runProcess localNode $ do
1399+ register "foobar" pid
1400+ unsafeNSend "foobar" ()
1401+
1402+ takeMVar clientDone
1403+
1404+testUnsafeSendChan :: TestTransport -> Assertion
1405+testUnsafeSendChan TestTransport{..} = do
1406+ serverAddr <- newEmptyMVar
1407+ clientDone <- newEmptyMVar
1408+
1409+ localNode <- newLocalNode testTransport initRemoteTable
1410+ void $ forkProcess localNode $ do
1411+ self <- getSelfPid
1412+ liftIO $ putMVar serverAddr self
1413+ sp <- expect
1414+ unsafeSendChan sp ()
1415+
1416+ void $ forkProcess localNode $ do
1417+ serverPid <- liftIO $ takeMVar serverAddr
1418+ (sp, rp) <- newChan
1419+ unsafeSend serverPid sp
1420+ () <- receiveChan rp
1421+ liftIO $ putMVar clientDone ()
1422+
1423+ takeMVar clientDone
1424+
1425+testCallLocal :: TestTransport -> Assertion
1426+testCallLocal TestTransport{..} = do
1427+ node <- newLocalNode testTransport initRemoteTable
1428+
1429+ -- Testing that (/=) <$> getSelfPid <*> callLocal getSelfPid.
1430+ result <- newEmptyMVar
1431+ runProcess node $ do
1432+ r <- (/=) <$> getSelfPid <*> callLocal getSelfPid
1433+ liftIO $ putMVar result r
1434+ True <- takeMVar result
1435+ return ()
1436+
1437+ -- Testing that when callLocal is interrupted, the worker is interrupted.
1438+ ibox <- newIORef False
1439+ runProcess node $ do
1440+ keeper <- getSelfPid
1441+ spawnLocal $ do
1442+ caller <- getSelfPid
1443+ send keeper caller
1444+ onException
1445+ (callLocal $ do
1446+ onException (do send keeper caller
1447+ expect)
1448+ (do liftIO $ writeIORef ibox True))
1449+ (send keeper ())
1450+ caller <- expect
1451+ exit caller "test"
1452+ () <- expect
1453+ return ()
1454+ True <- readIORef ibox
1455+ return ()
1456+
1457+ -- Testing that when the worker raises an exception, the exception is propagated to the parent.
1458+ ibox2 <- newIORef False
1459+ runProcess node $ do
1460+ r <- try (callLocal $ error "e" >> return ())
1461+ liftIO $ writeIORef ibox2 $ case r of
1462+ Left (ErrorCall "e") -> True
1463+ _ -> False
1464+ True <- readIORef ibox
1465+ return ()
1466+
1467+ -- Test that caller waits for the worker in correct situation
1468+ ibox3 <- newIORef False
1469+ result3 <- newEmptyMVar
1470+ runProcess node $ do
1471+ keeper <- getSelfPid
1472+ spawnLocal $ do
1473+ callLocal $
1474+ (do us <- getSelfPid
1475+ send keeper us
1476+ () <- expect
1477+ liftIO yield)
1478+ `finally` (liftIO $ writeIORef ibox3 True)
1479+ liftIO $ putMVar result3 =<< readIORef ibox3
1480+ worker <- expect
1481+ send worker ()
1482+ True <- takeMVar result3
1483+ return ()
1484+
1485+ -- Test that caller waits for the worker in case when caller gets an exception
1486+ ibox4 <- newIORef False
1487+ result4 <- newEmptyMVar
1488+ runProcess node $ do
1489+ keeper <- getSelfPid
1490+ spawnLocal $ do
1491+ caller <- getSelfPid
1492+ callLocal
1493+ ((do send keeper caller
1494+ expect)
1495+ `finally` (liftIO $ writeIORef ibox4 True))
1496+ `finally` (liftIO $ putMVar result4 =<< readIORef ibox4)
1497+ caller <- expect
1498+ exit caller "hi!"
1499+ True <- takeMVar result4
1500+ return ()
1501+ -- XXX: Testing that when mask_ $ callLocal p runs p in masked state.
1502+
1503+
1504+
1505+
1506+
1507+tests :: TestTransport -> IO [Test]
1508+tests testtrans = return [
1509+ testGroup "Basic features" [
1510+ testCase "Ping" (testPing testtrans)
1511+ , testCase "Math" (testMath testtrans)
1512+ , testCase "Timeout" (testTimeout testtrans)
1513+ , testCase "Timeout0" (testTimeout0 testtrans)
1514+ , testCase "SendToTerminated" (testSendToTerminated testtrans)
1515+ , testCase "TypedChannnels" (testTypedChannels testtrans)
1516+ , testCase "Terminate" (testTerminate testtrans)
1517+ , testCase "RegistryRemoteProcess" (testRegistryRemoteProcess testtrans)
1518+ , testCase "RemoteRegistry" (testRemoteRegistry testtrans)
1519+ , testCase "RemoteRegistryRemoteProcess" (testRemoteRegistryRemoteProcess testtrans)
1520+ , testCase "SpawnLocal" (testSpawnLocal testtrans)
1521+ , testCase "SpawnAsyncStrictness" (testSpawnAsyncStrictness testtrans)
1522+ , testCase "HandleMessageIf" (testHandleMessageIf testtrans)
1523+ , testCase "MatchAny" (testMatchAny testtrans)
1524+ , testCase "MatchAnyHandle" (testMatchAnyHandle testtrans)
1525+ , testCase "MatchAnyNoHandle" (testMatchAnyNoHandle testtrans)
1526+ , testCase "MatchAnyIf" (testMatchAnyIf testtrans)
1527+ , testCase "MatchMessageUnwrap" (testMatchMessageWithUnwrap testtrans)
1528+ , testCase "ReceiveChanTimeout" (testReceiveChanTimeout testtrans)
1529+ , testCase "ReceiveChanFeatures" (testReceiveChanFeatures testtrans)
1530+ , testCase "KillLocal" (testKillLocal testtrans)
1531+ , testCase "KillRemote" (testKillRemote testtrans)
1532+ , testCase "Die" (testDie testtrans)
1533+ , testCase "PrettyExit" (testPrettyExit testtrans)
1534+ , testCase "CatchesExit" (testCatchesExit testtrans)
1535+ , testCase "Catches" (testCatches testtrans)
1536+ , testCase "MaskRestoreScope" (testMaskRestoreScope testtrans)
1537+ , testCase "ExitLocal" (testExitLocal testtrans)
1538+ , testCase "ExitRemote" (testExitRemote testtrans)
1539+ , testCase "TextCallLocal" (testCallLocal testtrans)
1540+ -- Unsafe Primitives
1541+ , testCase "TestUnsafeSend" (testUnsafeSend testtrans)
1542+ , testCase "TestUnsafeNSend" (testUnsafeNSend testtrans)
1543+ , testCase "TestUnsafeSendChan" (testUnsafeSendChan testtrans)
1544+ -- usend
1545+ , testCase "USend" (testUSend usend testtrans 50)
1546+ , testCase "UForward"
1547+ (testUSend (\p m -> uforward (createUnencodedMessage m) p)
1548+ testtrans 50
1549+ )
1550+ ]
1551+ , testGroup "Monitoring and Linking" [
1552+ -- Monitoring processes
1553+ --
1554+ -- The "missing" combinations in the list below don't make much sense, as
1555+ -- we cannot guarantee that the monitor reply or link exception will not
1556+ -- happen before the unmonitor or unlink
1557+ testCase "MonitorNormalTermination" (testMonitorNormalTermination testtrans True False)
1558+ , testCase "MonitorAbnormalTermination" (testMonitorAbnormalTermination testtrans True False)
1559+ , testCase "MonitorLocalDeadProcess" (testMonitorLocalDeadProcess testtrans True False)
1560+ , testCase "MonitorRemoteDeadProcess" (testMonitorRemoteDeadProcess testtrans True False)
1561+ , testCase "MonitorDisconnect" (testMonitorDisconnect testtrans True False)
1562+ , testCase "LinkUnreachable" (testMonitorUnreachable testtrans False False)
1563+ , testCase "LinkNormalTermination" (testMonitorNormalTermination testtrans False False)
1564+ , testCase "LinkAbnormalTermination" (testMonitorAbnormalTermination testtrans False False)
1565+ , testCase "LinkLocalDeadProcess" (testMonitorLocalDeadProcess testtrans False False)
1566+ , testCase "LinkRemoteDeadProcess" (testMonitorRemoteDeadProcess testtrans False False)
1567+ , testCase "LinkDisconnect" (testMonitorDisconnect testtrans False False)
1568+ , testCase "UnmonitorNormalTermination" (testMonitorNormalTermination testtrans True True)
1569+ , testCase "UnmonitorAbnormalTermination" (testMonitorAbnormalTermination testtrans True True)
1570+ , testCase "UnmonitorDisconnect" (testMonitorDisconnect testtrans True True)
1571+ , testCase "UnlinkNormalTermination" (testMonitorNormalTermination testtrans False True)
1572+ , testCase "UnlinkAbnormalTermination" (testMonitorAbnormalTermination testtrans False True)
1573+ , testCase "UnlinkDisconnect" (testMonitorDisconnect testtrans False True)
1574+ -- Monitoring nodes and channels
1575+ , testCase "MonitorNode" (testMonitorNode testtrans)
1576+ , testCase "MonitorLiveNode" (testMonitorLiveNode testtrans)
1577+ , testCase "MonitorChannel" (testMonitorChannel testtrans)
1578+ -- Reconnect
1579+ ]
1580+
1581+ -- Tests that fail occasionally and should be revised
1582+ , testGroup "Flaky" [
1583+ testCase "Reconnect" (testReconnect testtrans)
1584+ , testCase "Registry" (testRegistry testtrans)
1585+ , testCase "MergeChannels" (testMergeChannels testtrans)
1586+ , testCase "MonitorUnreachable" (testMonitorUnreachable testtrans True False)
1587+ ]
1588+ ]
1589+
1590+syncBreakConnection :: (NT.EndPointAddress -> NT.EndPointAddress -> IO ()) -> LocalNode -> LocalNode -> IO ()
1591+syncBreakConnection breakConnection nid0 nid1 = do
1592+ m <- newEmptyMVar
1593+ _ <- forkProcess nid1 $ getSelfPid >>= liftIO . putMVar m
1594+ runProcess nid0 $ do
1595+ them <- liftIO $ takeMVar m
1596+ pinger <- spawnLocal $ forever $ send them ()
1597+ _ <- monitorNode (localNodeId nid1)
1598+ liftIO $ breakConnection (nodeAddress $ localNodeId nid0)
1599+ (nodeAddress $ localNodeId nid1)
1600+ NodeMonitorNotification _ _ _ <- expect
1601+ kill pinger "finished"
1602+ return ()