From 87eb8de00157d134837e019169a227374fc8eaec Mon Sep 17 00:00:00 2001 From: Ian Chua Date: Thu, 24 Sep 2026 20:28:13 +0800 Subject: [PATCH] fix: serialize MQTT ws I/O and dewcouple keepalive --- src/slic3r/Utils/OrcaMqttConnection.cpp | 338 ++++++++++++------ src/slic3r/Utils/OrcaMqttConnection.hpp | 23 +- tests/slic3rutils/orca_mqtt_mock_broker.hpp | 18 +- .../slic3rutils/test_orca_mqtt_connection.cpp | 14 + 4 files changed, 260 insertions(+), 133 deletions(-) diff --git a/src/slic3r/Utils/OrcaMqttConnection.cpp b/src/slic3r/Utils/OrcaMqttConnection.cpp index 396744655c..d33e4182c6 100644 --- a/src/slic3r/Utils/OrcaMqttConnection.cpp +++ b/src/slic3r/Utils/OrcaMqttConnection.cpp @@ -23,6 +23,12 @@ struct OrcaMqttConnection::Connection { boost::asio::io_context io_context; boost::asio::ssl::context ssl_context; boost::asio::ip::tcp::resolver resolver; + boost::asio::steady_timer keepalive_timer; + boost::beast::flat_buffer read_buffer; + std::deque>> outbound_packets; + boost::system::error_code terminal_error; + std::atomic_bool async_session_started{false}; + bool write_in_progress{false}; // Exactly one of these is engaged once ws_handshake() has run: wss for // wss:// endpoints, ws for plaintext ws://. std::optional wss; @@ -31,6 +37,7 @@ struct OrcaMqttConnection::Connection { Connection() : ssl_context(boost::asio::ssl::context::tls_client) , resolver(io_context) + , keepalive_timer(io_context) {} }; @@ -81,19 +88,31 @@ void OrcaMqttConnection::stop() { { std::lock_guard lock(connection_mutex); if (active_connection) { - // Generic so it accepts either the TLS or the plaintext websocket. - auto shutdown_socket = [](auto& websocket) { - auto& socket = boost::beast::get_lowest_layer(websocket).socket(); - boost::system::error_code socket_error; - socket.cancel(socket_error); - socket.shutdown(boost::asio::ip::tcp::socket::shutdown_both, socket_error); - socket.close(socket_error); - }; - if (active_connection->wss) - shutdown_socket(*active_connection->wss); - else if (active_connection->ws) - shutdown_socket(*active_connection->ws); - active_connection->resolver.cancel(); + if (active_connection->async_session_started.load()) { + // The worker owns the live WebSocket. Stop dispatching its + // asynchronous operations; the worker closes the socket after + // leaving the event loop. + active_connection->io_context.stop(); + } else { + // Setup still uses synchronous operations on the worker. Wake a + // pending resolve/connect/CONNACK read without competing with a + // live asynchronous session. + auto shutdown_socket = [](auto& websocket) { + auto& socket = boost::beast::get_lowest_layer(websocket).socket(); + boost::system::error_code socket_error; + if (socket.cancel(socket_error)) + return; + if (socket.shutdown(boost::asio::ip::tcp::socket::shutdown_both, socket_error)) + return; + if (socket.close(socket_error)) + return; + }; + if (active_connection->wss) + shutdown_socket(*active_connection->wss); + else if (active_connection->ws) + shutdown_socket(*active_connection->ws); + active_connection->resolver.cancel(); + } } } if (worker.joinable()) @@ -130,13 +149,10 @@ void OrcaMqttConnection::flush_subscription_change() { } if (!conn || !connacked) return; // no live MQTT session yet — the worker sends the set on CONNACK - - // beast permits a concurrent writer while the worker is blocked in - // websocket.read(); every write is serialised by write_mutex inside ws_write(). - try { - send_pending_subscriptions(*conn); - } catch (const std::exception&) { - } + boost::asio::post(conn->io_context, [this, conn] { + if (!stopping.load() && connected.load()) + send_pending_subscriptions(conn); + }); } bool OrcaMqttConnection::subscribe(const std::string& dev_id) { @@ -157,7 +173,7 @@ bool OrcaMqttConnection::subscribe(const std::string& dev_id) { pending_subscriptions.insert(topic); } state_cv.notify_all(); - flush_subscription_change(); // emit SUBSCRIBE now on the live socket (no reconnect) + flush_subscription_change(); // ask the worker to emit SUBSCRIBE now (no reconnect) return true; } @@ -183,7 +199,7 @@ bool OrcaMqttConnection::unsubscribe(const std::string& dev_id) { } } state_cv.notify_all(); - flush_subscription_change(); // emit UNSUBSCRIBE now on the live socket (no reconnect) + flush_subscription_change(); // ask the worker to emit UNSUBSCRIBE now (no reconnect) return true; } @@ -298,10 +314,6 @@ void OrcaMqttConnection::ws_write(Connection& conn, const std::vector& if (packet.empty()) { return; } - // Writes come from the worker thread AND, for dynamic (un)subscribes, the - // caller thread. Serialise them; the worker's concurrent read is fine (beast - // allows one reader + one writer). - std::lock_guard lock(write_mutex); if (conn.wss) { conn.wss->binary(true); conn.wss->write(boost::asio::buffer(packet)); @@ -311,6 +323,110 @@ void OrcaMqttConnection::ws_write(Connection& conn, const std::vector& } } +void OrcaMqttConnection::close_connection(Connection& conn) { + boost::system::error_code error; + if (conn.wss) { + auto& socket = boost::beast::get_lowest_layer(*conn.wss).socket(); + if (socket.cancel(error)) + return; + if (socket.shutdown(boost::asio::ip::tcp::socket::shutdown_both, error)) + return; + if (socket.close(error)) + return; + } else if (conn.ws) { + auto& socket = boost::beast::get_lowest_layer(*conn.ws).socket(); + if (socket.cancel(error)) + return; + if (socket.shutdown(boost::asio::ip::tcp::socket::shutdown_both, error)) + return; + if (socket.close(error)) + return; + } +} + +void OrcaMqttConnection::enqueue_packet(const std::shared_ptr& conn, + std::vector packet) { + if (!conn || packet.empty()) + return; + conn->outbound_packets.emplace_back(std::make_shared>(std::move(packet))); + start_async_write(conn); +} + +void OrcaMqttConnection::start_async_write(const std::shared_ptr& conn) { + if (!conn || conn->write_in_progress || conn->outbound_packets.empty() || stopping.load()) + return; + + conn->write_in_progress = true; + const auto packet = conn->outbound_packets.front(); + auto on_write = [this, conn](const boost::system::error_code& error, std::size_t) { + conn->write_in_progress = false; + if (error) { + if (!stopping.load()) + conn->terminal_error = error; + conn->io_context.stop(); + return; + } + conn->outbound_packets.pop_front(); + start_async_write(conn); + }; + if (conn->wss) { + conn->wss->binary(true); + conn->wss->async_write(boost::asio::buffer(*packet), std::move(on_write)); + } else if (conn->ws) { + conn->ws->binary(true); + conn->ws->async_write(boost::asio::buffer(*packet), std::move(on_write)); + } else { + conn->write_in_progress = false; + conn->outbound_packets.pop_front(); + } +} + +void OrcaMqttConnection::start_async_read(const std::shared_ptr& conn) { + if (!conn || stopping.load()) + return; + + auto on_read = [this, conn](const boost::system::error_code& error, std::size_t) { + if (error) { + if (!stopping.load()) + conn->terminal_error = error; + conn->io_context.stop(); + return; + } + const std::string packet = boost::beast::buffers_to_string(conn->read_buffer.data()); + conn->read_buffer.consume(conn->read_buffer.size()); + handle_packet(packet); + start_async_read(conn); + }; + if (conn->wss) + conn->wss->async_read(conn->read_buffer, std::move(on_read)); + else if (conn->ws) + conn->ws->async_read(conn->read_buffer, std::move(on_read)); +} + +void OrcaMqttConnection::schedule_keepalive(const std::shared_ptr& conn) { + const int keepalive = current_config.keepalive_seconds; + if (!conn || keepalive <= 0 || stopping.load()) + return; + + conn->keepalive_timer.expires_after(std::chrono::seconds(std::max(1, keepalive / 2))); + conn->keepalive_timer.async_wait([this, conn](const boost::system::error_code& error) { + if (error || stopping.load()) + return; + enqueue_packet(conn, make_ping_packet()); + schedule_keepalive(conn); + }); +} + +void OrcaMqttConnection::post_packet(const std::shared_ptr& conn, + std::vector packet) { + if (!conn || packet.empty()) + return; + boost::asio::post(conn->io_context, [this, conn, packet = std::move(packet)]() mutable { + if (!stopping.load()) + enqueue_packet(conn, std::move(packet)); + }); +} + std::size_t OrcaMqttConnection::ws_read(Connection& conn, boost::beast::flat_buffer& buffer, boost::system::error_code& ec) { if (conn.wss) @@ -321,14 +437,6 @@ std::size_t OrcaMqttConnection::ws_read(Connection& conn, boost::beast::flat_buf return 0; } -void OrcaMqttConnection::ws_close(Connection& conn) { - boost::system::error_code close_error; - if (conn.wss) - conn.wss->close(boost::beast::websocket::close_code::normal, close_error); - else if (conn.ws) - conn.ws->close(boost::beast::websocket::close_code::normal, close_error); -} - void OrcaMqttConnection::ws_handshake(Connection& conn, const Config& config, const Endpoint& endpoint) { const auto results = conn.resolver.resolve(endpoint.host, endpoint.port); @@ -415,12 +523,7 @@ bool OrcaMqttConnection::send_request(const std::string& dev_id, const std::stri return true; } } - try { - // ws_write() serialises the write via write_mutex; do not lock it here. - ws_write(*conn, make_publish_packet(request_topic(dev_id), payload)); - } catch (const std::exception&) { - return false; - } + post_packet(conn, make_publish_packet(request_topic(dev_id), payload)); return true; } @@ -429,94 +532,93 @@ void OrcaMqttConnection::connect_and_read() { { std::lock_guard lock(connection_mutex); active_connection = connection; - if (stopping.load()) + if (stopping.load()) { + active_connection.reset(); return; + } } - Endpoint endpoint; - if (!parse_endpoint(current_config.url, endpoint)) { - throw std::runtime_error("invalid Orca Cloud WebSocket endpoint"); - } + auto clear_connection = [this, connection] { + close_connection(*connection); + std::lock_guard lock(connection_mutex); + if (active_connection == connection) + active_connection.reset(); + }; + + try { + Endpoint endpoint; + if (!parse_endpoint(current_config.url, endpoint)) { + throw std::runtime_error("invalid Orca Cloud WebSocket endpoint"); + } - ws_handshake(*connection, current_config, endpoint); + ws_handshake(*connection, current_config, endpoint); - expires_never(*connection); - // Auth precedence: a bearer_provider authenticates the WebSocket upgrade, so the - // CONNECT username/password fields are omitted entirely (the cloud form). - const bool use_bearer = static_cast(current_config.bearer_provider); - ws_write(*connection, make_connect_packet(current_config.client_id, - use_bearer ? std::string() : current_config.username, - use_bearer ? std::string() : current_config.password, - current_config.keepalive_seconds)); + expires_never(*connection); + // Auth precedence: a bearer_provider authenticates the WebSocket upgrade, so the + // CONNECT username/password fields are omitted entirely (the cloud form). + const bool use_bearer = static_cast(current_config.bearer_provider); + ws_write(*connection, make_connect_packet(current_config.client_id, + use_bearer ? std::string() : current_config.username, + use_bearer ? std::string() : current_config.password, + current_config.keepalive_seconds)); - boost::beast::flat_buffer buffer; - expires_after(*connection, std::chrono::seconds(10)); - boost::system::error_code connack_error; - ws_read(*connection, buffer, connack_error); - if (connack_error) - throw boost::system::system_error(connack_error, "read Orca MQTT CONNACK"); - const std::string connack = boost::beast::buffers_to_string(buffer.data()); - // rc: 0 accepted, 1..5 refusal, -1 malformed/not a CONNACK. - const int rc = (connack.size() == 4 && static_cast(connack[0]) == 0x20) - ? static_cast(static_cast(connack[3])) - : -1; - m_last_connack_rc.store(rc); - if (rc != 0) { - if (rc == 4 || rc == 5) { - // Bad credentials / not authorized — retrying cannot help. Make run()'s - // loop exit and unblock any waiting start(). - stopping.store(true); - { - std::lock_guard lock(mutex); - initial_completed = true; - initial_result = false; + boost::beast::flat_buffer buffer; + expires_after(*connection, std::chrono::seconds(10)); + boost::system::error_code connack_error; + ws_read(*connection, buffer, connack_error); + if (connack_error) + throw boost::system::system_error(connack_error, "read Orca MQTT CONNACK"); + const std::string connack = boost::beast::buffers_to_string(buffer.data()); + // rc: 0 accepted, 1..5 refusal, -1 malformed/not a CONNACK. + const int rc = (connack.size() == 4 && static_cast(connack[0]) == 0x20) + ? static_cast(static_cast(connack[3])) + : -1; + m_last_connack_rc.store(rc); + if (rc != 0) { + if (rc == 4 || rc == 5) { + // Bad credentials / not authorized — retrying cannot help. Make run()'s + // loop exit and unblock any waiting start(). + stopping.store(true); + { + std::lock_guard lock(mutex); + initial_completed = true; + initial_result = false; + } + initial_cv.notify_all(); } - initial_cv.notify_all(); + throw std::runtime_error("Orca MQTT CONNECT refused rc=" + std::to_string(rc)); } - throw std::runtime_error("Orca MQTT CONNECT refused rc=" + std::to_string(rc)); - } - // The subscription acknowledgement belongs to this MQTT session. Clear - // the previous session's state before notifying the owner, because the - // reconnect callback immediately queues the printer's initial requests. - { - std::lock_guard lock(mutex); - acknowledged_subscriptions.clear(); - pending_subscribe_packets.clear(); - } - notify_state(true); - reconnect_delay_seconds.store(1); // a fresh CONNACK resets the backoff - send_current_subscriptions(*connection); - std::chrono::steady_clock::time_point next_ping = std::chrono::steady_clock::now() + std::chrono::seconds(30); - - while (!stopping.load()) { - send_pending_subscriptions(*connection); - // Keepalive is driven every iteration, not only from the read-timeout branch: - // a printer pushing faster than the 1s read deadline would otherwise keep the - // read hot and the broker would drop us at 1.5 x keepalive. - if (std::chrono::steady_clock::now() >= next_ping) { - ws_write(*connection, make_ping_packet()); - next_ping = std::chrono::steady_clock::now() + std::chrono::seconds(30); + // The subscription acknowledgement belongs to this MQTT session. Clear + // the previous session's state before notifying the owner, because the + // reconnect callback immediately queues the printer's initial requests. + { + std::lock_guard lock(mutex); + acknowledged_subscriptions.clear(); + pending_subscribe_packets.clear(); } - buffer.consume(buffer.size()); - expires_after(*connection, std::chrono::seconds(1)); - boost::system::error_code error; - ws_read(*connection, buffer, error); - if (error == boost::beast::error::timeout) - continue; - if (error) { - throw boost::system::system_error(error, "read Orca MQTT message"); - } - handle_packet(boost::beast::buffers_to_string(buffer.data())); + connection->async_session_started.store(true); + notify_state(true); + reconnect_delay_seconds.store(1); // a fresh CONNACK resets the backoff + send_current_subscriptions(connection); + start_async_read(connection); + schedule_keepalive(connection); + const std::size_t handlers_run = connection->io_context.run(); + if (handlers_run == 0 && !stopping.load()) + throw std::runtime_error("Orca MQTT event loop stopped unexpectedly"); + if (connection->terminal_error && !stopping.load()) + throw boost::system::system_error(connection->terminal_error, "read Orca MQTT message"); + clear_connection(); + if (!stopping.load()) + notify_state(false); + } catch (...) { + clear_connection(); + throw; } - - ws_close(*connection); - if (!stopping.load()) - notify_state(false); } -void OrcaMqttConnection::send_current_subscriptions(Connection& conn) { +void OrcaMqttConnection::send_current_subscriptions(const std::shared_ptr& conn) { std::vector topics; { std::lock_guard lock(mutex); @@ -532,11 +634,11 @@ void OrcaMqttConnection::send_current_subscriptions(Connection& conn) { std::lock_guard lock(mutex); pending_subscribe_packets[packet_id] = topic; } - ws_write(conn, make_subscribe_packet(packet_id, topic, 1)); + enqueue_packet(conn, make_subscribe_packet(packet_id, topic, 1)); } } -void OrcaMqttConnection::send_pending_subscriptions(Connection& conn) { +void OrcaMqttConnection::send_pending_subscriptions(const std::shared_ptr& conn) { std::vector subscribe_topics; std::vector unsubscribe_topics; { @@ -552,11 +654,11 @@ void OrcaMqttConnection::send_pending_subscriptions(Connection& conn) { std::lock_guard lock(mutex); pending_subscribe_packets[packet_id] = topic; } - ws_write(conn, make_subscribe_packet(packet_id, topic, 1)); + enqueue_packet(conn, make_subscribe_packet(packet_id, topic, 1)); } for (const std::string& topic : unsubscribe_topics) { const uint16_t packet_id = next_packet_id++; - ws_write(conn, make_unsubscribe_packet(packet_id, topic)); + enqueue_packet(conn, make_unsubscribe_packet(packet_id, topic)); } } diff --git a/src/slic3r/Utils/OrcaMqttConnection.hpp b/src/slic3r/Utils/OrcaMqttConnection.hpp index dd62a72964..1565542f69 100644 --- a/src/slic3r/Utils/OrcaMqttConnection.hpp +++ b/src/slic3r/Utils/OrcaMqttConnection.hpp @@ -96,19 +96,23 @@ private: static std::vector make_ping_packet(); // Transport dispatch: each forwards to conn.wss (TLS) or conn.ws (plaintext). - void ws_write(Connection& conn, const std::vector& packet); // locks write_mutex + // These synchronous operations are called only by the MQTT worker during + // connection setup. Once the MQTT session is established, all socket I/O is + // asynchronous and owned by that worker's io_context. + void ws_write(Connection& conn, const std::vector& packet); std::size_t ws_read(Connection& conn, boost::beast::flat_buffer& buffer, boost::system::error_code& ec); void ws_handshake(Connection& conn, const Config& config, const Endpoint& endpoint); - void ws_close(Connection& conn); - // Emit a queued SUBSCRIBE/UNSUBSCRIBE on the live socket right now (from the - // caller thread), so a selection change is applied without waiting for the - // blocking read loop to next return. No-op if no CONNACKed socket exists yet - // (the worker sends the set on connect). The WebSocket is never dropped for a - // subscription change. + void close_connection(Connection& conn); + void enqueue_packet(const std::shared_ptr& conn, std::vector packet); + void start_async_write(const std::shared_ptr& conn); + void start_async_read(const std::shared_ptr& conn); + void schedule_keepalive(const std::shared_ptr& conn); + void post_packet(const std::shared_ptr& conn, std::vector packet); + // Ask the MQTT worker to emit subscription changes on its own io_context. void flush_subscription_change(); void connect_and_read(); - void send_current_subscriptions(Connection& conn); - void send_pending_subscriptions(Connection& conn); + void send_current_subscriptions(const std::shared_ptr& conn); + void send_pending_subscriptions(const std::shared_ptr& conn); void handle_packet(const std::string& packet); void notify_state(bool is_now_connected); void run(); @@ -122,7 +126,6 @@ private: std::thread worker; std::mutex mutex; std::mutex connection_mutex; - std::mutex write_mutex; // serialises every websocket write (worker + caller threads) std::shared_ptr active_connection; std::condition_variable initial_cv; std::condition_variable state_cv; diff --git a/tests/slic3rutils/orca_mqtt_mock_broker.hpp b/tests/slic3rutils/orca_mqtt_mock_broker.hpp index 8545a4344f..555a4a9a05 100644 --- a/tests/slic3rutils/orca_mqtt_mock_broker.hpp +++ b/tests/slic3rutils/orca_mqtt_mock_broker.hpp @@ -127,6 +127,9 @@ public: // MQTT CONNECTs seen; increments again after a reconnect. int connect_count() const { return m_connect_count.load(); } + // MQTT PINGREQs seen while the client has no other traffic. + int ping_count() const { return m_ping_count.load(); } + private: void run() { @@ -231,6 +234,7 @@ private: return true; } case 0xc0: // PINGREQ + ++m_ping_count; write_packet(stream, {0xd0, 0x00}); return true; case 0xe0: // DISCONNECT @@ -277,8 +281,8 @@ private: } // Every write - the worker's own replies and push_report() from the test - // thread - is serialised by m_mutex. beast permits a writer while the worker - // is blocked in read(), which is the same arrangement OrcaMqttConnection uses. + // thread - is serialised by m_mutex. The production client keeps all of its + // WebSocket operations on its MQTT worker instead. void write_packet(ws::stream& stream, const std::vector& packet) { std::lock_guard lock(m_mutex); @@ -294,11 +298,14 @@ private: m_stream_ready = false; boost::system::error_code ec; auto& socket = beast::get_lowest_layer(*m_stream).socket(); - socket.cancel(ec); + if (socket.cancel(ec)) + return; // shutdown() before close() is what actually wakes a blocking read on the // worker thread; close() alone does not on POSIX. - socket.shutdown(tcp::socket::shutdown_both, ec); - socket.close(ec); + if (socket.shutdown(tcp::socket::shutdown_both, ec)) + return; + if (socket.close(ec)) + return; } const bool m_refuse_auth; @@ -308,6 +315,7 @@ private: std::thread m_thread; std::atomic_bool m_stopping{false}; std::atomic m_connect_count{0}; + std::atomic m_ping_count{0}; mutable std::mutex m_mutex; std::optional> m_stream; // guarded by m_mutex bool m_stream_ready = false; // guarded by m_mutex diff --git a/tests/slic3rutils/test_orca_mqtt_connection.cpp b/tests/slic3rutils/test_orca_mqtt_connection.cpp index 60c4e9303b..1bb4983901 100644 --- a/tests/slic3rutils/test_orca_mqtt_connection.cpp +++ b/tests/slic3rutils/test_orca_mqtt_connection.cpp @@ -182,6 +182,20 @@ static void run_round_trip(bool use_tls_flag_only) { TEST_CASE("OrcaMqtt round-trip — LAN-style config", "[OrcaMqtt][.integration]") { run_round_trip(false); } TEST_CASE("OrcaMqtt round-trip — cloud-style config", "[OrcaMqtt][.integration]") { run_round_trip(true); } +TEST_CASE("OrcaMqtt keepalive runs while the connection is idle", "[OrcaMqtt][.integration]") { + orca_mqtt_test::MockBroker broker; + OrcaMqttConnection conn; + OrcaMqttConnection::Config cfg; + cfg.url = broker.ws_url(); + cfg.keepalive_seconds = 2; + + REQUIRE(conn.start(cfg, [](const std::string&, const std::string&) {}, [](bool, bool) {})); + for (int i = 0; i < 200 && broker.ping_count() == 0; ++i) + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + CHECK(broker.ping_count() > 0); + conn.stop(); +} + TEST_CASE("OrcaMqtt reconnects and re-subscribes after a socket drop", "[OrcaMqtt][.integration]") { orca_mqtt_test::MockBroker broker; OrcaMqttConnection conn;