diff --git a/src/pipewire_sink.cpp b/src/pipewire_sink.cpp index b1c9f92..1f45fd1 100644 --- a/src/pipewire_sink.cpp +++ b/src/pipewire_sink.cpp @@ -359,6 +359,11 @@ bool PipeWireSink::configure(uint32_t sample_rate, uint8_t channels, uint8_t bit // Before anything can fail: recovery reopens at this format. this->last_format_ = {sample_rate, channels, bits_per_sample}; + // A new stream starts the producer from zero buffered frames, so an outage gap left over from + // the last one is owed to nobody, on either side of the handoff. Here rather than beside + // reset(), which a failed open skips. + this->recovery_.forget_discarded_frames(); + this->gap_handoff_.forget(); this->close_stream_(); if (!this->open_stream_(sample_rate, channels, bits_per_sample, PIPEWIRE_STREAM_TIMEOUT_MS)) { @@ -391,7 +396,11 @@ size_t PipeWireSink::write(const uint8_t* data, size_t length, uint32_t timeout_ ? this->bytes_per_frame_ : static_cast(this->last_format_.channels) * (static_cast(this->last_format_.bit_depth) / 8U); - return (frame == 0) ? length : length - (length % frame); + const size_t consumed = (frame == 0) ? length : length - (length % frame); + if (frame != 0) { + this->recovery_.discard_frames(static_cast(consumed / frame)); + } + return consumed; } const size_t bytes_per_frame = this->bytes_per_frame_; @@ -407,6 +416,10 @@ size_t PipeWireSink::write(const uint8_t* data, size_t length, uint32_t timeout_ const uint64_t target = this->target_multiplier_.load(std::memory_order_relaxed); const uint8_t* src = this->stage_(data, frames_total, start, target); + // The stream is alive, so an outage's gap can go to the process callback: it has the + // timestamp to retire the gap against, and this thread does not. + this->gap_handoff_.add(this->recovery_.take_discarded_frames()); + size_t done = 0; while (done < usable) { if (this->stopping_.load()) { @@ -452,6 +465,13 @@ void PipeWireSink::clear() { ++this->stream_generation_; this->space_available_.notify_all(); + // The player zeroes its buffered-frame count with a flush, so the gap is owed to nobody. Not + // reset(): the recovery budget is configure()'s to refill. A callback that has already taken + // the gap may still report it after this -- the same window request_clear() leaves for frames + // already read, and not worth a lock on the realtime path. + this->recovery_.forget_discarded_frames(); + this->gap_handoff_.forget(); + // Do not snap current_multiplier_: process() keeps running through a flush. if (this->stream_alive_()) { @@ -522,6 +542,7 @@ void PipeWireSink::poll(int64_t now_ms) { } const StreamFormat format = this->last_format_; + this->discard_ring_tail_(); this->close_stream_(); // Rebuild the loop too: a restarted daemon took the old connection with it. this->stop_loop_(); @@ -607,6 +628,7 @@ void PipeWireSink::stream_process_cb(void* userdata) { pw_time time{}; int64_t finish_us = 0; bool have_timing = false; + uint32_t gap_frames = 0; if (real_bytes > 0 && self->on_frames_played && pw_stream_get_time_n(self->stream_, &time, sizeof(time)) == 0 && time.rate.denom != 0) { const double rate_s = @@ -618,6 +640,8 @@ void PipeWireSink::stream_process_cb(void* userdata) { } finish_us = entered_us + static_cast(ahead_s * 1e6); have_timing = true; + // Only here, so a cycle the graph gave no timing for leaves the gap for the next one. + gap_frames = self->gap_handoff_.take(); } data.chunk->offset = 0; @@ -632,8 +656,10 @@ void PipeWireSink::stream_process_cb(void* userdata) { // Notified without the mutex; a missed wakeup costs at most one quantum. self->space_available_.notify_one(); + // The gap rides on this report rather than its own: the player sums the frames of reports it + // has not read yet but keeps only the last timestamp. if (have_timing) { - self->on_frames_played(static_cast(real_bytes / stride), finish_us); + self->on_frames_played(frames_with_gap(gap_frames, real_bytes / stride), finish_us); } } @@ -823,6 +849,7 @@ bool PipeWireSink::reopen_in_place_() { } const StreamFormat format = this->last_format_; + this->discard_ring_tail_(); this->close_stream_(); // Keep the loop: this cheap attempt recovers a node lost while the daemon stayed up. if (!this->open_stream_(format.sample_rate, format.channels, format.bit_depth, @@ -842,6 +869,14 @@ bool PipeWireSink::reopen_in_place_() { return true; } +void PipeWireSink::discard_ring_tail_() { + if (this->bytes_per_frame_ == 0) { + return; // no stream, so close_stream_() has already emptied the ring + } + this->recovery_.discard_frames( + static_cast(this->ring_.available() / this->bytes_per_frame_)); +} + bool PipeWireSink::stream_alive_() const { return this->stream_ != nullptr && this->bytes_per_frame_ != 0 && !this->stream_failed_.load(); } diff --git a/src/pipewire_sink.h b/src/pipewire_sink.h index d63f361..935cf55 100644 --- a/src/pipewire_sink.h +++ b/src/pipewire_sink.h @@ -125,6 +125,10 @@ class PipeWireSink final : public AudioSink { /// The one in-place reconnect a dead stream gets, from write(). Caller holds mutex_. /// @return true if a stream is running again. bool reopen_in_place_(); + /// Adds the frames still in the ring to the outage gap: the player has counted them, and the + /// lost stream recovery is about to close will never report them. Only for those closes -- + /// stop(), configure() and clear() end the stream the gap belonged to. Caller holds mutex_. + void discard_ring_tail_(); /// True while the stream is connected and the graph is still driving it. Caller holds mutex_. bool stream_alive_() const; /// Ring size in bytes. Caller holds mutex_ and the format fields are set. @@ -181,6 +185,9 @@ class PipeWireSink final : public AudioSink { StreamFormat last_format_{}; /// Guarded by mutex_, except SinkRecovery::pending(). SinkRecovery recovery_; + /// The outage gap once write() has handed it on, for the process callback to retire lock-free + /// with its next timed report. recovery_ cannot be read there; see OutageGapHandoff. + OutageGapHandoff gap_handoff_; std::atomic volume_{DEFAULT_SINK_VOLUME}; std::atomic muted_{false}; diff --git a/src/portaudio_sink.cpp b/src/portaudio_sink.cpp index f6b185f..e6796f7 100644 --- a/src/portaudio_sink.cpp +++ b/src/portaudio_sink.cpp @@ -412,6 +412,11 @@ bool PortAudioSink::configure(uint32_t sample_rate, uint8_t channels, uint8_t bi // Before anything can fail: recovery reopens at this format. this->last_format_ = {sample_rate, channels, bits_per_sample}; + // A new stream starts the producer from zero buffered frames, so an outage gap left over from + // the last one is owed to nobody, on either side of the handoff. Before the device is + // resolved, so the restart_stream_() reuse path and a failed open both drop it too. + this->recovery_.forget_discarded_frames(); + this->gap_handoff_.forget(); // Resolved per stream, so a bare -o portaudio follows the host's default. PaDeviceIndex device = paNoDevice; @@ -465,7 +470,11 @@ size_t PortAudioSink::write(const uint8_t* data, size_t length, uint32_t timeout ? this->bytes_per_frame_ : static_cast(this->last_format_.channels) * (static_cast(this->last_format_.bit_depth) / 8U); - return (frame == 0) ? length : length - (length % frame); + const size_t consumed = (frame == 0) ? length : length - (length % frame); + if (frame != 0) { + this->recovery_.discard_frames(static_cast(consumed / frame)); + } + return consumed; } const size_t bytes_per_frame = this->bytes_per_frame_; @@ -476,6 +485,10 @@ size_t PortAudioSink::write(const uint8_t* data, size_t length, uint32_t timeout return 0; } + // The stream is alive, so an outage's gap can go to the callback: it has the timestamp to + // retire the gap against, and this thread does not. + this->gap_handoff_.add(this->recovery_.take_discarded_frames()); + size_t done = 0; while (done < usable) { if (this->stopping_.load()) { @@ -518,6 +531,13 @@ void PortAudioSink::clear() { // A flush ends a parked write()'s stream. ++this->stream_generation_; + // The player zeroes its buffered-frame count with a flush, so the gap is owed to nobody. Not + // reset(): the recovery budget is configure()'s to refill. A callback that has already taken + // the gap may still report it after this -- the same window request_clear() leaves for frames + // already read, and not worth a lock on the callback. + this->recovery_.forget_discarded_frames(); + this->gap_handoff_.forget(); + // Do not snap current_multiplier_: the callback keeps running through a flush. if (this->stream_alive_()) { @@ -561,7 +581,11 @@ void PortAudioSink::poll(int64_t now_ms) { this->recovery_.rescan_done(true); const StreamFormat format = this->last_format_; - // Close first: Pa_Terminate() with a stream open is undefined and invalidates every index. + // The stream goes first whatever happens next: Pa_Terminate() with one open is undefined, + // and every PaDeviceIndex -- device_index_ among them, which this clears -- dies with it. + // Nothing playing is torn down by this, because only reopen_in_place_() can ask for a rescan + // and it only runs on a stream that has already died. + this->discard_ring_tail_(); this->close_stream_(); if (!this->pa_.reinitialize()) { @@ -759,6 +783,7 @@ bool PortAudioSink::reopen_in_place_() { } const StreamFormat format = this->last_format_; + this->discard_ring_tail_(); this->close_stream_(); if (!this->open_stream_(device, format.sample_rate, format.channels, format.bit_depth)) { this->recovery_.reopen_done(false); // open_stream_() has already said why, once @@ -777,6 +802,14 @@ bool PortAudioSink::reopen_in_place_() { return true; } +void PortAudioSink::discard_ring_tail_() { + if (this->bytes_per_frame_ == 0) { + return; // no stream, so close_stream_() has already emptied the ring + } + this->recovery_.discard_frames( + static_cast(this->ring_.available() / this->bytes_per_frame_)); +} + bool PortAudioSink::stream_alive_() const { if (this->stream_ == nullptr) { return false; @@ -840,7 +873,10 @@ int PortAudioSink::pa_callback(const void* /*input*/, void* output, unsigned lon (static_cast(frames_played) / self->stream_rate_); const int64_t finish_us = entered_us + static_cast(std::llround(dac_offset_s * 1e6)); - self->on_frames_played(frames_played, finish_us); + // The gap rides on this report rather than its own: the player sums the frames of reports + // it has not read yet but keeps only the last timestamp. + self->on_frames_played(frames_with_gap(self->gap_handoff_.take(), frames_played), + finish_us); } return paContinue; diff --git a/src/portaudio_sink.h b/src/portaudio_sink.h index bcf9279..c76b82c 100644 --- a/src/portaudio_sink.h +++ b/src/portaudio_sink.h @@ -103,6 +103,10 @@ class PortAudioSink final : public AudioSink { /// The one in-place reopen a dead stream gets, from write(). Caller holds mutex_. /// @return true if a stream is running again. bool reopen_in_place_(); + /// Adds the frames still in the ring to the outage gap: the player has counted them, and the + /// lost stream recovery is about to close will never report them. Only for those closes -- + /// stop(), configure() and clear() end the stream the gap belonged to. Caller holds mutex_. + void discard_ring_tail_(); /// True while the open stream is still being driven by PortAudio. Caller holds mutex_. bool stream_alive_() const; /// Ring size in bytes. Caller holds mutex_ and the format fields are set. @@ -146,6 +150,9 @@ class PortAudioSink final : public AudioSink { StreamFormat last_format_{}; /// Guarded by mutex_, except SinkRecovery::pending(). SinkRecovery recovery_; + /// The outage gap once write() has handed it on, for the callback to retire lock-free with + /// its next report. recovery_ cannot be read there; see OutageGapHandoff. + OutageGapHandoff gap_handoff_; std::atomic volume_{DEFAULT_SINK_VOLUME}; std::atomic muted_{false}; diff --git a/src/pulse_sink.cpp b/src/pulse_sink.cpp index 3891ec2..9e63849 100644 --- a/src/pulse_sink.cpp +++ b/src/pulse_sink.cpp @@ -383,6 +383,9 @@ bool PulseAudioSink::configure(uint32_t sample_rate, uint8_t channels, uint8_t b // Before anything can fail: recovery reopens at this format. this->last_format_ = {sample_rate, channels, bits_per_sample}; + // A new stream starts the producer from zero buffered frames, so an outage gap left over from + // the last one is owed to nobody. Here rather than beside reset(), which a failed open skips. + this->recovery_.forget_discarded_frames(); this->close_stream_(); if (!this->conn_.ready()) { @@ -422,7 +425,11 @@ size_t PulseAudioSink::write(const uint8_t* data, size_t length, uint32_t timeou ? this->bytes_per_frame_ : static_cast(this->last_format_.channels) * (static_cast(this->last_format_.bit_depth) / 8U); - return (frame == 0) ? length : length - (length % frame); + const size_t consumed = (frame == 0) ? length : length - (length % frame); + if (frame != 0) { + this->recovery_.discard_frames(static_cast(consumed / frame)); + } + return consumed; } const size_t bytes_per_frame = this->bytes_per_frame_; @@ -495,6 +502,7 @@ size_t PulseAudioSink::write(const uint8_t* data, size_t length, uint32_t timeou int64_t finish_us = 0; bool have_timing = false; + uint32_t gap_frames = 0; if (frames_done > 0 && this->stream_ != nullptr) { // pa_stream_get_latency() is the server's own measure of time until queued audio plays. const MainloopLock ml(this->conn_.mainloop()); @@ -503,13 +511,29 @@ size_t PulseAudioSink::write(const uint8_t* data, size_t length, uint32_t timeou if (pa_stream_get_latency(this->stream_, &latency_us, &negative) == 0 && negative == 0) { finish_us = now_us() + static_cast(latency_us); have_timing = true; + // Taken with the timestamp, under the same lock, so no report can land between the + // outage being retired and the first real playback being reported. + gap_frames = this->recovery_.take_discarded_frames(); } } + // Frames that reached a stream which then died will never be reported, so they join the gap. + // A clear() during the wait is a flush rather than a loss -- the player has dropped those + // frames already -- and is told apart by the generation. A poll() that closed and reopened the + // stream during the wait moves the generation too and so is taken for a flush: that rare case + // under-counts. + if (frames_done > 0 && !have_timing && + (this->stream_ == nullptr || + (this->stream_failed_.load() && this->stream_generation_ == generation))) { + this->recovery_.discard_frames(static_cast(frames_done)); + } + lock.unlock(); - // Outside the lock, so a callback that touches the sink cannot deadlock. + // Outside the lock, so a callback that touches the sink cannot deadlock. One report for the + // gap and this write together: the player keeps only the last timestamp, so a separate gap + // report could lose it. if (have_timing && this->on_frames_played) { - this->on_frames_played(static_cast(frames_done), finish_us); + this->on_frames_played(frames_with_gap(gap_frames, frames_done), finish_us); } return frames_done * bytes_per_frame; } @@ -524,6 +548,10 @@ void PulseAudioSink::clear() { ++this->stream_generation_; this->space_available_.notify_all(); + // Also before the early return: the player zeroes its buffered-frame count with a flush, so + // the gap is owed to nobody. Only the gap -- the recovery budget is configure()'s to refill. + this->recovery_.forget_discarded_frames(); + if (this->stream_ == nullptr) { return; } diff --git a/src/sink_recovery.cpp b/src/sink_recovery.cpp index 009ac13..f3aa9d7 100644 --- a/src/sink_recovery.cpp +++ b/src/sink_recovery.cpp @@ -107,6 +107,30 @@ void SinkRecovery::escalate_() { } } +void OutageGapHandoff::add(uint32_t frames) { + if (frames == 0) { + return; + } + uint32_t current = this->frames_.load(); + // A compare-exchange rather than load-then-store: the callback can take() between the two, and + // a plain store would hand back the gap it just retired. + while (true) { + const uint32_t room = std::numeric_limits::max() - current; + const uint32_t next = current + ((frames < room) ? frames : room); + if (this->frames_.compare_exchange_weak(current, next)) { + return; + } + } +} + +uint32_t OutageGapHandoff::take() { + return this->frames_.exchange(0); +} + +void OutageGapHandoff::forget() { + this->frames_.store(0); +} + int64_t SinkRecovery::delay_for_(int attempts_made) { int64_t delay = SINK_RESCAN_DELAY_MS; // Doubled in a loop rather than shifted, so raising SINK_RESCAN_ATTEMPTS cannot overflow. diff --git a/src/sink_recovery.h b/src/sink_recovery.h index c68a958..edee8ec 100644 --- a/src/sink_recovery.h +++ b/src/sink_recovery.h @@ -18,6 +18,7 @@ #include #include +#include namespace sendspin_cli { @@ -88,4 +89,35 @@ class SinkRecovery { int64_t rescan_at_ms_{NOT_STAMPED}; }; +/// @brief Carries an outage gap from a sink's locked write() to a callback that takes no lock. +/// +/// PortAudioSink and PipeWireSink report playback from a realtime callback, which SinkRecovery's +/// locking rule puts out of its reach. So write() moves the gap in here, under the sink's lock, +/// once the stream is alive again, and the callback takes it with the first report that has a +/// timestamp to retire it against. Saturates rather than wraps, like discard_frames(). +class OutageGapHandoff { +public: + /// Producer side, under the sink's lock. Safe against a concurrent take(). + void add(uint32_t frames); + + /// Consumer side, from the audio callback: the whole gap, leaving none behind. + uint32_t take(); + + /// For a stream that ended before its gap was retired; see forget_discarded_frames(). + void forget(); + +private: + std::atomic frames_{0}; +}; + +/// @brief The count for one on_frames_played() report that retires `gap_frames` alongside +/// `played_frames`. +/// +/// Saturated rather than wrapped: the report is 32-bit and a gap can already sit at the ceiling. +constexpr uint32_t frames_with_gap(uint32_t gap_frames, uint64_t played_frames) { + const uint64_t total = static_cast(gap_frames) + played_frames; + const uint64_t ceiling = std::numeric_limits::max(); + return static_cast((total < ceiling) ? total : ceiling); +} + } // namespace sendspin_cli diff --git a/tests/sink_recovery_test.cpp b/tests/sink_recovery_test.cpp index 3ecc696..a29e36c 100644 --- a/tests/sink_recovery_test.cpp +++ b/tests/sink_recovery_test.cpp @@ -151,6 +151,48 @@ TEST(SinkRecovery, ResetClearsTheDiscardedGap) { EXPECT_EQ(recovery.take_discarded_frames(), 0U); } +TEST(OutageGapHandoff, HandsTheGapToTheCallbackOnce) { + OutageGapHandoff handoff; + handoff.add(48'000U); + handoff.add(1'000U); + + // A second take() would be a second report of the same outage, which is the playhead jump + // this exists to prevent in the other direction. + EXPECT_EQ(handoff.take(), 49'000U); + EXPECT_EQ(handoff.take(), 0U); +} + +TEST(OutageGapHandoff, SaturatesInsteadOfWrapping) { + OutageGapHandoff handoff; + handoff.add(std::numeric_limits::max() - 10U); + handoff.add(100U); + + EXPECT_EQ(handoff.take(), std::numeric_limits::max()); +} + +TEST(OutageGapHandoff, ForgettingDropsAGapTheCallbackHasNotTaken) { + OutageGapHandoff handoff; + handoff.add(48'000U); + + // What clear() and configure() do: the player starts the next stream from zero, so a gap + // still waiting for the callback is owed to nobody. + handoff.forget(); + EXPECT_EQ(handoff.take(), 0U); + + // And the next outage is counted from nothing rather than refused. + handoff.add(480U); + EXPECT_EQ(handoff.take(), 480U); +} + +TEST(OutageGapHandoff, TheReportCountSaturatesInsteadOfWrapping) { + EXPECT_EQ(frames_with_gap(0U, 480U), 480U); + EXPECT_EQ(frames_with_gap(672'000U, 480U), 672'480U); + EXPECT_EQ(frames_with_gap(std::numeric_limits::max() - 5U, 480U), + std::numeric_limits::max()); + EXPECT_EQ(frames_with_gap(1U, std::numeric_limits::max() - 1U), + std::numeric_limits::max()); +} + TEST(SinkRecovery, EveryFurtherWriteOfTheOutageIsToldToDiscard) { SinkRecovery recovery; escalate(recovery);