hasql-pool 0.5.2.1 → 0.5.2.2

raw patch · 4 files changed

+66/−271 lines, 4 filesdep +resource-pooldep −basedep −stmdep −transformersdep ~base-preludePVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependencies added: resource-pool

Dependencies removed: base, stm, transformers

Dependency ranges changed: base-prelude

API changes (from Hackage documentation)

Files

• hasql-pool.cabal +13/−10
• library/Hasql/Pool.hs +29/−161
• library/Hasql/Pool/Prelude.hs +3/−100
• library/Hasql/Pool/ResourcePool.hs +21/−0

@@ -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

@@ -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

@@ -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)

@@ -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)