diff --git a/ChangeLog.md b/ChangeLog.md new file mode 100644 --- /dev/null +++ b/ChangeLog.md @@ -0,0 +1,5 @@ +# simpoole + +### 0.0.0 + +This is the root version. diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/lib/Simpoole.hs b/lib/Simpoole.hs new file mode 100644 --- /dev/null +++ b/lib/Simpoole.hs @@ -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, ())) diff --git a/simpoole.cabal b/simpoole.cabal new file mode 100644 --- /dev/null +++ b/simpoole.cabal @@ -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 +maintainer: Ole Krüger +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 diff --git a/test/Main.hs b/test/Main.hs new file mode 100644 --- /dev/null +++ b/test/Main.hs @@ -0,0 +1,6 @@ +module Main (main) where + +import Test.Hspec + +main :: IO () +main = hspec $ pure ()