How to correctly do simultaneous pub and sub?

[chipsik-afk]

I'm trying to use the client for simultaneous pub/sub, and it becomes unresponsive when exceeding Receive Maximum. This only happens when the client is subscribed with QoS1 or QoS2. Please find below some code reproducing it - on my default mosquitto installation, which has Receive Maximum equal to 20, only 30 or 40 messages are published, and then the client hangs. It seems to be constantly entering if (ec == disconnect_reason_code::receive_maximum_exceeded) in endpoint_send.hpp. I didn't diagnose it further, because this is probably a result of me using the client incorrectly. I apologize for this, and ask for some directions on how to use it correctly for this scenario.

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

#include <boost/asio.hpp>

#include <async_mqtt/all.hpp>

namespace as = boost::asio;
namespace am = async_mqtt;

using client_t = am::client<am::protocol_version::v5, am::protocol::mqtt>;

as::awaitable<void>
proc(
    client_t& amcl,
    std::string_view host,
    std::string_view port,
    std::string topic,
    am::qos sub_qos) {

    auto exe = co_await as::this_coro::executor;

    try {
        co_await amcl.async_underlying_handshake(host, port, as::use_awaitable);
        co_await amcl.async_start(
            am::v5::connect_packet{true, 0x1234, "", std::nullopt, "", ""},
            as::use_awaitable
        );

        std::vector<am::topic_subopts> sub_entry{{topic, sub_qos}};
        co_await amcl.async_subscribe(
            am::v5::subscribe_packet{
                *amcl.acquire_unique_packet_id(),
                am::force_move(sub_entry)
            },
            as::use_awaitable
        );

        std::cout << "begin receive loop" << std::endl;
        bool quit{false};
        while (!quit) {
            auto pv_opt = co_await amcl.async_recv(as::use_awaitable);
            BOOST_ASSERT(pv_opt);
            pv_opt->visit(
                am::overload{
                    [&](client_t::publish_packet& p) {
                        quit = (p.payload() == "quit");
                    },
                    [&](client_t::disconnect_packet& p) {
                        std::cout << p << std::endl;
                    },
                    [](auto&) {
                    }
                }
            );
        }
        std::cout << "receive loop ended" << std::endl;

        std::vector<am::topic_sharename> unsub_entry{topic};
        co_await amcl.async_unsubscribe(
            am::v5::unsubscribe_packet{
                *amcl.acquire_unique_packet_id(),
                am::force_move(unsub_entry)
            },
            as::use_awaitable
        );

        co_await amcl.async_disconnect(as::use_awaitable);
    }
    catch (boost::system::system_error const& se) {
    }
}

int main(int argc, char* argv[]) {
    am::setup_log(
        am::severity_level::warning,
        true // log colored
    );
    if (argc != 3) {
        std::cout << "Usage: " << argv[0] << " host port" << std::endl;
        return -1;
    }

    as::io_context ioc1;
    auto amcl1 = client_t{ioc1.get_executor()};
    auto guard1 = std::make_optional(as::make_work_guard(ioc1));

    as::io_context ioc2;
    auto amcl2 = client_t{ioc2.get_executor()};
    auto guard2 = std::make_optional(as::make_work_guard(ioc2));

    auto qos = am::qos::at_least_once;

    std::jthread worker1{
        [&](){
            as::co_spawn(amcl1.get_executor(), proc(amcl1, argv[1], argv[2], "topic1", qos), as::detached);
            ioc1.run();
        }
    };

    std::jthread worker2{
        [&](){
            as::co_spawn(amcl2.get_executor(), proc(amcl2, argv[1], argv[2], "topic2", qos), as::detached);
            ioc2.run();
        }
    };

    std::this_thread::sleep_for(std::chrono::milliseconds(1000));

    auto get_packet_id = [](client_t& amcl, am::qos qos) -> am::packet_id_type {
        if (qos == am::qos::at_most_once) {
            return 0;
        }
        return amcl.async_acquire_unique_packet_id_wait_until(as::use_future).get();
    };

    for (int i = 0; i != 100; ++i) {
        amcl1.async_publish(
            am::v5::publish_packet{
                get_packet_id(amcl1, qos),
                "topic2",
                std::to_string(i),
                qos
            },
            as::detached
        );

        amcl2.async_publish(
            am::v5::publish_packet{
                get_packet_id(amcl2, qos),
                "topic1",
                std::to_string(i),
                qos
            },
            as::detached
        );
    }

    amcl1.async_publish(
        am::v5::publish_packet{
            get_packet_id(amcl1, qos),
            "topic2",
            "quit",
            qos
        },
        as::use_future
    ).get();

    amcl2.async_publish(
        am::v5::publish_packet{
            get_packet_id(amcl2, qos),
            "topic1",
            "quit",
            qos
        },
        as::use_future
    ).get();

    guard1.reset();
    guard2.reset();
}

