Separate stream audio DSP from bounded PCM consumption

src/stream/stream_playback_audio.cpp

@@ -1,6 +1,7 @@
 #include "stream_playback_audio.h"
 #include "soundtouch/SoundTouch.h"
 #include <algorithm>
+#include <atomic>
 #include <cmath>
 #include <cstring>
 #include <new>
@@ -12,7 +13,7 @@ extern "C" {
 }
 
 namespace {
-constexpr unsigned rate = 48000, framesPerPacket = 960, maxFrames = 8192, mapCapacity = 1024;
+constexpr unsigned rate = 48000, framesPerPacket = 960, maxFrames = 8192, mapCapacity = 1024, pcmCapacity = 128;
 struct Span {
     double first, last, ptsFirst, ptsLast;
     Span(double begin = 0, double end = 0, double ptsBegin = 0, double ptsEnd = 0)
@@ -26,6 +27,15 @@ double mapped(const Span& span, double position) {
 }
 
 struct stream_playback_audio {
+    struct Pcm {
+        int16_t samples[framesPerPacket * 2];
+        Span spans[8];
+        unsigned frames = 0, count = 0;
+    } ready[pcmCapacity];
+    std::atomic<uint64_t> readyWrite{0}, readyRead{0}, preparedFrames{0}, consumedFrames{0};
+    std::atomic<double> consumedGain{0};
+    unsigned readyOffset = 0; // Только consumer; producer переиспользует слот после readyRead.store(release).
+    double preparedOutput = 0, preparedGain = 0; // Только producer.
     opus_codec_decoder_t* opus = nullptr;
     soundtouch::SoundTouch stretch;
     Span input[mapCapacity], output[mapCapacity];
@@ -35,7 +45,7 @@ struct stream_playback_audio {
     int64_t startPts = 0, expectedPts = -1;
     double expectedOutput = 0, producedOutput = 0, lastPts = 0;
     uint64_t written = 0;
-    bool finished = false, failed = false;
+    std::atomic<bool> finished{false}, failed{false}, preparedEnd{false};
     int16_t pcm[framesPerPacket * 2];
     float samples[maxFrames * 2];
 };
@@ -69,7 +79,7 @@ static int feedPcm(stream_playback_audio* a, int64_t pts, const uint8_t* packet,
     unsigned skip = pts < a->startPts ? (unsigned)((a->startPts - pts) * rate + 999999) / 1000000 : 0;
     if (skip >= framesPerPacket) return 0;
     unsigned frames = framesPerPacket - skip;
-    while (a->inputCount && a->input[a->inputHead].last <= a->producedOutput) {
+    while (a->inputCount && a->input[a->inputHead].last <= a->preparedOutput) {
         a->inputHead = (a->inputHead + 1) % mapCapacity; a->inputCount--;
     }
     if (a->inputCount == mapCapacity) {
@@ -103,7 +113,7 @@ struct stream_playback_audio* stream_playback_audio_create(int64_t start, unsign
     a->stretch.setSetting(SETTING_USE_QUICKSEEK, 1);
     a->stretch.setSetting(SETTING_SEQUENCE_MS, 20); a->stretch.setSetting(SETTING_SEEKWINDOW_MS, 10);
     a->stretch.setSetting(SETTING_OVERLAP_MS, 4); a->stretch.setSetting(SETTING_USE_AA_FILTER, 0);
-    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "playback audio created start=%lld stereo=1 tempo=1.0..1.1 PCM_buffer=%ums/%u_frames",
+    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "playback audio created start=%lld stereo=1 tempo=1.0..1.6 PCM_buffer=%ums/%u_frames output=SPSC",
                (long long)start, bufferMs, a->bufferFrames);
     return a;
 }
@@ -111,7 +121,7 @@ struct stream_playback_audio* stream_playback_audio_create(int64_t start, unsign
 void stream_playback_audio_destroy(struct stream_playback_audio* a) {
     if (!a) return;
     DEBUG_INFO(DEBUG_CATEGORY_STREAM, "playback audio destroy written=%llu produced=%.0f pts=%.0f failed=%d",
-               (unsigned long long)a->written, a->producedOutput, a->lastPts, a->failed);
+               (unsigned long long)a->written, a->producedOutput, a->lastPts, (int)a->failed.load());
     opus_codec_decoder_destroy(a->opus); a->~stream_playback_audio(); u_free(a);
 }
 
@@ -136,11 +146,11 @@ int stream_playback_audio_feed(struct stream_playback_audio* a, int64_t pts, con
     for (int i = 0; i < missing; i++) if (feedPcm(a, a->expectedPts + i * 20000, nullptr, 0) < 0) return -1;
     if (feedPcm(a, effectivePts, packet, (int)length) < 0) return -1;
     a->expectedPts = effectivePts + 20000;
-    return 0;
+    return stream_playback_audio_prepare(a);
 }
 
 void stream_playback_audio_set_tempo(struct stream_playback_audio* a, int tempo) {
-    if (!a || tempo < 1000 || tempo > 1100) {
+    if (!a || tempo < 1000 || tempo > 1600) {
         DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "playback audio invalid tempo=%d", tempo); return;
     }
     if (tempo == a->tempo) return;
@@ -148,41 +158,96 @@ void stream_playback_audio_set_tempo(struct stream_playback_audio* a, int tempo)
 }
 
 int stream_playback_audio_needs_input(const struct stream_playback_audio* a) {
-    return a && !a->finished && !a->failed && a->stretch.numSamples() < a->bufferFrames;
+    return a && !a->finished && !a->failed && stream_playback_audio_buffered(a) < a->bufferFrames;
 }
 
-/* Ограничение готовой границы применяется до выдачи PCM устройству, сохраняя общий A/V clock. */
+unsigned stream_playback_audio_buffered(const struct stream_playback_audio* a) {
+    if (!a) return 0;
+    uint64_t consumed = a->consumedFrames.load(std::memory_order_acquire);
+    uint64_t prepared = a->preparedFrames.load(std::memory_order_acquire);
+    return (unsigned)(prepared > consumed ? prepared - consumed : 0);
+}
+
+/* DSP принадлежит producer. Каждый слот содержит PCM и его исходные PTS, включая смены темпа. */
+int stream_playback_audio_prepare(struct stream_playback_audio* a) {
+    if (!a || a->failed) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "playback PCM prepare invalid state"); return -1; }
+    uint64_t write = a->readyWrite.load(std::memory_order_relaxed);
+    while (write - a->readyRead.load(std::memory_order_acquire) < pcmCapacity) {
+        double end = a->finished ? std::round(a->expectedOutput) : std::floor(a->expectedOutput);
+        unsigned count = (unsigned)std::max(0.0, std::min<double>(framesPerPacket, end - a->preparedOutput));
+        count = std::min(count, a->stretch.numSamples());
+        if (!count) break;
+        auto& slot = a->ready[write % pcmCapacity]; slot.count = 0;
+        slot.frames = a->stretch.receiveSamples(a->samples, count);
+        for (unsigned i = 0; i < slot.frames * 2; i++)
+            slot.samples[i] = (int16_t)std::lrint(std::max(-32768.0f, std::min(32767.0f, a->samples[i] * 32767.0f)));
+        double position = a->preparedOutput, until = position + slot.frames;
+        for (unsigned i = 0; i < a->inputCount && position < until; i++) {
+            const auto& span = a->input[(a->inputHead + i) % mapCapacity];
+            if (span.last <= position) continue;
+            double next = std::min(until, span.last);
+            // EOF округляется до целого выходного frame, последний PTS остаётся концом материала.
+            if (a->finished && i + 1 == a->inputCount && until <= std::round(a->expectedOutput)) next = until;
+            if (slot.count == 8) {
+                DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "playback PCM source map overflow frames=%u", slot.frames);
+                a->failed = true; return -1;
+            }
+            double firstPts = mapped(span, position), lastPts = mapped(span, next);
+            slot.spans[slot.count++] = {position, next, firstPts, lastPts};
+            a->preparedGain += std::max(0.0, lastPts - firstPts - (next - position) * 1000000.0 / rate);
+            position = next;
+        }
+        if (position != until || !slot.frames) {
+            DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "playback PCM source map missing begin=%.0f end=%.0f mapped=%.0f",
+                        a->preparedOutput, until, position);
+            a->failed = true; return -1;
+        }
+        a->preparedOutput = until;
+        a->preparedFrames.store((uint64_t)until, std::memory_order_release);
+        a->readyWrite.store(++write, std::memory_order_release);
+    }
+    if (a->finished && a->preparedOutput >= std::round(a->expectedOutput)) a->preparedEnd.store(true, std::memory_order_release);
+    return 0;
+}
+
+/* Consumer копирует только готовый PCM. Ни Opus, ни SoundTouch, ни producer-lock здесь нет. */
 int stream_playback_audio_pull(struct stream_playback_audio* a, int16_t* pcm, unsigned frames, int64_t limit) {
     if (!a || !pcm || frames > maxFrames || a->failed || limit < 0) {
         DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "playback audio invalid pull frames=%u limit=%lld", frames, (long long)limit); return -1;
     }
-    double until = a->producedOutput;
-    for (unsigned i = 0; i < a->inputCount; i++) {
-        const auto& span = a->input[(a->inputHead + i) % mapCapacity];
-        if (span.last <= a->producedOutput) continue;
-        if (limit >= span.ptsLast) until = span.last;
-        else if (limit > span.ptsFirst) {
-            until = span.first + (limit - span.ptsFirst) * (span.last - span.first) / (span.ptsLast - span.ptsFirst); break;
-        } else break;
+    unsigned done = 0;
+    uint64_t read = a->readyRead.load(std::memory_order_relaxed);
+    double consumedGain = a->consumedGain.load(std::memory_order_relaxed);
+    while (done < frames && read != a->readyWrite.load(std::memory_order_acquire)) {
+        const auto& slot = a->ready[read % pcmCapacity];
+        double until = a->producedOutput;
+        for (unsigned i = 0; i < slot.count; i++) {
+            const auto& span = slot.spans[i];
+            if (span.last <= a->producedOutput) continue;
+            if (limit >= span.ptsLast) until = span.last;
+            else if (limit > span.ptsFirst) {
+                until = span.first + (limit - span.ptsFirst) * (span.last - span.first) / (span.ptsLast - span.ptsFirst); break;
+            } else break;
+        }
+        unsigned got = std::min(frames - done, (unsigned)std::max(0.0, std::floor(until - a->producedOutput)));
+        if (!got) break;
+        std::memcpy(pcm + done * 2, slot.samples + a->readyOffset * 2, got * 2 * sizeof(int16_t));
+        double position = a->producedOutput, end = position + got;
+        for (unsigned i = 0; i < slot.count && position < end; i++) {
+            const auto& span = slot.spans[i];
+            if (span.last <= position) continue;
+            double next = std::min(end, span.last), firstPts = mapped(span, position), lastPts = mapped(span, next);
+            if (!appendOutput(a, {a->written + position - a->producedOutput,
+                                  a->written + next - a->producedOutput, firstPts, lastPts})) return -1;
+            consumedGain += std::max(0.0, lastPts - firstPts - (next - position) * 1000000.0 / rate);
+            a->lastPts = lastPts; position = next;
+        }
+        a->producedOutput = end; a->written += got; done += got; a->readyOffset += got;
+        if (a->readyOffset == slot.frames) { a->readyOffset = 0; a->readyRead.store(++read, std::memory_order_release); }
     }
-    unsigned available = (unsigned)std::max(0.0, std::floor(until - a->producedOutput));
-    frames = std::min(frames, std::min(available, a->stretch.numSamples()));
-    if (!frames) return 0;
-    unsigned got = a->stretch.receiveSamples(a->samples, frames);
-    for (unsigned i = 0; i < got * 2; i++)
-        pcm[i] = (int16_t)std::lrint(std::max(-32768.0f, std::min(32767.0f, a->samples[i] * 32767.0f)));
-    double position = a->producedOutput, end = position + got;
-    for (unsigned i = 0; i < a->inputCount && position < end; i++) {
-        const auto& span = a->input[(a->inputHead + i) % mapCapacity];
-        if (span.last <= position) continue;
-        double next = std::min(end, span.last);
-        double firstPts = mapped(span, position), lastPts = mapped(span, next);
-        if (!appendOutput(a, {a->written + position - a->producedOutput,
-                              a->written + next - a->producedOutput, firstPts, lastPts})) return -1;
-        a->lastPts = lastPts; position = next;
-    }
-    a->producedOutput = end; a->written += got;
-    return (int)got;
+    a->consumedGain.store(consumedGain, std::memory_order_release);
+    a->consumedFrames.store((uint64_t)a->producedOutput, std::memory_order_release);
+    return (int)done;
 }
 
 void stream_playback_audio_silence(struct stream_playback_audio* a, unsigned frames) {
@@ -203,7 +268,7 @@ int64_t stream_playback_audio_position(struct stream_playback_audio* a, uint64_t
 
 int64_t stream_playback_audio_pending_gain(struct stream_playback_audio* a, uint64_t played) {
     if (!a) return 0;
-    double gain = 0;
+    double gain = std::max(0.0, a->preparedGain - a->consumedGain.load(std::memory_order_acquire));
     for (unsigned i = 0; i < a->outputCount; i++) {
         const auto& span = a->output[(a->outputHead + i) % mapCapacity];
         double begin = std::max(span.first, (double)played);
@@ -212,7 +277,7 @@ int64_t stream_playback_audio_pending_gain(struct stream_playback_audio* a, uint
     }
     for (unsigned i = 0; i < a->inputCount; i++) {
         const auto& span = a->input[(a->inputHead + i) % mapCapacity];
-        double begin = std::max(span.first, a->producedOutput);
+        double begin = std::max(span.first, a->preparedOutput);
         if (span.last <= begin) continue;
         gain += std::max(0.0, span.ptsLast - mapped(span, begin) - (span.last - begin) * 1000000.0 / rate);
     }
@@ -234,10 +299,10 @@ int stream_playback_audio_starved(const struct stream_playback_audio* a, uint64_
 
 void stream_playback_audio_finish(struct stream_playback_audio* a) {
     if (!a || a->finished) return;
-    a->finished = true; a->stretch.flush();
+    a->stretch.flush(); a->finished = true; stream_playback_audio_prepare(a);
 }
 
 int stream_playback_audio_empty(const struct stream_playback_audio* a) {
-    return !a || (a->finished && a->producedOutput + 0.5 >= a->expectedOutput);
+    return !a || (a->preparedEnd.load(std::memory_order_acquire) && !stream_playback_audio_buffered(a));
 }
 }

src/stream/stream_playback_audio.h

@@ -1,5 +1,7 @@
-/* Короткий stereo Opus/SoundTouch конвейер плеера. Кодированный запас остаётся в RS-кольце.
- * Все вызовы одного экземпляра последовательны либо под внешним mutex.
+/* Stereo Opus/SoundTouch и ограниченный SPSC-запас готового PCM.
+ * feed/prepare/set_tempo/finish — один producer; pull/silence — один consumer.
+ * position/pending_gain сериализуют с consumer короткой внешней блокировкой; pending_gain вызывает producer.
+ * Producer не держит эту блокировку при feed/prepare/finish: callback не ждёт Opus/SoundTouch.
  * feed прогревает Opus перед start_pts, затем подаёт PCM с темпом 1.0..1.1 без изменения высоты звука.
  * pull записывает соответствие кадров устройства исходному PTS; silence продлевает hardware time, но не media time.
  * position/pending_gain используют реально проигранный frame counter, поэтому уже поставленный PCM сохраняет свой темп.
@@ -15,11 +17,15 @@ extern "C" {
 #endif
 
 struct stream_playback_audio;
-/* buffer_ms (1..1000) — порог пополнения готового PCM; трансляции используют 40 мс.
+/* buffer_ms (1..1000) — порог пополнения готового PCM, без обязательного ожидания перед стартом.
  * Это запас декодера, без аппаратной задержки и сетевого джиттер-буфера. */
 struct stream_playback_audio *stream_playback_audio_create(int64_t start_pts_us, unsigned buffer_ms);
 void stream_playback_audio_destroy(struct stream_playback_audio *audio);
 int stream_playback_audio_feed(struct stream_playback_audio *audio, int64_t pts_us, const uint8_t *packet, size_t length);
+/* Переносит готовый выход DSP в SPSC; вызывать producer после освобождения PCM, в том числе после finish. */
+int stream_playback_audio_prepare(struct stream_playback_audio *audio);
+/* Потокобезопасный снимок готового PCM, в stereo frames (48000 frames/s). */
+unsigned stream_playback_audio_buffered(const struct stream_playback_audio *audio);
 void stream_playback_audio_set_tempo(struct stream_playback_audio *audio, int tempo_milli);
 int stream_playback_audio_needs_input(const struct stream_playback_audio *audio);
 /* frames <=8192; limit_pts_us ограничивает продвижение исходного материала до готовой границы видео/аудио.

tools/chatgui/tests/test_stream_audio.cpp

@@ -3,6 +3,7 @@
 #include <cstdio>
 #include <cstdlib>
 #include <vector>
+#include <thread>
 
 extern "C" {
 #include "../../../lib/debug_config.h"
@@ -90,10 +91,10 @@ static void test_pcm_buffer() {
     auto* encoder = opus_codec_encoder_create(48000, 2); CHECK(encoder);
     int16_t pcm[1920] = {0}, output[960]; uint8_t packet[1275];
     int length = opus_codec_encode(encoder, pcm, 960, packet, sizeof(packet)); CHECK(length > 0);
-    for (unsigned bufferMs : {20u, 40u}) {
+    for (unsigned bufferMs : {20u, 40u, 250u}) {
         auto* audio = stream_playback_audio_create(0, bufferMs); CHECK(audio);
         unsigned packets = 0;
-        while (stream_playback_audio_needs_input(audio) && packets < 10) {
+        while (stream_playback_audio_needs_input(audio) && packets < 64) {
             CHECK(stream_playback_audio_feed(audio, packets * 20000, packet, (size_t)length) == 0); ++packets;
         }
         CHECK(!stream_playback_audio_needs_input(audio));
@@ -102,7 +103,14 @@ static void test_pcm_buffer() {
             int count = stream_playback_audio_pull(audio, output, 480, INT64_MAX); CHECK(count >= 0);
             played += (unsigned)count;
         }
-        CHECK(bufferMs == 40 ? played == 1920 : played < 1920);
+        CHECK(bufferMs >= 40 ? played == 1920 : played < 1920);
+        if (bufferMs == 250) {
+            for (unsigned callback = 4; callback < 20; callback++) {
+                int count = stream_playback_audio_pull(audio, output, 480, INT64_MAX); CHECK(count == 480);
+                played += count;
+            }
+            CHECK(played == 9600);
+        }
         CHECK(stream_playback_audio_needs_input(audio));
         DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] PCM buffer=%ums worker_gap=40ms played=%u/1920 frames", bufferMs, played);
         stream_playback_audio_destroy(audio);
@@ -110,6 +118,34 @@ static void test_pcm_buffer() {
     opus_codec_encoder_destroy(encoder);
 }
 
+/* Настоящие producer/consumer: callback не обращается к DSP даже при x1.6 и частичных блоках. */
+static void test_pcm_concurrent() {
+    auto* encoder = opus_codec_encoder_create(48000, 2); CHECK(encoder);
+    int16_t pcm[1920] = {}; uint8_t packet[1275];
+    int length = opus_codec_encode(encoder, pcm, 960, packet, sizeof(packet)); CHECK(length > 0);
+    auto* audio = stream_playback_audio_create(0, 250); CHECK(audio);
+    stream_playback_audio_set_tempo(audio, 1600);
+    std::thread producer([&] {
+        for (unsigned i = 0; i < 400; ++i) {
+            while (!stream_playback_audio_needs_input(audio)) std::this_thread::yield();
+            CHECK(stream_playback_audio_feed(audio, i * 20000, packet, length) == 0);
+        }
+        stream_playback_audio_finish(audio);
+    });
+    int16_t output[777 * 2]; uint64_t frames = 0; int64_t previous = 0;
+    while (!stream_playback_audio_empty(audio)) {
+        int count = stream_playback_audio_pull(audio, output, 777, INT64_MAX); CHECK(count >= 0);
+        frames += count;
+        int64_t position = stream_playback_audio_position(audio, frames);
+        CHECK(position >= previous); previous = position;
+        if (!count) std::this_thread::yield();
+    }
+    producer.join();
+    CHECK(frames == 240000 && previous == 8000000);
+    stream_playback_audio_destroy(audio); opus_codec_encoder_destroy(encoder);
+    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] independent PCM producer/consumer: stereo x1.6, 777-frame callbacks, exact EOF/PTS");
+}
+
 /* PTS из Windows→Android лога: звук начинается на 428ms позже KEY, audio clock ещё неподвижен. */
 static void test_late_audio_start() {
     constexpr int64_t videoPts = 20669261, audioPts = 21097059;
@@ -164,6 +200,7 @@ static void test_late_audio_start() {
 int main() {
     debug_config_init(); debug_set_level(DEBUG_LEVEL_ERROR); debug_set_category_level(DEBUG_CATEGORY_STREAM, DEBUG_LEVEL_INFO);
     size_t allocations = u_get_allocated_count();
-    test_stereo_tempo(); test_silence_clock(); test_pcm_buffer(); test_late_audio_start(); CHECK(u_get_allocated_count() == allocations);
+    test_stereo_tempo(); test_silence_clock(); test_pcm_buffer(); test_pcm_concurrent(); test_late_audio_start();
+    CHECK(u_get_allocated_count() == allocations);
     return 0;
 }