diff --git a/docs/integration-guide.md b/docs/integration-guide.md index 7fcc615..5ea2d62 100644 --- a/docs/integration-guide.md +++ b/docs/integration-guide.md @@ -51,7 +51,7 @@ SendspinClient client(std::move(config)); ## Step 2: Add Roles -Add only the roles your application needs. All roles must be added before calling `start_server()`. +Add only the roles your application needs. All roles must be added before the first call to `start()`. ### Player Role (Audio Playback) @@ -494,6 +494,12 @@ struct MyClientListener : SendspinClientListener { void on_release_high_performance() override { esp_wifi_set_ps(WIFI_PS_MIN_MODEM); } + + // Called when a request_stop() teardown completes (see Clean shutdown below). Everything + // is torn down at this point; it is safe to call start() again or destroy the client. + void on_stopped() override { + mark_client_stopped(); + } }; ``` @@ -522,7 +528,7 @@ client.set_persistence_provider(&persistence_provider); // Optional ```cpp // Start the WebSocket server and sync task. // Task priorities and PSRAM settings are taken from SendspinClientConfig. -if (!client.start_server()) { +if (!client.start()) { // Handle failure return 1; } @@ -537,8 +543,28 @@ while (running) { std::this_thread::sleep_for(std::chrono::milliseconds(10)); } -// Clean shutdown -client.disconnect(SendspinGoodbyeReason::SHUTDOWN); +// Clean shutdown: sends a goodbye, closes every connection, and stops the WebSocket +// server and background threads. Call start() again later to restart the whole stack; +// disconnect(SendspinGoodbyeReason) is still available to drop just the active +// connection while leaving the server listening. +client.stop(); +``` + +If blocking the main loop during shutdown is a concern (for example inside a firmware main +loop), use `request_stop()` instead: it returns immediately, and the teardown completes over +subsequent `loop()` calls once every connection has closed and the background threads have wound +down. A fixed grace deadline caps how long that wait runs before the teardown is forced, but not +the forced teardown itself: the `loop()` tick that reaches the deadline joins the role threads, +and a listener callback still running holds that join until it returns. Completion is reported +through `SendspinClientListener::on_stopped()` and can be polled via `get_run_state()`: + +```cpp +client.request_stop(); + +while (client.get_run_state() != SendspinRunState::STOPPED) { + client.loop(); // on_stopped() fires from here when the teardown finishes + std::this_thread::sleep_for(std::chrono::milliseconds(10)); +} ``` ## Sending Commands @@ -700,7 +726,7 @@ int main() { player.set_listener(&player_listener); client.set_network_provider(&network); - client.start_server(); + client.start(); while (true) { client.loop(); @@ -917,6 +943,18 @@ These represent commands the server can send to the player. The player advertise | `ERROR` | Error state | | `EXTERNAL_SOURCE` | Playing from an external source | +### SendspinRunState + +| Value | Description | +|---|---| +| `STOPPED` | Before the first `start()` and after a stop completes | +| `RUNNING` | After a successful `start()` | +| `STOPPING` | During either teardown path, until it completes | + +Returned by `client.get_run_state()`; `is_started()` is equivalent to `RUNNING`. `STOPPING` covers +both the window between `request_stop()` and its completion and the duration of a synchronous +`stop()` call, so a callback invoked during a teardown observes `STOPPING`, not `STOPPED`. + ### SendspinGoodbyeReason | Value | Description | diff --git a/docs/internals.md b/docs/internals.md index 2dc016c..ef7c244 100644 --- a/docs/internals.md +++ b/docs/internals.md @@ -43,25 +43,32 @@ On host builds, `platform_configure_thread()` is a no-op; threads use OS default ### Thread Lifecycle -**Sync task** (`src/sync_task.cpp:620`): +All three role threads stop through the same three entry points: the client's synchronous +`stop()` (signal + join via each role's `Impl::stop()`), the asynchronous `request_stop()` +(signal via `Impl::request_stop()`, join deferred to `loop()`'s completion tick once the thread +reports exit), and each `Impl`'s destructor. Threads are restartable: a later +`SendspinClient::start()` re-runs each role's `Impl::start()`, which clears the stale stop/exit +flags before spawning a fresh thread. + +**Sync task** (`src/sync_task.cpp`): 1. `SyncTask::start()` configures the thread and spawns it. 2. The caller blocks until the thread reaches IDLE state (`TASK_IDLE` event flag) or exits early due to an allocation failure (`TASK_STOPPED`). -3. The thread runs a persistent outer loop for the lifetime of the client. -4. `SyncTask::stop()` sets `COMMAND_STOP` and joins the thread. Called from `SyncTask`'s destructor, which is triggered by `sync_task_.reset()` in `PlayerRole::Impl`'s destructor. +3. The thread runs a persistent outer loop while the client is started, idling between streams. +4. `SyncTask::stop()` sets `COMMAND_STOP`, joins the thread, and drains the encoded ring buffer so no stale chunk survives into a restarted session. `SyncTask::request_stop()` only sets the flag; `has_thread_exited()` reports `TASK_STOPPED` so the eventual join is instant. The exit path returns any borrowed ring entry and clears `TASK_RUNNING` before announcing `TASK_STOPPED`. **Visualizer drain** (`src/visualizer_role.cpp`): 1. `VisualizerRole::Impl::start()` spawns the drain thread. 2. The thread blocks on ring buffer receives with a 50 ms timeout. -3. `VisualizerRole::Impl` destructor sets `COMMAND_STOP` and joins. +3. `stop()` sets `COMMAND_STOP`, joins, and then flushes the ring buffer so no stale entry (queued before the stop, or pushed by the network thread into the gap between the async `request_stop()` exit and this join) survives into a restarted session and gets decoded against the new session's `spectrum_bin_count`/`tracks_downbeats`. `request_stop()` only signals, and the thread sets `THREAD_EXITED` on exit for `has_stopped()`. **Artwork decode** (`src/artwork_role.cpp`): 1. `ArtworkRole::Impl::start()` spawns the decode thread. 2. The thread blocks on notification queue receives with a 100 ms timeout. 3. On notification: calls `on_image_decode()`, then merges an `ArtworkDisplayUpdate` (the slot's server display timestamp plus the `stream_epoch` it was decoded under) into the `ArtworkRole::Impl::EventState::display_slot` `InboxSlot` via `merge_artwork_display_update`. The main loop's `ArtworkRole::Impl::drain_events()` folds the taken update into its main-thread-only `held_display_*` state and fires `on_image_display()` once the timestamp is reached. Latest-wins per slot: if a newer frame's timestamp overwrites the pending one before the main loop takes it, only the newer display fires; the per-slot epoch lets the deadline sweep drop a display whose stream was replaced after the hand-off. -4. `ArtworkRole::Impl` destructor sets `COMMAND_STOP` and joins. +4. `stop()` sets `COMMAND_STOP` and joins; `request_stop()` only signals, and the thread sets `THREAD_EXITED` on exit for `has_stopped()`. **Destruction order** matters because external audio callbacks may still reference the sync task. `PlayerRole::Impl`'s destructor resets the sync task first (`sync_task_.reset()`) before tearing down anything else, so the thread is fully joined before any shared state is destroyed. @@ -219,26 +226,30 @@ The bump arena suits ArduinoJson's allocation pattern: during a parse the varian ├─ Acquire/release high-performance networking around burst └─ Notify listener of sync error when burst completes -3. Drain inbox event ring (when INBOX_TOPIC_EVENTS is set) - ├─ Feed TIME_RESPONSE events into time_burst_->on_time_response() - ├─ Fire CONTROLLER/METADATA/COLOR_CLEARED via each role's handle_cleared_event() - ├─ Dispatch PLAYER_STREAM via player_->impl_->on_stream_ring_event() - ├─ Dispatch ARTWORK_STREAM via artwork_->impl_->handle_stream_ring_event() - └─ Dispatch VISUALIZER_STREAM via visualizer_->impl_->handle_stream_ring_event() - -4. Role event draining (each role's impl_->drain_events(), gated on impl_->needs_drain(slot_bits)) - ├─ player_->impl_->drain_events() - ├─ controller_->impl_->drain_events() - ├─ metadata_->impl_->drain_events() - ├─ color_->impl_->drain_events() - └─ artwork_->impl_->drain_events() (display-deadline sweep; visualizer has no drain_events()) - -5. Drain group_slot (when INBOX_TOPIC_GROUP is set in slot_bits) - └─ Apply group deltas, fire on_group_update, persist last played server +3. drain_inbox_events() + ├─ Drain inbox event ring (when INBOX_TOPIC_EVENTS is set) + │ ├─ Feed TIME_RESPONSE events into time_burst_->on_time_response() + │ ├─ Fire CONTROLLER/METADATA/COLOR_CLEARED via each role's handle_cleared_event() + │ ├─ Dispatch PLAYER_STREAM via player_->impl_->on_stream_ring_event() + │ ├─ Dispatch ARTWORK_STREAM via artwork_->impl_->handle_stream_ring_event() + │ └─ Dispatch VISUALIZER_STREAM via visualizer_->impl_->handle_stream_ring_event() + ├─ Role event draining (each role's impl_->drain_events(), gated on impl_->needs_drain(slot_bits)) + │ ├─ player_->impl_->drain_events() + │ ├─ controller_->impl_->drain_events() + │ ├─ metadata_->impl_->drain_events() + │ ├─ color_->impl_->drain_events() + │ └─ artwork_->impl_->drain_events() (display-deadline sweep; visualizer has no drain_events()) + └─ Drain group_slot (when INBOX_TOPIC_GROUP is set in slot_bits) + └─ Apply group deltas, fire on_group_update, persist last played server + +4. Drive a pending request_stop() to completion (only while STOPPING): once every role thread has + reported exit and no connection remains, or the grace deadline passes, call finish_stop() ``` This ordering matters: connection lifecycle events are processed before role events, and time sync before audio processing, so that roles always see a consistent connection and time state. +`drain_inbox_events()` (step 3) is a standalone private method, not inlined into `loop()`: `finish_stop()` (`SendspinClient::stop()`/a completing `request_stop()`) calls it a second time, after tearing the connection and role threads down but before firing `on_stopped()`. This guarantees the roles' clear callbacks that `cleanup_connection_state()` just queued (STREAM_END / `*_CLEARED` / the artwork and visualizer stream events) are delivered before `on_stopped()` signals completion, rather than waiting for a consumer `loop()` call that may never come once the client reports itself stopped. See "Manager Shutdown" below. + Each `loop()` section that would otherwise take a mutex first consults a lock-free atomic hint, so an idle tick pays only for the atomic loads it needs to decide there is nothing to do. `ConnectionManager` keeps four such hints, each refreshed under the owning mutex right after the container/pointer it mirrors changes (always re-derived from `.size()` or the assigned value, never incremented in place, so the hint cannot drift from ground truth): - `has_pending_events_` (under `conn_mutex_`): set at every push into either deferred connection-event queue, cleared once `loop()` has swapped both out. Lets `loop()` skip the `conn_mutex_` acquisition when neither queue has anything pending. @@ -248,7 +259,7 @@ Each `loop()` section that would otherwise take a mutex first consults a lock-fr Steady state is therefore cheap: connected-and-idle costs one `conn_ptr_mutex_` acquisition (the current/nursery copy ahead of the `conn->loop()` calls) plus a handful of atomic loads; disconnected-and-idle costs zero mutex acquisitions. The mutex-protected containers and pointers remain the ground truth in every case; the hints only decide whether it is worth locking to look. -The Inbox drain steps gate the same way, off two lock-free `poll()` snapshots of the topic bitmask. `inbox_bits` is taken first and gates only the event-ring drain (step 3). `slot_bits` is taken *after* that drain completes and gates the role drains (step 4) and the group drain (step 5); the second snapshot catches topic bits a producer set while the ring drain was running (including a ring event's own side effects re-entering the inbox). A bit either snapshot races and misses is not lost - it stays set and the next tick's `poll()` observes it (bounded staleness, per `Inbox::poll()`). Each role's `needs_drain(slot_bits)` decides whether its drain runs: mostly a simple `slot_bits & INBOX_TOPIC_*` test, but the player, metadata, and artwork roles OR in a main-thread-only carry-over term (the player's `awaiting_sync_idle_events`, the metadata role's future-dated `held_delta`, the artwork role's nonzero `held_display_mask`) so that work waiting out a deadline no inbox bit tracks still gets a drain every tick until it fires. +The Inbox drain steps inside `drain_inbox_events()` gate the same way, off two lock-free `poll()` snapshots of the topic bitmask. `inbox_bits` is taken first and gates only the event-ring drain. `slot_bits` is taken *after* that drain completes and gates the role drains and the group drain; the second snapshot catches topic bits a producer set while the ring drain was running (including a ring event's own side effects re-entering the inbox). A bit either snapshot races and misses is not lost - it stays set and the next tick's `poll()` observes it (bounded staleness, per `Inbox::poll()`). Each role's `needs_drain(slot_bits)` decides whether its drain runs: mostly a simple `slot_bits & INBOX_TOPIC_*` test, but the player, metadata, and artwork roles OR in a main-thread-only carry-over term (the player's `awaiting_sync_idle_events`, the metadata role's future-dated `held_delta`, the artwork role's nonzero `held_display_mask`) so that work waiting out a deadline no inbox bit tracks still gets a drain every tick until it fires. ## Role Event Draining @@ -453,6 +464,23 @@ When a connection is lost (`on_connection_lost`): - **ESP server**: the goodbye text is queued as an httpd worker job. The worker resolves the connection by `lock()`ing the `weak_ptr` captured in the queued arg when the goodbye was enqueued; if it resolves it sends the frame, then runs the completion lambda that calls `trigger_close()`. The session slot installed in `open_callback` keeps the connection alive across that whole sequence even after `ConnectionManager`'s observer `shared_ptr` is dropped. The session is finally freed when httpd invokes the slot's `free_fn` (see [Server connection ownership (ESP)](#server-connection-ownership-esp)). The completion lambda also captures a `weak_ptr` to make this lifetime explicit — `trigger_close()` is skipped if the conn has already been freed. Goodbye is one of the two messages that pass `allow_before_hello=true`, so it is not blocked by the pre-hello send gate (a rejected connection is told to leave before it ever sends a hello). - **Host client**: the IXWebSocket send is synchronous, so the goodbye and close have both completed by the time `disconnect()` returns and the `shared_ptr` drops the last reference. +### Manager Shutdown + +`SendspinClient::stop()`/`request_stop()` tear the manager down through two entry points: + +- `ConnectionManager::begin_stop(reason)` (the `request_stop()` path) clears the `accepting_` atomic and sends a goodbye to every connected peer, but leaves the WebSocket server up so queued goodbye sends can still flush (on ESP they are httpd worker jobs). Connections then leave their slots through the normal close-event path over subsequent `loop()` ticks, polled via `has_connections()`. +- `ConnectionManager::stop(reason)` (the synchronous path, and `request_stop()`'s finisher) runs `begin_stop()`, destroys the WebSocket server, force-drops every remaining slot without a second goodbye, and discards pending lifecycle events whose connections were just released. + +While `accepting_` is false, `on_new_connection()` rejects a newly delivered peer with a goodbye instead of admitting it into a nursery that is about to be drained, and `loop()`'s server-start block is gated so a stop in progress cannot belatedly open the server. The rejection releases the connection immediately, so `has_connections()` never covers it; on ESP, where the goodbye is queued to an httpd worker, an asynchronous stop can therefore complete and tear the server down before that send runs, and this one peer loses its goodbye. That is accepted: tracking the peer to closure would mean admitting it to the bounded nursery (`loop()` copies the nursery into a fixed array sized `NURSERY_CAPACITY + 1`) and disconnecting it from the network thread rather than the main loop, which is too much to trade for a best-effort frame. `accepting_` is authoritative control state (set by `init_server()`, cleared by `begin_stop()`/`stop()`), unlike the manager's other atomics, which are lock-free hints over mutex-protected containers. + +`begin_stop()` quiesces the connections it goodbyes: it calls `disconnect_impl()` with `quiescing` set, which calls `disable_message_dispatch()` on the current slot and on every connected nursery entry before their goodbyes go out. A peer already told SHUTDOWN therefore cannot deliver `stream/start` or binary role data for the rest of the grace window. This matters because a nursery entry is released without `cleanup_connection_state()` (it was never admitted, so there is no admitted state to clear), so anything such a peer pushed into the roles would otherwise outlive the teardown whenever no connection had been promoted. The public per-connection `disconnect()` keeps dispatch enabled: a client dropping one session while still running has no teardown window for late frames to corrupt. + +`loop()`'s promotion scan is gated on `accepting_` as the structural backstop. With dispatch disabled an in-flight `server/hello` no longer flips `is_handshake_complete()` during the STOPPING window, but the scan stays gated so promotion is impossible while stopping rather than merely unreachable: promoting a peer there would hand it a `client/state` message after it was already told SHUTDOWN. It makes no difference to the teardown deadline either way, since `has_connections()` counts nursery entries as well as the current slot. `stop()`'s force-drain releases such a peer without a second goodbye, and `accepting_` is set again by the next `init_server()` before any future promotion. + +`loop()`'s two hello sites are gated the same way, so that a peer already told SHUTDOWN is not then handed a `client/hello`. `initiate_hello()` never sends inline: it arms a per-connection retry entry that a later tick's retry scan sends. Since `disconnect()` leaves a connected nursery peer parked with that entry intact, both the arm (from an outbound connection's transport-connected event) and the retry scan itself skip while `!accepting_`; otherwise a peer admitted on the tick before `request_stop()` would still get its hello out after the goodbye. Nothing leaks by skipping, because every path that erases from `nursery_` calls `remove_hello_retry()`, and `stop()`'s force-drain releases whatever is still parked. + +`SendspinClient::finish_stop()` (the shared teardown behind both `stop()` and a completing `request_stop()`) guards against re-entrancy with a `tearing_down_` flag and keeps `run_state_` at `STOPPING` (never `STOPPED`) for the duration: `connection_manager_->stop()` synchronously reaches `cleanup_connection_state()`, which can invoke a listener callback (`on_release_high_performance()`, whenever a time-sync burst is in flight) that calls back into `start()`/`stop()`/`request_stop()`/`connect_to()`. With `run_state_` still `STOPPING`, those calls refuse or no-op instead of racing the teardown in progress (`start()` in particular would otherwise see `STOPPED` and spin up a second set of role threads and server on top of the one being torn down). `finish_stop()` sets `run_state_ = STOPPED` and clears `tearing_down_` only at the very end, right before firing `on_stopped()`, so that calling `start()` again from inside `on_stopped()` -- a documented, supported pattern -- still works normally. + ### Server connection ownership (ESP) On the ESP build, `SendspinServerConnection` lifetime is pinned to the httpd session rather than to `ConnectionManager`: @@ -511,3 +539,5 @@ A listener callback fired from the main loop can synchronously re-enter connecti - `PlayerRole::Impl::cleanup_generation`, bumped by the player's `cleanup()`. `drain_events()` snapshots it around each `on_stream_start()` call; if it changes, the stream was torn down from inside the callback, so the player abandons the rest of the batch rather than re-arm the sync task for a dead stream. `stream_active` is set before the callback runs, so the STREAM_END that `cleanup()` enqueued still passes its gate and delivers a paired `on_stream_end()`. Neither counter needs atomics: they only let an in-flight drain notice that teardown ran underneath it and stop touching state that cleanup already reset. + +A related but distinct guard covers the teardown call itself rather than an in-flight drain: `SendspinClient::tearing_down_`, a plain `bool` set for the duration of `finish_stop()`. The same `cleanup_connection_state()` call above can invoke `on_release_high_performance()` (whenever a time-sync burst is in flight), and a listener that calls `stop()`/`request_stop()`/`start()`/`connect_to()` from inside that callback must not recurse into a second `finish_stop()` or observe `run_state_ == STOPPED` mid-teardown. See "Manager Shutdown" above. diff --git a/examples/basic_client/main.cpp b/examples/basic_client/main.cpp index 9f6889d..0ae5f2c 100644 --- a/examples/basic_client/main.cpp +++ b/examples/basic_client/main.cpp @@ -325,7 +325,7 @@ int main(int argc, char* argv[]) { // Start the server fprintf(stderr, "Starting Sendspin basic client on port %u...\n", server_port); - if (!client.start_server()) { + if (!client.start()) { fprintf(stderr, "Failed to start server\n"); return 1; } @@ -373,7 +373,7 @@ int main(int argc, char* argv[]) { #ifdef SENDSPIN_HAS_MDNS mdns.stop(); #endif - client.disconnect(SendspinGoodbyeReason::SHUTDOWN); + client.stop(); #ifndef SENDSPIN_HAS_PORTAUDIO fprintf(stderr, "Total audio bytes received: %zu\n", null_audio_total_bytes); diff --git a/examples/tui_client/main.cpp b/examples/tui_client/main.cpp index 87fd527..dee1174 100644 --- a/examples/tui_client/main.cpp +++ b/examples/tui_client/main.cpp @@ -763,7 +763,7 @@ int main(int argc, char* argv[]) { #endif // Start the server - if (!client.start_server()) { + if (!client.start()) { fprintf(stderr, "Failed to start server\n"); return 1; } @@ -951,7 +951,7 @@ int main(int argc, char* argv[]) { mdns_browser.stop(); mdns.stop(); #endif - client.disconnect(SendspinGoodbyeReason::SHUTDOWN); + client.stop(); return 0; } diff --git a/include/sendspin/client.h b/include/sendspin/client.h index 5f5fdf2..230888c 100644 --- a/include/sendspin/client.h +++ b/include/sendspin/client.h @@ -54,6 +54,17 @@ class VisualizerRole; // Forward declarations for listener types struct GroupUpdateObject; +/// @brief Lifecycle state of a SendspinClient +/// STOPPED before the first start() and after a stop completes; RUNNING after a successful +/// start(); STOPPING for the duration of either teardown path, whether that is request_stop()'s +/// deferred completion or a direct stop() call, until the teardown finishes and the state +/// becomes STOPPED. +enum class SendspinRunState : uint8_t { + STOPPED = 0, + RUNNING = 1, + STOPPING = 2, +}; + /// @brief Listener for SendspinClient events /// All methods fire on the main loop thread class SendspinClientListener { @@ -63,6 +74,14 @@ class SendspinClientListener { /// @brief Called when the group state is updated by the server virtual void on_group_update(const GroupUpdateObject& /*group*/) {} + /// @brief Called when a request_stop() teardown completes + /// Fires once per request_stop(), from loop() (or from inside a stop() call that finished + /// the teardown synchronously). When it fires, every connection is closed, the WebSocket + /// server and role threads are stopped, and it is safe to call start() again or destroy the + /// client. A synchronous stop() from the RUNNING state does not fire it; that call's return + /// is the completion signal. + virtual void on_stopped() {} + /// @brief Called after a time sync burst completes with the Kalman filter error virtual void on_time_sync_updated(float /*error*/) {} @@ -75,7 +94,7 @@ class SendspinClientListener { }; /// @brief Platform hook for network readiness -/// Must be set before start_server() +/// Must be set before start() class SendspinNetworkProvider { public: virtual ~SendspinNetworkProvider() = default; @@ -147,8 +166,11 @@ class SendspinTimeBurst; * 2. Construct a SendspinClient with that config * 3. Add roles via add_player(), add_controller(), add_metadata(), etc. * 4. Set listeners on each role and set the network provider on the client - * 5. Call start_server() to start the WebSocket server and background tasks + * 5. Call start() to start the WebSocket server and background tasks * 6. Call loop() periodically from the platform main loop + * 7. Call stop() to disconnect and shut everything down, or request_stop() for a non-blocking + * teardown that completes over loop() ticks (observe on_stopped() / get_run_state()); + * start() again to restart * * @code * struct MyPlayerListener : PlayerRoleListener { @@ -175,11 +197,13 @@ class SendspinTimeBurst; * player.set_listener(&player_listener); * client.add_controller(); * client.set_network_provider(&network_provider); - * client.start_server(); + * client.start(); * - * while (true) { + * while (running) { * client.loop(); * } + * + * client.stop(); * @endcode */ class SendspinClient { @@ -201,15 +225,73 @@ class SendspinClient { // Lifecycle // ======================================== - /// @brief Starts the WebSocket server and initializes the sync task (if audio is configured) + /// @brief Starts the client: role background threads and the WebSocket server + /// + /// Must be called from the main loop thread (see stop()). The server socket itself opens on + /// a later loop() tick, once the network provider reports ready. Add roles and set providers + /// before the first start(). Safe to call again after a stop completes; a no-op if already + /// running, and refused (returns false with a warning) while a request_stop() teardown is + /// still in progress. On failure every thread that did start is stopped again, so a + /// corrected retry begins clean. + /// @return true on success (or if already running), false on failure or while stopping + bool start(); + + /// @brief Stops the client: sends a goodbye, closes every connection, and stops the + /// WebSocket server and role background threads + /// + /// Must be called from the main loop thread (see disconnect()). Blocks until the role + /// background threads join. Every role is signaled before any is joined, so their stop + /// latencies overlap and the wait is the longest of them (the sync task's 500 ms idle poll) + /// rather than their sum. In-flight listener work extends it: a join cannot interrupt a + /// callback that is already running, so a slow on_audio_write(), on_image_decode(), or + /// visualizer per-frame callback adds its own duration on top. The goodbye is + /// best-effort: on ESP-IDF the send is queued to an httpd + /// worker and the server stop can close the session first; request_stop() avoids that by + /// keeping the server up through a grace window. loop() remains safe to call while stopped. + /// The roles' clear callbacks queued by the connection teardown are delivered before this + /// call returns, not on a later loop(). Called while a request_stop() is in progress, it + /// finishes that teardown synchronously and fires on_stopped() before returning. Call start() + /// to restart. No-op if already stopped. + void stop(); + + /// @brief Requests an asynchronous stop: goodbyes are sent and role threads are signaled + /// immediately, and the teardown completes over subsequent loop() ticks + /// + /// Must be called from the main loop thread (see disconnect()) and does not block: keep + /// calling loop() and the teardown finishes once every connection has closed and the role + /// threads have wound down, or after a fixed grace deadline (STOP_GRACE_MS in client.cpp), + /// whichever comes first. The deadline bounds when the final teardown + /// starts, not how long it takes: the loop() tick that reaches it joins the role threads, and + /// a listener callback still running (on_audio_write(), on_image_decode(), a visualizer + /// per-frame callback) holds that join for as long as it runs. Those callbacks gate shutdown, + /// so they must return promptly. Unlike stop(), the WebSocket server stays up through the grace + /// window, so on ESP-IDF the queued goodbye sends actually flush before the server is torn + /// down. Completion is signaled by SendspinClientListener::on_stopped() and observable via + /// get_run_state() == STOPPED, after which start() may be called again. No-op unless running. + /// + /// One exception to the non-blocking guarantee: tearing down an outbound connect_to() + /// connection stops its transport synchronously on both platforms (esp_websocket_client_stop() + /// on ESP-IDF, ix::WebSocket::stop() on host), so this call can stall the main loop for as + /// long as that transport takes to wind down. The stop cannot simply be deferred: the manager + /// releases its last reference to the connection as soon as the disconnect returns, so the + /// transport thread has to be joined before that happens. Inbound (server) connections are + /// unaffected on both platforms; they close asynchronously. + void request_stop(); + + /// @brief Legacy name for start(), kept for source compatibility + /// Unlike the pre-stop() start_server(), a repeat call while started is a no-op instead of + /// re-initializing the WebSocket server. /// @return true on success, false on failure - bool start_server(); + bool start_server() { + return this->start(); + } /// @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. + /// it concurrently with loop() would race those mutations. Ignored (with a warning) unless + /// the client is started. /// @param url WebSocket server URL (e.g., "ws://server.local:8927/sendspin") void connect_to(const std::string& url); @@ -225,7 +307,7 @@ class SendspinClient { void loop(); // ======================================== - // Role registration (call before start_server) + // Role registration (call before start()) // ======================================== #ifdef SENDSPIN_ENABLE_PLAYER @@ -339,29 +421,45 @@ class SendspinClient { // Queries // ======================================== + /// @brief Converts a server timestamp to the equivalent client timestamp + /// @param server_time Server-side timestamp in microseconds + /// @return Equivalent client-side timestamp in microseconds + int64_t get_client_time(int64_t server_time) const; + /// @brief Returns true if there is an active connection with completed handshake /// @return true if connected with a completed handshake, false otherwise bool is_connected() const; + /// @brief Returns the current group state + /// @return The current GroupUpdateObject (fields are optional and may be unset) + const GroupUpdateObject& get_group_state() const { + return this->group_state_; + } + + /// @brief Returns the client lifecycle state. Main-thread only + /// STOPPING covers either teardown path: the window between request_stop() and its + /// completion, and the duration of a synchronous stop() call. A callback invoked during that + /// window (a role's clear callback, or on_release_high_performance()) observes STOPPING. + /// @return The current SendspinRunState + SendspinRunState get_run_state() const { + return this->run_state_; + } + /// @brief Returns the server information from the active connection's hello handshake /// @return ServerInformationObject if connected with a completed handshake, nullopt otherwise std::optional get_server_information() const; + /// @brief Returns true if the client is running. Main-thread only + /// Equivalent to get_run_state() == RUNNING; false while a request_stop() is in progress. + /// @return true if start() succeeded and no stop has begun since + bool is_started() const { + return this->run_state_ == SendspinRunState::RUNNING; + } + /// @brief Returns true if the time filter has received at least one measurement /// @return true if time synchronization has been established, false otherwise bool is_time_synced() const; - /// @brief Converts a server timestamp to the equivalent client timestamp - /// @param server_time Server-side timestamp in microseconds - /// @return Equivalent client-side timestamp in microseconds - int64_t get_client_time(int64_t server_time) const; - - /// @brief Returns the current group state - /// @return The current GroupUpdateObject (fields are optional and may be unset) - const GroupUpdateObject& get_group_state() const { - return this->group_state_; - } - // ======================================== // State updates // ======================================== @@ -379,7 +477,7 @@ class SendspinClient { this->listener_ = listener; } - /// @brief Sets the network provider (required before start_server()) + /// @brief Sets the network provider (required before start()) /// The provider must outlive this client void set_network_provider(SendspinNetworkProvider* provider) { this->network_provider_ = provider; @@ -411,6 +509,30 @@ class SendspinClient { /// @brief Cleans up playback state when the active streaming connection is removed void cleanup_connection_state(); + /// @brief Stops (joins) every role background thread; roles restart them on the next start() + void stop_role_threads(); + + /// @brief Performs the full teardown shared by stop() and a completing request_stop() + /// + /// Guarded against re-entrancy by tearing_down_: a listener callback synchronously invoked + /// from inside the teardown (e.g. on_release_high_performance() from + /// cleanup_connection_state()) may call back into stop()/request_stop()/start()/connect_to(). + /// run_state_ stays STOPPING (never STOPPED) for the duration, so those re-entrant calls see + /// the client as still transitioning and refuse or no-op instead of racing the in-progress + /// teardown; see "Re-entrant Teardown During Callback Dispatch" in docs/internals.md for the + /// pattern. + /// @param notify Whether to fire the listener's on_stopped() once torn down + void finish_stop(bool notify); + + /// @brief Drains the inbox event ring, role event slots, and group-update slot + /// + /// Shared by loop() (every tick) and finish_stop() (once, after teardown, so the roles' clear + /// callbacks reach the listener before on_stopped() does). + void drain_inbox_events(); + + /// @brief Signals every role background thread to stop without joining it + void request_stop_role_threads(); + /// @brief Builds the formatted client hello message from config std::string build_hello_message(); @@ -496,13 +618,22 @@ class SendspinClient { std::unique_ptr visualizer_; #endif + // 64-bit fields + /// Deadline (us) for a request_stop() teardown; loop() force-finishes past it. Main-thread + /// only, meaningful only while run_state_ == STOPPING. + int64_t stop_deadline_us_{0}; + // 32-bit fields SendspinClientState state_{SendspinClientState::SYNCHRONIZED}; // 8-bit fields bool high_performance_held_for_time_{false}; std::atomic high_performance_ref_count_{0}; - bool started_{false}; + SendspinRunState run_state_{SendspinRunState::STOPPED}; + /// True for the duration of finish_stop(), guarding it against re-entrant recursion from a + /// listener callback invoked synchronously during teardown (see finish_stop()'s doc comment). + /// Main-thread only. + bool tearing_down_{false}; }; } // namespace sendspin diff --git a/include/sendspin/config.h b/include/sendspin/config.h index 89977c6..3d2cf7f 100644 --- a/include/sendspin/config.h +++ b/include/sendspin/config.h @@ -32,7 +32,7 @@ namespace sendspin { // ============================================================================ /// @brief Configuration for a SendspinClient instance -/// Filled in by the platform (e.g., ESPHome) before calling start_server() +/// Filled in by the platform (e.g., ESPHome) before calling start() struct SendspinClientConfig { /// Unique client identifier. When left empty, the library falls back to the detected local /// network interface MAC address (the same value used for device_info.mac_address). diff --git a/src/artwork_role.cpp b/src/artwork_role.cpp index ea4f576..86d77d8 100644 --- a/src/artwork_role.cpp +++ b/src/artwork_role.cpp @@ -37,6 +37,9 @@ static constexpr uint32_t DRAIN_RECEIVE_TIMEOUT_MS = 100U; // Event flag bits for decode thread signaling static constexpr uint32_t COMMAND_STOP = (1 << 0); +// Set by the decode thread as it exits, so an asynchronous stop can poll has_stopped() and only +// join once the join is known to be instant (mirrors SyncTask's TASK_STOPPED). +static constexpr uint32_t THREAD_EXITED = (1 << 1); // ============================================================================ // Big-endian helpers @@ -98,6 +101,18 @@ bool ArtworkRole::Impl::start() { return false; } + // The flag object survives a stop()/start() cycle. This clear only matters on the async + // path: on a plain stop() the drain thread's own exiting wait() call already self-clears + // COMMAND_STOP (EventFlags::wait's clear_on_exit clears matched bits unconditionally, even + // ones already set before the call), so the bit already reads 0 here. On the async path + // request_stop() lets the thread exit (and self-clear) on its own, then a later stop() calls + // event_flags.set(COMMAND_STOP) on the now-dead thread with nothing left to clear it, so the + // bit is still set by the time start() runs. Clearing it unconditionally here, rather than + // relying on which path preceded this start(), keeps a stale bit from making the new thread's + // first wait() see COMMAND_STOP and exit immediately. THREAD_EXITED is the previous thread's + // exit marker, stale for the new one either way. + this->drain_task->event_flags.clear(COMMAND_STOP | THREAD_EXITED); + platform_configure_thread("SsArt", 4096, static_cast(this->config.priority), this->config.psram_stack); this->drain_task->drain_thread = std::thread(drain_thread_func, this); @@ -112,6 +127,20 @@ void ArtworkRole::Impl::stop() const { this->drain_task->drain_thread.join(); } +void ArtworkRole::Impl::request_stop() const { + if (!this->drain_task || !this->drain_task->drain_thread.joinable()) { + return; + } + this->drain_task->event_flags.set(COMMAND_STOP); +} + +bool ArtworkRole::Impl::has_stopped() const { + if (!this->drain_task || !this->drain_task->drain_thread.joinable()) { + return true; + } + return (this->drain_task->event_flags.get() & THREAD_EXITED) != 0U; +} + void ArtworkRole::Impl::build_hello_fields(ClientHelloMessage& msg) const { if (this->artwork_channels.empty()) { return; @@ -703,6 +732,7 @@ void ArtworkRole::Impl::drain_thread_func(ArtworkRole::Impl* self) { self->process_notification(notif); } + flags.set(THREAD_EXITED); SS_LOGD(TAG, "Decode thread stopped"); } diff --git a/src/artwork_role_impl.h b/src/artwork_role_impl.h index 5d96931..c577f13 100644 --- a/src/artwork_role_impl.h +++ b/src/artwork_role_impl.h @@ -185,6 +185,8 @@ struct ArtworkRole::Impl { // ======================================== void stop() const; + void request_stop() const; + bool has_stopped() const; void enqueue_stream_event(ArtworkEventType event) const; // Merges a single-slot display delta into the accumulated cross-thread update. Called under // the Inbox mutex via InboxSlot::merge() (see process_notification), so it must stay a pure diff --git a/src/client.cpp b/src/client.cpp index 0f2c27c..a32d9f7 100644 --- a/src/client.cpp +++ b/src/client.cpp @@ -16,12 +16,14 @@ #include "connection.h" #include "connection_manager.h" +#include "constants.h" #include "inbox.h" #include "platform/compiler.h" #include "platform/json_arena.h" #include "platform/logging.h" #include "platform/memory.h" #include "platform/network_info.h" +#include "platform/time.h" #ifdef SENDSPIN_ENABLE_ARTWORK #include "artwork_role_impl.h" #endif @@ -48,6 +50,14 @@ static const char* const TAG = "sendspin.client"; namespace sendspin { +/// Grace deadline for a request_stop() teardown: covers the sync task's 500 ms idle poll +/// (IDLE_RECEIVE_TIMEOUT_MS in sync_task.cpp; keep the budget above it) plus margin for goodbye +/// sends to flush, close events to arrive, and the artwork/visualizer drain receive timeouts +/// (100 ms / 50 ms). Expiring early is bounded, not a fault: loop() force-finishes the stop, so a +/// peer that ignores its goodbye cannot hold the teardown open. +static constexpr int64_t STOP_GRACE_MS = 750; +static constexpr int64_t STOP_GRACE_US = STOP_GRACE_MS * US_PER_MS; + /// @brief Deferred event state for time responses and group updates on the main thread struct SendspinClient::EventState { Inbox inbox; @@ -116,43 +126,82 @@ LogLevel SendspinClient::get_log_level() { // Lifecycle // ============================================================================ -bool SendspinClient::start_server() { - this->started_ = true; +bool SendspinClient::start() { + if (this->run_state_ == SendspinRunState::RUNNING) { + return true; + } + if (this->run_state_ == SendspinRunState::STOPPING) { + SS_LOGW(TAG, "start() refused; a request_stop() teardown is still in progress"); + return false; + } // Load persisted state this->load_last_played_server(); + // Start role background threads. A failure part-way stops the threads that did start, so + // the client is back in the stopped state and a corrected retry begins clean. + bool roles_started = true; #ifdef SENDSPIN_ENABLE_PLAYER - if (this->player_) { - if (!this->player_->impl_->start()) { - return false; - } + if (roles_started && this->player_) { + roles_started = this->player_->impl_->start(); } #endif #ifdef SENDSPIN_ENABLE_VISUALIZER - if (this->visualizer_) { - if (!this->visualizer_->impl_->start()) { - return false; - } + if (roles_started && this->visualizer_) { + roles_started = this->visualizer_->impl_->start(); } #endif #ifdef SENDSPIN_ENABLE_ARTWORK - if (this->artwork_) { - if (!this->artwork_->impl_->start()) { - return false; - } + if (roles_started && this->artwork_) { + roles_started = this->artwork_->impl_->start(); } #endif + if (!roles_started) { + this->stop_role_threads(); + return false; + } + // Create and configure the WebSocket server (started later when network is ready) this->connection_manager_->init_server(this); + this->run_state_ = SendspinRunState::RUNNING; return true; } +void SendspinClient::stop() { + if (this->run_state_ == SendspinRunState::STOPPED) { + return; + } + // Finishing a pending request_stop() synchronously still announces its completion; a stop + // straight from RUNNING signals completion by returning instead. + this->finish_stop(/*notify=*/this->run_state_ == SendspinRunState::STOPPING); +} + +void SendspinClient::request_stop() { + if (this->run_state_ != SendspinRunState::RUNNING) { + return; + } + this->run_state_ = SendspinRunState::STOPPING; + this->stop_deadline_us_ = platform_time_us() + STOP_GRACE_US; + + // Goodbyes go out now (queued to httpd workers on ESP), with the server left up so they can + // flush; role threads start winding down concurrently. loop() finishes the teardown once + // every connection has closed and the sync task has exited, or at the deadline. + this->connection_manager_->begin_stop(SendspinGoodbyeReason::SHUTDOWN); + this->request_stop_role_threads(); +} + void SendspinClient::connect_to(const std::string& url) { + // Gated on the run state so a stopped (or stopping) client stays fully quiescent: an + // outbound connection created here would otherwise re-establish with no server or role + // threads running. + if (this->run_state_ != SendspinRunState::RUNNING) { + SS_LOGW(TAG, "connect_to() ignored; client is not running"); + return; + } this->connection_manager_->connect_to(url); } @@ -164,9 +213,15 @@ void SendspinClient::loop() { // Process connection lifecycle events (close, disconnect, hello, handoff, retry) this->connection_manager_->loop(); - // Handle time synchronization for the active connection via burst strategy + // Handle time synchronization for the active connection via burst strategy. Gated on RUNNING + // so a peer already told SHUTDOWN is sent nothing further: begin_stop() leaves the current + // connection in its slot until the close event arrives, and on ESP that close is queued to an + // httpd worker, so is_connected() stays true across at least one more tick. Only the burst is + // gated; the manager loop above and the teardown completion below must keep running while + // STOPPING to finish the stop. Nothing leaks if a burst is abandoned mid-flight: + // cleanup_connection_state() releases the high-performance hold unconditionally. auto* conn = this->connection_manager_->current(); - if (conn != nullptr) { + if (conn != nullptr && this->run_state_ == SendspinRunState::RUNNING) { auto result = this->time_burst_->loop(conn); if (result.sent && !this->high_performance_held_for_time_) { @@ -184,13 +239,47 @@ void SendspinClient::loop() { } // Process deferred events: all state mutations and user callbacks happen here, on the main - // loop thread, to avoid cross-thread data races. Two poll() snapshots gate the work below: - // inbox_bits (here) gates only the event-ring drain immediately following it; slot_bits - // (taken after that drain completes, below) gates the role drains and the group-update drain, - // since a role's InboxSlot can be written by a producer between this snapshot and that one. - // Both are lock-free atomic loads, so a tick with nothing pending performs zero inbox mutex - // acquisitions in this section. A bit either snapshot races and misses is picked up by the - // next tick's poll() -- bounded staleness, already documented on Inbox::poll(). + // loop thread, to avoid cross-thread data races. See drain_inbox_events() for the two-snapshot + // gating rationale. + this->drain_inbox_events(); + + // Drive a pending request_stop() to completion. Runs after the manager loop above so this + // tick's close events are already reflected in has_connections(). Completion waits for every + // role thread to report exit, so finish_stop()'s joins are instant on this path; the + // deadline caps the wait when a peer never delivers its close or a thread is stuck in + // in-flight work, at the cost of a bounded blocking join on that final tick. + if (this->run_state_ == SendspinRunState::STOPPING) { + bool roles_done = true; +#ifdef SENDSPIN_ENABLE_PLAYER + if (this->player_) { + roles_done = this->player_->impl_->has_stopped(); + } +#endif +#ifdef SENDSPIN_ENABLE_VISUALIZER + if (roles_done && this->visualizer_) { + roles_done = this->visualizer_->impl_->has_stopped(); + } +#endif +#ifdef SENDSPIN_ENABLE_ARTWORK + if (roles_done && this->artwork_) { + roles_done = this->artwork_->impl_->has_stopped(); + } +#endif + if ((roles_done && !this->connection_manager_->has_connections()) || + platform_time_us() >= this->stop_deadline_us_) { + this->finish_stop(/*notify=*/true); + } + } +} + +void SendspinClient::drain_inbox_events() { + // Two poll() snapshots gate the work below: inbox_bits (here) gates only the event-ring drain + // immediately following it; slot_bits (taken after that drain completes, below) gates the role + // drains and the group-update drain, since a role's InboxSlot can be written by a producer + // between this snapshot and that one. Both are lock-free atomic loads, so a tick with nothing + // pending performs zero inbox mutex acquisitions in this section. A bit either snapshot races + // and misses is picked up by the next tick's poll() -- bounded staleness, already documented on + // Inbox::poll(). const uint32_t inbox_bits = this->event_state_->inbox.poll(); // --- Time sync events --- @@ -391,13 +480,13 @@ void SendspinClient::loop() { } // ============================================================================ -// Role registration (call before start_server) +// Role registration (call before start()) // ============================================================================ #ifdef SENDSPIN_ENABLE_PLAYER PlayerRole& SendspinClient::add_player(PlayerRoleConfig config) { - if (this->started_) { - SS_LOGW(TAG, "add_player() called after start_server(); role may not initialize correctly"); + if (this->run_state_ != SendspinRunState::STOPPED) { + SS_LOGW(TAG, "add_player() called after start(); role may not initialize correctly"); } this->player_ = std::make_unique(std::move(config), this, this->persistence_provider_); @@ -408,8 +497,8 @@ PlayerRole& SendspinClient::add_player(PlayerRoleConfig config) { #ifdef SENDSPIN_ENABLE_CONTROLLER ControllerRole& SendspinClient::add_controller() { - if (this->started_) { - SS_LOGW(TAG, "add_controller() called after start_server()"); + if (this->run_state_ != SendspinRunState::STOPPED) { + SS_LOGW(TAG, "add_controller() called after start()"); } this->controller_ = std::make_unique(this); this->controller_->impl_->attach_inbox(this->event_state_->inbox); @@ -419,8 +508,8 @@ ControllerRole& SendspinClient::add_controller() { #ifdef SENDSPIN_ENABLE_METADATA MetadataRole& SendspinClient::add_metadata() { - if (this->started_) { - SS_LOGW(TAG, "add_metadata() called after start_server()"); + if (this->run_state_ != SendspinRunState::STOPPED) { + SS_LOGW(TAG, "add_metadata() called after start()"); } this->metadata_ = std::make_unique(this); this->metadata_->impl_->attach_inbox(this->event_state_->inbox); @@ -430,8 +519,8 @@ MetadataRole& SendspinClient::add_metadata() { #ifdef SENDSPIN_ENABLE_COLOR ColorRole& SendspinClient::add_color() { - if (this->started_) { - SS_LOGW(TAG, "add_color() called after start_server()"); + if (this->run_state_ != SendspinRunState::STOPPED) { + SS_LOGW(TAG, "add_color() called after start()"); } this->color_ = std::make_unique(this); this->color_->impl_->attach_inbox(this->event_state_->inbox); @@ -441,8 +530,8 @@ ColorRole& SendspinClient::add_color() { #ifdef SENDSPIN_ENABLE_ARTWORK ArtworkRole& SendspinClient::add_artwork(ArtworkRoleConfig config) { - if (this->started_) { - SS_LOGW(TAG, "add_artwork() called after start_server()"); + if (this->run_state_ != SendspinRunState::STOPPED) { + SS_LOGW(TAG, "add_artwork() called after start()"); } this->artwork_ = std::make_unique(std::move(config), this); this->artwork_->impl_->attach_inbox(this->event_state_->inbox); @@ -452,8 +541,8 @@ ArtworkRole& SendspinClient::add_artwork(ArtworkRoleConfig config) { #ifdef SENDSPIN_ENABLE_VISUALIZER VisualizerRole& SendspinClient::add_visualizer(VisualizerRoleConfig config) { - if (this->started_) { - SS_LOGW(TAG, "add_visualizer() called after start_server()"); + if (this->run_state_ != SendspinRunState::STOPPED) { + SS_LOGW(TAG, "add_visualizer() called after start()"); } this->visualizer_ = std::make_unique(std::move(config), this); this->visualizer_->impl_->attach_inbox(this->event_state_->inbox); @@ -465,23 +554,16 @@ VisualizerRole& SendspinClient::add_visualizer(VisualizerRoleConfig config) { // Queries // ============================================================================ -bool SendspinClient::is_connected() const { - return this->connection_manager_->is_connected(); -} - -bool SendspinClient::is_time_synced() const { - // 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 { // 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; } +bool SendspinClient::is_connected() const { + return this->connection_manager_->is_connected(); +} + std::optional SendspinClient::get_server_information() const { // current_shared(): public accessor, callable from any thread. auto conn = this->connection_manager_->current_shared(); @@ -491,6 +573,13 @@ std::optional SendspinClient::get_server_information() return conn->get_server_information(); } +bool SendspinClient::is_time_synced() const { + // 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(); +} + // ============================================================================ // State updates // ============================================================================ @@ -598,6 +687,106 @@ void SendspinClient::cleanup_connection_state() { } } +void SendspinClient::stop_role_threads() { + // Signal every role before joining any, so their receive timeouts elapse concurrently rather + // than in series: each role's stop() below both signals and joins, so without this an idle + // client would wait out the sum of the three timeouts instead of the longest one + // (ROLE_STOP_LATENCY_MS). Re-signalling inside stop() is harmless, the flags are idempotent. + // The request_stop() path has already signalled them by the time it reaches here; this makes + // the direct stop() path behave the same way. + this->request_stop_role_threads(); + +#ifdef SENDSPIN_ENABLE_PLAYER + if (this->player_) { + this->player_->impl_->stop(); + } +#endif +#ifdef SENDSPIN_ENABLE_VISUALIZER + if (this->visualizer_) { + this->visualizer_->impl_->stop(); + } +#endif +#ifdef SENDSPIN_ENABLE_ARTWORK + if (this->artwork_) { + this->artwork_->impl_->stop(); + } +#endif +} + +void SendspinClient::finish_stop(bool notify) { + // Re-entrancy guard: teardown below runs connection_manager_->stop(), which synchronously + // invokes listener callbacks (e.g. cleanup_connection_state() -> release_high_performance() + // -> on_release_high_performance() whenever a time-sync burst is in flight). A listener that + // calls stop()/request_stop()/start()/connect_to() from inside such a callback must not + // recurse into a second teardown or observe a state that lets it race the one in progress. + if (this->tearing_down_) { + return; + } + this->tearing_down_ = true; + + // Force (rather than assert) STOPPING: stop() can reach here directly from RUNNING, not only + // via a completing request_stop(). Staying in STOPPING (never STOPPED) for the duration means + // start() explicitly refuses and connect_to()/request_stop() no-op if a listener callback + // reached during teardown calls back into the client, instead of start() seeing STOPPED and + // spinning up a second set of role threads and server on top of the teardown still in flight. + this->run_state_ = SendspinRunState::STOPPING; + + // Goodbye and close every connection, then stop the WebSocket server so no new connections + // arrive. Dropping the current slot inside runs cleanup_connection_state(), which quiesces + // per-connection state (time burst, role state, high-performance holds) and queues the + // roles' clear callbacks for drain_inbox_events() below. + this->connection_manager_->stop(SendspinGoodbyeReason::SHUTDOWN); + + // Join role threads only after the connection teardown above has signaled their streams + // closed; a restart via start() re-creates them. + this->stop_role_threads(); + + // Reset session-scoped published state so a restart begins fresh: group_state_ is a delta + // accumulator that would otherwise keep serving the old session's group, and a stale state_ + // (e.g. ERROR) would be republished verbatim to the next server on handshake. Reset here + // rather than in cleanup_connection_state(), which also runs on reconnects and handoffs + // where carrying the state forward is intentional. Runs before drain_inbox_events() below so + // that drain cannot resurrect a stale group_state_/state_ even in the (already guarded-against + // by cleanup_connection_state()'s group_slot.reset()) case of a leftover group update. + this->group_state_ = GroupUpdateObject{}; + this->state_ = SendspinClientState::SYNCHRONIZED; + + // Deliver the roles' clear callbacks (STREAM_END / CONTROLLER_CLEARED / METADATA_CLEARED / + // COLOR_CLEARED / artwork+visualizer stream events) queued by cleanup_connection_state() above + // before on_stopped() fires below: the header documents on_stopped() as the signal that it is + // safe to destroy the client, so those callbacks must already have run by then rather than + // waiting for a consumer's next loop() call that may never come. + this->drain_inbox_events(); + + this->run_state_ = SendspinRunState::STOPPED; + // Clear before notifying: on_stopped() is documented to allow calling start() again from + // inside the callback, which requires run_state_ == STOPPED and tearing_down_ == false to + // proceed normally. + this->tearing_down_ = false; + + if (notify && this->listener_ != nullptr) { + this->listener_->on_stopped(); + } +} + +void SendspinClient::request_stop_role_threads() { +#ifdef SENDSPIN_ENABLE_PLAYER + if (this->player_) { + this->player_->impl_->request_stop(); + } +#endif +#ifdef SENDSPIN_ENABLE_VISUALIZER + if (this->visualizer_) { + this->visualizer_->impl_->request_stop(); + } +#endif +#ifdef SENDSPIN_ENABLE_ARTWORK + if (this->artwork_) { + this->artwork_->impl_->request_stop(); + } +#endif +} + std::string SendspinClient::build_hello_message() { ClientHelloMessage msg; msg.name = this->config_.name; diff --git a/src/connection_manager.cpp b/src/connection_manager.cpp index bd2dd8e..3782339 100644 --- a/src/connection_manager.cpp +++ b/src/connection_manager.cpp @@ -177,14 +177,20 @@ void ConnectionManager::connect_to(const std::string& url) { } void ConnectionManager::disconnect(SendspinGoodbyeReason reason) { - // Collect under the lock, send outside it: disconnect() can block on the transport (and on - // host outbound it joins the transport thread), which must not stall other manager entry - // points. The connections stay in their slots until their close events arrive (or the - // manager is destroyed). + this->disconnect_impl(reason, /*quiescing=*/false); +} + +void ConnectionManager::disconnect_impl(SendspinGoodbyeReason reason, bool quiescing) { + // Collect under the lock, send outside it: disconnect() can block on the transport, which must + // not stall other manager entry points. The connections stay in their slots until their close + // events arrive (or the manager is destroyed). std::vector> to_disconnect; { std::lock_guard lock(this->conn_ptr_mutex_); if (this->current_connection_ != nullptr && this->current_connection_->is_connected()) { + if (quiescing) { + this->current_connection_->disable_message_dispatch(); + } to_disconnect.push_back(this->current_connection_); } // Drain the nursery too. A connected entry gets a goodbye and leaves on its close event; an @@ -192,6 +198,9 @@ void ConnectionManager::disconnect(SendspinGoodbyeReason reason) { // release it here rather than leave it for the nursery deadline to reap. for (auto it = this->nursery_.begin(); it != this->nursery_.end();) { if (it->conn->is_connected()) { + if (quiescing) { + it->conn->disable_message_dispatch(); + } to_disconnect.push_back(it->conn); ++it; } else { @@ -211,6 +220,7 @@ void ConnectionManager::disconnect(SendspinGoodbyeReason reason) { void ConnectionManager::init_server(SendspinClient* client) { this->client_ = client; + this->accepting_.store(true, std::memory_order_release); this->ws_server_ = std::make_unique(); this->ws_server_->set_port(this->client_->config_.server_port); @@ -266,10 +276,70 @@ void ConnectionManager::init_server(SendspinClient* client) { }); } +void ConnectionManager::begin_stop(SendspinGoodbyeReason reason) { + // Close the admission door first: a peer delivered after this point is rejected with a + // goodbye in on_new_connection() rather than admitted into a nursery that is about to be + // torn down. + this->accepting_.store(false, std::memory_order_release); + + // Goodbye every connected peer and release unconnected nursery entries. The server stays + // up, so queued goodbye sends can flush and each connection leaves its slot through the + // normal close-event path; stop() later force-drops whatever remains. + // + // Quiescing: inbound dispatch is disabled on every connection before its goodbye goes out, so + // a peer told SHUTDOWN cannot push stream/role data into the client for the rest of the grace + // window. Without it a nursery peer's frames still reached the roles, and because a nursery + // entry is released without cleanup_connection_state(), that state outlived the teardown when + // no connection had been promoted. The plain disconnect() API keeps dispatch enabled, since a + // client that drops one session while running has no such window. + this->disconnect_impl(reason, /*quiescing=*/true); +} + +void ConnectionManager::stop(SendspinGoodbyeReason reason) { + // Same admission close + goodbye pass as begin_stop(); when stop() follows a begin_stop() + // this re-goodbyes only peers that ignored the first one for the whole grace window. + this->begin_stop(reason); + + // Stop accepting and tear down the transport server. Synchronous on both platforms: + // httpd_stop tears down every remaining session (ESP) and IXWebSocket's stop joins its + // worker threads (host). Close callbacks that fire during the teardown only queue pending + // events, which are discarded below. No manager lock is held here, so those callbacks + // cannot deadlock against this thread. + this->ws_server_.reset(); + this->ws_server_start_retry_time_us_ = 0; + + // Drop every remaining slot. The transports are gone (or their goodbye was already sent + // above), so no further goodbye is attempted. Dropping the current slot also quiesces the + // client's per-connection state (cleanup_connection_state), exactly as a connection-lost + // event would. + { + std::lock_guard lock(this->conn_ptr_mutex_); + this->drop_connection(this->current_connection_.get(), std::nullopt); + while (!this->nursery_.empty()) { + this->release_nursery_entry(this->nursery_.begin(), std::nullopt); + } + } + this->flush_deferred_releases(); + + // Discard pending lifecycle events: every connection they reference was just released, so + // draining them later would only no-op through drop_connection. Swapped out under the lock, + // destroyed outside it (a connection destructor can join its transport thread). + std::vector> stale_connected; + std::vector> stale_disconnects; + { + std::lock_guard lock(this->conn_mutex_); + stale_connected.swap(this->pending_connected_events_); + stale_disconnects.swap(this->pending_disconnect_events_); + this->has_pending_events_.store(false, std::memory_order_release); + } +} + void ConnectionManager::loop() { // 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()) { + // Gated on accepting_ so a stop in progress cannot belatedly open the server. + if (this->accepting_.load(std::memory_order_acquire) && this->ws_server_ != nullptr && + !this->ws_server_->is_started()) { 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()) { @@ -315,9 +385,14 @@ void ConnectionManager::loop() { // so arm its hello. (Inbound connections arrive already upgraded and armed theirs at // admission.) Guarded by nursery membership: a connection promoted or released by an // earlier event is skipped, and a duplicate event just re-arms the retry in place. - for (auto& conn : connected_events) { - if (this->find_in_nursery(conn.get()) != this->nursery_.end()) { - this->initiate_hello(conn.get()); + // Skipped while stopping, for the same reason as the hello retry scan below: a peer + // whose upgrade lands inside the goodbye window must not be handed a client/hello + // after it was already told SHUTDOWN. + if (this->accepting_.load(std::memory_order_acquire)) { + for (auto& conn : connected_events) { + if (this->find_in_nursery(conn.get()) != this->nursery_.end()) { + this->initiate_hello(conn.get()); + } } } @@ -329,44 +404,60 @@ void ConnectionManager::loop() { // of our client/hello is picked up on the tick after the send completes. A connection // leaves the nursery only here (handshake complete) or by being reaped, so the current // slot never holds a connection that has not proven itself. - for (auto it = this->nursery_.begin(); it != this->nursery_.end();) { - if (!it->conn->is_handshake_complete()) { - ++it; - continue; - } - auto conn = std::move(it->conn); - it = this->nursery_.erase(it); - this->nursery_size_.store(this->nursery_.size(), std::memory_order_release); - this->remove_hello_retry(conn.get()); - - if (this->current_connection_ == nullptr) { - this->set_current_connection(std::move(conn)); - } else if (this->should_switch_to_new_server(this->current_connection_.get(), - conn.get())) { - // Both sides of the comparison are established, so the PLAYBACK-reason and - // last-played-server preferences always ran on real data. No incumbent is - // ever evicted on timing alone. - SS_LOGI(TAG, "Handoff decision: switch to new server"); - this->drop_connection(this->current_connection_.get(), - SendspinGoodbyeReason::ANOTHER_SERVER); - this->set_current_connection(std::move(conn)); - } else { - SS_LOGI(TAG, "Handoff decision: keep current server"); - // Leaving management: block stale network-thread dispatch during the goodbye - // window (outgoing sends, including the goodbye itself, are unaffected). - conn->disable_message_dispatch(); - this->queue_deferred_release(std::move(conn), - SendspinGoodbyeReason::ANOTHER_SERVER); - continue; + // + // Skipped entirely while the manager is stopping (accepting_ cleared). begin_stop() + // disables message dispatch on every connected peer before its goodbye, so a + // server/hello arriving during the STOPPING window no longer flips + // is_handshake_complete(); this guard is the structural backstop, keeping promotion + // impossible while stopping rather than resting on that. Promoting a peer here would + // hand it a client/state message after it was already told SHUTDOWN. (It makes no + // difference to the teardown deadline either way: has_connections() counts nursery + // entries as well as the current slot, and both slots are released by the same close + // event or by stop()'s force-drain.) Leaving it in the nursery is safe: stop()'s + // force-drain releases it (no further goodbye needed, one was already sent), and + // accepting_ is set again by init_server() before any future promotion. + if (this->accepting_.load(std::memory_order_acquire)) { + for (auto it = this->nursery_.begin(); it != this->nursery_.end();) { + if (!it->conn->is_handshake_complete()) { + ++it; + continue; + } + auto conn = std::move(it->conn); + it = this->nursery_.erase(it); + this->nursery_size_.store(this->nursery_.size(), std::memory_order_release); + this->remove_hello_retry(conn.get()); + + if (this->current_connection_ == nullptr) { + this->set_current_connection(std::move(conn)); + } else if (this->should_switch_to_new_server(this->current_connection_.get(), + conn.get())) { + // Both sides of the comparison are established, so the PLAYBACK-reason and + // last-played-server preferences always ran on real data. No incumbent is + // ever evicted on timing alone. + SS_LOGI(TAG, "Handoff decision: switch to new server"); + this->drop_connection(this->current_connection_.get(), + SendspinGoodbyeReason::ANOTHER_SERVER); + this->set_current_connection(std::move(conn)); + } else { + SS_LOGI(TAG, "Handoff decision: keep current server"); + // Leaving management: block stale network-thread dispatch during the + // goodbye window (outgoing sends, including the goodbye itself, are + // unaffected). + conn->disable_message_dispatch(); + this->queue_deferred_release(std::move(conn), + SendspinGoodbyeReason::ANOTHER_SERVER); + continue; + } + + // Notify the client and publish state, only for the winner and never for a + // connection that is about to receive a goodbye. + this->client_->on_handshake_complete(this->current_connection_.get()); + + SS_LOGI(TAG, + "Connection handshake complete: server_id=%s, connection_reason=%s", + this->current_connection_->get_server_id().c_str(), + to_cstr(this->current_connection_->get_connection_reason())); } - - // Notify the client and publish state, only for the winner and never for a - // connection that is about to receive a goodbye. - this->client_->on_handshake_complete(this->current_connection_.get()); - - SS_LOGI(TAG, "Connection handshake complete: server_id=%s, connection_reason=%s", - this->current_connection_->get_server_id().c_str(), - to_cstr(this->current_connection_->get_connection_reason())); } } @@ -411,7 +502,16 @@ void ConnectionManager::loop() { // Check hello retry timers (one entry per managed connection, so a second connection // arriving mid-handshake cannot clobber the first connection's pending hello). - { + // + // Skipped entirely while the manager is stopping (accepting_ cleared), for the same + // reason the promotion scan above is: begin_stop() goodbyes every connected nursery peer + // but leaves it in the nursery with its retry entry intact (disconnect() only releases + // the unconnected entries). initiate_hello() never sends inline -- it arms an entry whose + // send happens on a later tick through this scan -- so without this guard a peer admitted + // in the tick before request_stop() would be sent a client/hello after it was already + // told SHUTDOWN. Nothing leaks by skipping: every path that erases from nursery_ calls + // remove_hello_retry(), and stop()'s force-drain releases whatever is still parked. + if (this->accepting_.load(std::memory_order_acquire)) { const int64_t now_us = platform_time_us(); for (auto it = this->hello_retries_.begin(); it != this->hello_retries_.end();) { HelloRetryState& retry = *it; @@ -543,22 +643,41 @@ void ConnectionManager::on_new_connection(std::shared_ptr lock(this->conn_ptr_mutex_); - // The newcomer has not completed the hello handshake, so it never touches the current - // slot; it enters the bounded nursery and is promoted only once it establishes. Only - // inbound entries count against the capacity (see NURSERY_CAPACITY). If the inbound slots - // are full, reject the newcomer: every occupant speaks WebSocket, so there is no safe - // eviction candidate. The goodbye reaches the peer because its session is already upgraded, - // provided the transport had a socket to accept it on (the NURSERY_CAPACITY + 2 budget). + // Reject peers delivered while the manager is stopping (between stop()'s goodbye + // snapshot and the server teardown): the nursery is about to be force-drained with no + // goodbye, so turning the peer away here, with one, is the only graceful exit left. + // + // Otherwise: the newcomer has not completed the hello handshake, so it never touches the + // current slot; it enters the bounded nursery and is promoted only once it establishes. + // Only inbound entries count against the capacity (see NURSERY_CAPACITY). If the inbound + // slots are full, reject the newcomer: every occupant speaks WebSocket, so there is no + // safe eviction candidate. Either goodbye reaches the peer because its session is already + // upgraded, provided the transport had a socket to accept it on (the NURSERY_CAPACITY + 2 + // budget). size_t inbound_count = 0; for (const auto& entry : this->nursery_) { if (entry.inbound) { ++inbound_count; } } - if (inbound_count >= NURSERY_CAPACITY) { - SS_LOGW(TAG, "Nursery full of live connections, rejecting new connection"); + if (!this->accepting_.load(std::memory_order_acquire)) { + SS_LOGD(TAG, "Manager stopping, rejecting new connection"); // Never managed, but its callbacks are already wired: block dispatch so it cannot // inject messages during the goodbye window. + // + // The release is immediate, so has_connections() never covers this peer. On ESP the + // goodbye is queued to an httpd worker, which means an asynchronous stop can report + // completion and tear the server down before that send runs, costing this one peer + // its goodbye. Tracking it to closure instead would mean admitting it to the nursery, + // which is bounded (loop() copies the nursery into a fixed array sized + // NURSERY_CAPACITY + 1) and whose entries are owned by the main loop, not this + // thread. Neither is worth breaking for a best-effort frame to a peer that arrived + // mid-teardown. + conn->disable_message_dispatch(); + this->queue_deferred_release(std::move(conn), SendspinGoodbyeReason::SHUTDOWN); + } else if (inbound_count >= NURSERY_CAPACITY) { + SS_LOGW(TAG, "Nursery full of live connections, rejecting new connection"); + // Never managed; same dispatch gating as the stopping rejection above. conn->disable_message_dispatch(); this->queue_deferred_release(std::move(conn), SendspinGoodbyeReason::ANOTHER_SERVER); } else { diff --git a/src/connection_manager.h b/src/connection_manager.h index c3c5366..7402c99 100644 --- a/src/connection_manager.h +++ b/src/connection_manager.h @@ -107,17 +107,18 @@ struct HelloRetryState { * 3. Call `loop()` periodically to drive connection state, process deferred events, and retry * hellos. * 4. Call `connect_to()` to initiate an outgoing client connection when needed. - * 5. Call `disconnect()` to gracefully close the active connection. + * 5. Call `disconnect()` to gracefully close the active connection, or `stop()` to also shut + * down the WebSocket server (call `init_server()` again before reuse). * * @code * ConnectionManager manager(client); - * manager.init_server(client, use_psram, priority); + * manager.init_server(client); * * while (running) { * manager.loop(); * } * - * manager.disconnect(SendspinGoodbyeReason::SHUTDOWN); + * manager.stop(SendspinGoodbyeReason::SHUTDOWN); * @endcode */ class ConnectionManager { @@ -136,9 +137,14 @@ class ConnectionManager { /// @brief Disconnects from the current server. /// /// Must be called from the main loop thread: conn->disconnect() runs outside conn_ptr_mutex_ - /// (it can block on the transport, and on host outbound it joins the transport thread), so + /// (it can block on the transport: outbound stops the transport synchronously on both + /// platforms, while inbound only starts an asynchronous close), so /// only the main loop's serialization keeps it from racing loop()'s reap/handoff release of /// the same connection into two concurrent transport stops. + /// + /// Inbound message dispatch stays enabled on the connections being goodbyed: a client that + /// drops one session while running has no teardown window for late frames to corrupt. + /// begin_stop() disables it instead (see disconnect_impl()). /// @param reason The goodbye reason to send before closing. void disconnect(SendspinGoodbyeReason reason); @@ -146,11 +152,33 @@ class ConnectionManager { // Server lifecycle // ======================================== - /// @brief Creates the WebSocket server and configures callbacks. Call once from start_server(). + /// @brief Creates the WebSocket server and configures callbacks. Called from + /// SendspinClient::start(); call again after stop() before reuse. /// Server configuration is read from client->config_. /// @param client The SendspinClient that owns this manager. void init_server(SendspinClient* client); + /// @brief Begins an asynchronous stop: stops admitting new peers and sends a goodbye to + /// every connected one, but leaves the WebSocket server up so queued goodbye sends can + /// still flush (the ESP transport sends via httpd workers). + /// + /// Must be called from the main loop thread (see disconnect()). Connections leave their + /// slots through the normal close-event path on subsequent loop() ticks; poll + /// has_connections() for completion and call stop() to finish the teardown. Calling stop() + /// after begin_stop() can re-send a goodbye to a peer that ignored the first one. + /// @param reason The goodbye reason to send before closing. + void begin_stop(SendspinGoodbyeReason reason); + + /// @brief Stops the manager: sends a goodbye to every connected peer, drops every managed + /// connection, and stops and destroys the WebSocket server so no new connections arrive. + /// + /// Must be called from the main loop thread (same transport-stop race rationale as + /// disconnect()). The goodbye is best-effort: on ESP the send is queued to an httpd worker + /// and the server stop can close the session first. Call init_server() again before reuse; + /// loop() stays safe to call while stopped (no server, no connections, nothing to drive). + /// @param reason The goodbye reason to send before closing. + void stop(SendspinGoodbyeReason reason); + /// @brief Drives connection state: starts server when network ready, processes lifecycle /// events, retries hello, calls loop() on active connections. /// @@ -171,6 +199,15 @@ class ConnectionManager { /// @return True if connected and handshake is complete, false otherwise. bool is_connected() const; + /// @brief Returns true while any managed connection exists (current slot or nursery). + /// Lock-free (reads the tick-gating hint atomics); used to poll an asynchronous stop for + /// completion. + /// @return True if the current slot is occupied or the nursery is non-empty. + bool has_connections() const { + return this->has_current_.load(std::memory_order_acquire) || + this->nursery_size_.load(std::memory_order_acquire) > 0; + } + /// @brief Returns the current active connection. Main-thread only. /// @return Pointer to the current connection, or nullptr if none. SendspinConnection* current() const { @@ -293,6 +330,14 @@ class ConnectionManager { // ======================================== // Connection lifecycle // ======================================== + /// @brief Goodbyes every connected connection and releases the unconnected nursery entries. + /// + /// Backs both disconnect() and begin_stop(); see disconnect() for the threading contract. + /// @param reason The goodbye reason to send before closing. + /// @param quiescing True to disable inbound message dispatch on each connection before its + /// goodbye, so a peer told SHUTDOWN cannot reach the roles for the rest of the teardown. + void disconnect_impl(SendspinGoodbyeReason reason, bool quiescing); + /// @brief Tears down a lost connection (current or nursery). Caller must hold conn_ptr_mutex_. /// @param conn The connection that was lost. void on_connection_lost(SendspinConnection* conn); @@ -363,6 +408,16 @@ class ConnectionManager { // 8-bit fields bool has_last_played_server_{false}; + // Atomic control state (authoritative: unlike the hint atomics below, there is no + // mutex-protected ground truth behind it) + + /// False while the manager is stopped. Cleared at the top of begin_stop()/stop(), before + /// the goodbye snapshot, and set again by init_server(). Closes the admission window during + /// shutdown: a peer the transport delivers between the goodbye snapshot and the server + /// teardown is rejected with a goodbye by on_new_connection() instead of entering the + /// nursery only to be force-dropped without one. Also gates loop()'s server-start block. + std::atomic accepting_{false}; + // Atomic fields (lock-free hints for loop() tick gating; ground truth remains the // mutex-protected containers/pointer above -- see the "Tick cost" note on loop()) diff --git a/src/host/client_connection.cpp b/src/host/client_connection.cpp index ee5503c..62c08c1 100644 --- a/src/host/client_connection.cpp +++ b/src/host/client_connection.cpp @@ -84,7 +84,14 @@ void SendspinClientConnection::disconnect(SendspinGoodbyeReason reason, return; } - // Send goodbye message then stop + // Send the goodbye, then stop the transport. This join is synchronous and blocks the caller, + // which is why request_stop() cannot promise a non-blocking teardown for an outbound + // connection (see SendspinClient::request_stop()). It must stay synchronous: the manager + // releases its last reference to the connection immediately after this returns + // (disconnect_and_release(), and stop()'s force-drop), so the transport thread has to be + // joined while a reference is still held. Deferring the join to the destructor instead lets + // the thread deliver its Close event into on_disconnected_cb() during that destructor, where + // shared_from_this() throws std::bad_weak_ptr on the already zero refcount and terminates. this->send_goodbye_reason(reason, [this, on_complete](bool /*success*/) { if (this->ws_) { this->ws_->stop(); diff --git a/src/player_role.cpp b/src/player_role.cpp index 8254a2e..8846412 100644 --- a/src/player_role.cpp +++ b/src/player_role.cpp @@ -175,20 +175,37 @@ void PlayerRole::Impl::attach_inbox(Inbox& inbox) { bool PlayerRole::Impl::start() { this->load_static_delay(); - if (!this->config.audio_formats.empty() && this->listener && - !this->sync_task->is_initialized()) { - if (!this->sync_task->init(this, this->client, this->config.audio_buffer_capacity)) { - SS_LOGE(TAG, "Failed to initialize sync task"); - return false; - } - if (!this->sync_task->start(this->config.psram_stack, this->config.priority)) { - SS_LOGE(TAG, "Failed to start sync task thread"); - return false; - } + if (this->config.audio_formats.empty() || !this->listener) { + return true; + } + + // Init once (queues, ring buffer), then start the thread; a restart after stop() skips the + // init and only re-creates the joined thread. + if (!this->sync_task->is_initialized() && + !this->sync_task->init(this, this->client, this->config.audio_buffer_capacity)) { + SS_LOGE(TAG, "Failed to initialize sync task"); + return false; + } + if (!this->sync_task->is_thread_running() && + !this->sync_task->start(this->config.psram_stack, this->config.priority)) { + SS_LOGE(TAG, "Failed to start sync task thread"); + return false; } return true; } +void PlayerRole::Impl::stop() const { + this->sync_task->stop(); +} + +void PlayerRole::Impl::request_stop() const { + this->sync_task->request_stop(); +} + +bool PlayerRole::Impl::has_stopped() const { + return this->sync_task->has_thread_exited(); +} + void PlayerRole::Impl::build_hello_fields(ClientHelloMessage& msg) { if (this->config.audio_formats.empty()) { return; diff --git a/src/player_role_impl.h b/src/player_role_impl.h index 494d705..871c5d6 100644 --- a/src/player_role_impl.h +++ b/src/player_role_impl.h @@ -64,6 +64,9 @@ struct PlayerRole::Impl { void attach_inbox(Inbox& inbox); bool start(); + void stop() const; + void request_stop() const; + bool has_stopped() const; void build_hello_fields(ClientHelloMessage& msg); void build_state_fields(ClientStateMessage& msg) const; void handle_binary(const uint8_t* data, size_t len) const; diff --git a/src/sync_task.cpp b/src/sync_task.cpp index 0b318b2..38d1d8c 100644 --- a/src/sync_task.cpp +++ b/src/sync_task.cpp @@ -96,7 +96,9 @@ bool SyncTask::init(PlayerRole::Impl* player_impl, SendspinClient* client, size_ this->player_impl_ = player_impl; this->client_ = client; - if (!this->event_flags_.create()) { + // Guarded so a retry after a partial failure (flags created, ring-buffer allocation failed) + // can call init() again without re-creating the flags. + if (!this->event_flags_.is_created() && !this->event_flags_.create()) { SS_LOGE(TAG, "Couldn't create event flags."); return false; } @@ -639,6 +641,8 @@ DecodeResult SyncTask::decode_chunk(SyncContext& sync_context) { bool SyncTask::wait_for_codec_header(SyncContext& sync_context) { // Wait for a codec header to arrive in the ring buffer, discarding stale audio chunks. // Uses a long timeout (500ms) so the task yields CPU and barely wakes when idle. + // Also the worst-case latency for the thread to observe a stop signal while idle: + // STOP_GRACE_MS in client.cpp budgets above this value; keep them in sync. static const uint32_t IDLE_RECEIVE_TIMEOUT_MS = 500; while ( @@ -784,6 +788,13 @@ void SyncTask::process_playback_progress(SyncContext& sync_context) { // Lifecycle // ============================================================================ +void SyncTask::request_stop() { + if (!this->sync_thread_.joinable()) { + return; + } + this->event_flags_.set(EventGroupBits::COMMAND_STOP); +} + void SyncTask::stop() { if (!this->sync_thread_.joinable()) { return; @@ -791,6 +802,14 @@ void SyncTask::stop() { this->event_flags_.set(EventGroupBits::COMMAND_STOP); this->sync_thread_.join(); + + // The ring buffer survives a stop()/start() cycle (init() runs once), so discard whatever a + // mid-stream stop left in it. Otherwise a restarted thread's wait_for_codec_header() could + // pick up a stale pre-stop codec header as the new stream's. Safe here: the consumer thread + // is joined, and the producers (network threads) are quiesced before role threads stop. + if (this->encoded_ring_buffer_ != nullptr) { + this->encoded_ring_buffer_->reset(); + } } // ============================================================================ @@ -930,6 +949,20 @@ void SyncTask::thread_entry(void* params) { // a codec header that arrived during a rapid seek (STREAM_END → STREAM_START). } + // Return any entry still borrowed at exit: the idle-path COMMAND_STOP breaks above can fire + // while a codec header is held awaiting COMMAND_START, and an un-returned borrow corrupts + // the ring's item accounting for the drain that stop() runs after the join. + if (sync_context.encoded_entry != nullptr) { + this_task->encoded_ring_buffer_->return_chunk(sync_context.encoded_entry); + sync_context.encoded_entry = nullptr; + } + + // Clear the state bits before announcing the stop: a COMMAND_STOP that interrupts an active + // stream breaks out of the inner loop above without passing the top-of-loop clear, so + // TASK_RUNNING would otherwise stay set after the thread is gone -- and is_running() reads + // that bit, which would wedge the player's sync-idle gate (a queued STREAM_END would never + // deliver its on_stream_end()). + this_task->event_flags_.clear(EventGroupBits::TASK_RUNNING | EventGroupBits::TASK_IDLE); this_task->event_flags_.set(EventGroupBits::TASK_STOPPED); } diff --git a/src/sync_task.h b/src/sync_task.h index 3f621b0..33a1a11 100644 --- a/src/sync_task.h +++ b/src/sync_task.h @@ -130,10 +130,23 @@ class SyncTask { /// @return true if thread started successfully, false otherwise. bool start(bool task_stack_in_psram, unsigned priority); + /// @brief Signals the task to stop and waits for the thread to finish + /// Restartable: start() may be called again afterwards. + void stop(); + + /// @brief Signals the task to stop without waiting for the thread to finish + /// Used by the client's asynchronous shutdown so the thread winds down concurrently with the + /// network grace period; the eventual stop() then joins an already-exited thread. Poll + /// has_thread_exited() for completion. No-op if the thread is not running. + void request_stop(); + /// @brief Returns true if init() has been called successfully + /// Checks every resource init() creates, not just the first: a partially failed init() + /// (flags created, ring-buffer allocation failed) reports false so the caller retries + /// init() instead of starting a thread that would dereference the missing ring buffer. /// @return true if the sync task has been initialized, false otherwise. bool is_initialized() const { - return this->event_flags_.is_created(); + return this->event_flags_.is_created() && this->encoded_ring_buffer_ != nullptr; } /// @brief Returns true if the sync task is actively processing a stream @@ -148,6 +161,25 @@ class SyncTask { return (this->event_flags_.get() & EventGroupBits::TASK_RUNNING) != 0U; } + /// @brief Returns true if the sync thread has finished, or never started + /// After request_stop(), true means a subsequent stop() joins without blocking. + /// @return true if no thread exists or the thread's entry function has returned. + bool has_thread_exited() const { + if (!this->sync_thread_.joinable()) { + return true; + } + return (this->event_flags_.get() & EventGroupBits::TASK_STOPPED) != 0U; + } + + /// @brief Returns true if the sync thread has been started and not yet stopped + /// Distinct from is_running(), which reports whether a stream is actively being processed. + /// Only meaningful on the thread that calls start()/stop(): joinable() reflects the + /// std::thread object's state, which only those calls mutate. + /// @return true if the sync thread exists (started, not yet joined by stop()). + bool is_thread_running() const { + return this->sync_thread_.joinable(); + } + /// @brief Signals the sync task to end the current stream. Non-blocking /// The task drains stale audio from the ring buffer and returns to idle. /// Thread-safe: may be called from any context. @@ -262,9 +294,6 @@ class SyncTask { /// playtime. void process_playback_progress(SyncContext& sync_context); - /// @brief Signals the task to stop and waits for the thread to finish - void stop(); - // Struct fields EventFlags event_flags_; // Latest-wins slot that merges (sum frames, keep latest finish_timestamp) diff --git a/src/visualizer_role.cpp b/src/visualizer_role.cpp index b77d947..f6006eb 100644 --- a/src/visualizer_role.cpp +++ b/src/visualizer_role.cpp @@ -55,6 +55,9 @@ static constexpr size_t BUFFER_ADVERTISE_DIVISOR = 3; static constexpr uint32_t COMMAND_STOP = (1 << 0); static constexpr uint32_t COMMAND_FLUSH = (1 << 1); // Drain to empty (producer already stopped) static constexpr uint32_t COMMAND_CLEAR = (1 << 2); // Discard up to the clear marker entry +// Set by the drain thread as it exits, so an asynchronous stop can poll has_stopped() and only +// join once the join is known to be instant (mirrors SyncTask's TASK_STOPPED). +static constexpr uint32_t THREAD_EXITED = (1 << 3); // Sentinel entry marking a stream/start or stream/clear boundary in the ring buffer. Entries // before the marker predate the boundary and are discarded; entries after it survive. The value @@ -173,6 +176,20 @@ bool VisualizerRole::Impl::start() { return false; } + // The flag object survives a stop()/start() cycle. This clear only matters on the async + // path: on a plain stop() the drain thread's own exiting wait() call already self-clears + // COMMAND_STOP (EventFlags::wait's clear_on_exit clears matched bits unconditionally, even + // ones already set before the call), so the bit already reads 0 here. On the async path + // request_stop() lets the thread exit (and self-clear) on its own, then a later stop() calls + // event_flags.set(COMMAND_STOP) on the now-dead thread with nothing left to clear it, so the + // bit is still set by the time start() runs. Clearing it (and any unconsumed flush/clear + // command) unconditionally here, rather than relying on which path preceded this start(), + // keeps a stale bit from making the new thread's first wait() see COMMAND_STOP and exit + // immediately. THREAD_EXITED is the previous thread's exit marker, stale for the new one + // either way. + this->drain_task->event_flags.clear(COMMAND_STOP | COMMAND_FLUSH | COMMAND_CLEAR | + THREAD_EXITED); + platform_configure_thread("SsVis", 4096, static_cast(this->config.priority), this->config.psram_stack); this->drain_task->drain_thread = std::thread(drain_thread_func, this); @@ -185,6 +202,30 @@ void VisualizerRole::Impl::stop() const { } this->drain_task->event_flags.set(COMMAND_STOP); this->drain_task->drain_thread.join(); + + // The ring buffer survives a stop()/start() cycle, so discard whatever is left in it -- + // including entries queued before this stop and anything the network thread pushes during the + // grace window after the drain thread has already exited (on the async request_stop() path the + // thread exits on COMMAND_STOP alone; cleanup()'s later COMMAND_FLUSH lands on a dead thread + // and never runs). Otherwise a restarted session would decode stale entries against the new + // session's spectrum_bin_count/tracks_downbeats. Safe here: the drain thread is already joined, + // so this call is the ring's only consumer (the single-consumer contract SpscRingBuffer + // requires). + this->flush_ring_buffer(); +} + +void VisualizerRole::Impl::request_stop() const { + if (!this->drain_task || !this->drain_task->drain_thread.joinable()) { + return; + } + this->drain_task->event_flags.set(COMMAND_STOP); +} + +bool VisualizerRole::Impl::has_stopped() const { + if (!this->drain_task || !this->drain_task->drain_thread.joinable()) { + return true; + } + return (this->drain_task->event_flags.get() & THREAD_EXITED) != 0U; } void VisualizerRole::Impl::build_hello_fields(ClientHelloMessage& msg) { @@ -619,6 +660,7 @@ void VisualizerRole::Impl::drain_thread_func(VisualizerRole::Impl* self) { } } + flags.set(THREAD_EXITED); SS_LOGD(TAG, "Drain thread stopped"); } diff --git a/src/visualizer_role_impl.h b/src/visualizer_role_impl.h index 56cad0c..ea70ecd 100644 --- a/src/visualizer_role_impl.h +++ b/src/visualizer_role_impl.h @@ -112,6 +112,8 @@ struct VisualizerRole::Impl { // ======================================== void stop() const; + void request_stop() const; + bool has_stopped() const; void flush_ring_buffer() const; void signal_clear_marker() const; void discard_to_clear_marker() const; diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index ac5113c..fc4751b 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -21,16 +21,29 @@ FetchContent_MakeAvailable(googletest) # registers the individual TEST() cases with CTest so failures are reported per case. add_executable(sendspin_tests test_audio_stream_info.cpp + test_client_internal.cpp test_connection_lifecycle.cpp + test_connection_manager.cpp test_time_filter.cpp test_protocol.cpp test_network_info.cpp test_spsc_ring_buffer.cpp test_inbox.cpp + test_sync_task.cpp test_visualizer_role.cpp test_artwork_role.cpp ) +# test_sync_task.cpp stages internal SyncTask states (partial init, mid-stream stop) directly; +# test_connection_manager.cpp reaches ConnectionManager's private nursery_/accepting_ state to +# reproduce STOPPING-window behavior not reliably reproducible through a real socket; +# test_client_internal.cpp stages the client's private session state and a role-start +# failure. Disable access control for these TUs instead of adding test seams to the production +# code. +set_source_files_properties(test_sync_task.cpp test_connection_manager.cpp + test_client_internal.cpp + PROPERTIES COMPILE_OPTIONS "-fno-access-control") + # Reach the library's private headers (protocol_messages.h, time_filter.h, ...). # The public include/ dir and ArduinoJson propagate transitively from `sendspin`. # Use CMAKE_CURRENT_SOURCE_DIR (not CMAKE_SOURCE_DIR) so the path stays correct even diff --git a/tests/test_artwork_role.cpp b/tests/test_artwork_role.cpp index c557f8c..6568775 100644 --- a/tests/test_artwork_role.cpp +++ b/tests/test_artwork_role.cpp @@ -955,3 +955,59 @@ TEST(ArtworkDisplayLateness, HugeLatenessSaturatesAtUint32Max) { // saturate there rather than wrap when narrowed to uint32_t. EXPECT_EQ(ArtworkRole::Impl::display_lateness_ms(1, INT64_MAX), UINT32_MAX); } + +// ============================================================================ +// Lifecycle: stop()/start() restart +// ============================================================================ + +// A stop()/start() cycle must yield a live decode thread again. This exercises the plain +// stop()/start() path, where the drain thread's own exiting wait() call already self-clears +// COMMAND_STOP, so start()'s clear of the surviving flag object is a no-op here; see +// RequestStopReportsExitAndRestarts below for the async path where that clear is load-bearing +// (stop() re-sets COMMAND_STOP on an already-dead thread, leaving it stale for the next start()). +TEST(ArtworkLifecycle, RestartDecodesAgain) { + auto impl = make_impl(make_single_slot_config(false)); + RecordingListener listener; + impl->listener = &listener; + ASSERT_TRUE(impl->start()); + impl->handle_stream_start(ServerArtworkStreamObject{}); + + send_frame(*impl, 0, 'A'); + ASSERT_TRUE(listener.wait_for([&] { return listener.decodes.size() >= 1; }, POSITIVE_TIMEOUT)); + EXPECT_EQ(listener.decode_marker_at(0), 'A'); + + impl->stop(); + ASSERT_TRUE(impl->start()); + impl->handle_stream_start(ServerArtworkStreamObject{}); + + send_frame(*impl, 0, 'B'); + ASSERT_TRUE(listener.wait_for([&] { return listener.decodes.size() >= 2; }, POSITIVE_TIMEOUT)); + EXPECT_EQ(listener.decode_marker_at(1), 'B'); +} + +// request_stop() signals the decode thread without joining it: the thread exits within its +// receive timeout and reports via has_stopped(), after which stop() joins instantly and a +// start() revives the role. This is the primitive behind the client's asynchronous +// request_stop(). +TEST(ArtworkLifecycle, RequestStopReportsExitAndRestarts) { + auto impl = make_impl(make_single_slot_config(false)); + RecordingListener listener; + impl->listener = &listener; + ASSERT_TRUE(impl->start()); + EXPECT_FALSE(impl->has_stopped()); + + impl->request_stop(); + const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(2); + while (!impl->has_stopped() && std::chrono::steady_clock::now() < deadline) { + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + EXPECT_TRUE(impl->has_stopped()); + impl->stop(); + + // The role revives: a fresh thread decodes again. + ASSERT_TRUE(impl->start()); + EXPECT_FALSE(impl->has_stopped()); + impl->handle_stream_start(ServerArtworkStreamObject{}); + send_frame(*impl, 0, 'R'); + ASSERT_TRUE(listener.wait_for([&] { return listener.decodes.size() >= 1; }, POSITIVE_TIMEOUT)); +} diff --git a/tests/test_client_internal.cpp b/tests/test_client_internal.cpp new file mode 100644 index 0000000..cbd368f --- /dev/null +++ b/tests/test_client_internal.cpp @@ -0,0 +1,133 @@ +// Copyright 2026 Sendspin Contributors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// White-box tests for the two SendspinClient lifecycle paths that no socket-level test can +// reach (compiled with -fno-access-control, see CMakeLists.txt, rather than adding test seams +// to the production code): +// +// - the session-state reset in finish_stop(). group_state_ is only ever filled from a server +// group/update and state_ has no public getter, so staging and reading them back needs +// private access. +// - start()'s rollback of already-started role threads when a later role fails to start. +// No role's start() can be made to fail through the public API on host (the failure paths +// are allocation and pthread-creation failures), so the failure is staged directly. + +#include "player_role_impl.h" +#include "sendspin/client.h" +#include "sendspin/config.h" +#include "sendspin/player_role.h" +#include "sendspin/visualizer_role.h" +#include "sync_task.h" +#include "visualizer_role_impl.h" + +#include + +#include +#include +#include + +using namespace sendspin; // NOLINT(google-build-using-namespace): test-local convenience + +namespace { + +SendspinClientConfig make_config() { + SendspinClientConfig config; + config.client_id = "client-lifecycle-internal-test"; + config.name = "Client Lifecycle Internal Test"; + return config; +} + +// Minimal player listener so a real player role (and thus a live sync task thread) can be +// attached without audio hardware: writes are accepted and discarded. +class NullPlayerListener : public PlayerRoleListener { +public: + size_t on_audio_write(uint8_t* /*data*/, size_t length, uint32_t /*timeout_ms*/) override { + return length; + } +}; + +} // namespace + +// Finding: finish_stop() must clear the session-scoped published state so a restarted client does +// not serve the previous session's group deltas or republish a stale ERROR client state to the +// next server on handshake. cleanup_connection_state() deliberately does not reset these (it also +// runs on reconnects and handoffs, where carrying them forward is intentional), so the stop path +// is the only place that does. +TEST(ClientLifecycleInternal, StopResetsPublishedSessionStateForRestart) { + SendspinClient client(make_config()); + ASSERT_TRUE(client.start()); + + // Stand in for a session that accumulated a group delta and then reported an error. + client.group_state_.group_id = "old-group"; + client.group_state_.group_name = "Old Group"; + client.update_state(SendspinClientState::ERROR); + + // Control: the state really is dirty going into the stop, so the assertions below cannot + // pass just because it was never set. + ASSERT_EQ(client.get_group_state().group_id, "old-group"); + ASSERT_EQ(client.get_group_state().group_name, "Old Group"); + ASSERT_EQ(client.state_, SendspinClientState::ERROR); + + client.stop(); + + EXPECT_FALSE(client.get_group_state().group_id.has_value()); + EXPECT_FALSE(client.get_group_state().group_name.has_value()); + EXPECT_FALSE(client.get_group_state().playback_state.has_value()); + EXPECT_EQ(client.state_, SendspinClientState::SYNCHRONIZED); + + // The restart begins from that clean state rather than re-deriving it on connect. + ASSERT_TRUE(client.start()); + EXPECT_FALSE(client.get_group_state().group_id.has_value()); + EXPECT_EQ(client.state_, SendspinClientState::SYNCHRONIZED); + + client.stop(); +} + +// Finding: when a role fails to start, start() must stop the role threads that already started +// before returning false, so the client is genuinely back in the stopped state and a corrected +// retry begins clean. Roles start in the order player, visualizer, artwork, so a sabotaged +// visualizer leaves the player's sync task thread as the one that must be rolled back. +TEST(ClientLifecycleInternal, StartRollsBackStartedRoleThreadsWhenALaterRoleFails) { + SendspinClient client(make_config()); + NullPlayerListener player_listener; + + PlayerRoleConfig player_config; + player_config.audio_formats = {{SendspinCodecFormat::PCM, 2, 44100, 16}}; + auto& player = client.add_player(std::move(player_config)); + player.set_listener(&player_listener); + + VisualizerRoleConfig visualizer_config; + visualizer_config.support.buffer_capacity = 4096; + client.add_visualizer(std::move(visualizer_config)); + + // Control: with both roles healthy, start() succeeds and the sync task thread is running. + // Without this the rollback assertion below would also pass if the thread had simply never + // started in the first place. + ASSERT_TRUE(client.start()); + ASSERT_TRUE(client.player_->impl_->sync_task->is_thread_running()); + client.stop(); + ASSERT_FALSE(client.player_->impl_->sync_task->is_thread_running()); + + // Sabotage the visualizer so its start() fails at the drain-task check, after the player's + // start() has already brought the sync task thread back up. + client.visualizer_->impl_->drain_task.reset(); + + EXPECT_FALSE(client.start()); + EXPECT_EQ(client.get_run_state(), SendspinRunState::STOPPED); + EXPECT_FALSE(client.is_started()); + + // The rollback joined the thread the player had already started. Without it the client + // reports itself stopped while a live sync task thread runs on. + EXPECT_FALSE(client.player_->impl_->sync_task->is_thread_running()); +} diff --git a/tests/test_connection_lifecycle.cpp b/tests/test_connection_lifecycle.cpp index be2bf82..a256ec3 100644 --- a/tests/test_connection_lifecycle.cpp +++ b/tests/test_connection_lifecycle.cpp @@ -22,6 +22,8 @@ #include "connection_manager.h" // fnv1_hash for the last-played preference #include "sendspin/client.h" #include "sendspin/config.h" +#include "sendspin/controller_role.h" +#include "sendspin/player_role.h" #include #include #include @@ -40,6 +42,7 @@ #include #include #include +#include using namespace sendspin; // NOLINT(google-build-using-namespace): test-local convenience @@ -56,6 +59,14 @@ constexpr uint16_t EVICT_TEST_PORT = 18972; constexpr uint16_t REJECT_TEST_PORT = 18973; constexpr uint16_t STALL_LISTEN_PORT = 18981; constexpr uint16_t ADMIT_TEST_PORT = 18982; +constexpr uint16_t STOP_TEST_PORT = 18991; +constexpr uint16_t STOP_NOOP_TEST_PORT = 18992; +constexpr uint16_t REQUEST_STOP_TEST_PORT = 18993; +constexpr uint16_t REQUEST_STOP_SYNC_TEST_PORT = 18994; +constexpr uint16_t REQUEST_STOP_PLAYER_TEST_PORT = 18995; +constexpr uint16_t REQUEST_STOP_LATE_PEER_TEST_PORT = 18996; +constexpr uint16_t STOP_ORDERING_TEST_PORT = 18997; +constexpr uint16_t RESTART_IN_CALLBACK_TEST_PORT = 18998; std::string server_url(uint16_t port) { return "ws://127.0.0.1:" + std::to_string(port) + "/sendspin"; @@ -598,3 +609,365 @@ TEST(ConnectionLifecycle, FullNurseryOfLivePeersRejectsNewcomer) { EXPECT_FALSE(mute_a.closed()); EXPECT_FALSE(mute_b.closed()); } + +// stop() is the full shutdown: the established peer receives a goodbye before its socket closes, +// the WS server stops listening (new connections are refused at the TCP level), and loop() stays +// safe to call while stopped. start() afterwards brings the whole stack back on the same port for +// a fresh establishment. +TEST(ConnectionLifecycle, StopShutsDownAndRestartAcceptsAgain) { + TestNetworkProvider network; + SendspinClient client(make_config(STOP_TEST_PORT)); + client.set_network_provider(&network); + ASSERT_TRUE(client.start()); + EXPECT_TRUE(client.is_started()); + pump_for(client, 50); + + { + FakeServer server(server_url(STOP_TEST_PORT), "server-before-stop"); + ASSERT_TRUE(pump_until( + client, [&] { return client.is_connected(); }, 4000)); + + client.stop(); + EXPECT_FALSE(client.is_started()); + EXPECT_FALSE(client.is_connected()); + + // Pumping while stopped is allowed and must not revive anything; the peer sees the + // goodbye and then the close. + EXPECT_TRUE(pump_until( + client, [&] { return server.closed(); }, 3000)); + EXPECT_TRUE(server.got_goodbye()); + } + + // The listener is gone: a raw TCP connect is refused outright. + int probe_fd = connect_loopback(STOP_TEST_PORT); + EXPECT_LT(probe_fd, 0); + if (probe_fd >= 0) { + ::close(probe_fd); + } + + // A second stop is a no-op: still stopped, nothing revived. + client.stop(); + EXPECT_EQ(client.get_run_state(), SendspinRunState::STOPPED); + + // Restart on the same port: a new peer establishes as if this were the first start. + ASSERT_TRUE(client.start()); + EXPECT_TRUE(client.is_started()); + pump_for(client, 50); + FakeServer server_after(server_url(STOP_TEST_PORT), "server-after-restart"); + EXPECT_TRUE(pump_until( + client, [&] { return client.is_connected(); }, 4000)); + auto info = client.get_server_information(); + ASSERT_TRUE(info.has_value()); + EXPECT_EQ(info->server_id, "server-after-restart"); + + client.stop(); +} + +// stop() before start(), and loop() while never started, must be safe no-ops. +TEST(ConnectionLifecycle, StopWithoutStartIsANoOp) { + SendspinClient client(make_config(STOP_NOOP_TEST_PORT)); + client.stop(); + client.loop(); + EXPECT_FALSE(client.is_started()); +} + +namespace { + +// Counts on_stopped() completions from request_stop() teardowns. +class StopListener : public SendspinClientListener { +public: + void on_stopped() override { + ++this->stopped_count; + } + + int stopped_count{0}; +}; + +// Restarts the client from inside on_stopped(), the pattern the header documents as supported. +// Only the first completion restarts, so the test's own final teardown is not undone. +class RestartOnStopListener : public SendspinClientListener { +public: + void on_stopped() override { + ++this->stopped_count; + if (this->client != nullptr && this->stopped_count == 1) { + this->state_seen_in_callback = this->client->get_run_state(); + this->restart_result = this->client->start(); + } + } + + SendspinClient* client{nullptr}; + int stopped_count{0}; + bool restart_result{false}; + SendspinRunState state_seen_in_callback{SendspinRunState::RUNNING}; +}; + +} // namespace + +// request_stop() must return without tearing anything down itself (state moves to STOPPING), +// then complete over loop() ticks: the established peer receives a goodbye before its socket +// closes, on_stopped() fires exactly once, the port stops accepting, and start() works again. +// start() during the teardown must be refused. +TEST(ConnectionLifecycle, RequestStopCompletesOverLoopAndNotifies) { + TestNetworkProvider network; + StopListener stop_listener; + SendspinClient client(make_config(REQUEST_STOP_TEST_PORT)); + client.set_network_provider(&network); + client.set_listener(&stop_listener); + ASSERT_TRUE(client.start()); + pump_for(client, 50); + + { + FakeServer server(server_url(REQUEST_STOP_TEST_PORT), "server-async-stop"); + ASSERT_TRUE(pump_until( + client, [&] { return client.is_connected(); }, 4000)); + + client.request_stop(); + EXPECT_EQ(client.get_run_state(), SendspinRunState::STOPPING); + EXPECT_FALSE(client.is_started()); + EXPECT_EQ(stop_listener.stopped_count, 0); + + // A start during the teardown is refused; a second request_stop is a no-op. + EXPECT_FALSE(client.start()); + client.request_stop(); + EXPECT_EQ(client.get_run_state(), SendspinRunState::STOPPING); + + ASSERT_TRUE(pump_until( + client, [&] { return client.get_run_state() == SendspinRunState::STOPPED; }, 3000)); + EXPECT_EQ(stop_listener.stopped_count, 1); + EXPECT_TRUE(pump_until( + client, [&] { return server.closed(); }, 2000)); + EXPECT_TRUE(server.got_goodbye()); + } + + // The listener is gone once the teardown completes. + int probe_fd = connect_loopback(REQUEST_STOP_TEST_PORT); + EXPECT_LT(probe_fd, 0); + if (probe_fd >= 0) { + ::close(probe_fd); + } + + // Restart works, and no further on_stopped() fires for it. + ASSERT_TRUE(client.start()); + pump_for(client, 50); + FakeServer server_after(server_url(REQUEST_STOP_TEST_PORT), "server-after-async"); + EXPECT_TRUE(pump_until( + client, [&] { return client.is_connected(); }, 4000)); + client.stop(); + EXPECT_EQ(stop_listener.stopped_count, 1); +} + +// A synchronous stop() while a request_stop() teardown is in progress finishes it immediately +// and still announces the completion exactly once. +TEST(ConnectionLifecycle, SyncStopFinishesPendingRequestStop) { + TestNetworkProvider network; + StopListener stop_listener; + SendspinClient client(make_config(REQUEST_STOP_SYNC_TEST_PORT)); + client.set_network_provider(&network); + client.set_listener(&stop_listener); + ASSERT_TRUE(client.start()); + pump_for(client, 50); + + client.request_stop(); + EXPECT_EQ(client.get_run_state(), SendspinRunState::STOPPING); + client.stop(); + EXPECT_EQ(client.get_run_state(), SendspinRunState::STOPPED); + EXPECT_EQ(stop_listener.stopped_count, 1); + + // With no connections and no player role, a fresh request_stop() completes on the next + // loop() tick, well inside the grace deadline. + ASSERT_TRUE(client.start()); + client.request_stop(); + EXPECT_TRUE(pump_until( + client, [&] { return client.get_run_state() == SendspinRunState::STOPPED; }, 1000)); + EXPECT_EQ(stop_listener.stopped_count, 2); + + // request_stop() from STOPPED is a no-op: no state change, no notification. + client.request_stop(); + client.loop(); + EXPECT_EQ(client.get_run_state(), SendspinRunState::STOPPED); + EXPECT_EQ(stop_listener.stopped_count, 2); +} + +namespace { + +// Minimal player listener so a lifecycle test can attach a real player role (and thus a live +// sync task thread) without any audio hardware: writes are accepted and discarded. +class NullPlayerListener : public PlayerRoleListener { +public: + size_t on_audio_write(uint8_t* /*data*/, size_t length, uint32_t /*timeout_ms*/) override { + return length; + } +}; + +} // namespace + +// Calling start() from inside on_stopped() is documented as supported, and it only works because +// finish_stop() assigns STOPPED before invoking the callback: start() refuses while STOPPING. +// That ordering is invisible to the callback-counting tests, which assert the end state after +// finish_stop() has fully returned and so hold no matter where inside it the write happens. This +// drives the restart from the callback itself, so moving the assignment below the notify turns +// the suite red. (Where tearing_down_ is cleared relative to the notify does not matter, since +// start() does not read it; that it is cleared at all is covered by the second teardown below.) +TEST(ConnectionLifecycle, StartFromStoppedCallbackRestartsClient) { + TestNetworkProvider network; + RestartOnStopListener stop_listener; + SendspinClient client(make_config(RESTART_IN_CALLBACK_TEST_PORT)); + stop_listener.client = &client; + client.set_network_provider(&network); + client.set_listener(&stop_listener); + ASSERT_TRUE(client.start()); + pump_for(client, 50); + + client.request_stop(); + ASSERT_TRUE(pump_until( + client, [&] { return stop_listener.stopped_count == 1; }, 3000)); + + // The callback saw STOPPED, not the STOPPING that start() refuses. + EXPECT_EQ(stop_listener.state_seen_in_callback, SendspinRunState::STOPPED); + EXPECT_TRUE(stop_listener.restart_result); + EXPECT_EQ(client.get_run_state(), SendspinRunState::RUNNING); + + // Restarted from inside the callback, the client is fully live: the server accepts a peer and + // the handshake completes. + pump_for(client, 50); + { + FakeServer server(server_url(RESTART_IN_CALLBACK_TEST_PORT), "server-restart-callback"); + EXPECT_TRUE(pump_until( + client, [&] { return client.is_connected(); }, 4000)); + } + + // A second teardown still completes and notifies. This is what catches a tearing_down_ that + // was never cleared: finish_stop() would return early and the state would never reach STOPPED. + client.request_stop(); + ASSERT_TRUE(pump_until( + client, [&] { return client.get_run_state() == SendspinRunState::STOPPED; }, 3000)); + EXPECT_EQ(stop_listener.stopped_count, 2); +} + +// With a player role attached, request_stop() must wait for the sync task thread to actually +// exit (has_stopped() gating in loop()) before completing, and a subsequent start() must bring +// the thread back. Covers the request-side of the role-thread stop path end to end. +TEST(ConnectionLifecycle, RequestStopWithPlayerRoleWaitsForSyncTask) { + TestNetworkProvider network; + StopListener stop_listener; + NullPlayerListener player_listener; + SendspinClient client(make_config(REQUEST_STOP_PLAYER_TEST_PORT)); + client.set_network_provider(&network); + client.set_listener(&stop_listener); + PlayerRoleConfig player_config; + player_config.audio_formats = {{SendspinCodecFormat::PCM, 2, 44100, 16}}; + auto& player = client.add_player(std::move(player_config)); + player.set_listener(&player_listener); + + ASSERT_TRUE(client.start()); + // Let the sync thread reach its idle receive before signaling, so the stop lands while it is + // parked and the thread is guaranteed not to exit for roughly one idle poll. + pump_for(client, 100); + client.request_stop(); + EXPECT_EQ(client.get_run_state(), SendspinRunState::STOPPING); + + // request_stop() is asynchronous, so this tick must hand the teardown to a later one rather + // than finishing it here: the sync thread is parked in its idle receive and cannot have + // exited yet, so the client is still STOPPING when loop() returns. + client.loop(); + EXPECT_EQ(client.get_run_state(), SendspinRunState::STOPPING); + EXPECT_EQ(stop_listener.stopped_count, 0); + + // Completion waits out the sync task's idle poll; well inside the grace deadline. + EXPECT_TRUE(pump_until( + client, [&] { return client.get_run_state() == SendspinRunState::STOPPED; }, 2000)); + EXPECT_EQ(stop_listener.stopped_count, 1); + + // The sync task thread restarts with the client. + ASSERT_TRUE(client.start()); + client.stop(); + EXPECT_EQ(stop_listener.stopped_count, 1); +} + +// A peer that connects during the STOPPING window (goodbyes sent, server still up so they can +// flush) must be rejected with a goodbye by the admission gate instead of entering the nursery +// only to be force-dropped without one. The rejection happens entirely on the transport thread, +// so no loop() pumping is needed for it. +TEST(ConnectionLifecycle, RequestStopRejectsLateArrivalWithGoodbye) { + TestNetworkProvider network; + StopListener stop_listener; + SendspinClient client(make_config(REQUEST_STOP_LATE_PEER_TEST_PORT)); + client.set_network_provider(&network); + client.set_listener(&stop_listener); + ASSERT_TRUE(client.start()); + pump_for(client, 50); // opens the WS server + + client.request_stop(); + + // No pumping here: the client stays in STOPPING (completion needs a loop() tick), so the + // still-listening server delivers the late peer to the admission gate. + FakeServer late(server_url(REQUEST_STOP_LATE_PEER_TEST_PORT), "server-late-arrival"); + const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(3); + while (!late.closed() && std::chrono::steady_clock::now() < deadline) { + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + } + EXPECT_TRUE(late.closed()); + EXPECT_TRUE(late.got_goodbye()); + EXPECT_FALSE(late.got_client_hello()); + + EXPECT_TRUE(pump_until( + client, [&] { return client.get_run_state() == SendspinRunState::STOPPED; }, 2000)); + EXPECT_EQ(stop_listener.stopped_count, 1); +} + +namespace { + +// Records the relative order of the controller's clear callback (queued by +// cleanup_connection_state() during teardown) against on_stopped(), so the test can assert one +// happens strictly before the other rather than merely that both eventually fire. +class OrderingListener : public SendspinClientListener, public ControllerRoleListener { +public: + void on_controller_state_clear() override { + this->order.push_back("controller_cleared"); + } + + void on_stopped() override { + this->order.push_back("on_stopped"); + } + + std::vector order; +}; + +} // namespace + +// on_stopped() is documented as the signal that every role's teardown has already been delivered +// and it is safe to destroy the client. The clear callbacks that the same teardown queues +// (cleanup_connection_state(), reached from finish_stop()'s connection_manager_->stop()) must +// therefore reach the listener before on_stopped() does, not on some later loop() tick. +TEST(ConnectionLifecycle, StoppedFiresAfterRoleClearCallbacks) { + TestNetworkProvider network; + OrderingListener listener; + SendspinClient client(make_config(STOP_ORDERING_TEST_PORT)); + client.set_network_provider(&network); + client.set_listener(&listener); + auto& controller = client.add_controller(); + controller.set_listener(&listener); + + ASSERT_TRUE(client.start()); + pump_for(client, 50); + + FakeServer server(server_url(STOP_ORDERING_TEST_PORT), "server-ordering"); + ASSERT_TRUE(pump_until( + client, [&] { return client.is_connected(); }, 4000)); + + // No loop() pumping between request_stop() and stop(): the close event that request_stop()'s + // goodbye eventually triggers has no chance to be processed first, so stop()'s teardown + // (finish_stop()) is guaranteed to run cleanup_connection_state() itself (current_connection_ + // is still set going in). That exercises the same-tick ordering finish_stop() must get right; + // pumping loop() until STOPPED instead would let an earlier tick's close-event handling run + // cleanup_connection_state() (and thus drain the clear via that tick's own drain_inbox_events() + // call) well before finish_stop() ever runs, which is already safe and would not catch a + // regression here. + client.request_stop(); + client.stop(); + + ASSERT_EQ(client.get_run_state(), SendspinRunState::STOPPED); + ASSERT_EQ(listener.order.size(), 2U); + EXPECT_EQ(listener.order[0], "controller_cleared"); + EXPECT_EQ(listener.order[1], "on_stopped"); +} diff --git a/tests/test_connection_manager.cpp b/tests/test_connection_manager.cpp new file mode 100644 index 0000000..1a9143a --- /dev/null +++ b/tests/test_connection_manager.cpp @@ -0,0 +1,285 @@ +// Copyright 2026 Sendspin Contributors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// White-box tests for what ConnectionManager::loop() must not do while the manager is stopping, +// plus the client-level stop deadline that depends on a peer the manager never releases. All of +// it reaches private state directly (compiled with -fno-access-control, see CMakeLists.txt) +// rather than going through a real socket, because none of these preconditions are reproducible +// through a real WebSocket transport on host: IXWebSocket's server-side disconnect is +// synchronous, so is_connected() flips false in the same call as the goodbye send and the close +// event lands immediately. That closes the window where a nursery peer is still connected with +// its dispatch enabled after begin_stop(), which is exactly the state the accepting_ gates on +// the promotion scan and the two hello sites exist for, and it makes a peer that never delivers +// a close (the only way to reach the grace deadline) impossible to stage. Driving the private +// nursery_/accepting_/hello_retries_/pending_connected_events_ state directly exercises those +// guarded paths under their exact preconditions without depending on transport timing. + +#include "connection.h" +#include "connection_manager.h" +#include "platform/time.h" +#include "sendspin/client.h" + +#include + +#include +#include +#include +#include +#include + +using namespace sendspin; // NOLINT(google-build-using-namespace): test-local convenience + +namespace { + +/// Minimal concrete SendspinConnection: no real transport, just enough state to drive the +/// handshake-complete and connected flags the promotion scan reads. +class FakeConnection : public SendspinConnection { +public: + void start() override {} + void loop() override {} + + void disconnect(SendspinGoodbyeReason /*reason*/, std::function on_complete) override { + this->connected_.store(false); + if (on_complete) { + on_complete(); + } + } + + bool is_connected() const override { + return this->connected_.load(); + } + + SsErr send_text_message(const std::string& /*message*/, SendCompleteCallback cb, + bool /*allow_before_hello*/) override { + ++this->send_count_; + if (cb) { + cb(true); + } + return SsErr::OK; + } + + bool send_time_message() override { + return true; + } + + void set_connected(bool connected) { + this->connected_.store(connected); + } + + int send_count() const { + return this->send_count_; + } + +private: + std::atomic connected_{false}; + int send_count_{0}; +}; + +SendspinClientConfig make_config() { + SendspinClientConfig config; + config.client_id = "conn-manager-test-client"; + config.name = "Connection Manager Test Client"; + return config; +} + +} // namespace + +// Finding: the promotion scan in ConnectionManager::loop() must not promote a nursery peer whose +// handshake completes while the manager is stopping (accepting_ cleared by begin_stop()). A +// begin_stop() disables dispatch on connected peers, so an in-flight server/hello no longer flips +// is_handshake_complete() by itself, but this guard is the structural backstop and is asserted +// here on its own: the scan is driven with dispatch left enabled (accepting_ cleared directly, +// as a stopping manager would have it) so the skip is attributable to the guard alone. Promoting +// a peer here would hand it a client/state message after it was already told SHUTDOWN. (Parking +// it costs nothing: has_connections() counts nursery entries too, so the teardown deadline +// behaves the same either way.) +TEST(ConnectionManagerInternal, PromotionSkippedWhileStopping) { + SendspinClientConfig config = make_config(); + SendspinClient client(config); + // Private member access via -fno-access-control: exercises the manager owned by a real + // client rather than standing up a duplicate one. + ConnectionManager& manager = *client.connection_manager_; + + auto fake = std::make_shared(); + fake->set_connected(true); + fake->set_client_hello_sent(true); + fake->set_server_hello_received(true); + ASSERT_TRUE(fake->is_handshake_complete()); + // Fresh timestamp so the nursery establish-deadline reap (30 s) does not fire first and mask + // whether the promotion scan itself was skipped. + fake->set_provisional_time_us(platform_time_us()); + + { + std::lock_guard lock(manager.conn_ptr_mutex_); + manager.push_nursery_entry(NurseryEntry{fake, /*inbound=*/true}); + } + + // Simulate begin_stop() having already cleared the admission door. + manager.accepting_.store(false, std::memory_order_release); + + manager.loop(); + + // Not promoted: the current slot stays empty and the entry is neither sent a client/state + // message nor released out of the nursery. + EXPECT_EQ(manager.current(), nullptr); + EXPECT_TRUE(manager.has_connections()); + EXPECT_EQ(fake->send_count(), 0); + { + std::lock_guard lock(manager.conn_ptr_mutex_); + ASSERT_EQ(manager.nursery_.size(), 1U); + EXPECT_EQ(manager.nursery_.front().conn.get(), fake.get()); + } + + // Once accepting_ is restored (as init_server() does for a fresh start()), the same + // already-established entry is free to promote on the very next tick: the fix leaves it + // parked, not permanently stranded. + manager.accepting_.store(true, std::memory_order_release); + manager.loop(); + + EXPECT_EQ(manager.current(), fake.get()); + EXPECT_EQ(fake->send_count(), 1); +} + +// Finding: the hello scans in ConnectionManager::loop() must not send a client/hello to a peer +// that begin_stop() has already goodbyed. initiate_hello() never sends inline -- it arms a retry +// entry whose send happens on a later tick through the retry scan -- and disconnect() leaves a +// connected nursery peer parked with its entry intact, so without the accepting_ guard a peer +// admitted in the tick before request_stop() is handed a hello after its SHUTDOWN goodbye. +TEST(ConnectionManagerInternal, HelloNotSentWhileStopping) { + SendspinClientConfig config = make_config(); + SendspinClient client(config); + ConnectionManager& manager = *client.connection_manager_; + + auto fake = std::make_shared(); + fake->set_connected(true); + // Fresh timestamp so the nursery establish-deadline reap (30 s) cannot release the entry and + // mask whether the hello scan itself was skipped. + fake->set_provisional_time_us(platform_time_us()); + ASSERT_FALSE(fake->is_handshake_complete()); + + { + std::lock_guard lock(manager.conn_ptr_mutex_); + manager.push_nursery_entry(NurseryEntry{fake, /*inbound=*/true}); + // Exactly what on_new_connection() does at admission: arm the hello, send nothing yet. + manager.initiate_hello(fake.get()); + } + ASSERT_EQ(fake->send_count(), 0); + + // Simulate begin_stop() having already cleared the admission door and goodbyed this peer. + manager.accepting_.store(false, std::memory_order_release); + + manager.loop(); + + EXPECT_EQ(fake->send_count(), 0); + EXPECT_TRUE(manager.has_connections()); + + // Control: the same armed entry sends its hello on the very next tick once accepting_ is + // restored, so the guard defers the send rather than silently dropping the handshake. + manager.accepting_.store(true, std::memory_order_release); + manager.loop(); + + EXPECT_EQ(fake->send_count(), 1); +} + +// Finding: loop()'s STOPPING completion gate must finish the teardown at the grace deadline even +// when a peer never delivers its close event. Without the deadline disjunct, has_connections() +// stays true forever and the client is stranded in STOPPING with on_stopped() never firing. +// Driven white-box because a real host socket cannot reproduce it: IXWebSocket's server-side +// disconnect is synchronous, so a real peer's close event always lands immediately. +TEST(ConnectionManagerInternal, StopDeadlineFinishesTeardownWhenPeerNeverCloses) { + SendspinClientConfig config = make_config(); + SendspinClient client(config); + ConnectionManager& manager = *client.connection_manager_; + + ASSERT_TRUE(client.start()); + + // A peer that is connected when begin_stop() runs is goodbyed and left parked in the nursery + // until its close event arrives. This one never delivers one. + auto fake = std::make_shared(); + fake->set_connected(true); + fake->set_provisional_time_us(platform_time_us()); + { + std::lock_guard lock(manager.conn_ptr_mutex_); + manager.push_nursery_entry(NurseryEntry{fake, /*inbound=*/true}); + } + + client.request_stop(); + ASSERT_EQ(client.get_run_state(), SendspinRunState::STOPPING); + ASSERT_TRUE(manager.has_connections()); + + // Control: pumping inside the grace window must not complete the teardown. Without this the + // assertion below would also pass if the has_connections() gate had finished it early, and + // the deadline would still be untested. + for (int i = 0; i < 20; ++i) { + client.loop(); + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + } + EXPECT_EQ(client.get_run_state(), SendspinRunState::STOPPING); + EXPECT_TRUE(manager.has_connections()); + + // Past STOP_GRACE_MS (750 ms, client.cpp) the deadline forces completion regardless. + std::this_thread::sleep_for(std::chrono::milliseconds(800)); + client.loop(); + + EXPECT_EQ(client.get_run_state(), SendspinRunState::STOPPED); + EXPECT_FALSE(manager.has_connections()); +} + +// Companion to the case above for the other hello site: an outbound connection arms its hello +// from the transport's connected event rather than at admission, so that arm needs the same +// accepting_ gate. Without it a connect_to() whose upgrade lands inside the goodbye window is +// armed during STOPPING and sent a client/hello on the same tick. +TEST(ConnectionManagerInternal, ConnectedEventDoesNotArmHelloWhileStopping) { + SendspinClientConfig config = make_config(); + SendspinClient client(config); + ConnectionManager& manager = *client.connection_manager_; + + auto fake = std::make_shared(); + fake->set_connected(true); + fake->mark_ws_upgraded(); + fake->set_provisional_time_us(platform_time_us()); + { + std::lock_guard lock(manager.conn_ptr_mutex_); + manager.push_nursery_entry(NurseryEntry{fake, /*inbound=*/false}); + } + { + std::lock_guard lock(manager.conn_mutex_); + manager.queue_pending_connected(fake); + } + + manager.accepting_.store(false, std::memory_order_release); + + manager.loop(); + + // Nothing sent, and nothing armed either, so no later tick can send one on its behalf. + EXPECT_EQ(fake->send_count(), 0); + { + std::lock_guard lock(manager.conn_ptr_mutex_); + EXPECT_TRUE(manager.hello_retries_.empty()); + } + manager.loop(); + EXPECT_EQ(fake->send_count(), 0); + + // Control: the same event arms and sends on one tick once the manager is accepting, so the + // guard is what suppressed it rather than the entry being unreachable from this path. + { + std::lock_guard lock(manager.conn_mutex_); + manager.queue_pending_connected(fake); + } + manager.accepting_.store(true, std::memory_order_release); + + manager.loop(); + + EXPECT_EQ(fake->send_count(), 1); +} diff --git a/tests/test_sync_task.cpp b/tests/test_sync_task.cpp new file mode 100644 index 0000000..6438677 --- /dev/null +++ b/tests/test_sync_task.cpp @@ -0,0 +1,155 @@ +// Copyright 2026 Sendspin Contributors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Lifecycle regression tests for SyncTask around the client's stop()/start() restart support. +// This translation unit is compiled with -fno-access-control (see tests/CMakeLists.txt) so the +// tests can stage the exact internal states the regressions arise from (a partially failed +// init(), a stop() landing mid-stream) without adding test seams to the production code. + +#include "audio_types.h" +#include "inbox.h" +#include "player_role_impl.h" +#include "sendspin/client.h" +#include "sendspin/config.h" +#include "sync_task.h" +#include + +#include +#include +#include +#include +#include + +using namespace sendspin; + +namespace { + +constexpr size_t RING_BUFFER_BYTES = 16 * 1024; + +// A real, never-started SendspinClient plus a bound PlayerRole::Impl for SyncTask::init() to +// point at. The sync thread only dereferences them once a stream goes active; these tests keep +// the thread idle, so inert instances suffice. Static deques give them program lifetime and +// stable addresses (mirroring make_impl() in test_artwork_role.cpp). +PlayerRole::Impl* make_player_impl() { + static std::deque clients; + static std::deque inboxes; + static std::deque> impls; + + clients.emplace_back(SendspinClientConfig{}); + impls.emplace_back( + std::make_unique(PlayerRoleConfig{}, &clients.back(), nullptr)); + inboxes.emplace_back(); + impls.back()->attach_inbox(inboxes.back()); + return impls.back().get(); +} + +} // namespace + +// A partially failed init() (event flags created, ring-buffer allocation failed) must not report +// initialized: PlayerRole::Impl::start() gates init() on !is_initialized(), so a stale true +// would make a retried start() skip init() and spawn a thread that dereferences the missing +// ring buffer. +TEST(SyncTaskLifecycle, PartialInitDoesNotReportInitialized) { + auto* player = make_player_impl(); + SyncTask task; + + // Stage the post-partial-failure state directly: flags exist, ring buffer does not. + ASSERT_TRUE(task.event_flags_.create()); + EXPECT_FALSE(task.is_initialized()); + + // A retried init() succeeds from that state, and the thread starts and stops cleanly. + ASSERT_TRUE(task.init(player, player->client, RING_BUFFER_BYTES)); + EXPECT_TRUE(task.is_initialized()); + ASSERT_TRUE(task.start(false, 0)); + EXPECT_TRUE(task.is_thread_running()); + task.stop(); + EXPECT_FALSE(task.is_thread_running()); +} + +// stop() must leave is_running() false even when it interrupts an active stream: the thread's +// COMMAND_STOP exit path skips the idle-state flag clear, so without an exit-path clear a stale +// TASK_RUNNING would wedge the player's sync-idle gate (a queued STREAM_END would never deliver +// its on_stream_end()). +TEST(SyncTaskLifecycle, IsRunningFalseAfterStopDuringActiveStream) { + auto* player = make_player_impl(); + SyncTask task; + ASSERT_TRUE(task.init(player, player->client, RING_BUFFER_BYTES)); + ASSERT_TRUE(task.start(false, 0)); + + // start() returns only once the thread has passed the idle-state flag clear, so a bit set + // here stays set until the thread exits -- the state a stop() landing mid-stream sees. + task.event_flags_.set(EventGroupBits::TASK_RUNNING); + ASSERT_TRUE(task.is_running()); + + task.stop(); + EXPECT_FALSE(task.is_running()); +} + +// stop() discards whatever an interrupted session left in the ring buffer: the buffer survives +// a stop()/start() cycle (init() runs once), and a stale pre-stop codec header would otherwise +// be picked up by the restarted thread as the new stream's. +TEST(SyncTaskLifecycle, StopDrainsRingBuffer) { + auto* player = make_player_impl(); + SyncTask task; + ASSERT_TRUE(task.init(player, player->client, RING_BUFFER_BYTES)); + + // Captured before start(), when nothing can be borrowed. chunks_waiting() alone cannot + // detect a borrowed-but-never-returned entry (items leave that count at receive time, not + // return time), so the free-byte check below is what catches a leaked borrow. + const size_t initial_free_bytes = task.encoded_ring_buffer_->ring_buffer_.free_bytes_; + + ASSERT_TRUE(task.start(false, 0)); + + // Queue header + audio + header, as a rapid seek interrupted by stop() would. The idle + // thread may receive (and hold) the first header while waiting for the stream start that + // never comes; everything behind it stays queued. + const uint8_t bytes[4] = {1, 2, 3, 4}; + ASSERT_TRUE(task.encoded_ring_buffer_->write_chunk(bytes, sizeof(bytes), 0, + CHUNK_TYPE_PCM_DUMMY_HEADER, 100)); + ASSERT_TRUE(task.encoded_ring_buffer_->write_chunk(bytes, sizeof(bytes), 0, + CHUNK_TYPE_ENCODED_AUDIO, 100)); + ASSERT_TRUE(task.encoded_ring_buffer_->write_chunk(bytes, sizeof(bytes), 0, + CHUNK_TYPE_PCM_DUMMY_HEADER, 100)); + + task.stop(); + EXPECT_EQ(task.encoded_ring_buffer_->chunks_waiting(), 0U); + // Every entry, including one the thread had borrowed at stop time, was returned: the ring's + // full capacity is available to the next session. + EXPECT_EQ(task.encoded_ring_buffer_->ring_buffer_.free_bytes_, initial_free_bytes); +} + +// request_stop() must signal the thread without joining it: the thread exits on its own within +// its idle poll interval and reports via has_thread_exited(), after which stop() joins without +// blocking. This is the primitive the client's asynchronous request_stop() is built on. +TEST(SyncTaskLifecycle, RequestStopSignalsWithoutJoin) { + auto* player = make_player_impl(); + SyncTask task; + ASSERT_TRUE(task.init(player, player->client, RING_BUFFER_BYTES)); + ASSERT_TRUE(task.start(false, 0)); + EXPECT_FALSE(task.has_thread_exited()); + + task.request_stop(); + // The thread object is intentionally not joined by request_stop(). + EXPECT_TRUE(task.is_thread_running()); + + // The idle poll observes the signal within IDLE_RECEIVE_TIMEOUT_MS (500 ms); allow margin. + const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(2); + while (!task.has_thread_exited() && std::chrono::steady_clock::now() < deadline) { + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + EXPECT_TRUE(task.has_thread_exited()); + + task.stop(); + EXPECT_FALSE(task.is_thread_running()); +} diff --git a/tests/test_visualizer_role.cpp b/tests/test_visualizer_role.cpp index bd6390a..fdd0277 100644 --- a/tests/test_visualizer_role.cpp +++ b/tests/test_visualizer_role.cpp @@ -13,14 +13,18 @@ // limitations under the License. #include "protocol_messages.h" +#include "sendspin/client.h" +#include "sendspin/config.h" #include "visualizer_role_impl.h" #include #include +#include #include #include #include +#include #include #include @@ -225,7 +229,9 @@ std::unique_ptr make_impl() { return impl; } -// Pops one entry from the ring buffer, or returns false if none is waiting. +// Pops one entry from the ring buffer, or returns false if none is waiting. Not safe to call +// while the drain thread may still be running: receive()/return_item() assume exactly one +// consumer (SPSC), and the drain thread is the other one. Callers must stop() first. bool pop_entry(VisualizerRole::Impl& impl, std::vector& out) { size_t size = 0; void* item = impl.drain_task->ring_buffer.receive(&size, 0); @@ -238,6 +244,21 @@ bool pop_entry(VisualizerRole::Impl& impl, std::vector& out) { return true; } +// Polls the ring's occupancy count -- not its content -- until it drains or the deadline elapses. +// is_empty() locks the same mutex as receive()/return_item() for a plain size read, so unlike +// pop_entry() above it is safe to call while the drain thread is live: it is a size query, not a +// second consumer. +bool wait_for_ring_empty(VisualizerRole::Impl& impl, std::chrono::milliseconds timeout) { + const auto deadline = std::chrono::steady_clock::now() + timeout; + do { + if (impl.drain_task->ring_buffer.is_empty()) { + return true; + } + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } while (std::chrono::steady_clock::now() < deadline); + return impl.drain_task->ring_buffer.is_empty(); +} + } // namespace TEST(VisualizerHandleBinary, ForwardsMessageVerbatim) { @@ -404,3 +425,113 @@ TEST(VisualizerClearMarker, DiscardDrainsToEmptyWithoutMarker) { std::vector entry; EXPECT_FALSE(pop_entry(*impl, entry)); } + +// ============================================================================ +// Lifecycle: stop()/start() restart +// ============================================================================ + +namespace { + +// A visualizer Impl bound to a real, never-started SendspinClient, for tests that run the live +// drain thread: the thread dereferences client (is_time_synced()/get_client_time()), so the +// nullptr client the decode-only make_impl() passes would crash it. A never-started client +// reports no time sync, so the drain thread consumes each entry and drops it unread -- which is +// exactly the observable these tests use for thread liveness. +std::unique_ptr make_impl_with_client() { + static std::deque clients; + static std::deque inboxes; + + VisualizerRoleConfig config; + config.support.buffer_capacity = 4096; + clients.emplace_back(SendspinClientConfig{}); + auto impl = std::make_unique(std::move(config), &clients.back()); + inboxes.emplace_back(); + impl->attach_inbox(inboxes.back()); + impl->stream_active = true; + impl->negotiated_types_mask = 0x1F; + return impl; +} + +} // namespace + +// A stop()/start() cycle must yield a live drain thread again. This exercises the plain +// stop()/start() path, where the drain thread's own exiting wait() call already self-clears +// COMMAND_STOP, so start()'s clear of the surviving flag object is a no-op here; see +// RequestStopReportsExitAndRestarts below for the async path where that clear is load-bearing +// (stop() re-sets COMMAND_STOP on an already-dead thread, leaving it stale for the next start()). +// Liveness is observed through consumption: the restarted thread drains the entry (dropping it +// for lack of time sync), so the ring empties on its own; a dead thread leaves the entry in place +// forever, which wait_for_ring_empty()'s deadline turns into a bounded failure instead of a hang. +TEST(VisualizerLifecycle, RestartDrainsAgain) { + auto impl = make_impl_with_client(); + ASSERT_TRUE(impl->start()); + impl->stop(); + ASSERT_TRUE(impl->start()); + + std::vector data; + put_be64(data, 123456); + put_be16(data, 0xABCD); + impl->handle_binary(SENDSPIN_BINARY_VISUALIZER_LOUDNESS, data.data(), data.size()); + + // Poll occupancy (safe with the drain thread live) rather than sleeping a fixed guess past + // its 50 ms receive timeout; stop() below then makes it safe for the test thread to pop_entry() + // as the sole consumer. + ASSERT_TRUE(wait_for_ring_empty(*impl, std::chrono::seconds(2))); + impl->stop(); + + std::vector entry; + EXPECT_FALSE(pop_entry(*impl, entry)); +} + +// An entry queued into the ring before stop() must not survive into a restarted session, where +// it would be decoded against the new session's spectrum_bin_count/tracks_downbeats. Forces the +// "already dead, not yet flushed" window precisely: request_stop() lets the drain thread exit +// (self-reported via has_stopped()) without stop() ever running, so the entry pushed right after +// is guaranteed to still be sitting in the ring, with no consumer thread live to race it, when +// stop() is finally called -- exactly the network-thread-pushes-during-the-grace-window scenario +// the fix (flush_ring_buffer() after join() in stop()) covers. +TEST(VisualizerLifecycle, StopDrainsEntryQueuedBeforeStop) { + auto impl = make_impl_with_client(); + ASSERT_TRUE(impl->start()); + + impl->request_stop(); + const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(2); + while (!impl->has_stopped() && std::chrono::steady_clock::now() < deadline) { + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + ASSERT_TRUE(impl->has_stopped()); + + // The drain thread has already exited, so this push has no consumer racing it. + std::vector data; + put_be64(data, 123456); + put_be16(data, 0xABCD); + impl->handle_binary(SENDSPIN_BINARY_VISUALIZER_LOUDNESS, data.data(), data.size()); + + // Joins the already-dead thread, then (with the fix) flushes the ring before returning. + impl->stop(); + + std::vector entry; + EXPECT_FALSE(pop_entry(*impl, entry)); +} + +// request_stop() signals the drain thread without joining it: the thread exits within its +// receive timeout and reports via has_stopped(), after which stop() joins instantly and a +// start() revives the role. This is the primitive behind the client's asynchronous +// request_stop(). +TEST(VisualizerLifecycle, RequestStopReportsExitAndRestarts) { + auto impl = make_impl_with_client(); + ASSERT_TRUE(impl->start()); + EXPECT_FALSE(impl->has_stopped()); + + impl->request_stop(); + const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(2); + while (!impl->has_stopped() && std::chrono::steady_clock::now() < deadline) { + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + EXPECT_TRUE(impl->has_stopped()); + impl->stop(); + + ASSERT_TRUE(impl->start()); + EXPECT_FALSE(impl->has_stopped()); + impl->stop(); +}