diff --git a/distributed-process-tests.cabal b/distributed-process-tests.cabal --- a/distributed-process-tests.cabal +++ b/distributed-process-tests.cabal @@ -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 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 diff --git a/src/Control/Distributed/Process/Tests/CH.hs b/src/Control/Distributed/Process/Tests/CH.hs --- a/src/Control/Distributed/Process/Tests/CH.hs +++ b/src/Control/Distributed/Process/Tests/CH.hs @@ -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" diff --git a/src/Control/Distributed/Process/Tests/Closure.hs b/src/Control/Distributed/Process/Tests/Closure.hs --- a/src/Control/Distributed/Process/Tests/Closure.hs +++ b/src/Control/Distributed/Process/Tests/Closure.hs @@ -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) diff --git a/src/Control/Distributed/Process/Tests/Internal/Utils.hs b/src/Control/Distributed/Process/Tests/Internal/Utils.hs --- a/src/Control/Distributed/Process/Tests/Internal/Utils.hs +++ b/src/Control/Distributed/Process/Tests/Internal/Utils.hs @@ -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)) + diff --git a/tests/runTCP.hs b/tests/runTCP.hs --- a/tests/runTCP.hs +++ b/tests/runTCP.hs @@ -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)