stm 2.3 → 2.4
raw patch · 6 files changed
+403/−10 lines, 6 files
Files
- Control/Concurrent/STM.hs +5/−1
- Control/Concurrent/STM/TBQueue.hs +179/−0
- Control/Concurrent/STM/TChan.hs +60/−7
- Control/Concurrent/STM/TQueue.hs +137/−0
- Control/Sequential/STM.hs +2/−0
- stm.cabal +20/−2
Control/Concurrent/STM.hs view
@@ -29,7 +29,9 @@ module Control.Concurrent.STM.TVar, #ifdef __GLASGOW_HASKELL__ module Control.Concurrent.STM.TMVar,- module Control.Concurrent.STM.TChan,+ module Control.Concurrent.STM.TChan,+ module Control.Concurrent.STM.TQueue,+ module Control.Concurrent.STM.TBQueue, #endif module Control.Concurrent.STM.TArray ) where@@ -41,3 +43,5 @@ import Control.Concurrent.STM.TChan #endif import Control.Concurrent.STM.TArray+import Control.Concurrent.STM.TQueue+import Control.Concurrent.STM.TBQueue
+ Control/Concurrent/STM/TBQueue.hs view
@@ -0,0 +1,179 @@+{-# OPTIONS_GHC -fno-warn-name-shadowing #-}+{-# LANGUAGE CPP, DeriveDataTypeable #-}++#if __GLASGOW_HASKELL__ >= 701+{-# LANGUAGE Trustworthy #-}+#endif++-----------------------------------------------------------------------------+-- |+-- Module : Control.Concurrent.STM.TBQueue+-- Copyright : (c) The University of Glasgow 2012+-- License : BSD-style (see the file libraries/base/LICENSE)+-- +-- Maintainer : libraries@haskell.org+-- Stability : experimental+-- Portability : non-portable (requires STM)+--+-- 'TBQueue' is a bounded version of 'TQueue'. The queue has a maximum+-- capacity set when it is created. If the queue already contains the+-- maximum number of elements, then 'writeTBQueue' blocks until an+-- element is removed from the queue.+--+-- The implementation is based on the traditional purely-functional+-- queue representation that uses two lists to obtain amortised /O(1)/+-- enqueue and dequeue operations.+--+-----------------------------------------------------------------------------++module Control.Concurrent.STM.TBQueue (+ -- * TBQueue+ TBQueue,+ newTBQueue,+ newTBQueueIO,+ readTBQueue,+ tryReadTBQueue,+ peekTBQueue,+ tryPeekTBQueue,+ writeTBQueue,+ unGetTBQueue,+ isEmptyTBQueue,+ ) where++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 do 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 -- NB. 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 do 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
Control/Concurrent/STM/TChan.hs view
@@ -24,16 +24,23 @@ #ifdef __GLASGOW_HASKELL__ -- * TChans TChan,- newTChan,++ -- ** Construction+ newTChan, newTChanIO,+ newBroadcastTChan,+ newBroadcastTChanIO,+ dupTChan,++ -- ** Reading and writing readTChan, tryReadTChan, peekTChan, tryPeekTChan, writeTChan,- dupTChan,- unGetTChan,- isEmptyTChan+ unGetTChan,+ isEmptyTChan,+ cloneTChan #endif ) where @@ -42,13 +49,17 @@ import Data.Typeable (Typeable) +#define _UPK_(x) {-# UNPACK #-} !(x)+ -- | 'TChan' is an abstract type representing an unbounded FIFO channel.-data TChan a = TChan (TVar (TVarList a)) (TVar (TVarList a)) deriving Typeable+data TChan a = TChan _UPK_(TVar (TVarList a))+ _UPK_(TVar (TVarList a))+ deriving (Eq, Typeable) type TVarList a = TVar (TList a)-data TList a = TNil | TCons a (TVarList a)+data TList a = TNil | TCons a _UPK_(TVarList a) --- |Build and returns a new instance of 'TChan'+-- |Build and return a new instance of 'TChan' newTChan :: STM (TChan a) newTChan = do hole <- newTVar TNil@@ -67,6 +78,40 @@ write <- newTVarIO hole return (TChan read write) +-- | Create a write-only 'TChan'. More precisely, 'readTChan' will 'retry'+-- even after items have been written to the channel. The only way to read+-- a broadcast channel is to duplicate it with 'dupTChan'.+--+-- Consider a server that broadcasts messages to clients:+--+-- >serve :: TChan Message -> Client -> IO loop+-- >serve broadcastChan client = do+-- > myChan <- dupTChan broadcastChan+-- > forever $ do+-- > message <- readTChan myChan+-- > send client message+--+-- The problem with using 'newTChan' to create the broadcast channel is that if+-- it is only written to and never read, items will pile up in memory. By+-- using 'newBroadcastTChan' to create the broadcast channel, items can be+-- garbage collected after clients have seen them.+newBroadcastTChan :: STM (TChan a)+newBroadcastTChan = do+ dummy_hole <- newTVar TNil+ write_hole <- newTVar TNil+ read <- newTVar dummy_hole+ write <- newTVar write_hole+ return (TChan read write)++-- | @IO@ version of 'newBroadcastTChan'.+newBroadcastTChanIO :: IO (TChan a)+newBroadcastTChanIO = do+ dummy_hole <- newTVarIO TNil+ write_hole <- newTVarIO TNil+ read <- newTVarIO dummy_hole+ write <- newTVarIO write_hole+ return (TChan read write)+ -- |Write a value to a 'TChan'. writeTChan :: TChan a -> a -> STM () writeTChan (TChan _read write) a = do@@ -143,4 +188,12 @@ case head of TNil -> return True TCons _ _ -> return False++-- |Clone a 'TChan': similar to dupTChan, but the cloned channel starts with the+-- same content available as the original channel.+cloneTChan :: TChan a -> STM (TChan a)+cloneTChan (TChan read write) = do+ readpos <- readTVar read+ new_read <- newTVar readpos+ return (TChan new_read write) #endif
+ Control/Concurrent/STM/TQueue.hs view
@@ -0,0 +1,137 @@+{-# OPTIONS_GHC -fno-warn-name-shadowing #-}+{-# LANGUAGE CPP, DeriveDataTypeable #-}++#if __GLASGOW_HASKELL__ >= 701+{-# LANGUAGE Trustworthy #-}+#endif++-----------------------------------------------------------------------------+-- |+-- Module : Control.Concurrent.STM.TQueue+-- Copyright : (c) The University of Glasgow 2012+-- License : BSD-style (see the file libraries/base/LICENSE)+-- +-- Maintainer : libraries@haskell.org+-- Stability : experimental+-- Portability : non-portable (requires STM)+--+-- A 'TQueue' is like a 'TChan', with two important differences:+--+-- * it has faster throughput than both 'TChan' and 'Chan' (although+-- the costs are amortised, so the cost of individual operations+-- can vary a lot).+--+-- * it does /not/ provide equivalents of the 'dupTChan' and+-- 'cloneTChan' operations.+--+-- The implementation is based on the traditional purely-functional+-- queue representation that uses two lists to obtain amortised /O(1)/+-- enqueue and dequeue operations.+--+-----------------------------------------------------------------------------++module Control.Concurrent.STM.TQueue (+ -- * TQueue+ TQueue,+ newTQueue,+ newTQueueIO,+ readTQueue,+ tryReadTQueue,+ peekTQueue,+ tryPeekTQueue,+ writeTQueue,+ unGetTQueue,+ isEmptyTQueue,+ ) where++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
Control/Sequential/STM.hs view
@@ -14,7 +14,9 @@ TVar, newTVar, newTVarIO, readTVar, readTVarIO, writeTVar ) where +#if __GLASGOW_HASKELL__ < 705 import Prelude hiding (catch)+#endif import Control.Applicative (Applicative(pure, (<*>))) import Control.Exception import Data.IORef
stm.cabal view
@@ -1,11 +1,27 @@ name: stm-version: 2.3+version: 2.4 license: BSD3 license-file: LICENSE maintainer: libraries@haskell.org synopsis: Software Transactional Memory category: Concurrency-description: A modular composable concurrency abstraction.+description:+ A modular composable concurrency abstraction.+ .+ Changes in version 2.4+ .+ * Added "Control.Concurrent.STM.TQueue" (a faster @TChan@)+ .+ * Added "Control.Concurrent.STM.TBQueue" (a bounded channel based on @TQueue@)+ .+ * @TChan@ has an @Eq@ instances+ .+ * Added @newBroadcastTChan@ and @newBroadcastTChanIO@+ .+ * Some performance improvements for @TChan@+ .+ * Added @cloneTChan@+ build-type: Simple cabal-version: >=1.6 @@ -22,6 +38,8 @@ Control.Concurrent.STM.TVar Control.Concurrent.STM.TChan Control.Concurrent.STM.TMVar+ Control.Concurrent.STM.TQueue+ Control.Concurrent.STM.TBQueue Control.Monad.STM other-modules: Control.Sequential.STM