diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,7 @@ +# resource-pool-0.5.1.0 (2026-08-15) +* Spawn a collector thread per stripe and make them wake up when appropriate + instead of polling every second. + # resource-pool-0.5.0.1 (2026-07-08) * Fix a bug where a thread waiting for a resource would get stuck in the queue indefinitely if resource creation failed in another thread. diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -7,3 +7,38 @@ A high-performance striped resource pooling implementation for Haskell based on [QSem](https://hackage.haskell.org/package/base/docs/Control-Concurrent-QSem.html). + +## Advice for library authors + +If your library creates a pool on behalf of its users, don't expose your own, +restricted set of pool parameters (size, TTL, ...) and construct the +`PoolConfig` internally. Such a config inevitably lags behind features of this +library (stripe count, labels, whatever comes next) and users can't take +advantage of them without waiting for you to mirror each one. + +Instead, take a function of type `IO a -> (a -> IO ()) -> PoolConfig a` as a +parameter. Your library supplies the resource creation and destruction actions: + +```haskell +createConnectionPool + :: ConnectionSettings + -> (IO Connection -> (Connection -> IO ()) -> PoolConfig Connection) + -> IO (Pool Connection) +createConnectionPool settings mkPoolConfig = + newPool $ mkPoolConfig connect disconnect + where + connect :: IO Connection + connect = ... + + disconnect :: Connection -> IO () + disconnect = ... +``` + +while users retain full control over the rest of the pool configuration: + +```haskell +pool <- createConnectionPool settings $ \create free -> + setPoolLabel "db" + . setNumStripes (Just 1) + $ defaultPoolConfig create free 60 10 +``` diff --git a/resource-pool.cabal b/resource-pool.cabal --- a/resource-pool.cabal +++ b/resource-pool.cabal @@ -1,7 +1,7 @@ cabal-version: 3.0 build-type: Simple name: resource-pool -version: 0.5.0.1 +version: 0.5.1.0 license: BSD-3-Clause license-file: LICENSE category: Data, Database, Network diff --git a/src/Data/Pool.hs b/src/Data/Pool.hs --- a/src/Data/Pool.hs +++ b/src/Data/Pool.hs @@ -73,10 +73,10 @@ q <- newEmptyTMVar writeTVar (stripeVar lp) $! stripe {queueR = Queue q (queueR stripe)} pure - $ waitForResource (stripeVar lp) q >>= \case + $ waitForResource lp q >>= \case Just a -> pure (a, lp) Nothing -> do - a <- createResource (poolConfig pool) `onException` restoreSize (stripeVar lp) + a <- createResource (poolConfig pool) `onException` restoreSize lp pure (a, lp) else takeAvailableResource pool lp stripe @@ -131,7 +131,7 @@ [] -> do writeTVar (stripeVar lp) $! stripe {available = available stripe - 1} pure $ do - a <- createResource (poolConfig pool) `onException` restoreSize (stripeVar lp) + a <- createResource (poolConfig pool) `onException` restoreSize lp pure (a, lp) Entry a _ : as -> do writeTVar (stripeVar lp) diff --git a/src/Data/Pool/Internal.hs b/src/Data/Pool/Internal.hs --- a/src/Data/Pool/Internal.hs +++ b/src/Data/Pool/Internal.hs @@ -11,6 +11,7 @@ import Control.Exception import Control.Monad import Data.Either +import Data.Function import Data.Hashable (hash) import Data.IORef import Data.List qualified as L @@ -23,13 +24,13 @@ data Pool a = Pool { poolConfig :: !(PoolConfig a) , localPools :: !(SmallArray (LocalPool a)) - , reaperRef :: !(IORef ()) } -- | A single, local pool. data LocalPool a = LocalPool { stripeId :: !Int , stripeVar :: !(TVar (Stripe a)) + , wakeupSem :: !WakeupSem , cleanerRef :: !(IORef ()) } @@ -39,6 +40,7 @@ data Stripe a = Stripe { available :: !Int , cache :: ![Entry a] + -- ^ Ordered by 'lastUsed', newest first (required by collector threads). , queue :: !(Queue a) , queueR :: !(Queue a) } @@ -79,9 +81,6 @@ -> Double -- ^ The number of seconds for which an unused resource is kept around. The -- smallest acceptable value is @0.5@. - -- - -- /Note:/ the elapsed time before destroying a resource may be a little - -- longer than requested, as the collector thread wakes at 1-second intervals. -> Int -- ^ The maximum number of resources to keep open __across all stripes__. The -- smallest acceptable value is @1@ per stripe. @@ -129,9 +128,10 @@ -- pool is garbage collected, it's recommended to manually call -- 'destroyAllResources' when you're done with the pool so that the resources -- are freed up as soon as possible. -newPool :: PoolConfig a -> IO (Pool a) +newPool :: forall a. PoolConfig a -> IO (Pool a) newPool pc = do - when (poolCacheTTL pc < 0.5) $ do + -- Arranged so that NaN is also rejected as it breaks the collector thread. + unless (poolCacheTTL pc >= 0.5) $ do error "poolCacheTTL must be at least 0.5" when (poolMaxResources pc < 1) $ do error "poolMaxResources must be at least 1" @@ -151,30 +151,36 @@ , queue = Empty , queueR = Empty } - -- When the local pool goes out of scope, free its resources. - void . mkWeakIORef ref $ cleanStripe (const True) (freeResource pc) stripe + sem <- newWakeupSem + mask_ $ do + -- The collector must not reference 'ref', otherwise the finalizer below + -- would never run. + collectorId <- forkIOWithUnmask $ \unmask -> unmask $ do + tid <- myThreadId + labelThread tid + $ "resource-pool: collector #" + ++ show n + ++ " (" + ++ T.unpack (pcLabel pc) + ++ ")" + collector sem stripe + void . mkWeakIORef ref $ do + -- When the local pool goes out of scope, stop its collector and free + -- its resources. + killThread collectorId + cleanStripe (const True) (freeResource pc) stripe pure LocalPool { stripeId = n , stripeVar = stripe + , wakeupSem = sem , cleanerRef = ref } - mask_ $ do - ref <- newIORef () - collectorA <- forkIOWithUnmask $ \unmask -> unmask $ do - tid <- myThreadId - labelThread tid $ "resource-pool: collector (" ++ T.unpack (pcLabel pc) ++ ")" - collector pools - void . mkWeakIORef ref $ do - -- When the pool goes out of scope, stop the collector. Resources existing - -- in stripes will be taken care by their cleaners. - killThread collectorA - pure - Pool - { poolConfig = pc - , localPools = pools - , reaperRef = ref - } + pure + Pool + { poolConfig = pc + , localPools = pools + } where stripeResources :: Int -> [(Int, Int)] stripeResources numStripes = @@ -186,12 +192,37 @@ 0 -> acc rest -> r + 1 : addRest rs (rest - 1) - -- Collect stale resources from the pool once per second. - collector pools = forever $ do - threadDelay 1000000 + collector :: WakeupSem -> TVar (Stripe a) -> IO r + collector sem stripe = forever $ do + atomically $ wakeupWait sem + -- The wakeup signal means that a resource was just put into the empty + -- cache, so neither it nor any resource cached after it can expire + -- earlier than TTL from now. + -- + -- Waiting a full TTL before looking at the cache also caps signal-driven + -- wakeups at one per TTL when resources are rapidly taken from and put + -- back into an almost-empty cache. + waitUntil . (+ poolCacheTTL pc) =<< getMonotonicTime + fix $ \loop -> + (cache <$> readTVarIO stripe) >>= \case + [] -> pure () + entries -> do + -- Nothing can expire before the last entry. + waitUntil $ lastUsed (L.last entries) + poolCacheTTL pc + now <- getMonotonicTime + let isStale e = now - lastUsed e > poolCacheTTL pc + cleanStripe isStale (freeResource pc) stripe + loop + + waitUntil :: Double -> IO () + waitUntil deadline = do now <- getMonotonicTime - let isStale e = now - lastUsed e > poolCacheTTL pc - mapM_ (cleanStripe isStale (freeResource pc) . stripeVar) pools + let micros = (deadline - now) * 1000000 + when (micros > 0) $ do + threadDelay + $ if micros >= fromIntegral (maxBound :: Int) + then maxBound + else ceiling micros -- | Destroy a resource. -- @@ -201,7 +232,7 @@ destroyResource pool lp a = mask_ $ do atomically $ do stripe <- readTVar (stripeVar lp) - newStripe <- signal stripe Nothing + newStripe <- signal lp stripe Nothing writeTVar (stripeVar lp) $! newStripe freeResource (poolConfig pool) a @@ -209,7 +240,7 @@ putResource :: LocalPool a -> a -> IO () putResource lp a = atomically $ do stripe <- readTVar (stripeVar lp) - newStripe <- signal stripe (Just a) + newStripe <- signal lp stripe (Just a) writeTVar (stripeVar lp) $! newStripe -- | Destroy all resources in all stripes in the pool. @@ -233,6 +264,24 @@ ---------------------------------------- -- Helpers +-- | Binary semaphore for signaling a collector thread to wake up. +newtype WakeupSem = WakeupSem (TVar Bool) + +newWakeupSem :: IO WakeupSem +newWakeupSem = WakeupSem <$> newTVarIO False + +wakeupSignal :: WakeupSem -> STM () +wakeupSignal (WakeupSem var) = writeTVar var True + +wakeupWait :: WakeupSem -> STM () +wakeupWait (WakeupSem var) = do + signaled <- readTVar var + if signaled + then writeTVar var False + else retry + +---------------------------------------- + -- | Get a local pool. getLocalPool :: SmallArray (LocalPool a) -> IO (LocalPool a) getLocalPool pools = do @@ -267,34 +316,34 @@ stripes = sizeofSmallArray pools -- | Wait for the resource to be put into a given 'TMVar'. -waitForResource :: TVar (Stripe a) -> TMVar (Maybe a) -> IO (Maybe a) -waitForResource mstripe q = atomically (takeTMVar q) `onException` cleanup +waitForResource :: LocalPool a -> TMVar (Maybe a) -> IO (Maybe a) +waitForResource lp q = atomically (takeTMVar q) `onException` cleanup where cleanup = atomically $ do - stripe <- readTVar mstripe + stripe <- readTVar (stripeVar lp) newStripe <- tryTakeTMVar q >>= \case Just ma -> do -- Between entering the exception handler and taking ownership of -- the stripe we got the resource we wanted. We don't need it -- anymore though, so pass it to someone else. - signal stripe ma + signal lp stripe ma Nothing -> do -- If we're still waiting, fill up the TMVar with an undefined value -- so that 'signal' can discard our TMVar from the queue. putTMVar q $ error "unreachable" pure stripe - writeTVar mstripe $! newStripe + writeTVar (stripeVar lp) $! newStripe -- | If an exception is received while a resource is being created, restore the -- original size of the stripe. -restoreSize :: TVar (Stripe a) -> IO () -restoreSize mstripe = atomically $ do - stripe <- readTVar mstripe +restoreSize :: LocalPool a -> IO () +restoreSize lp = atomically $ do + stripe <- readTVar (stripeVar lp) -- Signal needs to be called so that if there are threads waiting for a -- resource, one of them wakes up and attempts the creation itself. - newStripe <- signal stripe Nothing - writeTVar mstripe $! newStripe + newStripe <- signal lp stripe Nothing + writeTVar (stripeVar lp) $! newStripe -- | Free resource entries in the stripes that fulfil a given condition. cleanStripe @@ -312,14 +361,15 @@ -- the number of resources taken from the pool. writeTVar mstripe $! stripe {cache = fresh} pure $ map entry stale - -- We need to ignore exceptions in the 'free' function, otherwise if an - -- exception is thrown half-way, we leak the rest of the resources. Also, - -- asynchronous exceptions need to be hard masked here we need to run 'free' - -- for all resources. - uninterruptibleMask $ \release -> do - rs <- forM stale $ try @SomeException . release . free - -- If any async exception arrived in between, propagate it. - rethrowFirstAsyncException $ lefts rs + -- We need to catch all exceptions in the 'free' function, otherwise if an + -- exception was thrown half-way through the traversal of 'stale', we would + -- leak the rest of the resources. + -- + -- The loop outside of a call to 'free' is not interruptible, so asynchronous + -- exceptions can only be delivered inside 'free'. If such a situation arises, + -- propagate the first one we got. + rs <- forM stale $ try @SomeException . free + rethrowFirstAsyncException $ lefts rs where rethrowFirstAsyncException = \case [] -> pure () @@ -327,14 +377,18 @@ | Just SomeAsyncException {} <- fromException e -> throwIO e | otherwise -> rethrowFirstAsyncException es -signal :: forall a. Stripe a -> Maybe a -> STM (Stripe a) -signal stripe ma = +signal :: forall a. LocalPool a -> Stripe a -> Maybe a -> STM (Stripe a) +signal lp stripe ma = + -- When cache changes from empty to non-empty, the collector needs to be + -- signaled via wakeupSem. if available stripe == 0 then loop (queue stripe) (queueR stripe) else do newCache <- case ma of Just a -> do now <- unsafeIOToSTM getMonotonicTime + when (null $ cache stripe) $ do + wakeupSignal $ wakeupSem lp pure $ Entry a now : cache stripe Nothing -> pure $ cache stripe pure @@ -348,6 +402,7 @@ newCache <- case ma of Just a -> do now <- unsafeIOToSTM getMonotonicTime + wakeupSignal $ wakeupSem lp pure [Entry a now] Nothing -> pure [] pure diff --git a/src/Data/Pool/Introspection.hs b/src/Data/Pool/Introspection.hs --- a/src/Data/Pool/Introspection.hs +++ b/src/Data/Pool/Introspection.hs @@ -72,7 +72,7 @@ q <- newEmptyTMVar writeTVar (stripeVar lp) $! stripe {queueR = Queue q (queueR stripe)} pure - $ waitForResource (stripeVar lp) q >>= \case + $ waitForResource lp q >>= \case Just a -> do t2 <- getMonotonicTime let res = @@ -88,7 +88,7 @@ pure (res, lp) Nothing -> do t2 <- getMonotonicTime - a <- createResource (poolConfig pool) `onException` restoreSize (stripeVar lp) + a <- createResource (poolConfig pool) `onException` restoreSize lp t3 <- getMonotonicTime let res = Resource @@ -141,7 +141,7 @@ writeTVar (stripeVar lp) $! stripe {available = newAvailable} pure $ do t2 <- getMonotonicTime - a <- createResource (poolConfig pool) `onException` restoreSize (stripeVar lp) + a <- createResource (poolConfig pool) `onException` restoreSize lp t3 <- getMonotonicTime let res = Resource diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -39,6 +39,9 @@ "config validation" [ testCase "rejects too small poolCacheTTL" $ do expectError . newPool $ poolConfig_ 0.4 1 + , testCase "rejects a NaN poolCacheTTL" $ do + -- A NaN would send the collector into a busy loop. + expectError . newPool $ poolConfig_ (0 / 0) 1 , testCase "rejects non-positive poolMaxResources" $ do expectError . newPool $ poolConfig_ 100 0 , testCase "rejects non-positive number of stripes" $ do @@ -131,9 +134,45 @@ 0.5 5 _ <- withResource pool pure - -- The collector thread wakes up every second. + -- The collector should free the resource promptly after its TTL + -- expires. waitUntil "the resource is collected" $ (== 1) <$> readIORef freedC readIORef createdC >>= assertEqual "created resources" 1 + , testCase "the collector runs again after the cache is refilled" $ do + freedC <- newIORef (0 :: Int) + pool <- + newPool + $ defaultPoolConfig + (pure ()) + (\_ -> atomicModifyIORef' freedC $ \n -> (n + 1, ())) + 0.5 + 5 + _ <- withResource pool pure + waitUntil "the first resource is collected" $ (== 1) <$> readIORef freedC + -- The collector went back to sleep on an empty cache; putting a new + -- resource into it needs to wake it up again. + _ <- withResource pool pure + waitUntil "the second resource is collected" $ (== 2) <$> readIORef freedC + , testCase "entries not yet stale in a collection round are collected later" $ do + freedC <- newIORef (0 :: Int) + pool <- + newPool + $ defaultPoolConfig + (pure ()) + (\_ -> atomicModifyIORef' freedC $ \n -> (n + 1, ())) + 0.5 + 5 + (r1, lp1) <- takeResource pool + (r2, lp2) <- takeResource pool + putResource lp1 r1 + -- Put the second resource back only after a while, so that when the + -- collector wakes up to free the first one, the second one is not yet + -- stale and has to be freed in a later collection round, even though + -- the pool sees no further activity. + threadDelay 300000 + putResource lp2 r2 + waitUntil "the first resource is collected" $ (>= 1) <$> readIORef freedC + waitUntil "the second resource is collected" $ (== 2) <$> readIORef freedC ] ----------------------------------------