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

Log stream startup key age and buffer handoff delays

Browse Source
This commit is contained in:
evgeny committed 2026-10-20 21:27:51 +02:00
1 parent c92d36354a
commit f38aa3b08d
3 files changed
+69 -14
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/stream/stream_media_buffer.c
+22 -8
View File
@@ -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;
}
src/stream/stream_publisher.c
+11 -2
View File
@@ -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;
src/stream/stream_viewer.c
+36 -4
View File
@@ -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++;