concurrent-batch (empty) → 0.1.0.0

raw patch · 5 files changed

+207/−0 lines, 5 filesdep +basedep +clockdep +stmsetup-changed

Dependencies added: base, clock, stm

Files

• LICENSE +30/−0
• README.md +26/−0
• Setup.hs +2/−0
• concurrent-batch.cabal +39/−0
• src/Control/Concurrent/STM/Batch.hs +110/−0

@@ -0,0 +1,30 @@+Copyright Harpo Roeder (c) 2018++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 Author name here 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.

@@ -0,0 +1,26 @@+# Concurrent STM Batch++This package facilitates batch processing based on STM. Batches are both sized, and have automatic timeout functionality.++## Example Usage++The batch handler lives inside STM. Batches should not be processed at this stage but instead pushed somewhere else for processing. In this simple example we store the results in a TMVar and process them with ```putStrLn```. Requiring the initial handler to be in STM instead of IO increases async exception safety without having to mask on a potentially blocked action.++```+import Control.Concurrent.STM+import Control.Concurrent.STM.TMVar+import Control.Concurrent.STM.Batch++main :: IO ()+main = do+  -- Create variable that batches are pushed to+  output <- newEmptyTMVarIO+  -- Create batches of 3 elements with a timeout of 10 seconds+  batcher <- newBatch 3 (Just $ fromSecs 10) (putTMVar output)+  -- Write 3 items to the batcher+  mapM_ (writeBatch batcher) [1 2 3]+  -- Get the first result batch+  batch1 <- atomically $ takeTMVar output+  -- See [3 2 1]+  putStrLn batch1+```

@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain

@@ -0,0 +1,39 @@+-- This file has been generated from package.yaml by hpack version 0.28.2.+--+-- see: https://github.com/sol/hpack+--+-- hash: d746d6a6da591ee47cf9107e29d62627d0aeee09f3fe8c95ef8c1d2e5d79d4e3++name:           concurrent-batch+version:        0.1.0.0+synopsis:       Concurrent batching queue based on STM with timeout.+description:    Please see the README on GitHub at <https://github.com/harporoeder/concurrent-batch#readme>+category:       Data+homepage:       https://github.com/harporoeder/concurrent-batch#readme+bug-reports:    https://github.com/harporoeder/concurrent-batch/issues+author:         Harpo Roeder+maintainer:     roederharpo@protonmail.ch+copyright:      2018 Harpo Roeder+license:        BSD3+license-file:   LICENSE+build-type:     Simple+cabal-version:  >= 1.10+extra-source-files:+    README.md++source-repository head+  type: git+  location: https://github.com/harporoeder/concurrent-batch++library+  exposed-modules:+      Control.Concurrent.STM.Batch+  other-modules:+      Paths_concurrent_batch+  hs-source-dirs:+      src+  build-depends:+      base >=4.7 && <5+    , clock+    , stm+  default-language: Haskell2010

@@ -0,0 +1,110 @@+module Control.Concurrent.STM.Batch+  ( Batch+    -- * Batch Operations+  , newBatch+  , writeBatch+  , flushBatch+    -- * Time Utilities+  , fromMilliSecs+  , fromSecs+  , fromMicroSecs+    -- * Re-exports+  , TimeSpec(..)+  ) where++import Data.Maybe (isJust, fromJust)+import System.Clock+import Control.Concurrent (forkIO, threadDelay)+import Control.Monad (void, when, forever, unless)+import Control.Concurrent.STM+import Control.Concurrent.STM.TVar+import Control.Concurrent.STM.TMVar++-- | Opaque batch with buffer and settings.+data Batch a = Batch+  { batchAcc     :: TVar [a]+  , batchLength  :: TVar Int+  , batchLimit   :: Int+  , batchTimeout :: Maybe TimeSpec+  , batchStarted :: TMVar TimeSpec+  , batchHandler :: [a] -> STM ()+  }++-- | Constructs a new batcher state. If a batch timeout is configured this+-- operation will automatically spawn a timeout handler thread. The timeout+-- handler will automatically be killed when the batcher is garbage collected.++newBatch ::+     Int             -- ^ Max items in a batch+  -> Maybe TimeSpec  -- ^ Batch timeout+  -> ([a] -> STM ()) -- ^ Handler for complete batch+  -> IO (Batch a)    -- ^ Batch with settings++newBatch batchLimit' batchTimeout' batchHandler' = do+  batchLength'  <- newTVarIO 0+  batchAcc'     <- newTVarIO []+  batchStarted' <- newEmptyTMVarIO++  let+    batch = Batch+      { batchAcc     = batchAcc'+      , batchLength  = batchLength'+      , batchLimit   = batchLimit'+      , batchTimeout = batchTimeout'+      , batchStarted = batchStarted'+      , batchHandler = batchHandler'+      }++  when (isJust batchTimeout') $ void $ forkIO $ timeoutHandler batch++  return batch++-- | Fires the batchHandler for the current batch from the current thread.+-- This function is automatically called for a timeout or when buffer is filled+-- by a write operation.+flushBatch :: Batch a -> STM ()+flushBatch ctx = do+  acc <- readTVar $ batchAcc ctx+  when (not $ null acc) $ batchHandler ctx acc+  void $ takeTMVar $ batchStarted ctx+  writeTVar (batchAcc ctx) []+  writeTVar (batchLength ctx) 0++-- | Add a single item to the batch. The batch is automatically flushed when full.+writeBatch :: Batch a -> a -> IO ()+writeBatch ctx item = do+  batchInitial <- atomically $ do+    modifyTVar' (batchAcc ctx) (item :)+    modifyTVar' (batchLength ctx) (+ 1)+    len <- readTVar $ batchLength ctx+    unless (len < batchLimit ctx) $ flushBatch ctx+    return $ len == 1++  when (batchInitial && batchLimit ctx > 1) $ do+    now <- getTime Monotonic+    atomically $ putTMVar (batchStarted ctx) now++timeoutHandler :: Batch a -> IO ()+timeoutHandler ctx = let timeout = fromJust (batchTimeout ctx) in forever $ do+  now <- getTime Monotonic+  started <- atomically $ tryReadTMVar $ batchStarted ctx+  case started of+    Nothing -> threadDelay $ fromIntegral $ toMicroSecs now+    Just t  -> if now - t < timeout+      then threadDelay $ fromIntegral $ toMicroSecs $ timeout + t - now+      else atomically $ flushBatch ctx++-- | Convenience function for timeout in milliseconds.+fromMilliSecs :: Integer -> TimeSpec+fromMilliSecs ts = fromNanoSecs $ 1000000 * ts++-- | Convenience function for timeout in seconds.+fromSecs :: Integer -> TimeSpec+fromSecs ts = TimeSpec (fromIntegral ts) 0++-- | Highest resolution time supported by internal usage of @threadDelay@.+fromMicroSecs :: Integer -> TimeSpec+fromMicroSecs ts = fromNanoSecs $ 1000 * ts++toMicroSecs :: TimeSpec -> Integer+toMicroSecs ts = 1000 `quot` toNanoSecs ts