hpqtypes 1.12.0.0 → 1.13.0.0
raw patch · 15 files changed
+599/−380 lines, 15 filesdep −semigroupsdep −unordered-containersdep ~base
Dependencies removed: semigroups, unordered-containers
Dependency ranges changed: base
Files
- CHANGELOG.md +6/−0
- README.md +1/−2
- hpqtypes.cabal +5/−7
- src/Database/PostgreSQL/PQTypes.hs +1/−0
- src/Database/PostgreSQL/PQTypes/Class.hs +25/−18
- src/Database/PostgreSQL/PQTypes/Internal/BackendPid.hs +4/−0
- src/Database/PostgreSQL/PQTypes/Internal/Connection.hs +120/−145
- src/Database/PostgreSQL/PQTypes/Internal/Monad.hs +25/−40
- src/Database/PostgreSQL/PQTypes/Internal/Notification.hsc +16/−18
- src/Database/PostgreSQL/PQTypes/Internal/State.hs +244/−15
- src/Database/PostgreSQL/PQTypes/Internal/Utils.hs +8/−7
- src/Database/PostgreSQL/PQTypes/Transaction.hs +58/−93
- src/Database/PostgreSQL/PQTypes/Transaction/Settings.hs +32/−22
- src/Database/PostgreSQL/PQTypes/Utils.hs +1/−2
- test/Main.hs +53/−11
CHANGELOG.md view
@@ -1,3 +1,9 @@+# hpqtypes-1.13.0.0 (2025-11-26)+* Drop support for GHC < 9.2.+* Include time spent executing queries in `ConnectionStats`.+* Add `initialConnectionStats`.+* Introduce on-demand connection acquisition mode.+ # hpqtypes-1.12.0.0 (2024-03-18) * Drop support for GHC 8.8. * Attach `CallStack` and `BackendPid` to `DBException`.
README.md view
@@ -1,8 +1,7 @@ # hpqtypes -[](https://github.com/scrive/hpqtypes/actions?query=branch%3Amaster)+[](https://github.com/scrive/hpqtypes/actions/workflows/haskell-ci.yml) [](https://hackage.haskell.org/package/hpqtypes)-[](https://packdeps.haskellers.com/feed?needle=andrzej@rybczak.net) [](https://www.stackage.org/lts/package/hpqtypes) [](https://www.stackage.org/nightly/package/hpqtypes)
hpqtypes.cabal view
@@ -1,7 +1,7 @@ cabal-version: 3.0 build-type: Simple name: hpqtypes-version: 1.12.0.0+version: 1.13.0.0 synopsis: Haskell bindings to libpqtypes description: Efficient and easy-to-use bindings to (slightly modified)@@ -21,7 +21,7 @@ maintainer: Andrzej Rybczak <andrzej@rybczak.net> copyright: Scrive AB category: Database-tested-with: GHC == { 8.10.7, 9.0.2, 9.2.8, 9.4.8, 9.6.4, 9.8.2 }+tested-with: GHC == { 9.2.8, 9.4.8, 9.6.7, 9.8.4, 9.10.3, 9.12.2, 9.14.1 } extra-source-files: README.md , CHANGELOG.md@@ -93,12 +93,11 @@ , Database.PostgreSQL.PQTypes.Internal.C.Interface , Database.PostgreSQL.PQTypes.Internal.C.Get - build-depends: base >= 4.14 && < 5+ build-depends: base >= 4.16 && < 5 , text >= 0.11 , aeson >= 1.0 , async >= 2.1.1.1 , bytestring >= 0.9- , semigroups >= 0.16 , time >= 1.4 , vector >= 0.10 , transformers-base >= 0.4@@ -117,7 +116,7 @@ hs-source-dirs: src - ghc-options: -Wall -Wprepositive-qualified-module+ ghc-options: -Wall -Werror=prepositive-qualified-module include-dirs: libpqtypes/src @@ -173,7 +172,7 @@ test-suite hpqtypes-tests type: exitcode-stdio-1.0- ghc-options: -Wall -Wprepositive-qualified-module -threaded+ ghc-options: -Wall -Werror=prepositive-qualified-module -threaded hs-source-dirs: test main-is: Main.hs@@ -206,7 +205,6 @@ , text-show , time >= 1.4 , transformers-base >= 0.4- , unordered-containers , vector , uuid-types
src/Database/PostgreSQL/PQTypes.hs view
@@ -5,6 +5,7 @@ -- * Connection Connection , ConnectionStats (..)+ , initialConnectionStats , ConnectionSettings (..) , defaultConnectionSettings , ConnectionSourceM
src/Database/PostgreSQL/PQTypes/Class.hs view
@@ -30,18 +30,16 @@ -- given name. runPreparedQuery :: (HasCallStack, IsSQL sql) => QueryName -> sql -> m Int - -- | Get last SQL query that was executed.- getLastQuery :: m SomeSQL+ -- | Get last SQL query that was executed and ID of the server process+ -- attached to the session that executed it.+ getLastQuery :: m (BackendPid, SomeSQL) -- | Subsequent queries in the callback do not alter the result of -- 'getLastQuery'. withFrozenLastQuery :: m a -> m a - -- | Get ID of the server process attached to the current session.- getBackendPid :: m BackendPid- -- | Get current connection statistics.- getConnectionStats :: HasCallStack => m ConnectionStats+ getConnectionStats :: m ConnectionStats -- | Get current query result. getQueryResult :: FromRow row => m (Maybe (QueryResult row))@@ -49,14 +47,23 @@ -- | Clear current query result. clearQueryResult :: m () - -- | Get current transaction settings.- getTransactionSettings :: m TransactionSettings+ -- | Get current connection acquisition mode.+ getConnectionAcquisitionMode :: HasCallStack => m ConnectionAcquisitionMode - -- | Set transaction settings to supplied ones. Note that it- -- won't change any properties of currently running transaction,- -- only the subsequent ones.- setTransactionSettings :: TransactionSettings -> m ()+ -- | Acquire and hold a connection with a given isolation level and+ -- permissions.+ --+ -- If the connection is already held, a check is performed if the isolation+ -- level and permissions are the same as the ones currently in place. If so,+ -- nothing happens, otherwise an error is thrown.+ acquireAndHoldConnection :: HasCallStack => IsolationLevel -> Permissions -> m () + -- | Unsafely switch to the 'AcquireOnDemand' mode.+ --+ -- This function is unsafe because if a connection is already held, the+ -- transaction in progress is commited, so atomicity guarantee is lost.+ unsafeAcquireOnDemandConnection :: HasCallStack => m ()+ -- | Attempt to receive a notification from the server. This -- function waits until a notification arrives or specified -- number of microseconds has passed. If a negative number@@ -72,7 +79,7 @@ -- for further info), therefore calling this function within -- a transaction block will return 'Just' only if notifications -- were received before the transaction began.- getNotification :: Int -> m (Maybe Notification)+ getNotification :: HasCallStack => Int -> m (Maybe Notification) -- | Execute supplied monadic action with new connection -- using current connection source and transaction settings.@@ -80,7 +87,7 @@ -- Particularly useful when you want to spawn a new thread, but -- do not want the connection in child thread to be shared with -- the parent one.- withNewConnection :: m a -> m a+ withNewConnection :: HasCallStack => m a -> m a -- | Generic, overlappable instance. instance@@ -97,11 +104,11 @@ runPreparedQuery name = withFrozenCallStack $ lift . runPreparedQuery name getLastQuery = lift getLastQuery withFrozenLastQuery m = controlT $ \run -> withFrozenLastQuery (run m)- getBackendPid = lift getBackendPid- getConnectionStats = withFrozenCallStack $ lift getConnectionStats+ getConnectionStats = lift getConnectionStats getQueryResult = lift getQueryResult clearQueryResult = lift clearQueryResult- getTransactionSettings = lift getTransactionSettings- setTransactionSettings = lift . setTransactionSettings+ getConnectionAcquisitionMode = lift getConnectionAcquisitionMode+ acquireAndHoldConnection isoLevel = lift . acquireAndHoldConnection isoLevel+ unsafeAcquireOnDemandConnection = lift unsafeAcquireOnDemandConnection getNotification = lift . getNotification withNewConnection m = controlT $ \run -> withNewConnection (run m)
src/Database/PostgreSQL/PQTypes/Internal/BackendPid.hs view
@@ -1,7 +1,11 @@ module Database.PostgreSQL.PQTypes.Internal.BackendPid ( BackendPid (..)+ , noBackendPid ) where -- | Process ID of the server process attached to the current session. newtype BackendPid = BackendPid Int deriving newtype (Eq, Ord, Show)++noBackendPid :: BackendPid+noBackendPid = BackendPid 0
src/Database/PostgreSQL/PQTypes/Internal/Connection.hs view
@@ -1,13 +1,12 @@ module Database.PostgreSQL.PQTypes.Internal.Connection ( -- * Connection Connection (..)- , getBackendPidIO- , ConnectionData (..)- , withConnectionData , ConnectionStats (..)+ , initialConnectionStats , ConnectionSettings (..) , defaultConnectionSettings , ConnectionSourceM (..)+ , InternalConnectionSource (..) , ConnectionSource (..) , simpleSource , poolSource@@ -42,6 +41,7 @@ import Foreign.C.String import Foreign.ForeignPtr import Foreign.Ptr+import GHC.Clock (getMonotonicTime) import GHC.Conc (closeFdWith) import GHC.Stack @@ -95,17 +95,20 @@ -- ^ Number of values fetched from the database. , statsParams :: !Int -- ^ Number of parameters sent to the database.+ , statsTime :: !Double+ -- ^ Time spent executing queries (in seconds). } deriving (Eq, Ord, Show) -- | Initial connection statistics.-initialStats :: ConnectionStats-initialStats =+initialConnectionStats :: ConnectionStats+initialConnectionStats = ConnectionStats { statsQueries = 0 , statsRows = 0 , statsValues = 0 , statsParams = 0+ , statsTime = 0 } -- | Representation of a connection object.@@ -116,58 +119,41 @@ -- executing an SQL query). -- -- See https://gitlab.haskell.org/ghc/ghc/-/issues/10975 for more info.-data ConnectionData = ConnectionData- { cdPtr :: !(Ptr PGconn)+data Connection = Connection+ { connPtr :: !(Ptr PGconn) -- ^ Pointer to connection object.- , cdBackendPid :: !BackendPid+ , connBackendPid :: !BackendPid -- ^ Process ID of the server process attached to the current session.- , cdStats :: !ConnectionStats- -- ^ Statistics associated with the connection.- , cdPreparedQueries :: !(IORef (S.Set T.Text))+ , connPreparedQueries :: !(IORef (S.Set T.Text)) -- ^ A set of named prepared statements of the connection. } --- | Wrapper for hiding representation of a connection object.-newtype Connection = Connection- { unConnection :: MVar (Maybe ConnectionData)+data InternalConnectionSource m cdata = InternalConnectionSource+ { takeConnection :: !(m (Connection, cdata))+ , putConnection :: !(forall r. (Connection, cdata) -> ExitCase r -> m ()) } -getBackendPidIO :: Connection -> IO BackendPid-getBackendPidIO conn = do- withConnectionData conn "getBackendPidIO" $ \cd -> do- pure (cd, cdBackendPid cd)--withConnectionData- :: Connection- -> String- -> (ConnectionData -> IO (ConnectionData, r))- -> IO r-withConnectionData (Connection mvc) fname f = modifyMVar mvc $ \case- Nothing -> hpqTypesError $ fname ++ ": no connection"- Just cd -> do- (cd', r) <- f cd- cd' `seq` pure (Just cd', r)- -- | Database connection supplier.-newtype ConnectionSourceM m = ConnectionSourceM- { withConnection :: forall r. (Connection -> m r) -> m r- }+data ConnectionSourceM m+ = forall cdata. ConnectionSourceM !(InternalConnectionSource m cdata) -- | Wrapper for a polymorphic connection source. newtype ConnectionSource (cs :: [(Type -> Type) -> Constraint]) = ConnectionSource { unConnectionSource :: forall m. MkConstraint m cs => ConnectionSourceM m } --- | Default connection supplier. It establishes new--- database connection each time 'withConnection' is called.+-- | Default connection supplier. It establishes new database connection each+-- time 'withConnection' is called. simpleSource :: ConnectionSettings -> ConnectionSource [MonadBase IO, MonadMask] simpleSource cs = ConnectionSource $ ConnectionSourceM- { withConnection = bracket (liftBase $ connect cs) (liftBase . disconnect)- }+ InternalConnectionSource+ { takeConnection = (,()) <$> liftBase (connect cs)+ , putConnection = \(conn, ()) _ -> liftBase $ disconnect conn+ } -- | Pooled source. It uses striped pool from @resource-pool@ package to cache -- established connections and reuse them.@@ -185,23 +171,12 @@ where sourceM pool = ConnectionSourceM- { withConnection = doWithConnection pool . (clearStats >=>)- }-- doWithConnection pool m =- fst- <$> generalBracket- (liftBase $ takeResource pool)- ( \(resource, local) -> \case+ InternalConnectionSource+ { takeConnection = liftBase $ takeResource pool+ , putConnection = \(resource, local) -> \case ExitCaseSuccess _ -> liftBase $ putResource local resource _ -> liftBase $ destroyResource pool local resource- )- (\(resource, _) -> m resource)-- clearStats conn@(Connection mv) = do- liftBase . modifyMVar_ mv $ \mconn ->- pure $ (\cd -> cd {cdStats = initialStats}) <$> mconn- pure conn+ } ---------------------------------------- @@ -225,29 +200,22 @@ registerComposites connPtr csComposites conn <- do preparedQueries <- newIORef S.empty- fmap Connection . newMVar $- Just- ConnectionData- { cdPtr = connPtr- , cdBackendPid = noBackendPid- , cdStats = initialStats- , cdPreparedQueries = preparedQueries- }+ pure+ Connection+ { connPtr = connPtr+ , connBackendPid = noBackendPid+ , connPreparedQueries = preparedQueries+ } F.forM_ csRole $ \role -> runQueryIO conn $ "SET ROLE " <> role let selectPid = "SELECT pg_backend_pid()" :: RawSQL ()- (_, res) <- runQueryIO conn selectPid+ (_, res, _) <- runQueryIO conn selectPid case F.toList $ mkQueryResult @(Identity Int32) selectPid noBackendPid res of- [pid] -> withConnectionData conn fname $ \cd -> do- pure (cd {cdBackendPid = BackendPid $ fromIntegral pid}, ())+ [pid] -> pure $ conn {connBackendPid = BackendPid $ fromIntegral pid} pids -> do let err = HPQTypesError $ "unexpected backend pid: " ++ show pids rethrowWithContext selectPid noBackendPid $ toException err-- pure conn where- noBackendPid = BackendPid 0- fname = "connect" openConnection :: (forall r. IO r -> IO r) -> CString -> IO (Ptr PGconn)@@ -294,21 +262,16 @@ -- | Low-level function for disconnecting from the database. Useful if one wants -- to implement custom connection source. disconnect :: Connection -> IO ()-disconnect (Connection mvconn) = modifyMVar_ mvconn $ \mconn -> do- case mconn of- Just cd -> do- let conn = cdPtr cd- -- This covers the case when a connection is closed while other Haskell- -- threads are using GHC's IO manager to wait on the descriptor. This is- -- commonly the case with asynchronous notifications, for example. Since- -- libpq is responsible for opening and closing the file descriptor, GHC's- -- IO manager needs to be informed that the file descriptor has been- -- closed. The IO manager will then raise an exception in those threads.- c_PQsocket conn >>= \case- -1 -> c_PQfinish conn -- can happen if the connection is bad/lost- fd -> closeFdWith (\_ -> c_PQfinish conn) fd- Nothing -> E.throwIO (HPQTypesError "disconnect: no connection (shouldn't happen)")- pure Nothing+disconnect Connection {..} = do+ -- This covers the case when a connection is closed while other Haskell+ -- threads are using GHC's IO manager to wait on the descriptor. This is+ -- commonly the case with asynchronous notifications, for example. Since libpq+ -- is responsible for opening and closing the file descriptor, GHC's IO+ -- manager needs to be informed that the file descriptor has been closed. The+ -- IO manager will then raise an exception in those threads.+ c_PQsocket connPtr >>= \case+ -1 -> c_PQfinish connPtr -- can happen if the connection is bad/lost+ fd -> closeFdWith (\_ -> c_PQfinish connPtr) fd ---------------------------------------- -- Query running@@ -318,14 +281,14 @@ :: (HasCallStack, IsSQL sql) => Connection -> sql- -> IO (Int, ForeignPtr PGresult)-runQueryIO conn sql = do- runQueryImpl "runQueryIO" conn sql $ \ConnectionData {..} -> do- let allocParam = ParamAllocator $ withPGparam cdPtr+ -> IO (Int, ForeignPtr PGresult, ConnectionStats -> ConnectionStats)+runQueryIO conn@Connection {..} sql = do+ runQueryImpl conn sql $ do+ let allocParam = ParamAllocator $ withPGparam connPtr withSQL sql allocParam $ \param query -> (,) <$> (fromIntegral <$> c_PQparamCount param)- <*> c_PQparamExec cdPtr nullPtr param query c_RESULT_BINARY+ <*> c_PQparamExec connPtr nullPtr param query c_RESULT_BINARY -- | Name of a prepared query. newtype QueryName = QueryName T.Text@@ -337,91 +300,103 @@ => Connection -> QueryName -> sql- -> IO (Int, ForeignPtr PGresult)-runPreparedQueryIO conn (QueryName queryName) sql = do- runQueryImpl "runPreparedQueryIO" conn sql $ \ConnectionData {..} -> do+ -> IO (Int, ForeignPtr PGresult, ConnectionStats -> ConnectionStats)+runPreparedQueryIO conn@Connection {..} (QueryName queryName) sql = do+ runQueryImpl conn sql $ do when (T.null queryName) $ do E.throwIO DBException { dbeQueryContext = sql- , dbeBackendPid = cdBackendPid+ , dbeBackendPid = connBackendPid , dbeError = HPQTypesError "runPreparedQueryIO: unnamed prepared query is not supported" , dbeCallStack = callStack }- let allocParam = ParamAllocator $ withPGparam cdPtr+ let allocParam = ParamAllocator $ withPGparam connPtr withSQL sql allocParam $ \param query -> do- preparedQueries <- readIORef cdPreparedQueries+ preparedQueries <- readIORef connPreparedQueries BS.useAsCString (T.encodeUtf8 queryName) $ \cname -> do when (queryName `S.notMember` preparedQueries) . E.mask_ $ do -- Mask asynchronous exceptions, because if preparation of the query -- succeeds, we need to reflect that fact in cdPreparedQueries since -- you can't prepare a query with the same name more than once.- res <- c_PQparamPrepare cdPtr nullPtr param cname query- void . withForeignPtr res $ verifyResult sql cdBackendPid cdPtr- modifyIORef' cdPreparedQueries $ S.insert queryName+ res <- c_PQparamPrepare connPtr nullPtr param cname query+ void . withForeignPtr res $ verifyResult sql connBackendPid connPtr+ modifyIORef' connPreparedQueries $ S.insert queryName (,) <$> (fromIntegral <$> c_PQparamCount param)- <*> c_PQparamExecPrepared cdPtr nullPtr param cname c_RESULT_BINARY+ <*> c_PQparamExecPrepared connPtr nullPtr param cname c_RESULT_BINARY -- | Shared implementation of 'runQueryIO' and 'runPreparedQueryIO'. runQueryImpl :: (HasCallStack, IsSQL sql)- => String- -> Connection+ => Connection -> sql- -> (ConnectionData -> IO (Int, ForeignPtr PGresult)) -> IO (Int, ForeignPtr PGresult)-runQueryImpl fname conn sql execSql = do- withConnDo $ \cd@ConnectionData {..} -> E.mask $ \restore -> do- -- While the query runs, the current thread will not be able to receive- -- asynchronous exceptions. This prevents clients of the library from- -- interrupting execution of the query. To remedy that we spawn a separate- -- thread for the query execution and while we wait for its completion, we- -- are able to receive asynchronous exceptions (assuming that threaded GHC- -- runtime system is used) and react appropriately.- queryRunner <- async . restore $ do- (paramCount, res) <- execSql cd- affected <- withForeignPtr res $ verifyResult sql cdBackendPid cdPtr- stats' <- case affected of+ -> IO (Int, ForeignPtr PGresult, ConnectionStats -> ConnectionStats)+runQueryImpl Connection {..} sql execSql = do+ E.getMaskingState >>= \case+ E.MaskedUninterruptible -> do+ -- If asynchronous exceptions are already hard-masked, skip spawning a+ -- separate worker thread and the interruption logic as it won't do+ -- anything at this point.+ doRunQuery+ _ -> E.uninterruptibleMask $ \restore -> do+ -- While the query runs, the current thread will not be able to receive+ -- asynchronous exceptions. This prevents clients of the library from+ -- interrupting execution of the query. To remedy that we spawn a separate+ -- thread for the query execution and while we wait for its completion, we+ -- are able to receive asynchronous exceptions (assuming that threaded GHC+ -- runtime system is used) and react appropriately.+ queryRunner <- asyncWithUnmask $ \unmask -> do+ -- Unconditionally unmask asynchronous exceptions here so that 'cancel'+ -- operation potentially invoked below works as expected.+ unmask doRunQuery+ -- If we receive an exception while waiting for the execution to complete,+ -- we need to send a request to PostgreSQL for query cancellation and wait+ -- for the query runner thread to terminate. It is paramount we make the+ -- exception handler uninterruptible as we can't exit from the main block+ -- until the query runner thread has terminated.+ E.onException (restore $ wait queryRunner) $ do+ c_PQcancel connPtr >>= \case+ -- If query cancellation request was successfully processed, there is+ -- nothing else to do apart from waiting for the runner to terminate.+ Nothing -> cancel queryRunner+ -- Otherwise we check what happened with the runner. If it already+ -- finished we're fine, just ignore the result. If it didn't, something+ -- weird is going on. Maybe the cancellation request went through when+ -- the thread wasn't making a request to the server? In any case, try to+ -- cancel again and wait for the thread to terminate.+ Just _ ->+ poll queryRunner >>= \case+ Just _ -> pure ()+ Nothing -> do+ void $ c_PQcancel connPtr+ cancel queryRunner+ where+ doRunQuery = do+ t1 <- getMonotonicTime+ (paramCount, res) <- execSql+ t2 <- getMonotonicTime+ affected <- withForeignPtr res $ verifyResult sql connBackendPid connPtr+ updateStats <- case affected of Left _ ->- pure- cdStats- { statsQueries = statsQueries cdStats + 1- , statsParams = statsParams cdStats + paramCount+ pure $ \stats ->+ stats+ { statsQueries = statsQueries stats + 1+ , statsParams = statsParams stats + paramCount+ , statsTime = statsTime stats + (t2 - t1) } Right rows -> do columns <- fromIntegral <$> withForeignPtr res c_PQnfields- pure+ pure $ \stats -> ConnectionStats- { statsQueries = statsQueries cdStats + 1- , statsRows = statsRows cdStats + rows- , statsValues = statsValues cdStats + (rows * columns)- , statsParams = statsParams cdStats + paramCount+ { statsQueries = statsQueries stats + 1+ , statsRows = statsRows stats + rows+ , statsValues = statsValues stats + (rows * columns)+ , statsParams = statsParams stats + paramCount+ , statsTime = statsTime stats + (t2 - t1) }- pure (cd {cdStats = stats'}, (either id id affected, res))- -- If we receive an exception while waiting for the execution to complete,- -- we need to send a request to PostgreSQL for query cancellation and wait- -- for the query runner thread to terminate. It is paramount we make the- -- exception handler uninterruptible as we can't exit from the main block- -- until the query runner thread has terminated.- E.onException (restore $ wait queryRunner) . E.uninterruptibleMask_ $ do- c_PQcancel cdPtr >>= \case- -- If query cancellation request was successfully processed, there is- -- nothing else to do apart from waiting for the runner to terminate.- Nothing -> cancel queryRunner- -- Otherwise we check what happened with the runner. If it already- -- finished we're fine, just ignore the result. If it didn't, something- -- weird is going on. Maybe the cancellation request went through when- -- the thread wasn't making a request to the server? In any case, try to- -- cancel again and wait for the thread to terminate.- Just _ ->- poll queryRunner >>= \case- Just _ -> pure ()- Nothing -> do- void $ c_PQcancel cdPtr- cancel queryRunner- where- withConnDo = withConnectionData conn fname+ pure (either id id affected, res, updateStats) verifyResult :: (HasCallStack, IsSQL sql)
src/Database/PostgreSQL/PQTypes/Internal/Monad.hs view
@@ -6,7 +6,6 @@ ) where import Control.Applicative-import Control.Concurrent.MVar import Control.Monad import Control.Monad.Base import Control.Monad.Catch@@ -21,14 +20,9 @@ import Database.PostgreSQL.PQTypes.Class import Database.PostgreSQL.PQTypes.Internal.Connection-import Database.PostgreSQL.PQTypes.Internal.Error import Database.PostgreSQL.PQTypes.Internal.Notification import Database.PostgreSQL.PQTypes.Internal.State-import Database.PostgreSQL.PQTypes.SQL-import Database.PostgreSQL.PQTypes.SQL.Class-import Database.PostgreSQL.PQTypes.Transaction import Database.PostgreSQL.PQTypes.Transaction.Settings-import Database.PostgreSQL.PQTypes.Utils type InnerDBT m = StateT (DBState m) @@ -47,22 +41,8 @@ -> TransactionSettings -> DBT m a -> m a-runDBT cs ts m = withConnection cs $ \conn -> do- evalStateT action $- DBState- { dbConnection = conn- , dbConnectionSource = cs- , dbTransactionSettings = ts- , dbLastQuery = SomeSQL (mempty :: SQL)- , dbRecordLastQuery = True- , dbQueryResult = Nothing- }- where- action =- unDBT $- if tsAutoTransaction ts- then withTransaction' (ts {tsAutoTransaction = False}) m- else m+runDBT cs ts action = withConnectionData cs ts $ \cd -> do+ evalStateT (unDBT action) $ mkDBState cd ts -- | Transform the underlying monad. mapDBT@@ -76,11 +56,11 @@ instance (m ~ n, MonadBase IO m, MonadMask m) => MonadDB (DBT_ m n) where runQuery sql = withFrozenCallStack $ do- DBT . StateT $ \st -> liftBase $ do- updateStateWith st sql =<< runQueryIO (dbConnection st) sql+ DBT . StateT $ \st -> withConnection (dbConnectionData st) $ \conn -> do+ liftBase $ updateStateWith conn st sql =<< runQueryIO conn sql runPreparedQuery name sql = withFrozenCallStack $ do- DBT . StateT $ \st -> liftBase $ do- updateStateWith st sql =<< runPreparedQueryIO (dbConnection st) name sql+ DBT . StateT $ \st -> withConnection (dbConnectionData st) $ \conn -> do+ liftBase $ updateStateWith conn st sql =<< runPreparedQueryIO conn name sql getLastQuery = DBT . gets $ dbLastQuery @@ -89,27 +69,32 @@ (x, st'') <- runStateT (unDBT callback) st' pure (x, st'' {dbRecordLastQuery = dbRecordLastQuery st}) - getBackendPid = DBT . StateT $ \st -> do- (,st) <$> liftBase (getBackendPidIO $ dbConnection st)+ getConnectionStats = DBT $ gets dbConnectionStats - getConnectionStats = withFrozenCallStack $ do- mconn <- DBT $ liftBase . readMVar =<< gets (unConnection . dbConnection)- case mconn of- Nothing -> throwDB $ HPQTypesError "getConnectionStats: no connection"- Just cd -> pure $ cdStats cd+ getQueryResult = DBT $ gets dbQueryResult+ clearQueryResult = DBT . modify' $ \st -> st {dbQueryResult = Nothing} - getQueryResult = DBT . gets $ \st -> dbQueryResult st- clearQueryResult = DBT . modify $ \st -> st {dbQueryResult = Nothing}+ getConnectionAcquisitionMode = DBT . StateT $ \st -> do+ (,st) <$> liftBase (getConnectionAcquisitionModeIO $ dbConnectionData st) - getTransactionSettings = DBT . gets $ dbTransactionSettings- setTransactionSettings ts = DBT . modify $ \st -> st {dbTransactionSettings = ts}+ acquireAndHoldConnection isolationLevel permissions = DBT . StateT $ \st -> do+ (,st) <$> changeAcquisitionModeTo (AcquireAndHold isolationLevel permissions) (dbConnectionData st) + unsafeAcquireOnDemandConnection = DBT . StateT $ \st -> do+ (,st) <$> changeAcquisitionModeTo AcquireOnDemand (dbConnectionData st)+ getNotification time = DBT . StateT $ \st -> do- (,st) <$> liftBase (getNotificationIO st time)+ withConnection (dbConnectionData st) $ \conn -> do+ (,st) <$> liftBase (getNotificationIO conn time) withNewConnection m = DBT . StateT $ \st -> do- let cs = dbConnectionSource st- ts = dbTransactionSettings st+ cam <- liftBase . getConnectionAcquisitionModeIO $ dbConnectionData st+ let cs = getConnectionSource $ dbConnectionData st+ ts =+ TransactionSettings+ { tsRestartPredicate = dbRestartPredicate st+ , tsConnectionAcquisitionMode = cam+ } res <- runDBT cs ts m pure (res, st)
src/Database/PostgreSQL/PQTypes/Internal/Notification.hsc view
@@ -10,6 +10,7 @@ import Data.String import Foreign.Ptr import Foreign.Storable+import GHC.Stack import System.Posix.Types import System.Timeout import Control.Exception qualified as E@@ -20,7 +21,6 @@ import Database.PostgreSQL.PQTypes.Internal.C.Interface import Database.PostgreSQL.PQTypes.Internal.C.Types import Database.PostgreSQL.PQTypes.Internal.Connection-import Database.PostgreSQL.PQTypes.Internal.State import Database.PostgreSQL.PQTypes.Internal.Utils import Database.PostgreSQL.PQTypes.SQL.Raw @@ -70,23 +70,21 @@ -- | Low-level function that waits for a notification for a given -- number of microseconds (it uses 'timeout' function internally).-getNotificationIO :: DBState m -> Int -> IO (Maybe Notification)-getNotificationIO st n = timeout n $ do- withConnectionData (dbConnection st) fname $ \cd -> fix $ \loop -> do- let conn = cdPtr cd- mmsg <- tryGet conn- case mmsg of- Just msg -> pure (cd, msg)- Nothing -> do- fd <- c_PQsocket conn- if fd == -1- then hpqTypesError $ fname ++ ": invalid file descriptor"- else do- threadWaitRead fd- res <- c_PQconsumeInput conn- when (res /= 1) $ do- throwLibPQError conn fname- loop+getNotificationIO :: HasCallStack => Connection -> Int -> IO (Maybe Notification)+getNotificationIO conn n = timeout n $ fix $ \loop -> do+ mmsg <- tryGet $ connPtr conn+ case mmsg of+ Just msg -> pure msg+ Nothing -> do+ fd <- c_PQsocket $ connPtr conn+ if fd == -1+ then hpqTypesError $ fname ++ ": invalid file descriptor"+ else do+ threadWaitRead fd+ res <- c_PQconsumeInput $ connPtr conn+ when (res /= 1) $ do+ throwLibPQError (connPtr conn) fname+ loop where fname :: String fname = "getNotificationIO"
src/Database/PostgreSQL/PQTypes/Internal/State.hs view
@@ -1,49 +1,278 @@ -- | Definition of internal DBT state. module Database.PostgreSQL.PQTypes.Internal.State- ( DBState (..)+ ( -- * ConnectionData+ ConnectionData+ , getConnectionSource+ , getConnectionAcquisitionModeIO+ , withConnectionData+ , changeAcquisitionModeTo+ , withConnection++ -- * DBState+ , DBState (..)+ , mkDBState , updateStateWith ) where +import Control.Concurrent.MVar.Lifted+import Control.Monad+import Control.Monad.Base+import Control.Monad.Catch+import Data.Function+import Data.Typeable import Foreign.ForeignPtr+import GHC.Stack +import Data.Monoid.Utils import Database.PostgreSQL.PQTypes.FromRow+import Database.PostgreSQL.PQTypes.Internal.BackendPid import Database.PostgreSQL.PQTypes.Internal.C.Types import Database.PostgreSQL.PQTypes.Internal.Connection+import Database.PostgreSQL.PQTypes.Internal.Exception import Database.PostgreSQL.PQTypes.Internal.QueryResult+import Database.PostgreSQL.PQTypes.SQL import Database.PostgreSQL.PQTypes.SQL.Class import Database.PostgreSQL.PQTypes.Transaction.Settings +data ConnectionState cdata+ = OnDemand+ | Acquired !IsolationLevel !Permissions !Connection !cdata+ | Finalized++-- Note: initConnection{State,Data} and finalizeConnection{State,Data} need to+-- be invoked inside bracket and run with asynchronous exceptions softly+-- masked. In addition, they may run queries that start/finish a+-- transaction. Running queries is a blocking (and thus interruptible)+-- operation, but if these queries are interrupted with an asynchronous+-- exception, then a connection is leaked, so they need to be run with+-- asynchronous exceptions hard masked with uninterruptibleMask.++initConnectionState+ :: MonadBase IO m+ => InternalConnectionSource m cdata+ -> ConnectionAcquisitionMode+ -> m (ConnectionState cdata)+initConnectionState ics = \case+ AcquireOnDemand -> pure OnDemand+ AcquireAndHold tsIsolationLevel tsPermissions -> do+ let initSql =+ smconcat+ [ "BEGIN"+ , case tsIsolationLevel of+ DefaultLevel -> ""+ ReadCommitted -> "ISOLATION LEVEL READ COMMITTED"+ RepeatableRead -> "ISOLATION LEVEL REPEATABLE READ"+ Serializable -> "ISOLATION LEVEL SERIALIZABLE"+ , case tsPermissions of+ DefaultPermissions -> ""+ ReadOnly -> "READ ONLY"+ ReadWrite -> "READ WRITE"+ ]+ (conn, cdata) <- takeConnection ics+ _ <- liftBase . uninterruptibleMask_ $ runQueryIO @SQL conn initSql+ pure $ Acquired tsIsolationLevel tsPermissions conn cdata++finalizeConnectionState+ :: (HasCallStack, MonadBase IO m)+ => InternalConnectionSource m cdata+ -> ExitCase r+ -> ConnectionState cdata+ -> m ()+finalizeConnectionState ics ec = \case+ OnDemand -> pure ()+ Acquired _ _ conn cdata -> do+ let finalizeSql = case ec of+ ExitCaseSuccess _ -> "COMMIT"+ _ -> "ROLLBACK"+ _ <- liftBase . uninterruptibleMask_ $ runQueryIO @SQL conn finalizeSql+ putConnection ics (conn, cdata) ec+ Finalized -> error "finalized connection"++----------------------------------------++data ConnectionData m = forall cdata. ConnectionData+ { cdConnectionSource :: !(InternalConnectionSource m cdata)+ , cdConnectionState :: !(MVar (ConnectionState cdata))+ }++getConnectionSource :: ConnectionData m -> ConnectionSourceM m+getConnectionSource ConnectionData {..} = ConnectionSourceM cdConnectionSource++getConnectionAcquisitionModeIO+ :: HasCallStack+ => ConnectionData m+ -> IO ConnectionAcquisitionMode+getConnectionAcquisitionModeIO ConnectionData {..} = do+ readMVar cdConnectionState >>= \case+ OnDemand -> pure AcquireOnDemand+ Acquired isolationLevel permissions _ _ -> do+ pure $ AcquireAndHold isolationLevel permissions+ Finalized -> error "finalized connection"++withConnectionData+ :: (HasCallStack, MonadBase IO m, MonadMask m)+ => ConnectionSourceM m+ -> TransactionSettings+ -> (ConnectionData m -> m r)+ -> m r+withConnectionData cs ts action = (`fix` 1) $ \loop n -> do+ let maybeRestart = case tsRestartPredicate ts of+ Just _ -> handleJust (expred n) $ \_ -> loop $ n + 1+ Nothing -> id+ maybeRestart+ . fmap fst+ . generalBracket (initConnectionData cs cam) finalizeConnectionData+ $ action+ where+ cam = tsConnectionAcquisitionMode ts++ expred :: Integer -> SomeException -> Maybe ()+ expred n e = do+ -- check if the predicate exists+ RestartPredicate f <- tsRestartPredicate ts+ -- cast exception to the type expected by the predicate+ err <-+ msum+ [ -- either cast the exception itself...+ fromException e+ , -- ...or extract it from DBException+ fromException e >>= \DBException {..} -> cast dbeError+ ]+ -- check if the predicate allows for the restart+ guard $ f err n++changeAcquisitionModeTo+ :: (HasCallStack, MonadBase IO m, MonadMask m)+ => ConnectionAcquisitionMode+ -> ConnectionData m+ -> m ()+changeAcquisitionModeTo cam ConnectionData {..} = do+ bracketOnError (takeMVar cdConnectionState) (putMVar cdConnectionState) $ \case+ OnDemand -> case cam of+ AcquireOnDemand -> putMVar cdConnectionState OnDemand+ _ -> mask_ $ do+ -- Need to mask, if asynchronous exception arrives between+ -- initConnectionState and putMVar, the connection leaks.+ newConnState <- initConnectionState cdConnectionSource cam+ putMVar cdConnectionState newConnState+ connState@(Acquired isolationLevel permissions _ _) -> case cam of+ AcquireOnDemand -> mask_ $ do+ -- Need to mask, if asynchronous exception arrives between+ -- finalizeConnectionState and putMVar, we end up with an invalid+ -- (finalized) connection state.+ finalizeConnectionState cdConnectionSource (ExitCaseSuccess ()) connState+ putMVar cdConnectionState OnDemand+ AcquireAndHold newIsolationLevel newPermissions -> do+ when (isolationLevel /= newIsolationLevel) $ do+ error $+ "isolation level mismatch (current: "+ ++ show isolationLevel+ ++ ", new: "+ ++ show newIsolationLevel+ ++ ")"+ when (permissions /= newPermissions) $ do+ error $+ "permissions mismatch (current: "+ ++ show permissions+ ++ ", new: "+ ++ show newPermissions+ ++ ")"+ putMVar cdConnectionState connState+ Finalized -> error "finalized connection"++withConnection+ :: (HasCallStack, MonadBase IO m, MonadMask m)+ => ConnectionData m+ -> (Connection -> m r)+ -> m r+withConnection ConnectionData {..} action = do+ bracket (takeMVar cdConnectionState) (putMVar cdConnectionState) $ \case+ OnDemand ->+ fst+ <$> generalBracket+ (takeConnection cdConnectionSource)+ (putConnection cdConnectionSource)+ ( \(conn, _cdata) ->+ bracket_+ (liftBase . uninterruptibleMask_ $ runQueryIO @SQL conn "BEGIN READ ONLY")+ (liftBase . uninterruptibleMask_ $ runQueryIO @SQL conn "ROLLBACK")+ (action conn)+ )+ Acquired _ _ conn _ -> action conn+ Finalized -> error "finalized connection"++initConnectionData+ :: MonadBase IO m+ => ConnectionSourceM m+ -> ConnectionAcquisitionMode+ -> m (ConnectionData m)+initConnectionData (ConnectionSourceM ics) cam = do+ connState <- newMVar =<< initConnectionState ics cam+ pure $+ ConnectionData+ { cdConnectionSource = ics+ , cdConnectionState = connState+ }++finalizeConnectionData+ :: (HasCallStack, MonadBase IO m, MonadMask m)+ => ConnectionData m+ -> ExitCase r+ -> m ()+finalizeConnectionData ConnectionData {..} ec = do+ (`finally` putMVar cdConnectionState Finalized) $ do+ connState <- takeMVar cdConnectionState+ finalizeConnectionState cdConnectionSource ec connState++----------------------------------------+ -- | Internal DB state. data DBState m = DBState- { dbConnection :: !Connection+ { dbConnectionData :: !(ConnectionData m) -- ^ Active connection.- , dbConnectionSource :: !(ConnectionSourceM m)- -- ^ Supplied connection source.- , dbTransactionSettings :: !TransactionSettings- -- ^ Current transaction settings.- , dbLastQuery :: !SomeSQL- -- ^ Last SQL query that was executed.+ , dbConnectionStats :: !ConnectionStats+ -- ^ Statistics associated with the session.+ , dbRestartPredicate :: !(Maybe RestartPredicate)+ -- ^ Restart predicate from initial 'TransactionSettings'.+ , dbLastQuery :: !(BackendPid, SomeSQL)+ -- ^ Last SQL query that was executed along with ID of the server process+ -- attached to the session that executed it. , dbRecordLastQuery :: !Bool -- ^ Whether running query should override 'dbLastQuery'. , dbQueryResult :: !(forall row. FromRow row => Maybe (QueryResult row)) -- ^ Current query result. } +mkDBState+ :: ConnectionData m+ -> TransactionSettings+ -> DBState m+mkDBState cd ts =+ DBState+ { dbConnectionData = cd+ , dbConnectionStats = initialConnectionStats+ , dbRestartPredicate = tsRestartPredicate ts+ , dbLastQuery = (noBackendPid, SomeSQL (mempty :: SQL))+ , dbRecordLastQuery = True+ , dbQueryResult = Nothing+ }+ updateStateWith :: IsSQL sql- => DBState m+ => Connection+ -> DBState m -> sql- -> (r, ForeignPtr PGresult)+ -> (r, ForeignPtr PGresult, ConnectionStats -> ConnectionStats) -> IO (r, DBState m)-updateStateWith st sql (r, res) = do- pid <- getBackendPidIO $ dbConnection st+updateStateWith conn st sql (r, res, updateStats) = do pure ( r , st- { dbLastQuery =+ { dbConnectionStats = updateStats $ dbConnectionStats st+ , dbLastQuery = if dbRecordLastQuery st- then SomeSQL sql+ then (connBackendPid conn, SomeSQL sql) else dbLastQuery st- , dbQueryResult = Just $ mkQueryResult sql pid res+ , dbQueryResult = Just $ mkQueryResult sql (connBackendPid conn) res } )
src/Database/PostgreSQL/PQTypes/Internal/Utils.hs view
@@ -29,6 +29,7 @@ import Foreign.Ptr import Foreign.Storable import GHC.Exts+import GHC.Stack import Database.PostgreSQL.PQTypes.Internal.C.Interface import Database.PostgreSQL.PQTypes.Internal.C.Types@@ -82,12 +83,12 @@ -- | Check return value of a function from libpqtypes -- and if it indicates an error, throw appropriate exception.-verifyPQTRes :: Ptr PGerror -> String -> CInt -> IO ()+verifyPQTRes :: HasCallStack => Ptr PGerror -> String -> CInt -> IO () verifyPQTRes err ctx 0 = throwLibPQTypesError err ctx verifyPQTRes _ _ _ = pure () -- 'alloca'-like function for managing usage of 'PGparam' object.-withPGparam :: Ptr PGconn -> (Ptr PGparam -> IO r) -> IO r+withPGparam :: HasCallStack => Ptr PGconn -> (Ptr PGparam -> IO r) -> IO r withPGparam conn = E.bracket create c_PQparamClear where create = alloca $ \err -> do@@ -99,21 +100,21 @@ ---------------------------------------- -- | Throw libpq specific error.-throwLibPQError :: Ptr PGconn -> String -> IO a+throwLibPQError :: HasCallStack => Ptr PGconn -> String -> IO a throwLibPQError conn ctx = do msg <- safePeekCString' =<< c_PQerrorMessage conn E.throwIO . LibPQError $ if null ctx then msg else ctx ++ ": " ++ msg -- | Throw libpqtypes specific error.-throwLibPQTypesError :: Ptr PGerror -> String -> IO a+throwLibPQTypesError :: HasCallStack => Ptr PGerror -> String -> IO a throwLibPQTypesError err ctx = do msg <- pgErrorMsg <$> peek err E.throwIO . LibPQError $ if null ctx then msg else ctx ++ ": " ++ msg -- | Rethrow supplied exception enriched with array index.-rethrowWithArrayError :: CInt -> E.SomeException -> IO a+rethrowWithArrayError :: HasCallStack => CInt -> E.SomeException -> IO a rethrowWithArrayError i (E.SomeException e) = E.throwIO ArrayItemError@@ -122,9 +123,9 @@ } -- | Throw 'HPQTypesError exception.-hpqTypesError :: String -> IO a+hpqTypesError :: HasCallStack => String -> IO a hpqTypesError = E.throwIO . HPQTypesError -- | Throw 'unexpected NULL' exception.-unexpectedNULL :: IO a+unexpectedNULL :: HasCallStack => IO a unexpectedNULL = hpqTypesError "unexpected NULL"
src/Database/PostgreSQL/PQTypes/Transaction.hs view
@@ -1,26 +1,19 @@ module Database.PostgreSQL.PQTypes.Transaction ( Savepoint (..) , withSavepoint- , withTransaction , begin , commit , rollback- , withTransaction'- , begin'- , commit'- , rollback'+ , unsafeWithoutTransaction ) where -import Control.Monad import Control.Monad.Catch-import Data.Function import Data.String-import Data.Typeable import GHC.Stack import Data.Monoid.Utils import Database.PostgreSQL.PQTypes.Class-import Database.PostgreSQL.PQTypes.Internal.Exception+import Database.PostgreSQL.PQTypes.Internal.Error import Database.PostgreSQL.PQTypes.SQL.Raw import Database.PostgreSQL.PQTypes.Transaction.Settings import Database.PostgreSQL.PQTypes.Utils@@ -54,93 +47,65 @@ ---------------------------------------- --- | Same as 'withTransaction'' except that it uses current--- transaction settings instead of custom ones. It is worth--- noting that changing transaction settings inside supplied--- monadic action won't have any effect on the final 'commit'--- / 'rollback' as settings that were in effect during the call--- to 'withTransaction' will be used.-withTransaction :: (HasCallStack, MonadDB m, MonadMask m) => m a -> m a-withTransaction m = getTransactionSettings >>= flip withTransaction' m---- | Begin transaction using current transaction settings.-begin :: (HasCallStack, MonadDB m) => m ()-begin = getTransactionSettings >>= begin'---- | Commit active transaction using current transaction settings.-commit :: (HasCallStack, MonadDB m) => m ()-commit = getTransactionSettings >>= commit'---- | Rollback active transaction using current transaction settings.-rollback :: (HasCallStack, MonadDB m) => m ()-rollback = getTransactionSettings >>= rollback'---------------------------------------------- | Execute monadic action within a transaction using given transaction--- settings. Note that it won't work as expected if a transaction is already--- active (in such case 'withSavepoint' should be used instead).-withTransaction'- :: (HasCallStack, MonadDB m, MonadMask m)- => TransactionSettings- -> m a- -> m a-withTransaction' ts m = (`fix` 1) $ \loop n -> do- -- Optimization for squashing possible space leaks.- -- It looks like GHC doesn't like 'catch' and passes- -- on introducing strictness in some cases.- let maybeRestart = case tsRestartPredicate ts of- Just _ -> handleJust (expred n) (\_ -> loop $ n + 1)- Nothing -> id- maybeRestart $- fst- <$> generalBracket- (begin' ts)- ( \() -> \case- ExitCaseSuccess _ -> commit' ts- _ -> rollback' ts- )- (\() -> m)- where- expred :: Integer -> SomeException -> Maybe ()- expred !n e = do- -- check if the predicate exists- RestartPredicate f <- tsRestartPredicate ts- -- cast exception to the type expected by the predicate- err <-- msum- [ -- either cast the exception itself...- fromException e- , -- ...or extract it from DBException- fromException e >>= \DBException {..} -> cast dbeError- ]- -- check if the predicate allows for the restart- guard $ f err n+-- Note: sql queries in below functions that modify transaction state need to+-- not be interruptible so we don't end up in unexpected transaction+-- state. However, getConnectionAcquisitionMode should be interruptible to not+-- lead to deadlocks if a connection ends up being used from multiple threads. -- | Begin transaction using given transaction settings.-begin' :: (HasCallStack, MonadDB m) => TransactionSettings -> m ()-begin' ts = runSQL_ . mintercalate " " $ ["BEGIN", isolationLevel, permissions]- where- isolationLevel = case tsIsolationLevel ts of- DefaultLevel -> ""- ReadCommitted -> "ISOLATION LEVEL READ COMMITTED"- RepeatableRead -> "ISOLATION LEVEL REPEATABLE READ"- Serializable -> "ISOLATION LEVEL SERIALIZABLE"- permissions = case tsPermissions ts of- DefaultPermissions -> ""- ReadOnly -> "READ ONLY"- ReadWrite -> "READ WRITE"+begin :: (HasCallStack, MonadDB m, MonadMask m) => m ()+begin = do+ getConnectionAcquisitionMode >>= \case+ AcquireOnDemand -> do+ throwDB $ HPQTypesError "Can't begin a transaction in OnDemand mode"+ AcquireAndHold isolationLevel permissions -> uninterruptibleMask_ $ do+ runSQL_ $+ smconcat+ [ "BEGIN"+ , case isolationLevel of+ DefaultLevel -> ""+ ReadCommitted -> "ISOLATION LEVEL READ COMMITTED"+ RepeatableRead -> "ISOLATION LEVEL REPEATABLE READ"+ Serializable -> "ISOLATION LEVEL SERIALIZABLE"+ , case permissions of+ DefaultPermissions -> ""+ ReadOnly -> "READ ONLY"+ ReadWrite -> "READ WRITE"+ ] -- | Commit active transaction using given transaction settings.-commit' :: (HasCallStack, MonadDB m) => TransactionSettings -> m ()-commit' ts = do- runSQL_ "COMMIT"- when (tsAutoTransaction ts) $- begin' ts+commit :: (HasCallStack, MonadDB m, MonadMask m) => m ()+commit = do+ getConnectionAcquisitionMode >>= \case+ AcquireOnDemand -> do+ throwDB $ HPQTypesError "Can't commit a transaction in OnDemand mode"+ AcquireAndHold {} -> uninterruptibleMask_ $ do+ runSQL_ "COMMIT"+ begin -- | Rollback active transaction using given transaction settings.-rollback' :: (HasCallStack, MonadDB m) => TransactionSettings -> m ()-rollback' ts = do- runSQL_ "ROLLBACK"- when (tsAutoTransaction ts) $- begin' ts+rollback :: (HasCallStack, MonadDB m, MonadMask m) => m ()+rollback = do+ getConnectionAcquisitionMode >>= \case+ AcquireOnDemand -> do+ throwDB $ HPQTypesError "Can't rollback a transaction in OnDemand mode"+ AcquireAndHold {} -> uninterruptibleMask_ $ do+ runSQL_ "ROLLBACK"+ begin++-- | Run a block of code without an open transaction.+--+-- This function is unsafe, because if there is a transaction in progress, it's+-- commited, so the atomicity guarantee is lost.+unsafeWithoutTransaction+ :: (HasCallStack, MonadDB m, MonadMask m)+ => m a+ -> m a+unsafeWithoutTransaction action = do+ getConnectionAcquisitionMode >>= \case+ AcquireOnDemand -> action+ AcquireAndHold {} ->+ bracket_+ (uninterruptibleMask_ $ runSQL_ "COMMIT")+ begin+ action
src/Database/PostgreSQL/PQTypes/Transaction/Settings.hs view
@@ -1,6 +1,7 @@ module Database.PostgreSQL.PQTypes.Transaction.Settings ( RestartPredicate (..) , TransactionSettings (..)+ , ConnectionAcquisitionMode (..) , IsolationLevel (..) , Permissions (..) , defaultTransactionSettings@@ -18,25 +19,22 @@ showsPrec _ RestartPredicate {} = (++) "RestartPredicate" data TransactionSettings = TransactionSettings- { tsAutoTransaction :: !Bool- -- ^ If set to True, transaction will be automatically started at the- -- beginning of database action and after each 'commit' / 'rollback'. If- -- set to False, no transaction will automatically start in either of above- -- cases.- , tsIsolationLevel :: !IsolationLevel- -- ^ Isolation level of all transactions.- , tsRestartPredicate :: !(Maybe RestartPredicate)+ { tsRestartPredicate :: !(Maybe RestartPredicate) -- ^ Defines behavior of 'withTransaction' in case exceptions thrown within- -- supplied monadic action are not caught and reach its body. If set to- -- 'Nothing', exceptions will be propagated as usual. If set to 'Just' f,- -- exceptions will be intercepted and passed to f along with a number that- -- indicates how many times the transaction block already failed. If f- -- returns 'True', the transaction is restarted. Otherwise the exception is- -- further propagated. This allows for restarting transactions e.g. in case- -- of serialization failure. It is up to the caller to ensure that is it- -- safe to execute supplied monadic action multiple times.- , tsPermissions :: !Permissions- -- ^ Permissions of all transactions.+ -- supplied monadic action are not caught and reach its body.+ --+ -- If set to 'Nothing', exceptions will be propagated as usual.+ --+ -- If set to 'Just' f, exceptions will be intercepted and passed to f along+ -- with a number that indicates how many times the transaction block already+ -- failed.+ --+ -- If f returns 'True', the transaction is restarted. Otherwise the+ -- exception is further propagated. This allows for restarting transactions+ -- e.g. in case of serialization failure. It is up to the caller to ensure+ -- that is it safe to execute supplied monadic action multiple times.+ , tsConnectionAcquisitionMode :: !ConnectionAcquisitionMode+ -- ^ Acquisition mode of a database connection. } deriving (Show) @@ -46,12 +44,24 @@ data Permissions = DefaultPermissions | ReadOnly | ReadWrite deriving (Eq, Ord, Show) +-- | Acquisition mode of a database connection.+data ConnectionAcquisitionMode+ = -- | Acquire a connection on demand, i.e. only when a query needs to be+ -- run. This mode enables you to have a+ -- t'Database.PostgreSQL.PQTypes.Class.MonadDB' constraint in scope without+ -- keeping a transaction open, but is limited to executionn of read only+ -- queries.+ AcquireOnDemand+ | -- | Acquire a connection, start a transaction with a given isolation level+ -- and permissions and hold onto it for the duration of a+ -- t'Database.PostgreSQL.PQTypes.Class.MonadDB' constraint in scope.+ AcquireAndHold !IsolationLevel !Permissions+ deriving (Eq, Ord, Show)+ -- | Default transaction settings. defaultTransactionSettings :: TransactionSettings defaultTransactionSettings = TransactionSettings- { tsAutoTransaction = True- , tsIsolationLevel = DefaultLevel- , tsRestartPredicate = Nothing- , tsPermissions = DefaultPermissions+ { tsRestartPredicate = Nothing+ , tsConnectionAcquisitionMode = AcquireAndHold DefaultLevel DefaultPermissions }
src/Database/PostgreSQL/PQTypes/Utils.hs view
@@ -37,8 +37,7 @@ throwDB e = case fromException $ toException e of Just (dbe :: DBException) -> throwM dbe Nothing -> do- SomeSQL sql <- getLastQuery- pid <- getBackendPid+ (pid, SomeSQL sql) <- getLastQuery throwM DBException { dbeQueryContext = sql
test/Main.hs view
@@ -1,4 +1,4 @@-{-# OPTIONS_GHC -Wno-orphans #-}+{-# OPTIONS_GHC -Wno-deprecations -Wno-orphans #-} module Main (main) where @@ -208,9 +208,6 @@ ---------------------------------------- -tsNoTrans :: TransactionSettings-tsNoTrans = defaultTransactionSettings {tsAutoTransaction = False}- randomValue :: Arbitrary t => Int -> TestEnv t randomValue n = withQCGen $ \gen -> unGen arbitrary gen n @@ -294,7 +291,7 @@ moved withHoldCursorWorks = testCase "Cursor declared as WITH HOLD works" $ do- runTestEnv td tsNoTrans $ do+ runTestEnv td defaultTransactionSettings . unsafeWithoutTransaction $ do withCursor "ints" NoScroll Hold (sqlGenInts 10) $ \cursor -> do cursorFetch_ cursor CD_Forward_All sum_ :: Int32 <- sum . fmap runIdentity <$> queryResult@@ -310,7 +307,7 @@ queryInterruptionTest td = testCase "Queries are interruptible" $ do let sleep = "SELECT pg_sleep(2)" ints = sqlGenInts 5000000- runTestEnv td tsNoTrans $ do+ runTestEnv td defaultTransactionSettings . unsafeWithoutTransaction $ do testQuery id sleep testQuery id ints runTestEnv td defaultTransactionSettings $ do@@ -325,7 +322,8 @@ autocommitTest :: TestData -> Test autocommitTest td = testCase "Autocommit mode works"- . runTestEnv td tsNoTrans+ . runTestEnv td defaultTransactionSettings+ . unsafeWithoutTransaction $ do let sint = Identity (1 :: Int32) runQuery_ $ rawSQL "INSERT INTO test1_ (a) VALUES ($1)" sint@@ -395,7 +393,7 @@ readOnlyTest td = testCase "Read only transaction mode works" . runTestEnv td- defaultTransactionSettings {tsPermissions = ReadOnly}+ defaultTransactionSettings {tsConnectionAcquisitionMode = AcquireAndHold DefaultLevel ReadOnly} $ do let sint = Identity (2 :: Int32) eres <- try . runQuery_ $ rawSQL "INSERT INTO test1_ (a) VALUES ($1)" sint@@ -436,7 +434,7 @@ assertEqualEq "Result of all queries is visible" [int1, int2] res2 notifyTest :: TestData -> Test-notifyTest td = testCase "Notifications work" . runTestEnv td tsNoTrans $ do+notifyTest td = testCase "Notifications work" . runTestEnv td defaultTransactionSettings . unsafeWithoutTransaction $ do listen chan forkNewConn $ notify chan payload mnt1 <- getNotification 250000@@ -458,7 +456,17 @@ where chan = "test_channel" payload = "test_payload"- forkNewConn = void . fork . withNewConnection+ forkNewConn action = do+ sem <- newEmptyMVar+ void . fork . withNewConnection $ do+ -- withNewConnection needs access to the connection state to get current+ -- ConnectionAcquisitionMode, but getNotification called immediately+ -- after takes ownership of the connection state for its duration, so if+ -- CPU gets to it first, withNewConnection will block and notification+ -- will never be sent.+ putMVar sem ()+ action+ takeMVar sem transactionTest :: TestData -> IsolationLevel -> Test transactionTest td lvl =@@ -468,7 +476,7 @@ ) . runTestEnv td- defaultTransactionSettings {tsIsolationLevel = lvl}+ defaultTransactionSettings {tsConnectionAcquisitionMode = AcquireAndHold lvl DefaultPermissions} $ do let sint = Identity (5 :: Int32) runQuery_ $ rawSQL "INSERT INTO test1_ (a) VALUES ($1)" sint@@ -545,6 +553,39 @@ assertEqualEq "XML value correct" v v'' runSQL_ "SET CLIENT_ENCODING TO 'latin-1'" +onDemandTest :: TestData -> Test+onDemandTest td = testCase "OnDemand mode works" . runTestEnv td ts $ do+ runSQL_ "SELECT a FROM test1_"+ _ <- fetchMany $ id @(Identity Int32)++ er <- try . runSQL_ $ "INSERT INTO test1_ (a) VALUES (" <?> v <+> ")"+ liftBase $ case er of+ Left DBException {..}+ | Just DetailedQueryError {..} <- cast dbeError -> do+ assertEqualEq "Unexpected error code" ReadOnlySqlTransaction qeErrorCode+ | otherwise -> assertFailure $ "Unexpected exception: " ++ show dbeError+ Right () -> assertFailure "DBException wasn't thrown"++ acquireAndHoldConnection DefaultLevel DefaultPermissions+ runSQL_ "SHOW transaction_read_only"+ Identity ("off" :: T.Text) <- fetchOne id+ -- Switch twice to check idempotency.+ acquireAndHoldConnection DefaultLevel DefaultPermissions+ runSQL_ $ "INSERT INTO test1_ (a) VALUES (" <?> v <+> ")"++ unsafeAcquireOnDemandConnection+ runSQL_ "SHOW transaction_read_only"+ Identity ("on" :: T.Text) <- fetchOne id+ -- Switch twice to check idempotency.+ unsafeAcquireOnDemandConnection+ n <- runSQL $ "SELECT TRUE FROM test1_ WHERE a =" <?> v+ assertEqualEq "Unexpected amount of rows" 1 n+ where+ ts = defaultTransactionSettings {tsConnectionAcquisitionMode = AcquireOnDemand}++ v :: Int32+ v = 1337+ rowTest :: forall row . (Arbitrary row, Eq row, Show row, ToRow row, FromRow row)@@ -585,6 +626,7 @@ , queryInterruptionTest td , cursorTest td , uuidTest td+ , onDemandTest td , transactionTest td ReadCommitted , transactionTest td RepeatableRead , transactionTest td Serializable