chaselev-deque (empty) → 0.1.2
raw patch · 6 files changed
+500/−0 lines, 6 filesdep +HUnitdep +abstract-dequedep +arraysetup-changed
Dependencies added: HUnit, abstract-deque, array, atomic-primops, base, bits-atomic, containers, ghc-prim, test-framework, test-framework-hunit, transformers, vector
Files
- Data/Concurrent/Deque/ChaseLev.hs +321/−0
- Data/Concurrent/Deque/ChaseLev/DequeInstance.hs +24/−0
- LICENSE +34/−0
- Setup.hs +2/−0
- Test.hs +62/−0
- chaselev-deque.cabal +57/−0
+ Data/Concurrent/Deque/ChaseLev.hs view
@@ -0,0 +1,321 @@+{-# LANGUAGE FlexibleInstances, NamedFieldPuns, CPP, ScopedTypeVariables, BangPatterns, MagicHash #-}++-- | Chase-Lev work stealing Deques+-- +-- This implementation derives directly from the pseudocode in the 2005 SPAA paper:+--+-- http://citeseerx.ist.psu.edu/viewdoc/download?doi=10.1.1.170.1097&rep=rep1&type=pdf+--+-- TODO: local topBound optimization.+-- TODO: Do the more optimized version of growCirc+module Data.Concurrent.Deque.ChaseLev + (+ -- The convention here is to directly provide the concrete+ -- operations as well as providing the class instances.+ ChaseLevDeque(), newQ, nullQ, pushL, tryPopL, tryPopR,+ dbgInspectCLD+ )+ where++import Data.IORef+import Data.List (isInfixOf, intersperse)+import qualified Data.Concurrent.Deque.Class as PC++-- import Data.CAS (casIORef)+import qualified Data.Vector.Mutable as MV+import qualified Data.Vector as V+-- import Data.Vector.Unboxed.Mutable as V+-- import Data.Vector+import Text.Printf (printf)+import Control.Exception (catch, SomeException, throw, evaluate,try)+import Control.Monad (when, unless, forM_)+--import Data.Atomics (readArrayElem, readForCAS, casIORef, Ticket, peekTicket)+-- import Data.Atomics.Counter.IORef+import Data.Atomics (storeLoadBarrier, writeBarrier, loadLoadBarrier)+import Data.Atomics.Counter.Reference+ (AtomicCounter, newCounter, readCounter, writeCounter, casCounter, readCounterForCAS, peekCTicket)++-- Debugging:+import System.IO.Unsafe (unsafePerformIO)+import Text.Printf (printf)+import System.Mem.StableName (makeStableName, hashStableName)+import GHC.Exts (Int(I#))+import GHC.Prim (reallyUnsafePtrEquality#, unsafeCoerce#)++--------------------------------------------------------------------------------+-- Instances++instance PC.DequeClass ChaseLevDeque where + newQ = newQ+ nullQ = nullQ+ pushL = pushL+ tryPopR = tryPopR+ -- | Popping the left end is the "local" side:+ leftThreadSafe _ = False+ rightThreadSafe _ = True++instance PC.PopL ChaseLevDeque where + tryPopL = tryPopL++--------------------------------------------------------------------------------+-- Type definition++data ChaseLevDeque a = CLD {+ top :: {-# UNPACK #-} !AtomicCounter+ , bottom :: {-# UNPACK #-} !AtomicCounter+ -- This is a circular array:+ , activeArr :: {-# UNPACK #-} !(IORef (MV.IOVector a))+ }++dbgInspectCLD :: Show a => ChaseLevDeque a -> IO String+dbgInspectCLD CLD{top,bottom,activeArr} = do+ tp <- readCounter top+ bt <- readCounter bottom+ vc <- readIORef activeArr+ elems <- fmap V.toList$ V.freeze vc+ elems' <- mapM safePrint elems+ let sz = MV.length vc+ return$ " {DbgInspectCLD: top "++show tp++", bot "++show bt++", size "++show sz++"\n" +++ -- show elems ++ "\n"+++ " [ "++(concat $ intersperse " " elems')++" ]\n"+++ " end_DbgInspectCLD}"+ where+ -- Print any thunk, even if it raises an exception.+ safePrint :: Show a => a -> IO String+ safePrint val = do+ res <- try (evaluate val)+ case res of+ Left (e::SomeException)+ | isInfixOf "uninitialised element" (show e) -> return "<uninit>"+ | otherwise -> return$ "<"++ show e ++">"+ Right val' -> return (show val')+ +++--------------------------------------------------------------------------------+-- Debugging mode.+-- define DEBUGCL+--define FAKECAS++{-# INLINE rd #-}+{-# INLINE wr #-}+{-# INLINE nu #-}+{-# INLINE cpy #-}+{-# INLINE slc #-}+#ifndef DEBUGCL+dbg = False+nu = MV.unsafeNew+rd = MV.unsafeRead+wr = MV.unsafeWrite+slc = MV.unsafeSlice+cpy = MV.unsafeCopy+#else+#warning "Activating DEBUGCL!"+dbg = True+nu = MV.new +rd = MV.read+slc = MV.slice+cpy = MV.copy+wr = MV.write+-- Temp, debugging: Our own bounds checking, better error:+-- wr v i x = +-- if i >= MV.length v+-- then error (printf "ERROR: Out of bounds of top of vector index %d, vec length %d\n" i (MV.length v))+-- else MV.write v i x++-- [2013.06.25] Note Issue5 is not affected by this:+{-# NOINLINE pushL #-}+{-# NOINLINE tryPopL #-}+{-# NOINLINE tryPopR #-}+#endif+++#ifdef DEBUGCL+-- This simply localizes exceptions better:+tryit msg action = Control.Exception.catch action + (\e -> do putStrLn$ "ERROR inside "++msg++" "++ show e + throw (e::SomeException))+#else+{-# INLINE tryit #-}+tryit msg action = action+#endif++++--------------------------------------------------------------------------------+-- Circular array routines:+++-- TODO: make a "grow" that uses memcpy.+growCirc :: Int -> Int -> MV.IOVector a -> IO (MV.IOVector a)+growCirc strt end oldarr = do + -- let len = MV.length oldarr+ -- strtmod = strt`mod` len + -- endmod = end `mod` len+ -- newarr <- nu (len + len)+ -- if endmod < strtmod then do+ -- let elems1 = len - strtmod+ -- elems2 = endmod+ -- BS.putStrLn$ BS.pack$ printf "Copying segmented ... %d and %d" elems1 elems2++ -- -- Copy the upper then lower segments:+ -- copyOffset oldarr newarr strtmod 0 elems1+ -- copyOffset oldarr newarr 0 elems1 elems2+ -- else do+ -- BS.putStrLn$ BS.pack$ printf "Copying one seg into vec of size %d... size %d, strt %d, end %d, strtmod %d endmod %d" (MV.length newarr) (end - strt) strt end strtmod endmod+ -- -- Copy a single segment:+ -- copyOffset oldarr newarr strtmod 0 (end - strt)+ -- return newarr+ ----------------------------------------+ -- Easier version first:+ ---------------------------------------- + let len = MV.length oldarr+ elems = end - strt+ -- putStrLn$ "Grow to size "++show (len+len)++", copying over "++show elems+ newarr <- if dbg then+ nu (len + len)+ else -- Better errors:+ V.thaw $ V.generate (len+len) (\i -> error (" uninitialized element at position " ++ show i+ ++" had only initialized "++show elems++" elems: "+ ++show(strt`mod`(len+len),end`mod`(len+len))))+ -- Strictly matches what's in the paper:+ for_ strt end $ \ind -> do + x <- getCirc oldarr ind + evaluate x+ putCirc newarr ind x+ return newarr+{-# INLINE growCirc #-}++getCirc :: MV.IOVector a -> Int -> IO a+getCirc arr ind = rd arr (ind `mod` MV.length arr)+{-# INLINE getCirc #-}++putCirc :: MV.IOVector a -> Int -> a -> IO ()+putCirc arr ind x = wr arr (ind `mod` MV.length arr) x+{-# INLINE putCirc #-}++-- Use a potentially-optimized block-copy:+copyOffset :: MV.IOVector t -> MV.IOVector t -> Int -> Int -> Int -> IO ()+copyOffset from to iFrom iTo len =+ cpy (slc iTo len to)+ (slc iFrom len from)+{-# INLINE copyOffset #-}+++--------------------------------------------------------------------------------+-- Queue Operations+--------------------------------------------------------------------------------++newQ :: IO (ChaseLevDeque elt)+newQ = do+ -- Arbitrary Knob: We start as size 32 and double from there:+ v <- MV.new 32 + r1 <- newCounter+ r2 <- newCounter+ r3 <- newIORef v+ return$ CLD r1 r2 r3++nullQ :: ChaseLevDeque elt -> IO Bool+nullQ CLD{top,bottom} = do+ b <- readCounter bottom+ t <- readCounter top +-- return (b == t)+ let size = b - t + return (size <= 0)++-- | For a work-stealing queue `pushL` is the ``local'' push. Thus+-- only a single thread should perform this operation.+pushL :: ChaseLevDeque a -> a -> IO ()+pushL CLD{top,bottom,activeArr} obj = tryit "pushL" $ do+ b <- readCounter bottom+ t <- readCounter top+ arr <- readIORef activeArr+ let len = MV.length arr + size = b - t++-- when (dbg && size < 0) $ error$ "pushL: INVARIANT BREAKAGE - bottom, top: "++ show (b,t)++ arr' <- if (size >= len - 1) then do + arr' <- growCirc t b arr -- Double in size, don't change b/t.+ -- Only a single thread will do this!:+ writeIORef activeArr arr'+ return arr'+ else return arr++ putCirc arr' b obj+ {-+ KG: we need to put write barrier here since otherwise we might+ end with elem not added to q->elements, but q->bottom already+ modified (write reordering) and with stealWSDeque_ failing+ later when invoked from another thread since it thinks elem is+ there (in case there is just added element in the queue). This+ issue concretely hit me on ARMv7 multi-core CPUs+ -}+ writeBarrier+ writeCounter bottom (b+1)+ return ()++-- | This is the steal operation. Multiple threads may concurrently+-- attempt steals from the same thread.+tryPopR :: ChaseLevDeque elt -> IO (Maybe elt)+tryPopR CLD{top,bottom,activeArr} = tryit "tryPopR" $ do+ -- NB. these loads must be ordered, otherwise there is a race+ -- between steal and pop. + tt <- readCounterForCAS top+ loadLoadBarrier+ b <- readCounter bottom+ arr <- readIORef activeArr+ -- when (dbg && b < t) $ error$ "tryPopR: INVARIANT BREAKAGE - bottom < top: "++ show (b,t)++ let t = peekCTicket tt+ size = b - t+ if size <= 0 then + return Nothing+ else do + obj <- getCirc arr t+ (b,_) <- casCounter top tt (t+1)+ if b then + return (Just obj)+ else + return Nothing -- Someone beat us, abort++tryPopL :: ChaseLevDeque elt -> IO (Maybe elt)+tryPopL CLD{top,bottom,activeArr} = tryit "tryPopL" $ do+ b <- readCounter bottom+ arr <- readIORef activeArr+ b <- evaluate (b-1)+ writeCounter bottom b++ -- very important that the following read of q->top does not occur+ -- before the earlier write to q->bottom.+ storeLoadBarrier+ + tt <- readCounterForCAS top+-- when (dbg && b < t) $ error$ "tryPopL: INVARIANT BREAKAGE - bottom < top: "++ show (b,t)++ let t = peekCTicket tt+ size = b - t + if size < 0 then do+ writeCounter bottom t + return Nothing+ else do+ obj <- getCirc arr b+ if size > 0 then do+ return (Just obj)+ else do+ (b,ol) <- casCounter top tt (t+1)+ writeCounter bottom (t+1)+ if b then return$ Just obj+ else return$ Nothing ++------------------------------------------------------------++-- 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)
+ Data/Concurrent/Deque/ChaseLev/DequeInstance.hs view
@@ -0,0 +1,24 @@+{-# LANGUAGE TypeFamilies #-}++module Data.Concurrent.Deque.ChaseLev.DequeInstance () where++import Data.Array.IO+import Data.Concurrent.Deque.Class+import qualified Data.Concurrent.Deque.ChaseLev as R+-- import qualified Data.Concurrent.Deque.ReactorDeque as R++-- | Populate a slice of the configuration-space for `Deque`:+--+-- Work stealing queues are only threadsafe on one end (pop-only) and+-- double (push/pop) functionality on the other:++-- type instance Deque NT T D S Grow Safe elt = R.ChaseLevDeque elt -- Minimal slice++type instance Deque NT t dbl S grow safe elt = R.ChaseLevDeque elt -- Maximal slice++-- [2011.11.09] Presently having problems with this error when I try+-- to use these Deques:+--+-- Couldn't match type `Deque+-- Nonthreadsafe Threadsafe DoubleEnd SingleEnd Grow Safe (Par ())'+-- with `R.Deque IOArray (Par ())'
+ LICENSE view
@@ -0,0 +1,34 @@+Unless otherwise noted in individual files, the below+copyright/LICENSE applies to the source files in this repository.+--------------------------------------------------------------------------------++Copyright (c)2011, Ryan R. Newton++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++ * Redistributions of source code must retain the above copyright+ notice, this list of conditions and the following disclaimer.++ * Redistributions in binary form must reproduce the above+ copyright notice, this list of conditions and the following+ disclaimer in the documentation and/or other materials provided+ with the distribution.++ * Neither the name of Ryan R. Newton nor the names of other+ contributors may be used to endorse or promote products derived+ from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ Test.hs view
@@ -0,0 +1,62 @@+{-# LANGUAGE BangPatterns, NamedFieldPuns #-}+{-# OPTIONS_GHC -fwarn-incomplete-patterns #-}+{- Example build:+ ghc --make Test.hs -o Test.exe -rtsopts -fforce-recomp+-}+module Main where++import System.Environment (getEnvironment)+import Test.HUnit as HU++import Data.Concurrent.Deque.Tests +import Data.Concurrent.Deque.Class+import Data.Concurrent.Deque.Debugger (DebugDeque)+import qualified Data.Concurrent.Deque.ChaseLev as CL++import RegressionTests.Issue5 (standalone_pushPop)+import qualified RegressionTests.Issue5B ++main :: IO ()+main = stdTestHarness $ do + theEnv <- getEnvironment+ let wrapper = case lookup "NOWRAPPER" theEnv of+ Just _ -> False+ Nothing -> True+ let plain = case lookup "ONLYWRAPPER" theEnv of+ Just _ -> False+ Nothing -> True + let all_tests :: HU.Test+ all_tests = TestList $ + [ TestLabel "simplest_pushPop" $ TestCase simplest_pushPop+ , TestLabel "standalone_pushPop" $ TestCase standalone_pushPop+ , TestLabel "standalone_pushPop2" $ TestCase RegressionTests.Issue5B.standalone_pushPop ]+ -- This is very ugly and should be unnecessary:+ ++ if plain then+ [ TestLabel "ChaseLev" $ tests_wsqueue (newQ :: IO (CL.ChaseLevDeque a)) ]+ else [] + ++ if wrapper then+ [ TestLabel "ChaseLev(DbgWrapper)" $ tests_wsqueue (newQ :: IO (DebugDeque CL.ChaseLevDeque a)) ]+ else []+ + return all_tests++--------------------------------------------------------------------------------+-- Individual unit and regression tests:+-------------------------------------------------------------------------------++-- <Small Reproducer for recent debug wrapper problem>+-- This fails even without profiling on.+simplest_pushPop :: IO ()+simplest_pushPop =+ triv =<< (newQ :: IO (DebugDeque CL.ChaseLevDeque a)) + where + -- This is what's failing with the debug wrapper, WHY?+ triv :: PopL d => d [Char] -> IO ()+ triv q = do+ pushL q "hi" + x <- tryPopL q+ let y = case x of+ Just z -> z+ Nothing -> error "Even a single push/pop in isolation did not work!"+ assertEqual "test_ws_triv1" y "hi"+
+ chaselev-deque.cabal view
@@ -0,0 +1,57 @@+Name: chaselev-deque+Version: 0.1.2+License: BSD3+License-file: LICENSE+Author: Ryan R. Newton, Edward Kmett +Maintainer: rrnewton@gmail.com+Category: Data, Concurrent+Build-type: Simple+Cabal-version: >=1.8++-- Version history:+-- 0.1.1 -- bump for fixing bugs! First release candidate.++Homepage: https://github.com/rrnewton/haskell-lockfree-queue/wiki++Synopsis: Chase & Lev work-stealing lock-free double-ended queues (deques).+++Flag debug+ Description: Enable the extra internal checks.+ Default: False++Library+ exposed-modules: Data.Concurrent.Deque.ChaseLev.DequeInstance,+ Data.Concurrent.Deque.ChaseLev+-- Data.Concurrent.Deque.ChaseLev2+-- Disabling this [2012.03.08]. It got terrible performance anyway:+-- Data.Concurrent.Deque.ReactorDeque+ build-depends: base >= 4.4.0.0 && < 5, array, transformers, bits-atomic,+ abstract-deque, vector,+ atomic-primops+-- IORefCAS >= 0.2+ build-depends: ghc-prim+ ghc-options: -O2+ if flag(debug)+ cpp-options: -DDEBUGCL++Source-Repository head+ Type: git+ Location: git://github.com/rrnewton/haskell-lockfree-queue.git+++Test-Suite test-chaselev-deque+ type: exitcode-stdio-1.0+ main-is: Test.hs+ build-depends: base >= 4.4.0.0 && < 5, abstract-deque, + HUnit, test-framework, test-framework-hunit,+ atomic-primops+-- IORefCAS >= 0.2+ build-depends: containers+ ghc-options: -O2 -threaded -rtsopts ++ -- ghc-options: -O2 -threaded -rtsopts + -- -- Debugging generated code:+ -- 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+