From 0e17756e19cab9b747001b0967618fef195ff730 Mon Sep 17 00:00:00 2001 From: evgeny Date: Sun, 25 Oct 2026 15:21:23 +0100 Subject: [PATCH] Show stream playback buffer chains on desktop and Android Send coalesced per-hop telemetry every 200 ms ahead of video DATA using the shared ETCP transport. Measure pending frames before ETCP handoff, retain stale and unavailable node values, and display the worst path with playback tempo. Bump stream wire protocol to v7. Validate with 187 core tests, 21 desktop player tests, two Android indicator tests, and an Android APK build. --- doc/stream.txt | 37 ++++- src/stream/stream.c | 44 ++++- src/stream/stream_buffer.c | 85 ++++++++++ src/stream/stream_buffer.h | 15 ++ src/stream/stream_network.c | 150 +++++++++++++++++- src/stream/stream_network.h | 2 + src/stream/stream_priv.h | 3 + src/stream/stream_publisher.c | 16 ++ src/stream/stream_viewer.c | 55 +++++++ src/stream/stream_viewer.h | 7 + src/stream/stream_wire.c | 84 +++++++++- src/stream/stream_wire.h | 34 +++- tests/test_android_stream.c | 4 + tests/test_stream_core.c | 53 +++++++ tests/test_stream_service.c | 22 ++- .../main/java/com/utun/chat/data/NativeLib.kt | 2 + .../utun/chat/data/StreamPlaybackTelemetry.kt | 35 ++++ .../com/utun/chat/data/StreamPlayerEngine.kt | 19 ++- .../utun/chat/ui/screens/StreamGridScreen.kt | 6 +- .../ui/screens/StreamPlaybackIndicator.kt | 33 ++++ .../chat/ui/screens/StreamViewerScreen.kt | 11 +- .../chat/data/StreamPlaybackTelemetryTest.kt | 22 +++ .../jni_bridge/android_stream_jni.c | 28 ++++ .../libutun_lite/stream_bridge.c | 7 + .../libutun_lite/stream_bridge.h | 2 + tools/chatgui/src/streamplayer_window.cpp | 61 +++++++ tools/chatgui/src/streamplayer_window.h | 2 + tools/chatgui/tests/test_stream_view.cpp | 4 + tools/chatgui/transport/stream_bridge.cpp | 5 + tools/chatgui/transport/stream_bridge.h | 2 + 30 files changed, 831 insertions(+), 19 deletions(-) create mode 100644 tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamPlaybackTelemetry.kt create mode 100644 tools/chatgui-android/app/src/main/java/com/utun/chat/ui/screens/StreamPlaybackIndicator.kt create mode 100644 tools/chatgui-android/app/src/test/java/com/utun/chat/data/StreamPlaybackTelemetryTest.kt diff --git a/doc/stream.txt b/doc/stream.txt index 40a43a24..7358eea8 100644 --- a/doc/stream.txt +++ b/doc/stream.txt @@ -671,7 +671,7 @@ Stop/destroy выполняются по правилам ядра, вне callb Все stream-числа на wire — big-endian, без packed-структур и зависимости от host byte order. Версия stream wire — 4; обратная совместимость не поддерживается. -Заголовок raw (52 байта, version=6): cmd:1, version:1, command:1, subblock:1, +Заголовок raw (52 байта, version=7): cmd:1, version:1, command:1, subblock:1, group_id:8, source_node_id:8, source_epoch:8, sender_epoch:8, receiver_epoch:8, handle:8. INFO_REQ/RESP согласуют sender/receiver epochs; receiver_epoch=0 используется INFO_REQ для обнаружения поколения и REVERSE_REQ. Подписанный routed payload содержит тот же заголовок; его group_id сверяется с реальной группой router, @@ -694,7 +694,7 @@ CPU — 0..10000 (сотые доли процента). SUBSCRIBE: requested_seq:8; REJECT: reason:1, для SUBSCRIBE дополнительно first_available:8,last_available:8, для INFO дополнительно probe_nonce:8. Старый nonce не подтверждает новую пробу. RENEW: один header handle без payload (каждая кодограмма продлевает один элемент списка). -UNSUBSCRIBE/CLOSED: reason:1. DATA: path_count:1,node_ids:8*path_count,неизменяемая часть RS-пачки переменной длины. +UNSUBSCRIBE/CLOSED: reason:1. DATA: path_count:1,node_ids:8*path_count,frame_timing:48,неизменяемая часть RS-пачки переменной длины. GAP: old_cursor:8,next_cursor:8, уведомление включает диапазон [old_cursor,next_cursor). END: "stream_end":10,group_id:8,source_node_id:8,source_epoch:8,last_seq:8,Ed25519:64 — всего 106 байт. UINT64_MAX в END означает пустой источник; принимается только выдачей с next=0 и без ранее принятых DATA. @@ -1375,6 +1375,39 @@ Desktop/Android не закрывают соединение по отдельн - stream recovery resumed — продвижение после попытки без сброса медиа; - stream recovery KEY — пройденная граница, выбранный KEY и причина недоступности прежнего диапазона; - stream GAP — точные first/next и направление; stream NET/PLAY — сводки каждые >=500мс. +- stream BUFFER — худшая цепочка, сумма её буферов, число устаревших/недоступных/неизвестных узлов, максимум возраста раз в секунду. + +Буферы цепочки воспроизведения (raw protocol v7): + +TELEMETRY=18 идёт через общий raw ETCP отдельной служебной кодограммой. В очередь видео DATA он не ставится: +перед очередной DATA pump сначала отправляет просроченную/обновлённую телеметрию. Снимок строится в момент отправки, +не копится в control FIFO; при backpressure сохраняется только dirty-флаг. Уже принятые ETCP данные он не обгоняет. +Источник и relay обновляют собственные измерения каждые 200ms; получение нового upstream сразу будит downstream. +Если upstream не обновлялся 400ms, его значения сохраняются с STALE. Подтверждённый разрыв помечает последний переход DOWN, +предыдущую часть цепочки — STALE. Отсутствовавшее измерение отличается от известного нулевого буфера. + +Payload TELEMETRY: count:1, count записей по 33 байта: +node_id:8, buffered_us:8, sample:8, frames:4, age_ms:4, flags:1. Все числа big-endian. +Флаги KNOWN=1, STALE=2, DOWN=4. Путь начинается источником, заканчивается отправителем и не содержит повторов/получателя. +sample — монотонный номер измерения узла; возраст увеличивается при каждом хранении/пересылке, без синхронизации часов. +Потеря выдачи не удаляет последний снимок из общего кольца; новая эпоха получает отдельное кольцо. + +Подсказка frame_timing в DATA содержит first_frame:8,last_frame:8,first_pts:8,last_pts:8,period_us:4, +started:4,audio_us:4,flags:1,reserved:3. Флаги: valid=1,video=2,continuation=4; reserved=0. +Она формируется из uAV2 у источника и неизменно пересылается relay. Это диагностическая подсказка непосредственного пира, +а не часть подписанного RS-тела: на проверку подписи, выбор частей и декодирование видео она не влияет. +Считаются только части >= cursor конкретной выдачи, ещё не переданные в ETCP. История кольца и другие получатели не прибавляются. +Начала кадров плюс одно продолжение в начале очереди дают число уникальных кадров, независимо от FEC и дробления большого KEY. +Длительность определяется их PTS и последним измеренным периодом кадра; при постоянном FPS соответствует frames/FPS. +До измерения периода первого кадра время неизвестно. Аудио считается по сэмплам 48kHz; параллельные треки не суммируются. +Источник также учитывает ещё не упакованную publisher-очередь, объединяя границы частично упакованного кадра без дублирования. + +Для каждого viewer выбирается целый фактический входящий путь с максимальной суммой buffered_us. +Desktop/Android показывают две строки: «0.6S → 0.5S → 2S 1.1x», под каждым временем имя узла. +Имя длиннее восьми Unicode code points сокращается до первых восьми и одного символа «…». +Последняя колонка — собственный непрерывный запас до фактической playback-позиции, включая decoder/PCM/device; +перенос между локальными очередями не уменьшает запас повторно. Темп берётся из общего playback clock и виден только при ускорении. +STALE отмечается приглушением и «?», DOWN — приглушением и «×»; последнее известное время сохраняется. Регрессии test_stream_service --recovery-only используют настоящий CHAT/ETCP, подписи и RS 8+0. Медиатест создаёт 50 кадров/с по 6000 байт, KEY 55000 байт на нескольких пачках раз в 50 кадров. diff --git a/src/stream/stream.c b/src/stream/stream.c index 4236eedc..03021fa1 100644 --- a/src/stream/stream.c +++ b/src/stream/stream.c @@ -41,6 +41,10 @@ struct stream_session { int anchored, decoded, closing, invalidated, finished, end_delivered, reason; int restarting, restart_gap, notify_close; int audio_enabled, radio_muted; + uint64_t timing_frame, timing_pts; // Последний новый кадр источника; продолжения не меняют период + uint32_t timing_period; + int timing_seen; + struct stream_frame_timing pending; }; struct STREAM_HANDLE { @@ -961,13 +965,49 @@ int stream_publish_block(struct STREAM_HANDLE *h, const uint8_t *data, size_t le DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream block write not an active publisher"); return STREAM_ERR_STATE; } struct stream_session *s = h->session; s->holds++; - int result = stream_source_write_block(s->source, data, length); + struct stream_frame_timing timing = {0}; + int result = STREAM_OK; + if (s->media) { + struct stream_media_block media; + result = stream_media_parse(data, length, &media); + if (result == STREAM_OK) { + timing.valid = 1; timing.video = media.video_count != 0; + for (unsigned i = 0; i < media.video_count; i++) { + const struct stream_media_video *v = &media.video[i]; + if (!i) { timing.first_frame = v->frame_id; timing.first_pts = v->pts_us; timing.continuation = v->offset != 0; } + timing.last_frame = v->frame_id; timing.last_pts = v->pts_us; + if (!v->offset) timing.started++; + if (!s->timing_seen || v->frame_id > s->timing_frame) { + if (s->timing_seen && v->pts_us > s->timing_pts && v->pts_us - s->timing_pts <= UINT32_MAX) + s->timing_period = (uint32_t)(v->pts_us - s->timing_pts); + s->timing_frame = v->frame_id; s->timing_pts = v->pts_us; s->timing_seen = 1; + } + } + timing.period_us = s->timing_period; + for (unsigned i = 0; i < media.audio_count; i++) timing.audio_us += (uint32_t)media.audio[i].samples * 1000000u / 48000u; + } else DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "publisher frame timing cannot parse media result=%d", result); + } + if (result == STREAM_OK) result = stream_source_write_block(s->source, data, length); + if (result == STREAM_OK && timing.valid) { + struct stream_buffer_stats stats; + result = stream_buffer_get_stats(s->buffer, &stats); + if (result == STREAM_OK) result = stream_buffer_set_timing(s->buffer, stats.highest_seq, &timing); + if (result != STREAM_OK) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "publisher frame timing store failed result=%d", result); + } if (result != STREAM_OK && result != STREAM_ERR_ARGUMENT) session_close(s, STREAM_CLOSE_FAILED, 1, 1); s->holds--; if (s->closing && !s->holds && !s->delivery) session_free(s); return result; } +void stream_publish_pending(struct STREAM_HANDLE *h, const struct stream_frame_timing *pending) { + if (!stream_handle_active(h) || !h->publisher || !pending) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream pending frames invalid publisher/value"); return; + } + h->session->pending = *pending; + if (!h->session->pending.period_us) h->session->pending.period_us = h->session->timing_period; +} + int stream_publish_set_audio_enabled(struct STREAM_HANDLE *h, int enabled) { if (!stream_handle_active(h) || !h->publisher || h->session->finished || (enabled != 0 && enabled != 1)) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream audio change invalid publisher/state enabled=%d", enabled); return STREAM_ERR_STATE; @@ -1077,7 +1117,7 @@ int stream_local_get(struct UTUN_INSTANCE *inst, const struct stream_descriptor *local = (struct stream_local){.descriptor = s->descriptor, .buffer = s->buffer, .media = s->media, .start = s->start_seq, .last = s->last_seq, .next = s->next_seq, .progress_ms = s->progress_ms, .anchored = s->anchored, .decoded = s->decoded, .publisher = s->source != NULL, .finished = s->end_valid, - .relay_only = !s->source && s->owners == s->relay_owners, .end = s->end_valid ? s->end : NULL}; + .relay_only = !s->source && s->owners == s->relay_owners, .end = s->end_valid ? s->end : NULL, .pending = s->pending}; return STREAM_OK; } diff --git a/src/stream/stream_buffer.c b/src/stream/stream_buffer.c index ec4acb74..0bc19282 100644 --- a/src/stream/stream_buffer.c +++ b/src/stream/stream_buffer.c @@ -29,6 +29,7 @@ struct stream_cached_batch { int rejected; int verified; uint8_t verified_parts[(RS_MAX_BLOCKS + 7) / 8]; + struct stream_frame_timing timing; // Подсказка для очередей relay без восстановления FEC }; struct stream_buffer { @@ -42,6 +43,8 @@ struct stream_buffer { struct stream_cached_batch **heap; uint64_t latest[RS_MAX_BLOCKS]; uint8_t latest_valid[RS_MAX_BLOCKS]; + struct stream_telemetry *telemetry[RS_MAX_BLOCKS]; // Последнее измерение каждого получаемого субпотока + uint32_t frame_period_us; size_t block_bytes; uint32_t capacity; uint32_t hash_size; @@ -190,6 +193,7 @@ void stream_buffer_destroy(struct stream_buffer *b) { (unsigned long long)b->stats.rejected_batches); for (uint32_t i = 0; i < b->stats.batches; i++) u_free(b->heap[i]); if (b->slots) for (uint32_t i = 0; i < b->capacity; i++) u_free(b->slots[i].path); + for (unsigned i = 0; i < RS_MAX_BLOCKS; i++) u_free(b->telemetry[i]); rs_free(b->rs); u_free(b->decode_output); u_free(b->data); @@ -475,3 +479,84 @@ int stream_buffer_get_stats(const struct stream_buffer *b, struct stream_buffer_ *stats = b->stats; return STREAM_OK; } + +int stream_buffer_set_timing(struct stream_buffer *b, uint64_t seq, const struct stream_frame_timing *timing) { + if (!b || !timing) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream timing missing buffer/value"); return STREAM_ERR_ARGUMENT; } + struct stream_cached_batch *batch = buffer_batch(b, seq); + if (!batch) return STREAM_EXPIRED; + if (batch->timing.valid) { + /* Сравниваем wire, поскольку padding структуры не является данными протокола. */ + uint8_t a[STREAM_FRAME_TIMING_SIZE], c[STREAM_FRAME_TIMING_SIZE]; + stream_frame_timing_write(a, &batch->timing); stream_frame_timing_write(c, timing); + if (memcmp(a, c, sizeof(a))) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream conflicting frame timing seq=%llu", (unsigned long long)seq); + return STREAM_ERR_CONFLICT; + } + } + batch->timing = *timing; + if (timing->valid && timing->period_us) b->frame_period_us = timing->period_us; + return STREAM_OK; +} + +int stream_buffer_get_timing(struct stream_buffer *b, uint64_t seq, struct stream_frame_timing *timing) { + if (!b || !timing) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream timing missing buffer/output"); return STREAM_ERR_ARGUMENT; } + struct stream_cached_batch *batch = buffer_batch(b, seq); + if (!batch) return STREAM_EXPIRED; + *timing = batch->timing; return STREAM_OK; +} + +/* Число начал плюс один начатый ранее кадр; десятки частей большого KEY не становятся десятками кадров. */ +int stream_buffer_get_delay(struct stream_buffer *b, uint64_t seq, uint8_t sub, struct stream_buffer_delay *out) { + if (!b || !out || sub >= b->descriptor.n) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream delay invalid sub=%u", sub); return STREAM_ERR_ARGUMENT; + } + struct stream_buffer_delay delay = {.known = 1}; + const struct stream_frame_timing *first = NULL, *last = NULL; + uint64_t first_seq = UINT64_MAX, last_seq = 0; + for (uint32_t slot = b->sub_head[sub]; slot != STREAM_SLOT_NONE; slot = b->slots[slot].sub_next) { + if (b->slots[slot].seq < seq) continue; + struct stream_cached_batch *batch = buffer_batch(b, b->slots[slot].seq); + const struct stream_frame_timing *t = &batch->timing; + if (!t->valid) { delay.known = 0; continue; } + delay.audio_us += t->audio_us; + if (!t->video) continue; + delay.frames += t->started; + if (!first || batch->seq < first_seq) { first = t; first_seq = batch->seq; } + if (!last || batch->seq > last_seq) { last = t; last_seq = batch->seq; } + } + if (first) { + delay.frames += first->continuation; + uint32_t period = last->period_us ? last->period_us : b->frame_period_us; + delay.first_pts = first->first_pts; delay.last_pts = last->last_pts; delay.period_us = period; + if (!period || last->last_pts < first->first_pts) delay.known = 0; + else delay.video_us = last->last_pts - first->first_pts + period; + } + *out = delay; return STREAM_OK; +} + +int stream_buffer_set_telemetry(struct stream_buffer *b, uint8_t sub, const struct stream_telemetry *t) { + if (!b || !t || sub >= b->descriptor.n || !t->count || t->count > STREAM_PATH_MAX) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream telemetry cache invalid sub=%u", sub); return STREAM_ERR_ARGUMENT; + } + if (!b->telemetry[sub]) b->telemetry[sub] = u_malloc(sizeof(*t)); + if (!b->telemetry[sub]) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream telemetry cache allocation failed"); return STREAM_ERR_MEMORY; } + *b->telemetry[sub] = *t; return STREAM_OK; +} + +int stream_buffer_get_telemetry(struct stream_buffer *b, uint8_t sub, uint64_t now, struct stream_telemetry *t) { + if (!b || !t || sub >= b->descriptor.n) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream telemetry cache invalid output sub=%u", sub); return STREAM_ERR_ARGUMENT; + } + if (!b->telemetry[sub]) return STREAM_WAIT; + *t = *b->telemetry[sub]; stream_telemetry_age(t, now); return STREAM_OK; +} + +/* Потерян последний переход, доступность более ранних узлов неизвестна; числа остаются прежними. */ +void stream_buffer_telemetry_down(struct stream_buffer *b, uint8_t sub, uint64_t peer) { + if (!b || sub >= b->descriptor.n || !b->telemetry[sub]) return; + struct stream_telemetry *t = b->telemetry[sub]; + if (t->rows[t->count - 1].node != peer) return; + for (unsigned i = 0; i < t->count; i++) t->rows[i].flags |= STREAM_TELEMETRY_STALE; + t->rows[t->count - 1].flags |= STREAM_TELEMETRY_DOWN; + DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream telemetry DOWN sub=%u peer=%llx", sub, (unsigned long long)peer); +} diff --git a/src/stream/stream_buffer.h b/src/stream/stream_buffer.h index fee7b8cd..a19fbc10 100644 --- a/src/stream/stream_buffer.h +++ b/src/stream/stream_buffer.h @@ -14,6 +14,21 @@ extern "C" { #endif struct stream_buffer; +struct stream_buffer_delay { + uint64_t frames, video_us, audio_us; // Непереданные кадры и длительности параллельных треков + uint64_t first_pts, last_pts; // Границы видео для объединения с ещё не упакованной очередью источника + uint32_t period_us; // Измеренный период последнего кадра + int known; // 0 — отсутствует кадровая подсказка или ещё неизвестен период первого кадра +}; +/* Подсказки DATA хранятся один раз на пачку, независимо от числа RS-частей; на воспроизведение не влияют. */ +int stream_buffer_set_timing(struct stream_buffer *buffer, uint64_t seq, const struct stream_frame_timing *timing); +int stream_buffer_get_timing(struct stream_buffer *buffer, uint64_t seq, struct stream_frame_timing *timing); +/* Длительность очереди одного subblock от seq; история до cursor не учитывается. */ +int stream_buffer_get_delay(struct stream_buffer *buffer, uint64_t seq, uint8_t subblock, struct stream_buffer_delay *delay); +/* Последняя телеметрия субпотока переживает закрытие receive; память принадлежит кольцу. */ +int stream_buffer_set_telemetry(struct stream_buffer *buffer, uint8_t subblock, const struct stream_telemetry *telemetry); +int stream_buffer_get_telemetry(struct stream_buffer *buffer, uint8_t subblock, uint64_t now_ms, struct stream_telemetry *telemetry); +void stream_buffer_telemetry_down(struct stream_buffer *buffer, uint8_t subblock, uint64_t peer); /* Observer не выполняет I/O и не меняет кольцо; wire действует лишь внутри callback. Destroy observer не вызывает. */ typedef void (*stream_buffer_event_fn)(void *arg, const uint8_t *wire, size_t length, int added); void stream_buffer_set_observer(struct stream_buffer *buffer, stream_buffer_event_fn observer, void *arg); diff --git a/src/stream/stream_network.c b/src/stream/stream_network.c index bb062ad4..b836d404 100644 --- a/src/stream/stream_network.c +++ b/src/stream/stream_network.c @@ -73,6 +73,8 @@ struct STREAM_RECEIVE { struct stream_load load; uint64_t admitted_ms, pending_highest, probe, probe_sent_ms; int account_failed; + uint64_t telemetry_sent_ms, telemetry_sample; + int telemetry_dirty; }; struct stream_peer { struct stream_peer *next; @@ -106,6 +108,7 @@ struct stream_network { struct stream_relay *relays; uint64_t relay_epoch, info_order; unsigned relay_count, relay_request_count; + uint64_t telemetry_sequence; }; static void peer_pump(void *arg); @@ -283,6 +286,11 @@ static int route_control(struct stream_network *net, uint64_t peer, struct strea static void flow_close(struct STREAM_RECEIVE *f, int reason, int notify) { if (f->info.state == STREAM_RECEIVE_CLOSED) return; struct stream_peer *p = f->peer; + if (f->outgoing && !f->info.probe_only && p && reason != STREAM_REASON_STOPPED) { + struct stream_local local; + if (stream_local_get(p->network->inst, &f->descriptor, &local) == STREAM_OK) + stream_buffer_telemetry_down(local.buffer, f->subblock, p->node_id); + } enum stream_receive_state previous = f->info.state; if (!f->outgoing && p) { for (unsigned i = 0; i < p->pressure.count; i++) { @@ -613,6 +621,60 @@ static int raw_send(struct stream_peer *p, struct stream_message m, const uint8_ return STREAM_OK; } +/* Путь измерения относится к очередной DATA; пустая очередь сохраняет путь последнего доступного блока. */ +static int telemetry_send(struct STREAM_RECEIVE *f, uint64_t now) { + struct stream_peer *p = f->peer; + struct stream_local local; + if (stream_local_get(p->network->inst, &f->descriptor, &local) != STREAM_OK) return 0; + const uint8_t *wire; size_t length; uint64_t actual; + struct stream_path path = {.count = 1, .nodes = {p->network->inst->node_id}}; + int found = stream_buffer_read(local.buffer, f->info.next, f->subblock, &wire, &length, &actual); + if (found != STREAM_OK) found = stream_buffer_read(local.buffer, UINT64_MAX, f->subblock, &wire, &length, &actual); + if (found == STREAM_OK) { + if (stream_buffer_path(local.buffer, actual, f->subblock, &path) != STREAM_OK) return 0; + } else if (!local.publisher) return 0; + if (stream_path_validate(&path, f->descriptor.source_node_id, p->network->inst->node_id, p->node_id) != STREAM_OK) return 0; + struct stream_telemetry t = {0}; + if (path.count > 1) { + int matched = stream_buffer_get_telemetry(local.buffer, f->subblock, now, &t) == STREAM_OK && t.count + 1 == path.count; + for (unsigned i = 0; matched && i < t.count; i++) if (t.rows[i].node != path.nodes[i]) matched = 0; + if (!matched) { + t = (struct stream_telemetry){.count = (uint8_t)(path.count - 1), .received_ms = now}; + for (unsigned i = 0; i < t.count; i++) t.rows[i] = (struct stream_telemetry_row){.node = path.nodes[i], + .age_ms = UINT32_MAX, .flags = STREAM_TELEMETRY_STALE}; + } + } + struct stream_buffer_delay delay; + if (stream_buffer_get_delay(local.buffer, f->info.next, f->subblock, &delay) != STREAM_OK) return 0; + if (local.publisher && local.pending.valid) { + if (local.pending.video) { + uint64_t first_pts = delay.frames ? delay.first_pts : local.pending.first_pts; + uint32_t period = local.pending.period_us ? local.pending.period_us : delay.period_us; + delay.frames += local.pending.started + (local.pending.continuation && !delay.frames); + if (!period || local.pending.last_pts < first_pts) delay.known = 0; + else delay.video_us = local.pending.last_pts - first_pts + period; + } + delay.audio_us += local.pending.audio_us; + } + if (p->network->telemetry_sequence == UINT64_MAX) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream telemetry sequence exhausted"); return 0; + } + struct stream_telemetry_row *own = &t.rows[t.count++]; + *own = (struct stream_telemetry_row){.node = p->network->inst->node_id, .sample = ++p->network->telemetry_sequence, + .buffered_us = delay.known ? (delay.video_us > delay.audio_us ? delay.video_us : delay.audio_us) : 0, + .frames = delay.known ? (delay.frames > UINT32_MAX ? UINT32_MAX : (uint32_t)delay.frames) : 0, + .flags = delay.known ? STREAM_TELEMETRY_KNOWN : 0}; + uint8_t payload[1 + STREAM_PATH_MAX * STREAM_TELEMETRY_ROW_SIZE]; + struct stream_message m = {.descriptor = f->descriptor, .handle = f->info.handle, .subblock = f->subblock, + .command = STREAM_CMD_TELEMETRY}; + if (raw_send(p, m, payload, stream_telemetry_write(payload, &t)) != STREAM_OK) return 0; + f->telemetry_sent_ms = now; f->telemetry_dirty = 0; + DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "stream telemetry TX peer=%llx sub=%u nodes=%u own_frames=%u own_us=%llu total_us=%llu", + (unsigned long long)p->node_id, f->subblock, t.count, own->frames, + (unsigned long long)own->buffered_us, (unsigned long long)stream_telemetry_total(&t)); + return 1; +} + /* Нижняя граница прогресса всех выбранных выдач. INFO/REQUESTED ещё могут принести старую пачку и запрещают скачок. */ static void receive_progress(struct stream_network *net, const struct stream_descriptor *d) { uint8_t seen[RS_MAX_BLOCKS] = {0}; unsigned distinct = 0; uint64_t before = UINT64_MAX; @@ -641,8 +703,12 @@ static int downstream_send(struct STREAM_RECEIVE *f) { || stream_path_validate(&path, f->descriptor.source_node_id, p->network->inst->node_id, p->node_id) != STREAM_OK) { flow_close(f, STREAM_REASON_LOOP, 1); return 0; } - uint8_t payload[1 + STREAM_PATH_MAX * 8 + STREAM_BLOCK_MAX_SIZE]; - size_t path_bytes = stream_path_write(payload, &path); memcpy(payload + path_bytes, wire, length); + uint8_t payload[1 + STREAM_PATH_MAX * 8 + STREAM_FRAME_TIMING_SIZE + STREAM_BLOCK_MAX_SIZE]; + size_t path_bytes = stream_path_write(payload, &path); + struct stream_frame_timing timing = {0}; + if (stream_buffer_get_timing(local.buffer, actual, &timing) != STREAM_OK) return 0; + stream_frame_timing_write(payload + path_bytes, &timing); path_bytes += STREAM_FRAME_TIMING_SIZE; + memcpy(payload + path_bytes, wire, length); m.command = STREAM_CMD_DATA; if (raw_send(p, m, payload, path_bytes + length) != STREAM_OK) { flow_close(f, STREAM_REASON_TRANSPORT, 0); return 0; } if (f->info.state == STREAM_RECEIVE_ACTIVE) { @@ -740,9 +806,15 @@ static void peer_pump(void *arg) { } else sent = 1; queue_resume_callback(p->control); } else { + uint64_t now = get_time_tb() / 10; + for (struct STREAM_RECEIVE *flow = p->flows; flow; flow = flow->next) { + if (flow->outgoing || flow->info.state != STREAM_RECEIVE_ACTIVE) continue; + if ((flow->telemetry_dirty || !flow->telemetry_sent_ms || now - flow->telemetry_sent_ms >= STREAM_TELEMETRY_PERIOD_MS) + && telemetry_send(flow, now)) { sent = 1; break; } + } struct STREAM_RECEIVE *first = p->last_sent && p->last_sent->next ? p->last_sent->next : p->flows; struct STREAM_RECEIVE *f = first; - while (f) { + while (!sent && f) { if (f->account_failed) flow_close(f, STREAM_REASON_PROTOCOL, 1); if (!f->outgoing && f->info.state == STREAM_RECEIVE_ACTIVE && downstream_send(f)) { p->last_sent = f; sent = 1; break; } f = f->next ? f->next : p->flows; @@ -823,7 +895,7 @@ static void subscribe_request(struct stream_peer *p, struct stream_message m, st static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t *wire, size_t length, int routed) { struct stream_message m; struct stream_local local; if (stream_wire_parse(wire, length, &m) != STREAM_OK || !sender || sender == net->inst->node_id) return; - if (m.command >= STREAM_CMD_RELAY_REQUEST) { + if (m.command >= STREAM_CMD_RELAY_REQUEST && m.command <= STREAM_CMD_RELAY_CANCEL) { if (routed) { relay_dispatch(net, sender, &m); stream_network_wake(net); } else DEBUG_WARN(DEBUG_CATEGORY_STREAM, "relay command requires signed routed control peer=%llx command=%u", (unsigned long long)sender, m.command); @@ -932,17 +1004,44 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream receive ACCEPT peer=%llx handle=%llu sub=%u start=%llu", (unsigned long long)sender, (unsigned long long)m.handle, m.subblock, (unsigned long long)start); } + } else if (m.command == STREAM_CMD_TELEMETRY && f && f->info.state == STREAM_RECEIVE_ACTIVE) { + struct stream_telemetry t; struct stream_path path = {0}; + int valid = stream_telemetry_read(m.payload, m.length, &t) == STREAM_OK; + if (valid) { + path.count = t.count; + for (unsigned i = 0; i < t.count; i++) path.nodes[i] = t.rows[i].node; + valid = stream_path_validate(&path, f->descriptor.source_node_id, sender, net->inst->node_id) == STREAM_OK; + } + if (valid && t.rows[t.count - 1].sample > f->telemetry_sample) { + t.received_ms = get_time_tb() / 10; + f->telemetry_sample = t.rows[t.count - 1].sample; + if (stream_buffer_set_telemetry(local.buffer, f->subblock, &t) == STREAM_OK) { + for (struct stream_peer *down = net->peers; down; down = down->next) + for (struct STREAM_RECEIVE *flow = down->flows; flow; flow = flow->next) + if (!flow->outgoing && flow->subblock == f->subblock && same_stream(&flow->descriptor, &f->descriptor)) { + flow->telemetry_dirty = 1; peer_schedule(down); + } + DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "stream telemetry RX peer=%llx sub=%u nodes=%u total_us=%llu", + (unsigned long long)sender, f->subblock, t.count, (unsigned long long)stream_telemetry_total(&t)); + } + } else if (!valid) DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream telemetry rejected peer=%llx sub=%u", + (unsigned long long)sender, f->subblock); } else if (m.command == STREAM_CMD_DATA && f && f->info.state == STREAM_RECEIVE_ACTIVE) { struct stream_path path; size_t used; if (stream_path_parse(m.payload, m.length, &path, &used) != STREAM_OK || stream_path_validate(&path, f->descriptor.source_node_id, sender, net->inst->node_id) != STREAM_OK) { flow_close(f, STREAM_REASON_LOOP, 1); - } else if (m.length - used <= STREAM_BLOCK_OVERHEAD + RS_HEADER_SIZE - || m.length - used > stream_block_capacity(&f->descriptor)) { + } else if (m.length - used <= STREAM_FRAME_TIMING_SIZE + STREAM_BLOCK_OVERHEAD + RS_HEADER_SIZE + || m.length - used - STREAM_FRAME_TIMING_SIZE > stream_block_capacity(&f->descriptor)) { DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream DATA part size invalid peer=%llx bytes=%zu max=%zu", (unsigned long long)sender, m.length - used, stream_block_capacity(&f->descriptor)); flow_close(f, STREAM_REASON_PROTOCOL, 1); } else { + struct stream_frame_timing timing; + if (stream_frame_timing_read(m.payload + used, m.length - used, &timing) != STREAM_OK) { + flow_close(f, STREAM_REASON_PROTOCOL, 1); p->busy--; peer_sweep(p); return; + } + used += STREAM_FRAME_TIMING_SIZE; const uint8_t *block = m.payload + used; uint64_t seq = stream_read_be(block + STREAM_BLOCK_SEQ_OFF, 8); if (block[STREAM_BLOCK_SUB_OFF] != f->subblock || seq != f->info.next || stream_read_be(block + STREAM_BLOCK_GROUP_OFF, 8) != f->descriptor.group_id @@ -951,6 +1050,9 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t else { int result = stream_ingest_path(net->inst, block, m.length - used, &path); if (result == STREAM_OK || result == STREAM_DUPLICATE || result == STREAM_EXPIRED) { + if (result != STREAM_EXPIRED && stream_buffer_set_timing(local.buffer, seq, &timing) < 0) + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream received frame timing conflict peer=%llx seq=%llu", + (unsigned long long)sender, (unsigned long long)seq); stream_metrics_observe(&f->metrics, seq, (uint16_t)stream_read_be(block + STREAM_BLOCK_TIMESTAMP_OFF, 2), block[STREAM_BLOCK_FLAGS_OFF], (uint32_t)(m.length - used), get_time_tb() / 10); if (f->info.path.count != path.count || memcmp(f->info.path.nodes, path.nodes, path.count * sizeof(uint64_t))) @@ -1185,6 +1287,8 @@ static void network_tick(void *arg) { expire_controls(p, now_ms); for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next) { if (f->info.state == STREAM_RECEIVE_CLOSED) continue; + if (!f->outgoing && f->info.state == STREAM_RECEIVE_ACTIVE + && (!f->telemetry_sent_ms || now_ms - f->telemetry_sent_ms >= STREAM_TELEMETRY_PERIOD_MS)) peer_schedule(p); f->info.metrics_valid = stream_metrics_poll(&f->metrics, now_ms, &f->info.metrics) == STREAM_OK; if ((!f->outgoing || (f->info.state < STREAM_RECEIVE_ACTIVE && !(f->info.probe_only && f->info.info_ready)) || (f->outgoing && !f->info.probe_only && f->info.state == STREAM_RECEIVE_ACTIVE && !f->info.blocks @@ -1985,6 +2089,40 @@ uint64_t stream_network_pending_bytes(struct UTUN_INSTANCE *inst, const struct s return bytes; } +/* Выбираем целый путь; общие узлы разных субпотоков никогда не склеиваются в искусственную цепочку. */ +int stream_network_telemetry(struct UTUN_INSTANCE *inst, const struct stream_descriptor *d, struct stream_telemetry *out) { + struct stream_network *net = stream_network_get(inst); struct stream_local local; + if (!net || !d || !out || stream_local_get(inst, d, &local) != STREAM_OK) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream telemetry snapshot unavailable"); return STREAM_ERR_STATE; + } + struct stream_telemetry worst = {0}; uint64_t total = 0, now = get_time_tb() / 10; + int active = 0; + for (struct stream_peer *p = net->peers; p; p = p->next) + for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next) { + if (!f->outgoing || f->info.probe_only || f->info.state != STREAM_RECEIVE_ACTIVE + || !same_stream(&f->descriptor, d) || !f->info.path.count) continue; + active = 1; + struct stream_telemetry t; + int matched = stream_buffer_get_telemetry(local.buffer, f->subblock, now, &t) == STREAM_OK + && t.count == f->info.path.count; + for (unsigned i = 0; matched && i < t.count; i++) if (t.rows[i].node != f->info.path.nodes[i]) matched = 0; + if (!matched) { + t = (struct stream_telemetry){.count = f->info.path.count, .received_ms = now}; + for (unsigned i = 0; i < t.count; i++) t.rows[i] = (struct stream_telemetry_row){ + .node = f->info.path.nodes[i], .flags = STREAM_TELEMETRY_STALE, .age_ms = UINT32_MAX}; + } + uint64_t sum = stream_telemetry_total(&t); + if (!worst.count || sum > total) { worst = t; total = sum; } + } + if (!active) for (unsigned sub = 0; sub < d->n; sub++) { + struct stream_telemetry t; + if (stream_buffer_get_telemetry(local.buffer, (uint8_t)sub, now, &t) != STREAM_OK) continue; + uint64_t sum = stream_telemetry_total(&t); + if (!worst.count || sum > total) { worst = t; total = sum; } + } + *out = worst; return STREAM_OK; +} + int stream_network_buffer_positions(struct UTUN_INSTANCE *inst, const struct stream_descriptor *d, struct stream_buffer_positions *out) { struct stream_network *net = stream_network_get(inst); diff --git a/src/stream/stream_network.h b/src/stream/stream_network.h index ec392f06..39df9431 100644 --- a/src/stream/stream_network.h +++ b/src/stream/stream_network.h @@ -62,6 +62,8 @@ uint64_t stream_network_pending_bytes(struct UTUN_INSTANCE *inst, const struct s * out обязателен; STREAM_OK либо ошибка аргументов/отсутствия локальной подписки. Кольцо не сканируется. */ int stream_network_buffer_positions(struct UTUN_INSTANCE *inst, const struct stream_descriptor *descriptor, struct stream_buffer_positions *out); +/* Худший фактический входящий путь по сумме длительностей; после разрыва сохраняет последнюю цепочку. */ +int stream_network_telemetry(struct UTUN_INSTANCE *inst, const struct stream_descriptor *descriptor, struct stream_telemetry *out); int stream_network_init(struct UTUN_INSTANCE *inst, struct stream_network **out); void stream_network_destroy(struct stream_network *network); diff --git a/src/stream/stream_priv.h b/src/stream/stream_priv.h index 1bb7e3f8..a8e93d50 100644 --- a/src/stream/stream_priv.h +++ b/src/stream/stream_priv.h @@ -15,11 +15,14 @@ struct stream_local { uint64_t progress_ms; // Монотонное время последней полной проверенной пачки, не прихода отдельной RS-части int anchored, decoded, publisher, finished, relay_only; const uint8_t *end; + struct stream_frame_timing pending; // Ещё не упакованные publisher-кадры; отсутствуют у relay }; /* Заимствованные буфер/END действуют только до следующего callback/мутации сервиса. */ struct stream_network *stream_network_get(struct UTUN_INSTANCE *inst); void stream_optimizers_tick(struct UTUN_INSTANCE *inst, uint64_t now_ms); int stream_local_get(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key, struct stream_local *local); +/* Publisher обновляет диагностику своей ограниченной очереди в uasync; время не включает ETCP. */ +void stream_publish_pending(struct STREAM_HANDLE *handle, const struct stream_frame_timing *pending); /* По запросу media-получателя создать общий индекс поставщика; поиск истории выполняется отложенно с бюджетом. */ int stream_local_media_prepare(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key); int stream_handle_context(struct STREAM_HANDLE *handle, struct UTUN_INSTANCE **inst, struct stream_descriptor *descriptor); diff --git a/src/stream/stream_publisher.c b/src/stream/stream_publisher.c index 711cd760..7482c2dc 100644 --- a/src/stream/stream_publisher.c +++ b/src/stream/stream_publisher.c @@ -8,6 +8,7 @@ #include "stream.h" #include "stream_network.h" #include "stream_publisher.h" +#include "stream_priv.h" #define TX_LIMIT (2u * 1024u * 1024u) #define TX_RECORDS 128 @@ -246,6 +247,21 @@ static void publisher_process(void *arg) { p, publisher_process, "publisher_media"); if (!p->timer) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "publisher media timer allocation failed"); publisher_close(p, STREAM_CLOSE_FAILED); } } + if (p->handle && stream_handle_active(p->handle)) { + struct stream_frame_timing pending = {.valid = 1, .audio_us = p->audio.count * 20000u}; + struct publisher_record *previous = NULL; + for (struct publisher_record *r = p->video.head; r; r = r->next) { + if (!previous) { + pending.video = 1; pending.first_frame = r->frame_id; pending.first_pts = r->pts; + pending.continuation = r->offset != 0; + } else if (r->pts > previous->pts && r->pts - previous->pts <= UINT32_MAX) + pending.period_us = (uint32_t)(r->pts - previous->pts); + pending.last_frame = r->frame_id; pending.last_pts = r->pts; + if (!r->offset) pending.started++; + previous = r; + } + stream_publish_pending(p->handle, &pending); + } pthread_mutex_unlock(&p->lock); } diff --git a/src/stream/stream_viewer.c b/src/stream/stream_viewer.c index f5cdf556..a228113b 100644 --- a/src/stream/stream_viewer.c +++ b/src/stream/stream_viewer.c @@ -1,4 +1,5 @@ #include +#include #include #include "../../lib/debug_config.h" #include "../../lib/mem.h" @@ -6,6 +7,7 @@ #include "../utun_instance.h" #include "stream.h" #include "stream_viewer.h" +#include "stream_network.h" struct viewer { struct viewer *next; @@ -16,6 +18,9 @@ struct viewer { void (*notify)(void *); void *notify_arg; struct stream_media_status status; + struct stream_viewer_telemetry telemetry; + uint64_t telemetry_requested_ms; + uint64_t telemetry_report_ms; int opening, stopping, resync, requested, waiting; unsigned video, records, ended_tracks; int64_t position, audio_ahead; @@ -34,6 +39,38 @@ struct stream_viewers { }; static void process(void *arg); +/* Имена разрешаются в потоке ядра; UI получает owned строки и никогда не обращается к registry. */ +static void viewer_telemetry(struct stream_viewers *v, struct viewer *s) { + uint64_t now = get_time_tb() / 10; + if (s->telemetry.sample_ms && now - s->telemetry.sample_ms < STREAM_TELEMETRY_PERIOD_MS) return; + struct stream_descriptor descriptor = {.group_id = s->group, .source_node_id = s->source, .source_epoch = s->epoch}; + struct stream_info info; + if (stream_get_info(s->handle, &info) != STREAM_OK) return; + descriptor = info.descriptor; + if (stream_network_telemetry(v->inst, &descriptor, &s->telemetry.chain) != STREAM_OK) return; + s->telemetry.self = v->inst->node_id; s->telemetry.sample_ms = now; + for (unsigned i = 0; i <= s->telemetry.chain.count; i++) { + uint64_t id = i < s->telemetry.chain.count ? s->telemetry.chain.rows[i].node : v->inst->node_id; + struct TOPO_NODE *node = topo_node_registry_find(v->inst->topo_groups, id); + const char *name = node && node->node_name ? node->node_name : id == v->inst->node_id ? v->inst->name : NULL; + if (name && *name) snprintf(s->telemetry.names[i], sizeof(s->telemetry.names[i]), "%s", name); + else snprintf(s->telemetry.names[i], sizeof(s->telemetry.names[i]), "%016llx", (unsigned long long)id); + } + if (!s->telemetry_report_ms || now - s->telemetry_report_ms >= 1000) { + unsigned stale = 0, down = 0, unknown = 0; uint32_t age = 0; + for (unsigned i = 0; i < s->telemetry.chain.count; i++) { + const struct stream_telemetry_row *r = &s->telemetry.chain.rows[i]; + stale += !!(r->flags & STREAM_TELEMETRY_STALE); down += !!(r->flags & STREAM_TELEMETRY_DOWN); + unknown += !(r->flags & STREAM_TELEMETRY_KNOWN); if (r->age_ms > age) age = r->age_ms; + } + DEBUG_DEBUG(DEBUG_CATEGORY_VIDEO, + "stream BUFFER id=%llu group=%llx source=%llx epoch=%llx nodes=%u total_us=%llu stale=%u down=%u unknown=%u max_age_ms=%u", + (unsigned long long)s->id, (unsigned long long)s->group, (unsigned long long)s->source, (unsigned long long)s->epoch, + s->telemetry.chain.count, (unsigned long long)stream_telemetry_total(&s->telemetry.chain), stale, down, unknown, age); + s->telemetry_report_ms = now; + } +} + static struct viewer *find(struct stream_viewers *v, uint64_t id) { for (struct viewer *s = v->head; s; s = s->next) if (s->id == id) return s; return NULL; @@ -133,6 +170,7 @@ 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; + viewer_telemetry(v, s); if (s->tail && s->tail->kind == STREAM_VIEWER_GAP) continue; if (s->resync) { s->resync = s->waiting = 0; s->position = -1; s->ended_tracks = 0; @@ -371,6 +409,23 @@ int stream_viewer_status(struct stream_viewers *v, uint64_t id, struct stream_me pthread_mutex_unlock(&v->lock); return result; } +int stream_viewer_get_telemetry(struct stream_viewers *v, uint64_t id, struct stream_viewer_telemetry *out) { + if (!v || !out) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "viewer telemetry missing context/output"); return -1; } + pthread_mutex_lock(&v->lock); + struct viewer *s = find(v, id); + int result = s && !s->stopping ? 0 : -1; + if (!result) { + *out = s->telemetry; + uint64_t now = get_time_tb() / 10; + stream_telemetry_age(&out->chain, now); + if (s->handle && now - s->telemetry_requested_ms >= STREAM_TELEMETRY_PERIOD_MS) { + s->telemetry_requested_ms = now; + if (schedule(v) < 0) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "viewer telemetry refresh scheduling failed"); + } + } + pthread_mutex_unlock(&v->lock); return result; +} + int stream_viewer_resync(struct stream_viewers *v, uint64_t id) { if (!v) return -1; pthread_mutex_lock(&v->lock); diff --git a/src/stream/stream_viewer.h b/src/stream/stream_viewer.h index c06f040c..470bda67 100644 --- a/src/stream/stream_viewer.h +++ b/src/stream/stream_viewer.h @@ -11,6 +11,11 @@ extern "C" { #endif struct UTUN_INSTANCE; struct stream_viewers; +struct stream_viewer_telemetry { + struct stream_telemetry chain; /* Худшая входящая цепочка без локального playback-буфера. */ + uint64_t self, sample_ms; /* Локальный узел и монотонное время снимка. */ + char names[STREAM_PATH_MAX + 1][256]; /* Имена строк chain, затем имя локального узла; UTF-8. */ +}; #define STREAM_VIEWER_BATCH_RECORDS 16u enum stream_viewer_kind { STREAM_VIEWER_VIDEO = 1, STREAM_VIEWER_AUDIO, STREAM_VIEWER_GAP, @@ -53,6 +58,8 @@ int stream_viewer_request(struct stream_viewers *viewers, uint64_t id, int64_t p int stream_viewer_set_notify(struct stream_viewers *viewers, uint64_t id, void (*notify)(void *), void *arg); /* Последний снимок сохраняется после END до stop. 0 — получен, -1 — ID отсутствует. */ int stream_viewer_status(struct stream_viewers *viewers, uint64_t id, struct stream_media_status *status); +/* Последний снимок цепочки; чтение раз в 200ms запрашивает обновление в uasync даже при остановке DATA. */ +int stream_viewer_get_telemetry(struct stream_viewers *viewers, uint64_t id, struct stream_viewer_telemetry *telemetry); /* Очистить ответ и искать ближайший KEY только этого reader. 0 — принято, -1 — отказ. */ int stream_viewer_resync(struct stream_viewers *viewers, uint64_t id); #ifdef __cplusplus diff --git a/src/stream/stream_wire.c b/src/stream/stream_wire.c index 7bb9670a..ee44247b 100644 --- a/src/stream/stream_wire.c +++ b/src/stream/stream_wire.c @@ -7,7 +7,7 @@ /* Заголовок одинаков у raw и router payload; cmd raw сохраняется внутри routed-кодограммы. */ int stream_wire_parse(const uint8_t *wire, size_t length, struct stream_message *m) { if (!wire || !m || length < STREAM_WIRE_HEADER || wire[0] != ETCP_ID_STREAM || wire[1] != STREAM_WIRE_VERSION - || wire[2] < STREAM_CMD_INFO_REQ || wire[2] > STREAM_CMD_RELAY_CANCEL) { + || wire[2] < STREAM_CMD_INFO_REQ || wire[2] > STREAM_CMD_TELEMETRY) { DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream wire header invalid bytes=%zu", length); return STREAM_ERR_FORMAT; } struct stream_message result = {.command = wire[2], .subblock = wire[3], .payload = wire + STREAM_WIRE_HEADER, @@ -30,6 +30,88 @@ void stream_wire_header(uint8_t *wire, const struct stream_message *m) { stream_write_be(wire + 36, m->receiver_epoch, 8); stream_write_be(wire + 44, m->handle, 8); } +/* Подсказка не меняет подписанные RS-байты и не используется как доверенный media index. */ +void stream_frame_timing_write(uint8_t *wire, const struct stream_frame_timing *t) { + memset(wire, 0, STREAM_FRAME_TIMING_SIZE); + stream_write_be(wire, t->first_frame, 8); stream_write_be(wire + 8, t->last_frame, 8); + stream_write_be(wire + 16, t->first_pts, 8); stream_write_be(wire + 24, t->last_pts, 8); + stream_write_be(wire + 32, t->period_us, 4); stream_write_be(wire + 36, t->started, 4); + stream_write_be(wire + 40, t->audio_us, 4); + wire[44] = t->valid | (t->video << 1) | (t->continuation << 2); +} + +int stream_frame_timing_read(const uint8_t *wire, size_t length, struct stream_frame_timing *out) { + if (!wire || !out || length < STREAM_FRAME_TIMING_SIZE || (wire[44] & ~7u) || wire[45] || wire[46] || wire[47]) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream frame timing invalid header bytes=%zu", length); return STREAM_ERR_FORMAT; + } + struct stream_frame_timing t = {.first_frame = stream_read_be(wire, 8), .last_frame = stream_read_be(wire + 8, 8), + .first_pts = stream_read_be(wire + 16, 8), .last_pts = stream_read_be(wire + 24, 8), + .period_us = (uint32_t)stream_read_be(wire + 32, 4), .started = (uint32_t)stream_read_be(wire + 36, 4), + .audio_us = (uint32_t)stream_read_be(wire + 40, 4), .valid = wire[44] & 1, + .video = (wire[44] >> 1) & 1, .continuation = (wire[44] >> 2) & 1}; + if ((!t.valid && (t.video || t.continuation || t.audio_us || t.period_us || t.started || t.first_frame || t.last_frame + || t.first_pts || t.last_pts)) + || (!t.video && (t.continuation || t.started || t.first_frame || t.last_frame || t.first_pts || t.last_pts)) + || (t.video && (t.last_frame < t.first_frame || t.last_pts < t.first_pts || t.last_pts > INT64_MAX + || t.started > 16 || (!t.started && !t.continuation))) || t.audio_us > 320000) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream frame timing invalid values frames=%u video=%u continuation=%u audio_us=%u", + t.started, t.video, t.continuation, t.audio_us); return STREAM_ERR_FORMAT; + } + *out = t; return STREAM_OK; +} + +size_t stream_telemetry_write(uint8_t *wire, const struct stream_telemetry *t) { + wire[0] = t->count; + for (unsigned i = 0; i < t->count; i++) { + const struct stream_telemetry_row *r = &t->rows[i]; uint8_t *p = wire + 1 + i * STREAM_TELEMETRY_ROW_SIZE; + stream_write_be(p, r->node, 8); stream_write_be(p + 8, r->buffered_us, 8); stream_write_be(p + 16, r->sample, 8); + stream_write_be(p + 24, r->frames, 4); stream_write_be(p + 28, r->age_ms, 4); p[32] = r->flags; + } + return 1 + t->count * STREAM_TELEMETRY_ROW_SIZE; +} + +int stream_telemetry_read(const uint8_t *wire, size_t length, struct stream_telemetry *out) { + if (!wire || !out || !length || !wire[0] || wire[0] > STREAM_PATH_MAX + || length != 1u + wire[0] * STREAM_TELEMETRY_ROW_SIZE) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream telemetry invalid size=%zu", length); return STREAM_ERR_FORMAT; + } + struct stream_telemetry t = {.count = wire[0]}; + for (unsigned i = 0; i < t.count; i++) { + const uint8_t *p = wire + 1 + i * STREAM_TELEMETRY_ROW_SIZE; + struct stream_telemetry_row *r = &t.rows[i]; + *r = (struct stream_telemetry_row){.node = stream_read_be(p, 8), .buffered_us = stream_read_be(p + 8, 8), + .sample = stream_read_be(p + 16, 8), .frames = (uint32_t)stream_read_be(p + 24, 4), + .age_ms = (uint32_t)stream_read_be(p + 28, 4), .flags = p[32]}; + if (!r->node || r->buffered_us > INT64_MAX || (r->flags & ~7u) + || (!(r->flags & STREAM_TELEMETRY_KNOWN) && (r->buffered_us || r->frames))) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream telemetry invalid row=%u flags=%u", i, r->flags); return STREAM_ERR_FORMAT; + } + for (unsigned j = 0; j < i; j++) if (t.rows[j].node == r->node) { + DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream telemetry repeated node row=%u", i); return STREAM_ERR_FORMAT; + } + } + *out = t; return STREAM_OK; +} + +void stream_telemetry_age(struct stream_telemetry *t, uint64_t now) { + uint64_t elapsed = now >= t->received_ms ? now - t->received_ms : 0; + for (unsigned i = 0; i < t->count; i++) { + struct stream_telemetry_row *r = &t->rows[i]; + r->age_ms = elapsed > UINT32_MAX - r->age_ms ? UINT32_MAX : r->age_ms + (uint32_t)elapsed; + if (r->age_ms >= STREAM_TELEMETRY_STALE_MS) r->flags |= STREAM_TELEMETRY_STALE; + } + t->received_ms = now; +} + +uint64_t stream_telemetry_total(const struct stream_telemetry *t) { + uint64_t total = 0; + for (unsigned i = 0; i < t->count; i++) { + uint64_t value = t->rows[i].buffered_us; + total = value > UINT64_MAX - total ? UINT64_MAX : total + value; + } + return total; +} + /* История блока проверяется отдельно от BGP-пути; получатель не должен уже встречаться в ней. */ int stream_path_validate(const struct stream_path *path, uint64_t source, uint64_t sender, uint64_t forbidden) { if (!path || !path->count || path->count > STREAM_PATH_MAX || path->nodes[0] != source diff --git a/src/stream/stream_wire.h b/src/stream/stream_wire.h index d7f1b9ab..ecbe3dce 100644 --- a/src/stream/stream_wire.h +++ b/src/stream/stream_wire.h @@ -10,7 +10,7 @@ extern "C" { #endif -#define STREAM_WIRE_VERSION 6 +#define STREAM_WIRE_VERSION 7 #define STREAM_WIRE_HEADER 52 #define STREAM_PATH_MAX 32 #define STREAM_END_SIZE 106 @@ -22,12 +22,16 @@ extern "C" { #define STREAM_INFO_SIZE (STREAM_INFO_PATH_OFF + 1) /* probe:8, границы/FEC:28, метрики:49, предложение:25, KEY:16, count:1 */ #define STREAM_OVERLOAD_MAX 64 #define STREAM_OVERLOAD_HEADER 18 +#define STREAM_FRAME_TIMING_SIZE 48 +#define STREAM_TELEMETRY_PERIOD_MS 200 +#define STREAM_TELEMETRY_STALE_MS 400 +#define STREAM_TELEMETRY_ROW_SIZE 33 enum stream_command { STREAM_CMD_INFO_REQ = 1, STREAM_CMD_INFO_RESP, STREAM_CMD_SUBSCRIBE, STREAM_CMD_ACCEPT, STREAM_CMD_REJECT, STREAM_CMD_RENEW, STREAM_CMD_UNSUBSCRIBE, STREAM_CMD_CLOSED, STREAM_CMD_DATA, STREAM_CMD_GAP, STREAM_CMD_END, STREAM_CMD_REVERSE_REQ, STREAM_CMD_OVERLOAD, STREAM_CMD_OVERLOAD_SELECT, - STREAM_CMD_RELAY_REQUEST, STREAM_CMD_RELAY_STATUS, STREAM_CMD_RELAY_CANCEL + STREAM_CMD_RELAY_REQUEST, STREAM_CMD_RELAY_STATUS, STREAM_CMD_RELAY_CANCEL, STREAM_CMD_TELEMETRY }; /* Routed помощь: handle — nonce запроса; эпохи запроса/сервиса независимы от raw ETCP-сессии. */ enum stream_relay_status { STREAM_RELAY_STARTING = 1, STREAM_RELAY_READY, STREAM_RELAY_BOUND, @@ -38,6 +42,32 @@ enum stream_reject { STREAM_REASON_LEASE, STREAM_REASON_MEMBER, STREAM_REASON_PROTOCOL, STREAM_REASON_OVERLOAD }; struct stream_path { uint64_t nodes[STREAM_PATH_MAX]; uint8_t count; }; +/* Подсказка источника для учёта кадров без сборки FEC; только диагностика, не управление воспроизведением. + * started — начала кадров в пачке; continuation исключает повторный учёт кадра на границе пачек. */ +struct stream_frame_timing { + uint64_t first_frame, last_frame, first_pts, last_pts; + uint32_t period_us, started, audio_us; + uint8_t valid, video, continuation; +}; +enum stream_telemetry_flags { STREAM_TELEMETRY_KNOWN = 1, STREAM_TELEMETRY_STALE = 2, STREAM_TELEMETRY_DOWN = 4 }; +struct stream_telemetry_row { + uint64_t node, buffered_us, sample; // Узел, длительность очереди материала и номер его измерения + uint32_t frames, age_ms; // Кадры и накопленный возраст измерения; пересылка его не обнуляет + uint8_t flags; // KNOWN, STALE, DOWN; неизвестное значение отличается от нуля +}; +struct stream_telemetry { + struct stream_telemetry_row rows[STREAM_PATH_MAX]; // Фактический stream-путь до отправителя включительно + uint64_t received_ms; // Локальное монотонное время сохранения; по сети не передаётся + uint8_t count; +}; +/* Канонические кодировки подсказки DATA и отдельного TELEMETRY; read проверяет полный размер и поля. */ +void stream_frame_timing_write(uint8_t *wire, const struct stream_frame_timing *timing); +int stream_frame_timing_read(const uint8_t *wire, size_t length, struct stream_frame_timing *timing); +size_t stream_telemetry_write(uint8_t *wire, const struct stream_telemetry *telemetry); +int stream_telemetry_read(const uint8_t *wire, size_t length, struct stream_telemetry *telemetry); +/* Копия стареет по локальным часам; сумма служит только выбору худшего реального пути. */ +void stream_telemetry_age(struct stream_telemetry *telemetry, uint64_t now_ms); +uint64_t stream_telemetry_total(const struct stream_telemetry *telemetry); struct stream_message { struct stream_descriptor descriptor; uint64_t sender_epoch, receiver_epoch, handle; diff --git a/tests/test_android_stream.c b/tests/test_android_stream.c index f574c229..22bf08f7 100644 --- a/tests/test_android_stream.c +++ b/tests/test_android_stream.c @@ -12,6 +12,7 @@ #include "../src/chat/chat_setting.h" #include "../src/stream/stream.h" #include "../src/stream/stream_member.h" +#include "../src/stream/stream_viewer.h" #include "../src/utun_instance.h" #include "../tools/chatgui-android/libutun_lite/stream_bridge.h" #include "../src/stream/stream_proto.h" @@ -104,6 +105,9 @@ int main(void) { uint8_t *error = wait_event(5, 6, &size); CHECK(error[1] == STREAM_CLOSE_SOURCE_CHANGED); u_free(error); CHECK(android_stream_open(2, group, inst->node_id, 0, 0, 0, 1, member.source.source_epoch, STREAM_MEDIA_H264) == 0); u_free(wait_event(2, 4, &size)); + struct stream_viewer_telemetry telemetry; + CHECK(android_stream_get_telemetry(2, &telemetry) == 0); + CHECK(telemetry.self == inst->node_id && !telemetry.chain.count && telemetry.names[0][0] && telemetry.sample_ms); struct android_stream_signal *signal = android_stream_signal_create(2); CHECK(signal); CHECK(android_stream_signal_wait(signal, 0) == 0); android_stream_signal_notify(signal); android_stream_signal_notify(signal); diff --git a/tests/test_stream_core.c b/tests/test_stream_core.c index 56272f35..f970bd42 100644 --- a/tests/test_stream_core.c +++ b/tests/test_stream_core.c @@ -562,6 +562,58 @@ static void test_probe_schedule(void) { DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] probe schedule: one 32-node burst, 1/s, new peers, 30s minimum and unfinished cycles"); } +/* Длительности зависят от кадров/PTS, не от размера RS; продолжения большого кадра считаются один раз. */ +static void test_buffer_telemetry(void) { + struct stream_encoder *encoder = NULL; struct stream_buffer *buffer = NULL; + CHECK(stream_encoder_create(&encoder, &descriptor, private_key) == STREAM_OK); + CHECK(stream_buffer_create(&buffer, &descriptor, stream_encoder_public_key(encoder), 64 * 1300) == STREAM_OK); + struct stream_frame_timing timings[] = { + {.valid = 1, .video = 1, .started = 1, .first_frame = 1, .last_frame = 1, + .first_pts = 1000000, .last_pts = 1000000, .period_us = 33333, .audio_us = 20000}, + {.valid = 1, .video = 1, .continuation = 1, .first_frame = 1, .last_frame = 1, + .first_pts = 1000000, .last_pts = 1000000, .period_us = 33333, .audio_us = 20000}, + {.valid = 1, .video = 1, .started = 1, .first_frame = 2, .last_frame = 2, + .first_pts = 1040000, .last_pts = 1040000, .period_us = 40000, .audio_us = 20000}, + {.valid = 1, .video = 1, .started = 1, .first_frame = 3, .last_frame = 3, + .first_pts = 1090000, .last_pts = 1090000, .period_us = 50000, .audio_us = 20000} + }; + struct stream_buffer_delay delay; + for (unsigned i = 0; i < 4; i++) { + put_batch(encoder, buffer, i, 255); + if (!i) CHECK(stream_buffer_get_delay(buffer, 0, 0, &delay) == STREAM_OK && !delay.known); + uint8_t wire[STREAM_FRAME_TIMING_SIZE]; struct stream_frame_timing timing; + stream_frame_timing_write(wire, &timings[i]); + CHECK(stream_frame_timing_read(wire, sizeof(wire), &timing) == STREAM_OK); + CHECK(stream_buffer_set_timing(buffer, i, &timing) == STREAM_OK); + wire[45] = 1; CHECK(stream_frame_timing_read(wire, sizeof(wire), &timing) == STREAM_ERR_FORMAT); + } + CHECK(stream_buffer_get_delay(buffer, 0, 0, &delay) == STREAM_OK && delay.known && delay.frames == 3); + CHECK(delay.video_us == 140000 && delay.audio_us == 80000); + CHECK(stream_buffer_get_delay(buffer, 1, 0, &delay) == STREAM_OK && delay.frames == 3 && delay.video_us == 140000); + CHECK(stream_buffer_get_delay(buffer, 2, 0, &delay) == STREAM_OK && delay.frames == 2 && delay.video_us == 100000); + CHECK(stream_buffer_get_delay(buffer, 4, 0, &delay) == STREAM_OK && delay.known && !delay.frames && !delay.video_us); + struct stream_telemetry t = {.count = 2, .received_ms = 100, + .rows = {{.node = 1, .buffered_us = 600000, .frames = 18, .sample = 4, .flags = STREAM_TELEMETRY_KNOWN}, + {.node = 2, .buffered_us = 500000, .frames = 15, .sample = 8, .flags = STREAM_TELEMETRY_KNOWN}}}; + uint8_t wire[1 + STREAM_PATH_MAX * STREAM_TELEMETRY_ROW_SIZE]; + size_t length = stream_telemetry_write(wire, &t); struct stream_telemetry parsed; + CHECK(stream_telemetry_read(wire, length, &parsed) == STREAM_OK && stream_telemetry_total(&parsed) == 1100000); + CHECK(stream_telemetry_read(wire, length - 1, &parsed) == STREAM_ERR_FORMAT); + parsed.received_ms = 100; stream_telemetry_age(&parsed, 499); + CHECK(parsed.rows[0].age_ms == 399 && !(parsed.rows[0].flags & STREAM_TELEMETRY_STALE)); + CHECK(stream_buffer_set_telemetry(buffer, 0, &parsed) == STREAM_OK); + CHECK(stream_buffer_get_telemetry(buffer, 0, 500, &parsed) == STREAM_OK); + CHECK(parsed.rows[0].age_ms == 400 && (parsed.rows[0].flags & STREAM_TELEMETRY_STALE)); + CHECK(stream_telemetry_total(&parsed) == 1100000); + stream_buffer_telemetry_down(buffer, 0, 2); + CHECK(stream_buffer_get_telemetry(buffer, 0, 600, &parsed) == STREAM_OK && (parsed.rows[1].flags & STREAM_TELEMETRY_DOWN)); + CHECK(parsed.rows[1].buffered_us == 500000 && !(parsed.rows[0].flags & STREAM_TELEMETRY_DOWN)); + t.rows[1].node = 1; length = stream_telemetry_write(wire, &t); + CHECK(stream_telemetry_read(wire, length, &parsed) == STREAM_ERR_FORMAT); + stream_buffer_destroy(buffer); stream_encoder_destroy(encoder); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] frame buffers: FEC/continuation dedup, variable PTS, cursor, stale 400ms and retained DOWN values"); +} + int main(void) { debug_config_init(); debug_set_level(DEBUG_LEVEL_ERROR); @@ -579,6 +631,7 @@ int main(void) { test_variable_batches(); CHECK(u_get_allocated_count() == allocations); test_rejected_batches(); CHECK(u_get_allocated_count() == allocations); test_ring_and_conflicts(); CHECK(u_get_allocated_count() == allocations); + test_buffer_telemetry(); CHECK(u_get_allocated_count() == allocations); test_relative_metrics(); CHECK(u_get_allocated_count() == allocations); test_source_timer(); CHECK(u_get_allocated_count() == allocations); DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] stream core: all checks, no tracked allocations remain"); diff --git a/tests/test_stream_service.c b/tests/test_stream_service.c index f54b9084..14002683 100644 --- a/tests/test_stream_service.c +++ b/tests/test_stream_service.c @@ -71,7 +71,7 @@ static unsigned parallel_data; static uint64_t network_next[4] = {0, 1, 1, 0}; static uint8_t first_request[STREAM_WIRE_HEADER + 8]; static etcp_recv_fn saved_raw[4], saved_route; -static int drop_renew, gap_batch; +static int drop_renew, gap_batch, drop_telemetry; static unsigned pressure_data, overload_seen, selects_seen; static int ignore_select; static uint8_t pressure_payload[PRESSURE_INPUT_CAPACITY], stale_select[9 + STREAM_OVERLOAD_MAX * 8 + STREAM_WIRE_HEADER]; @@ -239,6 +239,9 @@ static void network_raw(struct ETCP_CONN *conn, struct ll_entry *entry) { unsigned node = 0; while (node < 4 && nodes[node] != conn->instance) node++; CHECK(node < 4); if (entry->len >= STREAM_WIRE_HEADER) { uint8_t command = entry->dgram[2]; + if (node == 1 && command == STREAM_CMD_TELEMETRY && drop_telemetry) { + queue_dgram_free(entry); queue_entry_free(entry); return; + } if (node == 3 && command == STREAM_CMD_SUBSCRIBE && entry->dgram[3] == 0 && !first_request[0]) { CHECK(entry->len == sizeof(first_request)); memcpy(first_request, entry->dgram, sizeof(first_request)); } @@ -248,7 +251,7 @@ static void network_raw(struct ETCP_CONN *conn, struct ll_entry *entry) { struct stream_message message; struct stream_path path; size_t used; CHECK(stream_wire_parse(entry->dgram, entry->len, &message) == STREAM_OK); CHECK(stream_path_parse(message.payload, message.length, &path, &used) == STREAM_OK); - if (stream_read_be(message.payload + used + STREAM_BLOCK_SEQ_OFF, 8) == 2) { + if (stream_read_be(message.payload + used + STREAM_FRAME_TIMING_SIZE + STREAM_BLOCK_SEQ_OFF, 8) == 2) { entry->dgram[2] = STREAM_CMD_GAP; entry->len = STREAM_WIRE_HEADER + 16; stream_write_be(entry->dgram + STREAM_WIRE_HEADER, 2, 8); stream_write_be(entry->dgram + STREAM_WIRE_HEADER + 8, 3, 8); @@ -700,6 +703,21 @@ static void test_network(void) { struct stream_local retained; CHECK(stream_local_get(nodes[2], &origin.descriptor, &retained) == STREAM_OK); struct stream_path path; CHECK(stream_buffer_path(retained.buffer, 1, 0, &path) == STREAM_OK); CHECK(path.count == 3 && path.nodes[0] == nodes[3]->node_id && path.nodes[1] == nodes[1]->node_id && path.nodes[2] == nodes[2]->node_id); + struct stream_telemetry telemetry; + deadline = get_time_tb() + 20000; + do { + test_clock_poll(ua, 100); + CHECK(stream_network_telemetry(nodes[2], &origin.descriptor, &telemetry) == STREAM_OK); + } while ((telemetry.count != 2 || !telemetry.rows[0].sample || !telemetry.rows[1].sample) && get_time_tb() < deadline); + CHECK(telemetry.count == 2 && telemetry.rows[0].node == nodes[3]->node_id && telemetry.rows[1].node == nodes[1]->node_id); + CHECK(telemetry.rows[0].sample && telemetry.rows[1].sample); + drop_telemetry = 1; test_clock_run_for(ua, 7000); + CHECK(stream_network_telemetry(nodes[2], &origin.descriptor, &telemetry) == STREAM_OK); + CHECK((telemetry.rows[0].flags & STREAM_TELEMETRY_STALE) && !(telemetry.rows[1].flags & STREAM_TELEMETRY_STALE)); + drop_telemetry = 0; test_clock_run_for(ua, 5000); + CHECK(stream_network_telemetry(nodes[2], &origin.descriptor, &telemetry) == STREAM_OK); + CHECK(!(telemetry.rows[0].flags & STREAM_TELEMETRY_STALE) && !(telemetry.rows[1].flags & STREAM_TELEMETRY_STALE)); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] telemetry through relay: 200ms updates, stale upstream after 400ms, fresh local and recovery"); /* Замкнуть сохранённую цепочку обратно в node1 нельзя, даже при актуальной Merkle-подписке. */ struct STREAM_RECEIVE *loop = NULL; CHECK(stream_receive_open(subscribers[1], nodes[2]->node_id, 0, 1, &loop) == STREAM_OK); diff --git a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/NativeLib.kt b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/NativeLib.kt index 734885fa..5c1cec5d 100644 --- a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/NativeLib.kt +++ b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/NativeLib.kt @@ -218,6 +218,7 @@ object NativeLib { fun streamPull(id: Long, positionUs: Long = -1, video: Int = 1, audioAheadUs: Long = 80000, records: Int = 1): ByteArray? = nativeStreamPull(id, positionUs, video, audioAheadUs, records) fun streamStatus(id: Long): LongArray? = nativeStreamStatus(id) + fun streamTelemetry(id: Long): ByteArray? = nativeStreamTelemetry(id) external fun videoLog(trace: Boolean, message: String) fun streamResync(id: Long): Boolean = nativeStreamResync(id) external fun streamSignalCreate(id: Long): Long @@ -252,6 +253,7 @@ object NativeLib { private external fun nativeStreamStop(id: Long) private external fun nativeStreamPull(id: Long, position: Long, video: Int, audioAhead: Long, records: Int): ByteArray? private external fun nativeStreamStatus(id: Long): LongArray? + private external fun nativeStreamTelemetry(id: Long): ByteArray? private external fun nativeStreamResync(id: Long): Boolean private external fun nativeStreamPlaybackCreate(id: Long, pts: Long, audio: Boolean): Long private external fun nativeStreamPlaybackFeed(handle: Long, pts: Long, packet: ByteArray): Int diff --git a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamPlaybackTelemetry.kt b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamPlaybackTelemetry.kt new file mode 100644 index 00000000..e163458c --- /dev/null +++ b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamPlaybackTelemetry.kt @@ -0,0 +1,35 @@ +package com.utun.chat.data + +import org.json.JSONObject +import java.util.Locale + +/** Входящая цепочка заканчивается собственным playback-буфером; все значения относятся к одному видео. */ +data class StreamBufferNode(val name: String, val durationUs: Long?, val frames: Long = 0, + val stale: Boolean = false, val down: Boolean = false) { + val shortName: String get() = if (name.codePointCount(0, name.length) <= 8) name + else name.substring(0, name.offsetByCodePoints(0, 8)) + "…" + val time: String get() = durationUs?.let { String.format(Locale.ROOT, "%.1f", it / 1_000_000.0).removeSuffix(".0") + "S" } ?: "?" +} + +data class StreamPlaybackTelemetry(val nodes: List = emptyList(), val tempoMilli: Int = 1000) { + val speed: String? get() = if (tempoMilli <= 1000) null + else String.format(Locale.ROOT, "%.2f", tempoMilli / 1000.0).removeSuffix("0") + "x" + + companion object { + fun fromNative(data: ByteArray, reserveUs: Long?, tempoMilli: Int): StreamPlaybackTelemetry { + val json = JSONObject(data.toString(Charsets.UTF_8)) + val array = json.getJSONArray("nodes") + require(array.length() <= 32) + val nodes = List(array.length()) { index -> + val row = array.getJSONObject(index) + val flags = row.getInt("flags") + val duration = row.getLong("us") + val frames = row.getLong("frames") + require(flags in 0..7 && duration >= 0 && frames >= 0) + StreamBufferNode(row.getString("name"), duration.takeIf { flags and 1 != 0 }, frames, + stale = flags and 2 != 0, down = flags and 4 != 0) + } + StreamBufferNode(json.getString("self"), reserveUs) + return StreamPlaybackTelemetry(nodes, tempoMilli) + } + } +} diff --git a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamPlayerEngine.kt b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamPlayerEngine.kt index 0c6bf76b..94764275 100644 --- a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamPlayerEngine.kt +++ b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamPlayerEngine.kt @@ -42,6 +42,8 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv val aspect: StateFlow = mutableAspect private val mutableStatus = MutableStateFlow(R.string.stream_connecting) val status: StateFlow = mutableStatus + private val mutableTelemetry = MutableStateFlow(StreamPlaybackTelemetry()) + val telemetry: StateFlow = mutableTelemetry private val id = StreamController.nextId() private var requestedAtUs = 0L @Volatile private var running = false @@ -100,6 +102,8 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv var videoRevision = 0L var position = -1L var ready = -1L + var readyTogether = -1L + var telemetryAtUs = 0L var needsAudio = true var audioInputs = 0L var lastAudioPts = -1L @@ -117,6 +121,7 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv fun observeStatus(meta: LongArray?) { if (meta == null || meta[5] and 1 == 0L) return ready = if (meta[5] and 2 != 0L && meta[5] and 16 == 0L) meta[2] else maxOf(meta[1], meta[2]) + readyTogether = if (meta[5] and 16 != 0L) meta[1] else meta[0] if (playback != null) return audioEnabled = meta[5] and 2 != 0L playback = StreamMixedPlayback(id, meta[3], audioEnabled) { NativeLib.streamSignalNotify(signal) } @@ -134,7 +139,8 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv fun resetTimeline(request: Boolean) { try { playback?.close() } catch (e: Exception) { LogManager.addLog("WARN", "stream", "audio reset id=$id: ${e.message}") } - playback = null; position = -1; generation = 0; needsAudio = true; clockState = null; playbackStarted = false + playback = null; position = -1; readyTogether = -1; generation = 0; needsAudio = true; clockState = null; playbackStarted = false + mutableTelemetry.value = StreamPlaybackTelemetry() underrunBase = 0; silenceBase = 0 records.clear(); ended = false; videoInputEnded = false; audioInputEnded = false; endUntil = 0 resetVideo() @@ -240,6 +246,16 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv if (System.nanoTime() >= endUntil) { mutableStatus.value = R.string.stream_ended; break } } val reportNow = System.nanoTime() / 1000 + if (reportNow - telemetryAtUs >= 200_000) { + telemetryAtUs = reportNow + val reserve = if (readyTogether >= 0 && position >= 0) (readyTogether - position).coerceAtLeast(0) else null + val tempo = if (playbackStarted && clockState?.get(4) == 0L) clockState?.get(3)?.toInt() ?: 1000 else 1000 + try { + NativeLib.streamTelemetry(id)?.let { mutableTelemetry.value = StreamPlaybackTelemetry.fromNative(it, reserve, tempo) } + } catch (e: Exception) { + LogManager.addLog("WARN", "stream", "viewer telemetry invalid id=$id: ${e.message}") + } + } if (reportNow - reported >= 500_000) { fun ms(us: Long?, valid: Boolean = true) = if (us != null && valid) String.format(Locale.ROOT, "%.1f", us / 1000.0) else "na" @@ -290,6 +306,7 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv try { playback?.close() } catch (e: Exception) { LogManager.addLog("WARN", "stream", "audio release id=$id: ${e.message}") } NativeLib.streamSignalClose(signal) + mutableTelemetry.value = mutableTelemetry.value.copy(tempoMilli = 1000) running = false LogManager.addLog("INFO", "stream", "viewer released id=$id video=${video?.state}") } diff --git a/tools/chatgui-android/app/src/main/java/com/utun/chat/ui/screens/StreamGridScreen.kt b/tools/chatgui-android/app/src/main/java/com/utun/chat/ui/screens/StreamGridScreen.kt index b3d63a78..8151605f 100644 --- a/tools/chatgui-android/app/src/main/java/com/utun/chat/ui/screens/StreamGridScreen.kt +++ b/tools/chatgui-android/app/src/main/java/com/utun/chat/ui/screens/StreamGridScreen.kt @@ -34,6 +34,7 @@ import com.utun.chat.data.LogManager import com.utun.chat.data.StreamController import com.utun.chat.data.StreamPhase import com.utun.chat.data.StreamPlayerEngine +import com.utun.chat.data.StreamPlaybackTelemetry import com.utun.chat.data.StreamSource import com.utun.chat.data.streamGridShape import kotlinx.coroutines.delay @@ -159,6 +160,7 @@ internal fun StreamGridScreen(group: Long, title: String, sources: List Unit) { var aspect by remember(tile.source.id, tile.source.epoch) { mutableStateOf(16f / 9) } + var telemetry by remember(tile.source.id, tile.source.epoch) { mutableStateOf(StreamPlaybackTelemetry()) } BoxWithConstraints(Modifier.fillMaxSize().padding(if (expanded) 0.dp else 2.dp).clipToBounds() .background(if (expanded) Color.Black else Color(0xff151515)) .then(if (covered) Modifier.clearAndSetSemantics {} else Modifier)) { @@ -172,7 +174,7 @@ private fun StreamGridTile(group: Long, tile: GridStream, visible: Boolean, expa onStatus = { tile.status = it if (it == R.string.stream_live) { tile.failures = 0; tile.retryAt = null } - }, modifier = video, onEngine = { tile.engine = it }) + }, modifier = video, onEngine = { tile.engine = it }, onTelemetry = { telemetry = it }) } } if (!expanded || tile.status != R.string.stream_live) { @@ -188,6 +190,8 @@ private fun StreamGridTile(group: Long, tile: GridStream, visible: Boolean, expa role = Role.Button, onClickLabel = stringResource(if (tile.source.self) R.string.stream_manage else if (expanded) R.string.collapse else R.string.expand), onClick = onClick) .semantics { contentDescription = tile.source.displayName }) + if (!tile.source.self) StreamPlaybackIndicator(telemetry, Modifier.align(Alignment.BottomStart).fillMaxWidth() + .then(if (expanded) Modifier.windowInsetsPadding(WindowInsets.safeDrawing) else Modifier)) if (expanded) Box(Modifier.align(Alignment.TopEnd).windowInsetsPadding(WindowInsets.safeDrawing)) { MediaAudioDeviceButton(tint = Color.White) } diff --git a/tools/chatgui-android/app/src/main/java/com/utun/chat/ui/screens/StreamPlaybackIndicator.kt b/tools/chatgui-android/app/src/main/java/com/utun/chat/ui/screens/StreamPlaybackIndicator.kt new file mode 100644 index 00000000..ddc55182 --- /dev/null +++ b/tools/chatgui-android/app/src/main/java/com/utun/chat/ui/screens/StreamPlaybackIndicator.kt @@ -0,0 +1,33 @@ +package com.utun.chat.ui.screens + +import androidx.compose.foundation.background +import androidx.compose.foundation.horizontalScroll +import androidx.compose.foundation.layout.* +import androidx.compose.foundation.rememberScrollState +import androidx.compose.material3.MaterialTheme +import androidx.compose.material3.Text +import androidx.compose.runtime.Composable +import androidx.compose.ui.Alignment +import androidx.compose.ui.Modifier +import androidx.compose.ui.graphics.Color +import androidx.compose.ui.unit.dp +import com.utun.chat.data.StreamPlaybackTelemetry + +/** Две строки с общими колонками: время над именем, ускорение только у локального узла справа. */ +@Composable +internal fun StreamPlaybackIndicator(state: StreamPlaybackTelemetry, modifier: Modifier = Modifier) { + if (state.nodes.isEmpty()) return + Row(modifier.background(Color.Black.copy(alpha = 0.55f)).horizontalScroll(rememberScrollState()).padding(4.dp), + verticalAlignment = Alignment.Top) { + state.nodes.forEachIndexed { index, node -> + if (index != 0) Text(" → ", color = Color.LightGray, style = MaterialTheme.typography.labelSmall) + val color = if (node.down) Color(0xffc88888) else if (node.stale) Color.Gray else Color.LightGray + Column(horizontalAlignment = Alignment.CenterHorizontally) { + val mark = if (node.down) " ×" else if (node.stale && node.durationUs != null) " ?" else "" + Text(node.time + mark, color = color, style = MaterialTheme.typography.labelSmall) + Text(node.shortName, color = color, style = MaterialTheme.typography.labelSmall, maxLines = 1) + } + } + state.speed?.let { Text(" $it", color = Color.LightGray, style = MaterialTheme.typography.labelSmall) } + } +} diff --git a/tools/chatgui-android/app/src/main/java/com/utun/chat/ui/screens/StreamViewerScreen.kt b/tools/chatgui-android/app/src/main/java/com/utun/chat/ui/screens/StreamViewerScreen.kt index 38dfaf17..027e50df 100644 --- a/tools/chatgui-android/app/src/main/java/com/utun/chat/ui/screens/StreamViewerScreen.kt +++ b/tools/chatgui-android/app/src/main/java/com/utun/chat/ui/screens/StreamViewerScreen.kt @@ -41,6 +41,7 @@ import androidx.lifecycle.LifecycleEventObserver import com.utun.chat.R import com.utun.chat.data.LogManager import com.utun.chat.data.StreamPlayerEngine +import com.utun.chat.data.StreamPlaybackTelemetry import kotlinx.coroutines.delay /** Полноэкранный viewer: видео разворачивается вдоль окна, ориентацию экрана не меняем. */ @@ -51,6 +52,7 @@ internal fun StreamViewerScreen(group: Long, source: Long, epoch: Long, title: S var visible by remember(owner) { mutableStateOf(owner.lifecycle.currentState.isAtLeast(Lifecycle.State.STARTED)) } var aspect by remember { mutableStateOf(0f) } var status by remember { mutableStateOf(R.string.stream_connecting) } + var telemetry by remember { mutableStateOf(StreamPlaybackTelemetry()) } var controls by remember { mutableStateOf(true) } var hideTick by remember { mutableStateOf(0) } val zoom = remember { VideoZoomState() } @@ -99,7 +101,7 @@ internal fun StreamViewerScreen(group: Long, source: Long, epoch: Long, title: S rotationZ = if (turn) 90f else 0f scaleX = zoom.scale; scaleY = zoom.scale translationX = zoom.offset.x; translationY = zoom.offset.y - }) + }, onTelemetry = { telemetry = it }) Box(Modifier.matchParentSize() .pointerInput(status) { detectTapGestures { if (status == R.string.stream_live) controls = !controls else controls = true } @@ -110,6 +112,8 @@ internal fun StreamViewerScreen(group: Long, source: Long, epoch: Long, title: S zoom.transform(centroid, pan, factor, videoSize, viewport) } }) + StreamPlaybackIndicator(telemetry, Modifier.align(Alignment.BottomStart).fillMaxWidth() + .windowInsetsPadding(WindowInsets.safeDrawing)) if (controls) { Row(Modifier.fillMaxWidth().align(Alignment.TopCenter).background(Color.Black.copy(alpha = 0.325f)) .windowInsetsPadding(WindowInsets.safeDrawing).padding(start = 16.dp, end = 4.dp, top = 4.dp, bottom = 4.dp), @@ -156,10 +160,12 @@ internal fun StreamFullscreenWindow(source: Long, epoch: Long) { /** Смена размеров окна и скрытие controls сохраняют TextureView, decoder и сетевой reader. */ @Composable internal fun StreamViewerPlayback(group: Long, source: Long, epoch: Long, onAspect: (Float) -> Unit, onStatus: (Int) -> Unit, - modifier: Modifier, onEngine: (StreamPlayerEngine) -> Unit = {}) { + modifier: Modifier, onEngine: (StreamPlayerEngine) -> Unit = {}, + onTelemetry: (StreamPlaybackTelemetry) -> Unit = {}) { var engine by remember(group, source, epoch) { mutableStateOf(null) } val aspect = engine?.aspect?.collectAsState()?.value ?: 0f val status = engine?.status?.collectAsState()?.value ?: R.string.stream_connecting + val telemetry = engine?.telemetry?.collectAsState()?.value ?: StreamPlaybackTelemetry() val resources = remember(group, source, epoch) { StreamViewerSurface() } fun releaseSurface(texture: SurfaceTexture? = null) { val surface = resources.surface ?: return @@ -173,6 +179,7 @@ internal fun StreamViewerPlayback(group: Long, source: Long, epoch: Long, onAspe else { surface.release(); currentTexture.release() } } LaunchedEffect(aspect, status) { onAspect(aspect); onStatus(status) } + LaunchedEffect(telemetry) { onTelemetry(telemetry) } DisposableEffect(resources) { onDispose { resources.disposed = true; releaseSurface() } } AndroidView(factory = { context -> TextureView(context).apply { diff --git a/tools/chatgui-android/app/src/test/java/com/utun/chat/data/StreamPlaybackTelemetryTest.kt b/tools/chatgui-android/app/src/test/java/com/utun/chat/data/StreamPlaybackTelemetryTest.kt new file mode 100644 index 00000000..63270d2d --- /dev/null +++ b/tools/chatgui-android/app/src/test/java/com/utun/chat/data/StreamPlaybackTelemetryTest.kt @@ -0,0 +1,22 @@ +package com.utun.chat.data + +import org.junit.Assert.* +import org.junit.Test + +class StreamPlaybackTelemetryTest { + @Test fun nodeNamesUseEightWholeUnicodeCharactersAndOneEllipsis() { + assertEquals("Источник", StreamBufferNode("Источник", 600000).shortName) + assertEquals("abcdefgh…", StreamBufferNode("abcdefghijk", 600000).shortName) + assertEquals("😀abcdefg…", StreamBufferNode("😀abcdefgh", 600000).shortName) + } + + @Test fun unknownIsDifferentFromZeroAndUnavailableKeepsTheLastTime() { + assertEquals("?", StreamBufferNode("node", null).time) + assertEquals("0S", StreamBufferNode("node", 0).time) + assertEquals("0.6S", StreamBufferNode("node", 600000, stale = true, down = true).time) + assertEquals("2S", StreamBufferNode("self", 2000000).time) + assertNull(StreamPlaybackTelemetry(tempoMilli = 1000).speed) + assertEquals("1.1x", StreamPlaybackTelemetry(tempoMilli = 1100).speed) + assertEquals("1.12x", StreamPlaybackTelemetry(tempoMilli = 1120).speed) + } +} diff --git a/tools/chatgui-android/jni_bridge/android_stream_jni.c b/tools/chatgui-android/jni_bridge/android_stream_jni.c index 56810cb1..ffe88806 100644 --- a/tools/chatgui-android/jni_bridge/android_stream_jni.c +++ b/tools/chatgui-android/jni_bridge/android_stream_jni.c @@ -4,15 +4,18 @@ #include #include #include +#include #include #include "../../../lib/debug_config.h" #include "../../../lib/mem.h" +#include "../../../lib/strbuf.h" #include "../../../lib/opus_codec.h" #include "../../../lib/audio_compressor.h" #include "../libutun_lite/stream_bridge.h" #include "../../../src/stream/stream_media.h" #include "../../../src/stream/stream_media_buffer.h" +#include "../../../src/stream/stream_viewer.h" #include "../../../src/stream/stream_playback_clock.h" #include "../../../src/stream/stream_playback_audio.h" @@ -123,6 +126,31 @@ JNIEXPORT jlongArray JNICALL Java_com_utun_chat_data_NativeLib_nativeStreamStatu return result; } +/* UTF-8 передаётся байтами: имена с supplementary Unicode не являются modified UTF-8 JNI. */ +JNIEXPORT jbyteArray JNICALL Java_com_utun_chat_data_NativeLib_nativeStreamTelemetry(JNIEnv *env, jobject obj, jlong id) { + (void)obj; + struct stream_viewer_telemetry t; + if (android_stream_get_telemetry((uint64_t)id, &t) < 0) return NULL; + struct strbuf json = strbuf_new(); int ok = strbuf_addf(&json, "{\"self\":") >= 0; + if (ok) ok = strbuf_add_json_string(&json, t.names[t.chain.count]) >= 0; + if (ok) ok = strbuf_addf(&json, ",\"nodes\":[") >= 0; + for (unsigned i = 0; ok && i < t.chain.count; i++) { + const struct stream_telemetry_row *r = &t.chain.rows[i]; + ok = strbuf_addf(&json, "%s{\"name\":", i ? "," : "") >= 0 + && strbuf_add_json_string(&json, t.names[i]) >= 0 + && strbuf_addf(&json, ",\"us\":%llu,\"frames\":%u,\"flags\":%u,\"age\":%u}", + (unsigned long long)r->buffered_us, r->frames, r->flags, r->age_ms) >= 0; + } + if (ok) ok = strbuf_addf(&json, "]}") >= 0; + char *data = ok ? strbuf_detach(&json) : NULL; strbuf_free(&json); + if (!data) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "JNI stream telemetry JSON allocation failed"); return NULL; } + size_t length = strlen(data); + jbyteArray result = (*env)->NewByteArray(env, (jsize)length); + if (result) (*env)->SetByteArrayRegion(env, result, 0, (jsize)length, (const jbyte *)data); + else DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "JNI stream telemetry array allocation failed bytes=%zu", length); + u_free(data); return result; +} + JNIEXPORT jboolean JNICALL Java_com_utun_chat_data_NativeLib_nativeStreamResync(JNIEnv *env, jobject obj, jlong id) { (void)env; (void)obj; return android_stream_resync((uint64_t)id) == 0; } diff --git a/tools/chatgui-android/libutun_lite/stream_bridge.c b/tools/chatgui-android/libutun_lite/stream_bridge.c index 7cf7389b..ea3be854 100644 --- a/tools/chatgui-android/libutun_lite/stream_bridge.c +++ b/tools/chatgui-android/libutun_lite/stream_bridge.c @@ -203,6 +203,13 @@ int android_stream_get_status(uint64_t id, struct stream_media_status *status) { pthread_mutex_unlock(&media_lock); return result; } +int android_stream_get_telemetry(uint64_t id, struct stream_viewer_telemetry *telemetry) { + if (!telemetry) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "Android stream telemetry missing output"); return -1; } + pthread_mutex_lock(&media_lock); + int result = stream_viewer_get_telemetry(viewers, id, telemetry); + pthread_mutex_unlock(&media_lock); return result; +} + int android_stream_resync(uint64_t id) { pthread_mutex_lock(&media_lock); int result = stream_viewer_resync(viewers, id); diff --git a/tools/chatgui-android/libutun_lite/stream_bridge.h b/tools/chatgui-android/libutun_lite/stream_bridge.h index 886790ed..b4f4c019 100644 --- a/tools/chatgui-android/libutun_lite/stream_bridge.h +++ b/tools/chatgui-android/libutun_lite/stream_bridge.h @@ -17,6 +17,7 @@ struct UTUN_INSTANCE; struct stream_media_format; struct stream_media_status; +struct stream_viewer_telemetry; struct android_stream_signal; #define ANDROID_STREAM_HEADER_SIZE 28 struct android_stream_tx_stats { @@ -45,6 +46,7 @@ uint8_t *android_stream_pull(uint64_t id, size_t *length); * Повторные запросы до получения ответа не создают дополнительные записи. Указатели кольца остаются в uasync. */ int android_stream_read_request(uint64_t id, int64_t position_us, unsigned video, int64_t audio_ahead_us, unsigned records); int android_stream_get_status(uint64_t id, struct stream_media_status *status); +int android_stream_get_telemetry(uint64_t id, struct stream_viewer_telemetry *telemetry); int android_stream_resync(uint64_t id); /* Worker владеет signal; create привязывает его к открытому viewer. Готовность/stop/detach будят wait. * signal также будит worker после расходования PCM. Уведомления сливаются, но не теряются до wait. diff --git a/tools/chatgui/src/streamplayer_window.cpp b/tools/chatgui/src/streamplayer_window.cpp index 48dd7db0..894d8650 100644 --- a/tools/chatgui/src/streamplayer_window.cpp +++ b/tools/chatgui/src/streamplayer_window.cpp @@ -6,6 +6,8 @@ #include #include #include +#include +#include #include #include #include @@ -38,6 +40,23 @@ int64_t nowUs() { } std::atomic nextId{0}; QString text(const char* value) { return QCoreApplication::translate("ChatGui", value); } + +/* Ровно восемь Unicode code points и один компактный знак многоточия; HTML имён экранируется. */ +QString shortNodeName(const char* name) { + QString value = QString::fromUtf8(name); int end = 0, count = 0; + while (end < value.size() && count < 8) { + if (value.at(end).isHighSurrogate() && end + 1 < value.size() && value.at(end + 1).isLowSurrogate()) end++; + end++; count++; + } + return (value.left(end) + (end < value.size() ? QStringLiteral("…") : QString())).toHtmlEscaped(); +} + +QString bufferTime(int64_t us) { + if (us < 0) return QStringLiteral("?"); + QString value = QString::number(us / 1000000.0, 'f', 1); + if (value.endsWith(QStringLiteral(".0"))) value.chop(2); + return value + QStringLiteral("S"); +} } struct StreamPlayerWindow::Playback { @@ -113,6 +132,12 @@ StreamPlayerWindow::StreamPlayerWindow(uint64_t group, uint64_t source, uint64_t auto* layout = new QVBoxLayout(this); auto* title = new QLabel(name, this); layout->addWidget(title); m_status = new QLabel(text("Connecting…"), this); layout->addWidget(m_status); m_render = new VideoRenderWidget(this, true); layout->addWidget(m_render, 1); + m_telemetry = new QLabel(this); m_telemetry->setObjectName(QStringLiteral("streamTelemetry")); + m_telemetry->setTextFormat(Qt::RichText); m_telemetry->setAlignment(Qt::AlignCenter); + auto* telemetryScroll = new QScrollArea(this); telemetryScroll->setFrameShape(QFrame::NoFrame); + telemetryScroll->setVerticalScrollBarPolicy(Qt::ScrollBarAlwaysOff); telemetryScroll->setAlignment(Qt::AlignCenter); + telemetryScroll->setFixedHeight(fontMetrics().height() * 2 + style()->pixelMetric(QStyle::PM_ScrollBarExtent) + 4); + telemetryScroll->setWidget(m_telemetry); layout->addWidget(telemetryScroll); auto* controls = new QHBoxLayout; auto* volume = new QSlider(Qt::Horizontal, this); volume->setRange(0, 100); volume->setValue(100); volume->setAccessibleName(text("Stream volume")); controls->addWidget(new QLabel(text("Volume"), this)); controls->addWidget(volume); @@ -220,6 +245,42 @@ void StreamPlayerWindow::tick() { finished = p.ended && p.videoEnded && p.videos.empty() && audioDone; } p.notify(); + if (nowUs() - m_telemetryAt >= STREAM_TELEMETRY_PERIOD_MS * 1000) { + m_telemetryAt = nowUs(); + stream_viewer_telemetry t{}; + if (gui_stream_telemetry(p.id, t)) { + int64_t reserve = -1; int tempo = 1000; + { + std::lock_guard guard(p.lock); + int64_t ready = p.activeAudio && !p.media.audio_ended ? p.media.ready_us : p.media.video_us; + if (p.started && ready >= 0) reserve = std::max(0, ready - p.clock(nowUs())); + if (p.presenting && !p.timing.waiting && !finished && !p.failed) tempo = p.adaptation.tempo_milli; + } + QString times, names, tooltip; + for (unsigned i = 0; i <= t.chain.count; i++) { + bool own = i == t.chain.count; + uint8_t flags = own ? STREAM_TELEMETRY_KNOWN : t.chain.rows[i].flags; + int64_t us = own ? reserve : (flags & STREAM_TELEMETRY_KNOWN) ? (int64_t)t.chain.rows[i].buffered_us : -1; + QString color = flags & STREAM_TELEMETRY_DOWN ? QStringLiteral("#c88888") + : flags & STREAM_TELEMETRY_STALE ? QStringLiteral("#888888") : m_telemetry->palette().color(QPalette::WindowText).name(); + QString value = bufferTime(us); + if (flags & STREAM_TELEMETRY_DOWN) value += QStringLiteral(" ×"); + else if ((flags & STREAM_TELEMETRY_STALE) && us >= 0) value += QStringLiteral(" ?"); + if (i) { times += QStringLiteral(" → "); names += QStringLiteral(""); tooltip += QStringLiteral(" → "); } + times += QStringLiteral("%2").arg(color, value); + names += QStringLiteral("%2").arg(color, shortNodeName(t.names[i])); + tooltip += QString::fromUtf8(t.names[i]) + QStringLiteral(": ") + bufferTime(us); + } + if (tempo > 1000) { + QString rate = QString::number(tempo / 1000.0, 'f', 2); + if (rate.endsWith(QLatin1Char('0'))) rate.chop(1); + times += QStringLiteral("  %1x").arg(rate); names += QStringLiteral(""); + } + m_telemetry->setText(QStringLiteral("%1%2
").arg(times, names)); + m_telemetry->adjustSize(); + m_telemetry->setToolTip(tooltip); + } + } if (!image.isNull()) { m_render->setFrame(image); m_status->setText(text("Live")); if (first) DEBUG_INFO(DEBUG_CATEGORY_STREAM, diff --git a/tools/chatgui/src/streamplayer_window.h b/tools/chatgui/src/streamplayer_window.h index a894d04d..8311a1ff 100644 --- a/tools/chatgui/src/streamplayer_window.h +++ b/tools/chatgui/src/streamplayer_window.h @@ -36,6 +36,8 @@ private: QTimer m_timer; VideoRenderWidget* m_render = nullptr; QLabel* m_status = nullptr; + QLabel* m_telemetry = nullptr; + int64_t m_telemetryAt = 0; void attachAudio(); void tick(); void decode(); diff --git a/tools/chatgui/tests/test_stream_view.cpp b/tools/chatgui/tests/test_stream_view.cpp index 094e9834..ace90a0b 100644 --- a/tools/chatgui/tests/test_stream_view.cpp +++ b/tools/chatgui/tests/test_stream_view.cpp @@ -715,6 +715,10 @@ private slots: sender.start(100); QTRY_VERIFY_WITH_TIMEOUT(shown.count() > 0, 5000); QTest::qWait(400); // Пройти старт декодера и физическую задержку тестового устройства. + auto* telemetry = window.findChild(QStringLiteral("streamTelemetry")); + QVERIFY(telemetry); QVERIFY(telemetry->text().contains(QStringLiteral("text().contains(QStringLiteral("S")) || telemetry->text().contains(QStringLiteral("StoolTip().isEmpty()); #ifdef TEST_NULL_AUDIO uint64_t gapsBefore = silenceGaps.load(); #endif diff --git a/tools/chatgui/transport/stream_bridge.cpp b/tools/chatgui/transport/stream_bridge.cpp index eb0327b2..18809360 100644 --- a/tools/chatgui/transport/stream_bridge.cpp +++ b/tools/chatgui/transport/stream_bridge.cpp @@ -139,6 +139,11 @@ bool gui_stream_status(uint64_t id, stream_media_status& status) { return stream_viewer_status(bridge.viewers, id, &status) == 0; } +bool gui_stream_telemetry(uint64_t id, stream_viewer_telemetry& telemetry) { + std::lock_guard guard(bridge.lock); + return stream_viewer_get_telemetry(bridge.viewers, id, &telemetry) == 0; +} + bool gui_stream_resync(uint64_t id) { std::lock_guard guard(bridge.lock); return stream_viewer_resync(bridge.viewers, id) == 0; diff --git a/tools/chatgui/transport/stream_bridge.h b/tools/chatgui/transport/stream_bridge.h index 957260e4..cb8b681a 100644 --- a/tools/chatgui/transport/stream_bridge.h +++ b/tools/chatgui/transport/stream_bridge.h @@ -11,6 +11,7 @@ #include "../../../src/stream/stream_media.h" #include "../../../src/stream/stream_media_buffer.h" #include "../../../src/stream/stream_publisher.h" +#include "../../../src/stream/stream_viewer.h" struct UTUN_INSTANCE; enum class StreamRecordKind { Video, Audio, Gap, Opened, End, Closed, VideoEnd, AudioEnd }; @@ -37,6 +38,7 @@ bool gui_stream_pull(uint64_t id, StreamRecord& record, int64_t position = -1, bool gui_stream_pull_batch(uint64_t id, std::deque& records, int64_t position, unsigned videoBudget, int64_t audioAhead); bool gui_stream_status(uint64_t id, stream_media_status& status); +bool gui_stream_telemetry(uint64_t id, stream_viewer_telemetry& telemetry); bool gui_stream_resync(uint64_t id); bool gui_stream_ready(); /* Publisher независим от viewer; lifetime ядра ограничивает общий bridge mutex. */