qthreadpool.cpp

Go to the documentation of this file.

// Copyright (C) 2016 The Qt Company Ltd.
// SPDX-License-Identifier: LicenseRef-Qt-Commercial OR LGPL-3.0-only OR GPL-2.0-only OR GPL-3.0-only
 
#include "qthreadpool.h"
#include "qthreadpool_p.h"
#include "qdeadlinetimer.h"
#include "qcoreapplication.h"
 
#include <QtCore/qpointer.h>
 
#include <algorithm>
#include <memory>
 
QT_BEGIN_NAMESPACE
 
using namespace Qt::StringLiterals;
 
/*
    QThread wrapper, provides synchronization against a ThreadPool
*/
class QThreadPoolThread : public QThread
{
    Q_OBJECT
public:
    QThreadPoolThread(QThreadPoolPrivate *manager);
    void run() override;
    void registerThreadInactive();
 
    QWaitCondition runnableReady;
    QThreadPoolPrivate *manager;
    QRunnable *runnable;
};
 
/*
    QThreadPool private class.
*/
 
 
QThreadPoolThread::QThreadPoolThread(QThreadPoolPrivate *manager)
    :manager(manager), runnable(nullptr)
{
    setStackSize(manager->stackSize);
}
 
/*
    \internal
*/
void QThreadPoolThread::run()
{
    QMutexLocker locker(&manager->mutex);
    for(;;) {
        QRunnable *r = runnable;
        runnable = nullptr;
 
        do {
            if (r) {
                // If autoDelete() is false, r might already be deleted after run(), so check status now.
                const bool del = r->autoDelete();
 
                // run the task
                locker.unlock();
#ifndef QT_NO_EXCEPTIONS
                try {
#endif
                    r->run();
#ifndef QT_NO_EXCEPTIONS
                } catch (...) {
                    qWarning("Qt Concurrent has caught an exception thrown from a worker thread.\n"
                             "This is not supported, exceptions thrown in worker threads must be\n"
                             "caught before control returns to Qt Concurrent.");
                    registerThreadInactive();
                    throw;
                }
#endif
 
                if (del)
                    delete r;
                locker.relock();
            }
 
            // if too many threads are active, stop working in this one
            if (manager->tooManyThreadsActive())
                break;
 
            // all work is done, time to wait for more
            if (manager->queue.isEmpty())
                break;
 
            QueuePage *page = manager->queue.constFirst();
            r = page->pop();
 
            if (page->isFinished()) {
                manager->queue.removeFirst();
                delete page;
            }
        } while (true);
 
        // this thread is about to be deleted, do not wait or expire
        if (!manager->allThreads.contains(this)) {
            registerThreadInactive();
            return;
        }
 
        // if too many threads are active, expire this thread
        if (manager->tooManyThreadsActive()) {
            manager->expiredThreads.enqueue(this);
            registerThreadInactive();
            return;
        }
        manager->waitingThreads.enqueue(this);
        registerThreadInactive();
        // wait for work, exiting after the expiry timeout is reached
        runnableReady.wait(locker.mutex(), QDeadlineTimer(manager->expiryTimeout));
        // this thread is about to be deleted, do not work or expire
        if (!manager->allThreads.contains(this)) {
            Q_ASSERT(manager->queue.isEmpty());
            return;
        }
        if (manager->waitingThreads.removeOne(this)) {
            manager->expiredThreads.enqueue(this);
            return;
        }
        ++manager->activeThreads;
    }
}
 
void QThreadPoolThread::registerThreadInactive()
{
    if (--manager->activeThreads == 0)
        manager->noActiveThreads.wakeAll();
}
 
 
/*
    \internal
*/
QThreadPoolPrivate:: QThreadPoolPrivate()
{ }
 
bool QThreadPoolPrivate::tryStart(QRunnable *task)
{
    Q_ASSERT(task != nullptr);
    if (allThreads.isEmpty()) {
        // always create at least one thread
        startThread(task);
        return true;
    }
 
    // can't do anything if we're over the limit
    if (areAllThreadsActive())
        return false;
 
    if (!waitingThreads.isEmpty()) {
        // recycle an available thread
        enqueueTask(task);
        waitingThreads.takeFirst()->runnableReady.wakeOne();
        return true;
    }
 
    if (!expiredThreads.isEmpty()) {
        // restart an expired thread
        QThreadPoolThread *thread = expiredThreads.dequeue();
        Q_ASSERT(thread->runnable == nullptr);
 
        ++activeThreads;
 
        thread->runnable = task;
 
        // Ensure that the thread has actually finished, otherwise the following
        // start() has no effect.
        thread->wait();
        Q_ASSERT(thread->isFinished());
        thread->start(threadPriority);
        return true;
    }
 
    // start a new thread
    startThread(task);
    return true;
}
 
inline bool comparePriority(int priority, const QueuePage *p)
{
    return p->priority() < priority;
}
 
