diff options
Diffstat (limited to 'src/corelib/thread/qthreadpool.cpp')
| -rw-r--r-- | src/corelib/thread/qthreadpool.cpp | 427 |
1 files changed, 244 insertions, 183 deletions
diff --git a/src/corelib/thread/qthreadpool.cpp b/src/corelib/thread/qthreadpool.cpp index 1478cfcb19..c7531111da 100644 --- a/src/corelib/thread/qthreadpool.cpp +++ b/src/corelib/thread/qthreadpool.cpp @@ -1,51 +1,20 @@ -/**************************************************************************** -** -** Copyright (C) 2016 The Qt Company Ltd. -** Contact: https://www.qt.io/licensing/ -** -** This file is part of the QtCore module of the Qt Toolkit. -** -** $QT_BEGIN_LICENSE:LGPL$ -** Commercial License Usage -** Licensees holding valid commercial Qt licenses may use this file in -** accordance with the commercial license agreement provided with the -** Software or, alternatively, in accordance with the terms contained in -** a written agreement between you and The Qt Company. For licensing terms -** and conditions see https://www.qt.io/terms-conditions. For further -** information use the contact form at https://www.qt.io/contact-us. -** -** GNU Lesser General Public License Usage -** Alternatively, this file may be used under the terms of the GNU Lesser -** General Public License version 3 as published by the Free Software -** Foundation and appearing in the file LICENSE.LGPL3 included in the -** packaging of this file. Please review the following information to -** ensure the GNU Lesser General Public License version 3 requirements -** will be met: https://www.gnu.org/licenses/lgpl-3.0.html. -** -** GNU General Public License Usage -** Alternatively, this file may be used under the terms of the GNU -** General Public License version 2.0 or (at your option) the GNU General -** Public license version 3 or any later version approved by the KDE Free -** Qt Foundation. The licenses are as published by the Free Software -** Foundation and appearing in the file LICENSE.GPL2 and LICENSE.GPL3 -** included in the packaging of this file. Please review the following -** information to ensure the GNU General Public License requirements will -** be met: https://www.gnu.org/licenses/gpl-2.0.html and -** https://www.gnu.org/licenses/gpl-3.0.html. -** -** $QT_END_LICENSE$ -** -****************************************************************************/ +// 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 */ @@ -88,9 +57,8 @@ void QThreadPoolThread::run() do { if (r) { + // If autoDelete() is false, r might already be deleted after run(), so check status now. const bool del = r->autoDelete(); - Q_ASSERT(!del || r->ref == 1); - // run the task locker.unlock(); @@ -113,16 +81,15 @@ void QThreadPoolThread::run() locker.relock(); } - // if too many threads are active, expire this thread + // if too many threads are active, stop working in this one if (manager->tooManyThreadsActive()) break; - if (manager->queue.isEmpty()) { - r = nullptr; + // all work is done, time to wait for more + if (manager->queue.isEmpty()) break; - } - QueuePage *page = manager->queue.first(); + QueuePage *page = manager->queue.constFirst(); r = page->pop(); if (page->isFinished()) { @@ -131,26 +98,32 @@ void QThreadPoolThread::run() } } while (true); - // if too many threads are active, expire this thread - bool expired = manager->tooManyThreadsActive(); - if (!expired) { - manager->waitingThreads.enqueue(this); + // this thread is about to be deleted, do not wait or expire + if (!manager->allThreads.contains(this)) { registerThreadInactive(); - // wait for work, exiting after the expiry timeout is reached - runnableReady.wait(locker.mutex(), QDeadlineTimer(manager->expiryTimeout)); - ++manager->activeThreads; - if (manager->waitingThreads.removeOne(this)) - expired = true; - if (!manager->allThreads.contains(this)) { - registerThreadInactive(); - break; - } + return; } - if (expired) { + + // if too many threads are active, expire this thread + if (manager->tooManyThreadsActive()) { manager->expiredThreads.enqueue(this); registerThreadInactive(); - break; + 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; } } @@ -177,10 +150,10 @@ bool QThreadPoolPrivate::tryStart(QRunnable *task) } // can't do anything if we're over the limit - if (activeThreadCount() >= maxThreadCount) + if (areAllThreadsActive()) return false; - if (waitingThreads.count() > 0) { + if (!waitingThreads.isEmpty()) { // recycle an available thread enqueueTask(task); waitingThreads.takeFirst()->runnableReady.wakeOne(); @@ -195,7 +168,12 @@ bool QThreadPoolPrivate::tryStart(QRunnable *task) ++activeThreads; thread->runnable = task; - thread->start(); + + // 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; } @@ -212,7 +190,7 @@ inline bool comparePriority(int priority, const QueuePage *p) void QThreadPoolPrivate::enqueueTask(QRunnable *runnable, int priority) { Q_ASSERT(runnable != nullptr); - for (QueuePage *page : qAsConst(queue)) { + for (QueuePage *page : std::as_const(queue)) { if (page->priority() == priority && !page->isFull()) { page->push(runnable); return; @@ -224,9 +202,9 @@ void QThreadPoolPrivate::enqueueTask(QRunnable *runnable, int priority) int QThreadPoolPrivate::activeThreadCount() const { - return (allThreads.count() - - expiredThreads.count() - - waitingThreads.count() + return (allThreads.size() + - expiredThreads.size() + - waitingThreads.size() + reservedThreads); } @@ -234,7 +212,7 @@ void QThreadPoolPrivate::tryToStartMoreThreads() { // try to push tasks on the queue to any available threads while (!queue.isEmpty()) { - QueuePage *page = queue.first(); + QueuePage *page = queue.constFirst(); if (!tryStart(page->first())) break; @@ -247,10 +225,16 @@ void QThreadPoolPrivate::tryToStartMoreThreads() } } +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; + return activeThreadCount > maxThreadCount() && (activeThreadCount - reservedThreads) > 1; } /*! @@ -259,32 +243,36 @@ bool QThreadPoolPrivate::tooManyThreadsActive() const void QThreadPoolPrivate::startThread(QRunnable *runnable) { Q_ASSERT(runnable != nullptr); - QScopedPointer <QThreadPoolThread> thread(new QThreadPoolThread(this)); - thread->setObjectName(QLatin1String("Thread (pooled)")); - Q_ASSERT(!allThreads.contains(thread.data())); // if this assert hits, we have an ABA problem (deleted threads don't get removed here) - allThreads.insert(thread.data()); + 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.take()->start(); + thread.release()->start(threadPriority); } /*! \internal - Helper function only to be called from waitForDone(int) + Helper function only to be called from waitForDone() + + Deletes all current threads. */ void QThreadPoolPrivate::reset() { // move the contents of the set out so that we can iterate without the lock - QSet<QThreadPoolThread *> allThreadsCopy; - allThreadsCopy.swap(allThreads); + auto allThreadsCopy = std::exchange(allThreads, {}); expiredThreads.clear(); waitingThreads.clear(); + mutex.unlock(); - for (QThreadPoolThread *thread: qAsConst(allThreadsCopy)) { - if (!thread->isFinished()) { + for (QThreadPoolThread *thread : std::as_const(allThreadsCopy)) { + if (thread->isRunning()) { thread->runnableReady.wakeAll(); thread->wait(); } @@ -297,47 +285,39 @@ void QThreadPoolPrivate::reset() /*! \internal - Helper function only to be called from waitForDone(int) + Helper function only to be called from the public waitForDone() */ bool QThreadPoolPrivate::waitForDone(const QDeadlineTimer &timer) { + QMutexLocker locker(&mutex); while (!(queue.isEmpty() && activeThreads == 0) && !timer.hasExpired()) noActiveThreads.wait(&mutex, timer); - return queue.isEmpty() && activeThreads == 0; -} - -bool QThreadPoolPrivate::waitForDone(int msecs) -{ - QMutexLocker locker(&mutex); - QDeadlineTimer timer(msecs); - do { - if (!waitForDone(timer)) - return false; - reset(); - // More threads can be started during reset(), in that case continue - // waiting if we still have time left. - } while ((!queue.isEmpty() || activeThreads) && !timer.hasExpired()); + if (!queue.isEmpty() || activeThreads) + return false; - return queue.isEmpty() && activeThreads == 0; + 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); - for (QueuePage *page : qAsConst(queue)) { + while (!queue.isEmpty()) { + auto *page = queue.takeLast(); while (!page->isFinished()) { QRunnable *r = page->pop(); if (r && r->autoDelete()) { - Q_ASSERT(r->ref == 1); locker.unlock(); delete r; locker.relock(); } } + delete page; } - qDeleteAll(queue); - queue.clear(); } /*! @@ -365,16 +345,12 @@ bool QThreadPool::tryTake(QRunnable *runnable) return false; QMutexLocker locker(&d->mutex); - for (QueuePage *page : qAsConst(d->queue)) { + for (QueuePage *page : std::as_const(d->queue)) { if (page->tryTake(runnable)) { if (page->isFinished()) { d->queue.removeOne(page); delete page; } - if (runnable->autoDelete()) { - Q_ASSERT(runnable->ref == 1); - --runnable->ref; // undo ++ref in start() - } return true; } } @@ -393,14 +369,13 @@ 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) { - Q_ASSERT(runnable->ref == 0); // tryTake already deref'ed + if (del) delete runnable; - } } /*! @@ -412,7 +387,7 @@ void QThreadPoolPrivate::stealAndRunRunnable(QRunnable *runnable) \ingroup thread - QThreadPool manages and recyles individual QThread objects to help reduce + QThreadPool manages and recycles individual QThread objects to help reduce thread creation costs in programs that use threads. Each Qt application has one global QThreadPool object, which can be accessed by calling globalInstance(). @@ -461,7 +436,14 @@ void QThreadPoolPrivate::stealAndRunRunnable(QRunnable *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; + }); +} /*! Destroys the QThreadPool. @@ -469,7 +451,10 @@ QThreadPool::QThreadPool(QObject *parent) */ QThreadPool::~QThreadPool() { + Q_D(QThreadPool); waitForDone(); + Q_ASSERT(d->queue.isEmpty()); + Q_ASSERT(d->allThreads.isEmpty()); } /*! @@ -477,8 +462,8 @@ QThreadPool::~QThreadPool() */ QThreadPool *QThreadPool::globalInstance() { - static QPointer<QThreadPool> theInstance; - static QBasicMutex theMutex; + Q_CONSTINIT static QPointer<QThreadPool> theInstance; + Q_CONSTINIT static QBasicMutex theMutex; const QMutexLocker locker(&theMutex); if (theInstance.isNull() && !QCoreApplication::closingDown()) @@ -487,6 +472,21 @@ QThreadPool *QThreadPool::globalInstance() } /*! + Returns the QThreadPool instance for Qt Gui. + \internal +*/ +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; +} + +/*! Reserves a thread and uses it to run \a runnable, unless this thread will make the current thread count exceed maxThreadCount(). In that case, \a runnable is added to a run queue instead. The \a priority argument can @@ -508,34 +508,27 @@ void QThreadPool::start(QRunnable *runnable, int priority) Q_D(QThreadPool); QMutexLocker locker(&d->mutex); - if (runnable->autoDelete()) { - Q_ASSERT(runnable->ref == 0); - ++runnable->ref; - } - if (!d->tryStart(runnable)) { + if (!d->tryStart(runnable)) d->enqueueTask(runnable, priority); - - if (!d->waitingThreads.isEmpty()) - d->waitingThreads.takeFirst()->runnableReady.wakeOne(); - } } /*! + \fn template<typename Callable, QRunnable::if_callable<Callable>> void QThreadPool::start(Callable &&callableToRun, int priority) \overload \since 5.15 - Reserves a thread and uses it to run \a functionToRun, unless this thread will + Reserves a thread and uses it to run \a callableToRun, unless this thread will make the current thread count exceed maxThreadCount(). In that case, - \a functionToRun is added to a run queue instead. The \a priority argument can + \a callableToRun is added to a run queue instead. The \a priority argument can be used to control the run queue's order of execution. + + \note This function participates in overload resolution only if \c Callable + is a function or function object which can be called with zero arguments. + + \note In Qt version prior to 6.6, this function took std::function<void()>, + and therefore couldn't handle move-only callables. */ -void QThreadPool::start(std::function<void()> functionToRun, int priority) -{ - if (!functionToRun) - return; - start(QRunnable::create(std::move(functionToRun)), priority); -} /*! Attempts to reserve a thread to run \a runnable. @@ -558,54 +551,35 @@ bool QThreadPool::tryStart(QRunnable *runnable) if (!runnable) return false; - if (runnable->autoDelete()) { - Q_ASSERT(runnable->ref == 0); - ++runnable->ref; - } - Q_D(QThreadPool); QMutexLocker locker(&d->mutex); if (d->tryStart(runnable)) return true; - // Undo the reference above as we did not start the runnable and - // take over ownership. - if (runnable->autoDelete()) { - --runnable->ref; - Q_ASSERT(runnable->ref == 0); - } return false; } /*! + \fn template<typename Callable, QRunnable::if_callable<Callable>> bool QThreadPool::tryStart(Callable &&callableToRun) \overload \since 5.15 - Attempts to reserve a thread to run \a functionToRun. + Attempts to reserve a thread to run \a callableToRun. If no threads are available at the time of calling, then this function - does nothing and returns \c false. Otherwise, \a functionToRun is run immediately + does nothing and returns \c false. Otherwise, \a callableToRun is run immediately using one available thread and this function returns \c true. -*/ -bool QThreadPool::tryStart(std::function<void()> functionToRun) -{ - if (!functionToRun) - return false; - Q_D(QThreadPool); - QMutexLocker locker(&d->mutex); - if (!d->allThreads.isEmpty() && d->activeThreadCount() >= d->maxThreadCount) - return false; + \note This function participates in overload resolution only if \c Callable + is a function or function object which can be called with zero arguments. - QRunnable *runnable = QRunnable::create(std::move(functionToRun)); - if (d->tryStart(runnable)) - return true; - delete runnable; - return false; -} + \note In Qt version prior to 6.6, this function took std::function<void()>, + and therefore couldn't handle move-only callables. +*/ /*! \property QThreadPool::expiryTimeout + \brief the thread expiry timeout value in milliseconds. - Threads that are unused for \a expiryTimeout milliseconds are considered + Threads that are unused for \e expiryTimeout milliseconds are considered to have expired and will exit. Such threads will be restarted as needed. The default \a expiryTimeout is 30000 milliseconds (30 seconds). If \a expiryTimeout is negative, newly created threads will not expire, e.g., @@ -619,22 +593,24 @@ bool QThreadPool::tryStart(std::function<void()> functionToRun) int QThreadPool::expiryTimeout() const { + using namespace std::chrono; Q_D(const QThreadPool); - return d->expiryTimeout; + QMutexLocker locker(&d->mutex); + return duration_cast<milliseconds>(d->expiryTimeout).count(); } void QThreadPool::setExpiryTimeout(int expiryTimeout) { Q_D(QThreadPool); - if (d->expiryTimeout == expiryTimeout) - return; - d->expiryTimeout = expiryTimeout; + QMutexLocker locker(&d->mutex); + d->expiryTimeout = std::chrono::milliseconds(expiryTimeout); } /*! \property QThreadPool::maxThreadCount - This property represents the maximum number of threads used by the thread - pool. + \brief the maximum number of threads used by the thread pool. This property + will default to the value of QThread::idealThreadCount() at the moment the + QThreadPool object is created. \note The thread pool will always use at least 1 thread, even if \a maxThreadCount limit is zero or negative. @@ -645,7 +621,8 @@ void QThreadPool::setExpiryTimeout(int expiryTimeout) int QThreadPool::maxThreadCount() const { Q_D(const QThreadPool); - return d->maxThreadCount; + QMutexLocker locker(&d->mutex); + return d->requestedMaxThreadCount; } void QThreadPool::setMaxThreadCount(int maxThreadCount) @@ -653,16 +630,16 @@ void QThreadPool::setMaxThreadCount(int maxThreadCount) Q_D(QThreadPool); QMutexLocker locker(&d->mutex); - if (maxThreadCount == d->maxThreadCount) + if (maxThreadCount == d->requestedMaxThreadCount) return; - d->maxThreadCount = maxThreadCount; + d->requestedMaxThreadCount = maxThreadCount; d->tryToStartMoreThreads(); } /*! \property QThreadPool::activeThreadCount - This property represents the number of active threads in the thread pool. + \brief the number of active threads in the thread pool. \note It is possible for this function to return a value that is greater than maxThreadCount(). See reserveThread() for more details. @@ -683,7 +660,10 @@ int QThreadPool::activeThreadCount() const Once you are done with the thread, call releaseThread() to allow it to be reused. - \note This function will always increase the number of active threads. + \note Even if reserving maxThreadCount() threads or more, the thread pool + will still allow a minimum of one thread. + + \note This function will increase the reported number of active threads. This means that by using this function, it is possible for activeThreadCount() to return a value greater than maxThreadCount() . @@ -697,9 +677,7 @@ void QThreadPool::reserveThread() } /*! \property QThreadPool::stackSize - - This property contains the stack size for the thread pool worker - threads. + \brief the stack size for the thread pool worker threads. The value of the property is only used when the thread pool creates new threads. Changing it has no effect for already created @@ -713,15 +691,45 @@ void QThreadPool::reserveThread() 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; } +/*! \property QThreadPool::threadPriority + \brief the thread priority for new worker threads. + + The value of the property is only used when the thread pool starts + new threads. Changing it has no effect for already running threads. + + The default value is QThread::InheritPriority, which makes QThread + use the same priority as the one the QThreadPool object lives in. + + \sa QThread::Priority + + \since 6.2 +*/ + +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; +} + /*! Releases a thread previously reserved by a call to reserveThread(). @@ -743,15 +751,75 @@ void QThreadPool::releaseThread() } /*! + Releases a thread previously reserved with reserveThread() and uses it + to run \a runnable. + + Note that the thread pool takes ownership of the \a runnable if + \l{QRunnable::autoDelete()}{runnable->autoDelete()} returns \c true, + and the \a runnable will be deleted automatically by the thread + pool after the \l{QRunnable::run()}{runnable->run()} returns. If + \l{QRunnable::autoDelete()}{runnable->autoDelete()} returns \c false, + ownership of \a runnable remains with the caller. Note that + changing the auto-deletion on \a runnable after calling this + functions results in undefined behavior. + + \note Calling this when no threads are reserved results in + undefined behavior. + + \since 6.3 + \sa reserveThread(), start() +*/ +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); + } +} + +/*! + \fn template<typename Callable, QRunnable::if_callable<Callable>> void QThreadPool::startOnReservedThread(Callable &&callableToRun) + \overload + \since 6.3 + + Releases a thread previously reserved with reserveThread() and uses it + to run \a callableToRun. + + \note This function participates in overload resolution only if \c Callable + is a function or function object which can be called with zero arguments. + + \note In Qt version prior to 6.6, this function took std::function<void()>, + and therefore couldn't handle move-only callables. +*/ + +/*! + \fn bool QThreadPool::waitForDone(int msecs) Waits up to \a msecs milliseconds for all threads to exit and removes all threads from the thread pool. Returns \c true if all threads were removed; - otherwise it returns \c false. If \a msecs is -1 (the default), the timeout - is ignored (waits for the last thread to exit). + otherwise it returns \c false. If \a msecs is -1, this function waits for + the last thread to exit. */ -bool QThreadPool::waitForDone(int msecs) + +/*! + \since 6.8 + + Waits until \a deadline expires for all threads to exit and removes all + threads from the thread pool. Returns \c true if all threads were removed; + otherwise it returns \c false. +*/ +bool QThreadPool::waitForDone(QDeadlineTimer deadline) { Q_D(QThreadPool); - return d->waitForDone(msecs); + return d->waitForDone(deadline); } /*! @@ -769,25 +837,18 @@ void QThreadPool::clear() d->clear(); } -#if QT_VERSION < QT_VERSION_CHECK(6,0,0) -/*! - \internal - - Returns \c true if \a thread is a thread managed by this thread pool. -*/ -#else /*! \since 6.0 Returns \c true if \a thread is a thread managed by this thread pool. */ -#endif 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)); } |