git.iko.soy/distributed-process
on send-direct help sign in
distributed-process/tests/TestCH.hs
1213 lines 41 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.Node
31import Control.Distributed.Process.Serializable (Serializable)
33import Test.HUnit (Assertion)
34import Test.Framework (Test, defaultMain, testGroup)
35import Test.Framework.Providers.HUnit (testCase)
37newtype Ping = Ping ProcessId
38 deriving (Typeable, Binary, Show)
40newtype Pong = Pong ProcessId
41 deriving (Typeable, Binary, Show)
43--------------------------------------------------------------------------------
44-- Supporting definitions --
45--------------------------------------------------------------------------------
47-- | Like fork, but throw exceptions in the child thread to the parent
48forkTry :: IO () -> IO ThreadId
49forkTry p = do
50 tid <- myThreadId
51 forkIO $ Ex.catch p (\e -> throwTo tid (e :: SomeException))
53-- | The ping server from the paper
54ping :: Process ()
55ping = do
56 Pong partner <- expect
57 self <- getSelfPid
58 send partner (Ping self)
59 ping
61-- | Quick and dirty synchronous version of whereisRemoteAsync
62whereisRemote :: NodeId -> String -> Process (Maybe ProcessId)
63whereisRemote nid string = do
64 whereisRemoteAsync nid string
65 WhereIsReply _ mPid <- expect
66 return mPid
68data Add = Add ProcessId Double Double deriving (Typeable)
69data Divide = Divide ProcessId Double Double deriving (Typeable)
70data DivByZero = DivByZero deriving (Typeable)
72instance Binary Add where
73 put (Add pid x y) = put pid >> put x >> put y
74 get = Add <$> get <*> get <*> get
76instance Binary Divide where
77 put (Divide pid x y) = put pid >> put x >> put y
78 get = Divide <$> get <*> get <*> get
80instance Binary DivByZero where
81 put DivByZero = return ()
82 get = return DivByZero
84-- The math server from the paper
85math :: Process ()
86math = do
87 receiveWait
88 [ match (\(Add pid x y) -> send pid (x + y))
89 , matchIf (\(Divide _ _ y) -> y /= 0)
90 (\(Divide pid x y) -> send pid (x / y))
91 , match (\(Divide pid _ _) -> send pid DivByZero)
92 ]
93 math
95-- | Monitor or link to a remote node
96monitorOrLink :: Bool -- ^ 'True' for monitor, 'False' for link
97 -> ProcessId -- Process to monitor/link to
98 -> Maybe (MVar ()) -- MVar to signal on once the monitor has been set up
99 -> Process (Maybe MonitorRef)
100monitorOrLink mOrL pid mSignal = do
101 result <- if mOrL then Just <$> monitor pid
102 else link pid >> return Nothing
103 -- Monitor is asynchronous, which usually does not matter but if we want a
104 -- *specific* signal then it does. Therefore we wait an arbitrary delay and
105 -- hope that this means the monitor has been set up
106 forM_ mSignal $ \signal -> liftIO . forkIO $ threadDelay 100000 >> putMVar signal ()
107 return result
109monitorTestProcess :: ProcessId -- Process to monitor/link to
110 -> Bool -- 'True' for monitor, 'False' for link
111 -> Bool -- Should we unmonitor?
112 -> DiedReason -- Expected cause of death
113 -> Maybe (MVar ()) -- Signal for 'monitor set up'
114 -> MVar () -- Signal for successful termination
115 -> Process ()
116monitorTestProcess theirAddr mOrL un reason monitorSetup done =
117 catch (do mRef <- monitorOrLink mOrL theirAddr monitorSetup
118 case (un, mRef) of
119 (True, Nothing) -> do
120 unlink theirAddr
121 liftIO $ putMVar done ()
122 (True, Just ref) -> do
123 unmonitor ref
124 liftIO $ putMVar done ()
125 (False, ref) -> do
126 ProcessMonitorNotification ref' pid reason' <- expect
127 True <- return $ Just ref' == ref && pid == theirAddr && mOrL && reason == reason'
128 liftIO $ putMVar done ()
129 )
130 (\(ProcessLinkException pid reason') -> do
131 True <- return $ pid == theirAddr && not mOrL && not un && reason == reason'
132 liftIO $ putMVar done ()
133 )
135--------------------------------------------------------------------------------
136-- The tests proper --
137--------------------------------------------------------------------------------
139-- | Basic ping test
140testPing :: NT.Transport -> Assertion
141testPing transport = do
142 serverAddr <- newEmptyMVar
143 clientDone <- newEmptyMVar
145 -- Server
146 forkIO $ do
147 localNode <- newLocalNode transport initRemoteTable
148 addr <- forkProcess localNode ping
149 putMVar serverAddr addr
151 -- Client
152 forkIO $ do
153 localNode <- newLocalNode transport initRemoteTable
154 pingServer <- readMVar serverAddr
156 let numPings = 10000
158 runProcess localNode $ do
159 pid <- getSelfPid
160 replicateM_ numPings $ do
161 send pingServer (Pong pid)
162 Ping _ <- expect
163 return ()
165 putMVar clientDone ()
167 takeMVar clientDone
169-- | Monitor a process on an unreachable node
170testMonitorUnreachable :: NT.Transport -> Bool -> Bool -> Assertion
171testMonitorUnreachable transport mOrL un = do
172 deadProcess <- newEmptyMVar
173 done <- newEmptyMVar
175 forkIO $ do
176 localNode <- newLocalNode transport initRemoteTable
177 addr <- forkProcess localNode . liftIO $ threadDelay 1000000
178 closeLocalNode localNode
179 putMVar deadProcess addr
181 forkIO $ do
182 localNode <- newLocalNode transport initRemoteTable
183 theirAddr <- readMVar deadProcess
184 runProcess localNode $
185 monitorTestProcess theirAddr mOrL un DiedDisconnect Nothing done
187 takeMVar done
189-- | Monitor a process which terminates normally
190testMonitorNormalTermination :: NT.Transport -> Bool -> Bool -> Assertion
191testMonitorNormalTermination transport mOrL un = do
192 monitorSetup <- newEmptyMVar
193 monitoredProcess <- newEmptyMVar
194 done <- newEmptyMVar
196 forkIO $ do
197 localNode <- newLocalNode transport initRemoteTable
198 addr <- forkProcess localNode $
199 liftIO $ readMVar monitorSetup
200 putMVar monitoredProcess addr
202 forkIO $ do
203 localNode <- newLocalNode transport initRemoteTable
204 theirAddr <- readMVar monitoredProcess
205 runProcess localNode $
206 monitorTestProcess theirAddr mOrL un DiedNormal (Just monitorSetup) done
208 takeMVar done
210-- | Monitor a process which terminates abnormally
211testMonitorAbnormalTermination :: NT.Transport -> Bool -> Bool -> Assertion
212testMonitorAbnormalTermination transport mOrL un = do
213 monitorSetup <- newEmptyMVar
214 monitoredProcess <- newEmptyMVar
215 done <- newEmptyMVar
217 let err = userError "Abnormal termination"
219 forkIO $ do
220 localNode <- newLocalNode transport initRemoteTable
221 addr <- forkProcess localNode . liftIO $ do
222 readMVar monitorSetup
223 throwIO err
224 putMVar monitoredProcess addr
226 forkIO $ do
227 localNode <- newLocalNode transport initRemoteTable
228 theirAddr <- readMVar monitoredProcess
229 runProcess localNode $
230 monitorTestProcess theirAddr mOrL un (DiedException (show err)) (Just monitorSetup) done
232 takeMVar done
234-- | Monitor a local process that is already dead
235testMonitorLocalDeadProcess :: NT.Transport -> Bool -> Bool -> Assertion
236testMonitorLocalDeadProcess transport mOrL un = do
237 processDead <- newEmptyMVar
238 processAddr <- newEmptyMVar
239 localNode <- newLocalNode transport initRemoteTable
240 done <- newEmptyMVar
242 forkIO $ do
243 addr <- forkProcess localNode . liftIO $ putMVar processDead ()
244 putMVar processAddr addr
246 forkIO $ do
247 theirAddr <- readMVar processAddr
248 readMVar processDead
249 runProcess localNode $ do
250 monitorTestProcess theirAddr mOrL un DiedUnknownId Nothing done
252 takeMVar done
254-- | Monitor a remote process that is already dead
255testMonitorRemoteDeadProcess :: NT.Transport -> Bool -> Bool -> Assertion
256testMonitorRemoteDeadProcess transport mOrL un = do
257 processDead <- newEmptyMVar
258 processAddr <- newEmptyMVar
259 done <- newEmptyMVar
261 forkIO $ do
262 localNode <- newLocalNode transport initRemoteTable
263 addr <- forkProcess localNode . liftIO $ putMVar processDead ()
264 putMVar processAddr addr
266 forkIO $ do
267 localNode <- newLocalNode transport initRemoteTable
268 theirAddr <- readMVar processAddr
269 readMVar processDead
270 runProcess localNode $ do
271 monitorTestProcess theirAddr mOrL un DiedUnknownId Nothing done
273 takeMVar done
275-- | Monitor a process that becomes disconnected
276testMonitorDisconnect :: NT.Transport -> Bool -> Bool -> Assertion
277testMonitorDisconnect transport mOrL un = do
278 processAddr <- newEmptyMVar
279 monitorSetup <- newEmptyMVar
280 done <- newEmptyMVar
282 forkIO $ do
283 localNode <- newLocalNode transport initRemoteTable
284 addr <- forkProcess localNode . liftIO $ threadDelay 1000000
285 putMVar processAddr addr
286 readMVar monitorSetup
287 NT.closeEndPoint (localEndPoint localNode)
289 forkIO $ do
290 localNode <- newLocalNode transport initRemoteTable
291 theirAddr <- readMVar processAddr
292 runProcess localNode $ do
293 monitorTestProcess theirAddr mOrL un DiedDisconnect (Just monitorSetup) done
295 takeMVar done
297-- | Test the math server (i.e., receiveWait)
298testMath :: NT.Transport -> Assertion
299testMath transport = do
300 serverAddr <- newEmptyMVar
301 clientDone <- newEmptyMVar
303 -- Server
304 forkIO $ do
305 localNode <- newLocalNode transport initRemoteTable
306 addr <- forkProcess localNode math
307 putMVar serverAddr addr
309 -- Client
310 forkIO $ do
311 localNode <- newLocalNode transport initRemoteTable
312 mathServer <- readMVar serverAddr
314 runProcess localNode $ do
315 pid <- getSelfPid
316 send mathServer (Add pid 1 2)
317 3 <- expect :: Process Double
318 send mathServer (Divide pid 8 2)
319 4 <- expect :: Process Double
320 send mathServer (Divide pid 8 0)
321 DivByZero <- expect
322 liftIO $ putMVar clientDone ()
324 takeMVar clientDone
326-- | Send first message (i.e. connect) to an already terminated process
327-- (without monitoring); then send another message to a second process on
328-- the same remote node (we're checking that the remote node did not die)
329testSendToTerminated :: NT.Transport -> Assertion
330testSendToTerminated transport = do
331 serverAddr1 <- newEmptyMVar
332 serverAddr2 <- newEmptyMVar
333 clientDone <- newEmptyMVar
335 forkIO $ do
336 terminated <- newEmptyMVar
337 localNode <- newLocalNode transport initRemoteTable
338 addr1 <- forkProcess localNode $ liftIO $ putMVar terminated ()
339 addr2 <- forkProcess localNode $ ping
340 readMVar terminated
341 putMVar serverAddr1 addr1
342 putMVar serverAddr2 addr2
344 forkIO $ do
345 localNode <- newLocalNode transport initRemoteTable
346 server1 <- readMVar serverAddr1
347 server2 <- readMVar serverAddr2
348 runProcess localNode $ do
349 pid <- getSelfPid
350 send server1 "Hi"
351 send server2 (Pong pid)
352 Ping pid' <- expect
353 True <- return $ pid' == server2
354 liftIO $ putMVar clientDone ()
356 takeMVar clientDone
358-- | Test (non-zero) timeout
359testTimeout :: NT.Transport -> Assertion
360testTimeout transport = do
361 localNode <- newLocalNode transport initRemoteTable
362 done <- newEmptyMVar
364 runProcess localNode $ do
365 Nothing <- receiveTimeout 1000000 [match (\(Add _ _ _) -> return ())]
366 liftIO $ putMVar done ()
368 takeMVar done
370-- | Test zero timeout
371testTimeout0 :: NT.Transport -> Assertion
372testTimeout0 transport = do
373 serverAddr <- newEmptyMVar
374 clientDone <- newEmptyMVar
375 messagesSent <- newEmptyMVar
377 forkIO $ do
378 localNode <- newLocalNode transport initRemoteTable
379 addr <- forkProcess localNode $ do
380 liftIO $ readMVar messagesSent >> threadDelay 1000000
381 -- Variation on the venerable ping server which uses a zero timeout
382 -- Since we wait for all messages to be sent before doing this receive,
383 -- we should nevertheless find the right message immediately
384 Just partner <- receiveTimeout 0 [match (\(Pong partner) -> return partner)]
385 self <- getSelfPid
386 send partner (Ping self)
387 putMVar serverAddr addr
389 forkIO $ do
390 localNode <- newLocalNode transport initRemoteTable
391 server <- readMVar serverAddr
392 runProcess localNode $ do
393 pid <- getSelfPid
394 -- Send a bunch of messages. A large number of messages that the server
395 -- is not interested in, and then a single message that it wants
396 replicateM_ 10000 $ send server "Irrelevant message"
397 send server (Pong pid)
398 liftIO $ putMVar messagesSent ()
399 Ping _ <- expect
400 liftIO $ putMVar clientDone ()
402 takeMVar clientDone
404-- | Test typed channels
405testTypedChannels :: NT.Transport -> Assertion
406testTypedChannels transport = do
407 serverChannel <- newEmptyMVar :: IO (MVar (SendPort (SendPort Bool, Int)))
408 clientDone <- newEmptyMVar
410 forkIO $ do
411 localNode <- newLocalNode transport initRemoteTable
412 forkProcess localNode $ do
413 (serverSendPort, rport) <- newChan
414 liftIO $ putMVar serverChannel serverSendPort
415 (clientSendPort, i) <- receiveChan rport
416 sendChan clientSendPort (even i)
417 return ()
419 forkIO $ do
420 localNode <- newLocalNode transport initRemoteTable
421 serverSendPort <- readMVar serverChannel
422 runProcess localNode $ do
423 (clientSendPort, rport) <- newChan
424 sendChan serverSendPort (clientSendPort, 5)
425 False <- receiveChan rport
426 liftIO $ putMVar clientDone ()
428 takeMVar clientDone
430-- | Test merging receive ports
431testMergeChannels :: NT.Transport -> Assertion
432testMergeChannels transport = do
433 localNode <- newLocalNode transport initRemoteTable
434 testFlat localNode True "aaabbbccc"
435 testFlat localNode False "abcabcabc"
436 testNested localNode True True "aaabbbcccdddeeefffggghhhiii"
437 testNested localNode True False "adgadgadgbehbehbehcficficfi"
438 testNested localNode False True "abcabcabcdefdefdefghighighi"
439 testNested localNode False False "adgbehcfiadgbehcfiadgbehcfi"
440 testBlocked localNode True
441 testBlocked localNode False
442 where
443 -- Single layer of merging
444 testFlat :: LocalNode -> Bool -> String -> IO ()
445 testFlat localNode biased expected = do
446 done <- newEmptyMVar
447 forkProcess localNode $ do
448 rs <- mapM charChannel "abc"
449 m <- mergePorts biased rs
450 xs <- replicateM 9 $ receiveChan m
451 True <- return $ xs == expected
452 liftIO $ putMVar done ()
453 takeMVar done
455 -- Two layers of merging
456 testNested :: LocalNode -> Bool -> Bool -> String -> IO ()
457 testNested localNode biasedInner biasedOuter expected = do
458 done <- newEmptyMVar
459 forkProcess localNode $ do
460 rss <- mapM (mapM charChannel) ["abc", "def", "ghi"]
461 ms <- mapM (mergePorts biasedInner) rss
462 m <- mergePorts biasedOuter ms
463 xs <- replicateM (9 * 3) $ receiveChan m
464 True <- return $ xs == expected
465 liftIO $ putMVar done ()
466 takeMVar done
468 -- Test that if no messages are (immediately) available, the scheduler makes no difference
469 testBlocked :: LocalNode -> Bool -> IO ()
470 testBlocked localNode biased = do
471 vs <- replicateM 3 newEmptyMVar
472 done <- newEmptyMVar
474 forkProcess localNode $ do
475 [sa, sb, sc] <- liftIO $ mapM readMVar vs
476 mapM_ ((>> liftIO (threadDelay 10000)) . uncurry sendChan)
477 [ -- a, b, c
478 (sa, 'a')
479 , (sb, 'b')
480 , (sc, 'c')
481 -- a, c, b
482 , (sa, 'a')
483 , (sc, 'c')
484 , (sb, 'b')
485 -- b, a, c
486 , (sb, 'b')
487 , (sa, 'a')
488 , (sc, 'c')
489 -- b, c, a
490 , (sb, 'b')
491 , (sc, 'c')
492 , (sa, 'a')
493 -- c, a, b
494 , (sc, 'c')
495 , (sa, 'a')
496 , (sb, 'b')
497 -- c, b, a
498 , (sc, 'c')
499 , (sb, 'b')
500 , (sa, 'a')
501 ]
503 forkProcess localNode $ do
504 (ss, rs) <- unzip <$> replicateM 3 newChan
505 liftIO $ mapM_ (uncurry putMVar) $ zip vs ss
506 m <- mergePorts biased rs
507 xs <- replicateM (6 * 3) $ receiveChan m
508 True <- return $ xs == "abcacbbacbcacabcba"
509 liftIO $ putMVar done ()
511 takeMVar done
513 mergePorts :: Serializable a => Bool -> [ReceivePort a] -> Process (ReceivePort a)
514 mergePorts True = mergePortsBiased
515 mergePorts False = mergePortsRR
517 charChannel :: Char -> Process (ReceivePort Char)
518 charChannel c = do
519 (sport, rport) <- newChan
520 replicateM_ 3 $ sendChan sport c
521 liftIO $ threadDelay 10000 -- Make sure messages have been sent
522 return rport
524testTerminate :: NT.Transport -> Assertion
525testTerminate transport = do
526 localNode <- newLocalNode transport initRemoteTable
527 done <- newEmptyMVar
529 pid <- forkProcess localNode $ do
530 liftIO $ threadDelay 100000
531 terminate
533 runProcess localNode $ do
534 ref <- monitor pid
535 ProcessMonitorNotification ref' pid' (DiedException ex) <- expect
536 True <- return $ ref == ref' && pid == pid' && ex == show ProcessTerminationException
537 liftIO $ putMVar done ()
539 takeMVar done
541testMonitorNode :: NT.Transport -> Assertion
542testMonitorNode transport = do
543 [node1, node2] <- replicateM 2 $ newLocalNode transport initRemoteTable
544 done <- newEmptyMVar
546 closeLocalNode node1
548 runProcess node2 $ do
549 ref <- monitorNode (localNodeId node1)
550 NodeMonitorNotification ref' nid DiedDisconnect <- expect
551 True <- return $ ref == ref' && nid == localNodeId node1
552 liftIO $ putMVar done ()
554 takeMVar done
556testMonitorChannel :: NT.Transport -> Assertion
557testMonitorChannel transport = do
558 [node1, node2] <- replicateM 2 $ newLocalNode transport initRemoteTable
559 gotNotification <- newEmptyMVar
561 pid <- forkProcess node1 $ do
562 sport <- expect :: Process (SendPort ())
563 ref <- monitorPort sport
564 PortMonitorNotification ref' port' reason <- expect
565 -- reason might be DiedUnknownId if the receive port is GCed before the
566 -- monitor is established (TODO: not sure that this is reasonable)
567 return $ ref' == ref && port' == sendPortId sport && (reason == DiedNormal || reason == DiedUnknownId)
568 liftIO $ putMVar gotNotification ()
570 runProcess node2 $ do
571 (sport, _) <- newChan :: Process (SendPort (), ReceivePort ())
572 send pid sport
573 liftIO $ threadDelay 100000
575 takeMVar gotNotification
577testRegistry :: NT.Transport -> Assertion
578testRegistry transport = do
579 node <- newLocalNode transport initRemoteTable
580 done <- newEmptyMVar
582 pingServer <- forkProcess node ping
584 runProcess node $ do
585 register "ping" pingServer
586 Just pid <- whereis "ping"
587 True <- return $ pingServer == pid
588 us <- getSelfPid
589 nsend "ping" (Pong us)
590 Ping pid' <- expect
591 True <- return $ pingServer == pid'
592 liftIO $ putMVar done ()
594 takeMVar done
596testRemoteRegistry :: NT.Transport -> Assertion
597testRemoteRegistry transport = do
598 node1 <- newLocalNode transport initRemoteTable
599 node2 <- newLocalNode transport initRemoteTable
600 done <- newEmptyMVar
602 pingServer <- forkProcess node1 ping
604 runProcess node2 $ do
605 let nid1 = localNodeId node1
606 registerRemoteAsync nid1 "ping" pingServer
607 receiveWait [
608 matchIf (\(RegisterReply label' _) -> "ping" == label')
609 (\(RegisterReply _ _) -> return ()) ]
611 Just pid <- whereisRemote nid1 "ping"
612 True <- return $ pingServer == pid
613 us <- getSelfPid
614 nsendRemote nid1 "ping" (Pong us)
615 Ping pid' <- expect
616 True <- return $ pingServer == pid'
617 liftIO $ putMVar done ()
619 takeMVar done
621testSpawnLocal :: NT.Transport -> Assertion
622testSpawnLocal transport = do
623 node <- newLocalNode transport initRemoteTable
624 done <- newEmptyMVar
626 runProcess node $ do
627 us <- getSelfPid
629 pid <- spawnLocal $ do
630 sport <- expect
631 sendChan sport (1234 :: Int)
633 sport <- spawnChannelLocal $ \rport -> do
634 (1234 :: Int) <- receiveChan rport
635 send us ()
637 send pid sport
638 () <- expect
639 liftIO $ putMVar done ()
641 takeMVar done
643testReconnect :: NT.Transport -> TransportInternals -> Assertion
644testReconnect transport transportInternals = do
645 [node1, node2] <- replicateM 2 $ newLocalNode transport initRemoteTable
646 let nid1 = localNodeId node1
647 nid2 = localNodeId node2
648 processA <- newEmptyMVar
649 [sendTestOk, registerTestOk] <- replicateM 2 newEmptyMVar
651 forkProcess node1 $ do
652 us <- getSelfPid
653 liftIO $ putMVar processA us
654 msg1 <- expect
655 msg2 <- expect
656 True <- return $ msg1 == "message 1" && msg2 == "message 3"
657 liftIO $ putMVar sendTestOk ()
659 forkProcess node2 $ do
660 {-
661 - Make sure there is no implicit reconnect on normal message sending
662 -}
664 them <- liftIO $ readMVar processA
665 send them "message 1" >> liftIO (threadDelay 100000)
667 -- Simulate network failure
668 liftIO $ do
669 sock <- socketBetween transportInternals (nodeAddress nid1) (nodeAddress nid2)
670 sClose sock
671 threadDelay 10000
673 -- Should not arrive
674 send them "message 2"
676 -- Should arrive
677 reconnect them
678 send them "message 3"
680 liftIO $ takeMVar sendTestOk
682 {-
683 - Test that there *is* implicit reconnect on node controller messages
684 -}
686 us <- getSelfPid
687 registerRemoteAsync nid1 "a" us -- registerRemote is asynchronous
688 receiveWait [
689 matchIf (\(RegisterReply label' _) -> "a" == label')
690 (\(RegisterReply _ _) -> return ()) ]
692 Just _ <- whereisRemote nid1 "a"
695 -- Simulate network failure
696 liftIO $ do
697 sock <- socketBetween transportInternals (nodeAddress nid1) (nodeAddress nid2)
698 sClose sock
699 threadDelay 10000
701 -- This will happen due to implicit reconnect
702 registerRemoteAsync nid1 "b" us
703 receiveWait [
704 matchIf (\(RegisterReply label' _) -> "b" == label')
705 (\(RegisterReply _ _) -> return ()) ]
707 -- Should happen
708 registerRemoteAsync nid1 "c" us
709 receiveWait [
710 matchIf (\(RegisterReply label' _) -> "c" == label')
711 (\(RegisterReply _ _) -> return ()) ]
713 -- Check
714 Nothing <- whereisRemote nid1 "a" -- this will fail because the name is removed when the node is disconnected
715 Just _ <- whereisRemote nid1 "b" -- this will suceed because the value is set after thereconnect
716 Just _ <- whereisRemote nid1 "c"
718 liftIO $ putMVar registerTestOk ()
720 takeMVar registerTestOk
722-- | Test 'matchAny'. This repeats the 'testMath' but with a proxy server
723-- in between
724testMatchAny :: NT.Transport -> Assertion
725testMatchAny transport = do
726 proxyAddr <- newEmptyMVar
727 clientDone <- newEmptyMVar
729 -- Math server
730 forkIO $ do
731 localNode <- newLocalNode transport initRemoteTable
732 mathServer <- forkProcess localNode math
733 proxyServer <- forkProcess localNode $ forever $ do
734 msg <- receiveWait [ matchAny return ]
735 forward msg mathServer
736 putMVar proxyAddr proxyServer
738 -- Client
739 forkIO $ do
740 localNode <- newLocalNode transport initRemoteTable
741 mathServer <- readMVar proxyAddr
743 runProcess localNode $ do
744 pid <- getSelfPid
745 send mathServer (Add pid 1 2)
746 3 <- expect :: Process Double
747 send mathServer (Divide pid 8 2)
748 4 <- expect :: Process Double
749 send mathServer (Divide pid 8 0)
750 DivByZero <- expect
751 liftIO $ putMVar clientDone ()
753 takeMVar clientDone
755-- | Test 'matchAny'. This repeats the 'testMath' but with a proxy server
756-- in between, however we block 'Divide' requests ....
757testMatchAnyHandle :: NT.Transport -> Assertion
758testMatchAnyHandle transport = do
759 proxyAddr <- newEmptyMVar
760 clientDone <- newEmptyMVar
762 -- Math server
763 forkIO $ do
764 localNode <- newLocalNode transport initRemoteTable
765 mathServer <- forkProcess localNode math
766 proxyServer <- forkProcess localNode $ forever $ do
767 receiveWait [
768 matchAny (maybeForward mathServer)
769 ]
770 putMVar proxyAddr proxyServer
772 -- Client
773 forkIO $ do
774 localNode <- newLocalNode transport initRemoteTable
775 mathServer <- readMVar proxyAddr
777 runProcess localNode $ do
778 pid <- getSelfPid
779 send mathServer (Add pid 1 2)
780 3 <- expect :: Process Double
781 send mathServer (Divide pid 8 2)
782 Nothing <- (expectTimeout 100000) :: Process (Maybe Double)
783 liftIO $ putMVar clientDone ()
785 takeMVar clientDone
786 where maybeForward :: ProcessId -> Message -> Process (Maybe ())
787 maybeForward s msg =
788 handleMessage msg (\m@(Add _ _ _) -> send s m)
790testMatchAnyNoHandle :: NT.Transport -> Assertion
791testMatchAnyNoHandle transport = do
792 addr <- newEmptyMVar
793 clientDone <- newEmptyMVar
794 serverDone <- newEmptyMVar
796 -- Math server
797 forkIO $ do
798 localNode <- newLocalNode transport initRemoteTable
799 server <- forkProcess localNode $ forever $ do
800 receiveWait [
801 matchAnyIf
802 -- the condition has type `Add -> Bool`
803 (\(Add _ _ _) -> True)
804 -- the match `AbstractMessage -> Process ()` will succeed!
805 (\m -> do
806 -- `String -> Process ()` does *not* match the input types however
807 r <- (handleMessage m (\(_ :: String) -> die "NONSENSE" ))
808 case r of
809 Nothing -> return ()
810 Just _ -> die "NONSENSE")
811 ]
812 -- we *must* have removed the message from our mailbox though!!!
813 Nothing <- receiveTimeout 100000 [ match (\(Add _ _ _) -> return ()) ]
814 liftIO $ putMVar serverDone ()
815 putMVar addr server
817 -- Client
818 forkIO $ do
819 localNode <- newLocalNode transport initRemoteTable
820 server <- readMVar addr
822 runProcess localNode $ do
823 pid <- getSelfPid
824 send server (Add pid 1 2)
825 -- we only care about the client having sent a message, so we're done
826 liftIO $ putMVar clientDone ()
828 takeMVar clientDone
829 takeMVar serverDone
831-- | Test 'matchAnyIf'. We provide an /echo/ server, but it ignores requests
832-- unless the text body @/= "bar"@ - this case should time out rather than
833-- removing the message from the process mailbox.
834testMatchAnyIf :: NT.Transport -> Assertion
835testMatchAnyIf transport = do
836 echoAddr <- newEmptyMVar
837 clientDone <- newEmptyMVar
839 -- echo server
840 forkIO $ do
841 localNode <- newLocalNode transport initRemoteTable
842 echoServer <- forkProcess localNode $ forever $ do
843 receiveWait [
844 matchAnyIf (\(_ :: ProcessId, (s :: String)) -> s /= "bar")
845 tryHandleMessage
846 ]
847 putMVar echoAddr echoServer
849 -- Client
850 forkIO $ do
851 localNode <- newLocalNode transport initRemoteTable
852 server <- readMVar echoAddr
854 runProcess localNode $ do
855 pid <- getSelfPid
856 send server (pid, "foo")
857 "foo" <- expect
858 send server (pid, "baz")
859 "baz" <- expect
860 send server (pid, "bar")
861 Nothing <- (expectTimeout 100000) :: Process (Maybe Double)
862 liftIO $ putMVar clientDone ()
864 takeMVar clientDone
865 where tryHandleMessage :: Message -> Process (Maybe ())
866 tryHandleMessage msg =
867 handleMessage msg (\(pid :: ProcessId, (m :: String))
868 -> do { send pid m; return () })
870testMatchMessageWithUnwrap :: NT.Transport -> Assertion
871testMatchMessageWithUnwrap transport = do
872 echoAddr <- newEmptyMVar
873 clientDone <- newEmptyMVar
875 -- echo server
876 forkIO $ do
877 localNode <- newLocalNode transport initRemoteTable
878 echoServer <- forkProcess localNode $ forever $ do
879 msg <- receiveWait [
880 matchMessage (\(m :: Message) -> do
881 return m)
882 ]
883 unwrapped <- unwrapMessage msg :: Process (Maybe (ProcessId, Message))
884 case unwrapped of
885 (Just (p, msg')) -> forward msg' p
886 Nothing -> die "unable to unwrap the message"
887 putMVar echoAddr echoServer
889 -- Client
890 forkIO $ do
891 localNode <- newLocalNode transport initRemoteTable
892 server <- readMVar echoAddr
894 runProcess localNode $ do
895 pid <- getSelfPid
896 send server (pid, wrapMessage ("foo" :: String))
897 "foo" <- expect
898 send server (pid, wrapMessage ("baz" :: String))
899 "baz" <- expect
900 liftIO $ putMVar clientDone ()
902 takeMVar clientDone
904-- Test 'receiveChanTimeout'
905testReceiveChanTimeout :: NT.Transport -> Assertion
906testReceiveChanTimeout transport = do
907 done <- newEmptyMVar
908 sendPort <- newEmptyMVar
910 forkTry $ do
911 localNode <- newLocalNode transport initRemoteTable
912 runProcess localNode $ do
913 -- Create a typed channel
914 (sp, rp) <- newChan :: Process (SendPort Bool, ReceivePort Bool)
915 liftIO $ putMVar sendPort sp
917 -- Wait for a message with a delay. No message arrives, we should get Nothing after 1 second
918 Nothing <- receiveChanTimeout 1000000 rp
920 -- Wait for a message with a delay again. Now a message arrives after 0.5 seconds
921 Just True <- receiveChanTimeout 1000000 rp
923 -- Wait for a message with zero timeout: non-blocking check. No message is available, we get Nothing
924 Nothing <- receiveChanTimeout 0 rp
926 -- Again, but now there is a message available
927 liftIO $ threadDelay 1000000
928 Just False <- receiveChanTimeout 0 rp
930 liftIO $ putMVar done ()
932 forkTry $ do
933 localNode <- newLocalNode transport initRemoteTable
934 runProcess localNode $ do
935 sp <- liftIO $ readMVar sendPort
937 liftIO $ threadDelay 1500000
938 sendChan sp True
940 liftIO $ threadDelay 500000
941 sendChan sp False
943 takeMVar done
945-- | Test Functor, Applicative, Alternative and Monad instances for ReceiveChan
946testReceiveChanFeatures :: NT.Transport -> Assertion
947testReceiveChanFeatures transport = do
948 done <- newEmptyMVar
950 forkTry $ do
951 localNode <- newLocalNode transport initRemoteTable
952 runProcess localNode $ do
953 (spInt, rpInt) <- newChan :: Process (SendPort Int, ReceivePort Int)
954 (spBool, rpBool) <- newChan :: Process (SendPort Bool, ReceivePort Bool)
956 -- Test Functor instance
958 sendChan spInt 2
959 sendChan spBool False
961 rp1 <- mergePortsBiased [even <$> rpInt, rpBool]
963 True <- receiveChan rp1
964 False <- receiveChan rp1
966 -- Test Applicative instance
968 sendChan spInt 3
969 sendChan spInt 4
971 let rp2 = pure (+) <*> rpInt <*> rpInt
973 7 <- receiveChan rp2
975 -- Test Alternative instance
977 sendChan spInt 3
978 sendChan spBool True
980 let rp3 = (even <$> rpInt) <|> rpBool
982 False <- receiveChan rp3
983 True <- receiveChan rp3
985 -- Test Monad instance
987 sendChan spBool True
988 sendChan spBool False
989 sendChan spInt 5
991 let rp4 :: ReceivePort Int
992 rp4 = do b <- rpBool
993 if b
994 then rpInt
995 else return 7
997 5 <- receiveChan rp4
998 7 <- receiveChan rp4
1000 liftIO $ putMVar done ()
1002 takeMVar done
1004testKillLocal :: NT.Transport -> Assertion
1005testKillLocal transport = do
1006 localNode <- newLocalNode transport initRemoteTable
1007 done <- newEmptyMVar
1009 pid <- forkProcess localNode $ do
1010 liftIO $ threadDelay 1000000
1012 runProcess localNode $ do
1013 ref <- monitor pid
1014 us <- getSelfPid
1015 kill pid "TestKill"
1016 ProcessMonitorNotification ref' pid' (DiedException ex) <- expect
1017 True <- return $ ref == ref' && pid == pid' && ex == "killed-by=" ++ show us ++ ",reason=TestKill"
1018 liftIO $ putMVar done ()
1020 takeMVar done
1022testKillRemote :: NT.Transport -> Assertion
1023testKillRemote transport = do
1024 node1 <- newLocalNode transport initRemoteTable
1025 node2 <- newLocalNode transport initRemoteTable
1026 done <- newEmptyMVar
1028 pid <- forkProcess node1 $ do
1029 liftIO $ threadDelay 1000000
1031 runProcess node2 $ do
1032 ref <- monitor pid
1033 us <- getSelfPid
1034 kill pid "TestKill"
1035 ProcessMonitorNotification ref' pid' (DiedException reason) <- expect
1036 True <- return $ ref == ref' && pid == pid' && reason == "killed-by=" ++ show us ++ ",reason=TestKill"
1037 liftIO $ putMVar done ()
1039 takeMVar done
1041testCatchesExit :: NT.Transport -> Assertion
1042testCatchesExit transport = do
1043 localNode <- newLocalNode transport initRemoteTable
1044 done <- newEmptyMVar
1046 _ <- forkProcess localNode $ do
1047 (die ("foobar", 123 :: Int))
1048 `catchesExit` [
1049 (\_ m -> handleMessage m (\(_ :: String) -> return ()))
1050 , (\_ m -> handleMessage m (\(_ :: Maybe Int) -> return ()))
1051 , (\_ m -> handleMessage m (\(_ :: String, _ :: Int)
1052 -> (liftIO $ putMVar done ()) >> return ()))
1055 takeMVar done
1057testCatches :: NT.Transport -> Assertion
1058testCatches transport = do
1059 localNode <- newLocalNode transport initRemoteTable
1060 done <- newEmptyMVar
1062 _ <- forkProcess localNode $ do
1063 node <- getSelfNode
1064 (liftIO $ throwIO (ProcessLinkException (nullProcessId node) DiedNormal))
1065 `catches` [
1066 Handler (\(ProcessLinkException _ _) -> liftIO $ putMVar done ())
1069 takeMVar done
1071testDie :: NT.Transport -> Assertion
1072testDie transport = do
1073 localNode <- newLocalNode transport initRemoteTable
1074 done <- newEmptyMVar
1076 _ <- forkProcess localNode $ do
1077 (die ("foobar", 123 :: Int))
1078 `catchExit` \_from reason -> do
1079 -- TODO: should verify that 'from' has the right value
1080 True <- return $ reason == ("foobar", 123 :: Int)
1081 liftIO $ putMVar done ()
1083 takeMVar done
1085testPrettyExit :: NT.Transport -> Assertion
1086testPrettyExit transport = do
1087 localNode <- newLocalNode transport initRemoteTable
1088 done <- newEmptyMVar
1090 _ <- forkProcess localNode $ do
1091 (die "timeout")
1092 `catch` \(ex :: ProcessExitException) ->
1093 let expected = "exit-from=" ++ (show (exitSource ex))
1094 in do
1095 True <- return $ (show ex) == expected
1096 liftIO $ putMVar done ()
1098 takeMVar done
1100testExitLocal :: NT.Transport -> Assertion
1101testExitLocal transport = do
1102 localNode <- newLocalNode transport initRemoteTable
1103 supervisedDone <- newEmptyMVar
1104 supervisorDone <- newEmptyMVar
1106 pid <- forkProcess localNode $ do
1107 (liftIO $ threadDelay 100000)
1108 `catchExit` \_from reason -> do
1109 -- TODO: should verify that 'from' has the right value
1110 True <- return $ reason == "TestExit"
1111 liftIO $ putMVar supervisedDone ()
1113 runProcess localNode $ do
1114 ref <- monitor pid
1115 exit pid "TestExit"
1116 -- This time the client catches the exception, so it dies normally
1117 ProcessMonitorNotification ref' pid' DiedNormal <- expect
1118 True <- return $ ref == ref' && pid == pid'
1119 liftIO $ putMVar supervisorDone ()
1121 takeMVar supervisedDone
1122 takeMVar supervisorDone
1124testExitRemote :: NT.Transport -> Assertion
1125testExitRemote transport = do
1126 node1 <- newLocalNode transport initRemoteTable
1127 node2 <- newLocalNode transport initRemoteTable
1128 supervisedDone <- newEmptyMVar
1129 supervisorDone <- newEmptyMVar
1131 pid <- forkProcess node1 $ do
1132 (liftIO $ threadDelay 100000)
1133 `catchExit` \_from reason -> do
1134 -- TODO: should verify that 'from' has the right value
1135 True <- return $ reason == "TestExit"
1136 liftIO $ putMVar supervisedDone ()
1138 runProcess node2 $ do
1139 ref <- monitor pid
1140 exit pid "TestExit"
1141 ProcessMonitorNotification ref' pid' DiedNormal <- expect
1142 True <- return $ ref == ref' && pid == pid'
1143 liftIO $ putMVar supervisorDone ()
1145 takeMVar supervisedDone
1146 takeMVar supervisorDone
1148tests :: (NT.Transport, TransportInternals) -> [Test]
1149tests (transport, transportInternals) = [
1150 testGroup "Basic features" [
1151 testCase "Ping" (testPing transport)
1152 , testCase "Math" (testMath transport)
1153 , testCase "Timeout" (testTimeout transport)
1154 , testCase "Timeout0" (testTimeout0 transport)
1155 , testCase "SendToTerminated" (testSendToTerminated transport)
1156 , testCase "TypedChannnels" (testTypedChannels transport)
1157 , testCase "MergeChannels" (testMergeChannels transport)
1158 , testCase "Terminate" (testTerminate transport)
1159 , testCase "Registry" (testRegistry transport)
1160 , testCase "RemoteRegistry" (testRemoteRegistry transport)
1161 , testCase "SpawnLocal" (testSpawnLocal transport)
1162 , testCase "MatchAny" (testMatchAny transport)
1163 , testCase "MatchAnyHandle" (testMatchAnyHandle transport)
1164 , testCase "MatchAnyNoHandle" (testMatchAnyNoHandle transport)
1165 , testCase "MatchAnyIf" (testMatchAnyIf transport)
1166 , testCase "MatchMessageUnwrap" (testMatchMessageWithUnwrap transport)
1167 , testCase "ReceiveChanTimeout" (testReceiveChanTimeout transport)
1168 , testCase "ReceiveChanFeatures" (testReceiveChanFeatures transport)
1169 , testCase "KillLocal" (testKillLocal transport)
1170 , testCase "KillRemote" (testKillRemote transport)
1171 , testCase "Die" (testDie transport)
1172 , testCase "PrettyExit" (testPrettyExit transport)
1173 , testCase "CatchesExit" (testCatchesExit transport)
1174 , testCase "Catches" (testCatches transport)
1175 , testCase "ExitLocal" (testExitLocal transport)
1176 , testCase "ExitRemote" (testExitRemote transport)
1178 , testGroup "Monitoring and Linking" [
1179 -- Monitoring processes
1180 --
1181 -- The "missing" combinations in the list below don't make much sense, as
1182 -- we cannot guarantee that the monitor reply or link exception will not
1183 -- happen before the unmonitor or unlink
1184 testCase "MonitorUnreachable" (testMonitorUnreachable transport True False)
1185 , testCase "MonitorNormalTermination" (testMonitorNormalTermination transport True False)
1186 , testCase "MonitorAbnormalTermination" (testMonitorAbnormalTermination transport True False)
1187 , testCase "MonitorLocalDeadProcess" (testMonitorLocalDeadProcess transport True False)
1188 , testCase "MonitorRemoteDeadProcess" (testMonitorRemoteDeadProcess transport True False)
1189 , testCase "MonitorDisconnect" (testMonitorDisconnect transport True False)
1190 , testCase "LinkUnreachable" (testMonitorUnreachable transport False False)
1191 , testCase "LinkNormalTermination" (testMonitorNormalTermination transport False False)
1192 , testCase "LinkAbnormalTermination" (testMonitorAbnormalTermination transport False False)
1193 , testCase "LinkLocalDeadProcess" (testMonitorLocalDeadProcess transport False False)
1194 , testCase "LinkRemoteDeadProcess" (testMonitorRemoteDeadProcess transport False False)
1195 , testCase "LinkDisconnect" (testMonitorDisconnect transport False False)
1196 , testCase "UnmonitorNormalTermination" (testMonitorNormalTermination transport True True)
1197 , testCase "UnmonitorAbnormalTermination" (testMonitorAbnormalTermination transport True True)
1198 , testCase "UnmonitorDisconnect" (testMonitorDisconnect transport True True)
1199 , testCase "UnlinkNormalTermination" (testMonitorNormalTermination transport False True)
1200 , testCase "UnlinkAbnormalTermination" (testMonitorAbnormalTermination transport False True)
1201 , testCase "UnlinkDisconnect" (testMonitorDisconnect transport False True)
1202 -- Monitoring nodes and channels
1203 , testCase "MonitorNode" (testMonitorNode transport)
1204 , testCase "MonitorChannel" (testMonitorChannel transport)
1205 -- Reconnect
1206 , testCase "Reconnect" (testReconnect transport transportInternals)
1210main :: IO ()
1211main = do
1212 Right transport <- createTransportExposeInternals "127.0.0.1" "8080" defaultTCPParameters
1213 defaultMain (tests transport)