diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,13 @@ # CHANGELOG +## master + + +# v0.2.0.0 + +* Catch errors thrown during popping, add new `qPopFailure` config field to + handle them and then restart the item popper. + ## v0.1.0.1 * Fix lower-bound on the `immortal` package diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # Immortal Queue -[![immortal-queue Build Status](https://travis-ci.org/prikhi/immortal-queue.svg?branch=master)](https://travis-ci.org/prikhi/immortal-queue) +[![immortal-queue Build Status](https://github.com/prikhi/immortal-queue/actions/workflows/main.yml/badge.svg?branch=master)](https://github.com/prikhi/immortal-queue/actions/workflows/main.yml) A Haskell library for building a pool of queue-processing worker threads, diff --git a/immortal-queue.cabal b/immortal-queue.cabal --- a/immortal-queue.cabal +++ b/immortal-queue.cabal @@ -1,13 +1,13 @@ cabal-version: 1.12 --- This file has been generated from package.yaml by hpack version 0.33.0. +-- This file has been generated from package.yaml by hpack version 0.38.1. -- -- see: https://github.com/sol/hpack -- --- hash: 8c517de3588e995edfea529de9301f2fd844e674f8016c429fba633de9f4c864 +-- hash: 39cceb7713d23b85fd53f67a2cf6156f99d4691c2a3deed4a56873542beff9eb name: immortal-queue -version: 0.1.0.1 +version: 0.2.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 @@ -26,7 +26,7 @@ bug-reports: https://github.com/prikhi/immortal-queue/issues author: Pavan Rikhi maintainer: pavan.rikhi@gmail.com -copyright: 2020 Pavan Rikhi +copyright: 2020-2025 Pavan Rikhi license: BSD3 license-file: LICENSE build-type: Simple @@ -47,17 +47,17 @@ src ghc-options: -Wall -O2 build-depends: - async >=2 && <3 + async ==2.* , base >=4.7 && <5 , immortal <1 && >=0.2.1 + default-language: Haskell2010 if impl(ghc >= 8.0) ghc-options: -Wcompat -Wincomplete-record-updates -Wincomplete-uni-patterns -Wredundant-constraints else ghc-options: -fwarn-incomplete-record-updates -fwarn-incomplete-uni-patterns if impl(ghc < 8.0) build-depends: - nats >=1 && <2 - default-language: Haskell2010 + nats ==1.* test-suite immortal-queue-test type: exitcode-stdio-1.0 @@ -74,8 +74,8 @@ , stm , tasty , tasty-hunit + default-language: Haskell2010 if impl(ghc >= 8.0) ghc-options: -Wcompat -Wincomplete-record-updates -Wincomplete-uni-patterns -Wredundant-constraints else ghc-options: -fwarn-incomplete-record-updates -fwarn-incomplete-uni-patterns - default-language: Haskell2010 diff --git a/src/Control/Immortal/Queue.hs b/src/Control/Immortal/Queue.hs --- a/src/Control/Immortal/Queue.hs +++ b/src/Control/Immortal/Queue.hs @@ -27,6 +27,7 @@ > , qPush = atomically . writeTQueue queue > , qHandler = performTask > , qFailure = printError +> , qPopFailure = \e -> putStrLn $ "Unexpected pop failure: " <> show e > } > where > performTask :: Task -> IO () @@ -72,10 +73,13 @@ import Control.Concurrent.Async ( Async , async , wait + , waitCatch , race , cancel ) -import Control.Exception ( Exception ) +import Control.Exception ( Exception + , SomeException + ) import Control.Immortal ( Thread ) import Control.Monad ( (>=>) , forever @@ -101,8 +105,12 @@ , 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. + -- ^ An error handler for when a processing thread encounters an + -- unhandled exception. + , qPopFailure :: SomeException -> IO () + -- ^ An error handler for when the queue popping thread encounters + -- an unhandled exception. The popping thread will be restarted + -- after this is run. } -- | Start a management thread that creates the queue-processing worker @@ -110,21 +118,30 @@ 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 + threads <- mapM (const makeWorker) [1 .. qThreadCount queue] + nextAction <- newEmptyMVar + let manageQueuePopper = do + asyncQueuePopper <- async $ popQueue nextAction threads + finishAction <- takeMVar shutdown `race` waitCatch asyncQueuePopper + case finishAction of + Left cleanClose -> do + 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 + Right (Left e) -> do + qPopFailure queue e + manageQueuePopper + Right (Right ()) -> do + manageQueuePopper + asyncQueue <- async manageQueuePopper + return QueueId { qiCloseCleanly = shutdown, qiAsyncQueue = asyncQueue } where -- Create the communication MVars for a worker & then start the diff --git a/tests/MockQueue.hs b/tests/MockQueue.hs --- a/tests/MockQueue.hs +++ b/tests/MockQueue.hs @@ -3,7 +3,9 @@ import Control.Concurrent ( threadDelay ) import Control.Concurrent.STM +import Control.Exception import Control.Immortal.Queue +import Data.Maybe data Task @@ -11,40 +13,59 @@ -- ^ Always succeeds with Integer after Int milliseconds | Fail String -- ^ Always fails + | PopFail String + -- ^ Throws error in `qPop` -queueConfig :: TVar ([Integer], [String]) -> TQueue Task -> ImmortalQueue Task +data PopFailException + = PopE String + deriving (Show, Eq) +instance Exception PopFailException + +queueConfig :: TVar ([Integer], [String], [PopFailException]) -> TQueue Task -> ImmortalQueue Task queueConfig output q = ImmortalQueue { qThreadCount = 2 , qPollWorkerTime = 200 - , qPop = atomically $ readTQueue q + , qPop = popQueue , qPush = atomically . writeTQueue q , qHandler = performTask , qFailure = handleError + , qPopFailure = atomically . addPopFailure } where performTask = \case - Log i t -> threadDelay (1000 * t) >> atomically (addSuccess i) - Fail _ -> error "failed" + Log i t -> threadDelay (1000 * t) >> atomically (addSuccess i) + Fail _ -> error "failed" + PopFail _ -> return () handleError t _ = atomically $ case t of - Log _ _ -> return () - Fail s -> addFailure s + Log _ _ -> return () + Fail s -> addFailure s + PopFail _ -> return () - addSuccess i = modifyTVar output $ \(s, f) -> (s ++ [i], f) - addFailure m = modifyTVar output $ \(s, f) -> (s, f ++ [m]) + popQueue = do + task <- atomically (readTQueue q) + case task of + PopFail errMsg -> + throw $ PopE errMsg + _ -> + return task + addSuccess i = modifyTVar output $ \(s, f, p) -> (s ++ [i], f, p) + addFailure m = modifyTVar output $ \(s, f, p) -> (s, f ++ [m], p) + addPopFailure e = modifyTVar output $ \(s, f, p) -> (s, f, p ++ maybeToList (fromException e)) + -- | Run pool that processes all the given tasks, splitting successes and -- failures. -runPool :: [Task] -> IO ([Integer], [String]) +runPool :: [Task] -> IO ([Integer], [String], [PopFailException]) 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_ :: Bool -> Maybe Int -> [Task] -> IO ([Integer], [String], [PopFailException]) runPool_ cleanClose waitTime tasks = do - output <- newTVarIO ([], []) + output <- newTVarIO ([], [], []) tqueue <- newTQueueIO workers <- processImmortalQueue $ queueConfig output tqueue atomically $ mapM_ (writeTQueue tqueue) tasks diff --git a/tests/Spec.hs b/tests/Spec.hs --- a/tests/Spec.hs +++ b/tests/Spec.hs @@ -11,38 +11,53 @@ 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 + [ testCase "All Succeed" allSuccess + , testCase "Error Handling" errorHandling + , testCase "Close Waits For Workers" closeWaits + , testCase "Kill Stops Workers Immediately" killExitsEarly + , testCase "Pop Failure Continues Processing" popFailureContinues ] where allSuccess :: Assertion allSuccess = do - (successes, failures) <- runPool + (successes, failures, popFailures) <- runPool [Log 1 100, Log 2 200, Log 3 300, Log 4 400, Log 5 500] successes @?= [1, 2, 3, 4, 5] failures @?= [] + popFailures @?= [] errorHandling :: Assertion errorHandling = do - (successes, failures) <- runPool + (successes, failures, popFailures) <- runPool [Log 1 0, Fail "hello", Log 2 0, Fail "world", Fail "9001"] successes @?= [1, 2] failures @?= ["hello", "world", "9001"] + popFailures @?= [] closeWaits :: Assertion closeWaits = do - (successes, failures) <- runPool_ True - (Just 500) - [Log 1 0, Log 2 200, Log 3 2000] + (successes, failures, popFailures) <- runPool_ True + (Just 500) + [Log 1 0, Log 2 200, Log 3 2000] successes @?= [1, 2, 3] failures @?= [] + popFailures @?= [] killExitsEarly :: Assertion killExitsEarly = do - (successes, failures) <- runPool_ False - (Just 500) - [Log 1 0, Log 2 200, Log 3 3000] + (successes, failures, popFailures) <- runPool_ False + (Just 500) + [Log 1 0, Log 2 200, Log 3 3000] successes @?= [1, 2] failures @?= [] + popFailures @?= [] + + popFailureContinues :: Assertion + popFailureContinues = do + (successes, failures, popFailures) <- runPool_ + True + (Just 800) + [Log 1 10, PopFail "died", Log 2 70, Log 3 100, PopFail "again", Log 4 200, Log 5 300, Log 6 400] + successes @?= [1, 2, 3, 4, 5, 6] + failures @?= [] + popFailures @?= [PopE "died", PopE "again"]