rolling-queue (empty) → 0.1
raw patch · 6 files changed
+884/−0 lines, 6 filesdep +basedep +stmsetup-changed
Dependencies added: base, stm
Files
- Data/STM/RollingQueue.hs +310/−0
- LICENSE +30/−0
- Setup.hs +2/−0
- rolling-queue.cabal +47/−0
- testing/trivial.expected +420/−0
- testing/trivial.hs +75/−0
+ Data/STM/RollingQueue.hs view
@@ -0,0 +1,310 @@+-- |+-- Module: Data.STM.RollingQueue+-- Copyright: (c) Joseph Adams 2012+-- License: BSD3+-- Maintainer: joeyadams3.14159@gmail.com+-- Portability: Requires STM+--+-- This module is usually imported qualified:+--+-- >import Data.STM.RollingQueue (RollingQueue)+-- >import qualified Data.STM.RollingQueue as RQ+{-# LANGUAGE BangPatterns, DeriveDataTypeable #-}+module Data.STM.RollingQueue (+ RollingQueue,+ new,+ newIO,+ write,+ read,+ tryRead,+ isEmpty,+ length,+ setLimit,+ getLimit,++ -- * Debugging+ checkInvariants,+ CheckException(..),+ dump,+) where++import Prelude hiding (length, read)+import qualified Prelude++import Control.Concurrent.STM hiding (check)+import Control.Exception (Exception)+import Control.Monad (join)+import Data.Typeable (Typeable)++-- | A 'RollingQueue' is a bounded FIFO channel. When the size limit is+-- exceeded, older entries are discarded to make way for newer entries.+--+-- Note: if the size limit is less than @1@, 'write' will have no effect, and+-- 'read' will always 'retry'.+data RollingQueue a = RQ (TVar (ReadEnd a)) (TVar (WriteEnd a))+ deriving Typeable++-- Invariants:+--+-- * writeCounter - readCounter = number of items in the queue+--+-- * writeCounter >= readCounter, since the queue count cannot be negative++instance Eq (RollingQueue a) where+ (==) (RQ r1 _) (RQ r2 _) = r1 == r2++-- | Invariants:+--+-- * readCounter >= 0, readDiscarded >= 0+data ReadEnd a =+ ReadEnd+ { readPtr :: !(TCell a)+ -- ^ Pointer to next item in the stream+ , readCounter :: !Int+ -- ^ Number of reads since we last synced with the writer+ , readDiscarded :: !Int+ -- ^ Number of log entries discarded since the last read+ }++-- | Invariants:+--+-- * readTVar writePtr ==> TNil+--+-- * writeCounter <= sizeLimit. However, this will normally be false for the+-- 'WriteEnd' passed to 'syncEnds'.+--+-- * writeCounter >= 0, sizeLimit >= 0+data WriteEnd a =+ WriteEnd+ { writePtr :: !(TCell a)+ -- ^ Pointer to the hole (which is a TNil)+ , writeCounter :: !Int+ -- ^ Write counter. Number of items in the queue, not taking into+ -- account reads performed since the last call to 'syncEnds'.+ , sizeLimit :: !Int+ -- ^ The size limit of the RollingQueue is stored here, in the+ -- 'WriteEnd'. This makes it convenient for 'write' to access.+ }++type TCell a = TVar (TList a)+data TList a = TNil | TCons a (TCell a)+++-- | Create a new, empty 'RollingQueue', with the given size limit.+--+-- To change the size limit later, use 'setLimit'.+new :: Int -> STM (RollingQueue a)+new limit = do+ hole <- newTVar TNil+ rv <- newTVar $ ReadEnd hole 0 0+ wv <- newTVar $ WriteEnd hole 0 (max 0 limit)+ return (RQ rv wv)++{- |+@IO@ variant of 'new'. This is useful for creating top-level+'RollingQueue's using 'System.IO.Unsafe.unsafePerformIO', because performing+'atomically' inside a pure computation is extremely dangerous (can lead to+'Control.Exception.NestedAtomically' errors and even segfaults, see GHC ticket+#5866).++Example:++@+logQueue :: 'RollingQueue' LogEntry+logQueue = 'System.IO.Unsafe.unsafePerformIO' (RQ.'newIO' 1000)+\{\-\# NOINLINE logQueue \#\-\}+@+-}+newIO :: Int -> IO (RollingQueue a)+newIO limit = do+ hole <- newTVarIO TNil+ rv <- newTVarIO $ ReadEnd hole 0 0+ wv <- newTVarIO $ WriteEnd hole 0 (max 0 limit)+ return (RQ rv wv)++-- | Write an entry to the rolling queue. If the queue is full, discard the+-- oldest entry.+--+-- There is no @tryWrite@, because 'write' never retries.+write :: RollingQueue a -> a -> STM ()+write rq@(RQ _ wv) x = do+ w <- readTVar wv+ new_hole <- newTVar TNil+ writeTVar (writePtr w) (TCons x new_hole)+ updateWriteEnd rq $ WriteEnd new_hole (writeCounter w + 1) (sizeLimit w)++-- | Read the next entry from the 'RollingQueue'. 'retry' if the queue is+-- empty.+--+-- The 'Int' is the number of entries discarded since the last read operation+-- (usually 0).+read :: RollingQueue a -> STM (a, Int)+read rq = tryRead rq >>= maybe retry return++-- | Like 'read', but do not 'retry'. Instead, return 'Nothing' if the queue+-- is empty.+tryRead :: RollingQueue a -> STM (Maybe (a, Int))+tryRead (RQ rv _) = do+ r <- readTVar rv+ xs <- readTVar (readPtr r)+ case xs of+ TNil -> return Nothing+ TCons x cell' -> do+ writeTVar rv $ ReadEnd cell' (readCounter r + 1) 0+ return $ Just (x, readDiscarded r)++-- | Test if the queue is empty.+isEmpty :: RollingQueue a -> STM Bool+isEmpty (RQ rv _) = do+ r <- readTVar rv+ xs <- readTVar (readPtr r)+ case xs of+ TNil -> return True+ TCons _ _ -> return False++-- | /O(1)/ Get the number of items in the queue.+length :: RollingQueue a -> STM Int+length (RQ rv wv) = do+ r <- readTVar rv+ w <- readTVar wv+ return (writeCounter w - readCounter r)++-- | Adjust the size limit. Queue entries will be discarded if necessary to+-- satisfy the new limit.+setLimit :: RollingQueue a -> Int -> STM ()+setLimit rq@(RQ _ wv) new_limit = do+ w <- readTVar wv+ updateWriteEnd rq w{sizeLimit = max 0 new_limit}++-- | Get the current size limit. This will return 0 if a negative value was+-- passed to 'new', 'newIO', or 'setLimit'.+getLimit :: RollingQueue a -> STM Int+getLimit (RQ _ wv) = do+ w <- readTVar wv+ return (sizeLimit w)++------------------------------------------------------------------------+-- Internal++-- | Update the 'WriteEnd'. If the size limit is exceeded, use 'syncEnds'.+updateWriteEnd :: RollingQueue a -> WriteEnd a -> STM ()+updateWriteEnd (RQ rv wv) w+ | writeCounter w <= sizeLimit w+ = writeTVar wv w+ | otherwise = do+ r <- readTVar rv+ (r', w') <- syncEnds r w+ writeTVar rv r'+ writeTVar wv w'++-- | Sync the reader and writer. This sets the ReadEnd's counter to 0, and+-- discards old log entries to satisfy the size limit (if necessary).+syncEnds :: ReadEnd a -> WriteEnd a -> STM (ReadEnd a, WriteEnd a)+syncEnds r w = do+ let count = writeCounter w - readCounter r+ limit = sizeLimit w+ if count > limit+ then do+ let drop_count = count - limit+ rp' <- dropItems drop_count (readPtr r)+ return ( ReadEnd rp' 0 (readDiscarded r + drop_count)+ , WriteEnd (writePtr w) limit limit+ )+ else+ return ( ReadEnd (readPtr r) 0 (readDiscarded r)+ , WriteEnd (writePtr w) count limit+ )++-- | TCell variant of 'drop'. This does not modify the cells themselves, it+-- just returns the new pointer.+dropItems :: Int -> TCell a -> STM (TCell a)+dropItems n cell+ | n <= 0 = return cell+ | otherwise = do+ xs <- readTVar cell+ case xs of+ TNil -> return cell+ TCons _ cell' -> dropItems (n-1) cell'++------------------------------------------------------------------------+-- Debugging++data CheckException = CheckException String+ deriving Typeable++instance Show CheckException where+ show (CheckException msg) = "Data.STM.RollingQueue checkInvariants: " ++ msg++instance Exception CheckException++-- | Verify internal structure. Throw a 'CheckException' if the check fails,+-- signifying a bug in the implementation.+checkInvariants :: RollingQueue a -> STM ()+checkInvariants (RQ rv wv) = do+ r <- readTVar rv+ w <- readTVar wv++ check (readCounter r >= 0) "readCounter >= 0"+ check (readDiscarded r >= 0) "readDiscarded >= 0"++ check (writeCounter w >= 0) "writeCounter >= 0"+ check (sizeLimit w >= 0) "sizeLimit >= 0"+ check (writeCounter w <= sizeLimit w) "writeCounter <= sizeLimit"+ hole <- readTVar (writePtr w)+ case hole of+ TNil -> return ()+ TCons _ _ -> throwSTM $ CheckException "writePtr does not point to a TNil"++ check (writeCounter w >= readCounter r) "writeCounter >= readCounter"+ len <- traverseLength (readPtr r)+ check (writeCounter w - readCounter r == len) "writeCounter - readCounter == length"++ where+ check b expr | b = return ()+ | otherwise = throwSTM $ CheckException $ expr ++ " does not hold"++traverseLength :: TCell a -> STM Int+traverseLength = loop 0+ where+ loop !n cell = do+ xs <- readTVar cell+ case xs of+ TNil -> return n+ TCons _ cell' -> loop (n+1) cell'++-- | Return a list of all items currently in the queue. This does not modify+-- the 'RollingQueue'.+getItems :: RollingQueue a -> STM [a]+getItems (RQ rv _) = do+ r <- readTVar rv+ loop id (readPtr r)+ where+ loop dl cell = do+ xs <- readTVar cell+ case xs of+ TNil -> return $ dl []+ TCons x cell' -> loop (dl . (x :)) cell'++-- | Return a list of internal values as key-value pairs.+getAttributes :: RollingQueue a -> STM [(String, String)]+getAttributes (RQ rv wv) = do+ r <- readTVar rv+ w <- readTVar wv+ return [ ("readCounter", show $ readCounter r)+ , ("readDiscarded", show $ readDiscarded r)+ , ("writeCounter", show $ writeCounter w)+ , ("sizeLimit", show $ sizeLimit w)+ ]++-- | Dump the RollingQueue (output and internal counters) to standard output.+-- This calls 'checkInvariants' first.+dump :: Show a => RollingQueue a -> IO ()+dump rq = join $ atomically $ do+ checkInvariants rq+ xs <- getItems rq+ attrs <- getAttributes rq+ return $ do+ print xs+ let c1width = maximum $ map (Prelude.length . fst) attrs+ mapM_ putStrLn+ [k ++ replicate (c1width - Prelude.length k) ' ' ++ " = " ++ v | (k, v) <- attrs]
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright (c) 2012, Joseph Adams++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++ * Redistributions of source code must retain the above copyright+ notice, this list of conditions and the following disclaimer.++ * Redistributions in binary form must reproduce the above+ copyright notice, this list of conditions and the following+ disclaimer in the documentation and/or other materials provided+ with the distribution.++ * Neither the name of Joseph Adams nor the names of other+ contributors may be used to endorse or promote products derived+ from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ rolling-queue.cabal view
@@ -0,0 +1,47 @@+name: rolling-queue+version: 0.1+synopsis: Bounded channel for STM that discards old entries when full+description:+ This package provides a FIFO channel for STM supporting a size limit. When+ this limit is reached, older entries are discarded to make way for newer+ entries.+ .+ The motivation for this is logging. If log entries are written to a plain+ @TChan@, the program will use a lot of memory if it produces log entries+ faster than they can be processed. If log entries are written to a bounded+ channel where writes block (e.g. the @stm-chans@ package), the program may+ deadlock if the log channel fills up. With 'Data.STM.RollingQueue', old+ entries will be discarded instead.+ .+ Possible improvements (not available in 'Data.STM.RollingQueue') include:+ .+ * Discard lower-priority entries first.+ .+ * Discard every other entry, so some of the older entries will still be+ available.+homepage: https://github.com/joeyadams/haskell-rolling-queue+license: BSD3+license-file: LICENSE+author: Joey Adams+maintainer: joeyadams3.14159@gmail.com+copyright: Copyright (c) Joseph Adams 2012+category: Data+build-type: Simple+cabal-version: >=1.8++extra-source-files:+ testing/trivial.hs+ testing/trivial.expected++source-repository head+ type: git+ location: git://github.com/joeyadams/haskell-rolling-queue.git++library+ exposed-modules:+ Data.STM.RollingQueue++ build-depends: base == 4.*+ , stm++ ghc-options: -Wall -O2
+ testing/trivial.expected view
@@ -0,0 +1,420 @@+newIO 5:+[]+readCounter = 0+readDiscarded = 0+writeCounter = 0+sizeLimit = 5++read rq, isEmpty:+[]+readCounter = 0+readDiscarded = 0+writeCounter = 0+sizeLimit = 5++write rq 1:+[1]+readCounter = 0+readDiscarded = 0+writeCounter = 1+sizeLimit = 5++write rq 2:+[1,2]+readCounter = 0+readDiscarded = 0+writeCounter = 2+sizeLimit = 5++write rq 3:+[1,2,3]+readCounter = 0+readDiscarded = 0+writeCounter = 3+sizeLimit = 5++write rq 4:+[1,2,3,4]+readCounter = 0+readDiscarded = 0+writeCounter = 4+sizeLimit = 5++write rq 5:+[1,2,3,4,5]+readCounter = 0+readDiscarded = 0+writeCounter = 5+sizeLimit = 5++write rq 6:+[2,3,4,5,6]+readCounter = 0+readDiscarded = 1+writeCounter = 5+sizeLimit = 5++write rq 7:+[3,4,5,6,7]+readCounter = 0+readDiscarded = 2+writeCounter = 5+sizeLimit = 5++write rq 8:+[4,5,6,7,8]+readCounter = 0+readDiscarded = 3+writeCounter = 5+sizeLimit = 5++write rq 9:+[5,6,7,8,9]+readCounter = 0+readDiscarded = 4+writeCounter = 5+sizeLimit = 5++write rq 10:+[6,7,8,9,10]+readCounter = 0+readDiscarded = 5+writeCounter = 5+sizeLimit = 5++read rq (6,5):+[7,8,9,10]+readCounter = 1+readDiscarded = 0+writeCounter = 5+sizeLimit = 5++read rq (7,0):+[8,9,10]+readCounter = 2+readDiscarded = 0+writeCounter = 5+sizeLimit = 5++read rq (8,0):+[9,10]+readCounter = 3+readDiscarded = 0+writeCounter = 5+sizeLimit = 5++read rq (9,0):+[10]+readCounter = 4+readDiscarded = 0+writeCounter = 5+sizeLimit = 5++read rq (10,0):+[]+readCounter = 5+readDiscarded = 0+writeCounter = 5+sizeLimit = 5++read rq, isEmpty:+[]+readCounter = 5+readDiscarded = 0+writeCounter = 5+sizeLimit = 5++write rq 11:+[11]+readCounter = 0+readDiscarded = 0+writeCounter = 1+sizeLimit = 5++write rq 12:+[11,12]+readCounter = 0+readDiscarded = 0+writeCounter = 2+sizeLimit = 5++write rq 13:+[11,12,13]+readCounter = 0+readDiscarded = 0+writeCounter = 3+sizeLimit = 5++write rq 14:+[11,12,13,14]+readCounter = 0+readDiscarded = 0+writeCounter = 4+sizeLimit = 5++write rq 15:+[11,12,13,14,15]+readCounter = 0+readDiscarded = 0+writeCounter = 5+sizeLimit = 5++write rq 16:+[12,13,14,15,16]+readCounter = 0+readDiscarded = 1+writeCounter = 5+sizeLimit = 5++read rq (12,1):+[13,14,15,16]+readCounter = 1+readDiscarded = 0+writeCounter = 5+sizeLimit = 5++setLimit rq 3:+[14,15,16]+readCounter = 0+readDiscarded = 1+writeCounter = 3+sizeLimit = 3++setLimit rq 10:+[14,15,16]+readCounter = 0+readDiscarded = 1+writeCounter = 3+sizeLimit = 10++write rq 17:+[14,15,16,17]+readCounter = 0+readDiscarded = 1+writeCounter = 4+sizeLimit = 10++write rq 18:+[14,15,16,17,18]+readCounter = 0+readDiscarded = 1+writeCounter = 5+sizeLimit = 10++write rq 19:+[14,15,16,17,18,19]+readCounter = 0+readDiscarded = 1+writeCounter = 6+sizeLimit = 10++write rq 20:+[14,15,16,17,18,19,20]+readCounter = 0+readDiscarded = 1+writeCounter = 7+sizeLimit = 10++write rq 21:+[14,15,16,17,18,19,20,21]+readCounter = 0+readDiscarded = 1+writeCounter = 8+sizeLimit = 10++write rq 22:+[14,15,16,17,18,19,20,21,22]+readCounter = 0+readDiscarded = 1+writeCounter = 9+sizeLimit = 10++write rq 23:+[14,15,16,17,18,19,20,21,22,23]+readCounter = 0+readDiscarded = 1+writeCounter = 10+sizeLimit = 10++write rq 24:+[15,16,17,18,19,20,21,22,23,24]+readCounter = 0+readDiscarded = 2+writeCounter = 10+sizeLimit = 10++write rq 25:+[16,17,18,19,20,21,22,23,24,25]+readCounter = 0+readDiscarded = 3+writeCounter = 10+sizeLimit = 10++write rq 26:+[17,18,19,20,21,22,23,24,25,26]+readCounter = 0+readDiscarded = 4+writeCounter = 10+sizeLimit = 10++write rq 27:+[18,19,20,21,22,23,24,25,26,27]+readCounter = 0+readDiscarded = 5+writeCounter = 10+sizeLimit = 10++write rq 28:+[19,20,21,22,23,24,25,26,27,28]+readCounter = 0+readDiscarded = 6+writeCounter = 10+sizeLimit = 10++write rq 29:+[20,21,22,23,24,25,26,27,28,29]+readCounter = 0+readDiscarded = 7+writeCounter = 10+sizeLimit = 10++write rq 30:+[21,22,23,24,25,26,27,28,29,30]+readCounter = 0+readDiscarded = 8+writeCounter = 10+sizeLimit = 10++read rq (21,8):+[22,23,24,25,26,27,28,29,30]+readCounter = 1+readDiscarded = 0+writeCounter = 10+sizeLimit = 10++read rq (22,0):+[23,24,25,26,27,28,29,30]+readCounter = 2+readDiscarded = 0+writeCounter = 10+sizeLimit = 10++read rq (23,0):+[24,25,26,27,28,29,30]+readCounter = 3+readDiscarded = 0+writeCounter = 10+sizeLimit = 10++setLimit rq -5:+[]+readCounter = 0+readDiscarded = 7+writeCounter = 0+sizeLimit = 0++write rq 31 (limit is 0):+[]+readCounter = 0+readDiscarded = 8+writeCounter = 0+sizeLimit = 0++write rq 32 (limit is 0):+[]+readCounter = 0+readDiscarded = 9+writeCounter = 0+sizeLimit = 0++write rq 33 (limit is 0):+[]+readCounter = 0+readDiscarded = 10+writeCounter = 0+sizeLimit = 0++write rq 34 (limit is 0):+[]+readCounter = 0+readDiscarded = 11+writeCounter = 0+sizeLimit = 0++write rq 35 (limit is 0):+[]+readCounter = 0+readDiscarded = 12+writeCounter = 0+sizeLimit = 0++read rq, isEmpty:+[]+readCounter = 0+readDiscarded = 12+writeCounter = 0+sizeLimit = 0++setLimit rq 1:+[]+readCounter = 0+readDiscarded = 12+writeCounter = 0+sizeLimit = 1++write rq 36:+[36]+readCounter = 0+readDiscarded = 12+writeCounter = 1+sizeLimit = 1++read rq (36,12):+[]+readCounter = 1+readDiscarded = 0+writeCounter = 1+sizeLimit = 1++write rq 37:+[37]+readCounter = 0+readDiscarded = 0+writeCounter = 1+sizeLimit = 1++read rq (37,0):+[]+readCounter = 1+readDiscarded = 0+writeCounter = 1+sizeLimit = 1++write rq 38:+[38]+readCounter = 0+readDiscarded = 0+writeCounter = 1+sizeLimit = 1++write rq 39:+[39]+readCounter = 0+readDiscarded = 1+writeCounter = 1+sizeLimit = 1++read rq (39,1):+[]+readCounter = 1+readDiscarded = 0+writeCounter = 1+sizeLimit = 1++read rq, isEmpty:+[]+readCounter = 1+readDiscarded = 0+writeCounter = 1+sizeLimit = 1+
+ testing/trivial.hs view
@@ -0,0 +1,75 @@+import Prelude hiding (read)++import Data.STM.RollingQueue (RollingQueue)+import qualified Data.STM.RollingQueue as RQ++import Control.Concurrent.STM+import Control.Monad++dump :: Show a => RollingQueue a -> String -> IO ()+dump rq label = do+ putStrLn $ label ++ ":"+ RQ.dump rq+ putStrLn ""++main :: IO ()+main = do+ rq <- RQ.newIO 5 :: IO (RollingQueue Int)+ dump rq "newIO 5"++ let empty = do+ atomically $ do+ Nothing <- RQ.tryRead rq+ True <- RQ.isEmpty rq+ 0 <- RQ.length rq+ return ()+ dump rq "read rq, isEmpty"++ write x = do+ atomically $ do+ RQ.write rq x+ False <- RQ.isEmpty rq+ return ()+ dump rq $ "write rq " ++ show x++ read = do+ p <- atomically $ do+ False <- RQ.isEmpty rq+ RQ.read rq+ dump rq $ "read rq " ++ show p++ setLimit n = do+ atomically $ RQ.setLimit rq n+ dump rq $ "setLimit rq " ++ show n++ empty+ mapM_ write [1..10]+ replicateM_ 5 read+ empty+ mapM_ write [11..16]+ read+ 4 <- atomically $ RQ.length rq++ 5 <- atomically $ RQ.getLimit rq+ setLimit 3+ setLimit 10+ mapM_ write [17..30]+ replicateM_ 3 read+ setLimit (-5)+ 0 <- atomically $ RQ.getLimit rq++ forM_ [31..35] $ \i -> do+ atomically $ RQ.write rq i+ True <- atomically $ RQ.isEmpty rq+ dump rq $ "write rq " ++ show i ++ " (limit is 0)"+ empty++ setLimit 1+ write 36+ read+ write 37+ read+ write 38+ write 39+ read+ empty