batch (empty) → 0.1.0.0

raw patch · 7 files changed

+300/−0 lines, 7 filesdep +asyncdep +basedep +batchsetup-changed

Dependencies added: async, base, batch, hspec, lifted-async, lifted-base, monad-control, mtl, stm, timespan, transformers-base

Files

ChangeLog.md +7/−0
LICENSE +30/−0
README.md +31/−0
Setup.hs +2/−0
batch.cabal +64/−0
src/Control/Batch.hs +102/−0
test/Spec.hs +64/−0

+ ChangeLog.md view

@@ -0,0 +1,7 @@+# Changelog for batch++## 0.1.0.0++* First release++## Unreleased changes

+ LICENSE view

@@ -0,0 +1,30 @@+Copyright Alexander Thiemann (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 Alexander Thiemann 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.

+ README.md view

@@ -0,0 +1,31 @@+# batch++[![CircleCI](https://circleci.com/gh/agrafix/batch.svg?style=svg)](https://circleci.com/gh/agrafix/batch)++Simplify queuing up data and processing it in batch.++```haskell+import Control.Batch+import Control.Concurrent.STM+import Control.Monad++example :: IO ()+example =+  do outVar <- atomically $ newTVar []+     let cfg =+             Batch+             { b_runEveryItems = Just 5+             , b_runAfterTimeout = Nothing+             , b_maxQueueLength = Nothing+             , b_runBatch =+                     \x -> atomically $ modifyTVar' outVar (++x)+             }+     withBatchRunner cfg $ \hdl ->+         do replicateM_ 5 $ bh_enqueue hdl True+            out <-+                atomically $+                do x <- readTVar outVar+                   when (length x /= 5) retry+                   pure x+            out `shouldBe` [True, True, True, True, True]+```

+ Setup.hs view

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

+ batch.cabal view

@@ -0,0 +1,64 @@+-- This file has been generated from package.yaml by hpack version 0.20.0.+--+-- see: https://github.com/sol/hpack+--+-- hash: d87b4be02cf9109e8c9efd7511dbf3c5f169411d9f890206d978a2de89349c21++name:           batch+version:        0.1.0.0+synopsis:       Simplify queuing up data and processing it in batch.+description:    Simplify queuing up data and processing it in batch.+category:       Data+homepage:       https://github.com/agrafix/batch#readme+bug-reports:    https://github.com/agrafix/batch/issues+author:         Alexander Thiemann+maintainer:     mail@athiemann.net+copyright:      2017 Alexander Thiemann <mail@athiemann.net>+license:        BSD3+license-file:   LICENSE+build-type:     Simple+cabal-version:  >= 1.10++extra-source-files:+    ChangeLog.md+    README.md++source-repository head+  type: git+  location: https://github.com/agrafix/batch++library+  hs-source-dirs:+      src+  ghc-options: -Wall+  build-depends:+      async >=2.0+    , base >=4.7 && <5+    , lifted-async >=0.9+    , lifted-base >=0.2.3+    , monad-control >=1.0+    , mtl >=2.2+    , stm >=2.4+    , timespan >=0.3+    , transformers-base >=0.4+  exposed-modules:+      Control.Batch+  other-modules:+      Paths_batch+  default-language: Haskell2010++test-suite batch-test+  type: exitcode-stdio-1.0+  main-is: Spec.hs+  hs-source-dirs:+      test+  ghc-options: -threaded -rtsopts -with-rtsopts=-N+  build-depends:+      base >=4.7 && <5+    , batch+    , hspec >=2.4+    , stm >=2.4+    , timespan >=0.3+  other-modules:+      Paths_batch+  default-language: Haskell2010

+ src/Control/Batch.hs view

@@ -0,0 +1,102 @@+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE StrictData #-}+module Control.Batch+    ( Batch(..)+    , BatchHandle(..)+    , withBatchRunner+    )+where++import Control.Concurrent.Async+import Control.Concurrent.STM+import Control.Exception.Lifted+import Control.Monad+import Control.Monad.Base+import Control.Monad.Trans.Control+import Data.Time.TimeSpan+import qualified Control.Concurrent.Async.Lifted as L++data Batch v m+    = Batch+    { b_runEveryItems :: Maybe Int+      -- ^ flush in memory items ever N items for processing+    , b_runAfterTimeout :: Maybe TimeSpan+      -- ^ flush after item stuck in queue for longer than this+    , b_maxQueueLength :: Maybe Int+      -- ^ if more than N items are enqueued, block enqueueing+    , b_runBatch :: [v] -> m ()+      -- ^ function describing how batches should be processed+    }++data BatchHandle v m+    = BatchHandle+    { bh_enqueue :: v -> m ()+      -- ^ enqueue a new element+    }++withBatchRunner ::+    forall v m a. MonadBaseControl IO m+    => Batch v m -> (BatchHandle v m -> m a) -> m a+withBatchRunner batch go =+    do dummy <- liftBase $ async $ pure ()+       (queueVar, sizeVar, flushThreadVar, triggerVar, stopVar) <-+           liftBase $ atomically $+           (,,,,) <$> newTQueue <*> newTVar 0 <*> newTVar dummy <*> newTVar False <*> newTVar False+       let flusher :: m ()+           flusher =+               do shouldStop <-+                      liftBase $+                      atomically $+                      do x <- readTVar triggerVar+                         stop <- readTVar stopVar+                         unless (stop || x) retry+                         writeTVar triggerVar False+                         pure stop+                  drain <-+                      liftBase $+                      atomically $+                      do let readLoop !accum =+                                 do x <- tryReadTQueue queueVar+                                    case x of+                                      Nothing -> pure (reverse accum)+                                      Just v -> readLoop (v : accum)+                         readLoop []+                  b_runBatch batch drain+                  unless shouldStop flusher+           flush = writeTVar triggerVar True+           enqueue item =+               liftBase $+               do case b_maxQueueLength batch of+                    Just maxLen ->+                        atomically $+                        do totalSize <- readTVar sizeVar+                           when (totalSize > maxLen) retry+                    Nothing -> pure ()+                  waiter <-+                      async $+                      case b_runAfterTimeout batch of+                        Nothing -> pure ()+                        Just ts ->+                            do sleepTS ts+                               atomically flush+                  (oldFlushThread :: Async ()) <-+                      atomically $+                      do writeTQueue queueVar item+                         modifyTVar' sizeVar (+1)+                         totalSize <- readTVar sizeVar+                         oldFlush <- readTVar flushThreadVar+                         case b_runEveryItems batch of+                           Just x | totalSize >= x -> flush+                           _ -> pure ()+                         writeTVar flushThreadVar waiter+                         pure oldFlush+                  uninterruptibleCancel oldFlushThread+       flushThread <- L.async flusher+       let bh = BatchHandle enqueue+           cleanup =+               liftBase $+               do atomically $ writeTVar stopVar True >> flush+                  wait flushThread+       go bh `finally` cleanup

+ test/Spec.hs view

@@ -0,0 +1,64 @@+import Control.Batch+import Control.Concurrent.STM+import Control.Monad+import Data.IORef+import Data.Time.TimeSpan++import Test.Hspec+++main :: IO ()+main = hspec spec++spec :: Spec+spec =+    describe "BatchJobs" $+    do it "batch job is processed on termination" $+           do outVar <- newIORef []+              let cfg =+                      Batch+                      { b_runEveryItems = Nothing+                      , b_runAfterTimeout = Nothing+                      , b_maxQueueLength = Nothing+                      , b_runBatch = \x -> writeIORef outVar x+                      }+              withBatchRunner cfg $ \hdl ->+                  do bh_enqueue hdl True+                     bh_enqueue hdl False+              readIORef outVar `shouldReturn` [True, False]+       it "batch jobs processes when N items reached" $+           do outVar <- atomically $ newTVar []+              let cfg =+                      Batch+                      { b_runEveryItems = Just 5+                      , b_runAfterTimeout = Nothing+                      , b_maxQueueLength = Nothing+                      , b_runBatch =+                              \x -> atomically $ modifyTVar' outVar (++x)+                      }+              withBatchRunner cfg $ \hdl ->+                  do replicateM_ 5 $ bh_enqueue hdl True+                     out <-+                         atomically $+                         do x <- readTVar outVar+                            when (length x /= 5) retry+                            pure x+                     out `shouldBe` [True, True, True, True, True]+       it "batch jobs processes when timeout reached" $+           do outVar <- atomically $ newTVar []+              let cfg =+                      Batch+                      { b_runEveryItems = Nothing+                      , b_runAfterTimeout = Just (seconds 1)+                      , b_maxQueueLength = Nothing+                      , b_runBatch =+                              \x -> atomically $ modifyTVar' outVar (++x)+                      }+              withBatchRunner cfg $ \hdl ->+                  do replicateM_ 5 $ bh_enqueue hdl True+                     out <-+                         atomically $+                         do x <- readTVar outVar+                            when (length x /= 5) retry+                            pure x+                     out `shouldBe` [True, True, True, True, True]