diff --git a/consumers.cabal b/consumers.cabal --- a/consumers.cabal +++ b/consumers.cabal @@ -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 +maintainer: Andrzej Rybczak 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 diff --git a/src/Database/PostgreSQL/Consumers/Components.hs b/src/Database/PostgreSQL/Consumers/Components.hs --- a/src/Database/PostgreSQL/Consumers/Components.hs +++ b/src/Database/PostgreSQL/Consumers/Components.hs @@ -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. diff --git a/src/Database/PostgreSQL/Consumers/Config.hs b/src/Database/PostgreSQL/Consumers/Config.hs --- a/src/Database/PostgreSQL/Consumers/Config.hs +++ b/src/Database/PostgreSQL/Consumers/Config.hs @@ -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 diff --git a/src/Database/PostgreSQL/Consumers/Consumer.hs b/src/Database/PostgreSQL/Consumers/Consumer.hs --- a/src/Database/PostgreSQL/Consumers/Consumer.hs +++ b/src/Database/PostgreSQL/Consumers/Consumer.hs @@ -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 diff --git a/src/Database/PostgreSQL/Consumers/Utils.hs b/src/Database/PostgreSQL/Consumers/Utils.hs --- a/src/Database/PostgreSQL/Consumers/Utils.hs +++ b/src/Database/PostgreSQL/Consumers/Utils.hs @@ -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