diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,15 @@ +# 0.8 + +`release` became reusable. You can use it to destroy the whole pool (same as before), but now also you can use it to reset the connections. + +Acquisition timeout added. + +Breaking changes in API: + +- Removed `PoolIsReleasedUsageError` +- `acquire` extended with the acquisition timeout parameter +- `acquireDynamically` extended with the acquisition timeout parameter + # 0.7.2 Added support for dynamic connection configuration ([issue #11](https://github.com/nikita-volkov/hasql-pool/issues/11)). diff --git a/hasql-pool.cabal b/hasql-pool.cabal --- a/hasql-pool.cabal +++ b/hasql-pool.cabal @@ -1,74 +1,53 @@ -name: - hasql-pool -version: - 0.7.2.1 -category: - Hasql, Database, PostgreSQL -synopsis: - A pool of connections for Hasql -homepage: - https://github.com/nikita-volkov/hasql-pool -bug-reports: - https://github.com/nikita-volkov/hasql-pool/issues -author: - Nikita Volkov -maintainer: - Nikita Volkov -copyright: - (c) 2015, Nikita Volkov -license: - MIT -license-file: - LICENSE -build-type: - Simple -cabal-version: - >=1.10 -extra-source-files: - CHANGELOG.md +cabal-version: 3.0 +name: hasql-pool +version: 0.8 +category: Hasql, Database, PostgreSQL +synopsis: Pool of connections for Hasql +homepage: https://github.com/nikita-volkov/hasql-pool +bug-reports: https://github.com/nikita-volkov/hasql-pool/issues +author: Nikita Volkov +maintainer: Nikita Volkov +copyright: (c) 2015, Nikita Volkov +license: MIT +license-file: LICENSE +extra-source-files: CHANGELOG.md + source-repository head - type: - git - location: - git://github.com/nikita-volkov/hasql-pool.git + type: git + location: git://github.com/nikita-volkov/hasql-pool.git +common base-settings + default-extensions: BangPatterns, BlockArguments, ConstraintKinds, DataKinds, DefaultSignatures, DeriveDataTypeable, DeriveFoldable, DeriveFunctor, DeriveGeneric, DerivingVia, DeriveTraversable, EmptyDataDecls, FlexibleContexts, FlexibleInstances, FunctionalDependencies, GADTs, GeneralizedNewtypeDeriving, InstanceSigs, LambdaCase, LiberalTypeSynonyms, MagicHash, MultiParamTypeClasses, MultiWayIf, NoImplicitPrelude, NoMonomorphismRestriction, OverloadedStrings, PatternGuards, ParallelListComp, QuasiQuotes, RankNTypes, RecordWildCards, ScopedTypeVariables, StandaloneDeriving, StrictData, TemplateHaskell, TupleSections, TypeApplications, TypeFamilies, TypeOperators, UnboxedTuples + default-language: Haskell2010 library - hs-source-dirs: - library - ghc-options: - default-extensions: - 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 + import: base-settings + hs-source-dirs: library exposed-modules: Hasql.Pool other-modules: Hasql.Pool.Prelude + Hasql.Pool.TimeExtras.IO + Hasql.Pool.TimeExtras.Conversions build-depends: base >=4.11 && <5, - hasql >=1.6 && <1.7, + hasql >=1.6.0.1 && <1.7, stm >=2.5 && <3, time >=1.5 && <2, - transformers >=0.5 && <0.7 - + transformers >=0.5 && <0.7, test-suite test - type: - exitcode-stdio-1.0 - hs-source-dirs: - test - main-is: - Main.hs - default-extensions: - 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 + import: base-settings + type: exitcode-stdio-1.0 + hs-source-dirs: test + main-is: Main.hs + ghc-options: -threaded build-depends: + async >=2.2 && <3, hasql, hasql-pool, hspec >=2.6 && <3, rerebase >=1.15 && <2, - stm >=2.5 && <3 + 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 @@ -17,48 +17,74 @@ import Hasql.Session (Session) import qualified Hasql.Session as Session +data ReuseConnection = Keep | Close + -- | A pool of connections to DB. data Pool = Pool { -- | Connection settings. poolFetchConnectionSettings :: IO Connection.Settings, + -- | Acquisition timeout, in microseconds. + poolAcquisitionTimeout :: Maybe Int, -- | Avail connections. poolConnectionQueue :: TQueue Connection, - -- | Capacity. + -- | Remaining capacity. + -- The pool size limits the sum of poolCapacity, the length + -- of length poolConnectionQueue and the number of in-flight + -- connections. poolCapacity :: TVar Int, - -- | Alive. - poolAlive :: TVar Bool + -- | Whether to return a connection to the pool. + poolReuseToggle :: TVar (TVar ReuseConnection) } --- | Given the pool-size and connection settings create a connection-pool. +-- | Create a connection-pool. -- -- No connections actually get established by this function. It is delegated -- to 'use'. -acquire :: Int -> Connection.Settings -> IO Pool -acquire poolSize connectionSettings = - acquireDynamically poolSize (pure connectionSettings) +acquire :: + -- | Pool size. + Int -> + -- | Connection acquisition timeout. + Maybe Int -> + -- | Connection settings. + Connection.Settings -> + IO Pool +acquire poolSize timeout connectionSettings = + acquireDynamically poolSize timeout (pure connectionSettings) --- | Given the pool-size and connection settings constructor action --- create a connection-pool. --- --- No connections actually get established by this function. It is delegated --- to 'use'. +-- | Create a connection-pool. -- -- In difference to 'acquire' new settings get fetched each time a connection -- is created. This may be useful for some security models. -acquireDynamically :: Int -> IO Connection.Settings -> IO Pool -acquireDynamically poolSize fetchConnectionSettings = do - Pool fetchConnectionSettings +-- +-- No connections actually get established by this function. It is delegated +-- to 'use'. +acquireDynamically :: + -- | Pool size. + Int -> + -- | Connection acquisition timeout. + Maybe Int -> + -- | Action fetching connection settings settings. + IO Connection.Settings -> + IO Pool +acquireDynamically poolSize timeout fetchConnectionSettings = do + Pool fetchConnectionSettings timeout <$> newTQueueIO <*> newTVarIO poolSize - <*> newTVarIO True + <*> (newTVarIO =<< newTVarIO Keep) --- | Release all the connections in the pool. +-- | Release all the idle connections in the pool, and mark the in-use connections +-- to be released on return. Any connections acquired after the call will be +-- newly established. release :: Pool -> IO () -release Pool {..} = do - connections <- atomically $ do - writeTVar poolAlive False - flushTQueue poolConnectionQueue - forM_ connections Connection.release +release Pool {..} = + join . atomically $ do + prevReuseToggle <- readTVar poolReuseToggle + writeTVar prevReuseToggle Close + newReuseToggle <- newTVar Keep + writeTVar poolReuseToggle newReuseToggle + conns <- flushTQueue poolConnectionQueue + modifyTVar' poolCapacity (+ (length conns)) + return $ forM_ conns Connection.release -- | Use a connection from the pool to run a session and return the connection -- to the pool, when finished. @@ -68,32 +94,40 @@ -- and a slot gets freed up for a new connection to be established the next -- time one is needed. The error still gets returned from this function. use :: Pool -> Session.Session a -> IO (Either UsageError a) -use Pool {..} sess = +use Pool {..} sess = do + timeout <- case poolAcquisitionTimeout of + Just delta -> do + delay <- registerDelay delta + return $ readTVar delay + Nothing -> + return $ return False join . atomically $ do - alive <- readTVar poolAlive - if alive - then - asum - [ readTQueue poolConnectionQueue <&> onConn, - do - capVal <- readTVar poolCapacity - if capVal > 0 - then do - writeTVar poolCapacity $! pred capVal - return onNewConn - else retry - ] - else return . return . Left $ PoolIsReleasedUsageError + reuseToggle <- readTVar poolReuseToggle + asum + [ readTQueue poolConnectionQueue <&> onConn reuseToggle, + do + capVal <- readTVar poolCapacity + if capVal > 0 + then do + writeTVar poolCapacity $! pred capVal + return $ onNewConn reuseToggle + else retry, + do + timedOut <- timeout + if timedOut + then return . return . Left $ AcquisitionTimeoutUsageError + else retry + ] where - onNewConn = do + onNewConn reuseToggle = do settings <- poolFetchConnectionSettings connRes <- Connection.acquire settings case connRes of Left connErr -> do atomically $ modifyTVar' poolCapacity succ return $ Left $ ConnectionUsageError connErr - Right conn -> onConn conn - onConn conn = do + Right conn -> onConn reuseToggle conn + onConn reuseToggle conn = do sessRes <- Session.run sess conn case sessRes of Left err -> case err of @@ -109,10 +143,12 @@ where returnConn = join . atomically $ do - alive <- readTVar poolAlive - if alive - then writeTQueue poolConnectionQueue conn $> return () - else return $ Connection.release conn + reuse <- readTVar reuseToggle + case reuse of + Keep -> writeTQueue poolConnectionQueue conn $> return () + Close -> do + modifyTVar' poolCapacity succ + return $ Connection.release conn -- | Union over all errors that 'use' can result in. data UsageError @@ -120,8 +156,8 @@ ConnectionUsageError Connection.ConnectionError | -- | Session execution failed. SessionUsageError Session.QueryError - | -- | Attempt to use a pool, which has already been called 'release' upon. - PoolIsReleasedUsageError + | -- | Timeout acquiring a connection. + AcquisitionTimeoutUsageError deriving (Show, Eq) instance Exception UsageError 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,6 +1,5 @@ module Hasql.Pool.Prelude ( module Exports, - getMillisecondsSinceEpoch, ) where @@ -81,10 +80,3 @@ 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/TimeExtras/Conversions.hs b/library/Hasql/Pool/TimeExtras/Conversions.hs new file mode 100644 --- /dev/null +++ b/library/Hasql/Pool/TimeExtras/Conversions.hs @@ -0,0 +1,10 @@ +module Hasql.Pool.TimeExtras.Conversions where + +import Hasql.Pool.Prelude + +class ToMilliseconds a where + toMilliseconds :: a -> Int + +instance ToMilliseconds SystemTime where + toMilliseconds (MkSystemTime s ns) = + fromIntegral s * 1000 + fromIntegral (div ns 1000000) diff --git a/library/Hasql/Pool/TimeExtras/IO.hs b/library/Hasql/Pool/TimeExtras/IO.hs new file mode 100644 --- /dev/null +++ b/library/Hasql/Pool/TimeExtras/IO.hs @@ -0,0 +1,8 @@ +module Hasql.Pool.TimeExtras.IO where + +import Hasql.Pool.Prelude +import Hasql.Pool.TimeExtras.Conversions + +getMillisecondsSinceEpoch :: IO Int +getMillisecondsSinceEpoch = + fmap toMilliseconds getSystemTime diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -1,35 +1,39 @@ module Main where +import Control.Concurrent.Async (race) +import qualified Data.ByteString.Char8 as B8 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 qualified System.Environment import Test.Hspec import Prelude -main = hspec $ do - describe "" $ do +main = do + connectionSettings <- getConnectionSettings + hspec . describe "" $ do it "Releases a spot in the pool when there is a query error" $ do - pool <- acquire 1 connectionSettings + pool <- acquire 1 Nothing connectionSettings use pool badQuerySession `shouldNotReturn` (Right ()) use pool selectOneSession `shouldReturn` (Right 1) it "Simulation of connection error works" $ do - pool <- acquire 3 connectionSettings + pool <- acquire 3 Nothing 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 connectionSettings + pool <- acquire 3 Nothing 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 connectionSettings + pool <- acquire 3 Nothing connectionSettings res <- use pool $ selectOneSession res <- use pool $ selectOneSession res <- use pool $ selectOneSession @@ -37,17 +41,75 @@ res <- use pool $ selectOneSession shouldSatisfy res $ isRight it "Connection gets returned to the pool after non-connection error" $ do - pool <- acquire 3 connectionSettings + pool <- acquire 3 Nothing connectionSettings res <- use pool $ badQuerySession res <- use pool $ badQuerySession res <- use pool $ badQuerySession res <- use pool $ badQuerySession res <- use pool $ selectOneSession shouldSatisfy res $ isRight + it "The pool remains usable after release" $ do + pool <- acquire 1 Nothing connectionSettings + res <- use pool $ selectOneSession + release pool + res <- use pool $ selectOneSession + shouldSatisfy res $ isRight + it "Getting and setting session variables works" $ do + pool <- acquire 1 Nothing connectionSettings + res <- use pool $ getSettingSession "testing.foo" + res `shouldBe` Right Nothing + res <- use pool $ do + setSettingSession "testing.foo" "hello world" + getSettingSession "testing.foo" + res `shouldBe` Right (Just "hello world") + it "Session variables stay set when a connection gets reused" $ do + pool <- acquire 1 Nothing connectionSettings + res <- use pool $ setSettingSession "testing.foo" "hello world" + res `shouldBe` Right () + res2 <- use pool $ getSettingSession "testing.foo" + res2 `shouldBe` Right (Just "hello world") + it "Releasing the pool resets session variables" $ do + pool <- acquire 1 Nothing connectionSettings + res <- use pool $ setSettingSession "testing.foo" "hello world" + res `shouldBe` Right () + release pool + res <- use pool $ getSettingSession "testing.foo" + res `shouldBe` Right Nothing + it "Times out connection acquisition" $ do + pool <- acquire 1 (Just 1000) connectionSettings -- 1ms timeout + sleeping <- newEmptyMVar + t0 <- getCurrentTime + res <- + race + ( use pool $ + liftIO $ do + putMVar sleeping () + threadDelay 1000000 -- 1s + ) + ( do + takeMVar sleeping + use pool $ selectOneSession + ) + t1 <- getCurrentTime + res `shouldBe` Right (Left AcquisitionTimeoutUsageError) + diffUTCTime t1 t0 `shouldSatisfy` (< 0.5) -- 0.5s -connectionSettings :: Connection.Settings -connectionSettings = - "host=localhost port=5432 user=postgres dbname=postgres" +getConnectionSettings :: IO Connection.Settings +getConnectionSettings = + B8.unwords . catMaybes + <$> sequence + [ setting "host" $ defaultEnv "POSTGRES_HOST" "localhost", + setting "port" $ defaultEnv "POSTGRES_PORT" "5432", + setting "user" $ defaultEnv "POSTGRES_USER" "postgres", + setting "password" $ maybeEnv "POSTGRES_PASSWORD", + setting "dbname" $ defaultEnv "POSTGRES_DBNAME" "postgres" + ] + where + maybeEnv env = fmap B8.pack <$> System.Environment.lookupEnv env + defaultEnv env val = Just . fromMaybe val <$> maybeEnv env + setting label getEnv = do + val <- getEnv + return $ (\v -> label <> "=" <> v) <$> val selectOneSession :: Session.Session Int64 selectOneSession = @@ -66,3 +128,20 @@ closeConnSession = do conn <- ask liftIO $ Connection.release conn + +setSettingSession :: Text -> Text -> Session.Session () +setSettingSession name value = do + Session.statement (name, value) statement + where + statement = Statement.Statement "SELECT set_config($1, $2, false)" encoder Decoders.noResult True + encoder = + contramap fst (Encoders.param (Encoders.nonNullable Encoders.text)) + <> contramap snd (Encoders.param (Encoders.nonNullable Encoders.text)) + +getSettingSession :: Text -> Session.Session (Maybe Text) +getSettingSession name = do + Session.statement name statement + where + statement = Statement.Statement "SELECT current_setting($1, true)" encoder decoder True + encoder = Encoders.param (Encoders.nonNullable Encoders.text) + decoder = Decoders.singleRow (Decoders.column (Decoders.nullable Decoders.text))