void QThreadPoolPrivate::enqueueTask(QRunnable *runnable, int priority)
{
    Q_ASSERT(runnable != nullptr);
    for (QueuePage *page : std::as_const(queue)) {
        if (page->priority() == priority && !page->isFull()) {
            page->push(runnable);
            return;
        }
    }
    auto it = std::upper_bound(queue.constBegin(), queue.constEnd(), priority, comparePriority);
    queue.insert(std::distance(queue.constBegin(), it), new QueuePage(runnable, priority));
}
 
int QThreadPoolPrivate::activeThreadCount() const
{
    return (allThreads.size()
            - expiredThreads.size()
            - waitingThreads.size()
            + reservedThreads);
}
 
void QThreadPoolPrivate::tryToStartMoreThreads()
{
    // try to push tasks on the queue to any available threads
    while (!queue.isEmpty()) {
        QueuePage *page = queue.constFirst();
        if (!tryStart(page->first()))
            break;
 
        page->pop();
 
        if (page->isFinished()) {
            queue.removeFirst();
            delete page;
        }
    }
}
 
bool QThreadPoolPrivate::areAllThreadsActive() const
{
    const int activeThreadCount = this->activeThreadCount();
    return activeThreadCount >= maxThreadCount() && (activeThreadCount - reservedThreads) >= 1;
}
 
bool QThreadPoolPrivate::tooManyThreadsActive() const
{
    const int activeThreadCount = this->activeThreadCount();
    return activeThreadCount > maxThreadCount() && (activeThreadCount - reservedThreads) > 1;
}
 
void QThreadPoolPrivate::startThread(QRunnable *runnable)
{
    Q_ASSERT(runnable != nullptr);
    auto thread = std::make_unique<QThreadPoolThread>(this);
    if (objectName.isEmpty())
        objectName = u"Thread (pooled)"_s;
    thread->setObjectName(objectName);
    Q_ASSERT(!allThreads.contains(thread.get())); // if this assert hits, we have an ABA problem (deleted threads don't get removed here)
    allThreads.insert(thread.get());
    ++activeThreads;
 
    thread->runnable = runnable;
    thread.release()->start(threadPriority);
}
 
void QThreadPoolPrivate::reset()
{
    // move the contents of the set out so that we can iterate without the lock
    auto allThreadsCopy = std::exchange(allThreads, {});
    expiredThreads.clear();
    waitingThreads.clear();
 
    mutex.unlock();
 
    for (QThreadPoolThread *thread : std::as_const(allThreadsCopy)) {
        if (thread->isRunning()) {
            thread->runnableReady.wakeAll();
            thread->wait();
        }
        delete thread;
    }
 
    mutex.lock();
}
 
bool QThreadPoolPrivate::waitForDone(const QDeadlineTimer &timer)
{
    QMutexLocker locker(&mutex);
    while (!(queue.isEmpty() && activeThreads == 0) && !timer.hasExpired())
        noActiveThreads.wait(&mutex, timer);
 
    if (!queue.isEmpty() || activeThreads)
        return false;
 
    reset();
    // New jobs might have started during reset, but return anyway
    // as the active thread and task count did reach 0 once, and
    // race conditions are outside our scope.
    return true;
}
 
void QThreadPoolPrivate::clear()
{
    QMutexLocker locker(&mutex);
    while (!queue.isEmpty()) {
        auto *page = queue.takeLast();
        while (!page->isFinished()) {
            QRunnable *r = page->pop();
            if (r && r->autoDelete()) {
                locker.unlock();
                delete r;
                locker.relock();
            }
        }
        delete page;
    }
}
 
bool QThreadPool::tryTake(QRunnable *runnable)
{
    Q_D(QThreadPool);
 
    if (runnable == nullptr)
        return false;
 
    QMutexLocker locker(&d->mutex);
    for (QueuePage *page : std::as_const(d->queue)) {
        if (page->tryTake(runnable)) {
            if (page->isFinished()) {
                d->queue.removeOne(page);
                delete page;
            }
            return true;
        }
    }
 
    return false;
}
 
void QThreadPoolPrivate::stealAndRunRunnable(QRunnable *runnable)
{
    Q_Q(QThreadPool);
    if (!q->tryTake(runnable))
        return;
    // If autoDelete() is false, runnable might already be deleted after run(), so check status now.
    const bool del = runnable->autoDelete();
 
    runnable->run();
 
    if (del)
        delete runnable;
}
 
QThreadPool::QThreadPool(QObject *parent)
    : QObject(*new QThreadPoolPrivate, parent)
{
    Q_D(QThreadPool);
    connect(this, &QObject::objectNameChanged, this, [d](const QString &newName) {
        // We keep a copy of the name under our own lock, so we can access it thread-safely.
        QMutexLocker locker(&d->mutex);
        d->objectName = newName;
    });
}
 
QThreadPool::~QThreadPool()
{
    Q_D(QThreadPool);
    waitForDone();
    Q_ASSERT(d->queue.isEmpty());
    Q_ASSERT(d->allThreads.isEmpty());
}
 
