计时等待引发的Bug: 一个不可避免的超时竟态

#include <QApplication>
#include <QMainWindow>
#include <QtConcurrentMap>
#include <QtConcurrentRun>
#include <QFuture>
#include <QThreadPool>
#include <QtTest/QTest>
#include <QFutureSynchronizer>

struct Task2 { // only calculation
    typedef void result_type;
    void operator()(int count) {
        int k = 0;
        for (int i = 0; i < count * 10; ++i) {
            for (int j = 0; j < count * 10; ++j) {
                k++;
            }
        }
        assert(k >= 0);
    }
};

struct Task1 { // will launch some other concurrent map
    typedef void result_type;
    void operator()(int count) {

        QVector<int> vec;
        for (int i = 0; i < 5; ++i) {
            vec.push_back(i+count);
        }
        Task2 task;

        QFuture<void> f = QtConcurrent::map(vec.begin(), vec.end(), task);
        {
            // without releaseThread before wait, it will hang directly
            QThreadPool::globalInstance()->releaseThread();
            f.waitForFinished(); // BUG: may hang there
            QThreadPool::globalInstance()->reserveThread();
        }
    }
};


int main() {
    QThreadPool* gtpool = QThreadPool::globalInstance();
    gtpool->setExpiryTimeout(50);
    int count = 0;
    for (;;) {
        QVector<int> vec;
        for (int i = 0; i < 40 ; i++) {
            vec.push_back(i);
        }
        // feature A, launch a task with nested map
        Task1 task; // Task1 will have nested concurrent map
        QFuture<void> f = QtConcurrent::map(vec.begin(), vec.end(),task);

        f.waitForFinished(); // BUG: may hang there

        count++;

        // waiting most of thread in thread pool expire
        while (QThreadPool::globalInstance()->activeThreadCount() > 0) {
            QTest::qSleep(50);
        }

        // feature B, launch a task only calculation
        Task2 task2;
        QFuture<void> f2 = QtConcurrent::map(vec.begin(), vec.end(), task2);

        f2.waitForFinished(); // BUG: may hang there

        qDebug() << count;
    }
    return 0;
}

Linux version 2.6.32-696.18.7.el6.x86_64; Qt4.7.4; GCC 3.4.5

Windows 7; Qt4.7.4; mingw 4.4.0

// qtlib4.7.4/src/corlib/concurrent/qthreadpool.cpp
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();
}

int QThreadPoolPrivate::activeThreadCount() const
{
    // To improve scalability this function is called without holding 
    // the mutex lock -- keep it thread-safe.
    return (allThreads.count()
            - expiredThreads.count()
            - waitingThreads
            + 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;
}

bool QThreadPoolPrivate::tryStart(QRunnable *task)
    QMutexLocker locker(&mutex);

    taskQueue.append(task); // Place the task on the task queue
    if (waitingThreads > 0) {
       // there are already running idle thread. They are waiting on the 'runnableReady' 
       // QWaitCondition. Wake one up them up.
       waitingThreads--;
       runnableReady.wakeOne();
    } else if (runningThreadCount < maxThreadCount) {
       startNewThread(task);
    }
}

void QThreadPoolThread::run()
{
    QQMutexLocker locker(&manager->mutex);
    for(;;) {
        QRunnable *r = manager->queue.takeFirst();
        do {
            if (r) {
                // run the task
                locker.unlock();
                r->run();
                locker.relock();
            }
            // if too many threads are active, expire this thread
            if (manager->tooManyThreadsActive())
                break;
            r = manager->queue.takeFirst();
        } while (r != 0);

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

if (waitingThreads.count() > 0) {
    // recycle an available thread
    enqueueTask(task);
    waitingThreads.takeFirst()->runnableReady.wakeOne();
    return true;
}