consumers 2.1.2.0 → 2.2.0.0

raw patch · 5 files changed

+53/−38 lines, 5 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• CHANGELOG.md +4/−0
• consumers.cabal +2/−2
• example/Example.hs +1/−0
• src/Database/PostgreSQL/Consumers/Components.hs +45/−36
• test/Test.hs +1/−0

@@ -1,3 +1,7 @@+# 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+ # consumers-2.1.2.0 (2018-07-11)  * Support hpqtypes-1.6.0.0.

@@ -1,5 +1,5 @@ name:               consumers-version:            2.1.2.0+version:            2.2.0.0 synopsis:           Concurrent PostgreSQL data consumers  description:        Library for setting up concurrent consumers of data@@ -18,7 +18,7 @@ category:           Concurrency, Database build-type:         Simple cabal-version:      >=1.10-tested-with:        GHC == 8.0.2, GHC == 8.2.2, GHC == 8.4.3+tested-with:        GHC == 8.0.2, GHC == 8.2.2, GHC == 8.4.4, GHC == 8.6.2  Source-repository head   Type:             git

@@ -19,6 +19,7 @@ import Data.Monoid import Log import Log.Backend.StandardOutput+import Prelude import System.Environment import TextShow 

@@ -39,7 +39,7 @@ -- seamlessly handle the finalization. runConsumer   :: ( MonadBaseControl IO m, MonadLog m, MonadMask m, Eq idx, Show idx-     , ToSQL idx )+     , FromSQL idx, ToSQL idx )   => ConsumerConfig m idx job   -> ConnectionSourceM m   -> m (m ())@@ -47,7 +47,7 @@  runConsumerWithIdleSignal   :: ( MonadBaseControl IO m, MonadLog m, MonadMask m, Eq idx, Show idx-     , ToSQL idx )+     , FromSQL idx, ToSQL idx )   => ConsumerConfig m idx job   -> ConnectionSourceM m   -> TMVar Bool@@ -56,35 +56,38 @@  -- | Run the consumer and also signal whenever the consumer is waiting for -- getNotification or threadDelay.- runConsumerWithMaybeIdleSignal   :: ( MonadBaseControl IO m, MonadLog m, MonadMask m, Eq idx, Show idx-     , ToSQL idx )+     , FromSQL idx, ToSQL idx )   => ConsumerConfig m idx job   -> ConnectionSourceM m   -> Maybe (TMVar Bool)   -> m (m ())-runConsumerWithMaybeIdleSignal cc cs mIdleSignal = do-  semaphore <- newMVar ()-  runningJobsInfo <- liftBase $ newTVarIO M.empty-  runningJobs <- liftBase $ newTVarIO 0+runConsumerWithMaybeIdleSignal cc cs mIdleSignal+  | ccMaxRunningJobs cc < 1 = do+      logInfo_ "ccMaxRunningJobs < 1, not starting the consumer"+      return $ return ()+  | otherwise = do+      semaphore <- newMVar ()+      runningJobsInfo <- liftBase $ newTVarIO M.empty+      runningJobs <- liftBase $ newTVarIO 0 -  skipLockedTest :: Either DBException () <- try . runDBT cs def $-    runSQL_ "SELECT TRUE FOR UPDATE SKIP LOCKED"-  let useSkipLocked = either (const False) (const True) skipLockedTest+      skipLockedTest :: Either DBException () <- try . runDBT cs def $+        runSQL_ "SELECT TRUE FOR UPDATE SKIP LOCKED"+      let useSkipLocked = either (const False) (const True) 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-      cs cid semaphore runningJobsInfo runningJobs mIdleSignal-    return . localDomain "finalizer" $ do-      stopExecution listener-      stopExecution dispatcher-      waitForRunningJobs runningJobsInfo runningJobs-      stopExecution monitor-      unregisterConsumer cc cs cid+      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+          cs cid semaphore runningJobsInfo runningJobs mIdleSignal+        return . localDomain "finalizer" $ do+          stopExecution listener+          stopExecution dispatcher+          waitForRunningJobs runningJobsInfo runningJobs+          stopExecution monitor+          unregisterConsumer cc cs cid   where     waitForRunningJobs runningJobsInfo runningJobs = do       initialJobs <- liftBase $ readTVarIO runningJobsInfo@@ -143,7 +146,8 @@ -- | Spawn a thread that monitors working consumers -- for activity and periodically updates its own. spawnMonitor-  :: (MonadBaseControl IO m, MonadLog m, MonadMask m)+  :: forall m idx job. (MonadBaseControl IO m, MonadLog m, MonadMask m,+                        Show idx, FromSQL idx)   => ConsumerConfig m idx job   -> ConnectionSourceM m   -> ConsumerID@@ -177,23 +181,28 @@       , "  RETURNING id::bigint"       ]     inactive :: [Int64] <- fetchMany runIdentity-    -- Reset reserved jobs manually, do not rely-    -- on the foreign key constraint to do its job.-    freed <- if null inactive-      then return 0-      else runSQL $ smconcat [-        "UPDATE" <+> raw ccJobsTable-      , "SET reserved_by = NULL"-      , "WHERE reserved_by = ANY(" <?> Array1 inactive <+> ")"-      ]-    return (length inactive, freed)+    -- 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+        runSQL_ $ smconcat+          [ "UPDATE" <+> raw ccJobsTable+          , "SET reserved_by = NULL"+          , "  , finished_at = NULL"+          , "WHERE reserved_by = ANY(" <?> Array1 inactive <+> ")"+          , "RETURNING id"+          ]+        fetchMany runIdentity+    return (length inactive, freedJobs)   when (inactiveConsumers > 0) $ do     logInfo "Unregistered inactive consumers" $ object [         "inactive_consumers" .= inactiveConsumers       ]-  when (freedJobs > 0) $ do+  unless (null freedJobs) $ do     logInfo "Freed locked jobs" $ object [-        "freed_jobs" .= freedJobs+        "freed_jobs" .= map show freedJobs       ]   liftBase . threadDelay $ 30 * 1000000 -- wait 30 seconds   where

@@ -30,6 +30,7 @@ import Data.Time import Log import Log.Backend.StandardOutput+import Prelude import System.Environment import TextShow