diff --git a/src/stream/stream_viewer.c b/src/stream/stream_viewer.c index 3c73bf16..b5e779f9 100644 --- a/src/stream/stream_viewer.c +++ b/src/stream/stream_viewer.c @@ -343,7 +343,7 @@ int stream_viewer_request(struct stream_viewers *v, uint64_t id, int64_t positio int changed = s->position != position || s->video != video || s->audio_ahead != audio_ahead || s->records != records; if (!s->requested && !s->record) s->demand_us = get_time_us(); s->position = position; s->video = video; s->audio_ahead = audio_ahead; s->records = records; - s->requested = video || audio_ahead >= 0; + s->requested = 1; /* Нулевой media-бюджет всё равно разрешает управляющие события. */ result = 0; if (s->requested && !s->record && (!s->waiting || changed)) { s->waiting = 0; result = schedule(v); } } diff --git a/tools/chatgui/src/streamplayer_window.cpp b/tools/chatgui/src/streamplayer_window.cpp index 7886580f..407a34f9 100644 --- a/tools/chatgui/src/streamplayer_window.cpp +++ b/tools/chatgui/src/streamplayer_window.cpp @@ -71,7 +71,7 @@ struct StreamPlayerWindow::Playback { int64_t decoderLead = 0, framePeriod = 33333, lastInputPts = -1, lastDecodedPts = -1; uint64_t audioFrameStart = 0, shown = 0, underruns = 0, endFrames = UINT64_MAX; bool started = false, presenting = false, videoReady = false, ended = false, videoEnded = false; - bool activeAudio = false, holdAudio = false, resync = false; + bool activeAudio = false, audioEnded = false, holdAudio = false, resync = false; int64_t audioFedAt = 0, audioFedPts = -1, workerAt = 0, guiAt = 0, guiGapMax = 0; ~Playback() { stream_playback_audio_destroy(sound); } @@ -90,7 +90,7 @@ struct StreamPlayerWindow::Playback { int64_t clock(int64_t now) { if (!started) return 0; if (!presenting) return position; - if (activeAudio && sound && audioAvailable) + if (activeAudio && sound && audioAvailable && !(audioEnded && endFrames != UINT64_MAX && played(now) >= endFrames)) return std::max(position, stream_playback_audio_position(sound, played(now))); if (holdAudio) return position; int64_t proposed = position + (now - clockAt) * adaptation.tempo_milli / 1000; @@ -152,11 +152,11 @@ void StreamPlayerWindow::attachAudio() { p.deviceLead = std::max(0, firstTimeUs - now); if (!p.sound || !p.started || p.resync) return; p.audioFrameStart = stream_playback_audio_written(p.sound); p.audioDacAt = firstTimeUs; - int count = !p.presenting || p.holdAudio ? 0 : stream_playback_audio_pull(p.sound, pcm, frames, p.ended ? INT64_MAX : p.ready); + int count = !p.presenting || p.holdAudio ? 0 : stream_playback_audio_pull(p.sound, pcm, frames, p.audioEnded ? INT64_MAX : p.ready); if (count < 0) { p.resync = true; count = 0; } if ((unsigned)count < frames) { stream_playback_audio_silence(p.sound, frames - (unsigned)count); - if (p.presenting && !p.ended) { + if (p.presenting && !p.audioEnded) { p.underruns++; p.audioGap.count++; p.audioGap.silentFrames += frames - (unsigned)count; p.audioGap.requested = frames; p.audioGap.received = (unsigned)count; p.audioGap.held = p.holdAudio; p.audioGap.reserve = p.ready - p.position; p.audioGap.videoQueue = p.videos.size(); @@ -170,7 +170,7 @@ void StreamPlayerWindow::attachAudio() { output[2 * i] += pcm[2 * i] * (p.outputVolume / 32768.0f); output[2 * i + 1] += pcm[2 * i + 1] * (p.outputVolume / 32768.0f); } - if (p.ended && p.endFrames == UINT64_MAX && stream_playback_audio_empty(p.sound)) + if (p.audioEnded && p.endFrames == UINT64_MAX && stream_playback_audio_empty(p.sound)) p.endFrames = stream_playback_audio_written(p.sound); }); if (!m_audioAttached) { @@ -250,7 +250,7 @@ void StreamPlayerWindow::decode() { SwsContext* converter = nullptr; bool waitKey = true, opened = false, finishing = false, eosSent = false; stream_media_format videoFormat{}; - std::deque records; // Только worker: кодированные записи двух треков, управление — барьер. + std::deque records; // Только worker; порядок внутри каждого трека, GAP/CLOSED отменяют старые данные. uint32_t revision = 0; uint64_t decoded = 0, discarded = 0, reportedShown = 0; int64_t deadline = nowUs() + 10000000, report = nowUs(), decoderProgress = -1, lastInputAt = -1; @@ -272,7 +272,7 @@ void StreamPlayerWindow::decode() { stream_playback_audio_destroy(p.sound); p.sound = nullptr; p.videos.clear(); p.videoBytes = 0; p.started = p.presenting = p.videoReady = false; p.position = p.ready = -1; p.decoderLead = 0; p.framePeriod = 33333; p.lastInputPts = p.lastDecodedPts = -1; - p.ended = p.videoEnded = false; + p.ended = p.videoEnded = p.audioEnded = false; p.audioDacAt = 0; p.audioFrameStart = 0; p.holdAudio = p.resync = false; p.endFrames = UINT64_MAX; p.audioFedAt = 0; p.audioFedPts = -1; } @@ -309,7 +309,7 @@ void StreamPlayerWindow::decode() { auto drainVideo = [&] { if (!decoder) return; for (unsigned attempt = 0; attempt < 8 && p.running; attempt++) { - { std::lock_guard guard(p.lock); if (p.videos.size() >= 2) return; } + { std::lock_guard guard(p.lock); if (p.videoEnded || p.videos.size() >= 2) return; } int64_t codecAt = nowUs(); int result = avcodec_receive_frame(decoder, frame); int64_t codecUs = nowUs() - codecAt; @@ -438,7 +438,7 @@ void StreamPlayerWindow::decode() { } } { std::lock_guard guard(p.lock); - if (finishing && p.videoEnded) break; + if (p.ended && p.videoEnded) break; if (p.videos.empty() && decoderProgress >= 0 && nowUs() - decoderProgress > 10000000 && (packet->size || finishing || (lastInputAt >= 0 && nowUs() - lastInputAt < 1000000))) { DEBUG_ERROR(DEBUG_CATEGORY_VIDEO, "desktop decoder stalled id=%llu input_PTS=%lld output_PTS=%lld pending=%d FIN=%d", @@ -449,13 +449,14 @@ void StreamPlayerWindow::decode() { { std::lock_guard guard(p.lock); int64_t now = nowUs(); position = p.started ? p.clock(now) : -1; - wantVideo = !packet->size && p.videos.size() < 2; - videoBudget = static_cast(std::max(0, 2 - (int)p.videos.size() - (packet->size ? 1 : 0))); - audioAhead = !p.activeAudio || !p.sound || stream_playback_audio_needs_input(p.sound) ? p.deviceLead + 80000 : -1; + wantVideo = !finishing && !packet->size && p.videos.size() < 2; + videoBudget = finishing ? 0 : static_cast(std::max(0, 2 - (int)p.videos.size() - (packet->size ? 1 : 0))); + audioAhead = !p.audioEnded && (!p.activeAudio || !p.sound || stream_playback_audio_needs_input(p.sound)) + ? p.deviceLead + 80000 : -1; if (p.presenting && p.ready >= 0) { uint64_t played = p.played(now); - p.holdAudio = !p.videoReady || (p.videos.empty() && p.lastDecodedPts < position - p.framePeriod) - || (!p.videos.empty() && p.videos.front().pts < position - 40000); + p.holdAudio = !p.videoEnded && (!p.videoReady || (p.videos.empty() && p.lastDecodedPts < position - p.framePeriod) + || (!p.videos.empty() && p.videos.front().pts < position - 40000)); int virtualEmpty = !p.activeAudio && p.position + (now - p.clockAt) * p.adaptation.tempo_milli / 1000 > p.ready; if (!p.activeAudio && p.holdAudio) position = p.position; int starved = p.holdAudio || virtualEmpty || (p.sound && stream_playback_audio_starved(p.sound, played)); @@ -469,19 +470,28 @@ void StreamPlayerWindow::decode() { } } } - // Каждый трек расходуется по своей готовности. GAP/FIN/CLOSED не обгоняют предыдущие media. + // Конец трека обгоняет только чужой трек; аварийные события отменяют всю прежнюю очередь. + auto urgent = std::find_if(records.begin(), records.end(), [](const StreamRecord& r) { + return r.kind == StreamRecordKind::Gap || r.kind == StreamRecordKind::Closed; + }); + if (urgent != records.end()) records.erase(records.begin(), urgent); auto next = records.end(); unsigned pendingVideo = 0, pendingAudio = 0; bool barrier = false; for (auto it = records.begin(); it != records.end(); ++it) { bool video = it->kind == StreamRecordKind::Video, audio = it->kind == StreamRecordKind::Audio; pendingVideo += video; pendingAudio += audio; + if (it->kind == StreamRecordKind::VideoEnd || it->kind == StreamRecordKind::AudioEnd) { + if (next == records.end() && (it->kind == StreamRecordKind::VideoEnd ? !pendingVideo : !pendingAudio)) next = it; + continue; + } if (!video && !audio) { barrier = true; if (it == records.begin()) next = it; break; } if (next == records.end() && ((video && wantVideo) || (audio && audioAhead >= 0))) next = it; } if (next == records.end()) { size_t previous = records.size(); int64_t pullAt = nowUs(); - bool received = !finishing && !barrier && gui_stream_pull_batch(p.id, records, position, + // После обоих EOF нужен ещё общий END, даже когда больше media не требуется. + bool received = !p.ended && !barrier && gui_stream_pull_batch(p.id, records, position, videoBudget > pendingVideo ? videoBudget - pendingVideo : 0, pendingAudio ? -1 : audioAhead); int64_t pullUs = nowUs() - pullAt; work.pull += pullUs; work.pullMax = std::max(work.pullMax, pullUs); @@ -507,10 +517,17 @@ void StreamPlayerWindow::decode() { : record.reason == STREAM_CLOSE_GROUP ? "The channel is closed" : "Could not continue the stream"; fail(text(reason), record.reason == STREAM_CLOSE_FAILED); break; } - if (record.kind == StreamRecordKind::End) { - std::lock_guard guard(p.lock); p.ended = true; p.holdAudio = false; - if (p.sound) stream_playback_audio_finish(p.sound); - finishing = true; p.videoEnded = !decoder; continue; + if (record.kind == StreamRecordKind::VideoEnd || record.kind == StreamRecordKind::AudioEnd || record.kind == StreamRecordKind::End) { + std::lock_guard guard(p.lock); + if (record.kind != StreamRecordKind::AudioEnd) { finishing = true; p.videoEnded = !decoder; } + if (record.kind != StreamRecordKind::VideoEnd && !p.audioEnded) { + p.audioEnded = true; + if (p.sound) stream_playback_audio_finish(p.sound); + } + if (record.kind == StreamRecordKind::End) p.ended = true; + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "desktop track end id=%llu kind=%d video_input_end=%d audio_input_end=%d pending=%zu", + (unsigned long long)p.id, (int)record.kind, finishing, p.audioEnded, records.size()); + continue; } observe(); if (record.pts > INT64_MAX) { fail(text("Invalid stream timestamp")); break; }