simpoole (empty) → 0.0.0

raw patch · 6 files changed

+279/−0 lines, 6 filesdep +basedep +concurrencydep +containerssetup-changed

Dependencies added: base, concurrency, containers, exceptions, hspec, simpoole, time

Files

ChangeLog.md +5/−0
LICENSE +27/−0
Setup.hs +2/−0
lib/Simpoole.hs +202/−0
simpoole.cabal +37/−0
test/Main.hs +6/−0

+ ChangeLog.md view

@@ -0,0 +1,5 @@+# simpoole++### 0.0.0++This is the root version.

+ LICENSE view

@@ -0,0 +1,27 @@+Copyright (c) Ole Krüger++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions+are met:++* Redistributions of source code must retain the above copyright+  notice, this list of conditions and the following disclaimer.+* Redistributions in binary form must reproduce the above copyright+  notice, this list of conditions and the following disclaimer in the+  documentation and/or other materials provided with the distribution.+* Neither the name of the author nor the+  names of its contributors may be used to endorse or promote products+  derived from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND+ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED+WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE+DISCLAIMED. IN NO EVENT SHALL THE AUTHORS OR CONTRIBUTORS BE LIABLE FOR ANY+DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES+(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;+LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND+ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS+SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

+ Setup.hs view

@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain

+ lib/Simpoole.hs view

@@ -0,0 +1,202 @@+{-# LANGUAGE DeriveTraversable #-}+{-# LANGUAGE DerivingStrategies #-}+{-# LANGUAGE NumericUnderscores #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE StrictData #-}+{-# LANGUAGE TypeApplications #-}++module Simpoole+  ( Pool+  , mapPool+  , newUnlimitedPool+  , newPool+  , withResource+  , poolMetrics++  , Metrics (..)+  )+where++import qualified Control.Concurrent.Classy as Concurrent+import qualified Control.Concurrent.Classy.Async as Async+import           Control.Monad (forever, unless, void)+import qualified Control.Monad.Catch as Catch+import           Control.Monad.IO.Class (MonadIO (liftIO))+import           Data.Foldable (for_)+import qualified Data.Sequence as Seq+import qualified Data.Time as Time+import           Numeric.Natural (Natural)++-- | Pool of resources+data Pool m a = Pool+  { pool_acquire :: m a+  , pool_return :: a -> m ()+  , pool_destroy :: a -> m ()+  , pool_metrics :: m (Metrics Natural)+  }++-- | Lift a natural transformation @m ~> n@ to @Pool m ~> Pool n@.+mapPool+  :: (forall x. m x -> n x)+  -> Pool m a+  -> Pool n a+mapPool to pool = Pool+  { pool_acquire = to $ pool_acquire pool+  , pool_return = to . pool_return pool+  , pool_destroy = to . pool_destroy pool+  , pool_metrics = to $ pool_metrics pool+  }++{-# INLINE mapPool #-}++-- | Pool resource+data Resource a =+  Resource+    Time.UTCTime+    -- ^ Last use time+    a+    -- ^ The resource itesemf++-- | Create a new pool that has no limit on how many resources it may create and hold.+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+  -> m (Pool m a)+newUnlimitedPool create destroy maxIdleTime = do+  leftOversRef <- Concurrent.newIORefN "leftOvers" Seq.empty++  metricRefs <- mkMetricRefs++  let+    wrappedCreate = do+      value <- create+      succIORef (metrics_createdResources metricRefs)+      pure value++    wrappedDestroy resource =+      destroy resource `Catch.finally` succIORef (metrics_destroyedResources metricRefs)++    acquireResource = do+      (mbResource, tailSize) <- Concurrent.atomicModifyIORef' leftOversRef $ \leftOvers ->+        case leftOvers of+          Resource _ head Seq.:<| tail -> (tail, (Just head, Seq.length tail))+          _empty -> (leftOvers, (Nothing, 0))+      resource <- maybe wrappedCreate pure mbResource+      maxIORef (metrics_maxLiveResources metricRefs) (fromIntegral tailSize + 1)+      pure resource++    returnResource value = do+      now <- liftIO Time.getCurrentTime+      Concurrent.atomicModifyIORef' leftOversRef $ \leftOvers ->+        (leftOvers Seq.:|> Resource now value, ())++  _reaperThread <- Async.asyncWithUnmaskN "reaperThread" $ \unmask -> unmask $ forever $ do+    now <- liftIO Time.getCurrentTime++    let isStillGood (Resource lastUse _) = Time.diffUTCTime now lastUse <= maxIdleTime+    oldResource <- Concurrent.atomicModifyIORef' leftOversRef (Seq.partition isStillGood)++    unless (null oldResource) $ void $+      Async.asyncN "destructionThread" $+        for_ oldResource $ \(Resource _ value) ->+          Catch.try @_ @Catch.SomeException $ wrappedDestroy value++    Concurrent.threadDelay 1_000_000++  pure Pool+    { pool_acquire = acquireResource+    , pool_return = returnResource+    , pool_destroy = wrappedDestroy+    , pool_metrics = readMetricRefs metricRefs+    }++-- | 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.+newPool+  :: (Concurrent.MonadConc m, MonadIO m, MonadFail m)+  => m a+  -- ^ Resource creation+  -> (a -> m ())+  -- ^ Resource destruction+  -> Int+  -- ^ Maximum number of resources to exist at the same time+  -> Time.NominalDiffTime+  -- ^ Maximum idle time (+-1s) after which a resource is destroyed+  -> m (Pool m a)+newPool create destroy maxElems maxIdleTime = do+  basePool <- newUnlimitedPool create destroy maxIdleTime+  maxElemBarrier <- Concurrent.newQSem maxElems++  let+    acquireResource = Catch.mask $ \restore -> do+      Concurrent.waitQSem maxElemBarrier+      restore (pool_acquire basePool)+        `Catch.onError` Concurrent.signalQSem maxElemBarrier++    giveBackResource f value = Catch.mask $ \restore ->+      restore (f basePool value)+        `Catch.finally` Concurrent.signalQSem maxElemBarrier++  pure Pool+    { pool_acquire = acquireResource+    , pool_return = giveBackResource pool_return+    , pool_destroy = giveBackResource pool_destroy+    , pool_metrics = pool_metrics basePool+    }++-- | Use a resource from the pool.+withResource :: Catch.MonadMask m => Pool m a -> (a -> m r) -> m r+withResource pool f =+  Catch.mask $ \restore -> do+    resource <- restore (pool_acquire pool)+    result <- restore (f resource) `Catch.onError` pool_destroy pool resource+    pool_return pool resource+    pure result++{-# INLINE withResource #-}++-- | Fetch pool metrics.+poolMetrics :: Pool m a -> m (Metrics Natural)+poolMetrics = pool_metrics++{-# INLINE poolMetrics #-}++---++-- | Pool metrics+data Metrics a = Metrics+  { metrics_createdResources :: a+  -- ^ Total number of resources created+  , metrics_destroyedResources :: a+  -- ^ Total number of resources destroyed+  , metrics_maxLiveResources :: a+  -- ^ Maximum number of resources that were alive simultaneously+  }+  deriving stock (Show, 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 ()+succIORef ref = Concurrent.atomicModifyIORef' ref (\x -> (succ x, ()))++-- | Replace the value in an IORef with the given value if the latter is greater.+maxIORef :: (Concurrent.MonadConc m, Ord a) => Concurrent.IORef m a -> a -> m ()+maxIORef ref y = Concurrent.atomicModifyIORef' ref (\x -> (max x y, ()))

+ simpoole.cabal view

@@ -0,0 +1,37 @@+cabal-version:      2.2+name:               simpoole+version:            0.0.0+category:           Data, Resources+synopsis:           Simple pool+description:        Provides a simple pool implementation.+author:             Ole Krüger <haskell-simpoole@vprsm.de>+maintainer:         Ole Krüger <haskell-simpoole@vprsm.de>+homepage:           https://github.com/vapourismo/simpoole+license:            BSD-3-Clause+license-file:       LICENSE+extra-source-files: ChangeLog.md+build-type:         Simple++source-repository head+  type:     git+  location: git://github.com/vapourismo/simpoole.git++common warnings+  ghc-options: -Wall -Wextra -Wno-name-shadowing++library+  import:           warnings+  default-language: Haskell2010+  ghc-options:      -Wall -Wextra -Wno-name-shadowing -Wredundant-constraints+  build-depends:    base >= 4.13 && < 5, time, exceptions, concurrency, containers+  hs-source-dirs:   lib+  exposed-modules:  Simpoole++test-suite simpoole-tests+  import:           warnings+  type:             exitcode-stdio-1.0+  default-language: Haskell2010+  ghc-options:      -Wall -Wextra -Wno-name-shadowing -threaded -with-rtsopts=-N+  build-depends:    base, simpoole, hspec >= 2.7.1+  hs-source-dirs:   test+  main-is:          Main.hs

+ test/Main.hs view

@@ -0,0 +1,6 @@+module Main (main) where++import Test.Hspec++main :: IO ()+main = hspec $ pure ()