diff --git a/src/alsa_sink.cpp b/src/alsa_sink.cpp index 00534f8..1ffef9e 100644 --- a/src/alsa_sink.cpp +++ b/src/alsa_sink.cpp @@ -27,6 +27,7 @@ #include #include #include +#include #include #include #include @@ -467,6 +468,9 @@ bool AlsaAudioSink::configure(uint32_t sample_rate, uint8_t channels, uint8_t bi // Before anything can fail: poll() reopens at this format. this->last_format_ = {sample_rate, channels, bits_per_sample}; + // A new stream starts from zero buffered frames; here, not beside reset(), so a failed open + // drops the gap too. + this->recovery_.forget_discarded_frames(); if (this->pcm_ != nullptr && this->rate_ == sample_rate && this->channels_ == channels && this->bits_ == bits_per_sample) { @@ -505,6 +509,7 @@ size_t AlsaAudioSink::write(const uint8_t* data, size_t length, uint32_t timeout size_t bytes_per_frame = 0; int64_t finish_us = 0; bool have_timing = false; + uint32_t gap_frames = 0; { const std::lock_guard lock(this->device_mutex_); @@ -522,7 +527,11 @@ size_t AlsaAudioSink::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; } bytes_per_frame = this->bytes_per_frame_; @@ -607,13 +616,23 @@ size_t AlsaAudioSink::write(const uint8_t* data, size_t length, uint32_t timeout if (snd_pcm_delay(this->pcm_, &delay) == 0 && delay >= 0) { finish_us = now_us() + ((static_cast(delay) * 1000000) / this->rate_); have_timing = true; + // Taken with the timestamp under the same lock, so nothing reports between them. + gap_frames = this->recovery_.take_discarded_frames(); } } + + // recover_() closed the device mid-write: these bytes are consumed but never timestamped. + if (frames_done > 0 && this->pcm_ == nullptr) { + this->recovery_.discard_frames(static_cast(frames_done)); + } } - // 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 covers + // the gap and this write: the player keeps only the last timestamp it has not read. if (have_timing && this->on_frames_played) { - this->on_frames_played(static_cast(frames_done), finish_us); + const uint64_t played = static_cast(gap_frames) + frames_done; + const uint64_t ceiling = std::numeric_limits::max(); + this->on_frames_played(static_cast(std::min(played, ceiling)), finish_us); } return frames_done * bytes_per_frame; @@ -625,6 +644,10 @@ void AlsaAudioSink::clear() { // Safe only under device_mutex_; PortAudioSink::clear() must not do the same. this->current_multiplier_ = this->target_multiplier_.load(std::memory_order_relaxed); + // Before the early return: the gap's stream is over. Only the gap goes; reset() owns the + // budget. + this->recovery_.forget_discarded_frames(); + if (this->pcm_ == nullptr) { return; } @@ -698,6 +721,7 @@ void AlsaAudioSink::poll(int64_t now_ms) { this->close_device_(); return; } + // The gap stays pending: reporting it here would race write()'s first timed report. this->recovery_.rescan_done(true); cli_log(LogLevel::INFO, "alsa: '%s' is back -- recovered without waiting for the next stream", this->device_.c_str()); diff --git a/src/sink_recovery.cpp b/src/sink_recovery.cpp index ef4bba8..009ac13 100644 --- a/src/sink_recovery.cpp +++ b/src/sink_recovery.cpp @@ -14,6 +14,8 @@ #include "sink_recovery.h" +#include + namespace sendspin_cli { bool SinkRecovery::reopen_due() { @@ -73,6 +75,21 @@ bool SinkRecovery::pending() const { return this->rescan_owed_.load(std::memory_order_relaxed); } +void SinkRecovery::discard_frames(uint32_t frames) { + const uint32_t room = std::numeric_limits::max() - this->discarded_frames_; + this->discarded_frames_ += (frames < room) ? frames : room; +} + +uint32_t SinkRecovery::take_discarded_frames() { + const uint32_t frames = this->discarded_frames_; + this->discarded_frames_ = 0; + return frames; +} + +void SinkRecovery::forget_discarded_frames() { + this->discarded_frames_ = 0; +} + void SinkRecovery::reset() { this->reopen_spent_ = false; this->rescan_spent_ = false; @@ -80,6 +97,7 @@ void SinkRecovery::reset() { this->rescan_attempts_ = 0; this->rescan_owed_.store(false, std::memory_order_relaxed); this->rescan_at_ms_ = NOT_STAMPED; + this->discarded_frames_ = 0; } void SinkRecovery::escalate_() { diff --git a/src/sink_recovery.h b/src/sink_recovery.h index f1490b7..c68a958 100644 --- a/src/sink_recovery.h +++ b/src/sink_recovery.h @@ -54,6 +54,16 @@ class SinkRecovery { /// True while a rescan is still owed. The one method safe to call without the lock. bool pending() const; + /// Records frames accepted and discarded with no device to play them; saturates. + /// Stays pending through a recovered rescan until a timed write takes it. + void discard_frames(uint32_t frames); + + /// Returns and clears the frames accumulated by discard_frames(). + uint32_t take_discarded_frames(); + + /// Drops the discarded-frame count, leaving the recovery budget alone. + void forget_discarded_frames(); + /// Refills the budget; call only when configure() really opened a stream. void reset(); @@ -71,6 +81,8 @@ class SinkRecovery { /// Set from handing out an attempt until rescan_done(); blocks re-arming and double counting. bool rescan_in_flight_{false}; int rescan_attempts_{0}; + /// Frames accepted with no device to play them, not yet retired; see discard_frames(). + uint32_t discarded_frames_{0}; /// Read by the main loop without the sink's lock; see pending(). std::atomic rescan_owed_{false}; int64_t rescan_at_ms_{NOT_STAMPED}; diff --git a/tests/sink_recovery_test.cpp b/tests/sink_recovery_test.cpp index 7326cf3..3ecc696 100644 --- a/tests/sink_recovery_test.cpp +++ b/tests/sink_recovery_test.cpp @@ -19,6 +19,7 @@ #include #include +#include namespace sendspin_cli { namespace { @@ -75,6 +76,81 @@ TEST(SinkRecovery, AFailedReopenEscalatesToTheRescan) { EXPECT_TRUE(recovery.pending()); } +TEST(SinkRecovery, ReturnsTheDiscardedGapOnceWhenADeviceComesBack) { + SinkRecovery recovery; + escalate(recovery); + + // Fourteen seconds at 48 kHz, accepted while no DAC played them. + recovery.discard_frames(14U * 48'000U); + + EXPECT_EQ(recovery.take_discarded_frames(), 672'000U); + EXPECT_EQ(recovery.take_discarded_frames(), 0U); +} + +TEST(SinkRecovery, DiscardedGapSaturatesInsteadOfWrapping) { + SinkRecovery recovery; + recovery.discard_frames(std::numeric_limits::max() - 10U); + recovery.discard_frames(100U); + + EXPECT_EQ(recovery.take_discarded_frames(), std::numeric_limits::max()); +} + +TEST(SinkRecovery, TheDiscardedGapOutlivesARecoveredRescanUntilTaken) { + SinkRecovery recovery; + escalate(recovery); + recovery.discard_frames(48'000U); + ASSERT_GE(rescan_fires_at(recovery, T0, T0 + 10 * SINK_RESCAN_DELAY_MS), T0); + + // The reopen has no device timestamp to retire the gap against; the first timed write does. + recovery.rescan_done(true); + + EXPECT_EQ(recovery.take_discarded_frames(), 48'000U); + EXPECT_EQ(recovery.take_discarded_frames(), 0U); +} + +TEST(SinkRecovery, ForgettingTheDiscardedGapLeavesTheBudgetAlone) { + SinkRecovery recovery; + escalate(recovery); + recovery.discard_frames(48'000U); + + // What a flush does with no device open: the outage is still owed what it was. + recovery.forget_discarded_frames(); + + EXPECT_EQ(recovery.take_discarded_frames(), 0U); + EXPECT_TRUE(recovery.pending()); + EXPECT_FALSE(recovery.reopen_due()); + EXPECT_EQ(rescan_fires_at(recovery, T0, T0 + 10 * SINK_RESCAN_DELAY_MS), + T0 + SINK_RESCAN_DELAY_MS); +} + +TEST(SinkRecovery, ForgettingTheDiscardedGapDoesNotRefillASpentBudget) { + SinkRecovery recovery; + escalate(recovery); + int64_t now = T0; + for (int attempt = 0; attempt < SINK_RESCAN_ATTEMPTS; ++attempt) { + now = rescan_fires_at(recovery, now, now + 100 * SINK_RESCAN_DELAY_MS); + ASSERT_GT(now, 0) << "attempt " << attempt << " never fired"; + recovery.rescan_done(false); + } + recovery.discard_frames(48'000U); + + recovery.forget_discarded_frames(); + + EXPECT_FALSE(recovery.pending()); + EXPECT_FALSE(recovery.reopen_due()); + EXPECT_EQ(rescan_fires_at(recovery, now, now + 100 * SINK_RESCAN_DELAY_MS), -1); +} + +TEST(SinkRecovery, ResetClearsTheDiscardedGap) { + SinkRecovery recovery; + escalate(recovery); + recovery.discard_frames(48'000U); + + recovery.reset(); + + EXPECT_EQ(recovery.take_discarded_frames(), 0U); +} + TEST(SinkRecovery, EveryFurtherWriteOfTheOutageIsToldToDiscard) { SinkRecovery recovery; escalate(recovery);