git.iko.soy/distributed-process
on fix-spawnChan help sign in
distributed-process/tests/TestCH.hs
1184 lines 40 kB history raw
1module Main where
3#if ! MIN_VERSION_base(4,6,0)
4import Prelude hiding (catch)
5#endif
7import Data.Binary (Binary(..))
8import Data.Typeable (Typeable)
9import Data.Foldable (forM_)
10import Control.Concurrent (forkIO, threadDelay, myThreadId, throwTo, ThreadId)
11import Control.Concurrent.MVar
12 ( MVar
13 , newEmptyMVar
14 , putMVar
15 , takeMVar
16 , readMVar
17 )
18import Control.Monad (replicateM_, replicateM, forever)
19import Control.Exception (SomeException, throwIO)
20import qualified Control.Exception as Ex (catch)
21import Control.Applicative ((<$>), (<*>), pure, (<|>))
22import qualified Network.Transport as NT (Transport, closeEndPoint)
23import Network.Socket (sClose)
24import Network.Transport.TCP
25 ( createTransportExposeInternals
26 , TransportInternals(socketBetween)
27 , defaultTCPParameters
28 )
29import Control.Distributed.Process
30import Control.Distributed.Process.Internal.Types
31 ( NodeId(nodeAddress)
32 , LocalNode(localEndPoint)
33 , ProcessExitException(..)
34 , nullProcessId
35 )
36import Control.Distributed.Process.Node
37import Control.Distributed.Process.Serializable (Serializable)
39import Test.HUnit (Assertion)
40import Test.Framework (Test, defaultMain, testGroup)
41import Test.Framework.Providers.HUnit (testCase)
43newtype Ping = Ping ProcessId
44 deriving (Typeable, Binary, Show)
46newtype Pong = Pong ProcessId
47 deriving (Typeable, Binary, Show)
49--------------------------------------------------------------------------------
50-- Supporting definitions --
51--------------------------------------------------------------------------------
53-- | Like fork, but throw exceptions in the child thread to the parent
54forkTry :: IO () -> IO ThreadId
55forkTry p = do
56 tid <- myThreadId
57 forkIO $ Ex.catch p (\e -> throwTo tid (e :: SomeException))
59-- | The ping server from the paper
60ping :: Process ()
61ping = do
62 Pong partner <- expect
63 self <- getSelfPid
64 send partner (Ping self)
65 ping
67-- | Quick and dirty synchronous version of whereisRemoteAsync
68whereisRemote :: NodeId -> String -> Process (Maybe ProcessId)
69whereisRemote nid string = do
70 whereisRemoteAsync nid string
71 WhereIsReply _ mPid <- expect
72 return mPid
74data Add = Add ProcessId Double Double deriving (Typeable)
75data Divide = Divide ProcessId Double Double deriving (Typeable)
76data DivByZero = DivByZero deriving (Typeable)
78instance Binary Add where
79 put (Add pid x y) = put pid >> put x >> put y
80 get = Add <$> get <*> get <*> get
82instance Binary Divide where
83 put (Divide pid x y) = put pid >> put x >> put y
84 get = Divide <$> get <*> get <*> get
86instance Binary DivByZero where
87 put DivByZero = return ()
88 get = return DivByZero
90-- The math server from the paper
91math :: Process ()
92math = do
93 receiveWait
94 [ match (\(Add pid x y) -> send pid (x + y))
95 , matchIf (\(Divide _ _ y) -> y /= 0)
96 (\(Divide pid x y) -> send pid (x / y))
97 , match (\(Divide pid _ _) -> send pid DivByZero)
98 ]
99 math
101-- | Monitor or link to a remote node
102monitorOrLink :: Bool -- ^ 'True' for monitor, 'False' for link
103 -> ProcessId -- Process to monitor/link to
104 -> Maybe (MVar ()) -- MVar to signal on once the monitor has been set up
105 -> Process (Maybe MonitorRef)
106monitorOrLink mOrL pid mSignal = do
107 result <- if mOrL then Just <$> monitor pid
108 else link pid >> return Nothing
109 -- Monitor is asynchronous, which usually does not matter but if we want a
110 -- *specific* signal then it does. Therefore we wait an arbitrary delay and
111 -- hope that this means the monitor has been set up
112 forM_ mSignal $ \signal -> liftIO . forkIO $ threadDelay 100000 >> putMVar signal ()
113 return result
115monitorTestProcess :: ProcessId -- Process to monitor/link to
116 -> Bool -- 'True' for monitor, 'False' for link
117 -> Bool -- Should we unmonitor?
118 -> DiedReason -- Expected cause of death
119 -> Maybe (MVar ()) -- Signal for 'monitor set up'
120 -> MVar () -- Signal for successful termination
121 -> Process ()
122monitorTestProcess theirAddr mOrL un reason monitorSetup done =
123 catch (do mRef <- monitorOrLink mOrL theirAddr monitorSetup
124 case (un, mRef) of
125 (True, Nothing) -> do
126 unlink theirAddr
127 liftIO $ putMVar done ()
128 (True, Just ref) -> do
129 unmonitor ref
130 liftIO $ putMVar done ()
131 (False, ref) -> do
132 ProcessMonitorNotification ref' pid reason' <- expect
133 True <- return $ Just ref' == ref && pid == theirAddr && mOrL && reason == reason'
134 liftIO $ putMVar done ()
135 )
136 (\(ProcessLinkException pid reason') -> do
137 True <- return $ pid == theirAddr && not mOrL && not un && reason == reason'
138 liftIO $ putMVar done ()
139 )
141--------------------------------------------------------------------------------
142-- The tests proper --
143--------------------------------------------------------------------------------
145-- | Basic ping test
146testPing :: NT.Transport -> Assertion
147testPing transport = do
148 serverAddr <- newEmptyMVar
149 clientDone <- newEmptyMVar
151 -- Server
152 forkIO $ do
153 localNode <- newLocalNode transport initRemoteTable
154 addr <- forkProcess localNode ping
155 putMVar serverAddr addr
157 -- Client
158 forkIO $ do
159 localNode <- newLocalNode transport initRemoteTable
160 pingServer <- readMVar serverAddr
162 let numPings = 10000
164 runProcess localNode $ do
165 pid <- getSelfPid
166 replicateM_ numPings $ do
167 send pingServer (Pong pid)
168 Ping _ <- expect
169 return ()
171 putMVar clientDone ()
173 takeMVar clientDone
175-- | Monitor a process on an unreachable node
176testMonitorUnreachable :: NT.Transport -> Bool -> Bool -> Assertion
177testMonitorUnreachable transport mOrL un = do
178 deadProcess <- newEmptyMVar
179 done <- newEmptyMVar
181 forkIO $ do
182 localNode <- newLocalNode transport initRemoteTable
183 addr <- forkProcess localNode . liftIO $ threadDelay 1000000
184 closeLocalNode localNode
185 putMVar deadProcess addr
187 forkIO $ do
188 localNode <- newLocalNode transport initRemoteTable
189 theirAddr <- readMVar deadProcess
190 runProcess localNode $
191 monitorTestProcess theirAddr mOrL un DiedDisconnect Nothing done
193 takeMVar done
195-- | Monitor a process which terminates normally
196testMonitorNormalTermination :: NT.Transport -> Bool -> Bool -> Assertion
197testMonitorNormalTermination transport mOrL un = do
198 monitorSetup <- newEmptyMVar
199 monitoredProcess <- newEmptyMVar
200 done <- newEmptyMVar
202 forkIO $ do
203 localNode <- newLocalNode transport initRemoteTable
204 addr <- forkProcess localNode $
205 liftIO $ readMVar monitorSetup
206 putMVar monitoredProcess addr
208 forkIO $ do
209 localNode <- newLocalNode transport initRemoteTable
210 theirAddr <- readMVar monitoredProcess
211 runProcess localNode $
212 monitorTestProcess theirAddr mOrL un DiedNormal (Just monitorSetup) done
214 takeMVar done
216-- | Monitor a process which terminates abnormally
217testMonitorAbnormalTermination :: NT.Transport -> Bool -> Bool -> Assertion
218testMonitorAbnormalTermination transport mOrL un = do
219 monitorSetup <- newEmptyMVar
220 monitoredProcess <- newEmptyMVar
221 done <- newEmptyMVar
223 let err = userError "Abnormal termination"
225 forkIO $ do
226 localNode <- newLocalNode transport initRemoteTable
227 addr <- forkProcess localNode . liftIO $ do
228 readMVar monitorSetup
229 throwIO err
230 putMVar monitoredProcess addr
232 forkIO $ do
233 localNode <- newLocalNode transport initRemoteTable
234 theirAddr <- readMVar monitoredProcess
235 runProcess localNode $
236 monitorTestProcess theirAddr mOrL un (DiedException (show err)) (Just monitorSetup) done
238 takeMVar done
240-- | Monitor a local process that is already dead
241testMonitorLocalDeadProcess :: NT.Transport -> Bool -> Bool -> Assertion
242testMonitorLocalDeadProcess transport mOrL un = do
243 processDead <- newEmptyMVar
244 processAddr <- newEmptyMVar
245 localNode <- newLocalNode transport initRemoteTable
246 done <- newEmptyMVar
248 forkIO $ do
249 addr <- forkProcess localNode . liftIO $ putMVar processDead ()
250 putMVar processAddr addr
252 forkIO $ do
253 theirAddr <- readMVar processAddr
254 readMVar processDead
255 runProcess localNode $ do
256 monitorTestProcess theirAddr mOrL un DiedUnknownId Nothing done
258 takeMVar done
260-- | Monitor a remote process that is already dead
261testMonitorRemoteDeadProcess :: NT.Transport -> Bool -> Bool -> Assertion
262testMonitorRemoteDeadProcess transport mOrL un = do
263 processDead <- newEmptyMVar
264 processAddr <- newEmptyMVar
265 done <- newEmptyMVar
267 forkIO $ do
268 localNode <- newLocalNode transport initRemoteTable
269 addr <- forkProcess localNode . liftIO $ putMVar processDead ()
270 putMVar processAddr addr
272 forkIO $ do
273 localNode <- newLocalNode transport initRemoteTable
274 theirAddr <- readMVar processAddr
275 readMVar processDead
276 runProcess localNode $ do
277 monitorTestProcess theirAddr mOrL un DiedUnknownId Nothing done
279 takeMVar done
281-- | Monitor a process that becomes disconnected
282testMonitorDisconnect :: NT.Transport -> Bool -> Bool -> Assertion
283testMonitorDisconnect transport mOrL un = do
284 processAddr <- newEmptyMVar
285 monitorSetup <- newEmptyMVar
286 done <- newEmptyMVar
288 forkIO $ do
289 localNode <- newLocalNode transport initRemoteTable
290 addr <- forkProcess localNode . liftIO $ threadDelay 1000000
291 putMVar processAddr addr
292 readMVar monitorSetup
293 NT.closeEndPoint (localEndPoint localNode)
295 forkIO $ do
296 localNode <- newLocalNode transport initRemoteTable
297 theirAddr <- readMVar processAddr
298 runProcess localNode $ do
299 monitorTestProcess theirAddr mOrL un DiedDisconnect (Just monitorSetup) done
301 takeMVar done
303-- | Test the math server (i.e., receiveWait)
304testMath :: NT.Transport -> Assertion
305testMath transport = do
306 serverAddr <- newEmptyMVar
307 clientDone <- newEmptyMVar
309 -- Server
310 forkIO $ do
311 localNode <- newLocalNode transport initRemoteTable
312 addr <- forkProcess localNode math
313 putMVar serverAddr addr
315 -- Client
316 forkIO $ do
317 localNode <- newLocalNode transport initRemoteTable
318 mathServer <- readMVar serverAddr
320 runProcess localNode $ do
321 pid <- getSelfPid
322 send mathServer (Add pid 1 2)
323 3 <- expect :: Process Double
324 send mathServer (Divide pid 8 2)
325 4 <- expect :: Process Double
326 send mathServer (Divide pid 8 0)
327 DivByZero <- expect
328 liftIO $ putMVar clientDone ()
330 takeMVar clientDone
332-- | Send first message (i.e. connect) to an already terminated process
333-- (without monitoring); then send another message to a second process on
334-- the same remote node (we're checking that the remote node did not die)
335testSendToTerminated :: NT.Transport -> Assertion
336testSendToTerminated transport = do
337 serverAddr1 <- newEmptyMVar
338 serverAddr2 <- newEmptyMVar
339 clientDone <- newEmptyMVar
341 forkIO $ do
342 terminated <- newEmptyMVar
343 localNode <- newLocalNode transport initRemoteTable
344 addr1 <- forkProcess localNode $ liftIO $ putMVar terminated ()
345 addr2 <- forkProcess localNode $ ping
346 readMVar terminated
347 putMVar serverAddr1 addr1
348 putMVar serverAddr2 addr2
350 forkIO $ do
351 localNode <- newLocalNode transport initRemoteTable
352 server1 <- readMVar serverAddr1
353 server2 <- readMVar serverAddr2
354 runProcess localNode $ do
355 pid <- getSelfPid
356 send server1 "Hi"
357 send server2 (Pong pid)
358 Ping pid' <- expect
359 True <- return $ pid' == server2
360 liftIO $ putMVar clientDone ()
362 takeMVar clientDone
364-- | Test (non-zero) timeout
365testTimeout :: NT.Transport -> Assertion
366testTimeout transport = do
367 localNode <- newLocalNode transport initRemoteTable
368 done <- newEmptyMVar
370 runProcess localNode $ do
371 Nothing <- receiveTimeout 1000000 [match (\(Add _ _ _) -> return ())]
372 liftIO $ putMVar done ()
374 takeMVar done
376-- | Test zero timeout
377testTimeout0 :: NT.Transport -> Assertion
378testTimeout0 transport = do
379 serverAddr <- newEmptyMVar
380 clientDone <- newEmptyMVar
381 messagesSent <- newEmptyMVar
383 forkIO $ do
384 localNode <- newLocalNode transport initRemoteTable
385 addr <- forkProcess localNode $ do
386 liftIO $ readMVar messagesSent >> threadDelay 1000000
387 -- Variation on the venerable ping server which uses a zero timeout
388 -- Since we wait for all messages to be sent before doing this receive,
389 -- we should nevertheless find the right message immediately
390 Just partner <- receiveTimeout 0 [match (\(Pong partner) -> return partner)]
391 self <- getSelfPid
392 send partner (Ping self)
393 putMVar serverAddr addr
395 forkIO $ do
396 localNode <- newLocalNode transport initRemoteTable
397 server <- readMVar serverAddr
398 runProcess localNode $ do
399 pid <- getSelfPid
400 -- Send a bunch of messages. A large number of messages that the server
401 -- is not interested in, and then a single message that it wants
402 replicateM_ 10000 $ send server "Irrelevant message"
403 send server (Pong pid)
404 liftIO $ putMVar messagesSent ()
405 Ping _ <- expect
406 liftIO $ putMVar clientDone ()
408 takeMVar clientDone
410-- | Test typed channels
411testTypedChannels :: NT.Transport -> Assertion
412testTypedChannels transport = do
413 serverChannel <- newEmptyMVar :: IO (MVar (SendPort (SendPort Bool, Int)))
414 clientDone <- newEmptyMVar
416 forkIO $ do
417 localNode <- newLocalNode transport initRemoteTable
418 forkProcess localNode $ do
419 (serverSendPort, rport) <- newChan
420 liftIO $ putMVar serverChannel serverSendPort
421 (clientSendPort, i) <- receiveChan rport
422 sendChan clientSendPort (even i)
423 return ()
425 forkIO $ do
426 localNode <- newLocalNode transport initRemoteTable
427 serverSendPort <- readMVar serverChannel
428 runProcess localNode $ do
429 (clientSendPort, rport) <- newChan
430 sendChan serverSendPort (clientSendPort, 5)
431 False <- receiveChan rport
432 liftIO $ putMVar clientDone ()
434 takeMVar clientDone
436-- | Test merging receive ports
437testMergeChannels :: NT.Transport -> Assertion
438testMergeChannels transport = do
439 localNode <- newLocalNode transport initRemoteTable
440 testFlat localNode True "aaabbbccc"
441 testFlat localNode False "abcabcabc"
442 testNested localNode True True "aaabbbcccdddeeefffggghhhiii"
443 testNested localNode True False "adgadgadgbehbehbehcficficfi"
444 testNested localNode False True "abcabcabcdefdefdefghighighi"
445 testNested localNode False False "adgbehcfiadgbehcfiadgbehcfi"
446 testBlocked localNode True
447 testBlocked localNode False
448 where
449 -- Single layer of merging
450 testFlat :: LocalNode -> Bool -> String -> IO ()
451 testFlat localNode biased expected = do
452 done <- newEmptyMVar
453 forkProcess localNode $ do
454 rs <- mapM charChannel "abc"
455 m <- mergePorts biased rs
456 xs <- replicateM 9 $ receiveChan m
457 True <- return $ xs == expected
458 liftIO $ putMVar done ()
459 takeMVar done
461 -- Two layers of merging
462 testNested :: LocalNode -> Bool -> Bool -> String -> IO ()
463 testNested localNode biasedInner biasedOuter expected = do
464 done <- newEmptyMVar
465 forkProcess localNode $ do
466 rss <- mapM (mapM charChannel) ["abc", "def", "ghi"]
467 ms <- mapM (mergePorts biasedInner) rss
468 m <- mergePorts biasedOuter ms
469 xs <- replicateM (9 * 3) $ receiveChan m
470 True <- return $ xs == expected
471 liftIO $ putMVar done ()
472 takeMVar done
474 -- Test that if no messages are (immediately) available, the scheduler makes no difference
475 testBlocked :: LocalNode -> Bool -> IO ()
476 testBlocked localNode biased = do
477 vs <- replicateM 3 newEmptyMVar
478 done <- newEmptyMVar
480 forkProcess localNode $ do
481 [sa, sb, sc] <- liftIO $ mapM readMVar vs
482 mapM_ ((>> liftIO (threadDelay 10000)) . uncurry sendChan)
483 [ -- a, b, c
484 (sa, 'a')
485 , (sb, 'b')
486 , (sc, 'c')
487 -- a, c, b
488 , (sa, 'a')
489 , (sc, 'c')
490 , (sb, 'b')
491 -- b, a, c
492 , (sb, 'b')
493 , (sa, 'a')
494 , (sc, 'c')
495 -- b, c, a
496 , (sb, 'b')
497 , (sc, 'c')
498 , (sa, 'a')
499 -- c, a, b
500 , (sc, 'c')
501 , (sa, 'a')
502 , (sb, 'b')
503 -- c, b, a
504 , (sc, 'c')
505 , (sb, 'b')
506 , (sa, 'a')
507 ]
509 forkProcess localNode $ do
510 (ss, rs) <- unzip <$> replicateM 3 newChan
511 liftIO $ mapM_ (uncurry putMVar) $ zip vs ss
512 m <- mergePorts biased rs
513 xs <- replicateM (6 * 3) $ receiveChan m
514 True <- return $ xs == "abcacbbacbcacabcba"
515 liftIO $ putMVar done ()
517 takeMVar done
519 mergePorts :: Serializable a => Bool -> [ReceivePort a] -> Process (ReceivePort a)
520 mergePorts True = mergePortsBiased
521 mergePorts False = mergePortsRR
523 charChannel :: Char -> Process (ReceivePort Char)
524 charChannel c = do
525 (sport, rport) <- newChan
526 replicateM_ 3 $ sendChan sport c
527 liftIO $ threadDelay 10000 -- Make sure messages have been sent
528 return rport
530testTerminate :: NT.Transport -> Assertion
531testTerminate transport = do
532 localNode <- newLocalNode transport initRemoteTable
533 done <- newEmptyMVar
535 pid <- forkProcess localNode $ do
536 liftIO $ threadDelay 100000
537 terminate
539 runProcess localNode $ do
540 ref <- monitor pid
541 ProcessMonitorNotification ref' pid' (DiedException ex) <- expect
542 True <- return $ ref == ref' && pid == pid' && ex == show ProcessTerminationException
543 liftIO $ putMVar done ()
545 takeMVar done
547testMonitorNode :: NT.Transport -> Assertion
548testMonitorNode transport = do
549 [node1, node2] <- replicateM 2 $ newLocalNode transport initRemoteTable
550 done <- newEmptyMVar
552 closeLocalNode node1
554 runProcess node2 $ do
555 ref <- monitorNode (localNodeId node1)
556 NodeMonitorNotification ref' nid DiedDisconnect <- expect
557 True <- return $ ref == ref' && nid == localNodeId node1
558 liftIO $ putMVar done ()
560 takeMVar done
562testMonitorChannel :: NT.Transport -> Assertion
563testMonitorChannel transport = do
564 [node1, node2] <- replicateM 2 $ newLocalNode transport initRemoteTable
565 gotNotification <- newEmptyMVar
567 pid <- forkProcess node1 $ do
568 sport <- expect :: Process (SendPort ())
569 ref <- monitorPort sport
570 PortMonitorNotification ref' port' reason <- expect
571 -- reason might be DiedUnknownId if the receive port is GCed before the
572 -- monitor is established (TODO: not sure that this is reasonable)
573 return $ ref' == ref && port' == sendPortId sport && (reason == DiedNormal || reason == DiedUnknownId)
574 liftIO $ putMVar gotNotification ()
576 runProcess node2 $ do
577 (sport, _) <- newChan :: Process (SendPort (), ReceivePort ())
578 send pid sport
579 liftIO $ threadDelay 100000
581 takeMVar gotNotification
583testRegistry :: NT.Transport -> Assertion
584testRegistry transport = do
585 node <- newLocalNode transport initRemoteTable
586 done <- newEmptyMVar
588 pingServer <- forkProcess node ping
590 runProcess node $ do
591 register "ping" pingServer
592 Just pid <- whereis "ping"
593 True <- return $ pingServer == pid
594 us <- getSelfPid
595 nsend "ping" (Pong us)
596 Ping pid' <- expect
597 True <- return $ pingServer == pid'
598 liftIO $ putMVar done ()
600 takeMVar done
602testRemoteRegistry :: NT.Transport -> Assertion
603testRemoteRegistry transport = do
604 node1 <- newLocalNode transport initRemoteTable
605 node2 <- newLocalNode transport initRemoteTable
606 done <- newEmptyMVar
608 pingServer <- forkProcess node1 ping
610 runProcess node2 $ do
611 let nid1 = localNodeId node1
612 registerRemoteAsync nid1 "ping" pingServer
613 receiveWait [
614 matchIf (\(RegisterReply label' _) -> "ping" == label')
615 (\(RegisterReply _ _) -> return ()) ]
617 Just pid <- whereisRemote nid1 "ping"
618 True <- return $ pingServer == pid
619 us <- getSelfPid
620 nsendRemote nid1 "ping" (Pong us)
621 Ping pid' <- expect
622 True <- return $ pingServer == pid'
623 liftIO $ putMVar done ()
625 takeMVar done
627testSpawnLocal :: NT.Transport -> Assertion
628testSpawnLocal transport = do
629 node <- newLocalNode transport initRemoteTable
630 done <- newEmptyMVar
632 runProcess node $ do
633 us <- getSelfPid
635 pid <- spawnLocal $ do
636 sport <- expect
637 sendChan sport (1234 :: Int)
639 sport <- spawnChannelLocal $ \rport -> do
640 (1234 :: Int) <- receiveChan rport
641 send us ()
643 send pid sport
644 () <- expect
645 liftIO $ putMVar done ()
647 takeMVar done
649testReconnect :: NT.Transport -> TransportInternals -> Assertion
650testReconnect transport transportInternals = do
651 [node1, node2] <- replicateM 2 $ newLocalNode transport initRemoteTable
652 let nid1 = localNodeId node1
653 nid2 = localNodeId node2
654 processA <- newEmptyMVar
655 [sendTestOk, registerTestOk] <- replicateM 2 newEmptyMVar
657 forkProcess node1 $ do
658 us <- getSelfPid
659 liftIO $ putMVar processA us
660 msg1 <- expect
661 msg2 <- expect
662 True <- return $ msg1 == "message 1" && msg2 == "message 3"
663 liftIO $ putMVar sendTestOk ()
665 forkProcess node2 $ do
666 {-
667 - Make sure there is no implicit reconnect on normal message sending
668 -}
670 them <- liftIO $ readMVar processA
671 send them "message 1" >> liftIO (threadDelay 100000)
673 -- Simulate network failure
674 liftIO $ do
675 sock <- socketBetween transportInternals (nodeAddress nid1) (nodeAddress nid2)
676 sClose sock
677 threadDelay 10000
679 -- Should not arrive
680 send them "message 2"
682 -- Should arrive
683 reconnect them
684 send them "message 3"
686 liftIO $ takeMVar sendTestOk
688 {-
689 - Test that there *is* implicit reconnect on node controller messages
690 -}
692 us <- getSelfPid
693 registerRemoteAsync nid1 "a" us -- registerRemote is asynchronous
694 receiveWait [
695 matchIf (\(RegisterReply label' _) -> "a" == label')
696 (\(RegisterReply _ _) -> return ()) ]
698 Just _ <- whereisRemote nid1 "a"
701 -- Simulate network failure
702 liftIO $ do
703 sock <- socketBetween transportInternals (nodeAddress nid1) (nodeAddress nid2)
704 sClose sock
705 threadDelay 10000
707 -- This will happen due to implicit reconnect
708 registerRemoteAsync nid1 "b" us
709 receiveWait [
710 matchIf (\(RegisterReply label' _) -> "b" == label')
711 (\(RegisterReply _ _) -> return ()) ]
713 -- Should happen
714 registerRemoteAsync nid1 "c" us
715 receiveWait [
716 matchIf (\(RegisterReply label' _) -> "c" == label')
717 (\(RegisterReply _ _) -> return ()) ]
719 -- Check
720 Nothing <- whereisRemote nid1 "a" -- this will fail because the name is removed when the node is disconnected
721 Just _ <- whereisRemote nid1 "b" -- this will suceed because the value is set after thereconnect
722 Just _ <- whereisRemote nid1 "c"
724 liftIO $ putMVar registerTestOk ()
726 takeMVar registerTestOk
728-- | Test 'matchAny'. This repeats the 'testMath' but with a proxy server
729-- in between
730testMatchAny :: NT.Transport -> Assertion
731testMatchAny transport = do
732 proxyAddr <- newEmptyMVar
733 clientDone <- newEmptyMVar
735 -- Math server
736 forkIO $ do
737 localNode <- newLocalNode transport initRemoteTable
738 mathServer <- forkProcess localNode math
739 proxyServer <- forkProcess localNode $ forever $ do
740 msg <- receiveWait [ matchAny return ]
741 forward msg mathServer
742 putMVar proxyAddr proxyServer
744 -- Client
745 forkIO $ do
746 localNode <- newLocalNode transport initRemoteTable
747 mathServer <- readMVar proxyAddr
749 runProcess localNode $ do
750 pid <- getSelfPid
751 send mathServer (Add pid 1 2)
752 3 <- expect :: Process Double
753 send mathServer (Divide pid 8 2)
754 4 <- expect :: Process Double
755 send mathServer (Divide pid 8 0)
756 DivByZero <- expect
757 liftIO $ putMVar clientDone ()
759 takeMVar clientDone
761-- | Test 'matchAny'. This repeats the 'testMath' but with a proxy server
762-- in between, however we block 'Divide' requests ....
763testMatchAnyHandle :: NT.Transport -> Assertion
764testMatchAnyHandle transport = do
765 proxyAddr <- newEmptyMVar
766 clientDone <- newEmptyMVar
768 -- Math server
769 forkIO $ do
770 localNode <- newLocalNode transport initRemoteTable
771 mathServer <- forkProcess localNode math
772 proxyServer <- forkProcess localNode $ forever $ do
773 receiveWait [
774 matchAny (maybeForward mathServer)
775 ]
776 putMVar proxyAddr proxyServer
778 -- Client
779 forkIO $ do
780 localNode <- newLocalNode transport initRemoteTable
781 mathServer <- readMVar proxyAddr
783 runProcess localNode $ do
784 pid <- getSelfPid
785 send mathServer (Add pid 1 2)
786 3 <- expect :: Process Double
787 send mathServer (Divide pid 8 2)
788 Nothing <- (expectTimeout 100000) :: Process (Maybe Double)
789 liftIO $ putMVar clientDone ()
791 takeMVar clientDone
792 where maybeForward :: ProcessId -> AbstractMessage -> Process (Maybe ())
793 maybeForward s msg =
794 maybeHandleMessage msg (\m@(Add _ _ _) -> send s m)
796testMatchAnyNoHandle :: NT.Transport -> Assertion
797testMatchAnyNoHandle transport = do
798 addr <- newEmptyMVar
799 clientDone <- newEmptyMVar
800 serverDone <- newEmptyMVar
802 -- Math server
803 forkIO $ do
804 localNode <- newLocalNode transport initRemoteTable
805 server <- forkProcess localNode $ forever $ do
806 receiveWait [
807 matchAnyIf
808 -- the condition has type `Add -> Bool`
809 (\(Add _ _ _) -> True)
810 -- the match `AbstractMessage -> Process ()` will succeed!
811 (\m -> do
812 -- `String -> Process ()` does *not* match the input types however
813 r <- (maybeHandleMessage m (\(_ :: String) -> die "NONSENSE" ))
814 case r of
815 Nothing -> return ()
816 Just _ -> die "NONSENSE")
817 ]
818 -- we *must* have removed the message from our mailbox though!!!
819 Nothing <- receiveTimeout 100000 [ match (\(Add _ _ _) -> return ()) ]
820 liftIO $ putMVar serverDone ()
821 putMVar addr server
823 -- Client
824 forkIO $ do
825 localNode <- newLocalNode transport initRemoteTable
826 server <- readMVar addr
828 runProcess localNode $ do
829 pid <- getSelfPid
830 send server (Add pid 1 2)
831 -- we only care about the client having sent a message, so we're done
832 liftIO $ putMVar clientDone ()
834 takeMVar clientDone
835 takeMVar serverDone
837-- | Test 'matchAnyIf'. We provide an /echo/ server, but it ignores requests
838-- unless the text body @/= "bar"@ - this case should time out rather than
839-- removing the message from the process mailbox.
840testMatchAnyIf :: NT.Transport -> Assertion
841testMatchAnyIf transport = do
842 echoAddr <- newEmptyMVar
843 clientDone <- newEmptyMVar
845 -- echo server
846 forkIO $ do
847 localNode <- newLocalNode transport initRemoteTable
848 echoServer <- forkProcess localNode $ forever $ do
849 receiveWait [
850 matchAnyIf (\(_ :: ProcessId, (s :: String)) -> s /= "bar")
851 handleMessage
852 ]
853 putMVar echoAddr echoServer
855 -- Client
856 forkIO $ do
857 localNode <- newLocalNode transport initRemoteTable
858 server <- readMVar echoAddr
860 runProcess localNode $ do
861 pid <- getSelfPid
862 send server (pid, "foo")
863 "foo" <- expect
864 send server (pid, "baz")
865 "baz" <- expect
866 send server (pid, "bar")
867 Nothing <- (expectTimeout 100000) :: Process (Maybe Double)
868 liftIO $ putMVar clientDone ()
870 takeMVar clientDone
871 where handleMessage :: AbstractMessage -> Process (Maybe ())
872 handleMessage msg =
873 maybeHandleMessage msg (\(pid :: ProcessId, (m :: String))
874 -> do { send pid m; return () })
876-- Test 'receiveChanTimeout'
877testReceiveChanTimeout :: NT.Transport -> Assertion
878testReceiveChanTimeout transport = do
879 done <- newEmptyMVar
880 sendPort <- newEmptyMVar
882 forkTry $ do
883 localNode <- newLocalNode transport initRemoteTable
884 runProcess localNode $ do
885 -- Create a typed channel
886 (sp, rp) <- newChan :: Process (SendPort Bool, ReceivePort Bool)
887 liftIO $ putMVar sendPort sp
889 -- Wait for a message with a delay. No message arrives, we should get Nothing after 1 second
890 Nothing <- receiveChanTimeout 1000000 rp
892 -- Wait for a message with a delay again. Now a message arrives after 0.5 seconds
893 Just True <- receiveChanTimeout 1000000 rp
895 -- Wait for a message with zero timeout: non-blocking check. No message is available, we get Nothing
896 Nothing <- receiveChanTimeout 0 rp
898 -- Again, but now there is a message available
899 liftIO $ threadDelay 1000000
900 Just False <- receiveChanTimeout 0 rp
902 liftIO $ putMVar done ()
904 forkTry $ do
905 localNode <- newLocalNode transport initRemoteTable
906 runProcess localNode $ do
907 sp <- liftIO $ readMVar sendPort
909 liftIO $ threadDelay 1500000
910 sendChan sp True
912 liftIO $ threadDelay 500000
913 sendChan sp False
915 takeMVar done
917-- | Test Functor, Applicative, Alternative and Monad instances for ReceiveChan
918testReceiveChanFeatures :: NT.Transport -> Assertion
919testReceiveChanFeatures transport = do
920 done <- newEmptyMVar
922 forkTry $ do
923 localNode <- newLocalNode transport initRemoteTable
924 runProcess localNode $ do
925 (spInt, rpInt) <- newChan :: Process (SendPort Int, ReceivePort Int)
926 (spBool, rpBool) <- newChan :: Process (SendPort Bool, ReceivePort Bool)
928 -- Test Functor instance
930 sendChan spInt 2
931 sendChan spBool False
933 rp1 <- mergePortsBiased [even <$> rpInt, rpBool]
935 True <- receiveChan rp1
936 False <- receiveChan rp1
938 -- Test Applicative instance
940 sendChan spInt 3
941 sendChan spInt 4
943 let rp2 = pure (+) <*> rpInt <*> rpInt
945 7 <- receiveChan rp2
947 -- Test Alternative instance
949 sendChan spInt 3
950 sendChan spBool True
952 let rp3 = (even <$> rpInt) <|> rpBool
954 False <- receiveChan rp3
955 True <- receiveChan rp3
957 -- Test Monad instance
959 sendChan spBool True
960 sendChan spBool False
961 sendChan spInt 5
963 let rp4 :: ReceivePort Int
964 rp4 = do b <- rpBool
965 if b
966 then rpInt
967 else return 7
969 5 <- receiveChan rp4
970 7 <- receiveChan rp4
972 liftIO $ putMVar done ()
974 takeMVar done
976testKillLocal :: NT.Transport -> Assertion
977testKillLocal transport = do
978 localNode <- newLocalNode transport initRemoteTable
979 done <- newEmptyMVar
981 pid <- forkProcess localNode $ do
982 liftIO $ threadDelay 1000000
984 runProcess localNode $ do
985 ref <- monitor pid
986 us <- getSelfPid
987 kill pid "TestKill"
988 ProcessMonitorNotification ref' pid' (DiedException ex) <- expect
989 True <- return $ ref == ref' && pid == pid' && ex == "killed-by=" ++ show us ++ ",reason=TestKill"
990 liftIO $ putMVar done ()
992 takeMVar done
994testKillRemote :: NT.Transport -> Assertion
995testKillRemote transport = do
996 node1 <- newLocalNode transport initRemoteTable
997 node2 <- newLocalNode transport initRemoteTable
998 done <- newEmptyMVar
1000 pid <- forkProcess node1 $ do
1001 liftIO $ threadDelay 1000000
1003 runProcess node2 $ do
1004 ref <- monitor pid
1005 us <- getSelfPid
1006 kill pid "TestKill"
1007 ProcessMonitorNotification ref' pid' (DiedException reason) <- expect
1008 True <- return $ ref == ref' && pid == pid' && reason == "killed-by=" ++ show us ++ ",reason=TestKill"
1009 liftIO $ putMVar done ()
1011 takeMVar done
1013testCatchesExit :: NT.Transport -> Assertion
1014testCatchesExit transport = do
1015 localNode <- newLocalNode transport initRemoteTable
1016 done <- newEmptyMVar
1018 _ <- forkProcess localNode $ do
1019 (die ("foobar", 123 :: Int))
1020 `catchesExit` [
1021 (\_ m -> maybeHandleMessage m (\(_ :: String) -> return ()))
1022 , (\_ m -> maybeHandleMessage m (\(_ :: Maybe Int) -> return ()))
1023 , (\_ m -> maybeHandleMessage m (\(_ :: String, _ :: Int)
1024 -> (liftIO $ putMVar done ()) >> return ()))
1027 takeMVar done
1029testCatches :: NT.Transport -> Assertion
1030testCatches transport = do
1031 localNode <- newLocalNode transport initRemoteTable
1032 done <- newEmptyMVar
1034 _ <- forkProcess localNode $ do
1035 node <- getSelfNode
1036 (liftIO $ throwIO (ProcessLinkException (nullProcessId node) DiedNormal))
1037 `catches` [
1038 Handler (\(ProcessLinkException _ _) -> liftIO $ putMVar done ())
1041 takeMVar done
1043testDie :: NT.Transport -> Assertion
1044testDie transport = do
1045 localNode <- newLocalNode transport initRemoteTable
1046 done <- newEmptyMVar
1048 _ <- forkProcess localNode $ do
1049 (die ("foobar", 123 :: Int))
1050 `catchExit` \_from reason -> do
1051 -- TODO: should verify that 'from' has the right value
1052 True <- return $ reason == ("foobar", 123 :: Int)
1053 liftIO $ putMVar done ()
1055 takeMVar done
1057testPrettyExit :: NT.Transport -> Assertion
1058testPrettyExit transport = do
1059 localNode <- newLocalNode transport initRemoteTable
1060 done <- newEmptyMVar
1062 _ <- forkProcess localNode $ do
1063 (die "timeout")
1064 `catch` \ex@(ProcessExitException from _) ->
1065 let expected = "exit-from=" ++ (show from)
1066 in do
1067 True <- return $ (show ex) == expected
1068 liftIO $ putMVar done ()
1070 takeMVar done
1072testExitLocal :: NT.Transport -> Assertion
1073testExitLocal transport = do
1074 localNode <- newLocalNode transport initRemoteTable
1075 supervisedDone <- newEmptyMVar
1076 supervisorDone <- newEmptyMVar
1078 pid <- forkProcess localNode $ do
1079 (liftIO $ threadDelay 100000)
1080 `catchExit` \_from reason -> do
1081 -- TODO: should verify that 'from' has the right value
1082 True <- return $ reason == "TestExit"
1083 liftIO $ putMVar supervisedDone ()
1085 runProcess localNode $ do
1086 ref <- monitor pid
1087 exit pid "TestExit"
1088 -- This time the client catches the exception, so it dies normally
1089 ProcessMonitorNotification ref' pid' DiedNormal <- expect
1090 True <- return $ ref == ref' && pid == pid'
1091 liftIO $ putMVar supervisorDone ()
1093 takeMVar supervisedDone
1094 takeMVar supervisorDone
1096testExitRemote :: NT.Transport -> Assertion
1097testExitRemote transport = do
1098 node1 <- newLocalNode transport initRemoteTable
1099 node2 <- newLocalNode transport initRemoteTable
1100 supervisedDone <- newEmptyMVar
1101 supervisorDone <- newEmptyMVar
1103 pid <- forkProcess node1 $ do
1104 (liftIO $ threadDelay 100000)
1105 `catchExit` \_from reason -> do
1106 -- TODO: should verify that 'from' has the right value
1107 True <- return $ reason == "TestExit"
1108 liftIO $ putMVar supervisedDone ()
1110 runProcess node2 $ do
1111 ref <- monitor pid
1112 exit pid "TestExit"
1113 ProcessMonitorNotification ref' pid' DiedNormal <- expect
1114 True <- return $ ref == ref' && pid == pid'
1115 liftIO $ putMVar supervisorDone ()
1117 takeMVar supervisedDone
1118 takeMVar supervisorDone
1120tests :: (NT.Transport, TransportInternals) -> [Test]
1121tests (transport, transportInternals) = [
1122 testGroup "Basic features" [
1123 testCase "Ping" (testPing transport)
1124 , testCase "Math" (testMath transport)
1125 , testCase "Timeout" (testTimeout transport)
1126 , testCase "Timeout0" (testTimeout0 transport)
1127 , testCase "SendToTerminated" (testSendToTerminated transport)
1128 , testCase "TypedChannnels" (testTypedChannels transport)
1129 , testCase "MergeChannels" (testMergeChannels transport)
1130 , testCase "Terminate" (testTerminate transport)
1131 , testCase "Registry" (testRegistry transport)
1132 , testCase "RemoteRegistry" (testRemoteRegistry transport)
1133 , testCase "SpawnLocal" (testSpawnLocal transport)
1134 , testCase "MatchAny" (testMatchAny transport)
1135 , testCase "MatchAnyHandle" (testMatchAnyHandle transport)
1136 , testCase "MatchAnyNoHandle" (testMatchAnyNoHandle transport)
1137 , testCase "MatchAnyIf" (testMatchAnyIf transport)
1138 , testCase "ReceiveChanTimeout" (testReceiveChanTimeout transport)
1139 , testCase "ReceiveChanFeatures" (testReceiveChanFeatures transport)
1140 , testCase "KillLocal" (testKillLocal transport)
1141 , testCase "KillRemote" (testKillRemote transport)
1142 , testCase "Die" (testDie transport)
1143 , testCase "PrettyExit" (testPrettyExit transport)
1144 , testCase "CatchesExit" (testCatchesExit transport)
1145 , testCase "Catches" (testCatches transport)
1146 , testCase "ExitLocal" (testExitLocal transport)
1147 , testCase "ExitRemote" (testExitRemote transport)
1149 , testGroup "Monitoring and Linking" [
1150 -- Monitoring processes
1151 --
1152 -- The "missing" combinations in the list below don't make much sense, as
1153 -- we cannot guarantee that the monitor reply or link exception will not
1154 -- happen before the unmonitor or unlink
1155 testCase "MonitorUnreachable" (testMonitorUnreachable transport True False)
1156 , testCase "MonitorNormalTermination" (testMonitorNormalTermination transport True False)
1157 , testCase "MonitorAbnormalTermination" (testMonitorAbnormalTermination transport True False)
1158 , testCase "MonitorLocalDeadProcess" (testMonitorLocalDeadProcess transport True False)
1159 , testCase "MonitorRemoteDeadProcess" (testMonitorRemoteDeadProcess transport True False)
1160 , testCase "MonitorDisconnect" (testMonitorDisconnect transport True False)
1161 , testCase "LinkUnreachable" (testMonitorUnreachable transport False False)
1162 , testCase "LinkNormalTermination" (testMonitorNormalTermination transport False False)
1163 , testCase "LinkAbnormalTermination" (testMonitorAbnormalTermination transport False False)
1164 , testCase "LinkLocalDeadProcess" (testMonitorLocalDeadProcess transport False False)
1165 , testCase "LinkRemoteDeadProcess" (testMonitorRemoteDeadProcess transport False False)
1166 , testCase "LinkDisconnect" (testMonitorDisconnect transport False False)
1167 , testCase "UnmonitorNormalTermination" (testMonitorNormalTermination transport True True)
1168 , testCase "UnmonitorAbnormalTermination" (testMonitorAbnormalTermination transport True True)
1169 , testCase "UnmonitorDisconnect" (testMonitorDisconnect transport True True)
1170 , testCase "UnlinkNormalTermination" (testMonitorNormalTermination transport False True)
1171 , testCase "UnlinkAbnormalTermination" (testMonitorAbnormalTermination transport False True)
1172 , testCase "UnlinkDisconnect" (testMonitorDisconnect transport False True)
1173 -- Monitoring nodes and channels
1174 , testCase "MonitorNode" (testMonitorNode transport)
1175 , testCase "MonitorChannel" (testMonitorChannel transport)
1176 -- Reconnect
1177 , testCase "Reconnect" (testReconnect transport transportInternals)
1181main :: IO ()
1182main = do
1183 Right transport <- createTransportExposeInternals "127.0.0.1" "8080" defaultTCPParameters
1184 defaultMain (tests transport)