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.1 + 0.5.2.2 category: Hasql, Database, PostgreSQL synopsis: @@ -41,16 +41,20 @@ 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 - exposed-modules: - Hasql.Pool other-modules: Hasql.Pool.Prelude + Hasql.Pool.ResourcePool + exposed-modules: + Hasql.Pool build-depends: - base >=4.11 && <5, - hasql >=1.3 && <1.6, - stm >=2.5 && <3, - time >=1.5 && <2, - transformers >=0.5 && <0.7 + -- resources: + resource-pool >= 0.2 && < 0.3, + -- database: + hasql >= 1.3 && < 1.6, + -- data: + time >= 1.5 && < 2, + -- general: + base-prelude >= 1 && < 2 test-suite test @@ -68,5 +72,4 @@ base-prelude, hasql, hasql-pool, - stm >=2.5 && <3, - hspec >=2.6 && <3 + hspec >= 2.6 && < 3 diff --git a/library/Hasql/Pool.hs b/library/Hasql/Pool.hs --- a/library/Hasql/Pool.hs +++ b/library/Hasql/Pool.hs @@ -10,98 +10,17 @@ where import Hasql.Pool.Prelude -import Hasql.Connection (Connection) -import Hasql.Session (Session) -import qualified Hasql.Connection as Connection -import qualified Hasql.Session as Session +import qualified Hasql.Connection +import qualified Hasql.Session +import qualified Data.Pool as ResourcePool +import qualified Hasql.Pool.ResourcePool as ResourcePool -- | -- A pool of connections to DB. -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 - -- Fetch the current value of available slots and - -- release this one and other connections. - then do - slotsAvail <- readTVar slotsAvailVar - collectAndRelease slotsAvail [connection] outdatingTs - -- Return it to the front of the queue and - -- wait until it's outdating time. - else 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 +newtype Pool = + Pool (ResourcePool.Pool (Either Hasql.Connection.ConnectionError Hasql.Connection.Connection)) + deriving (Show) -- | -- Settings of the connection pool. Consist of: @@ -109,99 +28,48 @@ -- * Pool-size. -- -- * Timeout. --- An amount of time in milliseconds for which the unused connections are kept open. +-- An amount of time for which an unused resource is kept open. +-- The smallest acceptable value is 0.5 seconds. -- -- * Connection settings. -- type Settings = - (Int, Int, Connection.Settings) + (Int, NominalDiffTime, Hasql.Connection.Settings) -- | -- Given the pool-size, timeout and connection settings -- create a connection-pool. acquire :: Settings -> IO Pool acquire (size, timeout, connectionSettings) = - do - establishedQueue <- newTQueueIO - slotsAvailVar <- newTVarIO size - aliveVar <- newTVarIO (size > 0) - forkIO $ loopCollectingGarbage timeout establishedQueue slotsAvailVar aliveVar - return (Pool connectionSettings establishedQueue slotsAvailVar aliveVar) + fmap Pool $ + ResourcePool.createPool acquire release stripes timeout size + where + acquire = + Hasql.Connection.acquire connectionSettings + release = + either (const (pure ())) Hasql.Connection.release + stripes = + 1 -- | -- Release the connection-pool. release :: Pool -> IO () -release (Pool _ _ _ aliveVar) = - atomically (writeTVar aliveVar False) +release (Pool pool) = + ResourcePool.destroyAllResources pool -- | -- A union over the connection establishment error and the session error. data UsageError = - ConnectionUsageError Connection.ConnectionError | - SessionUsageError Session.QueryError | - PoolIsReleasedUsageError + ConnectionError Hasql.Connection.ConnectionError | + SessionError Hasql.Session.QueryError deriving (Show, Eq) -- | -- Use a connection from the pool to run a session and -- return the connection to the pool, when finished. -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 - -- Reduce the available slots var and instruct to - -- establish and use a new connection. - then do - writeTVar slotsAvailVar $! pred slotsAvail - return acquireConnectionThenUseThenPutItToQueue - -- Wait until the state changes and retry. - else - 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 +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 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,111 +1,14 @@ module Hasql.Pool.Prelude -( +( module Exports, - getMillisecondsSinceEpoch, ) where --- base -------------------------- -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.Exception as Exports -import Control.Monad as Exports hiding (fail, mapM_, sequence_, forM_, msum, mapM, sequence, forM) -import Control.Monad.IO.Class as Exports -import Control.Monad.Fail as Exports -import Control.Monad.Fix as Exports hiding (fix) -import Control.Monad.ST as Exports -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.Int as Exports -import Data.IORef as Exports -import Data.Ix as Exports -import Data.List as Exports hiding (sortOn, isSubsequenceOf, uncons, concat, foldr, foldl1, maximum, minimum, product, sum, all, and, any, concatMap, elem, foldl, foldr1, notElem, or, find, maximumBy, minimumBy, mapAccumL, mapAccumR, foldl') -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.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, withMVar, threadWaitWriteSTM, threadWaitWrite, threadWaitReadSTM, threadWaitRead) -import GHC.Exts as Exports (IsList(..), lazy, inline, sortWith, groupWith) -import GHC.Generics as Exports (Generic) -import GHC.IO.Exception as Exports -import Numeric as Exports -import Prelude as Exports hiding (fail, concat, foldr, mapM_, sequence_, foldl1, maximum, minimum, product, sum, all, and, any, concatMap, elem, foldl, foldr1, notElem, or, mapM, sequence, id, (.)) -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, readPrec_to_P, readP_to_Prec, readPrec_to_S, readS_to_Prec) -import Text.Printf as Exports (printf, hPrintf) -import Text.Read as Exports (Read(..), readMaybe, readEither) -import Unsafe.Coerce as Exports --- stm +-- base-prelude ------------------------- -import Control.Concurrent.STM as Exports hiding (orElse) - --- -- text --- ------------------------- --- import Data.Text as Exports (Text) - --- -- bytestring --- ------------------------- --- import Data.ByteString as Exports (ByteString) +import BasePrelude as Exports hiding (assert, left, right, isLeft, isRight, error) -- time ------------------------- import Data.Time as Exports -import Data.Time.Clock.POSIX as Exports -import Data.Time.Clock.System as Exports - --- transformers -------------------------- -import Control.Monad.Trans.Class as Exports -import Control.Monad.Trans.Cont as Exports hiding (shift, callCC) -import Control.Monad.Trans.Except as Exports (ExceptT(ExceptT), Except, except, runExcept, runExceptT, mapExcept, mapExceptT, withExcept, withExceptT) -import Control.Monad.Trans.Maybe as Exports -import Control.Monad.Trans.Reader as Exports (Reader, runReader, mapReader, withReader, ReaderT(ReaderT), runReaderT, mapReaderT, withReaderT) -import Control.Monad.Trans.State.Strict as Exports (State, runState, evalState, execState, mapState, withState, StateT(StateT), runStateT, evalStateT, execStateT, mapStateT, withStateT) -import Control.Monad.Trans.Writer.Strict as Exports (Writer, runWriter, execWriter, mapWriter, WriterT(..), execWriterT, mapWriterT) - - -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 new file mode 100644 --- /dev/null +++ b/library/Hasql/Pool/ResourcePool.hs @@ -0,0 +1,21 @@ +{-| +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)