hasql-pool 0.6.0.1 → 0.7
raw patch · 4 files changed
+85/−179 lines, 4 filesdep ~hasqlPVP ok
version bump matches the API change (PVP)
Dependency ranges changed: hasql
API changes (from Hackage documentation)
- Hasql.Pool: type Settings = (Int, Int, Settings)
- Hasql.Pool: acquire :: Settings -> IO Pool
+ Hasql.Pool: acquire :: Int -> Settings -> IO Pool
Files
- CHANGELOG.md +8/−0
- hasql-pool.cabal +2/−2
- library/Hasql/Pool.hs +69/−171
- test/Main.hs +6/−6
CHANGELOG.md view
@@ -1,3 +1,11 @@+# 0.7++Simplified the implementation a lot by removing the notion of timeout.++Breaking:+- Removed the `Settings` type+- Changed the signature of `acquire`+ # 0.6 Moved away from "resource-pool" and fixed the handling of lost connections.
hasql-pool.cabal view
@@ -1,7 +1,7 @@ name: hasql-pool version:- 0.6.0.1+ 0.7 category: Hasql, Database, PostgreSQL synopsis:@@ -49,7 +49,7 @@ Hasql.Pool.Prelude build-depends: base >=4.11 && <5,- hasql >=1.6.0.1 && <1.7,+ hasql >=1.3 && <1.6, stm >=2.5 && <3, time >=1.5 && <2, transformers >=0.5 && <0.7
library/Hasql/Pool.hs view
@@ -1,6 +1,5 @@ module Hasql.Pool ( Pool,- Settings (..), acquire, release, UsageError (..),@@ -16,186 +15,85 @@ -- | -- 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+data Pool = Pool+ { -- | Connection settings.+ poolConnectionSettings :: Connection.Settings,+ -- | Avail connections.+ poolConnectionQueue :: TQueue Connection,+ -- | Capacity.+ poolCapacity :: TVar Int,+ -- | Alive.+ poolAlive :: TVar Bool } -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 in milliseconds for which the unused connections are kept open.------ * Connection settings.-type 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) =- do- establishedQueue <- newTQueueIO- slotsAvailVar <- newTVarIO size- aliveVar <- newTVarIO (size > 0)- forkIO $ loopCollectingGarbage timeout establishedQueue slotsAvailVar aliveVar- return (Pool connectionSettings establishedQueue slotsAvailVar aliveVar)+-- | Given the pool-size and connection settings create a connection-pool.+acquire :: Int -> Connection.Settings -> IO Pool+acquire poolSize connectionSettings = do+ Pool connectionSettings+ <$> newTQueueIO+ <*> newTVarIO poolSize+ <*> newTVarIO True -- |--- Release the connection-pool.+-- Release all the connections in the pool. release :: Pool -> IO ()-release (Pool _ _ _ aliveVar) =- atomically (writeTVar aliveVar False)+release Pool {..} = do+ connections <- atomically $ do+ writeTVar poolAlive False+ flushTQueue poolConnectionQueue+ forM_ connections Connection.release -- | -- A union over the connection establishment error and the session error. 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+ = 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. If the session fails--- with 'Session.ClientError' the connection gets reestablished.+-- |+-- 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- 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))+use Pool {..} sess =+ 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 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+ onNewConn = do+ connRes <- Connection.acquire poolConnectionSettings+ case connRes of+ Left connErr -> do+ atomically $ modifyTVar' poolCapacity succ+ return $ Left $ ConnectionUsageError connErr+ Right conn -> onConn conn+ onConn conn = do+ sessRes <- Session.run sess conn+ case sessRes of+ Left err -> case err of+ Session.QueryError _ _ (Session.ClientError _) -> do+ atomically $ modifyTVar' poolCapacity succ+ return $ Left $ SessionUsageError err+ _ -> do+ returnConn+ return $ Left $ SessionUsageError err+ Right res -> do+ returnConn+ return $ Right res+ where+ returnConn = do+ atomically $ do+ alive <- readTVar poolAlive+ when alive $ writeTQueue poolConnectionQueue conn
test/Main.hs view
@@ -12,24 +12,24 @@ main = hspec $ do describe "" $ do it "Releases a spot in the pool when there is a query error" $ do- pool <- acquire (1, 1, connectionSettings)+ pool <- acquire 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)+ pool <- acquire 3 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)+ pool <- acquire 3 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)+ pool <- acquire 3 connectionSettings res <- use pool $ selectOneSession res <- use pool $ selectOneSession res <- use pool $ selectOneSession@@ -37,7 +37,7 @@ res <- use pool $ selectOneSession shouldSatisfy res $ isRight it "Connection gets returned to the pool after non-connection error" $ do- pool <- acquire (3, 1, connectionSettings)+ pool <- acquire 3 connectionSettings res <- use pool $ badQuerySession res <- use pool $ badQuerySession res <- use pool $ badQuerySession@@ -60,7 +60,7 @@ badQuerySession = Session.statement () statement where- statement = Statement.Statement "zzz" Encoders.noParams Decoders.noResult True+ statement = Statement.Statement "" Encoders.noParams Decoders.noResult True closeConnSession :: Session.Session () closeConnSession = do