c++ - QtConcurrent:为什么releaseThread和reserveThread会导致死锁？

Question

在 Qt 4.7 Reference for QThreadPool 中，我们发现:

void QThreadPool::releaseThread()

Releases a thread previously reserved by a call to reserveThread().

Note: Calling this function without previously reserving a thread temporarily increases maxThreadCount(). This is useful when a thread goes to sleep waiting for more work, allowing other threads to continue. Be sure to call reserveThread() when done waiting, so that the thread pool can correctly maintain the activeThreadCount().

See also reserveThread().

---

void QThreadPool::reserveThread()

Reserves one thread, disregarding activeThreadCount() and maxThreadCount().

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. This means that by using this function, it is possible for activeThreadCount() to return a value greater than maxThreadCount().

See also releaseThread().

我想使用releaseThread()来实现嵌套并发映射，但是在下面的代码中，它卡在了waitForFinished()中:

#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);
        {
            // with out 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);
        }
        // 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);
        }

        // 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;
}

这段代码不会永远运行；它会在许多循环(1~10000)后挂起，所有线程都在等待条件变量。

我的问题是:

1. 为什么会挂起？
2. 我可以修复它并保留嵌套的并发映射吗？

开发环境:

Linux 版本 2.6.32-696.18.7.el6.x86_64； Qt4.7.4；海湾合作委员会 3.4.5

Windows 7； Qt4.7.4； mingw 4.4.0

Answer 1

当您尝试处理 expiryTimeout 时，由于 QThreadPool 中的竞争条件，程序挂起。下面是详 segmentation 析:

QThreadPool 中的问题 - source

When starting a task, QThreadPool did something along the lines of:

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

And the the thread's main loop looks like this:

void QThreadPoolThread::run()
{
  QMutexLocker locker(&manager->mutex);
  while (true) {
    /* ... */
    if (manager->taskQueue.isEmpty()) {
      // no pending task, wait for one.
      bool expired = !manager->runnableReady.wait(locker.mutex(), 
                                                  manager->expiryTimeout);
      if (expired) {
        manager->runningThreadCount--;
        return;
      } else {
        continue;
      }
    }
    QRunnable *r = manager->taskQueue.takeFirst();
    // run the task
    locker.unlock();
    r->run();
    locker.relock();
  }
}

The idea is that the thread will wait for a given amount of second for a task, but if no task was added in a given amount of time, the thread expires and is terminated. The problem here is that we rely on the return value of runnableReady. If there is a task that is scheduled at exactly the same time as the thread expires, then the thread will see false and will expire. But the main thread will not restart any other thread. That might let the application hang as the task will never be run.

快速解决方法是使用较长的 expiryTime(默认为 30000)并删除等待线程过期的 while 循环。

这里是修改的main函数，程序在Windows 7下运行流畅，默认使用4个线程:

int main() {
    QThreadPool* gtpool = QThreadPool::globalInstance();
    //gtpool->setExpiryTimeout(50); <-- don't set the expiry Timeout, use the default one.
    qDebug() << gtpool->maxThreadCount();

    int count = 0;
    for (;;) {

        QVector<int> vec;
        for (int i = 0; i < 40 ; i++) {
            vec.push_back(i);
        }
        // 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);
        }
        */

        // 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;
}