qthreadpool.cpp

Go to the documentation of this file.

/****************************************************************************
**
** Copyright (C) 2014 Digia Plc and/or its subsidiary(-ies).
** Contact: http://www.qt-project.org/legal
**
** 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 Digia.  For licensing terms and
** conditions see http://qt.digia.com/licensing.  For further information
** use the contact form at http://qt.digia.com/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 2.1 as published by the Free Software
** Foundation and appearing in the file LICENSE.LGPL included in the
** packaging of this file.  Please review the following information to
** ensure the GNU Lesser General Public License version 2.1 requirements
** will be met: http://www.gnu.org/licenses/old-licenses/lgpl-2.1.html.
**
** In addition, as a special exception, Digia gives you certain additional
** rights.  These rights are described in the Digia Qt LGPL Exception
** version 1.1, included in the file LGPL_EXCEPTION.txt in this package.
**
** GNU General Public License Usage
** Alternatively, this file may be used under the terms of the GNU
** General Public License version 3.0 as published by the Free Software
** Foundation and appearing in the file LICENSE.GPL included in the
** packaging of this file.  Please review the following information to
** ensure the GNU General Public License version 3.0 requirements will be
** met: http://www.gnu.org/copyleft/gpl.html.
**
**
** $QT_END_LICENSE$
**
****************************************************************************/

#include "qthreadpool.h"
#include "qthreadpool_p.h"
#include "qelapsedtimer.h"

#ifndef QT_NO_THREAD

QT_BEGIN_NAMESPACE

inline bool operator<(int priority, const QPair<QRunnable *, int> &p)
{
    return p.second < priority;
}
inline bool operator<(const QPair<QRunnable *, int> &p, int priority)
{
    return priority < p.second;
}

Q_GLOBAL_STATIC(QThreadPool, theInstance)

/*
    QThread wrapper, provides synchronization against a ThreadPool
*/
class QThreadPoolThread : public QThread
{
public:
    QThreadPoolThread(QThreadPoolPrivate *manager);
    void run();
    void registerThreadInactive();

    QWaitCondition runnableReady;
    QThreadPoolPrivate *manager;
    QRunnable *runnable;
};

/*
    QThreadPool private class.
*/


QThreadPoolThread::QThreadPoolThread(QThreadPoolPrivate *manager)
    :manager(manager), runnable(0)
{ }

/* \internal

*/
void QThreadPoolThread::run()
{
    QMutexLocker locker(&manager->mutex);
    for(;;) {
        QRunnable *r = runnable;
        runnable = 0;

        do {
            if (r) {
                const bool autoDelete = 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
                locker.relock();

                if (autoDelete && !--r->ref)
                    delete r;
            }

            // if too many threads are active, expire this thread
            if (manager->tooManyThreadsActive())
                break;

            r = !manager->queue.isEmpty() ? manager->queue.takeFirst().first : 0;
        } while (r != 0);

        if (manager->isExiting) {
            registerThreadInactive();
            break;
        }

        // if too many threads are active, expire this thread
        bool expired = manager->tooManyThreadsActive();
        if (!expired) {
            manager->waitingThreads.enqueue(this);
            registerThreadInactive();
            // wait for work, exiting after the expiry timeout is reached
            runnableReady.wait(locker.mutex(), manager->expiryTimeout);
            ++manager->activeThreads;
            if (manager->waitingThreads.removeOne(this))
                expired = true;
        }
        if (expired) {
            manager->expiredThreads.enqueue(this);
            registerThreadInactive();
            break;
        }
    }
}

void QThreadPoolThread::registerThreadInactive()
{
    if (--manager->activeThreads == 0)
        manager->noActiveThreads.wakeAll();
}


/* \internal

*/
QThreadPoolPrivate:: QThreadPoolPrivate()
    : isExiting(false),
      expiryTimeout(30000),
      maxThreadCount(qAbs(QThread::idealThreadCount())),
      reservedThreads(0),
      activeThreads(0)
{ }

bool QThreadPoolPrivate::tryStart(QRunnable *task)
{
    if (allThreads.isEmpty()) {
        // always create at least one thread
        startThread(task);
        return true;
    }

    // can't do anything if we're over the limit
    if (activeThreadCount() >= maxThreadCount)
        return false;

    if (waitingThreads.count() > 0) {
        // 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 == 0);

        ++activeThreads;

        if (task->autoDelete())
            ++task->ref;
        thread->runnable = task;
        thread->start();
        return true;
    }

    // start a new thread
    startThread(task);
    return true;
}

void QThreadPoolPrivate::enqueueTask(QRunnable *runnable, int priority)
{
    if (runnable->autoDelete())
        ++runnable->ref;

    // put it on the queue
    QList<QPair<QRunnable *, int> >::iterator at =
        qUpperBound(queue.begin(), queue.end(), priority);
    queue.insert(at, qMakePair(runnable, priority));
}

int QThreadPoolPrivate::activeThreadCount() const
{
    return (allThreads.count()
            - expiredThreads.count()
            - waitingThreads.count()
            + reservedThreads);
}

void QThreadPoolPrivate::tryToStartMoreThreads()
{
    // try to push tasks on the queue to any available threads
    while (!queue.isEmpty() && tryStart(queue.first().first))
        queue.removeFirst();
}

