Callbacks and avoiding Deep Linking

[Radagan]

I was updating my mqtt client code to 10.0.0 and realized that it's probably not safe to issue async_recv in the lambda that processes the async_recv for a long lived client, as shown in this basic use example:

https://github.com/redboltz/async_mqtt/blob/10.0.0/example/ep_cb_mqtt_client.cpp

My mqtt client maintains a persistent connection indefinitely. Won't this lead to deep linking of callbacks and an eventual fatal error?

Do I need to add code to loop while connected and call async_recv only after a packet has been received? Or am I over thinking this and the library handles these issues.

[redboltz]

Let me clarify your question: Are you asking whether invoking async_read within the async_recv callback could cause a stack overflow?

[Radagan]

Yes, exactly. It looks like it would, so I changed my code to use a loop with lock and condition_variable to wait on the receive before issuing a new endpoint.async_recv. Basically, this:

    void mqtt_client::process_receive_packets() {
        // Start receiving packets...
        std::atomic<bool> run{true};
        bool send_async_recv{true};
        std::mutex run_mutex{};
        std::condition_variable run_condition{};
        while(run) {
            std::unique_lock lock(run_mutex);
            if(send_async_recv) {
                send_async_recv = false;
                endpoint.async_recv(
                    [this, &](async_mqtt::error_code& error, std::optional<async_mqtt::packet_variant> packet) {
                        if(error) {
                            // Handle error and exit...
                            run = false;
                            return;
                        }

                        send_async_recv = true;
                        packet->visit(
                            // Handle messages...
                        );

                        std::unique_lock run_lock(run_mutex);
                        send_async_recv = true;
                        run_lock.unlock();
                        run_condition.notify_all();
                    }
                );
            }
            run_condition.wait_for(lock, 1s, [&]{ return send_async_recv; });
        }
    }

Is this a good approach?

[redboltz]

Short answer:

Calling the next async_mqtt::endpoint::async_recv() in the completion handler (e.g. callback) of async_mqtt::endpoint::async_recv() is safe.

Detailed answer:

dispatch() and post() are different.

dispatch() executes the handler immediately if the caller is running in the same executor. Consequently, a recursive dispatch() call can cause a stack overflow. See https://godbolt.org/z/1Thd73EvP

In contrast, post() simply adds the handler to the tail of the executor's internal queue, and the handler is later invoked via ioc.run(), ensuring that the call stack is not increased. See https://godbolt.org/z/8nEcfe1zY

---

The key question is whether endpoint::async_recv() behaves like dispatch() or like post().

See

async_mqtt/include/async_mqtt/asio_bind/impl/endpoint_recv.hpp

Lines 15 to 44 in 6f107e9

	template <role Role, std::size_t PacketIdBytes, typename NextLayer>
	template <typename CompletionToken>
	auto
	basic_endpoint<Role, PacketIdBytes, NextLayer>::async_recv(
	CompletionToken&& token
	) {
	ASYNC_MQTT_LOG("mqtt_api", info)
	<< ASYNC_MQTT_ADD_VALUE(address, this)
	<< "recv";
	BOOST_ASSERT(impl_);
	return
	as::async_initiate<
	CompletionToken,
	void(error_code, std::optional<packet_variant_type>)
	>(
	[](
	auto handler,
	std::shared_ptr<impl_type> impl
	) {
	impl_type::async_recv(
	force_move(impl),
	std::nullopt,
	std::set<control_packet_type>{},
	force_move(handler)
	);
	},
	token,
	impl_
	);
	}

async_mqtt/include/async_mqtt/asio_bind/impl/endpoint_recv.ipp

Lines 320 to 347 in 6f107e9

	template <role Role, std::size_t PacketIdBytes, typename NextLayer>
	ASYNC_MQTT_HEADER_ONLY_INLINE
	void
	basic_endpoint_impl<Role, PacketIdBytes, NextLayer>::
	async_recv(
	this_type_sp impl,
	std::optional<filter> fil,
	std::set<control_packet_type> types,
	as::any_completion_handler<
	void(error_code, std::optional<packet_variant_type>)
	> handler
	) {
	auto exe = impl->get_executor();
	as::async_compose<
	as::any_completion_handler<
	void(error_code, std::optional<packet_variant_type>)
	>,
	void(error_code, std::optional<packet_variant_type>)
	>(
	recv_op{
	force_move(impl),
	fil,
	force_move(types)
	},
	handler,
	exe
	);
	}

First, consider the dispatch phase:

async_mqtt/include/async_mqtt/asio_bind/impl/endpoint_recv.ipp

Lines 149 to 155 in 6f107e9

	case dispatch: {
	state = check_buf;
	as::dispatch(
	a_ep.get_executor(),
	force_move(self)
	);
	} break;

Then, the function calls check_buf():

async_mqtt/include/async_mqtt/asio_bind/impl/endpoint_recv.ipp

Lines 156 to 172 in 6f107e9

	case check_buf: {
	if (a_ep.read_buf_.size() == 0) {
	// read required
	state = read;
	as::dispatch(
	a_ep.get_executor(),
	force_move(self)
	);
	}
	else {
	state = read_istream;
	as::post(
	a_ep.get_executor(),
	force_move(self)
	);
	}
	} break;

If read_buf_ is empty (for example, during the initial read), the function is dispatched (jumped) directly to the read state. Otherwise, it transitions to the read_istream state using post(), which resets the call stack.

Next, consider the read phase:

async_mqtt/include/async_mqtt/asio_bind/impl/endpoint_recv.ipp

Lines 226 to 232 in 6f107e9

	case read: {
	state = finish_read;
	a_ep.stream_.async_read_some(
	a_ep.read_buf_.prepare(a_ep.read_buffer_size_),
	force_move(self)
	);
	} break;

At this point, async_read_some() is called. It ultimately calls asio::ip::tcp::socket::async_read_some(), and its completion handler is executed via ioc.run() (similar to using post()), which resets the call stack.

Thus, in both cases, the call stack is reset.

Hence, calling the next async_mqtt::endpoint::async_recv() in the completion handler (e.g. callback) of async_mqtt::endpoint::async_recv() is safe.

[Radagan]

Thank you! Not only is your library an excellent solution to my company's needs, but it has deepened my knowledge of modern C++ by working with it.

[redboltz]

You are welcome!
If your project is publicly available, I would greatly appreciate it if you could leave a comment on #403.

[Radagan]

Unfortunately, it is part of a commercial product. But I will try to contribute a use case there. I just noticed the discussion before I saw your reply.