diff --git a/src/stream/stream_playback_audio.cpp b/src/stream/stream_playback_audio.cpp index 5be185b9..0b218c00 100644 --- a/src/stream/stream_playback_audio.cpp +++ b/src/stream/stream_playback_audio.cpp @@ -1,6 +1,7 @@ #include "stream_playback_audio.h" #include "soundtouch/SoundTouch.h" #include +#include #include #include #include @@ -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 readyWrite{0}, readyRead{0}, preparedFrames{0}, consumedFrames{0}; + std::atomic 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 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(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)); } } diff --git a/src/stream/stream_playback_audio.h b/src/stream/stream_playback_audio.h index 544b64bb..1fceec2b 100644 --- a/src/stream/stream_playback_audio.h +++ b/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 ограничивает продвижение исходного материала до готовой границы видео/аудио. diff --git a/tools/chatgui/tests/test_stream_audio.cpp b/tools/chatgui/tests/test_stream_audio.cpp index 53a9d25b..a2d75d15 100644 --- a/tools/chatgui/tests/test_stream_audio.cpp +++ b/tools/chatgui/tests/test_stream_audio.cpp @@ -3,6 +3,7 @@ #include #include #include +#include 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; }