Integrate delay processing with call jitter and verify PCM handoffs

src/call/call_jitter.cpp

@@ -1,28 +1,22 @@
-// Буфер звонка: храним речь в порядке доставки, ускоряем накопление без изменения высоты голоса.
+// Сетевой запас и decode-on-pull. Обработкой PCM и фазой кадров владеет CallDelay.
 #include "call_jitter.h"
 #include "call_jitter_window.h"
 #include "call_timeline.h"
+#include "call_delay.h"
 #include <chrono>
 #include "../../lib/debug_config.h"
 #include "../../lib/mem.h"
-#include "soundtouch/SoundTouch.h"
 
 #include <algorithm>
 #include <cmath>
 #include <cstring>
 #include <mutex>
 #include <new>
-#include <vector>
 
 static constexpr int kSampleRate = 48000;
 static constexpr int kFrameSamples = 960;
 static constexpr int kMaxEncoded = 512;
-static constexpr double kMaxTempo = 1.6;
-static constexpr int kTempoRampMs = 500;
-
-template<typename T> static T clamp_value(T value, T low, T high) {
-    return std::max(low, std::min(value, high));
-}
+static unsigned jitter_decode(void* context, int16_t* pcm, unsigned capacity);
 
 struct call_jitter {
     std::mutex mtx;
@@ -41,14 +35,48 @@ struct call_jitter {
     call_jitter_range range = {0, 0, CALL_JITTER_MIN_RESERVE_MS}; /* опубликованный снимок под mtx */
     call_jitter_decode_fn decode;
     void* arg;
-    soundtouch::SoundTouch st;
-    int16_t pcm[kFrameSamples];
-    float input[kFrameSamples];
-    std::vector<float> output;
-    bool primed = false, catching_up = false;
-    double tempo = 1.0, ema_ms = CALL_JITTER_MIN_RESERVE_MS;
+    CallDelay delay{jitter_decode, this, [](void* context, const soundtouch::SourceSpan& span) {
+        static_cast<call_jitter*>(context)->timeline.source(span);
+    }, this};
+    bool primed = false;
+    int budget = 0, attempts = 0, error_result = 0, error_len = 0;
+    uint32_t decode_errors = 0;
 };
 
+/* Единственный владелец декодера — playback worker; сетевой mutex только на извлечении пакета. */
+static unsigned jitter_decode(void* context, int16_t* pcm, unsigned capacity) {
+    auto* j = static_cast<call_jitter*>(context);
+    uint8_t enc[kMaxEncoded];
+    while (j->attempts < j->budget) {
+        int len;
+        call_capture_time timing;
+        {
+            std::lock_guard<std::mutex> lock(j->mtx);
+            if (!j->count) return 0;
+            len = j->lens[j->head]; std::memcpy(enc, j->ring[j->head], (size_t)len);
+            timing = j->timing[j->head];
+            j->head = (j->head + 1) % CALL_JITTER_MAX_FRAMES; --j->count;
+        }
+        ++j->attempts;
+        int n = j->decode(j->arg, enc, len, pcm, (int)capacity);
+        if (n <= 0 || n > (int)capacity) {
+            ++j->decode_errors; j->error_result = n; j->error_len = len;
+            continue;
+        }
+        int64_t pts = timing.pts_us == INT64_MIN ? INT64_MIN : timing.pts_us - (int64_t)timing.delay_samples * 1000000 / kSampleRate;
+        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;
+        return (unsigned)n;
+    }
+    return 0;
+}
+
 extern "C" {
 struct call_jitter* call_jitter_create(call_jitter_decode_fn decode, void* arg, int max_reserve_ms) {
     if (max_reserve_ms < 500 || max_reserve_ms > 2000) {
@@ -67,21 +95,13 @@ struct call_jitter* call_jitter_create(call_jitter_decode_fn decode, void* arg,
     auto* j = new (memory) call_jitter;
     j->arrival_window.max_reserve = max_reserve_ms;
     j->decode = decode; j->arg = arg;
-    j->st.setSampleRate(kSampleRate); j->st.setChannels(1); j->st.setTempo(1.0);
-    j->st.setSetting(SETTING_USE_QUICKSEEK, 1);
-    j->st.setSetting(SETTING_SEQUENCE_MS, 20);
-    j->st.setSetting(SETTING_SEEKWINDOW_MS, 10);
-    j->st.setSetting(SETTING_OVERLAP_MS, 4);
-    j->st.setSetting(SETTING_USE_AA_FILTER, 0);
     j->timeline.reset(j->playback_epoch);
-    if (!j->st.setSourceCallback([](void* context, const soundtouch::SourceSpan& span) {
-            static_cast<call_jitter*>(context)->timeline.source(span);
-        }, j)) {
-        DEBUG_ERROR(DEBUG_CATEGORY_CALL, "call_jitter: SoundTouch source tracking unavailable");
+    if (!j->delay.valid()) {
+        DEBUG_ERROR(DEBUG_CATEGORY_CALL, "call_jitter: delay processor unavailable");
         j->~call_jitter(); u_free(j); return nullptr;
     }
-    DEBUG_INFO(DEBUG_CATEGORY_CALL, "call_jitter: created capacity=%dms target=%dms max_reserve=%dms tempo=1.0..%.1f ramp=%dms",
-               CALL_JITTER_MAX_FRAMES * CALL_JITTER_FRAME_MS, CALL_JITTER_MIN_RESERVE_MS, max_reserve_ms, kMaxTempo, kTempoRampMs);
+    DEBUG_INFO(DEBUG_CATEGORY_CALL, "call_jitter: created capacity=%dms reserve=%d..%dms adaptive_pcm=1",
+               CALL_JITTER_MAX_FRAMES * CALL_JITTER_FRAME_MS, CALL_JITTER_MIN_RESERVE_MS, max_reserve_ms);
     return j;
 }
 
@@ -116,9 +136,8 @@ int call_jitter_pull(struct call_jitter* j, int16_t* out, int max_samples) {
     if (!j || !out || max_samples <= 0) return 0;
     int count, reserve;
     { std::lock_guard<std::mutex> lock(j->mtx); count = j->count; reserve = j->arrival_window.reserve; }
-    // Готовый PCM учитываем в длительности исходной речи.
-    double depth = count * CALL_JITTER_FRAME_MS
-        + (j->st.numUnprocessedSamples() + j->st.numSamples() * j->tempo) * 1000.0 / kSampleRate;
+    double depth_samples = count * kFrameSamples + j->delay.stats().pending;
+    double depth = depth_samples / 48;
     int64_t now_ms = std::chrono::duration_cast<std::chrono::milliseconds>(
         std::chrono::steady_clock::now().time_since_epoch()).count();
     j->depth_window.observe(now_ms, (int)std::lround(depth));
@@ -128,81 +147,37 @@ int call_jitter_pull(struct call_jitter* j, int16_t* out, int max_samples) {
         j->range.min_ms = (int)j->depth_window.low; j->range.max_ms = (int)j->depth_window.high;
     }
     if (!j->primed) {
-        if (depth < reserve) {
+        // После первого полного блока остаётся сетевой запас, не скрытый запас DSP.
+        if (depth_samples < reserve * 48 + max_samples) {
             std::lock_guard<std::mutex> lock(j->mtx);
             j->diagnostics.depth_ms = (int)depth; j->diagnostics.reserve_ms = reserve;
             j->diagnostics.tempo_x100 = j->tempo_x100; j->diagnostics.waiting = 1;
             j->diagnostics.pcm_samples = 0;
             return 0;
         }
-        j->primed = true;
         resumed = 1;
     }
-    j->ema_ms += 0.2 * (depth - j->ema_ms);
-    double target = 1.0 + (kMaxTempo - 1.0) * clamp_value((std::min(depth, j->ema_ms) - reserve - 15.0) / 925.0, 0.0, 1.0);
-    // Линейный предел 1.0→1.6 за 500мс PCM. Снижение следует за глубиной буфера с тем же пределом.
-    double step = (kMaxTempo - 1.0) * max_samples * 1000.0 / (kSampleRate * kTempoRampMs);
-    j->tempo += clamp_value(target - j->tempo, -step, step);
-    j->st.setTempo(j->tempo);
-    bool catching = j->tempo > 1.05;
-    j->catching_up = catching;
-    uint32_t decode_errors = 0;
-    int error_result = 0, error_len = 0;
-    uint8_t enc[kMaxEncoded];
-    // Бюджет учитывает ускорение 1.6 и начальное накопление SoundTouch, включая битые пакеты.
-    int budget = std::min(CALL_JITTER_MAX_FRAMES, 8 + max_samples / 480);
-    int attempted = 0;
-    while (j->st.numSamples() < (uint)max_samples && attempted++ < budget) {
-        int len;
-        call_capture_time timing;
-        {
-            std::lock_guard<std::mutex> lock(j->mtx);
-            if (!j->count) break;
-            len = j->lens[j->head]; std::memcpy(enc, j->ring[j->head], (size_t)len);
-            timing = j->timing[j->head];
-            j->head = (j->head + 1) % CALL_JITTER_MAX_FRAMES; j->count--;
-        }
-        int n = j->decode(j->arg, enc, len, j->pcm, kFrameSamples);
-        if (n <= 0 || n > kFrameSamples) {
-            ++decode_errors;
-            error_result = n; error_len = len;
-            continue;
-        }
-        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;
-        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);
-    int n = (int)j->st.receiveSamples(j->output.data(), (uint)max_samples);
-    for (int i = 0; i < n; i++) out[i] = (int16_t)std::lrintf(clamp_value(j->output[i] * 32767.0f, -32768.0f, 32767.0f));
-    int processing_ms = (int)std::lround((j->st.numUnprocessedSamples() + j->st.numSamples() * j->tempo) * 1000.0 / kSampleRate);
+    j->budget = std::min(CALL_JITTER_MAX_FRAMES, 8 + max_samples / 480);
+    j->attempts = 0; j->decode_errors = 0;
+    int n = (int)j->delay.pull(out, (unsigned)max_samples, count * kFrameSamples, reserve * 48);
+    auto state = j->delay.stats();
+    j->primed = n == max_samples;
     {
         std::lock_guard<std::mutex> lock(j->mtx);
         j->timeline.read((unsigned)n);
-        j->processing_ms = processing_ms; j->tempo_x100 = (int)std::lround(j->tempo * 100);
+        j->processing_ms = (int)std::lround(state.pending / 48); j->tempo_x100 = (int)std::lround(state.tempo * 100);
         if (n < max_samples) j->underruns++;
         j->diagnostics.depth_ms = (int)depth;
         j->diagnostics.reserve_ms = reserve;
         j->diagnostics.tempo_x100 = j->tempo_x100;
-        j->diagnostics.decode_errors += decode_errors;
-        if (decode_errors) { j->diagnostics.error_result = error_result; j->diagnostics.error_len = error_len; }
-        if (attempted > budget) ++j->diagnostics.budget_hits;
-        j->diagnostics.resumed |= resumed;
-        j->diagnostics.catching_up = catching;
+        j->diagnostics.decode_errors += j->decode_errors;
+        if (j->decode_errors) { j->diagnostics.error_result = j->error_result; j->diagnostics.error_len = j->error_len; }
+        if (j->attempts == j->budget && n < max_samples) ++j->diagnostics.budget_hits;
+        j->diagnostics.resumed |= resumed && n == max_samples;
+        j->diagnostics.catching_up = state.tempo > 1.05;
         j->diagnostics.waiting = n < max_samples;
         j->diagnostics.pcm_samples = n;
     }
-    if (n < max_samples) {
-        j->primed = false;
-    }
     return n;
 }
 
@@ -223,6 +198,7 @@ void call_jitter_get_diagnostics(struct call_jitter* j, struct call_jitter_diagn
 void call_jitter_silence(struct call_jitter* j, unsigned samples) {
     if (!j) return;
     std::lock_guard<std::mutex> lock(j->mtx);
+    j->delay.silence(samples);
     j->timeline.silence(samples);
 }
 
@@ -246,11 +222,10 @@ void call_jitter_reset(struct call_jitter* j) {
     j->head = (j->head + discard) % CALL_JITTER_MAX_FRAMES;
     j->count = keep;
     j->dropped += discard;
-    j->st.clear(); j->st.setTempo(1.0);
+    j->delay.reset();
     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->primed = false;
     j->processing_ms = 0; j->tempo_x100 = 100;
     j->depth_window = {};
     j->diagnostics = {};