diff --git a/include/sendspin/client.h b/include/sendspin/client.h index a199f04..bc1e533 100644 --- a/include/sendspin/client.h +++ b/include/sendspin/client.h @@ -24,6 +24,7 @@ #include #include #include +#include #include #include #include @@ -205,6 +206,10 @@ class SendspinClient { bool start_server(); /// @brief Initiates a client connection to a Sendspin server at the given URL + /// + /// Must be called from the main loop thread: it tears down and replaces connection state + /// (time filter, dispatch, client state) directly rather than deferring to loop(), so calling + /// it concurrently with loop() would race those mutations. /// @param url WebSocket server URL (e.g., "ws://server.local:8927/sendspin") void connect_to(const std::string& url); @@ -470,6 +475,9 @@ class SendspinClient { /// Internal-RAM scratch arena for parsing incoming JSON; null unless config_.json_arena_size > /// 0 std::unique_ptr json_arena_; + /// Serializes process_json_message() (and its use of json_arena_) across the network threads + /// of concurrently live connections (current + pending during a handoff). + std::mutex json_processing_mutex_; SendspinClientListener* listener_{nullptr}; #ifdef SENDSPIN_ENABLE_METADATA std::unique_ptr metadata_; diff --git a/src/client.cpp b/src/client.cpp index cf28cfd..64d9faf 100644 --- a/src/client.cpp +++ b/src/client.cpp @@ -58,6 +58,12 @@ struct TimeResponseEvent { int64_t offset; int64_t max_error; int64_t timestamp; + /// Instance id of the connection the response arrived on, compared against the current + /// connection at drain time so a measurement from a displaced or pending server cannot + /// contaminate the current connection's time filter. An id (not a pointer) so a since-freed + /// connection cannot ABA-match a later connection reusing its address; 0 never matches a live + /// connection (ids start at 1). + uint64_t source_id; }; /// @brief Deferred event state for time responses and group updates on the main thread @@ -186,7 +192,10 @@ void SendspinClient::loop() { TimeResponseEvent time_event{}; while (this->event_state_->time_queue.receive(time_event, 0)) { auto* current = this->connection_manager_->current(); - if (current != nullptr) { + // Apply only measurements from the connection that is still current; a response queued + // by a since-displaced (or pending) server carries that server's clock and would + // contaminate this connection's Kalman filter. + if (current != nullptr && current->get_instance_id() == time_event.source_id) { this->time_burst_->on_time_response(current, time_event.offset, time_event.max_error, time_event.timestamp); } @@ -331,17 +340,21 @@ bool SendspinClient::is_connected() const { } bool SendspinClient::is_time_synced() const { - auto* conn = this->connection_manager_->current(); + // current_shared(): called from role threads (sync task, drain threads), so the shared_ptr + // must keep the connection alive while it is dereferenced. + auto conn = this->connection_manager_->current_shared(); return conn != nullptr && conn->is_time_synced(); } int64_t SendspinClient::get_client_time(int64_t server_time) const { - auto* conn = this->connection_manager_->current(); + // current_shared(): called from role threads; see is_time_synced(). + auto conn = this->connection_manager_->current_shared(); return conn != nullptr ? conn->get_client_time(server_time) : 0; } std::optional SendspinClient::get_server_information() const { - auto* conn = this->connection_manager_->current(); + // current_shared(): public accessor, callable from any thread. + auto conn = this->connection_manager_->current_shared(); if (conn == nullptr || !conn->is_handshake_complete()) { return std::nullopt; } @@ -499,9 +512,14 @@ std::string SendspinClient::build_hello_message() { void SendspinClient::process_json_message(SendspinConnection* conn, const char* data, size_t len, int64_t timestamp) { - // Reuse the internal-RAM scratch arena if configured. Safe to reset here: this runs only on - // the network task, and the JsonDocument from the previous call was already destroyed when - // that call returned. + // Two connections can deliver JSON concurrently on their own network threads (current + + // pending during a handoff, or an outbound connect_to() transport alongside the inbound + // server). Serialize the shared arena and the parse itself; JSON control messages are + // infrequent, so contention is negligible. + std::lock_guard lock(this->json_processing_mutex_); + + // Reuse the internal-RAM scratch arena if configured. Safe to reset here: the JsonDocument + // from the previous call was already destroyed when that call returned. if (this->json_arena_) { this->json_arena_->reset(); } @@ -656,7 +674,8 @@ void SendspinClient::process_json_message(SendspinConnection* conn, const char* int64_t offset{0}; int64_t max_error{0}; if (process_server_time_message(root, timestamp, &offset, &max_error)) { - if (!this->event_state_->time_queue.send({offset, max_error, timestamp}, 0)) { + if (!this->event_state_->time_queue.send( + {offset, max_error, timestamp, conn->get_instance_id()}, 0)) { SS_LOGW(TAG, "Time response queue full; dropping measurement"); } } diff --git a/src/connection.h b/src/connection.h index 904243b..0b63fd7 100644 --- a/src/connection.h +++ b/src/connection.h @@ -115,6 +115,15 @@ class SendspinConnection : public std::enable_shared_from_thisinstance_id; + } + /// @brief Sends a text message to the server with a completion callback /// @param message The message string to send. /// @param cb Callback invoked with the send result. On asynchronous transports it is not @@ -342,6 +351,13 @@ class SendspinConnection : public std::enable_shared_from_this counter{1}; + return counter.fetch_add(1, std::memory_order_relaxed); + } + // Struct fields /// Message buffering (for websocket frame assembly). @@ -361,6 +377,9 @@ class SendspinConnection : public std::enable_shared_from_this serialize_ema_us_{0}; + /// Process-unique connection identity (see get_instance_id()). Assigned once at construction. + const uint64_t instance_id{next_instance_id()}; + // size_t fields size_t websocket_write_offset_{0}; @@ -384,8 +403,9 @@ class SendspinConnection : public std::enable_shared_from_this message_dispatch_enabled_{true}; - /// Time message state. - bool pending_time_message_{false}; + /// Time message state. Written by the main-loop time burst and cleared by the transport + /// thread's disconnect handler, hence atomic. + std::atomic pending_time_message_{false}; /// Atomic for the same reason as client_hello_sent_: written on network threads, read from the /// main loop via is_handshake_complete(). diff --git a/src/connection_manager.cpp b/src/connection_manager.cpp index 94c1d4f..26fca30 100644 --- a/src/connection_manager.cpp +++ b/src/connection_manager.cpp @@ -31,6 +31,8 @@ static constexpr uint32_t FNV_OFFSET_BASIS = 2166136261UL; static constexpr uint32_t FNV_PRIME = 16777619UL; static constexpr int64_t HELLO_INITIAL_DELAY_MS = 100LL; static constexpr int64_t HELLO_INITIAL_DELAY_US = HELLO_INITIAL_DELAY_MS * US_PER_MS; +static constexpr int64_t WS_SERVER_START_RETRY_MS = 5000LL; +static constexpr int64_t WS_SERVER_START_RETRY_US = WS_SERVER_START_RETRY_MS * US_PER_MS; // ============================================================================ // Constructor / Destructor @@ -72,9 +74,27 @@ void ConnectionManager::connect_to(const std::string& url) { std::lock_guard lock(this->conn_ptr_mutex_); if (this->current_connection_ != nullptr && this->current_connection_->is_connected()) { SS_LOGD(TAG, "Existing connection active, new connection will go through handoff"); + // Release any existing pending connection (e.g. an inbound handoff candidate) before + // overwriting the slot, mirroring on_new_connection and complete_handoff; otherwise it + // is dropped with no goodbye and, on ESP, leaves its httpd session pinned. + if (this->pending_connection_ != nullptr) { + this->remove_hello_retry(this->pending_connection_.get()); + this->disconnect_and_release(std::move(this->pending_connection_), + SendspinGoodbyeReason::ANOTHER_SERVER); + } this->pending_connection_ = std::move(client_conn); this->pending_connection_->start(); } else { + // A present-but-disconnected current connection (its close event not yet processed) is + // being replaced. Tear its state down as on_connection_lost would, instead of + // overwriting the slot and orphaning dispatch/time-filter/client state and its retry. + if (this->current_connection_ != nullptr) { + this->current_connection_->disable_message_dispatch(); + this->client_->time_burst_->reset(); + this->client_->cleanup_connection_state(); + this->remove_hello_retry(this->current_connection_.get()); + this->current_connection_.reset(); + } this->current_connection_ = std::move(client_conn); this->current_connection_->start(); } @@ -132,12 +152,16 @@ void ConnectionManager::init_server(SendspinClient* client) { } void ConnectionManager::loop() { - // Start WS server when network becomes ready + // Start WS server when network becomes ready. A persistent failure (e.g. the server port is + // already in use) is retried with backoff instead of on every tick, which would spam the log. if (this->ws_server_ != nullptr && !this->ws_server_->is_started()) { - if (this->client_->network_provider_ && + const int64_t now_us = platform_time_us(); + if (now_us >= this->ws_server_start_retry_time_us_ && this->client_->network_provider_ && this->client_->network_provider_->is_network_ready()) { - this->ws_server_->start(this->client_, this->client_->config_.httpd_psram_stack, - this->client_->config_.httpd_priority); + if (!this->ws_server_->start(this->client_, this->client_->config_.httpd_psram_stack, + this->client_->config_.httpd_priority)) { + this->ws_server_start_retry_time_us_ = now_us + WS_SERVER_START_RETRY_US; + } } } diff --git a/src/connection_manager.h b/src/connection_manager.h index 8fe0780..8c27a6d 100644 --- a/src/connection_manager.h +++ b/src/connection_manager.h @@ -120,6 +120,16 @@ class ConnectionManager { return this->current_connection_.get(); } + /// @brief Returns a shared_ptr to the current connection. Thread-safe. + /// Role threads (sync task, artwork/visualizer drains) must use this instead of current(): + /// the shared_ptr keeps the connection alive for the duration of the caller's use even if the + /// main loop concurrently drops or replaces the current connection. + /// @return Shared pointer to the current connection, or nullptr if none. + std::shared_ptr current_shared() const { + std::lock_guard lock(this->conn_ptr_mutex_); + return this->current_connection_; + } + // ======================================== // Event queuing (thread-safe) // ======================================== @@ -208,6 +218,10 @@ class ConnectionManager { // 32-bit fields uint32_t last_played_server_hash_{0}; + // 64-bit fields + /// Earliest time (us) to attempt another WS server start after a failure. Main-loop only. + int64_t ws_server_start_retry_time_us_{0}; + // 8-bit fields bool has_last_played_server_{false}; }; diff --git a/src/esp/client_connection.cpp b/src/esp/client_connection.cpp index 8a31fd1..bdbc3c1 100644 --- a/src/esp/client_connection.cpp +++ b/src/esp/client_connection.cpp @@ -258,6 +258,11 @@ void SendspinClientConnection::handle_data(const esp_websocket_event_data_t* dat uint8_t* dest = this->prepare_receive_buffer(prepare_len); if (dest == nullptr) { SS_LOGE(TAG, "Allocation failed, dropping connection"); + // Stop processing frames that keep arriving on the still-open transport; the + // manager reacts to the disconnect callback by dropping the connection, whose + // destructor stops the transport (esp_websocket_client_stop cannot be called + // from the websocket task's own event handler). + this->disable_message_dispatch(); this->handle_disconnected(); return; } diff --git a/src/esp/client_connection.h b/src/esp/client_connection.h index ef70480..3db66c1 100644 --- a/src/esp/client_connection.h +++ b/src/esp/client_connection.h @@ -20,6 +20,7 @@ #include "connection.h" #include +#include #include #include @@ -162,8 +163,9 @@ class SendspinClientConnection : public SendspinConnection { /// @brief Whether to automatically reconnect after connection loss bool auto_reconnect_{true}; - /// @brief Whether the websocket is currently connected - bool connected_{false}; + /// @brief Whether the websocket is currently connected. Written by the transport task, + /// read cross-thread via is_connected(), hence atomic. + std::atomic connected_{false}; }; } // namespace sendspin diff --git a/src/esp/server_connection.cpp b/src/esp/server_connection.cpp index 256b939..e484a7b 100644 --- a/src/esp/server_connection.cpp +++ b/src/esp/server_connection.cpp @@ -110,7 +110,7 @@ void SendspinServerConnection::disconnect(SendspinGoodbyeReason reason, } bool SendspinServerConnection::is_connected() const { - return this->sockfd_ >= 0; + return this->sockfd_ >= 0 && !this->closed_.load(std::memory_order_acquire); } SsErr SendspinServerConnection::send_text_message(const std::string& message, @@ -211,6 +211,10 @@ SS_HOT esp_err_t SendspinServerConnection::handle_data(httpd_req_t* req, int64_t // Allocate/grow directly into the websocket payload buffer (zero-copy) uint8_t* dest = this->prepare_receive_buffer(ws_pkt.len); if (dest == nullptr) { + // Returning an error makes httpd close the session, which tears the slot down via + // the close notification on the main loop + SS_LOGE(TAG, "Allocation failed, dropping connection"); + this->disable_message_dispatch(); return ESP_ERR_NO_MEM; } diff --git a/src/esp/server_connection.h b/src/esp/server_connection.h index 6c86e30..2b70b16 100644 --- a/src/esp/server_connection.h +++ b/src/esp/server_connection.h @@ -20,6 +20,7 @@ #include "connection.h" #include +#include #include namespace sendspin { @@ -85,6 +86,16 @@ class SendspinServerConnection : public SendspinConnection { /// @return true if connected, false otherwise. bool is_connected() const override; + /// @brief Marks the connection closed after the httpd session ends + /// + /// Called from the ws server's close notification (httpd thread). Without this, + /// is_connected() stayed true until the manager dropped the connection on the main loop, + /// and a queued async send in that window could resolve the stale sockfd against a + /// recycled httpd session and write the frame to the wrong peer. + void mark_closed() { + this->closed_.store(true, std::memory_order_release); + } + /// @brief Sends a text message to the server with a completion callback /// @param message The message string to send. /// @param on_complete Callback invoked after send completes. @@ -150,6 +161,11 @@ class SendspinServerConnection : public SendspinConnection { /// @brief The socket file descriptor for this connection int sockfd_{-1}; + + // 8-bit fields + + /// @brief Set once the httpd session has closed (see mark_closed()) + std::atomic closed_{false}; }; } // namespace sendspin diff --git a/src/esp/ws_server.cpp b/src/esp/ws_server.cpp index 13ae936..6a8c7ec 100644 --- a/src/esp/ws_server.cpp +++ b/src/esp/ws_server.cpp @@ -151,6 +151,14 @@ esp_err_t SendspinWsServer::open_callback(httpd_handle_t handle, int sockfd) { void SendspinWsServer::close_callback(httpd_handle_t handle, int sockfd) { SS_LOGD(TAG, "Client closed connection on socket %d", sockfd); + // Flip the connection to disconnected immediately: queued async sends check is_connected() + // and must not resolve this sockfd once httpd may recycle it for a new session. + auto* slot = + static_cast*>(httpd_sess_get_ctx(handle, sockfd)); + if (slot != nullptr && *slot != nullptr) { + (*slot)->mark_closed(); + } + SendspinWsServer* server = (SendspinWsServer*)httpd_get_global_user_ctx(handle); // Notify ConnectionManager so it can drop its observer shared_ptr. The session slot (set in diff --git a/src/host/client_connection.cpp b/src/host/client_connection.cpp index 89b624d..ee5503c 100644 --- a/src/host/client_connection.cpp +++ b/src/host/client_connection.cpp @@ -173,10 +173,17 @@ void SendspinClientConnection::setup_callbacks() { uint8_t* dest = this->prepare_receive_buffer(data.size()); if (dest == nullptr) { SS_LOGE(TAG, "Allocation failed, dropping connection"); + // Stop processing further frames and initiate a real transport close + // (stop() would join IX's thread and deadlock here; close() is async + // and safe). The later Close event repeats the disconnect callback, + // which the manager tolerates (drop of an unmanaged connection is a + // no-op). + this->disable_message_dispatch(); this->connected_ = false; if (this->on_disconnected_cb) { this->on_disconnected_cb(this); } + this->ws_->close(); return; } std::copy(data.begin(), data.end(), dest); diff --git a/src/host/client_connection.h b/src/host/client_connection.h index acce691..7ec672c 100644 --- a/src/host/client_connection.h +++ b/src/host/client_connection.h @@ -20,6 +20,7 @@ #include "connection.h" #include +#include #include #include #include @@ -128,8 +129,9 @@ class SendspinClientConnection : public SendspinConnection { /// @brief Whether to automatically reconnect after connection loss bool auto_reconnect_{true}; - /// @brief Whether the websocket is currently connected - bool connected_{false}; + /// @brief Whether the websocket is currently connected. Written by the IXWebSocket + /// callback thread, read cross-thread via is_connected(), hence atomic. + std::atomic connected_{false}; }; } // namespace sendspin diff --git a/src/host/server_connection.cpp b/src/host/server_connection.cpp index c9613c6..110fa3b 100644 --- a/src/host/server_connection.cpp +++ b/src/host/server_connection.cpp @@ -22,9 +22,15 @@ namespace sendspin { +static const char* const TAG = "sendspin.server_conn"; + SendspinServerConnection::SendspinServerConnection(std::shared_ptr ws, int sockfd) : ws_(std::move(ws)), sockfd_(sockfd) { - // IXWebSocket handles TCP_NODELAY internally + // Note: IXWebSocket only sets TCP_NODELAY on outbound connects, not on accepted sockets, + // and its public API does not expose the accepted socket's fd, so Nagle stays enabled + // here. Time messages on this host-server path can therefore see up to ~40 ms of + // coalescing delay. The ESP server path sets TCP_NODELAY explicitly; revisit if + // IXWebSocket ever exposes the accepted socket. } void SendspinServerConnection::start() { @@ -104,10 +110,17 @@ void SendspinServerConnection::handle_message(const std::string& data, bool is_b int64_t receive_time) { if (!data.empty()) { uint8_t* dest = this->prepare_receive_buffer(data.size()); - if (dest != nullptr) { - std::copy(data.begin(), data.end(), dest); - this->commit_receive_buffer(data.size()); + if (dest == nullptr) { + // Dispatching would hand a stale/partial buffer to the protocol layer. Drop the + // connection instead: the close event tears the slot down on the main loop. + SS_LOGE(TAG, "Allocation failed, dropping connection"); + this->disable_message_dispatch(); + this->reset_websocket_payload(); + this->trigger_close(); + return; } + std::copy(data.begin(), data.end(), dest); + this->commit_receive_buffer(data.size()); } this->dispatch_completed_message(!is_binary, receive_time); } diff --git a/src/host/ws_server.cpp b/src/host/ws_server.cpp index 4c2f57e..b451190 100644 --- a/src/host/ws_server.cpp +++ b/src/host/ws_server.cpp @@ -19,7 +19,7 @@ #include "platform/time.h" #include "server_connection.h" -#include +#include namespace sendspin { @@ -49,9 +49,16 @@ bool SendspinWsServer::start(SendspinClient* client, bool /*task_stack_in_psram* return; } - // Generate a synthetic sockfd from the pointer for connection lookup - int synthetic_sockfd = static_cast(reinterpret_cast(ws.get()) & - std::numeric_limits::max()); + // Generate a synthetic sockfd from a monotonic counter for connection lookup. + // (A pointer-derived id could repeat when a freed ix::WebSocket's address is + // reused, briefly mis-routing close events to a live connection.) + static std::atomic next_synthetic_sockfd{1}; + int synthetic_sockfd = next_synthetic_sockfd.fetch_add(1); + if (synthetic_sockfd <= 0) { + // Wrapped after ~2 billion connections; restart the sequence + next_synthetic_sockfd.store(1); + synthetic_sockfd = next_synthetic_sockfd.fetch_add(1); + } SS_LOGD(TAG, "New client connection (synthetic sockfd %d)", synthetic_sockfd); diff --git a/tests/test_spsc_ring_buffer.cpp b/tests/test_spsc_ring_buffer.cpp index ae4adb0..66ee8af 100644 --- a/tests/test_spsc_ring_buffer.cpp +++ b/tests/test_spsc_ring_buffer.cpp @@ -116,7 +116,7 @@ TEST(SpscRingBuffer, UnalignedStorageSizeJustUnderAligned) { TEST(SpscRingBuffer, CreateRejectsTooSmallStorage) { std::vector storage(16); SpscRingBuffer rb; - // After rounding down, less than one header plus one aligned payload unit fits. + // After rounding down, less than one header plus one aligned byte of payload fits. EXPECT_FALSE(rb.create(15, storage.data())); EXPECT_FALSE(rb.create(8, storage.data())); EXPECT_TRUE(rb.create(16, storage.data()));