qgsconnectionpool.h

Go to the documentation of this file.

/***************************************************************************
    qgsconnectionpool.h
    ---------------------
    begin                : February 2014
    copyright            : (C) 2014 by Martin Dobias
    email                : wonder dot sk at gmail dot com
 ***************************************************************************
 *                                                                         *
 *   This program is free software; you can redistribute it and/or modify  *
 *   it under the terms of the GNU General Public License as published by  *
 *   the Free Software Foundation; either version 2 of the License, or     *
 *   (at your option) any later version.                                   *
 *                                                                         *
 ***************************************************************************/
 
#ifndef QGSCONNECTIONPOOL_H
#define QGSCONNECTIONPOOL_H
 
 
#include "qgis.h"
#include "qgsapplication.h"
#include "qgsfeedback.h"
#include "qgslogger.h"
 
#include <QCoreApplication>
#include <QElapsedTimer>
#include <QMap>
#include <QMutex>
#include <QSemaphore>
#include <QStack>
#include <QString>
#include <QThread>
#include <QTime>
#include <QTimer>
 
#define SIP_NO_FILE
 
using namespace Qt::StringLiterals;
 
#define CONN_POOL_EXPIRATION_TIME 60  // in seconds
#define CONN_POOL_SPARE_CONNECTIONS 2 // number of spare connections in case all the base connections are used but we have a nested request with the risk of a deadlock
 

template<typename T> class QgsConnectionPoolGroup
{
  public:
    struct Item
    {
        T c;
        QTime lastUsedTime;
    };

    QgsConnectionPoolGroup( const QString &ci )
      : connInfo( ci )
      , sem( QgsApplication::instance()->maxConcurrentConnectionsPerPool() + CONN_POOL_SPARE_CONNECTIONS )
    {}
 
    ~QgsConnectionPoolGroup()
    {
      QgsDebugMsgLevel( u"Destroying connection pool group"_s, 2 );
      for ( const Item &item : std::as_const( conns ) )
      {
        qgsConnectionPool_ConnectionDestroy( item.c );
      }
    }
 
    QgsConnectionPoolGroup( const QgsConnectionPoolGroup &other ) = delete;
    QgsConnectionPoolGroup &operator=( const QgsConnectionPoolGroup &other ) = delete;

    T acquire( int timeout, bool requestMayBeNested )
    {
      QgsDebugMsgLevel( u"Trying to acquire connection"_s, 2 );
      const int requiredFreeConnectionCount = requestMayBeNested ? 1 : 3;
      // we are going to acquire a resource - if no resource is available, we will block here
      if ( !sem.tryAcquire( requiredFreeConnectionCount, QDeadlineTimer( timeout ) ) )
      {
        QgsDebugMsgLevel( u"Failed to acquire semaphore"_s, 2 );
        return nullptr;
      }
 
      sem.release( requiredFreeConnectionCount - 1 );
 
      // quick (preferred) way - use cached connection
      {
        QMutexLocker locker( &connMutex );
 
        if ( !conns.isEmpty() )
        {
          QgsDebugMsgLevel( u"Trying to use existing connection"_s, 2 );
          Item i = conns.pop();
          if ( !qgsConnectionPool_ConnectionIsValid( i.c ) )
          {
            QgsDebugMsgLevel( u"Connection is not valid, destroying"_s, 2 );
            qgsConnectionPool_ConnectionDestroy( i.c );
            QgsDebugMsgLevel( u"Creating new connection"_s, 2 );
            qgsConnectionPool_ConnectionCreate( connInfo, i.c );
          }
 
 
          // no need to run if nothing can expire
          if ( conns.isEmpty() )
          {
            // will call the slot directly or queue the call (if the object lives in a different thread)
            QMetaObject::invokeMethod( expirationTimer->parent(), "stopExpirationTimer" );
          }
 
          QgsDebugMsgLevel( u"Acquired connection"_s, 2 );
          acquiredConns.append( i.c );
 
          return i.c;
        }
      }
 
      QgsDebugMsgLevel( u"Creating new connection"_s, 2 );
      T c;
      qgsConnectionPool_ConnectionCreate( connInfo, c );
      if ( !c )
      {
        // we didn't get connection for some reason, so release the lock
        sem.release();
        QgsDebugMsgLevel( u"Failed to create new connection"_s, 2 );
        return nullptr;
      }
 
      connMutex.lock();
      QgsDebugMsgLevel( u"Acquired connection with name: %1"_s.arg( qgsConnectionPool_ConnectionToName( c ) ), 2 );
      acquiredConns.append( c );
      connMutex.unlock();
      return c;
    }
 
    void release( T conn )
    {
      QgsDebugMsgLevel( u"Releasing connection"_s, 2 );
      connMutex.lock();
      acquiredConns.removeAll( conn );
      if ( !qgsConnectionPool_ConnectionIsValid( conn ) )
      {
        QgsDebugMsgLevel( u"Destroying invalid connection"_s, 2 );
        qgsConnectionPool_ConnectionDestroy( conn );
      }
      else
      {
        Item i;
        i.c = conn;
        i.lastUsedTime = QTime::currentTime();
        conns.push( i );
 
        if ( !expirationTimer->isActive() )
        {
          // will call the slot directly or queue the call (if the object lives in a different thread)
          QMetaObject::invokeMethod( expirationTimer->parent(), "startExpirationTimer" );
        }
      }
 
      connMutex.unlock();
 
      sem.release(); // this can unlock a thread waiting in acquire()
    }
 
    void invalidateConnections()
    {
      QgsDebugMsgLevel( u"Invalidating connections for group"_s, 2 );
      connMutex.lock();
      for ( const Item &i : std::as_const( conns ) )
      {
        qgsConnectionPool_ConnectionDestroy( i.c );
      }
      conns.clear();
      for ( T c : std::as_const( acquiredConns ) )
        qgsConnectionPool_InvalidateConnection( c );
      connMutex.unlock();
    }
 
  protected:
    template<typename U> void initTimer( U *parent )
    {
      expirationTimer = new QTimer( parent );
      expirationTimer->setInterval( CONN_POOL_EXPIRATION_TIME * 1000 );
      QObject::connect( expirationTimer, &QTimer::timeout, parent, &U::handleConnectionExpired );
 
      // just to make sure the object belongs to main thread and thus will get events
      if ( qApp )
        parent->moveToThread( qApp->thread() );
    }
 
    void onConnectionExpired()
    {
      connMutex.lock();
 
      QTime now = QTime::currentTime();
 
      // what connections have expired?
      QList<int> toDelete;
      for ( int i = 0; i < conns.count(); ++i )
      {
        if ( conns.at( i ).lastUsedTime.secsTo( now ) >= CONN_POOL_EXPIRATION_TIME )
          toDelete.append( i );
      }
 
      // delete expired connections
      for ( int j = toDelete.count() - 1; j >= 0; --j )
      {
        int index = toDelete[j];
        qgsConnectionPool_ConnectionDestroy( conns[index].c );
        conns.remove( index );
      }
 
      if ( conns.isEmpty() )
        expirationTimer->stop();
 
      connMutex.unlock();
    }
 
  protected:
    QString connInfo;
    QStack<Item> conns;
    QList<T> acquiredConns;
    QMutex connMutex;
    QSemaphore sem;
    QTimer *expirationTimer = nullptr;
};
 

