Update UI from multiple threads

[aiphae]

I have a function that is separated across multiple threads and I want to implement some progress bar that those threads can contribute to.
I have a thread pool:

class ThreadPool {
public:
    explicit ThreadPool(int threadNumber);
    ~ThreadPool();

    template<typename F, typename... Args>
    auto enqueue(F&& f, Args&&... args)
        -> std::future<typename std::invoke_result<F, Args...>::type>;

private:
    std::vector<std::thread> workers;
    std::queue<std::function<void()>> tasks;

    std::mutex queueMutex;
    std::condition_variable condition;
    bool stop = false;
};

The pool is initialized with std::thread::hardware_concurrency() - 1 number of threads (so there is one free thread for GUI).
In my function I do this:

std::atomic<int> framesAnalyzed = 0;
for (int i = 0; i < totalFrames; ++i) {
    int index = i;
    cv::Mat frame = getMatAtFrame(source.files, index);
    if (frame.empty()) {
        continue;
    }

    pool.enqueue([&]() {
        double quality = Frame::estimateQuality(frame);
        source.sorted[index].second = quality;
        int done = ++framesAnalyzed;
        emit sortingProgressUpdated(done);
    });
}

The sortingProgressUpdated(int current) signal is connected this way:

connect(this, &StackPage::sortingProgressUpdated, this, [this](int current) {
        ui->analyzingProgressEdit->setText(QString::number(current) + "/" + QString::number(totalFrames));
    });

However, the progress text does not change in real time, and after the function completes, the final number does not match totalFrames, though I'm using an atomic counter.

What am I doing wrong?

[aiphae]

@Christian-Ehrlicher

Okay, I finally saw the issue. In my code std::this_thread::sleep_for(std::chrono::milliseconds(100)); was outside the pool.enqueue() function. So the main thread was constantly sleeping and couldn't process the signals.

[Pl45m4]

@aiphae

You must not access the Qt GUI from other threads than the main thread.

[aiphae]

@Pl45m4 I know. That's why I'm emitting a signal to the main thread.

[JonB]

@aiphae said in Update UI from multiple threads:

However, the progress text does not change in real time, and after the function completes, the final number does not match totalFrames, though I'm using an atomic counter.

So if it were me in order to see whether I have the fundamentals right I would make the connect() use Qt::BlockingQueuedConnection and verify that makes both of these work? Then if that works and your example fails you know it is something to do with asynchronous signal processing?

[Christian Ehrlicher]

But you use the wrong context - it must be an object in the main thread. Fix you connect.

[aiphae]

@Christian-Ehrlicher said in Update UI from multiple threads:

But you use the wrong context - it must be an object in the main thread. Fix you connect.

The connecting itself happens in the main thread. What do you mean?

[Christian Ehrlicher]

@aiphae said in Update UI from multiple threads:

What do you mean?

Where does 'this' lives in? Not in the main thread I would guess (but to few code)

[aiphae]

@Christian-Ehrlicher 'this' is a widget class. It lives in the main thread.

I found the issue: getMatAtFrame() loads images from a disk and it is slower than actual processing. I'll need some different concurrency structute.

[JonB]

@aiphae
I don't see how this would cause your reported

However, the progress text does not change in real time, and after the function completes, the final number does not match totalFrames

It should still change in "real time" for whatever speed it manages, and the total ought be correct....

And if @Christian-Ehrlicher says you have the wrong context for a thread connection I would also believe him....

[aiphae]

@JonB

You were right.

Here's the minimum example of the program (mainwindow.h and mainwindow.cpp) :

#ifndef MAINWINDOW_H
#define MAINWINDOW_H

#include <QMainWindow>

#include <vector>
#include <thread>
#include <queue>
#include <mutex>
#include <condition_variable>
#include <functional>
#include <future>

class ThreadPool {
public:
    explicit ThreadPool(int threadNumber) {
        for (int i = 0; i < threadNumber; ++i) {
            workers.emplace_back([this] {
                for (;;) {
                    std::function<void()> task;
                    {
                        std::unique_lock<std::mutex> lock(queueMutex);
                        condition.wait(lock, [this] { return stop || !tasks.empty(); });
                        if (stop && tasks.empty()) {
                            return;
                        }
                        task = std::move(tasks.front());
                        tasks.pop();
                    }
                    task();
                }
            });
        }
    }
    ~ThreadPool() {
        {
            std::unique_lock<std::mutex> lock(queueMutex);
            stop = true;
        }
        condition.notify_all();
        for (auto &worker : workers) {
            worker.join();
        }
    }

