jeka/utun2
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Deliver independent track ends and invalidate batches on gaps

Browse Source
This commit is contained in:
evgeny committed 2026-10-20 23:53:23 +02:00
1 parent 23c4e4efbc
commit a54031081b
2 files changed
+31 -9
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/stream/stream_viewer.c
+26 -7
View File
@@ -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;
src/stream/stream_viewer.h
+5 -2
View File
@@ -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);