QThreadPool not always emitting (custom) finished signal

jbosak

I'm trying to make use of QThreadPool and want to be able to have a signal be emitted when the QRunnable passed in is started and finished. To this end I've created a custom QRunnableWrapper which literally does just that.

class QRunnableWrapper: public QObject, public QRunnable {
public:
    QRunnableWrapper(Worker *worker, bool autoDelete): m_worker(worker), m_autoDelete(autoDelete) {
        setAutoDelete(true);
    }

    virtual ~QRunnableWrapper() {
        if (m_autoDelete)
            delete(m_worker);
    }

    void run() override {
        emit started();
        m_worker->doWork();
        emit finished();
    }

signals:
    void started();
    void finished();

private:
    Worker *m_worker;
    bool m_autoDelete;
};

To help facilitate the starting of tasks in the QThreadPool I've created a little helper, which ensures that the signals/slots are properly connected.

class MyThreadPool {

public:
    MyThreadPool(): m_pool(QThreadPool::globalInstance()) {
    }

    void start(Worker *worker, const QObject *destination, const char *startedMember, const char *finishedMember, bool autoDelete = false) {
        QRunnableWrapper *wrapper = new QRunnable(wrapper, autoDelete);
        QObject::connect(wrapper, SIGNAL(started()), destination, startedMember);
        QObject::connect(wrapper, SIGNAL(finished()), destination, finishedMember);
        m_pool->start(wrapper);
    }

private:
    QThreadPool *m_pool;
};

Based on everything I've been able to find online, this should work and indeed it does work some of the time. However, sometimes the finished() signal from QRunnableWrapper is not received by the destination. From logging I can see that it is emitted (or at least that piece of code is executed), however it's not always received. To test this functionality I have the following QTest case:

void test() {
        // WaitWorker merely sleep for the specified number of milliseconds QThread::msleep(##)
        WaitWorker *work1= new WaitWorker (20);
        WaitWorker *work2 = new WaitWorker (40);
        WaitWorker *work3 = new WaitWorker (20);
        WaitWorker *work4 = new WaitWorker (40);
        WaitWorker *work5 = new WaitWorker (60);
        // It has a "valid instance" tracker for determining whether or not something is a valid instance
        // Probably could have used QPointer instead but ultimately this is not the problem.
        QVERIFY(WaitWorker ::areValidInstances({work1, work2, work3, work4, work5}));

        // Dummy receiver merely has the started() and finished() slots and counts how many times they were called
        DummyReceiver  receiver1;
        DummyReceiver receiver2;
        DummyReceiver  receiver3;
        DummyReceiver  receiver4;
        DummyReceiver  receiver5;

        // Start the work in the thread pool
        MyThreadPool pool;
        pool.start(work1, &start, SLOT(destination()), &receiver1, SLOT(started()), SLOT(finished()), true);
        pool.start(work2, &start, SLOT(destination()), &receiver2, SLOT(started()), SLOT(finished()), true);
        pool.start(work3, &start, SLOT(destination()), &receiver3, SLOT(started()), SLOT(finished()), true);
        pool.start(work4, &start, SLOT(destination()), &receiver4, SLOT(started()), SLOT(finished()), true);
        pool.start(work5, &start, SLOT(destination()), &receiver5, SLOT(started()), SLOT(finished()), true);
        QVERIFY(WaitWorker ::areValidInstances({work1, work2, work3, work4, work5}));

        // WaitWorker has a mechanism for determining whether or not it has been completed, so 
        // waitForFinish waits for  all of the specified workers to say that they were done
        // and cleaned up as well. It has a grace period that it'll wait for all to complete, returning
        // either true if all completed in time, or false if something still hadn't completed prior
        // to the grace period ending.
        QVERIFY(waitForFinish({work1, work2, work3, work4, work5}));
        
        // Ensure that all of the instances were destroyed and are no longer considered valid
        QVERIFY(WaitWorker ::areNotValidInstances({work1, work2, work3, work4, work5}));
        QCOMPARE(receiver1.getStartedCount(), 1);
        QCOMPARE(receiver1.getFinishedCount(), 1);
        QCOMPARE(receiver2.getStartedCount(), 1);
        QCOMPARE(receiver2.getFinishedCount(), 1);
        QCOMPARE(receiver3.getStartedCount(), 1);
        QCOMPARE(receiver3.getFinishedCount(), 1);
        QCOMPARE(receiver4.getStartedCount(), 1);
        QCOMPARE(receiver4.getFinishedCount(), 1);
        QCOMPARE(receiver5.getStartedCount(), 1);
        QCOMPARE(receiver5.getFinishedCount(), 1); // This has a ~50% chance to fail
}

From my experimenting making the finished() connection Qt::DirectConnection seems to make it be consistently received, but unless I'm mistaken then that means that the connected slot would be executed in the QThreadPool thread rather than the actual destination one. Thus, if the destination slot were to emit a signal of its own, then that signal would be at risk of exhibiting the same behavior (thus merely passing the problem on rather than outright fixing it). Not to mention, based on my understanding the Qt::QueuedConnection which I would expect to be used automatically should be tied to the lifecycle of the QThread provided by the QThreadPool, meaning that that thread should stay alive for 30 seconds by default if I'm understanding the documentation properly.

Note from additional logging I can see that the line emit finished() is called and the QRunnableWrapper is destroyed. I've also tried using the built-in QRunnableWrapper::destroyed() signal rather than my custom finished() signal, but that didn't have any appreciable impact.

Some additional details to note:

Qt Version: 6.7.2
OS: Windows (10 and 11)

SGaist

Hi and welcome to devnet,

How did you implement ˋwaitForFinishedˋ ?

jbosak

The waitForFinished is in a nutshell:

bool waitForFinished(const QList<WaitWorker*> &workers) {
    int timesWaited = 0;
    while (timesWaited < 100) {
        bool isDone = true;
        foreach (WaitWorker *w, workers) {
            // make sure that all workers have completed
            if (WaitWorker::isValidInstance(w) && !w->completed()) {
                isDone = false;
                break;
            }
        }
 
        // make sure that all workers have been destroyed
        if (isDone) {
            foreach(WaitWorker *w, workers) {
                if (WaitWorker::isValidInstance(w)) {
                    isDone = false;
                    break;
                }
            }
        }

        // At this point all must be complete and destroyed
        if (isDone)
            return true;

        // Not done yet, wait a moment and try again
        timesWaited++;
        QTest::qWait(100);
    }

    return false;
}

As mentioned, the WaitWorker has a flag for tracking when it's completed it's "work" (with its work being purely just to wait).

void WaitWorker::doWork() {
    QThread::msleep(delay);
    m_complete = true;
}

With the QRunnableWrapper handling the signaling

void QRunnableWrapper::run() {
    emit started();
    m_worker->doWork();
    emit finished();
}

Writing this out I can think of two problems (which may or not may not be real problems):

m_complete is marked true before the signal finished() is emitted, to it's theoretically possible for waitForFinished to see the WaitWorker as complete before finished() is emitted. Though from my debugging statements, that doesn't appear to be the case, not to mention that it's waiting for the instance to be destroyed before returning from the wait.
I'm waiting at the end of the loop rather than at the beginning, meaning that on the first iteration it'll start it's checks before giving the workers any time to do their thing. Might moving the QTest::qWait to the start of the loop make a difference?

jeremy_k

@jbosak said in QThreadPool not always emitting (custom) finished signal:

I'm trying to make use of QThreadPool and want to be able to have a signal be emitted when the QRunnable passed in is started and finished.

This looks like functionality provided by QtConcurrent::run() and QFutureWatcher.