diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,4 +1,11 @@ -## [_Unreleased_](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.1.2.7...main) +## [_Unreleased_](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.1.3.0...main) + +## [v1.1.3.0](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.1.2.7...v1.1.3.0) + +- Add `Faktory.Pool` + + This incurs the new dependencies, `unliftio`, `resource-pool`, and + `microlens`. ## [v1.1.2.7](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.1.2.6...v1.1.2.7) diff --git a/faktory.cabal b/faktory.cabal --- a/faktory.cabal +++ b/faktory.cabal @@ -1,6 +1,6 @@ cabal-version: 1.18 name: faktory -version: 1.1.2.7 +version: 1.1.3.0 license: MIT license-file: LICENSE copyright: 2018 Freckle Education @@ -51,6 +51,7 @@ library exposed-modules: + Data.Pool.Compat Faktory.Client Faktory.Connection Faktory.Ent.Batch @@ -61,6 +62,7 @@ Faktory.JobFailure Faktory.JobOptions Faktory.JobState + Faktory.Pool Faktory.Prelude Faktory.Producer Faktory.Protocol @@ -97,15 +99,18 @@ errors >=2.3.0, megaparsec >=8.0.0, memory >=0.15.0, + microlens >=0.4.11.2, mtl >=2.2.2, network >=3.1.1.1, random >=1.1, + resource-pool >=0.2.3.2, safe-exceptions >=0.1.7.1, scanner >=0.3.1, semigroups >=0.19.1, text >=1.2.4.0, time >=1.9.3, unix >=2.7.2.2, + unliftio >=0.2.13.1, unordered-containers >=0.2.10.0 if impl(ghc >=8.10) diff --git a/library/Data/Pool/Compat.hs b/library/Data/Pool/Compat.hs new file mode 100644 --- /dev/null +++ b/library/Data/Pool/Compat.hs @@ -0,0 +1,31 @@ +{-# LANGUAGE CPP #-} + +module Data.Pool.Compat + ( module Data.Pool + , createPool + ) where + +import Prelude + +import Data.Pool hiding (createPool) +#if MIN_VERSION_resource_pool(0,3,0) +#else +import Control.Concurrent (getNumCapabilities) +import qualified Data.Pool as Pool +#endif + +createPool + :: IO a + -> (a -> IO ()) + -> Double + -> Int + -> IO (Pool a) +createPool create destroy timeout size = do +#if MIN_VERSION_resource_pool(0,3,0) + newPool $ defaultPoolConfig create destroy timeout size +#else + -- Re-implement instead of using the deprecated compatibility function, so + -- that we can get a consistent numStripes and size behavior. + numStripes <- getNumCapabilities + Pool.createPool create destroy numStripes (realToFrac timeout) size +#endif diff --git a/library/Faktory/Pool.hs b/library/Faktory/Pool.hs new file mode 100644 --- /dev/null +++ b/library/Faktory/Pool.hs @@ -0,0 +1,126 @@ +module Faktory.Pool + ( FaktoryPool + , HasFaktoryPool (..) + + -- * Pool Construction + , Settings + , PoolSettings + , newFaktoryPool + + -- * Pool use + , perform + , buildJob + + -- * Direct access + , withProducer + , takeProducer + + -- * Re-exports + , module Faktory.Job + ) where + +import Faktory.Prelude + +import Control.Monad.IO.Class (MonadIO (..)) +import Control.Monad.Reader (MonadReader, asks) +import Data.Aeson (ToJSON) +import Data.Pool.Compat (Pool) +import qualified Data.Pool.Compat as Pool +import Faktory.Job hiding (buildJob, perform) +import qualified Faktory.Job as Job +import Faktory.Producer +import Faktory.Settings (PoolSettings (..), Settings) +import GHC.Stack (HasCallStack) +import Lens.Micro (Lens', (^.)) +import UnliftIO (MonadUnliftIO, withRunInIO) + +-- | +-- +-- @since 1.1.3.0 +newtype FaktoryPool = FaktoryPool (Pool Producer) + +-- | +-- +-- @since 1.1.3.0 +class HasFaktoryPool env where + faktoryPoolL :: Lens' env FaktoryPool + +instance HasFaktoryPool FaktoryPool where + faktoryPoolL = id + +-- | Build a 'FaktoryPool' with the given settings +-- +-- See 'Settings', 'envSettings', 'PoolSettings', and 'envPoolSettings'. +-- +-- @since 1.1.3.0 +newFaktoryPool + :: MonadIO m + => Settings + -> PoolSettings + -> m FaktoryPool +newFaktoryPool settings PoolSettings {..} = do + liftIO $ + FaktoryPool + <$> Pool.createPool + (newProducer settings) + closeProducer + (fromIntegral settingsTimeout) + (fromIntegral settingsSize) + +-- | 'Faktory.Job.perform' but using a 'Producer' from the pool +-- +-- @since 1.1.3.0 +perform + :: ( MonadUnliftIO m + , MonadReader env m + , HasFaktoryPool env + , ToJSON arg + , HasCallStack + ) + => JobOptions + -> arg + -> m JobId +perform options arg = do + withProducer $ \producer -> do + liftIO $ Job.perform options producer arg + +-- | 'Faktory.Job.buildJob' but using a 'Producer' from the pool +-- +-- @since 1.1.3.0 +buildJob + :: (MonadUnliftIO m, MonadReader env m, HasFaktoryPool env) + => JobOptions + -> arg + -> m (Job arg) +buildJob options arg = do + withProducer $ \producer -> do + liftIO $ Job.buildJob options producer arg + +-- | Acquire a 'Producer', use it, and return it to the pool +-- +-- @since 1.1.3.0 +withProducer + :: (MonadUnliftIO m, MonadReader env m, HasFaktoryPool env) + => (Producer -> m a) + -> m a +withProducer f = do + FaktoryPool p <- asks (^. faktoryPoolL) + withRunInIO $ \runInIO -> do + Pool.withResource p $ runInIO . f + +-- | Get a 'Producer' from the pool along with an action to return it +-- +-- You should prefer 'withProducer' if at all possible. With this function you +-- are responsible to ensure the return action is called (e.g. with 'finally'). +-- +-- This is only necessary if you are operating in a monad that doesn't have +-- 'MonadUnliftIO' (like 'ConduitT'), so you need to take and return a +-- 'Producer' separately (e.g. with 'bracketP'). +-- +-- @since 1.1.3.0 +takeProducer + :: (MonadIO m, MonadReader env m, HasFaktoryPool env) => m (Producer, m ()) +takeProducer = do + FaktoryPool p <- asks (^. faktoryPoolL) + (producer, lp) <- liftIO $ Pool.takeResource p + pure (producer, liftIO $ Pool.putResource lp producer) diff --git a/library/Faktory/Settings.hs b/library/Faktory/Settings.hs --- a/library/Faktory/Settings.hs +++ b/library/Faktory/Settings.hs @@ -2,6 +2,8 @@ ( Settings (..) , defaultSettings , envSettings + + -- * Worker , WorkerSettings (..) , defaultWorkerSettings , envWorkerSettings @@ -12,6 +14,10 @@ , WorkerId , randomWorkerId + -- * Pool + , PoolSettings (..) + , envPoolSettings + -- * Re-exports , ConnectionInfo (..) , Namespace (..) @@ -23,6 +29,7 @@ import Faktory.Connection import Faktory.JobOptions (JobOptions) import Faktory.Settings.Queue +import Numeric.Natural import System.Environment (lookupEnv) import System.IO (hPutStrLn, stderr) import System.Random @@ -82,3 +89,43 @@ randomWorkerId :: IO WorkerId randomWorkerId = WorkerId . take 8 . randomRs ('a', 'z') <$> newStdGen + +-- | +-- +-- @since 1.1.3.0 +data PoolSettings = PoolSettings + { settingsSize :: Natural + -- ^ Maximum pool size + -- + -- Default is @10@. Smallest acceptable value is @1@. Note that, due to the + -- striping behavior of @resource-pool@, a configured size @N@ may result in + -- @N - 1@ resources. + , settingsTimeout :: Natural + -- ^ How long before destroying a resource, in seconds + -- + -- Default is @600@. + } + +-- | +-- +-- @since 1.1.3.0 +defaultPoolSettings :: PoolSettings +defaultPoolSettings = + PoolSettings + { settingsSize = 10 + , settingsTimeout = 600 + } + +-- | Read 'PoolSettings' from the environment +-- +-- - @FAKTORY_POOL_SIZE@ +-- - @FAKTORY_POOL_TIMEOUT@ +-- +-- @since 1.1.3.0 +envPoolSettings :: IO PoolSettings +envPoolSettings = + PoolSettings + <$> (maybe settingsSize read <$> lookupEnv "FAKTORY_POOL_SIZE") + <*> (maybe settingsTimeout read <$> lookupEnv "FAKTORY_POOL_TIMEOUT") + where + PoolSettings {..} = defaultPoolSettings