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. */