abstract-deque 0.1.6 → 0.1.7
raw patch · 4 files changed
+389/−81 lines, 4 filesdep +arraydep +randomdep +test-frameworkPVP: major bump suggested
API removals or changes: PVP suggests a major version bump
Dependencies added: array, random, test-framework, test-framework-hunit
API changes (from Hackage documentation)
+ Data.Concurrent.Deque.Class: leftThreadSafe :: DequeClass d => d elt -> Bool
+ Data.Concurrent.Deque.Class: rightThreadSafe :: DequeClass d => d elt -> Bool
+ Data.Concurrent.Deque.Tests: numAgents :: Int
+ Data.Concurrent.Deque.Tests: numElems :: Int
+ Data.Concurrent.Deque.Tests: producerRatio :: Double
- Data.Concurrent.Deque.Reference: DQ :: {-# UNPACK #-} !Int -> !IORef (Seq elt) -> SimpleDeque elt
+ Data.Concurrent.Deque.Reference: DQ :: {-# UNPACK #-} !Int -> !(IORef (Seq elt)) -> SimpleDeque elt
Files
- Data/Concurrent/Deque/Class.hs +21/−6
- Data/Concurrent/Deque/Reference.hs +6/−1
- Data/Concurrent/Deque/Tests.hs +337/−72
- abstract-deque.cabal +25/−2
Data/Concurrent/Deque/Class.hs view
@@ -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
Data/Concurrent/Deque/Reference.hs view
@@ -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
Data/Concurrent/Deque/Tests.hs view
@@ -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")
abstract-deque.cabal view
@@ -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