bool QThreadPoolPrivate::tooManyThreadsActive() const
{
    const int activeThreadCount = this->activeThreadCount();
    return activeThreadCount > maxThreadCount && (activeThreadCount - reservedThreads) > 1;
}

void QThreadPoolPrivate::startThread(QRunnable *runnable)
{
    QScopedPointer <QThreadPoolThread> thread(new QThreadPoolThread(this));
    thread->setObjectName(QLatin1String("Thread (pooled)"));
    allThreads.insert(thread.data());
    ++activeThreads;

    if (runnable->autoDelete())
        ++runnable->ref;
    thread->runnable = runnable;
    thread.take()->start();
}

void QThreadPoolPrivate::reset()
{
    QMutexLocker locker(&mutex);
    isExiting = true;

    do {
        // make a copy of the set so that we can iterate without the lock
        QSet<QThreadPoolThread *> allThreadsCopy = allThreads;
        allThreads.clear();
        locker.unlock();

        foreach (QThreadPoolThread *thread, allThreadsCopy) {
            thread->runnableReady.wakeAll();
            thread->wait();
            delete thread;
        }

        locker.relock();
        // repeat until all newly arrived threads have also completed
    } while (!allThreads.isEmpty());

    waitingThreads.clear();
    expiredThreads.clear();

    isExiting = false;
}

bool QThreadPoolPrivate::waitForDone(int msecs)
{
    QMutexLocker locker(&mutex);
    if (msecs < 0) {
        while (!(queue.isEmpty() && activeThreads == 0))
            noActiveThreads.wait(locker.mutex());
    } else {
        QElapsedTimer timer;
        timer.start();
        int t;
        while (!(queue.isEmpty() && activeThreads == 0) &&
               ((t = msecs - timer.elapsed()) > 0))
            noActiveThreads.wait(locker.mutex(), t);
    }
    return queue.isEmpty() && activeThreads == 0;
}

bool QThreadPoolPrivate::startFrontRunnable()
{
    QMutexLocker locker(&mutex);
    if (queue.isEmpty())
        return false;

    QRunnable *runnable = queue.takeFirst().first;
    const bool autoDelete = runnable->autoDelete();
    bool del = autoDelete && !--runnable->ref;

    locker.unlock();
    runnable->run();
    locker.relock();

    if (del) {
        delete runnable;
    }

    return true;
}

void QThreadPoolPrivate::stealRunnable(QRunnable *runnable)
{
    if (runnable == 0 || queue.isEmpty())
        return;
    bool found = false;
    {
        QMutexLocker locker(&mutex);
        QList<QPair<QRunnable *, int> >::iterator it = queue.begin();
        QList<QPair<QRunnable *, int> >::iterator end = queue.end();

        while (it != end) {
            if (it->first == runnable) {
                found = true;
                queue.erase(it);
                break;
            }
            ++it;
        }
    }

    if (!found)
        return;

    const bool autoDelete = runnable->autoDelete();
    bool del = autoDelete && !--runnable->ref;

    runnable->run();

    if (del) {
        delete runnable;
    }
}

QThreadPool::QThreadPool(QObject *parent)
    : QObject(*new QThreadPoolPrivate, parent)
{ }

QThreadPool::~QThreadPool()
{
    d_func()->waitForDone();
    d_func()->reset();
}

QThreadPool *QThreadPool::globalInstance()
{
    return theInstance();
}

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);

        if (!d->waitingThreads.isEmpty())
            d->waitingThreads.takeFirst()->runnableReady.wakeOne();
    }
}

bool QThreadPool::tryStart(QRunnable *runnable)
{
    if (!runnable)
        return false;

    Q_D(QThreadPool);

    QMutexLocker locker(&d->mutex);

    if (d->allThreads.isEmpty() == false && d->activeThreadCount() >= d->maxThreadCount)
        return false;

    return d->tryStart(runnable);
}

int QThreadPool::expiryTimeout() const
{
    Q_D(const QThreadPool);
    return d->expiryTimeout;
}

void QThreadPool::setExpiryTimeout(int expiryTimeout)
{
    Q_D(QThreadPool);
    if (d->expiryTimeout == expiryTimeout)
        return;
    d->expiryTimeout = expiryTimeout;
}

int QThreadPool::maxThreadCount() const
{
    Q_D(const QThreadPool);
    return d->maxThreadCount;
}

void QThreadPool::setMaxThreadCount(int maxThreadCount)
{
    Q_D(QThreadPool);
    QMutexLocker locker(&d->mutex);

    if (maxThreadCount == d->maxThreadCount)
        return;

    d->maxThreadCount = 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::releaseThread()
{
    Q_D(QThreadPool);
    QMutexLocker locker(&d->mutex);
    --d->reservedThreads;
    d->tryToStartMoreThreads();
}

void QThreadPool::waitForDone()
{
    Q_D(QThreadPool);
    d->waitForDone();
    d->reset();
}

bool QThreadPool::waitForDone(int msecs)
{
    Q_D(QThreadPool);
    bool rc = d->waitForDone(msecs);
    if (rc)
      d->reset();
    return rc;
}

QT_END_NAMESPACE

#endif