diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,5 @@ +# CHANGELOG + +## v0.1.0.0 + +* Initial release diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -0,0 +1,56 @@ +# Immortal Queue + +[![immortal-queue Build Status](https://travis-ci.org/prikhi/immortal-queue.svg?branch=master)](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 diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/immortal-queue.cabal b/immortal-queue.cabal new file mode 100644 --- /dev/null +++ b/immortal-queue.cabal @@ -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 + . +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 diff --git a/src/Control/Immortal/Queue.hs b/src/Control/Immortal/Queue.hs new file mode 100644 --- /dev/null +++ b/src/Control/Immortal/Queue.hs @@ -0,0 +1,221 @@ +{-# LANGUAGE RankNTypes #-} +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE ScopedTypeVariables #-} +{- | This module uses the +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) diff --git a/tests/MockQueue.hs b/tests/MockQueue.hs new file mode 100644 --- /dev/null +++ b/tests/MockQueue.hs @@ -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 diff --git a/tests/Spec.hs b/tests/Spec.hs new file mode 100644 --- /dev/null +++ b/tests/Spec.hs @@ -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 @?= []