diff --git a/Data/Concurrent/Deque/ChaseLev.hs b/Data/Concurrent/Deque/ChaseLev.hs new file mode 100644 --- /dev/null +++ b/Data/Concurrent/Deque/ChaseLev.hs @@ -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 "" + | 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) diff --git a/Data/Concurrent/Deque/ChaseLev/DequeInstance.hs b/Data/Concurrent/Deque/ChaseLev/DequeInstance.hs new file mode 100644 --- /dev/null +++ b/Data/Concurrent/Deque/ChaseLev/DequeInstance.hs @@ -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 ())' diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/Test.hs b/Test.hs new file mode 100644 --- /dev/null +++ b/Test.hs @@ -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: +------------------------------------------------------------------------------- + +-- +-- 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" + diff --git a/chaselev-deque.cabal b/chaselev-deque.cabal new file mode 100644 --- /dev/null +++ b/chaselev-deque.cabal @@ -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 +