Skip to content
Permalink
Browse files

Fix libpq access from different threads

This is a forward port from 05f949b, following PR #8700.

In addition, two other fixes are added (that will be backported to
3.4):
- PQexecNR, openCursor and closeCursor are protected
- uniqueCursorName is also protected (I stumbled accross a bug where
the cursor name was reused by two different threads)
  • Loading branch information
Hugo Mercier
Hugo Mercier committed Jan 14, 2019
1 parent c263750 commit 648672d7dba0d405c2924cc8e14a2fca0690e73e
@@ -211,6 +211,7 @@ QgsPostgresConn::QgsPostgresConn( const QString &conninfo, bool readOnly, bool s
, mNextCursorId( 0 )
, mShared( shared )
, mTransaction( transaction )
, mLock( QMutex::Recursive )
{
QgsDebugMsg( QStringLiteral( "New PostgreSQL connection for " ) + conninfo );

@@ -1065,40 +1066,46 @@ PGresult *QgsPostgresConn::PQexec( const QString &query, bool logError ) const
}

QgsDebugMsgLevel( QStringLiteral( "Executing SQL: %1" ).arg( query ), 3 );
PGresult *res = ::PQexec( mConn, query.toUtf8() );

if ( res )
PGresult *res = nullptr;
{
int errorStatus = PQresultStatus( res );
if ( errorStatus != PGRES_COMMAND_OK && errorStatus != PGRES_TUPLES_OK )
QMutexLocker locker( &mLock );
res = ::PQexec( mConn, query.toUtf8() );

if ( res )
{
if ( logError )
int errorStatus = PQresultStatus( res );
if ( errorStatus != PGRES_COMMAND_OK && errorStatus != PGRES_TUPLES_OK )
{
QgsMessageLog::logMessage( tr( "Erroneous query: %1 returned %2 [%3]" )
.arg( query ).arg( errorStatus ).arg( PQresultErrorMessage( res ) ),
tr( "PostGIS" ) );
}
else
{
QgsDebugMsg( QStringLiteral( "Not logged erroneous query: %1 returned %2 [%3]" )
.arg( query ).arg( errorStatus ).arg( PQresultErrorMessage( res ) ) );
if ( logError )
{
QgsMessageLog::logMessage( tr( "Erroneous query: %1 returned %2 [%3]" )
.arg( query ).arg( errorStatus ).arg( PQresultErrorMessage( res ) ),
tr( "PostGIS" ) );
}
else
{
QgsDebugMsg( QStringLiteral( "Not logged erroneous query: %1 returned %2 [%3]" )
.arg( query ).arg( errorStatus ).arg( PQresultErrorMessage( res ) ) );
}
}
}
}
else if ( logError )
{
QgsMessageLog::logMessage( tr( "Query failed: %1\nError: no result buffer" ).arg( query ), tr( "PostGIS" ) );
}
else
{
QgsDebugMsg( QStringLiteral( "Not logged query failed: %1\nError: no result buffer" ).arg( query ) );
else if ( logError )
{
QgsMessageLog::logMessage( tr( "Query failed: %1\nError: no result buffer" ).arg( query ), tr( "PostGIS" ) );
}
else
{
QgsDebugMsg( QStringLiteral( "Not logged query failed: %1\nError: no result buffer" ).arg( query ) );
}
}

return res;
}

