diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,12 @@ +# 0.9 + +- Maximal lifetime added for connections. Allows to refresh the connections in time cleaning up the resources. + +Breaking: + +- The acquisition timeout is now non-optional. +- Moved to `DiffTime` for timeouts. + # 0.8.0.7 Fix excessive connections during releases due to race conditions. diff --git a/hasql-pool.cabal b/hasql-pool.cabal --- a/hasql-pool.cabal +++ b/hasql-pool.cabal @@ -1,50 +1,88 @@ -cabal-version: 3.0 - -name: hasql-pool -version: 0.8.0.7 - -category: Hasql, Database, PostgreSQL -synopsis: Pool of connections for Hasql -homepage: https://github.com/nikita-volkov/hasql-pool -bug-reports: https://github.com/nikita-volkov/hasql-pool/issues -author: Nikita Volkov -maintainer: Nikita Volkov -copyright: (c) 2015, Nikita Volkov -license: MIT -license-file: LICENSE +cabal-version: 3.0 +name: hasql-pool +version: 0.9 +category: Hasql, Database, PostgreSQL +synopsis: Pool of connections for Hasql +homepage: https://github.com/nikita-volkov/hasql-pool +bug-reports: https://github.com/nikita-volkov/hasql-pool/issues +author: Nikita Volkov +maintainer: Nikita Volkov +copyright: (c) 2015, Nikita Volkov +license: MIT +license-file: LICENSE extra-source-files: CHANGELOG.md source-repository head - type: git + type: git location: git://github.com/nikita-volkov/hasql-pool.git common base-settings - default-extensions: BangPatterns, BlockArguments, ConstraintKinds, DataKinds, DefaultSignatures, DeriveDataTypeable, DeriveFoldable, DeriveFunctor, DeriveGeneric, DerivingVia, DeriveTraversable, EmptyDataDecls, FlexibleContexts, FlexibleInstances, FunctionalDependencies, GADTs, GeneralizedNewtypeDeriving, InstanceSigs, LambdaCase, LiberalTypeSynonyms, MagicHash, MultiParamTypeClasses, MultiWayIf, NoImplicitPrelude, NoMonomorphismRestriction, OverloadedStrings, PatternGuards, ParallelListComp, QuasiQuotes, RankNTypes, RecordWildCards, ScopedTypeVariables, StandaloneDeriving, StrictData, TemplateHaskell, TupleSections, TypeApplications, TypeFamilies, TypeOperators, UnboxedTuples - default-language: Haskell2010 + default-extensions: + NoImplicitPrelude + NoMonomorphismRestriction + BangPatterns + BlockArguments + ConstraintKinds + DataKinds + DefaultSignatures + DeriveDataTypeable + DeriveFoldable + DeriveFunctor + DeriveGeneric + DeriveTraversable + DerivingVia + EmptyDataDecls + FlexibleContexts + FlexibleInstances + FunctionalDependencies + GADTs + GeneralizedNewtypeDeriving + InstanceSigs + LambdaCase + LiberalTypeSynonyms + MagicHash + MultiParamTypeClasses + MultiWayIf + NumericUnderscores + OverloadedStrings + ParallelListComp + PatternGuards + QuasiQuotes + RankNTypes + RecordWildCards + ScopedTypeVariables + StandaloneDeriving + StrictData + TemplateHaskell + TupleSections + TypeApplications + TypeFamilies + TypeOperators + UnboxedTuples + default-language: Haskell2010 + library - import: base-settings - hs-source-dirs: library - exposed-modules: - Hasql.Pool - other-modules: - Hasql.Pool.Prelude + import: base-settings + hs-source-dirs: library + exposed-modules: Hasql.Pool + other-modules: Hasql.Pool.Prelude build-depends: - base >=4.11 && <5, - hasql >=1.6.0.1 && <1.7, - stm >=2.5 && <3, - transformers >=0.5 && <0.7, + , base >=4.11 && <5 + , hasql >=1.6.0.1 && <1.7 + , stm >=2.5 && <3 + , time >=1.9 && <2 test-suite test - import: base-settings - type: exitcode-stdio-1.0 + import: base-settings + type: exitcode-stdio-1.0 hs-source-dirs: test - main-is: Main.hs - ghc-options: -threaded + main-is: Main.hs + ghc-options: -threaded build-depends: - async >=2.2 && <3, - hasql, - hasql-pool, - hspec >=2.6 && <3, - rerebase >=1.15 && <2, - stm >=2.5 && <3, + , async >=2.2 && <3 + , hasql + , hasql-pool + , hspec >=2.6 && <3 + , random >=1.2 && <2 + , rerebase >=1.15 && <2 diff --git a/library/Hasql/Pool.hs b/library/Hasql/Pool.hs --- a/library/Hasql/Pool.hs +++ b/library/Hasql/Pool.hs @@ -3,8 +3,8 @@ Pool, acquire, acquireDynamically, - release, use, + release, -- * Errors UsageError (..), @@ -14,62 +14,103 @@ import Hasql.Connection (Connection) import qualified Hasql.Connection as Connection import Hasql.Pool.Prelude -import Hasql.Session (Session) import qualified Hasql.Session as Session +-- | A connection tagged with metadata. +data Conn = Conn + { connConnection :: Connection, + connCreationTimeNSec :: Word64 + } + +isAlive :: Word64 -> Word64 -> Conn -> Bool +isAlive maxLifetime now conn = + now <= connCreationTimeNSec conn + maxLifetime + -- | Pool of connections to DB. data Pool = Pool - { -- | Connection settings. + { -- | Pool size. + poolSize :: Int, + -- | Connection settings. poolFetchConnectionSettings :: IO Connection.Settings, -- | Acquisition timeout, in microseconds. - poolAcquisitionTimeout :: Maybe Int, + poolAcquisitionTimeout :: Int, + -- | Maximal connection lifetime, in nanoseconds. + poolMaxLifetime :: Word64, -- | Avail connections. - poolConnectionQueue :: TQueue Connection, + poolConnectionQueue :: TQueue Conn, -- | Remaining capacity. -- The pool size limits the sum of poolCapacity, the length -- of poolConnectionQueue and the number of in-flight -- connections. poolCapacity :: TVar Int, -- | Whether to return a connection to the pool. - poolReuse :: TVar (TVar Bool) + poolReuseVar :: TVar (TVar Bool), + -- | To stop the manager thread via garbage collection. + poolReaperRef :: IORef () } --- | Create a connection-pool. +-- | Create a connection-pool, with default settings. -- -- No connections actually get established by this function. It is delegated -- to 'use'. acquire :: -- | Pool size. Int -> - -- | Connection acquisition timeout in microseconds. - Maybe Int -> + -- | Connection acquisition timeout. + DiffTime -> + -- | Maximal connection lifetime. + DiffTime -> -- | Connection settings. Connection.Settings -> IO Pool -acquire poolSize timeout connectionSettings = - acquireDynamically poolSize timeout (pure connectionSettings) +acquire poolSize acqTimeout maxLifetime connectionSettings = + acquireDynamically poolSize acqTimeout maxLifetime (pure connectionSettings) -- | Create a connection-pool. -- --- In difference to 'acquire' new settings get fetched each time a connection --- is created. This may be useful for some security models. +-- In difference to 'acquire' new connection settings get fetched each +-- time a connection is created. This may be useful for some security models. -- -- No connections actually get established by this function. It is delegated -- to 'use'. acquireDynamically :: -- | Pool size. Int -> - -- | Connection acquisition timeout in microseconds. - Maybe Int -> + -- | Connection acquisition timeout. + DiffTime -> + -- | Maximal connection lifetime. + DiffTime -> -- | Action fetching connection settings. IO Connection.Settings -> IO Pool -acquireDynamically poolSize timeout fetchConnectionSettings = do - Pool fetchConnectionSettings timeout - <$> newTQueueIO - <*> newTVarIO poolSize - <*> (newTVarIO =<< newTVarIO True) +acquireDynamically poolSize acqTimeout maxLifetime fetchConnectionSettings = do + connectionQueue <- newTQueueIO + capVar <- newTVarIO poolSize + reuseVar <- newTVarIO =<< newTVarIO True + reaperRef <- newIORef () + managerTid <- forkIOWithUnmask $ \unmask -> unmask $ forever $ do + threadDelay 1000000 + now <- getMonotonicTimeNSec + join . atomically $ do + conns <- flushTQueue connectionQueue + let (keep, close) = partition (isAlive maxLifetimeNanos now) conns + traverse_ (writeTQueue connectionQueue) keep + return $ forM_ close $ \conn -> do + Connection.release (connConnection conn) + atomically $ modifyTVar' capVar succ + + void . mkWeakIORef reaperRef $ do + -- When the pool goes out of scope, stop the manager. + killThread managerTid + + return $ Pool poolSize fetchConnectionSettings acqTimeoutMicros maxLifetimeNanos connectionQueue capVar reuseVar reaperRef + where + acqTimeoutMicros = + div (fromIntegral (diffTimeToPicoseconds acqTimeout)) 1_000_000 + maxLifetimeNanos = + div (fromIntegral (diffTimeToPicoseconds maxLifetime)) 1_000 + -- | Release all the idle connections in the pool, and mark the in-use connections -- to be released after use. Any connections acquired after the call will be -- freshly established. @@ -80,13 +121,13 @@ release :: Pool -> IO () release Pool {..} = join . atomically $ do - prevReuse <- readTVar poolReuse + prevReuse <- readTVar poolReuseVar writeTVar prevReuse False newReuse <- newTVar True - writeTVar poolReuse newReuse + writeTVar poolReuseVar newReuse conns <- flushTQueue poolConnectionQueue return $ forM_ conns $ \conn -> do - Connection.release conn + Connection.release (connConnection conn) atomically $ modifyTVar' poolCapacity succ -- | Use a connection from the pool to run a session and return the connection @@ -98,14 +139,11 @@ -- time one is needed. The error still gets returned from this function. use :: Pool -> Session.Session a -> IO (Either UsageError a) use Pool {..} sess = do - timeout <- case poolAcquisitionTimeout of - Just delta -> do - delay <- registerDelay delta - return $ readTVar delay - Nothing -> - return $ return False + timeout <- do + delay <- registerDelay poolAcquisitionTimeout + return $ readTVar delay join . atomically $ do - reuseVar <- readTVar poolReuse + reuseVar <- readTVar poolReuseVar asum [ readTQueue poolConnectionQueue <&> onConn reuseVar, do @@ -124,16 +162,26 @@ where onNewConn reuseVar = do settings <- poolFetchConnectionSettings + now <- getMonotonicTimeNSec connRes <- Connection.acquire settings case connRes of Left connErr -> do atomically $ modifyTVar' poolCapacity succ return $ Left $ ConnectionUsageError connErr - Right conn -> onConn reuseVar conn + Right conn -> onLiveConn reuseVar (Conn conn now) + onConn reuseVar conn = do + now <- getMonotonicTimeNSec + if isAlive poolMaxLifetime now conn + then onLiveConn reuseVar conn + else do + Connection.release (connConnection conn) + onNewConn reuseVar + + onLiveConn reuseVar conn = do sessRes <- - catch (Session.run sess conn) $ \(err :: SomeException) -> do - Connection.release conn + catch (Session.run sess (connConnection conn)) $ \(err :: SomeException) -> do + Connection.release (connConnection conn) atomically $ modifyTVar' poolCapacity succ throw err @@ -155,7 +203,7 @@ if reuse then writeTQueue poolConnectionQueue conn $> return () else return $ do - Connection.release conn + Connection.release (connConnection conn) atomically $ modifyTVar' poolCapacity succ -- | Union over all errors that 'use' can result in. diff --git a/library/Hasql/Pool/Prelude.hs b/library/Hasql/Pool/Prelude.hs --- a/library/Hasql/Pool/Prelude.hs +++ b/library/Hasql/Pool/Prelude.hs @@ -14,13 +14,6 @@ import Control.Monad.Fix as Exports hiding (fix) import Control.Monad.IO.Class as Exports import Control.Monad.ST as Exports -import Control.Monad.Trans.Class as Exports -import Control.Monad.Trans.Cont as Exports hiding (callCC, shift) -import Control.Monad.Trans.Except as Exports (Except, ExceptT (ExceptT), except, mapExcept, mapExceptT, runExcept, runExceptT, withExcept, withExceptT) -import Control.Monad.Trans.Maybe as Exports -import Control.Monad.Trans.Reader as Exports (Reader, ReaderT (ReaderT), mapReader, mapReaderT, runReader, runReaderT, withReader, withReaderT) -import Control.Monad.Trans.State.Strict as Exports (State, StateT (StateT), evalState, evalStateT, execState, execStateT, mapState, mapStateT, runState, runStateT, withState, withStateT) -import Control.Monad.Trans.Writer.Strict as Exports (Writer, WriterT (..), execWriter, execWriterT, mapWriter, mapWriterT, runWriter) import Data.Bifunctor as Exports import Data.Bits as Exports import Data.Bool as Exports @@ -47,6 +40,7 @@ import Data.Ratio as Exports import Data.STRef as Exports import Data.String as Exports +import Data.Time as Exports import Data.Traversable as Exports import Data.Tuple as Exports import Data.Unique as Exports @@ -58,6 +52,7 @@ import Foreign.Ptr as Exports import Foreign.StablePtr as Exports import Foreign.Storable as Exports +import GHC.Clock as Exports (getMonotonicTimeNSec) import GHC.Conc as Exports hiding (orElse, threadWaitRead, threadWaitReadSTM, threadWaitWrite, threadWaitWriteSTM, withMVar) import GHC.Exts as Exports (IsList (..), groupWith, inline, lazy, sortWith) import GHC.Generics as Exports (Generic) @@ -71,8 +66,6 @@ import System.Mem as Exports import System.Mem.StableName as Exports import System.Timeout as Exports -import Text.ParserCombinators.ReadP as Exports (ReadP, ReadS, readP_to_S, readS_to_P) -import Text.ParserCombinators.ReadPrec as Exports (ReadPrec, readP_to_Prec, readPrec_to_P, readPrec_to_S, readS_to_Prec) import Text.Printf as Exports (hPrintf, printf) import Text.Read as Exports (Read (..), readEither, readMaybe) import Unsafe.Coerce as Exports diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -2,6 +2,7 @@ import Control.Concurrent.Async (race) import qualified Data.ByteString.Char8 as B8 +import qualified Data.Text as Text import qualified Hasql.Connection as Connection import qualified Hasql.Decoders as Decoders import qualified Hasql.Encoders as Encoders @@ -9,90 +10,115 @@ import qualified Hasql.Session as Session import qualified Hasql.Statement as Statement import qualified System.Environment +import qualified System.Random.Stateful as Random import Test.Hspec import Prelude +main :: IO () main = do connectionSettings <- getConnectionSettings + let withPool poolSize acqTimeout maxLifetime connectionSettings = + bracket (acquire poolSize acqTimeout maxLifetime connectionSettings) release + withDefaultPool = + withPool 3 10 1_800 connectionSettings + hspec . describe "" $ do - it "Releases a spot in the pool when there is a query error" $ do - pool <- acquire 1 Nothing connectionSettings + it "Releases a spot in the pool when there is a query error" $ withDefaultPool $ \pool -> do use pool badQuerySession `shouldNotReturn` (Right ()) use pool selectOneSession `shouldReturn` (Right 1) - it "Simulation of connection error works" $ do - pool <- acquire 3 Nothing connectionSettings + it "Simulation of connection error works" $ withDefaultPool $ \pool -> do res <- use pool $ closeConnSession >> selectOneSession shouldSatisfy res $ \case Left (SessionUsageError (Session.QueryError _ _ (Session.ClientError _))) -> True _ -> False - it "Connection errors cause eviction of connection" $ do - pool <- acquire 3 Nothing connectionSettings + it "Connection errors cause eviction of connection" $ withDefaultPool $ \pool -> do res <- use pool $ closeConnSession >> selectOneSession res <- use pool $ closeConnSession >> selectOneSession res <- use pool $ closeConnSession >> selectOneSession res <- use pool $ selectOneSession shouldSatisfy res $ isRight - it "Connection gets returned to the pool after normal use" $ do - pool <- acquire 3 Nothing connectionSettings + it "Connection gets returned to the pool after normal use" $ withDefaultPool $ \pool -> do res <- use pool $ selectOneSession res <- use pool $ selectOneSession res <- use pool $ selectOneSession res <- use pool $ selectOneSession res <- use pool $ selectOneSession shouldSatisfy res $ isRight - it "Connection gets returned to the pool after non-connection error" $ do - pool <- acquire 3 Nothing connectionSettings + it "Connection gets returned to the pool after non-connection error" $ withDefaultPool $ \pool -> do res <- use pool $ badQuerySession res <- use pool $ badQuerySession res <- use pool $ badQuerySession res <- use pool $ badQuerySession res <- use pool $ selectOneSession shouldSatisfy res $ isRight - it "The pool remains usable after release" $ do - pool <- acquire 1 Nothing connectionSettings + it "The pool remains usable after release" $ withDefaultPool $ \pool -> do res <- use pool $ selectOneSession release pool res <- use pool $ selectOneSession shouldSatisfy res $ isRight - it "Getting and setting session variables works" $ do - pool <- acquire 1 Nothing connectionSettings + it "Getting and setting session variables works" $ withDefaultPool $ \pool -> do res <- use pool $ getSettingSession "testing.foo" res `shouldBe` Right Nothing res <- use pool $ do setSettingSession "testing.foo" "hello world" getSettingSession "testing.foo" res `shouldBe` Right (Just "hello world") - it "Session variables stay set when a connection gets reused" $ do - pool <- acquire 1 Nothing connectionSettings + it "Session variables stay set when a connection gets reused" $ withPool 1 10 1_800 connectionSettings $ \pool -> do res <- use pool $ setSettingSession "testing.foo" "hello world" res `shouldBe` Right () res2 <- use pool $ getSettingSession "testing.foo" res2 `shouldBe` Right (Just "hello world") - it "Releasing the pool resets session variables" $ do - pool <- acquire 1 Nothing connectionSettings + it "Releasing the pool resets session variables" $ withPool 1 10 1_800 connectionSettings $ \pool -> do res <- use pool $ setSettingSession "testing.foo" "hello world" res `shouldBe` Right () release pool res <- use pool $ getSettingSession "testing.foo" res `shouldBe` Right Nothing - it "Times out connection acquisition" $ do - pool <- acquire 1 (Just 1000) connectionSettings -- 1ms timeout - sleeping <- newEmptyMVar - t0 <- getCurrentTime - res <- - race - ( use pool $ - liftIO $ do - putMVar sleeping () - threadDelay 1000000 -- 1s - ) - ( do - takeMVar sleeping - use pool $ selectOneSession - ) - t1 <- getCurrentTime - res `shouldBe` Right (Left AcquisitionTimeoutUsageError) - diffUTCTime t1 t0 `shouldSatisfy` (< 0.5) -- 0.5s + it "Times out connection acquisition" $ + -- 1ms timeout + withPool 1 0.001 1_800 connectionSettings $ \pool -> do + sleeping <- newEmptyMVar + t0 <- getCurrentTime + res <- + race + ( use pool $ + liftIO $ do + putMVar sleeping () + threadDelay 1_000_000 -- 1s + ) + ( do + takeMVar sleeping + use pool $ selectOneSession + ) + t1 <- getCurrentTime + res `shouldBe` Right (Left AcquisitionTimeoutUsageError) + diffUTCTime t1 t0 `shouldSatisfy` (< 0.5) -- 0.5s + it "Passively times out old connections" $ + -- 0.5s connection lifetime + withPool 1 10 0.5 connectionSettings $ \pool -> do + res <- use pool $ setSettingSession "testing.foo" "hello world" + res `shouldBe` Right () + res2 <- use pool $ getSettingSession "testing.foo" + res2 `shouldBe` Right (Just "hello world") + threadDelay 1_000_000 -- 1s + res3 <- use pool $ getSettingSession "testing.foo" + res3 `shouldBe` Right Nothing + it "Counts active connections" $ do + (taggedConnectionSettings, appName) <- tagConnection connectionSettings + withPool 3 10 1_800 taggedConnectionSettings $ \pool -> do + res <- use pool $ countConnectionsSession appName + res `shouldBe` Right 1 + it "Times out old connections" $ do + withDefaultPool $ \countPool -> do + (taggedConnectionSettings, appName) <- tagConnection connectionSettings + withPool 3 10 0.5 taggedConnectionSettings $ \limitedPool -> do + res <- use limitedPool $ selectOneSession + res `shouldBe` Right 1 + res2 <- use countPool $ countConnectionsSession appName + res2 `shouldBe` Right 1 + threadDelay 1_000_000 -- 1s + res3 <- use countPool $ countConnectionsSession appName + res3 `shouldBe` Right 0 getConnectionSettings :: IO Connection.Settings getConnectionSettings = @@ -111,6 +137,12 @@ val <- getEnv return $ (\v -> label <> "=" <> v) <$> val +tagConnection :: Connection.Settings -> IO (Connection.Settings, Text) +tagConnection connectionSettings = do + tag <- Random.uniformWord32 Random.globalStdGen + let appName = "hasql-pool-test-" <> show tag + return (connectionSettings <> " application_name=" <> B8.pack appName, Text.pack appName) + selectOneSession :: Session.Session Int64 selectOneSession = Session.statement () statement @@ -145,3 +177,11 @@ statement = Statement.Statement "SELECT current_setting($1, true)" encoder decoder True encoder = Encoders.param (Encoders.nonNullable Encoders.text) decoder = Decoders.singleRow (Decoders.column (Decoders.nullable Decoders.text)) + +countConnectionsSession :: Text -> Session.Session Int64 +countConnectionsSession appName = do + Session.statement appName statement + where + statement = Statement.Statement "SELECT count(*) FROM pg_stat_activity WHERE application_name = $1" encoder decoder True + encoder = Encoders.param (Encoders.nonNullable Encoders.text) + decoder = Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.int8))