template<typename T, typename T_Group> class QgsConnectionPool
{
  public:
    typedef QMap<QString, T_Group *> T_Groups;
 
    virtual ~QgsConnectionPool()
    {
      QgsDebugMsgLevel( u"Destroying connection pool"_s, 2 );
      mMutex.lock();
      for ( auto it = mGroups.constBegin(); it != mGroups.constEnd(); ++it )
      {
        QgsDebugMsgLevel( u"Destroying connection pool group with key %1"_s.arg( it.key() ), 2 );
        delete it.value();
      }
      QgsDebugMsgLevel( u"Connection pool groups destroyed"_s, 2 );
      mGroups.clear();
      mMutex.unlock();
    }

    T acquireConnection( const QString &connInfo, int timeout = -1, bool requestMayBeNested = false, QgsFeedback *feedback = nullptr )
    {
      QgsDebugMsgLevel( u"Trying to acquire connection for %1"_s.arg( connInfo ), 2 );
      mMutex.lock();
      typename T_Groups::iterator it = mGroups.find( connInfo );
      if ( it == mGroups.end() )
      {
        QgsDebugMsgLevel( u"Could not find existing group, adding new one"_s, 2 );
        it = mGroups.insert( connInfo, new T_Group( connInfo ) );
      }
      else
      {
        QgsDebugMsgLevel( u"Found existing group"_s, 2 );
      }
      T_Group *group = *it;
      mMutex.unlock();
 
      if ( feedback )
      {
        QElapsedTimer timer;
        timer.start();
 
        while ( !feedback->isCanceled() )
        {
          if ( T conn = group->acquire( 300, requestMayBeNested ) )
            return conn;
 
          if ( timeout > 0 && timer.elapsed() >= timeout )
            return nullptr;
        }
        return nullptr;
      }
      else
      {
        return group->acquire( timeout, requestMayBeNested );
      }
    }

    void releaseConnection( T conn )
    {
      mMutex.lock();
      const QString groupName = qgsConnectionPool_ConnectionToName( conn );
      QgsDebugMsgLevel( u"Releasing connection for %1"_s.arg( groupName ), 2 );
      typename T_Groups::iterator it = mGroups.find( groupName );
      Q_ASSERT( it != mGroups.end() );
      T_Group *group = *it;
      mMutex.unlock();
 
      QgsDebugMsgLevel( u"Group found, releasing..."_s, 2 );
      group->release( conn );
    }

    void invalidateConnections( const QString &connInfo )
    {
      QgsDebugMsgLevel( u"Invalidating connections for %1"_s.arg( connInfo ), 2 );
      mMutex.lock();
 
      auto it = mGroups.constFind( connInfo );
      if ( it != mGroups.constEnd() )
      {
        QgsDebugMsgLevel( u"Found group, invalidating..."_s, 2 );
        it.value()->invalidateConnections();
      }
      else
      {
        QgsDebugMsgLevel( u"Could not find matching group!"_s, 2 );
      }
      mMutex.unlock();
    }
 
 
  protected:
    T_Groups mGroups;
    QMutex mMutex;
};
 
 
#endif // QGSCONNECTIONPOOL_H