consumers 2.3.0.0 → 2.3.1.0
raw patch · 5 files changed
+123/−120 lines, 5 filesdep −extradep ~containersdep ~exceptionsdep ~hpqtypesPVP: major bump suggested
API removals or changes: PVP suggests a major version bump
Dependencies removed: extra
Dependency ranges changed: containers, exceptions, hpqtypes, hpqtypes-extras, lifted-base, lifted-threads, log-base, monad-control, monad-time, mtl, stm, time, transformers-base
API changes (from Hackage documentation)
- Database.PostgreSQL.Consumers.Components: spawnDispatcher :: forall m idx job. (MonadBaseControl IO m, MonadLog m, MonadMask m, MonadTime m, Show idx, ToSQL idx) => ConsumerConfig m idx job -> Bool -> ConnectionSourceM m -> ConsumerID -> MVar () -> TVar (Map ThreadId idx) -> TVar Int -> Maybe (TMVar Bool) -> m ThreadId
+ Database.PostgreSQL.Consumers.Components: spawnDispatcher :: forall m idx job. (MonadBaseControl IO m, MonadLog m, MonadMask m, MonadTime m, Show idx, ToSQL idx) => ConsumerConfig m idx job -> ConnectionSourceM m -> ConsumerID -> MVar () -> TVar (Map ThreadId idx) -> TVar Int -> Maybe (TMVar Bool) -> m ThreadId
- Database.PostgreSQL.Consumers.Components: spawnMonitor :: forall m idx job. (MonadBaseControl IO m, MonadLog m, MonadMask m, MonadTime m, Show idx, FromSQL idx) => ConsumerConfig m idx job -> ConnectionSourceM m -> ConsumerID -> m ThreadId
+ Database.PostgreSQL.Consumers.Components: spawnMonitor :: forall m idx job. (MonadBaseControl IO m, MonadLog m, MonadMask m, MonadTime m, Show idx, FromSQL idx, ToSQL idx) => ConsumerConfig m idx job -> ConnectionSourceM m -> ConsumerID -> m ThreadId
Files
- CHANGELOG.md +10/−5
- consumers.cabal +18/−16
- example/Example.hs +1/−1
- src/Database/PostgreSQL/Consumers/Components.hs +93/−97
- test/Test.hs +1/−1
CHANGELOG.md view
@@ -1,7 +1,12 @@+# consumers-2.3.1.0 (2023-11-27)+* Drop support for advisory locks.+* Add support for GHC 9.6 and 9.8.+* Apply `ccOnException` when freeing jobs reserved by an inactive consumer.+ # consumers-2.3.0.0 (2022-07-07)-* Update to monad-time 0.4.0.0-* Support for GHC 9.2-* Drop support for GHC < 8.8+* Update to monad-time 0.4.0.0.+* Support for GHC 9.2.+* Drop support for GHC < 8.8. # consumers-2.2.0.6 (2021-10-22) * Improve efficiency of the SQL query that reserves jobs.@@ -22,8 +27,8 @@ * Support hpqtypes-1.7.0.0 and hpqtypes-extras-1.9.0.0. # consumers-2.2.0.0 (2019-02-19)-* When jobs are relased from a dead consumer, reset their finished_at column-* Start the consumer only when ccMaxRunningJobs >= 1+* When jobs are relased from a dead consumer, reset their finished_at column.+* Start the consumer only when ccMaxRunningJobs >= 1. # consumers-2.1.2.0 (2018-07-11)
consumers.cabal view
@@ -1,5 +1,5 @@ name: consumers-version: 2.3.0.0+version: 2.3.1.0 synopsis: Concurrent PostgreSQL data consumers description: Library for setting up concurrent consumers of data@@ -18,7 +18,8 @@ category: Concurrency, Database build-type: Simple cabal-version: >=1.10-tested-with: GHC ==8.8.4 || ==8.10.7 || ==9.0.2 || ==9.2.3+tested-with: GHC ==8.8.4 || ==8.10.7 || ==9.0.2 || ==9.2.8 || ==9.4.7 || ==9.6.3+ || ==9.8.1 Source-repository head Type: git@@ -32,19 +33,18 @@ Database.PostgreSQL.Consumers.Utils build-depends: base >= 4.13 && < 5- , containers >= 0.5 && < 0.7- , exceptions >= 0.10 && < 0.11- , extra >= 1.6 && < 1.8- , hpqtypes >= 1.7 && < 2.0- , lifted-base >= 0.2 && < 0.3- , lifted-threads >= 1.0 && < 1.1- , log-base >= 0.11 && < 0.12- , monad-control >= 1.0 && < 1.1- , monad-time >= 0.4 && < 0.5- , mtl >= 2.2 && < 2.3- , stm >= 2.4 && < 2.6- , time >= 1.6 && < 2.0- , transformers-base >= 0.4 && < 0.5+ , containers >= 0.5+ , exceptions >= 0.10+ , hpqtypes >= 1.11+ , lifted-base >= 0.2+ , lifted-threads >= 1.0+ , log-base >= 0.11+ , monad-control >= 1.0+ , monad-time >= 0.4+ , mtl >= 2.2+ , stm >= 2.4+ , time >= 1.6+ , transformers-base >= 0.4 hs-source-dirs: src @@ -54,12 +54,14 @@ default-extensions: DeriveDataTypeable , FlexibleContexts , GeneralizedNewtypeDeriving+ , LambdaCase , NoImplicitPrelude , OverloadedStrings , RankNTypes , RecordWildCards , ScopedTypeVariables , TupleSections+ , TypeApplications , TypeFamilies , UndecidableInstances @@ -74,7 +76,7 @@ build-depends: base, consumers, hpqtypes,- hpqtypes-extras >= 1.9 && < 2.0,+ hpqtypes-extras, log-base, text, text-show,
example/Example.hs view
@@ -44,7 +44,7 @@ ConnectionSource connSource = simpleSource connSettings -- Monad stack initialisation.- withSimpleStdOutLogger $ \logger ->+ withStdOutLogger $ \logger -> runLogT "consumers-example" logger defaultLogLevel $ runDBT connSource defaultTransactionSettings $ do
src/Database/PostgreSQL/Consumers/Components.hs view
@@ -75,13 +75,14 @@ skipLockedTest :: Either DBException () <- try . runDBT cs defaultTransactionSettings $ runSQL_ "SELECT TRUE FOR UPDATE SKIP LOCKED"- let useSkipLocked = either (const False) (const True) skipLockedTest+ -- If we can't lock rows using 'skip locked' throw an exception+ either (const $ error "PostgreSQL version with support for SKIP LOCKED is required") pure skipLockedTest cid <- registerConsumer cc cs localData ["consumer_id" .= show cid] $ do listener <- spawnListener cc cs semaphore monitor <- localDomain "monitor" $ spawnMonitor cc cs cid- dispatcher <- localDomain "dispatcher" $ spawnDispatcher cc useSkipLocked+ dispatcher <- localDomain "dispatcher" $ spawnDispatcher cc cs cid semaphore runningJobsInfo runningJobs mIdleSignal return . localDomain "finalizer" $ do stopExecution listener@@ -148,7 +149,7 @@ -- for activity and periodically updates its own. spawnMonitor :: forall m idx job. (MonadBaseControl IO m, MonadLog m, MonadMask m, MonadTime m,- Show idx, FromSQL idx)+ Show idx, FromSQL idx, ToSQL idx) => ConsumerConfig m idx job -> ConnectionSourceM m -> ConsumerID@@ -168,35 +169,42 @@ else do logInfo_ $ "Consumer is not registered" throwM ThreadKilled- -- Freeing jobs locked by inactive consumers needs to happen- -- exactly once, otherwise it's possible to free it twice, after- -- it was already marked as reserved by other consumer, so let's- -- run it in serializable transaction.- (inactiveConsumers, freedJobs) <- runDBT cs tsSerializable $ do+ (inactiveConsumers, freedJobs) <- runDBT cs ts $ do now <- currentTime- -- Delete all inactive (assumed dead) consumers and get their ids.- runSQL_ $ smconcat [- "DELETE FROM" <+> raw ccConsumersTable- , " WHERE last_activity +" <?> iminutes 1 <+> "<= " <?> now- , " AND name =" <?> unRawSQL ccJobsTable- , " RETURNING id::bigint"+ -- Reserve all inactive (assumed dead) consumers and get their ids. We don't+ -- delete them here, because if the coresponding reserved_by column in the+ -- jobs table has an IMMEDIATE foreign key with the ON DELETE SET NULL+ -- property, we will not be able to determine stuck jobs in the next step.+ runSQL_ $ smconcat+ [ "SELECT id::bigint"+ , "FROM" <+> raw ccConsumersTable+ , "WHERE last_activity +" <?> iminutes 1 <+> "<= " <?> now+ , " AND name =" <?> unRawSQL ccJobsTable+ , "FOR UPDATE SKIP LOCKED" ]- inactive :: [Int64] <- fetchMany runIdentity- -- Reset reserved jobs manually, do not rely on the foreign key constraint- -- to do its job. We also reset finished_at to correctly bump number of- -- attempts on the next try.- freedJobs :: [idx] <- if null inactive- then return []- else do+ fetchMany (runIdentity @Int64) >>= \case+ [] -> pure (0, [])+ inactive -> do+ -- Fetch all stuck jobs and run ccOnException on them to determine+ -- actions. This is necessary e.g. to be able to apply exponential+ -- backoff to them correctly. runSQL_ $ smconcat- [ "UPDATE" <+> raw ccJobsTable- , "SET reserved_by = NULL"- , " , finished_at = NULL"+ [ "SELECT" <+> mintercalate ", " ccJobSelectors+ , "FROM" <+> raw ccJobsTable , "WHERE reserved_by = ANY(" <?> Array1 inactive <+> ")"- , "RETURNING id"+ , "FOR UPDATE SKIP LOCKED" ]- fetchMany runIdentity- return (length inactive, freedJobs)+ stuckJobs <- fetchMany ccJobFetcher+ unless (null stuckJobs) $ do+ results <- forM stuckJobs $ \job -> do+ action <- lift $ ccOnException (toException ThreadKilled) job+ pure (ccJobIndex job, Failed action)+ runSQL_ $ updateJobsQuery ccJobsTable results now+ runSQL_ $ smconcat+ [ "DELETE FROM" <+> raw ccConsumersTable+ , "WHERE id = ANY(" <?> Array1 inactive <+> ")"+ ]+ pure (length inactive, map ccJobIndex stuckJobs) when (inactiveConsumers > 0) $ do logInfo "Unregistered inactive consumers" $ object [ "inactive_consumers" .= inactiveConsumers@@ -206,17 +214,12 @@ "freed_jobs" .= map show freedJobs ] liftBase . threadDelay $ 30 * 1000000 -- wait 30 seconds- where- tsSerializable = ts {- tsIsolationLevel = Serializable- } -- | Spawn a thread that reserves and processes jobs. spawnDispatcher :: forall m idx job. ( MonadBaseControl IO m, MonadLog m, MonadMask m, MonadTime m , Show idx, ToSQL idx ) => ConsumerConfig m idx job- -> Bool -> ConnectionSourceM m -> ConsumerID -> MVar ()@@ -224,7 +227,7 @@ -> TVar Int -> Maybe (TMVar Bool) -> m ThreadId-spawnDispatcher ConsumerConfig{..} useSkipLocked cs cid semaphore+spawnDispatcher ConsumerConfig{..} cs cid semaphore runningJobsInfo runningJobs mIdleSignal = forkP "dispatcher" . forever $ do void $ takeMVar semaphore@@ -288,29 +291,13 @@ reservedJobs :: UTCTime -> SQL reservedJobs now = smconcat [ "SELECT id FROM" <+> raw ccJobsTable- -- Converting id to text and hashing it may seem silly,- -- especially when we're dealing with integers in the- -- first place, but even in such case the overhead is- -- small enough (converting 100k integers to text and- -- hashing them takes around 15 ms on i7) to be worth the- -- generality. Note that even if IDs of two pending jobs- -- produce the same hash, it just means that in the worst- -- case they will be processed by the same consumer.- , if useSkipLocked- then "WHERE TRUE"- else "WHERE pg_try_advisory_xact_lock("- <?> unRawSQL ccJobsTable- <> "::regclass::integer, hashtext(id::text))"- , " AND reserved_by IS NULL"+ , "WHERE"+ , " reserved_by IS NULL" , " AND run_at IS NOT NULL" , " AND run_at <= " <?> now , " ORDER BY run_at" , "LIMIT" <?> limit- -- Use SKIP LOCKED if available. Otherwise utilise- -- advisory locks.- , if useSkipLocked- then "FOR UPDATE SKIP LOCKED"- else "FOR UPDATE"+ , "FOR UPDATE SKIP LOCKED" ] -- | Spawn each job in a separate thread.@@ -345,56 +332,65 @@ updateJobs :: [(idx, Result)] -> m () updateJobs results = runDBT cs ts $ do now <- currentTime- runSQL_ $ smconcat [- "WITH removed AS ("- , " DELETE FROM" <+> raw ccJobsTable- , " WHERE id = ANY(" <?> Array1 deletes <+> ")"- , ")"- , "UPDATE" <+> raw ccJobsTable <+> "SET"- , " reserved_by = NULL"- , ", run_at = CASE"- , " WHEN FALSE THEN run_at"- , smconcat $ M.foldrWithKey (retryToSQL now) [] retries- , " ELSE NULL" -- processed- , " END"- , ", finished_at = CASE"- , " WHEN id = ANY(" <?> Array1 successes <+> ") THEN " <?> now- , " ELSE NULL"- , " END"- , "WHERE id = ANY(" <?> Array1 (map fst updates) <+> ")"- ]- where- retryToSQL now (Left int) ids =- ("WHEN id = ANY(" <?> Array1 ids <+> ") THEN " <?> now <> " +" <?> int :)- retryToSQL _ (Right time) ids =- ("WHEN id = ANY(" <?> Array1 ids <+> ") THEN" <?> time :)+ runSQL_ $ updateJobsQuery ccJobsTable results now - retries = foldr step M.empty $ map f updates- where- f (idx, result) = case result of- Ok action -> (idx, action)- Failed action -> (idx, action)+---------------------------------------- - step (idx, action) iretries = case action of- MarkProcessed -> iretries- RerunAfter int -> M.insertWith (++) (Left int) [idx] iretries- RerunAt time -> M.insertWith (++) (Right time) [idx] iretries- Remove -> error- "updateJobs: Remove should've been filtered out"+-- | Generate a single SQL query for updating all given jobs.+updateJobsQuery+ :: (Show idx, ToSQL idx)+ => RawSQL ()+ -> [(idx, Result)]+ -> UTCTime+ -> SQL+updateJobsQuery jobsTable results now = smconcat+ [ "WITH removed AS ("+ , " DELETE FROM" <+> raw jobsTable+ , " WHERE id = ANY(" <?> Array1 deletes <+> ")"+ , ")"+ , "UPDATE" <+> raw jobsTable <+> "SET"+ , " reserved_by = NULL"+ , ", run_at = CASE"+ , " WHEN FALSE THEN run_at"+ , smconcat $ M.foldrWithKey retryToSQL [] retries+ , " ELSE NULL" -- processed+ , " END"+ , ", finished_at = CASE"+ , " WHEN id = ANY(" <?> Array1 successes <+> ") THEN " <?> now+ , " ELSE NULL"+ , " END"+ , "WHERE id = ANY(" <?> Array1 (map fst updates) <+> ")"+ ]+ where+ retryToSQL (Left int) ids =+ ("WHEN id = ANY(" <?> Array1 ids <+> ") THEN " <?> now <> " +" <?> int :)+ retryToSQL (Right time) ids =+ ("WHEN id = ANY(" <?> Array1 ids <+> ") THEN" <?> time :) - successes = foldr step [] updates- where- step (idx, Ok _) acc = idx : acc- step (_, Failed _) acc = acc+ retries = foldr step M.empty $ map getAction updates+ where+ getAction (idx, result) = case result of+ Ok action -> (idx, action)+ Failed action -> (idx, action) - (deletes, updates) = foldr step ([], []) results- where- step job@(idx, result) (ideletes, iupdates) = case result of- Ok Remove -> (idx : ideletes, iupdates)- Failed Remove -> (idx : ideletes, iupdates)- _ -> (ideletes, job : iupdates)+ step (idx, action) iretries = case action of+ MarkProcessed -> iretries+ RerunAfter int -> M.insertWith (++) (Left int) [idx] iretries+ RerunAt time -> M.insertWith (++) (Right time) [idx] iretries+ Remove -> error "updateJobs: Remove should've been filtered out" -----------------------------------------+ successes = foldr step [] updates+ where+ step (idx, Ok _) acc = idx : acc+ step (_, Failed _) acc = acc++ (deletes, updates) = foldr step ([], []) results+ where+ step job@(idx, result) (ideletes, iupdates) = case result of+ Ok Remove -> (idx : ideletes, iupdates)+ Failed Remove -> (idx : ideletes, iupdates)+ _ -> (ideletes, job : iupdates)+ ts :: TransactionSettings ts = defaultTransactionSettings {
test/Test.hs view
@@ -86,7 +86,7 @@ { csConnInfo = connString } ConnectionSource connSource = simpleSource connSettings - withSimpleStdOutLogger $ \logger ->+ withStdOutLogger $ \logger -> runTestEnv connSource logger $ do createTables idleSignal <- liftIO $ atomically $ newEmptyTMVar