diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,8 @@ +# 0.6 + +Moved away from "resource-pool" and fixed the handling of lost connections. + +Breaking: + +- Changed the suffix of `UsageError` constructors from `Error` to `UsageError` +- Added `PoolIsReleasedUsageError` diff --git a/Setup.hs b/Setup.hs deleted file mode 100644 --- a/Setup.hs +++ /dev/null @@ -1,2 +0,0 @@ -import Distribution.Simple -main = defaultMain diff --git a/hasql-pool.cabal b/hasql-pool.cabal --- a/hasql-pool.cabal +++ b/hasql-pool.cabal @@ -1,7 +1,7 @@ name: hasql-pool version: - 0.5.2.2 + 0.6 category: Hasql, Database, PostgreSQL synopsis: @@ -24,6 +24,8 @@ Simple cabal-version: >=1.10 +extra-source-files: + CHANGELOG.md source-repository head @@ -41,20 +43,16 @@ Arrows, BangPatterns, ConstraintKinds, DataKinds, DefaultSignatures, DeriveDataTypeable, DeriveFoldable, DeriveFunctor, DeriveGeneric, DeriveTraversable, EmptyDataDecls, FlexibleContexts, FlexibleInstances, FunctionalDependencies, GADTs, GeneralizedNewtypeDeriving, LambdaCase, LiberalTypeSynonyms, MagicHash, MultiParamTypeClasses, MultiWayIf, NoImplicitPrelude, NoMonomorphismRestriction, OverloadedStrings, PatternGuards, ParallelListComp, QuasiQuotes, RankNTypes, RecordWildCards, ScopedTypeVariables, StandaloneDeriving, TemplateHaskell, TupleSections, TypeFamilies, TypeOperators, UnboxedTuples default-language: Haskell2010 - other-modules: - Hasql.Pool.Prelude - Hasql.Pool.ResourcePool exposed-modules: Hasql.Pool + other-modules: + Hasql.Pool.Prelude build-depends: - -- resources: - resource-pool >= 0.2 && < 0.3, - -- database: - hasql >= 1.3 && < 1.6, - -- data: - time >= 1.5 && < 2, - -- general: - base-prelude >= 1 && < 2 + base >=4.11 && <5, + hasql >=1.3 && <1.6, + stm >=2.5 && <3, + time >=1.5 && <2, + transformers >=0.5 && <0.7 test-suite test @@ -69,7 +67,8 @@ default-language: Haskell2010 build-depends: - base-prelude, hasql, hasql-pool, - hspec >= 2.6 && < 3 + hspec >=2.6 && <3, + rerebase >=1.15 && <2, + stm >=2.5 && <3 diff --git a/library/Hasql/Pool.hs b/library/Hasql/Pool.hs --- a/library/Hasql/Pool.hs +++ b/library/Hasql/Pool.hs @@ -1,75 +1,201 @@ module Hasql.Pool -( - Pool, - Settings(..), - acquire, - release, - UsageError(..), - use, -) + ( Pool, + Settings (..), + acquire, + release, + UsageError (..), + use, + ) where +import Hasql.Connection (Connection) +import qualified Hasql.Connection as Connection import Hasql.Pool.Prelude -import qualified Hasql.Connection -import qualified Hasql.Session -import qualified Data.Pool as ResourcePool -import qualified Hasql.Pool.ResourcePool as ResourcePool - +import Hasql.Session (Session) +import qualified Hasql.Session as Session -- | -- A pool of connections to DB. -newtype Pool = - Pool (ResourcePool.Pool (Either Hasql.Connection.ConnectionError Hasql.Connection.Connection)) - deriving (Show) +data Pool + = Pool + Connection.Settings + -- ^ Connection settings. + (TQueue ActiveConnection) + -- ^ Queue of established connections. + (TVar Int) + -- ^ Slots available for establishing new connections. + (TVar Bool) + -- ^ Flag signaling whether pool's alive. +data ActiveConnection = ActiveConnection + { activeConnectionLastUseTimestamp :: Int, + activeConnectionConnection :: Connection + } + +loopCollectingGarbage :: Int -> TQueue ActiveConnection -> TVar Int -> TVar Bool -> IO () +loopCollectingGarbage timeout establishedQueue slotsAvailVar aliveVar = + decide + where + decide = + do + ts <- getMillisecondsSinceEpoch + join $ + atomically $ do + alive <- readTVar aliveVar + if alive + then + let tryToRelease = + tryReadTQueue establishedQueue >>= \case + -- The queue is empty. Just wait for changes in the state. + Nothing -> + retry + Just entry@(ActiveConnection lastUseTs connection) -> + let outdatingTs = + lastUseTs + timeout + in -- Check whether it's outdated. + if outdatingTs < ts + then -- Fetch the current value of available slots and + -- release this one and other connections. + do + slotsAvail <- readTVar slotsAvailVar + collectAndRelease slotsAvail [connection] outdatingTs + else -- Return it to the front of the queue and + -- wait until it's outdating time. + do + unGetTQueue establishedQueue entry + return (sleep outdatingTs *> decide) + collectAndRelease !slotsAvail !outdatedList outdatingTs = + tryReadTQueue establishedQueue >>= \case + Nothing -> + finalizeAndRelease slotsAvail outdatedList outdatingTs + Just entry@(ActiveConnection lastUseTs connection) -> + let outdatingTs = + lastUseTs + timeout + in if outdatingTs < ts + then do + unGetTQueue establishedQueue entry + finalizeAndRelease slotsAvail outdatedList outdatingTs + else collectAndRelease (succ slotsAvail) (connection : outdatedList) outdatingTs + finalizeAndRelease slotsAvail outdatedList outdatingTs = + do + writeTVar slotsAvailVar slotsAvail + return (release outdatedList *> sleep outdatingTs *> decide) + in tryToRelease + else do + list <- flushTQueue establishedQueue + return (release (fmap activeConnectionConnection list)) + sleep untilTs = + do + ts <- getMillisecondsSinceEpoch + let diff = + untilTs - ts + in if diff > 0 + then threadDelay (diff * 1000) + else return () + release = + traverse_ Connection.release + -- | -- Settings of the connection pool. Consist of: --- +-- -- * Pool-size. --- --- * Timeout. --- An amount of time for which an unused resource is kept open. --- The smallest acceptable value is 0.5 seconds. --- +-- +-- * Timeout. +-- An amount of time in milliseconds for which the unused connections are kept open. +-- -- * Connection settings. --- type Settings = - (Int, NominalDiffTime, Hasql.Connection.Settings) + (Int, Int, Connection.Settings) -- | -- Given the pool-size, timeout and connection settings -- create a connection-pool. acquire :: Settings -> IO Pool acquire (size, timeout, connectionSettings) = - fmap Pool $ - ResourcePool.createPool acquire release stripes timeout size - where - acquire = - Hasql.Connection.acquire connectionSettings - release = - either (const (pure ())) Hasql.Connection.release - stripes = - 1 + do + establishedQueue <- newTQueueIO + slotsAvailVar <- newTVarIO size + aliveVar <- newTVarIO (size > 0) + forkIO $ loopCollectingGarbage timeout establishedQueue slotsAvailVar aliveVar + return (Pool connectionSettings establishedQueue slotsAvailVar aliveVar) -- | -- Release the connection-pool. release :: Pool -> IO () -release (Pool pool) = - ResourcePool.destroyAllResources pool +release (Pool _ _ _ aliveVar) = + atomically (writeTVar aliveVar False) -- | -- A union over the connection establishment error and the session error. -data UsageError = - ConnectionError Hasql.Connection.ConnectionError | - SessionError Hasql.Session.QueryError +data UsageError + = -- | Error during an attempt to connect. + ConnectionUsageError Connection.ConnectionError + | -- | Error during session execution. + SessionUsageError Session.QueryError + | -- | Pool has been released and can no longer be used. + PoolIsReleasedUsageError deriving (Show, Eq) --- | --- Use a connection from the pool to run a session and --- return the connection to the pool, when finished. -use :: Pool -> Hasql.Session.Session a -> IO (Either UsageError a) -use (Pool pool) session = - fmap (either (Left . ConnectionError) (either (Left . SessionError) Right)) $ - ResourcePool.withResourceOnEither pool $ - traverse $ - Hasql.Session.run session +-- | Use a connection from the pool to run a session and return the connection +-- to the pool, when finished. If the session fails +-- with 'Session.ClientError' the connection gets reestablished. +use :: Pool -> Session.Session a -> IO (Either UsageError a) +use (Pool connectionSettings establishedQueue slotsAvailVar aliveVar) session = + join $ + atomically $ do + alive <- readTVar aliveVar + if alive + then + tryReadTQueue establishedQueue >>= \case + -- No established connection avail at the moment. + Nothing -> do + slotsAvail <- readTVar slotsAvailVar + -- Do we have any slots left for establishing new connections? + if slotsAvail > 0 + then -- Reduce the available slots var and instruct to + -- establish and use a new connection. + do + writeTVar slotsAvailVar $! pred slotsAvail + return acquireConnectionThenUseThenPutItToQueue + else -- Wait until the state changes and retry. + + retry + Just (ActiveConnection _ connection) -> + return (useConnectionThenPutItToQueue connection) + else return (return (Left PoolIsReleasedUsageError)) + where + acquireConnectionThenUseThenPutItToQueue = + do + res <- Connection.acquire connectionSettings + case res of + -- Failed to acquire, so release an availability slot, + -- returning the error details. + Left acquisitionError -> do + atomically $ modifyTVar' slotsAvailVar succ + return (Left (ConnectionUsageError acquisitionError)) + Right connection -> + useConnectionThenPutItToQueue connection + useConnectionThenPutItToQueue connection = + do + res <- Session.run session connection + case res of + Left queryError -> do + -- Check whether the error is on client-side, + -- and in that case release the connection. + case queryError of + Session.QueryError _ _ (Session.ClientError _) -> + releaseConnection connection + _ -> + putConnectionToPool connection + return (Left (SessionUsageError queryError)) + Right res -> do + putConnectionToPool connection + return (Right res) + putConnectionToPool connection = + do + ts <- getMillisecondsSinceEpoch + atomically $ writeTQueue establishedQueue (ActiveConnection ts connection) + releaseConnection connection = + do + atomically $ modifyTVar' slotsAvailVar succ + Connection.release connection 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 @@ -1,14 +1,90 @@ module Hasql.Pool.Prelude -( - module Exports, -) + ( module Exports, + getMillisecondsSinceEpoch, + ) where - --- base-prelude -------------------------- -import BasePrelude as Exports hiding (assert, left, right, isLeft, isRight, error) - --- time -------------------------- +import Control.Applicative as Exports hiding (WrappedArrow (..)) +import Control.Arrow as Exports hiding (first, second) +import Control.Category as Exports +import Control.Concurrent as Exports +import Control.Concurrent.STM as Exports hiding (orElse) +import Control.Exception as Exports +import Control.Monad as Exports hiding (fail, forM, forM_, mapM, mapM_, msum, sequence, sequence_) +import Control.Monad.Fail as Exports +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 +import Data.Char as Exports +import Data.Coerce as Exports +import Data.Complex as Exports +import Data.Data as Exports +import Data.Dynamic as Exports +import Data.Either as Exports +import Data.Fixed as Exports +import Data.Foldable as Exports hiding (toList) +import Data.Function as Exports hiding (id, (.)) +import Data.Functor as Exports +import Data.Functor.Compose as Exports +import Data.IORef as Exports +import Data.Int as Exports +import Data.Ix as Exports +import Data.List as Exports hiding (all, and, any, concat, concatMap, elem, find, foldl, foldl', foldl1, foldr, foldr1, isSubsequenceOf, mapAccumL, mapAccumR, maximum, maximumBy, minimum, minimumBy, notElem, or, product, sortOn, sum, uncons) +import Data.List.NonEmpty as Exports (NonEmpty (..)) +import Data.Maybe as Exports +import Data.Monoid as Exports hiding (Alt) +import Data.Ord as Exports +import Data.Proxy as Exports +import Data.Ratio as Exports +import Data.STRef as Exports +import Data.String as Exports import Data.Time as Exports +import Data.Time.Clock.POSIX as Exports +import Data.Time.Clock.System as Exports +import Data.Traversable as Exports +import Data.Tuple as Exports +import Data.Unique as Exports +import Data.Version as Exports +import Data.Void as Exports +import Data.Word as Exports +import Debug.Trace as Exports +import Foreign.ForeignPtr as Exports +import Foreign.Ptr as Exports +import Foreign.StablePtr as Exports +import Foreign.Storable as Exports +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) +import GHC.IO.Exception as Exports +import Numeric as Exports +import System.Environment as Exports +import System.Exit as Exports +import System.IO as Exports (Handle, hClose) +import System.IO.Error as Exports +import System.IO.Unsafe as Exports +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 +import Prelude as Exports hiding (all, and, any, concat, concatMap, elem, fail, foldl, foldl1, foldr, foldr1, id, mapM, mapM_, maximum, minimum, notElem, or, product, sequence, sequence_, sum, (.)) + +getMillisecondsSinceEpoch :: IO Int +getMillisecondsSinceEpoch = + fmap (fromIntegral . systemTimeToMicros) getSystemTime + where + systemTimeToMicros (MkSystemTime s ns) = + s * 1000 + fromIntegral (div ns 1000000) diff --git a/library/Hasql/Pool/ResourcePool.hs b/library/Hasql/Pool/ResourcePool.hs deleted file mode 100644 --- a/library/Hasql/Pool/ResourcePool.hs +++ /dev/null @@ -1,21 +0,0 @@ -{-| -Extras for the resource-pool library. --} -module Hasql.Pool.ResourcePool -where - -import Hasql.Pool.Prelude -import Data.Pool - - -withResourceOnEither :: Pool resource -> (resource -> IO (Either failure success)) -> IO (Either failure success) -withResourceOnEither pool act = mask_ $ do - (resource, localPool) <- takeResource pool - failureOrSuccess <- act resource `onException` destroyResource pool localPool resource - case failureOrSuccess of - Right success -> do - putResource localPool resource - return (Right success) - Left failure -> do - destroyResource pool localPool resource - return (Left failure) diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -1,29 +1,68 @@ module Main where -import BasePrelude -import Test.Hspec -import Hasql.Pool import qualified Hasql.Connection as Connection import qualified Hasql.Decoders as Decoders import qualified Hasql.Encoders as Encoders +import Hasql.Pool import qualified Hasql.Session as Session import qualified Hasql.Statement as Statement - +import Test.Hspec +import Prelude main = hspec $ do - describe "Hasql.Pool.use" $ do - it "releases a spot in the pool when there is an error" $ do - pool <- acquire (1, 1, "host=localhost port=5432 user=postgres dbname=postgres") - let - statement = Statement.Statement "" Encoders.noParams Decoders.noResult True - session = Session.statement () statement - in do - use pool session `shouldNotReturn` (Right ()) - let - session = let - statement = let - decoder = Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.int8)) - in Statement.Statement "SELECT 1" Encoders.noParams decoder True - in Session.statement () statement - in do - use pool session `shouldReturn` (Right 1) + describe "" $ do + it "Releases a spot in the pool when there is a query error" $ do + pool <- acquire (1, 1, connectionSettings) + use pool badQuerySession `shouldNotReturn` (Right ()) + use pool selectOneSession `shouldReturn` (Right 1) + it "Simulation of connection error works" $ do + pool <- acquire (3, 1, connectionSettings) + 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, 1, connectionSettings) + 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, 1, connectionSettings) + 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, 1, connectionSettings) + res <- use pool $ badQuerySession + res <- use pool $ badQuerySession + res <- use pool $ badQuerySession + res <- use pool $ badQuerySession + res <- use pool $ selectOneSession + shouldSatisfy res $ isRight + +connectionSettings :: Connection.Settings +connectionSettings = + "host=localhost port=5432 user=postgres dbname=postgres" + +selectOneSession :: Session.Session Int64 +selectOneSession = + Session.statement () statement + where + statement = Statement.Statement "SELECT 1" Encoders.noParams decoder True + decoder = Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.int8)) + +badQuerySession :: Session.Session () +badQuerySession = + Session.statement () statement + where + statement = Statement.Statement "" Encoders.noParams Decoders.noResult True + +closeConnSession :: Session.Session () +closeConnSession = do + conn <- ask + liftIO $ Connection.release conn