How to integrate with boost::asio?

[SeeRich]

I currently have all my applications networking code written with boost::asio and returning futures. For example:

std::future<std::string> restRequest(const HttpRequest& req);

However, this doesn't integrate the best with my Qt UI and I would love to integrate this with QCoro.

If I modify the method above to return a boost coroutine instead:

boost::asio::awaitable<std::string> restRequest(const HttpRequest& req);

What would I need to do to co_await it inside a QCoro::Task<>?
e.g. (barrowed from QCoro networking example)

QCoro::Task<> start()
    {
        mPb->setVisible(true);
        mBtn->setEnabled(false);
        mBtn->setText(tr("Downloading ..."));

        auto request = HttpRequest();
        ... fill in request ....

        auto response = co_await restRequest(request);
        std::cout << response << std::endl;

        mPb->setVisible(false);
        mBtn->setEnabled(true);
        mBtn->setText(tr("Done, download again"));
    }

I did try this and the code doesn't compile because the boost::asio::awaitable type doesn't meet the requirements of QCoro::Task.

Also, I am somewhat concerned that this will not work anyways due to the restRequest using a boost::asio::io_context under the hood. This means the coroutine being awaited above will actually have its IO handled by a thread assigned to the io_context.

Any suggestions appreciated!

[danvratil]

In theory the Promise::await_transform() overload should accept a type that satisfies the QCoro's Awaitable concept. But it's fairly possible that the concept is incorrect or incomplete so the boost's awaitable is not accepted, I'll look into that.

What's more problematic is the integration of the two event loops - boost's io_context and the Qt event loop. This is a problem unrelated to QCoro - you would have the same problem even without coroutines. I think the only solution is to run the boost's io_context in a separate thread - which I believe is what you did previously with the std::future-based implementation?

If you are using Qt6, you could replace std::future with QFuture and QPromise. QFuture is supported by QCoro, so this would work out-of-the-box.

