Branch data     Line data    Source code

       1                 :            : /***************************************************************************
       2                 :            :     qgsconnectionpool.h
       3                 :            :     ---------------------
       4                 :            :     begin                : February 2014
       5                 :            :     copyright            : (C) 2014 by Martin Dobias
       6                 :            :     email                : wonder dot sk at gmail dot com
       7                 :            :  ***************************************************************************
       8                 :            :  *                                                                         *
       9                 :            :  *   This program is free software; you can redistribute it and/or modify  *
      10                 :            :  *   it under the terms of the GNU General Public License as published by  *
      11                 :            :  *   the Free Software Foundation; either version 2 of the License, or     *
      12                 :            :  *   (at your option) any later version.                                   *
      13                 :            :  *                                                                         *
      14                 :            :  ***************************************************************************/
      15                 :            : 
      16                 :            : #ifndef QGSCONNECTIONPOOL_H
      17                 :            : #define QGSCONNECTIONPOOL_H
      18                 :            : 
      19                 :            : #define SIP_NO_FILE
      20                 :            : 
      21                 :            : #include "qgis.h"
      22                 :            : #include "qgsapplication.h"
      23                 :            : #include <QCoreApplication>
      24                 :            : #include <QMap>
      25                 :            : #include <QMutex>
      26                 :            : #include <QSemaphore>
      27                 :            : #include <QStack>
      28                 :            : #include <QTime>
      29                 :            : #include <QTimer>
      30                 :            : #include <QThread>
      31                 :            : 
      32                 :            : 
      33                 :            : #define CONN_POOL_EXPIRATION_TIME           60    // in seconds
      34                 :            : #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
      35                 :            : 
      36                 :            : 
      37                 :            : /**
      38                 :            :  * \ingroup core
      39                 :            :  * \brief Template that stores data related to a connection to a single server or datasource.
      40                 :            :  *
      41                 :            :  * It is assumed that following functions exist:
      42                 :            :  *
      43                 :            :  * - void qgsConnectionPool_ConnectionCreate(QString name, T& c)  ... create a new connection
      44                 :            :  * - void qgsConnectionPool_ConnectionDestroy(T c)                ... destroy the connection
      45                 :            :  * - QString qgsConnectionPool_ConnectionToName(T c)              ... lookup connection's name (path)
      46                 :            :  * - void qgsConnectionPool_InvalidateConnection(T c)             ... flag a connection as invalid
      47                 :            :  * - bool qgsConnectionPool_ConnectionIsValid(T c)                ... return whether a connection is valid
      48                 :            :  *
      49                 :            :  * Because of issues with templates and QObject's signals and slots, this class only provides helper functions for QObject-related
      50                 :            :  * functionality - the place which uses the template is resonsible for:
      51                 :            :  *
      52                 :            :  * - being derived from QObject
      53                 :            :  * - calling initTimer( this ) in constructor
      54                 :            :  * - having handleConnectionExpired() slot that calls onConnectionExpired()
      55                 :            :  * - having startExpirationTimer(), stopExpirationTimer() slots to start/stop the expiration timer
      56                 :            :  *
      57                 :            :  * For an example on how to use the template class, have a look at the implementation in Postgres/SpatiaLite providers.
      58                 :            :  * \note not available in Python bindings
      59                 :            :  */
      60                 :            : template <typename T>
      61                 :            : class QgsConnectionPoolGroup
      62                 :            : {
      63                 :            :   public:
      64                 :            : 
      65                 :         92 :     struct Item
      66                 :            :     {
      67                 :            :       T c;
      68                 :            :       QTime lastUsedTime;
      69                 :            :     };
      70                 :            : 
      71                 :         64 :     QgsConnectionPoolGroup( const QString &ci )
      72                 :         64 :       : connInfo( ci )
      73                 :         64 :       , sem( QgsApplication::instance()->maxConcurrentConnectionsPerPool() + CONN_POOL_SPARE_CONNECTIONS )
      74                 :            :     {
      75                 :         64 :     }
      76                 :            : 
      77                 :         64 :     ~QgsConnectionPoolGroup()
      78                 :            :     {
      79                 :         67 :       for ( const Item &item : std::as_const( conns ) )
      80                 :            :       {
      81                 :          3 :         qgsConnectionPool_ConnectionDestroy( item.c );
      82                 :            :       }
      83                 :         64 :     }
      84                 :            : 
      85                 :            :     //! QgsConnectionPoolGroup cannot be copied
      86                 :            :     QgsConnectionPoolGroup( const QgsConnectionPoolGroup &other ) = delete;
      87                 :            :     //! QgsConnectionPoolGroup cannot be copied
      88                 :            :     QgsConnectionPoolGroup &operator=( const QgsConnectionPoolGroup &other ) = delete;
      89                 :            : 
      90                 :            :     /**
      91                 :            :      * Try to acquire a connection for a maximum of \a timeout milliseconds.
      92                 :            :      * If \a timeout is a negative value the calling thread will be blocked
      93                 :            :      * until a connection becomes available. This is the default behavior.
      94                 :            :      *
      95                 :            :      * \returns initialized connection or NULLPTR if unsuccessful
      96                 :            :      */
      97                 :         92 :     T acquire( int timeout, bool requestMayBeNested )
      98                 :            :     {
      99                 :         92 :       const int requiredFreeConnectionCount = requestMayBeNested ? 1 : 3;
     100                 :            :       // we are going to acquire a resource - if no resource is available, we will block here
     101                 :         92 :       if ( timeout >= 0 )
     102                 :            :       {
     103                 :          0 :         if ( !sem.tryAcquire( requiredFreeConnectionCount, timeout ) )
     104                 :          0 :           return nullptr;
     105                 :          0 :       }
     106                 :            :       else
     107                 :            :       {
     108                 :            :         // we should still be able to use tryAcquire with a negative timeout here, but
     109                 :            :         // tryAcquire is broken on Qt > 5.8 with negative timeouts - see
     110                 :            :         // https://bugreports.qt.io/browse/QTBUG-64413
     111                 :            :         // https://lists.osgeo.org/pipermail/qgis-developer/2017-November/050456.html
     112                 :         92 :         sem.acquire( requiredFreeConnectionCount );
     113                 :            :       }
     114                 :         92 :       sem.release( requiredFreeConnectionCount - 1 );
     115                 :            : 
     116                 :            :       // quick (preferred) way - use cached connection
     117                 :            :       {
     118                 :         92 :         QMutexLocker locker( &connMutex );
     119                 :            : 
     120                 :         92 :         if ( !conns.isEmpty() )
     121                 :            :         {
     122                 :          5 :           Item i = conns.pop();
     123                 :          5 :           if ( !qgsConnectionPool_ConnectionIsValid( i.c ) )
     124                 :            :           {
     125                 :          0 :             qgsConnectionPool_ConnectionDestroy( i.c );
     126                 :          0 :             qgsConnectionPool_ConnectionCreate( connInfo, i.c );
     127                 :          0 :           }
     128                 :            : 
     129                 :            : 
     130                 :            :           // no need to run if nothing can expire
     131                 :          5 :           if ( conns.isEmpty() )
     132                 :            :           {
     133                 :            :             // will call the slot directly or queue the call (if the object lives in a different thread)
     134                 :          5 :             QMetaObject::invokeMethod( expirationTimer->parent(), "stopExpirationTimer" );
     135                 :          5 :           }
     136                 :            : 
     137                 :          5 :           acquiredConns.append( i.c );
     138                 :            : 
     139                 :          5 :           return i.c;
     140                 :            :         }
     141                 :         92 :       }
     142                 :            : 
     143                 :            :       T c;
     144                 :         87 :       qgsConnectionPool_ConnectionCreate( connInfo, c );
     145                 :         87 :       if ( !c )
     146                 :            :       {
     147                 :            :         // we didn't get connection for some reason, so release the lock
     148                 :          0 :         sem.release();
     149                 :          0 :         return nullptr;
     150                 :            :       }
     151                 :            : 
     152                 :         87 :       connMutex.lock();
     153                 :         87 :       acquiredConns.append( c );
     154                 :         87 :       connMutex.unlock();
     155                 :         87 :       return c;
     156                 :         92 :     }
     157                 :            : 
     158                 :         92 :     void release( T conn )
     159                 :            :     {
     160                 :         92 :       connMutex.lock();
     161                 :         92 :       acquiredConns.removeAll( conn );
     162                 :         92 :       if ( !qgsConnectionPool_ConnectionIsValid( conn ) )
     163                 :            :       {
     164                 :          0 :         qgsConnectionPool_ConnectionDestroy( conn );
     165                 :          0 :       }
     166                 :            :       else
     167                 :            :       {
     168                 :         92 :         Item i;
     169                 :         92 :         i.c = conn;
     170                 :         92 :         i.lastUsedTime = QTime::currentTime();
     171                 :         92 :         conns.push( i );
     172                 :            : 
     173                 :         92 :         if ( !expirationTimer->isActive() )
     174                 :            :         {
     175                 :            :           // will call the slot directly or queue the call (if the object lives in a different thread)
     176                 :         69 :           QMetaObject::invokeMethod( expirationTimer->parent(), "startExpirationTimer" );
     177                 :         69 :         }
     178                 :            :       }
     179                 :            : 
     180                 :         92 :       connMutex.unlock();
     181                 :            : 
     182                 :         92 :       sem.release(); // this can unlock a thread waiting in acquire()
     183                 :         92 :     }
     184                 :            : 
     185                 :        241 :     void invalidateConnections()
     186                 :            :     {
     187                 :        241 :       connMutex.lock();
     188                 :        325 :       for ( const Item &i : std::as_const( conns ) )
     189                 :            :       {
     190                 :         84 :         qgsConnectionPool_ConnectionDestroy( i.c );
     191                 :            :       }
     192                 :        241 :       conns.clear();
     193                 :        241 :       for ( T c : std::as_const( acquiredConns ) )
     194                 :          0 :         qgsConnectionPool_InvalidateConnection( c );
     195                 :        241 :       connMutex.unlock();
     196                 :        241 :     }
     197                 :            : 
     198                 :            :   protected:
     199                 :            : 
     200                 :         64 :     void initTimer( QObject *parent )
     201                 :            :     {
     202                 :         64 :       expirationTimer = new QTimer( parent );
     203                 :         64 :       expirationTimer->setInterval( CONN_POOL_EXPIRATION_TIME * 1000 );
     204                 :         64 :       QObject::connect( expirationTimer, SIGNAL( timeout() ), parent, SLOT( handleConnectionExpired() ) );
     205                 :            : 
     206                 :            :       // just to make sure the object belongs to main thread and thus will get events
     207                 :         64 :       if ( qApp )
     208                 :         64 :         parent->moveToThread( qApp->thread() );
     209                 :         64 :     }
     210                 :            : 
     211                 :          0 :     void onConnectionExpired()
     212                 :            :     {
     213                 :          0 :       connMutex.lock();
     214                 :            : 
     215                 :          0 :       QTime now = QTime::currentTime();
     216                 :            : 
     217                 :            :       // what connections have expired?
     218                 :          0 :       QList<int> toDelete;
     219                 :          0 :       for ( int i = 0; i < conns.count(); ++i )
     220                 :            :       {
     221                 :          0 :         if ( conns.at( i ).lastUsedTime.secsTo( now ) >= CONN_POOL_EXPIRATION_TIME )
     222                 :          0 :           toDelete.append( i );
     223                 :          0 :       }
     224                 :            : 
     225                 :            :       // delete expired connections
     226                 :          0 :       for ( int j = toDelete.count() - 1; j >= 0; --j )
     227                 :            :       {
     228                 :          0 :         int index = toDelete[j];
     229                 :          0 :         qgsConnectionPool_ConnectionDestroy( conns[index].c );
     230                 :          0 :         conns.remove( index );
     231                 :          0 :       }
     232                 :            : 
     233                 :          0 :       if ( conns.isEmpty() )
     234                 :          0 :         expirationTimer->stop();
     235                 :            : 
     236                 :          0 :       connMutex.unlock();
     237                 :          0 :     }
     238                 :            : 
     239                 :            :   protected:
     240                 :            : 
     241                 :            :     QString connInfo;
     242                 :            :     QStack<Item> conns;
     243                 :            :     QList<T> acquiredConns;
     244                 :            :     QMutex connMutex;
     245                 :            :     QSemaphore sem;
     246                 :         64 :     QTimer *expirationTimer = nullptr;
     247                 :            : 
     248                 :            : };
     249                 :            : 
     250                 :            : 
     251                 :            : /**
     252                 :            :  * \ingroup core
     253                 :            :  * \brief Template class responsible for keeping a pool of open connections.
     254                 :            :  *
     255                 :            :  * This is desired to avoid the overhead of creation of new connection every time.
     256                 :            :  *
     257                 :            :  * The methods are thread safe.
     258                 :            :  *
     259                 :            :  * The connection pool has a limit on maximum number of concurrent connections
     260                 :            :  * (per server), once the limit is reached, the acquireConnection() function
     261                 :            :  * will block. All connections that have been acquired must be then released
     262                 :            :  * with releaseConnection() function.
     263                 :            :  *
     264                 :            :  * When the connections are not used for some time, they will get closed automatically
     265                 :            :  * to save resources.
     266                 :            :  * \note not available in Python bindings
     267                 :            :  */
     268                 :            : template <typename T, typename T_Group>
     269                 :          1 : class QgsConnectionPool
     270                 :            : {
     271                 :            :   public:
     272                 :            : 
     273                 :            :     typedef QMap<QString, T_Group *> T_Groups;
     274                 :            : 
     275                 :          1 :     virtual ~QgsConnectionPool()
     276                 :          1 :     {
     277                 :          1 :       mMutex.lock();
     278                 :          4 :       for ( T_Group *group : std::as_const( mGroups ) )
     279                 :            :       {
     280                 :          3 :         delete group;
     281                 :            :       }
     282                 :          1 :       mGroups.clear();
     283                 :          1 :       mMutex.unlock();
     284                 :          1 :     }
     285                 :            : 
     286                 :            :     /**
     287                 :            :      * Try to acquire a connection for a maximum of \a timeout milliseconds.
     288                 :            :      * If \a timeout is a negative value the calling thread will be blocked
     289                 :            :      * until a connection becomes available. This is the default behavior.
     290                 :            :      *
     291                 :            :      * \returns initialized connection or NULLPTR if unsuccessful
     292                 :            :      */
     293                 :         92 :     T acquireConnection( const QString &connInfo, int timeout = -1, bool requestMayBeNested = false )
     294                 :            :     {
     295                 :         92 :       mMutex.lock();
     296                 :         92 :       typename T_Groups::iterator it = mGroups.find( connInfo );
     297                 :         92 :       if ( it == mGroups.end() )
     298                 :            :       {
     299                 :          0 :         it = mGroups.insert( connInfo, new T_Group( connInfo ) );
     300                 :          0 :       }
     301                 :         92 :       T_Group *group = *it;
     302                 :         92 :       mMutex.unlock();
     303                 :            : 
     304                 :         92 :       return group->acquire( timeout, requestMayBeNested );
     305                 :          0 :     }
     306                 :            : 
     307                 :            :     //! Release an existing connection so it will get back into the pool and can be reused
     308                 :         92 :     void releaseConnection( T conn )
     309                 :            :     {
     310                 :         92 :       mMutex.lock();
     311                 :         92 :       typename T_Groups::iterator it = mGroups.find( qgsConnectionPool_ConnectionToName( conn ) );
     312                 :            :       Q_ASSERT( it != mGroups.end() );
     313                 :         92 :       T_Group *group = *it;
     314                 :         92 :       mMutex.unlock();
     315                 :            : 
     316                 :         92 :       group->release( conn );
     317                 :         92 :     }
     318                 :            : 
     319                 :            :     /**
     320                 :            :      * Invalidates all connections to the specified resource.
     321                 :            :      * The internal state of certain handles (for instance OGR) are altered
     322                 :            :      * when a dataset is modified. Consquently, all open handles need to be
     323                 :            :      * invalidated when such datasets are changed to ensure the handles are
     324                 :            :      * refreshed. See the OGR provider for an example where this is needed.
     325                 :            :      */
     326                 :        305 :     void invalidateConnections( const QString &connInfo )
     327                 :            :     {
     328                 :        305 :       mMutex.lock();
     329                 :        305 :       if ( mGroups.contains( connInfo ) )
     330                 :        241 :         mGroups[connInfo]->invalidateConnections();
     331                 :        305 :       mMutex.unlock();
     332                 :        305 :     }
     333                 :            : 
     334                 :            : 
     335                 :            :   protected:
     336                 :            :     T_Groups mGroups;
     337                 :            :     QMutex mMutex;
     338                 :            : };
     339                 :            : 
     340                 :            : 
     341                 :            : #endif // QGSCONNECTIONPOOL_H