diff --git a/AUTHORS b/AUTHORS --- a/AUTHORS +++ b/AUTHORS @@ -1,9 +1,13 @@ === Haskell stm-chans package AUTHORS/THANKS file === The stm-chans package was written (predominantly) by wren ng thornton -and is released under the terms in the LICENSE file. Some important -improvements to TBChan and TBMChan were contributed by Thomas -DuBuisson, incorporating parts of his bounded-tchan package. These -improvements reduce contention between readers and writers, improving -throughput by 2--3 times when producers and consumers are running -in separate OS threads. +and is released under the terms in the LICENSE file. I would also +like to give thanks to the following contributers: + +Thomas DuBuisson --- For important improvements to TBChan and + TBMChan, incorporating parts of his bounded-tchan package. These + improvements reduce contention between readers and writers, + improving throughput by 2--3 times when producers and consumers + are running in separate OS threads. + +kudah --- For adding TQueue support. diff --git a/LICENSE b/LICENSE --- a/LICENSE +++ b/LICENSE @@ -1,6 +1,6 @@ === stm-chans license === -Copyright (c) 2011, 2012, wren ng thornton. +Copyright (c) 2011--2013, wren ng thornton. ALL RIGHTS RESERVED. Redistribution and use in source and binary forms, with or without diff --git a/README b/README --- a/README +++ b/README @@ -15,6 +15,10 @@ $> runhaskell Setup.hs build $> runhaskell Setup.hs test $> runhaskell Setup.hs haddock --hyperlink-source - $> runhaskell Setup.hs install + $> runhaskell Setup.hs copy + $> runhaskell Setup.hs register + +The test step is optional and currently does nothing. The Haddock +step is also optional. ----------------------------------------------------------- fin. diff --git a/VERSION b/VERSION --- a/VERSION +++ b/VERSION @@ -1,3 +1,6 @@ +2.0.0 (2013-05-12): + - Add TQueue support + 1.3.1 (2012-02-29): - Corrected the CPP macros now that stm-2.3 is released. 1.3.0 (2012-02-25): diff --git a/src/Control/Concurrent/STM/TBChan.hs b/src/Control/Concurrent/STM/TBChan.hs --- a/src/Control/Concurrent/STM/TBChan.hs +++ b/src/Control/Concurrent/STM/TBChan.hs @@ -1,10 +1,14 @@ {-# OPTIONS_GHC -Wall -fwarn-tabs #-} {-# LANGUAGE CPP, DeriveDataTypeable #-} + +#if __GLASGOW_HASKELL__ >= 701 +{-# LANGUAGE Trustworthy #-} +#endif ---------------------------------------------------------------- -- 2011.04.06 -- | -- Module : Control.Concurrent.STM.TBChan --- Copyright : Copyright (c) 2011--2012 wren ng thornton +-- Copyright : Copyright (c) 2011--2013 wren ng thornton -- License : BSD -- Maintainer : wren@community.haskell.org -- Stability : provisional diff --git a/src/Control/Concurrent/STM/TBMChan.hs b/src/Control/Concurrent/STM/TBMChan.hs --- a/src/Control/Concurrent/STM/TBMChan.hs +++ b/src/Control/Concurrent/STM/TBMChan.hs @@ -1,10 +1,14 @@ {-# OPTIONS_GHC -Wall -fwarn-tabs #-} {-# LANGUAGE CPP, DeriveDataTypeable #-} + +#if __GLASGOW_HASKELL__ >= 701 +{-# LANGUAGE Trustworthy #-} +#endif ---------------------------------------------------------------- -- 2011.04.06 -- | -- Module : Control.Concurrent.STM.TBMChan --- Copyright : Copyright (c) 2011--2012 wren ng thornton +-- Copyright : Copyright (c) 2011--2013 wren ng thornton -- License : BSD -- Maintainer : wren@community.haskell.org -- Stability : provisional diff --git a/src/Control/Concurrent/STM/TBMQueue.hs b/src/Control/Concurrent/STM/TBMQueue.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/STM/TBMQueue.hs @@ -0,0 +1,359 @@ +{-# OPTIONS_GHC -Wall -fwarn-tabs #-} +{-# LANGUAGE CPP, DeriveDataTypeable #-} + +#if __GLASGOW_HASKELL__ >= 701 +{-# LANGUAGE Trustworthy #-} +#endif +---------------------------------------------------------------- +-- 2013.05.12 +-- | +-- Module : Control.Concurrent.STM.TBMQueue +-- Copyright : Copyright (c) 2011--2013 wren ng thornton +-- License : BSD +-- Maintainer : wren@community.haskell.org +-- Stability : provisional +-- Portability : non-portable (GHC STM, DeriveDataTypeable) +-- +-- A version of "Control.Concurrent.STM.TQueue" where the queue is +-- bounded in length and closeable. This combines the abilities of +-- "Control.Concurrent.STM.TBQueue" and "Control.Concurrent.STM.TMQueue". +-- This variant incorporates ideas from Thomas M. DuBuisson's +-- @bounded-tqueue@ package in order to reduce contention between +-- readers and writers. +-- +-- /Since: 2.0.0/ +---------------------------------------------------------------- +module Control.Concurrent.STM.TBMQueue + ( + -- * The TBMQueue type + TBMQueue() + -- ** Creating TBMQueues + , newTBMQueue + , newTBMQueueIO + -- ** Reading from TBMQueues + , readTBMQueue + , tryReadTBMQueue + , peekTBMQueue + , tryPeekTBMQueue + -- ** Writing to TBMQueues + , writeTBMQueue + , tryWriteTBMQueue + , unGetTBMQueue + -- ** Closing TBMQueues + , closeTBMQueue + -- ** Predicates + , isClosedTBMQueue + , isEmptyTBMQueue + , isFullTBMQueue + -- ** Other functionality + , estimateFreeSlotsTBMQueue + , freeSlotsTBMQueue + ) where + +import Prelude hiding (reads) +import Data.Typeable (Typeable) +import Control.Applicative ((<$>)) +import Control.Monad.STM (STM, retry) +import Control.Concurrent.STM.TVar.Compat +import Control.Concurrent.STM.TQueue.Compat -- N.B., GHC only + +-- N.B., we need a Custom cabal build-type for this to work. +#ifdef __HADDOCK__ +import Control.Monad.STM (atomically) +import System.IO.Unsafe (unsafePerformIO) +#endif +---------------------------------------------------------------- + +-- | @TBMQueue@ is an abstract type representing a bounded closeable +-- FIFO queue. +data TBMQueue a = TBMQueue !(TVar Bool) !(TVar Int) !(TVar Int) !(TQueue a) + deriving (Typeable) +-- The components are: +-- * Whether the queue has been closed. +-- * How many free slots we /know/ we have available. +-- * How many slots have been freed up by successful reads since +-- the last time the slot count was synchronized by 'isFullTBQueue'. +-- * The underlying TQueue. + + +-- | Build and returns a new instance of @TBMQueue@ with the given +-- capacity. /N.B./, we do not verify the capacity is positive, but +-- if it is non-positive then 'writeTBMQueue' will always retry and +-- 'isFullTBMQueue' will always be true. +newTBMQueue :: Int -> STM (TBMQueue a) +newTBMQueue n = do + closed <- newTVar False + slots <- newTVar n + reads <- newTVar 0 + queue <- newTQueue + return (TBMQueue closed slots reads queue) + + +-- | @IO@ version of 'newTBMQueue'. This is useful for creating +-- top-level @TBMQueue@s using 'unsafePerformIO', because using +-- 'atomically' inside 'unsafePerformIO' isn't possible. +newTBMQueueIO :: Int -> IO (TBMQueue a) +newTBMQueueIO n = do + closed <- newTVarIO False + slots <- newTVarIO n + reads <- newTVarIO 0 + queue <- newTQueueIO + return (TBMQueue closed slots reads queue) + + +-- | Read the next value from the @TBMQueue@, retrying if the queue +-- is empty (and not closed). We return @Nothing@ immediately if +-- the queue is closed and empty. +readTBMQueue :: TBMQueue a -> STM (Maybe a) +readTBMQueue (TBMQueue closed _slots reads queue) = do + b <- readTVar closed + if b + then do + mx <- tryReadTQueue queue + case mx of + Nothing -> return mx + Just _x -> do + modifyTVar' reads (1 +) + return mx + else do + x <- readTQueue queue + modifyTVar' reads (1 +) + return (Just x) +{- +-- The above is slightly optimized over the clearer: +readTBMQueue (TBMQueue closed _slots reads queue) = + b <- readTVar closed + b' <- isEmptyTQueue queue + if b && b' + then return Nothing + else do + x <- readTQueue queue + modifyTVar' reads (1 +) + return (Just x) +-- TODO: compare Core and benchmarks; is the loss of clarity worth it? +-} + + +-- | A version of 'readTBMQueue' which does not retry. Instead it +-- returns @Just Nothing@ if the queue is open but no value is +-- available; it still returns @Nothing@ if the queue is closed +-- and empty. +tryReadTBMQueue :: TBMQueue a -> STM (Maybe (Maybe a)) +tryReadTBMQueue (TBMQueue closed _slots reads queue) = do + b <- readTVar closed + if b + then do + mx <- tryReadTQueue queue + case mx of + Nothing -> return Nothing + Just _x -> do + modifyTVar' reads (1 +) + return (Just mx) + else do + mx <- tryReadTQueue queue + case mx of + Nothing -> return (Just mx) + Just _x -> do + modifyTVar' reads (1 +) + return (Just mx) +{- +-- The above is slightly optimized over the clearer: +tryReadTBMQueue (TBMQueue closed _slots reads queue) = + b <- readTVar closed + b' <- isEmptyTQueue queue + if b && b' + then return Nothing + else do + mx <- tryReadTBMQueue queue + case mx of + Nothing -> return (Just mx) + Just _x -> do + modifyTVar' reads (1 +) + return (Just mx) +-- TODO: compare Core and benchmarks; is the loss of clarity worth it? +-} + + +-- | Get the next value from the @TBMQueue@ without removing it, +-- retrying if the queue is empty. +peekTBMQueue :: TBMQueue a -> STM (Maybe a) +peekTBMQueue (TBMQueue closed _slots _reads queue) = do + b <- readTVar closed + if b + then do + b' <- isEmptyTQueue queue + if b' + then return Nothing + else Just <$> peekTQueue queue + else Just <$> peekTQueue queue +{- +-- The above is lazier reading from @queue@ than the clearer: +peekTBMQueue (TBMQueue closed _slots _reads queue) = do + b <- isEmptyTQueue queue + b' <- readTVar closed + if b && b' + then return Nothing + else Just <$> peekTQueue queue +-- TODO: compare Core and benchmarks; is the loss of clarity worth it? +-} + + +-- | A version of 'peekTBMQueue' which does not retry. Instead it +-- returns @Just Nothing@ if the queue is open but no value is +-- available; it still returns @Nothing@ if the queue is closed +-- and empty. +tryPeekTBMQueue :: TBMQueue a -> STM (Maybe (Maybe a)) +tryPeekTBMQueue (TBMQueue closed _slots _reads queue) = do + b <- readTVar closed + if b + then fmap Just <$> tryPeekTQueue queue + else Just <$> tryPeekTQueue queue +{- +-- The above is lazier reading from @queue@ (and removes an extraneous isEmptyTQueue when using the compatibility layer) than the clearer: +tryPeekTBMQueue (TBMQueue closed _slots _reads queue) = do + b <- isEmptyTQueue queue + b' <- readTVar closed + if b && b' + then return Nothing + else Just <$> tryPeekTQueue queue +-- TODO: compare Core and benchmarks; is the loss of clarity worth it? +-} + + +-- | Write a value to a @TBMQueue@, retrying if the queue is full. +-- If the queue is closed then the value is silently discarded. +-- Use 'isClosedTBMQueue' to determine if the queue is closed +-- before writing, as needed. +writeTBMQueue :: TBMQueue a -> a -> STM () +writeTBMQueue self@(TBMQueue closed slots _reads queue) x = do + b <- readTVar closed + if b + then return () -- Discard silently + else do + n <- estimateFreeSlotsTBMQueue self + if n <= 0 + then retry + else do + writeTVar slots $! n - 1 + writeTQueue queue x + + +-- | A version of 'writeTBMQueue' which does not retry. Returns @Just +-- True@ if the value was successfully written, @Just False@ if it +-- could not be written (but the queue was open), and @Nothing@ +-- if it was discarded (i.e., the queue was closed). +tryWriteTBMQueue :: TBMQueue a -> a -> STM (Maybe Bool) +tryWriteTBMQueue self@(TBMQueue closed slots _reads queue) x = do + b <- readTVar closed + if b + then return Nothing + else do + n <- estimateFreeSlotsTBMQueue self + if n <= 0 + then return (Just False) + else do + writeTVar slots $! n - 1 + writeTQueue queue x + return (Just True) + + +-- | Put a data item back onto a queue, where it will be the next +-- item read. If the queue is closed then the value is silently +-- discarded; you can use 'peekTBMQueue' to circumvent this in certain +-- circumstances. /N.B./, this could allow the queue to temporarily +-- become longer than the specified limit, which is necessary to +-- ensure that the item is indeed the next one read. +unGetTBMQueue :: TBMQueue a -> a -> STM () +unGetTBMQueue (TBMQueue closed slots _reads queue) x = do + b <- readTVar closed + if b + then return () -- Discard silently + else do + modifyTVar' slots (subtract 1) + unGetTQueue queue x + + +-- | Closes the @TBMQueue@, preventing any further writes. +closeTBMQueue :: TBMQueue a -> STM () +closeTBMQueue (TBMQueue closed _slots _reads _queue) = + writeTVar closed True + + +-- | Returns @True@ if the supplied @TBMQueue@ has been closed. +isClosedTBMQueue :: TBMQueue a -> STM Bool +isClosedTBMQueue (TBMQueue closed _slots _reads _queue) = + readTVar closed + +{- +-- | Returns @True@ if the supplied @TBMQueue@ has been closed. +isClosedTBMQueueIO :: TBMQueue a -> IO Bool +isClosedTBMQueueIO (TBMQueue closed _slots _reads _queue) = + readTVarIO closed +-} + + +-- | Returns @True@ if the supplied @TBMQueue@ is empty (i.e., has +-- no elements). /N.B./, a @TBMQueue@ can be both ``empty'' and +-- ``full'' at the same time, if the initial limit was non-positive. +isEmptyTBMQueue :: TBMQueue a -> STM Bool +isEmptyTBMQueue (TBMQueue _closed _slots _reads queue) = + isEmptyTQueue queue + + +-- | Returns @True@ if the supplied @TBMQueue@ is full (i.e., is +-- over its limit). /N.B./, a @TBMQueue@ can be both ``empty'' and +-- ``full'' at the same time, if the initial limit was non-positive. +-- /N.B./, a @TBMQueue@ may still be full after reading, if +-- 'unGetTBMQueue' was used to go over the initial limit. +-- +-- This is equivalent to: @liftM (<= 0) estimateFreeSlotsTBMQueue@ +isFullTBMQueue :: TBMQueue a -> STM Bool +isFullTBMQueue (TBMQueue _closed slots reads _queue) = do + n <- readTVar slots + if n <= 0 + then do + m <- readTVar reads + let n' = n + m + writeTVar slots $! n' + writeTVar reads 0 + return $! n' <= 0 + else return False + + +-- | Estimate the number of free slots. If the result is positive, +-- then it's a minimum bound; if it's non-positive then it's exact. +-- It will only be negative if the initial limit was negative or +-- if 'unGetTBMQueue' was used to go over the initial limit. +-- +-- This function always contends with writers, but only contends +-- with readers when it has to; compare against 'freeSlotsTBMQueue'. +estimateFreeSlotsTBMQueue :: TBMQueue a -> STM Int +estimateFreeSlotsTBMQueue (TBMQueue _closed slots reads _queue) = do + n <- readTVar slots + if n > 0 + then return n + else do + m <- readTVar reads + let n' = n + m + writeTVar slots $! n' + writeTVar reads 0 + return n' + + +-- | Return the exact number of free slots. The result can be +-- negative if the initial limit was negative or if 'unGetTBMQueue' +-- was used to go over the initial limit. +-- +-- This function always contends with both readers and writers; +-- compare against 'estimateFreeSlotsTBMQueue'. +freeSlotsTBMQueue :: TBMQueue a -> STM Int +freeSlotsTBMQueue (TBMQueue _closed slots reads _queue) = do + n <- readTVar slots + m <- readTVar reads + let n' = n + m + writeTVar slots $! n' + writeTVar reads 0 + return n' + +---------------------------------------------------------------- +----------------------------------------------------------- fin. diff --git a/src/Control/Concurrent/STM/TBQueue/Compat.hs b/src/Control/Concurrent/STM/TBQueue/Compat.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/STM/TBQueue/Compat.hs @@ -0,0 +1,201 @@ +{-# OPTIONS_GHC -Wall -fwarn-tabs -fno-warn-name-shadowing #-} +{-# LANGUAGE CPP, DeriveDataTypeable #-} + +#if __GLASGOW_HASKELL__ >= 701 +{-# LANGUAGE Trustworthy #-} +#endif +---------------------------------------------------------------- +-- 2013.05.12 +-- | +-- Module : Control.Concurrent.STM.TBQueue.Compat +-- Copyright : Copyright (c) 2011--2013 wren ng thornton +-- License : BSD +-- Maintainer : wren@community.haskell.org +-- Stability : provisional +-- Portability : non-portable (CPP, STM, DeriveDataTypeable) +-- +-- Compatibility layer for older versions of the @stm@ library. +-- Namely, we copy "Control.Concurrent.STM.TBQueue" module which +-- @stm < 2.4.0@ lacks. This module uses Cabal-style CPP macros in +-- order to use the package versions when available. +-- +-- /Since: 2.0.0/ +---------------------------------------------------------------- + +module Control.Concurrent.STM.TBQueue.Compat + ( + -- * The TBQueue type + TBQueue + -- ** Creating TBQueues + , newTBQueue -- :: Int -> STM (TBQueue a) + , newTBQueueIO -- :: Int -> IO (TBQueue a) + -- ** Reading from TBQueues + , readTBQueue -- :: TBQueue a -> STM a + , tryReadTBQueue -- :: TBQueue a -> STM (Maybe a) + , peekTBQueue -- :: TBQueue a -> STM a + , tryPeekTBQueue -- :: TBQueue a -> STM (Maybe a) + -- ** Writing to TBQueues + , writeTBQueue -- :: TBQueue a -> a -> STM () + , unGetTBQueue -- :: TBQueue a -> a -> STM () + -- ** Predicates + , isEmptyTBQueue -- :: TBQueue a -> STM Bool + ) where + +#if MIN_VERSION_stm(2,4,0) +import Control.Concurrent.STM.TBQueue +#else +import Data.Typeable +import GHC.Conc + +#define _UPK_(x) {-# UNPACK #-} !(x) + +-- | 'TBQueue' is an abstract type representing a bounded FIFO channel. +data TBQueue a = + TBQueue _UPK_(TVar Int) -- CR: read capacity + _UPK_(TVar [a]) -- R: elements waiting to be read + _UPK_(TVar Int) -- CW: write capacity + _UPK_(TVar [a]) -- W: elements written (head is most recent) + deriving Typeable + +instance Eq (TBQueue a) where + TBQueue a _ _ _ == TBQueue b _ _ _ = a == b + + +-- Total channel capacity remaining is CR + CW. Reads only need to +-- access CR, writes usually need to access only CW but sometimes need +-- CR. So in the common case we avoid contention between CR and CW. +-- +-- - when removing an element from R: +-- CR := CR + 1 +-- +-- - when adding an element to W: +-- if CW is non-zero +-- then CW := CW - 1 +-- then if CR is non-zero +-- then CW := CR - 1; CR := 0 +-- else **FULL** + + +-- | Build and returns a new instance of 'TBQueue'. +newTBQueue + :: Int -- ^ maximum number of elements the queue can hold + -> STM (TBQueue a) +newTBQueue size = do + read <- newTVar [] + write <- newTVar [] + rsize <- newTVar 0 + wsize <- newTVar size + return (TBQueue rsize read wsize write) + + +-- | @IO@ version of 'newTBQueue'. This is useful for creating +-- top-level 'TBQueue's using 'System.IO.Unsafe.unsafePerformIO', +-- because using 'atomically' inside 'System.IO.Unsafe.unsafePerformIO' +-- isn't possible. +newTBQueueIO :: Int -> IO (TBQueue a) +newTBQueueIO size = do + read <- newTVarIO [] + write <- newTVarIO [] + rsize <- newTVarIO 0 + wsize <- newTVarIO size + return (TBQueue rsize read wsize write) + + +-- | Write a value to a 'TBQueue'; blocks if the queue is full. +writeTBQueue :: TBQueue a -> a -> STM () +writeTBQueue (TBQueue rsize _read wsize write) a = do + w <- readTVar wsize + if w /= 0 + then writeTVar wsize (w - 1) + else do + r <- readTVar rsize + if r /= 0 + then do + writeTVar rsize 0 + writeTVar wsize (r - 1) + else retry + listend <- readTVar write + writeTVar write (a:listend) + + +-- | Read the next value from the 'TBQueue'. +readTBQueue :: TBQueue a -> STM a +readTBQueue (TBQueue rsize read _wsize write) = do + xs <- readTVar read + r <- readTVar rsize + writeTVar rsize (r + 1) + case xs of + (x:xs') -> do + writeTVar read xs' + return x + [] -> do + ys <- readTVar write + case ys of + [] -> retry + _ -> do + let (z:zs) = reverse ys + -- N.B., lazy: we want the transaction to + -- be short, otherwise it will conflict. + writeTVar write [] + writeTVar read zs + return z + + +-- | A version of 'readTBQueue' which does not retry. Instead it +-- returns @Nothing@ if no value is available. +tryReadTBQueue :: TBQueue a -> STM (Maybe a) +tryReadTBQueue c = fmap Just (readTBQueue c) `orElse` return Nothing + + +-- | Get the next value from the @TBQueue@ without removing it, +-- retrying if the channel is empty. +peekTBQueue :: TBQueue a -> STM a +peekTBQueue c = do + x <- readTBQueue c + unGetTBQueue c x + return x + + +-- | A version of 'peekTBQueue' which does not retry. Instead it +-- returns @Nothing@ if no value is available. +tryPeekTBQueue :: TBQueue a -> STM (Maybe a) +tryPeekTBQueue c = do + m <- tryReadTBQueue c + case m of + Nothing -> return Nothing + Just x -> do + unGetTBQueue c x + return m + + +-- | Put a data item back onto a channel, where it will be the next +-- item read. Blocks if the queue is full. +unGetTBQueue :: TBQueue a -> a -> STM () +unGetTBQueue (TBQueue rsize read wsize _write) a = do + r <- readTVar rsize + if r > 0 + then writeTVar rsize (r - 1) + else do + w <- readTVar wsize + if w > 0 + then writeTVar wsize (w - 1) + else retry + xs <- readTVar read + writeTVar read (a:xs) + + +-- | Returns 'True' if the supplied 'TBQueue' is empty. +isEmptyTBQueue :: TBQueue a -> STM Bool +isEmptyTBQueue (TBQueue _rsize read _wsize write) = do + xs <- readTVar read + case xs of + (_:_) -> return False + [] -> do + ys <- readTVar write + case ys of + [] -> return True + _ -> return False +#endif + +---------------------------------------------------------------- +----------------------------------------------------------- fin. diff --git a/src/Control/Concurrent/STM/TChan/Compat.hs b/src/Control/Concurrent/STM/TChan/Compat.hs --- a/src/Control/Concurrent/STM/TChan/Compat.hs +++ b/src/Control/Concurrent/STM/TChan/Compat.hs @@ -1,10 +1,14 @@ {-# OPTIONS_GHC -Wall -fwarn-tabs #-} {-# LANGUAGE CPP #-} + +#if __GLASGOW_HASKELL__ >= 701 +{-# LANGUAGE Trustworthy #-} +#endif ---------------------------------------------------------------- -- 2012.02.29 -- | -- Module : Control.Concurrent.STM.TChan.Compat --- Copyright : Copyright (c) 2011--2012 wren ng thornton +-- Copyright : Copyright (c) 2011--2013 wren ng thornton -- License : BSD -- Maintainer : wren@community.haskell.org -- Stability : provisional @@ -12,7 +16,7 @@ -- -- Compatibility layer for older versions of the @stm@ library. -- Namely, we define 'tryReadTChan', 'peekTChan', and 'tryPeekTChan' --- which @stm<2.3.0@ lacks. These implementations are less efficient +-- which @stm < 2.3.0@ lacks. These implementations are less efficient -- than the package versions due to the 'TChan' type being abstract. -- However, this module uses Cabal-style CPP macros in order to use -- the package versions when available. @@ -22,19 +26,19 @@ -- * The TChan type TChan -- ** Creating TChans - , newTChan -- :: STM (TChan a) - , newTChanIO -- :: IO (TChan a) - , dupTChan -- :: TChan a -> STM (TChan a) + , newTChan -- :: STM (TChan a) + , newTChanIO -- :: IO (TChan a) + , dupTChan -- :: TChan a -> STM (TChan a) -- ** Reading from TChans - , readTChan -- :: TChan a -> STM a - , tryReadTChan -- :: TChan a -> STM (Maybe a) - , peekTChan -- :: TChan a -> STM a - , tryPeekTChan -- :: TChan a -> STM (Maybe a) + , readTChan -- :: TChan a -> STM a + , tryReadTChan -- :: TChan a -> STM (Maybe a) + , peekTChan -- :: TChan a -> STM a + , tryPeekTChan -- :: TChan a -> STM (Maybe a) -- ** Writing to TChans - , unGetTChan -- :: TChan a -> a -> STM () - , writeTChan -- :: TChan a -> a -> STM () + , unGetTChan -- :: TChan a -> a -> STM () + , writeTChan -- :: TChan a -> a -> STM () -- ** Predicates - , isEmptyTChan -- :: TChan a -> STM Bool + , isEmptyTChan -- :: TChan a -> STM Bool ) where import Control.Concurrent.STM.TChan -- N.B., GHC only diff --git a/src/Control/Concurrent/STM/TMChan.hs b/src/Control/Concurrent/STM/TMChan.hs --- a/src/Control/Concurrent/STM/TMChan.hs +++ b/src/Control/Concurrent/STM/TMChan.hs @@ -1,10 +1,14 @@ {-# OPTIONS_GHC -Wall -fwarn-tabs #-} {-# LANGUAGE CPP, DeriveDataTypeable #-} + +#if __GLASGOW_HASKELL__ >= 701 +{-# LANGUAGE Trustworthy #-} +#endif ---------------------------------------------------------------- -- 2012.02.12 -- | -- Module : Control.Concurrent.STM.TMChan --- Copyright : Copyright (c) 2011--2012 wren ng thornton +-- Copyright : Copyright (c) 2011--2013 wren ng thornton -- License : BSD -- Maintainer : wren@community.haskell.org -- Stability : provisional diff --git a/src/Control/Concurrent/STM/TMQueue.hs b/src/Control/Concurrent/STM/TMQueue.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/STM/TMQueue.hs @@ -0,0 +1,219 @@ +{-# OPTIONS_GHC -Wall -fwarn-tabs #-} +{-# LANGUAGE CPP, DeriveDataTypeable #-} + +#if __GLASGOW_HASKELL__ >= 701 +{-# LANGUAGE Trustworthy #-} +#endif +---------------------------------------------------------------- +-- 2013.05.12 +-- | +-- Module : Control.Concurrent.STM.TMQueue +-- Copyright : Copyright (c) 2011--2013 wren ng thornton +-- License : BSD +-- Maintainer : wren@community.haskell.org +-- Stability : provisional +-- Portability : non-portable (GHC STM, DeriveDataTypeable) +-- +-- A version of "Control.Concurrent.STM.TQueue" where the queue is +-- closeable. This is similar to a @TQueue (Maybe a)@ with a +-- monotonicity guarantee that once there's a @Nothing@ there will +-- always be @Nothing@. +-- +-- /Since: 2.0.0/ +---------------------------------------------------------------- +module Control.Concurrent.STM.TMQueue + ( + -- * The TMQueue type + TMQueue() + -- ** Creating TMQueues + , newTMQueue + , newTMQueueIO + -- ** Reading from TMQueues + , readTMQueue + , tryReadTMQueue + , peekTMQueue + , tryPeekTMQueue + -- ** Writing to TMQueues + , writeTMQueue + , unGetTMQueue + -- ** Closing TMQueues + , closeTMQueue + -- ** Predicates + , isClosedTMQueue + , isEmptyTMQueue + ) where + +import Data.Typeable (Typeable) +import Control.Applicative ((<$>)) +import Control.Monad.STM (STM) +import Control.Concurrent.STM.TVar.Compat +import Control.Concurrent.STM.TQueue.Compat -- N.B., GHC only + +-- N.B., we need a Custom cabal build-type for this to work. +#ifdef __HADDOCK__ +import Control.Monad.STM (atomically) +import System.IO.Unsafe (unsafePerformIO) +#endif +---------------------------------------------------------------- + +-- | @TMQueue@ is an abstract type representing a closeable FIFO +-- queue. +data TMQueue a = TMQueue !(TVar Bool) !(TQueue a) + deriving Typeable + + +-- | Build and returns a new instance of @TMQueue@. +newTMQueue :: STM (TMQueue a) +newTMQueue = do + closed <- newTVar False + queue <- newTQueue + return (TMQueue closed queue) + + +-- | @IO@ version of 'newTMQueue'. This is useful for creating +-- top-level @TMQueue@s using 'unsafePerformIO', because using +-- 'atomically' inside 'unsafePerformIO' isn't possible. +newTMQueueIO :: IO (TMQueue a) +newTMQueueIO = do + closed <- newTVarIO False + queue <- newTQueueIO + return (TMQueue closed queue) + +-- | Read the next value from the @TMQueue@, retrying if the queue +-- is empty (and not closed). We return @Nothing@ immediately if +-- the queue is closed and empty. +readTMQueue :: TMQueue a -> STM (Maybe a) +readTMQueue (TMQueue closed queue) = do + b <- readTVar closed + if b + then tryReadTQueue queue + else Just <$> readTQueue queue +{- +-- The above is lazier reading from @queue@, and slightly optimized, compared to the clearer: +readTMQueue (TMQueue closed queue) = do + b <- isEmptyTQueue queue + b' <- readTVar closed + if b && b' + then return Nothing + else Just <$> readTQueue queue +-- TODO: compare Core and benchmarks; is the loss of clarity worth it? +-} + + +-- | A version of 'readTMQueue' which does not retry. Instead it +-- returns @Just Nothing@ if the queue is open but no value is +-- available; it still returns @Nothing@ if the queue is closed +-- and empty. +tryReadTMQueue :: TMQueue a -> STM (Maybe (Maybe a)) +tryReadTMQueue (TMQueue closed queue) = do + b <- readTVar closed + if b + then fmap Just <$> tryReadTQueue queue + else Just <$> tryReadTQueue queue +{- +-- The above is lazier reading from @queue@ (and removes an extraneous isEmptyTQueue when using the compatibility layer) than the clearer: +tryReadTMQueue (TMQueue closed queue) = do + b <- isEmptyTQueue queue + b' <- readTVar closed + if b && b' + then return Nothing + else Just <$> tryReadTQueue queue +-- TODO: compare Core and benchmarks; is the loss of clarity worth it? +-} + + +-- | Get the next value from the @TMQueue@ without removing it, +-- retrying if the queue is empty. +peekTMQueue :: TMQueue a -> STM (Maybe a) +peekTMQueue (TMQueue closed queue) = do + b <- readTVar closed + if b + then do + b' <- isEmptyTQueue queue + if b' + then return Nothing + else Just <$> peekTQueue queue + else Just <$> peekTQueue queue +{- +-- The above is lazier reading from @queue@ than the clearer: +peekTMQueue (TMQueue closed queue) = do + b <- isEmptyTQueue queue + b' <- readTVar closed + if b && b' + then return Nothing + else Just <$> peekTQueue queue +-- TODO: compare Core and benchmarks; is the loss of clarity worth it? +-} + + +-- | A version of 'peekTMQueue' which does not retry. Instead it +-- returns @Just Nothing@ if the queue is open but no value is +-- available; it still returns @Nothing@ if the queue is closed +-- and empty. +tryPeekTMQueue :: TMQueue a -> STM (Maybe (Maybe a)) +tryPeekTMQueue (TMQueue closed queue) = do + b <- readTVar closed + if b + then fmap Just <$> tryPeekTQueue queue + else Just <$> tryPeekTQueue queue +{- +-- The above is lazier reading from @queue@ (and removes an extraneous isEmptyTQueue when using the compatibility layer) than the clearer: +tryPeekTMQueue (TMQueue closed queue) = do + b <- isEmptyTQueue queue + b' <- readTVar closed + if b && b' + then return Nothing + else Just <$> tryPeekTQueue queue +-- TODO: compare Core and benchmarks; is the loss of clarity worth it? +-} + + +-- | Write a value to a @TMQueue@. If the queue is closed then the +-- value is silently discarded. Use 'isClosedTMQueue' to determine +-- if the queue is closed before writing, as needed. +writeTMQueue :: TMQueue a -> a -> STM () +writeTMQueue (TMQueue closed queue) x = do + b <- readTVar closed + if b + then return () -- discard silently + else writeTQueue queue x + + +-- | Put a data item back onto a queue, where it will be the next +-- item read. If the queue is closed then the value is silently +-- discarded; you can use 'peekTMQueue' to circumvent this in certain +-- circumstances. +unGetTMQueue :: TMQueue a -> a -> STM () +unGetTMQueue (TMQueue closed queue) x = do + b <- readTVar closed + if b + then return () -- discard silently + else unGetTQueue queue x + + +-- | Closes the @TMQueue@, preventing any further writes. +closeTMQueue :: TMQueue a -> STM () +closeTMQueue (TMQueue closed _queue) = + writeTVar closed True + + +-- | Returns @True@ if the supplied @TMQueue@ has been closed. +isClosedTMQueue :: TMQueue a -> STM Bool +isClosedTMQueue (TMQueue closed _queue) = + readTVar closed + +{- +-- | Returns @True@ if the supplied @TMQueue@ has been closed. +isClosedTMQueueIO :: TMQueue a -> IO Bool +isClosedTMQueueIO (TMQueue closed _queue) = + readTVarIO closed +-} + + +-- | Returns @True@ if the supplied @TMQueue@ is empty. +isEmptyTMQueue :: TMQueue a -> STM Bool +isEmptyTMQueue (TMQueue _closed queue) = + isEmptyTQueue queue + +---------------------------------------------------------------- +----------------------------------------------------------- fin. diff --git a/src/Control/Concurrent/STM/TMVar/Compat.hs b/src/Control/Concurrent/STM/TMVar/Compat.hs --- a/src/Control/Concurrent/STM/TMVar/Compat.hs +++ b/src/Control/Concurrent/STM/TMVar/Compat.hs @@ -1,17 +1,21 @@ {-# OPTIONS_GHC -Wall -fwarn-tabs #-} {-# LANGUAGE CPP #-} + +#if __GLASGOW_HASKELL__ >= 701 +{-# LANGUAGE Trustworthy #-} +#endif ---------------------------------------------------------------- -- 2012.02.29 -- | -- Module : Control.Concurrent.STM.TMVar.Compat --- Copyright : Copyright (c) 2011--2012 wren ng thornton +-- Copyright : Copyright (c) 2011--2013 wren ng thornton -- License : BSD -- Maintainer : wren@community.haskell.org -- Stability : provisional -- Portability : non-portable (STM, CPP) -- -- Compatibility layer for older versions of the @stm@ library. --- Namely, we define 'tryReadTMVar' which @stm<2.3.0@ lacks. This +-- Namely, we define 'tryReadTMVar' which @stm < 2.3.0@ lacks. This -- module uses Cabal-style CPP macros in order to use the package -- versions when available. This isn't actually used by the @stm-chans@ -- package, but we provide it anyways since we provide compatibility @@ -24,22 +28,22 @@ -- * The TMVar type TMVar() -- ** Creating TMVars - , newTMVar -- :: a -> STM (TMVar a) - , newTMVarIO -- :: a -> IO (TMVar a) - , newEmptyTMVar -- :: STM (TMVar a) - , newEmptyTMVarIO -- :: IO (TMVar a) + , newTMVar -- :: a -> STM (TMVar a) + , newTMVarIO -- :: a -> IO (TMVar a) + , newEmptyTMVar -- :: STM (TMVar a) + , newEmptyTMVarIO -- :: IO (TMVar a) -- ** Reading from TMVars - , readTMVar -- :: TMVar a -> STM a - , tryReadTMVar -- :: TMVar a -> STM (Maybe a) - , takeTMVar -- :: TMVar a -> STM a - , tryTakeTMVar -- :: TMVar a -> STM (Maybe a) + , readTMVar -- :: TMVar a -> STM a + , tryReadTMVar -- :: TMVar a -> STM (Maybe a) + , takeTMVar -- :: TMVar a -> STM a + , tryTakeTMVar -- :: TMVar a -> STM (Maybe a) -- ** Writing to TMVars - , putTMVar -- :: TMVar a -> a -> STM () - , tryPutTMVar -- :: TMVar a -> a -> STM Bool - , swapTMVar -- :: TMVar a -> a -> STM a + , putTMVar -- :: TMVar a -> a -> STM () + , tryPutTMVar -- :: TMVar a -> a -> STM Bool + , swapTMVar -- :: TMVar a -> a -> STM a -- TODO: make another patch for trySwapTMVar? -- ** Other capabilities - , isEmptyTMVar -- :: TMVar a -> STM Bool + , isEmptyTMVar -- :: TMVar a -> STM Bool ) where import Control.Concurrent.STM.TMVar diff --git a/src/Control/Concurrent/STM/TQueue/Compat.hs b/src/Control/Concurrent/STM/TQueue/Compat.hs new file mode 100644 --- /dev/null +++ b/src/Control/Concurrent/STM/TQueue/Compat.hs @@ -0,0 +1,156 @@ +{-# OPTIONS_GHC -Wall -fwarn-tabs -fno-warn-name-shadowing #-} +{-# LANGUAGE CPP, DeriveDataTypeable #-} + +#if __GLASGOW_HASKELL__ >= 701 +{-# LANGUAGE Trustworthy #-} +#endif +---------------------------------------------------------------- +-- 2013.05.12 +-- | +-- Module : Control.Concurrent.STM.TQueue.Compat +-- Copyright : Copyright (c) 2011--2013 wren ng thornton +-- License : BSD +-- Maintainer : wren@community.haskell.org +-- Stability : provisional +-- Portability : non-portable (CPP, STM, DeriveDataTypeable) +-- +-- Compatibility layer for older versions of the @stm@ library. +-- Namely, we copy "Control.Concurrent.STM.TQueue" module which +-- @stm < 2.4.0@ lacks. This module uses Cabal-style CPP macros in +-- order to use the package versions when available. +-- +-- /Since: 2.0.0/ +---------------------------------------------------------------- + +module Control.Concurrent.STM.TQueue.Compat + ( + -- * The TQueue type + TQueue + -- ** Creating TQueues + , newTQueue -- :: STM (TQueue a) + , newTQueueIO -- :: IO (TQueue a) + -- ** Reading from TQueues + , readTQueue -- :: TQueue a -> STM a + , tryReadTQueue -- :: TQueue a -> STM (Maybe a) + , peekTQueue -- :: TQueue a -> STM a + , tryPeekTQueue -- :: TQueue a -> STM (Maybe a) + -- ** Writing to TQueues + , writeTQueue -- :: TQueue a -> a -> STM () + , unGetTQueue -- :: TQueue a -> a -> STM () + -- ** Predicates + , isEmptyTQueue -- :: TQueue a -> STM Bool + ) where + +#if MIN_VERSION_stm(2,4,0) +import Control.Concurrent.STM.TQueue +#else +import GHC.Conc +import Data.Typeable (Typeable) + + +-- | 'TQueue' is an abstract type representing an unbounded FIFO channel. +data TQueue a = TQueue {-# UNPACK #-} !(TVar [a]) + {-# UNPACK #-} !(TVar [a]) + deriving Typeable + +instance Eq (TQueue a) where + TQueue a _ == TQueue b _ = a == b + + +-- | Build and returns a new instance of 'TQueue'. +newTQueue :: STM (TQueue a) +newTQueue = do + read <- newTVar [] + write <- newTVar [] + return (TQueue read write) + + +-- | @IO@ version of 'newTQueue'. This is useful for creating +-- top-level 'TQueue's using 'System.IO.Unsafe.unsafePerformIO', +-- because using 'atomically' inside 'System.IO.Unsafe.unsafePerformIO' +-- isn't possible. +newTQueueIO :: IO (TQueue a) +newTQueueIO = do + read <- newTVarIO [] + write <- newTVarIO [] + return (TQueue read write) + + +-- | Write a value to a 'TQueue'. +writeTQueue :: TQueue a -> a -> STM () +writeTQueue (TQueue _read write) a = do + listend <- readTVar write + writeTVar write (a:listend) + + +-- | Read the next value from the 'TQueue'. +readTQueue :: TQueue a -> STM a +readTQueue (TQueue read write) = do + xs <- readTVar read + case xs of + (x:xs') -> do + writeTVar read xs' + return x + [] -> do + ys <- readTVar write + case ys of + [] -> retry + _ -> + case reverse ys of + [] -> error "readTQueue" + (z:zs) -> do + writeTVar write [] + writeTVar read zs + return z + + +-- | A version of 'readTQueue' which does not retry. Instead it +-- returns @Nothing@ if no value is available. +tryReadTQueue :: TQueue a -> STM (Maybe a) +tryReadTQueue c = fmap Just (readTQueue c) `orElse` return Nothing + + +-- | Get the next value from the @TQueue@ without removing it, +-- retrying if the channel is empty. +peekTQueue :: TQueue a -> STM a +peekTQueue c = do + x <- readTQueue c + unGetTQueue c x + return x + + +-- | A version of 'peekTQueue' which does not retry. Instead it +-- returns @Nothing@ if no value is available. +tryPeekTQueue :: TQueue a -> STM (Maybe a) +tryPeekTQueue c = do + m <- tryReadTQueue c + case m of + Nothing -> return Nothing + Just x -> do + unGetTQueue c x + return m + + +-- | Put a data item back onto a channel, where it will be the next +-- item read. +unGetTQueue :: TQueue a -> a -> STM () +unGetTQueue (TQueue read _write) a = do + xs <- readTVar read + writeTVar read (a:xs) + + +-- | Returns 'True' if the supplied 'TQueue' is empty. +isEmptyTQueue :: TQueue a -> STM Bool +isEmptyTQueue (TQueue read write) = do + xs <- readTVar read + case xs of + (_:_) -> return False + [] -> do + ys <- readTVar write + case ys of + [] -> return True + _ -> return False +#endif + +---------------------------------------------------------------- +----------------------------------------------------------- fin. diff --git a/src/Control/Concurrent/STM/TVar/Compat.hs b/src/Control/Concurrent/STM/TVar/Compat.hs --- a/src/Control/Concurrent/STM/TVar/Compat.hs +++ b/src/Control/Concurrent/STM/TVar/Compat.hs @@ -1,18 +1,22 @@ {-# OPTIONS_GHC -Wall -fwarn-tabs #-} {-# LANGUAGE CPP #-} + +#if __GLASGOW_HASKELL__ >= 701 +{-# LANGUAGE Trustworthy #-} +#endif ---------------------------------------------------------------- -- 2012.02.29 -- | -- Module : Control.Concurrent.STM.TVar.Compat --- Copyright : Copyright (c) 2011--2012 wren ng thornton +-- Copyright : Copyright (c) 2011--2013 wren ng thornton -- License : BSD -- Maintainer : wren@community.haskell.org -- Stability : provisional -- Portability : non-portable (STM, CPP) -- -- Compatibility layer for older versions of the @stm@ library. --- Namely, we define 'readTVarIO' which @stm<2.1.2@ lacks; and we --- define 'modifyTVar', 'modifyTVar'', and 'swapTVar' which @stm<2.3.0@ +-- Namely, we define 'readTVarIO' which @stm < 2.1.2@ lacks; and we +-- define 'modifyTVar', 'modifyTVar'', and 'swapTVar' which @stm < 2.3.0@ -- lacks. This module uses Cabal-style CPP macros in order to use -- the package versions when available. ---------------------------------------------------------------- diff --git a/stm-chans.cabal b/stm-chans.cabal --- a/stm-chans.cabal +++ b/stm-chans.cabal @@ -1,5 +1,5 @@ ---------------------------------------------------------------- --- wren ng thornton ~ 2012.02.25 +-- wren ng thornton ~ 2013.05.12 ---------------------------------------------------------------- -- By and large Cabal >=1.2 is fine; but >= 1.6 gives tested-with: @@ -9,12 +9,12 @@ Build-Type: Custom Name: stm-chans -Version: 1.3.1 +Version: 2.0.0 Stability: provisional Homepage: http://code.haskell.org/~wren/ Author: wren ng thornton, Thomas DuBuisson Maintainer: wren@community.haskell.org -Copyright: Copyright (c) 2011--2012 wren ng thornton +Copyright: Copyright (c) 2011--2013 wren ng thornton License: BSD3 License-File: LICENSE @@ -23,7 +23,7 @@ Description: Additional types of channels for STM. Tested-With: - GHC == 6.12.1, GHC == 6.12.3 + GHC == 6.12.1, GHC == 6.12.3, GHC == 7.6.1 Extra-source-files: AUTHORS, README, VERSION Source-Repository head @@ -32,19 +32,25 @@ ---------------------------------------------------------------- Library - -- Not sure what the real minbounds are for base and stm... + -- N.B., the following versions are required for: + -- * stm >= 2.4: TQueue and TBQueue. + -- * stm >= 2.3.0: fast tryReadTChan, peekTChan, tryPeekTChan, + -- tryReadTMVar, modifyTVar, modifyTVar', swapTVar. + -- * stm >= 2.1.2: fast readTVarIO. -- - -- N.B., stm >= 2.1.2 is required for fast readTVarIO. And - -- stm >= 2.3.0 is required for fast tryReadTChan, peekTChan, - -- tryPeekTChan, tryReadTMVar, modifyTVar, modifyTVar', swapTVar. + -- Not sure what the real minbound is for base... Build-Depends: base >= 4.1 && < 5 , stm >= 2.1.1 - + Hs-Source-Dirs: src Exposed-Modules: Control.Concurrent.STM.TBChan , Control.Concurrent.STM.TBMChan , Control.Concurrent.STM.TMChan + , Control.Concurrent.STM.TBMQueue + , Control.Concurrent.STM.TMQueue , Control.Concurrent.STM.TChan.Compat + , Control.Concurrent.STM.TQueue.Compat + , Control.Concurrent.STM.TBQueue.Compat , Control.Concurrent.STM.TMVar.Compat , Control.Concurrent.STM.TVar.Compat