bool QgsPostgresConn::openCursor( const QString &cursorName, const QString &sql )
{
QMutexLocker locker( &mLock ); // to protect access to mOpenCursors

if ( mOpenCursors++ == 0 && !mTransaction )
{
QgsDebugMsgLevel( QStringLiteral( "Starting read-only transaction: %1" ).arg( mPostgresqlVersion ), 4 );
@@ -1114,6 +1121,8 @@ bool QgsPostgresConn::openCursor( const QString &cursorName, const QString &sql

bool QgsPostgresConn::closeCursor( const QString &cursorName )
{
QMutexLocker locker( &mLock ); // to protect access to mOpenCursors

if ( !PQexecNR( QStringLiteral( "CLOSE %1" ).arg( cursorName ) ) )
return false;

@@ -1128,11 +1137,14 @@ bool QgsPostgresConn::closeCursor( const QString &cursorName )

QString QgsPostgresConn::uniqueCursorName()
{
QMutexLocker locker( &mLock ); // to protect access to mNextCursorId
return QStringLiteral( "qgis_%1" ).arg( ++mNextCursorId );
}

bool QgsPostgresConn::PQexecNR( const QString &query, bool retry )
{
QMutexLocker locker( &mLock ); // to protect access to mOpenCursors

QgsPostgresResult res( PQexec( query, false ) );

ExecStatusType errorStatus = res.PQresultStatus();
@@ -1194,11 +1206,15 @@ PGresult *QgsPostgresConn::PQgetResult()

PGresult *QgsPostgresConn::PQprepare( const QString &stmtName, const QString &query, int nParams, const Oid *paramTypes )
{
QMutexLocker locker( &mLock );

return ::PQprepare( mConn, stmtName.toUtf8(), query.toUtf8(), nParams, paramTypes );
}

PGresult *QgsPostgresConn::PQexecPrepared( const QString &stmtName, const QStringList &params )
{
QMutexLocker locker( &mLock );

const char **param = new const char *[ params.size()];
QList<QByteArray> qparam;

@@ -1222,19 +1238,25 @@ PGresult *QgsPostgresConn::PQexecPrepared( const QString &stmtName, const QStrin

void QgsPostgresConn::PQfinish()
{
QMutexLocker locker( &mLock );

Q_ASSERT( mConn );
::PQfinish( mConn );
mConn = nullptr;
}

int QgsPostgresConn::PQstatus() const
{
QMutexLocker locker( &mLock );

Q_ASSERT( mConn );
return ::PQstatus( mConn );
}

QString QgsPostgresConn::PQerrorMessage() const
{
QMutexLocker locker( &mLock );

Q_ASSERT( mConn );
return QString::fromUtf8( ::PQerrorMessage( mConn ) );
}
@@ -244,16 +244,26 @@ class QgsPostgresConn : public QObject
// libpq wrapper
//

// run a query and check for errors
// run a query and check for errors, thread-safe
PGresult *PQexec( const QString &query, bool logError = true ) const;
void PQfinish();
QString PQerrorMessage() const;
int PQsendQuery( const QString &query );
int PQstatus() const;
PGresult *PQgetResult();
PGresult *PQprepare( const QString &stmtName, const QString &query, int nParams, const Oid *paramTypes );
PGresult *PQexecPrepared( const QString &stmtName, const QStringList &params );

/**
* PQsendQuery is used for asynchronous queries (with PQgetResult)
* Thread safety must be ensured by the caller by calling QgsPostgresConn::lock() and QgsPostgresConn::unlock()
*/
int PQsendQuery( const QString &query );

/**
* PQgetResult is used for asynchronous queries (with PQsendQuery)
* Thread safety must be ensured by the caller by calling QgsPostgresConn::lock() and QgsPostgresConn::unlock()
*/
PGresult *PQgetResult();

bool begin();
bool commit();
bool rollback();
@@ -425,7 +435,7 @@ class QgsPostgresConn : public QObject

bool mTransaction;

QMutex mLock;
mutable QMutex mLock;
};

// clazy:excludeall=qstring-allocations
@@ -386,9 +386,7 @@ bool QgsPostgresFeatureIterator::rewind()

// move cursor to first record

lock();
mConn->PQexecNR( QStringLiteral( "move absolute 0 in %1" ).arg( mCursorName ) );
unlock();
mFeatureQueue.clear();
mFetched = 0;
mLastFetch = false;
@@ -401,9 +399,7 @@ bool QgsPostgresFeatureIterator::close()
if ( !mConn )
return false;

lock();
mConn->closeCursor( mCursorName );
unlock();

if ( !mIsTransactionConnection )
{
@@ -651,17 +647,14 @@ bool QgsPostgresFeatureIterator::declareCursor( const QString &whereClause, long
if ( !orderBy.isEmpty() )
query += QStringLiteral( " ORDER BY %1 " ).arg( orderBy );

lock();
if ( !mConn->openCursor( mCursorName, query ) )
{
unlock();
// reloading the fields might help next time around
// TODO how to cleanly force reload of fields? P->loadFields();
if ( closeOnFail )
close();
return false;
}
unlock();

mLastFetch = false;
return true;
@@ -4095,9 +4095,7 @@ QgsCoordinateReferenceSystem QgsPostgresProvider::crs() const
else
{
QgsPostgresConn *conn = connectionRO();
conn->lock();
QgsPostgresResult result( conn->PQexec( QStringLiteral( "SELECT proj4text FROM spatial_ref_sys WHERE srid=%1" ).arg( srid ) ) );
conn->unlock();
if ( result.PQresultStatus() == PGRES_TUPLES_OK )
{
srs = QgsCoordinateReferenceSystem::fromProj4( result.PQgetvalue( 0, 0 ) );
@@ -71,9 +71,7 @@ bool QgsPostgresTransaction::executeSql( const QString &sql, QString &errorMsg,
}

QgsDebugMsg( QStringLiteral( "Transaction sql: %1" ).arg( sql ) );
mConn->lock();
QgsPostgresResult r( mConn->PQexec( sql, true ) );
mConn->unlock();
if ( r.PQresultStatus() == PGRES_BAD_RESPONSE ||
r.PQresultStatus() == PGRES_FATAL_ERROR )
{

0 comments on commit 648672d

Please sign in to comment.
You can’t perform that action at this time.