powerqueue (empty) → 0.1.0.0
raw patch · 6 files changed
+371/−0 lines, 6 filesdep +asyncdep +basedep +contravariantsetup-changed
Dependencies added: async, base, contravariant, hspec, powerqueue, stm
Files
- LICENSE +30/−0
- Setup.hs +2/−0
- powerqueue.cabal +40/−0
- src/Data/PowerQueue.hs +184/−0
- test/Data/PowerQueueSpec.hs +108/−0
- test/Spec.hs +7/−0
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright Alexander Thiemann (c) 2017++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.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ powerqueue.cabal view
@@ -0,0 +1,40 @@+name: powerqueue+version: 0.1.0.0+synopsis: A flexible job queue with exchangeable backends+description: A flexible job queue with exchangeable backends+homepage: https://github.com/agrafix/powerqueue#readme+license: BSD3+license-file: LICENSE+author: Alexander Thiemann+maintainer: mail@athiemann.net+copyright: 2017 Alexander Thiemann <mail@athiemann.net>+category: Web+build-type: Simple+cabal-version: >=1.10++library+ hs-source-dirs: src+ exposed-modules: Data.PowerQueue+ build-depends:+ base >= 4.7 && < 5+ , contravariant >= 1.4+ , async >= 2.1+ default-language: Haskell2010++test-suite powerqueue-test+ type: exitcode-stdio-1.0+ hs-source-dirs: test+ main-is: Spec.hs+ other-modules: Data.PowerQueueSpec+ build-depends:+ base+ , powerqueue+ , hspec >= 2.2+ , stm >= 2.4+ , async >= 2.1+ ghc-options: -threaded -rtsopts -with-rtsopts=-N+ default-language: Haskell2010++source-repository head+ type: git+ location: https://github.com/agrafix/powerqueue
+ src/Data/PowerQueue.hs view
@@ -0,0 +1,184 @@+{-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+module Data.PowerQueue+ ( -- * Worker descriptions+ QueueWorker, newQueueWorker, JobResult(..)+ -- * Queue control+ , Queue, newQueue, mapQueue, enqueueJob+ -- * (persistent) queue backends+ , QueueBackend(..), mapBackend, basicChanBackend+ -- * execution strategies+ , workStep+ -- ** A local worker+ , LocalWorkerConfig(..), localQueueWorker+ )+where++import Control.Concurrent.Async+import Control.Exception+import Control.Monad+import Data.Bifunctor+import Data.Functor.Contravariant+import Data.Functor.Contravariant.Divisible+import Data.IORef+import qualified Control.Concurrent.Chan as C++-- | Result of the job+data JobResult+ = JOk+ -- ^ job is complete+ | JRetry+ -- ^ job execution should be retried+ deriving (Show, Eq)++data QueueBackend j+ = forall tx m. Monad m =>+ QueueBackend+ { qb_lift :: forall a. m a -> IO a+ -- ^ lift an action from the backend monad into 'IO'+ , qb_enqueue :: j -> m Bool+ -- ^ enqueue a job+ , qb_dequeue :: m (tx, j)+ -- ^ dequeue a single job, block if no job available+ , qb_confirm :: tx -> m ()+ -- ^ mark a single job as confirmed+ , qb_rollback :: tx -> m ()+ -- ^ mark a single job as failed+ }++-- | A very basic in memory backend using only data structures from the base library.+-- It should only be used for testing and serves as an implementation example+basicChanBackend :: forall j. IO (QueueBackend j)+basicChanBackend =+ do jobChannel <- C.newChan+ inProgress <- newIORef (0, [])+ let dequeueHandler :: j -> (Int, [(Int, j)]) -> ((Int, [(Int, j)]), (Int, j))+ dequeueHandler job st@(idx, _) =+ let entry = (idx, job)+ in ( bimap (+1) (entry:) st+ , entry+ )+ forget txId =+ atomicModifyIORef' inProgress $ \st ->+ ( second (filter (\x -> fst x == txId)) st+ , snd $ second (lookup txId) st+ )+ pure+ QueueBackend+ { qb_lift = id+ , qb_enqueue = \x -> C.writeChan jobChannel x >> pure True+ , qb_dequeue =+ do nextJob <- C.readChan jobChannel+ atomicModifyIORef' inProgress (dequeueHandler nextJob)+ , qb_confirm = void . forget+ , qb_rollback =+ \txId ->+ do oldVal <- forget txId+ case oldVal of+ Just j -> C.writeChan jobChannel j+ Nothing -> pure () -- should not really happen ...+ }++mapBackend :: (a -> b) -> (b -> a) -> QueueBackend a -> QueueBackend b+mapBackend f g (QueueBackend qlift qenq qdeq qconf qroll) =+ QueueBackend+ { qb_lift = qlift+ , qb_enqueue = qenq . g+ , qb_dequeue = fmap (second f) qdeq+ , qb_confirm = qconf+ , qb_rollback = qroll+ }++data QueueWorker j+ = QueueWorker+ { qw_execute :: j -> IO JobResult+ -- ^ run a job+ }++newQueueWorker :: (j -> IO JobResult) -> QueueWorker j+newQueueWorker exec =+ QueueWorker+ { qw_execute = exec+ }++instance Contravariant QueueWorker where+ contramap f (QueueWorker qexec) =+ QueueWorker+ { qw_execute = \val -> qexec (f val)+ }++instance Divisible QueueWorker where+ divide f (QueueWorker qe1) (QueueWorker qe2) =+ QueueWorker+ { qw_execute = \val ->+ do let (l, r) = f val+ ok1 <- qe1 l+ ok2 <- qe2 r+ pure $+ if ok1 /= JOk || ok2 /= JOk+ then JRetry+ else JOk+ }+ conquer =+ newQueueWorker $ \_ -> pure JOk++data Queue j+ = Queue+ { q_worker :: !(QueueWorker j)+ , q_backend :: !(QueueBackend j)+ }++mapQueue :: (a -> b) -> (b -> a) -> Queue a -> Queue b+mapQueue f g q =+ Queue+ { q_worker = contramap g (q_worker q)+ , q_backend = mapBackend f g (q_backend q)+ }++-- | Create a new queue description+newQueue :: QueueBackend j -> QueueWorker j -> Queue j+newQueue qb qw =+ Queue+ { q_worker = qw+ , q_backend = qb+ }++-- | Add a 'Job' to the 'Queue'+enqueueJob :: j -> Queue j -> IO Bool+enqueueJob j q =+ let enqueue QueueBackend{..} = qb_lift (qb_enqueue j)+ in enqueue (q_backend q)++-- | Execute a single work step: attempt a dequeue and run the job. Use+-- to implement a queue worker, such as 'localQueueWorker'+workStep :: Queue j -> IO ()+workStep q = workStepInternal (q_backend q) (q_worker q)++workStepInternal :: QueueBackend j -> QueueWorker j -> IO ()+workStepInternal QueueBackend{..} QueueWorker{..} =+ let acquire = qb_lift qb_dequeue+ onError (txId, _) = qb_lift (qb_rollback txId)+ go (txId, job) =+ do execRes <-+ try $ qw_execute job+ case execRes of+ Left (_ :: SomeException) ->+ qb_lift (qb_rollback txId)+ Right res ->+ case res of+ JOk -> qb_lift (qb_confirm txId)+ JRetry -> qb_lift (qb_rollback txId)+ in bracketOnError acquire onError go++data LocalWorkerConfig+ = LocalWorkerConfig+ { lwc_concurrentJobs :: !Int+ }++-- | (Concurrently) run pending jobs on local machine in current process+localQueueWorker :: LocalWorkerConfig -> Queue j -> IO ()+localQueueWorker cfg q =+ replicateConcurrently_ (lwc_concurrentJobs cfg) $+ forever $ workStep q
+ test/Data/PowerQueueSpec.hs view
@@ -0,0 +1,108 @@+module Data.PowerQueueSpec+ ( spec )+where++import Data.PowerQueue++import Control.Concurrent.Async+import Control.Concurrent.STM+import Control.Monad+import Data.IORef+import Test.Hspec++dummyRefWorker :: (a -> b -> b) -> IORef b -> QueueWorker a+dummyRefWorker f ioRef =+ newQueueWorker $ \v -> atomicModifyIORef' ioRef $ \i -> (f v i, JOk)++dummyRefQueue :: (a -> b -> b) -> IORef b -> IO (Queue a)+dummyRefQueue f ioRef =+ do be <- basicChanBackend+ pure $ newQueue be (dummyRefWorker f ioRef)++dummyCounterQueue :: IORef Int -> IO (Queue ())+dummyCounterQueue = dummyRefQueue (\_ i -> i + 1)++dummyStmCounterQueue :: TVar Int -> IO (Queue ())+dummyStmCounterQueue tv =+ do be <- basicChanBackend+ pure $ newQueue be $ newQueueWorker $ \() ->+ do atomically $ modifyTVar' tv (+1)+ pure JOk++dummyReverseAccumQueue :: IORef [Int] -> IO (Queue Int)+dummyReverseAccumQueue = dummyRefQueue (:)++spec :: Spec+spec =+ do specCore+ specLocalWorker++specLocalWorker :: Spec+specLocalWorker =+ describe "localQueueWorker" $+ it "should work with 100 jobs" $+ do ref <- atomically $ newTVar 0+ queue <- dummyStmCounterQueue ref+ handler <- async $ localQueueWorker (LocalWorkerConfig 10) queue+ replicateM_ 100 $ enqueueJob () queue `shouldReturn` True+ result <-+ atomically $+ do val <- readTVar ref+ when (val /= 100) retry+ pure val+ cancel handler+ result `shouldBe` 100++specCore :: Spec+specCore =+ describe "PowerQueue Core" $+ do it "providing 100 jobs and stepping 100 times works" $+ do ref <- newIORef 0+ queue <- dummyCounterQueue ref+ replicateM_ 100 $ enqueueJob () queue `shouldReturn` True+ replicateM_ 100 $ workStep queue+ result <- readIORef ref+ result `shouldBe` 100+ it "all jobs are executed" $+ do ref <- newIORef []+ queue <- dummyReverseAccumQueue ref+ let out = [1..100]+ forM_ out $ flip enqueueJob queue+ replicateM_ 100 $ workStep queue+ result <- reverse <$> readIORef ref+ result `shouldBe` out+ it "failed jobs result in reenqueue" $+ do be <- basicChanBackend+ crashed <- newIORef False+ out <- newIORef False+ let worker :: QueueWorker ()+ worker =+ newQueueWorker $ \() ->+ do x <- readIORef crashed+ writeIORef crashed True+ when x $ writeIORef out True+ pure (if x then JOk else JRetry)+ let queue = newQueue be worker+ enqueueJob () queue `shouldReturn` True+ workStep queue+ workStep queue+ result <- readIORef out+ result `shouldBe` True+ it "unhandled crashes result in reenqueue" $+ do be <- basicChanBackend+ crashed <- newIORef False+ out <- newIORef False+ let worker :: QueueWorker ()+ worker =+ newQueueWorker $ \() ->+ do x <- readIORef crashed+ writeIORef crashed True+ unless x $ fail "OOOOPS"+ writeIORef out True+ pure JOk+ let queue = newQueue be worker+ enqueueJob () queue `shouldReturn` True+ workStep queue+ workStep queue+ result <- readIORef out+ result `shouldBe` True
+ test/Spec.hs view
@@ -0,0 +1,7 @@+import Test.Hspec+import qualified Data.PowerQueueSpec++main :: IO ()+main =+ hspec $+ Data.PowerQueueSpec.spec