diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,28 @@ +# v1.5.1.0 + +## Fixes + +- Fix pool capacity leak when `use` gets interrupted during connection + establishment, connection refresh, or an observation (#60) +- Fix pool state corruption from throwing observation handlers; handlers + are now contained (see the updated `observationHandler` docs) +- Fix background reaper dying permanently from a failed connection + release or observation +- Fix `release` leaking flushed entries on failure +- Fix waiters waiting forever on a garbage-collected pool + +## Added + +- `reuseOrder` setting (#59): LIFO reuse lets the pool shrink to the + connections the load needs and keeps caches warm; FIFO stays the default +- One background timer for acquisition timeouts instead of one + uncancelable timer per waiting request (#58) + +## Changed + +- Connections are handed to waiting requests in arrival order; a request + no longer takes a connection while others wait (#58) + # v1.5.0.1 - Allow newer pqi. diff --git a/hasql-pool.cabal b/hasql-pool.cabal --- a/hasql-pool.cabal +++ b/hasql-pool.cabal @@ -1,6 +1,6 @@ cabal-version: 3.0 name: hasql-pool -version: 1.5.0.1 +version: 1.5.1.0 category: Hasql, Database, PostgreSQL synopsis: Pool of connections for Hasql homepage: https://github.com/nikita-volkov/hasql-pool @@ -78,6 +78,7 @@ other-modules: Hasql.Pool.Config.Config + Hasql.Pool.Config.ReuseOrder Hasql.Pool.Config.Setting Hasql.Pool.Prelude Hasql.Pool.SessionErrorDestructors @@ -107,6 +108,7 @@ Specs.Config.InitSessionSpec Specs.Helpers.Sessions.CountConnectionsSpec Specs.Helpers.Sessions.GetSettingSpec + Specs.InterruptionSpec Specs.ReleaseSpec Specs.SpecHook Specs.UsageError.AcquisitionTimeoutSpec diff --git a/src/integration-tests/Helpers/Scripts.hs b/src/integration-tests/Helpers/Scripts.hs --- a/src/integration-tests/Helpers/Scripts.hs +++ b/src/integration-tests/Helpers/Scripts.hs @@ -3,6 +3,7 @@ import Hasql.Connection.Settings qualified as Connection.Settings import Hasql.Pool qualified as Pool import Hasql.Pool.Config qualified as Config +import Hasql.Pool.Observation (Observation) import Hasql.Session qualified as Session import Pqi qualified import Prelude @@ -44,6 +45,36 @@ appName <- generateName "hasql-pool-test-" onTaggedPool poolSize acqTimeout maxLifetime maxIdletime appName scopeParams (cont appName) +onTaggedPoolWithObservationHandler :: Int -> DiffTime -> DiffTime -> DiffTime -> (Observation -> IO ()) -> Text -> ScopeParams -> (Pool.Pool -> IO ()) -> IO () +onTaggedPoolWithObservationHandler poolSize acqTimeout maxLifetime maxIdletime observationHandler appName (adapter, host, port) = + bracket + ( Pool.acquire + adapter + ( Config.settings + [ Config.size poolSize, + Config.acquisitionTimeout acqTimeout, + Config.agingTimeout maxLifetime, + Config.idlenessTimeout maxIdletime, + Config.observationHandler observationHandler, + Config.staticConnectionSettings + ( mconcat + [ Connection.Settings.hostAndPort host (fromIntegral port), + Connection.Settings.user "postgres", + Connection.Settings.password "", + Connection.Settings.dbname "postgres", + Connection.Settings.applicationName appName + ] + ) + ] + ) + ) + Pool.release + +onAutotaggedPoolWithObservationHandler :: Int -> DiffTime -> DiffTime -> DiffTime -> (Observation -> IO ()) -> ScopeParams -> (Text -> Pool.Pool -> IO ()) -> IO () +onAutotaggedPoolWithObservationHandler poolSize acqTimeout maxLifetime maxIdletime observationHandler scopeParams cont = do + appName <- generateName "hasql-pool-test-" + onTaggedPoolWithObservationHandler poolSize acqTimeout maxLifetime maxIdletime observationHandler appName scopeParams (cont appName) + onTaggedPoolWithInitSession :: Int -> DiffTime -> DiffTime -> DiffTime -> Session.Session () -> Text -> ScopeParams -> (Pool.Pool -> IO ()) -> IO () onTaggedPoolWithInitSession poolSize acqTimeout maxLifetime maxIdletime initSession appName (adapter, host, port) = bracket @@ -73,6 +104,36 @@ onAutotaggedPoolWithInitSession poolSize acqTimeout maxLifetime maxIdletime initSession (adapter, host, port) cont = do appName <- generateName "hasql-pool-test-" onTaggedPoolWithInitSession poolSize acqTimeout maxLifetime maxIdletime initSession appName (adapter, host, port) (cont appName) + +onTaggedPoolWithReuseOrder :: Int -> DiffTime -> DiffTime -> DiffTime -> Config.ReuseOrder -> Text -> ScopeParams -> (Pool.Pool -> IO ()) -> IO () +onTaggedPoolWithReuseOrder poolSize acqTimeout maxLifetime maxIdletime reuseOrder appName (adapter, host, port) = + bracket + ( Pool.acquire + adapter + ( Config.settings + [ Config.size poolSize, + Config.acquisitionTimeout acqTimeout, + Config.agingTimeout maxLifetime, + Config.idlenessTimeout maxIdletime, + Config.reuseOrder reuseOrder, + Config.staticConnectionSettings + ( mconcat + [ Connection.Settings.hostAndPort host (fromIntegral port), + Connection.Settings.user "postgres", + Connection.Settings.password "", + Connection.Settings.dbname "postgres", + Connection.Settings.applicationName appName + ] + ) + ] + ) + ) + Pool.release + +onAutotaggedPoolWithReuseOrder :: Int -> DiffTime -> DiffTime -> DiffTime -> Config.ReuseOrder -> ScopeParams -> (Text -> Pool.Pool -> IO ()) -> IO () +onAutotaggedPoolWithReuseOrder poolSize acqTimeout maxLifetime maxIdletime reuseOrder scopeParams cont = do + appName <- generateName "hasql-pool-test-" + onTaggedPoolWithReuseOrder poolSize acqTimeout maxLifetime maxIdletime reuseOrder appName scopeParams (cont appName) onDefaultTaggedPool :: ScopeParams -> (Text -> Pool.Pool -> IO ()) -> IO () onDefaultTaggedPool = diff --git a/src/integration-tests/Helpers/Sessions.hs b/src/integration-tests/Helpers/Sessions.hs --- a/src/integration-tests/Helpers/Sessions.hs +++ b/src/integration-tests/Helpers/Sessions.hs @@ -6,6 +6,7 @@ getSetting, countConnections, sleep, + backendPid, ) where @@ -71,3 +72,10 @@ where statement = Statement.preparable "SELECT pg_sleep($1)" encoder Decoders.noResult encoder = Encoders.param (Encoders.nonNullable Encoders.float8) + +backendPid :: Session.Session Int32 +backendPid = + Session.statement () statement + where + statement = Statement.preparable "SELECT pg_backend_pid()" Encoders.noParams decoder + decoder = Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.int4)) diff --git a/src/integration-tests/Specs/Config/IdlenessTimeoutSpec.hs b/src/integration-tests/Specs/Config/IdlenessTimeoutSpec.hs --- a/src/integration-tests/Specs/Config/IdlenessTimeoutSpec.hs +++ b/src/integration-tests/Specs/Config/IdlenessTimeoutSpec.hs @@ -1,6 +1,7 @@ module Specs.Config.IdlenessTimeoutSpec where import Hasql.Pool +import Hasql.Pool.Config qualified as Config import Helpers.Scripts qualified as Scripts import Helpers.Sessions qualified as Sessions import Prelude @@ -42,3 +43,20 @@ threadDelay 1_500_000 -- 1.5s res3 <- use pool1 $ Sessions.countConnections appName2 res3 `shouldBe` Right 0 + + it "Times out the connections that aren't needed with LIFO" \scopeParams -> do + -- 0.5s connection idle time + Scripts.onAutotaggedPoolWithReuseOrder 2 10 1_800 0.5 Config.LIFO scopeParams \appName pool -> do + held <- newEmptyMVar + done <- newEmptyMVar + -- Establish two connections. + _ <- forkIO (void (use pool (liftIO (putMVar held () *> takeMVar done)))) + takeMVar held + use pool Sessions.selectOne `shouldReturn` Right 1 + putMVar done () + -- Keep using the pool one pool request at a time, which needs only one + -- connection, for longer than the idle time and the reaper's 1s tick. + forM_ [1 :: Int .. 15] \_ -> do + use pool Sessions.selectOne `shouldReturn` Right 1 + threadDelay 100_000 -- 0.1s + use pool (Sessions.countConnections appName) `shouldReturn` Right 1 diff --git a/src/integration-tests/Specs/InterruptionSpec.hs b/src/integration-tests/Specs/InterruptionSpec.hs new file mode 100644 --- /dev/null +++ b/src/integration-tests/Specs/InterruptionSpec.hs @@ -0,0 +1,163 @@ +module Specs.InterruptionSpec where + +import Control.Exception (SomeException, try) +import Hasql.Pool +import Hasql.Pool qualified as Pool +import Hasql.Pool.Config qualified as Config +import Hasql.Pool.Observation +import Helpers.Scripts qualified as Scripts +import Helpers.Sessions qualified as Sessions +import Prelude +import Test.Hspec + +-- | Regression tests for the pool capacity getting lost when a pool request +-- gets interrupted anywhere between the enrollment and the session +-- (https://github.com/nikita-volkov/hasql-pool/issues/60), and for the pool +-- surviving failing observation handlers. +spec :: SpecWith Scripts.ScopeParams +spec = do + it "Restores the pool capacity when the pool request is interrupted while observing the connection attempt" \scopeParams -> + onGatedPool connectingObs 1 10 1_800 1_800 scopeParams \gate pool -> do + interruptAt gate (\pool -> void (use pool Sessions.selectOne)) pool + use pool Sessions.selectOne `shouldReturn` Right 1 + + it "Restores the pool capacity when the pool request is interrupted while observing the idle-eviction" \scopeParams -> + onGatedPool idleEvictionObs 1 10 1_800 0.3 scopeParams \gate pool -> do + -- Establish a connection and let it go idle. + _ <- use pool Sessions.selectOne + threadDelay 600_000 -- 0.6s + interruptAt gate (\pool -> void (use pool Sessions.selectOne)) pool + use pool Sessions.selectOne `shouldReturn` Right 1 + + it "Preserves the dequeued connection when the pool request is interrupted while observing its use" \scopeParams -> + onGatedPool inUseObs 1 10 1_800 1_800 scopeParams \gate pool -> do + -- Establish a connection and return it to the pool. + _ <- use pool Sessions.selectOne + interruptAt gate (\pool -> void (use pool Sessions.selectOne)) pool + use pool Sessions.selectOne `shouldReturn` Right 1 + + it "Succeeds the session when the observation handler throws synchronously during the connection attempt" \scopeParams -> + Scripts.onAutotaggedPoolWithObservationHandler 1 10 1_800 1_800 (throwAt connectingObs) scopeParams \_ pool -> do + use pool Sessions.selectOne `shouldReturn` Right 1 + + it "Succeeds the session when the observation handler throws synchronously during the idle-eviction" \scopeParams -> + Scripts.onAutotaggedPoolWithObservationHandler 1 10 1_800 0.3 (throwAt idleEvictionObs) scopeParams \_ pool -> do + _ <- use pool Sessions.selectOne + threadDelay 600_000 -- 0.6s + use pool Sessions.selectOne `shouldReturn` Right 1 + + it "Succeeds the session when the observation handler throws synchronously while observing the use" \scopeParams -> + Scripts.onAutotaggedPoolWithObservationHandler 1 10 1_800 1_800 (throwAt inUseObs) scopeParams \_ pool -> do + use pool Sessions.selectOne `shouldReturn` Right 1 + + it "The reaper survives a throwing observation handler and keeps reaping on later ticks" \scopeParams -> + Scripts.onAutotaggedPool 1 10 1_800 1_800 scopeParams \_ counterPool -> + Scripts.onAutotaggedPoolWithObservationHandler 1 10 1_800 0.3 (throwAt idleEvictionObs) scopeParams \appName pool -> do + -- Establish a connection and let it go idle past the reaper's tick. + _ <- use pool Sessions.selectOne + threadDelay 1_600_000 -- 1.6s + -- The connection got reaped and the reaper is still alive. + use counterPool (Sessions.countConnections appName) `shouldReturn` Right 0 + use pool Sessions.selectOne `shouldReturn` Right 1 + -- It keeps reaping on the later ticks. + threadDelay 1_600_000 -- 1.6s + use counterPool (Sessions.countConnections appName) `shouldReturn` Right 0 + + it "The pool remains usable after the release with a throwing observation handler" \scopeParams -> + Scripts.onAutotaggedPoolWithObservationHandler 1 10 1_800 1_800 (throwAt releaseObs) scopeParams \_ pool -> do + _ <- use pool Sessions.selectOne + release pool + use pool Sessions.selectOne `shouldReturn` Right 1 + + it "Resolves the pool requests waiting on a garbage-collected pool" \scopeParams -> do + (adapter, host, port) <- pure scopeParams + result <- newEmptyMVar + -- Inside this block the pool is only reachable from the waiting pool + -- request. Once collected, the finalizer resolves it, and if it is not + -- collectible, the acquisition timeout does. Either way it does not wait + -- forever. + do + pool <- + Pool.acquire + adapter + ( Config.settings + [ -- No capacity: every pool request waits. + Config.size 0, + Config.acquisitionTimeout 0.5 + ] + ) + _ <- forkIO (putMVar result =<< use pool Sessions.selectOne) + threadDelay 100_000 -- 0.1s, to let the pool request start waiting + pure () + performGC + performGC + res <- takeMVar result + res `shouldBe` Left AcquisitionTimeoutUsageError + +-- | Runs the action in a pool request interrupted at the gated observation. +-- The pool request is expected to die with the interruption. +interruptAt :: Gate -> (Pool.Pool -> IO ()) -> Pool.Pool -> IO () +interruptAt gate action pool = do + armGate gate + done <- newEmptyMVar + thread <- forkIO do + _ <- try @SomeException (action pool) + putMVar done () + takeMVar (gateEntered gate) + killThread thread + disarmGate gate + takeMVar done + +-- | Runs the continuation in a pool with a gated observation handler. +onGatedPool :: (Observation -> Bool) -> Int -> DiffTime -> DiffTime -> DiffTime -> Scripts.ScopeParams -> (Gate -> Pool.Pool -> IO ()) -> IO () +onGatedPool matches poolSize acqTimeout maxLifetime maxIdletime scopeParams cont = do + gate <- newGate matches + Scripts.onAutotaggedPoolWithObservationHandler poolSize acqTimeout maxLifetime maxIdletime (gateHandler gate) scopeParams \appName pool -> cont gate pool + +-- | A gate for stopping a pool request at a chosen observation and letting the +-- test interrupt it there. While the gate is armed, the pool request stops at +-- the first matching observation, waiting to be released or interrupted. +data Gate = Gate + { gateMatches :: Observation -> Bool, + gateEntered :: MVar (), + gateProceed :: MVar (), + gateArmed :: IORef Bool + } + +newGate :: (Observation -> Bool) -> IO Gate +newGate matches = Gate matches <$> newEmptyMVar <*> newEmptyMVar <*> newIORef False + +armGate, disarmGate :: Gate -> IO () +armGate gate = writeIORef (gateArmed gate) True +disarmGate gate = writeIORef (gateArmed gate) False + +-- | The observation handler stopping at the gate. +gateHandler :: Gate -> Observation -> IO () +gateHandler gate observation = + when (gateMatches gate observation) do + armed <- readIORef (gateArmed gate) + when armed do + putMVar (gateEntered gate) () + takeMVar (gateProceed gate) + +-- | The observations the tests stop or throw at. +connectingObs, idleEvictionObs, inUseObs, releaseObs :: Observation -> Bool +connectingObs = \case + ConnectionObservation _ ConnectingConnectionStatus -> True + _ -> False +idleEvictionObs = \case + ConnectionObservation _ (TerminatedConnectionStatus IdlenessConnectionTerminationReason) -> True + _ -> False +inUseObs = \case + ConnectionObservation _ InUseConnectionStatus -> True + _ -> False +releaseObs = \case + ConnectionObservation _ (TerminatedConnectionStatus ReleaseConnectionTerminationReason) -> True + _ -> False + +-- | An observation handler that throws synchronously at the matching +-- observations, and passes all the other observations through. +throwAt :: (Observation -> Bool) -> Observation -> IO () +throwAt matches observation = + when (matches observation) do + error "Synchronous exception from the observation handler" diff --git a/src/integration-tests/Specs/UsageError/AcquisitionTimeoutSpec.hs b/src/integration-tests/Specs/UsageError/AcquisitionTimeoutSpec.hs --- a/src/integration-tests/Specs/UsageError/AcquisitionTimeoutSpec.hs +++ b/src/integration-tests/Specs/UsageError/AcquisitionTimeoutSpec.hs @@ -1,6 +1,6 @@ module Specs.UsageError.AcquisitionTimeoutSpec where -import Control.Concurrent.Async (race) +import Control.Concurrent.Async (forConcurrently, race) import Hasql.Pool import Helpers.Scripts qualified as Scripts import Helpers.Sessions qualified as Sessions @@ -31,3 +31,15 @@ res `shouldBe` Right (Left AcquisitionTimeoutUsageError) -- 0.5s diffUTCTime t1 t0 `shouldSatisfy` (< 0.5) + + it "Gets produced for every waiting pool request, and the pool remains usable" \scopeParams -> + -- 100ms timeout + Scripts.onAutotaggedPool 1 0.1 1_800 1_800 scopeParams \_ pool -> do + held <- newEmptyMVar + done <- newEmptyMVar + _ <- forkIO (void (use pool (liftIO (putMVar held () *> takeMVar done)))) + takeMVar held + results <- forConcurrently [1 .. 3 :: Int] \_ -> use pool Sessions.selectOne + results `shouldBe` replicate 3 (Left AcquisitionTimeoutUsageError) + putMVar done () + use pool Sessions.selectOne `shouldReturn` Right 1 diff --git a/src/integration-tests/Specs/UseSpec.hs b/src/integration-tests/Specs/UseSpec.hs --- a/src/integration-tests/Specs/UseSpec.hs +++ b/src/integration-tests/Specs/UseSpec.hs @@ -6,6 +6,7 @@ import Hasql.Encoders qualified as Encoders import Hasql.Errors qualified as Errors import Hasql.Pool +import Hasql.Pool.Config qualified as Config import Hasql.Session qualified as Session import Hasql.Statement qualified as Statement import Helpers.Scripts qualified as Scripts @@ -89,6 +90,52 @@ res <- use pool Sessions.selectOne res `shouldSatisfy` isRight + it "Gives a freed connection to a pool request waiting for it rather than to a later one" \scopeParams -> + Scripts.onAutotaggedPool 1 10 1_800 1_800 scopeParams \_ pool -> do + order <- newIORef [] + held <- newEmptyMVar + done <- newEmptyMVar + finished <- newEmptyMVar + let record name = use pool (liftIO (atomicModifyIORef' order \names -> (names <> [name], ()))) + -- Requests a connection again as soon as it returns one, before the + -- waiting pool request gets to run. + _ <- forkIO do + _ <- use pool (liftIO (putMVar held () *> takeMVar done)) + _ <- record "later" + putMVar finished () + takeMVar held + _ <- forkIO (record "waiting" *> putMVar finished ()) + -- Give the waiting pool request time to start waiting. + threadDelay 100_000 + putMVar done () + replicateM_ 2 (takeMVar finished) + readIORef order `shouldReturn` ["waiting", "later"] + + it "Loses no capacity when a pool request waiting for a connection is interrupted" \scopeParams -> + Scripts.onAutotaggedPool 1 10 1_800 1_800 scopeParams \_ pool -> do + held <- newEmptyMVar + done <- newEmptyMVar + _ <- forkIO (void (use pool (liftIO (putMVar held () *> takeMVar done)))) + takeMVar held + waiting <- forkIO (void (use pool Sessions.selectOne)) + -- Give the pool request time to start waiting. + threadDelay 100_000 + killThread waiting + putMVar done () + use pool Sessions.selectOne `shouldReturn` Right 1 + + it "Reuses the most recently returned connection with LIFO" \scopeParams -> + Scripts.onAutotaggedPoolWithReuseOrder 2 10 1_800 1_800 Config.LIFO scopeParams \_ pool -> do + (first, second) <- returnTwoConnections pool + first `shouldNotBe` second + use pool Sessions.backendPid `shouldReturn` Right second + + it "Reuses the least recently returned connection by default" \scopeParams -> + Scripts.onAutotaggedPool 2 10 1_800 1_800 scopeParams \_ pool -> do + (first, second) <- returnTwoConnections pool + first `shouldNotBe` second + use pool Sessions.backendPid `shouldReturn` Right first + it "Cached type errors cause eviction of connection" \scopeParams -> do typeName <- Text.replace "-" "_" <$> Scripts.generateName "cached_type_" Scripts.onAutotaggedPool 1 10 1_800 1_800 scopeParams \_ pool -> do @@ -131,3 +178,24 @@ driverError = Session.onLibpqConnection \connection -> pure (Left (Errors.DriverSessionError "synthetic driver error"), connection) + +-- | Establish two connections, return the first before the second, and give +-- their backend PIDs. +returnTwoConnections :: Pool -> IO (Int32, Int32) +returnTwoConnections pool = do + held <- newEmptyMVar + done <- newEmptyMVar + returned <- newEmptyMVar + _ <- forkIO do + res <- use pool (Sessions.backendPid <* liftIO (putMVar held () *> takeMVar done)) + putMVar returned res + takeMVar held + res <- use pool do + second <- Sessions.backendPid + first <- liftIO do + putMVar done () + takeMVar returned + pure (first, second) + case res of + Right (Right first, second) -> pure (first, second) + _ -> fail ("Unexpected result: " <> show res) diff --git a/src/library/Hasql/Pool.hs b/src/library/Hasql/Pool.hs --- a/src/library/Hasql/Pool.hs +++ b/src/library/Hasql/Pool.hs @@ -16,6 +16,7 @@ import Hasql.Connection.Settings qualified as Connection.Settings import Hasql.Errors qualified as Errors import Hasql.Pool.Config.Config qualified as Config +import Hasql.Pool.Config.ReuseOrder (ReuseOrder (..)) import Hasql.Pool.Observation import Hasql.Pool.Prelude import Hasql.Pool.SessionErrorDestructors qualified as ErrorsDestruction @@ -38,6 +39,33 @@ entryIsIdle maxIdletime now Entry {..} = now > entryUseTimeNSec + maxIdletime +-- | A pool request waiting for a connection. +data Waiter = Waiter + { -- | When the pool request times out, in nanoseconds of the monotonic clock. + waiterDeadline :: Word64, + -- | What the pool request gets. + waiterGiven :: TMVar Given, + -- | Whether the pool request stopped waiting, so that it gets passed over. + waiterGone :: TVar Bool + } + +-- | What a waiting pool request gets: an idle connection, or the slot of a +-- closed one to establish a connection in, with the reuse flag at the time, +-- or a timeout. +data Given + = Given (TVar Bool) (Maybe Entry) + | TimedOut + +-- | What a pool request holds after the enrollment transaction commits and +-- has to give back on any failure. +data Holding + = -- | An entry dequeued from the idle connections. + EntryHolding (TVar Bool) Entry + | -- | A capacity unit taken from the pool. + SlotHolding (TVar Bool) + | -- | A spot in the waiting queue. + WaitingHolding Waiter + -- | Pool of connections to DB. data Pool = Pool { -- | Pool size. @@ -46,22 +74,28 @@ poolAdapter :: Pqi.Adapter, -- | Connection settings. poolFetchConnectionSettings :: IO Connection.Settings.Settings, - -- | Acquisition timeout, in microseconds. - poolAcquisitionTimeout :: Int, + -- | Acquisition timeout, in nanoseconds. + poolAcquisitionTimeout :: Word64, -- | Maximal connection lifetime, in nanoseconds. poolMaxLifetime :: Word64, -- | Maximal connection idle time, in nanoseconds. poolMaxIdletime :: Word64, - -- | Avail connections. + -- | Avail connections, in the order they get reused. poolConnectionQueue :: TQueue Entry, + -- | Which end of the queue returned connections go to. + poolReuseOrder :: ReuseOrder, -- | Remaining capacity. -- The pool size limits the sum of poolCapacity, the length -- of poolConnectionQueue and the number of in-flight -- connections. poolCapacity :: TVar Int, + -- | Pool requests waiting for a connection, the longest waiting first. + -- Freed connections and slots go to them before anyone else, + -- so that only the pool request given one wakes up. + poolWaiting :: TQueue Waiter, -- | Whether to return a connection to the pool. poolReuseVar :: TVar (TVar Bool), - -- | To stop the manager thread via garbage collection. + -- | To stop the manager and expirer threads via garbage collection. poolReaperRef :: IORef (), -- | Action for reporting the observations. poolObserver :: Observation -> IO (), @@ -69,6 +103,17 @@ poolInitSession :: Session.Session () } +-- | Contain the failures of the user-supplied observation handler. Synchronous +-- exceptions from it are discarded, asynchronous ones are honored, so that +-- neither affects the pool's state or the session result, and the pool's +-- threads stay killable. +containObserver :: (Observation -> IO ()) -> Observation -> IO () +containObserver observer observation = + catch @SomeException (observer observation) \exc -> + case fromException exc of + Just SomeAsyncException {} -> throwIO exc + Nothing -> return () + -- | Create a connection-pool. -- -- No connections actually get established by this function. It is delegated @@ -82,7 +127,9 @@ connectionQueue <- newTQueueIO capVar <- newTVarIO (Config.size config) reuseVar <- newTVarIO =<< newTVarIO True + waiting <- newTQueueIO reaperRef <- newIORef () + let observer = containObserver (Config.observationHandler config) managerTid <- forkIOWithUnmask $ \unmask -> unmask $ forever $ do threadDelay 1000000 @@ -93,23 +140,22 @@ (idleEntries, liveEntries) = partition (entryIsIdle maxIdletimeNanos now) unagedEntries traverse_ (writeTQueue connectionQueue) liveEntries return $ do - forM_ agedEntries $ \entry -> do - Connection.release (entryConnection entry) - atomically $ modifyTVar' capVar succ - (Config.observationHandler config) (ConnectionObservation (entryId entry) (TerminatedConnectionStatus AgingConnectionTerminationReason)) - forM_ idleEntries $ \entry -> do - Connection.release (entryConnection entry) - atomically $ modifyTVar' capVar succ - (Config.observationHandler config) (ConnectionObservation (entryId entry) (TerminatedConnectionStatus IdlenessConnectionTerminationReason)) + forM_ agedEntries $ reap observer reuseVar waiting capVar AgingConnectionTerminationReason + forM_ idleEntries $ reap observer reuseVar waiting capVar IdlenessConnectionTerminationReason + expirerTid <- forkIOWithUnmask $ \unmask -> unmask $ expire waiting + void . mkWeakIORef reaperRef $ do - -- When the pool goes out of scope, stop the manager. + -- When the pool goes out of scope, resolve the pool requests still + -- waiting, and stop the manager and the expirer. + atomically $ drainWaiting waiting killThread managerTid + killThread expirerTid - return $ Pool (Config.size config) adapter (Config.connectionSettingsProvider config) acqTimeoutMicros agingTimeoutNanos maxIdletimeNanos connectionQueue capVar reuseVar reaperRef (Config.observationHandler config) (Config.initSession config) + return $ Pool (Config.size config) adapter (Config.connectionSettingsProvider config) acqTimeoutNanos agingTimeoutNanos maxIdletimeNanos connectionQueue (Config.reuseOrder config) capVar waiting reuseVar reaperRef observer (Config.initSession config) where - acqTimeoutMicros = - div (fromIntegral (diffTimeToPicoseconds (Config.acquisitionTimeout config))) 1_000_000 + acqTimeoutNanos = + div (fromIntegral (diffTimeToPicoseconds (Config.acquisitionTimeout config))) 1_000 agingTimeoutNanos = div (fromIntegral (diffTimeToPicoseconds (Config.agingTimeout config))) 1_000 maxIdletimeNanos = @@ -130,10 +176,19 @@ newReuse <- newTVar True writeTVar poolReuseVar newReuse entries <- flushTQueue poolConnectionQueue - return $ forM_ entries $ \entry -> do - Connection.release (entryConnection entry) - atomically $ modifyTVar' poolCapacity succ - poolObserver (ConnectionObservation (entryId entry) (TerminatedConnectionStatus ReleaseConnectionTerminationReason)) + return $ forM_ entries $ \entry -> + mask \restore -> + catch @SomeException + ( do + restore $ Connection.release (entryConnection entry) + restore $ poolObserver (ConnectionObservation (entryId entry) (TerminatedConnectionStatus ReleaseConnectionTerminationReason)) + atomically $ freeSlot poolReuseVar poolWaiting poolCapacity + ) + \exc -> do + atomically $ freeSlot poolReuseVar poolWaiting poolCapacity + case fromException exc of + Just SomeAsyncException {} -> throwIO exc + Nothing -> return () -- | Use a connection from the pool to run a session and return the connection -- to the pool, when finished. @@ -144,113 +199,279 @@ -- time one is needed. The error still gets returned from this function. -- -- __Warning:__ Due to the mechanism mentioned above you should avoid intercepting this error type from within sessions. -use :: Pool -> Session.Session a -> IO (Either UsageError a) +-- +-- When no connection is available, the pool requests wait for one in the +-- order they came, and a later pool request doesn't take a connection while +-- others wait. +use :: forall a. Pool -> Session.Session a -> IO (Either UsageError a) use Pool {..} sess = do - timeout <- do - delay <- registerDelay poolAcquisitionTimeout - return $ readTVar delay - join . atomically $ do - reuseVar <- readTVar poolReuseVar - asum - [ readTQueue poolConnectionQueue <&> onConn reuseVar, - do - capVal <- readTVar poolCapacity - if capVal > 0 - then do - writeTVar poolCapacity $! pred capVal - return $ onNewConn reuseVar - else retry, - do - timedOut <- timeout - if timedOut - then return . return . Left $ AcquisitionTimeoutUsageError - else retry - ] + now <- getMonotonicTimeNSec + mask \unmask -> do + holding <- atomically $ do + reuseVar <- readTVar poolReuseVar + waited <- someoneWaiting poolWaiting + -- Behind the pool requests already waiting, which get what's freed first. + if waited + then enqueue now <&> WaitingHolding + else + asum + [ readTQueue poolConnectionQueue <&> EntryHolding reuseVar, + do + capVal <- readTVar poolCapacity + if capVal > 0 + then do + writeTVar poolCapacity $! pred capVal + return $ SlotHolding reuseVar + else retry, + enqueue now <&> WaitingHolding + ] + case holding of + EntryHolding reuseVar entry -> onEntry unmask reuseVar entry + SlotHolding reuseVar -> onSlot unmask reuseVar + WaitingHolding waiter -> do + given <- atomically (takeTMVar (waiterGiven waiter)) `onException` abandon waiter + case given of + TimedOut -> return $ Left AcquisitionTimeoutUsageError + Given reuseVar Nothing -> onSlot unmask reuseVar + Given reuseVar (Just entry) -> onEntry unmask reuseVar entry where - onNewConn reuseVar = do - settings <- poolFetchConnectionSettings + enqueue :: Word64 -> STM Waiter + enqueue now = do + waiter <- Waiter (now + poolAcquisitionTimeout) <$> newEmptyTMVar <*> newTVar False + writeTQueue poolWaiting waiter + return waiter + + abandon :: Waiter -> IO () + abandon waiter = do + given <- atomically $ do + writeTVar (waiterGone waiter) True + tryTakeTMVar (waiterGiven waiter) + forM_ given \case + Given _ Nothing -> atomically returnSlot + Given reuseVar (Just entry) -> returnConn reuseVar entry + TimedOut -> return () + + returnSlot :: STM () + returnSlot = freeSlot poolReuseVar poolWaiting poolCapacity + + -- \| Holding one entry dequeued from the idle connections. If it is still + -- good, run the session on it; if it expired, replace it with a fresh + -- connection on the same slot. All the way, the holding is given back on + -- interruption. + onEntry :: (forall x. IO x -> IO x) -> TVar Bool -> Entry -> IO (Either UsageError a) + onEntry unmask reuseVar entry = do now <- getMonotonicTimeNSec + if entryIsAged poolMaxLifetime now entry + then renewEntry unmask reuseVar entry AgingConnectionTerminationReason + else + if entryIsIdle poolMaxIdletime now entry + then renewEntry unmask reuseVar entry IdlenessConnectionTerminationReason + else onLiveConn unmask reuseVar entry {entryUseTimeNSec = now} + + -- \| Holding an expired entry. Release its connection and establish a + -- fresh one on the same slot. On interruption the slot gets given back: + -- the released connection cannot return to the pool. + renewEntry :: (forall x. IO x -> IO x) -> TVar Bool -> Entry -> ConnectionTerminationReason -> IO (Either UsageError a) + renewEntry unmask reuseVar entry reason = do + outcome <- + onException + ( do + Connection.release (entryConnection entry) + unmask $ poolObserver (ConnectionObservation (entryId entry) (TerminatedConnectionStatus reason)) + establish unmask + ) + (atomically returnSlot) + case outcome of + Left err -> return $ Left err + Right newEntry -> onLiveConn unmask reuseVar newEntry + + -- \| Holding one capacity unit. Establish a connection on it and run the + -- session. On interruption the slot gets given back. + onSlot :: (forall x. IO x -> IO x) -> TVar Bool -> IO (Either UsageError a) + onSlot unmask reuseVar = do + outcome <- onException (establish unmask) (atomically returnSlot) + case outcome of + Left err -> return $ Left err + Right entry -> onLiveConn unmask reuseVar entry + + -- Run with one capacity unit held: establish the connection and run the + -- init session on it. On any failure the slot gets given back and the + -- error gets returned. On interruption the slot gets given back and the + -- exception gets rethrown. + establish :: (forall x. IO x -> IO x) -> IO (Either UsageError Entry) + establish unmask = do + settings <- unmask poolFetchConnectionSettings + now <- getMonotonicTimeNSec id <- Uuid.nextRandom - poolObserver (ConnectionObservation id ConnectingConnectionStatus) - Connection.acquire poolAdapter settings >>= \case + unmask $ poolObserver (ConnectionObservation id ConnectingConnectionStatus) + unmask (Connection.acquire poolAdapter settings) >>= \case Left connErr -> do let connErrText = case connErr of Errors.NetworkingConnectionError details -> Just details Errors.AuthenticationConnectionError details -> Just details Errors.CompatibilityConnectionError details -> Just details Errors.OtherConnectionError details -> if details == "" then Nothing else Just details - poolObserver (ConnectionObservation id (TerminatedConnectionStatus (NetworkErrorConnectionTerminationReason connErrText))) - atomically $ modifyTVar' poolCapacity succ + unmask $ poolObserver (ConnectionObservation id (TerminatedConnectionStatus (NetworkErrorConnectionTerminationReason connErrText))) + atomically returnSlot return $ Left $ ConnectionUsageError connErr - Right connection -> do - Connection.use connection poolInitSession >>= \case + Right connection -> + unmask (Connection.use connection poolInitSession) >>= \case Left err -> do Connection.release connection - atomically $ modifyTVar' poolCapacity succ - ErrorsDestruction.reset - ( \details -> do - poolObserver (ConnectionObservation id (TerminatedConnectionStatus (NetworkErrorConnectionTerminationReason (Just details)))) - ) - (poolObserver (ConnectionObservation id (TerminatedConnectionStatus (InitializationErrorTerminationReason err)))) - err + unmask + $ ErrorsDestruction.reset + (\details -> poolObserver (ConnectionObservation id (TerminatedConnectionStatus (NetworkErrorConnectionTerminationReason (Just details))))) + (poolObserver (ConnectionObservation id (TerminatedConnectionStatus (InitializationErrorTerminationReason err)))) + err + atomically returnSlot return $ Left $ SessionUsageError err Right () -> do - poolObserver (ConnectionObservation id (ReadyForUseConnectionStatus EstablishedConnectionReadyForUseReason)) - onLiveConn reuseVar (Entry connection now now id) - - onConn reuseVar entry = do - now <- getMonotonicTimeNSec - if entryIsAged poolMaxLifetime now entry - then do - Connection.release (entryConnection entry) - poolObserver (ConnectionObservation (entryId entry) (TerminatedConnectionStatus AgingConnectionTerminationReason)) - onNewConn reuseVar - else - if entryIsIdle poolMaxIdletime now entry - then do - Connection.release (entryConnection entry) - poolObserver (ConnectionObservation (entryId entry) (TerminatedConnectionStatus IdlenessConnectionTerminationReason)) - onNewConn reuseVar - else do - onLiveConn reuseVar entry {entryUseTimeNSec = now} + unmask $ poolObserver (ConnectionObservation id (ReadyForUseConnectionStatus EstablishedConnectionReadyForUseReason)) + return $ Right (Entry connection now now id) - onLiveConn reuseVar entry = do - poolObserver (ConnectionObservation (entryId entry) InUseConnectionStatus) - sessRes <- try @SomeException (Connection.use (entryConnection entry) sess) + -- \| Holding one entry ready for use. Run the session on it, return the + -- entry to the pool, and return the session result. Interrupted, the + -- session exception gets rethrown after the entry is back in the pool. + onLiveConn :: (forall x. IO x -> IO x) -> TVar Bool -> Entry -> IO (Either UsageError a) + onLiveConn unmask reuseVar entry = do + sessRes <- + onException + ( unmask $ do + poolObserver (ConnectionObservation (entryId entry) InUseConnectionStatus) + try @SomeException (Connection.use (entryConnection entry) sess) + ) + (returnConn reuseVar entry) case sessRes of Left exc -> do - returnConn + returnConn reuseVar entry throwIO exc Right (Left err) -> if ErrorsDestruction.requiresConnectionDiscard err then do Connection.release (entryConnection entry) - atomically $ modifyTVar' poolCapacity succ - poolObserver - ( ConnectionObservation - (entryId entry) - (TerminatedConnectionStatus (NetworkErrorConnectionTerminationReason (ErrorsDestruction.discardDetails err))) - ) + atomically returnSlot + unmask + $ poolObserver + ( ConnectionObservation + (entryId entry) + (TerminatedConnectionStatus (NetworkErrorConnectionTerminationReason (ErrorsDestruction.discardDetails err))) + ) return $ Left $ SessionUsageError err else do - returnConn - poolObserver (ConnectionObservation (entryId entry) (ReadyForUseConnectionStatus (SessionFailedConnectionReadyForUseReason err))) + returnConn reuseVar entry + unmask $ poolObserver (ConnectionObservation (entryId entry) (ReadyForUseConnectionStatus (SessionFailedConnectionReadyForUseReason err))) return $ Left $ SessionUsageError err Right (Right res) -> do - returnConn - poolObserver (ConnectionObservation (entryId entry) (ReadyForUseConnectionStatus SessionSucceededConnectionReadyForUseReason)) + returnConn reuseVar entry + unmask $ poolObserver (ConnectionObservation (entryId entry) (ReadyForUseConnectionStatus SessionSucceededConnectionReadyForUseReason)) return $ Right res - where - returnConn = - join . atomically $ do - reuse <- readTVar reuseVar - if reuse - then writeTQueue poolConnectionQueue entry $> return () - else return $ do - Connection.release (entryConnection entry) - atomically $ modifyTVar' poolCapacity succ - poolObserver (ConnectionObservation (entryId entry) (TerminatedConnectionStatus ReleaseConnectionTerminationReason)) + + -- Back to the pool request waiting longest or to the idle connections, + -- unless the pool was released meanwhile. + returnConn reuseVar entry = + join . atomically $ do + reuse <- readTVar reuseVar + if reuse + then do + nextWaiter poolWaiting >>= \case + Just waiter -> do + currentReuseVar <- readTVar poolReuseVar + putTMVar (waiterGiven waiter) (Given currentReuseVar (Just entry)) + Nothing -> + case poolReuseOrder of + LIFO -> unGetTQueue poolConnectionQueue entry + FIFO -> writeTQueue poolConnectionQueue entry + return $ return () + else return $ do + Connection.release (entryConnection entry) + atomically returnSlot + poolObserver (ConnectionObservation (entryId entry) (TerminatedConnectionStatus ReleaseConnectionTerminationReason)) + +-- | Whether a pool request is waiting. The ones that stopped waiting get +-- dropped. +someoneWaiting :: TQueue Waiter -> STM Bool +someoneWaiting waiting = + tryPeekTQueue waiting >>= \case + Nothing -> return False + Just waiter -> + readTVar (waiterGone waiter) >>= \case + True -> readTQueue waiting *> someoneWaiting waiting + False -> return True + +-- | Take the pool request waiting longest, passing over the ones that stopped +-- waiting. +nextWaiter :: TQueue Waiter -> STM (Maybe Waiter) +nextWaiter waiting = + tryReadTQueue waiting >>= \case + Nothing -> return Nothing + Just waiter -> + readTVar (waiterGone waiter) >>= \case + True -> nextWaiter waiting + False -> return (Just waiter) + +-- | Give the slot of a closed connection to the pool request waiting longest, +-- or back to the pool if no pool request waits. +freeSlot :: TVar (TVar Bool) -> TQueue Waiter -> TVar Int -> STM () +freeSlot reuseVarVar waiting capVar = + nextWaiter waiting >>= \case + Just waiter -> do + reuseVar <- readTVar reuseVarVar + putTMVar (waiterGiven waiter) (Given reuseVar Nothing) + Nothing -> modifyTVar' capVar succ + +-- | Time out the pool requests waiting too long. All wait as long, so the pool +-- request waiting longest times out first: one thread sleeps until its +-- deadline, rather than every waiting pool request registering a timer, which +-- GHC can't cancel. +expire :: TQueue Waiter -> IO () +expire waiting = forever $ do + deadline <- atomically $ do + waited <- someoneWaiting waiting + if waited then waiterDeadline <$> peekTQueue waiting else retry + now <- getMonotonicTimeNSec + -- In microseconds, at most an hour at a time. + when (deadline > now) $ threadDelay (fromIntegral (min 3_600_000_000 (div (deadline - now) 1_000 + 1))) + atomically . timeOut =<< getMonotonicTimeNSec + where + timeOut now = + tryPeekTQueue waiting >>= \case + Just waiter | waiterDeadline waiter <= now -> do + _ <- readTQueue waiting + gone <- readTVar (waiterGone waiter) + unless gone $ putTMVar (waiterGiven waiter) TimedOut + timeOut now + _ -> return () + +-- | Resolve the pool requests still waiting: with the pool gone, nothing will +-- ever be given to them. +drainWaiting :: TQueue Waiter -> STM () +drainWaiting waiting = + tryReadTQueue waiting >>= \case + Nothing -> return () + Just waiter -> do + gone <- readTVar (waiterGone waiter) + unless gone $ void $ tryPutTMVar (waiterGiven waiter) TimedOut + drainWaiting waiting + +-- | Terminate one expired connection on the reaper's tick: release it, report +-- the termination, and give its slot back. The slot gets given back even when +-- the release or the observation fails, so no capacity is lost, and a failing +-- entry neither skips the other entries of the tick nor kills the reaper. +reap :: (Observation -> IO ()) -> TVar (TVar Bool) -> TQueue Waiter -> TVar Int -> ConnectionTerminationReason -> Entry -> IO () +reap observer reuseVar waiting capVar reason entry = + mask \restore -> + catch @SomeException + ( do + restore $ Connection.release (entryConnection entry) + restore $ observer (ConnectionObservation (entryId entry) (TerminatedConnectionStatus reason)) + atomically $ freeSlot reuseVar waiting capVar + ) + \exc -> do + atomically $ freeSlot reuseVar waiting capVar + case fromException exc of + Just SomeAsyncException {} -> throwIO exc + Nothing -> return () -- | Union over all errors that 'use' can result in. data UsageError diff --git a/src/library/Hasql/Pool/Config.hs b/src/library/Hasql/Pool/Config.hs --- a/src/library/Hasql/Pool/Config.hs +++ b/src/library/Hasql/Pool/Config.hs @@ -11,10 +11,13 @@ Setting.dynamicConnectionSettings, Setting.observationHandler, Setting.initSession, + Setting.reuseOrder, + ReuseOrder.ReuseOrder (..), ) where import Hasql.Pool.Config.Config qualified as Config +import Hasql.Pool.Config.ReuseOrder qualified as ReuseOrder import Hasql.Pool.Config.Setting qualified as Setting import Hasql.Pool.Prelude diff --git a/src/library/Hasql/Pool/Config/Config.hs b/src/library/Hasql/Pool/Config/Config.hs --- a/src/library/Hasql/Pool/Config/Config.hs +++ b/src/library/Hasql/Pool/Config/Config.hs @@ -2,6 +2,7 @@ import Hasql.Connection.Settings qualified as Connection.Settings import Hasql.Pool.Config.Defaults qualified as Defaults +import Hasql.Pool.Config.ReuseOrder (ReuseOrder) import Hasql.Pool.Observation (Observation) import Hasql.Pool.Prelude import Hasql.Session qualified as Session @@ -14,7 +15,8 @@ idlenessTimeout :: DiffTime, connectionSettingsProvider :: IO Connection.Settings.Settings, observationHandler :: Observation -> IO (), - initSession :: Session.Session () + initSession :: Session.Session (), + reuseOrder :: ReuseOrder } -- | Reasonable defaults, which can be built upon. @@ -27,5 +29,6 @@ idlenessTimeout = Defaults.idlenessTimeout, connectionSettingsProvider = Defaults.dynamicConnectionSettings, observationHandler = Defaults.observationHandler, - initSession = Defaults.initSession + initSession = Defaults.initSession, + reuseOrder = Defaults.reuseOrder } diff --git a/src/library/Hasql/Pool/Config/Defaults.hs b/src/library/Hasql/Pool/Config/Defaults.hs --- a/src/library/Hasql/Pool/Config/Defaults.hs +++ b/src/library/Hasql/Pool/Config/Defaults.hs @@ -1,6 +1,7 @@ module Hasql.Pool.Config.Defaults where import Hasql.Connection.Settings qualified as Connection.Settings +import Hasql.Pool.Config.ReuseOrder (ReuseOrder (..)) import Hasql.Pool.Observation (Observation) import Hasql.Pool.Prelude import Hasql.Session qualified as Session @@ -45,3 +46,8 @@ -- > pure () initSession :: Session.Session () initSession = pure () + +-- | +-- The idle connection returned longest ago first. +reuseOrder :: ReuseOrder +reuseOrder = FIFO diff --git a/src/library/Hasql/Pool/Config/ReuseOrder.hs b/src/library/Hasql/Pool/Config/ReuseOrder.hs new file mode 100644 --- /dev/null +++ b/src/library/Hasql/Pool/Config/ReuseOrder.hs @@ -0,0 +1,17 @@ +module Hasql.Pool.Config.ReuseOrder where + +import Hasql.Pool.Prelude + +-- | Which of the idle connections a pool request gets. +data ReuseOrder + = -- | Last in, first out: the one returned most recently. + -- + -- The pool keeps reusing the few connections the load needs, so the rest + -- time out as idle, and the ones in use keep their caches warm. + LIFO + | -- | First in, first out: the one returned longest ago. + -- + -- Pool requests cycle through all the connections, so steady load keeps + -- every connection open. + FIFO + deriving (Show, Eq) diff --git a/src/library/Hasql/Pool/Config/Setting.hs b/src/library/Hasql/Pool/Config/Setting.hs --- a/src/library/Hasql/Pool/Config/Setting.hs +++ b/src/library/Hasql/Pool/Config/Setting.hs @@ -3,6 +3,7 @@ import Hasql.Connection.Settings qualified as Connection.Settings import Hasql.Pool.Config.Config (Config) import Hasql.Pool.Config.Config qualified as Config +import Hasql.Pool.Config.ReuseOrder (ReuseOrder (..)) import Hasql.Pool.Observation (Observation) import Hasql.Pool.Prelude import Hasql.Session qualified as Session @@ -74,6 +75,8 @@ -- -- Typically it's used for monitoring the state of the pool via metrics and logging. -- +-- An observation handler must not throw. Synchronous exceptions from it are discarded; asynchronous exceptions are honored. Its failure never affects the session result or the pool's state. +-- -- If the provided action is not lightweight, it's recommended to use intermediate bufferring via channels like TBQueue to avoid occupying the pool management thread for too long. -- E.g., if the action is @'atomically' . 'writeTBQueue' yourQueue@, then reading from it and processing can be done on a separate thread. -- @@ -95,3 +98,14 @@ initSession :: Session.Session () -> Setting initSession x = Setting (\config -> config {Config.initSession = x}) + +-- | Which of the idle connections a pool request gets. +-- +-- With 'LIFO', the pool shrinks to the connections the load needs: the rest +-- time out as idle (see 'idlenessTimeout'). With 'FIFO', steady load keeps +-- every connection open. +-- +-- 'FIFO' by default. +reuseOrder :: ReuseOrder -> Setting +reuseOrder x = + Setting (\config -> config {Config.reuseOrder = x})