diff --git a/Data/Concurrent/Deque/Debugger.hs b/Data/Concurrent/Deque/Debugger.hs new file mode 100644 --- /dev/null +++ b/Data/Concurrent/Deque/Debugger.hs @@ -0,0 +1,65 @@ + + +-- | This module provides a wrapper around a deque that can enforce additional +-- invariants at runtime for debugging purposes. + +module Data.Concurrent.Deque.Debugger + (DebugDeque(DebugDeque)) + where + +import Data.IORef +import Control.Concurrent +import Data.Concurrent.Deque.Class + +-- newtype DebugDeque d = DebugDeque d + +-- | Warning, this enforces the excessively STRONG invariant that if any end of the +-- deque is non-threadsafe then it may ever only be touched by one thread during its +-- entire lifetime. +-- +-- This extreme form of monagamy is easier to verify, because we don't have enough +-- information to know if two operations on different threads are racing with one +-- another or are properly synchronized. +-- +-- The wrapper data structure has two IORefs to track the last thread that touched +-- the left and right end of the deque, respectively. +data DebugDeque d elt = DebugDeque (IORef (Maybe ThreadId), IORef (Maybe ThreadId)) (d elt) + + +instance DequeClass d => DequeClass (DebugDeque d) where + pushL (DebugDeque (ref,_) q) elt = do + markThread (leftThreadSafe q) ref + pushL q elt + + tryPopR (DebugDeque (_,ref) q) = do + markThread (rightThreadSafe q) ref + tryPopR q + + newQ = do l <- newIORef Nothing + r <- newIORef Nothing + fmap (DebugDeque (l,r)) newQ + + -- FIXME: What are the threadsafe rules for nullQ? + nullQ (DebugDeque _ q) = nullQ q + + leftThreadSafe (DebugDeque _ q) = leftThreadSafe q + rightThreadSafe (DebugDeque _ q) = rightThreadSafe q + + +instance PopL d => PopL (DebugDeque d) where + tryPopL (DebugDeque (ref,_) q) = do + markThread (leftThreadSafe q) ref + tryPopL q + +-- | Mark the last thread to use this endpoint. +markThread True _ = return () -- Don't bother tracking. +markThread False ref = do + last <- readIORef ref + tid <- myThreadId +-- putStrLn$"Marking! "++show tid + atomicModifyIORef ref $ \ x -> + case x of + Nothing -> (Just tid, ()) + Just tid2 + | tid == tid2 -> (Just tid,()) + | otherwise -> error$ "DebugDeque: invariant violated, thread safety not allowed but accessed by: "++show (tid,tid2) 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 #-} +{-# LANGUAGE BangPatterns, RankNTypes, CPP #-} -- | This module contains a battery of simple tests for queues -- implementing the interface defined in @@ -17,7 +17,13 @@ tests_all, -- * Testing parameters - numElems, numAgents, producerRatio + numElems, getNumAgents, producerRatio, + + -- * Utility for controlling the number of threads used by generated tests. + setTestThreads, + + -- * Test initialization, reading common configs + stdTestHarness ) where @@ -25,24 +31,40 @@ import qualified Data.Concurrent.Deque.Reference as R import Control.Monad --- import qualified Data.Vector V --- import qualified Data.Vector.Mutable as MV 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 (numCapabilities, throwTo, threadDelay, myThreadId) +import GHC.Conc (throwTo, threadDelay, myThreadId) import Control.Concurrent.MVar import Control.Concurrent (yield, forkOS, forkIO, ThreadId) -import Control.Exception (catch, SomeException, fromException, AsyncException(ThreadKilled)) -import System.Environment (getEnvironment) +import Control.Exception (catch, SomeException, fromException, bracket, AsyncException(ThreadKilled)) +import System.Environment (withArgs, getArgs, getEnvironment) import System.IO (hPutStrLn, stderr, hFlush, stdout) import System.IO.Unsafe (unsafePerformIO) import System.Random (randomIO, randomRIO) -import Test.HUnit +import qualified Test.Framework as TF +import Test.Framework.Providers.HUnit (hUnitTestToTests) +import Test.HUnit as HU import Debug.Trace (trace) +#if __GLASGOW_HASKELL__ >= 704 +import GHC.Conc (getNumCapabilities, setNumCapabilities, getNumProcessors) +#else +import GHC.Conc (numCapabilities) +getNumCapabilities :: IO Int +getNumCapabilities = return numCapabilities + +setNumCapabilities :: Int -> IO () +setNumCapabilities = error "setNumCapabilities not supported in this older GHC! Set NUMTHREADS and +RTS -N to match." + +getNumProcessors :: IO Int +getNumProcessors = return 1 +#endif + theEnv :: [(String, String)] theEnv = unsafePerformIO getEnvironment @@ -71,11 +93,11 @@ -- | How many communicating agents are there? By default one per -- thread used by the RTS. -numAgents :: Int -numAgents = case lookup "NUMAGENTS" theEnv of - Nothing -> numCapabilities - Just str -> warnUsing ("NUMAGENTS = "++str) $ - read str +getNumAgents :: IO Int +getNumAgents = case lookup "NUMAGENTS" theEnv of + Nothing -> getNumCapabilities + Just str -> warnUsing ("NUMAGENTS = "++str) $ + return (read str) -- | It is possible to have imbalanced concurrency where there is more -- contention on the producing or consuming side (which corresponds to @@ -90,19 +112,75 @@ warnUsing str a = trace (" [Warning]: Using environment variable "++str) a +-- | Dig through the test constructors to find the leaf IO actions and bracket them +-- with a thread-setting action. +setTestThreads :: Int -> HU.Test -> HU.Test +setTestThreads nm tst = loop False tst + where + loop flg x = + case x of + TestLabel lb t2 -> TestLabel (decor flg lb) (loop True t2) + TestList ls -> TestList (map (loop flg) ls) + TestCase io -> TestCase (bracketThreads nm io) + -- We only need to insert the numcapabilities in the description string ONCE: + decor False lb = "N"++show nm++"_"++ lb + decor True lb = lb + + bracketThreads :: Int -> IO a -> IO a + bracketThreads n act = + bracket (getNumCapabilities) + setNumCapabilities + (\_ -> do dbgPrint 1 ("\n [Setting # capabilities to "++show n++" before test] \n") + setNumCapabilities n + act) + +stdTestHarness :: (IO Test) -> IO () +stdTestHarness genTests = do + numAgents <- getNumAgents + putStrLn$ "Running with numElems "++show numElems++" and numAgents "++ show numAgents + putStrLn "Use NUMELEMS, NUMAGENTS, NUMTHREADS to control the size of this benchmark." + args <- getArgs + + np <- getNumProcessors + putStrLn $"Running on a machine with "++show np++" hardware threads." + + -- We allow the user to set this directly, because the "-t" based regexp selection + -- of benchmarks is quite limited. + let all_threads = case lookup "NUMTHREADS" theEnv of + Just str -> [read str] + Nothing -> S.toList$ S.fromList$ + [1, 2, np `quot` 2, np, 2*np ] + putStrLn $"Running tests for these thread settings: " ++show all_threads + all_tests <- genTests + + -- Don't allow concurent tests (the tests are concurrent!): + withArgs (args ++ ["-j1","--jxml=test-results.xml"]) $ do + + -- Hack, this shouldn't be necessary, but I'm having problems with -t: + tests <- case all_threads of + [one] -> do cap <- getNumCapabilities + unless (cap == one) $ setNumCapabilities one + return all_tests + _ -> return$ TestList [ setTestThreads n all_tests | n <- all_threads ] + TF.defaultMain$ hUnitTestToTests tests + + ---------------------------------------------------------------------------------------------------- -- Test a plain FIFO queue: ---------------------------------------------------------------------------------------------------- +-- The type of data that we push over the queues. +type Elt = Int64 + -- | This test serially fills up a queue and then drains it. -test_fifo_filldrain :: DequeClass d => d Int -> IO () +test_fifo_filldrain :: DequeClass d => d Elt -> IO () test_fifo_filldrain q = do dbgPrintLn 1 "\nTest FIFO queue: sequential fill and then drain" dbgPrintLn 1 "===============================================" -- let n = 1000 - let n = numElems + let n = fromIntegral numElems dbgPrintLn 1$ "Done creating queue. Pushing "++show n++" elements:" forM_ [1..n] $ \i -> do pushL q i @@ -115,7 +193,7 @@ when (i < 200) $ dbgPrint 1 $ printf " %d" x loop (i-1) (sumR + x) s <- loop n 0 - let expected = sum [1..n] :: Int + let expected = sum [1..n] :: Elt dbgPrint 1 $ printf "\nSum of popped vals: %d should be %d\n" s expected when (s /= expected) (assertFailure "Incorrect sum!") return () @@ -126,7 +204,7 @@ -- thread performs its designated operation as fast as possible. The -- 'Int' argument 'total' designates how many total items should be -- communicated (irrespective of 'numAgents'). -test_fifo_OneBottleneck :: DequeClass d => Bool -> Int -> d Int -> IO () +test_fifo_OneBottleneck :: DequeClass d => Bool -> Int -> d Elt -> IO () test_fifo_OneBottleneck doBackoff total q = do assertBool "test_fifo_OneBottleneck requires thread safe left end" (leftThreadSafe q) @@ -138,12 +216,15 @@ bl <- nullQ q dbgPrintLn 1$ "Check that queue is initially null: "++show bl + + numAgents <- getNumAgents let producers = max 1 (round$ producerRatio * (fromIntegral numAgents) / (producerRatio + 1)) consumers = max 1 (numAgents - producers) perthread = total `quot` producers (perthread2,remain) = total `quotRem` consumers - - when (not doBackoff && (numCapabilities == 1 || numCapabilities < producers + consumers)) $ + + numCap <- getNumCapabilities + when (not doBackoff && (numCap == 1 || numCap < producers + consumers)) $ error$ "The aggressively busy-waiting version of the test can only run with the right thread settings." dbgPrint 1 $ printf "Forking %d producer threads, each producing %d elements.\n" producers perthread @@ -153,9 +234,9 @@ myfork "producer thread" $ do let start = ind * perthread dbgPrint 1 $ printf " * Producer thread %d pushing ints from %d to %d \n" ind start (start+perthread - 1) - for_ start (start+perthread) $ \ i -> do + forI_ start (start+perthread) $ \ i -> do pushL q i - when (i < start + 10) $ dbgPrint 1 $ printf " [p%d] pushed %d \n" ind i + -- when (i < start + 10) $ dbgPrint 1 $ printf " [p%d] pushed %d \n" ind i ls <- forkJoin consumers $ \ ind -> let mymax = if ind==0 then perthread2 + remain else perthread2 @@ -170,28 +251,30 @@ let finalSum = Prelude.sum (map fst ls) dbgPrintLn 1$ "Consumers DONE. Maximum retries for each consumer thread: "++ show (map snd ls) dbgPrintLn 1$ "Final sum: "++ show finalSum - assertEqual "Correct final sum" (expectedSum (producers * perthread)) finalSum + assertEqual "Correct final sum" (expectedSum (fromIntegral$ producers * perthread)) finalSum dbgPrintLn 1$ "Checking that queue is finally null..." b <- nullQ q if b then dbgPrintLn 1$ "Sum matched expected, test passed." else assertFailure "Queue was not empty!!" --- | This test uses a separate queue per consumer thread. The queues --- are used in a single-writer multiple-reader fashion (mailboxes). -test_contention_free_parallel :: DequeClass d => Bool -> Int -> IO (d Int) -> IO () +-- | This test uses a separate queue per producer/consumer pair. The queues +-- are used in a single-writer single-reader. +test_contention_free_parallel :: DequeClass d => Bool -> Int -> IO (d Elt) -> IO () test_contention_free_parallel doBackoff total newqueue = do dbgPrintLn 1$ "\nTest FIFO queue: producers & consumers thru N queues" ++(if doBackoff then " (with backoff)" else "(hard busy wait)") dbgPrintLn 1 "======================================================" mv <- newEmptyMVar - + + numAgents <- getNumAgents let producers = max 1 (round$ producerRatio * (fromIntegral numAgents) / (producerRatio + 1)) consumers = producers -- Must be matched perthread = total `quot` producers qs <- sequence (replicate consumers newqueue) - - when (not doBackoff && (numCapabilities == 1 || numCapabilities < producers + consumers)) $ + + numCap <- getNumCapabilities + when (not doBackoff && (numCap == 1 || numCap < producers + consumers)) $ error$ "The aggressively busy-waiting version of the test can only run with the right thread settings." dbgPrint 1 $ printf "Forking %d producer threads, each producing %d elements.\n" producers perthread @@ -200,9 +283,9 @@ forM_ (zip [0..producers-1] qs) $ \ (id, q) -> myfork "producer thread" $ let start = id*perthread in - for_ 0 perthread $ \ i -> do + forI_ 0 perthread $ \ i -> do pushL q i - when (i - id*producers < 10) $ dbgPrint 1 $ printf " [%d] pushed %d \n" id i + -- when (i - id*producers < 10) $ dbgPrint 1 $ printf " [%d] pushed %d \n" id i forM_ (zip [0..consumers-1] qs) $ \ (id, q) -> myfork "consumer thread" $ do @@ -211,7 +294,8 @@ (x,iters) <- if doBackoff then spinPopBkoff q else spinPopHard q when (i < 10) $ dbgPrint 1 $ printf " [%d] popped #%d = %d \n" id i x - unless (x == i) $ error $ "Message out of order! Expected "++show i++" recevied "++show x + unless (x == fromIntegral i) $ + error $ "Message out of order! Expected "++show i++" recevied "++show x consume_loop (sum+x) (max maxiters iters) (i+1) pr <- consume_loop 0 0 0 putMVar mv pr @@ -221,7 +305,7 @@ let finalSum = Prelude.sum (map fst ls) dbgPrintLn 1$ "Consumers DONE. Maximum retries for each consumer thread: "++ show (map snd ls) dbgPrintLn 1$ "All messages received in order. Final sum: "++ show finalSum - assertEqual "Correct final sum" (producers * expectedSum perthread) finalSum + assertEqual "Correct final sum" (fromIntegral producers * expectedSum (fromIntegral perthread)) finalSum dbgPrintLn 1$ "Checking that queue is finally null..." bs <- mapM nullQ qs if all id bs @@ -232,9 +316,10 @@ -- | 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 Int) -> IO () -test_random_array_comm size total newqueue | size > 0 = do - +test_random_array_comm :: DequeClass d => Int -> Int -> IO (d Elt) -> IO () +test_random_array_comm size total newqueue = do + assertBool "positive size" (size > 0) + qs <- sequence (replicate size newqueue) -- arr <- V.thaw $ V.fromlist qs let arr = A.listArray (0,size-1) qs @@ -245,9 +330,10 @@ dbgPrintLn 1$ "\nTest FIFO queue: producers & consumers select random queues" dbgPrintLn 1 "======================================================" + numAgents <- getNumAgents let producers = max 1 (round$ producerRatio * (fromIntegral numAgents) / (producerRatio + 1)) consumers = max 1 (numAgents - producers) - perthread = total `quot` producers + perthread = fromIntegral (total `quot` producers) (perthread2,remain) = total `quotRem` consumers dbgPrintLn 1 $ printf "Forking %d producer threads, each producing %d elements.\n" producers perthread @@ -257,9 +343,9 @@ myfork "producer thread" $ for_ 0 perthread $ \ i -> do -- Randomly pick a position: - ix <- randomRIO (0,size-1) :: IO Int + ix <- randomRIO (0,size - 1) :: IO Int pushL (arr ! ix) i - when (i - ind*producers < 10) $ dbgPrint 1 $ printf " [%d] pushed %d \n" ind i + -- when (i - ind*producers < 10) $ dbgPrint 1 $ printf " [%d] pushed %d \n" ind i -- Each consumer doesn't quit until it has popped "perthread": sums <- forkJoin consumers $ \ ind -> @@ -282,7 +368,7 @@ let finalSum = Prelude.sum sums dbgPrintLn 1$ "Final sum: "++ show finalSum ++ ", per-consumer sums: "++show sums dbgPrintLn 1$ "Checking that queue is finally null..." - assertEqual "Correct final sum" (producers * expectedSum perthread) finalSum + assertEqual "Correct final sum" (fromIntegral producers * expectedSum perthread) finalSum bs <- mapM nullQ qs if all id bs then dbgPrintLn 1$ "Sum matched expected, test passed." @@ -298,7 +384,7 @@ -- | 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 = TestList $ +tests_fifo newq = TestLabel "single-ended-queue-tests"$ TestList $ tests_basic newq ++ tests_fifo_exclusive newq @@ -319,7 +405,7 @@ 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_contention_free_parallel" (TestCase$ assert $ test_contention_free_parallel True numElems newq) ] ---------------------------------------------------------------------------------------------------- @@ -347,18 +433,24 @@ [Just "one",Just "two",Just "four",Just "three",Nothing,Nothing] --- | 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_work_stealing :: (DequeClass d, PopL d) => Int -> IO (d Int) -> IO () + +-- | 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. +test_random_work_stealing :: (DequeClass d, PopL d) => Int -> IO (d Elt) -> IO () test_random_work_stealing total newqueue = do dbgPrintLn 1$ "\nTest FIFO queue: producers & consumers select random queues" dbgPrintLn 1 "======================================================" - let producers = max 1 (round$ producerRatio * (fromIntegral numAgents) / (producerRatio + 1)) - consumers = max 1 (numAgents - producers) - perthread = total `quot` producers - (perthread2,remain) = total `quotRem` consumers + numAgents <- getNumAgents + let producers, consumers :: Int + perthread, realtotal, perthread2 :: Elt + producers = max 1 (round$ producerRatio * (fromIntegral numAgents) / (producerRatio + 1)) + consumers = max 1 (fromIntegral numAgents - producers) + perthread = fromIntegral total `quot` fromIntegral producers + realtotal = perthread * fromIntegral producers + (perthread2,remain) = realtotal `quotRem` fromIntegral consumers qs <- sequence (replicate producers newqueue) -- A work-stealing deque only has ONE threadsafe end: @@ -366,58 +458,72 @@ let arr = A.listArray (0,producers - 1) qs dbgPrint 1 $ printf "Forking %d producer threads, each producing %d elements.\n" producers perthread - dbgPrint 1 $ printf "Forking %d consumer threads, each consuming %d elements.\n" consumers perthread2 + dbgPrint 1 $ printf "Forking %d consumer threads, each consuming ~%d elements.\n" consumers perthread2 prod_results <- newEmptyMVar forM_ (zip [0..producers-1] qs) $ \ (ind,myQ) -> do myfork "producer thread" $ - let loop i !acc | i == perthread = putMVar prod_results acc - loop i !acc = - do -- Randomly push or pop: + let loop :: Elt -> Elt -> Elt -> IO () + loop i !pops !acc + | i == perthread = putMVar prod_results (pops,acc) + | otherwise = do + -- Randomly push or pop: b <- randomIO :: IO Bool if b then do x <- spinPopN 100 (tryPopL myQ) + -- In the case where we pop, we do NOT advance 'i', saving that + -- for the next iteration that acutally does a push. case x of - Nothing -> loop i acc - Just n -> loop i (n+acc) + Nothing -> loop i pops acc + Just n -> loop i (pops+1) (n+acc) else do pushL myQ i - when (i - ind*producers < 10) $ dbgPrint 1 $ printf " [%d] pushed %d \n" ind i - loop (i+1) acc - in loop 0 0 + -- when (i - ind*producers < 10) $ dbgPrint 1 $ printf " [%d] pushed %d \n" ind i + loop (i+1) pops acc + in loop 0 0 0 - -- Each consumer doesn't quit until it has popped "perthread": + -- In this version consumer_sums <- forkJoin consumers $ \ ind -> - let mymax = if ind==0 then perthread2 + remain else perthread2 - consume_loop summ i | i == mymax = return summ - consume_loop !summ i = do + let mymax = if ind==0 then perthread2 + remain else perthread2 + consume_loop !summ !successes i + | i == mymax = return (successes, summ) + | otherwise = do -- Randomly pick a position: ix <- randomRIO (0,producers - 1) :: IO Int -- Try to pop something, but not too hard: m <- spinPopN 100 (tryPopR (arr ! ix)) case m of Just x -> do - when (i < 10) $ dbgPrint 1 $ printf " [%d] popped #%d = %d\n" ind i x - consume_loop (summ+x) (i+1) + when (i < 10) $ dbgPrint 1 $ printf " [%d] popped try#%d = %d\n" ind i x + consume_loop (summ+x) (successes+1) (i+1) Nothing -> - consume_loop summ (i+1) -- Increment even if we don't get a result. - in consume_loop 0 0 + consume_loop summ successes (i+1) -- Increment even if we don't get a result. + in do dbgPrintLn 1 (" Beginning consumer thread loop for "++show mymax ++" attempts.") + consume_loop 0 0 0 + -- <-- Consumers are finished as of here. dbgPrintLn 1 "Reading sums from MVar..." prod_ls <- mapM (\_ -> takeMVar prod_results) [1..producers] + -- We sequentially read out the leftovers after all the parallel tasks are done: leftovers <- forM qs $ \ q -> - let loop !acc = do - x <- tryPopR q + let loop !cnt !acc = do + x <- tryPopR q -- This should work as a popR OR a popL. case x of - Nothing -> return acc - Just n -> loop (acc+n) - in loop 0 + Nothing -> return (cnt,acc) + Just n -> loop (cnt+1) (acc+n) + in loop 0 0 - let finalSum = Prelude.sum (consumer_sums ++ prod_ls ++ leftovers) - dbgPrintLn 1$ "Final sum: "++ show finalSum ++ ", producer/consumer/leftover sums: "++show (prod_ls, consumer_sums, leftovers) + let finalSum = Prelude.sum (map snd consumer_sums ++ + map snd prod_ls ++ + map snd leftovers) + dbgPrintLn 0$ "Final sum: "++ show finalSum ++ ", producer/consumer/leftover sums: "++show (prod_ls, consumer_sums, leftovers) + dbgPrintLn 1$ "Total pop events: "++ show (Prelude.sum (map fst consumer_sums ++ + map fst prod_ls ++ + map fst leftovers)) + ++" should be "++ (show$ realtotal) dbgPrintLn 1$ "Checking that queue is finally null..." - assertEqual "Correct final sum" (producers * expectedSum perthread) finalSum + assertEqual "Correct final sum" (fromIntegral producers * expectedSum perthread) finalSum bs <- mapM nullQ qs if all id bs then dbgPrintLn 1$ "Sum matched expected, test passed." @@ -428,9 +534,9 @@ -- | 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 = TestList $ - tests_basic newq ++ - tests_wsqueue_exclusive newq +tests_wsqueue newq = TestLabel "work-stealing-deque-tests"$ TestList $ + tests_wsqueue_exclusive newq ++ + tests_basic newq -- Internal: factoring this out. tests_wsqueue_exclusive :: (PopL d) => (forall elt. IO (d elt)) -> [Test] @@ -446,7 +552,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 = TestList $ +tests_all newq = TestLabel "full-deque-tests"$ TestList $ tests_basic newq ++ tests_fifo_exclusive newq ++ tests_wsqueue_exclusive newq @@ -487,6 +593,7 @@ ls <- mapM readMVar answers return ls +-- | Pop from the right end more gently. spinPopBkoff :: DequeClass d => d t -> IO (t, Int) spinPopBkoff q = loop 1 where @@ -505,18 +612,20 @@ Nothing -> do -- Every `sleepevery` times do a significant delay: if n `mod` sleepevery == 0 - then do putStr "!" + then do dbgPrint 1 "!" threadDelay n -- 1ms after 1K fails, 2 after 2K... else when (n > hardspinfor) $ do - putStr "." + dbgPrint 1 "." hFlush stdout yield -- At LEAST yield... you'd think this is pretty strong backoff. loop (n+1) Just x -> return (x, n) --- | This is always a crazy and dangerous thing to do -- GHC cannot --- deschedule this spinning thread because it does not allocate: +-- | This is always a crazy and dangerous thing to do -- GHC cannot deschedule this +-- spinning thread because it does not allocate. Thus one can run into a deadlock +-- situation where the consumer holds the capabilitity hostage, not letting the +-- producer run (and therefore unblock it). spinPopHard :: (DequeClass d) => d t -> IO (t, Int) spinPopHard q = loop 1 where @@ -537,19 +646,21 @@ -- My own forM for numeric ranges (not requiring deforestation optimizations). -- Inclusive start, exclusive end. {-# INLINE for_ #-} -for_ :: Monad m => Int -> Int -> (Int -> m ()) -> m () +for_ :: Monad m => Elt -> Elt -> (Elt -> m ()) -> m () for_ start end _fn | start > end = error "for_: start is greater than end" for_ start end fn = loop start where loop !i | i == end = return () | otherwise = do fn i; loop (i+1) +forI_ :: Monad m => Int -> Int -> (Elt -> m ()) -> m () +forI_ st en = for_ (fromIntegral st) (fromIntegral en) ---------------------------------------------------------------------------------------------------- -- DEBUGGING ---------------------------------------------------------------------------------------------------- --- | Debugging flag shared by all accelerate-backend-kit modules. +-- | Debugging flag shared by several modules. -- This is activated by setting the environment variable DEBUG=1..5 dbg :: Int dbg = case lookup "DEBUG" theEnv of diff --git a/Test.hs b/Test.hs --- a/Test.hs +++ b/Test.hs @@ -9,20 +9,24 @@ import Test.Framework (defaultMain) import Test.Framework.Providers.HUnit (hUnitTestToTests) -import Test.HUnit (assert, assertEqual, Test(TestCase, TestList)) +import Test.HUnit (assert, assertEqual, Test(TestCase, TestList, TestLabel)) import qualified Data.Concurrent.Deque.Tests as T import qualified Data.Concurrent.Deque.Reference as R +import qualified Data.Concurrent.Deque.Class as C +import Data.Concurrent.Deque.Debugger (DebugDeque) +import System.Environment (withArgs) -- Import the instances: -import qualified Data.Concurrent.Deque.Reference.DequeInstance -import System.Exit +import Data.Concurrent.Deque.Reference.DequeInstance () +test_1 :: Test test_1 = TestCase $ assert $ do q <- R.newQ -- Select a specific implementation. pushR q 3 Just x <- tryPopR q - assertEqual "test_1 result" x 3 + assertEqual "test_1 result" x (3::Integer) +test_2 :: Test test_2 = TestCase $ assert $ do -- Here's an example of type-based restriction of the queue implementation: @@ -32,16 +36,26 @@ Just x <- tryPopR q assertEqual "test_2 result" x 33 +main :: IO () #if __GLASGOW_HASKELL__ >= 700 -main = do - putStrLn "[ Test executable: test reference deque implementation... ]" - defaultMain$ hUnitTestToTests$ - TestList $ - [ - T.tests_all R.newQ - , test_1 - , test_2 - ] +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 + -- 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)) + ] + +-- main = do +-- putStrLn "[ Test executable: test reference deque implementation... ]" +-- withArgs ["-j1","--jxml=test-results.xml"] $ +-- defaultMain$ hUnitTestToTests$ + #else main = putStrLn "WARNING: Tests disabled for GHC < 7" #endif 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 +Version: 0.2.2 License: BSD3 License-file: LICENSE Author: Ryan R. Newton @@ -20,6 +20,7 @@ -- 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 Synopsis: Abstract, parameterized interface to mutable Deques. @@ -55,10 +56,11 @@ exposed-modules: Data.Concurrent.Deque.Class, Data.Concurrent.Deque.Tests, Data.Concurrent.Deque.Reference, - Data.Concurrent.Deque.Reference.DequeInstance + Data.Concurrent.Deque.Reference.DequeInstance, + Data.Concurrent.Deque.Debugger build-depends: base >= 4 && < 5, array, random, - containers, HUnit - + containers, + HUnit, test-framework >= 0.6, test-framework-hunit >= 0.2.7 if flag(useCAS) && impl( ghc >= 7.4 ) && !os(mingw32) { build-depends: IORefCAS >= 0.2 cpp-options: -DUSE_CAS -DDEFAULT_SIGNATURES @@ -77,8 +79,9 @@ type: exitcode-stdio-1.0 main-is: Test.hs build-depends: base >= 4 && < 5, containers, HUnit, array, random, - test-framework >= 0.6, test-framework-hunit >= 0.2.7 + HUnit, test-framework >= 0.6, test-framework-hunit >= 0.2.7 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 -- ghc-options: -ddump-ds -ddump-simpl -ddump-stg -ddump-asm -ddump-bcos -ddump-cmm -ddump-opt-cmm -ddump-inlinings