    template<typename F, typename... Args>
    auto enqueue(F&& f, Args&&... args)
        -> std::future<typename std::invoke_result<F, Args...>::type> {
        using return_type = typename std::invoke_result<F, Args...>::type;

        auto task = std::make_shared<std::packaged_task<return_type()>>(
            std::bind(std::forward<F>(f), std::forward<Args>(args)...)
            );

        std::future<return_type> res = task->get_future();
        {
            std::unique_lock<std::mutex> lock(queueMutex);
            if (stop) {
                throw std::runtime_error("enqueue on stopped ThreadPool");
            }
            tasks.emplace([task]() { (*task)(); });
        }
        condition.notify_one();
        return res;
    }

private:
    std::vector<std::thread> workers;
    std::queue<std::function<void()>> tasks;

    std::mutex queueMutex;
    std::condition_variable condition;
    bool stop = false;
};

QT_BEGIN_NAMESPACE
namespace Ui {
class MainWindow;
}
QT_END_NAMESPACE

class MainWindow : public QMainWindow
{
    Q_OBJECT

public:
    MainWindow(QWidget *parent = nullptr);
    ~MainWindow();

signals:
    void update(int value);

private:
    Ui::MainWindow *ui;
    void start();
    std::atomic<int> total = 0;
};
#endif // MAINWINDOW_H

#include "mainwindow.h"
#include "ui_mainwindow.h"

MainWindow::MainWindow(QWidget *parent)
    : QMainWindow(parent)
    , ui(new Ui::MainWindow)
{
    ui->setupUi(this);
    connect(this, &MainWindow::update, this, [](int value) {
        qDebug() << value;
    });
    start();
}

MainWindow::~MainWindow()
{
    delete ui;
}

void MainWindow::start() {
    ThreadPool pool(std::thread::hardware_concurrency() - 1);
    total = 0;
    for (int i = 0; i < 100; ++i) {
        pool.enqueue([this]() {
            int value = ++total;
            std::this_thread::sleep_for(std::chrono::milliseconds(1000));
            emit update(value);
        });
    }
}

[Christian Ehrlicher]

@aiphae said in Update UI from multiple threads:

void MainWindow::start() {
    ThreadPool pool(std::thread::hardware_concurrency() - 1);
    ....
}

~ThreadPool() {
    {
        std::unique_lock<std::mutex> lock(queueMutex);
        stop = true;
    }
    condition.notify_all();
    for (auto &worker : workers) {
        worker.join();
    }
}

What do you expect? You block the main event loop in the dtor.

/edit:
My Widget I used so it's a fully functional example which does not block:

class MyWidget : public QLabel
{
    Q_OBJECT
public:
    MyWidget()
        : pool(std::thread::hardware_concurrency() - 1)
    {
        connect(this, &MyWidget::updateValue, this, [this](int value) {
            setText(QString::number(value));
            });
    }
    void start() {
        total = 0;
        for (int i = 0; i < 100; ++i) {
            pool.enqueue([this]() {
                int value = ++total;
                std::this_thread::sleep_for(std::chrono::milliseconds(1000));
                emit updateValue(value);
                });
        }
    }
    std::atomic<int> total = 0;
    ThreadPool pool;
Q_SIGNALS:
    void updateValue(int);
};

[aiphae]

@Christian-Ehrlicher Okay, I moved ThreadPool to the private members of StackPage and now I connect the signal this way:

connect(this, &StackPage::sortingProgressUpdated, this, [this](int current) {
        qDebug() << current;
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
        ui->analyzingProgressEdit->setText(QString::number(current) + "/" + QString::number(totalFrames));
    }, Qt::QueuedConnection);

qDebug() prints concurrently as expected, but the ui->analyzingProgressEdit doesn't get updated. Why?

UPD:
I added ui->analyzingProgressEdit->repaint() right after changing its value. It looks like all the slots connected to the signal are queued and executed only after the threads complete their tasks because the UI doesn't change at all at the beginning of the function (actual concurrent processing) and at the end updates very rapidly (processing queued signals).

