hasql-pool 0.5.2 → 0.5.2.1

raw patch · 4 files changed

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

API removals or changes: PVP suggests a major version bump

Dependencies added: base, stm, transformers

Dependencies removed: resource-pool

Dependency ranges changed: base-prelude, hasql

API changes (from Hackage documentation)

Files

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

@@ -1,7 +1,7 @@ name:   hasql-pool version:-  0.5.2+  0.5.2.1 category:   Hasql, Database, PostgreSQL synopsis:@@ -41,20 +41,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.5,-    -- 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@@ -72,4 +68,5 @@     base-prelude,     hasql,     hasql-pool,-    hspec >= 2.6 && < 3+    stm >=2.5 && <3,+    hspec >=2.6 && <3

@@ -10,66 +10,198 @@ where  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.Connection (Connection)+import Hasql.Session (Session)+import qualified Hasql.Connection as Connection+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+                          -- 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+ -- | -- 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.+-- 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+  ConnectionUsageError Connection.ConnectionError |+  SessionUsageError Session.QueryError |+  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 :: 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

@@ -1,14 +1,111 @@ 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 --- base-prelude+-- stm --------------------------import BasePrelude as Exports hiding (assert, left, right, isLeft, isRight, error)+import Control.Concurrent.STM as Exports hiding (orElse) +-- -- text+-- -------------------------+-- import Data.Text as Exports (Text)++-- -- bytestring+-- -------------------------+-- import Data.ByteString as Exports (ByteString)+ -- 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)

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