[redboltz]

@chipsik-afk
Thank you for reporting the issue. I reproduced the problem. I will take a look it.

[redboltz]

All async_mqtt member funcitons that start with async_ dispatch the request to the corresponding executor.
See https://redboltz.github.io/async_mqtt/doc/latest/async_mqtt/functionality/thread_safe.html

However, even if the async_mqtt do that, coroutine mechanism override it by coroutine's executor.
So the code example you post is thread unsafe.
amcl1.async_publish() is called from main() function. Even if amcl1 is running on ioc1.executor() and async_mqtt post the request to it, main()'s implicit executor is used. It is coroutine's limitation.

In order to solve the issue, you can use co_spawn for publish processing.
And it is the correct way to pub/sub simultaneously.
Instead of calling async_mqtt functions from the main thread, creating a coroutine for calling async_mqtt functions and the croutine's executor should be the same as async_mqtt client's one.

Here is updated code based on your code. Could you try this?

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

#include <boost/asio.hpp>

#include <async_mqtt/all.hpp>

namespace as = boost::asio;
namespace am = async_mqtt;

using client_t = am::client<am::protocol_version::v5, am::protocol::mqtt>;

as::awaitable<void>
proc(
    client_t& amcl,
    std::string_view host,
    std::string_view port,
    std::string topic,
    am::qos sub_qos) {

    auto exe = co_await as::this_coro::executor;

    try {
        co_await amcl.async_underlying_handshake(host, port, as::use_awaitable);
        co_await amcl.async_start(
            am::v5::connect_packet{true, 0x1234, "", std::nullopt, "", ""},
            as::use_awaitable
        );

        std::vector<am::topic_subopts> sub_entry{{topic, sub_qos}};
        co_await amcl.async_subscribe(
            am::v5::subscribe_packet{
                *amcl.acquire_unique_packet_id(),
                am::force_move(sub_entry)
            },
            as::use_awaitable
        );

        std::cout << "begin receive loop" << std::endl;
        bool quit{false};
        while (!quit) {
            auto pv_opt = co_await amcl.async_recv(as::use_awaitable);
            BOOST_ASSERT(pv_opt);
            pv_opt->visit(
                am::overload{
                    [&](client_t::publish_packet& p) {
                        quit = (p.payload() == "quit");
                    },
                    [&](client_t::disconnect_packet& p) {
                        std::cout << p << std::endl;
                    },
                    [](auto&) {
                    }
                }
            );
        }
        std::cout << "receive loop ended" << std::endl;

        std::vector<am::topic_sharename> unsub_entry{topic};
        co_await amcl.async_unsubscribe(
            am::v5::unsubscribe_packet{
                *amcl.acquire_unique_packet_id(),
                am::force_move(unsub_entry)
            },
            as::use_awaitable
        );

        co_await amcl.async_disconnect(as::use_awaitable);
    }
    catch (boost::system::system_error const& se) {
    }
}

