distributed-process-tests 0.4.3.3 → 0.4.4
raw patch · 6 files changed
+540/−58 lines, 6 filesdep +network-transport-inmemorydep +setenvdep ~distributed-processdep ~networkdep ~network-transport-tcp
Dependencies added: network-transport-inmemory, setenv
Dependency ranges changed: distributed-process, network, network-transport-tcp
Files
- distributed-process-tests.cabal +36/−13
- src/Control/Distributed/Process/Tests/CH.hs +333/−32
- src/Control/Distributed/Process/Tests/Closure.hs +40/−11
- src/Control/Distributed/Process/Tests/Tracing.hs +93/−0
- tests/runInMemory.hs +31/−0
- tests/runTCP.hs +7/−2
distributed-process-tests.cabal view
@@ -1,5 +1,5 @@ name: distributed-process-tests-version: 0.4.3.3+version: 0.4.4 synopsis: Tests and test support tools for distributed-process. homepage: http://github.com/haskell-distributed/distributed-process-tests description: Tests and test suite for Cloud Haskell libraries.@@ -12,6 +12,10 @@ build-type: Simple cabal-version: >=1.8 +flag tcp+ Description: build and run TCP tests+ Default: False+ library exposed-modules: Network.Transport.Test Control.Distributed.Process.Tests.CH@@ -25,13 +29,14 @@ ansi-terminal >= 0.5 && < 0.7, binary >= 0.5 && < 0.8, bytestring >= 0.9 && < 0.11,- distributed-process >= 0.5.3 && < 0.6,+ distributed-process >= 0.6.0 && < 0.7, distributed-static, HUnit >= 1.2 && < 1.4, network-transport >= 0.4.1.0 && < 0.5, network >= 2.5 && < 2.7, random >= 1.0 && < 1.2, rematch >= 0.1.2.1,+ setenv >= 0.1.1.3, test-framework >= 0.6 && < 0.9, test-framework-hunit >= 0.2.0 && < 0.4, stm@@ -50,29 +55,47 @@ if impl(ghc <= 7.4.2) Build-Depends: ghc-prim == 0.2.0.0 -Test-Suite TestCH+Test-Suite TestCHInMemory Type: exitcode-stdio-1.0- Main-Is: runTCP.hs+ Main-Is: runInMemory.hs CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.CH Build-Depends: base >= 4.4 && < 5, distributed-process-tests, network >= 2.3 && < 2.7, network-transport >= 0.4.1.0 && < 0.5,- network-transport-tcp >= 0.3 && < 0.5,+ network-transport-inmemory >= 0.5, test-framework >= 0.6 && < 0.9 Extensions: CPP ghc-options: -Wall -threaded -eventlog -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind HS-Source-Dirs: tests -Test-Suite TestClosure+Test-Suite TestCHInTCP Type: exitcode-stdio-1.0 Main-Is: runTCP.hs+ CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.CH+ if flag(tcp)+ Build-Depends: base >= 4.4 && < 5,+ distributed-process-tests,+ network >= 2.3 && < 2.7,+ network-transport >= 0.4.1.0 && < 0.5,+ network-transport-tcp >= 0.5 && < 0.6,+ test-framework >= 0.6 && < 0.9+ else+ Buildable: False+ Extensions: CPP+ ghc-options: -Wall -threaded -eventlog -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind+ HS-Source-Dirs: tests+++Test-Suite TestClosure+ Type: exitcode-stdio-1.0+ Main-Is: runInMemory.hs CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Closure Build-Depends: base >= 4.4 && < 5, distributed-process-tests, network >= 2.3 && < 2.7, network-transport >= 0.4.1.0 && < 0.5,- network-transport-tcp >= 0.3 && < 0.5,+ network-transport-inmemory >= 0.5, test-framework >= 0.6 && < 0.9 Extensions: CPP ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind@@ -80,13 +103,13 @@ Test-Suite TestStats Type: exitcode-stdio-1.0- Main-Is: runTCP.hs+ Main-Is: runInMemory.hs CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Stats Build-Depends: base >= 4.4 && < 5, distributed-process-tests, network >= 2.3 && < 2.7, network-transport >= 0.4.1.0 && < 0.5,- network-transport-tcp >= 0.3 && < 0.5,+ network-transport-inmemory >= 0.5, test-framework >= 0.6 && < 0.9 Extensions: CPP ghc-options: -Wall -eventlog -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind@@ -95,13 +118,13 @@ Test-Suite TestMx Type: exitcode-stdio-1.0- Main-Is: runTCP.hs+ Main-Is: runInMemory.hs CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Mx Build-Depends: base >= 4.4 && < 5, distributed-process-tests, network >= 2.3 && < 2.7, network-transport >= 0.4.1.0 && < 0.5,- network-transport-tcp >= 0.3 && < 0.5,+ network-transport-inmemory >= 0.5, test-framework >= 0.6 && < 0.9 Extensions: CPP ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind@@ -109,13 +132,13 @@ Test-Suite TestTracing Type: exitcode-stdio-1.0- Main-Is: runTCP.hs+ Main-Is: runInMemory.hs CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Tracing Build-Depends: base >= 4.4 && < 5, distributed-process-tests, network >= 2.3 && < 2.7, network-transport >= 0.4.1.0 && < 0.5,- network-transport-tcp >= 0.3 && < 0.5,+ network-transport-inmemory >= 0.5, test-framework >= 0.6 && < 0.9 Extensions: CPP ghc-options: -Wall -eventlog -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind
src/Control/Distributed/Process/Tests/CH.hs view
@@ -9,7 +9,13 @@ import Data.Binary (Binary(..)) import Data.Typeable (Typeable) import Data.Foldable (forM_)-import Control.Concurrent (forkIO, threadDelay, myThreadId, throwTo, ThreadId)+import Data.Function (fix)+import Data.IORef+ ( readIORef+ , writeIORef+ , newIORef+ )+import Control.Concurrent (forkIO, threadDelay, myThreadId, throwTo, ThreadId, yield) import Control.Concurrent.MVar ( MVar , newEmptyMVar@@ -17,17 +23,18 @@ , takeMVar , readMVar )-import Control.Monad (replicateM_, replicateM, forever, void, unless)-import Control.Exception (SomeException, throwIO)+import Control.Monad (replicateM_, replicateM, forever, void, unless, join)+import Control.Exception (SomeException, throwIO, ErrorCall(..)) import qualified Control.Exception as Ex (catch) import Control.Applicative ((<$>), (<*>), pure, (<|>))-import qualified Network.Transport as NT (closeEndPoint)+import qualified Network.Transport as NT (closeEndPoint, EndPointAddress) import Control.Distributed.Process import Control.Distributed.Process.Internal.Types ( NodeId(nodeAddress) , LocalNode(localEndPoint) , ProcessExitException(..) , nullProcessId+ , createUnencodedMessage ) import Control.Distributed.Process.Node import Control.Distributed.Process.Serializable (Serializable)@@ -199,7 +206,7 @@ forkIO $ do localNode <- newLocalNode testTransport initRemoteTable- addr <- forkProcess localNode . liftIO $ threadDelay 1000000+ addr <- forkProcess localNode expect closeLocalNode localNode putMVar deadProcess addr @@ -302,19 +309,25 @@ testMonitorDisconnect :: TestTransport -> Bool -> Bool -> Assertion testMonitorDisconnect TestTransport{..} mOrL un = do processAddr <- newEmptyMVar+ processAddr2 <- newEmptyMVar monitorSetup <- newEmptyMVar done <- newEmptyMVar forkIO $ do localNode <- newLocalNode testTransport initRemoteTable- addr <- forkProcess localNode . liftIO $ threadDelay 1000000+ addr <- forkProcess localNode $ expect+ addr2 <- forkProcess localNode $ return () putMVar processAddr addr readMVar monitorSetup NT.closeEndPoint (localEndPoint localNode)+ putMVar processAddr2 addr2 forkIO $ do localNode <- newLocalNode testTransport initRemoteTable theirAddr <- readMVar processAddr+ forkProcess localNode $ do+ lc <- liftIO $ readMVar processAddr2+ send lc () runProcess localNode $ do monitorTestProcess theirAddr mOrL un DiedDisconnect (Just monitorSetup) done @@ -398,16 +411,14 @@ testTimeout0 TestTransport{..} = do serverAddr <- newEmptyMVar clientDone <- newEmptyMVar- messagesSent <- newEmptyMVar forkIO $ do localNode <- newLocalNode testTransport initRemoteTable addr <- forkProcess localNode $ do- liftIO $ readMVar messagesSent >> threadDelay 1000000 -- Variation on the venerable ping server which uses a zero timeout- -- Since we wait for all messages to be sent before doing this receive,- -- we should nevertheless find the right message immediately- Just partner <- receiveTimeout 0 [match (\(Pong partner) -> return partner)]+ partner <- fix $ \loop ->+ receiveTimeout 0 [match (\(Pong partner) -> return partner)]+ >>= maybe (liftIO (threadDelay 100000) >> loop) return self <- getSelfPid send partner (Ping self) putMVar serverAddr addr@@ -421,7 +432,6 @@ -- is not interested in, and then a single message that it wants replicateM_ 10000 $ send server "Irrelevant message" send server (Pong pid)- liftIO $ putMVar messagesSent () Ping _ <- expect liftIO $ putMVar clientDone () @@ -583,17 +593,22 @@ testMonitorLiveNode TestTransport{..} = do [node1, node2] <- replicateM 2 $ newLocalNode testTransport initRemoteTable ready <- newEmptyMVar+ readyr <- newEmptyMVar done <- newEmptyMVar + p <- forkProcess node1 $ return () forkProcess node2 $ do ref <- monitorNode (localNodeId node1) liftIO $ putMVar ready ()+ liftIO $ takeMVar readyr+ send p () NodeMonitorNotification ref' nid _ <- expect True <- return $ ref == ref' && nid == localNodeId node1 liftIO $ putMVar done () takeMVar ready closeLocalNode node1+ putMVar readyr () takeMVar done @@ -624,6 +639,7 @@ done <- newEmptyMVar pingServer <- forkProcess node ping+ deadProcess <- forkProcess node (return ()) runProcess node $ do register "ping" pingServer@@ -633,10 +649,34 @@ nsend "ping" (Pong us) Ping pid' <- expect True <- return $ pingServer == pid'+ Left (ProcessRegistrationException "dead" Nothing) <- try $ register "dead" deadProcess+ Left (ProcessRegistrationException "ping" (Just x)) <- try $ register "ping" deadProcess+ True <- return $ x == pingServer+ Left (ProcessRegistrationException "dead" Nothing) <- try $ unregister "dead" liftIO $ putMVar done () takeMVar done +testRegistryRemoteProcess :: TestTransport -> Assertion+testRegistryRemoteProcess TestTransport{..} = do+ node1 <- newLocalNode testTransport initRemoteTable+ node2 <- newLocalNode testTransport initRemoteTable+ done <- newEmptyMVar++ pingServer <- forkProcess node1 ping++ runProcess node2 $ do+ register "ping" pingServer+ Just pid <- whereis "ping"+ True <- return $ pingServer == pid+ us <- getSelfPid+ nsend "ping" (Pong us)+ Ping pid' <- expect+ True <- return $ pingServer == pid'+ liftIO $ putMVar done ()++ takeMVar done+ testRemoteRegistry :: TestTransport -> Assertion testRemoteRegistry TestTransport{..} = do node1 <- newLocalNode testTransport initRemoteTable@@ -644,13 +684,15 @@ done <- newEmptyMVar pingServer <- forkProcess node1 ping+ deadProcess <- forkProcess node1 (return ()) runProcess node2 $ do let nid1 = localNodeId node1 registerRemoteAsync nid1 "ping" pingServer receiveWait [- matchIf (\(RegisterReply label' _) -> "ping" == label')- (\(RegisterReply _ _) -> return ()) ]+ matchIf (\(RegisterReply label' _ (Just pid)) ->+ "ping" == label' && pid == pingServer)+ (\(RegisterReply _ _ _) -> return ()) ] Just pid <- whereisRemote nid1 "ping" True <- return $ pingServer == pid@@ -658,10 +700,51 @@ nsendRemote nid1 "ping" (Pong us) Ping pid' <- expect True <- return $ pingServer == pid'++ -- test that if process was not registered Nothing is returned+ -- in owner field.+ registerRemoteAsync nid1 "dead" deadProcess+ receiveWait [+ matchIf (\(RegisterReply label' False Nothing) -> "dead" == label')+ (\(RegisterReply _ _ _) -> return ()) ]+ registerRemoteAsync nid1 "ping" deadProcess+ receiveWait [+ matchIf (\(RegisterReply label' False (Just pid)) ->+ "ping" == label' && pid == pingServer)+ (\(RegisterReply _ _ _) -> return ()) ]+ unregisterRemoteAsync nid1 "dead"+ receiveWait [+ matchIf (\(RegisterReply label' False Nothing) ->+ "dead" == label' && pid == pingServer)+ (\(RegisterReply _ _ _) -> return ()) ] liftIO $ putMVar done () takeMVar done +testRemoteRegistryRemoteProcess :: TestTransport -> Assertion+testRemoteRegistryRemoteProcess TestTransport{..} = do+ node1 <- newLocalNode testTransport initRemoteTable+ node2 <- newLocalNode testTransport initRemoteTable+ done <- newEmptyMVar++ pingServer <- forkProcess node2 ping++ runProcess node2 $ do+ let nid1 = localNodeId node1+ registerRemoteAsync nid1 "ping" pingServer+ receiveWait [+ matchIf (\(RegisterReply label' _ _) -> "ping" == label')+ (\(RegisterReply _ _ _) -> return ()) ]+ Just pid <- whereisRemote nid1 "ping"+ True <- return $ pingServer == pid+ us <- getSelfPid+ nsendRemote nid1 "ping" (Pong us)+ Ping pid' <- expect+ True <- return $ pingServer == pid'+ liftIO $ putMVar done ()++ takeMVar done+ testSpawnLocal :: TestTransport -> Assertion testSpawnLocal TestTransport{..} = do node <- newLocalNode testTransport initRemoteTable@@ -684,11 +767,25 @@ takeMVar done +testSpawnAsyncStrictness :: TestTransport -> Assertion+testSpawnAsyncStrictness TestTransport{..} = do+ node <- newLocalNode testTransport initRemoteTable+ done <- newEmptyMVar++ runProcess node $ do+ here <-getSelfNode++ ev <- try $ spawnAsync here (error "boom")+ liftIO $ case ev of+ Right _ -> putMVar done (error "Exception didn't fire")+ Left (_::SomeException) -> putMVar done (return ())++ join $ takeMVar done+ testReconnect :: TestTransport -> Assertion testReconnect TestTransport{..} = do [node1, node2] <- replicateM 2 $ newLocalNode testTransport initRemoteTable let nid1 = localNodeId node1- nid2 = localNodeId node2 processA <- newEmptyMVar [sendTestOk, registerTestOk] <- replicateM 2 newEmptyMVar @@ -709,8 +806,9 @@ send them "message 1" >> liftIO (threadDelay 100000) -- Simulate network failure- liftIO $ testBreakConnection (nodeAddress nid1) (nodeAddress nid2)+ liftIO $ syncBreakConnection testBreakConnection node1 node2 + -- Should not arrive send them "message 2" @@ -727,26 +825,26 @@ us <- getSelfPid registerRemoteAsync nid1 "a" us -- registerRemote is asynchronous receiveWait [- matchIf (\(RegisterReply label' _) -> "a" == label')- (\(RegisterReply _ _) -> return ()) ]+ matchIf (\(RegisterReply label' _ _) -> "a" == label')+ (\(RegisterReply _ _ _) -> return ()) ] Just _ <- whereisRemote nid1 "a" -- Simulate network failure- liftIO $ testBreakConnection (nodeAddress nid1) (nodeAddress nid2)+ liftIO $ syncBreakConnection testBreakConnection node1 node2 -- This will happen due to implicit reconnect registerRemoteAsync nid1 "b" us receiveWait [- matchIf (\(RegisterReply label' _) -> "b" == label')- (\(RegisterReply _ _) -> return ()) ]+ matchIf (\(RegisterReply label' _ _) -> "b" == label')+ (\(RegisterReply _ _ _) -> return ()) ] -- Should happen registerRemoteAsync nid1 "c" us receiveWait [- matchIf (\(RegisterReply label' _) -> "c" == label')- (\(RegisterReply _ _) -> return ()) ]+ matchIf (\(RegisterReply label' _ _) -> "c" == label')+ (\(RegisterReply _ _ _) -> return ()) ] -- Check Nothing <- whereisRemote nid1 "a" -- this will fail because the name is removed when the node is disconnected@@ -757,6 +855,55 @@ takeMVar registerTestOk +-- | Tests that unreliable messages arrive sorted even when there are connection+-- failures.+testUSend :: (ProcessId -> Int -> Process ())+ -> TestTransport -> Int -> Assertion+testUSend usendPrim TestTransport{..} numMessages = do+ [node1, node2] <- replicateM 2 $ newLocalNode testTransport initRemoteTable+ let nid1 = localNodeId node1+ nid2 = localNodeId node2+ processA <- newEmptyMVar+ usendTestOk <- newEmptyMVar++ forkProcess node1 $ flip catch (\e -> liftIO $ print (e :: SomeException) ) $ do+ us <- getSelfPid+ liftIO $ putMVar processA us+ them <- expect+ send them ()+ _ <- monitor them+ let -- Collects messages from 'them' until the sender dies.+ -- Disconnection notifications are ignored.+ receiveMessages :: Process [Int]+ receiveMessages = receiveWait+ [ match $ \mn -> case mn of+ ProcessMonitorNotification _ _ DiedDisconnect -> do+ monitor them+ receiveMessages+ _ -> return []+ , match $ \i -> fmap (i :) receiveMessages+ ]+ msgs <- receiveMessages+ let -- Checks that the input list is sorted.+ isSorted :: [Int] -> Bool+ isSorted (x : xs@(y : _)) = x <= y && isSorted xs+ isSorted _ = True+ -- The list can't be null since there are no failures after sending+ -- the last message.+ True <- return $ isSorted msgs && not (null msgs)+ liftIO $ putMVar usendTestOk ()++ forkProcess node2 $ do+ them <- liftIO $ readMVar processA+ getSelfPid >>= send them+ () <- expect+ forM_ [1..numMessages] $ \i -> do+ liftIO $ testBreakConnection (nodeAddress nid1) (nodeAddress nid2)+ usendPrim them i+ liftIO (threadDelay 30000)++ takeMVar usendTestOk+ -- | Test 'matchAny'. This repeats the 'testMath' but with a proxy server -- in between testMatchAny :: TestTransport -> Assertion@@ -1170,15 +1317,18 @@ localNode <- newLocalNode testTransport initRemoteTable supervisedDone <- newEmptyMVar supervisorDone <- newEmptyMVar+ -- XXX: we guarantee that exception handler will be set up+ -- regardless if forkProcess preserve masking state or not.+ handlerSetUp <- newEmptyMVar pid <- forkProcess localNode $ do- (liftIO $ threadDelay 100000)- `catchExit` \_from reason -> do+ (liftIO (putMVar handlerSetUp ()) >> expect) `catchExit` \_from reason -> do -- TODO: should verify that 'from' has the right value True <- return $ reason == "TestExit" liftIO $ putMVar supervisedDone () runProcess localNode $ do+ liftIO $ takeMVar handlerSetUp ref <- monitor pid exit pid "TestExit" -- This time the client catches the exception, so it dies normally@@ -1233,6 +1383,26 @@ takeMVar clientDone +testUnsafeUSend :: TestTransport -> Assertion+testUnsafeUSend TestTransport{..} = do+ serverAddr <- newEmptyMVar+ clientDone <- newEmptyMVar++ localNode <- newLocalNode testTransport initRemoteTable+ void $ forkProcess localNode $ do+ self <- getSelfPid+ liftIO $ putMVar serverAddr self+ clientAddr <- expect+ unsafeUSend clientAddr ()++ void $ forkProcess localNode $ do+ serverPid <- liftIO $ takeMVar serverAddr+ getSelfPid >>= unsafeUSend serverPid+ () <- expect+ liftIO $ putMVar clientDone ()++ takeMVar clientDone+ testUnsafeNSend :: TestTransport -> Assertion testUnsafeNSend TestTransport{..} = do clientDone <- newEmptyMVar@@ -1249,6 +1419,25 @@ takeMVar clientDone +testUnsafeNSendRemote :: TestTransport -> Assertion+testUnsafeNSendRemote TestTransport{..} = do+ clientDone <- newEmptyMVar++ localNode1 <- newLocalNode testTransport initRemoteTable+ localNode2 <- newLocalNode testTransport initRemoteTable++ _ <- forkProcess localNode1 $ do+ getSelfPid >>= register "foobar"+ liftIO $ putMVar clientDone ()+ () <- expect+ liftIO $ putMVar clientDone ()++ takeMVar clientDone+ void $ runProcess localNode2 $ do+ unsafeNSendRemote (localNodeId localNode1) "foobar" ()++ takeMVar clientDone+ testUnsafeSendChan :: TestTransport -> Assertion testUnsafeSendChan TestTransport{..} = do serverAddr <- newEmptyMVar@@ -1270,20 +1459,103 @@ takeMVar clientDone +testCallLocal :: TestTransport -> Assertion+testCallLocal TestTransport{..} = do+ node <- newLocalNode testTransport initRemoteTable++ -- Testing that (/=) <$> getSelfPid <*> callLocal getSelfPid.+ result <- newEmptyMVar+ runProcess node $ do+ r <- (/=) <$> getSelfPid <*> callLocal getSelfPid+ liftIO $ putMVar result r+ True <- takeMVar result+ return ()++ -- Testing that when callLocal is interrupted, the worker is interrupted.+ ibox <- newIORef False+ runProcess node $ do+ keeper <- getSelfPid+ spawnLocal $ do+ caller <- getSelfPid+ send keeper caller+ onException+ (callLocal $ do+ onException (do send keeper caller+ expect)+ (do liftIO $ writeIORef ibox True))+ (send keeper ())+ caller <- expect+ exit caller "test"+ () <- expect+ return ()+ True <- readIORef ibox+ return ()++ -- Testing that when the worker raises an exception, the exception is propagated to the parent.+ ibox2 <- newIORef False+ runProcess node $ do+ r <- try (callLocal $ error "e" >> return ())+ liftIO $ writeIORef ibox2 $ case r of+ Left (ErrorCall "e") -> True+ _ -> False+ True <- readIORef ibox+ return ()++ -- Test that caller waits for the worker in correct situation+ ibox3 <- newIORef False+ result3 <- newEmptyMVar+ runProcess node $ do+ keeper <- getSelfPid+ spawnLocal $ do+ callLocal $+ (do us <- getSelfPid+ send keeper us+ () <- expect+ liftIO yield)+ `finally` (liftIO $ writeIORef ibox3 True)+ liftIO $ putMVar result3 =<< readIORef ibox3+ worker <- expect+ send worker ()+ True <- takeMVar result3+ return ()++ -- Test that caller waits for the worker in case when caller gets an exception+ ibox4 <- newIORef False+ result4 <- newEmptyMVar+ runProcess node $ do+ keeper <- getSelfPid+ spawnLocal $ do+ caller <- getSelfPid+ callLocal+ ((do send keeper caller+ expect)+ `finally` (liftIO $ writeIORef ibox4 True))+ `finally` (liftIO $ putMVar result4 =<< readIORef ibox4)+ caller <- expect+ exit caller "hi!"+ True <- takeMVar result4+ return ()+ -- XXX: Testing that when mask_ $ callLocal p runs p in masked state.+++++ tests :: TestTransport -> IO [Test] tests testtrans = return [- testGroup "Basic features" [+ testGroup "Basic features" [ testCase "Ping" (testPing testtrans) , testCase "Math" (testMath testtrans) , testCase "Timeout" (testTimeout testtrans) , testCase "Timeout0" (testTimeout0 testtrans) , testCase "SendToTerminated" (testSendToTerminated testtrans) , testCase "TypedChannnels" (testTypedChannels testtrans)- , testCase "MergeChannels" (testMergeChannels testtrans) , testCase "Terminate" (testTerminate testtrans)- , testCase "Registry" (testRegistry testtrans)+ , testCase "RegistryRemoteProcess" (testRegistryRemoteProcess testtrans) , testCase "RemoteRegistry" (testRemoteRegistry testtrans)+ , testCase "RemoteRegistryRemoteProcess" (testRemoteRegistryRemoteProcess testtrans) , testCase "SpawnLocal" (testSpawnLocal testtrans)+ , testCase "SpawnAsyncStrictness" (testSpawnAsyncStrictness testtrans) , testCase "HandleMessageIf" (testHandleMessageIf testtrans) , testCase "MatchAny" (testMatchAny testtrans) , testCase "MatchAnyHandle" (testMatchAnyHandle testtrans)@@ -1301,19 +1573,27 @@ , testCase "MaskRestoreScope" (testMaskRestoreScope testtrans) , testCase "ExitLocal" (testExitLocal testtrans) , testCase "ExitRemote" (testExitRemote testtrans)+ , testCase "TextCallLocal" (testCallLocal testtrans) -- Unsafe Primitives , testCase "TestUnsafeSend" (testUnsafeSend testtrans)+ , testCase "TestUnsafeUSend" (testUnsafeUSend testtrans) , testCase "TestUnsafeNSend" (testUnsafeNSend testtrans)+ , testCase "TestUnsafeNSendRemote" (testUnsafeNSendRemote testtrans) , testCase "TestUnsafeSendChan" (testUnsafeSendChan testtrans)+ -- usend+ , testCase "USend" (testUSend usend testtrans 50)+ , testCase "UForward"+ (testUSend (\p m -> uforward (createUnencodedMessage m) p)+ testtrans 50+ ) ]- , testGroup "Monitoring and Linking" [+ , testGroup "Monitoring and Linking" [ -- Monitoring processes -- -- The "missing" combinations in the list below don't make much sense, as -- we cannot guarantee that the monitor reply or link exception will not -- happen before the unmonitor or unlink- testCase "MonitorUnreachable" (testMonitorUnreachable testtrans True False)- , testCase "MonitorNormalTermination" (testMonitorNormalTermination testtrans True False)+ testCase "MonitorNormalTermination" (testMonitorNormalTermination testtrans True False) , testCase "MonitorAbnormalTermination" (testMonitorAbnormalTermination testtrans True False) , testCase "MonitorLocalDeadProcess" (testMonitorLocalDeadProcess testtrans True False) , testCase "MonitorRemoteDeadProcess" (testMonitorRemoteDeadProcess testtrans True False)@@ -1335,6 +1615,27 @@ , testCase "MonitorLiveNode" (testMonitorLiveNode testtrans) , testCase "MonitorChannel" (testMonitorChannel testtrans) -- Reconnect- , testCase "Reconnect" (testReconnect testtrans) ]++ -- Tests that fail occasionally and should be revised+ , testGroup "Flaky" [+ testCase "Reconnect" (testReconnect testtrans)+ , testCase "Registry" (testRegistry testtrans)+ , testCase "MergeChannels" (testMergeChannels testtrans)+ , testCase "MonitorUnreachable" (testMonitorUnreachable testtrans True False)+ ] ]++syncBreakConnection :: (NT.EndPointAddress -> NT.EndPointAddress -> IO ()) -> LocalNode -> LocalNode -> IO ()+syncBreakConnection breakConnection nid0 nid1 = do+ m <- newEmptyMVar+ _ <- forkProcess nid1 $ getSelfPid >>= liftIO . putMVar m+ runProcess nid0 $ do+ them <- liftIO $ takeMVar m+ pinger <- spawnLocal $ forever $ send them ()+ _ <- monitorNode (localNodeId nid1)+ liftIO $ breakConnection (nodeAddress $ localNodeId nid0)+ (nodeAddress $ localNodeId nid1)+ NodeMonitorNotification _ _ _ <- expect+ kill pinger "finished"+ return ()
src/Control/Distributed/Process/Tests/Closure.hs view
@@ -6,7 +6,8 @@ import Data.ByteString.Lazy (empty) import Data.IORef import Data.Typeable (Typeable)-import Control.Monad (join, replicateM, forever, replicateM_, void, when)+import Data.Maybe+import Control.Monad (join, replicateM, forever, replicateM_, void, when, unless) import Control.Exception (IOException, throw) import Control.Concurrent (forkIO, threadDelay) import Control.Concurrent.MVar@@ -63,6 +64,9 @@ wait :: Int -> Process () wait = liftIO . threadDelay +expectUnit :: Process ()+expectUnit = expect+ isPrime :: Integer -> Process Bool isPrime n = return . (n `elem`) . takeWhile (<= n) . sieve $ [2..] where@@ -85,6 +89,7 @@ , 'sendPid , 'sdictInt , 'wait+ , 'expectUnit , 'typedPingServer , 'isPrime , 'quintuple@@ -141,11 +146,19 @@ waitClosure :: Int -> Closure (Process ()) waitClosure = $(mkClosure 'wait) -simulateNetworkFailure :: TestTransport -> NodeId -> NodeId -> Process ()+simulateNetworkFailure :: TestTransport -> LocalNode -> LocalNode -> Process () simulateNetworkFailure TestTransport{..} from to = liftIO $ do- threadDelay 10000- testBreakConnection (nodeAddress from) (nodeAddress to)- threadDelay 10000+ m <- newEmptyMVar+ _ <- forkProcess to $ getSelfPid >>= liftIO . putMVar m+ runProcess from $ do+ them <- liftIO $ takeMVar m+ pinger <- spawnLocal $ forever $ send them ()+ _ <- monitorNode (localNodeId to)+ liftIO $ testBreakConnection (nodeAddress $ localNodeId from)+ (nodeAddress $ localNodeId to)+ NodeMonitorNotification _ _ _ <- expect+ kill pinger "finished"+ return () -------------------------------------------------------------------------------- -- The tests proper --@@ -363,19 +376,35 @@ [node1, node2] <- replicateM 2 $ newLocalNode testTransport rtable [superPid, childPid] <- replicateM 2 $ newEmptyMVar thirdProcessDone <- newEmptyMVar+ linkUp <- newEmptyMVar forkProcess node1 $ do us <- getSelfPid liftIO $ putMVar superPid us- (child, _ref) <- spawnSupervised (localNodeId node2) (waitClosure 1000000)- liftIO $ do- putMVar childPid child- threadDelay 500000 -- Give the child a chance to link to us- throw supervisorDeath+ (child, _ref) <- spawnSupervised (localNodeId node2)+ (sendPidClosure us `seqCP` $(mkStaticClosure 'expectUnit))+ _ <- expect :: Process ProcessId + liftIO $ do putMVar childPid child+ -- Give the child a chance to link to us+ takeMVar linkUp+ throw supervisorDeath+ forkProcess node2 $ do [super, child] <- liftIO $ mapM readMVar [superPid, childPid] ref <- monitor child+ self <- getSelfPid+ let waitForMOrL = do+ liftIO $ threadDelay 10000+ mpinfo <- getProcessInfo child+ case mpinfo of+ Nothing -> waitForMOrL+ Just pinfo ->+ unless (isJust $ lookup self (infoMonitors pinfo)) waitForMOrL+ waitForMOrL+ liftIO $ putMVar linkUp ()+ -- because monitor message was sent before message to process+ -- we hope that it will be processed before ProcessMonitorNotification ref' pid' (DiedException e) <- expect True <- return $ ref' == ref && pid' == child@@ -465,7 +494,7 @@ forkProcess node2 $ do _pid1 <- spawn nid1 ($(mkClosure 'signal) incr)- simulateNetworkFailure testtrans nid2 nid1+ simulateNetworkFailure testtrans node2 node1 _pid2 <- spawn nid1 ($(mkClosure 'signal) incr) _pid3 <- spawn nid1 ($(mkClosure 'signal) incr)
src/Control/Distributed/Process/Tests/Tracing.hs view
@@ -5,6 +5,7 @@ import Control.Distributed.Process.Tests.Internal.Utils import Network.Transport.Test (TestTransport(..)) +import Control.Applicative ((<*)) import Control.Concurrent (threadDelay) import Control.Concurrent.MVar ( MVar@@ -19,9 +20,13 @@ import Control.Distributed.Process.Management ( MxEvent(..) )+import qualified Control.Exception as IO (bracket)+import Data.List (isPrefixOf, isSuffixOf) #if ! MIN_VERSION_base(4,6,0) import Prelude hiding (catch, log)+#else+import Prelude hiding ((<*)) #endif import Test.Framework@@ -29,7 +34,11 @@ , testGroup ) import Test.Framework.Providers.HUnit (testCase)+import System.Environment (getEnvironment)+-- These are available in System.Environment only since base 4.7+import System.SetEnv (setEnv, unsetEnv) + testSpawnTracing :: TestResult Bool -> Process () testSpawnTracing result = do setTraceFlags defaultTraceFlags {@@ -282,6 +291,82 @@ -- and just to be polite... liftIO $ closeLocalNode node2 +-- | Sets the value of an environment variable while executing the given IO+-- computation and restores the preceeding value upon completion.+withEnv :: String -> String -> IO a -> IO a+withEnv var val =+ IO.bracket (fmap (lookup var) getEnvironment <* setEnv var val)+ (maybe (unsetEnv var) (setEnv var))+ . const++-- | Tests that one and only one interesting trace message is produced when a+-- given action is performed. A message is considered interesting when the given+-- function return @True@.+testSystemLoggerMsg :: TestTransport+ -> Process a+ -> (a -> String -> Bool)+ -> IO ()+testSystemLoggerMsg t action interestingMessage =+ withEnv "DISTRIBUTED_PROCESS_TRACE_CONSOLE" "yes" $+ withEnv "DISTRIBUTED_PROCESS_TRACE_FLAGS" "pdnusrl" $ do+ n <- newLocalNode (testTransport t) initRemoteTable++ runProcess n $ do+ self <- getSelfPid+ reregister "trace.logger" self+ a <- action+ let interestingMessage' (_ :: String, msg) = interestingMessage a msg+ -- Wait for the trace message.+ receiveWait [ matchIf interestingMessage' $ const $ return () ]+ -- Only one interesting message should arrive.+ Nothing <- receiveTimeout 100000+ [ matchIf interestingMessage' $ const $ return () ]+ return ()+++-- | Tests that one and only one trace message is produced when a message is+-- received.+testSystemLoggerMxReceive :: TestTransport -> IO ()+testSystemLoggerMxReceive t = testSystemLoggerMsg t+ (getSelfPid >>= flip send ())+ (\_ msg -> "MxReceived" `isPrefixOf` msg+ -- discard traces of internal messages+ && not (":: RegisterReply" `isSuffixOf` msg)+ )++-- | Tests that one and only one trace message is produced when a message is+-- sent.+testSystemLoggerMxSent :: TestTransport -> IO ()+testSystemLoggerMxSent t = testSystemLoggerMsg t+ (getSelfPid >>= flip send ())+ (const $ isPrefixOf "MxSent")++-- | Tests that one and only one trace message is produced when a process dies.+testSystemLoggerMxProcessDied :: TestTransport -> IO ()+testSystemLoggerMxProcessDied t = testSystemLoggerMsg t+ (spawnLocal $ return ())+ (\pid -> isPrefixOf $ "MxProcessDied " ++ show pid)++-- | Tests that one and only one trace message appears when a process spawns.+testSystemLoggerMxSpawned :: TestTransport -> IO ()+testSystemLoggerMxSpawned t = testSystemLoggerMsg t+ (spawnLocal $ return ())+ (\pid -> isPrefixOf $ "MxSpawned " ++ show pid)++-- | Tests that one and only one trace message appears when a process is+-- registered.+testSystemLoggerMxRegistered :: TestTransport -> IO ()+testSystemLoggerMxRegistered t = testSystemLoggerMsg t+ (getSelfPid >>= register "a" >> getSelfPid)+ (\self -> isPrefixOf $ "MxRegistered " ++ show self ++ " " ++ show "a")++-- | Tests that one and only one trace message appears when a process is+-- unregistered.+testSystemLoggerMxUnRegistered :: TestTransport -> IO ()+testSystemLoggerMxUnRegistered t = testSystemLoggerMsg t+ (getSelfPid >>= register "a" >> unregister "a" >> getSelfPid)+ (\self -> isPrefixOf $ "MxUnRegistered " ++ show self ++ " " ++ show "a")+ tests :: TestTransport -> IO [Test] tests testtrans@TestTransport{..} = do node1 <- newLocalNode testTransport initRemoteTable@@ -323,4 +408,12 @@ (synchronisedAssertion "expected blah" node1 True (testRemoteTraceRelay testtrans) lock)+ , testGroup "SystemLoggerTracer"+ [ testCase "MxReceive" $ testSystemLoggerMxReceive testtrans+ , testCase "MxSent" $ testSystemLoggerMxSent testtrans+ , testCase "MxProcessDied" $ testSystemLoggerMxProcessDied testtrans+ , testCase "MxSpawned" $ testSystemLoggerMxSpawned testtrans+ , testCase "MxRegistered" $ testSystemLoggerMxRegistered testtrans+ , testCase "MxUnRegistered" $ testSystemLoggerMxUnRegistered testtrans+ ] ] ]
+ tests/runInMemory.hs view
@@ -0,0 +1,31 @@+-- Run tests using the TCP transport.++module Main where++import TEST_SUITE_MODULE (tests)++import Network.Transport.Test (TestTransport(..))+import Network.Transport.InMemory+import Test.Framework (defaultMainWithArgs)++import Control.Concurrent (threadDelay)+import System.Environment (getArgs)++main :: IO ()+main = do+ (transport, internals) <- createTransportExposeInternals+ ts <- tests TestTransport+ { testTransport = transport+ , testBreakConnection = \addr1 addr2 -> breakConnection internals addr1 addr2 "user error"+ }+ args <- getArgs+ -- Tests are time sensitive. Running the tests concurrently can slow them+ -- down enough that threads using threadDelay would wake up later than+ -- expected, thus changing the order in which messages were expected.+ -- Therefore we run the tests sequentially by passing "-j 1" to+ -- test-framework. This does not solve the issue but makes it less likely.+ --+ -- The problem was first detected with+ -- 'Control.Distributed.Process.Tests.CH.testMergeChannels'+ -- in particular.+ defaultMainWithArgs ts ("-j" : "1" : args)
tests/runTCP.hs view
@@ -1,5 +1,5 @@ -- Run tests using the TCP transport.-+-- module Main where import TEST_SUITE_MODULE (tests)@@ -10,16 +10,21 @@ ( createTransportExposeInternals , TransportInternals(socketBetween) , defaultTCPParameters+ , TCPParameters(..) ) import Test.Framework (defaultMainWithArgs) import Control.Concurrent (threadDelay) import System.Environment (getArgs)+import System.IO main :: IO () main = do+ hSetBuffering stdout LineBuffering+ hSetBuffering stderr LineBuffering Right (transport, internals) <-- createTransportExposeInternals "127.0.0.1" "8080" defaultTCPParameters+ createTransportExposeInternals "127.0.0.1" "8080"+ defaultTCPParameters { transportConnectTimeout = Just 3000000 } ts <- tests TestTransport { testTransport = transport , testBreakConnection = \addr1 addr2 -> do