diff --git a/CHANGELOG.rst b/CHANGELOG.rst --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -7,6 +7,38 @@ .. _PVP: https://pvp.haskell.org/ +1.6.1.0 (2018-09-23) +-------------------- + +* Git: :tag:`concurrency-1.6.1.0` +* Hackage: :hackage:`concurrency-1.6.1.0` + +Added +~~~~~ + +* (:issue:`286`) Copy across additions from the :hackage:`stm` package: + + * ``Control.Concurrent.Classy.STM.TQueue.flushTQueue`` + * ``Control.Concurrent.Classy.STM.TBQueue.flushTBQueue`` + * ``Control.Concurrent.Classy.STM.TBQueue.lengthTBQueue`` + * ``Control.Concurrent.Classy.STM.TVar.stateTVar`` + +* (:issue:`287`) The ``Control.Concurrent.Classy.STM.TSem`` module. + +Changed +~~~~~~~ + +* (:issue:`286`) Copy across changes from the :hackage:`stm` package: + + * Make definition of ``readTQueue`` consistent with + ``readTBQueue`` + +Miscellaneous +~~~~~~~~~~~~~ + +* The upper bound on :hackage:`stm` is <2.6. + + 1.6.0.0 - IORefs (2018-07-01) ----------------------------- diff --git a/Control/Concurrent/Classy/STM.hs b/Control/Concurrent/Classy/STM.hs --- a/Control/Concurrent/Classy/STM.hs +++ b/Control/Concurrent/Classy/STM.hs @@ -15,6 +15,7 @@ , module Control.Concurrent.Classy.STM.TQueue , module Control.Concurrent.Classy.STM.TBQueue , module Control.Concurrent.Classy.STM.TArray + , module Control.Concurrent.Classy.STM.TSem ) where import Control.Concurrent.Classy.STM.TArray @@ -22,5 +23,6 @@ import Control.Concurrent.Classy.STM.TChan import Control.Concurrent.Classy.STM.TMVar import Control.Concurrent.Classy.STM.TQueue +import Control.Concurrent.Classy.STM.TSem import Control.Concurrent.Classy.STM.TVar import Control.Monad.STM.Class diff --git a/Control/Concurrent/Classy/STM/TBQueue.hs b/Control/Concurrent/Classy/STM/TBQueue.hs --- a/Control/Concurrent/Classy/STM/TBQueue.hs +++ b/Control/Concurrent/Classy/STM/TBQueue.hs @@ -25,10 +25,12 @@ , newTBQueue , readTBQueue , tryReadTBQueue + , flushTBQueue , peekTBQueue , tryPeekTBQueue , writeTBQueue , unGetTBQueue + , lengthTBQueue , isEmptyTBQueue , isFullTBQueue ) where @@ -44,6 +46,7 @@ (TVar stm [a]) (TVar stm Int) (TVar stm [a]) + {-# UNPACK #-} !Int -- | Build and returns a new instance of 'TBQueue' -- @@ -56,22 +59,22 @@ writeT <- newTVar [] rsize <- newTVar 0 wsize <- newTVar size - pure (TBQueue rsize readT wsize writeT) + pure (TBQueue rsize readT wsize writeT size) -- | Write a value to a 'TBQueue'; retries if the queue is full. -- -- @since 1.0.0.0 writeTBQueue :: MonadSTM stm => TBQueue stm a -> a -> stm () -writeTBQueue (TBQueue rsize _ wsize writeT) a = do +writeTBQueue (TBQueue rsize _ wsize writeT _) a = do w <- readTVar wsize if w /= 0 - then writeTVar wsize (w - 1) + then writeTVar wsize $! w - 1 else do r <- readTVar rsize if r /= 0 then do writeTVar rsize 0 - writeTVar wsize (r - 1) + writeTVar wsize $! r - 1 else retry listend <- readTVar writeT writeTVar writeT (a:listend) @@ -80,10 +83,10 @@ -- -- @since 1.0.0.0 readTBQueue :: MonadSTM stm => TBQueue stm a -> stm a -readTBQueue (TBQueue rsize readT _ writeT) = do +readTBQueue (TBQueue rsize readT _ writeT _) = do xs <- readTVar readT r <- readTVar rsize - writeTVar rsize (r + 1) + writeTVar rsize $! r + 1 case xs of (x:xs') -> do writeTVar readT xs' @@ -105,6 +108,23 @@ tryReadTBQueue :: MonadSTM stm => TBQueue stm a -> stm (Maybe a) tryReadTBQueue c = (Just <$> readTBQueue c) `orElse` pure Nothing +-- | Efficiently read the entire contents of a 'TBQueue' into a list. This +-- function never retries. +-- +-- @since 1.6.1.0 +flushTBQueue :: MonadSTM stm => TBQueue stm a -> stm [a] +flushTBQueue (TBQueue rsize r wsize w size) = do + xs <- readTVar r + ys <- readTVar w + if null xs && null ys + then pure [] + else do + writeTVar r [] + writeTVar w [] + writeTVar rsize 0 + writeTVar wsize size + pure (xs ++ reverse ys) + -- | Get the next value from the @TBQueue@ without removing it, -- retrying if the channel is empty. -- @@ -133,23 +153,32 @@ -- -- @since 1.0.0.0 unGetTBQueue :: MonadSTM stm => TBQueue stm a -> a -> stm () -unGetTBQueue (TBQueue rsize readT wsize _) a = do +unGetTBQueue (TBQueue rsize readT wsize _ _) a = do r <- readTVar rsize if r > 0 - then writeTVar rsize (r - 1) + then writeTVar rsize $! r - 1 else do w <- readTVar wsize if w > 0 - then writeTVar wsize (w - 1) + then writeTVar wsize $! w - 1 else retry xs <- readTVar readT writeTVar readT (a:xs) +-- |Return the length of a 'TBQueue'. +-- +-- @since 1.6.1.0 +lengthTBQueue :: MonadSTM stm => TBQueue stm a -> stm Int +lengthTBQueue (TBQueue rsize _ wsize _ size) = do + r <- readTVar rsize + w <- readTVar wsize + pure $! size - r - w + -- | Returns 'True' if the supplied 'TBQueue' is empty. -- -- @since 1.0.0.0 isEmptyTBQueue :: MonadSTM stm => TBQueue stm a -> stm Bool -isEmptyTBQueue (TBQueue _ readT _ writeT) = do +isEmptyTBQueue (TBQueue _ readT _ writeT _) = do xs <- readTVar readT case xs of (_:_) -> pure False @@ -159,7 +188,7 @@ -- -- @since 1.0.0.0 isFullTBQueue :: MonadSTM stm => TBQueue stm a -> stm Bool -isFullTBQueue (TBQueue rsize _ wsize _) = do +isFullTBQueue (TBQueue rsize _ wsize _ _) = do w <- readTVar wsize if w > 0 then pure False diff --git a/Control/Concurrent/Classy/STM/TQueue.hs b/Control/Concurrent/Classy/STM/TQueue.hs --- a/Control/Concurrent/Classy/STM/TQueue.hs +++ b/Control/Concurrent/Classy/STM/TQueue.hs @@ -29,6 +29,7 @@ , newTQueue , readTQueue , tryReadTQueue + , flushTQueue , peekTQueue , tryPeekTQueue , writeTQueue @@ -36,6 +37,7 @@ , isEmptyTQueue ) where +import Control.Monad (unless) import Control.Monad.STM.Class -- | 'TQueue' is an abstract type representing an unbounded FIFO channel. @@ -75,12 +77,12 @@ ys <- readTVar writeT case ys of [] -> retry - _ -> case reverse ys of - [] -> error "readTQueue" - (z:zs) -> do - writeTVar writeT [] - writeTVar readT zs - pure z + _ -> do + let (z:zs) = reverse ys -- NB. lazy: we want the transaction to be + -- short, otherwise it will conflict + writeTVar writeT [] + writeTVar readT zs + pure z -- | A version of 'readTQueue' which does not retry. Instead it -- returns @Nothing@ if no value is available. @@ -88,6 +90,18 @@ -- @since 1.0.0.0 tryReadTQueue :: MonadSTM stm => TQueue stm a -> stm (Maybe a) tryReadTQueue c = (Just <$> readTQueue c) `orElse` pure Nothing + +-- | Efficiently read the entire contents of a 'TQueue' into a list. This +-- function never retries. +-- +-- @since 1.6.1.0 +flushTQueue :: MonadSTM stm => TQueue stm a -> stm [a] +flushTQueue (TQueue r w) = do + xs <- readTVar r + ys <- readTVar w + unless (null xs) $ writeTVar r [] + unless (null ys) $ writeTVar w [] + pure (xs ++ reverse ys) -- | Get the next value from the @TQueue@ without removing it, -- retrying if the channel is empty. diff --git a/Control/Concurrent/Classy/STM/TSem.hs b/Control/Concurrent/Classy/STM/TSem.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/Classy/STM/TSem.hs @@ -0,0 +1,90 @@ +-- | +-- Module : Control.Concurrent.Classy.STM.TSem +-- Copyright : (c) 2018 Michael Walker +-- License : MIT +-- Maintainer : Michael Walker +-- Stability : stable +-- Portability : portable +-- +-- 'TSem': transactional semaphores. +-- +-- __Deviations:__ There is no @Eq@ instance for @TSem@ type. +module Control.Concurrent.Classy.STM.TSem + ( TSem + , newTSem + , waitTSem + , signalTSem + , signalTSemN + ) where + +import Control.Monad (when) +import Control.Monad.STM.Class +import Numeric.Natural (Natural) + +-- | 'TSem' is a transactional semaphore. It holds a certain number +-- of units, and units may be acquired or released by 'waitTSem' and +-- 'signalTSem' respectively. When the 'TSem' is empty, 'waitTSem' +-- blocks. +-- +-- Note that 'TSem' has no concept of fairness, and there is no +-- guarantee that threads blocked in `waitTSem` will be unblocked in +-- the same order; in fact they will all be unblocked at the same time +-- and will fight over the 'TSem'. Hence 'TSem' is not suitable if +-- you expect there to be a high number of threads contending for the +-- resource. However, like other STM abstractions, 'TSem' is +-- composable. +-- +-- @since 1.6.1.0 +newtype TSem stm = TSem (TVar stm Integer) + +-- | Construct new 'TSem' with an initial counter value. +-- +-- A positive initial counter value denotes availability of +-- units 'waitTSem' can acquire. +-- +-- The initial counter value can be negative which denotes a resource +-- \"debt\" that requires a respective amount of 'signalTSem' +-- operations to counter-balance. +-- +-- @since 1.6.1.0 +newTSem :: MonadSTM stm => Integer -> stm (TSem stm) +newTSem i = fmap TSem (newTVar $! i) + +-- | Wait on 'TSem' (aka __P__ operation). +-- +-- This operation acquires a unit from the semaphore (i.e. decreases +-- the internal counter) and blocks (via 'retry') if no units are +-- available (i.e. if the counter is /not/ positive). +-- +-- @since 2.4.2 +waitTSem :: MonadSTM stm => TSem stm -> stm () +waitTSem (TSem t) = do + i <- readTVar t + when (i <= 0) retry + writeTVar t $! (i-1) + +-- | Signal a 'TSem' (aka __V__ operation). +-- +-- This operation adds\/releases a unit back to the semaphore +-- (i.e. increments the internal counter). +-- +-- @since 1.6.1.0 +signalTSem :: MonadSTM stm => TSem stm -> stm () +signalTSem (TSem t) = do + i <- readTVar t + writeTVar t $! i+1 + +-- | Multi-signal a 'TSem' +-- +-- This operation adds\/releases multiple units back to the semaphore +-- (i.e. increments the internal counter). +-- +-- > signalTSem == signalTSemN 1 +-- +-- @since 1.6.1.0 +signalTSemN :: MonadSTM stm => Natural -> TSem stm -> stm () +signalTSemN 0 _ = pure () +signalTSemN 1 s = signalTSem s +signalTSemN n (TSem t) = do + i <- readTVar t + writeTVar t $! i + toInteger n diff --git a/Control/Concurrent/Classy/STM/TVar.hs b/Control/Concurrent/Classy/STM/TVar.hs --- a/Control/Concurrent/Classy/STM/TVar.hs +++ b/Control/Concurrent/Classy/STM/TVar.hs @@ -21,6 +21,7 @@ , writeTVar , modifyTVar , modifyTVar' + , stateTVar , swapTVar , registerDelay ) where @@ -46,6 +47,17 @@ modifyTVar' ctvar f = do a <- readTVar ctvar writeTVar ctvar $! f a + +-- | Like 'modifyTVar'' but the function is a simple state transition that can +-- return a side value which is passed on as the result of the STM. +-- +-- @since 1.6.1.0 +stateTVar :: MonadSTM stm => TVar stm s -> (s -> (a, s)) -> stm a +stateTVar var f = do + s <- readTVar var + let (a, s') = f s -- since we destructure this, we are strict in f + writeTVar var s' + pure a -- | Swap the contents of a 'TVar', returning the old value. -- diff --git a/concurrency.cabal b/concurrency.cabal --- a/concurrency.cabal +++ b/concurrency.cabal @@ -2,7 +2,7 @@ -- documentation, see http://haskell.org/cabal/users-guide/ name: concurrency -version: 1.6.0.0 +version: 1.6.1.0 synopsis: Typeclasses, functions, and data types for concurrency and STM. description: @@ -32,7 +32,7 @@ source-repository this type: git location: https://github.com/barrucadu/dejafu.git - tag: concurrency-1.6.0.0 + tag: concurrency-1.6.1.0 library exposed-modules: Control.Monad.Conc.Class @@ -53,6 +53,7 @@ , Control.Concurrent.Classy.STM.TQueue , Control.Concurrent.Classy.STM.TBQueue , Control.Concurrent.Classy.STM.TArray + , Control.Concurrent.Classy.STM.TSem -- other-modules: -- other-extensions: @@ -62,7 +63,7 @@ , exceptions >=0.7 && <0.11 , monad-control >=1.0 && <1.1 , mtl >=2.2 && <2.3 - , stm >=2.4 && <2.5 + , stm >=2.4 && <2.6 , transformers >=0.5 && <0.6 -- hs-source-dirs: default-language: Haskell2010