diff --git a/Data/Concurrent/Deque/Tests.hs b/Data/Concurrent/Deque/Tests.hs --- a/Data/Concurrent/Deque/Tests.hs +++ b/Data/Concurrent/Deque/Tests.hs @@ -1,4 +1,4 @@ -{-# LANGUAGE BangPatterns, RankNTypes, CPP #-} +{-# LANGUAGE BangPatterns, RankNTypes, CPP, BangPatterns #-} -- | This module contains a battery of simple tests for queues -- implementing the interface defined in @@ -12,6 +12,7 @@ -- * Tests for Work-stealing queues. test_ws_triv1, test_ws_triv2, tests_wsqueue, + test_parfib_work_stealing, -- * All deque tests, aggregated. tests_all, @@ -19,23 +20,23 @@ -- * Testing parameters numElems, getNumAgents, producerRatio, - -- * Utility for controlling the number of threads used by generated tests. - setTestThreads, + -- * Utility for tweaking test suites + setTestThreads, appendLabels, appendLabel, -- * Test initialization, reading common configs - stdTestHarness + stdTestHarness, Elt, + + -- * Misc helpers + forkJoin, timeit, fibSize ) where -import Data.Concurrent.Deque.Class as C -import qualified Data.Concurrent.Deque.Reference as R - import Control.Monad -import Data.Array as A -import Data.IORef -import Data.Int +import Data.Time.Clock +import Data.Array as A +import Data.IORef +import Data.Int import qualified Data.Set as S --- import System.Mem.StableName import Text.Printf import GHC.Conc (throwTo, threadDelay, myThreadId) import Control.Concurrent.MVar @@ -46,11 +47,15 @@ import System.IO.Unsafe (unsafePerformIO) import System.Random (randomIO, randomRIO) import qualified Test.Framework as TF -import Test.Framework.Providers.HUnit (hUnitTestToTests) -import Test.HUnit as HU +import Test.Framework.Providers.HUnit (hUnitTestToTests) +import Test.HUnit as HU import Debug.Trace (trace) +import Data.Concurrent.Deque.Class as C +import qualified Data.Concurrent.Deque.Reference as R + + #if __GLASGOW_HASKELL__ >= 704 import GHC.Conc (getNumCapabilities, setNumCapabilities, getNumProcessors) #else @@ -79,6 +84,10 @@ Nothing -> 100 * 1000 -- 500000 Just str -> warnUsing ("NUMELEMS = "++str) $ read str +fibSize :: Int64 +fibSize = case lookup "FIBSIZE" theEnv of + Just s -> read s + Nothing -> 32 forkThread :: IO () -> IO ThreadId forkThread = case lookup "OSTHREADS" theEnv of @@ -95,7 +104,9 @@ -- thread used by the RTS. getNumAgents :: IO Int getNumAgents = case lookup "NUMAGENTS" theEnv of - Nothing -> getNumCapabilities + Nothing -> do x <- getNumCapabilities + putStrLn$"Defaulting numAgents to numCapabilities: "++show x + return x Just str -> warnUsing ("NUMAGENTS = "++str) $ return (read str) @@ -135,10 +146,25 @@ setNumCapabilities n act) +-- | Dig through the test constructors and add a new string to the first label found. +-- If no such label exists, add one. +appendLabel :: String -> HU.Test -> HU.Test +appendLabel newLab tst = loop tst + where + loop x = + case x of + TestLabel lb t2 -> TestLabel (newLab ++"_"++ lb) t2 + TestList ls -> TestList (map loop ls) + TestCase io -> TestLabel newLab x + +-- | This version has the option of being smarter about how it handles uniformly +-- labeling many tests. +appendLabels :: String -> [HU.Test] -> HU.Test +appendLabels newLab tst = TestList $ map (appendLabel newLab) tst + stdTestHarness :: (IO Test) -> IO () stdTestHarness genTests = do - numAgents <- getNumAgents - putStrLn$ "Running with numElems "++show numElems++" and numAgents "++ show numAgents + putStrLn$ "Running with numElems "++show numElems putStrLn "Use NUMELEMS, NUMAGENTS, NUMTHREADS to control the size of this benchmark." args <- getArgs @@ -313,7 +339,7 @@ else assertFailure "Queue was not empty!!" - +{-# INLINE test_random_array_comm #-} -- | This test uses a number of producer and consumer threads which push and pop -- elements from random positions in an array of FIFOs. test_random_array_comm :: DequeClass d => Int -> Int -> IO (d Elt) -> IO () @@ -384,30 +410,34 @@ -- | This creates an HUnit test list to perform all the tests that apply to a -- single-ended (threadsafe) queue. It requires thread safety at /both/ ends. tests_fifo :: DequeClass d => (forall elt. IO (d elt)) -> Test -tests_fifo newq = TestLabel "single-ended-queue-tests"$ TestList $ +tests_fifo newq = appendLabels "single-ended-queue-tests" $ tests_basic newq ++ tests_fifo_exclusive newq +{-# INLINE tests_fifo_exclusive #-} -- | Tests exclusive to single-ended (threadsafe) queues: tests_fifo_exclusive :: DequeClass d => (forall elt. IO (d elt)) -> [Test] tests_fifo_exclusive newq = - [ TestLabel "test_fifo_OneBottleneck_backoff" (TestCase$ assert $ newq >>= test_fifo_OneBottleneck True numElems) + [ TestLabel "test_fifo_OneBottleneck_backoff" (TestCase$ assertT $ newq >>= test_fifo_OneBottleneck True numElems) -- , TestLabel "test_fifo_OneBottleneck_aggressive" (TestCase$ assert $ newq >>= test_fifo_OneBottleneck False numElems) -- , TestLabel "test the tests" (TestCase$ assert $ assertFailure "This SHOULD fail.") ] ++ [ TestLabel ("test_random_array_comm_"++show size) - (TestCase$ assert $ test_random_array_comm size numElems newq) + (TestCase$ assertT $ test_random_array_comm size numElems newq) | size <- [10,100] ] - +{-# INLINE tests_basic #-} -- | Tests that don't require thread safety at either end. tests_basic :: DequeClass d => (forall elt. IO (d elt)) -> [Test] tests_basic newq = - [ TestLabel "test_fifo_filldrain" (TestCase$ assert $ newq >>= test_fifo_filldrain) - , TestLabel "test_contention_free_parallel" (TestCase$ assert $ test_contention_free_parallel True numElems newq) + [ TestLabel "test_fifo_filldrain" (TestCase$ assertT $ newq >>= test_fifo_filldrain) + , TestLabel "test_contention_free_parallel" (TestCase$ assertT $ test_contention_free_parallel True numElems newq) ] +assertT :: IO () -> IO () +assertT = timeit . assert + ---------------------------------------------------------------------------------------------------- -- Test a Work-stealing queue: ---------------------------------------------------------------------------------------------------- @@ -433,7 +463,7 @@ [Just "one",Just "two",Just "four",Just "three",Nothing,Nothing] - +{-# INLINE test_random_work_stealing #-} -- | This test uses a number of worker threads which randomly either push or pop work -- to their local work stealing deque. Also, there are consumer (always thief) -- threads which never produce and only consume. @@ -530,20 +560,63 @@ else assertFailure "Queue was not empty!!" +{-# INLINE test_parfib_work_stealing #-} +test_parfib_work_stealing :: (DequeClass d, PopL d) => Elt -> IO (d Elt) -> IO () +test_parfib_work_stealing origInput newqueue = do + putStrLn$ " [parfib] Computing fib("++show origInput++")" + numAgents <- getNumAgents + qs <- sequence (replicate numAgents newqueue) + let arr = A.listArray (0,numAgents - 1) qs + + let parfib !myId !myQ !mySum !num + | num <= 2 = + do x <- tryPopL myQ + case x of + Nothing -> trySteal myId myQ (mySum+1) + Just n -> parfib myId myQ (mySum+1) n + | otherwise = do + pushL myQ (num-1) + parfib myId myQ mySum (num-2) + + trySteal !myId !myQ !mySum = + let loop ind + -- After we finish one sweep... we're completely done. + | ind == myId = return mySum + | ind == size arr = loop 0 + | otherwise = do + x <- tryPopR (arr ! ind) + case x of + Just n -> parfib myId myQ mySum n + Nothing -> do yield + loop (ind+1) + in loop (myId+1) + size a = let (st,en) = A.bounds a in en - st + 1 + + partial_sums <- forkJoin numAgents $ \ myId -> + if myId == 0 + then parfib myId (arr ! myId) 0 origInput + else trySteal myId (arr ! myId) 0 + + putStrLn$ " [parfib] fib("++show origInput++") = "++show (sum partial_sums) + + +{-# INLINE tests_wsqueue #-} -- | Aggregate tests for work stealing queues. None of these require thread-safety -- on the left end. There is some duplication with tests_fifo. tests_wsqueue :: (PopL d) => (forall elt. IO (d elt)) -> Test -tests_wsqueue newq = TestLabel "work-stealing-deque-tests"$ TestList $ +tests_wsqueue newq = appendLabels "work-stealing-deque-tests"$ tests_wsqueue_exclusive newq ++ tests_basic newq +{-# INLINE tests_wsqueue_exclusive #-} -- Internal: factoring this out. tests_wsqueue_exclusive :: (PopL d) => (forall elt. IO (d elt)) -> [Test] tests_wsqueue_exclusive newq = - [ TestLabel "test_ws_triv1" (TestCase$ assert $ newq >>= test_ws_triv1) - , TestLabel "test_ws_triv2" (TestCase$ assert $ newq >>= test_ws_triv2) - , TestLabel "test_random_work_stealing" (TestCase$ assert $ test_random_work_stealing numElems newq) + [ TestLabel "test_ws_triv1" (TestCase$ assertT $ newq >>= test_ws_triv1) + , TestLabel "test_ws_triv2" (TestCase$ assertT $ newq >>= test_ws_triv2) + , TestLabel "test_random_work_stealing" (TestCase$ assertT $ test_random_work_stealing numElems newq) + , TestLabel "test_parfib_generic" (TestCase$ assertT $ test_parfib_work_stealing fibSize newq) ] ---------------------------------------------------------------------------------------------------- @@ -552,7 +625,7 @@ -- | This requires double ended queues that are threadsafe on BOTH ends. tests_all :: (PopL d) => (forall elt. IO (d elt)) -> Test -tests_all newq = TestLabel "full-deque-tests"$ TestList $ +tests_all newq = appendLabels "full-deque-tests" $ tests_basic newq ++ tests_fifo_exclusive newq ++ tests_wsqueue_exclusive newq @@ -687,3 +760,11 @@ dbgPrintLn :: Int -> String -> IO () dbgPrintLn lvl str = dbgPrint lvl (str++"\n") + +timeit :: IO a -> IO a +timeit ioact = do + start <- getCurrentTime + res <- ioact + end <- getCurrentTime + putStrLn$ "SELFTIMED: " ++ show (diffUTCTime end start) + return res diff --git a/Test.hs b/Test.hs --- a/Test.hs +++ b/Test.hs @@ -1,4 +1,4 @@ -{-# LANGUAGE CPP, ScopedTypeVariables, NamedFieldPuns #-} +{-# LANGUAGE CPP, ScopedTypeVariables, NamedFieldPuns, BangPatterns #-} #if __GLASGOW_HASKELL >= 700 {-# OPTIONS_GHC -with-rtsopts=-K32M #-} #endif @@ -6,7 +6,9 @@ import Data.Concurrent.Deque.Class -- import Data.Concurrent.Deque.Class.Reference (newQueue) -- import Data.Concurrent.MegaDeque - +import Data.Int +import Data.Array as A +import Control.Concurrent (yield) import Test.Framework (defaultMain) import Test.Framework.Providers.HUnit (hUnitTestToTests) import Test.HUnit (assert, assertEqual, Test(TestCase, TestList, TestLabel)) @@ -36,19 +38,61 @@ Just x <- tryPopR q assertEqual "test_2 result" x 33 +test_parfib_work_stealing_specialized :: T.Elt -> IO T.Elt +test_parfib_work_stealing_specialized origInput = do + putStrLn$ " [parfib] Computing fib("++show origInput++")" + numAgents <- T.getNumAgents + qs <- sequence (replicate numAgents R.newQ) + let arr = A.listArray (0,numAgents - 1) qs + + let parfib !myId !myQ !mySum !num + | num <= 2 = + do x <- R.tryPopL myQ + case x of + Nothing -> trySteal myId myQ (mySum+1) + Just n -> parfib myId myQ (mySum+1) n + | otherwise = do + R.pushL myQ (num-1) + parfib myId myQ mySum (num-2) + + trySteal !myId !myQ !mySum = + let loop ind + -- After we finish one sweep... we're completely done. + | ind == myId = return mySum + | ind == size arr = loop 0 + | otherwise = do + x <- R.tryPopR (arr ! ind) + case x of + Just n -> parfib myId myQ mySum n + Nothing -> do yield + loop (ind+1) + in loop (myId+1) + + size a = let (st,en) = A.bounds a in en - st + 1 + + partial_sums <- T.forkJoin numAgents $ \ myId -> + if myId == 0 + then parfib myId (arr ! myId) 0 origInput + else trySteal myId (arr ! myId) 0 + + return (sum partial_sums) + main :: IO () #if __GLASGOW_HASKELL__ >= 700 main = T.stdTestHarness $ return all_tests where all_tests :: Test all_tests = - TestLabel "Reference_Deque" $ TestList $ - [ TestLabel "test_1" test_1 - , TestLabel "test_2" test_2 - , TestLabel "direct"$ T.tests_all R.newQ + T.appendLabels "Reference_Deque" $ + [ T.appendLabel "test_1" test_1 + , T.appendLabel "test_2" test_2 + , T.appendLabel "direct"$ T.tests_all R.newQ -- Test going through the class interface as well: - , TestLabel "thru_class"$ T.tests_all (C.newQ :: IO (R.SimpleDeque a)) - , TestLabel "with_debug"$ T.tests_all (C.newQ :: IO (DebugDeque R.SimpleDeque a)) + , T.appendLabel "thru_class"$ T.tests_all (C.newQ :: IO (R.SimpleDeque a)) + , T.appendLabel "with_debug"$ T.tests_all (C.newQ :: IO (DebugDeque R.SimpleDeque a)) + + , TestLabel "parfib_specialized" $ TestCase $ T.timeit$ + print =<< test_parfib_work_stealing_specialized T.fibSize ] -- main = do diff --git a/abstract-deque.cabal b/abstract-deque.cabal --- a/abstract-deque.cabal +++ b/abstract-deque.cabal @@ -1,5 +1,5 @@ Name: abstract-deque -Version: 0.2.2 +Version: 0.2.2.1 License: BSD3 License-file: LICENSE Author: Ryan R. Newton @@ -20,7 +20,8 @@ -- 0.1.6: Make useCAS default FALSE. -- 0.1.7: Add leftThreadSafe / rightThreadSafe -- 0.2: [breaking] Refactor names of exposed Tests --- 0.2.2: Add Debugger +-- 0.2.2: Add Debugger +-- 0.2.2.1: Add some extra testing helpers. Synopsis: Abstract, parameterized interface to mutable Deques. @@ -60,7 +61,7 @@ Data.Concurrent.Deque.Debugger build-depends: base >= 4 && < 5, array, random, containers, - HUnit, test-framework >= 0.6, test-framework-hunit >= 0.2.7 + HUnit, test-framework >= 0.6, test-framework-hunit >= 0.2.7, time if flag(useCAS) && impl( ghc >= 7.4 ) && !os(mingw32) { build-depends: IORefCAS >= 0.2 cpp-options: -DUSE_CAS -DDEFAULT_SIGNATURES @@ -79,7 +80,7 @@ type: exitcode-stdio-1.0 main-is: Test.hs build-depends: base >= 4 && < 5, containers, HUnit, array, random, - HUnit, test-framework >= 0.6, test-framework-hunit >= 0.2.7 + HUnit, test-framework >= 0.6, test-framework-hunit >= 0.2.7, time ghc-options: -O2 -threaded -rtsopts -- ghc-options: -keep-tmp-files -dsuppress-module-prefixes -ddump-to-file -ddump-core-stats -ddump-simpl-stats -dcore-lint -dcmm-lint