From a54031081bd297cb7b201afd085d6f329fecaf0f Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 20 Oct 2026 23:53:23 +0200 Subject: [PATCH] Deliver independent track ends and invalidate batches on gaps --- src/stream/stream_viewer.c | 33 ++++++++++++++++++++++++++------- src/stream/stream_viewer.h | 7 +++++-- 2 files changed, 31 insertions(+), 9 deletions(-) diff --git a/src/stream/stream_viewer.c b/src/stream/stream_viewer.c index f65c4061..3c73bf16 100644 --- a/src/stream/stream_viewer.c +++ b/src/stream/stream_viewer.c @@ -17,7 +17,7 @@ struct viewer { void *notify_arg; struct stream_media_status status; int opening, stopping, resync, requested, waiting; - unsigned video, records; + unsigned video, records, ended_tracks; int64_t position, audio_ahead; uint64_t replies, fulfilled_repeats; /* Выданные ответы и погашенные ими повторные запросы. */ uint64_t created_us, prepared_us, report_us; @@ -133,8 +133,9 @@ static void process(void *arg) { (unsigned long long)s->epoch, s->tracks, (unsigned long long)(get_time_us() - s->created_us)); } if (!s->handle) continue; + if (s->tail && s->tail->kind == STREAM_VIEWER_GAP) continue; if (s->resync) { - s->resync = s->waiting = 0; s->position = -1; + s->resync = s->waiting = 0; s->position = -1; s->ended_tracks = 0; if (stream_media_resync(s->handle) < 0) { close_viewer(s, STREAM_VIEWER_CLOSED, STREAM_CLOSE_FAILED); continue; } } if (stream_media_get_status(s->handle, &s->status) < 0) { @@ -168,14 +169,29 @@ static void process(void *arg) { size_t bytes = 0; uint64_t began = get_time_us(); do { + unsigned ended = s->status.finished ? (s->status.video_ended ? STREAM_MEDIA_TRACK_VIDEO : 0) + | (s->status.audio_ended ? STREAM_MEDIA_TRACK_AUDIO : 0) : 0; + unsigned notify = ended & ~s->ended_tracks; + if (notify) { + unsigned track = notify & STREAM_MEDIA_TRACK_VIDEO ? STREAM_MEDIA_TRACK_VIDEO : STREAM_MEDIA_TRACK_AUDIO; + if (output(s, track == STREAM_MEDIA_TRACK_VIDEO ? STREAM_VIEWER_VIDEO_END : STREAM_VIEWER_AUDIO_END, 0, NULL) < 0) { + close_viewer(s, STREAM_VIEWER_CLOSED, STREAM_CLOSE_FAILED); break; + } + s->ended_tracks |= track; count++; + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "viewer track end id=%llu track=%s batch_records=%u", + (unsigned long long)s->id, track == STREAM_MEDIA_TRACK_VIDEO ? "video" : "audio", count); + continue; + } struct stream_media_packet p; int result = stream_media_read(s->handle, s->position, video > 0, s->audio_ahead, &p); + if (stream_media_get_status(s->handle, &s->status) < 0) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "viewer read status failed id=%llu", (unsigned long long)s->id); + close_viewer(s, STREAM_VIEWER_CLOSED, STREAM_CLOSE_FAILED); break; + } if (result == STREAM_OK) { if (p.kind == STREAM_MEDIA_READ_END) { - if (stream_media_get_status(s->handle, &s->status) < 0) { - DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "viewer final status failed id=%llu", (unsigned long long)s->id); - close_viewer(s, STREAM_VIEWER_CLOSED, STREAM_CLOSE_FAILED); - } else close_viewer(s, STREAM_VIEWER_END, 0); + if (s->status.initialized && s->ended_tracks != (STREAM_MEDIA_TRACK_VIDEO | STREAM_MEDIA_TRACK_AUDIO)) continue; + close_viewer(s, STREAM_VIEWER_END, 0); break; } int kind = p.kind == STREAM_MEDIA_READ_VIDEO ? STREAM_VIEWER_VIDEO : STREAM_VIEWER_AUDIO; @@ -183,6 +199,8 @@ static void process(void *arg) { count++; bytes += s->tail->length; if (kind == STREAM_VIEWER_VIDEO) video--; } else if (result == STREAM_MORE || result == STREAM_WAIT) { + if (s->status.finished && ((s->status.video_ended && !(s->ended_tracks & STREAM_MEDIA_TRACK_VIDEO)) + || (s->status.audio_ended && !(s->ended_tracks & STREAM_MEDIA_TRACK_AUDIO)))) continue; if (!s->record) { s->requested = 1; s->waiting = result == STREAM_WAIT; if (!s->waiting && schedule(v) < 0) close_viewer(s, STREAM_VIEWER_CLOSED, STREAM_CLOSE_FAILED); @@ -190,7 +208,8 @@ static void process(void *arg) { break; } else if (result == STREAM_GAP || result == STREAM_EXPIRED) { DEBUG_INFO(DEBUG_CATEGORY_STREAM, "viewer resync id=%llu result=%d", (unsigned long long)s->id, result); - s->position = -1; memset(&s->status, 0, sizeof(s->status)); + stream_viewer_records_free(s->record); s->record = s->tail = NULL; + s->position = -1; s->ended_tracks = 0; memset(&s->status, 0, sizeof(s->status)); if (stream_media_resync(s->handle) < 0 || output(s, STREAM_VIEWER_GAP, 0, NULL) < 0) close_viewer(s, STREAM_VIEWER_CLOSED, STREAM_CLOSE_FAILED); break; diff --git a/src/stream/stream_viewer.h b/src/stream/stream_viewer.h index bcc281b2..c06f040c 100644 --- a/src/stream/stream_viewer.h +++ b/src/stream/stream_viewer.h @@ -14,7 +14,8 @@ struct stream_viewers; #define STREAM_VIEWER_BATCH_RECORDS 16u enum stream_viewer_kind { STREAM_VIEWER_VIDEO = 1, STREAM_VIEWER_AUDIO, STREAM_VIEWER_GAP, - STREAM_VIEWER_OPENED, STREAM_VIEWER_END, STREAM_VIEWER_CLOSED + STREAM_VIEWER_OPENED, STREAM_VIEWER_END, STREAM_VIEWER_CLOSED, + STREAM_VIEWER_VIDEO_END, STREAM_VIEWER_AUDIO_END }; struct stream_viewer_record { struct stream_viewer_record *next; /* Следующая запись батча; NULL завершает цепочку. */ @@ -33,7 +34,9 @@ void stream_viewers_destroy(struct stream_viewers *viewers); int stream_viewer_open(struct stream_viewers *viewers, uint64_t id, uint64_t group, uint64_t source, uint64_t epoch, unsigned tracks); void stream_viewer_stop(struct stream_viewers *viewers, uint64_t id); /* Owned батч или NULL. Выдача батча погашает demand, включая повторный request до pull. - * NULL сохраняет ожидающий запрос. Следующие данные требуют нового request; idle не копирует кольцо. */ + * NULL сохраняет ожидающий запрос. Следующие данные требуют нового request; idle не копирует кольцо. + * VIDEO_END/AUDIO_END следуют за последней записью своего трека; другой трек их не задерживает. + * END завершает оба входа. GAP/CLOSED отменяют старый батч и требуют немедленного сброса frontend. */ struct stream_viewer_record *stream_viewer_pull(struct stream_viewers *viewers, uint64_t id); /* Освобождает всю owned цепочку и payload каждой записи; NULL допустим. */ void stream_viewer_records_free(struct stream_viewer_record *records);