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 @@ -9,7 +9,7 @@ An abstract, parameterizable interface for queues. This interface includes a non-associated type family for Deques - plus separate type classes encapusulating the Deque operations. + plus separate type classes encapsulating the Deque operations. This design strives to hide the extra phantom-type parameters from the Class constraints and therefore from the type signatures of client code. @@ -20,9 +20,9 @@ -- * Highly parameterized Deque type(s) Deque -- ** The choices that select a queue-variant. - -- *** Choice #1 -- thread saftety. + -- *** Choice #1 -- thread safety. , Threadsafe, Nonthreadsafe - -- *** Choice #2 -- double or or single functionality on an end. + -- *** Choice #2 -- double or single functionality on an end. , SingleEnd, DoubleEnd -- *** Choice #3 -- bounded or growing queues: , Bound, Grow @@ -62,7 +62,7 @@ several choices. For example, a work stealing deque is threadsafe only on one end and - supports push/pop on one end (and popo-only) on the other: + supports push/pop on one end (and pop-only) on the other: >> (Deque NT T D S Grow elt) @@ -92,7 +92,7 @@ -- (right) functionality. Thus a 'SingleEnd' / 'SingleEnd' combination -- is what is commonly referred to as a /single ended queue/, whereas -- 'DoubleEnd' / 'DoubleEnd' is --- a /double ended queue/. Heterogenous combinations are sometimes +-- a /double ended queue/. Heterogeneous combinations are sometimes -- colloquially referred to as \"1.5 ended queues\". data SingleEnd -- | This end of the queue supports both push and pop. 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 @@ -1,25 +1,28 @@ -{-# LANGUAGE TypeFamilies, CPP #-} +{-# LANGUAGE TypeFamilies, CPP, BangPatterns #-} {-| - A strawman implementation of concurrent Dequeus. This + A strawman implementation of concurrent Dequeues. This implementation is so simple that it also makes a good reference implementation for debugging. The queue representation is simply an IORef containing a Data.Sequence. - Note, alse see "Data.Concurrent.Deque.Reference.DequeInstance". + Also see "Data.Concurrent.Deque.Reference.DequeInstance". By convention a module of this name is also provided. -} module Data.Concurrent.Deque.Reference (SimpleDeque(..), - newQ, nullQ, newBoundedQ, pushL, pushR, tryPopR, tryPopL, tryPushL, tryPushR + newQ, nullQ, newBoundedQ, pushL, pushR, tryPopR, tryPopL, tryPushL, tryPushR, + + _is_using_CAS -- Internal ) where 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 @@ -27,26 +30,42 @@ import Data.CAS (atomicModifyIORefCAS) -- Toggle these and compare performance: modify = atomicModifyIORefCAS +_is_using_CAS = True #else modify = atomicModifyIORef +_is_using_CAS = False #endif {-# INLINE modify #-} +modify :: IORef a -> (a -> (a, b)) -> IO b +_is_using_CAS :: Bool + -- | Stores a size bound (if any) as well as a mutable Seq. data SimpleDeque elt = DQ {-# UNPACK #-} !Int !(IORef (Seq elt)) +newQ :: IO (SimpleDeque elt) newQ = do r <- newIORef empty - return (DQ 0 r) + return $! DQ 0 r +newBoundedQ :: Int -> IO (SimpleDeque elt) newBoundedQ lim = do r <- newIORef empty - return (DQ lim r) + return $! DQ lim r -pushL (DQ 0 qr) x = modify qr (\s -> (x <| s, ())) +pushL :: SimpleDeque t -> t -> IO () +pushL (DQ 0 qr) !x = do + () <- modify qr addleft + return () + where + -- Here we are very strict to avoid stack leaks. + addleft !s = extended `seq` pair + where extended = x <| s + pair = (extended, ()) pushL (DQ n _) _ = error$ "should not call pushL on Deque with size bound "++ show n -tryPopR (DQ _ qr) = modify qr $ \s -> +tryPopR :: SimpleDeque a -> IO (Maybe a) +tryPopR (DQ _ qr) = modify qr $ \ s -> case viewr s of EmptyR -> (empty, Nothing) s' :> x -> (s', Just x) diff --git a/Data/Concurrent/Deque/Reference/DequeInstance.hs b/Data/Concurrent/Deque/Reference/DequeInstance.hs --- a/Data/Concurrent/Deque/Reference/DequeInstance.hs +++ b/Data/Concurrent/Deque/Reference/DequeInstance.hs @@ -1,10 +1,12 @@ {-# LANGUAGE TypeFamilies, TypeSynonymInstances #-} {- | + By convention, every provider of the "Data.Concurrent.Deque.Class" interface optionally provides a module that provides the relevant - instances of the 'Deque' type class, covering the portion of the - configuration space that the implementation is able to handle. + instances of the 'Deque' type class, covering the [maximum] portion + of the configuration space that the implementation is able to + handle. This is kept in a separate package because importing instances is unconditional and the user may well want to assemble their own diff --git a/Data/Concurrent/Deque/Tests.hs b/Data/Concurrent/Deque/Tests.hs --- a/Data/Concurrent/Deque/Tests.hs +++ b/Data/Concurrent/Deque/Tests.hs @@ -1,8 +1,14 @@ {-# LANGUAGE BangPatterns, RankNTypes #-} + +-- | 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_HalfToHalf, test_fifo, + test_fifo_filldrain, test_fifo_OneBottleneck, test_fifo, -- * Tests for Work-stealing queues. test_ws_triv1, test_ws_triv2, test_wsqueue, @@ -17,18 +23,90 @@ import Control.Monad import Data.IORef -import System.Mem.StableName +-- import System.Mem.StableName import Text.Printf -import GHC.IO (unsafePerformIO) -import GHC.Conc +import GHC.Conc (numCapabilities, throwTo, threadDelay, myThreadId) import Control.Concurrent.MVar -import Control.Concurrent (yield, forkOS) - -import System.Environment +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.Unsafe (unsafePerformIO) import Test.HUnit +import Debug.Trace (trace) +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 -> 50 * 1000 -- 500000 + Just str -> warnUsing ("NUMELEMS = "++str) $ + read str + +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. +numAgents :: Int +numAgents = case lookup "NUMAGENTS" theEnv of + Nothing -> numCapabilities + Just str -> warnUsing ("NUMAGENTS = "++str) $ + 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 + + +---------------------------------------------------------------------------------------------------- +-- 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: ---------------------------------------------------------------------------------------------------- @@ -38,47 +116,51 @@ do -- q <- newQ putStrLn "\nTest FIFO queue: sequential fill and then drain" putStrLn "===============================================" - let n = 1000 +-- let n = 1000 + let n = numElems putStrLn$ "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..." - sumR <- newIORef 0 - forM_ [1..n] $ \i -> do - (x,_) <- spinPop q - when (i < 200) $ printf " %d" x - modifyIORef sumR (+x) - s <- readIORef sumR + + let loop 0 !sumR = return sumR + loop i !sumR = do + (x,_) <- spinPopBkoff q + when (i < 200) $ 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 when (s /= expected) (assertFailure "Incorrect sum!") --- return s return () --- myfork = forkIO -myfork = forkOS --- | This one splits the 'numCapabilities' threads into producers and --- consumers. Each thread performs its designated operation as fast --- as possible. The 'Int' argument 'total' designates how many total --- items should be communicated (irrespective of 'numCapabilities'). -test_fifo_HalfToHalf :: DequeClass d => Int -> d Int -> IO () -test_fifo_HalfToHalf total q = +-- | 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 Int -> IO () +test_fifo_OneBottleneck doBackoff total q = do -- q <- newQ - putStrLn$ "\nTest FIFO queue: producer/consumer Half-To-Half" - putStrLn "===============================================" + putStrLn$ "\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 - let producers = max 1 (numCapabilities `quot` 2) - consumers = producers + let producers = max 1 (round$ producerRatio * (fromIntegral numAgents) / (producerRatio + 1)) + consumers = max 1 (numAgents - producers) perthread = total `quot` producers + 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 forM_ [0..producers-1] $ \ id -> - myfork $ + myfork "producer thread" $ forM_ (take perthread [id * producers .. ]) $ \ i -> do pushL q i when (i - id*producers < 10) $ printf " [%d] pushed %d \n" id i @@ -86,10 +168,11 @@ printf "Forking %d consumer threads.\n" consumers forM_ [0..consumers-1] $ \ id -> - myfork $ do + myfork "consumer thread" $ do - let fn (!sum,!maxiters) i = do - (x,iters) <- spinPop q + 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) @@ -106,13 +189,21 @@ if b then putStrLn$ "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_fifo_mailboxes + +------------------------------------------------------------ + -- | 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 [ TestLabel "test_fifo_filldrain" (TestCase$ assert $ newq >>= test_fifo_filldrain) -- Do half a million elements by default: - , TestLabel "test_fifo_HalfToHalf" (TestCase$ assert $ newq >>= test_fifo_HalfToHalf (500 * 1000)) + , 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.") ] @@ -164,21 +255,43 @@ ---------------------------------------------------------------------------------------------------- -- Helpers -spinPop q = loop 1 +spinPopBkoff q = loop 1 where - warnevery = 5000 + hardspinfor = 10 + sleepevery = 1000 + warnafter = 5000 errorafter = 1 * 1000 * 1000 loop n = do - when (n == warnevery) - (putStrLn$ "Warning: Failed to pop "++ show warnevery ++ + when (n == warnafter) + (putStrLn$ "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 - Nothing -> do putStr "." - yield + -- 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 putStr "!" + threadDelay n -- 1ms after 1K fails, 2 after 2K... + else when (n > hardspinfor) $ do + putStr "." + 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: +spinPopHard q = loop 1 + where + loop n = do + x <- tryPopR q + case x of + Nothing -> do loop (n+1) + Just x -> return (x, n) + + ---------------------------------------------------------------------------------------------------- diff --git a/Test.hs b/Test.hs new file mode 100644 --- /dev/null +++ b/Test.hs @@ -0,0 +1,47 @@ +{-# LANGUAGE CPP, ScopedTypeVariables, NamedFieldPuns #-} +#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 Test.Framework (defaultMain) +import Test.Framework.Providers.HUnit (hUnitTestToTests) +import Test.HUnit (assert, assertEqual, Test(TestCase, TestList)) +import qualified Data.Concurrent.Deque.Tests as T +import qualified Data.Concurrent.Deque.Reference as R + +-- Import the instances: +import qualified Data.Concurrent.Deque.Reference.DequeInstance +import System.Exit + +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 + +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 + +#if __GLASGOW_HASKELL__ >= 700 +main = do + putStrLn "[ Test executable: test reference deque implementation... ]" + defaultMain$ hUnitTestToTests$ + TestList $ + [ + T.test_all R.newQ + , test_1 + , test_2 + ] +#else +main = putStrLn "WARNING: Tests disabled for GHC < 7" +#endif diff --git a/abstract-deque.cabal b/abstract-deque.cabal --- a/abstract-deque.cabal +++ b/abstract-deque.cabal @@ -1,5 +1,5 @@ Name: abstract-deque -Version: 0.1.5 +Version: 0.1.6 License: BSD3 License-file: LICENSE Author: Ryan R. Newton @@ -8,6 +8,8 @@ Build-type: Simple Cabal-version: >= 1.8 +Homepage: https://github.com/rrnewton/haskell-lockfree-queue/wiki + -- Version History: -- 0.1: -- 0.1.1: Added nullQ to interface. [First release] @@ -15,6 +17,7 @@ -- 0.1.3: Actually turned on real CAS! DANGER -- 0.1.4: Another release. -- 0.1.5: Fix for 6.12 and 7.0. +-- 0.1.6: Make useCAS default FALSE. Synopsis: Abstract, parameterized interface to mutable Deques. @@ -39,6 +42,13 @@ This package also includes a simple reference implementation based on 'IORef' and "Data.Sequence". + +-- Making this default False because abstract-deque should be VERY depndency-safe. +-- The reference implementation can be factored out in the future. +Flag useCAS + Description: Enable the reference implementation to use hardware compare-and-swap. + Default: False + Library exposed-modules: Data.Concurrent.Deque.Class, Data.Concurrent.Deque.Tests, @@ -47,7 +57,7 @@ build-depends: base >= 4 && < 5, containers, HUnit - if impl( ghc >= 7.2) { + if flag(useCAS) && impl( ghc >= 7.4 ) && !os(mingw32) { build-depends: IORefCAS >= 0.2 cpp-options: -DUSE_CAS -DDEFAULT_SIGNATURES -- extensions: DefaultSignatures @@ -57,10 +67,6 @@ extensions: CPP ghc-options: -O2 - - -Test-Suite test-abstract-deque - type: exitcode-stdio-1.0 - main-is: Test.hs - build-depends: base >= 4 && < 5, IORefCAS >= 0.2, containers, HUnit - ghc-options: -O2 -threaded -rtsopts +Source-Repository head + Type: git + Location: git://github.com/rrnewton/haskell-lockfree-queue.git