diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,9 @@ +# hpqtypes-1.5.3.0 (2018-06-04) +* add INLINE/INLINEABLE pragmas for call site specialization +* remove -O2 -funbox-strict-fields from ghc-options +* make query execution interruptible with asynchronous exceptions +* make connect interruptible with asynchronous exceptions + # hpqtypes-1.5.2.0 (2018-03-18) * support GHC 8.4.1 diff --git a/hpqtypes.cabal b/hpqtypes.cabal --- a/hpqtypes.cabal +++ b/hpqtypes.cabal @@ -1,5 +1,5 @@ name: hpqtypes -version: 1.5.2.0 +version: 1.5.3.0 synopsis: Haskell bindings to libpqtypes description: Efficient and easy-to-use bindings to (slightly modified) @@ -35,7 +35,7 @@ copyright: Scrive AB category: Database build-type: Custom -cabal-version: >= 1.18 +cabal-version: 1.18 tested-with: GHC == 7.8.4, GHC == 7.10.3, GHC == 8.0.1, GHC == 8.2.2, GHC == 8.4.1 @@ -79,7 +79,7 @@ source-repository this type: git location: https://github.com/scrive/hpqtypes.git - tag: 1.5.2.0 + tag: 1.5.3.0 library exposed-modules: Data.Monoid.Utils @@ -122,6 +122,7 @@ build-depends: base >= 4.7 && < 4.12 , text >= 0.11 , aeson >= 0.6.2.0 + , async >= 2.1.1.1 , bytestring >= 0.9 , semigroups >= 0.16 , time >= 1.4 @@ -139,7 +140,7 @@ hs-source-dirs: src - ghc-options: -O2 -Wall -funbox-strict-fields + ghc-options: -Wall build-tools: hsc2hs include-dirs: libpqtypes/src @@ -177,7 +178,9 @@ , ForeignFunctionInterface , GADTs , GeneralizedNewtypeDeriving + , LambdaCase , MultiParamTypeClasses + , MultiWayIf , NoImplicitPrelude , OverloadedStrings , RankNTypes @@ -191,7 +194,7 @@ test-suite hpqtypes-tests type: exitcode-stdio-1.0 - ghc-options: -O2 -Wall -funbox-strict-fields -threaded + ghc-options: -Wall -threaded hs-source-dirs: test main-is: Main.hs @@ -231,7 +234,9 @@ , ForeignFunctionInterface , GADTs , GeneralizedNewtypeDeriving + , LambdaCase , MultiParamTypeClasses + , MultiWayIf , NoImplicitPrelude , OverloadedStrings , RankNTypes diff --git a/src/Data/Monoid/Utils.hs b/src/Data/Monoid/Utils.hs --- a/src/Data/Monoid/Utils.hs +++ b/src/Data/Monoid/Utils.hs @@ -12,22 +12,27 @@ import Prelude -- | Generalization of 'intercalate' to arbitrary 'Monoid'. +{-# INLINE mintercalate #-} mintercalate :: Monoid m => m -> [m] -> m mintercalate m = mconcat . intersperse m -- | Generalization of separator to arbitrary 'Monoid'. +{-# INLINE mspace #-} mspace :: (IsString m, Monoid m) => m mspace = fromString " " -- | Concatenate two elements with separator between them. +{-# INLINE smappend #-} smappend :: (IsString m, Monoid m) => m -> m -> m smappend a b = mconcat [a, mspace, b] -- | Concatenate a list of elements, inserting separators between them. +{-# INLINE smconcat #-} smconcat :: (IsString m, Monoid m) => [m] -> m smconcat = mintercalate mspace -- | Infix version of 'smappend'. +{-# INLINE (<+>) #-} (<+>) :: (IsString m, Monoid m) => m -> m -> m (<+>) = smappend infixr 6 <+> diff --git a/src/Database/PostgreSQL/PQTypes/Class.hs b/src/Database/PostgreSQL/PQTypes/Class.hs --- a/src/Database/PostgreSQL/PQTypes/Class.hs +++ b/src/Database/PostgreSQL/PQTypes/Class.hs @@ -82,6 +82,15 @@ setTransactionSettings = lift . setTransactionSettings getNotification = lift . getNotification withNewConnection m = controlT $ \run -> withNewConnection (run m) + {-# INLINE runQuery #-} + {-# INLINE getLastQuery #-} + {-# INLINE getConnectionStats #-} + {-# INLINE getQueryResult #-} + {-# INLINE clearQueryResult #-} + {-# INLINE getTransactionSettings #-} + {-# INLINE setTransactionSettings #-} + {-# INLINE getNotification #-} + {-# INLINE withNewConnection #-} controlT :: (MonadTransControl t, Monad (t m), Monad m) => (Run t -> m (StT t a)) -> t m a diff --git a/src/Database/PostgreSQL/PQTypes/Fold.hs b/src/Database/PostgreSQL/PQTypes/Fold.hs --- a/src/Database/PostgreSQL/PQTypes/Fold.hs +++ b/src/Database/PostgreSQL/PQTypes/Fold.hs @@ -20,6 +20,7 @@ import Database.PostgreSQL.PQTypes.Utils -- | Get current 'QueryResult' or throw an exception if there isn't one. +{-# INLINABLE queryResult #-} queryResult :: (MonadDB m, MonadThrow m, FromRow row) => m (QueryResult row) queryResult = getQueryResult >>= maybe (throwDB . HPQTypesError $ "queryResult: no query result") return @@ -27,25 +28,30 @@ ---------------------------------------- -- | Specialization of 'F.foldrM' for convenient query results fetching. +{-# INLINABLE foldrDB #-} foldrDB :: (MonadDB m, FromRow row) => (row -> acc -> m acc) -> acc -> m acc foldrDB f acc = maybe (return acc) (F.foldrM f acc) =<< getQueryResult -- | Specialization of 'F.foldlM' for convenient query results fetching. +{-# INLINABLE foldlDB #-} foldlDB :: (MonadDB m, FromRow row) => (acc -> row -> m acc) -> acc -> m acc foldlDB f acc = maybe (return acc) (F.foldlM f acc) =<< getQueryResult -- | Specialization of 'F.mapM_' for convenient mapping over query results. +{-# INLINABLE mapDB_ #-} mapDB_ :: (MonadDB m, FromRow row) => (row -> m t) -> m () mapDB_ f = maybe (return ()) (F.mapM_ f) =<< getQueryResult ---------------------------------------- -- | Specialization of 'foldrDB' that fetches a list of rows. +{-# INLINABLE fetchMany #-} fetchMany :: (MonadDB m, FromRow row) => (row -> t) -> m [t] fetchMany f = foldrDB (\row acc -> return $ f row : acc) [] -- | Specialization of 'foldlDB' that fetches one or zero rows. If -- more rows are delivered, 'AffectedRowsMismatch' exception is thrown. +{-# INLINABLE fetchMaybe #-} fetchMaybe :: (MonadDB m, MonadThrow m, FromRow row) => (row -> t) -> m (Maybe t) fetchMaybe f = getQueryResult >>= \mqr -> case mqr of Nothing -> return Nothing @@ -59,6 +65,7 @@ -- | Specialization of 'fetchMaybe' that fetches exactly one row. If -- no row is delivered, 'AffectedRowsMismatch' exception is thrown. +{-# INLINABLE fetchOne #-} fetchOne :: (MonadDB m, MonadThrow m, FromRow row) => (row -> t) -> m t fetchOne f = do mt <- fetchMaybe f diff --git a/src/Database/PostgreSQL/PQTypes/Internal/C/Interface.hs b/src/Database/PostgreSQL/PQTypes/Internal/C/Interface.hs --- a/src/Database/PostgreSQL/PQTypes/Internal/C/Interface.hs +++ b/src/Database/PostgreSQL/PQTypes/Internal/C/Interface.hs @@ -16,9 +16,12 @@ , c_PQgetisnull , c_PQfname , c_PQclear - -- * libpqtypes imports + , c_PQcancel + , c_PQconnectStart + , c_PQconnectPoll , c_PQfinishPtr - , c_PQconnectdb + , c_ptr_PQfinishPtr + -- * libpqtypes imports , c_PQinitTypes , c_PQregisterTypes , c_PQparamExec @@ -29,10 +32,11 @@ , nullStringCStringLen ) where +import Control.Applicative import Foreign.C import Foreign.ForeignPtr +import Foreign.Marshal.Alloc import Foreign.Ptr -import Foreign.Storable import Prelude import System.Posix.Types import qualified Control.Exception as E @@ -88,30 +92,45 @@ ---------------------------------------- --- | May block in case of network problems, hence marked 'safe'. -foreign import ccall safe "PQconnectdb" - c_rawPQconnectdb :: CString -> IO (Ptr PGconn) +foreign import ccall unsafe "PQgetCancel" + c_PQgetCancel :: Ptr PGconn -> IO (Ptr PGcancel) +foreign import ccall unsafe "PQfreeCancel" + c_PQfreeCancel :: Ptr PGcancel -> IO () + +foreign import ccall unsafe "PQcancel" + c_rawPQcancel :: Ptr PGcancel -> CString -> CInt -> IO CInt + +-- | Attempt to cancel currently running query. If the request is successfully +-- dispatched Nothing is returned, otherwise a textual explanation of what +-- happened. +c_PQcancel :: Ptr PGconn -> IO (Maybe String) +c_PQcancel conn = E.mask $ \restore -> do + cancel <- c_PQgetCancel conn + (`E.finally` c_PQfreeCancel cancel) . restore $ do + allocaBytes errbufsize $ \errbuf -> do + c_rawPQcancel cancel errbuf (fromIntegral errbufsize) >>= \case + 0 -> Just <$> peekCString errbuf + _ -> return Nothing + where + -- Size recommended by + -- https://www.postgresql.org/docs/current/static/libpq-cancel.html + errbufsize :: Int + errbufsize = 256 + +---------------------------------------- + +foreign import ccall unsafe "PQconnectStart" + c_PQconnectStart :: CString -> IO (Ptr PGconn) + +foreign import ccall unsafe "PQconnectPoll" + c_PQconnectPoll :: Ptr PGconn -> IO PostgresPollingStatusType + foreign import ccall unsafe "PQfinishPtr" c_PQfinishPtr :: Ptr (Ptr PGconn) -> IO () foreign import ccall unsafe "&PQfinishPtr" c_ptr_PQfinishPtr :: FunPtr (Ptr (Ptr PGconn) -> IO ()) - --- | Safe wrapper for 'c_rawPQconnectdb', returns --- 'ForeignPtr' instead of 'Ptr'. -c_PQconnectdb :: CString -> IO (ForeignPtr (Ptr PGconn)) -c_PQconnectdb conninfo = E.mask_ $ do - conn <- c_rawPQconnectdb conninfo - -- Work around a bug in GHC that causes foreign pointer - -- finalizers to be run multiple times under random - -- circumstances by providing another level of indirection - -- and a wrapper for PQfinish that can be safely called - -- multiple times. - connPtr <- mallocForeignPtr - withForeignPtr connPtr $ flip poke conn - addForeignPtrFinalizer c_ptr_PQfinishPtr connPtr - return connPtr -- libpqtypes imports diff --git a/src/Database/PostgreSQL/PQTypes/Internal/C/Types.hsc b/src/Database/PostgreSQL/PQTypes/Internal/C/Types.hsc --- a/src/Database/PostgreSQL/PQTypes/Internal/C/Types.hsc +++ b/src/Database/PostgreSQL/PQTypes/Internal/C/Types.hsc @@ -1,6 +1,7 @@ -- | Mappings of types used by libpq/libpqtypes to Haskell ADTs. module Database.PostgreSQL.PQTypes.Internal.C.Types ( - PGconn + PGcancel + , PGconn , PGparam , PGresult , PGtypeArgs @@ -8,6 +9,9 @@ , c_CONNECTION_OK, c_CONNECTION_BAD, c_CONNECTION_STARTED , c_CONNECTION_MADE, c_CONNECTION_AWAITING_RESPONSE, c_CONNECTION_AUTH_OK , c_CONNECTION_SETENV, c_CONNECTION_SSL_STARTUP, c_CONNECTION_NEEDED + , PostgresPollingStatusType(..) + , c_PGRES_POLLING_FAILED, c_PGRES_POLLING_READING, c_PGRES_POLLING_WRITING + , c_PGRES_POLLING_OK, c_PGRES_POLLING_ACTIVE , ResultFormat(..) , c_RESULT_TEXT, c_RESULT_BINARY , ExecStatusType(..) @@ -44,6 +48,7 @@ #let alignment t = "%lu", (unsigned long)offsetof(struct {char x__; t (y__);}, y__) +data PGcancel data PGconn data PGparam data PGresult @@ -68,6 +73,19 @@ , c_CONNECTION_SSL_STARTUP = CONNECTION_SSL_STARTUP , c_CONNECTION_NEEDED = CONNECTION_NEEDED } + +---------------------------------------- + +newtype PostgresPollingStatusType = PostgresPollingStatusType CInt + deriving Eq + +#{enum PostgresPollingStatusType, PostgresPollingStatusType + , c_PGRES_POLLING_FAILED = PGRES_POLLING_FAILED + , c_PGRES_POLLING_READING = PGRES_POLLING_READING + , c_PGRES_POLLING_WRITING = PGRES_POLLING_WRITING + , c_PGRES_POLLING_OK = PGRES_POLLING_OK + , c_PGRES_POLLING_ACTIVE = PGRES_POLLING_ACTIVE + } ---------------------------------------- diff --git a/src/Database/PostgreSQL/PQTypes/Internal/Connection.hs b/src/Database/PostgreSQL/PQTypes/Internal/Connection.hs --- a/src/Database/PostgreSQL/PQTypes/Internal/Connection.hs +++ b/src/Database/PostgreSQL/PQTypes/Internal/Connection.hs @@ -14,14 +14,16 @@ ) where import Control.Applicative -import Control.Arrow -import Control.Concurrent.MVar +import Control.Arrow (first) +import Control.Concurrent import Control.Monad import Control.Monad.Base import Control.Monad.Catch import Data.Default.Class +import Data.Function import Data.Pool import Data.Time.Clock +import Foreign.C.String import Foreign.ForeignPtr import Foreign.Ptr import Foreign.Storable @@ -170,16 +172,16 @@ -- Useful if one wants to implement custom connection source. connect :: ConnectionSettings -> IO Connection connect ConnectionSettings{..} = do - fconn <- BS.useAsCString (T.encodeUtf8 csConnInfo) c_PQconnectdb + fconn <- BS.useAsCString (T.encodeUtf8 csConnInfo) openConnection withForeignPtr fconn $ \connPtr -> do conn <- peek connPtr status <- c_PQstatus conn when (status /= c_CONNECTION_OK) $ - throwLibPQError conn "connect" + throwLibPQError conn fname F.forM_ csClientEncoding $ \enc -> do res <- BS.useAsCString (T.encodeUtf8 enc) (c_PQsetClientEncoding conn) when (res == -1) $ - throwLibPQError conn "connect" + throwLibPQError conn fname c_PQinitTypes conn registerComposites conn csComposites Connection <$> newMVar (Just ConnectionData { @@ -187,6 +189,42 @@ , cdPtr = conn , cdStats = initialStats }) + where + fname = "connect" + + openConnection :: CString -> IO (ForeignPtr (Ptr PGconn)) + openConnection conninfo = E.mask $ \restore -> do + -- We want to use non-blocking C functions to be able to observe + -- incoming asynchronous exceptions, hence we don't use + -- PQconnectdb here. + conn <- c_PQconnectStart conninfo + when (conn == nullPtr) $ + throwError "PQconnectStart returned a null pointer" + -- Work around a bug in GHC that causes foreign pointer + -- finalizers to be run multiple times under random + -- circumstances (see + -- https://ghc.haskell.org/trac/ghc/ticket/7170 for more + -- details) by providing another level of indirection and a + -- wrapper for PQfinish that can be safely called multiple + -- times. + connPtr <- mallocForeignPtr + withForeignPtr connPtr (`poke` conn) + addForeignPtrFinalizer c_ptr_PQfinishPtr connPtr + -- Wait until connection status is resolved (to either + -- established or failed state). + restore $ fix $ \loop -> do + ps <- c_PQconnectPoll conn + if | ps == c_PGRES_POLLING_READING -> (threadWaitRead =<< getFd conn) >> loop + | ps == c_PGRES_POLLING_WRITING -> (threadWaitWrite =<< getFd conn) >> loop + | otherwise -> return connPtr + where + getFd conn = do + fd <- c_PQsocket conn + when (fd == -1) $ + throwError "invalid file descriptor" + return fd + + throwError = hpqTypesError . (fname ++) . (": " ++) -- | Low-level function for disconnecting from the database. -- Useful if one wants to implement custom connection source. diff --git a/src/Database/PostgreSQL/PQTypes/Internal/Monad.hs b/src/Database/PostgreSQL/PQTypes/Internal/Monad.hs --- a/src/Database/PostgreSQL/PQTypes/Internal/Monad.hs +++ b/src/Database/PostgreSQL/PQTypes/Internal/Monad.hs @@ -42,6 +42,7 @@ -- | Evaluate monadic action with supplied -- connection source and transaction settings. +{-# INLINABLE runDBT #-} runDBT :: (MonadBase IO m, MonadMask m) => ConnectionSourceM m @@ -62,6 +63,7 @@ else m -- | Transform the underlying monad. +{-# INLINABLE mapDBT #-} mapDBT :: (DBState n -> DBState m) -> (m (a, DBState m) -> n (b, DBState n)) @@ -81,12 +83,12 @@ Nothing -> throwDB $ HPQTypesError "getConnectionStats: no connection" Just cd -> return $ cdStats cd - getTransactionSettings = DBT . gets $ dbTransactionSettings - setTransactionSettings ts = DBT . modify $ \st -> st { dbTransactionSettings = ts } - getQueryResult = DBT . gets $ dbQueryResult clearQueryResult = DBT . modify $ \st -> st { dbQueryResult = Nothing } + getTransactionSettings = DBT . gets $ dbTransactionSettings + setTransactionSettings ts = DBT . modify $ \st -> st { dbTransactionSettings = ts } + getNotification time = DBT . StateT $ \st -> (, st) <$> liftBase (getNotificationIO st time) @@ -96,6 +98,16 @@ res <- runDBT cs ts m return (res, st) + {-# INLINABLE runQuery #-} + {-# INLINABLE getLastQuery #-} + {-# INLINABLE getConnectionStats #-} + {-# INLINABLE getQueryResult #-} + {-# INLINABLE clearQueryResult #-} + {-# INLINABLE getTransactionSettings #-} + {-# INLINABLE setTransactionSettings #-} + {-# INLINABLE getNotification #-} + {-# INLINABLE withNewConnection #-} + ---------------------------------------- instance MonadTransControl (DBT_ m) where @@ -131,19 +143,31 @@ instance (m ~ n, MonadError e m) => MonadError e (DBT_ m n) where throwError = lift . throwError catchError m h = DBT $ S.liftCatch catchError (unDBT m) (unDBT . h) + {-# INLINE throwError #-} + {-# INLINE catchError #-} instance (m ~ n, MonadReader r m) => MonadReader r (DBT_ m n) where ask = lift ask local f = mapDBT id (local f) reader = lift . reader + {-# INLINE ask #-} + {-# INLINE local #-} + {-# INLINE reader #-} instance (m ~ n, MonadState s m) => MonadState s (DBT_ m n) where get = lift get put = lift . put state = lift . state + {-# INLINE get #-} + {-# INLINE put #-} + {-# INLINE state #-} instance (m ~ n, MonadWriter w m) => MonadWriter w (DBT_ m n) where writer = lift . writer tell = lift . tell listen = DBT . S.liftListen listen . unDBT pass = DBT . S.liftPass pass . unDBT + {-# INLINE writer #-} + {-# INLINE tell #-} + {-# INLINE listen #-} + {-# INLINE pass #-} diff --git a/src/Database/PostgreSQL/PQTypes/Internal/Query.hs b/src/Database/PostgreSQL/PQTypes/Internal/Query.hs --- a/src/Database/PostgreSQL/PQTypes/Internal/Query.hs +++ b/src/Database/PostgreSQL/PQTypes/Internal/Query.hs @@ -3,6 +3,8 @@ ) where import Control.Applicative +import Control.Concurrent.Async +import Control.Monad import Foreign.ForeignPtr import Foreign.Ptr import Prelude @@ -24,28 +26,55 @@ -- | Low-level function for running SQL query. runQueryIO :: IsSQL sql => sql -> DBState m -> IO (Int, DBState m) runQueryIO sql st = do - (affected, res) <- withConnectionData (dbConnection st) "runQueryIO" $ \cd -> do - let ConnectionData{..} = cd - allocParam = ParamAllocator $ withPGparam cdPtr - (paramCount, res) <- withSQL sql allocParam $ \param query -> (,) - <$> (fromIntegral <$> c_PQparamCount param) - <*> c_PQparamExec cdPtr nullPtr param query c_RESULT_BINARY - affected <- withForeignPtr res $ verifyResult cdPtr - stats' <- case affected of - Left _ -> return cdStats { - statsQueries = statsQueries cdStats + 1 - , statsParams = statsParams cdStats + paramCount - } - Right rows -> do - columns <- fromIntegral <$> withForeignPtr res c_PQnfields - return ConnectionStats { + (affected, res) <- 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 + let allocParam = ParamAllocator $ withPGparam cdPtr + (paramCount, res) <- withSQL sql allocParam $ \param query -> (,) + <$> (fromIntegral <$> c_PQparamCount param) + <*> c_PQparamExec cdPtr nullPtr param query c_RESULT_BINARY + affected <- withForeignPtr res $ verifyResult cdPtr + stats' <- case affected of + Left _ -> return cdStats { statsQueries = statsQueries cdStats + 1 - , statsRows = statsRows cdStats + rows - , statsValues = statsValues cdStats + (rows * columns) , statsParams = statsParams cdStats + paramCount } - -- Force evaluation of modified stats to squash space leak. - stats' `seq` return (cd { cdStats = stats' }, (either id id affected, res)) + Right rows -> do + columns <- fromIntegral <$> withForeignPtr res c_PQnfields + return ConnectionStats { + statsQueries = statsQueries cdStats + 1 + , statsRows = statsRows cdStats + rows + , statsValues = statsValues cdStats + (rows * columns) + , statsParams = statsParams cdStats + paramCount + } + -- Force evaluation of modified stats to squash a space leak. + stats' `seq` return (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, there is + -- something wrong - cancellation request didn't go through, yet the + -- query is still running? It's not clear how this might happen, but in + -- such case we must wait for its completion anyway. + Just err -> poll queryRunner >>= \case + Just _ -> return () + Nothing -> do + cancel queryRunner + rethrowWithContext sql . E.toException $ + HPQTypesError ("PQcancel failed: " ++ err) return (affected, st { dbLastQuery = SomeSQL sql , dbQueryResult = Just QueryResult { @@ -55,6 +84,8 @@ } }) where + withConnDo = withConnectionData (dbConnection st) "runQueryIO" + verifyResult :: Ptr PGconn -> Ptr PGresult -> IO (Either Int Int) verifyResult conn res = do -- works even if res is NULL diff --git a/src/Database/PostgreSQL/PQTypes/Notification.hs b/src/Database/PostgreSQL/PQTypes/Notification.hs --- a/src/Database/PostgreSQL/PQTypes/Notification.hs +++ b/src/Database/PostgreSQL/PQTypes/Notification.hs @@ -17,18 +17,22 @@ import Database.PostgreSQL.PQTypes.Utils -- | Start listening for notifications on a given channel. +{-# INLINABLE listen #-} listen :: MonadDB m => Channel -> m () listen (Channel chan) = runQuery_ $ "LISTEN" <+> chan -- | Stop listening for notifications on a given channel. +{-# INLINABLE unlisten #-} unlisten :: MonadDB m => Channel -> m () unlisten (Channel chan) = runQuery_ $ "UNLISTEN" <+> chan -- | Cancel all listener registrations for the current session. +{-# INLINABLE unlistenAll #-} unlistenAll :: MonadDB m => m () unlistenAll = runSQL_ "UNLISTEN *" -- | Generate a notification on a given channel. +{-# INLINABLE notify #-} notify :: MonadDB m => Channel -> Text -> m () notify (Channel chan) payload = runQuery_ $ rawSQL "SELECT pg_notify($1, $2)" (unRawSQL chan, payload) diff --git a/src/Database/PostgreSQL/PQTypes/Transaction.hs b/src/Database/PostgreSQL/PQTypes/Transaction.hs --- a/src/Database/PostgreSQL/PQTypes/Transaction.hs +++ b/src/Database/PostgreSQL/PQTypes/Transaction.hs @@ -36,6 +36,7 @@ -- provides something like \"nested transaction\". -- -- See +{-# INLINABLE withSavepoint #-} withSavepoint :: (MonadDB m, MonadMask m) => Savepoint -> m a -> m a withSavepoint (Savepoint savepoint) m = mask $ \restore -> do runQuery_ $ "SAVEPOINT" <+> savepoint @@ -56,18 +57,22 @@ -- 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. +{-# INLINABLE withTransaction #-} withTransaction :: (MonadDB m, MonadMask m) => m a -> m a withTransaction m = getTransactionSettings >>= flip withTransaction' m -- | Begin transaction using current transaction settings. +{-# INLINABLE begin #-} begin :: MonadDB m => m () begin = getTransactionSettings >>= begin' -- | Commit active transaction using current transaction settings. +{-# INLINABLE commit #-} commit :: MonadDB m => m () commit = getTransactionSettings >>= commit' -- | Rollback active transaction using current transaction settings. +{-# INLINABLE rollback #-} rollback :: MonadDB m => m () rollback = getTransactionSettings >>= rollback' @@ -76,6 +81,7 @@ -- | 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). +{-# INLINABLE withTransaction' #-} withTransaction' :: (MonadDB m, MonadMask m) => TransactionSettings -> m a -> m a withTransaction' ts m = mask $ \restore -> (`fix` 1) $ \loop n -> do @@ -106,6 +112,7 @@ guard $ f err n -- | Begin transaction using given transaction settings. +{-# INLINABLE begin' #-} begin' :: MonadDB m => TransactionSettings -> m () begin' ts = runSQL_ . mintercalate " " $ ["BEGIN", isolationLevel, permissions] where @@ -120,6 +127,7 @@ ReadWrite -> "READ WRITE" -- | Commit active transaction using given transaction settings. +{-# INLINABLE commit' #-} commit' :: MonadDB m => TransactionSettings -> m () commit' ts = do runSQL_ "COMMIT" @@ -127,6 +135,7 @@ begin' ts -- | Rollback active transaction using given transaction settings. +{-# INLINABLE rollback' #-} rollback' :: MonadDB m => TransactionSettings -> m () rollback' ts = do runSQL_ "ROLLBACK" diff --git a/src/Database/PostgreSQL/PQTypes/Utils.hs b/src/Database/PostgreSQL/PQTypes/Utils.hs --- a/src/Database/PostgreSQL/PQTypes/Utils.hs +++ b/src/Database/PostgreSQL/PQTypes/Utils.hs @@ -26,6 +26,7 @@ -- | When given 'DBException', throw it immediately. Otherwise -- wrap it in 'DBException' with the current query context first. +{-# INLINABLE throwDB #-} throwDB :: (Exception e, MonadDB m, MonadThrow m) => e -> m a throwDB e = case fromException $ toException e of Just (dbe::DBException) -> throwM dbe @@ -45,12 +46,14 @@ ---------------------------------------- -- | Specialization of 'runQuery' that discards the result. +{-# INLINABLE runQuery_ #-} runQuery_ :: (IsSQL sql, MonadDB m) => sql -> m () runQuery_ = void . runQuery -- | Specialization of 'runQuery' that checks whether affected/returned -- number of rows is in range [0, 1] and returns appropriate 'Bool' value. -- Otherwise, 'AffectedRowsMismatch' exception is thrown. +{-# INLINABLE runQuery01 #-} runQuery01 :: (IsSQL sql, MonadDB m, MonadThrow m) => sql -> m Bool runQuery01 sql = do n <- runQuery sql @@ -61,25 +64,28 @@ return $ n == 1 -- | Specialization of 'runQuery01' that discards the result. +{-# INLINABLE runQuery01_ #-} runQuery01_ :: (IsSQL sql, MonadDB m, MonadThrow m) => sql -> m () runQuery01_ = void . runQuery01 ---------------------------------------- -- | Specialization of 'runQuery' to 'SQL' type. +{-# INLINABLE runSQL #-} runSQL :: MonadDB m => SQL -> m Int runSQL = runQuery -- | Specialization of 'runQuery_' to 'SQL' type. +{-# INLINABLE runSQL_ #-} runSQL_ :: MonadDB m => SQL -> m () runSQL_ = runQuery_ -- | Specialization of 'runQuery01' to 'SQL' type. +{-# INLINABLE runSQL01 #-} runSQL01 :: (MonadDB m, MonadThrow m) => SQL -> m Bool runSQL01 = runQuery01 -- | Specialization of 'runQuery01_' to 'SQL' type. +{-# INLINABLE runSQL01_ #-} runSQL01_ :: (MonadDB m, MonadThrow m) => SQL -> m () runSQL01_ = runQuery01_ - ----------------------------------------- diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -20,6 +20,7 @@ import System.Environment import System.Exit import System.Random +import System.Timeout.Lifted import Test.Framework import Test.Framework.Providers.HUnit import Test.HUnit hiding (Test, assertEqual) @@ -223,6 +224,26 @@ ---------------------------------------- +queryInterruptionTest :: TestData -> Test +queryInterruptionTest td = testCase "Queries are interruptible" $ do + let sleep = "SELECT pg_sleep(2)" + ints = smconcat + [ "WITH RECURSIVE ints(n) AS" + , "( VALUES (1) UNION ALL SELECT n+1 FROM ints WHERE n < 5000000" + , ") SELECT n FROM ints" + ] + runTestEnv td tsNoTrans $ do + testQuery id sleep + testQuery id ints + runTestEnv td def $ do + testQuery (withSavepoint "ints") ints + testQuery (withSavepoint "sleep") sleep + where + testQuery m sql = timeout 500000 (m . runQuery_ $ rawSQL sql ()) >>= \case + Just _ -> liftBase $ do + assertFailure $ "Query" <+> T.unpack sql <+> "wasn't interrupted in time" + Nothing -> return () + autocommitTest :: TestData -> Test autocommitTest td = testCase "Autocommit mode works" . runTestEnv td tsNoTrans $ do let sint = Identity (1::Int32) @@ -355,6 +376,7 @@ , readOnlyTest td , savepointTest td , notifyTest td + , queryInterruptionTest td ---------------------------------------- , transactionTest td ReadCommitted , transactionTest td RepeatableRead