int main(int argc, char* argv[]) {
    am::setup_log(
        am::severity_level::warning,
        true // log colored
    );
    if (argc != 3) {
        std::cout << "Usage: " << argv[0] << " host port" << std::endl;
        return -1;
    }

    as::io_context ioc1;
    auto amcl1 = client_t{ioc1.get_executor()};
    auto guard1 = std::make_optional(as::make_work_guard(ioc1));

    as::io_context ioc2;
    auto amcl2 = client_t{ioc2.get_executor()};
    auto guard2 = std::make_optional(as::make_work_guard(ioc2));

    auto qos = am::qos::at_least_once;

    std::jthread worker1{
        [&](){
            as::co_spawn(amcl1.get_executor(), proc(amcl1, argv[1], argv[2], "topic1", qos), as::detached);
            ioc1.run();
        }
    };
    std::jthread worker2{
        [&](){
            as::co_spawn(amcl2.get_executor(), proc(amcl2, argv[1], argv[2], "topic2", qos), as::detached);
            ioc2.run();
        }
    };
    std::this_thread::sleep_for(std::chrono::milliseconds(1000));

    as::co_spawn(
        ioc1.get_executor(),
        [&]() -> as::awaitable<void> {
            for (int i = 0; i != 100; ++i) {
                am::packet_id_type pid = 0;
                if (qos != am::qos::at_most_once) {
                    pid = co_await amcl1.async_acquire_unique_packet_id_wait_until(as::use_awaitable);
                }
                co_await amcl1.async_publish(
                    am::v5::publish_packet{
                        pid,
                        "topic2",
                        std::to_string(i),
                        qos
                    },
                    as::use_awaitable
                );
            }
            am::packet_id_type pid = 0;
            if (qos != am::qos::at_most_once) {
                pid = co_await amcl1.async_acquire_unique_packet_id_wait_until(as::use_awaitable);
            }
            co_await amcl1.async_publish(
                    am::v5::publish_packet{
                    pid,
                    "topic2",
                    "quit",
                    qos
                },
                as::use_awaitable
            );
            co_return;
        }(),
        as::detached
    );

    as::co_spawn(
        ioc2.get_executor(),
        [&]() -> as::awaitable<void> {
            for (int i = 0; i != 100; ++i) {
                am::packet_id_type pid = 0;
                if (qos != am::qos::at_most_once) {
                    pid = co_await amcl2.async_acquire_unique_packet_id_wait_until(as::use_awaitable);
                }
                co_await amcl2.async_publish(
                    am::v5::publish_packet{
                        pid,
                        "topic1",
                        std::to_string(i),
                        qos
                    },
                    as::use_awaitable
                );
            }
            am::packet_id_type pid = 0;
            if (qos != am::qos::at_most_once) {
                pid = co_await amcl2.async_acquire_unique_packet_id_wait_until(as::use_awaitable);
            }
            co_await amcl2.async_publish(
                    am::v5::publish_packet{
                    pid,
                    "topic1",
                    "quit",
                    qos
                },
                as::use_awaitable
            );
            co_return;
        }(),
        as::detached
    );

    guard1.reset();
    guard2.reset();
}

Also you can refine the main function like this. The meaning of the code is the same as above one.

int main(int argc, char* argv[]) {
    am::setup_log(
        am::severity_level::warning,
        true // log colored
    );
    if (argc != 3) {
        std::cout << "Usage: " << argv[0] << " host port" << std::endl;
        return -1;
    }

    as::io_context ioc1;
    auto amcl1 = client_t{ioc1.get_executor()};
    auto guard1 = std::make_optional(as::make_work_guard(ioc1));

    as::io_context ioc2;
    auto amcl2 = client_t{ioc2.get_executor()};
    auto guard2 = std::make_optional(as::make_work_guard(ioc2));

    auto qos = am::qos::at_least_once;

    std::jthread worker1{
        [&](){
            as::co_spawn(amcl1.get_executor(), proc(amcl1, argv[1], argv[2], "topic1", qos), as::detached);
            ioc1.run();
        }
    };
    std::jthread worker2{
        [&](){
            as::co_spawn(amcl2.get_executor(), proc(amcl2, argv[1], argv[2], "topic2", qos), as::detached);
            ioc2.run();
        }
    };
    std::this_thread::sleep_for(std::chrono::milliseconds(1000));

    auto publish_proc = [](auto& amcl, auto qos, auto topic, auto payload) -> as::awaitable<void> {
        am::packet_id_type pid = 0;
        if (qos != am::qos::at_most_once) {
            pid = co_await amcl.async_acquire_unique_packet_id_wait_until(as::use_awaitable);
        }
        co_await amcl.async_publish(
            am::v5::publish_packet{
                pid,
                am::force_move(topic),
                am::force_move(payload),
                qos
            },
            as::use_awaitable
        );
        co_return;
    };

    as::co_spawn(
        ioc1.get_executor(),
        [&]() -> as::awaitable<void> {
            for (int i = 0; i != 100; ++i) {
                co_await publish_proc(amcl1, qos, "topic2", std::to_string(i));
            }
            co_await publish_proc(amcl1, qos, "topic2", "quit");
            co_return;
        }(),
        as::detached
    );

    as::co_spawn(
        ioc2.get_executor(),
        [&]() -> as::awaitable<void> {
            for (int i = 0; i != 100; ++i) {
                co_await publish_proc(amcl2, qos, "topic1", std::to_string(i));
            }
            co_await publish_proc(amcl2, qos, "topic1", "quit");
            co_return;
        }(),
        as::detached
    );

    guard1.reset();
    guard2.reset();
}

