lockfree-queue 0.2 → 0.2.0.1
raw patch · 3 files changed
+153/−145 lines, 3 filesdep +atomic-primops
Dependencies added: atomic-primops
Files
- Data/Concurrent/Queue/MichaelScott.hs +131/−141
- Test.hs +1/−1
- lockfree-queue.cabal +21/−3
Data/Concurrent/Queue/MichaelScott.hs view
@@ -1,5 +1,4 @@-{-# LANGUAGE BangPatterns, CPP #-}-{-# LANGUAGE MagicHash, UnboxedTuples #-}+{-# LANGUAGE BangPatterns, CPP, MagicHash, UnboxedTuples, ScopedTypeVariables #-} -- TypeFamilies, FlexibleInstances -- | Michael and Scott lock-free, single-ended queues.@@ -20,121 +19,112 @@ ) where -import Data.IORef (readIORef, IORef, newIORef)+import Data.IORef (readIORef, newIORef) import System.IO (stderr) import Data.ByteString.Char8 (hPutStrLn, pack) -import GHC.Prim -- (MutVar#, RealWorld, sameMutVar#, newMutVar#, casMutVar#, readMutVar#)-import GHC.IO (IO(IO))+-- import GHC.Types (Word(W#))+import GHC.Prim (sameMutVar#)+import GHC.IORef(IORef(IORef))+import GHC.STRef(STRef(STRef)) import qualified Data.Concurrent.Deque.Class as C--- NOTE: you can switch which CAS implementation is used here:----------------------------------------------------------------import Data.CAS (casIORef, ptrEq)--- import Data.CAS.Internal.Fake (casIORef, ptrEq)--- #warning "Using fake CAS"--- import Data.CAS.Internal.Native (casIORef, ptrEq)--- #warning "Using NATIVE CAS"---------------------------------------------------------------+import Data.Atomics (readForCAS, casIORef, Ticket, peekTicket) -- Considering using the Queue class definition: -- import Data.MQueue.Class data LinkedQueue a = LQ - { head :: MutVar# RealWorld (Pair a)- , tail :: MutVar# RealWorld (Pair a)+ { head :: {-# UNPACK #-} !(IORef (Pair a))+ , tail :: {-# UNPACK #-} !(IORef (Pair a)) } -data Pair a = Null | Cons a (MutVar# RealWorld (Pair a))+data Pair a = Null | Cons a {-# UNPACK #-}!(IORef (Pair a)) --- Only checks that the node type is the same and in the case of a Cons Pair checks that--- the IORefs are pointer-equal. This suffices to check equality since IORefs are never used in different --- Pair values.+{-# INLINE pairEq #-}+-- | This only checks that the node type is the same and in the case of a Cons Pair+-- checks that the underlying MutVar#s are pointer-equal. This suffices to check+-- equality since each IORef is never used in multiple Pair values. pairEq :: Pair a -> Pair a -> Bool pairEq Null Null = True-pairEq (Cons _ r) (Cons _ r') = sameMutVar# r r'+pairEq (Cons _ (IORef (STRef mv1)))+ (Cons _ (IORef (STRef mv2))) = sameMutVar# mv1 mv2 pairEq _ _ = False - -- | Push a new element onto the queue. Because the queue can grow, -- this always succeeds.-pushL :: LinkedQueue a -> a -> IO ()-pushL q@(LQ headPtr tailPtr) val = IO $ \ st1 ->- case newMutVar# Null st1 of- (# st2, mv #) ->- let newp = Cons val mv in -- Create the new cell that stores val.- case loop st2 newp of- (# st3, tail #) -> - -- After the loop, enqueue is done. Try to swing the tail.- -- If we fail, that is ok. Whoever came in after us deserves it. - case casMutVar# tailPtr tail newp st2 of- (# st3, flag, res #) -> (# st3, () #)- where- loop s1 newp = - case readMutVar# tailPtr s1 of -- [Re]read the tailptr from the queue structure.- (# s2, tail #) ->- case tail of- -- The head and tail pointers should never themselves be NULL:- Null -> error "push: LinkedQueue invariants broken. Internal error."- Cons _ nextMV ->- case readMutVar# nextMV s2 of- (# s3, next #) ->- {-- -- Optimization: The algorithm can reread tailPtr here to make sure it is still good:- #ifdef RECHECK_ASSUMPTIONS- -- There's a possibility for an infinite loop here with StableName based ptrEq.- -- (And at one point I observed such an infinite loop.)- -- But with one based on reallyUnsafePtrEquality# we should be ok.- tail' <- readIORef tailPtr -- ANDREAS: used atomicModifyIORef here- if not (pairEq tail tail') then loop newp - else case next of - #else- case next of - #endif- -}- case next of- -- Here tail points (or pointed!) to the last node. Try to link our new node.- Null -> case casMutVar# nextMV next newp s3 of- (# s4, flag, newtail #) ->- if flag ==# 0#- then (# s4, tail #)- else loop s4 newp - Cons _ _ -> - -- Someone has beat us by extending the tail. Here we- -- might have to do some community service by updating the tail ptr.- case casMutVar# tailPtr tail next s3 of- (# s4, _, _ #) -> loop s4 newp ---- tryPopR :: LinkedQueue a -> IO (Maybe a)--- tryPopR = error "tryPopR Unimplemented"+pushL :: forall a . LinkedQueue a -> a -> IO ()+pushL q@(LQ headPtr tailPtr) val = do+ r <- newIORef Null+ let newp = Cons val r -- Create the new cell that stores val.+ -- Enqueue loop: repeatedly read the tail pointer and attempt to extend the last pair.+ loop :: IO ()+ loop = do + tailTicket <- readForCAS tailPtr -- [Re]read the tailptr from the queue structure.+ case peekTicket tailTicket of+ -- The head and tail pointers should never themselves be NULL:+ Null -> error "push: LinkedQueue invariants broken. Internal error."+ Cons _ nextPtr -> do+ nextTicket <- readForCAS nextPtr --- -- Andreas's checked this invariant in several places--- -- Check for: head /= tail, and head->next == NULL--- checkInvariant :: String -> LinkedQueue a -> IO ()--- checkInvariant s (LQ headPtr tailPtr) = --- do head <- readIORef headPtr--- tail <- readIORef tailPtr--- if (not (pairEq head tail))--- then case head of --- Null -> error (s ++ " checkInvariant: LinkedQueue invariants broken. Internal error.")--- Cons _ next -> do--- next' <- readIORef next--- case next' of --- Null -> error (s ++ " checkInvariant: next' should not be null")--- _ -> return ()--- else return ()+ -- The algorithm can reread tailPtr here to make sure it is still good:+ -- [UPDATE: This is actually a necessary part of the algorithm's "hand-over-hand"+ -- locking, NOT an optimization.]+#ifdef RECHECK_ASSUMPTIONS+ -- There's a possibility for an infinite loop here with StableName based ptrEq.+ -- (And at one point I observed such an infinite loop.)+ -- But with one based on reallyUnsafePtrEquality# we should be ok.+ (tailTicket', tail') <- readForCAS tailPtr -- ANDREAS: used atomicModifyIORef here+ if not (pairEq tail tail') then loop+ else case next of +#else+ case peekTicket nextTicket of +#endif+ -- Here tail points (or pointed!) to the last node. Try to link our new node.+ Null -> do (b,newtick) <- casIORef nextPtr nextTicket newp+ case b of + True -> do + --------------------Exit Loop------------------+ -- After the loop, enqueue is done. Try to swing the tail.+ -- If we fail, that is ok. Whoever came in after us deserves it.+ _ <- casIORef tailPtr tailTicket newp+ return ()+ -----------------------------------------------+ False -> loop + nxt@(Cons _ _) -> do + -- Someone has beat us by extending the tail. Here we+ -- might have to do some community service by updating the tail ptr.+ _ <- casIORef tailPtr tailTicket nxt+ loop + loop -- Start the loop. +-- Andreas's checked this invariant in several places+-- Check for: head /= tail, and head->next == NULL+checkInvariant :: String -> LinkedQueue a -> IO ()+checkInvariant s (LQ headPtr tailPtr) = + do head <- readIORef headPtr+ tail <- readIORef tailPtr+ if (not (pairEq head tail))+ then case head of + Null -> error (s ++ " checkInvariant: LinkedQueue invariants broken. Internal error.")+ Cons _ next -> do+ next' <- readIORef next+ case next' of + Null -> error (s ++ " checkInvariant: next' should not be null")+ _ -> return ()+ else return ()+ -- | Attempt to pop an element from the queue if one is available. -- tryPop will return semi-promptly (depending on contention), but -- will return 'Nothing' if the queue is empty.-tryPopR :: LinkedQueue a -> IO (Maybe a)--- FIXME / TODO -- add some kind of backoff. This should probably at least--- yield after a certain number of failures.-tryPopR q@(LQ headPtr tailPtr) = IO $ \st -> loop (0::Int) st+tryPopR :: forall a . LinkedQueue a -> IO (Maybe a)+-- FIXME -- this version+-- TODO -- add some kind of backoff. This should probably at least+-- yield after a certain number of failures.+tryPopR q@(LQ headPtr tailPtr) = loop 0 where- loop !tries st =- {- + loop :: Int -> IO (Maybe a) #ifdef DEBUG -- loop 10 = do hPutStrLn stderr (pack "tryPopR: tried ~10 times!!"); loop 11 -- This one happens a lot on -N32 loop 25 = do hPutStrLn stderr (pack "tryPopR: tried ~25 times!!"); loop 26@@ -142,64 +132,63 @@ loop 100 = do hPutStrLn stderr (pack "tryPopR: tried ~100 times!!"); loop 101 loop 1000 = do hPutStrLn stderr (pack "tryPopR: tried ~1000 times!!"); loop 1001 #endif- -}- case readMutVar# headPtr st of- (# st, head #) ->- case readMutVar# tailPtr st of- (# st, tail #) -> - case head of - Null -> error "tryPopR: LinkedQueue invariants broken. Internal error."- Cons _ next -> - case readMutVar# next st of- (# st, next' #) -> + loop !tries = do + headTicket <- readForCAS headPtr+ tailTicket <- readForCAS tailPtr+ case peekTicket headTicket of + Null -> error "tryPopR: LinkedQueue invariants broken. Internal error."+ head@(Cons _ next) -> do+ nextTicket' <- readForCAS next #ifdef RECHECK_ASSUMPTIONS- -- As with push, double-check our information is up-to-date. (head,tail,next consistent)- head' <- readIORef headPtr -- ANDREAS: used atomicModifyIORef headPtr (\x -> (x,x))- if not (pairEq head head') then loop (tries+1) else do + -- As with push, double-check our information is up-to-date. (head,tail,next consistent)+ head' <- readIORef headPtr -- ANDREAS: used atomicModifyIORef headPtr (\x -> (x,x))+ if not (pairEq head head') then loop (tries+1) else do #else- let head' = head in+ let head' = head+ do #endif - -- Is queue empty or tail falling behind?:- if pairEq head tail then do - case next' of -- Is queue empty?- Null -> (# st, Nothing #) -- Queue is empty, couldn't dequeue- Cons _ _ -> - -- Tail is falling behind. Try to advance it:- case casMutVar# tailPtr tail next' st of- (# st, _, _ #) -> loop (tries+1) st- else -- head /= tail- -- No need to deal with Tail. Read value before CAS.- -- Otherwise, another dequeue might free the next node- case next' of - Null -> error "tryPop: Internal error. Next should not be null if head/=tail."- Cons value _ ->- -- Try to swing Head to the next node:- case casMutVar# headPtr head next' st of- (# st, b, _ #) -> - if b ==# 0#- then (# st, Just value #) -- Dequeue done; exit loop.- else loop (tries+1) st--+ -- Is queue empty or tail falling behind?:+ if pairEq head (peekTicket tailTicket) then do + -- if ptrEq head tail then do + case peekTicket nextTicket' of -- Is queue empty?+ Null -> return Nothing -- Queue is empty, couldn't dequeue+ next'@(Cons _ _) -> do+ -- Tail is falling behind. Try to advance it:+ casIORef tailPtr tailTicket next'+ loop (tries+1)+ + else do -- head /= tail+ -- No need to deal with Tail. Read value before CAS.+ -- Otherwise, another dequeue might free the next node+ case peekTicket nextTicket' of + Null -> error "tryPop: Internal error. Next should not be null if head/=tail."+-- Null -> loop (tries+1)+ next'@(Cons value _) -> do + -- Try to swing Head to the next node+ (b,_) <- casIORef headPtr headTicket next'+ case b of+ -- [2013.04.24] Looking at the STG, I can't see a way to get rid of the allocation on this Just:+ True -> return (Just value) -- Dequeue done; exit loop.+ False -> loop (tries+1) -- ANDREAS: observed this loop being taken >1M times+ -- | Create a new queue. newQ :: IO (LinkedQueue a)-newQ = IO$ \ s1 ->- case newMutVar# Null s1 of- (# s2, mv #) ->- let newp = Cons (error "LinkedQueue: Used uninitialized magic value.") mv in- case newMutVar# newp s2 of- (# s3, hd #) ->- case newMutVar# newp s3 of- (# s4, tl #) -> (# s4, LQ hd tl #)+newQ = do + r <- newIORef Null+ let newp = Cons (error "LinkedQueue: Used uninitialized magic value.") r+ hd <- newIORef newp+ tl <- newIORef newp+ return (LQ hd tl) -- | Is the queue currently empty? Beware that this can be a highly transient state. nullQ :: LinkedQueue a -> IO Bool-nullQ (LQ headPtr tailPtr) = IO $ \ st ->- case readMutVar# headPtr st of- (# st, head #) ->- case readMutVar# tailPtr st of- (# st, tail #) -> (# st, pairEq head tail #)+nullQ (LQ headPtr tailPtr) = do + head <- readIORef headPtr+ tail <- readIORef tailPtr+ return (pairEq head tail) ++ -------------------------------------------------------------------------------- -- Instance(s) of abstract deque interface --------------------------------------------------------------------------------@@ -210,6 +199,7 @@ nullQ = nullQ pushL = pushL tryPopR = tryPopR+ leftThreadSafe _ = True+ rightThreadSafe _ = True ---------------------------------------------------------------------------------
Test.hs view
@@ -8,4 +8,4 @@ import Data.Concurrent.Deque.Tests (test_fifo) import Data.Concurrent.Queue.MichaelScott (newQ) -main = defaultMain$ hUnitTestToTests$ test_fifo newQ +main = defaultMain$ hUnitTestToTests$ test_fifo newQ
lockfree-queue.cabal view
@@ -1,5 +1,5 @@ Name: lockfree-queue-Version: 0.2+Version: 0.2.0.1 License: BSD3 License-file: LICENSE Author: Ryan R. Newton@@ -13,6 +13,7 @@ -- Version History: -- 0.1 -- initial version -- 0.2 -- switched to MutVar +-- 0.2.0.1 -- use ticketed CAS internally; add support for -prof mode Synopsis: Michael and Scott lock-free queues. @@ -34,7 +35,8 @@ exposed-modules: Data.Concurrent.Queue.MichaelScott, Data.Concurrent.Queue.MichaelScott.DequeInstance build-depends: base >= 4.4.0.0 && < 5,- ghc-prim, IORefCAS >= 0.2, abstract-deque, bytestring+ ghc-prim, IORefCAS >= 0.2, abstract-deque, bytestring,+ atomic-primops -- Build failure on GHC 7.2: -- queuelike ghc-options: -O2@@ -48,6 +50,22 @@ type: exitcode-stdio-1.0 main-is: Test.hs build-depends: base >= 4.4.0.0 && < 5, IORefCAS >= 0.2, abstract-deque, bytestring,- HUnit, test-framework, test-framework-hunit+ HUnit, test-framework, test-framework-hunit,+ ghc-prim, atomic-primops+ 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 +++-- Executable benchmark-lockfree-queue+-- main-is: Benchmark.hs+-- build-depends: base >= 4.4.0.0 && < 5, IORefCAS >= 0.2, abstract-deque, bytestring,+-- HUnit, test-framework, test-framework-hunit,+-- ghc-prim, atomic-primops+-- 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