From f38aa3b08dcb637b7b15bd9a8b83d0998198f3b2 Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 20 Oct 2026 21:27:51 +0200 Subject: [PATCH] Log stream startup key age and buffer handoff delays --- src/stream/stream_media_buffer.c | 30 +++++++++++++++++------- src/stream/stream_publisher.c | 13 +++++++++-- src/stream/stream_viewer.c | 40 ++++++++++++++++++++++++++++---- 3 files changed, 69 insertions(+), 14 deletions(-) diff --git a/src/stream/stream_media_buffer.c b/src/stream/stream_media_buffer.c index 0349d068..3ad20582 100644 --- a/src/stream/stream_media_buffer.c +++ b/src/stream/stream_media_buffer.c @@ -24,7 +24,7 @@ struct stream_media_store { uint64_t pending_seq, key_seq, last_frame_id; uint32_t pending_used; int64_t video_us, audio_us, key_us, last_audio_us; - uint64_t pressure_report_us; + uint64_t pressure_report_us, pending_started_us, key_completed_us; }; struct cursor_track { uint64_t seq; unsigned entry; int ended; }; struct stream_media_cursor { @@ -36,6 +36,7 @@ struct stream_media_cursor { int64_t start_us; int audio_started; int64_t audio_start_us; + uint64_t created_us; uint8_t *scratch[2], *frame; size_t scratch_capacity, frame_capacity; uint64_t scratch_seq[2]; @@ -198,7 +199,7 @@ int stream_media_store_feed(struct stream_media_store *s, uint64_t seq, const ui if (media_discontinuity(s) != STREAM_OK) return STREAM_ERR_FORMAT; } if (s->waiting_key && !(v->flags & STREAM_MEDIA_KEY)) continue; - s->pending = *v; s->pending.data = NULL; s->pending_seq = seq; + s->pending = *v; s->pending.data = NULL; s->pending_seq = seq; s->pending_started_us = get_time_us(); } if (s->waiting_key && !(v->flags & STREAM_MEDIA_KEY)) continue; if (v->frame_id != s->pending.frame_id || v->offset != s->pending_used || v->pts_us != s->pending.pts_us @@ -211,7 +212,14 @@ int stream_media_store_feed(struct stream_media_store *s, uint64_t seq, const ui s->pending_used += v->length; if (s->pending_used != s->pending.frame_size) continue; s->video_us = (int64_t)v->pts_us; s->last_frame_id = v->frame_id; s->waiting_key = 0; s->pending_used = 0; - if (v->flags & STREAM_MEDIA_KEY) { s->key_seq = s->pending_seq; s->key_us = (int64_t)v->pts_us; } + if (v->flags & STREAM_MEDIA_KEY) { + s->key_seq = s->pending_seq; s->key_us = (int64_t)v->pts_us; s->key_completed_us = get_time_us(); + DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, + "media KEY indexed source=%llx epoch=%llx pts=%lld seq=%llu..%llu bytes=%u assemble_us=%llu", + (unsigned long long)s->descriptor.source_node_id, (unsigned long long)s->descriptor.source_epoch, + (long long)s->key_us, (unsigned long long)s->key_seq, (unsigned long long)seq, v->frame_size, + (unsigned long long)(s->key_completed_us - s->pending_started_us)); + } } for (unsigned i = 0; i < b.audio_count; i++) { const struct stream_media_audio *a = &b.audio[i]; @@ -238,7 +246,7 @@ int stream_media_cursor_create(struct stream_media_cursor **out, struct stream_m } struct stream_media_cursor *c = u_calloc(1, sizeof(*c)); if (!c) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "media cursor allocation failed"); return STREAM_ERR_MEMORY; } - c->store = s; c->tracks = tracks; c->scratch_capacity = stream_input_size(&s->descriptor); + c->store = s; c->tracks = tracks; c->scratch_capacity = stream_input_size(&s->descriptor); c->created_us = get_time_us(); c->audio.ended = !(tracks & STREAM_MEDIA_TRACK_AUDIO); if (c->scratch_capacity > STREAM_MEDIA_MAX_BLOCK) c->scratch_capacity = STREAM_MEDIA_MAX_BLOCK; for (unsigned i = 0; i < 2; i++) c->scratch[i] = u_malloc((uint32_t)c->scratch_capacity); @@ -264,7 +272,7 @@ void stream_media_cursor_reset(struct stream_media_cursor *c) { c->pending = (struct stream_media_video){0}; c->pending_format = (struct stream_media_format){0}; c->scratch_valid[0] = c->scratch_valid[1] = 0; c->video = (struct cursor_track){0}; c->audio = (struct cursor_track){0}; - c->audio.ended = !(c->tracks & STREAM_MEDIA_TRACK_AUDIO); + c->audio.ended = !(c->tracks & STREAM_MEDIA_TRACK_AUDIO); c->created_us = get_time_us(); } /* Выбирается последний полный KEY; тёплая история аудио перед ним остаётся в том же RS-кольце. */ @@ -283,9 +291,15 @@ static int cursor_start(struct stream_media_cursor *c) { } } c->initialized = 1; - DEBUG_INFO(DEBUG_CATEGORY_STREAM, "media cursor start source=%llx key_seq=%llu pts=%lld audio_seq=%llu generation=%u", - (unsigned long long)s->descriptor.source_node_id, (unsigned long long)c->video.seq, - (long long)c->start_us, (unsigned long long)c->audio.seq, c->generation); + uint64_t now = get_time_us(); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, + "media cursor start source=%llx epoch=%llx key_seq=%llu pts=%lld audio_seq=%llu generation=%u " + "wait_key_us=%llu key_cached_us=%llu video_ready_us=%lld audio_ready_us=%lld key_backlog_us=%lld", + (unsigned long long)s->descriptor.source_node_id, (unsigned long long)s->descriptor.source_epoch, + (unsigned long long)c->video.seq, (long long)c->start_us, + (unsigned long long)c->audio.seq, c->generation, (unsigned long long)(now - c->created_us), + (unsigned long long)(s->key_completed_us ? now - s->key_completed_us : 0), + (long long)s->video_us, (long long)s->audio_us, (long long)(s->video_us - c->start_us)); return STREAM_OK; } diff --git a/src/stream/stream_publisher.c b/src/stream/stream_publisher.c index 8e4effb7..d0e461a9 100644 --- a/src/stream/stream_publisher.c +++ b/src/stream/stream_publisher.c @@ -108,11 +108,16 @@ static void publisher_stats(void *arg) { if (now - p->report_us >= 1000000) { p->report_us = now; DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, - "publisher TX id=%llu video=%llu audio=%llu RS=%llu ETCP=%llu inflight=%llu total=%llu ACK_1s=%llu conns=%u KEY=%d", + "publisher TX id=%llu video=%llu audio=%llu RS=%llu ETCP=%llu inflight=%llu total=%llu ACK_1s=%llu conns=%u KEY=%d " + "source=%llx epoch=%llx video_age_us=%llu audio_age_us=%llu video_pts=%llu audio_pts=%llu", (unsigned long long)p->status.id, (unsigned long long)p->status.video_bytes, (unsigned long long)p->status.audio_bytes, (unsigned long long)p->status.stream_bytes, (unsigned long long)p->status.transport_bytes, (unsigned long long)p->status.inflight_bytes, (unsigned long long)p->status.total_bytes, - (unsigned long long)p->status.acked_1s, p->status.connections, p->status.need_key); + (unsigned long long)p->status.acked_1s, p->status.connections, p->status.need_key, + (unsigned long long)info.descriptor.source_node_id, (unsigned long long)p->status.epoch, + (unsigned long long)(p->video.head ? now - p->video.head->created : 0), + (unsigned long long)(p->audio.head ? now - p->audio.head->created : 0), + (unsigned long long)p->last_pts[0], (unsigned long long)p->last_pts[1]); } } p->stats_timer = uasync_set_timeout(p->inst->ua, 1000, p, publisher_stats, "publisher_stats"); @@ -334,6 +339,10 @@ int stream_publisher_feed(struct stream_publisher *p, uint64_t id, int video, ui r->length = length; r->pts = pts; r->created = get_time_us(); r->key = key ? STREAM_MEDIA_KEY : 0; if (video) { r->frame_id = ++p->next_frame; p->status.need_key = 0; } memcpy(r->data, data, length); record_append(video ? &p->video : &p->audio, r); + if (video && key) DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, + "publisher KEY accepted id=%llu source=%llx epoch=%llx pts=%llu frame=%llu bytes=%zu queued_video=%u queued_audio=%u", + (unsigned long long)id, (unsigned long long)p->inst->node_id, (unsigned long long)p->status.epoch, + (unsigned long long)pts, (unsigned long long)r->frame_id, length, p->video.count, p->audio.count); p->seen[track] = 1; p->last_pts[track] = pts; result = 1; done: pthread_mutex_unlock(&p->lock); return result; diff --git a/src/stream/stream_viewer.c b/src/stream/stream_viewer.c index 76d58bdd..9eb20f62 100644 --- a/src/stream/stream_viewer.c +++ b/src/stream/stream_viewer.c @@ -18,6 +18,8 @@ struct viewer { int video; int64_t position, audio_ahead; uint64_t replies, fulfilled_repeats; /* Выданные ответы и погашенные ими повторные запросы. */ + uint64_t created_us, prepared_us, report_us; + unsigned first_records; /* Диагностика первой выдачи VIDEO/AUDIO; не влияет на cursor. */ }; struct stream_viewers { pthread_mutex_t lock; @@ -58,7 +60,7 @@ static int output(struct viewer *s, enum stream_viewer_kind kind, int reason, co r->flags = kind == STREAM_VIEWER_VIDEO ? p->video.flags : 0; memcpy(r->data, kind == STREAM_VIEWER_VIDEO ? p->video.data : p->audio.data, length); } - u_free(s->record); s->record = r; + u_free(s->record); s->record = r; s->prepared_us = get_time_us(); return 0; } @@ -111,9 +113,9 @@ static void process(void *arg) { continue; } if (output(s, STREAM_VIEWER_OPENED, 0, NULL) < 0) { close_viewer(s, STREAM_VIEWER_CLOSED, STREAM_CLOSE_FAILED); continue; } - DEBUG_INFO(DEBUG_CATEGORY_STREAM, "viewer opened id=%llu group=%llx source=%llx epoch=%llx tracks=%u", + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "viewer opened id=%llu group=%llx source=%llx epoch=%llx tracks=%u open_us=%llu", (unsigned long long)s->id, (unsigned long long)s->group, (unsigned long long)s->source, - (unsigned long long)s->epoch, s->tracks); + (unsigned long long)s->epoch, s->tracks, (unsigned long long)(get_time_us() - s->created_us)); } if (!s->handle) continue; if (s->resync) { @@ -124,6 +126,21 @@ static void process(void *arg) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "viewer index failed id=%llu", (unsigned long long)s->id); close_viewer(s, STREAM_VIEWER_CLOSED, STREAM_CLOSE_FAILED); continue; } + uint64_t now = get_time_us(); + if (now - s->report_us >= 1000000) { + s->report_us = now; + DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, + "viewer buffer id=%llu source=%llx epoch=%llx elapsed_us=%llu initialized=%d generation=%llu " + "start_us=%lld position_us=%lld ready_us=%lld video_us=%lld audio_us=%lld reserve_us=%lld " + "requested=%d waiting=%d slot=%d slot_age_us=%llu", + (unsigned long long)s->id, (unsigned long long)s->source, (unsigned long long)s->epoch, + (unsigned long long)(now - s->created_us), s->status.initialized, (unsigned long long)s->status.generation, + (long long)s->status.start_us, (long long)s->position, (long long)s->status.ready_us, + (long long)s->status.video_us, (long long)s->status.audio_us, + (long long)(s->status.initialized && s->position >= 0 ? s->status.ready_us - s->position : -1), + s->requested, s->waiting, s->record ? s->record->kind : 0, + (unsigned long long)(s->record ? now - s->prepared_us : 0)); + } if (!s->requested || s->waiting || s->record) continue; s->requested = 0; struct stream_media_packet p; @@ -186,8 +203,11 @@ int stream_viewer_open(struct stream_viewers *v, uint64_t id, uint64_t group, ui else if (schedule(v) < 0) u_free(s); else { *s = (struct viewer){.next = v->head, .id = id, .group = group, .source = source, .epoch = epoch, - .tracks = tracks, .opening = 1, .position = -1, .audio_ahead = STREAM_MEDIA_READ_AHEAD_US}; + .tracks = tracks, .opening = 1, .position = -1, .audio_ahead = STREAM_MEDIA_READ_AHEAD_US, + .created_us = get_time_us()}; v->head = s; result = 0; + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "viewer requested id=%llu group=%llx source=%llx epoch=%llx tracks=%u", + (unsigned long long)id, (unsigned long long)group, (unsigned long long)source, (unsigned long long)epoch, tracks); } } pthread_mutex_unlock(&v->lock); return result; @@ -210,6 +230,18 @@ struct stream_viewer_record *stream_viewer_pull(struct stream_viewers *v, uint64 struct viewer *s = find(v, id); struct stream_viewer_record *r = s && !s->stopping ? s->record : NULL; if (r) { + unsigned bit = r->kind == STREAM_VIEWER_VIDEO ? 1 : r->kind == STREAM_VIEWER_AUDIO ? 2 : 0; + if (bit && !(s->first_records & bit)) { + uint64_t now = get_time_us(); s->first_records |= bit; + DEBUG_INFO(DEBUG_CATEGORY_STREAM, + "viewer first record id=%llu source=%llx epoch=%llx kind=%d key=%d pts=%llu bytes=%zu " + "elapsed_us=%llu handoff_us=%llu start_us=%lld ready_us=%lld ready_ahead_us=%lld", + (unsigned long long)s->id, (unsigned long long)s->source, (unsigned long long)s->epoch, + r->kind, !!(r->flags & STREAM_MEDIA_KEY), (unsigned long long)r->pts, r->length, + (unsigned long long)(now - s->created_us), (unsigned long long)(now - s->prepared_us), + (long long)s->status.start_us, (long long)s->status.ready_us, + (long long)(s->status.initialized ? s->status.ready_us - (int64_t)r->pts : -1)); + } /* Ответ обслуживает и повторный request: AVAILABLE не должен прочитать следующий кадр до нового demand. */ int repeated = s->requested; s->record = NULL; s->requested = 0; s->replies++;