immortal-queue 0.1.0.1 → 0.2.0.0

raw patch · 6 files changed

+111/−50 lines, 6 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• CHANGELOG.md +8/−0
• README.md +1/−1
• immortal-queue.cabal +8/−8
• src/Control/Immortal/Queue.hs +35/−18
• tests/MockQueue.hs +32/−11
• tests/Spec.hs +27/−12

@@ -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

@@ -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,

@@ -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

@@ -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

@@ -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

@@ -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"]