utun2/src/call/call_jitter.cpp

// Сетевой запас и 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 <algorithm>
#include <cmath>
#include <cstring>
#include <mutex>
#include <new>

static constexpr int kSampleRate = 48000;
static constexpr int kFrameSamples = 960;
static constexpr int kMaxEncoded = 512;
static unsigned jitter_decode(void* context, int16_t* pcm, unsigned capacity);

struct call_jitter {
    std::mutex mtx;
    uint8_t ring[CALL_JITTER_MAX_FRAMES][kMaxEncoded];
    int lens[CALL_JITTER_MAX_FRAMES];
    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{};
    int processing_ms = 0, tempo_x100 = 100; /* снимок аудиопотока под mtx */
    call_jitter_window depth_window, arrival_window;
    call_jitter_range range = {0, 0, CALL_JITTER_MIN_RESERVE_MS}; /* опубликованный снимок под mtx */
    call_jitter_decode_fn decode;
    void* arg;
    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) {
        DEBUG_ERROR(DEBUG_CATEGORY_CALL, "call_jitter: invalid max reserve=%dms", max_reserve_ms);
        return nullptr;
    }
    if (!decode) {
        DEBUG_ERROR(DEBUG_CATEGORY_CALL, "call_jitter: missing decoder");
        return nullptr;
    }
    void* memory = u_malloc(sizeof(call_jitter));
    if (!memory) {
        DEBUG_ERROR(DEBUG_CATEGORY_CALL, "call_jitter: allocation failed");
        return nullptr;
    }
    auto* j = new (memory) call_jitter;
    j->arrival_window.max_reserve = max_reserve_ms;
    j->decode = decode; j->arg = arg;
    j->timeline.reset(j->playback_epoch);
    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 reserve=%d..%dms adaptive_pcm=1",
               CALL_JITTER_MAX_FRAMES * CALL_JITTER_FRAME_MS, CALL_JITTER_MIN_RESERVE_MS, max_reserve_ms);
    return j;
}

void call_jitter_destroy(struct call_jitter* j) {
    if (!j) return;
    j->~call_jitter(); u_free(j);
}

void call_jitter_push(struct call_jitter* j, const uint8_t* enc, int len, uint32_t sent_ms, int64_t received_ms,
                      const struct call_capture_time* timing) {
    if (!j || !enc || len <= 0 || len > kMaxEncoded || received_ms < 0) {
        DEBUG_WARN(DEBUG_CATEGORY_CALL, "call_jitter: invalid frame len=%d received=%lldms", len, (long long)received_ms);
        return;
    }
    std::lock_guard<std::mutex> lock(j->mtx);
    if (received_ms < j->arrival_window.last_received_ms) {
        DEBUG_WARN(DEBUG_CATEGORY_CALL, "call_jitter: receive clock moved backwards previous=%lldms now=%lldms",
                   (long long)j->arrival_window.last_received_ms, (long long)received_ms);
        return;
    }
    j->arrival_window.observe_arrival(received_ms, sent_ms);
    j->range.reserve_ms = j->arrival_window.reserve;
    if (j->count == CALL_JITTER_MAX_FRAMES) {
        j->head = (j->head + 1) % CALL_JITTER_MAX_FRAMES; j->count--; j->dropped++;
    }
    int idx = (j->head + j->count) % CALL_JITTER_MAX_FRAMES;
    j->timing[idx] = timing ? *timing : call_capture_time{INT64_MIN, 0, 0};
    std::memcpy(j->ring[idx], enc, (size_t)len); j->lens[idx] = len; j->count++;
}

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; }
    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));
    int resumed = 0;
    {
        std::lock_guard<std::mutex> lock(j->mtx);
        j->range.min_ms = (int)j->depth_window.low; j->range.max_ms = (int)j->depth_window.high;
    }
    if (!j->primed) {
        // После первого полного блока остаётся сетевой запас, не скрытый запас 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;
        }
        resumed = 1;
    }
    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 = (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 += 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;
        j->diagnostics.delay_mode = (int)state.mode;
        j->diagnostics.tempo_ppm = (int)std::lround(state.tempo * 1000000);
        j->diagnostics.phase_samples = state.phase;
        j->diagnostics.cut_samples = state.cuts;
        j->diagnostics.stretch_entries = state.entered; j->diagnostics.stretch_exits = state.exited;
    }
    return n;
}

void call_jitter_get_diagnostics(struct call_jitter* j, struct call_jitter_diagnostics* out, int reset_events) {
    std::lock_guard<std::mutex> lock(j->mtx);
    *out = j->diagnostics;
    out->encoded_ms = j->count * CALL_JITTER_FRAME_MS;
    out->processing_ms = j->processing_ms;
    out->reserve_ms = j->arrival_window.reserve;
    out->arrival_spread_ms = j->arrival_window.high - j->arrival_window.low;
    out->underruns = j->underruns;
    if (reset_events) {
        j->diagnostics.decode_errors = j->diagnostics.budget_hits = 0;
        j->diagnostics.resumed = 0;
    }
}

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);
}

void call_jitter_progress(struct call_jitter* j, uint64_t epoch, uint64_t position, int64_t at_us, int64_t now_us, int measured) {
    if (!j) return;
    std::lock_guard<std::mutex> lock(j->mtx);
    j->timeline.progress(epoch, position, at_us, now_us, measured != 0);
}

void call_jitter_clock(struct call_jitter* j, int64_t now_us, struct call_clock_snapshot* out) {
    if (!j || !out) return;
    std::lock_guard<std::mutex> lock(j->mtx);
    *out = j->timeline.clock(now_us);
}

void call_jitter_reset(struct call_jitter* j) {
    if (!j) return;
    std::lock_guard<std::mutex> lock(j->mtx);
    int keep = std::min(j->count, (j->arrival_window.reserve + CALL_JITTER_FRAME_MS - 1) / CALL_JITTER_FRAME_MS);
    int discard = j->count - keep;
    j->head = (j->head + discard) % CALL_JITTER_MAX_FRAMES;
    j->count = keep;
    j->dropped += discard;
    j->delay.reset();
    j->timeline.reset(++j->playback_epoch);
    j->source_epoch = 0; j->capture_end = INT64_MIN;
    j->primed = false;
    j->processing_ms = 0; j->tempo_x100 = 100;
    j->depth_window = {};
    j->diagnostics = {};
    j->range.min_ms = j->range.max_ms = keep * CALL_JITTER_FRAME_MS;
    DEBUG_INFO(DEBUG_CATEGORY_CALL, "call_jitter: device discontinuity discarded=%dms kept=%dms reserve=%dms",
               discard * CALL_JITTER_FRAME_MS, keep * CALL_JITTER_FRAME_MS, j->arrival_window.reserve);
}

void call_jitter_get_stats(struct call_jitter* j, int* depth_ms, int* tempo_x100,
                           uint32_t* dropped, uint32_t* underruns, struct call_jitter_range* range) {
    if (!j) return;
    std::lock_guard<std::mutex> lock(j->mtx);
    if (depth_ms) *depth_ms = j->count * CALL_JITTER_FRAME_MS + j->processing_ms;
    if (tempo_x100) *tempo_x100 = j->tempo_x100;
    if (range) *range = j->range;
    if (dropped) *dropped = j->dropped;
    if (underruns) *underruns = j->underruns;
}
}