abstract-deque 0.1.5 → 0.1.6

raw patch · 6 files changed

+249/−62 lines, 6 filesPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

API changes (from Hackage documentation)

Files

• Data/Concurrent/Deque/Class.hs +5/−5
• Data/Concurrent/Deque/Reference.hs +27/−8
• Data/Concurrent/Deque/Reference/DequeInstance.hs +4/−2
• Data/Concurrent/Deque/Tests.hs +151/−38
• Test.hs +47/−0
• abstract-deque.cabal +15/−9

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

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

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

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

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

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