[РЕШЕНО] connection pool плюсы и минусы

[ballard]

Если вы блокируете текущую нить на время выполнения запроса, то не понятно для чего его выполнять в другой нитке. Ну и выполняйте этот запрос сразу в текущей? 

Изначально суть в том, как вы понимаете, чтобы открыть и держать открытыми несколько соединений с БД.
Если я блокирую какую-то одну нитку, то остальные остаются свободны для обращения из других потоков.

[Old]

Изначально суть в том, как вы понимаете, чтобы открыть и держать открытыми несколько соединений с БД.
Если я блокирую какую-то одну нитку, то остальные остаются свободны для обращения из других потоков.

Так откройте сразу 10 соединений и используйте их. Зачем вам рабочие нитки?

[ballard]

Так откройте сразу 10 соединений и используйте их. Зачем вам рабочие нитки?

Всмысле открывать сразу в вызывающих потоках? или вы что имеете в виду?
Идея такая, что у меня может быть к примеру 20 вызывающих при 5 открытых нитках в пуле, но каждых из этих 20-ти будет класть по задания последовательно, после выполнения предыдущего

[Old]

Всмысле открывать сразу в вызывающих потоках? или вы что имеете в виду?
Идея такая, что у меня может быть к примеру 20 вызывающих при 5 открытых нитках в пуле, но каждых из этих 20-ти будет класть по задания последовательно, после выполнения предыдущего

Пожалуйста, пулу все равно, дожидается ли клиент отработки предудущего запроса или нет.
Нужно ждать хорошо, не нужно - тоже хорошо.

[ballard]

Пожалуйста, пулу все равно, дожидается ли клиент отработки предудущего запроса или нет.
Нужно ждать хорошо, не нужно - тоже хорошо.

Ну да, но получается мне нужно как раз организовать промежуточный класс между вызывающими потоками и пулом, у которого и будет нужный мне метод, возвращающий QSqlRequest, Вы это имеете в виду?

[Old]

Ну да, но получается мне нужно как раз организовать промежуточный класс между вызывающими потоками и пулом, у которого и будет нужный мне метод, возвращающий QSqlRequest, Вы это имеете в виду?

Да, он будет запускать нитки-воркеры и нагружать их работой (через очередь), в ответ сигналя о результатах. Как у вас и было реализовано.
Добавьте в него очередь и все.

[ballard]

Да, он будет запускать нитки-воркеры и нагружать их работой (через очередь), в ответ сигналя о результатах. Как у вас и было реализовано.
Добавьте в него очередь и все.

Ok, спасибо, буду делать, потом поделюсь результатом

[ballard]

В общем что-то нет прогресса с очередью

Только сделал так выполнение запроса (QueryWrapper - это обернутый QSqlRequest):

Код

C++ (Qt)
QueryWrapper DBConnectionPool::execute(const QString &statement, const QMap<QString, QVariant> &values)
{
    QMutexLocker locker(&mutex);
    bool executed = false;
    QueryWrapper wrapper;
    while (!executed) {
        for (auto& connection : databaseConnections) {
            if (!connection->isBusy()) {
                wrapper = connection->executeSync(statement, values);
                executed = true;
                break;
            }
        }
    }
    return wrapper;
}
 
QueryWrapper DBConnectionPool::executeQuery(const QString &statement, const QMap<QString, QVariant> &values)
{
    wrappedFuture = QtConcurrent::run(this, &DBConnectionPool::execute, statement, values);
    wrappedFuture.waitForFinished();
    return(wrappedFuture.result());
}

Очередь непонятно, как сюда прикрутить: т.к., например future создается через qtconcurrent::run, который сразу исполняет запрос и, соответственно, в QList для некого воркера future не добавить.

Т.е. для синхронного запроса с очередью мне нужно что-то типа:

Код

C++ (Qt)
QueryWrapper DBConnectionPool::executeQuery(const QString &statement, const QMap<QString, QVariant> &values)
{
    Task task(statement, values);
    taskList.append(task);
    task.waitForResult();
    return task.result();
}

taskList это как раз qlist из которого некий consumer заберет запрос и выполнит

Пока что не понимаю, можно ли это сделать, и, если да, то как

[ballard]

Да, он будет запускать нитки-воркеры и нагружать их работой (через очередь), в ответ сигналя о результатах. Как у вас и было реализовано.
Добавьте в него очередь и все.

В общем-то пришла идея сделать эту функцию блокирующей, и тогда можно встроить работу с очередью. Либо по таймауту, либо цикл с проверкой выполнения запроса, либо через Qt::BlockingQueuedConnection, как Вы думаете, какой вариант лучше?

Код

