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