From e835ccabf204e0900cae839eb6ddad4025be7a46 Mon Sep 17 00:00:00 2001 From: Kevin Ahrendt Date: Thu, 15 Oct 2026 19:48:28 -0400 Subject: [PATCH] Keep a segment's boundary time from overflowing and resync a source after its ASRC refuses A segment's elapsed time from T0 was one int64 count of picoseconds, which overflows after about 106.7 days of one anchor. It is now whole seconds plus the picoseconds past them, carried at each segment, so the boundary keeps its picosecond precision for any run; the math still runs only per format change and per instance start. When a live source sink's ASRC refused a chunk, the chunk was written unaltered but the ASRC's position did not move with it, so the start of its next output was cut as already written. The sink now restarts the correction after a refusal, engaged or not. Closes #24 --- CLAUDE.md | 2 +- src/sched/timeline.cpp | 23 +++-- src/sched/timeline.h | 7 +- src/source_clock.h | 2 +- src/source_sink.cpp | 10 +- src/source_sink.h | 6 +- tests/CMakeLists.txt | 7 ++ tests/source_sink_rig.h | 128 +++++++++++++++++++++++++ tests/test_sched_timeline.cpp | 46 +++++++++ tests/test_source_sink.cpp | 114 ++-------------------- tests/test_source_sink_internals.cpp | 135 +++++++++++++++++++++++++++ 11 files changed, 356 insertions(+), 124 deletions(-) create mode 100644 tests/source_sink_rig.h create mode 100644 tests/test_source_sink_internals.cpp diff --git a/CLAUDE.md b/CLAUDE.md index 0eae2f0..b119cb9 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -95,7 +95,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/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, 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). - 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/src/sched/timeline.cpp b/src/sched/timeline.cpp index daa1662..0bf6df8 100644 --- a/src/sched/timeline.cpp +++ b/src/sched/timeline.cpp @@ -29,7 +29,7 @@ const Segment& segment_for(const Anchor& anchor, uint64_t position) { return it == anchor.segments.begin() ? anchor.segments.front() : *std::prev(it); } -/// Picoseconds in a second, the unit of a segment's elapsed time +/// Picoseconds in a second, the unit of a segment's elapsed time past its whole seconds constexpr int64_t PS_PER_SECOND = 1000000000000; /// `floor((p - p_j) * 10^6 / r_j)` for the given segment @@ -57,11 +57,15 @@ void reset_anchor(Anchor& anchor, int64_t t0_us, uint64_t p0, uint32_t rate, uin void add_segment(Anchor& anchor, uint64_t position, uint32_t rate, uint8_t channels) { const Segment& last = anchor.segments.back(); const int64_t unshifted = last.unshifted_us + offset_us(last, position); - // Whole seconds and the rest apart, so neither product overflows + // The segment's whole seconds and the rest apart, so the rest's product stays small const auto samples = static_cast(position - last.position); - const int64_t elapsed_ps = last.elapsed_ps + ((samples / last.rate) * PS_PER_SECOND) + - ((samples % last.rate) * PS_PER_SECOND / last.rate); - anchor.segments.push_back({position, unshifted, rate, channels, elapsed_ps}); + int64_t elapsed_s = last.elapsed_s + (samples / last.rate); + int64_t elapsed_ps = last.elapsed_ps + ((samples % last.rate) * PS_PER_SECOND / last.rate); + if (elapsed_ps >= PS_PER_SECOND) { + ++elapsed_s; + elapsed_ps -= PS_PER_SECOND; + } + anchor.segments.push_back({position, unshifted, rate, channels, elapsed_s, elapsed_ps}); } void add_shift(Anchor& anchor, int64_t index, int64_t delta_us) { @@ -140,11 +144,10 @@ int64_t first_index_at_or_after_position(const Anchor& anchor, uint64_t position } int64_t segment_output_index(const Anchor& anchor, size_t segment, uint32_t rate) { - const int64_t elapsed = anchor.segments[segment].elapsed_ps; - // Whole seconds and the rest apart, so neither product overflows; the rest rounds half up - const int64_t rest = elapsed % PS_PER_SECOND; - return ((elapsed / PS_PER_SECOND) * int64_t{rate}) + - (((2 * rest * int64_t{rate}) + PS_PER_SECOND) / (2 * PS_PER_SECOND)); + const Segment& boundary = anchor.segments[segment]; + // The picoseconds past the whole seconds round half up + return (boundary.elapsed_s * int64_t{rate}) + + (((2 * boundary.elapsed_ps * int64_t{rate}) + PS_PER_SECOND) / (2 * PS_PER_SECOND)); } int64_t instance_start(const Anchor& anchor, int64_t requested, uint32_t cells) { diff --git a/src/sched/timeline.h b/src/sched/timeline.h index e2f797a..aebe0e6 100644 --- a/src/sched/timeline.h +++ b/src/sched/timeline.h @@ -49,9 +49,10 @@ struct Segment { int64_t unshifted_us{0}; uint32_t rate{0}; uint8_t channels{0}; - /// The time from `T0` to `position` in picoseconds, floored at each segment: the boundary - /// time output indices derive from, which `u_j`'s whole microseconds would drift by up to - /// one per segment + /// The time from `T0` to `position`, whole seconds and the picoseconds past them (below + /// 10^12), floored to the picosecond at each segment: the boundary time output indices derive + /// from, which `u_j`'s whole microseconds would drift by up to one per segment + int64_t elapsed_s{0}; int64_t elapsed_ps{0}; }; diff --git a/src/source_clock.h b/src/source_clock.h index 24fcbb6..166137b 100644 --- a/src/source_clock.h +++ b/src/source_clock.h @@ -154,7 +154,7 @@ class SourceAsrc { /// what the filter holds at the current ratio double next_position() const; - /// @brief The output frames written since create() + /// @brief The output frames written since create() or the last reset() uint64_t produced() const { return this->produced_; } diff --git a/src/source_sink.cpp b/src/source_sink.cpp index df840d6..b77f18f 100644 --- a/src/source_sink.cpp +++ b/src/source_sink.cpp @@ -159,7 +159,15 @@ void LiveSourceSink::correct(const uint8_t* pcm, size_t frames, int64_t capture_ const uint64_t first = this->asrc_->produced(); const std::optional processed = this->asrc_->process( floats, frames, engaged ? ppm : 0.0, corrected, SourceAsrc::max_output(frames)); - if (!engaged || !processed) { + if (!processed) { + // Written unaltered; the ASRC's position, which did not move with it, and its state past + // the blocks it took start over, else its next output would be cut as written already. What + // its filter held of the chunk before is dropped. + (void)this->view_->write(floats, frames); + this->restart_correction(); + return; + } + if (!engaged) { (void)this->view_->write(floats, frames); this->written_ += frames; return; diff --git a/src/source_sink.h b/src/source_sink.h index 78340a1..3585121 100644 --- a/src/source_sink.h +++ b/src/source_sink.h @@ -113,7 +113,7 @@ class LiveSourceSink : public SourcePcmSink { private: /// The run ends unless a stop by the server holds it or the input closed void stop_run(); - /// The corrected positions start over, for a new run + /// The corrected positions start over: for a new run, or after the ASRC refuses a chunk void restart_correction(); /// Corrects one chunk's frames and writes them void correct(const uint8_t* pcm, size_t frames, int64_t capture_us); @@ -128,8 +128,8 @@ class LiveSourceSink : public SourcePcmSink { /// The corrected frames written since the corrected positions started uint64_t written_{0}; SampleSlip slip_; - /// With profile::SOURCE_ASRC: runs from the stream's start, its output written once the - /// correction starts, from the position the unaltered frames reached + /// With profile::SOURCE_ASRC: runs from the stream's start or its last restart, its output + /// written once the correction starts, from the position the unaltered frames reached std::unique_ptr asrc_; /// A chunk as floats, then corrected (profile::SOURCE_ASRC), or slipped; in PSRAM where there /// is one diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 30a1e59..36632ec 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -98,6 +98,7 @@ add_executable(sendspin_server_cpp_tests test_source_decoder.cpp test_source_feed.cpp test_source_sink.cpp + test_source_sink_internals.cpp test_source_stream.cpp test_soak_wire_checker.cpp test_task_queue.cpp @@ -236,6 +237,12 @@ set_source_files_properties(test_session_driver_internals.cpp PROPERTIES COMPILE # them, which nothing the session sends or reports shows. set_source_files_properties(test_session_internals.cpp PROPERTIES COMPILE_OPTIONS -fno-access-control) +# test_source_sink_internals.cpp reads LiveSourceSink::clock_ to learn when its correction has +# started and sets SourceAsrc::max_block_ of LiveSourceSink::asrc_ to zero for one chunk, so +# micro-resampler takes nothing and process() refuses: no input the server receives makes it +# refuse. +set_source_files_properties(test_source_sink_internals.cpp PROPERTIES COMPILE_OPTIONS + -fno-access-control) # sendspin_server_cpp_beast last: a system Boost's include directory (e.g. /opt/homebrew/include) must # not shadow the fetched dependencies' headers. diff --git a/tests/source_sink_rig.h b/tests/source_sink_rig.h new file mode 100644 index 0000000..9175b59 --- /dev/null +++ b/tests/source_sink_rig.h @@ -0,0 +1,128 @@ +// 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 source_sink_rig.h +/// @brief SinkRig, a live input a LiveSourceSink writes into and the rig reads back, with the +/// source's test tone + +#pragma once + +#include "dispatcher.h" +#include "host_view.h" +#include "job_view.h" +#include "pcm_convert.h" +#include "source_sink.h" +#include + +#include +#include +#include +#include +#include + +namespace sendspin::server::test { + +/// The source's write jitter, the staleness bound its chunks are judged by +inline constexpr int64_t JITTER_US = 60000; +/// The path from capture end to receipt when nothing holds the audio back +inline constexpr int64_t PATH_US = 3000; + +/// The rig's rate, and a source's 20 ms chunk at it +inline constexpr uint32_t RATE = 48000; +inline constexpr size_t CHUNK = RATE / 50; +/// The test tone's largest step between two frames at RATE: 100 Hz at half scale +inline const double TONE_STEP = 0.5 * 2.0 * std::numbers::pi * 100.0 / RATE; + +/// A source's tone at frame position, as a 16-bit sample +inline int16_t tone(uint64_t position) { + return static_cast(std::lround( + 16383.0 * std::sin(2.0 * std::numbers::pi * 100.0 * static_cast(position) / RATE))); +} + +/// A mono live input a LiveSourceSink writes into, whose staged audio the rig reads back as +/// floats and releases, as the pipelines would. Its group lead is raised past everything written +/// once the run is anchored, so the input never cuts a write to lower its latency. +class SinkRig { +public: + explicit SinkRig(uint32_t rate = RATE, ResamplerQuality quality = ResamplerQuality::MEDIUM) + : rate_(rate) { + InputOptions options{{rate, 1, SampleFormat::S16}}; + options.kind = InputKind::LIVE; + this->view = std::make_shared(GroupId{1}, InputId{2}, options, + GroupParams{300000, 0, WRITE_AHEAD_US}, + this->dispatcher, nullptr); + this->sink = std::make_unique(this->view, JITTER_US, quality); + } + + ~SinkRig() { + this->dispatcher->finish(); + this->dispatcher->join(); + } + + SinkRig(const SinkRig&) = delete; + SinkRig& operator=(const SinkRig&) = delete; + + /// One chunk of the tone from position, captured at capture_us and received on its path + void write(uint64_t position, size_t frames, int64_t capture_us, int64_t late_us = 0) { + std::vector pcm(frames); + for (size_t i = 0; i < frames; ++i) { + pcm[i] = tone(position + i); + } + const int64_t end = capture_us + static_cast(frames) * 1000000 / this->rate_; + this->sink->write(reinterpret_cast(pcm.data()), frames, capture_us, + end + PATH_US + late_us); + if (!this->anchored_) { + this->view->set_group_params({LEAD_PAST_ALL_US, 0, WRITE_AHEAD_US}); + this->anchored_ = true; + } + this->collect(); + } + + std::shared_ptr dispatcher = std::make_shared(); + std::shared_ptr view; + std::unique_ptr sink; + /// What the input staged, as floats, unless only counted + std::vector staged; + bool count_only{false}; + uint64_t staged_count{0}; + +private: + static constexpr int64_t WRITE_AHEAD_US = int64_t{1000} * 1000000; + static constexpr int64_t LEAD_PAST_ALL_US = int64_t{1000} * 1000000; + + void collect() { + const JobView job = job_view_of(*this->view); + if (job.head <= this->read_to_) { + return; + } + const auto frames = static_cast(job.head - this->read_to_); + this->staged_count += frames; + if (!this->count_only) { + std::vector bytes; + ASSERT_TRUE(this->view->read(job.run.epoch, this->read_to_, frames, bytes)); + const SampleFormat format = this->view->format().format.sample_format; + std::vector floats(frames); + to_float(bytes, format, floats); + this->staged.insert(this->staged.end(), floats.begin(), floats.end()); + } + this->read_to_ = job.head; + this->view->release_before(job.run.epoch, job.head); + } + + uint32_t rate_; + bool anchored_{false}; + uint64_t read_to_{0}; +}; + +} // namespace sendspin::server::test diff --git a/tests/test_sched_timeline.cpp b/tests/test_sched_timeline.cpp index 99b12bb..35c5845 100644 --- a/tests/test_sched_timeline.cpp +++ b/tests/test_sched_timeline.cpp @@ -142,6 +142,7 @@ TEST(SchedTimeline, ResetAnchorLeavesOneSegmentAndNoShifts) { EXPECT_EQ(reused.segments[0].unshifted_us, 1000 * MS); EXPECT_EQ(reused.segments[0].rate, 48000U); EXPECT_EQ(reused.segments[0].channels, 2U); + EXPECT_EQ(reused.segments[0].elapsed_s, 0); EXPECT_EQ(reused.segments[0].elapsed_ps, 0); EXPECT_EQ(reused.shift_base_us, 0); EXPECT_TRUE(reused.shifts.empty()); @@ -244,6 +245,51 @@ TEST(SchedTimeline, SegmentOutputIndexFromTheBoundaryTime) { EXPECT_EQ(sched::segment_output_index(alternating, 4, 48000), 4181); } +// A boundary months into one run keeps its time to the picosecond, past the 2^63 picoseconds +// (about 106.7 days) one count of them would hold: whole seconds and the picoseconds past them, +// carried at each segment +TEST(SchedTimeline, SegmentOutputIndexMonthsIntoARun) { + constexpr uint64_t DAY_S = 86400; + struct Row { + const char* name; + /// The first segment's rate and samples, then the second's; the third is the boundary + uint32_t first_rate; + uint64_t first_samples; + uint32_t second_rate; + uint64_t second_samples; + int64_t elapsed_s; + int64_t elapsed_ps; + uint32_t rate; + int64_t index; + }; + const Row rows[] = { + {"Control: an hour", 48000, 3600 * 48000, 48000, 0, 3600, 0, 44100, 158760000}, + {"106 days, under the limit", 48000, 106 * DAY_S * 48000, 48000, 0, 9158400, 0, 44100, + 403885440000}, + {"107 days, past it", 48000, 107 * DAY_S * 48000, 48000, 0, 9244800, 0, 44100, + 407695680000}, + {"past it, over two segments", 44100, 107 * DAY_S * 44100, 48000, 48000, 9244801, 0, 11025, + 101923931025}, + // 10^12 / 44,100 is 22,675,736.96, floored; 22.68 us at 48 kHz is 1.088 samples + {"past it, one sample over", 44100, (107 * DAY_S * 44100) + 1, 48000, 0, 9244800, 22675736, + 48000, 443750400001}, + // 9,244,800.5 s at 11,025 Hz is 101,923,925,512.5 + {"past it, half a second over rounds up", 44100, (107 * DAY_S * 44100) + 22050, 48000, 0, + 9244800, 500000000000, 11025, 101923925513}, + {"two half seconds carry into a whole one", 44100, (107 * DAY_S * 44100) + 22050, 48000, + 24000, 9244801, 0, 48000, 443750448000}, + }; + for (const Row& row : rows) { + SCOPED_TRACE(row.name); + Anchor anchor = sched::make_anchor(0, 0, row.first_rate, 2); + sched::add_segment(anchor, row.first_samples, row.second_rate, 2); + sched::add_segment(anchor, row.first_samples + row.second_samples, 48000, 2); + EXPECT_EQ(anchor.segments[2].elapsed_s, row.elapsed_s); + EXPECT_EQ(anchor.segments[2].elapsed_ps, row.elapsed_ps); + EXPECT_EQ(sched::segment_output_index(anchor, 2, row.rate), row.index); + } +} + // An instance starts on its own frame boundary, and past a shift whose index its frame length does // not divide, so none of its frames straddles one TEST(SchedTimeline, InstanceStartPastAStraddledShift) { diff --git a/tests/test_source_sink.cpp b/tests/test_source_sink.cpp index 3ba6edb..62928d3 100644 --- a/tests/test_source_sink.cpp +++ b/tests/test_source_sink.cpp @@ -18,19 +18,13 @@ /// new run; and LiveSourceSink, which passes a source on the server's clock through unaltered and /// corrects one whose clock drifts, with the build's correction -#include "dispatcher.h" -#include "host_view.h" -#include "job_view.h" -#include "pcm_convert.h" #include "profile.h" #include "source_sink.h" +#include "source_sink_rig.h" #include -#include #include #include -#include -#include #include #include #include @@ -42,10 +36,10 @@ using sendspin::server::SOURCE_RESTART_GAP_US; using sendspin::server::SourceChunkAction; using sendspin::server::SourceTiming; +using sendspin::server::test::JITTER_US; +using sendspin::server::test::PATH_US; + constexpr int64_t CHUNK_US = 20000; -constexpr int64_t JITTER_US = 60000; -/// The path from capture end to receipt when nothing holds the audio back -constexpr int64_t PATH_US = 3000; /// One chunk: captured at capture_us, received late_us after its path's usual time struct Chunk { @@ -118,103 +112,13 @@ TEST(SourceChunkJudgement, RebasesAfterStaleAudioOutlastsARestartsGap) { // LiveSourceSink // ============================================================================== -using sendspin::server::Dispatcher; -using sendspin::server::GroupId; -using sendspin::server::GroupParams; -using sendspin::server::HostView; -using sendspin::server::InputId; -using sendspin::server::InputKind; -using sendspin::server::InputOptions; -using sendspin::server::LiveSourceSink; using sendspin::server::ResamplerQuality; using sendspin::server::SampleFormat; -using sendspin::server::test::job_view_of; - -constexpr uint32_t RATE = 48000; -constexpr size_t CHUNK = RATE / 50; -/// The test tone's largest step between two frames at RATE: 100 Hz at half scale -const double TONE_STEP = 0.5 * 2.0 * std::numbers::pi * 100.0 / RATE; - -/// A source's tone at frame position, as a 16-bit sample -int16_t tone(uint64_t position) { - return static_cast(std::lround( - 16383.0 * std::sin(2.0 * std::numbers::pi * 100.0 * static_cast(position) / RATE))); -} - -/// A mono live input a LiveSourceSink writes into, whose staged audio the rig reads back as -/// floats and releases, as the pipelines would. Its group lead is raised past everything written -/// once the run is anchored, so the input never cuts a write to lower its latency. -class SinkRig { -public: - explicit SinkRig(uint32_t rate = RATE, ResamplerQuality quality = ResamplerQuality::MEDIUM) - : rate_(rate) { - InputOptions options{{rate, 1, SampleFormat::S16}}; - options.kind = InputKind::LIVE; - this->view = std::make_shared(GroupId{1}, InputId{2}, options, - GroupParams{300000, 0, WRITE_AHEAD_US}, - this->dispatcher, nullptr); - this->sink = std::make_unique(this->view, JITTER_US, quality); - } - - ~SinkRig() { - this->dispatcher->finish(); - this->dispatcher->join(); - } - - SinkRig(const SinkRig&) = delete; - SinkRig& operator=(const SinkRig&) = delete; - - /// One chunk of the tone from position, captured at capture_us and received on its path - void write(uint64_t position, size_t frames, int64_t capture_us, int64_t late_us = 0) { - std::vector pcm(frames); - for (size_t i = 0; i < frames; ++i) { - pcm[i] = tone(position + i); - } - const int64_t end = capture_us + static_cast(frames) * 1000000 / this->rate_; - this->sink->write(reinterpret_cast(pcm.data()), frames, capture_us, - end + PATH_US + late_us); - if (!this->anchored_) { - this->view->set_group_params({LEAD_PAST_ALL_US, 0, WRITE_AHEAD_US}); - this->anchored_ = true; - } - this->collect(); - } - - std::shared_ptr dispatcher = std::make_shared(); - std::shared_ptr view; - std::unique_ptr sink; - /// What the input staged, as floats, unless only counted - std::vector staged; - bool count_only{false}; - uint64_t staged_count{0}; - -private: - static constexpr int64_t WRITE_AHEAD_US = int64_t{1000} * 1000000; - static constexpr int64_t LEAD_PAST_ALL_US = int64_t{1000} * 1000000; - - void collect() { - const sendspin::server::JobView job = job_view_of(*this->view); - if (job.head <= this->read_to_) { - return; - } - const auto frames = static_cast(job.head - this->read_to_); - this->staged_count += frames; - if (!this->count_only) { - std::vector bytes; - ASSERT_TRUE(this->view->read(job.run.epoch, this->read_to_, frames, bytes)); - const SampleFormat format = this->view->format().format.sample_format; - std::vector floats(frames); - sendspin::server::to_float(bytes, format, floats); - this->staged.insert(this->staged.end(), floats.begin(), floats.end()); - } - this->read_to_ = job.head; - this->view->release_before(job.run.epoch, job.head); - } - - uint32_t rate_; - bool anchored_{false}; - uint64_t read_to_{0}; -}; +using sendspin::server::test::CHUNK; +using sendspin::server::test::RATE; +using sendspin::server::test::SinkRig; +using sendspin::server::test::tone; +using sendspin::server::test::TONE_STEP; /// Where staged and written first differ, past the start: where the correction began size_t first_difference(const std::vector& staged, const std::vector& written) { diff --git a/tests/test_source_sink_internals.cpp b/tests/test_source_sink_internals.cpp new file mode 100644 index 0000000..b20a14e --- /dev/null +++ b/tests/test_source_sink_internals.cpp @@ -0,0 +1,135 @@ +// 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_source_sink_internals.cpp +/// @brief A LiveSourceSink whose ASRC refuses a chunk. Compiled with -fno-access-control +/// (tests/CMakeLists.txt). + +#include "profile.h" +#include "source_clock.h" +#include "source_sink_rig.h" +#include + +#include +#include +#include + +namespace { + +using sendspin::server::SampleFormat; +using sendspin::server::test::CHUNK; +using sendspin::server::test::RATE; +using sendspin::server::test::SinkRig; +using sendspin::server::test::tone; +using sendspin::server::test::TONE_STEP; + +/// A source's tone at position as the input stages it, unaltered +float staged_tone(uint64_t position) { + return static_cast(tone(position)) / 32768.0F; +} + +/// A source whose clock runs 2,000 ppm fast, writing 20 ms chunks into a SinkRig, whose ASRC +/// refuses a chunk on request. The refusal comes from a block size of zero, which takes nothing: no +/// input the server receives makes micro-resampler refuse. +class RefusingSource { +public: + RefusingSource() { + EXPECT_TRUE(this->rig.sink->begin({RATE, 1, SampleFormat::S16})); + } + + void next() { + this->rig.write(this->position, CHUNK, this->capture); + this->position += CHUNK; + this->capture += std::llround(20000.0 / 1.002); + } + + /// One chunk the ASRC refuses, staged unaltered + void refused() { + const size_t refused_at = this->rig.staged.size(); + const uint64_t refused_position = this->position; + const size_t max_block = this->rig.sink->asrc_->max_block_; + this->rig.sink->asrc_->max_block_ = 0; + this->next(); + this->rig.sink->asrc_->max_block_ = max_block; + ASSERT_EQ(this->rig.staged.size(), refused_at + CHUNK); + for (size_t i = 0; i < CHUNK; ++i) { + EXPECT_EQ(this->rig.staged[refused_at + i], staged_tone(refused_position + i)) << i; + } + } + + /// Chunks until the correction has started, then 50 more + void engage() { + for (int k = 0; k < 750 && !this->rig.sink->clock_.engaged(); ++k) { + this->next(); + } + ASSERT_TRUE(this->rig.sink->clock_.engaged()); + for (int k = 0; k < 50; ++k) { + this->next(); + } + } + + SinkRig rig; + uint64_t position{0}; + int64_t capture{1000000}; +}; + +// A chunk the ASRC refuses once the correction has started is staged unaltered, and the +// correction starts over from the next chunk, whose tone then continues it rather than being cut +// as staged already +TEST(LiveSourceSink, StartsTheCorrectionOverAfterTheAsrcRefuses) { + if (!sendspin::server::profile::SOURCE_ASRC) { + GTEST_SKIP() << "the build corrects by slipping frames"; + } + RefusingSource source; + source.engage(); + source.refused(); + + // The next chunks continue the tone, at most the ASRC's hold behind + const size_t resumed_at = source.rig.staged.size(); + const uint64_t resumed_position = source.position; + constexpr int AFTER = 10; + for (int k = 0; k < AFTER; ++k) { + source.next(); + } + EXPECT_GE(source.rig.staged.size(), resumed_at + (AFTER * CHUNK) - 64); + ASSERT_GT(source.rig.staged.size(), resumed_at + 200); + for (size_t i = 0; i < 200; ++i) { + // 200 frames at the clamp move the source's position by a tenth of a frame + EXPECT_NEAR(source.rig.staged[resumed_at + i], staged_tone(resumed_position + i), + (0.1 * TONE_STEP) + 0.002) + << i; + } +} + +// A chunk the ASRC refuses before the correction starts leaves nothing to cut where it starts: the +// input stages every frame but what the correction drops, at most SOURCE_CLOCK_MAX_PPM of them, +// and what the ASRC holds +TEST(LiveSourceSink, StartsTheCorrectionOverAfterTheAsrcRefusesBeforeIt) { + if (!sendspin::server::profile::SOURCE_ASRC) { + GTEST_SKIP() << "the build corrects by slipping frames"; + } + RefusingSource source; + for (int k = 0; k < 25; ++k) { + source.next(); + } + ASSERT_FALSE(source.rig.sink->clock_.engaged()); + source.refused(); + source.engage(); + const auto written = static_cast(source.position); + const auto most_dropped = + static_cast(std::ceil(written * sendspin::server::SOURCE_CLOCK_MAX_PPM / 1e6) + 64); + EXPECT_GE(source.rig.staged.size() + most_dropped, source.position); +} + +} // namespace