Асинхронні API Qt 6

QFuture, Asynchronous, QtConcurrent, Qt, QThreadPool, Qt 6

Як читачі, можливо, вже знають, Qt надає кілька багатопоточних конструкцій (потоки, м'ютекси, стан очікування і т.д.), а також API вищого рівня, такі як QThreadPool, Qt Concurrent та інші родинні класи. У цій статті буде розказано про асинхронні API вищого рівня та зміни, внесені до Qt 6.

API паралелізму вищого рівня Qt

Qt Concurrent спрощує багатопотокове програмування, усуваючи необхідність у низькорівневій синхронізації (примітиви, такі як м'ютекси та блокування), і керує декількома потоками вручну. Він надає алгоритми зіставлення, фільтрації та скорочення (відоміші з функціонального програмування) для паралельної обробки повторюваних контейнерів. Крім того, існують такі класи, як QFuture, QFutureWatcher та QFutureSynchronizer для доступу та відстеження результатів асинхронних обчислень. Хоча всі вони дуже корисні, все ж таки були деякі недоліки, такі як неможливість використовувати QFuture поза Qt Concurrent, відсутність підтримки для об'єднання декількох обчислень для більш простого і чистого коду, відсутність гнучкості Qt Concurrent API і т. д. Що стосується Qt 6, розробники постаралися врахувати відгуки, отримані за останні роки, і зробити багатопотокове програмування на Qt більш приємним і захоплюючим!

Приєднання продовжень до QFuture

Поширеним сценарієм у багатопотоковому програмуванні є виконання асинхронного обчислення, яке, своєю чергою, має викликати інше асинхронне обчислення та передавати йому дані, які залежать від іншого, і так далі. Оскільки кожен етап вимагає результатів попереднього, вам потрібно або зачекати (шляхом блокування або опитування), поки попередній етап не завершиться і використовувати його результати, або структурувати свій код у стилі "call-callback" (зворотний виклик). Жоден з цих варіантів не ідеальний: ви або витрачаєте ресурси на очікування, або отримуєте складний код, що не підтримується. Додавання нових етапів або логіки (для обробки помилок тощо) ще більше збільшує складність.

Щоб краще зрозуміти проблему, розглянемо такий приклад. Припустимо, ви хочете завантажити велике зображення з мережі, провести з ним важку обробку і показати зображення, що вийшло у вашому додатку. Отже, ви повинні зробити такі кроки:

• зробити мережевий запит та дочекатися отримання всіх даних;
• створити зображення із необроблених даних;
• обробити зображення;
• показати зображення.

Є такі методи кожного кроку, які потрібно викликати послідовно:

QByteArray download(const QUrl &url);
QImage createImage(const QByteArray &data);
QImage processImage(const QImage &image);
void show(const QImage &image);

Можна використовувати QtConcurrent для асинхронного виконання цих завдань та QFutureWatcher для відстеження прогресу:

void loadImage(const QUrl &url) {
    QFuture data = QtConcurrent::run(download, url);
    QFutureWatcher dataWatcher;
    dataWatcher.setFuture(data);

    connect(&dataWatcher, &QFutureWatcher ::finished, this, [=] {
        // handle possible errors
        // ...
        QImage image = createImage(data);
        // Process the image
        // ...
        QFuture processedImage = QtConcurrent::run(processImage, image);
        QFutureWatcher<QImage> imageWatcher;
        imageWatcher.setFuture(processedImage);

        connect(&imageWatcher, &QFutureWatcher::finished, this, [=] {
            // handle possible errors
            // ...
            show(processedImage);
        });
    });
}

Погано виглядає, правда? Логіка програми змішана зі стандартним кодом, необхідним об'єднання компонентів програми разом. І ви знаєте, що чим більше кроків додаєте в ланцюжок, тим страшніше це стає. QFuture допомагає вирішити цю проблему, додаючи підтримку приєднання продовжень через метод QFuture::then():

auto future = QtConcurrent::run(download, url)
            .then(createImage)
            .then(processImage)
            .then(show);

Це, безперечно, виглядає набагато краще! Але не вистачає одного – обробки помилок. Ви можете зробити щось на зразок:

auto future = QtConcurrent::run(download, url)
            .then([](QByteArray data) {
                // handle possible errors from the previous step
                // ...
                return createImage(data);
            })    
            .then(...)    
            ...

Це буде працювати, але код обробки помилок, як і раніше, змішаний з логікою програми. Також, ймовірно, не варто запускати весь ланцюжок, якщо один із кроків не вдався. Це можна вирішити за допомогою методу QFuture::onFailed(), який дозволяє прикріплювати певні обробники помилок кожного можливого типу помилки:

auto future = QtConcurrent::run(download, url)
            .then(createImage)
            .then(processImage)
            .then(show)
            .onFailed([](QNetworkReply::NetworkError) {
                // handle network errors
            })
            .onFailed([](ImageProcessingError) {
                // handle image processing errors
            })
            .onFailed([] {
                // handle any other error
            });

Зверніть увагу, що для використання .onFailed() необхідно увімкнути винятки. Якщо якийсь із кроків завершується невдало з винятком, ланцюжок переривається, і викликається обробник помилок, що відповідає типу згенерованого виключення.

Подібно .then() і onFailed(), існує також .onCanceled()для прикріплення обробників у разі скасування ф'ючерса.

Створення QFuture із сигналів

Подібно до ф'ючерсів, сигнали також являють собою те, що стане доступним коли-небудь у майбутньому, тому здається природним мати можливість працювати з ними, як з ф'ючерсами, продовженнями приєднання, обробниками збоїв і т. д. Враховуючи клас MyObject на основі QObject з сигналом void mySignal(int), ви можете використовувати цей сигнал як ф'ючерс наступним чином:

QFuture intFuture = QtFuture::connect(&object, &MyObject::mySignal);

Тепер ви можете прикріпити обробники продовження, збою або скасування отриманого ф'ючерсу.

Зверніть увагу, що тип результуючого ф'ючерсу збігається із типом аргументу сигналу. Якщо немає аргументів, повертається QFuture . У разі кількох аргументів результат зберігається у std::tuple.

Повернемося до першого кроку (тобто завантаження) прикладу обробки зображень, щоб побачити, як це може бути корисно на практиці. Є багато способів реалізувати це, давайте використовуємо QNetworkAccessManager для надсилання мережного запиту та отримання даних:

QNetworkAccessManager manager;    
...

QByteArray download(const QUrl &url) {        
    QNetworkReply *reply = manager.get(QNetworkRequest(url));
    QObject::connect(reply, &QNetworkReply::finished, [reply] {...});

    // wait until we've received all data
    // ...    
    return data;        
}

Але очікування блокування вище це не дуже добре, було б краще, якби можна було позбутися цього і натомість сказати "коли QNetworkAccessManager отримає дані, створіть зображення, потім обробіть його та покажіть»). Це можна зробити, підключивши сигнал finished() диспетчера мережевого доступу до QFuture:

QNetworkReply *reply = manager.get(QNetworkRequest(url));

auto future = QtFuture::connect(reply, &QNetworkReply::finished)
        .then([reply] {
            return reply->readAll();
        })
        .then(QtFuture::Launch::Async, createImage)
        .then(processImage)
        .then(show)        
        ...

Ви можете помітити, що замість використання QtConcurrent::run() для асинхронного завантаження і повернення даних у новому потоці просто підключаємося до сигналу QNetworkAccessManager::finished(), який запускає ланцюжок обчислень. Також зверніть увагу на додатковий параметр у наступному рядку:

 .then(QtFuture::Launch::Async, createImage)

За промовчанням продовження, прикріплені за допомогою .then(), викликаються у тому потоці, у якому був запущений батьківський потік (у разі основний потік). Тепер, коли не використовується QtConcurrent::run() для асинхронного запуску ланцюжка, потрібно передати додатковий параметр QtFuture::Launch::Async, щоб запустити ланцюжок продовжень в окремому потоці і уникнути блокування інтерфейсу користувача.

Створення QFuture

До цього часу єдиним «офіційним» способом створення та збереження значення всередині QFuture було використання одного з методів QtConcurrent. Отже поза QtConcurrent, QFuture був дуже корисний. У Qt 6 нарешті з'явився "setter" ("сеттер") аналог QFuture: QPromise, представлений Андрієм Голубєвим. Його можна використовувати для встановлення значень, ходу виконання та винятків для асинхронних обчислень, яких пізніше можна буде отримати доступ через QFuture. Щоб продемонструвати, як це працює, давайте ще раз перепишемо приклад обробки зображення та скористаємося класом QPromise:

QFuture download(const QUrl &url) {
    QPromise promise;
    QFuture future = promise.future();

    promise.reportStarted(); // notify that download is started

    QNetworkReply *reply = manager.get(QNetworkRequest(url));
    QObject::connect(reply, &QNetworkReply::finished,
            [reply, p = std::move(promise)] {
                p.addResult(reply->readAll());
                p.reportFinished(); // notify that download is finished
                reply->deleteLater();
            });

    return future;
}
auto future = download()
        .then(QtFuture::Launch::Async, createImage)
        .then(processImage)
        .then(show)
        ...

Зміни в QtConcurrent

Завдяки Яреку Кобусу, Мартену Нордхейму, Карстену Хаймріху, Тимуру Почепцову та Віталію Фанаскову, QtConcurrent також отримав хороші оновлення. Існуючі API отримали деякі покращення, зокрема:

• Тепер ви можете встановити власний пул потоків для всіх методів QtConcurrent замість того, щоб завжди запускати їх у глобальному пулі потоків та потенційно блокувати виконання інших завдань.
• Map об'єкти алгоритми їх скорочення та фільтрації тепер можуть приймати початкове значення, тому вам не потрібно шукати обхідні шляхи для типів, які не мають конструктора за замовчуванням.
• QtConcurrent::run був покращений для роботи зі змінною кількістю аргументів та типами, призначеними тільки для переміщення.

Крім того, були додані два нових API QtConcurrent, щоб надати користувачам велику гнучкість. Давайте подивимося на них докладніше.

QtConcurrent::runWithPromise

Новий метод QtConcurrent::runWithPromise(), розроблений Яреком Кобусом, є ще одним приємним доповненням до фреймворку QtConcurrent. Він дуже схожий на QtConcurrent::run(), за винятком того, що робить об'єкт QPromise, пов'язаний з даним завданням, доступним для користувача. Це досягається за рахунок того, що runnable, переданий QtConcurrent::runWithPromise(), приймає посилання на об'єкт QPromise:

auto future = QtConcurrent::runWithPromise(
            [] (QPromise &promise, /* other arguments may follow */ ...) {
                // ...
                for (auto value : listOfValues) {
                    if (promise.isCanceled())
                        // handle the cancellation

                // do some processing...

                promise.addResult(...);
                promise.setProgressValue(...);
                }
            },
            /* pass other arguments */ ...);

Як бачите, з runWithPromise() користувачі мають більший контроль над завданням і можуть реагувати на запити скасування або призупинення, робити звіти про хід виконання тощо, що неможливо з QtConcurrent::run().

QtConcurrent::task

QtConcurrent::task() надає вільний інтерфейс для виконання завдання в окремому потоці. Це більш сучасна альтернатива QtConcurrent::run(), яка дозволяє зручніше налаштовувати завдання. Замість використання одного з небагатьох перезавантажень QtConcurrent::run()для передачі параметрів, призначених для запуску завдання, ви можете вказати їх у будь-якому порядку, пропустити ті, які не потрібні, і так далі. Наприклад:

QFuture future = QtConcurrent::task(doSomething)
        .withArguments(1, 2, 3)
        .onThreadPool(pool)
        .withPriority(10)
        .spawn();

Зверніть увагу, що на відміну від run(), ви також можете передати пріоритет завдання.