If you are using Qt5 (which doesn't have QPromise), you could write your own thread-save Awaitable, which would resume the awaiting coroutine from the main thread.

[SeeRich]

Thankfully, the Qt part of my application is pretty new so I am free to keep moving forward with the new Qt 6 releases.

[SeeRich]

Currently, are the exceptions from a future i.e. (QPromise::setException()) propagated to the calling code? I made some simple changes to my code base to test this out but I can't seem to catch the exception in my QCoro::Task. The QFuture/QPromise documentation seems to indicate this is done with the continuation onFailed; however, it doesn't seem that qcoro is using this?

Here are the changes I made:

auto operator()() -> QFuture<T>
{
    auto promise = std::make_shared<QPromise<T>>();
    auto future = promise->future();
    _fut.then([p = promise](boost::future<T> completeFut) -> void {
        std::cout << "then callback..." << std::endl;
        try {
            throw std::runtime_error("TESTING EXCEPTION");
            p->addResult(std::move(completeFut.get()));
            p->finish();
        } catch(const std::exception& e) {
            std::cout << "then callback error: " << e.what() << std::endl;
            p->setException(std::current_exception());
            // p->finish();
        }
        std::cout << "then callback exit..." << std::endl;
    });
    return future;
}

and

try {
    auto adapter = QFutureAdapter<std::string>(_async_manager->asyncOperation(300ms));
    auto result = co_await adapter();
    std::cout << "ASYNC RESULT: " << result << std::endl;
} catch(const std::exception& e) {
    std::cout << "HANDLING EXCEPTION: " << e.what() << std::endl;
    co_return;
}

[danvratil]

QCoro supports only exceptions normally thrown inside a coroutine. Explicit support for rethrowing an exception stored in QFuture sounds like a sensible feature to have, and should be fairly straightforward to implement inside QCoroFuture. I'll look into it in the next days.

[danvratil]

You need to call p->finish() in the exception-handling path, otherwise the Future will never finish (and re-throw the exception).

On QCoro side, there's a simple change needed, I already merged it into main and added a bunch of tests to make sure it keeps working :)

[SeeRich]

This works exactly as expected. Thank you for all the support!

Below is a copy of the final working code with exception handling in case it is helpful to someone else.

#include "qcoro/qcoronetworkreply.h"
#include "qcoro/qcorotask.h"

#include <QApplication>
#ifdef __clang__
    #pragma clang diagnostic push
    #pragma clang diagnostic ignored \
        "-Wdeprecated-enum-enum-conversion"  // QSizePolicy seems problematic...
#endif                                       // __clang__
#include <QBoxLayout>
#ifdef __clang__
    #pragma clang diagnostic pop
#endif  // __clang__
#include <QCoroFuture>
#include <QMainWindow>
#include <QMessageBox>
#include <QNetworkAccessManager>
#include <QNetworkReply>
#include <QNetworkRequest>
#include <QProgressBar>
#include <QPushButton>

#include <boost/asio.hpp>
#include <boost/thread/future.hpp>

#include <chrono>
#include <iostream>
#include <memory>
#include <optional>
#include <string>

using namespace std::chrono_literals;
namespace net = boost::asio;  // from <boost/asio.hpp>

// Programming langauges
constexpr char raw_url[] = "http://www.wikidata.org/wiki/Special:EntityData/Q9143.json";
static const QUrl wikiUrl = QUrl{raw_url};

/// Convenience function to convert a std::exception_ptr to a boost::exception_ptr
boost::exception_ptr convert_exception_s2b(std::exception_ptr ex)
{
    try {
        std::rethrow_exception(ex);
    } catch(...) {
        return boost::current_exception();
    }
}

/// Convenience function to convert a boost::exception_ptr to a std::exception_ptr
std::exception_ptr convert_exception_b2s(boost::exception_ptr ex)
{
    try {
        boost::rethrow_exception(ex);
    } catch(...) {
        return std::current_exception();
    }
}

class AsyncManager
{
public:
    AsyncManager()
        : _work(net::require(_ioc.get_executor(), net::execution::outstanding_work.tracked))
    {
        _thread = std::thread([this]() { _ioc.run(); });
    }

    ~AsyncManager()
    {
        // Gracefully close down.
        // Reset the work object, this should block until all work is complete.
        _work = net::any_io_executor();
        if(!_ioc.stopped())
            _ioc.stop();
        _thread.join();
    }

    net::awaitable<std::string> async_op_coro(net::io_context& ioc,
                                              std::chrono::milliseconds timeRequired,
                                              bool simulateException)
    {
        // Just simulate an asynchronous operation
        net::steady_timer timer(ioc);
        timer.expires_after(std::chrono::milliseconds(timeRequired));
        co_await timer.async_wait(net::use_awaitable);
        if(simulateException)
            BOOST_THROW_EXCEPTION(std::runtime_error("Async Error Timeout"));
        co_return "Async Operation Complete!!!!";
    }

    boost::future<std::string> asyncOperation(std::chrono::milliseconds timeRequired,
                                              bool simulateException)
    {
        auto promise = std::make_shared<boost::promise<std::string>>();
        auto result = promise->get_future();
        // Spawn coroutine that will handle request.
        net::co_spawn(_ioc,
                      async_op_coro(_ioc, timeRequired, simulateException),
                      [p = std::move(promise)](std::exception_ptr e, auto&& result) {
                          if(!e) {
                              p->set_value(result);
                          } else {
                              // Convert exception types to maintain exception information
                              auto bexp = convert_exception_s2b(e);
                              p->set_exception(bexp);
                          }
                      });
        return result;
    }

private:
    /// Thread responsible for servicing the requests
    std::thread _thread;
    /// I/O context responsible for handling all I/O.
    net::io_context _ioc;
    /// Executor that will prevent the service from stopping when it
    /// runs out of work.
    net::any_io_executor _work;
};

template <typename T>
class QFutureAdapter
{
public:
    QFutureAdapter(boost::future<T>&& fut) : _fut(std::move(fut)) {}

    ~QFutureAdapter() {}

    auto operator()() -> QFuture<T>
    {
        auto promise = std::make_shared<QPromise<T>>();
        auto future = promise->future();
        _fut.then([p = promise](boost::future<T> completeFut) -> void {
            std::cout << "then callback..." << std::endl;
            if(auto exp = completeFut.get_exception_ptr()) {
                auto sexp = convert_exception_b2s(exp);
                p->setException(sexp);
            } else {
                p->addResult(std::move(completeFut.get()));
            }
            // Always mark as finished.
            p->finish();
            std::cout << "then callback exit..." << std::endl;
        });
        return future;
    }

private:
    boost::future<T> _fut;
};

class MainWindow : public QMainWindow
{
    Q_OBJECT
public:
    MainWindow()
    {
        mPb = new QProgressBar();
        mPb->setVisible(false);
        mPb->setMinimumWidth(200);
        mPb->setMinimum(0);
        mPb->setMaximum(0);

        mBtn = new QPushButton(tr("Start Download"));
        connect(mBtn, &QPushButton::clicked, this, &MainWindow::start);

        auto* vbox = new QVBoxLayout();
        vbox->addStretch(1);
        vbox->addWidget(mPb);
        vbox->addWidget(mBtn);
        vbox->addStretch(1);

        auto* hbox = new QHBoxLayout();
        hbox->addStretch(1);
        hbox->addLayout(vbox);
        hbox->addStretch(1);

        QWidget* w = new QWidget;
        w->setLayout(hbox);
        setCentralWidget(w);

        _async_manager = std::make_shared<AsyncManager>();
    }

private Q_SLOTS:
    QCoro::Task<void> start()
    {
        try {
            mPb->setVisible(true);
            mBtn->setEnabled(false);
            mBtn->setText(tr("Downloading ..."));

            // Make request expecting success
            try {
                auto adapter =
                    QFutureAdapter<std::string>(_async_manager->asyncOperation(300ms, false));
                auto result = co_await adapter();
                std::cout << "ASYNC RESULT: " << result << std::endl;
            } catch(const boost::exception& e) {
                std::cout << boost::diagnostic_information(e) << std::endl;
            } catch(const std::exception& se) {
                std::cout << se.what() << std::endl;
            }

            // Make request expecting error
            try {
                auto adapter =
                    QFutureAdapter<std::string>(_async_manager->asyncOperation(300ms, true));
                auto result = co_await adapter();
                std::cout << "ASYNC RESULT: " << result << std::endl;
            } catch(const boost::exception& e) {
                std::cout << boost::diagnostic_information(e) << std::endl;
            } catch(const std::exception& se) {
                std::cout << se.what() << std::endl;
            }

            mPb->setVisible(false);
            mBtn->setEnabled(true);
            mBtn->setText(tr("Done, download again"));

            co_return;
        } catch(const std::exception& e) {
            std::cout << "Start error: " << e.what() << std::endl;
        }
    }

private:
    std::shared_ptr<AsyncManager> _async_manager;
    QPushButton* mBtn = {};
    QProgressBar* mPb = {};
};

int main(int argc, char** argv)
{
    QApplication app(argc, argv);
    MainWindow window;
    window.showNormal();
    return app.exec();
}

#include "Main.moc"

[SeeRich]

Hey, I just want to say that your help in this discussion has helped me refactor my application tremendously, thank you!

I have one additional question related to this setup. The application that I am integrating QCoro requires that exceptions occurring in asynchronous contexts must propagate to the main thread and cause the application to close so that the stack trace and exception can be logged and reported.

All that to say I am experiencing somewhat strange behavior when using coroutines that throw exceptions.

The following code is a refactored version of the above to demonstrate the issue. I expect that calling both error functions (i.e. coAwaitErrorFunction & nonCoAwaitErrorFunction) should be caught by the try/catch block around the main application. However, the function that does something useful and has a co_await keyword (i.e. Function 1), seems to just ignore the exception.

#include "qcoro/qcoronetworkreply.h"
#include "qcoro/qcorotask.h"

#include <QApplication>
#ifdef __clang__
    #pragma clang diagnostic push
    #pragma clang diagnostic ignored \
        "-Wdeprecated-enum-enum-conversion"  // QSizePolicy seems problematic...
#endif                                       // __clang__
#include <QBoxLayout>
#ifdef __clang__
    #pragma clang diagnostic pop
#endif  // __clang__
#include <QCoroFuture>
#include <QMainWindow>
#include <QMessageBox>
#include <QNetworkAccessManager>
#include <QNetworkReply>
#include <QNetworkRequest>
#include <QProgressBar>
#include <QPushButton>

#include <boost/asio.hpp>
#include <boost/thread/future.hpp>

#include <chrono>
#include <iostream>
#include <memory>
#include <optional>
#include <string>

using namespace std::chrono_literals;
namespace net = boost::asio;  // from <boost/asio.hpp>

/// Convenience function to convert a std::exception_ptr to a boost::exception_ptr
boost::exception_ptr convert_exception_s2b(std::exception_ptr ex)
{
    try {
        std::rethrow_exception(ex);
    } catch(...) {
        return boost::current_exception();
    }
}

/// Convenience function to convert a boost::exception_ptr to a std::exception_ptr
std::exception_ptr convert_exception_b2s(boost::exception_ptr ex)
{
    try {
        boost::rethrow_exception(ex);
    } catch(...) {
        return std::current_exception();
    }
}

class AsyncManager
{
public:
    AsyncManager()
        : _work(net::require(_ioc.get_executor(), net::execution::outstanding_work.tracked))
    {
        _thread = std::thread([this]() { _ioc.run(); });
    }

    ~AsyncManager()
    {
        // Gracefully close down.
        // Reset the work object, this should block until all work is complete.
        _work = net::any_io_executor();
        if(!_ioc.stopped())
            _ioc.stop();
        _thread.join();
    }

    net::awaitable<std::string> async_op_coro(net::io_context& ioc,
                                              std::chrono::milliseconds timeRequired,
                                              bool simulateException)
    {
        // Just simulate an asynchronous operation
        net::steady_timer timer(ioc);
        timer.expires_after(std::chrono::milliseconds(timeRequired));
        co_await timer.async_wait(net::use_awaitable);
        if(simulateException)
            BOOST_THROW_EXCEPTION(std::runtime_error("Async Error Timeout"));
        co_return "Async Operation Complete!!!!";
    }

    boost::future<std::string> asyncOperation(std::chrono::milliseconds timeRequired,
                                              bool simulateException)
    {
        auto promise = std::make_shared<boost::promise<std::string>>();
        auto result = promise->get_future();
        // Spawn coroutine that will handle request.
        net::co_spawn(_ioc,
                      async_op_coro(_ioc, timeRequired, simulateException),
                      [p = std::move(promise)](std::exception_ptr e, auto&& result) {
                          if(!e) {
                              p->set_value(result);
                          } else {
                              // Convert exception types to maintain exception information
                              auto bexp = convert_exception_s2b(e);
                              p->set_exception(bexp);
                          }
                      });
        return result;
    }

private:
    /// Thread responsible for servicing the requests
    std::thread _thread;
    /// I/O context responsible for handling all I/O.
    net::io_context _ioc;
    /// Executor that will prevent the service from stopping when it
    /// runs out of work.
    net::any_io_executor _work;
};

template <typename T>
class QFutureAdapter
{
public:
    QFutureAdapter(boost::future<T>&& fut) : _fut(std::move(fut)) {}

    ~QFutureAdapter() {}

    auto operator()() -> QFuture<T>
    {
        auto promise = std::make_shared<QPromise<T>>();
        auto future = promise->future();
        _fut.then([p = promise](boost::future<T> completeFut) -> void {
            std::cout << "then callback..." << std::endl;
            if(auto exp = completeFut.get_exception_ptr()) {
                auto sexp = convert_exception_b2s(exp);
                p->setException(sexp);
            } else {
                p->addResult(std::move(completeFut.get()));
            }
            // Always mark as finished.
            p->finish();
            std::cout << "then callback exit..." << std::endl;
        });
        return future;
    }

private:
    boost::future<T> _fut;
};

class MainWindow : public QMainWindow
{
    Q_OBJECT
public:
    MainWindow()
    {
        mPb = new QProgressBar();
        mPb->setVisible(false);
        mPb->setMinimumWidth(200);
        mPb->setMinimum(0);
        mPb->setMaximum(0);

        mBtn = new QPushButton(tr("Call Error Function 1"));
        connect(mBtn, &QPushButton::clicked, this, &MainWindow::coAwaitErrorFunction);

        mBtn2 = new QPushButton(tr("Call Error Function 2"));
        connect(mBtn2, &QPushButton::clicked, this, &MainWindow::nonCoAwaitErrorFunction);

        auto* vbox = new QVBoxLayout();
        vbox->addStretch(1);
        vbox->addWidget(mPb);
        vbox->addWidget(mBtn);
        vbox->addWidget(mBtn2);
        vbox->addStretch(1);

        auto* hbox = new QHBoxLayout();
        hbox->addStretch(1);
        hbox->addLayout(vbox);
        hbox->addStretch(1);

        QWidget* w = new QWidget;
        w->setLayout(hbox);
        setCentralWidget(w);

        _async_manager = std::make_shared<AsyncManager>();
    }

private Q_SLOTS:
    QCoro::Task<void> coAwaitErrorFunction()
    {
        try {
            mPb->setVisible(true);
            mBtn->setEnabled(false);
            mBtn2->setEnabled(false);
            mBtn->setText(tr("Downloading ..."));

            // Make request expecting error
            auto adapter = QFutureAdapter<std::string>(_async_manager->asyncOperation(300ms, true));
            auto result = co_await adapter();
            std::cout << "ASYNC RESULT: " << result << std::endl;
            co_return;
        } catch(const std::exception& e) {
            std::cout << "Start error: " << e.what() << std::endl;
            mPb->setVisible(false);
            mBtn->setEnabled(true);
            mBtn2->setEnabled(true);
            mBtn->setText(tr("Done, download again"));
            // Re throw the exception
            throw;
        }
    }

    QCoro::Task<void> nonCoAwaitErrorFunction() { throw std::runtime_error("I'm a bad coroutine"); }

private:
    std::shared_ptr<AsyncManager> _async_manager;
    QPushButton* mBtn = {};
    QPushButton* mBtn2 = {};
    QProgressBar* mPb = {};
};

int main(int argc, char** argv)
{
    QApplication app(argc, argv);
    try {
        MainWindow window;
        window.showNormal();
        return app.exec();
    } catch(...) {
        std::cout << "CHANCE TO HANDLE GLOBAL UNEXPECTED ERRORS" << std::endl;
    }
}

#include "Main.moc"

[danvratil]

I'm glad that I can help and that QCoro works for you nicely 🙂

I did not try to run your code, and won't be able to investigate in depth until next week, but from your description I think you are hitting a known limitation - non-awaited coroutines don't rethrow exceptions (#97)

Basically, when a coroutine throws, the exception gets stored, the coroutine is finished immediately, and if someone is awaiting the coroutine, they get resumed. The resumed awaiter tries to obtain the result of the coroutine and at this point, the stored exception gets rethrown inside the awaiter.

In your case, the exception is thrown inside the coAwaitErrorFunction which is a coroutine slot invoked from the event loop. However the event loop doesn't explicitly co_await the slot, it just discards the returned QCoro::Task<> right away. Thus, the exception is stored, but is never rethrown, because there's nothing trying to read the result of the coroutine slot.

There's more in issue #97, and there's also an experimental branch that makes the behavior of throwing exception from non-awaited coroutine optional. However I'm quite unhappy with the implementation and haven't yet figured out another way.