From 7bac999d9338a14257c22b8d4bf1c8ec63c436ca Mon Sep 17 00:00:00 2001 From: evgeny Date: Wed, 14 Oct 2026 19:27:54 +0200 Subject: [PATCH] Handle source discontinuities and reject stale device epochs --- src/call/call_jitter.cpp | 12 +++++++++++- src/call/call_timeline.cpp | 23 +++++++++++++++++------ src/call/call_timeline.h | 4 ++-- 3 files changed, 30 insertions(+), 9 deletions(-) diff --git a/src/call/call_jitter.cpp b/src/call/call_jitter.cpp index 284e27a5..2e049eb5 100644 --- a/src/call/call_jitter.cpp +++ b/src/call/call_jitter.cpp @@ -32,6 +32,8 @@ struct call_jitter { call_capture_time timing[CALL_JITTER_MAX_FRAMES]; CallTimeline timeline; uint64_t playback_epoch = 1; + uint64_t source_epoch = 0; + int64_t capture_end = INT64_MIN; int head = 0, count = 0; uint32_t dropped = 0, underruns = 0; call_jitter_diagnostics diagnostics{}; @@ -168,7 +170,14 @@ int call_jitter_pull(struct call_jitter* j, int16_t* out, int max_samples) { } for (int i = 0; i < n; i++) j->input[i] = j->pcm[i] / 32768.0f; int64_t pts = timing.pts_us == INT64_MIN ? INT64_MIN : timing.pts_us - (int64_t)timing.delay_samples * 1000000 / kSampleRate; - j->timeline.feed(pts, (unsigned)n, timing.epoch); + bool discontinuity = j->source_epoch != timing.epoch || (timing.pts_us != INT64_MIN && j->capture_end != INT64_MIN + && (timing.pts_us < j->capture_end - 2000 || timing.pts_us > j->capture_end + 2000)); + unsigned prefix = discontinuity ? std::min((unsigned)n, timing.delay_samples) : 0; + if (prefix) j->timeline.feed(INT64_MIN, prefix, timing.epoch); + if (prefix < (unsigned)n) j->timeline.feed(pts == INT64_MIN ? pts : pts + prefix * 1000000ll / kSampleRate, + n - prefix, timing.epoch); + j->source_epoch = timing.epoch; + j->capture_end = timing.pts_us == INT64_MIN ? INT64_MIN : timing.pts_us + n * 1000000ll / kSampleRate; j->st.putSamples(j->input, (uint)n); } j->output.resize((size_t)max_samples); @@ -231,6 +240,7 @@ void call_jitter_reset(struct call_jitter* j) { j->dropped += discard; j->st.clear(); j->st.setTempo(1.0); j->timeline.reset(++j->playback_epoch); + j->source_epoch = 0; j->capture_end = INT64_MIN; j->primed = false; j->catching_up = false; j->tempo = 1.0; j->ema_ms = j->arrival_window.reserve; j->processing_ms = 0; j->tempo_x100 = 100; diff --git a/src/call/call_timeline.cpp b/src/call/call_timeline.cpp index 1ffb7182..423d4dac 100644 --- a/src/call/call_timeline.cpp +++ b/src/call/call_timeline.cpp @@ -11,10 +11,10 @@ double CallTimeline::at(const Span& s, double position) { void CallTimeline::reset(uint64_t newEpoch) { input.clear(); ready.clear(); output.clear(); - inputWritten = generated = consumed = written = anchorPosition = minimumPosition = lastSourceEpoch = 0; + inputWritten = generated = consumed = written = anchorPosition = minimumPosition = 0; epoch = newEpoch; anchorTime = anchorReceived = 0; lastPts = -1; endPts = 0; endEpoch = 0; sourceError = 0; - outputFailed = haveAnchor = measured = endValid = false; + outputFailed = deviceFailed = haveAnchor = measured = endValid = false; DEBUG_DEBUG(DEBUG_CATEGORY_CALL, "audio timeline reset epoch=%llu", (unsigned long long)epoch); } @@ -78,6 +78,17 @@ void CallTimeline::source(const soundtouch::SourceSpan& s) { void CallTimeline::appendOutput(const Span& s) { if (outputFailed) return; + // Источники без PTS и длинные локальные паузы не расходуют по записи на каждый pull. + if (output.count) { + Span& previous = output.data[(output.head + output.count - 1) % output.data.size()]; + bool sameHold = previous.pts == previous.end && s.pts == s.end && previous.end == s.pts; + if (previous.first + previous.count == s.first && previous.epoch == s.epoch && previous.valid == s.valid + && (!s.valid || sameHold) && previous.count <= UINT32_MAX - s.count) { + previous.count += s.count; previous.end = s.end; + endPts = s.end; endEpoch = s.epoch; endValid = s.valid; + return; + } + } if (!output.push(s)) { outputFailed = true; DEBUG_ERROR(DEBUG_CATEGORY_CALL, "audio timeline output overflow epoch=%llu written=%llu anchor=%llu spans=%zu", @@ -118,12 +129,12 @@ void CallTimeline::silence(unsigned samples) { /* at_us — время конца известного устройству участка; будущая отметка учитывает очередь backend. */ void CallTimeline::progress(uint64_t deviceEpoch, uint64_t position, int64_t atUs, int64_t receivedUs, bool isMeasured) { - if (deviceEpoch != epoch) return; + if (deviceEpoch != epoch || deviceFailed) return; if (position > written || atUs < 0 || receivedUs < 0 || (haveAnchor && (position < anchorPosition || receivedUs < anchorReceived))) { DEBUG_WARN(DEBUG_CATEGORY_CALL, "audio clock invalid device progress epoch=%llu position=%llu written=%llu time=%lld received=%lld", (unsigned long long)epoch, (unsigned long long)position, (unsigned long long)written, (long long)atUs, (long long)receivedUs); - haveAnchor = false; return; + deviceFailed = true; haveAnchor = false; return; } anchorPosition = position; anchorTime = atUs; anchorReceived = receivedUs; measured = isMeasured; haveAnchor = true; uint64_t pending = atUs > receivedUs ? (uint64_t)std::ceil((atUs - receivedUs) / sampleUs) : 0; @@ -135,7 +146,7 @@ void CallTimeline::progress(uint64_t deviceEpoch, uint64_t position, int64_t atU call_clock_snapshot CallTimeline::clock(int64_t nowUs) { call_clock_snapshot result{}; result.epoch = epoch; result.written_samples = written; result.pts_us = lastPts; result.measured = measured; - if (outputFailed || !haveAnchor || !output.count || nowUs < anchorReceived || nowUs - anchorReceived > 250000) return result; + if (outputFailed || deviceFailed || !haveAnchor || !output.count || nowUs < anchorReceived || nowUs - anchorReceived > 250000) return result; uint64_t pending = anchorTime > nowUs ? (uint64_t)std::ceil((anchorTime - nowUs) / sampleUs) : 0; uint64_t position = pending > anchorPosition ? 0 : anchorPosition - pending; position = std::max(position, minimumPosition); @@ -147,7 +158,7 @@ call_clock_snapshot CallTimeline::clock(int64_t nowUs) { int64_t pts = (int64_t)std::llround(at(s, position)); if (pts < 0 || (lastPts >= 0 && pts < lastPts - 100000)) return result; result.held_back_us = pts < lastPts ? (uint32_t)(lastPts - pts) : 0; - lastPts = std::max(pts, lastPts); lastSourceEpoch = s.epoch; + lastPts = std::max(pts, lastPts); result.pts_us = lastPts; result.valid = 1; return result; } diff --git a/src/call/call_timeline.h b/src/call/call_timeline.h index a340b8a1..67ae058c 100644 --- a/src/call/call_timeline.h +++ b/src/call/call_timeline.h @@ -45,12 +45,12 @@ private: Ring<1024> ready; Ring<2048> output; uint64_t inputWritten = 0, generated = 0, consumed = 0, written = 0, epoch = 0; - uint64_t anchorPosition = 0, minimumPosition = 0, lastSourceEpoch = 0; + uint64_t anchorPosition = 0, minimumPosition = 0; int64_t anchorTime = 0, anchorReceived = 0, lastPts = -1; double endPts = 0; uint64_t endEpoch = 0; int sourceError = 0; - bool outputFailed = false, haveAnchor = false, measured = false, endValid = false; + bool outputFailed = false, deviceFailed = false, haveAnchor = false, measured = false, endValid = false; static double at(const Span& s, double position); const Span* findInput(uint64_t position) const; void appendOutput(const Span& s);