hasql-pool 0.5.2.2 → 0.6

raw patch · 7 files changed

+339/−114 lines, 7 filesdep +basedep +rerebasedep +stmdep −base-preludedep −resource-poolsetup-changedPVP ok

version bump matches the API change (PVP)

Dependencies added: base, rerebase, stm, transformers

Dependencies removed: base-prelude, resource-pool

API changes (from Hackage documentation)

Files

• CHANGELOG.md +8/−0
• Setup.hs +0/−2
• hasql-pool.cabal +13/−14
• library/Hasql/Pool.hs +173/−47
• library/Hasql/Pool/Prelude.hs +86/−10
• library/Hasql/Pool/ResourcePool.hs +0/−21
• test/Main.hs +59/−20

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

@@ -1,2 +0,0 @@-import Distribution.Simple-main = defaultMain

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

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

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

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

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