[chipsik-afk]

Thank you very much! I have a better understanding now, and I also see how I was using the client not-thread-safely in other ways.

If anyone else reads this, my code above has additional bug, because one of the "quit" messages can be received while there are still publishes queued. For this toy example it's enough to add some sleep after the receiving loop:

std::cout << "receive loop ended" << std::endl;
boost::asio::steady_timer timer(exe);
timer.expires_after(std::chrono::seconds(2));
co_await timer.async_wait(boost::asio::use_awaitable);

I've been struggling with this for some time, and I used a workaround of publishing one-by-one (and waiting for completion), I'm very happy to know that there is a solution.

I really like your code, and learned a lot about C++ and Boost by reading it. If you don't mind, I have follow-up questions to verify my understanding:

• in this situation, other async_ methods (e.g. async_subscribe) must also be called using a coroutine wrapper?
• in non-coroutine (callback based) usage, there is no need for special approach, and async_ methods can be called directly from any thread?

Thank you very much.

[redboltz]

• in this situation, other async_ methods (e.g. async_subscribe) must also be called using a coroutine wrapper?

If you some_amcl.call async_*() function from the outside of the executor equivalent to amcl.get_executor(), you always need to use co_spawn approach. Note that after co_await is called, the executor is back to the co_spawn's 1st argument one.

• in non-coroutine (callback based) usage, there is no need for special approach, and async_ methods can be called directly from any thread?

Yes.

[redboltz]

Coroutine is very difficult.

Here is a code that demonstrate coroutine and thread and executor behavior.
The first two approaches, Line 107 and 117 work on unexpected executor.
The third approach (co_spawn), works on expected executor.
https://godbolt.org/z/eevfhTPje

Here is callback based code. Line 105 and 115 work in the valid (expected) executor.
https://godbolt.org/z/GdhhrWxhh

You need to understand this complecated behavior if you want to use both multi threading and coroutne.
I am not sure but in the most case, just calling co_spawn would be a solution like as::post().

NOTE: I worte the codes to understand the coroutine and async operation behavior for me long time ago. They contains some of unrelated features to your question. Please ignore them.

[chipsik-afk]

I will study the examples to gain better understanding.

I'm sorry for taking your time, and thank you for your help.

[chipsik-afk]

I'm really sorry, but I encounter the same problem when using callbacks - after ~30 messages, the client becomes stuck in endpoint_send.hpp:44. I tried wrapping async_ calls in as::post(), but that didn't help. Could you take a look at this code (based on example/cl_cpp17_mqtt_*) and tell me what I'm doing wrong?

#include <iostream>
#include <optional>
#include <string>
#include <thread>

#include <boost/asio.hpp>

#include <async_mqtt/all.hpp>

namespace as = boost::asio;
namespace am = async_mqtt;

using client_t = am::client<am::protocol_version::v5, am::protocol::mqtt>;

constexpr auto qos = am::qos::at_least_once;

struct app {
    app(as::any_io_executor exe, std::string_view host, std::string_view port, std::string_view sub_topic, std::string_view pub_topic)
        : cli_{exe}, host_{host}, port_{port}, sub_topic_{sub_topic}, pub_topic_{pub_topic}
    {
    }