C++ (Qt)
QueryWrapper DBConnectionPool::executeQuery(const QString &statement, const QMap<QString, QVariant> &values)
{
    Task task(statement, values);
    taskList.append(task);
 
    // block
 
    return task.result();
}

[ballard]

В общем сделал пока вариант с одним воркером...из одного потока пока все работает.
Блокирующая функция выполняется. Можете посмотреть момент с ::run() у QueryThread и по Tasking, все ли верно?

worker.h

Код

C++ (Qt)
class DatabaseWorker : public QObject
{
    Q_OBJECT
public:
    explicit DatabaseWorker(QObject* parent = 0);
    DatabaseWorker(const QString& threadId, QObject* parent = 0);
    void init(const QString& threadId);
 
private:
    QSqlDatabase m_database;    
 
signals:
    void results( const QList<QSqlRecord>& records );
 
public slots:
    void slotExecute( const QString& query );
};

worker.cpp

Код

C++ (Qt)
DatabaseWorker::DatabaseWorker(const QString& threadId, QObject* parent ) : QObject(parent)
{
    init(threadId);
}
 
void DatabaseWorker::init(const QString& threadId)
{
    QString dbThreadId;
    QString emptyStr = "";
    if (threadId == emptyStr) {
        dbThreadId = QString::number((qintptr)QThread::currentThreadId());
    } else {
        dbThreadId = threadId;
    }
    m_database = QSqlDatabase::addDatabase( "QPSQL", dbThreadId );
    m_database.setDatabaseName( "test" );
    m_database.setHostName( "localhost" );
    m_database.setUserName( "postgres" );
    m_database.setPassword( "postgres" );
    if ( !m_database.open() )
    {
        return;
    }
}
 
void DatabaseWorker::slotExecute( const QString& query )
{
    QList<QSqlRecord> recs;
    QSqlQuery sql( query, m_database );
    while ( sql.next() )
    {
        recs.push_back( sql.record() );
    }
    emit results( recs );
}

querythread.h

Код

C++ (Qt)
class QueryThread : public QThread
{
    Q_OBJECT
public:
    explicit QueryThread(QThread* parent = 0);
    QueryThread(Tasking* taskManager, QThread* parent = 0);
    void execute( const QString& query );
    ~QueryThread();
    void prepare();
 
protected:
    void run();
 
private:
    Tasking* taskManager;
    DatabaseWorker* m_worker;    
 
signals:
    void progress( const QString& msg );
    void ready(bool);
    void results( const QList<QSqlRecord>& records );
    void executefwd( const QString& query );
 
public slots:
    void getResults( const QList<QSqlRecord>& records );
};

querythread.cpp

Код

C++ (Qt)
QueryThread::QueryThread(Tasking* taskManager, QThread *parent) : QThread(parent)
{
    this->taskManager = taskManager;
 
}
 
QueryThread::~QueryThread()
{
    delete m_worker;
}
 
void QueryThread::prepare()
{
    emit ready(false);
    m_worker = new DatabaseWorker;
    connect( this, &QueryThread::executefwd, m_worker, &DatabaseWorker::slotExecute );
    qRegisterMetaType< QList<QSqlRecord> >( "QList<QSqlRecord>" );
    connect( m_worker, &DatabaseWorker::results, this, &QueryThread::getResults );
    emit ready(true);
}
 
void QueryThread::execute( const QString& query )
{
    emit executefwd( query );
}
 
void QueryThread::getResults( const QList<QSqlRecord>& records )
{
    emit results( records );    
}
 
void QueryThread::run()
{
    forever {
        execute(taskManager->getTask());
    }
}

tasking.h

Код

C++ (Qt)
class Tasking : public QObject
{
    Q_OBJECT
 
public:
    explicit Tasking(QObject *parent = 0);
    QMutex mutex;
    QWaitCondition cond;
    QString getTask();
    void addTask(const QString &task);
    QList<QString> taskList;
 
signals:
 
public slots:
};

tasking.cpp

Код

C++ (Qt)
void Tasking::addTask(const QString &task)
{
    QMutexLocker locker( &mutex );
    taskList.append( task );
    cond.wakeAll();
}
 
QString Tasking::getTask()
{
    QMutexLocker locker( &mutex );
 
    while( taskList.empty() ) {
        cond.wait( &mutex );
    }
 
    return taskList.takeFirst();
}

pool.h

Код

C++ (Qt)
class WorkerPool : public QObject
{
    Q_OBJECT
public:
    explicit WorkerPool(QObject *parent = 0);
    QList<QSqlRecord> executeRequest( const QString& request );
 
private:
    QueryThread *querythread;
 
    bool isResultsReady = false;
    QList<QSqlRecord> recs;
    void saveResults(const QList<QSqlRecord> &recs);
 
