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 <QCoreApplication>
#include <QMap>
#include <QMutex>
#include <QSemaphore>
#include <QStack>
#include <QTime>
#include <QTimer>
#include <QThread>

#include "qgslogger.h"

#define CONN_POOL_MAX_CONCURRENT_CONNS      4
#define CONN_POOL_EXPIRATION_TIME           60    // in seconds


template <typename T>
class QgsConnectionPoolGroup
{
  public:

    static const int maxConcurrentConnections;

    struct Item
    {
      T c;
      QTime lastUsedTime;
    };

    QgsConnectionPoolGroup( const QString& ci )
        : connInfo( ci )
        , sem( CONN_POOL_MAX_CONCURRENT_CONNS )
        , expirationTimer( nullptr )
    {
    }

    ~QgsConnectionPoolGroup()
    {
      Q_FOREACH ( Item item, conns )
      {
        qgsConnectionPool_ConnectionDestroy( item.c );
      }
    }

    T acquire()
    {
      // we are going to acquire a resource - if no resource is available, we will block here
      sem.acquire();

      // quick (preferred) way - use cached connection
      {
        QMutexLocker locker( &connMutex );

        if ( !conns.isEmpty() )
        {
          Item i = conns.pop();
          if ( !qgsConnectionPool_ConnectionIsValid( i.c ) )
          {
            qgsConnectionPool_ConnectionDestroy( i.c );
            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" );
          }

          acquiredConns.append( i.c );

          return i.c;
        }
      }

      T c;
      qgsConnectionPool_ConnectionCreate( connInfo, c );
      if ( !c )
      {
        // we didn't get connection for some reason, so release the lock
        sem.release();
        return nullptr;
      }

      connMutex.lock();
      acquiredConns.append( c );
      connMutex.unlock();
      return c;
    }

    void release( T conn )
    {
      connMutex.lock();
      acquiredConns.removeAll( conn );
      if ( !qgsConnectionPool_ConnectionIsValid( conn ) )
      {
        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()
    {
      connMutex.lock();
      Q_FOREACH ( Item i, conns )
      {
        qgsConnectionPool_ConnectionDestroy( i.c );
      }
      conns.clear();
      Q_FOREACH ( T c, acquiredConns )
        qgsConnectionPool_InvalidateConnection( c );
      connMutex.unlock();
    }

  protected:

    void initTimer( QObject* parent )
    {
      expirationTimer = new QTimer( parent );
      expirationTimer->setInterval( CONN_POOL_EXPIRATION_TIME * 1000 );
      QObject::connect( expirationTimer, SIGNAL( timeout() ), parent, SLOT( 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;
};


template <typename T, typename T_Group>
class QgsConnectionPool
{
  public:

    typedef QMap<QString, T_Group*> T_Groups;

    virtual ~QgsConnectionPool()
    {
      mMutex.lock();
      Q_FOREACH ( T_Group* group, mGroups )
      {
        delete group;
      }
      mGroups.clear();
      mMutex.unlock();
    }

    T acquireConnection( const QString& connInfo )
    {
      mMutex.lock();
      typename T_Groups::iterator it = mGroups.find( connInfo );
      if ( it == mGroups.end() )
      {
        it = mGroups.insert( connInfo, new T_Group( connInfo ) );
      }
      T_Group* group = *it;
      mMutex.unlock();

      return group->acquire();
    }

    void releaseConnection( T conn )
    {
      mMutex.lock();
      typename T_Groups::iterator it = mGroups.find( qgsConnectionPool_ConnectionToName( conn ) );
      Q_ASSERT( it != mGroups.end() );
      T_Group* group = *it;
      mMutex.unlock();

      group->release( conn );
    }

    void invalidateConnections( const QString& connInfo )
    {
      mMutex.lock();
      if ( mGroups.contains( connInfo ) )
        mGroups[connInfo]->invalidateConnections();
      mMutex.unlock();
    }


  protected:
    T_Groups mGroups;
    QMutex mMutex;
};


#endif // QGSCONNECTIONPOOL_H