    void connect() {
        cli_.async_underlying_handshake(
            host_,
            port_,
            [this](auto&&... args) {
                handle_underlying_handshake(
                    std::forward<std::remove_reference_t<decltype(args)>>(args)...
                );
            }
        );
    }

    void publish(std::string payload) {
        if (qos == am::qos::at_most_once)
        {
            cli_.async_publish(
                pub_topic_,
                payload,
                qos,
                [this](auto&&... args) {
                    handle_publish_response(
                        std::forward<std::remove_reference_t<decltype(args)>>(args)...
                    );
                }
            );

        }
        else
        {
            // need to use async version, or we'll get BOOST_ASSERT in endpoint_send.hpp:56
            cli_.async_acquire_unique_packet_id(
                [this, payload](auto&&... args) {
                    handle_acquire_unique_packet_id(
                        payload, 
                        std::forward<std::remove_reference_t<decltype(args)>>(args)...
                    );
                }
            );
        }
    }

private:
    void handle_acquire_unique_packet_id(
        std::string payload,
        am::error_code ec,
        std::optional<am::packet_id_type> pid
    ) {
        std::cout << "acquire_unique_packet_id:" << ec.message() << std::endl;
        if (ec) return;
        if (pid) {
            std::cout << "pid: " << *pid << std::endl;
            cli_.async_publish(
                *pid,
                pub_topic_,
                payload,
                qos,
                [this](auto&&... args) {
                    handle_publish_response(
                        std::forward<std::remove_reference_t<decltype(args)>>(args)...
                    );
                }
            );
        }
    }

    void handle_publish_response(
        am::error_code ec,
        client_t::pubres_type pubres
    ) {
        std::cout << "publish:" << ec.message() << std::endl;
        if (ec) return;
        if (pubres.puback_opt) {
            std::cout << *pubres.puback_opt << std::endl;
        }
        if (pubres.pubrec_opt) {
            std::cout << *pubres.pubrec_opt << std::endl;
        }
        if (pubres.pubcomp_opt) {
            std::cout << *pubres.pubcomp_opt << std::endl;
        }
    }

    void reconnect() {
        tim_.expires_after(std::chrono::seconds{1});
        tim_.async_wait(
            [this](am::error_code const& ec) {
                if (!ec) {
                    connect();
                }
            }
        );
    }

    void handle_underlying_handshake(
        am::error_code ec
    ) {
        std::cout << "underlying_handshake:" << ec.message() << std::endl;
        if (ec) {
            reconnect();
            return;
        }
        cli_.async_start(
            true,                // clean_start
            std::uint16_t(0),    // keep_alive
            "",                  // Client Identifier, empty means generated by the broker
            std::nullopt,        // will
            "UserName1",
            "Password1",
            [this](auto&&... args) {
                handle_start_response(std::forward<decltype(args)>(args)...);
            }
        );
    }

    void handle_start_response(
        am::error_code ec,
        std::optional<client_t::connack_packet> connack_opt
    ) {
        std::cout << "start:" << ec.message() << std::endl;
        if (ec) {
            reconnect();
            return;
        }
        if (connack_opt) {
            std::cout << *connack_opt << std::endl;
        }

        // subscribe
        // MQTT send subscribe and wait suback
        std::vector<am::topic_subopts> sub_entry{
            {sub_topic_, qos},
        };
        cli_.async_subscribe(
            *cli_.acquire_unique_packet_id(), // sync version only works thread safe context
            am::force_move(sub_entry),
            [this](auto&&... args) {
                handle_subscribe_response(
                    std::forward<std::remove_reference_t<decltype(args)>>(args)...
                );
            }
        );
    }

    void handle_subscribe_response(
        am::error_code ec,
        std::optional<client_t::suback_packet> suback_opt
    ) {
        std::cout << "subscribe:" << ec.message() << std::endl;
        if (ec) {
            reconnect();
            return;
        }
        if (suback_opt) {
            std::cout << *suback_opt << std::endl;
        }
        cli_.async_recv(
            [this](auto&&... args) {
                handle_recv(
                    std::forward<std::remove_reference_t<decltype(args)>>(args)...
                );
            }
        );
    }