QThreadPool *QThreadPool::globalInstance()
{
    Q_CONSTINIT static QPointer<QThreadPool> theInstance;
    Q_CONSTINIT static QBasicMutex theMutex;
 
    const QMutexLocker locker(&theMutex);
    if (theInstance.isNull() && !QCoreApplication::closingDown())
        theInstance = new QThreadPool();
    return theInstance;
}
 
QThreadPool *QThreadPoolPrivate::qtGuiInstance()
{
    Q_CONSTINIT static QPointer<QThreadPool> guiInstance;
    Q_CONSTINIT static QBasicMutex theMutex;
 
    const QMutexLocker locker(&theMutex);
    if (guiInstance.isNull() && !QCoreApplication::closingDown())
        guiInstance = new QThreadPool();
    return guiInstance;
}
 
void QThreadPool::start(QRunnable *runnable, int priority)
{
    if (!runnable)
        return;
 
    Q_D(QThreadPool);
    QMutexLocker locker(&d->mutex);
 
    if (!d->tryStart(runnable))
        d->enqueueTask(runnable, priority);
}
 
bool QThreadPool::tryStart(QRunnable *runnable)
{
    if (!runnable)
        return false;
 
    Q_D(QThreadPool);
    QMutexLocker locker(&d->mutex);
    if (d->tryStart(runnable))
        return true;
 
    return false;
}
 
int QThreadPool::expiryTimeout() const
{
    using namespace std::chrono;
    Q_D(const QThreadPool);
    QMutexLocker locker(&d->mutex);
    return duration_cast<milliseconds>(d->expiryTimeout).count();
}
 
void QThreadPool::setExpiryTimeout(int expiryTimeout)
{
    Q_D(QThreadPool);
    QMutexLocker locker(&d->mutex);
    d->expiryTimeout = std::chrono::milliseconds(expiryTimeout);
}
 
int QThreadPool::maxThreadCount() const
{
    Q_D(const QThreadPool);
    QMutexLocker locker(&d->mutex);
    return d->requestedMaxThreadCount;
}
 
void QThreadPool::setMaxThreadCount(int maxThreadCount)
{
    Q_D(QThreadPool);
    QMutexLocker locker(&d->mutex);
 
    if (maxThreadCount == d->requestedMaxThreadCount)
        return;
 
    d->requestedMaxThreadCount = maxThreadCount;
    d->tryToStartMoreThreads();
}
 
int QThreadPool::activeThreadCount() const
{
    Q_D(const QThreadPool);
    QMutexLocker locker(&d->mutex);
    return d->activeThreadCount();
}
 
void QThreadPool::reserveThread()
{
    Q_D(QThreadPool);
    QMutexLocker locker(&d->mutex);
    ++d->reservedThreads;
}
 
void QThreadPool::setStackSize(uint stackSize)
{
    Q_D(QThreadPool);
    QMutexLocker locker(&d->mutex);
    d->stackSize = stackSize;
}
 
uint QThreadPool::stackSize() const
{
    Q_D(const QThreadPool);
    QMutexLocker locker(&d->mutex);
    return d->stackSize;
}
 
void QThreadPool::setThreadPriority(QThread::Priority priority)
{
    Q_D(QThreadPool);
    QMutexLocker locker(&d->mutex);
    d->threadPriority = priority;
}
 
QThread::Priority QThreadPool::threadPriority() const
{
    Q_D(const QThreadPool);
    QMutexLocker locker(&d->mutex);
    return d->threadPriority;
}
 
void QThreadPool::releaseThread()
{
    Q_D(QThreadPool);
    QMutexLocker locker(&d->mutex);
    --d->reservedThreads;
    d->tryToStartMoreThreads();
}
 
void QThreadPool::startOnReservedThread(QRunnable *runnable)
{
    if (!runnable)
        return releaseThread();
 
    Q_D(QThreadPool);
    QMutexLocker locker(&d->mutex);
    Q_ASSERT(d->reservedThreads > 0);
    --d->reservedThreads;
 
    if (!d->tryStart(runnable)) {
        // This can only happen if we reserved max threads,
        // and something took the one minimum thread.
        d->enqueueTask(runnable, INT_MAX);
    }
}
 
bool QThreadPool::waitForDone(QDeadlineTimer deadline)
{
    Q_D(QThreadPool);
    return d->waitForDone(deadline);
}
 
void QThreadPool::clear()
{
    Q_D(QThreadPool);
    d->clear();
}
 
bool QThreadPool::contains(const QThread *thread) const
{
    Q_D(const QThreadPool);
    const QThreadPoolThread *poolThread = qobject_cast<const QThreadPoolThread *>(thread);
    if (!poolThread)
        return false;
    QMutexLocker locker(&d->mutex);
    return d->allThreads.contains(const_cast<QThreadPoolThread *>(poolThread));
}
 
QT_END_NAMESPACE
 
#include "moc_qthreadpool.cpp"
#include "qthreadpool.moc"