I changed the number of threads for the thread pool to just 2 (though it could take all available threads?) - nothing changed.

UPD2:
Current code:

#include "mainwindow.h"
#include "ui_mainwindow.h"
#include <QThread>

MainWindow::MainWindow(QWidget *parent)
    : QMainWindow(parent)
    , ui(new Ui::MainWindow)
    , pool(std::thread::hardware_concurrency() - 1)
{
    ui->setupUi(this);
    connect(this, &MainWindow::update, this, [](int value) {
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
        qDebug() << value << "from thread" << QThread::currentThread();
    });
    start();
}

MainWindow::~MainWindow()
{
    delete ui;
}

void MainWindow::start() {
    total = 0;
    for (int i = 0; i < 100; ++i) {
        pool.enqueue([this]() {
            int value = ++total;
            qDebug() << "Emitting signal with value" << value;
            emit update(value);
        });
    }
}

Application output shows this:

Emitting signal with value 80
Emitting signal with value 90
Emitting signal with value 91
2 from thread QThread(0x2140d6406f0, name = "Qt mainThread")
4 from thread QThread(0x2140d6406f0, name = "Qt mainThread")
12 from thread QThread(0x2140d6406f0, name = "Qt mainThread")
...

so only after all signals from corresponding values are emitted (up to several hundreds or thousands depending on the data), their slots are executed.

[Christian Ehrlicher]

I don't know what you are doing wrong but my code works and even updates the ui. So either provide a minimal compilable example or fix your code...

#include <queue>
class ThreadPool {
public:
    explicit ThreadPool(int threadNumber) {
        for (int i = 0; i < threadNumber; ++i) {
            workers.emplace_back([this] {
                for (;;) {
                    std::function<void()> task;
                    {
                        std::unique_lock<std::mutex> lock(queueMutex);
                        condition.wait(lock, [this] { return stop || !tasks.empty(); });
                        if (stop && tasks.empty())
                            return;
                        task = std::move(tasks.front());
                        tasks.pop();
                    }
                    task();
                }
                });
        }
    }
    ~ThreadPool() {
        {
            std::unique_lock<std::mutex> lock(queueMutex);
            stop = true;
        }
        condition.notify_all();
        for (auto& worker : workers)
            worker.join();
    }

    template<typename F, typename... Args>
    auto enqueue(F&& f, Args&&... args)
        -> std::future<typename std::invoke_result<F, Args...>::type> {
        using return_type = typename std::invoke_result<F, Args...>::type;

        auto task = std::make_shared<std::packaged_task<return_type()>>(
            std::bind(std::forward<F>(f), std::forward<Args>(args)...)
        );

        std::future<return_type> res = task->get_future();
        {
            std::unique_lock<std::mutex> lock(queueMutex);
            if (stop)
                throw std::runtime_error("enqueue on stopped ThreadPool");
            tasks.emplace([task]() { (*task)(); });
        }
        condition.notify_one();
        return res;
    }
private:
    std::vector<std::thread> workers;
    std::queue<std::function<void()>> tasks;

    std::mutex queueMutex;
    std::condition_variable condition;
    bool stop = false;
};

class MyWidget : public QLabel
{
    Q_OBJECT
public:
    MyWidget()
        : pool(std::thread::hardware_concurrency() - 1)
    {
        connect(this, &MyWidget::updateValue, this, [this](int value) {
            setText(QString::number(value));
            qDebug() << QThread::currentThread() << value;
        });
    }
    void start() {
        for (int i = 0; i < 100; ++i) {
            pool.enqueue([this]() {
                int value = ++total;
                std::this_thread::sleep_for(std::chrono::milliseconds(1000));
                emit updateValue(value);
                });
        }
    }
    std::atomic<int> total = 0;
    ThreadPool pool;
Q_SIGNALS:
    void updateValue(int);
};

int main(int argc, char* argv[])
{
    QApplication a(argc, argv);
    MyWidget w;
    w.show();
    qDebug() << QThread::currentThread();
    QTimer::singleShot(0, &w, &MyWidget::start);
    return a.exec();
}
#include <main.moc>

[aiphae]

@Christian-Ehrlicher

Okay, I finally saw the issue. In my code std::this_thread::sleep_for(std::chrono::milliseconds(100)); was outside the pool.enqueue() function. So the main thread was constantly sleeping and couldn't process the signals.