simpoole 0.0.1 → 0.1.0

raw patch · 4 files changed

+152/−38 lines, 4 files

Files

• ChangeLog.md +7/−0
• lib/Simpoole.hs +136/−32
• simpoole.cabal +1/−1
• test/SimpooleSpec.hs +8/−5

@@ -1,5 +1,12 @@ # simpoole +## 0.1.0 (Unreleased)++* Expose helper functions: acquireResource, returnResource, destroyResource+* Track number of idle resources as part of metrics+* Unify pool settings into Settings type+* Allow configuration of "return policy" which helps to optimize the pool+ ## 0.0.1  * Fix max live metric

@@ -12,8 +12,15 @@   , newUnlimitedPool   , newPool   , withResource+  , acquireResource+  , returnResource+  , destroyResource   , poolMetrics +  , Settings (..)+  , defaultSettings+  , ReturnPolicy (..)+   , Metrics (..)   ) where@@ -28,6 +35,64 @@ import qualified Data.Time as Time import           Numeric.Natural (Natural) +-- | Strategy to use when returning resources to the pool+--+-- @since 0.1.0+data ReturnPolicy+  = ReturnToFront+  -- ^ Return resources to the front. Resources that have been used recently are more likely to be+  -- reused again quicker. This strategy is good if you want to scale down the pool more quickly in+  -- case resources are not needed.+  --+  -- @since 0.1.0+  | ReturnToBack+  -- ^ Return resources to the back. Resources that have been used recently are less likely to be+  -- used again quicker. Use this strategy if you want to keep more resources in the pool fresh, or+  -- when maintaining the pool in order to be ready for burst workloads.+  -- This strategy can lead to no resources ever been freed when all resources are used within the+  -- idle timeout.+  --+  -- @since 0.1.0+  | ReturnToMiddle+  -- ^ Return resources to the middle. This offers a middleground between 'ReturnToFront' and+  -- 'ReturnToBack'. By ensuring that the starting sub-sequence of resources is reused quicker but+  -- the trailing sub-sequence is not and therefore released more easily.+  --+  -- @since 0.1.0+  deriving stock (Show, Read, Eq, Ord, Enum, Bounded)++-- | Insert a value based on the return policy.+applyReturnPolicy :: ReturnPolicy -> a -> Seq.Seq a -> Seq.Seq a+applyReturnPolicy policy value seq =+  case policy of+    ReturnToFront -> value Seq.<| seq+    ReturnToBack -> seq Seq.|> value+    ReturnToMiddle -> Seq.insertAt middleIndex value seq+  where+    middleIndex+      | even (Seq.length seq) = div (Seq.length seq) 2+      | otherwise = div (Seq.length seq) 2 + 1++-- | Lets you configure certain behaviours of the pool+--+-- @since 0.1.0+data Settings = PoolSettings+  { settings_idleTimeout :: Time.NominalDiffTime+  -- ^ Maximum idle time after which a resource is destroyed+  --+  -- @since 0.1.0+  , settings_returnPolicy :: ReturnPolicy+  }++-- | Default pool settings+--+-- @since 0.1.0+defaultSettings :: Settings+defaultSettings = PoolSettings+  { settings_idleTimeout = 60 -- 60 seconds+  , settings_returnPolicy = ReturnToMiddle+  }+ -- | Pool of resources -- -- @since 0.0.0@@ -60,56 +125,77 @@     Time.UTCTime     -- ^ Last use time     a-    -- ^ The resource itesemf+    -- ^ The resource item  -- | Create a new pool that has no limit on how many resources it may create and hold. ----- @since 0.0.0+-- @since 0.1.0 newUnlimitedPool   :: (Concurrent.MonadConc m, MonadIO m)   => m a   -- ^ Resource creation   -> (a -> m ())   -- ^ Resource destruction-  -> Time.NominalDiffTime-  -- ^ Maximum idle time (+-1s) after which a resource is destroyed+  -> Settings+  -- ^ Pool settings   -> m (Pool m a)-newUnlimitedPool create destroy maxIdleTime = do+newUnlimitedPool create destroy settings = do   leftOversRef <- Concurrent.newIORefN "leftOvers" Seq.empty -  metricRefs <- mkMetricRefs+  createdRef <- Concurrent.newIORefN "created" 0+  destroyedRef <- Concurrent.newIORefN "destroyed" 0+  maxLiveRef <- Concurrent.newIORefN "maxLive" 0    let+    getMetrics = do+      created <- Concurrent.readIORef createdRef+      destroyed <- Concurrent.readIORef destroyedRef+      maxLive <- Concurrent.readIORef maxLiveRef+      leftOvers <- Concurrent.readIORef leftOversRef++      pure Metrics+        { metrics_createdResources = created+        , metrics_destroyedResources = destroyed+        , metrics_maxLiveResources = maxLive+        , metrics_idleResources = fromIntegral (Seq.length leftOvers)+        }+     wrappedCreate = do       value <- create-      succIORef (metrics_createdResources metricRefs)+      succIORef createdRef       pure value      wrappedDestroy resource =-      destroy resource `Catch.finally` succIORef (metrics_destroyedResources metricRefs)+      destroy resource `Catch.finally` succIORef destroyedRef      acquireResource = do       mbResource <- Concurrent.atomicModifyIORef' leftOversRef $ \leftOvers ->         case leftOvers of           Resource _ head Seq.:<| tail -> (tail, Just head)           _empty -> (leftOvers, Nothing)+       resource <- maybe wrappedCreate pure mbResource -      numDestroyed <- Concurrent.readIORef (metrics_destroyedResources metricRefs)-      numCreated <- Concurrent.readIORef (metrics_createdResources metricRefs)-      maxIORef (metrics_maxLiveResources metricRefs) (numCreated - numDestroyed)+      numDestroyed <- Concurrent.readIORef destroyedRef+      numCreated <- Concurrent.readIORef createdRef+      maxIORef maxLiveRef (numCreated - numDestroyed)        pure resource      returnResource value = do       now <- liftIO Time.getCurrentTime       Concurrent.atomicModifyIORef' leftOversRef $ \leftOvers ->-        (leftOvers Seq.:|> Resource now value, ())+        ( applyReturnPolicy (settings_returnPolicy settings) (Resource now value) leftOvers+        , ()+        )    _reaperThread <- Async.asyncWithUnmaskN "reaperThread" $ \unmask -> unmask $ forever $ do     now <- liftIO Time.getCurrentTime -    let isStillGood (Resource lastUse _) = Time.diffUTCTime now lastUse <= maxIdleTime+    let+      isStillGood (Resource lastUse _) =+        Time.diffUTCTime now lastUse <= settings_idleTimeout settings+     oldResource <- Concurrent.atomicModifyIORef' leftOversRef (Seq.partition isStillGood)      unless (null oldResource) $ void $@@ -123,14 +209,14 @@     { pool_acquire = acquireResource     , pool_return = returnResource     , pool_destroy = wrappedDestroy-    , pool_metrics = readMetricRefs metricRefs+    , pool_metrics = getMetrics     }  -- | Similar to 'newUnlimitedPool' but allows you to limit the number of resources that will exist -- at the same time. When all resources are currently in use, further resource acquisition will -- block until one is no longer in use. ----- @since 0.0.0+-- @since 0.1.0 newPool   :: (Concurrent.MonadConc m, MonadIO m, MonadFail m)   => m a@@ -139,11 +225,11 @@   -- ^ Resource destruction   -> Int   -- ^ Maximum number of resources to exist at the same time-  -> Time.NominalDiffTime-  -- ^ Maximum idle time after which a resource is destroyed+  -> Settings+  -- ^ Pool settings   -> m (Pool m a)-newPool create destroy maxElems maxIdleTime = do-  basePool <- newUnlimitedPool create destroy maxIdleTime+newPool create destroy maxElems settings = do+  basePool <- newUnlimitedPool create destroy settings   maxElemBarrier <- Concurrent.newQSem maxElems    let@@ -163,7 +249,9 @@     , pool_metrics = pool_metrics basePool     } --- | Use a resource from the pool.+-- | Use a resource from the pool. Once the continuation returns, the resource will be returned to+-- the pool. If the given continuation throws an error then the acquired resource will be destroyed+-- instead. -- -- @since 0.0.0 withResource :: Catch.MonadMask m => Pool m a -> (a -> m r) -> m r@@ -176,6 +264,30 @@  {-# INLINE withResource #-} +-- | Acquire a resource.+--+-- @since 0.1.0+acquireResource :: Pool m a -> m a+acquireResource = pool_acquire++{-# INLINE acquireResource #-}++-- | Return a resource to the pool.+--+-- @since 0.1.0+returnResource :: Pool m a -> a -> m ()+returnResource = pool_return++{-# INLINE returnResource #-}++-- | Destroy a resource.+--+-- @since 0.1.0+destroyResource :: Pool m a -> a -> m ()+destroyResource = pool_destroy++{-# INLINE destroyResource #-}+ -- | Fetch pool metrics. -- -- @since 0.0.0@@ -202,20 +314,12 @@   -- ^ Maximum number of resources that were alive simultaneously   --   -- @since 0.0.0+  , metrics_idleResources :: a+  -- ^ Number of resources currently idle+  --+  -- @since 0.1.0   }   deriving stock (Show, Read, Eq, Ord, Functor, Foldable, Traversable)---- | Create the IORefs which capture the metric values.-mkMetricRefs :: Concurrent.MonadConc m => m (Metrics (Concurrent.IORef m Natural))-mkMetricRefs =-  Metrics-    <$> Concurrent.newIORefN "created" 0-    <*> Concurrent.newIORefN "destroyed" 0-    <*> Concurrent.newIORefN "maxLive" 0---- | Read all the metric values.-readMetricRefs :: Concurrent.MonadConc m => Metrics (Concurrent.IORef m a) -> m (Metrics a)-readMetricRefs = traverse Concurrent.readIORef  -- | Increase a value held by an IORef by one. succIORef :: (Concurrent.MonadConc m, Enum a) => Concurrent.IORef m a -> m ()

