distributed-process-tests 0.4.1 → 0.4.3.1
raw patch · 5 files changed
+301/−48 lines, 5 filesdep +ghc-primdep +stmdep ~distributed-processdep ~networkdep ~randomnew-uploader
Dependencies added: ghc-prim, stm
Dependency ranges changed: distributed-process, network, random
Files
- distributed-process-tests.cabal +20/−17
- src/Control/Distributed/Process/Tests/CH.hs +26/−10
- src/Control/Distributed/Process/Tests/Closure.hs +53/−3
- src/Control/Distributed/Process/Tests/Internal/Utils.hs +188/−16
- tests/runTCP.hs +14/−2
distributed-process-tests.cabal view
@@ -1,13 +1,13 @@ name: distributed-process-tests-version: 0.4.1-synopsis: Tests for distributed-process.+version: 0.4.3.1+synopsis: Tests and test support tools for distributed-process. homepage: http://github.com/haskell-distributed/distributed-process-tests license: BSD3 license-file: LICENSE Author: Duncan Coutts, Nicolas Wu, Edsko de Vries-Maintainer: watson.timothy@gmail.com, edsko@well-typed.com, duncan@well-typed.com+Maintainer: Facundo Domínguez <facundo.dominguez@tweag.io> copyright: Well-Typed LLP-category: Control+category: Control, Cloud Haskell build-type: Simple cabal-version: >=1.8 @@ -24,15 +24,16 @@ ansi-terminal >= 0.5 && < 0.7, binary >= 0.5 && < 0.8, bytestring >= 0.9 && < 0.11,- distributed-process,+ distributed-process >= 0.5.3 && < 0.6, distributed-static, HUnit >= 1.2 && < 1.3, network-transport >= 0.4.1.0 && < 0.5,- network >= 2.3 && < 2.7,- random >= 1.0 && < 1.1,+ network >= 2.5 && < 2.7,+ random >= 1.0 && < 1.2, rematch >= 0.1.2.1, test-framework >= 0.6 && < 0.9,- test-framework-hunit >= 0.2.0 && < 0.4+ test-framework-hunit >= 0.2.0 && < 0.4,+ stm hs-source-dirs: src ghc-options: -Wall -fno-warn-unused-do-bind extensions: CPP,@@ -45,6 +46,8 @@ RankNTypes, RecordWildCards, ScopedTypeVariables+ if impl(ghc <= 7.4.2)+ Build-Depends: ghc-prim == 0.2.0.0 Test-Suite TestCH Type: exitcode-stdio-1.0@@ -52,12 +55,12 @@ CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.CH Build-Depends: base >= 4.4 && < 5, distributed-process-tests,- network >= 2.3 && < 2.5,+ network >= 2.3 && < 2.7, network-transport >= 0.4.1.0 && < 0.5, network-transport-tcp >= 0.3 && < 0.5, test-framework >= 0.6 && < 0.9 Extensions: CPP- ghc-options: -Wall -threaded -debug -eventlog -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind+ ghc-options: -Wall -threaded -eventlog -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind HS-Source-Dirs: tests Test-Suite TestClosure@@ -66,7 +69,7 @@ CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Closure Build-Depends: base >= 4.4 && < 5, distributed-process-tests,- network >= 2.3 && < 2.5,+ network >= 2.3 && < 2.7, network-transport >= 0.4.1.0 && < 0.5, network-transport-tcp >= 0.3 && < 0.5, test-framework >= 0.6 && < 0.9@@ -80,12 +83,12 @@ CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Stats Build-Depends: base >= 4.4 && < 5, distributed-process-tests,- network >= 2.3 && < 2.5,+ network >= 2.3 && < 2.7, network-transport >= 0.4.1.0 && < 0.5, network-transport-tcp >= 0.3 && < 0.5, test-framework >= 0.6 && < 0.9 Extensions: CPP- ghc-options: -Wall -debug -eventlog -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind+ ghc-options: -Wall -eventlog -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind HS-Source-Dirs: tests @@ -95,12 +98,12 @@ CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Mx Build-Depends: base >= 4.4 && < 5, distributed-process-tests,- network >= 2.3 && < 2.5,+ network >= 2.3 && < 2.7, network-transport >= 0.4.1.0 && < 0.5, network-transport-tcp >= 0.3 && < 0.5, test-framework >= 0.6 && < 0.9 Extensions: CPP- ghc-options: -Wall -debug -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind+ ghc-options: -Wall --threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind HS-Source-Dirs: tests Test-Suite TestTracing@@ -109,10 +112,10 @@ CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Tracing Build-Depends: base >= 4.4 && < 5, distributed-process-tests,- network >= 2.3 && < 2.5,+ network >= 2.3 && < 2.7, network-transport >= 0.4.1.0 && < 0.5, network-transport-tcp >= 0.3 && < 0.5, test-framework >= 0.6 && < 0.9 Extensions: CPP- ghc-options: -Wall -debug -eventlog -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind+ ghc-options: -Wall -eventlog -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind HS-Source-Dirs: tests
src/Control/Distributed/Process/Tests/CH.hs view
@@ -17,7 +17,7 @@ , takeMVar , readMVar )-import Control.Monad (replicateM_, replicateM, forever, void)+import Control.Monad (replicateM_, replicateM, forever, void, unless) import Control.Exception (SomeException, throwIO) import qualified Control.Exception as Ex (catch) import Control.Applicative ((<$>), (<*>), pure, (<|>))@@ -104,16 +104,31 @@ -- | Monitor or link to a remote node monitorOrLink :: Bool -- ^ 'True' for monitor, 'False' for link- -> ProcessId -- Process to monitor/link to- -> Maybe (MVar ()) -- MVar to signal on once the monitor has been set up+ -> ProcessId -- ^ Process to monitor/link to+ -> Maybe (MVar ()) -- ^ MVar to signal on once the monitor has been set up -> Process (Maybe MonitorRef) monitorOrLink mOrL pid mSignal = do result <- if mOrL then Just <$> monitor pid else link pid >> return Nothing -- Monitor is asynchronous, which usually does not matter but if we want a- -- *specific* signal then it does. Therefore we wait an arbitrary delay and- -- hope that this means the monitor has been set up- forM_ mSignal $ \signal -> liftIO . forkIO $ threadDelay 100000 >> putMVar signal ()+ -- *specific* signal then it does. Therefore we wait until the MonitorRef is+ -- listed in the ProcessInfo and hope that this means the monitor has been set+ -- up.+ forM_ mSignal $ \signal -> do+ self <- getSelfPid+ spawnLocal $ do+ let waitForMOrL = do+ liftIO $ threadDelay 100000+ mpinfo <- getProcessInfo pid+ case mpinfo of+ Nothing -> waitForMOrL+ Just pinfo ->+ if mOrL then+ unless (result == lookup self (infoMonitors pinfo)) waitForMOrL+ else+ unless (elem self $ infoLinks pinfo) waitForMOrL+ waitForMOrL+ liftIO $ putMVar signal () return result monitorTestProcess :: ProcessId -- Process to monitor/link to@@ -244,19 +259,20 @@ -- | Monitor a local process that is already dead testMonitorLocalDeadProcess :: TestTransport -> Bool -> Bool -> Assertion testMonitorLocalDeadProcess TestTransport{..} mOrL un = do- processDead <- newEmptyMVar processAddr <- newEmptyMVar localNode <- newLocalNode testTransport initRemoteTable done <- newEmptyMVar forkIO $ do- addr <- forkProcess localNode . liftIO $ putMVar processDead ()+ addr <- forkProcess localNode $ return () putMVar processAddr addr forkIO $ do theirAddr <- readMVar processAddr- readMVar processDead runProcess localNode $ do+ monitor theirAddr+ -- wait for the process to die+ ProcessMonitorNotification _ _ _ <- expect monitorTestProcess theirAddr mOrL un DiedUnknownId Nothing done takeMVar done@@ -1181,7 +1197,7 @@ supervisorDone <- newEmptyMVar pid <- forkProcess node1 $ do- (liftIO $ threadDelay 100000)+ (receiveWait [] :: Process ()) -- block forever `catchExit` \_from reason -> do -- TODO: should verify that 'from' has the right value True <- return $ reason == "TestExit"
src/Control/Distributed/Process/Tests/Closure.hs view
@@ -1,11 +1,12 @@-{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE TemplateHaskell, KindSignatures #-} module Control.Distributed.Process.Tests.Closure (tests) where import Network.Transport.Test (TestTransport(..)) import Data.ByteString.Lazy (empty)+import Data.IORef import Data.Typeable (Typeable)-import Control.Monad (join, replicateM, forever, replicateM_, void)+import Control.Monad (join, replicateM, forever, replicateM_, void, when) import Control.Exception (IOException, throw) import Control.Concurrent (forkIO, threadDelay) import Control.Concurrent.MVar@@ -22,8 +23,13 @@ import Control.Distributed.Process import Control.Distributed.Process.Closure import Control.Distributed.Process.Node-import Control.Distributed.Process.Internal.Types (NodeId(nodeAddress))+import Control.Distributed.Process.Internal.Types+ ( NodeId(nodeAddress)+ , createMessage+ , messageToPayload+ ) import Control.Distributed.Static (staticLabel, staticClosure)+import qualified Network.Transport as NT import Test.HUnit (Assertion) import Test.Framework (Test)@@ -255,6 +261,49 @@ takeMVar clientDone +-- | Tests that spawn executes the supplied closure even if the caller dies+-- immediately after calling spawn.+--+-- This situation is of interest because the implementation of spawn has the+-- remote peer monitor the caller. See DP-99.+--+-- The condition is tested by using a transport which refuses to send to the+-- remote peer the message that it is waiting to stop monitoring the caller,+-- namely @()@.+--+testSpawnRace :: TestTransport -> RemoteTable -> Assertion+testSpawnRace TestTransport{..} rtable = do+ node1 <- newLocalNode (wrapTransport testTransport) rtable+ node2 <- newLocalNode testTransport rtable++ runProcess node1 $ do+ pid <- getSelfPid+ spawnLocal $ spawn (localNodeId node2) (sendPidClosure pid) >>= send pid+ pid' <- expect :: Process ProcessId+ pid'' <- expect :: Process ProcessId+ True <- return $ pid' == pid''+ return ()++ where++ wrapTransport (NT.Transport ne ct) = NT.Transport (fmap (fmap wrapEP) ne) ct++ wrapEP :: NT.EndPoint -> NT.EndPoint+ wrapEP e =+ e { NT.connect = \x y z -> do+ healthy <- newIORef True+ fmap (fmap $ wrapConnection healthy) $ NT.connect e x y z+ }++ wrapConnection :: IORef Bool -> NT.Connection -> NT.Connection+ wrapConnection healthy (NT.Connection s closeC) =+ flip NT.Connection closeC $ \msg -> do+ when (msg == messageToPayload (createMessage ())) $ do+ writeIORef healthy False+ isHealthy <- readIORef healthy+ if isHealthy then s msg+ else return $ Left $ NT.TransportError NT.SendFailed ""+ testCall :: TestTransport -> RemoteTable -> Assertion testCall TestTransport{..} rtable = do serverNodeAddr <- newEmptyMVar@@ -455,6 +504,7 @@ , testCase "SendIOClosure" (testSendIOClosure testtrans rtable) , testCase "SendProcClosure" (testSendProcClosure testtrans rtable) , testCase "Spawn" (testSpawn testtrans rtable)+ , testCase "SpawnRace" (testSpawnRace testtrans rtable) , testCase "Call" (testCall testtrans rtable) , testCase "CallBind" (testCallBind testtrans rtable) , testCase "Seq" (testSeq testtrans rtable)
src/Control/Distributed/Process/Tests/Internal/Utils.hs view
@@ -1,28 +1,116 @@--- | General testing support.-module Control.Distributed.Process.Tests.Internal.Utils where+{-# LANGUAGE DeriveDataTypeable #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE CPP #-} +-----------------------------------------------------------------------------+-- |+-- Module : Control.Distributed.Process.Tests.Internal.Utils+-- Copyright : (c) Tim Watson, Jeff Epstein 2013+-- License : BSD3 (see the file LICENSE)+--+-- Maintainer : Tim Watson+-- Stability : experimental+-- Portability : non-portable (requires concurrency)+--+-- This module provides basic building blocks for testing Cloud Haskell programs.+-----------------------------------------------------------------------------+module Control.Distributed.Process.Tests.Internal.Utils+ ( TestResult+ -- ping !+ , Ping(Ping)+ , ping+ , shouldBe+ , shouldMatch+ , shouldContain+ , shouldNotContain+ , expectThat+ , synchronisedAssertion+ -- test process utilities+ , TestProcessControl+ , startTestProcess+ , runTestProcess+ , testProcessGo+ , testProcessStop+ , testProcessReport+ , delayedAssertion+ , assertComplete+ -- logging+ , Logger()+ , newLogger+ , putLogMsg+ , stopLogger+ -- runners+ , tryRunProcess+ , tryForkProcess+ , noop+ , stash+ ) where+ #if ! MIN_VERSION_base(4,6,0) import Prelude hiding (catch) #endif+import Control.Concurrent+ ( ThreadId+ , myThreadId+ , forkIO+ )+import Control.Concurrent.STM+ ( TQueue+ , newTQueueIO+ , readTQueue+ , writeTQueue+ ) import Control.Concurrent.MVar ( MVar , newEmptyMVar , takeMVar- , putMVar ) +import Control.Concurrent+ ( throwTo+ )+import Control.Concurrent.MVar+ ( putMVar+ ) import Control.Distributed.Process import Control.Distributed.Process.Node-import Control.Distributed.Process.Serializable ()-import Test.HUnit (Assertion)+import Control.Distributed.Process.Serializable()++import Control.Exception (AsyncException(ThreadKilled), SomeException)+import Control.Monad (forever)+import Control.Monad.STM (atomically)+import Control.Rematch hiding (match)+import Control.Rematch.Run +import Data.Binary+import Data.Typeable (Typeable)++import Test.HUnit (Assertion, assertFailure) import Test.HUnit.Base (assertBool) --- these utilities have been cribbed from distributed-process-platform--- we should really find a way to share them...+import GHC.Generics -- | A mutable cell containing a test result. type TestResult a = MVar a +-- | A simple @Ping@ signal+data Ping = Ping+ deriving (Typeable, Generic, Eq, Show)+instance Binary Ping where++ping :: ProcessId -> Process ()+ping pid = send pid Ping++-- | Control signals used to manage /test processes/+data TestProcessControl = Stop | Go | Report ProcessId+ deriving (Typeable, Generic)++instance Binary TestProcessControl where++-- | Does exactly what it says on the tin, doing so in the @Process@ monad.+noop :: Process ()+noop = return ()+ synchronisedAssertion :: Eq a => String -> LocalNode@@ -40,21 +128,105 @@ where acquire lock' = liftIO $ takeMVar lock' release lock' = liftIO $ putMVar lock' () -delayedAssertion :: Eq a- => String- -> LocalNode- -> a- -> (TestResult a -> Process ())- -> Assertion+stash :: TestResult a -> a -> Process ()+stash mvar x = liftIO $ putMVar mvar x++expectThat :: a -> Matcher a -> Process ()+expectThat a matcher = case res of+ MatchSuccess -> return ()+ (MatchFailure msg) -> liftIO $ assertFailure msg+ where res = runMatch matcher a++shouldBe :: a -> Matcher a -> Process ()+shouldBe = expectThat++shouldContain :: (Show a, Eq a) => [a] -> a -> Process ()+shouldContain xs x = expectThat xs $ hasItem (equalTo x)++shouldNotContain :: (Show a, Eq a) => [a] -> a -> Process ()+shouldNotContain xs x = expectThat xs $ isNot (hasItem (equalTo x))++shouldMatch :: a -> Matcher a -> Process ()+shouldMatch = expectThat++-- | Run the supplied @testProc@ using an @MVar@ to collect and assert+-- against its result. Uses the supplied @note@ if the assertion fails.+delayedAssertion :: (Eq a) => String -> LocalNode -> a ->+ (TestResult a -> Process ()) -> Assertion delayedAssertion note localNode expected testProc = do result <- newEmptyMVar _ <- forkProcess localNode $ testProc result assertComplete note result expected -assertComplete :: Eq a => String -> MVar a -> a -> IO ()+-- | Takes the value of @mv@ (using @takeMVar@) and asserts that it matches @a@+assertComplete :: (Eq a) => String -> MVar a -> a -> IO () assertComplete msg mv a = do b <- takeMVar mv assertBool msg (a == b) -stash :: TestResult a -> a -> Process ()-stash mvar x = liftIO $ putMVar mvar x+-- synchronised logging++data Logger = Logger { _tid :: ThreadId, msgs :: TQueue String }++-- | Create a new Logger.+-- Logger uses a 'TQueue' to receive and process messages on a worker thread.+newLogger :: IO Logger+newLogger = do+ tid <- liftIO $ myThreadId+ q <- liftIO $ newTQueueIO+ _ <- forkIO $ logger q+ return $ Logger tid q+ where logger q' = forever $ do+ msg <- atomically $ readTQueue q'+ putStrLn msg++-- | Send a message to the Logger+putLogMsg :: Logger -> String -> Process ()+putLogMsg logger msg = liftIO $ atomically $ writeTQueue (msgs logger) msg++-- | Stop the worker thread for the given Logger+stopLogger :: Logger -> IO ()+stopLogger = (flip throwTo) ThreadKilled . _tid++-- | Starts a test process on the local node.+startTestProcess :: Process () -> Process ProcessId+startTestProcess proc =+ spawnLocal $ do+ getSelfPid >>= register "test-process"+ runTestProcess proc++-- | Runs a /test process/ around the supplied @proc@, which is executed+-- whenever the outer process loop receives a 'Go' signal.+runTestProcess :: Process () -> Process ()+runTestProcess proc = do+ ctl <- expect+ case ctl of+ Stop -> return ()+ Go -> proc >> runTestProcess proc+ Report p -> receiveWait [matchAny (\m -> forward m p)] >> runTestProcess proc++-- | Tell a /test process/ to continue executing+testProcessGo :: ProcessId -> Process ()+testProcessGo pid = send pid Go++-- | Tell a /test process/ to stop (i.e., 'terminate')+testProcessStop :: ProcessId -> Process ()+testProcessStop pid = send pid Stop++-- | Tell a /test process/ to send a report (message)+-- back to the calling process+testProcessReport :: ProcessId -> Process ()+testProcessReport pid = do+ self <- getSelfPid+ send pid $ Report self++tryRunProcess :: LocalNode -> Process () -> IO ()+tryRunProcess node p = do+ tid <- liftIO myThreadId+ runProcess node $ catch p (\e -> liftIO $ throwTo tid (e::SomeException))++tryForkProcess :: LocalNode -> Process () -> IO ProcessId+tryForkProcess node p = do+ tid <- liftIO myThreadId+ forkProcess node $ catch p (\e -> liftIO $ throwTo tid (e::SomeException))+
tests/runTCP.hs view
@@ -11,18 +11,30 @@ , TransportInternals(socketBetween) , defaultTCPParameters )-import Test.Framework (defaultMain)+import Test.Framework (defaultMainWithArgs) import Control.Concurrent (threadDelay)+import System.Environment (getArgs) main :: IO () main = do Right (transport, internals) <- createTransportExposeInternals "127.0.0.1" "8080" defaultTCPParameters- defaultMain =<< tests TestTransport+ ts <- tests TestTransport { testTransport = transport , testBreakConnection = \addr1 addr2 -> do sock <- socketBetween internals addr1 addr2 sClose sock threadDelay 10000 }+ 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)