stm-ringbuffer (empty) → 0.1.0.0
raw patch · 10 files changed
+922/−0 lines, 10 filesdep +QuickCheckdep +arraydep +async
Dependencies added: QuickCheck, array, async, base, deepseq, dlist, generic-random, hspec, random, stm, stm-ringbuffer, tasty, tasty-bench
Files
- CHANGELOG.md +5/−0
- README.md +48/−0
- benchmark/Main.hs +6/−0
- benchmark/StmBench.hs +77/−0
- src/Control/Concurrent/STM/TRingBuffer.hs +230/−0
- src/Control/Concurrent/STM/TRingBuffer/TBQueue.hs +98/−0
- stm-ringbuffer.cabal +87/−0
- test/Main.hs +10/−0
- test/ReferenceTest.hs +168/−0
- test/TestTRingBuffer.hs +193/−0
+ CHANGELOG.md view
@@ -0,0 +1,5 @@+# Revision history for stm-queue++## 0.1.0.0 -- 2026-04-18++* First version.
+ README.md view
@@ -0,0 +1,48 @@+# STM ringbuffer++## About+This is an implementation of a ring buffer for STM. A wrapper exposing a TBQueue-like interface is provided, offering improved performance compared to the current implementation of TBQueue in the STM package.++This work is similar to https://github.com/haskell/stm/pull/70 . However, this implementation is hopefully correct and supports ringbuffers with size == 0.++## Usage+Mixins can be used to quickly replace the standard TBQueue implementation with the one in this library. Add the following fields to your cabal file:+```+library foo+ build-depends:+ stm+ stm-queue+ mixins:+ stm-queue (Control.Concurrent.STM.TRingBuffer.TBQueue as Control.Concurrent.STM.TBQueue),+ stm hiding (Control.Concurrent.STM.TBQueue)+```++## Development++It is recommended to add the following to your cabal.project.local file:+```+tests: True+benchmarks: True+semaphore: True+```++## Benchmark results+In benchmarks, TRingBuffer.TBQueue demonstrates improved performance compared to STM's TBQueue:+```+All+ concurrent spsc+ TBQueue: OK+ 119 ms ± 6.8 ms+ RB.TBQueue: OK+ 109 ms ± 7.3 ms+ concurrent mpmc+ TBQueue: OK+ 193 ms ± 14 ms+ RB.TBQueue: OK+ 136 ms ± 7.0 ms+ burst+ TBQueue: OK+ 114 ms ± 7.2 ms+ RB.TBQueue: OK+ 108 ms ± 6.8 ms+```
+ benchmark/Main.hs view
@@ -0,0 +1,6 @@+module Main where++import StmBench ( stmBench )++main :: IO ()+main = stmBench
+ benchmark/StmBench.hs view
@@ -0,0 +1,77 @@+-- Adapted from STM benchmarks: https://github.com/haskell/stm/blob/master/bench/ChanBench.hs++{-# LANGUAGE TypeApplications #-}+{-# LANGUAGE AllowAmbiguousTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}++module StmBench (stmBench) where++import Control.Concurrent.STM+ ( atomically, newTBQueueIO, readTBQueue, writeTBQueue, TBQueue )+import Control.Concurrent.STM.TRingBuffer.TBQueue qualified as RB+import Control.Concurrent.Async ( async, wait )+import Control.Monad ( replicateM, replicateM_ )+import Test.Tasty (localOption)+import Test.Tasty.Bench+ ( bench, bgroup, defaultMain, whnfAppIO, TimeMode(WallTime) )+import Data.Foldable (traverse_)+++stmBench :: IO ()+stmBench = defaultMain+ [ localOption WallTime $ bgroup "concurrent spsc"+ [ bench "TBQueue" $ whnfAppIO (concurrentSpsc @TBQueue) n+ , bench "RB.TBQueue" $ whnfAppIO (concurrentSpsc @RB.TBQueue) n+ ]+ , localOption WallTime $ bgroup "concurrent mpmc"+ [ bench "TBQueue" $ whnfAppIO (concurrentMpmc @TBQueue) n+ , bench "RB.TBQueue" $ whnfAppIO (concurrentMpmc @RB.TBQueue) n+ ]+ , bgroup "burst"+ [ bench "TBQueue" $ whnfAppIO (burst @TBQueue 1000) n+ , bench "RB.TBQueue" $ whnfAppIO (burst @RB.TBQueue 1000) n+ ]+ ]+ where+ n = 2000000++class Channel c where+ newc :: IO (c a)+ readc :: c a -> IO a+ writec :: c a -> a -> IO ()++instance Channel TBQueue where+ newc = newTBQueueIO 4096+ readc c = atomically $ readTBQueue c+ writec c x = atomically $ writeTBQueue c x++instance Channel RB.TBQueue where+ newc = RB.newTBQueueIO 4096+ readc c = atomically $ RB.readTBQueue c+ writec c x = atomically $ RB.writeTBQueue c x++-- concurrent writing and reading with single producer, single consumer+concurrentSpsc :: forall c. (Channel c) => Int -> IO ()+concurrentSpsc n = do+ c :: c Int <- newc+ writer <- async $ replicateM_ n $ writec c 1+ reader <- async $ replicateM_ n $ readc c+ wait writer+ wait reader++-- concurrent writing and reading with multiple producers, multiple consumers+concurrentMpmc :: forall c. (Channel c) => Int -> IO ()+concurrentMpmc n = do+ c :: c Int <- newc+ writers <- replicateM 10 $ async $ replicateM_ (n `div` 10) $ writec c 1+ readers <- replicateM 10 $ async $ replicateM_ (n `div` 10) $ readc c+ traverse_ wait writers+ traverse_ wait readers++-- bursts of bulk writes, then bulk reads+burst :: forall c. (Channel c) => Int -> Int -> IO ()+burst k n = do+ c :: c Int <- newc+ replicateM_ k $ do+ replicateM_ (n `div` k) $ writec c 1+ replicateM_ (n `div` k) $ readc c
+ src/Control/Concurrent/STM/TRingBuffer.hs view
@@ -0,0 +1,230 @@+-- |+-- Module : Control.Concurrent.STM.TRingBuffer+-- Copyright : (c) Greg Baimetov 2026+-- License : BSD-style (see the file LICENSE)+--+-- Stability : experimental+-- Portability : non-portable (requires STM)+--+-- 'TRingBuffer' is an STM ring buffer. The implementation allows for+-- simultaneous operations on the front and back of the buffer.++{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE InstanceSigs #-}++module Control.Concurrent.STM.TRingBuffer+ ( TRingBuffer,+ new,+ newIO,+ length,+ isEmpty,+ isFull,+ size,+ pushFront,+ pushBack,+ popFront,+ popBack,+ peekFront,+ peekBack,+ flushFront,+ flushBack,+ )+where++import Control.Concurrent.STM+ ( STM,+ TArray,+ TVar,+ newTVar,+ newTVarIO,+ readTVar,+ retry,+ writeTVar,+ )+import Control.Concurrent.STM.TArray ()+import Data.Array.Base+ ( MArray (newArray, unsafeRead, unsafeWrite),+ )+import Data.Array.MArray ()+import Data.DList as DList (empty, snoc, toList)+import Prelude hiding (length, null)++-- | STM ring buffer.+data TRingBuffer a+ = MkTRingBuffer+ { -- | Maximum number of elements the ring buffer can hold.+ size :: {-# UNPACK #-} !Int,+ -- Underlying TArray+ -- index of frontmost item item+ front :: {-# UNPACK #-} !(TVar Int),+ -- index of rearmost item+ back :: {-# UNPACK #-} !(TVar Int),+ arr :: {-# UNPACK #-} !(TArray Int (Maybe a))+ }++instance Eq (TRingBuffer a) where+ (==) :: TRingBuffer a -> TRingBuffer a -> Bool+ a == b = front a == front b++-- | Initialize a ring buffer in STM. O(size)+new :: Int -> STM (TRingBuffer a)+new size = check size $ do+ arr <- newArray (0, size - 1) Nothing+ front <- newTVar $! incr size 0+ back <- newTVar 0+ pure $! MkTRingBuffer {arr, front, back, size}++-- | Initialize a ring buffer in IO. This may be used with UnsafePerformIO. O(size)+newIO :: Int -> IO (TRingBuffer a)+newIO size = check size $ do+ arr <- newArray (0, size - 1) Nothing+ front <- newTVarIO $! incr size 0+ back <- newTVarIO 0+ pure $! MkTRingBuffer {arr, front, back, size}++check :: (Ord a, Num a) => a -> p -> p+check n a =+ if n >= 0+ then a+ else error "Attempted to initialize RingBuffer with size < 0"++-- | Current length. O(1)+length :: TRingBuffer a -> STM Int+length rb@MkTRingBuffer {front, back, size} = do+ rbIsFull <- isFull rb+ if rbIsFull+ then pure size+ else do+ f <- readTVar front+ b <- readTVar back+ pure $ (b + 1 - f) `mod` size++-- | Is the ring buffer empty?+isEmpty :: TRingBuffer a -> STM Bool+isEmpty MkTRingBuffer {arr, front, size} = testPure size True $ do+ ix <- readTVar front+ res <- unsafeRead arr ix+ pure $ case res of+ Nothing -> True+ Just _ -> False++-- | Is the ring buffer full?+isFull :: TRingBuffer a -> STM Bool+isFull MkTRingBuffer {arr, front, size} = testPure size True $ do+ ix <- decr size <$> readTVar front+ res <- unsafeRead arr ix+ pure $ case res of+ Nothing -> False+ Just _ -> True++-- | Push an element to the front of the buffer. Retry if the buffer is full. O(1)+pushFront :: TRingBuffer a -> a -> STM ()+pushFront MkTRingBuffer {arr, front, size} e = testRetry size $ do+ ix <- decr size <$> readTVar front+ cur <- unsafeRead arr ix+ case cur of+ Just _ -> retry+ Nothing -> do+ unsafeWrite arr ix (Just e)+ writeTVar front ix++-- | Push an element to the back of the buffer. Retry if the buffer is full. O(1)+pushBack :: TRingBuffer a -> a -> STM ()+pushBack MkTRingBuffer {arr, back, size} e = testRetry size $ do+ ix <- incr size <$> readTVar back+ cur <- unsafeRead arr ix+ case cur of+ Just _ -> retry+ Nothing -> do+ unsafeWrite arr ix (Just e)+ writeTVar back ix++-- | Pop an element from the front of the buffer. Retry if the buffer is empty. O(1)+popFront :: TRingBuffer b -> STM b+popFront MkTRingBuffer {arr, front, size} = testRetry size $ do+ ix <- readTVar front+ cur <- unsafeRead arr ix+ case cur of+ Nothing -> retry+ Just e -> do+ unsafeWrite arr ix Nothing+ writeTVar front $! incr size ix+ pure e++-- | Pop an element from the back of the buffer. Retry if the buffer is empty. O(1)+popBack :: TRingBuffer b -> STM b+popBack MkTRingBuffer {arr, back, size} = testRetry size $ do+ ix <- readTVar back+ cur <- unsafeRead arr ix+ case cur of+ Nothing -> retry+ Just e -> do+ unsafeWrite arr ix Nothing+ writeTVar back $! decr size ix+ pure e++-- | Peek at the frontmost element of the buffer. Retry if the buffer is empty. O(1)+peekFront :: TRingBuffer b -> STM b+peekFront MkTRingBuffer {arr, front, size} = testRetry size $ do+ ix <- readTVar front+ cur <- unsafeRead arr ix+ maybe retry pure cur++-- | Peek at the backmost element of the buffer. Retry if the buffer is empty. O(1)+peekBack :: TRingBuffer b -> STM b+peekBack MkTRingBuffer {arr, back, size} = testRetry size $ do+ ix <- readTVar back+ cur <- unsafeRead arr ix+ maybe retry pure cur++-- | Pop all elements from the front of the buffer. This operation may succeed even if new elements are pushed to the back in the meantime. O(length)+flushFront :: TRingBuffer a -> STM [a]+flushFront MkTRingBuffer {arr, front, size} = testPure size [] $ do+ initIx <- readTVar front+ go initIx DList.empty+ where+ go !ix !xs = do+ cur <- unsafeRead arr ix+ case cur of+ Nothing -> do+ writeTVar front ix+ pure (toList xs)+ Just x -> do+ unsafeWrite arr ix Nothing+ go (incr size ix) (snoc xs x)++-- | Pop all elements from the back of the buffer. This operation may succeed even if new elements are pushed to the front in the meantime. O(length)+flushBack :: TRingBuffer a -> STM [a]+flushBack MkTRingBuffer {arr, back, size} = testPure size [] $ do+ initIx <- readTVar back+ go initIx DList.empty+ where+ go !ix !xs = do+ cur <- unsafeRead arr ix+ case cur of+ Nothing -> do+ writeTVar back ix+ pure (toList xs)+ Just x -> do+ unsafeWrite arr ix Nothing+ go (decr size ix) (snoc xs x)++-- Functions for moving pointer in ring buffer+-- For n /= 0, these are equivalent to (x \pm 1) `mod` n+-- For n == 0, these are equivalent to (x \pm 1)++incr :: (Eq a, Num a) => a -> a -> a+incr n x = if x == (n - 1) then 0 else x + 1++decr :: (Eq a, Num a) => a -> a -> a+decr n x = if x == 0 then n - 1 else x - 1++-- Functions for handling the case where the buffer's length is zero.+-- Overhead should be negligible when the buffer length is never actually zero.++testRetry :: (Eq a1, Num a1) => a1 -> STM a2 -> STM a2+testRetry !size m = if size == 0 then retry else m++testPure :: (Eq a1, Num a1, Applicative f) => a1 -> a2 -> f a2 -> f a2+testPure !size default' m = if size == 0 then pure default' else m
+ src/Control/Concurrent/STM/TRingBuffer/TBQueue.hs view
@@ -0,0 +1,98 @@+-- |+-- Module : Control.Concurrent.STM.TRingUnMkTBQueuefer.TBQueue+-- Copyright : (c) Greg Baimetov 2026+-- License : BSD-style (see the file LICENSE)+--+-- Stability : experimental+-- Portability : non-portable (requires STM)+--+-- 'TBQueue' is a newtype wrapper around 'TRingBuffer' with an interface+-- that matches that of 'Control.Concurrent.STM.TBQueue' . As such, it can+-- be used as a drop-in replacement, for example with mixins.++module Control.Concurrent.STM.TRingBuffer.TBQueue+ ( TBQueue,+ newTBQueue,+ newTBQueueIO,+ readTBQueue,+ tryReadTBQueue,+ flushTBQueue,+ peekTBQueue,+ tryPeekTBQueue,+ writeTBQueue,+ unGetTBQueue,+ lengthTBQueue,+ isEmptyTBQueue,+ isFullTBQueue,+ capacityTBQueue,+ )+where++import Control.Concurrent.STM.TRingBuffer as TRB+ ( TRingBuffer,+ flushFront,+ isEmpty,+ isFull,+ length,+ new,+ newIO,+ peekFront,+ popFront,+ pushBack,+ pushFront,+ size,+ )+import Control.Monad.STM (STM, orElse)+import Numeric.Natural (Natural)++newtype TBQueue a = MkTBQueue {unMkTBQueue :: TRingBuffer a}+ deriving (Eq)++-- | WARNING: size must not exceed maxBound :: Int+newTBQueue :: Natural -> STM (TBQueue a)+newTBQueue n+ | n > fromIntegral (maxBound :: Int) = error "TBQueue size cannot exceed maxBound :: Int"+ | otherwise = MkTBQueue <$> new (fromIntegral n)++-- | WARNING: size must not exceed maxBound :: Int+newTBQueueIO :: Natural -> IO (TBQueue a)+newTBQueueIO n+ | n > fromIntegral (maxBound :: Int) = error "TBQueue size cannot exceed maxBound :: Int"+ | otherwise = MkTBQueue <$> newIO (fromIntegral n)++readTBQueue :: TBQueue a -> STM a+readTBQueue = popFront . unMkTBQueue++tryReadTBQueue :: TBQueue a -> STM (Maybe a)+tryReadTBQueue = try . readTBQueue++flushTBQueue :: TBQueue a -> STM [a]+flushTBQueue = flushFront . unMkTBQueue++peekTBQueue :: TBQueue a -> STM a+peekTBQueue = peekFront . unMkTBQueue++tryPeekTBQueue :: TBQueue a -> STM (Maybe a)+tryPeekTBQueue = try . peekTBQueue++writeTBQueue :: TBQueue a -> a -> STM ()+writeTBQueue = pushBack . unMkTBQueue++unGetTBQueue :: TBQueue a -> a -> STM ()+unGetTBQueue = pushFront . unMkTBQueue++lengthTBQueue :: TBQueue a -> STM Natural+lengthTBQueue = fmap fromIntegral . TRB.length . unMkTBQueue++isEmptyTBQueue :: TBQueue a -> STM Bool+isEmptyTBQueue = isEmpty . unMkTBQueue++isFullTBQueue :: TBQueue a -> STM Bool+isFullTBQueue = isFull . unMkTBQueue++capacityTBQueue :: TBQueue a -> Natural+capacityTBQueue = fromIntegral . size . unMkTBQueue++-- helper function+try :: STM a -> STM (Maybe a)+try m = fmap Just m `orElse` pure Nothing
+ stm-ringbuffer.cabal view
@@ -0,0 +1,87 @@+cabal-version: 3.4+name: stm-ringbuffer+category: Concurrency+version: 0.1.0.0+license: BSD-3-Clause+author: Greg Baimetov+maintainer: Greg Baimetov+synopsis:+ Ring buffer implementation in STM+description:+ This package provides a ring buffer in STM and a wrapper which mimics the interface to TBQueue.++copyright: (c) Greg Baimetov 2026+build-type: Simple+extra-doc-files:+ CHANGELOG.md+ README.md++tested-with:+ GHC == 9.6.7,+ GHC == 9.8.4,+ GHC == 9.10.3,+ GHC == 9.12.2,+ GHC == 9.14.1++common warnings+ ghc-options: -Wall++source-repository head+ type:git+ location: github.com/Greg-Bm/stm-ringbuffer.git++library+ hs-source-dirs: src+ default-extensions: ImportQualifiedPost+ exposed-modules:+ Control.Concurrent.STM.TRingBuffer+ Control.Concurrent.STM.TRingBuffer.TBQueue++ build-depends:+ , array >=0.5.8 && <0.6+ , base >=4.18.3 && <4.23+ , dlist >=1.0 && <1.1+ , stm >=2.5.3 && <2.6++ default-language: Haskell2010++ ghc-options: -Wall++test-suite test+ type: exitcode-stdio-1.0+ default-extensions: ImportQualifiedPost+ hs-source-dirs: test+ main-is: Main.hs+ other-modules:+ ReferenceTest+ TestTRingBuffer++ build-depends:+ , async >=2.2.6 && <2.3+ , base+ , generic-random >=1.5.0 && <1.6+ , hspec >=2.11.17 && <2.12+ , QuickCheck >=2.16.0 && <2.17+ , random >=1.3.1 && <1.4+ , stm+ , stm-ringbuffer++ default-language: Haskell2010++benchmark benchmark+ type: exitcode-stdio-1.0+ default-extensions: ImportQualifiedPost+ hs-source-dirs: benchmark+ main-is: Main.hs+ other-modules: StmBench+ build-depends:+ , async >=2.2.6 && <2.3+ , base+ , deepseq >=1.4.8 && <1.6+ , random+ , stm+ , stm-ringbuffer+ , tasty >=1.5.4 && <1.6+ , tasty-bench >=0.5 && <0.6++ default-language: Haskell2010
+ test/Main.hs view
@@ -0,0 +1,10 @@+module Main (main) where++import Test.Hspec ( hspec )+import TestTRingBuffer ( testTRingBuffer )+import ReferenceTest ( referenceTest )++main :: IO ()+main = hspec $ do+ testTRingBuffer+ referenceTest
+ test/ReferenceTest.hs view
@@ -0,0 +1,168 @@+-- Simulation that ensures that TRingBuffer.TBQueue matches the behavior of TBQueue+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE RecordWildCards #-}++module ReferenceTest (referenceTest) where++import Control.Concurrent.STM.TBQueue qualified as A+import Control.Concurrent.STM.TRingBuffer as RB+ ( TRingBuffer (..),+ flushBack,+ flushFront,+ isEmpty,+ isFull,+ length,+ newIO,+ peekBack,+ peekFront,+ popBack,+ popFront,+ pushBack,+ pushFront,+ )+import Control.Concurrent.STM.TRingBuffer qualified as RB+import Control.Concurrent.STM.TRingBuffer.TBQueue qualified as B+import Control.Monad (forM_)+import Control.Monad.STM (STM, atomically, orElse)+import Data.Functor (($>))+import Data.List (singleton)+import GHC.Generics (Generic)+import Generic.Random (genericArbitrary, uniform)+import Numeric.Natural (Natural)+import Test.Hspec (SpecWith, describe, shouldBe)+import Test.Hspec.QuickCheck (prop)+import Test.QuickCheck (Arbitrary (..), genericShrink)++referenceTest :: SpecWith ()+referenceTest = describe "Compare to reference behavior" $ do+ describe "TBQueue" $ do+ forM_ [0, 1, 2, 4, 8, 16] $ \size -> do+ let title = "matches reference behavior for size == " <> show size+ prop title $ testImplsMatch size implReferenceTBQueue implRBTBQueue+ describe "TRingBuffer" $ do+ forM_ [0, 1, 2, 4, 8, 16] $ \size -> do+ let title = "matches reference behavior for size == " <> show size+ prop title $ testImplsMatch size implReferenceTBQueue implTRingBuffer+ forM_ [0, 1, 2, 4, 8, 16] $ \size -> do+ let titleR = "(reverse order) matches reference behavior for size == " <> show size+ prop titleR $ testImplsMatch size implReferenceTBQueue implTRingBufferReverse++testImplsMatch :: Natural -> Impl a -> Impl b -> [Command] -> IO ()+testImplsMatch size a b commands = do+ bufA <- newBuf a size+ capacity a bufA `shouldBe` size+ bufB <- newBuf b size+ capacity b bufB `shouldBe` size+ resA <- traverse (runCommand a bufA) commands+ resB <- traverse (runCommand b bufB) commands+ resA `shouldBe` resB++data Command+ = Read+ | TryRead+ | Flush+ | Peek+ | TryPeek+ | Write !Int+ | UnGet !Int+ | Length+ | IsEmpty+ | IsFull+ deriving (Show, Generic)++instance Arbitrary Command where+ arbitrary = genericArbitrary uniform+ shrink = genericShrink++-- using this instead of a typeclass lets us test a ringbuffer both+-- forwards and backwards+data Impl a = MkImpl+ { newBuf :: Natural -> IO (a Int),+ runCommand :: a Int -> Command -> IO [Int],+ capacity :: a Int -> Natural+ }++implReferenceTBQueue :: Impl A.TBQueue+implReferenceTBQueue =+ let newBuf = A.newTBQueueIO+ runCommand q =+ atomically . \case+ Read -> attempt $ A.readTBQueue q+ TryRead -> attempt' $ A.tryReadTBQueue q+ Flush -> A.flushTBQueue q+ Peek -> attempt $ A.peekTBQueue q+ TryPeek -> attempt' $ A.tryPeekTBQueue q+ Write e -> attempt $ A.writeTBQueue q e $> 0+ UnGet e -> attempt $ A.unGetTBQueue q e $> 0+ Length -> singleton . fromIntegral <$> A.lengthTBQueue q+ IsEmpty -> fromBool <$> A.isEmptyTBQueue q+ IsFull -> fromBool <$> A.isFullTBQueue q+ capacity = A.capacityTBQueue+ in MkImpl {..}++implRBTBQueue :: Impl B.TBQueue+implRBTBQueue =+ let newBuf = B.newTBQueueIO+ runCommand q =+ atomically . \case+ Read -> attempt $ B.readTBQueue q+ TryRead -> attempt' $ B.tryReadTBQueue q+ Flush -> B.flushTBQueue q+ Peek -> attempt $ B.peekTBQueue q+ TryPeek -> attempt' $ B.tryPeekTBQueue q+ Write e -> attempt $ B.writeTBQueue q e $> 0+ UnGet e -> attempt $ B.unGetTBQueue q e $> 0+ Length -> singleton . fromIntegral <$> B.lengthTBQueue q+ IsEmpty -> fromBool <$> B.isEmptyTBQueue q+ IsFull -> fromBool <$> B.isFullTBQueue q+ capacity = B.capacityTBQueue+ in MkImpl {..}++implTRingBuffer :: Impl RB.TRingBuffer+implTRingBuffer =+ let newBuf = RB.newIO . fromIntegral+ runCommand q =+ atomically . \case+ Read -> attempt $ RB.popFront q+ TryRead -> attempt' $ try $ RB.popFront q+ Flush -> RB.flushFront q+ Peek -> attempt $ RB.peekFront q+ TryPeek -> attempt' $ try $ RB.peekFront q+ Write e -> attempt $ RB.pushBack q e $> 0+ UnGet e -> attempt $ RB.pushFront q e $> 0+ Length -> singleton . fromIntegral <$> RB.length q+ IsEmpty -> fromBool <$> RB.isEmpty q+ IsFull -> fromBool <$> RB.isFull q+ capacity = fromIntegral . RB.size+ in MkImpl {..}++implTRingBufferReverse :: Impl RB.TRingBuffer+implTRingBufferReverse =+ let newBuf = RB.newIO . fromIntegral+ runCommand q =+ atomically . \case+ Read -> attempt $ RB.popBack q+ TryRead -> attempt' $ try $ RB.popBack q+ Flush -> RB.flushBack q+ Peek -> attempt $ RB.peekBack q+ TryPeek -> attempt' $ try $ RB.peekBack q+ Write e -> attempt $ RB.pushFront q e $> 0+ UnGet e -> attempt $ RB.pushBack q e $> 0+ Length -> singleton . fromIntegral <$> RB.length q+ IsEmpty -> fromBool <$> RB.isEmpty q+ IsFull -> fromBool <$> RB.isFull q+ capacity = fromIntegral . RB.size+ in MkImpl {..}++fromBool :: (Num a) => Bool -> [a]+fromBool b = [if b then 1 else 0]++attempt :: STM a -> STM [a]+attempt m = fmap singleton m `orElse` pure []++attempt' :: (Functor f) => f (Maybe a) -> f [a]+attempt' m = maybe [] singleton <$> m++try :: STM a -> STM (Maybe a)+try m = fmap Just m `orElse` pure Nothing
+ test/TestTRingBuffer.hs view
@@ -0,0 +1,193 @@+-- Unit tests for TRingBuffer.+{-# OPTIONS_GHC -Wno-type-defaults -Wno-unused-do-bind #-}++module TestTRingBuffer (testTRingBuffer) where++import Control.Concurrent.Async ( async )+import Control.Concurrent.STM ( atomically, orElse )+import Control.Concurrent.STM.TRingBuffer as TRB+ ( TRingBuffer(..),+ new,+ isEmpty,+ length,+ isFull,+ newIO,+ pushBack,+ popFront,+ pushFront,+ peekFront,+ flushFront,+ popBack,+ peekBack,+ flushBack )+import Control.Monad+ ( Monad((>>), (>>=)), replicateM, replicateM_, forM_ )+import Data.Functor ( (<&>) )+import Test.Hspec+ ( SpecWith, describe, it, shouldReturn, shouldBe )+import Prelude hiding (length)++testTRingBuffer :: SpecWith ()+testTRingBuffer = describe "Unit tests for TRingBuffer" $ do+ describe "new" $ do+ let newBuf = atomically $ new 10+ it "should be empty upon init" $+ (newBuf >>= atomically . isEmpty) `shouldReturn` True+ it "should have length zero upon init" $+ (newBuf >>= atomically . length) `shouldReturn` 0+ it "should not be full upon init" $+ (newBuf >>= atomically . isFull) `shouldReturn` False+ it "should have size it is initialized with" $+ (newBuf <&> size) `shouldReturn` 10++ describe "newIO" $ do+ let newBuf = newIO 10+ it "should be empty upon init" $+ (newBuf >>= atomically . isEmpty) `shouldReturn` True+ it "should have length zero upon init" $+ (newBuf >>= atomically . length) `shouldReturn` 0+ it "should not be full upon init" $+ (newBuf >>= atomically . isFull) `shouldReturn` False+ it "should have size it is initialized with" $+ (newBuf <&> size) `shouldReturn` 10++ describe "Eq instance" $ do+ it "returns True for same ring buffer" $ do+ buf <- newIO 10 :: IO (TRingBuffer Int)+ (buf == buf) `shouldBe` True+ it "returns False for different ring buffers" $ do+ buf1 <- newIO 10 :: IO (TRingBuffer Int)+ buf2 <- newIO 10+ (buf1 == buf2) `shouldBe` False++ describe "length" $ it "returns length of array" $ do+ buf <- newIO 3+ atomically (length buf) `shouldReturn` 0+ atomically (replicateM 3 (pushBack buf 0 >> length buf))+ `shouldReturn` [1, 2, 3]++ describe "isEmpty" $ it "returns whether array is empty" $ do+ buf <- newIO 3+ atomically (isEmpty buf) `shouldReturn` True+ atomically (replicateM 3 (pushBack buf 0 >> isEmpty buf))+ `shouldReturn` [False, False, False]++ describe "isFull" $ it "returns whether array is full" $ do+ buf <- newIO 3+ atomically (isFull buf) `shouldReturn` False+ atomically (replicateM 3 (pushBack buf 0 >> isFull buf))+ `shouldReturn` [False, False, True]++ describe "pushBack" $ do+ let newBuf = newIO 10+ let pushN n b = replicateM_ n (pushBack b 0)+ it "increases the length" $ do+ buf <- newBuf+ atomically $ pushN 3 buf+ atomically (length buf) `shouldReturn` 3+ atomically $ pushN 7 buf+ atomically (length buf) `shouldReturn` 10+ it "retries when buffer is full" $ do+ buf <- newBuf+ atomically ((pushN 11 buf >> pure True) `orElse` pure False)+ `shouldReturn` False++ describe "popFront" $ do+ let testSeq = [1 .. 10]+ it "takes elements from pushBack in same order" $ do+ buf <- newIO 1+ async $ forM_ testSeq (atomically . pushBack buf)+ replicateM 10 (atomically (popFront buf)) `shouldReturn` testSeq+ it "takes elements from pushFront in reverse order" $ do+ buf <- newIO 10+ forM_ testSeq (atomically . pushFront buf)+ replicateM 10 (atomically (popFront buf)) `shouldReturn` reverse testSeq+ it "retries when buffer is empty" $ do+ buf <- newIO 10+ atomically ((popFront buf >> pure True) `orElse` pure False)+ `shouldReturn` False++ describe "peekFront" $ do+ let testHead = 1+ testTail = [2 .. 10]+ it "looks at the first element pushed back" $ do+ buf <- newIO 10+ forM_ (testHead : testTail) (atomically . pushBack buf)+ atomically (peekFront buf) `shouldReturn` testHead+ it "retries when buffer is empty" $ do+ buf <- newIO 10+ atomically ((popFront buf >> pure True) `orElse` pure False)+ `shouldReturn` False++ describe "flushFront" $ do+ let testSeq = [1 .. 10]+ it "takes elements from pushBack in same order" $ do+ buf <- newIO 10+ forM_ testSeq (atomically . pushBack buf)+ atomically (flushFront buf) `shouldReturn` testSeq+ it "takes elements from pushFront in reverse order" $ do+ buf <- newIO 10+ forM_ testSeq (atomically . pushFront buf)+ atomically (flushFront buf) `shouldReturn` reverse testSeq+ it "leaves the queue empty" $ do+ buf <- newIO 10+ forM_ testSeq (atomically . pushFront buf)+ atomically (flushFront buf)+ atomically (isEmpty buf) `shouldReturn` True++ describe "pushFront" $ do+ let newBuf = newIO 10+ let pushN n b = replicateM_ n (pushFront b 0)+ it "increases the length" $ do+ buf <- newBuf+ atomically $ pushN 3 buf+ atomically (length buf) `shouldReturn` 3+ atomically $ pushN 7 buf+ atomically (length buf) `shouldReturn` 10+ it "retries when buffer is full" $ do+ buf <- newBuf+ atomically ((pushN 11 buf >> pure True) `orElse` pure False)+ `shouldReturn` False++ describe "popBack" $ do+ let testSeq = [1 .. 10]+ it "takes elements from pushFront in same order" $ do+ buf <- newIO 1+ async $ forM_ testSeq (atomically . pushFront buf)+ replicateM 10 (atomically (popBack buf)) `shouldReturn` testSeq+ it "takes elements from pushBack in reverse order" $ do+ buf <- newIO 10+ forM_ testSeq (atomically . pushBack buf)+ replicateM 10 (atomically (popBack buf)) `shouldReturn` reverse testSeq+ it "retries when buffer is empty" $ do+ buf <- newIO 10+ atomically ((popBack buf >> pure True) `orElse` pure False)+ `shouldReturn` False++ describe "peekBack" $ do+ let testHead = 1+ testTail = [2..10]+ it "looks at the first element pushed front" $ do+ buf <- newIO 10+ forM_ (testHead : testTail) (atomically . pushFront buf)+ atomically (peekBack buf) `shouldReturn` testHead+ it "retries when buffer is empty" $ do+ buf <- newIO 10+ atomically ((popBack buf >> pure True) `orElse` pure False)+ `shouldReturn` False++ describe "flushBack" $ do+ let testSeq = [1 .. 10]+ it "takes elements from pushFront in same order" $ do+ buf <- newIO 10+ forM_ testSeq (atomically . pushFront buf)+ atomically (flushBack buf) `shouldReturn` testSeq+ it "takes elements from pushBack in reverse order" $ do+ buf <- newIO 10+ forM_ testSeq (atomically . pushBack buf)+ atomically (flushBack buf) `shouldReturn` reverse testSeq+ it "leaves the queue empty" $ do+ buf <- newIO 10+ forM_ testSeq (atomically . pushBack buf)+ atomically (flushBack buf)+ atomically (isEmpty buf) `shouldReturn` True