@@ -1,6 +1,6 @@ cabal-version:      2.2 name:               simpoole-version:            0.0.1+version:            0.1.0 category:           Data, Resources synopsis:           Simple pool description:        Provides a simple pool implementation.

@@ -5,7 +5,6 @@ import qualified Control.Concurrent.Classy as Concurrent import qualified Control.Concurrent.Classy.Async as Async import           Control.Monad (join)-import           Control.Monad.IO.Class (liftIO) import           Numeric.Natural (Natural) import qualified Simpoole as Pool import           Test.Hspec (Spec, describe, it, shouldBe, shouldSatisfy)@@ -23,7 +22,8 @@         destroy _ =           Concurrent.atomicModifyIORef' counterRef $ \count -> (pred count, ()) -      pool <- Pool.newUnlimitedPool create destroy 1+      pool <- Pool.newUnlimitedPool create destroy+        Pool.defaultSettings { Pool.settings_idleTimeout = 1 }        Async.replicateConcurrently_ 200 $         Pool.withResource pool $ const $ Concurrent.threadDelay 1_000@@ -61,16 +61,19 @@           Concurrent.atomicModifyIORef' destroyedRef $ \count -> (count + 1, ())           Concurrent.atomicModifyIORef' counterRef $ \count -> (count - 1, ()) -      pool <- Pool.newUnlimitedPool create destroy 60+      pool <- Pool.newUnlimitedPool create destroy Pool.defaultSettings        Async.replicateConcurrently_ 200 $ do         Pool.withResource pool $ const $ Concurrent.threadDelay 1_000        metrics <- Pool.poolMetrics pool-      liftIO $ print metrics +      -- After finishing the workload, all resources should be idle.+      Pool.metrics_maxLiveResources metrics `shouldBe` Pool.metrics_idleResources metrics+       created <- Concurrent.readIORef createdRef       Pool.metrics_createdResources metrics `shouldBe` created+      Pool.metrics_createdResources metrics `shouldBe` Pool.metrics_maxLiveResources metrics        destroyed <- Concurrent.readIORef destroyedRef       Pool.metrics_destroyedResources metrics `shouldBe` destroyed@@ -91,7 +94,7 @@         destroy _ =           Concurrent.atomicModifyIORef' counterRef $ \count -> (pred count, ()) -      pool <- Pool.newPool create destroy 10 60+      pool <- Pool.newPool create destroy 10 Pool.defaultSettings        Async.replicateConcurrently_ 200 $ do         Pool.withResource pool $ const $ Concurrent.threadDelay 1_000