consumers 1.1 → 2.0

raw patch · 5 files changed

+48/−38 lines, 5 filesdep ~hpqtypesPVP ok

version bump matches the API change (PVP)

Dependency ranges changed: hpqtypes

API changes (from Hackage documentation)

Files

• consumers.cabal +14/−14
• src/Database/PostgreSQL/Consumers/Components.hs +29/−22
• src/Database/PostgreSQL/Consumers/Config.hs +1/−0
• src/Database/PostgreSQL/Consumers/Consumer.hs +3/−2
• src/Database/PostgreSQL/Consumers/Utils.hs +1/−0

@@ -1,12 +1,12 @@ name:               consumers-version:            1.1+version:            2.0 synopsis:           Concurrent PostgreSQL data consumers description:        Library for setting up concurrent consumers of data stored inside PostgreSQL database in a simple, declarative manner. homepage:           https://github.com/scrive/consumers license:            BSD3 license-file:       LICENSE author:             Scrive-maintainer:         Andrzej Rybczak <andrzej@scrive.com>+maintainer:         Andrzej Rybczak <andrzej@rybczak.net> category:           Concurrency, Database build-type:         Simple cabal-version:      >=1.10@@ -25,7 +25,7 @@   build-depends:    base <5,                     containers,                     exceptions,-                    hpqtypes >=1.4,+                    hpqtypes >=1.5,                     lifted-base,                     lifted-threads,                     log >= 0.4,@@ -40,14 +40,14 @@   ghc-options:      -O2 -Wall -funbox-strict-fields    default-language: Haskell2010-  default-extensions:-    DeriveDataTypeable,-    FlexibleContexts,-    GeneralizedNewtypeDeriving,-    OverloadedStrings,-    RankNTypes,-    RecordWildCards,-    ScopedTypeVariables,-    TupleSections,-    TypeFamilies,-    UndecidableInstances+  default-extensions: DeriveDataTypeable+                    , FlexibleContexts+                    , GeneralizedNewtypeDeriving+                    , NoImplicitPrelude+                    , OverloadedStrings+                    , RankNTypes+                    , RecordWildCards+                    , ScopedTypeVariables+                    , TupleSections+                    , TypeFamilies+                    , UndecidableInstances

@@ -20,6 +20,7 @@ import Data.Monoid.Utils import Database.PostgreSQL.PQTypes import Log+import Prelude import qualified Control.Concurrent.STM as STM import qualified Control.Concurrent.Thread.Lifted as T import qualified Data.Foldable as F@@ -36,18 +37,22 @@ runConsumer   :: (MonadBaseControl IO m, MonadLog m, MonadMask m, Eq idx, Show idx, ToSQL idx)   => ConsumerConfig m idx job-  -> ConnectionSource+  -> ConnectionSourceM m   -> m (m ()) runConsumer cc cs = 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+   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 cs cid semaphore runningJobsInfo runningJobs+    dispatcher <- localDomain "dispatcher" $ spawnDispatcher cc useSkipLocked cs cid semaphore runningJobsInfo runningJobs     return . localDomain "finalizer" $ do       stopExecution listener       stopExecution dispatcher@@ -83,7 +88,7 @@ spawnListener   :: (MonadBaseControl IO m, MonadMask m)   => ConsumerConfig m idx job-  -> ConnectionSource+  -> ConnectionSourceM m   -> MVar ()   -> m ThreadId spawnListener cc cs semaphore = forkP "listener" $ case ccNotificationChannel cc of@@ -110,7 +115,7 @@ spawnMonitor   :: (MonadBaseControl IO m, MonadLog m, MonadMask m)   => ConsumerConfig m idx job-  -> ConnectionSource+  -> ConnectionSourceM m   -> ConsumerID   -> m ThreadId spawnMonitor ConsumerConfig{..} cs cid = forkP "monitor" . forever $ do@@ -168,13 +173,14 @@ spawnDispatcher   :: forall m idx job. (MonadBaseControl IO m, MonadLog m, MonadMask m, Show idx, ToSQL idx)   => ConsumerConfig m idx job-  -> ConnectionSource+  -> Bool+  -> ConnectionSourceM m   -> ConsumerID   -> MVar ()   -> TVar (M.Map ThreadId idx)   -> TVar Int   -> m ThreadId-spawnDispatcher ConsumerConfig{..} cs cid semaphore runningJobsInfo runningJobs =+spawnDispatcher ConsumerConfig{..} useSkipLocked cs cid semaphore runningJobsInfo runningJobs =   forkP "dispatcher" . forever $ do     void $ takeMVar semaphore     loop 1@@ -223,23 +229,24 @@         reservedJobs :: SQL         reservedJobs = 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.-            -- Also: after PostgreSQL 9.5 is released, we can use SELECT-            -- FOR UPDATE SKIP LOCKED instead of advisory locks (see-            -- http://michael.otacoo.com/postgresql-2/postgres-9-5-feature-highlight-skip-locked-row-level/-            -- for more details). Also, 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.-          , "WHERE pg_try_advisory_xact_lock(" <?> unRawSQL ccJobsTable <> "::regclass::integer, hashtext(id::text))"-          , "  AND reserved_by IS NULL"-          , "  AND run_at IS NOT NULL"-          , "  AND run_at <= now()"+            -- 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"+          , "       AND run_at IS NOT NULL"+          , "       AND run_at <= now()"           , "LIMIT" <?> limit-          , "FOR UPDATE"+            -- Use SKIP LOCKED if available. Otherwise utilize advisory locks.+          , if useSkipLocked+            then "FOR UPDATE SKIP LOCKED"+            else "FOR UPDATE"           ]      -- | Spawn each job in a separate thread.

@@ -8,6 +8,7 @@ import Control.Exception (SomeException) import Data.Time import Database.PostgreSQL.PQTypes+import Prelude  -- | Action to take after a job was processed. data Action

@@ -11,6 +11,7 @@ import Data.Monoid import Data.Monoid.Utils import Database.PostgreSQL.PQTypes+import Prelude  import Database.PostgreSQL.Consumers.Config @@ -32,7 +33,7 @@ registerConsumer   :: (MonadBase IO m, MonadMask m)   => ConsumerConfig n idx job-  -> ConnectionSource+  -> ConnectionSourceM m   -> m ConsumerID registerConsumer ConsumerConfig{..} cs = runDBT cs ts $ do   runSQL_ $ smconcat [@@ -50,7 +51,7 @@ unregisterConsumer   :: (MonadBase IO m, MonadMask m)   => ConsumerConfig n idx job-  -> ConnectionSource+  -> ConnectionSourceM m   -> ConsumerID   -> m () unregisterConsumer ConsumerConfig{..} cs wid = runDBT cs ts $ do

@@ -11,6 +11,7 @@ import Control.Monad.Catch import Control.Monad.Trans.Control import Data.Typeable+import Prelude import qualified Control.Concurrent.Thread.Group.Lifted as TG import qualified Control.Concurrent.Thread.Lifted as T import qualified Control.Exception.Lifted as E