diff --git a/Data/Concurrent/Deque/Class.hs b/Data/Concurrent/Deque/Class.hs --- a/Data/Concurrent/Deque/Class.hs +++ b/Data/Concurrent/Deque/Class.hs @@ -10,6 +10,9 @@ This interface includes a non-associated type family for Deques plus separate type classes encapsulating the Deque operations. + That is, we separate type selection (type family) from function overloading + (vanilla type classes). + This design strives to hide the extra phantom-type parameters from the Class constraints and therefore from the type signatures of client code. @@ -34,18 +37,22 @@ -- ** Aliases for commonly used Deque configurations: , Queue, ConcQueue, ConcDeque, WSDeque - -- * Classes containing Deque operations + -- * Class for basic Queue operations , DequeClass(..) - -- * Auxilary type classes. + -- * Extra capabilities: type classes - -- | In spite of hiding the extra phantom type - -- parameters in the DequeClass, we wish to retain the ability for - -- clients to constrain the set of implementations they work with - -- **statically**. + -- | These classes provide a more programmer-friendly constraints than directly + -- using the phantom type parameters to constrain queues in user code. Also note + -- that instances can be provided for types outside the type `Deque` type family. + -- + -- We still make a distinction between the different capabilities + -- (e.g. single-ended / double ended), and thus we need the below type classes for + -- the additional operations unsupported by the minimal "DequeClass". -- ** The \"unnatural\" double ended cases: pop left, push right. , PopL(..), PushR(..) + -- ** Operations that only make sense for bounded queues. , BoundedL(..), BoundedR(..) ) @@ -157,6 +164,14 @@ -- TODO: It would also be possible to include blocking/spinning pops. -- But maybe those should go in separate type classes... + + -- | Runtime indication of thread saftey for `pushL` (and `popL`). + -- (Argument unused.) + leftThreadSafe :: d elt -> Bool + + -- | Runtime indication of thread saftey for `tryPopR` (and `pushR`). + -- (Argument unused.) + rightThreadSafe :: d elt -> Bool class DequeClass d => PopL d where -- | PopL is not the native operation for the left end, so it requires diff --git a/Data/Concurrent/Deque/Reference.hs b/Data/Concurrent/Deque/Reference.hs --- a/Data/Concurrent/Deque/Reference.hs +++ b/Data/Concurrent/Deque/Reference.hs @@ -22,7 +22,6 @@ import Prelude hiding (length) import qualified Data.Concurrent.Deque.Class as C -import Control.Exception (evaluate) import Data.Sequence import Data.IORef #ifdef USE_CAS @@ -88,14 +87,17 @@ -- Nothing -> popL q -- Just x -> return x +tryPopL :: SimpleDeque a -> IO (Maybe a) tryPopL (DQ _ qr) = modify qr $ \s -> case viewl s of EmptyL -> (empty, Nothing) x :< s' -> (s', Just x) +pushR :: SimpleDeque t -> t -> IO () pushR (DQ 0 qr) x = modify qr (\s -> (s |> x, ())) pushR (DQ n _) _ = error$ "should not call pushR on Deque with size bound "++ show n +tryPushL :: SimpleDeque a -> a -> IO Bool tryPushL q@(DQ 0 _) v = pushL q v >> return True tryPushL (DQ lim qr) v = modify qr $ \s -> @@ -103,6 +105,7 @@ then (s, False) else (v <| s, True) +tryPushR :: SimpleDeque a -> a -> IO Bool tryPushR q@(DQ 0 _) v = pushR q v >> return True tryPushR (DQ lim qr) v = modify qr $ \s -> @@ -119,6 +122,8 @@ nullQ = nullQ pushL = pushL tryPopR = tryPopR + leftThreadSafe _ = True + rightThreadSafe _ = True instance C.PopL SimpleDeque where tryPopL = tryPopL instance C.PushR SimpleDeque where 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 @@ -14,7 +14,10 @@ test_ws_triv1, test_ws_triv2, test_wsqueue, -- * All deque tests, aggregated. - test_all + test_all, + + -- * Testing parameters + numElems, numAgents, producerRatio ) where @@ -22,6 +25,9 @@ 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 System.Mem.StableName import Text.Printf @@ -30,8 +36,9 @@ import Control.Concurrent (yield, forkOS, forkIO, ThreadId) import Control.Exception (catch, SomeException, fromException, AsyncException(ThreadKilled)) import System.Environment (getEnvironment) -import System.IO (hPutStrLn, stderr) +import System.IO (hPutStrLn, stderr, hFlush, stdout) import System.IO.Unsafe (unsafePerformIO) +import System.Random (randomIO, randomRIO) import Test.HUnit import Debug.Trace (trace) @@ -47,7 +54,7 @@ -- How many elements should each of the tests pump through the queue(s)? numElems :: Int numElems = case lookup "NUMELEMS" theEnv of - Nothing -> 50 * 1000 -- 500000 + Nothing -> 100 * 1000 -- 500000 Just str -> warnUsing ("NUMELEMS = "++str) $ read str @@ -83,29 +90,7 @@ warnUsing str a = trace (" [Warning]: Using environment variable "++str) a ----------------------------------------------------------------------------------------------------- --- Misc Helpers ----------------------------------------------------------------------------------------------------- -myfork :: String -> IO () -> IO ThreadId -myfork msg = forkWithExceptions forkThread msg - --- Exceptions that walk up the fork tree of threads: -forkWithExceptions :: (IO () -> IO ThreadId) -> String -> IO () -> IO ThreadId -forkWithExceptions forkit descr action = do - parent <- myThreadId - forkit $ - Control.Exception.catch action - (\ e -> - case fromException e of - -- Let threadKilled exceptions through. - Just ThreadKilled -> return () - _ -> do - hPutStrLn stderr $ "Exception inside child thread "++show descr++": "++show e - throwTo parent (e::SomeException) - ) - - ---------------------------------------------------------------------------------------------------- -- Test a plain FIFO queue: ---------------------------------------------------------------------------------------------------- @@ -113,29 +98,29 @@ -- | This test serially fills up a queue and then drains it. test_fifo_filldrain :: DequeClass d => d Int -> IO () test_fifo_filldrain q = - do -- q <- newQ - putStrLn "\nTest FIFO queue: sequential fill and then drain" - putStrLn "===============================================" + do + dbgPrintLn 1 "\nTest FIFO queue: sequential fill and then drain" + dbgPrintLn 1 "===============================================" -- let n = 1000 let n = numElems - putStrLn$ "Done creating queue. Pushing elements:" + dbgPrintLn 1$ "Done creating queue. Pushing elements:" forM_ [1..n] $ \i -> do pushL q i - when (i < 200) $ printf " %d" i - putStrLn "\nDone filling queue with elements. Now popping..." + when (i < 200) $ dbgPrint 1 $ printf " %d" i + dbgPrintLn 1 "\nDone filling queue with elements. Now popping..." let loop 0 !sumR = return sumR loop i !sumR = do (x,_) <- spinPopBkoff q - when (i < 200) $ printf " %d" x + when (i < 200) $ dbgPrint 1 $ printf " %d" x loop (i-1) (sumR + x) s <- loop n 0 let expected = sum [1..n] :: Int - printf "\nSum of popped vals: %d should be %d\n" s expected + dbgPrint 1 $ printf "\nSum of popped vals: %d should be %d\n" s expected when (s /= expected) (assertFailure "Incorrect sum!") return () - +{-# INLINABLE test_fifo_OneBottleneck #-} -- | This test splits the 'numAgents' threads into producers and -- consumers which all communicate through a SINGLE queue. Each -- thread performs its designated operation as fast as possible. The @@ -143,68 +128,188 @@ -- communicated (irrespective of 'numAgents'). test_fifo_OneBottleneck :: DequeClass d => Bool -> Int -> d Int -> IO () test_fifo_OneBottleneck doBackoff total q = - do -- q <- newQ - putStrLn$ "\nTest FIFO queue: producers & consumers thru 1 queue" + do + assertBool "test_fifo_OneBottleneck requires thread safe left end" (leftThreadSafe q) + assertBool "test_fifo_OneBottleneck requires thread safe right end" (rightThreadSafe q) + + dbgPrintLn 1$ "\nTest FIFO queue: producers & consumers thru 1 queue" ++(if doBackoff then " (with backoff)" else "(hard busy wait)") - putStrLn "======================================================" - mv <- newEmptyMVar - x <- nullQ q - putStrLn$ "Check that queue is initially null: "++show x + dbgPrintLn 1 "======================================================" + + bl <- nullQ q + dbgPrintLn 1$ "Check that queue is initially null: "++show bl let producers = max 1 (round$ producerRatio * (fromIntegral numAgents) / (producerRatio + 1)) consumers = max 1 (numAgents - producers) - perthread = total `quot` producers - + perthread = total `quot` producers + (perthread2,remain) = total `quotRem` consumers + when (not doBackoff && (numCapabilities == 1 || numCapabilities < producers + consumers)) $ error$ "The aggressively busy-waiting version of the test can only run with the right thread settings." - printf "Forking %d producer threads, each producing %d elements.\n" producers perthread + 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 - forM_ [0..producers-1] $ \ id -> - myfork "producer thread" $ - forM_ (take perthread [id * producers .. ]) $ \ i -> do + forM_ [0..producers-1] $ \ ind -> + 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 pushL q i - when (i - id*producers < 10) $ printf " [%d] pushed %d \n" id i - - printf "Forking %d consumer threads.\n" consumers - - forM_ [0..consumers-1] $ \ id -> - myfork "consumer thread" $ do + when (i < start + 10) $ dbgPrint 1 $ printf " [p%d] pushed %d \n" ind i - let fn (!sum,!maxiters) i = do - (x,iters) <- if doBackoff then spinPopBkoff q - else spinPopHard q - when (i - id*producers < 10) $ printf " [%d] popped %d \n" id i - return (sum+x, max maxiters iters) - - pr <- foldM fn (0,0) (take perthread [id * producers .. ]) - putMVar mv pr + ls <- forkJoin consumers $ \ ind -> + let mymax = if ind==0 then perthread2 + remain else perthread2 + consume_loop summ maxiters i | i == mymax = return (summ, maxiters) + consume_loop !summ !maxiters i = do + (x,iters) <- if doBackoff then spinPopBkoff q + else spinPopHard q + when (i >= mymax - 10) $ dbgPrint 1 $ printf " [c%d] popped #%d = %d \n" ind i x + consume_loop (summ+x) (max maxiters iters) (i+1) + in consume_loop 0 0 0 - printf "Reading sums from MVar...\n" - ls <- mapM (\_ -> takeMVar mv) [1..consumers] let finalSum = Prelude.sum (map fst ls) - putStrLn$ "Consumers DONE. Maximum retries for each consumer thread: "++ show (map snd ls) - putStrLn$ "Final sum: "++ show finalSum - putStrLn$ "Checking that queue is finally null..." + 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 + dbgPrintLn 1$ "Checking that queue is finally null..." b <- nullQ q - if b then putStrLn$ "Sum matched expected, test passed." + 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 () +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 --- test_fifo_mailboxes + 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)) $ + 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 + dbgPrint 1 $ printf "Forking %d consumer threads, each consuming %d elements.\n" consumers perthread + + forM_ (zip [0..producers-1] qs) $ \ (id, q) -> + myfork "producer thread" $ + let start = id*perthread in + for_ 0 perthread $ \ i -> do + pushL q 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 + let consume_loop sum maxiters i | i == perthread = return (sum, maxiters) + consume_loop !sum !maxiters i = do + (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 + consume_loop (sum+x) (max maxiters iters) (i+1) + pr <- consume_loop 0 0 0 + putMVar mv pr + + dbgPrint 1 $ printf "Reading sums from MVar...\n" + ls <- mapM (\_ -> takeMVar mv) [1..consumers] + 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 + dbgPrintLn 1$ "Checking that queue is finally null..." + bs <- mapM nullQ qs + if all id bs + then dbgPrintLn 1$ "Sum matched expected, test passed." + else assertFailure "Queue was not empty!!" + + + +-- | 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 + + qs <- sequence (replicate size newqueue) +-- arr <- V.thaw $ V.fromlist qs + let arr = A.listArray (0,size-1) qs + + assertBool "test_random_array_comm requires thread safe left end" (leftThreadSafe (head qs)) + assertBool "test_random_array_comm requires thread safe right end" (rightThreadSafe (head qs)) + + 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 + + dbgPrintLn 1 $ printf "Forking %d producer threads, each producing %d elements.\n" producers perthread + dbgPrintLn 1 $ printf "Forking %d consumer threads, each consuming ~%d elements.\n" consumers perthread2 + + forM_ [0..producers-1] $ \ ind -> + myfork "producer thread" $ + for_ 0 perthread $ \ i -> do + -- Randomly pick a position: + 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 + + -- Each consumer doesn't quit until it has popped "perthread": + 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 + -- Randomly pick a position: + ix <- randomRIO (0,size-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) + Nothing -> + consume_loop summ i + in consume_loop 0 0 + + dbgPrintLn 1 "Reading sums from MVar..." + 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 + bs <- mapM nullQ qs + if all id bs + then dbgPrintLn 1$ "Sum matched expected, test passed." + else assertFailure "Queue was not empty!!" + +-- Sum of first N numbers, starting with ZERO +expectedSum :: Integral a => a -> a +expectedSum n = (n * (n - 1)) `quot` 2 + + ------------------------------------------------------------ -- | This creates an HUnit test list to perform all the tests above. test_fifo :: DequeClass d => (forall elt. IO (d elt)) -> Test -test_fifo newq = TestList +test_fifo newq = TestList $ [ TestLabel "test_fifo_filldrain" (TestCase$ assert $ newq >>= test_fifo_filldrain) -- Do half a million elements by default: - , TestLabel "test_fifo_OneBottleneck_backoff" (TestCase$ assert $ newq >>= test_fifo_OneBottleneck True numElems) + , TestLabel "test_fifo_OneBottleneck_backoff" (TestCase$ assert $ 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_contention_free_parallel" (TestCase$ assert $ test_contention_free_parallel True numElems newq) + ] ++ + [ TestLabel ("test_random_array_comm_"++show size) + (TestCase$ assert $ test_random_array_comm size numElems newq) + | size <- [10,100] ] @@ -233,12 +338,91 @@ [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 () +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 + + qs <- sequence (replicate producers newqueue) + -- A work-stealing deque only has ONE threadsafe end: + assertBool "test_random_array_comm requires thread safe right end" (rightThreadSafe (head qs)) + 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 + + 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: + b <- randomIO :: IO Bool + if b then do + x <- spinPopN 100 (tryPopL myQ) + case x of + Nothing -> loop i acc + Just n -> loop i (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 + + -- Each consumer doesn't quit until it has popped "perthread": + 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 + -- 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) + Nothing -> + consume_loop summ (i+1) -- Increment even if we don't get a result. + in consume_loop 0 0 + + dbgPrintLn 1 "Reading sums from MVar..." + prod_ls <- mapM (\_ -> takeMVar prod_results) [1..producers] + + leftovers <- forM qs $ \ q -> + let loop !acc = do + x <- tryPopR q + case x of + Nothing -> return acc + Just n -> loop (acc+n) + in loop 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) + dbgPrintLn 1$ "Checking that queue is finally null..." + assertEqual "Correct final sum" (producers * expectedSum perthread) finalSum + bs <- mapM nullQ qs + if all id bs + then dbgPrintLn 1$ "Sum matched expected, test passed." + else assertFailure "Queue was not empty!!" + + + -- | Aggregate tests for work stealing queues. test_wsqueue :: (PopL d) => (forall elt. IO (d elt)) -> Test -test_wsqueue newq = TestList +test_wsqueue newq = TestList $ [ 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) ] ---------------------------------------------------------------------------------------------------- @@ -253,8 +437,42 @@ ] ---------------------------------------------------------------------------------------------------- --- Helpers +-- Misc Helpers +---------------------------------------------------------------------------------------------------- +---------------------------------------------------------------------------------------------------- +-- Misc Helpers +---------------------------------------------------------------------------------------------------- + +myfork :: String -> IO () -> IO ThreadId +myfork msg = forkWithExceptions forkThread msg + +-- Exceptions that walk up the fork tree of threads: +forkWithExceptions :: (IO () -> IO ThreadId) -> String -> IO () -> IO ThreadId +forkWithExceptions forkit descr action = do + parent <- myThreadId + forkit $ + Control.Exception.catch action + (\ e -> + case fromException e of + -- Let threadKilled exceptions through. + Just ThreadKilled -> return () + _ -> do + hPutStrLn stderr $ "Exception inside child thread "++show descr++": "++show e + throwTo parent (e::SomeException) + ) + +forkJoin :: Int -> (Int -> IO b) -> IO [b] +forkJoin numthreads action = + do + answers <- sequence (replicate numthreads newEmptyMVar) -- padding? + forM_ (zip [0..] answers) $ \ (ix,mv) -> + myfork "forkJoin worker" (action ix >>= putMVar mv) + -- Reading answers: + ls <- mapM readMVar answers + return ls + +spinPopBkoff :: DequeClass d => d t -> IO (t, Int) spinPopBkoff q = loop 1 where hardspinfor = 10 @@ -263,7 +481,7 @@ errorafter = 1 * 1000 * 1000 loop n = do when (n == warnafter) - (putStrLn$ "Warning: Failed to pop "++ show warnafter ++ + (dbgPrintLn 0$ "Warning: Failed to pop "++ show warnafter ++ " times consecutively. That shouldn't happen in this benchmark.") -- when (n == errorafter) (error "This can't be right. A queue consumer spun 1M times.") x <- tryPopR q @@ -283,6 +501,7 @@ -- | This is always a crazy and dangerous thing to do -- GHC cannot -- deschedule this spinning thread because it does not allocate: +spinPopHard :: (DequeClass d) => d t -> IO (t, Int) spinPopHard q = loop 1 where loop n = do @@ -291,7 +510,53 @@ Nothing -> do loop (n+1) Just x -> return (x, n) +spinPopN :: Int -> IO (Maybe t) -> IO (Maybe t) +spinPopN 0 _ = return Nothing +spinPopN tries act = do + x <- act + case x of + Nothing -> spinPopN (tries-1) act + res@(Just _) -> return res +-- 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_ 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) + ---------------------------------------------------------------------------------------------------- +-- DEBUGGING +---------------------------------------------------------------------------------------------------- +-- | Debugging flag shared by all accelerate-backend-kit modules. +-- This is activated by setting the environment variable DEBUG=1..5 +dbg :: Int +dbg = case lookup "DEBUG" theEnv of + Nothing -> defaultDbg + Just "" -> defaultDbg + Just "0" -> defaultDbg + Just s -> + trace (" ! Responding to env Var: DEBUG="++s)$ + case reads s of + ((n,_):_) -> n + [] -> error$"Attempt to parse DEBUG env var as Int failed: "++show s + +defaultDbg :: Int +defaultDbg = 0 + +-- | Print if the debug level is at or above a threshold. +dbgPrint :: Int -> String -> IO () +dbgPrint lvl str = if dbg < lvl then return () else do +-- hPutStrLn stderr str + -- hPrintf stderr str + -- hFlush stderr + printf str + hFlush stdout + +dbgPrintLn :: Int -> String -> IO () +dbgPrintLn lvl str = dbgPrint lvl (str++"\n") 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.1.6 +Version: 0.1.7 License: BSD3 License-file: LICENSE Author: Ryan R. Newton @@ -18,6 +18,7 @@ -- 0.1.4: Another release. -- 0.1.5: Fix for 6.12 and 7.0. -- 0.1.6: Make useCAS default FALSE. +-- 0.1.7: Add leftThreadSafe / rightThreadSafe Synopsis: Abstract, parameterized interface to mutable Deques. @@ -54,7 +55,7 @@ Data.Concurrent.Deque.Tests, Data.Concurrent.Deque.Reference, Data.Concurrent.Deque.Reference.DequeInstance - build-depends: base >= 4 && < 5, + build-depends: base >= 4 && < 5, array, random, containers, HUnit if flag(useCAS) && impl( ghc >= 7.4 ) && !os(mingw32) { @@ -70,3 +71,25 @@ Source-Repository head Type: git Location: git://github.com/rrnewton/haskell-lockfree-queue.git + +Test-Suite test-abstract-deque + 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 + 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 + + + -- Disabling this for now due to a limitation of cabal-install-0.10.2: + -- Examining generated code: + -- If + -------------------------------------------------------------------------------- + -- if flag(useCAS) && impl( ghc >= 7.4 ) && !os(mingw32) { + -- ghc-options: -with-rtsopts=-K32M + -- build-depends: IORefCAS >= 0.2 + -- cpp-options: -DUSE_CAS -DDEFAULT_SIGNATURES + -- } + -- if impl( ghc < 7 ) + -- buildable: False