    void handle_recv(
        am::error_code ec,
        std::optional<am::packet_variant> pv_opt
    ) {
        std::cout << "recv:" << ec.message() << std::endl;
        if (ec) {
            reconnect();
            return;
        }
        BOOST_ASSERT(pv_opt);
        std::cout << *pv_opt << std::endl;
        // next receive
        cli_.async_recv(
            [this](auto&&... args) {
                handle_recv(
                    std::forward<std::remove_reference_t<decltype(args)>>(args)...
                );
            }
        );
    }

    client_t cli_;
    std::string host_;
    std::string port_;
    std::string sub_topic_;
    std::string pub_topic_;
    as::steady_timer tim_{cli_.get_executor()};
};

int main(int argc, char* argv[]) {
    am::setup_log(
        am::severity_level::warning,
        true // log colored
    );
    if (argc != 3) {
        std::cout << "Usage: " << argv[0] << " host port" << std::endl;
        return -1;
    }

    as::io_context ioc1;
    app a1{ioc1.get_executor(), argv[1], argv[2], "topic1", "topic2"};
    auto guard1 = std::make_optional(as::make_work_guard(ioc1));

    as::io_context ioc2;
    app a2{ioc2.get_executor(), argv[1], argv[2], "topic2", "topic1"};
    auto guard2 = std::make_optional(as::make_work_guard(ioc2));

    std::jthread worker1{
        [&](){
            ioc1.run();
        }
    };
    std::jthread worker2{
        [&](){
            ioc2.run();
        }
    };

    a1.connect();
    a2.connect();

    std::this_thread::sleep_for(std::chrono::milliseconds(1000));

    for (int i = 0; i != 100; ++i) {
        auto payload = std::to_string(i);
        a1.publish(payload);
        a2.publish(payload);
    }

    guard1.reset();
    guard2.reset();
}

[redboltz]

I'm really sorry, but I encounter the same problem when using callbacks - after ~30 messages, the client becomes stuck in endpoint_send.hpp:44. I tried wrapping async_ calls in as::post(), but that didn't help. Could you take a look at this code (based on example/cl_cpp17_mqtt_*) and tell me what I'm doing wrong?

No problem, it is a great chance to improve the library :)

It seems that there is some bug at receive_maximum processing part of the library. I will start debugging. Please wait a moment.

[redboltz]

@chipsik-afk , I've fixed the bug. Could you try #422 ?

[chipsik-afk]

@chipsik-afk , I've fixed the bug. Could you try #422 ?

Thank you, everything works now! The coroutine version also works, but after your warnings I won't be using the coroutine. It simplifies the main flow, but having to wrap all calls in a coroutine adds too much complexity.

No problem, it is a great chance to improve the library :)

Thank you for this attitude :)

This may be offtopic for this thread, but I spent ~100 hours testing the stability and reliability of this library, especially focusing on scenarios like different connection failures, lost packets etc. The library is stable and reliable, the only issues are with underlying boost/networking layers. For example, when we have client looping on async_recv, and we want to shut it down from another thread:

• Sometimes (when using TLS, if I remember correctly), it's not enough to call async_disconnect and async_close. It seems to be the issue with boost::asio::ssl::stream. Workaround: after some time, call lowest_layer().close().
• Sometimes (I don't remember exactly the scenario, I think it was dropped packets), it's not enough to call lowest_layer().close(). It also seems to be the issue with boost streams. Workaround: call lowest_layer().cancel() first. Even then, it may take several seconds before the network layer cancels.
• When using TLS (really using it, providing client/server certificates etc.), the client can't reconnect after disconnecting. It also seems to be the issue with boost::asio::ssl::stream, which can be used only for one connection. Workaround: destroy the client and create a new one when reconnecting (just remember to wait for the io_context to complete first).
• When async_handshake fails really fast (for example, connecting to a closed port on localhost), io_context never stops running. It seems to be some kind of race condition in boost::asio::io_context. Workaround: after some time, call io_context::stop().

This problem with Receive Maximum was the only thing that turned out to be the issue with the library itself. Thank you so much for fixing it! ❤️