packages feed

flush-queue (empty) → 1.0.0

raw patch · 12 files changed

+819/−0 lines, 12 filesdep +QuickCheckdep +asyncdep +atomic-primopssetup-changed

Dependencies added: QuickCheck, async, atomic-primops, base, containers, deepseq, flush-queue, hspec, old-time, stm

Files

+ ChangeLog.md view
@@ -0,0 +1,5 @@+# Changelog for flush-queue++## 1.0.0++Initial release on Hackage
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright FP Complete (c) 2018-2019++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 Alexey Kuleshevich 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.
+ README.md view
@@ -0,0 +1,4 @@+# flush-queue++Two bounded blocking queues, one for `STM` another for `IO`, which are optimized for taking many+elements at the same time, instead of popping individual ones of the queue.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ bench/Benchmark.hs view
@@ -0,0 +1,269 @@+{-# LANGUAGE LambdaCase #-}+module Main where++import           Control.Concurrent.Async+import           Control.Concurrent.BFQueue+import           Control.Concurrent.MVar+import           Control.Concurrent.STM          (STM, atomically, check,+                                                  orElse)+import           Control.Concurrent.STM.TBFQueue+import           Control.Concurrent.STM.TBQueue+import           Control.DeepSeq+import           Control.Monad+import           Data.Foldable+import           Data.IORef+import           System.CPUTime+import           System.IO                       (BufferMode (LineBuffering),+                                                  hSetBuffering, stdout)+import           System.Time+++writeQueueUsing :: (Int -> IO ()) -> Int -> IO ()+writeQueueUsing f = go+  where go n | n > 0 = f n >> go (n-1)+             | otherwise = return ()++fillFlushQueue :: Int -- ^ Queue bound+               -> Int -- ^ Number of threads filling the queue+               -> (Int -> IO ()) -- ^ Queue writer+               -> IO [Int] -- ^ Queue flusher+               -> IO (Time, Time)+fillFlushQueue bound n write flush = do+  ((), fillTime) <- time $ replicateConcurrently_ n (writeQueueUsing write x)+  ((), flushTime) <- time $ do+    res <- flush+    res `deepseq` return ()+  return (fillTime, flushTime)+  where+    x = bound `div` n+++runBench :: [Char] -> IO (Time, Time) -> IO ()+runBench name runCycle = do+  let cycles = 30 :: Int+  putStrLn $ replicate 80 '-'+  putStrLn $ name ++ " (cycles " ++ show cycles ++ ")"+  (tFill, tFlush) <- unzip <$> mapM (const runCycle) [1..cycles]+  putStrLn "Average Fill:"+  putStrLn $ prettyTime $ avg tFill+  putStrLn "Average Flush:"+  putStrLn $ prettyTime $ avg tFlush+++-- | A rundown of a benchmark:+-- * Fill out the queue (`bound` is the limit) without making it block (1. benchmark)+-- * Writing to the queue is done concurrently by number of `threads`+-- * Flush the queue (2. benchmark)+main :: IO ()+main = do+  hSetBuffering stdout LineBuffering+  let bound = 100000+      threads = 16+      runFlushBFQueueMVar = do+        q <- newBFQueueMVar bound+        fillFlushQueue bound threads (void . writeBFQueueMVar q) (flushBFQueueMVar q)+      runFlushSQueue = do+        q <- newSQueue bound+        fillFlushQueue bound threads (writeSQueue q) (flushSQueue q)+      runFlushTBQueue = do+        q <- newTBQueueIO $ fromIntegral bound+        fillFlushQueue bound threads (atomically . writeTBQueue q) (atomically $ flushTBQueue q)+      runFlushTBFQueue = do+        q <- atomically $ newTBFQueue $ fromIntegral bound+        fillFlushQueue bound threads (atomically . writeTBFQueue q) (atomically $ flushTBFQueue q)+      runFlushBFQueue = do+        q <- newBFQueue $ fromIntegral bound+        fillFlushQueue bound threads (writeBFQueue q) (flushBFQueue q)+  putStrLn "==== Fill and Flush all ===="+  runBench "BFQueueMVar (MVar + no blocking)" runFlushBFQueueMVar+  runBench "SQueue (IORef + MVar for blocking)" runFlushSQueue+  runBench "STM TBQueue" runFlushTBQueue+  runBench "STM TBFQueue" runFlushTBFQueue+  runBench "BFQueue (IORef + MVar)" runFlushBFQueue+  let runTakeTBQueue = do+        q <- newTBQueueIO $ fromIntegral bound+        fillFlushQueue+          bound+          threads+          (atomically . void . tryWriteTBQueue q)+          (atomically $ takeTBQueue (bound `div` 2) q)+      runTakeTBFQueue = do+        q <- atomically $ newTBFQueue $ fromIntegral bound+        fillFlushQueue+          bound+          threads+          (atomically . void . tryWriteTBFQueue q)+          (atomically $ takeTBFQueue (fromIntegral bound `div` 2) q)+      runTakeBFQueue = do+        q <- newBFQueue $ fromIntegral bound+        fillFlushQueue+          bound+          threads+          (void . tryWriteBFQueue q)+          (takeBFQueue (fromIntegral bound `div` 2) q)+  putStrLn "==== Try Fill and Take half ===="+  runBench "STM TBQueue" runTakeTBQueue+  runBench "STM TBFQueue" runTakeTBFQueue+  runBench "BFQueue (IORef + MVar)" runTakeBFQueue+++-----------------------------------+-- Missing STM TBQueue functions --+-----------------------------------++tryWriteTBQueue :: TBQueue a -> a -> STM Bool+tryWriteTBQueue tbQueue x =+  orElse (writeTBQueue tbQueue x >> return True) (isFullTBQueue tbQueue >>= check >> return False)+++takeTBQueue :: (Ord a1, Num a1) => a1 -> TBQueue a2 -> STM [a2]+takeTBQueue i tbQueue+  | i <= 0 = return []+  | otherwise = do+    let tryReadN n acc =+          tryReadTBQueue tbQueue >>= \case+            Just v+              | n < i -> tryReadN (n + 1) (v : acc)+            Just v -> return $ reverse (v : acc)+            _ -> return $ reverse acc+    tryReadN 1 []++---------------------------------+-- Alternative implementations --+---------------------------------++type SQueue' a = IORef (SQueue a)++-- | Simple Queue+data SQueue a = SQueue+  { sqCount    :: !Int+  , sqStack    :: ![a]+  , sqMaxCount :: !Int+  , sqLock     :: !(MVar ())+  }++newSQueue :: Int -> IO (SQueue' a)+newSQueue bound = newEmptyMVar >>= newIORef . SQueue 0 [] bound++writeSQueue :: SQueue' a -> a -> IO ()+writeSQueue queue x = inner+  where+    inner = join $ atomicModifyIORef' queue $ \foo0@(SQueue cnt list bound baton) ->+      if cnt < bound+        then (SQueue (cnt + 1) (x:list) bound baton, pure ())+        else (foo0, readMVar baton >> inner)++flushSQueue :: SQueue' a -> IO [a]+flushSQueue queue = do+  newBaton <- newEmptyMVar+  join $ atomicModifyIORef' queue $ \(SQueue _ list bound oldBaton) ->+    (SQueue 0 [] bound newBaton, reverse list <$ putMVar oldBaton ())++++-- | Bounded Flush queue based on MVar, that does not support blocking+newtype BFQueueMVar a = BFQueueMVar (MVar (BList a))++-- | Simple Queue+data BList a = BList+  { bqCount    :: !Int+  , bqStack    :: ![a]+  , bqMaxCount :: !Int+  }++newBFQueueMVar :: Int -> IO (BFQueueMVar a)+newBFQueueMVar bound = BFQueueMVar <$> newMVar (BList 0 [] bound)++writeBFQueueMVar :: BFQueueMVar a -> a -> IO Bool+writeBFQueueMVar (BFQueueMVar bListMVar) x =+  modifyMVar bListMVar $ \ blist@(BList cnt list bound) ->+      if cnt < bound+        then return (BList (cnt + 1) (x:list) bound, True)+        else return (blist, False)++flushBFQueueMVar :: BFQueueMVar a -> IO [a]+flushBFQueueMVar (BFQueueMVar bListMVar) = do+  modifyMVar bListMVar $ \ (BList _ list bound) ->+    return (BList 0 [] bound, reverse list)+++++--------------------+-- Time functions --+--------------------+-- Temporarely borrowed from:+-- https://github.com/haskell-repa/repa/blob/master/repa-io/Data/Array/Repa/IO/Timing.hs++-- Time -----------------------------------------------------------------------+-- | Abstract representation of process time.+data Time+        = Time+        { cpu_time  :: Integer+        , wall_time :: Integer+        }++zipT :: (Integer -> Integer -> Integer) -> Time -> Time -> Time+zipT f (Time cpu1 wall1) (Time cpu2 wall2)+        = Time (f cpu1 cpu2) (f wall1 wall2)++-- | Subtract second time from the first.+minus :: Time -> Time -> Time+minus = zipT (-)+++-- | Add two times.+plus :: Time -> Time -> Time+plus  = zipT (+)++avg :: [Time] -> Time+avg ts = zipT div (foldl' plus (Time 0 0) ts) (Time len len)+  where len = fromIntegral $ length ts++-- TimeUnit -------------------------------------------------------------------+-- | Conversion+type TimeUnit = Integer -> Integer++microseconds :: TimeUnit+microseconds n = n `div` 1000000++milliseconds :: TimeUnit+milliseconds n = n `div` 1000000000++cpuTime :: TimeUnit -> Time -> Integer+cpuTime f = f . cpu_time++wallTime :: TimeUnit -> Time -> Integer+wallTime f = f . wall_time+++-- | Get the current time.+getTime :: IO Time+getTime =+  do+    cpu          <- getCPUTime+    TOD sec pico <- getClockTime+    return $ Time cpu (pico + sec * 1000000000000)+++-- | Pretty print the times, in milliseconds.+prettyTime :: Time -> String+prettyTime t+        = "elapsedTimeMS   = " ++ (show $ wallTime milliseconds t) +++          "\ncpuTimeMS       = " ++ (show $ cpuTime  milliseconds t)++-- Timing benchmarks ----------------------------------------------------------++-- | Time some IO action.+--   Make sure to deepseq the result before returning it from the action. If you+--   don't do this then there's a good chance that you'll just pass a suspension+--   out of the action, and the computation time will be zero.+time :: IO a -> IO (a, Time)+{-# NOINLINE time #-}+time p = do+           start <- getTime+           x     <- p+           ()    <- x `seq` return ()+           end   <- getTime+           return (x, end `minus` start)+
+ flush-queue.cabal view
@@ -0,0 +1,60 @@+name:           flush-queue+version:        1.0.0+synopsis:       Concurrent bouded blocking queues optimized for flushing. Both IO and STM implementations.+description:    Please see the README on GitHub at <https://github.com/fpco/flush-queue#readme>+homepage:       https://github.com/fpco/flush-queue#readme+bug-reports:    https://github.com/fpco/flush-queue/issues+author:         Alexey Kuleshevich+maintainer:     alexey@fpcomplete.com+copyright:      2018-2019 FP Complete+category:       Concurrency+license:        BSD3+license-file:   LICENSE+build-type:     Simple+cabal-version:  >= 1.10+extra-source-files:+    ChangeLog.md+    README.md++source-repository head+  type: git+  location: https://github.com/fpco/flush-queue++library+  exposed-modules: Control.Concurrent.BFQueue+                   Control.Concurrent.STM.TBFQueue+  other-modules: Control.Concurrent.BQueue+  hs-source-dirs: src+  build-depends: base >=4.8 && <5+               , stm+               , atomic-primops+               , containers+  default-language: Haskell2010++test-suite tests+  type: exitcode-stdio-1.0+  main-is: Spec.hs+  other-modules: Control.Concurrent.BFQueueSpec+               , Control.Concurrent.STM.TBFQueueSpec+  hs-source-dirs: test+  ghc-options:  -Wall -threaded -rtsopts -with-rtsopts=-N+  build-depends: base >=4.9 && <5+               , async >= 2.1.1+               , stm+               , flush-queue+               , QuickCheck+               , hspec+  default-language: Haskell2010++benchmark bench+  type:                exitcode-stdio-1.0+  hs-source-dirs:      bench+  main-is:             Benchmark.hs+  ghc-options:         -Wall -threaded -O2 -rtsopts -with-rtsopts=-N+  build-depends:       base+                     , stm >= 2.4.5+                     , async >= 2.1.1+                     , flush-queue+                     , old-time+                     , deepseq+  default-language:    Haskell2010
+ src/Control/Concurrent/BFQueue.hs view
@@ -0,0 +1,92 @@+{-# LANGUAGE BangPatterns #-}+-- |+-- Module      : Control.Concurrent.BFQueue+-- Copyright   : (c) FP Complete 2018+-- License     : BSD3+-- Maintainer  : Alexey Kuleshevich <alexey@fpcomplete.com>+-- Stability   : experimental+-- Portability : non-portable+--+-- Bouded Flush Queue is a very efficient queue that supports pushing elements concurrently, but has+-- no support for popping elements from the queue. The only way to get elements from the queue is to+-- flush it and get all the elements in FIFO order.+--+module Control.Concurrent.BFQueue+  ( BFQueue+  , newBFQueue+  , writeBFQueue+  , tryWriteBFQueue+  , takeBFQueue+  , flushBFQueue+  , lengthBFQueue+  , isEmptyBFQueue+  ) where++import           Control.Concurrent.BQueue+import           Control.Concurrent.MVar+import           Control.Monad+import           Data.Atomics              (atomicModifyIORefCAS)+import           Data.IORef+import           Numeric.Natural++-- | Bounded Flush Queue. It's a queue that allows pushing elements onto, but popping elements is+-- not an option. Only flushing or non-blocking taking from the queue will make space for new+-- elements and unlbock any concurrent writers..+newtype BFQueue a = BFQueue (IORef (BQueue a, MVar ()))++-- | Create new empty BFQueue+newBFQueue :: Natural -> IO (BFQueue a)+newBFQueue bound = do+  baton <- newEmptyMVar+  bQueueIORef <- newIORef (newBQueue $ fromIntegral bound, baton)+  return $ BFQueue bQueueIORef++-- | /O(1)/ - Push an element onto the queue. Will block if maximum bound has been reached.+writeBFQueue :: BFQueue a -> a -> IO ()+writeBFQueue (BFQueue bQueueIORef) x = inner+  where+    inner = join $ atomicModifyIORefCAS bQueueIORef $ \bbQueue@(bQueue, baton) ->+      case pushBQueue x bQueue of+        Just newQueue -> ((newQueue, baton), pure ())+        Nothing       -> (bbQueue, readMVar baton >> inner)++-- | /O(1)/ - Try to push an element onto the queue without blocking. Will return `True` if element+-- was pushed successfully, and `False` in case when the queue is full.+tryWriteBFQueue :: BFQueue a -> a -> IO Bool+tryWriteBFQueue (BFQueue bQueueIORef) x =+  atomicModifyIORefCAS bQueueIORef $ \bbQueue@(bQueue, baton) ->+    case pushBQueue x bQueue of+      Just newQueue -> ((newQueue, baton), True)+      Nothing       -> (bbQueue, False)++-- | /O(n)/ - Flush the queue, unblock all the possible writers and return all the elements from the+-- queue in FIFO order.+flushBFQueue :: BFQueue a -> IO [a]+flushBFQueue (BFQueue bQueueIORef) = do+  newBaton <- newEmptyMVar+  join $+    atomicModifyIORefCAS bQueueIORef $ \(bQueue, baton) ->+      let !(queue, newQueue) = flushBQueue bQueue+       in ((newQueue, newBaton), queue <$ putMVar baton ())++-- | /O(i)/ - Take @i@ elements from the queue, unblock all the possible writers and return all the+-- elements from the queue in FIFO order.+takeBFQueue :: Natural -> BFQueue a -> IO [a]+takeBFQueue i (BFQueue bQueueIORef)+  | i == 0 = return []+  | otherwise = do+    newBaton <- newEmptyMVar+    join $+      atomicModifyIORefCAS bQueueIORef $ \(bQueue, baton) ->+        let !(queue, newQueue) = takeBQueue (fromIntegral i) bQueue+         in ((newQueue, newBaton), queue <$ putMVar baton ())+++-- | /O(1)/ - Extract number of elements that is currently on the queue+lengthBFQueue :: BFQueue a -> IO Natural+lengthBFQueue (BFQueue bQueueIORef) = fromIntegral . lengthBQueue . fst <$> readIORef bQueueIORef+++-- | /O(1)/ - Check if queue is empty+isEmptyBFQueue :: BFQueue a -> IO Bool+isEmptyBFQueue = fmap (==0) . lengthBFQueue
+ src/Control/Concurrent/BQueue.hs view
@@ -0,0 +1,70 @@+{-# LANGUAGE NamedFieldPuns #-}+module Control.Concurrent.BQueue+  ( BQueue+  , newBQueue+  , pushBQueue+  , popBQueue+  , takeBQueue+  , flushBQueue+  , lengthBQueue+  ) where++-- | FIFO Bounded Queue with O(1) amortized popping.+data BQueue a = BQueue+  { bqRead      :: ![a]+  , bqReadSize  :: {-# UNPACK #-}!Int+  , bqWrite     :: ![a]+  , bqWriteSize :: {-# UNPACK #-}!Int+  , bqMaxSize   :: {-# UNPACK #-}!Int+  }++-- | Create new Bounded Queue+newBQueue :: Int -- ^ Upper bound on the numer of elements the queue can hold.+          -> BQueue a+newBQueue bqMaxSize =+  BQueue {bqRead = [], bqReadSize = 0, bqWrite = [], bqWriteSize = 0, bqMaxSize}++-- | Push an element onto the queue. Returns the new queue with the element placed onto the right+-- side of the source queue, but only if the maximum bound hasn't been reached, otherwise it will+-- return `Nothing`+pushBQueue :: a -> BQueue a -> Maybe (BQueue a)+pushBQueue x bq@BQueue {bqWrite, bqWriteSize}+  | bqReadSize bq + bqWriteSize < bqMaxSize bq =+    Just bq {bqWrite = x : bqWrite, bqWriteSize = bqWriteSize + 1}+  | otherwise = Nothing+++-- | Pop an element from the queue. Returns the leftmost element from the queue together with new+-- queue, lacking that element, `Nothing` if the queue was empty.+popBQueue :: BQueue a -> Maybe (a, BQueue a)+popBQueue bq@BQueue {bqRead, bqReadSize, bqWrite, bqWriteSize} =+  case bqRead of+    (x:xs) -> Just (x, bq {bqRead = xs, bqReadSize = bqReadSize - 1})+    [] ->+      case reverse bqWrite of+        (y:ys) ->+          Just (y, bq {bqRead = ys, bqReadSize = bqWriteSize - 1, bqWrite = [], bqWriteSize = 0})+        [] -> Nothing++-- | /O(n) - Get all the elements from the Bounded Queue.+flushBQueue :: BQueue a -> ([a], BQueue a)+flushBQueue bq = (bqRead bq ++ reverse (bqWrite bq), newBQueue (bqMaxSize bq))++-- | /O(i)/ - Take @i@ elements from the Bounded Queue. This function doesn't fail - it returns empty+-- list on negative @i@ and all elements there is if requested more than available.+takeBQueue :: Int -> BQueue a -> ([a], BQueue a)+takeBQueue i bq@BQueue {bqRead, bqReadSize, bqWrite, bqWriteSize}+  | i < bqReadSize =+    let (taken, leftover) = splitAt i bqRead+     in (taken, bq {bqRead = leftover, bqReadSize = bqReadSize - max 0 i})+  | i < totalSize =+    let (taken, leftover) = splitAt i (bqRead ++ reverse bqWrite)+     in (taken, bq {bqRead = leftover, bqReadSize = totalSize - i, bqWrite = [], bqWriteSize = 0})+  | otherwise = flushBQueue bq+  where+    totalSize = bqReadSize + bqWriteSize+++-- | /O(1)/ - Get the current length of a queue+lengthBQueue :: BQueue a -> Int+lengthBQueue bq = bqReadSize bq + bqWriteSize bq
+ src/Control/Concurrent/STM/TBFQueue.hs view
@@ -0,0 +1,96 @@+-- |+-- Module      : Control.Concurrent.STM.TBFQueue+-- Copyright   : (c) FP Complete 2018+-- License     : BSD3+-- Maintainer  : Alexey Kuleshevich <alexey@fpcomplete.com>+-- Stability   : experimental+-- Portability : non-portable+--+-- Transactional Bouded Flush Queue is a very similar to `Control.Concurrent.BFQueue.BFQueue`, with+-- an exception that it runs in `STM` and is also less efficient, but is still faster than+-- `Control.Concurrent.STM.TBQueue.TBQueue`.+--+module Control.Concurrent.STM.TBFQueue+  ( TBFQueue+  , newTBFQueue+  , newTBFQueueIO+  , writeTBFQueue+  , tryWriteTBFQueue+  , readTBFQueue+  , takeTBFQueue+  , flushTBFQueue+  , lengthTBFQueue+  , isEmptyTBFQueue+  ) where++import           Control.Concurrent.BQueue+import           Control.Concurrent.STM+import           Numeric.Natural+++-- | Bounded Flush Queue. It's a queue that allows pushing elements onto, popping elements from it,+-- but is mostly optimizied for flushing the queue or taking in bulk.+newtype TBFQueue a = TBFQueue (TVar (BQueue a))++-- | Construct a new empty Flush Bounded Queue+newTBFQueue :: Natural -- ^ Maximum number of elements, that this queue can hold.+           -> STM (TBFQueue a)+newTBFQueue bound = TBFQueue <$> newTVar (newBQueue (fromIntegral bound))+++-- | Construct a new empty Flush Bounded Queue inside IO monad.+newTBFQueueIO :: Natural -- ^ Maximum number of elements, that this queue can hold.+              -> IO (TBFQueue a)+newTBFQueueIO bound = TBFQueue <$> newTVarIO (newBQueue (fromIntegral bound))++-- | /O(1)/ - Push an element onto the queue. Will block if maximum bound has been reached.+writeTBFQueue :: TBFQueue a -> a -> STM ()+writeTBFQueue (TBFQueue bQueueTVar) x = do+  bQueue <- readTVar bQueueTVar+  case pushBQueue x bQueue of+    Just newQueue -> writeTVar bQueueTVar newQueue+    Nothing       -> retry++-- | /O(1)/ - Try to push an element onto the queue without blocking. Will return `True` if element+-- was pushed successfully, and `False` in case when the queue is full.+tryWriteTBFQueue :: TBFQueue a -> a -> STM Bool+tryWriteTBFQueue (TBFQueue bQueueTVar) x = do+  bQueue <- readTVar bQueueTVar+  case pushBQueue x bQueue of+    Just newQueue -> writeTVar bQueueTVar newQueue >> return True+    Nothing       -> return False++-- | /Amortized O(1)/ - Pop an element from the queue. Will block if queue is empty.+readTBFQueue :: TBFQueue a -> STM a+readTBFQueue (TBFQueue bQueueTVar) = do+  bQueue <- readTVar bQueueTVar+  case popBQueue bQueue of+    Just (x, newQueue) -> writeTVar bQueueTVar newQueue >> return x+    Nothing            -> retry++-- | /O(n)/ - Flush the queue, unblock all the possible writers and return all the elements from the+-- queue in FIFO order.+flushTBFQueue :: TBFQueue a -> STM [a]+flushTBFQueue (TBFQueue bQueueTVar) = do+  bQueue <- readTVar bQueueTVar+  let (xs, newQueue) = flushBQueue bQueue+  writeTVar bQueueTVar newQueue+  return xs++-- | /O(i)/ - Take @i@ elements from the queue, unblock all the possible writers and return all the+-- elements from the queue in FIFO order.+takeTBFQueue :: Natural -> TBFQueue a -> STM [a]+takeTBFQueue i (TBFQueue bQueueTVar) = do+  bQueue <- readTVar bQueueTVar+  let (xs, newQueue) = takeBQueue (fromIntegral i) bQueue+  writeTVar bQueueTVar newQueue+  return xs++-- | /O(1)/ - Extract number of elements that is currently on the queue+lengthTBFQueue :: TBFQueue a -> STM Natural+lengthTBFQueue (TBFQueue bQueueTVar) = fromIntegral . lengthBQueue <$> readTVar bQueueTVar+++-- | /O(1)/ - Check if queue is empty+isEmptyTBFQueue :: TBFQueue a -> STM Bool+isEmptyTBFQueue = fmap (==0) . lengthTBFQueue
+ test/Control/Concurrent/BFQueueSpec.hs view
@@ -0,0 +1,60 @@+module Control.Concurrent.BFQueueSpec (spec) where++import Control.Concurrent.Async+import Control.Concurrent.BFQueue+import Data.List++import Test.Hspec+import Test.QuickCheck+import Test.QuickCheck.Monadic+++prop_FillFlushNonBlocking :: [[Int]] -> Property+prop_FillFlushNonBlocking lss = monadicIO $ do+  ls <- run $ do+    q <- newBFQueue (fromIntegral (sum (map length lss)))+    mapConcurrently_ (foldMap (writeBFQueue q)) lss+    flushBFQueue q+  return (sort ls === sort (concat lss))++prop_FillAndBlockFlush :: Positive Int -> [Int] -> Int -> Property+prop_FillAndBlockFlush (Positive bound) ls oneExtra =+  bound < length ls ==> monadicIO $ do+    let (fillWith, leftOver) = splitAt bound ls+    run $ do+      q <- newBFQueue $ fromIntegral bound+      isSuccess' <- and <$> mapConcurrently (tryWriteBFQueue q) fillWith+      hasSpace <- or <$> mapConcurrently (tryWriteBFQueue q) leftOver+      len <- lengthBFQueue q+      eLs <- race (writeBFQueue q oneExtra >> flushBFQueue q) (flushBFQueue q)+      return $+        conjoin+          [ counterexample "Queue wasn't fully filled up" isSuccess'+          , counterexample "Left over was placed on the queue" (not hasSpace)+          , fromIntegral len === length fillWith+          , either+              (\o ->+                 o === [oneExtra] .||.+                 counterexample "Placed an element on the full queue concurrently" False)+              (\ls' -> sort ls' === sort fillWith)+              eLs+          ]++prop_FillReadTakeNonBlocking :: NonEmptyList Int -> Property+prop_FillReadTakeNonBlocking (NonEmpty xs) =+  monadicIO $+  run $ do+    let i = fromIntegral $ length xs - 1+    q <- newBFQueue (i + 1)+    mapM_ (writeBFQueue q) xs+    [x'] <- takeBFQueue 1 q+    xs' <- takeBFQueue i q+    isEmpty <- isEmptyBFQueue q+    return (head xs === x' .&&. tail xs === xs' .&&. counterexample "Queue is non-empty" isEmpty)++spec :: Spec+spec =+  describe "Fill+Flush" $ do+    it "FillFlushNonBlocking" $ property prop_FillFlushNonBlocking+    it "FillAndBlockFlush" $ property prop_FillAndBlockFlush+    it "FillReadTakeNonBlocking" $ property prop_FillReadTakeNonBlocking
+ test/Control/Concurrent/STM/TBFQueueSpec.hs view
@@ -0,0 +1,114 @@+module Control.Concurrent.STM.TBFQueueSpec (spec) where++import Control.Concurrent.Async+import Control.Concurrent.STM+import Control.Concurrent.STM.TBFQueue+import Data.List++import Test.Hspec+import Test.QuickCheck+import Test.QuickCheck.Monadic++prop_FillFlushNonBlocking :: [[Int]] -> Property+prop_FillFlushNonBlocking lss =+  monadicIO $ do+    ls <-+      run $ do+        q <- newTBFQueueIO (fromIntegral (sum (map length lss)))+        mapConcurrently_ (foldMap (atomically . writeTBFQueue q)) lss+        atomically $ flushTBFQueue q+    return (sort ls === sort (concat lss))++prop_FillAndBlockFlush :: Positive Int -> [Int] -> Int -> Property+prop_FillAndBlockFlush (Positive bound) ls oneExtra =+  bound < length ls ==> monadicIO $ do+    let (fillWith, leftOver) = splitAt bound ls+    run $ do+      q <- atomically $ newTBFQueue $ fromIntegral bound+      isSuccess' <- and <$> mapConcurrently (atomically . tryWriteTBFQueue q) fillWith+      hasSpace <- or <$> mapConcurrently (atomically . tryWriteTBFQueue q) leftOver+      len <- atomically $ lengthTBFQueue q+      eLs <- atomically $ orElse (Left <$> writeTBFQueue q oneExtra) (Right <$> flushTBFQueue q)+      return $+        conjoin+          [ counterexample "Queue wasn't fully filled up" isSuccess'+          , counterexample "Left over was placed on the queue" (not hasSpace)+          , fromIntegral len === length fillWith+          , either+              (\_ -> counterexample "Placed an element on the full queue" False)+              (\ls' -> sort ls' === sort fillWith)+              eLs+          ]++newFullQueueFromList :: Foldable t => t a -> IO (TBFQueue a)+newFullQueueFromList xs = do+    q <- newTBFQueueIO $ fromIntegral $ length xs+    mapM_ (atomically . writeTBFQueue q) xs+    pure q++prop_FillReadTakeNonBlocking :: NonEmptyList Int -> Property+prop_FillReadTakeNonBlocking (NonEmpty xs) =+  monadicIO $+  run $ do+    let i = fromIntegral (length xs - 1)+    q <- newFullQueueFromList xs+    x' <- atomically $ readTBFQueue q+    xs' <- atomically $ takeTBFQueue i q+    isEmpty <- atomically $ isEmptyTBFQueue q+    return (head xs === x' .&&. tail xs === xs' .&&. counterexample "Queue is non-empty" isEmpty)+++prop_FillReadTakeBlocking :: NonEmptyList Int -> Int -> Property+prop_FillReadTakeBlocking (NonEmpty xs) y =+  monadicIO $+  run $ do+    let i = fromIntegral (length xs - 1)+    q <- newFullQueueFromList xs+    ((), x') <- concurrently (atomically $ writeTBFQueue q y) (atomically $ readTBFQueue q)+    xs' <- atomically $ takeTBFQueue i q+    y' <- atomically $ readTBFQueue q+    return (head xs === x' .&&. tail xs === xs' .&&. y === y')+++prop_FillTakeNonBlocking :: [Int] -> NonNegative Int -> Property+prop_FillTakeNonBlocking xs (NonNegative i) =+  monadicIO $+  run $ do+    let n = length xs+    q <- newTBFQueueIO $ fromIntegral n+    mapM_ (atomically . writeTBFQueue q) xs+    xs1 <- atomically $ takeTBFQueue (fromIntegral i) q+    xs2 <- atomically $ takeTBFQueue (fromIntegral (max 0 (n - i))) q+    return (xs === xs1 ++ xs2)+++prop_PushPopConcurrently1 :: Int -> Positive Int -> Property+prop_PushPopConcurrently1 x (Positive bound) =+  monadicIO $+  run $ do+    q <- newTBFQueueIO $ fromIntegral bound+    (x', ()) <- concurrently (atomically $ readTBFQueue q) (atomically $ writeTBFQueue q x)+    return (x === x')++prop_PushPopConcurrentlyMany :: [Int] -> Positive Int -> Property+prop_PushPopConcurrentlyMany xs (Positive bound) =+  monadicIO $+  run $ do+    q <- newTBFQueueIO $ fromIntegral bound+    (xs', ()) <-+      concurrently+        (mapM (const (atomically (readTBFQueue q))) xs)+        (mapM_ (atomically . writeTBFQueue q) xs)+    return (sort xs === sort xs')+++spec :: Spec+spec =+  describe "Fill+Flush" $ do+    it "FillFlushNonBlocking" $ property prop_FillFlushNonBlocking+    it "FillAndBlockFlush" $ property prop_FillAndBlockFlush+    it "FillReadTakeNonBlocking" $ property prop_FillReadTakeNonBlocking+    it "FillReadTakeBlocking" $ property prop_FillReadTakeBlocking+    it "FillTakeNonBlocking" $ property prop_FillTakeNonBlocking+    it "PushPopConcurrently1" $ property prop_PushPopConcurrently1+    it "PushPopConcurrentlyMany" $ property prop_PushPopConcurrentlyMany
+ test/Spec.hs view
@@ -0,0 +1,17 @@+module Main where++import System.IO (BufferMode(LineBuffering), hSetBuffering, stdout)+import Test.Hspec++import Control.Concurrent.BFQueueSpec as BFQueueSpec+import Control.Concurrent.STM.TBFQueueSpec as TBFQueueSpec+++main :: IO ()+main = do+  hSetBuffering stdout LineBuffering+  hspec $ do+    describe "BFQueue" $ do+      BFQueueSpec.spec+    describe "TBFQueue" $ do+      TBFQueueSpec.spec