abstract-deque 0.2 → 0.2.2

raw patch · 4 files changed

+294/−101 lines, 4 filesPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

API changes (from Hackage documentation)

Files

• Data/Concurrent/Deque/Debugger.hs +65/−0
• Data/Concurrent/Deque/Tests.hs +194/−83
• Test.hs +27/−13
• abstract-deque.cabal +8/−5

@@ -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)

@@ -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

@@ -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

@@ -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