    Tasking* taskManager;
 
 
signals:
    void requestDone();

pool.cpp

Код

C++ (Qt)
WorkerPool::WorkerPool(QObject *parent) : QObject(parent)
{
    taskManager = new Tasking;
    querythread = new QueryThread(taskManager);
    connect(querythread, &QueryThread::results, this, &WorkerPool::saveResults );
    querythread->prepare();
    querythread->start();
}
 
QList<QSqlRecord> WorkerPool::executeRequest(const QString &request)
{
    isResultsReady = false;
    taskManager->addTask(request);
 
    QEventLoop loop;
    connect(this, &WorkerPool::requestDone, &loop, &QEventLoop::quit);
    loop.exec();
 
    return recs;
}
 
void WorkerPool::saveResults(const QList<QSqlRecord> &recs)
{
    this->recs = recs;
    isResultsReady = true;
    emit requestDone();
}

[ballard]

Теперь надо еще продумать, как идентифицировать сигналы от разных воркеров, при обращении из разных потоков

[ballard]

А результат можно возвращать с помощью сигнала, добавив каждому запросу некий ID.

Код

C++ (Qt)
public:
    ID    executeQuery( const QString &query );    // Возвращает ID запроса
 
signals:
    void    queryFinished( ID id, const QSqlQuery &result );    // Испускается после выполнения запроса
 

Обновил пул для работы с несколькими воркерами, можете посоветовать, как вернуть данные именно от того воркера, которому отправлен запрос?

Код

C++ (Qt)
WorkerPool::WorkerPool(const int &workersCount, QObject *parent) : QObject(parent)
{
    taskManager = new Tasking;
    for (auto i=0; i<workersCount; i++) {
        QueryThread* thread = new QueryThread(taskManager, "worker@" + QString::number(i));
        connect(thread, &QueryThread::results, this, &WorkerPool::saveResults);
        workersList << thread;
    }
 
    foreach (QueryThread* worker, workersList) {
        worker->prepare();
        worker->start();
    }
}
 
QList<QSqlRecord> WorkerPool::executeRequest(const QString &request)
{
    isResultsReady = false;
    taskManager->addTask(request);
 
    QEventLoop loop;
    connect(this, &WorkerPool::requestDone, &loop, &QEventLoop::quit);
    loop.exec();
 
    return recs;
}
 
void WorkerPool::saveResults(const QList<QSqlRecord> &recs)
{
    this->recs = recs;
    isResultsReady = true;
    emit requestDone();
}

[ballard]

В общем добавил пока сортрировку следующим образом, записывая запрос , блокируя мьютексом и проверяя потом на совпадение, но с таким способом пул зависает через несколько запросов, которые производятся из разных потоков, похоже из-за мьютекса возникает deadlock, без блокирования мьютексом запросы идут нормально, но, естественно порядок не совпадает...что тут можно сделать? :

Код

C++ (Qt)
WorkerPool::WorkerPool(const int &workersCount, QObject *parent) : QObject(parent)
{
    taskManager = new Tasking;
    for (auto i=0; i<workersCount; i++) {
        QueryThread* thread = new QueryThread(taskManager, "worker@" + QString::number(i));
        connect(thread, &QueryThread::results, this, &WorkerPool::saveResults);
        workersList << thread;
    }
 
    foreach (QueryThread* worker, workersList) {
        worker->prepare();
        worker->start();
    }
}
 
QList<QSqlRecord> WorkerPool::executeRequest(const QString &request)
{
    isResultsReady = false;
    taskManager->addTask(request);
 
    QMutexLocker locker(&mutex);
    lastRequest = request;
 
    QEventLoop loop;
    connect(this, &WorkerPool::requestDone, &loop, &QEventLoop::quit);
    loop.exec();
 
    return recs;
}
 
void WorkerPool::saveResults(const QList<QSqlRecord> &recs, const QString &request)
{
    if (request == lastRequest) {
        this->recs = recs;
        isResultsReady = true;
        emit requestDone();
    }
}

[ballard]

все, вроде бы победил: сделал мапу с результатами

Код

C++ (Qt)
QList<QSqlRecord> WorkerPool::executeRequest(const QString &request)
{
    isResultsReady = false;
 
    QMutexLocker locker(&mutex);
    taskManager->addTask(request);
 
    bool completed = false;
 
    while (!completed) {
        if (resultsMap.contains(request)) {
            completed = true;
        }
    }
    return resultsMap.take(request);
}
 
void WorkerPool::saveResults(const QList<QSqlRecord> &recs, const QString &request)
{
    resultsMap[request] = recs;
    isResultsReady = true;
    emit requestDone();
}