From 5dc64d602e6d0fe07d474da98521bdba0ca6d5aa Mon Sep 17 00:00:00 2001 From: Kevin Ahrendt Date: Fri, 16 Oct 2026 09:16:49 -0400 Subject: [PATCH] Keep frame payloads in slots placed with each pipeline instance Each encoded frame and visualizer cell carried its payload in a vector the group audio threads allocated per frame and core freed when its ring dropped it. Each pipeline instance now places a slab of slots of its format's largest frame when it is built; a frame's FramePayload holds one slot and returns it when destroyed, on whichever thread. An instance with every slot taken posts PayloadsFull and waits; core prunes its ring and wakes the pipelines once a slot is free, or arms a timer for its earliest frame. The group-to-core FIFO signals a preallocated Wakeup instead of posting its drain, which allocated on every post on the host. --- .claude/skills/realtime-review/SKILL.md | 2 +- CLAUDE.md | 5 +- cmake/sources.cmake | 2 + docs/conventions.md | 10 +- docs/internals.md | 7 +- src/frame_encoder.cpp | 73 +++++----- src/frame_encoder.h | 24 ++-- src/frame_payload.h | 64 +++++++++ src/group.cpp | 48 +++++++ src/group.h | 10 ++ src/group_audio.cpp | 125 +++++++++++------- src/group_audio.h | 59 ++++++--- src/payload_slots.cpp | 117 +++++++++++++++++ src/payload_slots.h | 86 ++++++++++++ src/player_stream.cpp | 4 +- src/profile.h | 3 + src/sched/frames.h | 21 ++- src/visualizer_stream.cpp | 20 +-- tests/CMakeLists.txt | 11 +- tests/frame_payloads.h | 79 +++++++++++ tests/freed_blocks.h | 70 ++++++++-- tests/group_pipelines.h | 25 +++- tests/test_frame_encoder.cpp | 53 ++++++-- tests/test_group_audio.cpp | 153 ++++++++++++++++----- tests/test_group_audio_internals.cpp | 90 ++++++++++++- tests/test_group_internals.cpp | 168 +++++++++++++++++++++++- tests/test_payload_slots.cpp | 144 ++++++++++++++++++++ tests/test_player_stream.cpp | 3 +- tests/test_sched_frames.cpp | 66 +++++++++- tests/test_sched_property.cpp | 7 +- tests/test_server_source.cpp | 4 +- tests/test_source_decoder.cpp | 8 +- tests/test_visualizer_stream.cpp | 5 +- 33 files changed, 1351 insertions(+), 215 deletions(-) create mode 100644 src/frame_payload.h create mode 100644 src/payload_slots.cpp create mode 100644 src/payload_slots.h create mode 100644 tests/frame_payloads.h create mode 100644 tests/test_payload_slots.cpp diff --git a/.claude/skills/realtime-review/SKILL.md b/.claude/skills/realtime-review/SKILL.md index ac639a3..2d6d135 100644 --- a/.claude/skills/realtime-review/SKILL.md +++ b/.claude/skills/realtime-review/SKILL.md @@ -16,7 +16,7 @@ Determine the diff: if `$ARGUMENTS` names a commit range or branch, diff that; o ## Core handler budget - Every core handler stays within about 200 microseconds: AEAD over one message, one X25519 step, a few KB of JSON. Receive stamps are read on the network thread, so a long handler does not move them, but it holds every time reply and send behind it. Flag new work on core that scales with the number of connections, members, frames or queued items without a bound per handler. -- Work that can exceed the budget is cut into posted steps, as `GroupToCoreFifo` drains (`DRAIN_BUDGET` events per handler, then it posts itself again), or moved off core, as the pipeline jobs are (to the group audio threads, `AudioWorkers`). Flag a loop that drains a whole queue in one handler, and encoding or resampling on core. +- Work that can exceed the budget is cut into posted steps, as `GroupToCoreFifo` drains (`DRAIN_BUDGET` events per run of its `Wakeup`, then it signals itself again), or moved off core, as the pipeline jobs are (to the group audio threads, `AudioWorkers`). Flag a loop that drains a whole queue in one handler, and encoding or resampling on core. - A pipeline job stays one frame per instance with at most one resampler built, so groups sharing the group audio threads take turns. Flag a job that loops until its input runs out. - When the verdict depends on a number, say what to measure, in a `RelWithDebInfo` build. diff --git a/CLAUDE.md b/CLAUDE.md index b119cb9..48ea8cd 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -46,6 +46,7 @@ script/clang-tidy.sh - clang-tidy over src/ using build/compile_commands.js - `host_view.h`: `HostView`, the host side of one input under one mutex: its kind (buffered or live), staging ring and its input segments, runs and epochs, anchoring and the anchor's shifts, drain, stop latch, the group parameters (`GroupParams`, in `group_params.h`) and a live input's latency slew. - `live_slew.h`: `LiveSlew`, a live input played slightly fast while its raised latency comes down (`sched/lowering.h` decides how fast): micro-resampler's ASRC on a host, single frames dropped on the embedded profile. - `group_audio.h`: `GroupAudio`, a group's pipelines and visualizer analyses, one job at a time, with the group-to-core FIFO and the pipeline-set slot. +- `payload_slots.h`: `PayloadSlots`, one pipeline instance's frame payloads in fixed slots placed when the instance is built, returned from any thread; `frame_payload.h` holds `FramePayload`, a frame's bytes in one slot. - `audio_workers.h`: `AudioWorkers`, the group audio threads: a fixed pool that runs each group's pipeline jobs off core, a group's jobs one at a time, groups taking turns. - `player_stream.h`: `PlayerStream`, one member's player stream: markers, cursor, ledger, chunk packing and the switch to another format in place, over a `StreamSource` (`stream_source.h`). - `playback_policy.h`: a session's activation (`decide_activation()`), a player's stream format, the group parameters and what a host's move does with a client (`move_action()`), as pure functions. @@ -78,7 +79,7 @@ script/clang-tidy.sh - clang-tidy over src/ using build/compile_commands.js - `inbound_hold.h`: `InboundHold`, a connection's message held for the inbound ring by a transport whose receive task reads for several connections (the ESP-IDF one): who pushes it, the pause tied to it, the close after it; platform-free and host-tested. - `inbound_ring.h`: `InboundRing`, how a transport's network threads hand received messages, with their receive stamps, and connection events to core: bounded, a message it refuses held back in the transport, independent of any transport. - `clock.h`: `now_us()`, the clock every timestamp is taken from. -- `profile.h`: the build's resource profile (write-ahead, staging ring, connection caps, the internal RAM a new connection needs, inbound ring, FLAC LPC, a source's clock correction and a live input's slew, audio threads, where each task's stack lives), the one place `SENDSPIN_SERVER_EMBEDDED_PROFILE` changes the library's limits. +- `profile.h`: the build's resource profile (write-ahead, staging ring, a pipeline instance's payload slots, connection caps, the internal RAM a new connection needs, inbound ring, FLAC LPC, a source's clock correction and a live input's slew, audio threads, where each task's stack lives), the one place `SENDSPIN_SERVER_EMBEDDED_PROFILE` changes the library's limits. - `memory.h`: placed allocation, each naming the memory it prefers (`MemoryLocation`: internal RAM or PSRAM, the other as the fallback): `placed_malloc()`, `PlacedAllocator` and `ExternalBytes` for containers, `make_placed_array()`, `make_shared_external()`, the `ExternalNew` base, and `memory_stats()`; `src/esp/memory_esp.cpp` on ESP-IDF (heap_caps, whatever the project's malloc policy), `memory_host.cpp` on the host (one heap). - `shared.h`: `SharedPtr`, `WeakPtr` and `EnableSharedFromThis`, the transport's shared pointers: with the atomic lock policy where libstdc++ is built without it (the ESP-IDF riscv32 toolchain), `std::shared_ptr` elsewhere. - `thread_setup.h`: `start_thread()`, each library thread's name and placement, and `current_thread_token()`, the calling thread's identity from any task; `thread_setup_host.cpp` on the host, `src/esp/thread_setup_esp.cpp` on ESP-IDF (core, priority, and the stack in internal RAM or PSRAM as `profile.h` places it; the token is the FreeRTOS task handle), with the connections' writer tasks. @@ -95,7 +96,7 @@ script/clang-tidy.sh - clang-tidy over src/ using build/compile_commands.js - Library target `sendspin_server_cpp` (alias `sendspin_server_cpp::sendspin_server_cpp`); CMake options are prefixed `SENDSPIN_SERVER_`. `SENDSPIN_SERVER_ENABLE_WERROR` (default off) makes warnings in this project's targets errors; CI builds with it on, so local builds should too. `SENDSPIN_SERVER_EMBEDDED_PROFILE` (default off) builds with `src/profile.h`'s embedded limits, as the ESP-IDF component (`cmake/esp_idf.cmake`) always does. - Plain: `cmake -B build && cmake --build build`, which builds `examples/` too (`SENDSPIN_SERVER_BUILD_EXAMPLES`, default on when top-level). Uses ccache when installed (`SENDSPIN_SERVER_USE_CCACHE`, default on when top-level). `SENDSPIN_SERVER_ENABLE_DISCOVERY` (default on) builds the mDNS backend: dns_sd on macOS, avahi-client through pkg-config on Linux (`libavahi-client-dev`). `SENDSPIN_SERVER_ENABLE_OPUS` (default on) builds the Opus encoder and the source role's Opus decoder; off compiles them out (`src/frame_encoder.cpp` and `src/source_decoder.cpp` are the places it is guarded) and is refused with the tests on. - Tests never use the network's mDNS: `tests/test_recorder.h` `local_config()` turns advertising and browsing off and listens on a port the system picks, and `test_server_discovery.cpp` supplies its own `DiscoveryProvider` through `ServerConfig::discovery`. -- Tests: `cmake -B build-asan -DSENDSPIN_SERVER_ENABLE_WERROR=ON -DSENDSPIN_SERVER_BUILD_TESTS=ON -DSENDSPIN_SERVER_ENABLE_SANITIZERS=ON`, then `cmake --build build-asan` and `ctest --test-dir build-asan --output-on-failure`. The suites whose assertions depend on players keeping up with real time carry the CTest label `realtime` (`tests/CMakeLists.txt`); CI runs them on their own and the rest in parallel (`-LE realtime -j`). The tests also build `server_cli`'s WAV and FLAC readers, feeders, `--one-group`, timing figures and log and `--web` dashboard; `tests/temp_dir.h` gives a test a temporary directory, and `tests/job_view.h` a copy of what a group's pipelines read of a host view. `tests/test_ports.h` hands out loopback ports for test peers, and holds one that nothing listens on for a dial that must be refused; `tests/raw_client.h` is the raw test client (noise-c and Beast, either suite, scriptable misbehavior), `tests/silent_listener.h` a listener that holds the server's dials in their upgrade or handshake, `tests/test_recorder.h` a recording listener, an in-memory provider and `LocalServer`, a server started with them, `tests/wire_messages.h` the spec's binary message types and fields as the tests decode them, `tests/flac_reader.h` decodes a FLAC stream as a player gets it, `tests/opus_reader.h` an Opus stream, with the test tones and an SNR to compare lossy audio by, `tests/player_sink.h` holds the test signal and a real-time audio output for sendspin-cpp players, `tests/peer.h` what each in-process sendspin-cpp client shares (its network provider and blob store, its config and the URL the server dials, its main loop run on the test thread, the client_id at a URL), `tests/trust_fixtures.h` distinct PSKs and the client record a server saved, `tests/hex.h` hex text for known answers, `tests/gate.h` a gate a job waits at, the fixtures and harnesses a class's two test files share (`tests/host_view_fixture.h`, `tests/group_pipelines.h`, `tests/source_sink_rig.h`, `tests/session_harness.h`, `tests/trust_harness.h`), `tests/raw_roles.h` the messages a raw client sends to drive the roles and the readers of what it receives, `tests/players_rig.h` the rig, writes and checks of the in-process player tests, `tests/group_job.h` a group's pipeline job wrapped between runs, for the files compiled with `-fno-access-control` alone, and `tests/freed_blocks.h` counts the heap blocks freed still holding a secret, through the sanitizer's free hook (a test that needs it skips itself without ASan or TSan). +- Tests: `cmake -B build-asan -DSENDSPIN_SERVER_ENABLE_WERROR=ON -DSENDSPIN_SERVER_BUILD_TESTS=ON -DSENDSPIN_SERVER_ENABLE_SANITIZERS=ON`, then `cmake --build build-asan` and `ctest --test-dir build-asan --output-on-failure`. The suites whose assertions depend on players keeping up with real time carry the CTest label `realtime` (`tests/CMakeLists.txt`); CI runs them on their own and the rest in parallel (`-LE realtime -j`). The tests also build `server_cli`'s WAV and FLAC readers, feeders, `--one-group`, timing figures and log and `--web` dashboard; `tests/temp_dir.h` gives a test a temporary directory, and `tests/job_view.h` a copy of what a group's pipelines read of a host view. `tests/test_ports.h` hands out loopback ports for test peers, and holds one that nothing listens on for a dial that must be refused; `tests/raw_client.h` is the raw test client (noise-c and Beast, either suite, scriptable misbehavior), `tests/silent_listener.h` a listener that holds the server's dials in their upgrade or handshake, `tests/test_recorder.h` a recording listener, an in-memory provider and `LocalServer`, a server started with them, `tests/wire_messages.h` the spec's binary message types and fields as the tests decode them, `tests/flac_reader.h` decodes a FLAC stream as a player gets it, `tests/opus_reader.h` an Opus stream, with the test tones and an SNR to compare lossy audio by, `tests/player_sink.h` holds the test signal and a real-time audio output for sendspin-cpp players, `tests/peer.h` what each in-process sendspin-cpp client shares (its network provider and blob store, its config and the URL the server dials, its main loop run on the test thread, the client_id at a URL), `tests/trust_fixtures.h` distinct PSKs and the client record a server saved, `tests/hex.h` hex text for known answers, `tests/gate.h` a gate a job waits at, the fixtures and harnesses a class's two test files share (`tests/host_view_fixture.h`, `tests/group_pipelines.h`, `tests/source_sink_rig.h`, `tests/session_harness.h`, `tests/trust_harness.h`), `tests/raw_roles.h` the messages a raw client sends to drive the roles and the readers of what it receives, `tests/players_rig.h` the rig, writes and checks of the in-process player tests, `tests/group_job.h` a group's pipeline job wrapped between runs, for the files compiled with `-fno-access-control` alone, `tests/frame_payloads.h` frame payloads a test owns, in slots of their own, and `tests/freed_blocks.h` counts the heap blocks freed still holding a secret, and those a thread allocates, through the sanitizer's free and malloc hooks (a test that needs them skips itself without ASan or TSan). - Soak: `tests/soak.cpp` (its trace checker in `tests/soak_wire_checker.h`, tested in the default suite), target `sendspin_server_cpp_soak` with its own entry point, registered as the CTest test `Soak` (label `soak`), which skips itself unless `SENDSPIN_SERVER_SOAK_SECONDS` is set (`SENDSPIN_SERVER_SOAK_SEED` picks the script); README.md has the command. - CI also builds the library with Opus and discovery off and memory statistics on; as a subproject (`tests/consumer/`); and as the ESP-IDF component (`tests/esp_idf/`: `idf.py -C tests/esp_idf set-target esp32s3`, then `idf.py -C tests/esp_idf build`). - Memory statistics: `SENDSPIN_SERVER_MEMORY_STATS` (default off; `CONFIG_SENDSPIN_SERVER_MEMORY_STATS` in the component's `Kconfig`) logs the heaps and the tasks' stacks at the server's start (on the thread calling `start()`) and, on the core thread, at a connection's start and end and a group's playback (`profile::MEMORY_STATS`); the tests count log lines and expect it off. diff --git a/cmake/sources.cmake b/cmake/sources.cmake index 01d236a..2ad1f92 100644 --- a/cmake/sources.cmake +++ b/cmake/sources.cmake @@ -52,6 +52,8 @@ set(SENDSPIN_SERVER_SOURCES src/noise.cpp # One pairing attempt: the decisions as pure functions, and its state and crypto src/pairing.cpp + # A pipeline instance's frame payloads, in slots placed with the instance + src/payload_slots.cpp src/pcm_convert.cpp src/pipeline_stages.cpp src/playback_policy.cpp diff --git a/docs/conventions.md b/docs/conventions.md index e7d3a58..75bc9c1 100644 --- a/docs/conventions.md +++ b/docs/conventions.md @@ -10,19 +10,19 @@ The design standards that code in this repository is held to. This document is n - **Out to the host: the dispatch thread.** `ServerListener`, `PersistenceProvider`, the message trace and the log sink are called only on the `Dispatcher`'s thread, one call at a time, in posting order. Core posts and never waits for them; the `Dispatcher` FIFO is unbounded so the host sets the rate it drains at. `start()`'s load is the one wait on the dispatch thread, from a host thread. - **Host reads: snapshots.** `clients()` and `groups()` copy a list core swaps under a mutex (`Snapshot`, latest wins). Core publishes the snapshot before posting the event that reports the change, so a listener sees the change in the snapshot. - **The host view.** A group input's state shared by host threads, core and the group audio threads lives in its `HostView`, one mutex and a condition variable; each `GroupInput` call is one step under it, but for a live write through the slew, which resamples with the mutex released between the steps that stage it. A callback the step decides goes into the view's outbox under the mutex and is posted to the dispatch thread after the mutex is released; a change the pipelines act on wakes the group's pipeline task and signals its `Wakeup` after release. -- **Pipelines to core.** Pipeline code runs on the group audio threads (`AudioWorkers`) and talks to core only through the `GroupToCoreFifo` (ordered events and frames, one posted drain handler) and the `PipelineSetSlot` (latest wins), and reads the host view; it does not reach into `Group` or `ServerCore`. What a job holds is shared with the group or outlives it (the FIFO's producer, the slot, the view), so core stops a group's task and erases the group without waiting for a job running. +- **Pipelines to core.** Pipeline code runs on the group audio threads (`AudioWorkers`) and talks to core only through the `GroupToCoreFifo` (ordered events and frames, drained by its `Wakeup`) and the `PipelineSetSlot` (latest wins), and reads the host view; it does not reach into `Group` or `ServerCore`. What a job holds is shared with the group or outlives it (the FIFO's producer, the slot, the view), so core stops a group's task and erases the group without waiting for a job running. - **Every library lock is a leaf.** A lock is held only to copy or update its own state, never across a listener or provider call, a post, a signal, a send or another library lock, so there is no lock order. A change that would take a second lock under one is a design change, not a local trade-off. - **Listeners may call back in.** A listener may call any `SendspinServer` or `GroupInput` method except `start()` and the destructor. A method a listener may call never waits for core or the dispatch thread (`stop()` only begins the shutdown). - **Any task may call in.** On ESP-IDF a host's calls come from tasks that are not pthreads (an application's main task, a task of its own), and `std::this_thread::get_id()` asserts on those; which thread a caller is goes through `current_thread_token()` (`thread_setup.h`). -- **Teardown discards, it does not deliver.** Work posted after its owner is gone runs as a no-op (a `Wakeup` signal after the `Wakeup` or its loop is gone, a FIFO handler after the FIFO, a task posted after `Dispatcher::finish()`), and the owner's declaration says so. A result the host is owed at teardown (a pending drain's `STOPPED`, a client's `SERVER_STOPPED`) is posted before the dispatch thread finishes. +- **Teardown discards, it does not deliver.** Work posted after its owner is gone runs as a no-op (a `Wakeup` signal after the `Wakeup` or its loop is gone, a FIFO's `Wakeup` run after the FIFO, a task posted after `Dispatcher::finish()`), and the owner's declaration says so. A result the host is owed at teardown (a pending drain's `STOPPED`, a client's `SERVER_STOPPED`) is posted before the dispatch thread finishes. ## Real-time and core budget -- **Core handlers have a budget** of about 200 microseconds. Receive stamps are read on the network thread, so an overrun does not move them, but every reply and send behind it waits. Work that can exceed the budget is cut into posted steps, as the FIFO drains at most `GroupToCoreFifo::DRAIN_BUDGET` events per handler before it posts itself again, or moved off core, as the pipelines are. A pipeline job produces at most one frame per instance and builds at most one resampler, so groups sharing the group audio threads take turns. +- **Core handlers have a budget** of about 200 microseconds. Receive stamps are read on the network thread, so an overrun does not move them, but every reply and send behind it waits. Work that can exceed the budget is cut into posted steps, as the FIFO drains at most `GroupToCoreFifo::DRAIN_BUDGET` events per run before it signals itself again, or moved off core, as the pipelines are. A pipeline job produces at most one frame per instance and builds at most one resampler, so groups sharing the group audio threads take turns. - **Core never blocks.** No waits, joins, sleeps, blocking I/O, name resolution (Asio's resolver runs on its own thread; browsed clients are dialed at their resolved addresses), listener or provider calls on core. The one exception is shutdown: core calls the discovery provider's `stop()`, which joins a built-in provider's thread. - **The write invariant:** a write starts only after the transport reported room and carries the messages ready as it starts, each its own WebSocket frame, within `write_batch_has_room()`; a time reply ends it; the next wait starts after the write completes. Nothing waits for more messages to fill a write. A message is built, stamped and encrypted when the pump takes it, with the stamp read immediately before encryption. - **`write()` is real-time safe:** it never allocates or frees, waits or does I/O. It copies into the staging ring preallocated by `open_input()` and `set_format()`, under the host-view mutex held for the copy and its bookkeeping, and signals core through the preallocated `Wakeup`. Of the `GroupInput` methods only `wait_writable()` waits. -- **A running pipeline instance does not reallocate.** The resampler is fed at most the block size its history buffer is sized for; filter banks and encoders are allocated when an instance is built. +- **A running pipeline instance does not reallocate.** The resampler is fed at most the block size its history buffer is sized for; filter banks, encoders and the frames' payload slots are allocated when an instance is built. - **Memory is placed.** On ESP-IDF a buffer, a per-connection or per-group object and a task stack name where they live through `memory.h` (`ExternalBytes`, `make_placed_array()`, `make_shared_external()`, `ExternalNew`) or `profile.h`, rather than leaving it to the project's malloc policy; internal RAM only for what must be there (a stack of a task that may write flash, a FreeRTOS object) or is faster there. - **Capacities are justified numbers.** Queues, rings and caches have a fixed capacity and an intentional overflow behavior, or a stated reason they need none at their declaration (the `Dispatcher` FIFO above). An overflow is observable (a `SessionEnd`, `on_error()`, a `false` return) or is an ignore documented at the cap's declaration (`discovery_limits::MAX_CLIENTS`). A queue is a `Ring` (`ring.h`: allocated with its owner, at its cap where it has one, else growing only past what its owner expects), for variable-length bytes a `ByteRing` (`byte_ring.h`), or, for closures, a `TaskQueue` (`task_queue.h`), never a `std::deque`, which allocates a block as it slides. Related constants are derived from each other (`constexpr`), not restated. Timing constants are named `constexpr` values (`src/sched/constants.h`) or, where the host tunes them, `ServerTiming` fields; never literals at the use site. - Trivially copyable types are copied, not `std::move`d; heap-backed types are moved. @@ -33,7 +33,7 @@ The design standards that code in this repository is held to. This document is n - **Fail closed and uniformly.** A malformed or missing required field rejects the whole enclosing object; nothing is silently defaulted, clamped or repaired. Sibling fields validate to the same standard. A part validated on its own (a `client/command`'s role objects, a `client/state` player object) is documented as such at its parser in `messages.h`. A value a newer revision may add (an unknown codec, an unrecognized message type) is skipped or ignored as the spec's ignore rules say, at one documented site. - **A misbehaving peer is closed, a well-behaved one never is** for the server's own backlog: the session ends with its `SessionEnd` reason. Passing a send queue's cap (`SendQueueCaps`) means the client misbehaves; audio never overflows because cursors pull. - **Reuse the parse and build helpers** in `messages.cpp` instead of per-field logic. Two paths that build or parse the same structure share one helper. -- **Pure logic is pure.** Scheduler math, accounting, activation and reconnect policy, format selection, controller state and message builders are free functions or value types over plain data, with no I/O, threads, locks or clock reads; a timing decision takes `now` as an argument and its caller reads `now_us()`. `src/sched/`, `discovery_policy.h`, `playback_policy.h` and `controller.h` are the shape. +- **Pure logic is pure.** Scheduler math, accounting, activation and reconnect policy, format selection, controller state and message builders are free functions or value types over plain data, with no I/O, threads, locks or clock reads; a timing decision takes `now` as an argument and its caller reads `now_us()`. `src/sched/`, `discovery_policy.h`, `playback_policy.h` and `controller.h` are the shape. The one lock `src/sched/` reaches is `PayloadSlots`' leaf mutex, which a frame's `FramePayload` takes to return its slot when the frame is destroyed. - **One clock.** Every timestamp comes from `now_us()` (`clock.h`). There is no clock injection. - Spec-driven behavior cites the spec section it implements (`docs/comment-style.md`). diff --git a/docs/internals.md b/docs/internals.md index 965c535..834b855 100644 --- a/docs/internals.md +++ b/docs/internals.md @@ -40,7 +40,8 @@ Asio's resolver runs host name lookups for `connect_url()` on a thread of its ow | Group ids, and the input, run, epoch and drain ids | Host threads, core (group ids), group audio (a routed source's writes) | The same | Atomic counters: the run's group ids, which `create_group()` and core take alike; the others process-wide, so ids from two inputs never compare equal | | An input's `on_drained()` and `on_writable()` | Host threads and core, under the host-view mutex | Dispatch | The view's outbox, posted after the mutex is released | | Host view changes the pipelines act on | Host threads, core | Group audio, core | The group's pipeline task (`AudioWorkers::Task::wake()`) and its `Wakeup`, preallocated, so `write()` never allocates | -| Group-to-core events and frames | The group's pipelines | Core | `GroupToCoreFifo`: an ordered FIFO with one posted drain handler | +| Group-to-core events and frames | The group's pipelines | Core | `GroupToCoreFifo`: an ordered FIFO whose pushes signal its preallocated `Wakeup`, whose run drains it | +| A pipeline instance's frame payloads | The instance (group audio) | Core (the streams' reads); a slot returns wherever its frame is destroyed | `PayloadSlots` (`payload_slots.h`): slots of the format's largest frame, placed when the instance is built, and a list of the free ones under one mutex, held to take or return a slot; the instance writes its slot without it. An instance with every slot taken posts `PayloadsFull` and waits; core prunes its ring and wakes the pipelines, or arms a timer for its earliest frame | | The pipeline set | Core | The group's pipelines | `PipelineSetSlot`, latest wins | | Artwork images | Host threads (`set_artwork()`) | Core | `EventLoop::post()`; the image is copied, checked and hashed on the host thread | | Artwork jobs and results | Core, then the artwork worker | The artwork worker, then core | `ArtworkWorker`'s bounded FIFO; results come back with `EventLoop::post()` | @@ -89,7 +90,7 @@ At `client/init` the session asks which PSK message 1 references (`select_handsh 1. `open_input()` creates the input's `HostView` on the host thread and posts its attachment to core. Attached, the group hands its `GroupAudio` to the group audio threads and writes its parameters (group lead, write-ahead) into the view as its members change them. 2. `write()` copies into the view's staging ring; the write that begins a run anchors it. `set_format()` starts an input segment at the head, so the timeline continues at the new rate. A live input's write also measures how much later its first sample lands than a fresh anchor would place it, against `G_next` or the input's least lead where higher (`HostView::lead_next()`). Its writes fall in consecutive windows of `LOWERING_WINDOW_US` from the epoch's first; once every write of a window kept such an excess, the window's least is owed, never more than a later write's own excess, and the view plays the writes that follow `LOWERING_SLEW_PPM` fast until it is paid (`sched::live_slew_ppm()`). `LiveSlew` (`live_slew.h`) plays them: micro-resampler's ASRC where `profile::SOURCE_ASRC`, started from the frames staged before it and run on until the epoch ends, so its output stays continuous, else single frames dropped (`SampleSlip`). The write runs the slew with the view's mutex released and stages each block's output under it, so core and the group audio threads never wait on the resampling; a flush, stop, pause, drain, close, release, server stop, new epoch or format meanwhile ends its staging, and a write made meanwhile takes nothing (`InputState::FULL`). A write then stages fewer positions than it was given and reports those, and `G_eff` falls by the time taken out (`sched::record_slew()`), never below `G_next` or the least lead, nor does the lead, as the rate leaves a frame of the debt for the part frame a frame drop carries: the lead keeps its distance from the stall rebase's threshold and from admission's horizon, so no frame goes late and no rebase follows. A joiner's raise during the slew lowers the debt to the excess it leaves; a stall's shift is measured by the windows after it. A latency a joiner, a stall or a fast clock raised thus comes back down within the epoch, starting one to two windows after it rose; an excess that straddles a window's start waits for the next. -3. A `GroupAudio` job reads the view, posts the epoch events, shifts the timeline when a frame's lead falls too low (a rebase), produces at most one frame per pipeline instance and lets the view drop staged PCM no instance needs. Core's FIFO handler appends frames to the frame rings and pumps the members. +3. A `GroupAudio` job reads the view, posts the epoch events, shifts the timeline when a frame's lead falls too low (a rebase), produces at most one frame per pipeline instance, each into a slot of the instance's `PayloadSlots`, and lets the view drop staged PCM no instance needs. The FIFO's `Wakeup` run on core appends frames to the frame rings, which drop each once it plays and so return its slot, and pumps the members. 4. Each member's `PlayerStream` reads the group's frames through a cursor, packs chunks within its ledger, and emits `stream/start`, `stream/clear` and `stream/end` from markers at the head of its control FIFO. A `VisualizerStream` runs beside it on the same markers. 5. A drain completes once no member's stream is active; core ends the run before the view releases `on_drained`. @@ -121,7 +122,7 @@ Send: after every event the driver runs its pump. With a message to send and no | --- | --- | | `constants.h` | The send path's constants and the tunable `SchedTiming` defaults | | `types.h` | Run, epoch and instance ids, formats, roles, and the input, run and drain states | -| `frames.h` | The frame grid (`frame_layout()`), the `FrameRing`, the `Cursor` and its switch point (`plan_switch()`) | +| `frames.h` | The frame grid (`frame_layout()`), the `FrameRing` and the payload slots an instance needs (`payload_slots()`), the `Cursor` and its switch point (`plan_switch()`) | | `timeline.h` | The `Anchor` with its segments and shifts: timestamps, positions, the stall rebase and the pause position | | `leads.h` | Per-player leads, pipeline latency, the group lead, the startup bound and the write-ahead | | `ledger.h` | `Ledger`, a session's byte accounting for one stream role | diff --git a/src/frame_encoder.cpp b/src/frame_encoder.cpp index 3e6a62f..051b0f7 100644 --- a/src/frame_encoder.cpp +++ b/src/frame_encoder.cpp @@ -216,15 +216,17 @@ FrameEncoder::FrameEncoder(const sched::AudioFormat& format, uint32_t frame_samp frame_samples_(frame_samples), flac_(std::move(flac)), opus_(std::move(opus)) { + const size_t frame_bytes = + size_t{frame_samples} * format.channels * pcm_output_bytes(quantized_depth(format)); + this->max_frame_bytes_ = frame_bytes; if (this->flac_ || this->opus_) { - this->quantized_.resize(size_t{frame_samples} * format.channels * - pcm_output_bytes(quantized_depth(format))); + this->quantized_.resize(frame_bytes); } if (this->flac_) { - this->encoded_.resize(this->flac_->get_max_output_bytes()); + this->max_frame_bytes_ = this->flac_->get_max_output_bytes(); } if (this->opus_) { - this->encoded_.resize(OPUS_MAX_PACKET_BYTES); + this->max_frame_bytes_ = OPUS_MAX_PACKET_BYTES; this->opus_input_.resize(size_t{frame_samples} * format.channels); } } @@ -240,64 +242,65 @@ void FrameEncoder::reset() { } } -bool FrameEncoder::encode(std::span input, SampleFormat in_format, - std::vector& payload) { +std::optional FrameEncoder::encode(std::span input, SampleFormat in_format, + std::span out) { // Samples of every channel const size_t values = input.size() / sample_bytes(in_format); const uint32_t depth = quantized_depth(this->format_); const size_t bytes = values * pcm_output_bytes(depth); if (!this->flac_ && !this->opus_) { - payload.resize(bytes); - convert_pcm(input, in_format, depth, payload); - return true; + if (bytes > out.size()) { + return std::nullopt; + } + convert_pcm(input, in_format, depth, out.first(bytes)); + return bytes; } - payload.clear(); if (bytes > this->quantized_.size()) { - return false; + return std::nullopt; } const std::span quantized(this->quantized_.data(), bytes); convert_pcm(input, in_format, depth, quantized); - return this->encode_quantized(quantized, payload); + return this->encode_quantized(quantized, out); } -bool FrameEncoder::encode_quantized(std::span samples, - std::vector& payload) { +std::optional FrameEncoder::encode_quantized(std::span samples, + std::span out) { if (this->opus_) { - return this->encode_opus(samples, payload); + return this->encode_opus(samples, out); } if (!this->flac_) { - payload.assign(samples.begin(), samples.end()); - return true; + if (samples.size() > out.size()) { + return std::nullopt; + } + std::copy(samples.begin(), samples.end(), out.begin()); + return samples.size(); } - payload.clear(); const size_t full = this->quantized_.size(); if (samples.size() > full) { - return false; + return std::nullopt; } size_t written = 0; micro_flac::FLACEncoderResult result = micro_flac::FLAC_ENCODER_SUCCESS; if (samples.size() == full) { size_t consumed = 0; - result = this->flac_->encode(samples.data(), samples.size(), this->encoded_.data(), - this->encoded_.size(), consumed, written); + result = this->flac_->encode(samples.data(), samples.size(), out.data(), out.size(), + consumed, written); } else { // A short frame is the stream's last - result = this->flac_->finish(samples.data(), samples.size(), this->encoded_.data(), - this->encoded_.size(), written); + result = + this->flac_->finish(samples.data(), samples.size(), out.data(), out.size(), written); } if (result != micro_flac::FLAC_ENCODER_SUCCESS || written == 0) { - return false; + return std::nullopt; } - payload.assign(this->encoded_.begin(), - this->encoded_.begin() + static_cast(written)); - return true; + return written; } -bool FrameEncoder::encode_opus(std::span samples, std::vector& payload) { - payload.clear(); +std::optional FrameEncoder::encode_opus(std::span samples, + std::span out) { const size_t full = this->opus_input_.size() * sizeof(int16_t); if (samples.size() > full) { - return false; + return std::nullopt; } // Little-endian 16-bit samples, then silence to a whole frame: an Opus frame has a fixed // duration (spec player/v1.md "Codec framing": one packet per chunk) @@ -306,14 +309,12 @@ bool FrameEncoder::encode_opus(std::span samples, std::vector(samples[i * 2] | (samples[(i * 2) + 1] << 8)); this->opus_input_[i] = static_cast(value); } - const size_t written = opus_encode_frame(this->opus_.get(), this->opus_input_.data(), - this->frame_samples_, this->encoded_); + const size_t written = + opus_encode_frame(this->opus_.get(), this->opus_input_.data(), this->frame_samples_, out); if (written == 0) { - return false; + return std::nullopt; } - payload.assign(this->encoded_.begin(), - this->encoded_.begin() + static_cast(written)); - return true; + return written; } } // namespace sendspin::server diff --git a/src/frame_encoder.h b/src/frame_encoder.h index 71cf0f2..db2396e 100644 --- a/src/frame_encoder.h +++ b/src/frame_encoder.h @@ -84,17 +84,24 @@ class FrameEncoder { /// cleared. At every epoch start. void reset(); - /// @brief Encodes one frame + /// @brief Encodes one frame into out /// @param input whole samples in in_format at the format's channel count: one frame's, or /// fewer for a drained epoch's last frame, which ends a FLAC stream and which Opus pads with /// silence to a whole frame - /// @return false when the encoder refused the call; payload is then empty - bool encode(std::span input, SampleFormat in_format, - std::vector& payload); + /// @param out room for max_frame_bytes() + /// @return the bytes written, or nullopt when the encoder refused the call + std::optional encode(std::span input, SampleFormat in_format, + std::span out); /// @brief encode() for samples the quantize stage (convert_pcm()) already wrote at /// quantized_depth() - bool encode_quantized(std::span samples, std::vector& payload); + std::optional encode_quantized(std::span samples, + std::span out); + + /// @brief The most bytes one frame takes: output_shape()'s `B_f` + size_t max_frame_bytes() const { + return this->max_frame_bytes_; + } private: struct OpusDeleter { @@ -105,7 +112,7 @@ class FrameEncoder { std::unique_ptr flac, std::unique_ptr opus); - bool encode_opus(std::span samples, std::vector& payload); + std::optional encode_opus(std::span samples, std::span out); sched::AudioFormat format_; uint32_t frame_samples_; @@ -113,10 +120,9 @@ class FrameEncoder { std::unique_ptr flac_; /// Opus only std::unique_ptr opus_; - /// FLAC and Opus: the quantize stage's output, one full frame's, and the encoder's, at its - /// largest + size_t max_frame_bytes_{0}; + /// FLAC and Opus: the quantize stage's output, one full frame's std::vector quantized_; - std::vector encoded_; /// Opus only: one frame of 16-bit samples, aligned for the encoder std::vector opus_input_; }; diff --git a/src/frame_payload.h b/src/frame_payload.h new file mode 100644 index 0000000..bd6a2e2 --- /dev/null +++ b/src/frame_payload.h @@ -0,0 +1,64 @@ +// 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. + +/// @file frame_payload.h +/// @brief FramePayload: a frame's bytes, held in a slot of its pipeline instance's PayloadSlots + +#pragma once + +#include "shared.h" + +#include +#include +#include + +namespace sendspin::server { + +class PayloadSlots; + +/// @brief A frame's bytes in one slot of its instance's PayloadSlots (payload_slots.h). Move-only; +/// destroying it returns the slot, on whatever thread that happens. Empty when default-made. +class FramePayload { +public: + FramePayload() noexcept = default; + ~FramePayload(); + FramePayload(FramePayload&& other) noexcept; + FramePayload& operator=(FramePayload&& other) noexcept; + FramePayload(const FramePayload&) = delete; + FramePayload& operator=(const FramePayload&) = delete; + + std::span bytes() const { + return {this->data_, this->size_}; + } + + size_t size() const { + return this->size_; + } + +private: + friend class PayloadSlots; + + FramePayload(SharedPtr slots, const uint8_t* data, size_t size, + uint32_t slot) noexcept; + + /// Returns the slot, leaving this empty + void reset() noexcept; + + SharedPtr slots_; + const uint8_t* data_{nullptr}; + size_t size_{0}; + uint32_t slot_{0}; +}; + +} // namespace sendspin::server diff --git a/src/group.cpp b/src/group.cpp index 989774a..617a317 100644 --- a/src/group.cpp +++ b/src/group.cpp @@ -115,6 +115,7 @@ Group::Group(EventLoop& loop, AudioWorkers& workers, GroupId id, std::string nam stall_timer_(loop), writable_timer_(loop), drain_timer_(loop), + payloads_timer_(loop), metadata_timer_(loop), color_timer_(loop), artwork_cache_(artwork), @@ -667,6 +668,8 @@ void Group::deliver(GroupEvent& event) { this->on_frame(value); } else if constexpr (std::is_same_v) { this->on_instance_final(value); + } else if constexpr (std::is_same_v) { + this->on_payloads_full(value); } else { this->on_pipeline_failed(value); } @@ -689,6 +692,7 @@ void Group::on_epoch_started(EpochStarted& started) { for (auto& [instance, ring] : this->rings_) { ring.restart(this->epoch_, 0); } + this->prune_full(); if (started.drain) { this->on_drain_started(*started.drain); } @@ -722,6 +726,7 @@ void Group::on_epoch_cut(const EpochCut& cut) { for (auto& [instance, ring] : this->rings_) { ring.restart(this->epoch_, 0); } + this->prune_full(); this->queue_markers(sched::TriggerKind::GROUP); } @@ -810,6 +815,48 @@ void Group::on_instance_final(const InstanceFinal& final) { this->schedule_check(); } +void Group::on_payloads_full(const PayloadsFull& full) { + const auto held = + std::find_if(this->full_.begin(), this->full_.end(), + [&](const PayloadsFull& f) { return f.instance == full.instance; }); + if (held == this->full_.end()) { + this->full_.push_back(full); + } + this->prune_full(); +} + +void Group::prune_full() { + if (this->full_.empty()) { + return; + } + const int64_t now = now_us(); + bool freed = false; + std::optional due; + std::erase_if(this->full_, [&](const PayloadsFull& full) { + const auto it = this->rings_.find(full.instance); + if (it == this->rings_.end()) { + return true; + } + sched::FrameRing& ring = it->second; + ring.prune(now); + if (ring.size() < full.slots) { + freed = true; + return true; + } + const int64_t ts = ring.at(ring.epoch(), ring.base())->ts_us; + due = due ? std::min(*due, ts) : ts; + return false; + }); + if (freed && this->audio_) { + this->audio_->wake(); + } + if (due) { + this->payloads_timer_.start(*due, this->guarded([this] { this->prune_full(); })); + } else { + this->payloads_timer_.cancel(); + } +} + void Group::on_pipeline_failed(const PipelineFailed& failed) { // Matched by instance, whatever its epoch: ids are never reused, and a broken instance stays // broken @@ -859,6 +906,7 @@ void Group::end_run(sched::EpochId epoch) { for (auto& [instance, ring] : this->rings_) { ring.restart({}, 0); } + this->prune_full(); for (const Member& member : this->members_) { this->send_group_update(*member.member); } diff --git a/src/group.h b/src/group.h index fac064b..e617ede 100644 --- a/src/group.h +++ b/src/group.h @@ -218,6 +218,12 @@ class Group : public ExternalNew { /// stall_timer_ for counts it holds back void report_stalls(int64_t now); void on_instance_final(const InstanceFinal& final); + /// Holds the instance among those waiting for payload slots, then prune_full() + void on_payloads_full(const PayloadsFull& full); + /// Prunes the rings of the instances waiting for payload slots and wakes the pipelines once + /// one holds fewer frames than its slots; payloads_timer_ fires at the earliest frame still + /// due, for the others. Also after a restart or erase of rings, which frees their slots. + void prune_full(); /// Excludes the format until the run ends, ends the streams on it and reports each player that /// streamed or wanted it void on_pipeline_failed(const PipelineFailed& failed); @@ -313,12 +319,16 @@ class Group : public ExternalNew { std::shared_ptr slot_; PipelineSet pipelines_; std::map rings_; + /// The instances whose payload slots were all taken when they last asked, with their slots + std::vector full_; bool drain_check_posted_{false}; sched::StallCount stalls_; /// Fires once a stall report held back is due Timer stall_timer_; Timer writable_timer_; Timer drain_timer_; + /// Fires at the earliest frame of a full_ instance's ring + Timer payloads_timer_; Timer metadata_timer_; Timer color_timer_; diff --git a/src/group_audio.cpp b/src/group_audio.cpp index ccf0859..7ce5d24 100644 --- a/src/group_audio.cpp +++ b/src/group_audio.cpp @@ -62,8 +62,13 @@ uint32_t segment_rate(const sched::Anchor& anchor, uint64_t position) { // ============================================================================== GroupToCoreFifo::GroupToCoreFifo(EventLoop& loop, std::function deliver) - : loop_(loop), state_(std::make_shared()) { + : state_(std::make_shared()), wakeup_(loop, [weak = std::weak_ptr(this->state_)] { + if (const std::shared_ptr state = weak.lock()) { + GroupToCoreFifo::drain(state); + } + }) { this->state_->deliver = std::move(deliver); + this->state_->wake = this->wakeup_.handle(); } GroupToCoreFifo::~GroupToCoreFifo() { @@ -74,61 +79,52 @@ GroupToCoreFifo::~GroupToCoreFifo() { dropped.swap(this->state_->events); } this->state_->deliver = nullptr; + this->state_->wake.reset(); } GroupToCoreFifo::Producer GroupToCoreFifo::producer() const { - return {this->loop_, this->state_}; + return {this->state_, this->wakeup_.handle()}; } -GroupToCoreFifo::Producer::Producer(EventLoop& loop, std::shared_ptr state) - : loop_(&loop), state_(std::move(state)) {} +GroupToCoreFifo::Producer::Producer(std::shared_ptr state, Wakeup::Handle wake) + : state_(std::move(state)), wake_(std::move(wake)) {} void GroupToCoreFifo::Producer::push(GroupEvent event) const { - bool post = false; { const std::lock_guard lock(this->state_->mutex); if (this->state_->closed) { return; } this->state_->events.push(std::move(event)); - post = !std::exchange(this->state_->drain_posted, true); - } - if (post) { - GroupToCoreFifo::drain(*this->loop_, this->state_); } + // A signal while a run is pending adds nothing + Wakeup::signal(this->wake_); } -// At most one handler is pending, and none leaves an event stranded -void GroupToCoreFifo::drain(EventLoop& loop, const std::shared_ptr& state) { - loop.post([&loop, weak = std::weak_ptr(state)] { - const std::shared_ptr state = weak.lock(); - if (!state) { - return; - } - Ring& batch = state->taking; - { - const std::lock_guard lock(state->mutex); - while (!state->events.empty() && batch.try_push(std::move(state->events.front()))) { - state->events.pop_front(); - } - } - for (; !batch.empty(); batch.pop_front()) { - if (!state->deliver) { - batch.clear(); - return; - } - state->deliver(batch.front()); - } - bool again = false; - { - const std::lock_guard lock(state->mutex); - again = !state->events.empty(); - state->drain_posted = again; +void GroupToCoreFifo::drain(const std::shared_ptr& state) { + Ring& batch = state->taking; + { + const std::lock_guard lock(state->mutex); + while (!state->events.empty() && batch.try_push(std::move(state->events.front()))) { + state->events.pop_front(); } - if (again) { - GroupToCoreFifo::drain(loop, state); + } + for (; !batch.empty(); batch.pop_front()) { + // A delivery may destroy the FIFO + if (!state->deliver) { + batch.clear(); + return; } - }); + state->deliver(batch.front()); + } + bool again = false; + { + const std::lock_guard lock(state->mutex); + again = !state->events.empty(); + } + if (again) { + Wakeup::signal(state->wake); + } } // ============================================================================== @@ -477,6 +473,12 @@ bool GroupAudio::configure_one(const JobView& view) { this->fail(instance); continue; } + instance.payloads = PayloadSlots::create(instance.encoder->max_frame_bytes(), + sched::payload_slots(instance.cells)); + if (!instance.payloads) { + this->fail(instance); + continue; + } } const Stage stage = this->ensure_resampler(instance, this->start_rate(instance, view)); if (stage == Stage::FAILED) { @@ -587,6 +589,9 @@ bool GroupAudio::produce(Instance& instance, const JobView& view) { if (!instance.configured) { return false; } + if (!this->reserve_slot(*instance.payloads, instance.id)) { + return false; + } if (!instance.begun && !this->begin(instance, view)) { return instance.failed; } @@ -922,10 +927,13 @@ bool GroupAudio::emit(Instance& instance, const JobView& view, size_t samples) { frame.duration_us = sched::frame_duration_us(frame.samples, instance.format.sample_rate); // With a lead, a drained epoch's last frames take what pending still holds const size_t bytes = std::min(samples * instance.sample_bytes, instance.pending.size()); - if (!instance.encoder->encode_quantized(std::span(instance.pending.data(), bytes), - frame.payload)) { + // The slot produce() reserved + const std::optional written = instance.encoder->encode_quantized( + std::span(instance.pending.data(), bytes), instance.payloads->reserve()); + if (!written) { return false; } + frame.payload = instance.payloads->commit(*written); this->fifo_.push(FrameReady{instance.id, std::move(frame)}); instance.pending.erase(instance.pending.begin(), instance.pending.begin() + static_cast(bytes)); @@ -966,6 +974,12 @@ void GroupAudio::follow_analyses(const JobView& view) { analysis.cell_bytes += visualizer_message_bytes(type, bins); } } + analysis.messages.reserve(analysis.cell_bytes); + analysis.payloads = + PayloadSlots::create(std::max(1, analysis.cell_bytes), sched::payload_slots(1)); + if (!analysis.payloads) { + this->fail(analysis); + } // from_index applies in its own epoch once anchored if (this->started_) { analysis.start = this->start_cell( @@ -984,6 +998,9 @@ bool GroupAudio::analyze(Analysis& analysis, const JobView& view, int64_t now_us if (view.drain_end && analysis_finished(analysis, view)) { return this->post_analysis_final(analysis); } + if (!this->reserve_slot(*analysis.payloads, analysis.id)) { + return false; + } const sched::Anchor& anchor = view.anchor; // A cell that ends by now has every instant, shifted as it is, at or before now, so every // cursor skips its messages: an analysis that fell behind with its input staged restarts at the @@ -1048,7 +1065,8 @@ bool GroupAudio::analyze(Analysis& analysis, const JobView& view, int64_t now_us frame.epoch = this->epoch_; frame.index = cell; frame.ts_us = cell_start + shift; - frame.payload.reserve(analysis.cell_bytes); + std::vector& messages = analysis.messages; + messages.clear(); const auto instant = [&](int64_t at, bool is_periodic, bool onset) { // A drained epoch's last cell has no input past the drain position to analyze const uint64_t position = sched::time_position(anchor, at); @@ -1058,9 +1076,9 @@ bool GroupAudio::analyze(Analysis& analysis, const JobView& view, int64_t now_us if (!this->feed_analysis(analysis, view, position)) { return false; } - const size_t before = frame.payload.size(); - analysis.analysis->analyze(at + shift, is_periodic, onset, frame.payload); - if (frame.payload.size() != before) { + const size_t before = messages.size(); + analysis.analysis->analyze(at + shift, is_periodic, onset, messages); + if (messages.size() != before) { frame.ts_us = at + shift; } return true; @@ -1075,6 +1093,14 @@ bool GroupAudio::analyze(Analysis& analysis, const JobView& view, int64_t now_us analysis.start = cell; return analysis.failed; } + // The slot reserved above, of cell_bytes + const std::span room = analysis.payloads->reserve(); + if (messages.size() > room.size()) { + this->fail(analysis); + return true; + } + std::copy(messages.begin(), messages.end(), room.begin()); + frame.payload = analysis.payloads->commit(messages.size()); this->fifo_.push(FrameReady{analysis.id, std::move(frame)}); analysis.last = cell; analysis.next = cell + 1; @@ -1175,6 +1201,17 @@ void GroupAudio::fail(Analysis& analysis) { this->fifo_.push(PipelineFailed{this->epoch_, analysis.id}); } +bool GroupAudio::reserve_slot(PayloadSlots& payloads, sched::InstanceId instance) { + const bool starved = payloads.starved(); + if (!payloads.reserve().empty()) { + return true; + } + if (!starved) { + this->fifo_.push(PayloadsFull{instance, payloads.slots()}); + } + return false; +} + int64_t GroupAudio::frame_start(const Instance& instance, int64_t index) { return (index / instance.cells) * instance.frame_samples; } diff --git a/src/group_audio.h b/src/group_audio.h index e532e3d..cd4d494 100644 --- a/src/group_audio.h +++ b/src/group_audio.h @@ -23,6 +23,7 @@ #include "frame_encoder.h" #include "host_view.h" #include "memory.h" +#include "payload_slots.h" #include "pipeline_stages.h" #include "ring.h" #include "sched/frames.h" @@ -101,23 +102,32 @@ struct PipelineFailed { sched::InstanceId instance; }; +/// @brief Every payload slot of the instance holds a frame core has not dropped: it produces +/// nothing until core drops one and wakes it. Its frames posted before it are in the FIFO ahead +/// of it, so once core holds fewer than slots of them, one is free. +struct PayloadsFull { + sched::InstanceId instance; + size_t slots{0}; +}; + using GroupEvent = std::variant; + FrameReady, InstanceFinal, PipelineFailed, PayloadsFull>; -/// @brief The ordered group-to-core FIFO: its producers push from any thread, and one posted -/// handler at a time delivers the events on core, at most a budget per handler. +/// @brief The ordered group-to-core FIFO: its producers push from any thread and signal its Wakeup, +/// whose run delivers the events on core, at most a budget per run. A push allocates nothing once +/// the ring has grown to the events a drain finds. /// -/// It needs no cap, since its producers bound it: a frame comes only from staged input, which -/// admission keeps within the group's lead and write-ahead past the time playing, and a job posts -/// at most a few epoch events, however many host calls preceded it. +/// It needs no cap, since its producers bound it: a frame holds one of its instance's payload slots +/// until core drops it (sched::payload_slots()), and a job posts at most a few epoch events, +/// however many host calls preceded it. class GroupToCoreFifo { struct State; public: - /// The most events one handler delivers before it posts itself again + /// The most events one run delivers before it signals the next static constexpr size_t DRAIN_BUDGET = 64; - /// The events the FIFO and a handler's batch have room for before they grow: a few jobs' frames - /// and epoch events between two drains + /// The events the FIFO and a run's batch have room for before they grow: a few jobs' frames and + /// epoch events between two drains static constexpr size_t QUEUED_EVENTS = 16; /// @brief The pushing end, which a job holds on its thread and which may outlive the FIFO @@ -129,17 +139,16 @@ class GroupToCoreFifo { private: friend class GroupToCoreFifo; - Producer(EventLoop& loop, std::shared_ptr state); + Producer(std::shared_ptr state, Wakeup::Handle wake); - /// Outlives every producer: the group audio threads are joined before the loop goes - EventLoop* loop_; std::shared_ptr state_; + Wakeup::Handle wake_; }; /// @param deliver called on core, in push order GroupToCoreFifo(EventLoop& loop, std::function deliver); - /// @brief A posted handler still pending delivers nothing, and a later push is dropped. On - /// core. + /// @brief A run still pending delivers nothing, and a later push is dropped. On core, from a + /// delivery too. ~GroupToCoreFifo(); GroupToCoreFifo(const GroupToCoreFifo&) = delete; GroupToCoreFifo& operator=(const GroupToCoreFifo&) = delete; @@ -152,19 +161,21 @@ class GroupToCoreFifo { /// Its producers bound it (above); room for QUEUED_EVENTS at first Ring events{QUEUED_EVENTS, Ring::UNLIMITED, MemoryLocation::PREFER_EXTERNAL}; - bool drain_posted{false}; /// The FIFO is gone bool closed{false}; /// Core only std::function deliver; - /// Core only: the batch a handler delivers + /// Core only: the batch a run delivers Ring taking{QUEUED_EVENTS, DRAIN_BUDGET, MemoryLocation::PREFER_EXTERNAL}; + /// Core only: the FIFO's Wakeup, which a run signals for the events past its budget + Wakeup::Handle wake; }; - static void drain(EventLoop& loop, const std::shared_ptr& state); + /// On core, as the Wakeup's run + static void drain(const std::shared_ptr& state); - EventLoop& loop_; std::shared_ptr state_; + Wakeup wakeup_; }; // ============================================================================== @@ -256,6 +267,8 @@ class GroupAudio { /// ahead of cell next's first sample, so a frame decodes to its own cells uint32_t lead{0}; std::unique_ptr encoder; + /// Built with the encoder, a slot of its largest frame for each frame alive at once + SharedPtr payloads; /// A resampler built for input at resampler_rate; kept while a segment at the output's /// rate needs none std::unique_ptr resampler; @@ -311,6 +324,10 @@ class GroupAudio { std::unique_ptr analysis; /// The most bytes one cell's messages take size_t cell_bytes{0}; + /// A slot of cell_bytes for each cell alive at once + SharedPtr payloads; + /// One cell's messages, room for cell_bytes, before they go to their slot + std::vector messages; bool failed{false}; /// Its input in the current epoch has started bool begun{false}; @@ -385,6 +402,10 @@ class GroupAudio { bool cross(Instance& instance, const JobView& view, size_t segment); /// At the drain position: flushes the resampler and cuts the output there bool finish_drain(Instance& instance, uint64_t drain_end); + /// Reserves a slot for the next frame; posts PayloadsFull when none is left and the last try + /// found one + /// @return false while every slot is taken + bool reserve_slot(PayloadSlots& payloads, sched::InstanceId instance); /// The output index of cell index's first sample static int64_t frame_start(const Instance& instance, int64_t index); /// Appends count output samples, dropping those before cell next's first and the lead @@ -396,7 +417,7 @@ class GroupAudio { /// its last sample up to it /// @return false when that would repeat more than MAX_BOUNDARY_EXTENSION samples bool cut_to(Instance& instance, int64_t total); - /// Encodes the instance's next frame from pending and posts it + /// Encodes the instance's next frame from pending into the slot produce() reserved and posts it /// @param samples the frame's output samples: its frame length, or a drained epoch's last /// frame's /// @return false when the encoder refused diff --git a/src/payload_slots.cpp b/src/payload_slots.cpp new file mode 100644 index 0000000..268c18f --- /dev/null +++ b/src/payload_slots.cpp @@ -0,0 +1,117 @@ +// 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. + +#include "payload_slots.h" + +#include +#include +#include + +namespace sendspin::server { + +// ============================================================================== +// FramePayload +// ============================================================================== + +FramePayload::FramePayload(SharedPtr slots, const uint8_t* data, size_t size, + uint32_t slot) noexcept + : slots_(std::move(slots)), data_(data), size_(size), slot_(slot) {} + +FramePayload::~FramePayload() { + this->reset(); +} + +FramePayload::FramePayload(FramePayload&& other) noexcept + : slots_(std::move(other.slots_)), + data_(std::exchange(other.data_, nullptr)), + size_(std::exchange(other.size_, 0)), + slot_(other.slot_) {} + +FramePayload& FramePayload::operator=(FramePayload&& other) noexcept { + if (this != &other) { + this->reset(); + this->slots_ = std::move(other.slots_); + this->data_ = std::exchange(other.data_, nullptr); + this->size_ = std::exchange(other.size_, 0); + this->slot_ = other.slot_; + } + return *this; +} + +void FramePayload::reset() noexcept { + if (this->slots_) { + this->slots_->release(this->slot_); + this->slots_.reset(); + } + this->data_ = nullptr; + this->size_ = 0; +} + +// ============================================================================== +// PayloadSlots +// ============================================================================== + +SharedPtr PayloadSlots::create(size_t slot_bytes, size_t slots) { + if (slot_bytes == 0 || slots == 0 || slots >= NONE || slot_bytes > SIZE_MAX / slots) { + return nullptr; + } + try { + PlacedArray bytes = + make_placed_array(slot_bytes * slots, MemoryLocation::PREFER_EXTERNAL); + Ring free(slots, slots, MemoryLocation::PREFER_EXTERNAL); + for (uint32_t slot = 0; slot < slots; ++slot) { + static_cast(free.try_push(slot)); + } + return make_shared_external(std::move(bytes), std::move(free), slot_bytes, + slots); + } catch (const std::bad_alloc&) { + return nullptr; + } +} + +PayloadSlots::PayloadSlots(PlacedArray bytes, Ring free, size_t slot_bytes, + size_t slots) noexcept + : bytes_(std::move(bytes)), free_(std::move(free)), slot_bytes_(slot_bytes), slots_(slots) {} + +std::span PayloadSlots::reserve() { + if (this->reserved_ == NONE) { + const std::lock_guard lock(this->mutex_); + if (!this->free_.empty()) { + this->reserved_ = this->free_.front(); + this->free_.pop_front(); + } + } + this->starved_ = this->reserved_ == NONE; + if (this->starved_) { + return {}; + } + return {this->bytes_.get() + (size_t{this->reserved_} * this->slot_bytes_), this->slot_bytes_}; +} + +FramePayload PayloadSlots::commit(size_t size) { + assert(this->reserved_ != NONE && size <= this->slot_bytes_); + const uint32_t slot = std::exchange(this->reserved_, NONE); + return {this->shared_from_this(), this->bytes_.get() + (size_t{slot} * this->slot_bytes_), size, + slot}; +} + +void PayloadSlots::release(uint32_t slot) noexcept { + const std::lock_guard lock(this->mutex_); + // Holds every slot, so a return always fits + const bool returned = this->free_.try_push(slot); + assert(returned); + static_cast(returned); +} + +} // namespace sendspin::server diff --git a/src/payload_slots.h b/src/payload_slots.h new file mode 100644 index 0000000..4d65012 --- /dev/null +++ b/src/payload_slots.h @@ -0,0 +1,86 @@ +// 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. + +/// @file payload_slots.h +/// @brief PayloadSlots: one pipeline instance's frame payloads, in fixed slots placed when the +/// instance is built and returned from any thread + +#pragma once + +#include "frame_payload.h" +#include "memory.h" +#include "ring.h" +#include "shared.h" + +#include +#include +#include +#include + +namespace sendspin::server { + +/// @brief Slots of a format's largest frame, one per frame alive at once, placed once, in PSRAM +/// where there is one. The instance (its one producer) reserves a slot, writes the frame +/// into it without the lock and commits it as a FramePayload; destroying the payload returns the +/// slot, in any order and from any thread. The mutex is a leaf, held only to take or return a slot. +class PayloadSlots : public EnableSharedFromThis { +public: + /// @return null when neither memory has room + static SharedPtr create(size_t slot_bytes, size_t slots); + + /// @brief Use create() + PayloadSlots(PlacedArray bytes, Ring free, size_t slot_bytes, + size_t slots) noexcept; + + /// @brief Producer only. A slot's room for one frame, the same slot until it is committed; + /// empty when every slot is taken, which sets starved() + std::span reserve(); + + /// @brief Producer only. The first size bytes of the slot reserve() gave, which it must have + /// given since the last commit + FramePayload commit(size_t size); + + /// @brief Producer only: the last reserve() found no slot + bool starved() const { + return this->starved_; + } + + size_t slot_bytes() const { + return this->slot_bytes_; + } + + size_t slots() const { + return this->slots_; + } + +private: + friend class FramePayload; + + static constexpr uint32_t NONE = UINT32_MAX; + + /// Any thread + void release(uint32_t slot) noexcept; + + std::mutex mutex_; + PlacedArray bytes_; + /// The slots not taken; its limit is every slot, so a return never grows it + Ring free_; + const size_t slot_bytes_; + const size_t slots_; + /// Producer only: the slot reserve() gave, until it is committed + uint32_t reserved_{NONE}; + bool starved_{false}; +}; + +} // namespace sendspin::server diff --git a/src/player_stream.cpp b/src/player_stream.cpp index d1ab531..c52a733 100644 --- a/src/player_stream.cpp +++ b/src/player_stream.cpp @@ -306,9 +306,9 @@ PackedChunk PlayerStream::take_chunk(const StreamSource* source, size_t max_mess bytes.resize(sched::CHUNK_HEADER_BYTES + payload); size_t offset = sched::CHUNK_HEADER_BYTES; for (size_t i = 0; i < frames; ++i) { - const std::vector& from = + const std::span from = ring->at(this->sent_.epoch, this->cursor_.next + (static_cast(i) * cells)) - ->payload; + ->payload.bytes(); std::memcpy(bytes.data() + offset, from.data(), from.size()); offset += from.size(); } diff --git a/src/profile.h b/src/profile.h index 62c1086..0b9530c 100644 --- a/src/profile.h +++ b/src/profile.h @@ -45,6 +45,9 @@ constexpr int64_t STAGING_RING_US = WRITE_AHEAD_MAX_US + STAGING_LEAD_ALLOWANCE_ static_assert(STAGING_RING_US > WRITE_AHEAD_MAX_US, "admission must stay within the staging ring"); /// @brief ...or this many bytes, whichever is less (in PSRAM on the embedded profile) constexpr size_t STAGING_RING_MAX_BYTES = (EMBEDDED ? size_t{2} : size_t{64}) * 1024 * 1024; +/// @brief How long core may stall without pruning a pipeline instance's frames before the +/// instance's payload slots run out and it waits; past it, the wait costs a rebase +constexpr int64_t PAYLOAD_LAG_US = EMBEDDED ? 500000 : 1000000; /// @brief Accepted connections in their WebSocket upgrade or handshake /// (connection_limits::MAX_ACCEPTED_HANDSHAKES); lwIP's socket table is small on the embedded diff --git a/src/sched/frames.h b/src/sched/frames.h index 9740dfb..80a735e 100644 --- a/src/sched/frames.h +++ b/src/sched/frames.h @@ -18,6 +18,7 @@ #pragma once +#include "frame_payload.h" #include "ring.h" #include "sched/constants.h" #include "sched/types.h" @@ -25,7 +26,6 @@ #include #include #include -#include namespace sendspin::server::sched { @@ -60,7 +60,7 @@ struct Frame { int64_t duration_us{0}; /// Per channel: `r * m / 50`, fewer only in a drained epoch's last frame uint32_t samples{0}; - std::vector payload; + FramePayload payload; }; /// @brief What packing reads of a frame @@ -73,21 +73,29 @@ struct FrameView { FrameView view_of(const Frame& frame); -/// @brief The frames of `cells` cells a ring has room for before it first grows: those over the -/// staging ring's span (the write-ahead and the lead the profile allows), and one more +/// @brief The frames of `cells` cells over the staging ring's span (the write-ahead and the lead +/// the profile allows), and one more constexpr size_t ring_frames(uint32_t cells) { return static_cast(profile::STAGING_RING_US / (int64_t{cells > 0 ? cells : 1} * CELL_US)) + 1; } +/// @brief The frames of `cells` cells alive at once, each in a payload slot: ring_frames(), and +/// those up to profile::PAYLOAD_LAG_US past their play time that core has not pruned +constexpr size_t payload_slots(uint32_t cells) { + const int64_t frame_us = int64_t{cells > 0 ? cells : 1} * CELL_US; + return ring_frames(cells) + + static_cast((profile::PAYLOAD_LAG_US + frame_us - 1) / frame_us); +} + /// @brief Core's store of one pipeline instance's frames in one epoch, keyed by cell index. The /// instance produces every `m`-th index from its start index, so the ring has no holes. Frames are /// dropped once `ts <= now`, which is late for every player since output delays are not negative. class FrameRing { public: FrameRing(InstanceId instance, uint32_t cells) - : frames_(ring_frames(cells), Ring::UNLIMITED, MemoryLocation::PREFER_EXTERNAL), + : frames_(payload_slots(cells), Ring::UNLIMITED, MemoryLocation::PREFER_EXTERNAL), instance_(instance), cells_(cells) {} @@ -152,7 +160,8 @@ class FrameRing { private: /// Bounded by the send path's horizon: a frame is dropped once its `ts` passes, and admission - /// keeps the input within the group lead and the write-ahead ahead + /// keeps the input within the group lead and the write-ahead ahead; at most payload_slots(), + /// since each frame holds one of its instance's payload slots Ring frames_; InstanceId instance_; EpochId epoch_; diff --git a/src/visualizer_stream.cpp b/src/visualizer_stream.cpp index 0f8b6d7..319a00a 100644 --- a/src/visualizer_stream.cpp +++ b/src/visualizer_stream.cpp @@ -19,6 +19,7 @@ #include "sched/timeline.h" #include +#include #include namespace sendspin::server { @@ -26,7 +27,7 @@ namespace sendspin::server { namespace { /// The big-endian timestamp of the message at offset -int64_t message_ts(const std::vector& payload, size_t offset) { +int64_t message_ts(std::span payload, size_t offset) { return static_cast(read_be(&payload[offset + 1])); } @@ -122,12 +123,13 @@ bool VisualizerStream::prepare(StreamSource* source, uint32_t markers_queued, in const uint32_t bins = bins_of(this->cursor_config()); for (const sched::Frame* cell = ring->at(this->sent_.epoch, this->cursor_.next); cell != nullptr; cell = ring->at(this->sent_.epoch, this->cursor_.next)) { - if (this->offset_ < cell->payload.size()) { - const int64_t ts = message_ts(cell->payload, this->offset_); + const std::span payload = cell->payload.bytes(); + if (this->offset_ < payload.size()) { + const int64_t ts = message_ts(payload, this->offset_); if (ts > now_us && ts >= this->cursor_.min_ts) { break; } - const uint8_t type = visualizer_type_of(cell->payload[this->offset_]); + const uint8_t type = visualizer_type_of(payload[this->offset_]); this->offset_ += visualizer_message_bytes(type, bins); continue; } @@ -297,7 +299,8 @@ OutboundPlaintext VisualizerStream::take_message(const StreamSource* source) { if (!next) { return message; } - const auto from = next->cell->payload.begin() + static_cast(next->offset); + const auto from = + next->cell->payload.bytes().begin() + static_cast(next->offset); reserve_plaintext(message.bytes, next->bytes); message.bytes.assign(from, from + static_cast(next->bytes)); // Counted until its own timestamp @@ -336,10 +339,11 @@ std::optional VisualizerStream::message_at(const sche index == this->cursor_.next && ring.instance() == this->offset_ring_ ? this->offset_ : 0; for (const sched::Frame* cell = ring.at(this->sent_.epoch, index); cell != nullptr; cell = ring.at(this->sent_.epoch, ++index), offset = 0) { - if (offset < cell->payload.size()) { - const uint8_t type = visualizer_type_of(cell->payload[offset]); + const std::span payload = cell->payload.bytes(); + if (offset < payload.size()) { + const uint8_t type = visualizer_type_of(payload[offset]); return Message{cell, offset, visualizer_message_bytes(type, bins), - message_ts(cell->payload, offset)}; + message_ts(payload, offset)}; } } return std::nullopt; diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 36632ec..62732b7 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -51,6 +51,7 @@ add_executable(sendspin_server_cpp_tests test_pairing.cpp test_pairing_crypto.cpp test_pairing_crypto_internals.cpp + test_payload_slots.cpp test_pcm_convert.cpp test_peer.cpp test_pipeline_stages.cpp @@ -163,7 +164,9 @@ set_source_files_properties(test_dispatcher_internals.cpp PROPERTIES COMPILE_OPT # resampler. It calls GroupAudio::cut_to() on a GroupAudio::Instance of its own to extend a short # stretch, which the stages never give, and sets HostView::p_pause_ to resume from a chosen # position, which pause() takes from the time. It reads HostView::rebase_latency_us_, the `L_p` -# the pipelines publish for admission, which only shows in admission's bounds. +# the pipelines publish for admission, which only shows in admission's bounds. It gives an instance +# and an analysis two payload slots each (GroupAudio::instances_, analyses_), since admission keeps +# the frames alive below the slots they are built with. set_source_files_properties(test_group_audio_internals.cpp PROPERTIES COMPILE_OPTIONS -fno-access-control) # test_group_internals.cpp wraps a group's pipeline job (Group::audio_) with group_job.h, to hold @@ -176,7 +179,11 @@ set_source_files_properties(test_group_audio_internals.cpp PROPERTIES COMPILE_OP # rather than after 10 s, and take from it to read the counts, which only the log line shows # otherwise. It calls Group::on_wake() as the group's wakeup does, since a write's signal can fold # into a pipeline job's that is not posted yet, and reads HostView::slew_ and next_slew_ under -# HostView::mutex_, as only their storage shows that core freed the slew a write left it. +# HostView::mutex_, as only their storage shows that core freed the slew a write left it. Its +# payload tests replace Group::audio_ with a task that counts its runs, fill rings_ directly and +# call on_payloads_full(), on_epoch_cut(), on_epoch_started() and end_run() as the FIFO does, +# reading Group::full_, the instances held for the timer; the run's end empties pipelines_ first, +# so its recompute keeps the rings. # group_job.h, which only the files given the flag include, waits for a task between runs under # its pool's mutex (AudioWorkers::Task::pool_, running_ and queued_) to replace its job # (Task::job_), and reads a group's pipelines from the job (Group::PipelineJob::audio). diff --git a/tests/frame_payloads.h b/tests/frame_payloads.h new file mode 100644 index 0000000..64e0fd7 --- /dev/null +++ b/tests/frame_payloads.h @@ -0,0 +1,79 @@ +// 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. + +/// @file frame_payloads.h +/// @brief Frame payloads a test owns: bytes in a slot of their own, so a frame a test holds never +/// keeps a pipeline instance's slot + +#pragma once + +#include "frame_encoder.h" +#include "payload_slots.h" +#include "sched/frames.h" +#include + +#include +#include +#include +#include +#include + +namespace sendspin::server::test { + +/// bytes as a payload in a one-slot store of its own +inline FramePayload payload_of(std::span bytes) { + const SharedPtr slots = + PayloadSlots::create(std::max(1, bytes.size()), 1); + EXPECT_TRUE(slots); + const std::span room = slots->reserve(); + std::copy(bytes.begin(), bytes.end(), room.begin()); + return slots->commit(bytes.size()); +} + +/// bytes zero bytes as a payload in a store of its own +inline FramePayload zero_payload(size_t bytes) { + return payload_of(std::vector(bytes, 0)); +} + +/// frame with its payload copied into a store of its own +inline sched::Frame copy_of(const sched::Frame& frame) { + sched::Frame copy; + copy.epoch = frame.epoch; + copy.index = frame.index; + copy.ts_us = frame.ts_us; + copy.duration_us = frame.duration_us; + copy.samples = frame.samples; + copy.payload = payload_of(frame.payload.bytes()); + return copy; +} + +/// Bytes copied out, to compare +inline std::vector bytes_of(std::span bytes) { + return {bytes.begin(), bytes.end()}; +} + +inline std::vector bytes_of(const FramePayload& payload) { + return bytes_of(payload.bytes()); +} + +/// FrameEncoder::encode() into out, sized to what it wrote +inline bool encode_into(FrameEncoder& encoder, std::span input, + SampleFormat in_format, std::vector& out) { + out.resize(encoder.max_frame_bytes()); + const std::optional written = encoder.encode(input, in_format, out); + out.resize(written.value_or(0)); + return written.has_value(); +} + +} // namespace sendspin::server::test diff --git a/tests/freed_blocks.h b/tests/freed_blocks.h index e4037d8..cdde27c 100644 --- a/tests/freed_blocks.h +++ b/tests/freed_blocks.h @@ -14,7 +14,8 @@ /// @file freed_blocks.h /// @brief FreedBlockScan, which counts the heap blocks the test thread frees while it is alive that -/// still hold a secret, through the sanitizer's free hook; only in a build with ASan or TSan +/// still hold a secret, and AllocationCount, which counts the blocks a thread allocates while it +/// counts, through the sanitizer's free and malloc hooks; only in a build with ASan or TSan #pragma once @@ -53,6 +54,10 @@ inline std::atomic needle{nullptr}; /// The thread whose frees the live scan counts inline std::atomic scanning_thread{}; inline std::atomic holding{0}; +/// The thread whose allocations are counted while counting is set +inline std::atomic counting_thread{}; +inline std::atomic counting{false}; +inline std::atomic allocated{0}; #ifdef SENDSPIN_TEST_FREE_HOOKS // Runs before the allocator takes the block back, so its bytes are still there. A container's @@ -84,24 +89,37 @@ __attribute__((no_sanitize_address, no_sanitize_thread)) inline void on_free( } } -inline void on_malloc(const volatile void* /*pointer*/, size_t /*size*/) {} +inline void on_malloc(const volatile void* /*pointer*/, size_t /*size*/) { + if (counting.load() && pthread_equal(pthread_self(), counting_thread.load()) != 0) { + allocated.fetch_add(1); + } +} #endif } // namespace freed_blocks +namespace freed_blocks { + +/// @brief Whether the build has the hooks; a test that needs them skips itself otherwise +inline bool hooks_available() { +#ifdef SENDSPIN_TEST_FREE_HOOKS + // The sanitizer takes a few hook pairs and never removes one: installed once per process + static const bool installed = + __sanitizer_install_malloc_and_free_hooks(&on_malloc, &on_free) != 0; + return installed; +#else + return false; +#endif +} + +} // namespace freed_blocks + /// Counts the blocks the calling thread frees while it is alive that hold `secret`. One at a time. class FreedBlockScan { public: /// @brief Whether the build has the hook; a test that needs it skips itself otherwise static bool available() { -#ifdef SENDSPIN_TEST_FREE_HOOKS - // The sanitizer takes a few hook pairs and never removes one: installed once per process - static const bool installed = __sanitizer_install_malloc_and_free_hooks( - &freed_blocks::on_malloc, &freed_blocks::on_free) != 0; - return installed; -#else - return false; -#endif + return freed_blocks::hooks_available(); } explicit FreedBlockScan(std::string_view secret) : secret_(secret) { @@ -123,3 +141,35 @@ class FreedBlockScan { private: std::string_view secret_; }; + +/// Counts the blocks the calling thread allocates between start() and stop(). One at a time. +class AllocationCount { +public: + /// @brief Whether the build has the hook; a test that needs it skips itself otherwise + static bool available() { + return freed_blocks::hooks_available(); + } + + AllocationCount() { + freed_blocks::allocated.store(0); + freed_blocks::counting_thread.store(pthread_self()); + } + ~AllocationCount() { + this->stop(); + } + AllocationCount(const AllocationCount&) = delete; + AllocationCount& operator=(const AllocationCount&) = delete; + + void start() { + freed_blocks::counting.store(true); + } + + void stop() { + freed_blocks::counting.store(false); + } + + /// @brief The blocks allocated while counting so far + int blocks() const { + return freed_blocks::allocated.load(); + } +}; diff --git a/tests/group_pipelines.h b/tests/group_pipelines.h index 7278800..12e2139 100644 --- a/tests/group_pipelines.h +++ b/tests/group_pipelines.h @@ -14,13 +14,14 @@ /// @file group_pipelines.h /// @brief Pipelines, a group's pipelines over an input of any format and kind, with their jobs and -/// the FIFO's handler run on the test thread, and what the tests read of their output: frames, +/// the FIFO's `Wakeup` run on the test thread, and what the tests read of their output: frames, /// samples, a reference resampling, a test tone #pragma once #include "clock.h" #include "dispatcher.h" +#include "frame_payloads.h" #include "group_audio.h" #include "host_view.h" #include "resampler_reference.h" @@ -33,9 +34,11 @@ #include #include #include +#include #include #include #include +#include #include #include #include @@ -48,12 +51,21 @@ inline std::map> frames_by_instance std::map> out; for (const GroupEvent& event : events) { if (const auto* ready = std::get_if(&event)) { - out[ready->instance].push_back(ready->frame); + out[ready->instance].push_back(copy_of(ready->frame)); } } return out; } +/// Appends event, a frame in a store of its own, so a frame the test holds returns its instance's +/// slot +inline void keep_event(std::vector& events, GroupEvent& event) { + if (auto* ready = std::get_if(&event)) { + ready->frame = copy_of(ready->frame); + } + events.push_back(std::move(event)); +} + /// A group's pipelines over an input of any format and kind struct Pipelines { /// The group lead `G` @@ -66,7 +78,7 @@ struct Pipelines { : view(std::make_shared(GroupId{1}, InputId{1}, options, GroupParams{LEAD_US, 0, write_ahead_us}, std::make_shared(), nullptr)), - fifo(this->loop, [this](GroupEvent& event) { this->events.push_back(event); }), + fifo(this->loop, [this](GroupEvent& event) { keep_event(this->events, event); }), audio(this->view, this->fifo.producer(), this->slot, ResamplerQuality::MEDIUM) {} /// @param late_us how far past now_us() the jobs run @@ -86,7 +98,7 @@ struct Pipelines { }; /// A little-endian signed sample of bytes bytes -inline int32_t sample_at(const std::vector& payload, size_t index, size_t bytes) { +inline int32_t sample_at(std::span payload, size_t index, size_t bytes) { uint32_t value = 0; for (size_t i = 0; i < bytes; ++i) { value |= uint32_t{payload[(index * bytes) + i]} << (8 * i); @@ -99,7 +111,7 @@ inline int32_t sample_at(const std::vector& payload, size_t index, size inline std::vector concatenated(const std::vector& frames) { std::vector out; for (const sched::Frame& frame : frames) { - out.insert(out.end(), frame.payload.begin(), frame.payload.end()); + out.insert(out.end(), frame.payload.bytes().begin(), frame.payload.bytes().end()); } return out; } @@ -206,7 +218,8 @@ inline void expect_resampled(const std::vector& out, size_t at, inline std::vector run_all(Pipelines& p) { std::vector events = p.run(); for (std::vector more = p.run(); !more.empty(); more = p.run()) { - events.insert(events.end(), more.begin(), more.end()); + events.insert(events.end(), std::make_move_iterator(more.begin()), + std::make_move_iterator(more.end())); } return events; } diff --git a/tests/test_frame_encoder.cpp b/tests/test_frame_encoder.cpp index 9288016..e89c968 100644 --- a/tests/test_frame_encoder.cpp +++ b/tests/test_frame_encoder.cpp @@ -18,6 +18,7 @@ #include "flac_reader.h" #include "frame_encoder.h" +#include "frame_payloads.h" #include "opus_reader.h" #include "pcm_convert.h" #include @@ -109,7 +110,11 @@ TEST(OutputShape, GivesEachEncodersFrameBound) { EXPECT_EQ(shape.encoder_available, row.available); EXPECT_EQ(shape.max_frame_bytes, row.max_frame_bytes); EXPECT_EQ(shape.layout.samples, row.format.sample_rate / 50); - EXPECT_EQ(static_cast(FrameEncoder::create(row.format)), row.available); + const std::unique_ptr encoder = FrameEncoder::create(row.format); + EXPECT_EQ(static_cast(encoder), row.available); + if (encoder) { + EXPECT_EQ(static_cast(encoder->max_frame_bytes()), row.max_frame_bytes); + } } // A FLAC bound is a real one: above the PCM frame of the same format EXPECT_GT(flac_bound(FLAC48_16), 960 * 4); @@ -126,6 +131,27 @@ TEST(OutputShape, GivesEachEncodersFrameBound) { EXPECT_FALSE(sched::producible(deep, 100000)); } +// A PCM or FLAC frame is written into room of the frame bound, never past it; room one byte short +// of the bound is refused, as micro-flac takes no less than its worst case +TEST(FrameEncoderOutput, RefusesRoomShortOfTheFrameBound) { + const sched::AudioFormat formats[] = {{sched::Codec::PCM, 48000, 2, 24}, FLAC48_16}; + const Bytes input = float_signal(48000, 960, 0); + for (const sched::AudioFormat& format : formats) { + SCOPED_TRACE(format.codec == sched::Codec::PCM ? "PCM" : "FLAC"); + // Control: room of the bound takes the frame + const std::unique_ptr whole = FrameEncoder::create(format); + ASSERT_TRUE(whole); + Bytes room(whole->max_frame_bytes()); + const std::optional written = whole->encode(input, SampleFormat::F32, room); + ASSERT_TRUE(written); + EXPECT_GT(*written, 0U); + EXPECT_LE(*written, whole->max_frame_bytes()); + const std::unique_ptr short_of = FrameEncoder::create(format); + Bytes less(short_of->max_frame_bytes() - 1); + EXPECT_FALSE(short_of->encode(input, SampleFormat::F32, less)); + } +} + // spec player/v1.md "Codec framing": FLAC's codec_header is `fLaC` and STREAMINFO in standard // Base64 with padding, for the stream's rate, channels, depth and block size; PCM has none TEST(CodecHeader, IsTheStreaminfoOfTheFormatInStandardBase64) { @@ -160,7 +186,7 @@ TEST(FrameEncoderPcm, IsTheQuantizedInput) { Bytes expected(960 * 2 * 3); sendspin::server::convert_pcm(input, SampleFormat::F32, 24, expected); Bytes payload; - ASSERT_TRUE(encoder->encode(input, SampleFormat::F32, payload)); + ASSERT_TRUE(sendspin::server::test::encode_into(*encoder, input, SampleFormat::F32, payload)); EXPECT_EQ(payload, expected); } @@ -192,7 +218,7 @@ TEST(FrameEncoderFlac, EncodesCellsThatDecodeToTheQuantizedInput) { Bytes expected(frames * 2 * out_bytes); sendspin::server::convert_pcm(input, row.input, row.format.bit_depth, expected); Bytes payload; - ASSERT_TRUE(encoder->encode(input, row.input, payload)); + ASSERT_TRUE(sendspin::server::test::encode_into(*encoder, input, row.input, payload)); EXPECT_LT(payload.size(), expected.size()); const auto decoded = reader.decode(payload); ASSERT_TRUE(decoded); @@ -211,7 +237,7 @@ TEST(FrameEncoderFlac, EncodesCellsThatDecodeToTheQuantizedInput) { // The stream ended until the next reset Bytes payload; const Bytes more = float_signal(row.format.sample_rate, block, 0); - EXPECT_FALSE(encoder->encode(more, row.input, payload)); + EXPECT_FALSE(sendspin::server::test::encode_into(*encoder, more, row.input, payload)); EXPECT_TRUE(payload.empty()); encoder->reset(); check(0, block, 0); @@ -241,7 +267,8 @@ TEST(OpusFrameEncoder, DecodesToTheInputBehindItsLookahead) { for (size_t k = 0; k < FRAMES; ++k) { Bytes payload; const std::span frame(input.data() + (k * block * 4), block * 4); - ASSERT_TRUE(encoder->encode(frame, SampleFormat::S16, payload)); + ASSERT_TRUE( + sendspin::server::test::encode_into(*encoder, frame, SampleFormat::S16, payload)); EXPECT_LE(payload.size(), 1276U); const std::optional> samples = reader.decode(payload); ASSERT_TRUE(samples); @@ -265,13 +292,14 @@ TEST(OpusFrameEncoder, DecodesToTheInputBehindItsLookahead) { // A short last frame is a whole packet, and reset() starts a new stream Bytes payload; - ASSERT_TRUE(encoder->encode(std::span(input.data(), 100 * 4), SampleFormat::S16, payload)); + ASSERT_TRUE(sendspin::server::test::encode_into(*encoder, std::span(input.data(), 100 * 4), + SampleFormat::S16, payload)); const std::optional> last = reader.decode(payload); ASSERT_TRUE(last); EXPECT_EQ(last->size(), size_t{block} * 2); encoder->reset(); - ASSERT_TRUE( - encoder->encode(std::span(input.data(), block * 4), SampleFormat::S16, payload)); + ASSERT_TRUE(sendspin::server::test::encode_into( + *encoder, std::span(input.data(), block * 4), SampleFormat::S16, payload)); EXPECT_FALSE(payload.empty()); } } @@ -299,7 +327,8 @@ TEST(OpusFrameEncoder, DecodesAtTheInputsLevelFromAnyPacket) { const auto encode = [&](uint64_t position) { const Bytes input = sendspin::server::test::tone_frames(position, BLOCK, RATE); Bytes payload; - EXPECT_TRUE(encoder->encode(input, SampleFormat::S16, payload)); + EXPECT_TRUE( + sendspin::server::test::encode_into(*encoder, input, SampleFormat::S16, payload)); return payload; }; // The tones' level, which every frame's input has @@ -334,7 +363,8 @@ TEST(OpusFrameEncoder, DecodesAtTheInputsLevelFromAnyPacket) { Bytes primed; { const Bytes input = sendspin::server::test::tone_frames(seek - lead, BLOCK, RATE); - ASSERT_TRUE(encoder->encode(input, SampleFormat::S16, primed)); + ASSERT_TRUE( + sendspin::server::test::encode_into(*encoder, input, SampleFormat::S16, primed)); } EXPECT_NEAR(heard_db(first, primed), reference_db, 1.0) << "seek, frame 0"; for (uint64_t k = 1; k < 10; ++k) { @@ -363,7 +393,8 @@ TEST(OpusFrameEncoder, CarriesSixtyFourKilobitsPerChannel) { stereo.begin() + static_cast(at + (2 * channels))); } Bytes payload; - ASSERT_TRUE(encoder->encode(input, SampleFormat::S16, payload)); + ASSERT_TRUE( + sendspin::server::test::encode_into(*encoder, input, SampleFormat::S16, payload)); bytes += payload.size(); } // 50 packets a second: 160 bytes per channel per packet at 64 kbit/s diff --git a/tests/test_group_audio.cpp b/tests/test_group_audio.cpp index 0a43edc..03956e5 100644 --- a/tests/test_group_audio.cpp +++ b/tests/test_group_audio.cpp @@ -21,6 +21,7 @@ #include "dispatcher.h" #include "flac_reader.h" #include "frame_encoder.h" +#include "freed_blocks.h" #include "group_audio.h" #include "group_pipelines.h" #include "host_view.h" @@ -41,6 +42,7 @@ #include #include #include +#include #include #include #include @@ -67,6 +69,7 @@ using sendspin::server::InputId; using sendspin::server::InstanceFinal; using sendspin::server::now_us; using sendspin::server::SampleFormat; +using sendspin::server::test::bytes_of; using sendspin::server::test::concatenated; using sendspin::server::test::drained_last_cell; using sendspin::server::test::expect_resampled; @@ -98,7 +101,10 @@ class GroupAudioTest : public testing::Test { sendspin::server::InputOptions{InputFormat{RATE, 1, SampleFormat::S16}}, sendspin::server::GroupParams{LEAD_US, 0, 1000000}, std::make_shared(), nullptr)), - fifo(this->loop, [this](GroupEvent& event) { this->events.push_back(event); }), + fifo(this->loop, + [this](GroupEvent& event) { + sendspin::server::test::keep_event(this->events, event); + }), audio(this->view, this->fifo.producer(), this->slot, sendspin::server::ResamplerQuality::MEDIUM) { // One 24-bit mono instance from the start @@ -130,7 +136,7 @@ class GroupAudioTest : public testing::Test { static std::string kinds(const std::vector& events) { std::string out; for (const GroupEvent& event : events) { - out += "CSEADFfP"[event.index()]; + out += "CSEADFfPW"[event.index()]; } return out; } @@ -159,8 +165,8 @@ TEST_F(GroupAudioTest, CutsFramesOnTheCellGridThroughADrain) { EXPECT_EQ(frame.duration_us, 20000); // 16-bit samples shifted up to 24 bits, little-endian ASSERT_EQ(frame.payload.size(), CELL * 3); - EXPECT_EQ(frame.payload[3], 0); - EXPECT_EQ(frame.payload[4], static_cast((k * CELL) + 1)); + EXPECT_EQ(frame.payload.bytes()[3], 0); + EXPECT_EQ(frame.payload.bytes()[4], static_cast((k * CELL) + 1)); } // DrainStarted once the last frame is produced, so no rebase follows @@ -209,11 +215,11 @@ TEST_F(GroupAudioTest, EncodesFlacFramesBesidePcmOnes) { EXPECT_EQ(encoded[i].duration_us, same->duration_us); // Compressed: the ramp is predictable EXPECT_LT(encoded[i].payload.size(), same->payload.size()); - const auto decoded = decoder.decode(encoded[i].payload); + const auto decoded = decoder.decode(encoded[i].payload.bytes()); ASSERT_TRUE(decoded); ASSERT_EQ(decoded->size(), 1U); EXPECT_EQ(decoded->front().number, numbers[i]); - EXPECT_EQ(decoded->front().pcm, same->payload); + EXPECT_EQ(decoded->front().pcm, bytes_of(same->payload)); } }; @@ -245,7 +251,8 @@ TEST_F(GroupAudioTest, EncodesFlacFramesBesidePcmOnes) { EXPECT_EQ(kinds(events), "FFFFFFDfff"); for (auto& [instance, list] : frames_by_instance(events)) { auto& all = instance == JOINER ? joined[instance] : second[instance]; - all.insert(all.end(), list.begin(), list.end()); + all.insert(all.end(), std::make_move_iterator(list.begin()), + std::make_move_iterator(list.end())); } ASSERT_EQ(second[PCM].size(), 4U); EXPECT_EQ(second[PCM].back().samples, 100U); @@ -283,7 +290,8 @@ TEST_F(GroupAudioTest, FollowsFlushPauseAndResume) { ASSERT_EQ(kinds(events), "SFF"); EXPECT_NE(std::get(events[0]).run, first.run); // The same audio again, from the start of the resumed epoch - EXPECT_EQ(std::get(events[1]).frame.payload[1], static_cast(500 & 0xFF)); + EXPECT_EQ(std::get(events[1]).frame.payload.bytes()[1], + static_cast(500 & 0xFF)); EXPECT_EQ(std::get(events[1]).frame.index, 0); this->view->stop(); @@ -377,7 +385,7 @@ using sendspin::server::PipelineRequest; std::string kinds_of(const std::vector& events) { std::string out; for (const GroupEvent& event : events) { - out += "CSEADFfP"[event.index()]; + out += "CSEADFfPW"[event.index()]; } return out; } @@ -499,10 +507,10 @@ TEST(GroupAudioPipelines, CutEveryFormatOnTheCellGridThroughADrain) { const std::vector& pcm_frames = frames[sched::InstanceId{21}]; const std::vector& flac_frames = frames[sched::InstanceId{22}]; for (size_t k = 0; k < flac_frames.size(); ++k) { - const auto decoded = reader.decode(flac_frames[k].payload); + const auto decoded = reader.decode(flac_frames[k].payload.bytes()); ASSERT_TRUE(decoded); ASSERT_EQ(decoded->size(), 1U); - EXPECT_EQ(decoded->front().pcm, pcm_frames[k].payload) << k; + EXPECT_EQ(decoded->front().pcm, bytes_of(pcm_frames[k].payload)) << k; } } @@ -562,10 +570,10 @@ TEST(GroupAudioPipelines, CutsTwoCellFramesAt11025Hz) { for (size_t i = 0; i < frames[flac].size(); ++i) { EXPECT_EQ(frames[flac][i].index, frames[pcm][i].index); EXPECT_EQ(frames[flac][i].ts_us, frames[pcm][i].ts_us); - const auto decoded = reader.decode(frames[flac][i].payload); + const auto decoded = reader.decode(frames[flac][i].payload.bytes()); ASSERT_TRUE(decoded); ASSERT_EQ(decoded->size(), 1U); - EXPECT_EQ(decoded->front().pcm, frames[pcm][i].payload) << i; + EXPECT_EQ(decoded->front().pcm, bytes_of(frames[pcm][i].payload)) << i; } ASSERT_EQ(frames[resampled].size(), 52U); @@ -655,7 +663,7 @@ std::map payloads_from(const std::vector& frames, std::map out; for (const sched::Frame& frame : frames) { if (frame.index >= index) { - out[frame.index] = frame.payload; + out[frame.index] = bytes_of(frame.payload); } } return out; @@ -883,11 +891,12 @@ void run_segments_seed(uint64_t seed) { } total += list[i].samples; if (request.format.codec == sched::Codec::FLAC) { - const auto decoded = reader.decode(list[i].payload); + const auto decoded = reader.decode(list[i].payload.bytes()); ASSERT_TRUE(decoded && decoded->size() == 1U) << i; out.insert(out.end(), decoded->front().pcm.begin(), decoded->front().pcm.end()); } else { - out.insert(out.end(), list[i].payload.begin(), list[i].payload.end()); + out.insert(out.end(), list[i].payload.bytes().begin(), + list[i].payload.bytes().end()); } } EXPECT_EQ(finals.at(request.instance), list.back().index); @@ -940,14 +949,14 @@ TEST_F(GroupAudioTest, DuplicatesAMonoInputIntoEveryChannel) { constexpr sched::InstanceId STEREO{30}; this->slot->publish({{STEREO, {}, {sched::Codec::PCM, RATE, 2, 16}, 0}}); this->write(2 * CELL, 100); - const std::vector frames = frames_by_instance(this->run())[STEREO]; + const std::vector frames = std::move(frames_by_instance(this->run())[STEREO]); ASSERT_EQ(frames.size(), 2U); for (const sched::Frame& frame : frames) { ASSERT_EQ(frame.payload.size(), CELL * 4); for (uint64_t i = 0; i < CELL; ++i) { const int32_t expected = static_cast(100 + (frame.index * CELL) + i); - ASSERT_EQ(sample_at(frame.payload, i * 2, 2), expected); - ASSERT_EQ(sample_at(frame.payload, (i * 2) + 1, 2), expected); + ASSERT_EQ(sample_at(frame.payload.bytes(), i * 2, 2), expected); + ASSERT_EQ(sample_at(frame.payload.bytes(), (i * 2) + 1, 2), expected); } } } @@ -983,7 +992,7 @@ std::optional impulse_time_us(const std::vector& frames, u int32_t peak = 0; for (const sched::Frame& frame : frames) { for (uint32_t i = 0; i < frame.samples; ++i) { - const int32_t value = std::abs(sample_at(frame.payload, i, 3)); + const int32_t value = std::abs(sample_at(frame.payload.bytes(), i, 3)); if (value > peak) { peak = value; at = static_cast(frame.ts_us) + (i * 1e6 / rate); @@ -1057,7 +1066,7 @@ TEST(GroupAudioPipelines, ResumesAResampledInstanceFromRealHistory) { p.run(); p.view->resume(); p.view->drain(); - std::vector frames = frames_by_instance(p.run())[resampled]; + std::vector frames = std::move(frames_by_instance(p.run())[resampled]); ASSERT_FALSE(frames.empty()); EXPECT_EQ(frames.front().index, 0); @@ -1101,14 +1110,15 @@ TEST(GroupAudioPipelines, CatchesUpAResampledInstanceAddedMidEpoch) { ASSERT_EQ(p.view->write(pcm.data(), 10 * 882).frames, 10 * 882U); std::vector events = p.run(); const sched::EpochId epoch = std::get(events.front()).epoch; - std::vector first = frames_by_instance(events)[running]; + std::vector first = std::move(frames_by_instance(events)[running]); p.slot->publish({{running, {}, pcm48, 0}, {caught_up, epoch, pcm48, 4}}); ASSERT_EQ(p.view->write(pcm.data() + (10 * 882 * 4), 10 * 882).frames, 10 * 882U); p.view->drain(); events = p.run(); std::map> frames = frames_by_instance(events); - first.insert(first.end(), frames[running].begin(), frames[running].end()); + first.insert(first.end(), std::make_move_iterator(frames[running].begin()), + std::make_move_iterator(frames[running].end())); ASSERT_FALSE(frames[caught_up].empty()); EXPECT_EQ(frames[caught_up].front().index, 4); for (const sched::Frame& frame : frames[caught_up]) { @@ -1116,7 +1126,7 @@ TEST(GroupAudioPipelines, CatchesUpAResampledInstanceAddedMidEpoch) { return f.index == frame.index; }); ASSERT_NE(same, first.end()); - EXPECT_EQ(frame.payload, same->payload) << frame.index; + EXPECT_EQ(bytes_of(frame.payload), bytes_of(same->payload)) << frame.index; } std::map finals = final_indices(events); EXPECT_EQ(drained_last_cell(events), 19); @@ -1366,7 +1376,7 @@ TEST(GroupAudioLive, TakesEveryWriteWhilePausedAndResumesAtTheHead) { EXPECT_EQ(std::get(events[0]).anchor.segments.front().position, head); const sched::Frame& frame = std::get(events[1]).frame; EXPECT_EQ(frame.index, 0); - EXPECT_EQ(sample_at(frame.payload, 0, 3), 1000 * 256); + EXPECT_EQ(sample_at(frame.payload.bytes(), 0, 3), 1000 * 256); } // Through a live pause, the staging ring keeps the history before the head a resampled instance @@ -1392,7 +1402,7 @@ TEST(GroupAudioLive, ResumesAResampledInstanceFromTheHistoryBeforeTheHead) { p.view->resume(); ASSERT_EQ(p.view->write(pcm.data() + HEAD, SAMPLES - HEAD).frames, SAMPLES - HEAD); p.view->drain(); - std::vector frames = frames_by_instance(p.run())[resampled]; + std::vector frames = std::move(frames_by_instance(p.run())[resampled]); ASSERT_FALSE(frames.empty()); EXPECT_EQ(frames.front().index, 0); @@ -1456,7 +1466,7 @@ TEST(GroupAudioOpus, EncodesPacketsAlignedWithTheInput) { EXPECT_EQ(frame.ts_us, frames[pcm][k].ts_us); // A packet decodes to a whole frame, the last one too EXPECT_EQ(frame.samples, BLOCK); - const std::optional> samples = reader.decode(frame.payload); + const std::optional> samples = reader.decode(frame.payload.bytes()); ASSERT_TRUE(samples); ASSERT_EQ(samples->size(), BLOCK * 2); decoded.insert(decoded.end(), samples->begin(), samples->end()); @@ -1479,7 +1489,7 @@ TEST(GroupAudioOpus, EncodesPacketsAlignedWithTheInput) { } // Events pushed before core drains, more than the FIFO has room for at first and more than one -// handler's budget, are delivered once each, in push order +// run's budget, are delivered once each, in push order TEST(GroupToCoreFifo, DeliversInOrderPastItsRoomAndItsBudget) { using sendspin::server::AnchorShift; using sendspin::server::GroupToCoreFifo; @@ -1518,7 +1528,7 @@ TEST(GroupAudioOnItsThread, PostsInOrderWhenWokenByTheHost) { std::vector events; sendspin::server::GroupToCoreFifo fifo(loop, [&](GroupEvent& event) { const std::lock_guard lock(mutex); - events.push_back(event); + sendspin::server::test::keep_event(events, event); delivered.notify_all(); }); const auto view = std::make_shared( @@ -1580,7 +1590,7 @@ std::vector message_timestamps(const std::vector& events, u if (ready == nullptr || ready->instance != ANALYSIS) { continue; } - const Bytes& payload = ready->frame.payload; + const Bytes payload = bytes_of(ready->frame.payload); for (size_t at = 0; at < payload.size(); at += bytes) { EXPECT_EQ(payload[at], id); out.push_back(wire::timestamp_at(payload, at + wire::TIMESTAMP_AT)); @@ -1635,7 +1645,8 @@ TEST_F(GroupAudioTest, ShiftsAnalysisMessagesWithTheFrames) { constexpr int64_t STALL_US = 2000000; this->write(2 * CELL, 0); std::vector stalled = this->run(STALL_US); - events.insert(events.end(), stalled.begin(), stalled.end()); + events.insert(events.end(), std::make_move_iterator(stalled.begin()), + std::make_move_iterator(stalled.end())); ASSERT_TRUE(std::any_of(events.begin(), events.end(), [](const GroupEvent& event) { return std::holds_alternative(event); })); @@ -1828,4 +1839,84 @@ TEST_F(GroupAudioTest, BuildsOneAnalysisPerJobAndHoldsTheOthersInput) { } } +// Once warmed up, producing and dropping frames allocates nothing: the jobs of a PCM, a FLAC and an +// Opus instance and a visualizer analysis, each frame's payload in its instance's slots, and core's +// handler appending each frame to its ring, which drops it as it plays. Counted through the +// sanitizer's malloc hook on the thread that runs both; skipped without ASan or TSan. +TEST(GroupAudioAllocations, ProducesAndDropsFramesWithoutAllocating) { + if (!AllocationCount::available()) { + GTEST_SKIP() << "needs the sanitizer's malloc hook"; + } + // Control: a block allocated while counting is counted + { + AllocationCount control; + control.start(); + // Through a volatile pointer, so the compiler keeps the allocation + std::vector* volatile block = new std::vector(64); + control.stop(); + delete block; + EXPECT_EQ(control.blocks(), 2); + } + constexpr uint32_t STEREO = 2; + const std::array instances{ + {sched::InstanceId{51}, sched::InstanceId{52}, sched::InstanceId{53}}}; + const sched::InstanceId analysis{54}; + std::map rings; + for (const sched::InstanceId id : instances) { + rings.emplace(id, sched::FrameRing(id, 1)); + } + rings.emplace(analysis, sched::FrameRing(analysis, 1)); + size_t dropped = 0; + sendspin::server::EventLoop loop; + sendspin::server::GroupToCoreFifo fifo(loop, [&](GroupEvent& event) { + if (const auto* started = std::get_if(&event)) { + for (auto& [id, ring] : rings) { + ring.restart(started->epoch, 0); + } + } else if (auto* ready = std::get_if(&event)) { + // Appended as it plays, so the ring drops it at once + const int64_t ts = ready->frame.ts_us; + EXPECT_TRUE(rings.at(ready->instance).append(std::move(ready->frame), ts)); + EXPECT_EQ(rings.at(ready->instance).size(), 0U); + ++dropped; + } + }); + const auto view = std::make_shared( + GroupId{1}, InputId{1}, + sendspin::server::InputOptions{InputFormat{RATE, STEREO, SampleFormat::S16}}, + sendspin::server::GroupParams{LEAD_US, 0, 10000000}, + std::make_shared(), nullptr); + auto slot = std::make_shared(); + slot->publish( + {{instances[0], {}, {sched::Codec::PCM, RATE, STEREO, 16}, 0}, + {instances[1], {}, {sched::Codec::FLAC, RATE, STEREO, 16}, 0}, + {instances[2], {}, {sched::Codec::OPUS, RATE, STEREO, 16}, 0}}, + {{analysis, + {}, + {sendspin::server::visualizer_type::LOUDNESS | sendspin::server::visualizer_type::F_PEAK, + 50, + {}}, + 0}}); + sendspin::server::GroupAudio audio(view, fifo.producer(), slot, + sendspin::server::ResamplerQuality::MEDIUM); + const Tone tone(RATE, STEREO, CELL); + AllocationCount count; + constexpr int WARM_UP = 20; + constexpr int COUNTED = 50; + for (int cell = 0; cell < WARM_UP + COUNTED; ++cell) { + ASSERT_EQ(view->write(tone.pcm.data(), CELL).frames, CELL); + if (cell == WARM_UP) { + dropped = 0; + count.start(); + } + while (audio.run_job(now_us())) {} + loop.impl().io().restart(); + loop.impl().io().poll(); + } + count.stop(); + // Every instance's frame and the analysis's cell for each counted cell + EXPECT_GE(dropped, size_t{COUNTED} * 4); + EXPECT_EQ(count.blocks(), 0); +} + } // namespace diff --git a/tests/test_group_audio_internals.cpp b/tests/test_group_audio_internals.cpp index 26ed698..a87505c 100644 --- a/tests/test_group_audio_internals.cpp +++ b/tests/test_group_audio_internals.cpp @@ -15,14 +15,15 @@ /// @file test_group_audio_internals.cpp /// @brief A group's pipelines on their own, reached inside: a resume from a paused position the /// test sets, a stretch cut or extended directly, the resamplers a job builds, an instance whose -/// encoder is swapped for one that refuses, and the `L_p` a live rebase publishes. Compiled with -/// -fno-access-control (tests/CMakeLists.txt). +/// encoder is swapped for one that refuses, the `L_p` a live rebase publishes, and an instance and +/// analysis given too few payload slots. Compiled with -fno-access-control (tests/CMakeLists.txt). #include "clock.h" #include "frame_encoder.h" #include "group_audio.h" #include "group_pipelines.h" #include "host_view.h" +#include "payload_slots.h" #include "pipeline_stages.h" #include "sched/leads.h" #include @@ -33,6 +34,7 @@ #include #include #include +#include #include #include @@ -278,4 +280,88 @@ TEST(GroupAudioLive, RebasesByTheResampledInstancesLookahead) { EXPECT_EQ(p.view->rebase_latency_us_, RESAMPLED_LATENCY_US); } +// An instance or analysis whose payload slots are all taken posts PayloadsFull once, however many +// jobs try again, and produces nothing more until its frames are dropped, consuming no input +// meanwhile: its frames then go on where they stopped, the same as from pipelines with room. +// White-box: the instance's and the analysis's slots (GroupAudio::instances_, analyses_) replaced +// with two each. +TEST(GroupAudioPipelines, WaitsForPayloadSlotsWithoutLosingInput) { + using sendspin::server::PayloadsFull; + using sendspin::server::PayloadSlots; + namespace visualizer_type = sendspin::server::visualizer_type; + const InputFormat format{48000, 1, SampleFormat::S16}; + const sched::InstanceId pcm{40}; + const sched::InstanceId loudness{41}; + constexpr size_t CELLS = 10; + const Tone tone(48000, 1, 960 * CELLS); + const auto start = [&](Pipelines& p) { + p.slot->publish({{pcm, {}, {sched::Codec::PCM, 48000, 1, 16}, 0}}, + {{loudness, {}, {visualizer_type::LOUDNESS, 50, {}}, 0}}); + // Builds the instance and the analysis, which produce nothing with no input + EXPECT_TRUE(p.run().empty()); + }; + Pipelines roomy(format); + start(roomy); + Pipelines tight(format); + start(tight); + ASSERT_EQ(tight.audio.instances_.size(), 1U); + ASSERT_EQ(tight.audio.analyses_.size(), 1U); + tight.audio.instances_[0].payloads = + PayloadSlots::create(tight.audio.instances_[0].payloads->slot_bytes(), 2); + tight.audio.analyses_[0].payloads = + PayloadSlots::create(tight.audio.analyses_[0].payloads->slot_bytes(), 2); + + ASSERT_EQ(roomy.view->write(tone.pcm.data(), 960 * CELLS).frames, 960 * CELLS); + ASSERT_EQ(tight.view->write(tone.pcm.data(), 960 * CELLS).frames, 960 * CELLS); + std::map> expected = + frames_by_instance(run_all(roomy)); + ASSERT_EQ(expected[pcm].size(), CELLS); + ASSERT_EQ(expected[loudness].size(), CELLS); + + // Each round's jobs hold the frames in the FIFO, a job more trying again with every slot + // taken, then the round drops them as it delivers them + const auto round = [&] { + while (tight.audio.run_job(now_us())) {} + EXPECT_FALSE(tight.audio.run_job(now_us())); + return tight.run(); + }; + std::map> got; + std::map waits; + for (std::vector events = round(); !events.empty(); events = round()) { + std::map full; + for (GroupEvent& event : events) { + if (const auto* waiting = std::get_if(&event)) { + ++full[waiting->instance]; + } else if (auto* ready = std::get_if(&event)) { + // Nothing after a wait in the same run + EXPECT_FALSE(full.contains(ready->instance)); + got[ready->instance].push_back(std::move(ready->frame)); + } + } + for (const auto& [instance, count] : full) { + EXPECT_EQ(count, 1) << instance.value; + waits[instance] += count; + } + ASSERT_LT(waits[pcm], static_cast(CELLS)); + } + EXPECT_GE(waits[pcm], 4); + EXPECT_GE(waits[loudness], 4); + ASSERT_EQ(got[pcm].size(), CELLS); + ASSERT_EQ(got[loudness].size(), CELLS); + EXPECT_EQ(concatenated(got[pcm]), tone.pcm); + EXPECT_EQ(concatenated(got[pcm]), concatenated(expected[pcm])); + // Each cell's loudness message, but for its timestamp, which follows each run's anchor + const size_t message = sendspin::server::visualizer_message_bytes(visualizer_type::LOUDNESS, 0); + for (size_t k = 0; k < CELLS; ++k) { + SCOPED_TRACE(k); + EXPECT_EQ(got[loudness][k].index, static_cast(k)); + const std::span a = got[loudness][k].payload.bytes(); + const std::span b = expected[loudness][k].payload.bytes(); + ASSERT_EQ(a.size(), message); + ASSERT_EQ(b.size(), message); + EXPECT_EQ(a[0], b[0]); + EXPECT_TRUE(std::equal(a.begin() + 9, a.end(), b.begin() + 9)); + } +} + } // namespace diff --git a/tests/test_group_internals.cpp b/tests/test_group_internals.cpp index 269e3e8..66b783a 100644 --- a/tests/test_group_internals.cpp +++ b/tests/test_group_internals.cpp @@ -15,7 +15,8 @@ /// @file test_group_internals.cpp /// @brief A group on core with its pipeline job on the group audio threads: a group erased on /// core while its job runs, an input released while its epoch start is queued, a full ring -/// answered once the job frees it, a slew freed on core, and a group's stall counts and reports. +/// answered once the job frees it, a slew freed on core, a group's stall counts and reports, and +/// the rings of an instance whose payload slots are all taken. /// Compiled with -fno-access-control (tests/CMakeLists.txt). #include "artwork_cache.h" @@ -104,7 +105,7 @@ TEST(GroupAudioThread, ErasesAGroupWhileItsJobRuns) { return more; }; }); - // Core waits, then erases the group, so the FIFO handler the job posts runs after the erase + // Core waits, then erases the group, so the FIFO's run the job signals comes after the erase std::promise core_gate; std::promise erased; loop.post([&] { @@ -118,7 +119,7 @@ TEST(GroupAudioThread, ErasesAGroupWhileItsJobRuns) { gate.wait_reached(); core_gate.set_value(); erased.get_future().wait(); - // The FIFO handler the job posted has run once a handler posted now runs, while the job, + // The FIFO's run the job signalled has run once a handler posted now runs, while the job, // still at its gate, holds the FIFO's state std::promise handled; loop.post([&] { handled.set_value(); }); @@ -204,8 +205,8 @@ TEST(GroupAudioThread, ReleasesAnInputWhoseEpochStartIsStillQueued) { }; }); on_core(loop, [] {}); - // Core waits until the release, so the write's wakeup and the FIFO handler the job posts - // run after it, in that order + // Core waits until the release, so the write's wakeup and the FIFO's run the job signals + // come after it, in that order std::promise core_gate; if (queued) { loop.post([gate = core_gate.get_future().share()] { gate.wait(); }); @@ -214,7 +215,7 @@ TEST(GroupAudioThread, ReleasesAnInputWhoseEpochStartIsStillQueued) { const std::vector pcm(size_t{960} * 4, 1); ASSERT_EQ(view->write(pcm.data(), 960).frames, 960U); ran.get_future().wait(); - // The FIFO handler, posted before the job returned, has run once a later handler runs + // The FIFO's run, signalled before the job returned, has run once a later handler runs if (!queued) { EXPECT_TRUE(on_core(loop, [&] { return group->playing_; })); } @@ -231,7 +232,7 @@ TEST(GroupAudioThread, ReleasesAnInputWhoseEpochStartIsStillQueued) { busy.get_future().wait(); view->release(); core_gate.set_value(); - // Core handles the release's wakeup, then the FIFO handler, then this + // Core handles the release's wakeup, then the FIFO's run, then this on_core(loop, [] {}); const auto [playing, attached] = on_core(loop, [&] { return std::pair{group->playing_, group->input_ != nullptr}; }); @@ -486,4 +487,157 @@ TEST(GroupStalls, ReportsAtMostEveryPeriodAndHoldsNothingBack) { core.join(); } +// ============================================================================== +// An instance whose payload slots are all taken +// ============================================================================== + +// Core holds a full instance until its ring holds fewer frames than its slots, then wakes the +// pipelines: at once for a ring empty or pruned below them, and by its timer at the earliest +// frame's time when every frame is still ahead; an instance with no ring is let go without a wake. +// White-box: the group's pipeline task (Group::audio_) replaced with one that counts its runs, its +// rings_ filled directly, on_payloads_full() called as the group-to-core FIFO does, and full_ read +// on core. +TEST(GroupPayloads, WakesAFullInstanceOnceItsRingHoldsFewerFramesThanItsSlots) { + using sendspin::server::PayloadsFull; + namespace sched = sendspin::server::sched; + constexpr size_t SLOTS = 2; + StallGroup s; + sendspin::server::Group& group = *s.group; + std::atomic runs{0}; + group.audio_ = s.workers.add([&runs] { + ++runs; + return false; + }); + std::thread core([&] { s.loop.run(); }); + const auto wait_for_runs = [&](int count) { + // A regression hangs into the suite watchdog + while (runs.load() < count) { + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } + }; + sched::FrameRing& ring = group.rings_.at(StallGroup::PLAYER); + // Two frames, the first at first_us; on core + const auto hold = [&](int64_t first_us, int64_t second_us) { + ring.restart(StallGroup::EPOCH, 0); + const int64_t now = sendspin::server::now_us(); + for (int64_t index = 0; index < 2; ++index) { + sched::Frame frame; + frame.epoch = StallGroup::EPOCH; + frame.index = index; + frame.ts_us = index == 0 ? first_us : second_us; + EXPECT_TRUE(ring.append(std::move(frame), now)); + } + }; + + // No ring: let go. The one worker runs tasks in the order woken, so once a probe woken after + // has run, a wake would have run too. + on_core(s.loop, [&] { + group.on_payloads_full(PayloadsFull{sched::InstanceId{99}, SLOTS}); + EXPECT_TRUE(group.full_.empty()); + }); + std::promise probed; + const std::shared_ptr probe = s.workers.add([&probed] { + probed.set_value(); + return false; + }); + probe->wake(); + probed.get_future().wait(); + EXPECT_EQ(runs.load(), 0); + + // An empty ring holds fewer frames than its slots + on_core(s.loop, [&] { + group.on_payloads_full(PayloadsFull{StallGroup::ANALYSIS, SLOTS}); + EXPECT_TRUE(group.full_.empty()); + }); + wait_for_runs(1); + + // One frame due and one ahead: the due one is pruned + const int64_t now = sendspin::server::now_us(); + on_core(s.loop, [&] { hold(now + 1, now + 10'000'000); }); + std::this_thread::sleep_for(std::chrono::milliseconds(2)); + on_core(s.loop, [&] { + group.on_payloads_full(PayloadsFull{StallGroup::PLAYER, SLOTS}); + EXPECT_TRUE(group.full_.empty()); + EXPECT_EQ(ring.size(), 1U); + }); + wait_for_runs(2); + + // Every frame ahead: held, nothing pruned, until the timer at the earliest one + on_core(s.loop, [&] { + const int64_t at = sendspin::server::now_us(); + hold(at + 50'000, at + 10'000'000); + group.on_payloads_full(PayloadsFull{StallGroup::PLAYER, SLOTS}); + ASSERT_EQ(group.full_.size(), 1U); + EXPECT_EQ(ring.size(), 2U); + EXPECT_EQ(runs.load(), 2); + }); + wait_for_runs(3); + on_core(s.loop, [&] { + EXPECT_TRUE(group.full_.empty()); + EXPECT_EQ(ring.size(), 1U); + }); + + s.loop.post([&] { + s.group.reset(); + s.loop.shutdown(0); + }); + core.join(); +} + +// A full instance whose frames are all far ahead is let go as soon as its ring restarts and so +// returns every slot: at a flush's cut, an epoch's start or the run's end. White-box: rings_ filled +// directly, on_payloads_full() and the restarts called as the FIFO does, and full_ read; the run's +// end with no pipelines (pipelines_), so its recompute keeps the rings. +TEST(GroupPayloads, LetsAFullInstanceGoWhenItsRingRestarts) { + using sendspin::server::PayloadsFull; + namespace sched = sendspin::server::sched; + struct Row { + const char* name; + void (*restart)(sendspin::server::Group&); + bool held; + }; + const Row rows[] = { + {"Control: no restart", [](sendspin::server::Group&) {}, true}, + {"a flush's cut", + [](sendspin::server::Group& group) { + group.on_epoch_cut(sendspin::server::EpochCut{sched::EpochId{2}}); + }, + false}, + {"an epoch's start", + [](sendspin::server::Group& group) { + sendspin::server::EpochStarted started{sched::RunId{1}, sched::EpochId{2}, + sched::make_anchor(0, 0, 48000, 2), + std::nullopt}; + group.on_epoch_started(started); + }, + false}, + {"the run's end", + [](sendspin::server::Group& group) { + group.pipelines_ = {}; + group.end_run(StallGroup::EPOCH); + }, + false}, + }; + for (const Row& row : rows) { + SCOPED_TRACE(row.name); + StallGroup s; + sendspin::server::Group& group = *s.group; + sched::FrameRing& ring = group.rings_.at(StallGroup::PLAYER); + ring.restart(StallGroup::EPOCH, 0); + const int64_t now = sendspin::server::now_us(); + for (int64_t index = 0; index < 2; ++index) { + sched::Frame frame; + frame.epoch = StallGroup::EPOCH; + frame.index = index; + frame.ts_us = now + ((index + 1) * 10'000'000); + ASSERT_TRUE(ring.append(std::move(frame), now)); + } + group.on_payloads_full(PayloadsFull{StallGroup::PLAYER, 2}); + ASSERT_EQ(group.full_.size(), 1U); + row.restart(group); + EXPECT_EQ(group.rings_.size(), 2U); + EXPECT_EQ(group.full_.size(), row.held ? 1U : 0U); + } +} + } // namespace diff --git a/tests/test_payload_slots.cpp b/tests/test_payload_slots.cpp new file mode 100644 index 0000000..f6a7eea --- /dev/null +++ b/tests/test_payload_slots.cpp @@ -0,0 +1,144 @@ +// 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. + +/// @file test_payload_slots.cpp +/// @brief PayloadSlots and FramePayload: a slot reserved, written and committed, returned in any +/// order by destroying its payload, and the store kept alive by the payloads it gave + +#include "frame_payload.h" +#include "payload_slots.h" +#include + +#include +#include +#include +#include +#include +#include +#include +#include + +namespace { + +using sendspin::server::FramePayload; +using sendspin::server::PayloadSlots; +using sendspin::server::SharedPtr; + +constexpr size_t SLOT_BYTES = 16; + +/// A payload of size bytes, each byte value, in the next free slot +FramePayload filled(PayloadSlots& slots, size_t size, uint8_t value) { + const std::span room = slots.reserve(); + EXPECT_EQ(room.size(), SLOT_BYTES); + std::fill_n(room.begin(), size, value); + return slots.commit(size); +} + +// A commit holds the first bytes written into the slot reserve() gave; reserve() gives the same +// slot until it is committed, and another after +TEST(PayloadSlots, CommitsTheBytesWrittenIntoTheReservedSlot) { + const SharedPtr slots = PayloadSlots::create(SLOT_BYTES, 2); + ASSERT_TRUE(slots); + EXPECT_EQ(slots->slot_bytes(), SLOT_BYTES); + const std::span room = slots->reserve(); + ASSERT_EQ(room.size(), SLOT_BYTES); + // Uncommitted: the same slot again + EXPECT_EQ(slots->reserve().data(), room.data()); + room[0] = 7; + room[1] = 8; + const FramePayload payload = slots->commit(2); + EXPECT_EQ(payload.size(), 2U); + EXPECT_EQ(payload.bytes().data(), room.data()); + EXPECT_EQ(payload.bytes()[0], 7); + EXPECT_EQ(payload.bytes()[1], 8); + // Control: committed, the next reserve gives the other slot + EXPECT_NE(slots->reserve().data(), room.data()); + EXPECT_FALSE(slots->starved()); +} + +// With every slot held, reserve() gives nothing and the store is starved; destroying any one +// payload, in any order, returns exactly its slot +TEST(PayloadSlots, ReturnsEachSlotInAnyOrder) { + struct Row { + const char* name; + size_t released; + }; + const std::array rows{{{"the first", 0}, {"the middle", 1}, {"the last", 2}}}; + for (const Row& row : rows) { + SCOPED_TRACE(row.name); + const SharedPtr slots = PayloadSlots::create(SLOT_BYTES, 3); + ASSERT_TRUE(slots); + std::vector> held; + for (uint8_t i = 0; i < 3; ++i) { + held.emplace_back(filled(*slots, 4, i)); + } + EXPECT_TRUE(slots->reserve().empty()); + EXPECT_TRUE(slots->starved()); + const uint8_t* freed = held[row.released]->bytes().data(); + held[row.released].reset(); + const std::span room = slots->reserve(); + EXPECT_EQ(room.data(), freed); + EXPECT_FALSE(slots->starved()); + // The others still hold their bytes + for (size_t i = 0; i < held.size(); ++i) { + if (held[i]) { + EXPECT_EQ(held[i]->bytes()[3], i); + } + } + const FramePayload refilled = slots->commit(0); + EXPECT_TRUE(slots->reserve().empty()); + } +} + +// A payload moved from returns nothing; the one moved to returns the slot once, and so does one +// assigned over +TEST(PayloadSlots, ReturnsASlotOnceThroughMoves) { + const SharedPtr slots = PayloadSlots::create(SLOT_BYTES, 1); + ASSERT_TRUE(slots); + std::optional first(filled(*slots, 3, 1)); + std::optional second(std::move(*first)); + EXPECT_EQ(first->size(), 0U); + EXPECT_TRUE(first->bytes().empty()); + first.reset(); + EXPECT_TRUE(slots->reserve().empty()); + EXPECT_EQ(second->size(), 3U); + // Assigned an empty payload: its slot returns + *second = FramePayload{}; + EXPECT_EQ(second->size(), 0U); + EXPECT_FALSE(slots->reserve().empty()); + second.reset(); + // Control: the slot came back once, so one commit fills the store again + const FramePayload again = slots->commit(1); + EXPECT_TRUE(slots->reserve().empty()); +} + +// A payload keeps its store, and its bytes, alive after the store's owner lets it go +TEST(PayloadSlots, OutlivesItsOwnerThroughItsPayloads) { + SharedPtr slots = PayloadSlots::create(SLOT_BYTES, 2); + ASSERT_TRUE(slots); + const FramePayload payload = filled(*slots, 5, 9); + slots.reset(); + ASSERT_EQ(payload.size(), 5U); + EXPECT_EQ(payload.bytes()[4], 9); +} + +// A store of no slots, or of slots of no bytes, is refused +TEST(PayloadSlots, RefusesAnEmptyStore) { + EXPECT_FALSE(PayloadSlots::create(0, 4)); + EXPECT_FALSE(PayloadSlots::create(SLOT_BYTES, 0)); + // Control + EXPECT_TRUE(PayloadSlots::create(1, 1)); +} + +} // namespace diff --git a/tests/test_player_stream.cpp b/tests/test_player_stream.cpp index c4986b6..6d3a7ea 100644 --- a/tests/test_player_stream.cpp +++ b/tests/test_player_stream.cpp @@ -17,6 +17,7 @@ /// cursor emit around a switch in place, and how long its cursor holds a short chunk #include "fragmentation.h" +#include "frame_payloads.h" #include "player_stream.h" #include "wire_messages.h" #include @@ -88,7 +89,7 @@ class FakeGroup : public StreamSource { frame.ts_us = T0 + (k * sched::CELL_US); frame.duration_us = sched::CELL_US; frame.samples = 960; - frame.payload.assign(PAYLOAD, 0); + frame.payload = sendspin::server::test::zero_payload(PAYLOAD); EXPECT_TRUE(ring.append(std::move(frame), START_US)); } if (final) { diff --git a/tests/test_sched_frames.cpp b/tests/test_sched_frames.cpp index 8281a89..4c3274b 100644 --- a/tests/test_sched_frames.cpp +++ b/tests/test_sched_frames.cpp @@ -15,11 +15,15 @@ /// @file test_sched_frames.cpp /// @brief The frame grid, the frame ring and the cursor's prepare step +#include "frame_payloads.h" +#include "payload_slots.h" +#include "profile.h" #include "sched/frames.h" #include #include #include +#include #include namespace { @@ -45,7 +49,7 @@ Frame frame(int64_t index, uint32_t cells = 1) { f.ts_us = 1000 * MS + index * 20 * MS; f.duration_us = cells * 20 * MS; f.samples = 960 * cells; - f.payload.assign(100, 0); + f.payload = sendspin::server::test::zero_payload(100); return f; } @@ -99,7 +103,7 @@ TEST(SchedFrameRing, KeysFramesByCellIndex) { EXPECT_FALSE(ring.append(frame(126, 2), 0)); Frame stale = frame(130, 2); stale.epoch = OTHER_EPOCH; - EXPECT_FALSE(ring.append(stale, 0)); + EXPECT_FALSE(ring.append(std::move(stale), 0)); EXPECT_EQ(ring.base(), 128); ASSERT_TRUE(ring.append(frame(130, 2), 0)); EXPECT_EQ(ring.base(), 130); @@ -170,7 +174,7 @@ TEST(SchedFrameRing, SkipsToALaterIndexOnceEmptied) { TEST(SchedFrameRing, HoldsFramesPastItsFirstRoom) { FrameRing ring(InstanceId{1}, 1); ring.restart(EPOCH, 0); - const auto frames = static_cast(sched::ring_frames(1)) * 2; + const auto frames = static_cast(sched::payload_slots(1)) * 2; for (int64_t k = 0; k < frames; ++k) { ASSERT_TRUE(ring.append(frame(k), 0)); } @@ -182,6 +186,62 @@ TEST(SchedFrameRing, HoldsFramesPastItsFirstRoom) { } } +// An instance's payload slots cover the ring's span and the frames core may leave unpruned past +// their play time, a part frame rounded up: 3 s and 0.5 s on the embedded profile, 12 s and 1 s +// on a host +TEST(SchedFrameRing, SizesPayloadSlotsPastTheRingsSpan) { + constexpr bool EMBEDDED = sendspin::server::profile::EMBEDDED; + EXPECT_EQ(sched::payload_slots(1), EMBEDDED ? 151U + 25U : 601U + 50U); + EXPECT_EQ(sched::payload_slots(2), EMBEDDED ? 76U + 13U : 301U + 25U); + EXPECT_EQ(sched::payload_slots(7), EMBEDDED ? 22U + 4U : 86U + 8U); +} + +// A frame the ring drops gives its payload's slot back: pruned at its play time, cleared by a +// restart, or refused on append +TEST(SchedFrameRing, ReturnsTheSlotOfEachFrameItDrops) { + using sendspin::server::FramePayload; + using sendspin::server::PayloadSlots; + struct Row { + const char* name; + /// Drops a frame, or not; spare holds the store's last slot + void (*drop)(FrameRing&, FramePayload& spare); + bool returned; + }; + const Row rows[] = { + {"Control: none dropped", [](FrameRing& ring, FramePayload&) { ring.prune(0); }, false}, + {"pruned", [](FrameRing& ring, FramePayload&) { ring.prune(frame(0).ts_us); }, true}, + {"restarted", [](FrameRing& ring, FramePayload&) { ring.restart(EPOCH, 0); }, true}, + {"refused", + [](FrameRing& ring, FramePayload& spare) { + Frame stale = frame(2); + stale.epoch = OTHER_EPOCH; + stale.payload = std::move(spare); + EXPECT_FALSE(ring.append(std::move(stale), 0)); + }, + true}, + }; + for (const Row& row : rows) { + SCOPED_TRACE(row.name); + const auto slots = PayloadSlots::create(8, 3); + ASSERT_TRUE(slots); + const auto take = [&] { + static_cast(slots->reserve()); + return slots->commit(1); + }; + FrameRing ring(InstanceId{1}, 1); + ring.restart(EPOCH, 0); + for (int64_t k = 0; k < 2; ++k) { + Frame f = frame(k); + f.payload = take(); + ASSERT_TRUE(ring.append(std::move(f), 0)); + } + FramePayload spare = take(); + EXPECT_TRUE(slots->reserve().empty()); + row.drop(ring, spare); + EXPECT_EQ(!slots->reserve().empty(), row.returned); + } +} + TEST(SchedCursor, PrepareClampsSkipsAndFinishes) { FrameRing ring(InstanceId{1}, 1); ring.restart(EPOCH, 0); diff --git a/tests/test_sched_property.cpp b/tests/test_sched_property.cpp index 72873f5..76d65cc 100644 --- a/tests/test_sched_property.cpp +++ b/tests/test_sched_property.cpp @@ -23,6 +23,7 @@ /// from slow clocks and with rising group leads, check the live rebase, admission as if rebased /// and the folding of shifts. +#include "frame_payloads.h" #include "sched/admission.h" #include "sched/chunk.h" #include "sched/constants.h" @@ -128,7 +129,8 @@ void run_seed(uint64_t seed) { frame.index = produced; frame.ts_us = T0 + produced * sched::CELL_US; frame.duration_us = cells * sched::CELL_US; - frame.payload.assign(static_cast(uniform(1, max_frame)), 0); + frame.payload = + sendspin::server::test::zero_payload(static_cast(uniform(1, max_frame))); ASSERT_TRUE(ring.append(std::move(frame), now)); produced += cells; } @@ -273,7 +275,8 @@ void run_switch_seed(uint64_t seed) { frame.index = instance.produced; frame.ts_us = T0 + instance.produced * sched::CELL_US; frame.duration_us = instance.cells * sched::CELL_US; - frame.payload.assign(static_cast(uniform(1, 3000)), 0); + frame.payload = + sendspin::server::test::zero_payload(static_cast(uniform(1, 3000))); // A frame already late is dropped at insertion instance.ring.append(std::move(frame), now); instance.produced += instance.cells; diff --git a/tests/test_server_source.cpp b/tests/test_server_source.cpp index 6fddc05..9d91cd1 100644 --- a/tests/test_server_source.cpp +++ b/tests/test_server_source.cpp @@ -25,6 +25,7 @@ #include "crypto/keys.h" #include "fragmentation.h" #include "frame_encoder.h" +#include "frame_payloads.h" #include "opus_reader.h" #include "peer.h" #include "player_sink.h" @@ -497,7 +498,8 @@ TEST(ServerSources, StreamFlacAndRideOutAStallAndAGap) { signal.put(&pcm[i * signal.frame_bytes()], (k * CHUNK) + i); } Bytes frame; - ASSERT_TRUE(encoder->encode(pcm, sendspin::server::SampleFormat::S24, frame)); + ASSERT_TRUE(sendspin::server::test::encode_into( + *encoder, pcm, sendspin::server::SampleFormat::S24, frame)); send_chunk(raw, base + (int64_t{k} * 20000), frame); }; const auto live = [&](int from, int to) { diff --git a/tests/test_source_decoder.cpp b/tests/test_source_decoder.cpp index 9826065..4b01421 100644 --- a/tests/test_source_decoder.cpp +++ b/tests/test_source_decoder.cpp @@ -21,6 +21,7 @@ #include "crypto/base64url.h" #include "dispatcher.h" #include "frame_encoder.h" +#include "frame_payloads.h" #include "host_view.h" #include "job_view.h" #include "source_decoder.h" @@ -233,8 +234,8 @@ FlacStream encode_flac(const Bytes& audio) { constexpr size_t FRAME_BYTES = 960 * 4; for (size_t at = 0; at < audio.size(); at += FRAME_BYTES) { Bytes payload; - EXPECT_TRUE( - encoder->encode(std::span(audio).subspan(at, FRAME_BYTES), SampleFormat::S16, payload)); + EXPECT_TRUE(sendspin::server::test::encode_into( + *encoder, std::span(audio).subspan(at, FRAME_BYTES), SampleFormat::S16, payload)); stream.frames.push_back(payload); } return stream; @@ -302,7 +303,8 @@ TEST(SourceDecoder, DecodesOpusPackets) { const std::unique_ptr encoder = sendspin::server::FrameEncoder::create(format); Bytes packet; - ASSERT_TRUE(encoder->encode(pcm16(960), SampleFormat::S16, packet)); + ASSERT_TRUE( + sendspin::server::test::encode_into(*encoder, pcm16(960), SampleFormat::S16, packet)); Rig r; r.start({Codec::OPUS, 2, 48000, 0}); ASSERT_EQ(r.sink->formats.size(), 1U); diff --git a/tests/test_visualizer_stream.cpp b/tests/test_visualizer_stream.cpp index 27ad322..d249b0a 100644 --- a/tests/test_visualizer_stream.cpp +++ b/tests/test_visualizer_stream.cpp @@ -18,6 +18,7 @@ /// a configuration change in place. #include "fragmentation.h" +#include "frame_payloads.h" #include "visualizer_stream.h" #include "wire_messages.h" #include @@ -129,7 +130,9 @@ class FakeGroup : public StreamSource { frame.epoch = EPOCH; frame.index = cell; frame.ts_us = T0 + (cell * sched::CELL_US) + offset_us; - append_message(frame.payload, id, bytes, frame.ts_us); + std::vector payload; + append_message(payload, id, bytes, frame.ts_us); + frame.payload = sendspin::server::test::payload_of(payload); EXPECT_TRUE(ring.append(std::move(frame), 0)); }