abstract-deque 0.2.2.1 → 0.3
raw patch · 4 files changed
+9/−904 lines, 4 filesdep +atomic-primopsdep −HUnitdep −IORefCASdep −test-framework
Dependencies added: atomic-primops
Dependencies removed: HUnit, IORefCAS, test-framework, test-framework-hunit
Files
- Data/Concurrent/Deque/Reference.hs +2/−0
- Data/Concurrent/Deque/Tests.hs +0/−770
- Test.hs +0/−105
- abstract-deque.cabal +7/−29
Data/Concurrent/Deque/Reference.hs view
@@ -24,6 +24,7 @@ import qualified Data.Concurrent.Deque.Class as C import Data.Sequence import Data.IORef+ #ifdef USE_CAS #warning "abstract-deque: reference implementation using CAS..." import Data.CAS (atomicModifyIORefCAS)@@ -34,6 +35,7 @@ modify = atomicModifyIORef _is_using_CAS = False #endif+ {-# INLINE modify #-} modify :: IORef a -> (a -> (a, b)) -> IO b _is_using_CAS :: Bool
− Data/Concurrent/Deque/Tests.hs
@@ -1,770 +0,0 @@-{-# LANGUAGE BangPatterns, RankNTypes, CPP, BangPatterns #-}---- | This module contains a battery of simple tests for queues--- implementing the interface defined in--- ` Data.Concurrent.Deque.Class`.--{-# OPTIONS_GHC -fwarn-incomplete-patterns #-}-module Data.Concurrent.Deque.Tests - ( - -- * Tests for simple FIFOs.- test_fifo_filldrain, test_fifo_OneBottleneck, tests_fifo,-- -- * Tests for Work-stealing queues.- test_ws_triv1, test_ws_triv2, tests_wsqueue,- test_parfib_work_stealing,-- -- * All deque tests, aggregated.- tests_all,-- -- * Testing parameters- numElems, getNumAgents, producerRatio,-- -- * Utility for tweaking test suites- setTestThreads, appendLabels, appendLabel,-- -- * Test initialization, reading common configs- stdTestHarness, Elt,-- -- * Misc helpers- forkJoin, timeit, fibSize- )- where --import Control.Monad-import Data.Time.Clock-import Data.Array as A-import Data.IORef-import Data.Int-import qualified Data.Set as S-import Text.Printf-import GHC.Conc (throwTo, threadDelay, myThreadId)-import Control.Concurrent.MVar-import Control.Concurrent (yield, forkOS, forkIO, ThreadId)-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 qualified Test.Framework as TF-import Test.Framework.Providers.HUnit (hUnitTestToTests)-import Test.HUnit as HU--import Debug.Trace (trace)--import Data.Concurrent.Deque.Class as C-import qualified Data.Concurrent.Deque.Reference as R---#if __GLASGOW_HASKELL__ >= 704-import GHC.Conc (getNumCapabilities, setNumCapabilities, getNumProcessors)-#else-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--------------------------------------------------------------------------------------------------------- TODO: In addition to setting these parameters from environment--- variables, it would be nice to route all of this through a--- configuration record, so that it can be changed programmatically.---- How many elements should each of the tests pump through the queue(s)?-numElems :: Int-numElems = case lookup "NUMELEMS" theEnv of - Nothing -> 100 * 1000 -- 500000- Just str -> warnUsing ("NUMELEMS = "++str) $ - read str-fibSize :: Int64-fibSize = case lookup "FIBSIZE" theEnv of- Just s -> read s- Nothing -> 32--forkThread :: IO () -> IO ThreadId-forkThread = case lookup "OSTHREADS" theEnv of - Nothing -> forkIO- Just x -> warnUsing ("OSTHREADS = "++x) $ - case x of - "0" -> forkIO- "False" -> forkIO- "1" -> forkOS- "True" -> forkOS- oth -> error$"OSTHREAD environment variable set to unrecognized option: "++oth---- | How many communicating agents are there? By default one per--- thread used by the RTS.-getNumAgents :: IO Int-getNumAgents = case lookup "NUMAGENTS" theEnv of - Nothing -> do x <- getNumCapabilities- putStrLn$"Defaulting numAgents to numCapabilities: "++show x- return x- 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--- settings of this parameter less than or greater than 1).-producerRatio :: Double-producerRatio = case lookup "PRODUCERRATIO" theEnv of - Nothing -> 1.0- Just str -> warnUsing ("PRODUCERRATIO = "++str) $ - read str--warnUsing :: String -> a -> a-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)---- | Dig through the test constructors and add a new string to the first label found.--- If no such label exists, add one. -appendLabel :: String -> HU.Test -> HU.Test-appendLabel newLab tst = loop tst- where- loop x = - case x of- TestLabel lb t2 -> TestLabel (newLab ++"_"++ lb) t2- TestList ls -> TestList (map loop ls)- TestCase io -> TestLabel newLab x---- | This version has the option of being smarter about how it handles uniformly--- labeling many tests.-appendLabels :: String -> [HU.Test] -> HU.Test-appendLabels newLab tst = TestList $ map (appendLabel newLab) tst--stdTestHarness :: (IO Test) -> IO ()-stdTestHarness genTests = do - putStrLn$ "Running with numElems "++show numElems- 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 Elt -> IO ()-test_fifo_filldrain q = - do- dbgPrintLn 1 "\nTest FIFO queue: sequential fill and then drain"- dbgPrintLn 1 "==============================================="--- let n = 1000- let n = fromIntegral numElems- dbgPrintLn 1$ "Done creating queue. Pushing "++show n++" elements:"- forM_ [1..n] $ \i -> do - pushL q i- 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) $ dbgPrint 1 $ printf " %d" x- loop (i-1) (sumR + x)- s <- loop n 0- 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 ()--{-# 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--- 'Int' argument 'total' designates how many total items should be--- communicated (irrespective of 'numAgents').-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)- 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)")- dbgPrintLn 1 "======================================================" -- 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-- 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- dbgPrint 1 $ printf "Forking %d consumer threads, each consuming ~%d elements.\n" consumers perthread2- - 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)- forI_ start (start+perthread) $ \ i -> do - pushL q 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- 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-- 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 (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 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)-- 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- 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- forI_ 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 == 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-- 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" (fromIntegral producers * expectedSum (fromIntegral 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!!"---{-# INLINE test_random_array_comm #-}--- | This test uses a number of producer and consumer threads which push and pop--- elements from random positions in an array of FIFOs.-test_random_array_comm :: DequeClass d => Int -> Int -> IO (d Elt) -> IO ()-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-- 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 "======================================================" -- numAgents <- getNumAgents- let producers = max 1 (round$ producerRatio * (fromIntegral numAgents) / (producerRatio + 1))- consumers = max 1 (numAgents - 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- 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" (fromIntegral 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 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 = appendLabels "single-ended-queue-tests" $ - tests_basic newq ++ - tests_fifo_exclusive newq- -{-# INLINE tests_fifo_exclusive #-}--- | Tests exclusive to single-ended (threadsafe) queues:-tests_fifo_exclusive :: DequeClass d => (forall elt. IO (d elt)) -> [Test]-tests_fifo_exclusive newq = - [ TestLabel "test_fifo_OneBottleneck_backoff" (TestCase$ assertT $ newq >>= test_fifo_OneBottleneck True numElems)--- , TestLabel "test_fifo_OneBottleneck_aggressive" (TestCase$ assert $ newq >>= test_fifo_OneBottleneck False numElems)--- , TestLabel "test the tests" (TestCase$ assert $ assertFailure "This SHOULD fail.")- ] ++- [ TestLabel ("test_random_array_comm_"++show size)- (TestCase$ assertT $ test_random_array_comm size numElems newq)- | size <- [10,100]- ]--{-# INLINE tests_basic #-}--- | Tests that don't require thread safety at either end.-tests_basic :: DequeClass d => (forall elt. IO (d elt)) -> [Test]-tests_basic newq =- [ TestLabel "test_fifo_filldrain" (TestCase$ assertT $ newq >>= test_fifo_filldrain)- , TestLabel "test_contention_free_parallel" (TestCase$ assertT $ test_contention_free_parallel True numElems newq)- ]--assertT :: IO () -> IO ()-assertT = timeit . assert--------------------------------------------------------------------------------------------------------- Test a Work-stealing queue:--------------------------------------------------------------------------------------------------------- | Trivial test: push then pop.-test_ws_triv1 :: PopL d => d [Char] -> IO ()-test_ws_triv1 q = do- pushL q "hi" - Just x <- tryPopL q - assertEqual "test_ws_triv1" x "hi"---- | Trivial test: push left, pop left and right.-test_ws_triv2 :: PopL d => d [Char] -> IO ()-test_ws_triv2 q = do- pushL q "one" - pushL q "two" - pushL q "three" - pushL q "four" - ls <- sequence [tryPopR q, tryPopR q, - tryPopL q, tryPopL q,- tryPopL q, tryPopR q ]- assertEqual "test_ws_triv2" ls - [Just "one",Just "two",Just "four",Just "three",Nothing,Nothing]---{-# INLINE test_random_work_stealing #-}--- | This test uses a number of worker threads which randomly either push or pop work--- to their local work stealing deque. Also, there are consumer (always thief)--- threads which never produce and only consume.-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 "======================================================" -- 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:- 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 :: 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 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) pops acc- in loop 0 0 0-- -- In this version- consumer_sums <- forkJoin consumers $ \ ind ->- 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 try#%d = %d\n" ind i x- consume_loop (summ+x) (successes+1) (i+1)- Nothing ->- 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 !cnt !acc = do- x <- tryPopR q -- This should work as a popR OR a popL.- case x of- Nothing -> return (cnt,acc)- Just n -> loop (cnt+1) (acc+n)- in loop 0 0- - 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" (fromIntegral 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!!"---{-# INLINE test_parfib_work_stealing #-}-test_parfib_work_stealing :: (DequeClass d, PopL d) => Elt -> IO (d Elt) -> IO ()-test_parfib_work_stealing origInput newqueue = do- putStrLn$ " [parfib] Computing fib("++show origInput++")"- numAgents <- getNumAgents- qs <- sequence (replicate numAgents newqueue)- let arr = A.listArray (0,numAgents - 1) qs - - let parfib !myId !myQ !mySum !num- | num <= 2 =- do x <- tryPopL myQ- case x of- Nothing -> trySteal myId myQ (mySum+1)- Just n -> parfib myId myQ (mySum+1) n- | otherwise = do - pushL myQ (num-1)- parfib myId myQ mySum (num-2)- - trySteal !myId !myQ !mySum =- let loop ind- -- After we finish one sweep... we're completely done.- | ind == myId = return mySum- | ind == size arr = loop 0- | otherwise = do- x <- tryPopR (arr ! ind)- case x of- Just n -> parfib myId myQ mySum n- Nothing -> do yield- loop (ind+1)- in loop (myId+1)-- size a = let (st,en) = A.bounds a in en - st + 1 - - partial_sums <- forkJoin numAgents $ \ myId ->- if myId == 0- then parfib myId (arr ! myId) 0 origInput- else trySteal myId (arr ! myId) 0 -- putStrLn$ " [parfib] fib("++show origInput++") = "++show (sum partial_sums)---{-# INLINE tests_wsqueue #-}--- | Aggregate tests for work stealing queues. None of these require thread-safety--- on the left end. There is some duplication with tests_fifo.-tests_wsqueue :: (PopL d) => (forall elt. IO (d elt)) -> Test-tests_wsqueue newq = appendLabels "work-stealing-deque-tests"$ - tests_wsqueue_exclusive newq ++- tests_basic newq --{-# INLINE tests_wsqueue_exclusive #-}--- Internal: factoring this out.-tests_wsqueue_exclusive :: (PopL d) => (forall elt. IO (d elt)) -> [Test]-tests_wsqueue_exclusive newq = - [ TestLabel "test_ws_triv1" (TestCase$ assertT $ newq >>= test_ws_triv1)- , TestLabel "test_ws_triv2" (TestCase$ assertT $ newq >>= test_ws_triv2)- , TestLabel "test_random_work_stealing" (TestCase$ assertT $ test_random_work_stealing numElems newq)- , TestLabel "test_parfib_generic" (TestCase$ assertT $ test_parfib_work_stealing fibSize newq)- ]--------------------------------------------------------------------------------------------------------- Combine all tests -- for a deques supporting all capabilities.--------------------------------------------------------------------------------------------------------- | This requires double ended queues that are threadsafe on BOTH ends.-tests_all :: (PopL d) => (forall elt. IO (d elt)) -> Test-tests_all newq = appendLabels "full-deque-tests" $ - tests_basic newq ++- tests_fifo_exclusive newq ++- tests_wsqueue_exclusive newq --------------------------------------------------------------------------------------------------------- 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---- | Pop from the right end more gently.-spinPopBkoff :: DequeClass d => d t -> IO (t, Int)-spinPopBkoff q = loop 1- where - hardspinfor = 10- sleepevery = 1000- warnafter = 5000- errorafter = 1 * 1000 * 1000- loop n = do- when (n == 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 - case x of - -- This yields EVERY time. And yet we get these real bursts / runs of failure.- Nothing -> do - -- Every `sleepevery` times do a significant delay:- if n `mod` sleepevery == 0 - then do dbgPrint 1 "!"- threadDelay n -- 1ms after 1K fails, 2 after 2K...- else when (n > hardspinfor) $ do - 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. 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 - loop n = do- x <- tryPopR q - case x of - 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 => 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 several 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")--timeit :: IO a -> IO a -timeit ioact = do - start <- getCurrentTime- res <- ioact- end <- getCurrentTime- putStrLn$ "SELFTIMED: " ++ show (diffUTCTime end start)- return res
− Test.hs
@@ -1,105 +0,0 @@-{-# LANGUAGE CPP, ScopedTypeVariables, NamedFieldPuns, BangPatterns #-}-#if __GLASGOW_HASKELL >= 700-{-# OPTIONS_GHC -with-rtsopts=-K32M #-}-#endif--import Data.Concurrent.Deque.Class--- import Data.Concurrent.Deque.Class.Reference (newQueue)--- import Data.Concurrent.MegaDeque -import Data.Int-import Data.Array as A-import Control.Concurrent (yield)-import Test.Framework (defaultMain)-import Test.Framework.Providers.HUnit (hUnitTestToTests)-import Test.HUnit (assert, assertEqual, Test(TestCase, TestList, TestLabel)) -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 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::Integer)--test_2 :: Test-test_2 = TestCase $ assert $ - do - -- Here's an example of type-based restriction of the queue implementation:- q <- newQ :: IO (Deque NT T D S Bound Safe Int)- pushL q 33--- pushR q 33 -- This would cause a type error because the Right end is not Double-capable.- Just x <- tryPopR q- assertEqual "test_2 result" x 33--test_parfib_work_stealing_specialized :: T.Elt -> IO T.Elt-test_parfib_work_stealing_specialized origInput = do- putStrLn$ " [parfib] Computing fib("++show origInput++")"- numAgents <- T.getNumAgents- qs <- sequence (replicate numAgents R.newQ)- let arr = A.listArray (0,numAgents - 1) qs - - let parfib !myId !myQ !mySum !num- | num <= 2 =- do x <- R.tryPopL myQ- case x of- Nothing -> trySteal myId myQ (mySum+1)- Just n -> parfib myId myQ (mySum+1) n- | otherwise = do - R.pushL myQ (num-1)- parfib myId myQ mySum (num-2)- - trySteal !myId !myQ !mySum =- let loop ind- -- After we finish one sweep... we're completely done.- | ind == myId = return mySum- | ind == size arr = loop 0- | otherwise = do- x <- R.tryPopR (arr ! ind)- case x of- Just n -> parfib myId myQ mySum n- Nothing -> do yield- loop (ind+1)- in loop (myId+1)-- size a = let (st,en) = A.bounds a in en - st + 1 - - partial_sums <- T.forkJoin numAgents $ \ myId ->- if myId == 0- then parfib myId (arr ! myId) 0 origInput- else trySteal myId (arr ! myId) 0 - - return (sum partial_sums)--main :: IO ()-#if __GLASGOW_HASKELL__ >= 700-main = T.stdTestHarness $ return all_tests- where - all_tests :: Test- all_tests = - T.appendLabels "Reference_Deque" $ - [ T.appendLabel "test_1" test_1- , T.appendLabel "test_2" test_2 - , T.appendLabel "direct"$ T.tests_all R.newQ- -- Test going through the class interface as well: - , T.appendLabel "thru_class"$ T.tests_all (C.newQ :: IO (R.SimpleDeque a))- , T.appendLabel "with_debug"$ T.tests_all (C.newQ :: IO (DebugDeque R.SimpleDeque a))- - , TestLabel "parfib_specialized" $ TestCase $ T.timeit$- print =<< test_parfib_work_stealing_specialized T.fibSize- ]---- main = do --- 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
abstract-deque.cabal view
@@ -1,5 +1,5 @@ Name: abstract-deque-Version: 0.2.2.1 +Version: 0.3 License: BSD3 License-file: LICENSE Author: Ryan R. Newton@@ -8,7 +8,8 @@ Build-type: Simple Cabal-version: >= 1.8 -Homepage: https://github.com/rrnewton/haskell-lockfree-queue/wiki+Homepage: https://github.com/rrnewton/haskell-lockfree/wiki+Bug-Reports: https://github.com/rrnewton/haskell-lockfree/issues -- Version History: -- 0.1: @@ -22,6 +23,7 @@ -- 0.2: [breaking] Refactor names of exposed Tests -- 0.2.2: Add Debugger -- 0.2.2.1: Add some extra testing helpers.+-- 0.3: Remove testing utilities to their own package. Synopsis: Abstract, parameterized interface to mutable Deques. @@ -55,18 +57,14 @@ Library exposed-modules: Data.Concurrent.Deque.Class,- Data.Concurrent.Deque.Tests, Data.Concurrent.Deque.Reference, Data.Concurrent.Deque.Reference.DequeInstance, Data.Concurrent.Deque.Debugger build-depends: base >= 4 && < 5, array, random,- containers, - HUnit, test-framework >= 0.6, test-framework-hunit >= 0.2.7, time+ containers, time if flag(useCAS) && impl( ghc >= 7.4 ) && !os(mingw32) {- build-depends: IORefCAS >= 0.2 + build-depends: atomic-primops >= 0.5.0.2 cpp-options: -DUSE_CAS -DDEFAULT_SIGNATURES--- extensions: DefaultSignatures--- GHC-Options: -XDefaultSignatures } extensions: CPP@@ -74,27 +72,7 @@ Source-Repository head Type: git- Location: git://github.com/rrnewton/haskell-lockfree-queue.git+ Location: git://github.com/rrnewton/haskell-lockfree.git -Test-Suite test-abstract-deque- type: exitcode-stdio-1.0- main-is: Test.hs- build-depends: base >= 4 && < 5, containers, HUnit, array, random,- HUnit, test-framework >= 0.6, test-framework-hunit >= 0.2.7, time- ghc-options: -O2 -threaded -rtsopts - -- ghc-options: -keep-tmp-files -dsuppress-module-prefixes -ddump-to-file -ddump-core-stats -ddump-simpl-stats -dcore-lint -dcmm-lint- -- 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