immortal-queue (empty) → 0.1.0.0
raw patch · 8 files changed
+492/−0 lines, 8 filesdep +asyncdep +basedep +immortalsetup-changed
Dependencies added: async, base, immortal, immortal-queue, stm, tasty, tasty-hunit
Files
- CHANGELOG.md +5/−0
- LICENSE +30/−0
- README.md +56/−0
- Setup.hs +2/−0
- immortal-queue.cabal +70/−0
- src/Control/Immortal/Queue.hs +221/−0
- tests/MockQueue.hs +60/−0
- tests/Spec.hs +48/−0
+ CHANGELOG.md view
@@ -0,0 +1,5 @@+# CHANGELOG++## v0.1.0.0++* Initial release
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright Pavan Rikhi (c) 2020++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 Pavan Rikhi 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,56 @@+# Immortal Queue++[](https://travis-ci.org/prikhi/immortal-queue)+++A Haskell library for building a pool of queue-processing worker threads,+leveraging the [immortal][immortal] package.+++## Usage++To use this library, build an `ImmortalQueue` value describing how to+manipulate and process your queue. Then you start start the pool using the+`processImmortalQueue` function and close or kill it with `closeImmortalQueue`+or `killImmortalQueue`.++For a simple example using a `TQueue`, please refer to the [haddock+documentation][hackage] for the module.++For a more complex example that uses a persistent database as a queue backend,+see the [Southern Exposure Seed Exchange's Workers module][sese-workers].+++## Developing++You can build the project with stack:++```+stack build+```++For development, you can enable fast builds with file-watching,+documentation-building, & test-running:+```+stack test --haddock --fast --file-watch --pedantic+````++To build & open the documentation, run:++```+stack haddock --open immortal-queue+````+++## LICENSE++BSD-3++The original code for this package was lifted from [Southern Exposure Seed+Exchange's website][sese].+++[hackage]: https://hackage.haskell.org/package/immortal-queue/docs/Control-Immortal-Queue.html+[sese-workers]: https://github.com/Southern-Exposure-Seed-Exchange/southernexposure.com/blob/develop/server/src/Workers.hs+[immortal]: https://hackage.haskell.org/package/immortal+[sese]: https://github.com/Southern-Exposure-Seed-Exchange/southernexposure.com
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ immortal-queue.cabal view
@@ -0,0 +1,70 @@+cabal-version: 1.12++-- This file has been generated from package.yaml by hpack version 0.33.0.+--+-- see: https://github.com/sol/hpack+--+-- hash: 35b56f6e700d6af7666256f0b7ee41af2dd8046088935e24944bd55310342d3a++name: immortal-queue+version: 0.1.0.0+synopsis: Build a pool of queue-processing worker threads.+description: @immortal-queue@ is a library for build an asynchronous worker pool that+ processes action from a generic queue. You can use any thread-safe datatype+ with a push and pop like a @TQueue@ or a @persistent@ database table.+ .+ The worker pool is configured by building an @ImmortalQueue@ type, which+ describes how to push and pop from the queue as well as how to process+ items and handle errors.+ .+ For a simple usage example using a TQueue, see the module documentation.+ For a more complex example that uses a @persistent@ database as a queue,+ see+ <https://github.com/Southern-Exposure-Seed-Exchange/southernexposure.com/blob/develop/server/src/Workers.hs Southern Exposure's website code>.+category: Concurrency+homepage: https://github.com/prikhi/immortal-queue#readme+bug-reports: https://github.com/prikhi/immortal-queue/issues+author: Pavan Rikhi+maintainer: pavan.rikhi@gmail.com+copyright: 2020 Pavan Rikhi+license: BSD3+license-file: LICENSE+build-type: Simple+extra-source-files:+ README.md+ CHANGELOG.md++source-repository head+ type: git+ location: https://github.com/prikhi/immortal-queue++library+ exposed-modules:+ Control.Immortal.Queue+ other-modules:+ Paths_immortal_queue+ hs-source-dirs:+ src+ ghc-options: -Wall -Wcompat -Wincomplete-record-updates -Wincomplete-uni-patterns -Wredundant-constraints -O2+ build-depends:+ async >=2 && <3+ , base >=4.7 && <5+ , immortal <1 && >=0.2+ default-language: Haskell2010++test-suite immortal-queue-test+ type: exitcode-stdio-1.0+ main-is: Spec.hs+ other-modules:+ MockQueue+ Paths_immortal_queue+ hs-source-dirs:+ tests+ ghc-options: -Wall -Wcompat -Wincomplete-record-updates -Wincomplete-uni-patterns -Wredundant-constraints -O2 -threaded -rtsopts -with-rtsopts "-N -T"+ build-depends:+ base >=4.7 && <5+ , immortal-queue+ , stm+ , tasty+ , tasty-hunit+ default-language: Haskell2010
+ src/Control/Immortal/Queue.hs view
@@ -0,0 +1,221 @@+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE ScopedTypeVariables #-}+{- | This module uses the <https://hackage.haskell.org/package/immortal immortal>+library to build a pool of worker threads that process a queue of tasks+asynchronously.++First build an 'ImmortalQueue' for your task type and queue backend. Then+you can launch the pool using 'processImmortalQueue' and stop the pool with+'closeImmortalQueue'.++> import Control.Concurrent.STM (atomically)+> import Control.Concurrent.STM.TQueue+> import Control.Exception (Exception)+> import Control.Immortal.Queue+>+> data Task+> = Print String+> deriving (Show)+>+> queueConfig :: TQueue Task -> ImmortalQueue Task+> queueConfig queue =+> ImmortalQueue+> { qThreadCount = 2+> , qPollWorkerTime = 1000+> , qPop = atomically $ readTQueue queue+> , qPush = atomically . writeTQueue queue+> , qHandler = performTask+> , qFailure = printError+> }+> where+> performTask :: Task -> IO ()+> performTask t = case t of+> Print str ->+> putStrLn str+> printError :: Exception e => Task -> e -> IO ()+> printError t err =+> let description = case t of+> Print str ->+> "print"+> in putStrLn $ "Task `" ++ description ++ "` failed with: " ++ show err+>+> main :: IO ()+> main = do+> queue <- newTQueueIO+> workers <- processImmortalQueue $ queueConfig queue+> atomically $ mapM_ (writeTQueue queue . Print) ["hello", "world"]+> closeImmortalQueue workers++-}+module Control.Immortal.Queue+ ( -- * Config+ ImmortalQueue(..)+ -- * Run+ , processImmortalQueue+ , QueueId+ -- * Stop+ , closeImmortalQueue+ , killImmortalQueue+ )+where++import Control.Concurrent ( MVar+ , newEmptyMVar+ , takeMVar+ , putMVar+ , threadDelay+ , readMVar+ , tryPutMVar+ , tryTakeMVar+ )+import Control.Concurrent.Async ( Async+ , async+ , wait+ , race+ , cancel+ )+import Control.Exception ( Exception )+import Control.Immortal ( Thread )+import Control.Monad ( (>=>)+ , forever+ , void+ )+import Numeric.Natural ( Natural )++import qualified Control.Immortal as Immortal+++-- | The configuration data required for initializing a worker pool.+data ImmortalQueue a =+ ImmortalQueue+ { qThreadCount :: Natural+ -- ^ Number of worker threads to run.+ , qPollWorkerTime :: Int+ -- ^ Wait time in milliseconds for polling for a free worker.+ , qPop :: IO a+ -- ^ A blocking action to pop the next item off of the queue.+ , qPush :: a -> IO ()+ -- ^ An action to enqueue a task. Used during shutdown if we've+ -- popped an item but haven't assigned it to a worker yet.+ , qHandler :: a -> IO ()+ -- ^ The handler to perform a queued task.+ , qFailure :: forall e. Exception e => a -> e -> IO ()+ -- ^ An error handler for when a thread encounters an unhandled+ -- exception.+ }++-- | Start a management thread that creates the queue-processing worker+-- threads & return a QueueId that can be used to stop the workers.+processImmortalQueue :: forall a . ImmortalQueue a -> IO QueueId+processImmortalQueue queue = do+ shutdown <- newEmptyMVar+ asyncQueue <- async $ do+ threads <- mapM (const makeWorker) [1 .. qThreadCount queue]+ nextAction <- newEmptyMVar+ asyncQueuePopper <- async $ popQueue nextAction threads+ cleanClose <- takeMVar shutdown+ cancel asyncQueuePopper+ if cleanClose+ then do+ mapM_ (Immortal.mortalize . wdThread) threads+ mapM_ (flip putMVar () . wdCloseMVar) threads+ else mapM_ (Immortal.stop . wdThread) threads+ mapM_ (Immortal.wait . wdThread) threads+ tryTakeMVar nextAction >>= \case+ Nothing -> return ()+ Just action -> qPush queue action+ return QueueId { qiCloseCleanly = shutdown, qiAsyncQueue = asyncQueue }+ where+ -- Create the communication MVars for a worker & then start the+ -- Immortal thread.+ makeWorker :: IO (WorkerData a)+ makeWorker = do+ closeMVar <- newEmptyMVar+ inputMVar <- newEmptyMVar+ let makeData thread = WorkerData { wdCloseMVar = closeMVar+ , wdInputMVar = inputMVar+ , wdThread = thread+ }+ makeData <$> Immortal.create (processAction . makeData)++ -- Wait for an input action or the close signal, then process the input+ -- or close by returning.+ processAction :: WorkerData a -> IO ()+ processAction workerData = do+ actionOrClose <- race (takeMVar $ wdCloseMVar workerData)+ (readMVar $ wdInputMVar workerData)+ case actionOrClose of+ Left () -> return ()+ Right action ->+ let flushInput = void $ takeMVar $ wdInputMVar workerData+ finalize = errorHandler action >=> const flushInput+ in Immortal.onFinish finalize $ qHandler queue action++ -- Ignore successful results & run the error handler on any exceptions.+ errorHandler :: Exception e => a -> Either e () -> IO ()+ errorHandler action = either (qFailure queue action) (const $ return ())++ -- Pop an item from the queue, put it in the enqueued action MVar,+ -- assign it to a worker, & then remove it as the currently enqueued+ -- action.+ popQueue :: MVar a -> [WorkerData a] -> IO ()+ popQueue actionMVar workers = forever $ do+ nextAction <- qPop queue+ putMVar actionMVar nextAction+ assignWork workers nextAction+ void $ takeMVar actionMVar++ -- Assign the action to the first free worker, retrying until a worker+ -- is available.+ assignWork :: [WorkerData a] -> a -> IO ()+ assignWork workers action = do+ assigned <- assignToFreeWorker action workers+ if assigned+ then return ()+ else+ threadDelay (1000 * qPollWorkerTime queue)+ >> assignWork workers action++ -- Iterate through the workers, trying to assign the action & returning+ -- whether assigning was successful or not.+ assignToFreeWorker :: a -> [WorkerData a] -> IO Bool+ assignToFreeWorker action = \case+ [] -> return False+ worker : ws -> do+ successfulPut <- tryPutMVar (wdInputMVar worker) action+ if successfulPut then return True else assignToFreeWorker action ws+++data WorkerData a =+ WorkerData+ { wdCloseMVar :: MVar ()+ -- ^ Signal to shut down the worker+ , wdInputMVar :: MVar a+ -- ^ The action to process+ , wdThread :: Thread+ -- ^ The "Immortal" thread for the worker+ }+++-- | An identifier created by a queue manager that can be used to stop the+-- worker processes.+data QueueId =+ QueueId+ { qiCloseCleanly :: MVar Bool+ -- ^ Signal to close the queue cleanly or immediately.+ , qiAsyncQueue :: Async ()+ -- ^ The pool's management thread.+ }+++-- | Cleanly close the worker pool, allowing them to complete their+-- actions.+closeImmortalQueue :: QueueId -> IO ()+closeImmortalQueue queueId =+ putMVar (qiCloseCleanly queueId) True >> wait (qiAsyncQueue queueId)++-- | Uncleanly close the worker pool, aborting current actions.+killImmortalQueue :: QueueId -> IO ()+killImmortalQueue queueId =+ putMVar (qiCloseCleanly queueId) False >> wait (qiAsyncQueue queueId)
+ tests/MockQueue.hs view
@@ -0,0 +1,60 @@+{-# LANGUAGE LambdaCase #-}+module MockQueue where++import Control.Concurrent ( threadDelay )+import Control.Concurrent.STM+import Control.Immortal.Queue+++data Task+ = Log Integer Int+ -- ^ Always succeeds with Integer after Int milliseconds+ | Fail String+ -- ^ Always fails++queueConfig :: TVar ([Integer], [String]) -> TQueue Task -> ImmortalQueue Task+queueConfig output q = ImmortalQueue { qThreadCount = 2+ , qPollWorkerTime = 200+ , qPop = atomically $ readTQueue q+ , qPush = atomically . writeTQueue q+ , qHandler = performTask+ , qFailure = handleError+ }+ where+ performTask = \case+ Log i t -> threadDelay (1000 * t) >> atomically (addSuccess i)+ Fail _ -> error "failed"++ handleError t _ = atomically $ case t of+ Log _ _ -> return ()+ Fail s -> addFailure s++ addSuccess i = modifyTVar output $ \(s, f) -> (s ++ [i], f)+ addFailure m = modifyTVar output $ \(s, f) -> (s, f ++ [m])+++-- | Run pool that processes all the given tasks, splitting successes and+-- failures.+runPool :: [Task] -> IO ([Integer], [String])+runPool = runPool_ True Nothing+++-- | Run a pool that processes the given tasks. `cleanClose` indicates if the+-- pool should be closed cleanly and `waitTime` will wait the specified+-- time before closing/killing or wait until the queue is empty if Nothing.+runPool_ :: Bool -> Maybe Int -> [Task] -> IO ([Integer], [String])+runPool_ cleanClose waitTime tasks = do+ output <- newTVarIO ([], [])+ tqueue <- newTQueueIO+ workers <- processImmortalQueue $ queueConfig output tqueue+ atomically $ mapM_ (writeTQueue tqueue) tasks+ maybe (waitEmpty tqueue) (threadDelay . (* 1000)) waitTime+ if cleanClose then closeImmortalQueue workers else killImmortalQueue workers+ readTVarIO output+++-- | Wait until the given queue is empty+waitEmpty :: TQueue a -> IO ()+waitEmpty q = do+ isEmpty <- atomically $ isEmptyTQueue q+ if isEmpty then return () else threadDelay 1000000 >> waitEmpty q
+ tests/Spec.hs view
@@ -0,0 +1,48 @@+import Test.Tasty+import Test.Tasty.HUnit++import MockQueue+++main :: IO ()+main = defaultMain tests+++tests :: TestTree+tests = testGroup+ "Immortal Queue Tests"+ [ testCase "All Succeed" allSuccess+ , testCase "Error Handling" errorHandling+ , testCase "Close Waits For Workers" closeWaits+ , testCase "Kill Stops Workers Immediately" killExitsEarly+ ]+ where+ allSuccess :: Assertion+ allSuccess = do+ (successes, failures) <- runPool+ [Log 1 100, Log 2 200, Log 3 300, Log 4 400, Log 5 500]+ successes @?= [1, 2, 3, 4, 5]+ failures @?= []++ errorHandling :: Assertion+ errorHandling = do+ (successes, failures) <- runPool+ [Log 1 0, Fail "hello", Log 2 0, Fail "world", Fail "9001"]+ successes @?= [1, 2]+ failures @?= ["hello", "world", "9001"]++ closeWaits :: Assertion+ closeWaits = do+ (successes, failures) <- runPool_ True+ (Just 500)+ [Log 1 0, Log 2 200, Log 3 2000]+ successes @?= [1, 2, 3]+ failures @?= []++ killExitsEarly :: Assertion+ killExitsEarly = do+ (successes, failures) <- runPool_ False+ (Just 500)+ [Log 1 0, Log 2 200, Log 3 3000]+ successes @?= [1, 2]+ failures @?= []