Improve stream recovery, delivery diagnostics and buffer indicators

Preserve essential FEC parts under pressure, coordinate stale-data gaps and reopen source subflows in parallel after transport failures.

Add per-peer network and playback diagnostics, Android and desktop buffer indicators, and regression coverage for Wi-Fi switching and congestion.
This commit is contained in:
evgeny committed 2026-10-25 01:18:02 +02:00
1 parent b7d74d8f87
commit 4a785c2598
46 files changed
+916 -134

No files matched your search

+71 -11
View File
@@ -396,21 +396,25 @@ A — поступившая нагрузка, D — передано в ETCP, X
в интервале без смены выдач g = max(0, (A-D)/elapsed_seconds).
Rate выдачи — её средний вклад в поступающую нагрузку, а не урезанная скорость выхода из перегруженного канала.
Для разгрузки нужно снять суммарный rate больше g, чтобы очередь начала убывать.
Число выдач выбирается по их rates: закрываем достаточно выдач для требуемого сокращения;
при сильном росте это сразу несколько, не фиксированная одна и не случайное постоянное число.
Число выдач оценивается по их rates, но закрывать можно только избыточные части сверх k.
Недостаточная полоса без избыточных частей требует пропуска устаревших данных, а не разрушения FEC-набора.
OVERLOAD содержит request_id, расчётное число выдач, требуемое сокращение rate и список handles с фиксированными rates.
Получатель выбирает handles с учётом FEC и альтернативных поставщиков и отправляет OVERLOAD_SELECT.
Получатель выбирает handles только при сохранении k различных работающих частей и отправляет OVERLOAD_SELECT.
Альтернатива должна уже получать DATA с действительными метриками без остановки; ожидающий INFO не является заменой.
Одновременный выбор для других перегруженных пиров тоже исключается из доступных альтернатив.
Отправитель исполняет выбор, только пока актуален исходный request_id, и сообщает CLOSED.
Получатель не отсоединяет выбранные handles до CLOSED: поздний SELECT после принудительного закрытия не режет дополнительные выдачи.
При разных rates одного количества закрытий недостаточно: суммарный снятый rate должен покрыть требование.
Получатель не отсоединяет выбранные handles до CLOSED; поздний SELECT после завершения запроса не меняет выдачи.
При разных rates снятая нагрузка считается по сумме rates; безопасность FEC важнее полного исполнения требования.
Rate закрываемых выдач фиксируется по оценке на момент запроса, чтобы условие исполнения не менялось на каждом тике.
Срок реакции — 2 секунды с первого запроса; повторы не сбрасывают срок и не начинают новое требование.
Новые выдачи, компенсирующие запрошенное сокращение, до разгрузки не принимаются.
Если требование не выполнено, отправитель сам закрывает достаточный остаток выдач и сообщает handles/причину.
Далее снова измеряется рост; автоматический выбор учитывает освобождаемую нагрузку, а при равенстве допустима случайность.
Если очередь опустела, запрос завершается сразу. По истечении 2 секунд отправитель завершает запрос без слепых закрытий.
Далее снова измеряется рост; автоматический выбор учитывает освобождаемую нагрузку.
При глобальном отставании сокращение выбирается по всем пирам модуля.
Память и возраст очереди остаются ограниченными во время ожидания; просроченные задания удаляются с GAP/EXPIRED.
При возрасте ожидающих DATA >1 секунды все выдачи одной точной эпохи одному получателю переходят на общий следующий seq.
Устаревшие задания учитываются как X; GAP сохраняет самую раннюю границу нескольких пропусков до его отправки.
Принятые ETCP пакеты, кольцо и подписки сохраняются. Завершённый источник дочитывается без такого пропуска.
13. Статусы через опции мембера и Merkle
@@ -502,7 +506,7 @@ Stop/destroy выполняются по правилам ядра, вне callb
- Перекрытие поставщиков без дублирования потребителю, rollback, одновременное восстановление нескольких субпотоков.
- Цикл по устаревшим метрикам, одновременная перестройка и выдача сохранённых путей: DATA не циркулирует.
- Глитч короче подтверждения, устойчивое отставание 500 мс, нулевой выход и локальная перегрузка uasync.
- Выбор числа закрытий по g/rates, разные rates, повтор OVERLOAD, частичное выполнение и закрытие остатка через 2 секунды.
- Выбор безопасных закрытий по g/rates, разные rates, повтор OVERLOAD, частичное выполнение и сохранение k после 2 секунд.
- 32 совместимых кандидата на старте, далее 1/с, маленький/большой список и запрет повторного полного цикла раньше 30 с.
- Merkle-опции до/после BGP-события, офлайн-опция не кандидат, поздний участник, перезапуск и сохранение других полей userinfo.
- Мобильный источник/получатель и исключение мобильных из фоновых проб ретрансляторов.
@@ -592,13 +596,14 @@ Stop/destroy выполняются по правилам ядра, вне callb
Rates фиксируются при создании OVERLOAD. Расчётное число — подсказка, исполнение проверяется по суммарному rate.
Получатель предпочитает закрытия, сохраняющие k с учётом других поставщиков, затем parity при равных условиях.
OVERLOAD_SELECT предлагает handles; source закрывает только актуальное требование, поздний/повторный выбор безопасен.
Срок 2с начинается с первого запроса и не меняется при повторе. Невыполненный остаток source закрывает сам.
Срок 2с начинается с первого запроса и не меняется при повторе. С 25.10 остаток без безопасного SELECT не закрывается.
Новые SUBSCRIBE не компенсируют разгрузку: запрет действует до исполнения и 500мс снятого давления;
stream_peer_get_load.accepting сообщает этот статус отдельно от request_active.
Глобальный бюджет включается только для backlog нескольких пиров вместе с CPU/опозданием callbacks;
CPU отдельно не отключает выдачи, уже запрошенное сокращение повторно не учитывается.
Control FIFO ограничена 128 кодограммами по 1094 байта, возрастом 4с и отменой неподтверждённых запросов.
Просроченные ожидающие DATA при активном OVERLOAD учитываются X и пропускаются явным GAP, кольцо не стирается.
С 25.10 DATA старше 1с пропускаются согласованно для всех частей одной эпохи одному получателю, независимо от OVERLOAD.
Пропуск учитывается X и передаётся явным GAP, кольцо не стирается; после END ожидающие DATA дочитываются.
Диагностика сообщает A/D/X/Q, age, input/output/g, длительность интервала, cpu/late callbacks,
frozen/released rates, request/deadline и нижние очереди/inflight; D остаётся принятием ETCP, а не доставкой.
test_stream_core: баланс, X отдельно от D, граница 500мс, глитч, смена состава, 10-секундное окно и широкие wire-метрики.
@@ -1293,3 +1298,58 @@ callback обнаружения поставщика без секундной
раздачу нескольким зрителям, общий owner с локальным viewer, остановку при уходе последнего,
поздний повтор запроса до/после удаления tombstone, CANCEL до REQUEST, истечение READY без SUBSCRIBE,
откат неуспешного добавления viewer, исключение relay-only кандидатов и передачу пустого подписанного END.
Диагностика доставки и индикатор буфера
[debug] video=debug включает две сводки с периодом не меньше 500 мс и фактическим dt_ms.
stream NET — одна строка на соседа для всех его трансляций и субпотоков вместе:
- TX: sent/drop/gap за интервал, remain до головы локального кольца, возраст очереди и скорость её роста.
- ETCP: очередь до нормализации/отправки, уникальные unacked DATA, физический flight и сумма окон линков,
уникальные acked_B, принятые транспортом повторы и число физических RTO за интервал; очередь пересборки RX.
Все поля ETCP относятся к общему соединению с пиром, включая другие сервисы. unacked и flight не складываются.
errors_total — суммы счётчиков send/recv/crypto существующих линков; удаление/сброс линка меняет базу этих сумм.
- PATH: рабочие UDP/TCP, stalled, заполненные окна для первого ожидающего DATA, pacing, RTT/jitter/RTO и причина ожидания.
- RX: received/gap/dup/expired за интервал, время без DATA, максимальное относительное отставание и число stopped.
worst=group/source/epoch/sub обозначает самый отставший измеренный субпоток. connecting — все ещё открывающиеся выдачи пира.
Счётчики блоков/байтов записаны как blocks/bytes. sent означает принятие ETCP, не ACK.
na — нет измерения. После смены эпохи changed=1, транспортные дельты na; следующая сводка имеет новую базу.
ack_idle_ms начинается только после реального логического ACK, не после отправки/keepalive.
Метрика lag относительна исходной базе timestamp, не является абсолютной задержкой от источника.
stream PLAY — одна строка на плеер/источник: TIME (позиция, непрерывная готовая граница, запас, минимум, guard,
темп и причина решения), VIDEO (показ/отброс за интервал, PTS показанного/переданного поверхности кадра,
av_ms=video PTS минус media clock, очередь и ожидание KEY), AUDIO (PCM, устройство, underruns/тишина за интервал).
Длительности — мс, tempo=1000 означает 1x. generation/epoch отделяют новые временные шкалы.
GAP, перегрузка, смена поставщика и restart выводятся отдельными событиями. Подробности кодеков/кадров/GUI — video=trace.
Kotlin videoDebug/videoTrace используют native-фильтр и общий log hook, включая файловый/UDP вывод.
Индикатор публикации обновляется раз в 500 мс независимо от уровня логирования:
Android — 96x48dp слева от Start/Stop в шапке, desktop — рядом с этой кнопкой над preview.
Горизонталь [first, head] показывает сохранённое окно, где head — следующий seq после последнего блока.
Красная/жёлтая/зелёная трети обозначают близость к вытеснению/середину окна/близость к голове.
Линия объединяет субпотоки одного получателя и отмечает худшую и лучшую позиции передачи в ETCP.
Показываются до 16 наиболее отставших получателей, порядок при равенстве — ID. Позиции вне окна прижимаются к краям.
Индикатор отражает чтение кольца для отправки; неподтверждённые транспортные данные учитываются отдельно в inflight.
Восстановление после смены Wi-Fi и временного застоя — 25.10.2026
test_stream_service --recovery-only запускает настоящее ядро/CHAT/ETCP и подписанные RS-пачки 8+0.
Источник создаёт uAV2 video по 6000 байт каждые 20мс, KEY раз в секунду; media reader проверяет порядок, PTS и каждый байт.
Это проверка доставки/сборки и восстановления media cursor; видеодекодер H.264 в этом тесте не запускается.
Три сценария: блокировка send_input_q на 850мс, REINIT с задержкой входящих stream-кодограмм на 100мс,
блокировка send_input_q на 2600мс. Задержка stream-кодограмм проверяет управление; она не меняет физический RTT ETCP.
После каждого сценария проверяются возобновление кадров, предел паузы, отсутствие OVERLOAD-закрытий необходимых частей,
а также отсутствие лишних restart при последующей 2-секундной паузе автора.
Длительный застой обязан привести к явному GAP без полного restart; баланс A/D/X/Q проверяется остальными pressure-тестами.
На прежнем коде короткое торможение закрывало 6 из 8 необходимых частей, а после REINIT кадры не возобновлялись за 3,5с.
Автор публикует все n частей вместе: один проверенный INFO разрешает параллельное открытие частей,
каждая DATA-выдача всё равно проходит собственное INFO/ACCEPT. Реле проверяются отдельно для каждой части.
Повтор после транспортного сбоя уже работавшей выдачи — через 250мс, без ожидания 30-секундного цикла проб.
При отсутствии прогресса 1с restart возможен только при нехватке k работающих частей и известном более новом seq;
неизвестная граница UINT64_MAX и обычная пауза автора не являются свидетельством новых данных.
Проверка 25.10: ./check.sh — 187/187 и proxy/burst/load; Qt stream_view/audio/publish — 3/3; Android assembleDebug — PASS.
Изолированный сетевой стенд с 16 узлами: baseline 10с, WAN 2Мбит/с + RTT 300мс + loss 5% на 10с,
восстановление, закрытие двух подписок и подписанный END — PASS, основная сеть не изменена.
Результат: /tmp/stream-recovery-netns/result.json; обычные проверки: /tmp/stream-recovery-check.log.
+4 -4
View File
@@ -425,7 +425,7 @@ static void session_deliver(void *arg) {
} else if (result == STREAM_REJECTED) {
event.type = STREAM_EVENT_GAP; event.first_seq = event.last_seq = s->next_seq++;
emit = again = 1;
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream rejected batch GAP source=%llx epoch=%llx seq=%llu; receiving continues",
DEBUG_WARN(DEBUG_CATEGORY_VIDEO, "stream rejected batch GAP source=%llx epoch=%llx seq=%llu; receiving continues",
(unsigned long long)s->descriptor.source_node_id, (unsigned long long)s->descriptor.source_epoch,
(unsigned long long)event.first_seq);
} else if (result == STREAM_EXPIRED) {
@@ -436,7 +436,7 @@ static void session_deliver(void *arg) {
if (stream_buffer_next_decodable(s->buffer, s->next_seq + 1, &decoded) == STREAM_OK && decoded < next) next = decoded;
event.type = STREAM_EVENT_GAP; event.first_seq = s->next_seq; event.last_seq = next - 1;
s->next_seq = next; emit = 1; again = 1;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream consumer GAP source=%llx first=%llu last=%llu next=%llu",
DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "stream consumer GAP source=%llx first=%llu last=%llu next=%llu",
(unsigned long long)s->descriptor.source_node_id, (unsigned long long)event.first_seq,
(unsigned long long)event.last_seq, (unsigned long long)next);
} else if (result < 0) {
@@ -1033,7 +1033,7 @@ int stream_ingest_path(struct UTUN_INSTANCE *inst, const uint8_t *wire, size_t l
uint64_t start = s->restarting && received < s->restart_from ? s->restart_from : received;
s->anchored = 1; s->start_seq = s->next_seq = start;
if (s->restarting && s->next_seq > s->restart_from) { s->restart_gap = 1; s->restart_last = s->next_seq - 1; }
if (s->restarting) DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream restart anchored source=%llx received=%llu floor=%llu start=%llu",
if (s->restarting) DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "stream restart anchored source=%llx received=%llu floor=%llu start=%llu",
(unsigned long long)source, (unsigned long long)received, (unsigned long long)s->restart_from, (unsigned long long)start);
s->restarting = 0;
}
@@ -1121,7 +1121,7 @@ int stream_local_restart(struct UTUN_INSTANCE *inst, const struct stream_descrip
}
s->restart_gap = 0;
s->anchored = s->decoded = s->end_delivered = 0; s->passed_before = s->next_seq = s->start_seq = 0;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream receiver restart from end group=%llx source=%llx epoch=%llx retained=%u floor=%llu known=%d",
DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "stream receiver restart from end group=%llx source=%llx epoch=%llx retained=%u floor=%llu known=%d",
(unsigned long long)key->group_id, (unsigned long long)key->source_node_id,
(unsigned long long)key->source_epoch, local.buffer != NULL, (unsigned long long)s->restart_from, s->restarting); return STREAM_OK;
}
+13
View File
@@ -34,6 +34,19 @@ struct stream_buffer_stats {
uint64_t rejected_batches;
};
#define STREAM_BUFFER_POSITION_PEERS 16
/* Одна линия индикатора — получатель; next обозначает позицию передачи в ETCP, а не подтверждённой доставки. */
struct stream_buffer_peer_position {
uint64_t peer; // ID получателя
uint64_t worst, best; // Минимальный/максимальный следующий seq его субпотоков
unsigned subflows; // Число действующих выдач получателю
};
struct stream_buffer_positions {
uint64_t first, head; // [first, head) — сохранённое окно; head — позиция после последней пачки
unsigned count, total; // Показанные/все получатели; показаны до 16 наиболее отставших
struct stream_buffer_peer_position peers[STREAM_BUFFER_POSITION_PEERS];
};
/* out заранее NULL; bytes=0 выбирает 32 МБ, меньший бюджет допускается для стенда.
* Должна помещаться хотя бы одна полная пачка n блоков; верхний предел — STREAM_BUFFER_BYTES. */
int stream_buffer_create(struct stream_buffer **out, const struct stream_descriptor *descriptor,
+1
View File
@@ -16,6 +16,7 @@ extern "C" {
#define STREAM_LOAD_CONFIRM_MS 500
#define STREAM_LOAD_GRACE_MS 2000
#define STREAM_LOAD_MAX_AGE_MS 4000
#define STREAM_LOAD_CATCHUP_MS 1000 // Старые RS-пачки пропускаются целиком, необходимые выдачи сохраняются
struct stream_load_counts { uint64_t packets, bytes; };
struct stream_load_interval {
+1 -1
View File
@@ -29,7 +29,7 @@ static void metrics_detect_stop(struct stream_metrics *m, uint64_t now_ms) {
uint64_t started = metrics_add(m->last_receive_ms, metrics_add(threshold, 1));
metrics_bucket(m, started)->stops++;
m->stopped = 1;
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "subflow stopped after seq=%llu elapsed=%llums expected=%llums lag=%llums",
DEBUG_WARN(DEBUG_CATEGORY_VIDEO, "subflow stopped after seq=%llu elapsed=%llums expected=%llums lag=%llums",
(unsigned long long)m->last_seq, (unsigned long long)elapsed,
(unsigned long long)m->expected_interval_ms, (unsigned long long)m->lag_ms);
}
+248 -29
View File
@@ -1,3 +1,4 @@
#include <stdio.h>
#include <string.h>
#include "../../lib/debug_config.h"
@@ -55,6 +56,11 @@ struct stream_pressure_request {
unsigned count, required_count, released_count;
int active, sent;
};
/* Интервальные счётчики переживают закрытие выдач; транспортные базы принадлежат одной эпохе ETCP. */
struct stream_peer_diagnostics {
uint64_t report_ms, tx_blocks, tx_bytes, rx_blocks, rx_bytes, duplicates, expired, tx_gaps, rx_gaps, dropped_blocks, dropped_bytes;
uint64_t last_rx_ms, reset_id, peer_reset_id, acked_bytes, retries, rtos;
};
struct STREAM_RECEIVE {
struct STREAM_RECEIVE *next;
struct stream_peer *peer;
@@ -83,6 +89,7 @@ struct stream_peer {
struct stream_pressure_request pressure, requested;
uint64_t pressure_id, pressure_retry_ms, clear_since_ms;
int cooling;
struct stream_peer_diagnostics diagnostic;
};
struct stream_network {
struct UTUN_INSTANCE *inst;
@@ -139,6 +146,11 @@ static uint64_t flow_age(struct STREAM_RECEIVE *f, uint64_t now) {
/* Один блок принадлежит логической очереди выдачи/пира/модуля; ни одна из них не хранит копию DATA. */
static void load_account(struct STREAM_RECEIVE *f, uint64_t packets, uint64_t bytes, int added, int sent) {
if (!packets) return;
if (!added) {
struct stream_peer_diagnostics *d = &f->peer->diagnostic;
if (sent) { d->tx_blocks += packets; d->tx_bytes += bytes; }
else { d->dropped_blocks += packets; d->dropped_bytes += bytes; }
}
struct stream_load *loads[] = {&f->load, &f->peer->load, &f->peer->network->load};
for (unsigned i = 0; i < 3; i++) {
int result = added ? stream_load_add(loads[i], packets, bytes) : stream_load_remove(loads[i], packets, bytes, sent);
@@ -155,6 +167,14 @@ static int same_stream(const struct stream_descriptor *a, const struct stream_de
return a->group_id == b->group_id && a->source_node_id == b->source_node_id && a->source_epoch == b->source_epoch;
}
static void flow_gap_report(struct STREAM_RECEIVE *f, uint64_t first, uint64_t next, const char *reason) {
if (f->outgoing) f->peer->diagnostic.rx_gaps++; else f->peer->diagnostic.tx_gaps++;
DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "stream GAP peer=%llx group=%llx source=%llx epoch=%llx sub=%u direction=%s first=%llu next=%llu reason=%s",
(unsigned long long)f->peer->node_id, (unsigned long long)f->descriptor.group_id,
(unsigned long long)f->descriptor.source_node_id, (unsigned long long)f->descriptor.source_epoch,
f->subblock, f->outgoing ? "RX" : "TX", (unsigned long long)first, (unsigned long long)next, reason);
}
/* Водяной знак и управляющая очередь принадлежат сессии пира, а не отдельной группе. */
static struct stream_peer *peer_find(struct stream_network *net, uint64_t node) {
for (struct stream_peer *p = net->peers; p; p = p->next) if (p->node_id == node) return p;
@@ -286,7 +306,7 @@ static void flow_close(struct STREAM_RECEIVE *f, int reason, int notify) {
}
f->info.state = STREAM_RECEIVE_CLOSED; f->info.reason = reason;
if (reason != STREAM_REASON_STOPPED)
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream issuance failure peer=%llx handle=%llu sub=%u reason=%d next=%llu"
DEBUG_WARN(DEBUG_CATEGORY_VIDEO, "stream issuance failure peer=%llx handle=%llu sub=%u reason=%d next=%llu"
" previous_state=%d probe_only=%d info_sent=%d info_ready=%d blocks=%llu transport_up=%d local_epoch=%llx peer_epoch=%llx",
(unsigned long long)f->info.peer, (unsigned long long)f->info.handle, f->subblock,
reason, (unsigned long long)f->info.next, previous, f->info.probe_only, f->info_sent, f->info.info_ready,
@@ -319,6 +339,7 @@ static void peer_sweep(struct stream_peer *p) {
/* Смена поколения может сохранить запросы до SUBSCRIBE; разрыв/ошибка отправки отменяют и эти запросы. */
static void peer_reset(struct stream_peer *p, int reason, unsigned flags, struct STREAM_RECEIVE *keep) {
peer_cancel_wait(p);
p->diagnostic.last_rx_ms = 0;
uint64_t now = get_time_tb(); unsigned pending = 0;
struct ll_entry *e;
while ((e = queue_data_get(p->control))) { queue_dgram_free(e); queue_entry_free(e); }
@@ -342,7 +363,7 @@ static void peer_reset(struct stream_peer *p, int reason, unsigned flags, struct
peer_sweep(p);
memset(&p->pressure, 0, sizeof(p->pressure)); memset(&p->requested, 0, sizeof(p->requested));
p->pressure_id = 0; p->cooling = 0; stream_load_init(&p->load, get_time_tb() / 10);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream peer session reset peer=%llx epoch=%llx rotate=%d reason=%d pending=%u",
DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "stream peer session reset peer=%llx epoch=%llx rotate=%d reason=%d pending=%u",
(unsigned long long)p->node_id, (unsigned long long)p->own_epoch,
!!(flags & STREAM_RESET_ROTATE), reason, pending);
}
@@ -394,6 +415,13 @@ static struct stream_peer *peer_open(struct stream_network *net, uint64_t node)
if (p->control) queue_free(p->control); u_free(p); return NULL;
}
p->next = net->peers; net->peers = p; net->peer_count++;
p->diagnostic.report_ms = p->born / 10;
struct ETCP_CONN *conn = node_conn_direct_get_conn(p->ncd);
if (conn) {
p->diagnostic.reset_id = conn->reset_id; p->diagnostic.peer_reset_id = conn->peer_reset_id;
p->diagnostic.acked_bytes = conn->diagnostic_acked_bytes;
p->diagnostic.retries = conn->diagnostic_retries; p->diagnostic.rtos = conn->diagnostic_rtos;
}
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream peer opened peer=%llx epoch=%llx", (unsigned long long)node, (unsigned long long)p->own_epoch);
return p;
}
@@ -587,6 +615,7 @@ static int downstream_send(struct STREAM_RECEIVE *f) {
uint8_t payload[16]; stream_write_be(payload, f->gap_from, 8); stream_write_be(payload + 8, f->info.next, 8);
m.command = STREAM_CMD_GAP;
if (raw_send(p, m, payload, sizeof(payload)) != STREAM_OK) { flow_close(f, STREAM_REASON_TRANSPORT, 0); return 0; }
flow_gap_report(f, f->gap_from, f->info.next, "overload");
f->gap_pending = 0; f->info.gaps++; return 1;
}
const uint8_t *wire; size_t length; uint64_t actual;
@@ -617,6 +646,7 @@ static int downstream_send(struct STREAM_RECEIVE *f) {
uint8_t payload[16]; stream_write_be(payload, f->info.next, 8); stream_write_be(payload + 8, next, 8);
m.command = STREAM_CMD_GAP;
if (raw_send(p, m, payload, sizeof(payload)) == STREAM_OK) {
flow_gap_report(f, f->info.next, next, result == STREAM_EXPIRED ? "expired" : "missing");
f->info.next = next; f->info.gaps++; return 1;
}
flow_close(f, STREAM_REASON_TRANSPORT, 0); return 0;
@@ -908,13 +938,21 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t
if (f->info.path.count != path.count || memcmp(f->info.path.nodes, path.nodes, path.count * sizeof(uint64_t)))
f->info.path_rtt_valid = 0;
f->info.next = seq + 1; f->info.blocks++; f->info.path = path; f->deadline = get_time_tb() + STREAM_LEASE_TB;
p->diagnostic.rx_blocks++; p->diagnostic.rx_bytes += m.length - used;
p->diagnostic.last_rx_ms = get_time_tb() / 10;
p->diagnostic.duplicates += result == STREAM_DUPLICATE;
p->diagnostic.expired += result == STREAM_EXPIRED;
} else flow_close(f, STREAM_REASON_PROTOCOL, 1);
}
}
} else if (m.command == STREAM_CMD_GAP && f && f->info.state == STREAM_RECEIVE_ACTIVE) {
if (m.length != 16 || stream_read_be(m.payload, 8) != f->info.next
|| stream_read_be(m.payload + 8, 8) <= f->info.next) flow_close(f, STREAM_REASON_PROTOCOL, 1);
else { f->info.next = stream_read_be(m.payload + 8, 8); f->info.gaps++; }
else {
uint64_t next = stream_read_be(m.payload + 8, 8);
flow_gap_report(f, f->info.next, next, "supplier");
f->info.next = next; f->info.gaps++;
}
} else if (m.command == STREAM_CMD_END && f && f->info.state == STREAM_RECEIVE_ACTIVE) {
uint64_t last = m.length == STREAM_END_SIZE ? stream_read_be(m.payload + 34, 8) : UINT64_MAX;
if (m.length == STREAM_END_SIZE && ((last != UINT64_MAX && f->info.next == last + 1)
@@ -988,7 +1026,7 @@ static void expire_controls(struct stream_peer *p, uint64_t now) {
struct STREAM_RECEIVE *f = flow_find(p, &m, outgoing);
if (f && m.command == STREAM_CMD_INFO_REQ && (m.length != 8 || stream_read_be(m.payload, 8) != f->probe)) f = NULL;
if (f && (outgoing || m.command == STREAM_CMD_ACCEPT)) flow_close(f, STREAM_REASON_OVERLOAD, 0);
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream control expired peer=%llx command=%u handle=%llu age=%llu",
DEBUG_WARN(DEBUG_CATEGORY_VIDEO, "stream control expired peer=%llx command=%u handle=%llu age=%llu",
(unsigned long long)p->node_id, m.command, (unsigned long long)m.handle, (unsigned long long)(now - queued_ms));
}
p->load.rejected++; p->network->load.rejected++;
@@ -996,6 +1034,125 @@ static void expire_controls(struct stream_peer *p, uint64_t now) {
}
}
/* Неизвестное время не смешивается с достоверным нулём; все числовые длительности — мс. */
static void diagnostic_value(char *text, size_t capacity, uint64_t value, int valid) {
if (valid) snprintf(text, capacity, "%llu", (unsigned long long)value);
else snprintf(text, capacity, "na");
}
/* Один снимок на peer, не на выдачу: транспорт общий также с другими сервисами.
* Очереди берём из ledger, без обхода кольца; счётчики интервала сохраняются при отключённом выводе только до следующей границы. */
static void peer_report(struct stream_peer *p, uint64_t now) {
struct stream_peer_diagnostics *d = &p->diagnostic;
struct ETCP_CONN *conn = node_conn_direct_get_conn(p->ncd);
uint64_t elapsed = d->report_ms ? now - d->report_ms : 0;
if (d->report_ms && elapsed < 500) return;
int same_epoch = conn && d->reset_id == conn->reset_id && d->peer_reset_id == conn->peer_reset_id;
uint64_t acked = same_epoch ? conn->diagnostic_acked_bytes - d->acked_bytes : 0;
uint64_t retries = same_epoch ? conn->diagnostic_retries - d->retries : 0;
uint64_t rtos = same_epoch ? conn->diagnostic_rtos - d->rtos : 0;
unsigned tx = 0, rx = 0, connecting = 0, stopped = 0;
uint64_t lag = 0, worst_group = 0, worst_source = 0, worst_epoch = 0;
unsigned worst_sub = 0;
int lag_valid = 0;
int enabled = debug_should_output(DEBUG_LEVEL_DEBUG, DEBUG_CATEGORY_VIDEO);
if (enabled && d->report_ms) {
for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next) {
if (f->info.probe_only || f->info.state >= STREAM_RECEIVE_ENDED) continue;
if (f->info.state != STREAM_RECEIVE_ACTIVE) { connecting++; continue; }
if (!f->outgoing) { tx++; continue; }
rx++;
if (!f->info.metrics_valid) continue;
const struct stream_metric_snapshot *m = &f->info.metrics;
stopped += m->stopped;
if (!lag_valid || m->lag_ms > lag) {
lag_valid = 1; lag = m->lag_ms; worst_group = f->descriptor.group_id;
worst_source = f->descriptor.source_node_id; worst_epoch = f->descriptor.source_epoch; worst_sub = f->subblock;
}
}
if (tx || rx || connecting || d->tx_blocks || d->rx_blocks || d->tx_gaps || d->rx_gaps || d->dropped_blocks) {
unsigned udp = 0, tcp = 0, stalled = 0, full = 0, paced = 0, rtt_count = 0, jitter_count = 0;
uint64_t flight = 0, window = 0, pacing = 0, rtt_min = UINT64_MAX, rtt_max = 0, jitter_max = 0, rto_max = 0;
uint64_t send_errors = 0, recv_errors = 0, crypto_errors = 0;
if (conn) for (struct ETCP_LINK *link = conn->links; link; link = link->next) {
flight += link->inflight_bytes;
send_errors += link->send_errors; recv_errors += link->recv_errors;
crypto_errors += link->encrypt_errors + link->decrypt_errors;
if (!link->initialized || !link->link_status) continue;
if (link->is_tcp) tcp++; else udp++;
stalled += link->tx_stalled;
uint32_t need = conn->input_send_q && conn->input_send_q->head ? conn->input_send_q->head->len : 0;
full += need && (need > link->inflight_lim_bytes || link->inflight_bytes > link->inflight_lim_bytes - need);
paced += !link->burst_active && link->shaper_timer != NULL;
window += link->inflight_lim_bytes; pacing += link->bbr_pacing_rate;
uint64_t rtt = link->data_rtt_us;
if (!rtt) rtt = (uint64_t)link->rtt_last * 100;
uint64_t rto = etcp_tx_rto_us(link);
if (rto > rto_max) rto_max = rto;
if (link->data_rtt_us) {
jitter_count++;
if (link->data_jitter_us > jitter_max) jitter_max = link->data_jitter_us;
}
if (rtt) {
rtt_count++;
if (rtt < rtt_min) rtt_min = rtt;
if (rtt > rtt_max) rtt_max = rtt;
}
}
char ack_idle[24], rx_idle[24], lag_text[24], rtt_text[48], jitter_text[24], rto_text[24], worst[80];
char acked_text[24], retries_text[24], rtos_text[24];
diagnostic_value(acked_text, sizeof(acked_text), acked, same_epoch);
diagnostic_value(retries_text, sizeof(retries_text), retries, same_epoch);
diagnostic_value(rtos_text, sizeof(rtos_text), rtos, same_epoch);
uint64_t now_us = get_time_us();
diagnostic_value(ack_idle, sizeof(ack_idle), conn && conn->diagnostic_ack_us
? (now_us - conn->diagnostic_ack_us) / 1000 : 0, conn && conn->diagnostic_ack_us);
diagnostic_value(rx_idle, sizeof(rx_idle), now - d->last_rx_ms, d->last_rx_ms != 0);
diagnostic_value(lag_text, sizeof(lag_text), lag, lag_valid);
if (jitter_count) snprintf(jitter_text, sizeof(jitter_text), "%.1f", jitter_max / 1000.0);
else snprintf(jitter_text, sizeof(jitter_text), "na");
diagnostic_value(rto_text, sizeof(rto_text), rto_max / 1000, udp + tcp != 0);
if (rtt_count) snprintf(rtt_text, sizeof(rtt_text), "%.1f..%.1f", rtt_min / 1000.0, rtt_max / 1000.0);
else snprintf(rtt_text, sizeof(rtt_text), "na");
if (lag_valid) snprintf(worst, sizeof(worst), "%llx/%llx/%llx/%u", (unsigned long long)worst_group,
(unsigned long long)worst_source, (unsigned long long)worst_epoch, worst_sub);
else snprintf(worst, sizeof(worst), "na");
const char *blocked = !p->up ? "down" : !conn || !conn->initialized ? "session"
: p->pressure.active ? "overload" : p->wait_q ? "etcp_queue"
: conn->tx_state == ETCP_TX_STATE_LINK_WAIT ? "link_wait" : p->load.queued.packets ? "ready" : "head";
DEBUG_DEBUG(DEBUG_CATEGORY_VIDEO,
"stream NET self=%llx peer=%llx dt_ms=%llu session=%llx/%llx changed=%d "
"TX{flows=%u sent=%llu/%llu remain=%llu/%llu oldest_ms=%llu growth_Bps=%llu drop=%llu/%llu gap=%llu} "
"ETCP{q_B=%llu unacked_B=%u flight_B=%llu window_B=%llu acked_B=%s ack_idle_ms=%s retry=%s rto=%s "
"rx_reorder=%d/%zu errors_total=%llu/%llu/%llu} "
"PATH{udp=%u tcp=%u stalled=%u full=%u paced=%u pacing_Bps=%llu rtt_ms=%s jitter_ms=%s rto_ms=%s blocked=%s} "
"RX{flows=%u connecting=%u received=%llu/%llu idle_ms=%s lag_max_ms=%s stopped=%u worst=%s gap=%llu dup=%llu expired=%llu} "
"overload=%d loop_late_ms=%llu counts=blocks/bytes errors=send/recv/crypto",
(unsigned long long)p->network->inst->node_id, (unsigned long long)p->node_id, (unsigned long long)elapsed,
(unsigned long long)(conn ? conn->reset_id : 0), (unsigned long long)(conn ? conn->peer_reset_id : 0), !same_epoch,
tx, (unsigned long long)d->tx_blocks, (unsigned long long)d->tx_bytes,
(unsigned long long)p->load.queued.packets, (unsigned long long)p->load.queued.bytes,
(unsigned long long)p->load.snapshot.oldest_age_ms, (unsigned long long)p->load.snapshot.growth_rate,
(unsigned long long)d->dropped_blocks, (unsigned long long)d->dropped_bytes,
(unsigned long long)d->tx_gaps,
(unsigned long long)(conn ? etcp_tx_queued(conn) : 0), conn ? conn->unacked_bytes : 0,
(unsigned long long)flight, (unsigned long long)window, acked_text, ack_idle,
retries_text, rtos_text, conn && conn->recv_q ? conn->recv_q->count : 0,
conn && conn->recv_q ? conn->recv_q->total_bytes : 0,
(unsigned long long)send_errors, (unsigned long long)recv_errors, (unsigned long long)crypto_errors,
udp, tcp, stalled, full, paced, (unsigned long long)pacing, rtt_text, jitter_text, rto_text, blocked,
rx, connecting, (unsigned long long)d->rx_blocks, (unsigned long long)d->rx_bytes, rx_idle, lag_text, stopped, worst,
(unsigned long long)d->rx_gaps, (unsigned long long)d->duplicates, (unsigned long long)d->expired,
p->pressure.active, (unsigned long long)p->network->callback_late_ms);
}
}
d->report_ms = now; d->reset_id = conn ? conn->reset_id : 0; d->peer_reset_id = conn ? conn->peer_reset_id : 0;
d->acked_bytes = conn ? conn->diagnostic_acked_bytes : 0;
d->retries = conn ? conn->diagnostic_retries : 0; d->rtos = conn ? conn->diagnostic_rtos : 0;
d->tx_blocks = d->tx_bytes = d->rx_blocks = d->rx_bytes = d->duplicates = d->expired = d->tx_gaps = d->rx_gaps = 0;
d->dropped_blocks = d->dropped_bytes = 0;
}
/* Один таймер обслуживает сроки/продления и ограниченные reverse-попытки; DATA не опрашивается таймером. */
static void network_tick(void *arg) {
struct stream_network *net = arg; net->timer = NULL;
@@ -1035,6 +1192,7 @@ static void network_tick(void *arg) {
p = next;
}
pressure_tick(net, now_ms);
for (p = net->peers; p; p = p->next) peer_report(p, now_ms);
stream_network_wake(net);
net->next_tick_ms = get_time_tb() / 10 + STREAM_TICK_TB / 10;
net->timer = uasync_set_timeout(net->inst->ua, STREAM_TICK_TB, net, network_tick, "stream_network");
@@ -1579,13 +1737,13 @@ static int pressure_begin(struct stream_peer *p, uint64_t required, uint64_t now
request.cuts[at] = cut; total += cut.rate;
}
if (!request.count) return 0;
if (p->pressure_id == UINT64_MAX) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream OVERLOAD id exhausted"); return 0; }
if (p->pressure_id == UINT64_MAX) { DEBUG_ERROR(DEBUG_CATEGORY_VIDEO, "stream OVERLOAD id exhausted"); return 0; }
request.required_rate = required && required < total ? required : total;
uint64_t picked = 0;
while (request.required_count < request.count && picked < request.required_rate) picked += request.cuts[request.required_count++].rate;
request.id = ++p->pressure_id; request.deadline_ms = now + STREAM_LOAD_GRACE_MS; request.active = 1;
p->pressure = request; p->pressure_retry_ms = 0; p->clear_since_ms = 0;
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream OVERLOAD begin peer=%llx id=%llu count=%u cut=%llu g=%llu A=%llu D=%llu X=%llu Q=%llu age=%llu global=%d",
DEBUG_WARN(DEBUG_CATEGORY_VIDEO, "stream OVERLOAD begin peer=%llx id=%llu count=%u cut=%llu g=%llu A=%llu D=%llu X=%llu Q=%llu age=%llu global=%d",
(unsigned long long)p->node_id, (unsigned long long)request.id, request.required_count,
(unsigned long long)request.required_rate, (unsigned long long)p->load.snapshot.growth_rate,
(unsigned long long)p->load.arrivals.bytes, (unsigned long long)p->load.sent.bytes,
@@ -1612,16 +1770,17 @@ static void pressure_send(struct stream_peer *p, uint64_t now) {
if (control_put(p, m, payload, STREAM_OVERLOAD_HEADER + p->pressure.count * 16) == STREAM_OK) p->pressure_retry_ms = now;
}
/* Оценить возможность убрать один субпоток без потери k: учитываем дубликаты и остальных поставщиков. */
/* Закрывать можно только избыток над k реально работающих частей, включая выбор других перегруженных пиров. */
static int removal_cost(struct stream_peer *p, struct STREAM_RECEIVE *candidate) {
uint8_t seen[RS_MAX_BLOCKS] = {0}; unsigned distinct = 0;
for (struct stream_peer *other = p->network->peers; other; other = other->next) {
for (struct STREAM_RECEIVE *f = other->flows; f; f = f->next) {
if (!f->outgoing || f == candidate || f->info.state != STREAM_RECEIVE_ACTIVE
|| !f->info.blocks || !f->info.metrics_valid || f->info.metrics.stopped
|| !same_stream(&f->descriptor, &candidate->descriptor)) continue;
int selected = 0;
if (other == p) for (unsigned i = 0; i < p->requested.count; i++)
if (p->requested.cuts[i].selected && p->requested.cuts[i].handle == f->info.handle) selected = 1;
if (other->requested.active) for (unsigned i = 0; i < other->requested.count; i++)
if (other->requested.cuts[i].selected && other->requested.cuts[i].handle == f->info.handle) selected = 1;
if (!selected && !seen[f->subblock]) { seen[f->subblock] = 1; distinct++; }
}
}
@@ -1629,7 +1788,7 @@ static int removal_cost(struct stream_peer *p, struct STREAM_RECEIVE *candidate)
return (distinct >= (unsigned)(candidate->descriptor.n - candidate->descriptor.r) ? 0 : 4) + (parity ? 0 : 1);
}
/* Выбор лишь предлагает source закрытия. Клиент не отсоединяет handles до CLOSED: поздний SELECT не режет лишние после fallback. */
/* Выбор предлагает только безопасные закрытия. Клиент сохраняет handles до CLOSED; поздний SELECT игнорируется отправителем. */
static void pressure_select(struct stream_peer *p, uint64_t now) {
struct stream_pressure_request *r = &p->requested;
if (!r->active || r->sent || now >= r->deadline_ms) return;
@@ -1641,6 +1800,7 @@ static void pressure_select(struct stream_peer *p, uint64_t now) {
struct STREAM_RECEIVE *f = flow_by_handle(p, r->cuts[i].handle, 1);
if (r->cuts[i].selected || !f) continue;
int cost = removal_cost(p, f);
if (cost >= 4) continue;
if (cost < best_cost || (cost == best_cost && (best < 0 || r->cuts[i].rate > r->cuts[best].rate))) {
best = (int)i; best_cost = cost;
}
@@ -1659,7 +1819,7 @@ static void pressure_select(struct stream_peer *p, uint64_t now) {
struct stream_message m = {.descriptor = first->descriptor, .command = STREAM_CMD_OVERLOAD_SELECT};
if (control_put(p, m, payload, 9 + at * 8) == STREAM_OK) {
r->sent = 1;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream OVERLOAD selected peer=%llx id=%llu handles=%u rate=%llu requested=%llu",
DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "stream OVERLOAD selected peer=%llx id=%llu handles=%u rate=%llu requested=%llu",
(unsigned long long)p->node_id, (unsigned long long)r->id, at,
(unsigned long long)reduction, (unsigned long long)r->required_rate);
}
@@ -1685,7 +1845,11 @@ static void pressure_receive(struct stream_peer *p, const struct stream_message
total += request.cuts[i].rate;
}
if (required > total) goto malformed;
p->requested = request; pressure_select(p, now); return;
p->requested = request; pressure_select(p, now);
if (!p->requested.sent) DEBUG_INFO(DEBUG_CATEGORY_VIDEO,
"stream OVERLOAD keep subscriptions peer=%llx id=%llu candidates=%u requested_rate=%llu; safe selection pending",
(unsigned long long)p->node_id, (unsigned long long)id, count, (unsigned long long)required);
return;
}
if (m->length < 9) goto malformed;
unsigned count = m->payload[8]; uint64_t id = stream_read_be(m->payload, 8);
@@ -1710,15 +1874,32 @@ malformed:
(unsigned long long)p->node_id, m->length);
}
/* Просроченные ключи удаляются X; сохранённое кольцо не стирается и GAP обязательно предшествует дальнейшим DATA. */
static void expire_pending(struct STREAM_RECEIVE *f) {
if (!f->gap_pending) f->gap_from = f->info.next;
f->info.next = f->pending_highest + 1; f->gap_pending = 1;
load_account(f, f->load.queued.packets, f->load.queued.bytes, 0, 0);
peer_schedule(f->peer);
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream pending expired peer=%llx handle=%llu first=%llu next=%llu",
(unsigned long long)f->peer->node_id, (unsigned long long)f->info.handle,
(unsigned long long)f->gap_from, (unsigned long long)f->info.next);
/* Один общий seq всех выдач этой эпохи получателю: частичная чистка оставляет невосстанавливаемые RS-пачки.
* Удаляем только не переданное в ETCP; кольцо и сами подписки сохраняются, GAP идёт перед следующими DATA. */
static void expire_pending(struct STREAM_RECEIVE *trigger) {
struct stream_peer *p = trigger->peer; struct stream_local local;
if (stream_local_get(p->network->inst, &trigger->descriptor, &local) != STREAM_OK) {
DEBUG_WARN(DEBUG_CATEGORY_VIDEO, "stream catchup missing local source=%llx", (unsigned long long)trigger->descriptor.source_node_id);
return;
}
if (local.finished) return;
struct stream_buffer_stats stats;
if (stream_buffer_get_stats(local.buffer, &stats) != STREAM_OK || !stats.blocks) return;
uint64_t next = stats.highest_seq + 1, dropped = 0, bytes = 0; unsigned flows = 0;
for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next) {
if (f->outgoing || f->info.state != STREAM_RECEIVE_ACTIVE || f->info.next >= next
|| !same_stream(&f->descriptor, &trigger->descriptor)) continue;
dropped += f->load.queued.packets; bytes += f->load.queued.bytes; flows++;
if (!f->gap_pending) f->gap_from = f->info.next;
f->info.next = next; f->gap_pending = 1;
load_account(f, f->load.queued.packets, f->load.queued.bytes, 0, 0);
}
peer_schedule(p);
DEBUG_WARN(DEBUG_CATEGORY_VIDEO, "stream catchup peer=%llx group=%llx source=%llx epoch=%llx flows=%u next=%llu dropped=%llu/%llu age_ms=%llu",
(unsigned long long)p->node_id, (unsigned long long)trigger->descriptor.group_id,
(unsigned long long)trigger->descriptor.source_node_id, (unsigned long long)trigger->descriptor.source_epoch,
flows, (unsigned long long)next, (unsigned long long)dropped, (unsigned long long)bytes,
(unsigned long long)p->load.snapshot.oldest_age_ms);
}
/* 100-мс обход не отправляет DATA и не заменяет waiter. Метрики фиксируются с фактическим периодом >=200мс. */
@@ -1771,22 +1952,23 @@ static void pressure_tick(struct stream_network *net, uint64_t now) {
}
for (struct stream_peer *p = net->peers; p; p = p->next) {
p->busy++;
if (p->pressure.active && !p->load.queued.packets) {
DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "stream OVERLOAD drained peer=%llx id=%llu; preserve subscriptions",
(unsigned long long)p->node_id, (unsigned long long)p->pressure.id);
p->pressure.active = 0; p->cooling = 1;
}
if (!net->global_pressure && p->load.snapshot.confirmed && !p->pressure.active && !p->cooling) {
uint64_t g = p->load.snapshot.growth_rate;
pressure_begin(p, g ? g + g / 100 + 1 : p->load.queued.bytes / 2 + 1, now);
}
if (p->pressure.active && now >= p->pressure.deadline_ms) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream OVERLOAD deadline peer=%llx id=%llu released=%llu required=%llu",
DEBUG_WARN(DEBUG_CATEGORY_VIDEO, "stream OVERLOAD deadline peer=%llx id=%llu released=%llu required=%llu; preserve essential subscriptions",
(unsigned long long)p->node_id, (unsigned long long)p->pressure.id,
(unsigned long long)p->pressure.released_rate, (unsigned long long)p->pressure.required_rate);
for (unsigned i = 0; i < p->pressure.count && p->pressure.released_rate < p->pressure.required_rate; i++) {
struct STREAM_RECEIVE *f = flow_by_handle(p, p->pressure.cuts[i].handle, 0);
if (f) flow_close(f, STREAM_REASON_OVERLOAD, 1);
}
}
if (p->pressure.active && (p->pressure.released_rate >= p->pressure.required_rate || now >= p->pressure.deadline_ms)) {
p->pressure.active = 0; p->cooling = 1;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream OVERLOAD complete peer=%llx id=%llu closed=%u rate=%llu requested=%llu",
DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "stream OVERLOAD complete peer=%llx id=%llu closed=%u rate=%llu requested=%llu",
(unsigned long long)p->node_id, (unsigned long long)p->pressure.id, p->pressure.released_count,
(unsigned long long)p->pressure.released_rate, (unsigned long long)p->pressure.required_rate);
}
@@ -1799,8 +1981,8 @@ static void pressure_tick(struct stream_network *net, uint64_t now) {
}
for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next) {
if (f->account_failed) flow_close(f, STREAM_REASON_PROTOCOL, 1);
if (!f->outgoing && f->info.state == STREAM_RECEIVE_ACTIVE && p->pressure.active
&& flow_age(f, now) > STREAM_LOAD_MAX_AGE_MS) expire_pending(f);
if (!f->outgoing && f->info.state == STREAM_RECEIVE_ACTIVE
&& flow_age(f, now) > STREAM_LOAD_CATCHUP_MS) expire_pending(f);
}
p->busy--; peer_sweep(p);
}
@@ -1815,6 +1997,43 @@ uint64_t stream_network_pending_bytes(struct UTUN_INSTANCE *inst, const struct s
return bytes;
}
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);
struct stream_local local;
if (!net || !d || !out) {
DEBUG_ERROR(DEBUG_CATEGORY_VIDEO, "stream buffer positions invalid arguments"); return STREAM_ERR_ARGUMENT;
}
if (stream_local_get(inst, d, &local) != STREAM_OK) {
DEBUG_WARN(DEBUG_CATEGORY_VIDEO, "stream buffer positions missing source=%llx", (unsigned long long)d->source_node_id);
return STREAM_ERR_STATE;
}
struct stream_buffer_stats stats;
int result = stream_buffer_get_stats(local.buffer, &stats);
if (result != STREAM_OK) return result;
struct stream_buffer_positions positions = {.first = stats.expired_before, .head = stats.blocks ? stats.highest_seq + 1 : 0};
for (struct stream_peer *p = net->peers; p; p = p->next) {
struct stream_buffer_peer_position row = {.peer = p->node_id, .worst = UINT64_MAX};
for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next) {
if (f->outgoing || f->info.state != STREAM_RECEIVE_ACTIVE || f->info.next == UINT64_MAX
|| !same_stream(&f->descriptor, d)) continue;
row.subflows++;
if (f->info.next < row.worst) row.worst = f->info.next;
if (f->info.next > row.best) row.best = f->info.next;
}
if (!row.subflows) continue;
positions.total++;
unsigned at = 0;
while (at < positions.count && (positions.peers[at].worst < row.worst
|| (positions.peers[at].worst == row.worst && positions.peers[at].peer < row.peer))) at++;
if (at == STREAM_BUFFER_POSITION_PEERS) continue;
if (positions.count < STREAM_BUFFER_POSITION_PEERS) positions.count++;
for (unsigned i = positions.count - 1; i > at; i--) positions.peers[i] = positions.peers[i - 1];
positions.peers[at] = row;
}
*out = positions; return STREAM_OK;
}
int stream_peer_get_load(struct UTUN_INSTANCE *inst, uint64_t node, struct stream_peer_load_info *info) {
struct stream_network *net = stream_network_get(inst); struct stream_peer *p = net ? peer_find(net, node) : NULL;
if (!p || !info) { DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream load unavailable peer=%llx", (unsigned long long)node); return STREAM_ERR_STATE; }
+4
View File
@@ -57,6 +57,10 @@ struct stream_peer_load_info {
int stream_peer_get_load(struct UTUN_INSTANCE *inst, uint64_t peer, struct stream_peer_load_info *info);
/* Оставшиеся RS-части всех локальных выдач точной публикации; принятые ETCP части сюда не входят. */
uint64_t stream_network_pending_bytes(struct UTUN_INSTANCE *inst, const struct stream_descriptor *descriptor);
/* Снимок позиций downstream в кольце точной эпохи, худшие получатели первыми. Только uasync.
* 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_init(struct UTUN_INSTANCE *inst, struct stream_network **out);
void stream_network_destroy(struct stream_network *network);
+30 -6
View File
@@ -13,6 +13,8 @@
#define STREAM_PROBE_CYCLE_MS 30000
#define STREAM_OFFER_TTL_MS 60000
#define STREAM_MIGRATION_TIMEOUT_MS 5000
#define STREAM_TRANSPORT_RETRY_MS 250
#define STREAM_RECOVERY_STALL_MS 1000
#define STREAM_BOOTSTRAP_RELAY_MS 100
#define STREAM_AVAILABILITY_RETRY_MS 100
#define STREAM_AVAILABILITY_WAIT_MS 5000
@@ -149,7 +151,9 @@ static void probes_poll(struct stream_optimizer *o, struct STREAM_HANDLE *h, uin
}
}
if (info.reason == STREAM_REASON_TRANSPORT || info.reason == STREAM_REASON_MEMBER || info.reason == STREAM_REASON_PROTOCOL) {
s->failures++; s->retry_ms = now + backoff(s->failures);
s->failures++;
s->retry_ms = now + (info.reason == STREAM_REASON_TRANSPORT && s->local_epoch
? STREAM_TRANSPORT_RETRY_MS : backoff(s->failures));
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream optimizer probe closed source=%llx peer=%llx sub=%u reason=%d"
" info_ready=%d local_epoch=%llx peer_epoch=%llx failures=%u retry_in_ms=%llu cycle=%llu",
(unsigned long long)o->descriptor.source_node_id, (unsigned long long)s->node, s->probe_sub, info.reason,
@@ -160,6 +164,18 @@ static void probes_poll(struct stream_optimizer *o, struct STREAM_HANDLE *h, uin
} else {
s->offers[s->probe_sub].info = info; s->failures = 0; s->retry_ms = 0;
s->local_epoch = info.local_epoch; s->peer_epoch = info.peer_epoch;
/* Автор создаёт все n частей одной пачки и вытесняет их вместе. Его диапазон/путь общий;
* новые DATA-запросы всё равно проходят собственный INFO/ACCEPT, но открываются параллельно. */
if (s->source) {
for (unsigned sub = 0; sub < o->descriptor.n; sub++) {
s->offers[sub].info = info; s->offers[sub].info.subblock = sub;
s->offers[sub].retry_ms = 0; s->offers[sub].failures = 0;
}
DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "stream source offers ready peer=%llx parts=%u range=%llu..%llu parallel=1",
(unsigned long long)s->node, o->descriptor.n,
(unsigned long long)info.first_available, (unsigned long long)info.last_available);
stream_receive_close(s->probe); s->probe = NULL; continue;
}
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream optimizer offer peer=%llx sub=%u range=%llu..%llu hops=%u"
" rtt=%u/%d chain=%u/%d slots=%u accepting=%d lag=%llu stopped=%d age=%llu/%d queue=%llu penalty_ms=%llu score_ms=%llu",
(unsigned long long)s->node, s->probe_sub, (unsigned long long)info.first_available,
@@ -185,6 +201,9 @@ static void probes_poll(struct stream_optimizer *o, struct STREAM_HANDLE *h, uin
static void probes_schedule(struct stream_optimizer *o, struct STREAM_HANDLE *h, uint64_t now) {
unsigned pending = 0, active = 0;
for (struct stream_supplier *s = o->suppliers; s; s = s->next) {
if (s->retry_ms && now >= s->retry_ms) {
s->retry_ms = 0; s->cycle = 0; o->schedule.next_ms = 0;
}
if (!s->probe && s->availability_retry_ms && now >= s->availability_retry_ms) {
s->availability_retry_ms = 0;
if (s->compatible && now < s->availability_until_ms && now >= s->retry_ms) {
@@ -302,7 +321,11 @@ static void receive_failed(struct stream_optimizer *o, struct STREAM_RECEIVE **h
int resynchronized = info.reason == STREAM_REASON_TRANSPORT && info.local_epoch && info.peer_epoch
&& s->local_epoch && s->peer_epoch && (info.local_epoch != s->local_epoch || info.peer_epoch != s->peer_epoch);
if (resynchronized) { offer->failures = 0; offer->retry_ms = 0; }
else { offer->failures++; offer->retry_ms = now + backoff(offer->failures); }
else {
offer->failures++;
offer->retry_ms = now + (info.reason == STREAM_REASON_TRANSPORT && info.blocks
? STREAM_TRANSPORT_RETRY_MS : backoff(offer->failures));
}
if (info.reason == STREAM_REASON_OVERLOAD) s->retry_ms = offer->retry_ms;
}
o->info.failures++;
@@ -448,7 +471,7 @@ void stream_optimizer_tick(struct stream_optimizer *o, struct STREAM_HANDLE *h,
uint64_t previous = pick->current ? info.peer : 0;
stream_receive_close(pick->current); pick->current = pick->replacement; pick->replacement = NULL;
pick->opened_ms = now; pick->better_ms = pick->better_node = 0; o->info.migrations++;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream optimizer moved sub=%u from=%llx to=%llx verified_blocks=%llu next=%llu",
DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "stream optimizer moved sub=%u from=%llx to=%llx verified_blocks=%llu next=%llu",
sub, (unsigned long long)previous, (unsigned long long)replacement.peer,
(unsigned long long)replacement.blocks, (unsigned long long)replacement.next);
}
@@ -456,12 +479,13 @@ void stream_optimizer_tick(struct stream_optimizer *o, struct STREAM_HANDLE *h,
if (!o->progress_valid || o->next_seq != local.next) {
o->progress_valid = 1; o->next_seq = local.next; o->progress_ms = now;
}
if (local.anchored && now - o->progress_ms >= STREAM_MIGRATION_TIMEOUT_MS && received < (unsigned)(o->descriptor.n - o->descriptor.r)) {
if (local.anchored && now - o->progress_ms >= STREAM_RECOVERY_STALL_MS && received < (unsigned)(o->descriptor.n - o->descriptor.r)) {
int newer = 0;
for (struct stream_supplier *s = o->suppliers; s; s = s->next)
if (s->offers) for (unsigned sub = 0; sub < o->descriptor.n; sub++) {
struct stream_receive_info *offer = &s->offers[sub].info;
if (offer->info_ms && now - offer->info_ms < STREAM_OFFER_TTL_MS && offer->last_available > local.next) newer = 1;
if (offer->info_ms && now - offer->info_ms < STREAM_OFFER_TTL_MS
&& offer->last_available != UINT64_MAX && offer->last_available > local.next) newer = 1;
}
if (newer) {
selections_close(o);
@@ -526,7 +550,7 @@ void stream_optimizer_tick(struct stream_optimizer *o, struct STREAM_HANDLE *h,
usage(o, info.peer, sub), usage(o, s->node, sub), (unsigned long long)(now - pick->better_ms));
if (stream_receive_open(h, s->node, (uint8_t)sub, seq, &pick->replacement) == STREAM_OK) {
pick->replacement_ms = now; migrating = 1;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream optimizer overlap sub=%u old=%llx new=%llx index=%llu",
DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "stream optimizer overlap sub=%u old=%llx new=%llx index=%llu",
sub, (unsigned long long)info.peer, (unsigned long long)s->node, (unsigned long long)seq);
}
}
+1 -1
View File
@@ -311,7 +311,7 @@ int stream_playback_audio_starved(const struct stream_playback_audio* a, uint64_
void stream_playback_audio_finish(struct stream_playback_audio* a) {
if (!a || a->finished) return;
a->endPts = std::max(a->startPts, a->expectedPts);
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "playback audio finish pts=%lld expected_frames=%.6f prepared_frames=%.0f",
DEBUG_TRACE(DEBUG_CATEGORY_VIDEO, "playback audio finish pts=%lld expected_frames=%.6f prepared_frames=%.0f",
(long long)a->endPts, a->expectedOutput, a->preparedOutput);
a->stretch.flush(); a->finished = true; stream_playback_audio_prepare(a);
}
+18 -4
View File
@@ -3,6 +3,18 @@
#include "../../lib/debug_config.h"
#include "stream_playback_clock.h"
const char *stream_playback_reason_name(enum stream_playback_reason reason) {
switch (reason) {
case STREAM_PLAYBACK_STARVED: return "starved";
case STREAM_PLAYBACK_HISTORY_WAIT: return "history";
case STREAM_PLAYBACK_COOLDOWN: return "cooldown";
case STREAM_PLAYBACK_RESERVE: return "reserve";
case STREAM_PLAYBACK_CATCHUP: return "catchup";
}
DEBUG_ERROR(DEBUG_CATEGORY_VIDEO, "playback invalid diagnostic reason=%d", reason);
return "invalid";
}
/* Новая временная шкала не наследует минимумы прежней позиции. */
void stream_playback_clock_reset(struct stream_playback_clock *c, uint64_t id) {
if (!c) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "playback clock reset: null clock id=%llu", (unsigned long long)id); return; }
@@ -89,18 +101,20 @@ int stream_playback_clock_update(struct stream_playback_clock *c, int64_t now, i
if (c->waiting) c->tempo_milli = 1000;
struct stream_playback_state result = {.position_us = position, .reserve_us = reserve, .minimum_us = minimum,
.correction_us = c->correction_us, .stable_us = stable, .removable_us = removable,
.guard_us = guard, .pending_gain_us = pending_gain,
.reason = c->waiting ? STREAM_PLAYBACK_STARVED : !complete ? STREAM_PLAYBACK_HISTORY_WAIT
: stable < STREAM_PLAYBACK_SETTLE_US ? STREAM_PLAYBACK_COOLDOWN
: removable <= braking ? STREAM_PLAYBACK_RESERVE : STREAM_PLAYBACK_CATCHUP,
.tempo_milli = c->tempo_milli, .waiting = c->waiting, .history_ready = complete};
if (!c->report_us || now - c->report_us >= 1000000) {
const char *reason = c->waiting ? "starved" : !complete ? "history" : stable < STREAM_PLAYBACK_SETTLE_US ? "cooldown"
: removable <= braking ? "reserve" : "catchup";
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG,
DEBUG_TRACE(DEBUG_CATEGORY_VIDEO,
"playback reserve id=%llu pts=%lld ready=%lld reserve=%lld minimum=%lld correction=%lld "
"stable_ms=%lld history=%u tempo=%d pending_gain=%lld waiting=%d "
"guard_us=%lld removable_us=%lld braking_us=%lld reason=%s",
(unsigned long long)c->id, (long long)position, (long long)ready, (long long)reserve,
(long long)minimum, (long long)c->correction_us, (long long)(stable / 1000), samples,
c->tempo_milli, (long long)pending_gain, c->waiting,
(long long)guard, (long long)removable, (long long)braking, reason);
(long long)guard, (long long)removable, (long long)braking, stream_playback_reason_name(result.reason));
c->report_us = now;
}
c->last_us = now; c->last_position_us = position; *state = result;
+10
View File
@@ -32,11 +32,21 @@ struct stream_playback_clock {
int tempo_milli, initialized, waiting;
};
/* Причина решения регулятора; значения передаются также через JNI. */
enum stream_playback_reason {
STREAM_PLAYBACK_STARVED, STREAM_PLAYBACK_HISTORY_WAIT, STREAM_PLAYBACK_COOLDOWN,
STREAM_PLAYBACK_RESERVE, STREAM_PLAYBACK_CATCHUP
};
struct stream_playback_state {
int64_t position_us, reserve_us, minimum_us, correction_us, stable_us, removable_us;
int64_t guard_us, pending_gain_us; // Защищённый запас и ещё не проигранное сокращение PCM
enum stream_playback_reason reason; // Причина последнего решения регулятора
int tempo_milli, waiting, history_ready;
};
/* Короткое постоянное имя причины для лога; неизвестное значение логируется и возвращает "invalid". */
const char *stream_playback_reason_name(enum stream_playback_reason reason);
// ── Управление временной шкалой ──
// Обнулить историю и вернуть темп 1x; id — идентификатор плеера для диагностики.
+7 -1
View File
@@ -33,7 +33,7 @@ struct stream_publisher {
struct publisher_queue video, audio;
uint8_t *wire;
size_t capacity;
uint64_t group, next_frame, last_pts[2], drain_until, report_us;
uint64_t group, next_frame, last_pts[2], drain_until, report_us, buffer_report_us;
int seen[2], stopping, audio_enabled, audio_dirty;
};
static void publisher_process(void *arg);
@@ -87,6 +87,7 @@ static void publisher_close(struct stream_publisher *p, int reason) {
p->status.state = reason ? STREAM_PUBLISHER_FAILED : STREAM_PUBLISHER_ENDED; p->status.reason = reason;
p->status.video_bytes = p->status.audio_bytes = p->status.total_bytes = 0;
p->status.stream_bytes = p->status.transport_bytes = p->status.inflight_bytes = p->status.acked_1s = 0;
memset(&p->status.buffer, 0, sizeof(p->status.buffer)); p->buffer_report_us = 0;
}
/* Один снимок всех соединений, без повторного учёта input_send_q внутри inflight. */
@@ -105,6 +106,11 @@ static void publisher_stats(void *arg) {
p->status.acked_1s = transport.acked_1s; p->status.connections = transport.connections;
p->status.total_bytes = p->status.video_bytes + p->status.audio_bytes + p->status.stream_bytes
+ p->status.transport_bytes + p->status.inflight_bytes;
if (!p->buffer_report_us || now - p->buffer_report_us >= 500000) {
p->buffer_report_us = now;
if (stream_network_buffer_positions(p->inst, &info.descriptor, &p->status.buffer) != STREAM_OK)
DEBUG_WARN(DEBUG_CATEGORY_VIDEO, "publisher buffer positions unavailable id=%llu", (unsigned long long)p->status.id);
}
if (now - p->report_us >= 1000000) {
p->report_us = now;
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM,
+2
View File
@@ -6,6 +6,7 @@
#ifndef STREAM_PUBLISHER_H
#define STREAM_PUBLISHER_H
#include "stream_media.h"
#include "stream_buffer.h"
#ifdef __cplusplus
extern "C" {
#endif
@@ -19,6 +20,7 @@ struct stream_publisher_status {
int state, reason, need_key;
int compressor_enabled, compressor_max_gain_db, compressor_rise_rate;
int audio_enabled;
struct stream_buffer_positions buffer; // Позиции получателей в кольце; обновление раз в 500 мс
};
int stream_publisher_create(struct stream_publisher **out, struct UTUN_INSTANCE *inst);
void stream_publisher_destroy(struct stream_publisher *publisher);
+1 -1
View File
@@ -145,7 +145,7 @@ static void process(void *arg) {
uint64_t now = get_time_us();
if (now - s->report_us >= 1000000) {
s->report_us = now;
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG,
DEBUG_TRACE(DEBUG_CATEGORY_VIDEO,
"viewer buffer id=%llu source=%llx epoch=%llx elapsed_us=%llu initialized=%d generation=%llu "
"start_us=%lld position_us=%lld ready_us=%lld video_us=%lld audio_us=%lld reserve_us=%lld "
"requested=%d waiting=%d slot=%d slot_age_us=%llu records=%llu prepare_us=%llu/%llu handoff_us=%llu/%llu "
+2
View File
@@ -532,6 +532,7 @@ void etcp_conn_reset(struct ETCP_CONN* conn) {
conn->got_initial_pkt = 0;
conn->unacked_bytes = conn->rtt_last = conn->rtt_avg_10 = conn->jitter = conn->last_rtt_cb_time = 0;
conn->bytes_sent_total = conn->retransmissions_count = conn->ack_packets_count = 0;
conn->diagnostic_ack_us = 0;
conn->rx_dup_count = conn->tx_dup_count = 0;
memset(conn->rtt_history, 0, sizeof(conn->rtt_history)); conn->rtt_history_idx = 0;
conn->last_rr_link = NULL; conn->reset_count++;
@@ -1341,6 +1342,7 @@ void etcp_ack_recv(struct ETCP_CONN* conn, uint32_t seq) {
abort();
}
conn->unacked_bytes -= packet->ll.len; conn->bytes_sent_total += packet->ll.len; conn->ack_packets_count++;
conn->diagnostic_acked_bytes += packet->ll.len; conn->diagnostic_ack_us = get_time_us();
etcp_tx_record_ack(&conn->instance->tx_history, get_time_us(), packet->ll.len);
DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] logical delivery seq=%u bytes=%u active_attempts=%u unacked=%u",
conn->log_name, seq, packet->ll.len, packet->active_attempts, conn->unacked_bytes);
+5
View File
@@ -190,6 +190,11 @@ struct ETCP_CONN {
uint32_t reinit_count;
uint64_t setup_start_tb; // время начала/рестарта подключения (0.1ms)
uint32_t reset_count;
// Диагностика за lifetime соединения: reset не обнуляет накопительные счётчики.
uint64_t diagnostic_acked_bytes; // Уникальные логические DATA-байты, подтверждённые пиром
uint64_t diagnostic_ack_us; // Последний реальный логический ACK; 0 — нет измерения текущей сессии
uint64_t diagnostic_retries; // Повторные DATA, действительно принятые транспортом
uint64_t diagnostic_rtos; // Физические DATA-попытки, снятые по RTO
// uint32_t bytes_sent_norx; // сколько отправили байт без единого ответного пакета (для детекции запроса реконнекта)
// Window and inflight management
+2
View File
@@ -68,6 +68,7 @@ static void attempt_retire(struct ETCP_LINK* link, struct ETCP_TX_ATTEMPT* attem
link->inflight_bytes -= attempt->ll.len; link->inflight_packets--; conn->active_attempts--;
if (packet) packet->active_attempts--;
if (loss) {
conn->diagnostic_rtos++;
if (link->data_rto_streak < ETCP_TX_STALL_THRESHOLD) link->data_rto_streak++;
if (!link->tx_stalled && link->data_rto_streak == ETCP_TX_STALL_THRESHOLD) {
link->tx_stalled = 1; link->tx_stalled_since_us = get_time_us();
@@ -227,6 +228,7 @@ void etcp_tx_complete(struct ETCP_DGRAM* dgram, int success) {
conn->log_name, packet->seq, (unsigned)(uint16_t)attempt->attempt_id, link->local_link_id,
packet->last_link ? packet->last_link->local_link_id : 0, packet->ll.len, link->inflight_bytes, link->inflight_packets);
if (packet->last_attempt_id && packet->last_link) packet->last_link->total_retransmissions++;
if (packet->send_count) conn->diagnostic_retries++;
packet->send_hist[packet->send_count % 8] = link->local_link_id;
if (packet->send_count != UINT32_MAX) packet->send_count++;
if (etcp_packet_link_used(packet, link->local_link_id)) {
+4 -4
View File
@@ -51,7 +51,7 @@ static __thread struct { unsigned depth, repeated; char last[1024]; } g_codec_pr
static void video_codec_probe_flush(void) {
if (g_codec_probe.repeated)
DEBUG_DEBUG(DEBUG_CATEGORY_VIDEO, "ffmpeg probe: repeated=%u message=%s", g_codec_probe.repeated, g_codec_probe.last);
DEBUG_TRACE(DEBUG_CATEGORY_VIDEO, "ffmpeg probe: repeated=%u message=%s", g_codec_probe.repeated, g_codec_probe.last);
g_codec_probe.repeated = 0;
}
@@ -128,7 +128,7 @@ static void video_av_log_cb(void* avcl, int level, const char* fmt, va_list vl)
if (g_codec_probe.depth && level <= AV_LOG_WARNING) {
if (!strcmp(g_codec_probe.last, buf)) { ++g_codec_probe.repeated; return; }
video_codec_probe_flush(); snprintf(g_codec_probe.last, sizeof(g_codec_probe.last), "%s", buf);
DEBUG_DEBUG(DEBUG_CATEGORY_VIDEO, "ffmpeg probe source=%s level=%d: %s", avcl ? av_default_item_name(avcl) : "none", level, buf);
DEBUG_TRACE(DEBUG_CATEGORY_VIDEO, "ffmpeg probe source=%s level=%d: %s", avcl ? av_default_item_name(avcl) : "none", level, buf);
return;
}
if (headers) {
@@ -216,7 +216,7 @@ int video_get_info(const char* path, struct video_info* out) {
}
vc_metadata(format, video, vc_stream_index(format, AVMEDIA_TYPE_AUDIO), out);
avformat_close_input(&format);
DEBUG_DEBUG(DEBUG_CATEGORY_VIDEO, "video metadata path=%s dimensions=%dx%d codec=%s bitrate=%lld audio=%s/%lld",
DEBUG_TRACE(DEBUG_CATEGORY_VIDEO, "video metadata path=%s dimensions=%dx%d codec=%s bitrate=%lld audio=%s/%lld",
path, out->width, out->height, out->codec, (long long)out->bitrate,
out->audio_codec, (long long)out->audio_bitrate);
return 0;
@@ -736,7 +736,7 @@ static void vc_done_trampoline(void* raw, int error) {
file_io_temp_discard(&ctx->file);
vc_progress_set(-1);
vc_active_set(0);
DEBUG_DEBUG(DEBUG_CATEGORY_VIDEO, "video completion destination=%s error=%d", ctx->destination, error);
DEBUG_TRACE(DEBUG_CATEGORY_VIDEO, "video completion destination=%s error=%d", ctx->destination, error);
ctx->done(ctx->arg, error, error ? NULL : &ctx->result);
u_free(ctx);
}
+184 -7
View File
@@ -33,6 +33,33 @@
} } while (0)
static struct UTUN_INSTANCE *nodes[4];
static struct {
uint64_t last_ms[4][4];
unsigned lines, tx, rx, idle;
} reports;
/* Проверяем живые сводки: один peer, все субпотоки вместе; таймер продолжает писать при полном простое. */
static void capture_stream_report(int level, const char *category, const char *message) {
const char *line = strstr(message, "stream NET self=");
if (!line || strcmp(category, "video")) return;
CHECK(level == DEBUG_LEVEL_DEBUG);
unsigned long long self, peer, dt, blocks, bytes;
CHECK(sscanf(line, "stream NET self=%llx peer=%llx dt_ms=%llu", &self, &peer, &dt) == 3 && dt >= 500);
unsigned a = 0, b = 0;
while (a < 4 && nodes[a]->node_id != self) a++;
while (b < 4 && nodes[b]->node_id != peer) b++;
CHECK(a < 4 && b < 4);
uint64_t now = get_time_tb() / 10;
CHECK(!reports.last_ms[a][b] || now - reports.last_ms[a][b] >= 500);
reports.last_ms[a][b] = now; reports.lines++;
const char *sent = strstr(line, " sent="), *received = strstr(line, " received=");
CHECK(sent && received && strstr(line, "errors=send/recv/crypto"));
CHECK(sscanf(sent, " sent=%llu/%llu", &blocks, &bytes) == 2);
reports.tx += blocks != 0;
int no_tx = blocks == 0;
CHECK(sscanf(received, " received=%llu/%llu", &blocks, &bytes) == 2);
reports.rx += blocks != 0; reports.idle += no_tx && blocks == 0;
}
static struct UASYNC *ua;
static uint8_t group_private[32], group_public[32], group_x[32], expected[1000];
static struct STREAM_HANDLE *reader1, *reader2;
@@ -579,6 +606,8 @@ static void test_pending_down(void) {
/* Источник -> relay -> получатель; данные передаются через raw ETCP, никаких прямых stream_ingest в этой проверке. */
static void test_network(void) {
debug_level_t previous_video = g_debug_config.category_levels[DEBUG_CATEGORY_VIDEO];
debug_set_category_level(DEBUG_CATEGORY_VIDEO, DEBUG_LEVEL_DEBUG); debug_set_log_hook(capture_stream_report);
for (unsigned i = 0; i < sizeof(expected); i++) expected[i] = (uint8_t)(i * 71 + 3);
struct TOPO_PEER_REQUEST *connections[2] = {0};
struct TOPO_GROUP *root = topo_groups_find(nodes[3]->topo_groups, GROUP2);
@@ -626,6 +655,11 @@ static void test_network(void) {
}
CHECK(info.blocks == 1 && info.start == 1 && info.next == 2 && info.handle);
}
struct stream_buffer_positions positions;
CHECK(stream_network_buffer_positions(nodes[3], &origin.descriptor, &positions) == STREAM_OK);
CHECK(positions.first == 0 && positions.head == 2 && positions.count == 1 && positions.total == 1);
CHECK(positions.peers[0].peer == nodes[1]->node_id && positions.peers[0].subflows == 8);
CHECK(positions.peers[0].worst == 2 && positions.peers[0].best == 2);
/* Меняется только локальная stream generation; физический ETCP и поколение источника сохраняются. */
struct stream_receive_info previous_generation; CHECK(stream_receive_get_info(up[0], &previous_generation) == STREAM_OK);
struct ETCP_CONN *stable = instance_find_conn(nodes[1], nodes[3]->node_id); CHECK(stable && stable->links_up);
@@ -761,9 +795,142 @@ static void test_network(void) {
test_clock_run_for(ua, 1000);
stream_set_event_callback(nodes[1], on_stream, NULL); stream_set_event_callback(nodes[2], NULL, NULL);
for (unsigned i = 0; i < 2; i++) topo_group_peer_close(connections[i]);
debug_set_log_hook(NULL); debug_set_category_level(DEBUG_CATEGORY_VIDEO, previous_video);
CHECK(reports.lines && reports.tx && reports.rx && reports.idle);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] live ETCP chain: common index/restart, FEC, paths/LOOP, GAP, duplicate/reverse, END, leases, REINIT");
}
struct recovery_pending {
struct recovery_pending *next;
struct ETCP_CONN *conn;
struct ll_entry *entry;
uint64_t due_ms;
};
struct recovery_test {
struct STREAM_HANDLE *publisher, *viewer;
struct stream_descriptor descriptor;
struct recovery_pending *head, *tail;
etcp_recv_fn receive;
uint64_t origin_us, next_frame_us, sent, frames, last_pts, last_frame_ms, maximum_gap_ms, maximum_lag_ms;
unsigned delay_ms, overloads, overload_closed, media_gaps;
int paused;
};
static struct recovery_test *recovery;
/* Реальные raw ETCP-пакеты сохраняют порядок и владение; задержка имитирует путь, а не ответы selector. */
static void recovery_receive(struct ETCP_CONN *conn, struct ll_entry *entry) {
struct stream_message message;
if (stream_wire_parse(entry->dgram, entry->len, &message) == STREAM_OK
&& message.descriptor.source_epoch == recovery->descriptor.source_epoch) {
if (message.command == STREAM_CMD_OVERLOAD) recovery->overloads++;
if (message.command == STREAM_CMD_CLOSED && message.length && message.payload[0] == STREAM_REASON_OVERLOAD)
recovery->overload_closed++;
}
if (!recovery->delay_ms && !recovery->head) { recovery->receive(conn, entry); return; }
struct recovery_pending *pending = u_calloc(1, sizeof(*pending)); CHECK(pending && etcp_conn_ref_take(conn) == 0);
pending->conn = conn; pending->entry = entry; pending->due_ms = get_time_tb() / 10 + recovery->delay_ms;
if (recovery->tail && pending->due_ms < recovery->tail->due_ms) pending->due_ms = recovery->tail->due_ms;
if (recovery->tail) recovery->tail->next = pending; else recovery->head = pending;
recovery->tail = pending;
}
/* 50 кадров/с, KEY раз в секунду, n=8/r=0: проверяем сборку настоящих подписанных uAV2/RS-пачек и media cursor. */
static void recovery_run(struct recovery_test *r, unsigned duration_ms) {
uint64_t until = get_time_tb() / 10 + duration_ms;
while (get_time_tb() / 10 < until) {
uint64_t now_us = get_time_us(), now_ms = get_time_tb() / 10;
while (r->head && now_ms >= r->head->due_ms) {
struct recovery_pending *pending = r->head; r->head = pending->next;
if (!r->head) r->tail = NULL;
r->receive(pending->conn, pending->entry); etcp_conn_ref_free(pending->conn); u_free(pending);
}
if (!r->paused && now_us >= r->next_frame_us) {
uint8_t video[6000] = {0}, wire[8192]; size_t length = 0;
memset(video + 4, (uint8_t)r->sent, sizeof(video) - 4);
struct stream_media_block block = {.format = {320, 180, 0, STREAM_MEDIA_H264, 0, 1}, .video_count = 1};
block.video[0] = (struct stream_media_video){r->sent + 1, now_us - r->origin_us, sizeof(video), 0,
sizeof(video), r->sent % 50 ? 0 : STREAM_MEDIA_KEY, video};
CHECK(stream_media_encode(&block, wire, sizeof(wire), &length) == STREAM_OK);
CHECK(stream_publish_block(r->publisher, wire, length) == STREAM_OK);
r->sent++; r->next_frame_us = now_us + 20000;
}
for (unsigned work = 0; work < 32; work++) {
struct stream_media_packet packet;
int result = stream_media_read(r->viewer, -1, 1, -1, &packet);
if (result == STREAM_WAIT || result == STREAM_MORE) break;
if (result == STREAM_GAP || result == STREAM_EXPIRED) {
r->media_gaps++; CHECK(stream_media_resync(r->viewer) == STREAM_OK); continue;
}
CHECK(result == STREAM_OK && packet.kind == STREAM_MEDIA_READ_VIDEO && packet.video.length == 6000);
CHECK(!r->frames || packet.video.pts_us > r->last_pts);
for (unsigned i = 4; i < packet.video.length; i++) CHECK(packet.video.data[i] == (uint8_t)(packet.video.frame_id - 1));
uint64_t gap = r->last_frame_ms ? now_ms - r->last_frame_ms : 0;
uint64_t lag = (now_us - r->origin_us - packet.video.pts_us) / 1000;
if (gap > r->maximum_gap_ms) r->maximum_gap_ms = gap;
if (lag > r->maximum_lag_ms) r->maximum_lag_ms = lag;
r->last_frame_ms = now_ms; r->last_pts = packet.video.pts_us; r->frames++;
}
if (r->last_frame_ms && now_ms - r->last_frame_ms > r->maximum_gap_ms)
r->maximum_gap_ms = now_ms - r->last_frame_ms;
test_clock_poll(ua, 50);
}
}
static void test_recovery_scenarios(void) {
debug_level_t previous_video = g_debug_config.category_levels[DEBUG_CATEGORY_VIDEO];
debug_set_category_level(DEBUG_CATEGORY_VIDEO, DEBUG_LEVEL_DEBUG);
unsigned failed = 0;
for (unsigned scenario = 0; scenario < 3; scenario++) {
int wifi = scenario == 1;
struct recovery_test r = {0}; recovery = &r;
CHECK(stream_publish_open(nodes[0], GROUP1, NULL, &r.publisher) == STREAM_OK);
struct stream_info info; CHECK(stream_get_info(r.publisher, &info) == STREAM_OK); r.descriptor = info.descriptor;
CHECK(r.descriptor.n == 8 && !r.descriptor.r);
wait_state(1, 0, GROUP1, r.descriptor.source_epoch, 0);
CHECK(stream_media_open(nodes[1], GROUP1, nodes[0]->node_id, r.descriptor.source_epoch,
STREAM_MEDIA_TRACK_VIDEO, &r.viewer) == STREAM_OK);
r.receive = nodes[1]->api_bindings.callbacks[ETCP_ID_STREAM];
CHECK(etcp_bind(nodes[1], ETCP_ID_STREAM, recovery_receive) == 0);
r.delay_ms = wifi ? 100 : 0; r.origin_us = get_time_us(); r.next_frame_us = r.origin_us;
recovery_run(&r, 3000); CHECK(r.frames >= 50);
r.maximum_gap_ms = r.maximum_lag_ms = 0;
struct ETCP_CONN *conn = instance_find_conn(nodes[0], nodes[1]->node_id); CHECK(conn && conn->links_up);
struct ll_queue *queue = conn->send_input_q;
queue_callback_fn callback = queue->callback; void *arg = queue->callback_arg; CHECK(callback);
if (wifi) {
etcp_conn_reinit(conn, "test Wi-Fi network switch");
recovery_run(&r, 3500);
} else {
/* Очередь растёт 850мс: как SM Q=165KB/671мс, но реальный транспорт остаётся UP. */
queue_set_callback(queue, NULL, NULL); recovery_run(&r, scenario == 2 ? 2600 : 850);
queue_set_callback(queue, callback, arg); queue_resume_callback(queue);
recovery_run(&r, 3500); CHECK(r.overloads);
}
struct stream_selection_info selection; CHECK(stream_selection_get_info(r.viewer, &selection) == STREAM_OK);
DEBUG_INFO(DEBUG_CATEGORY_VIDEO,
"recovery regression scenario=%s sent=%llu frames=%llu freeze_ms=%llu lag_ms=%llu overload=%u closed=%u gaps=%u restarts=%llu",
wifi ? "wifi-switch-control-delay100ms" : scenario == 2 ? "stalled-wifi-2600ms" : "slow-wifi-850ms",
(unsigned long long)r.sent, (unsigned long long)r.frames,
(unsigned long long)r.maximum_gap_ms, (unsigned long long)r.maximum_lag_ms,
r.overloads, r.overload_closed, r.media_gaps, (unsigned long long)selection.restarts);
failed += r.overload_closed != 0 || r.maximum_gap_ms > (wifi ? 1500 : scenario == 2 ? 4000 : 1600)
|| get_time_tb() / 10 - r.last_frame_ms >= 300;
if (scenario == 2) CHECK(r.media_gaps && selection.restarts == 0);
/* Пауза автора не является поломкой пути и не должна перезапускать приёмник. */
uint64_t restarts = selection.restarts;
r.paused = 1; recovery_run(&r, 2000);
CHECK(stream_selection_get_info(r.viewer, &selection) == STREAM_OK && selection.restarts == restarts);
r.delay_ms = 0;
while (r.head) recovery_run(&r, 10);
CHECK(etcp_bind(nodes[1], ETCP_ID_STREAM, r.receive) == 0);
stream_close(r.viewer); stream_close(r.publisher); test_clock_run_for(ua, 10000);
recovery = NULL;
}
debug_set_category_level(DEBUG_CATEGORY_VIDEO, previous_video);
CHECK(!failed);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] recovery: 8+0 media survives Wi-Fi switch, 850ms congestion and 2600ms stall without essential cuts");
}
/* Настоящий send_input_q намеренно держим непустым: короткий глитч не режет, устойчивый рост вызывает OVERLOAD. */
static void test_pressure(void) {
memset(pressure_payload, 42, sizeof(pressure_payload));
@@ -865,15 +1032,15 @@ static void test_pressure(void) {
deadline = get_time_tb() + 100000;
while (!selects_seen && get_time_tb() < deadline) test_clock_poll(ua, 100);
CHECK(overload_seen && selects_seen && stale_select_length);
/* Повторы SELECT намеренно не доходят: source исполняет достаточный остаток на исходном deadline. */
/* SELECT потерян: source сохраняет необходимые части, а запоздалый ответ больше не меняет выдачи. */
while (get_time_tb() / 10 < expires + 100) test_clock_poll(ua, 100);
info = source_load(); CHECK(!info.request_active && info.released_count > 1 && info.released_rate >= info.required_rate);
info = source_load(); CHECK(!info.request_active && !info.released_count && !info.load.queued.bytes);
unsigned closed = 0;
for (unsigned i = 0; i < 8; i++) {
struct stream_receive_info receive; CHECK(stream_receive_get_info(receives[i], &receive) == STREAM_OK);
if (receive.state == STREAM_RECEIVE_CLOSED) { CHECK(receive.reason == STREAM_REASON_OVERLOAD); closed++; }
}
CHECK(closed == info.released_count);
CHECK(!closed);
/* Поздний выбор старого id не закрывает дополнительную выдачу. */
ignore_select = 0;
struct ETCP_CONN *back = instance_find_conn(nodes[1], nodes[3]->node_id); CHECK(back != NULL);
@@ -886,7 +1053,7 @@ static void test_pressure(void) {
for (unsigned i = 1; i <= 3; i += 2) CHECK(etcp_bind(nodes[i], ETCP_ID_STREAM, pressure_handlers[i]) == 0);
stream_set_event_callback(nodes[1], on_stream, NULL);
CHECK(stream_handle_active(reader1) && stream_handle_active(reader2));
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] live overload: glitch, parity-only cooperation, frozen rates, fixed deadline, fallback and stale SELECT");
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] live overload: glitch, safe parity cooperation, fixed deadline, lost/stale SELECT preserves essential parts");
}
/* CPU сам по себе не режет выдачи. Подтверждённый backlog нескольких пиров с локальной загрузкой использует общий бюджет. */
@@ -951,12 +1118,20 @@ static void test_global_pressure(void) {
if (done || get_time_tb() >= deadline) break;
test_clock_poll(ua, 100);
}
CHECK(info[1].released_rate >= info[1].required_rate && info[2].released_rate >= info[2].required_rate);
for (unsigned i = 1; i < 3; i++) {
CHECK(!info[i].load.queued.bytes && info[i].released_count <= 1);
unsigned active = 0;
for (unsigned sub = 0; sub < 2; sub++) {
struct stream_receive_info receive; CHECK(stream_receive_get_info(receives[i][sub], &receive) == STREAM_OK);
active += receive.state == STREAM_RECEIVE_ACTIVE;
}
CHECK(active >= 1 && global_data[i] >= 1);
}
simulated_cpu = 0;
for (unsigned i = 1; i < 3; i++) { for (unsigned sub = 0; sub < 2; sub++) stream_receive_close(receives[i][sub]); stream_close(subscribers[i]); }
stream_close(publisher); test_clock_run_for(ua, 1000);
stream_set_event_callback(nodes[1], on_stream, NULL); stream_set_event_callback(nodes[2], NULL, NULL);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] global overload: CPU alone preserves flows; confirmed multi-peer backlog releases a shared rate budget");
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] global overload: CPU alone preserves flows; multi-peer backlog drains without cutting essential parts");
}
/* Default API сам выбирает выдачи: пустой источник, общий старт, цикл 30 с, overlap, stall и реальный CLOSED. */
@@ -1830,10 +2005,11 @@ int main(int argc, char **argv) {
CHECK(member_sync_start(nodes[i], nodes[0]->node_id, "710001", NULL, NULL) == 0);
}
if (argc == 2 && (!strcmp(argv[1], "--relay-only") || !strcmp(argv[1], "--lifecycle-only")
|| !strcmp(argv[1], "--wake-only") || !strcmp(argv[1], "--bootstrap-only"))) {
|| !strcmp(argv[1], "--wake-only") || !strcmp(argv[1], "--bootstrap-only") || !strcmp(argv[1], "--recovery-only"))) {
if (!strcmp(argv[1], "--relay-only")) test_supernode_relay();
else if (!strcmp(argv[1], "--lifecycle-only")) test_partial_bootstrap_and_end();
else if (!strcmp(argv[1], "--bootstrap-only")) test_bootstrap();
else if (!strcmp(argv[1], "--recovery-only")) test_recovery_scenarios();
else test_sources_after_wake();
for (unsigned i = 0; i < 3; i++) topo_group_peer_close(connections[i]);
for (unsigned i = 0; i < 4; i++) { stream_set_event_callback(nodes[i], NULL, NULL); utun_instance_destroy(nodes[i]); }
@@ -1847,6 +2023,7 @@ int main(int argc, char **argv) {
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] isolated stream checks complete, no tracked allocations remain");
return 0;
}
test_recovery_scenarios();
test_sources_after_wake();
test_profile_atomicity();
stream_set_event_callback(nodes[0], on_stream, NULL); stream_set_event_callback(nodes[1], on_stream, NULL);
+4
View File
@@ -40,6 +40,10 @@ int main(void) {
CHECK(etcp_tx_get_snapshot(inst, now, &s) == 0 && s.connections == 2 && s.inflight_bytes == 200 && s.queued_bytes == 10);
etcp_ack_recv(a, 1); etcp_ack_recv(a, 1);
CHECK(inst->tx_history.total == 80 && !a->unacked_bytes);
CHECK(a->diagnostic_acked_bytes == 80 && a->diagnostic_ack_us);
uint64_t acknowledged_at = a->diagnostic_ack_us;
etcp_ack_recv(a, 1);
CHECK(a->diagnostic_acked_bytes == 80 && a->diagnostic_ack_us == acknowledged_at);
etcp_ack_recv(b, 2); CHECK(inst->tx_history.total == 200);
etcp_connection_close(a); etcp_connection_close(b);
CHECK(etcp_tx_get_snapshot(inst, get_time_us() + 100000, &s) == 0 && !s.connections && !s.inflight_bytes && s.acked_1s == 200);
@@ -13,6 +13,10 @@ data class LogEntry(
)
object LogManager {
/** Категория video проходит общий native-фильтр и доставку в файл/UDP, как C-логи. */
fun videoDebug(message: String) { NativeLib.videoLog(false, message) }
fun videoTrace(message: String) { NativeLib.videoLog(true, message) }
private const val MAX_BUFFER = 100000
private val _buffer = ArrayDeque<LogEntry>(MAX_BUFFER)
private val _events = MutableSharedFlow<LogEntry>(replay = 0, extraBufferCapacity = 512)
@@ -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)
external fun videoLog(trace: Boolean, message: String)
fun streamResync(id: Long): Boolean = nativeStreamResync(id)
external fun streamSignalCreate(id: Long): Long
external fun streamSignalNotify(handle: Long)
@@ -84,6 +84,7 @@ internal object StreamAudioMixer {
* Каждый 10ms блок общего выхода отражён в своей шкале, включая тишину при starvation.
* Reset создаёт новый Input с новой origin; AudioTrack и остальные источники продолжают играть. */
internal class StreamMixedPlayback(val id: Long, pts: Long, private val audioEnabled: Boolean, private val wake: () -> Unit) {
data class Diagnostics(val pcmFrames: Int, val underruns: Long, val silenceFrames: Long)
private val lock = Any()
private var handle = NativeLib.streamPlaybackCreate(id, pts, audioEnabled).also { check(it != 0L) }
private val timeline = StreamMixTimeline()
@@ -110,6 +111,9 @@ internal class StreamMixedPlayback(val id: Long, pts: Long, private val audioEna
"played=${timeline.played(StreamAudioMixer.output.played)} leadUs=$leadUs " +
"pcmFrames=${NativeLib.streamPlaybackBuffered(handle)} underruns=$underruns silentFrames=$silentFrames"
}
val diagnostics: Diagnostics get() = synchronized(lock) {
Diagnostics(NativeLib.streamPlaybackBuffered(handle), underruns, silentFrames)
}
// Только audio worker; он же закрывает handle, поэтому DSP не нуждается в consumer lock.
fun prepare(): Int { check(handle != 0L); return NativeLib.streamPlaybackPrepare(handle) }
@@ -101,7 +101,7 @@ class StreamCaptureEngine(
if (!enabled) orientationListener.disable()
else if (orientationListener.canDetectOrientation()) orientationListener.enable()
else LogManager.addLog("WARN", "video", "stream orientation sensor unavailable id=$id; using ${orientation.degrees}")
LogManager.addLog("DEBUG", "video", "stream orientation monitoring id=$id enabled=$enabled")
LogManager.videoTrace("stream orientation monitoring id=$id enabled=$enabled")
}
fun start() {
@@ -111,7 +111,7 @@ class StreamCaptureEngine(
if (target === previewTarget) { target?.release(); return@post }
val previous = previewTarget; previewTarget = target; previous?.release()
val config = activeConfig; val device = camera
LogManager.addLog("DEBUG", "video", "preview output change id=$id target=${target?.let(System::identityHashCode)} " +
LogManager.videoTrace("preview output change id=$id target=${target?.let(System::identityHashCode)} " +
"generation=$generation camera=${device != null} opening=$opening")
if (device != null && config != null && current(generation)) createSession(device, config, generation)
}) {
@@ -175,7 +175,7 @@ class StreamCaptureEngine(
activeConfig = config
observeOrientation(config.orientation == StreamOrientation.AUTO)
preview.update(null)
LogManager.addLog("DEBUG", "video", "capture camera id=$id revision=$videoRevision generation=$token camera=${mode.camera} " +
LogManager.videoTrace("capture camera id=$id revision=$videoRevision generation=$token camera=${mode.camera} " +
"sensor=${cameraPose.sensor} front=${cameraPose.front} orientation=${config.orientation} broadcastRotation=$videoRotation")
csd = byteArrayOf(); partial.reset(); partialConfig.reset(); requireKey = true; lastKeyRequest = 0; waitingFirst = true
confirming = mode.highSpeed; confirmStartNs = 0
@@ -242,7 +242,7 @@ class StreamCaptureEngine(
override fun onClosed(device: CameraDevice) {
if (camera === device) camera = null
val pending = cameraSessions.values.filter { it.first === device }.map { it.second }
LogManager.addLog("DEBUG", "video", "camera closed id=$id generation=$token retiringSessions=${pending.size}")
LogManager.videoTrace("camera closed id=$id generation=$token retiringSessions=${pending.size}")
pending.forEach { it("cameraClosed") }
if (releaseDone != null && camera == null && !opening) finishRelease(true)
maybeQuit()
@@ -273,13 +273,13 @@ class StreamCaptureEngine(
fun complete(reason: String) {
if (completed) return
completed = true; cameraSessions.remove(serial); output?.release()
LogManager.addLog("DEBUG", "video", "camera session output released id=$id session=$serial reason=$reason remaining=${cameraSessions.size}")
LogManager.videoTrace("camera session output released id=$id session=$serial reason=$reason remaining=${cameraSessions.size}")
maybeQuit()
}
cameraSessions[serial] = device to ::complete
fun valid() = current(token) && serial == sessionGeneration
fun rejected(detail: String) {
if (!valid()) { LogManager.addLog("DEBUG", "video", "stale camera session failure id=$id session=$serial: $detail"); return }
if (!valid()) { LogManager.videoTrace("stale camera session failure id=$id session=$serial: $detail"); return }
if (output == null) { captureError(token, detail); return }
LogManager.addLog("ERROR", "video", "direct preview failed id=$id session=$serial; keep encoder output: $detail")
preview.update(format, failed = true)
@@ -320,7 +320,7 @@ class StreamCaptureEngine(
val now = System.nanoTime()
if (now - sensorReportNs < 1_000_000_000) return
val fps = (sensorResults - sensorReportCount) * 1_000_000_000.0 / (now - sensorReportNs)
LogManager.addLog("DEBUG", "video", "sensor id=$id generation=$token session=$serial results=$sensorResults " +
LogManager.videoTrace("sensor id=$id generation=$token session=$serial results=$sensorResults " +
"callbackFps=${"%.1f".format(fps)} range=${result.get(CaptureResult.CONTROL_AE_TARGET_FPS_RANGE)} " +
"durationNs=${result.get(CaptureResult.SENSOR_FRAME_DURATION)} " +
"exposureNs=${result.get(CaptureResult.SENSOR_EXPOSURE_TIME)} timestamp=${result.get(CaptureResult.SENSOR_TIMESTAMP)}")
@@ -350,7 +350,7 @@ class StreamCaptureEngine(
complete("sessionClosed")
}
}
LogManager.addLog("DEBUG", "video", "camera session configuring id=$id generation=$token session=$serial outputs=${targets.size}")
LogManager.videoTrace("camera session configuring id=$id generation=$token session=$serial outputs=${targets.size}")
if (mode.highSpeed) device.createConstrainedHighSpeedCaptureSession(targets, callback, handler)
else device.createCaptureSession(targets, callback, handler)
} catch (e: Exception) { complete("exception"); rejected("camera session: ${e.message}") }
@@ -364,7 +364,7 @@ class StreamCaptureEngine(
}
private fun captureError(token: Long, detail: String) {
if (!current(token)) { LogManager.addLog("DEBUG", "video", "stale capture error generation=$token: $detail"); return }
if (!current(token)) { LogManager.videoTrace("stale capture error generation=$token: $detail"); return }
LogManager.addLog("ERROR", "video", "capture error id=$id generation=$token: $detail")
val previous = lastGood
if (previous == null || reverting) { failed(detail); return }
@@ -424,7 +424,7 @@ class StreamCaptureEngine(
}
if (now - reportNs >= 1_000_000_000) {
val seconds = (now - reportNs) / 1_000_000_000.0
LogManager.addLog("DEBUG", "video", "capture id=$id revision=$videoRevision targetFps=${mode.fps} " +
LogManager.videoTrace("capture id=$id revision=$videoRevision targetFps=${mode.fps} " +
"fps=${"%.1f".format((encodedFrames - reportFrames) / seconds)} bitrate=${((bytesSent - reportBytes) * 8 / seconds).toLong()} " +
"frames=$frames dropped=$dropped pts=$pts waitingKey=$requireKey")
reportNs = now; reportFrames = encodedFrames; reportBytes = bytesSent
@@ -471,7 +471,7 @@ class StreamCaptureEngine(
try { encoder?.stop() } catch (e: Exception) { LogManager.addLog("WARN", "video", "encoder stop id=$id: ${e.message}") }
try { encoder?.release() } catch (e: Exception) { LogManager.addLog("WARN", "video", "encoder release id=$id: ${e.message}") }
encoder = null; input?.release(); input = null; partial.reset(); partialConfig.reset(); csd = byteArrayOf(); releaseDone = null
LogManager.addLog("DEBUG", "video", "video resources released id=$id closed=$closed generation=$generation")
LogManager.videoTrace("video resources released id=$id closed=$closed generation=$generation")
done(closed)
}
/** Меняем только свою копию PCM; общий микрофон и рация продолжают работать. */
@@ -10,6 +10,7 @@ import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.withContext
import java.nio.ByteBuffer
import java.nio.ByteOrder
import java.util.Locale
/** Audio worker читает независимые треки из RS-кольца и готовит PCM; MediaCodec работает отдельно.
* Общий native controller убирает избыточный запас ускорением, AudioTrack задаёт часы каждого источника.
@@ -106,7 +107,11 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv
var generation = 0L
var playbackStarted = false
var endUntil = 0L
var reported = 0L
var reported = System.nanoTime() / 1000
var shownBase = 0L
var discardedBase = 0L
var underrunBase = 0L
var silenceBase = 0L
var workMaxUs = 0L
var deadline = System.nanoTime() + 10_000_000_000
var clockState: LongArray? = null
@@ -125,12 +130,14 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv
fun resetVideo() {
video?.reset(); waitingKey = true
shownBase = 0; discardedBase = 0
mutableStatus.value = R.string.stream_waiting_key
}
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
underrunBase = 0; silenceBase = 0
records.clear(); ended = false; videoInputEnded = false; audioInputEnded = false; endUntil = 0
resetVideo(); deadline = System.nanoTime() + 10_000_000_000
if (request) check(NativeLib.streamResync(id)) { "Media resync refused" }
@@ -235,13 +242,36 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv
if (endUntil == 0L) endUntil = System.nanoTime() + 100_000_000
if (System.nanoTime() >= endUntil) { mutableStatus.value = R.string.stream_ended; break }
}
if (System.nanoTime() - reported >= 1_000_000_000) {
reported = System.nanoTime()
LogManager.addLog("DEBUG", "stream", "android audio worker id=$id generation=$generation revision=$videoRevision " +
"clock=$position ready=$ready audio=$audioEnabled audioInputs=$audioInputs audioPTS=$lastAudioPts needsAudio=$needsAudio " +
"reserve=${clockState?.get(1)} minimum=${clockState?.get(2)} tempo=${clockState?.get(3)} " +
"${playback?.description} workMaxUs=$workMaxUs waitingKey=$waitingKey " +
"batches=$batches batchMax=$batchMax pending=${records.size} videoEnd=$videoInputEnded audioEnd=$audioInputEnded")
val reportNow = System.nanoTime() / 1000
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"
val audio = playback?.diagnostics
val state = clockState
val reason = if (state == null) { if (waitingKey) "key" else "opening" }
else when (state[8].toInt()) {
0 -> "starved"; 1 -> "history"; 2 -> "cooldown"; 3 -> "reserve"; 4 -> "catchup"; else -> "invalid"
}
LogManager.videoDebug("stream PLAY group=${group.toULong().toString(16)} source=${source.toULong().toString(16)} " +
"epoch=${epoch.toULong().toString(16)} id=$id dt_ms=${(reportNow - reported) / 1000} generation=$generation " +
"TIME{position_ms=${ms(position, position >= 0)} ready_ms=${ms(ready, ready >= 0)} " +
"reserve_ms=${ms(ready - position, ready >= 0 && position >= 0)} minimum_ms=${ms(state?.get(2))} " +
"guard_ms=${ms(state?.get(7))} tempo=${state?.get(3) ?: 1000} reason=$reason " +
"correction_ms=${ms(state?.get(9))} pending_gain_ms=${ms(state?.get(10))}} " +
"VIDEO{shown=${(decoded.shown - shownBase).coerceAtLeast(0)} drop=${(decoded.discarded - discardedBase).coerceAtLeast(0)} " +
"pts_ms=${ms(decoded.shownPts, decoded.shownPts >= 0)} " +
"av_ms=${ms(decoded.shownPts - position, decoded.shownPts >= 0 && position >= 0)} " +
"input=${records.videos} inflight=${decoded.inflight} decoder_lead_ms=${ms(decoded.decoderLeadUs, playbackStarted)} " +
"wait_key=$waitingKey} AUDIO{enabled=$audioEnabled pcm_ms=${ms(audio?.pcmFrames?.toLong()?.times(1000)?.div(48))} " +
"device_ms=${ms(playback?.leadUs, audioEnabled && playbackStarted)} " +
"underruns=${((audio?.underruns ?: 0) - underrunBase).coerceAtLeast(0)} " +
"silence_frames=${((audio?.silenceFrames ?: 0) - silenceBase).coerceAtLeast(0)}} " +
"waiting=${state?.get(4) ?: 0} ended=$ended")
LogManager.videoTrace("android audio worker id=$id audioInputs=$audioInputs audioPTS=$lastAudioPts " +
"needsAudio=$needsAudio ${playback?.description} workMaxUs=$workMaxUs batches=$batches " +
"batchMax=$batchMax pending=${records.size} videoEnd=$videoInputEnded audioEnd=$audioInputEnded")
reported = reportNow; shownBase = decoded.shown; discardedBase = decoded.discarded
underrunBase = audio?.underruns ?: 0; silenceBase = audio?.silenceFrames ?: 0
batches = 0; batchMax = 0; workMaxUs = 0
}
// PCM consumption, media и video state будят worker; без аудио часы идут по wall clock.
@@ -37,7 +37,7 @@ class StreamPreviewTarget internal constructor(val texture: SurfaceTexture) {
val surface = Surface(texture)
private val lease = StreamPreviewLease {
surface.release(); texture.release()
LogManager.addLog("DEBUG", "video", "direct preview surface released target=${System.identityHashCode(this)}")
LogManager.videoTrace("direct preview surface released target=${System.identityHashCode(this)}")
}
internal fun retain(): StreamPreviewTarget { lease.retain(); return this }
fun release() = lease.release()
@@ -1,13 +1,27 @@
package com.utun.chat.data
data class StreamBufferPeer(val peer: Long, val worst: Long, val best: Long, val subflows: Int)
data class StreamBufferPositions(val first: Long = 0, val head: Long = 0, val total: Int = 0,
val peers: List<StreamBufferPeer> = emptyList()) {
fun fraction(next: Long): Float = if (head <= first) 1f
else ((next.coerceIn(first, head) - first).toDouble() / (head - first)).toFloat()
}
/** Все поля в байтах; ACK — десять завершённых интервалов по 100ms, для всех соединений ядра. */
data class StreamTxStats(val id: Long, val sampleUs: Long, val videoBytes: Long, val audioBytes: Long,
val streamBytes: Long, val transportBytes: Long, val inflightBytes: Long,
val acked1s: Long, val totalBytes: Long, val connections: Long) {
val acked1s: Long, val totalBytes: Long, val connections: Long, val buffer: StreamBufferPositions) {
companion object {
fun fromNative(id: Long, data: LongArray): StreamTxStats {
require(data.size == 9 && data.all { it >= 0 })
return StreamTxStats(id, data[0], data[1], data[2], data[3], data[4], data[5], data[6], data[7], data[8])
require(data.size >= 13 && data.take(13).all { it >= 0 } && data[12] <= 16)
require(data.size == 13 + data[12].toInt() * 4 && data[9] <= data[10] && data[12] <= data[11])
val peers = List(data[12].toInt()) { i ->
val offset = 13 + i * 4
require(data[offset + 1] >= 0 && data[offset + 2] >= data[offset + 1] && data[offset + 3] in 1..255)
StreamBufferPeer(data[offset], data[offset + 1], data[offset + 2], data[offset + 3].toInt())
}
return StreamTxStats(id, data[0], data[1], data[2], data[3], data[4], data[5], data[6], data[7], data[8],
StreamBufferPositions(data[9], data[10], data[11].toInt(), peers))
}
}
}
@@ -132,7 +132,7 @@ internal class StreamVideoPlayback(
workMaxUs = maxOf(workMaxUs, now - started)
if (now - reported >= 1_000_000) {
reported = now
LogManager.addLog("DEBUG", "stream", "android video worker id=$id generation=$activeGeneration revision=$videoRevision " +
LogManager.videoTrace("android video worker id=$id generation=$activeGeneration revision=$videoRevision " +
"inputs=${video.inputCount} decoded=${video.outputCount} shown=${video.shown} discarded=${video.discarded} " +
"inflight=${video.inflight} decoderLeadUs=${video.decoderLeadUs} workMaxUs=$workMaxUs " +
"clock=$position shownPTS=$lastShownPts displayLagUs=${if (lastShownPts >= 0) position - lastShownPts else -1}")
@@ -0,0 +1,60 @@
package com.utun.chat.ui.screens
import androidx.compose.foundation.Canvas
import androidx.compose.foundation.layout.size
import androidx.compose.material3.ExperimentalMaterial3Api
import androidx.compose.material3.MaterialTheme
import androidx.compose.material3.PlainTooltip
import androidx.compose.material3.Text
import androidx.compose.material3.TooltipBox
import androidx.compose.material3.TooltipDefaults
import androidx.compose.material3.rememberTooltipState
import androidx.compose.runtime.Composable
import androidx.compose.ui.Modifier
import androidx.compose.ui.geometry.CornerRadius
import androidx.compose.ui.geometry.Offset
import androidx.compose.ui.geometry.Size
import androidx.compose.ui.graphics.Color
import androidx.compose.ui.res.stringResource
import androidx.compose.ui.semantics.contentDescription
import androidx.compose.ui.semantics.semantics
import androidx.compose.ui.unit.dp
import com.utun.chat.R
import com.utun.chat.data.StreamBufferPositions
/** Одна линия на получателя: диапазон худшей/лучшей позиции отправки в сохранённом окне источника. */
@OptIn(ExperimentalMaterial3Api::class)
@Composable
internal fun StreamBufferIndicator(positions: StreamBufferPositions) {
val description = stringResource(R.string.stream_buffer_indicator, positions.total)
val background = MaterialTheme.colorScheme.surfaceVariant.copy(alpha = 0.55f)
val marker = MaterialTheme.colorScheme.onSurface
val empty = MaterialTheme.colorScheme.onSurfaceVariant.copy(alpha = 0.3f)
val zones = listOf(Color(0xFFE15B55), Color(0xFFE1AD36), Color(0xFF40B878))
TooltipBox(positionProvider = TooltipDefaults.rememberPlainTooltipPositionProvider(),
tooltip = { PlainTooltip { Text(description) } }, state = rememberTooltipState()) {
Canvas(Modifier.size(96.dp, 48.dp).semantics { contentDescription = description }) {
drawRoundRect(background, cornerRadius = CornerRadius(12.dp.toPx()))
val left = 6.dp.toPx()
val width = size.width - left * 2
val height = size.height - 12.dp.toPx()
if (positions.peers.isEmpty()) {
drawLine(empty, Offset(left, size.height / 2), Offset(left + width, size.height / 2), 2.dp.toPx())
} else {
val pitch = height / positions.peers.size
val thickness = minOf(4.dp.toPx(), pitch * 0.65f)
positions.peers.forEachIndexed { index, peer ->
val y = 6.dp.toPx() + pitch * (index + 0.5f)
zones.forEachIndexed { zone, color ->
drawRect(color.copy(alpha = 0.4f), Offset(left + width * zone / 3, y - thickness / 2), Size(width / 3, thickness))
}
val worst = left + width * positions.fraction(peer.worst)
val best = left + width * positions.fraction(peer.best)
drawLine(marker.copy(alpha = 0.8f), Offset(worst, y), Offset(best, y), thickness)
drawLine(marker, Offset(worst, y - thickness), Offset(worst, y + thickness), 1.dp.toPx())
drawLine(marker, Offset(best, y - thickness), Offset(best, y + thickness), 1.dp.toPx())
}
}
}
}
}
@@ -66,6 +66,7 @@ import com.utun.chat.data.StreamCaptureSettings
import com.utun.chat.data.StreamOrientation
import com.utun.chat.data.StreamCamera
import com.utun.chat.data.StreamCameraModes
import com.utun.chat.data.StreamBufferPositions
import com.utun.chat.data.LensType
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.Dispatchers
@@ -115,6 +116,7 @@ private fun StreamPublishScreen(group: Long, channelName: String, onBack: () ->
val context = LocalContext.current
val displayDegrees = StreamDisplayDegrees()
val state by StreamController.state.collectAsState()
val txStats by StreamController.txStats.collectAsState()
val local = state?.takeIf { it.group == group }
var audio by remember { mutableStateOf(true) }
var config by remember { mutableStateOf<StreamCaptureConfig?>(null) }
@@ -165,6 +167,8 @@ private fun StreamPublishScreen(group: Long, channelName: String, onBack: () ->
}, navigationIcon = {
IconButton(onClick = onBack) { Icon(Icons.AutoMirrored.Filled.ArrowBack, stringResource(R.string.back)) }
}, actions = {
StreamBufferIndicator(txStats?.takeIf { it.id == ongoing?.id }?.buffer ?: StreamBufferPositions())
Spacer(Modifier.width(4.dp))
val action = if (ongoing == null) R.string.stream_start
else if (ongoing.phase == StreamPhase.STOPPING) R.string.stream_stopping else R.string.stream_stop
CaptureControlButton(stringResource(action),
@@ -524,7 +528,7 @@ internal class CameraPreviewView(context: Context, private val preview: StreamPr
LogManager.addLog("ERROR", "video", "direct preview transform failed view=${width}x$height format=$next"); return
}
setTransform(matrix)
LogManager.addLog("DEBUG", "video", "direct preview transform view=${width}x$height buffer=${next.width}x${next.height} " +
LogManager.videoTrace("direct preview transform view=${width}x$height buffer=${next.width}x${next.height} " +
"sensor=${next.camera.sensor} display=$displayDegrees textureRotation=${next.textureRotation(displayDegrees)} front=${next.camera.front}")
}
@@ -47,6 +47,7 @@
<string name="stream_bitrate_auto">Битрейт видео: Авто</string>
<string name="stream_bitrate_value">%1$.2f Мбит/с</string>
<string name="stream_tx_queues">Очередь видео: %1$.1f КиБ · Неподтверждено: %2$.1f КиБ</string>
<string name="stream_buffer_indicator">Буфер отправки: получателей — %1$d. Линия показывает худшую и лучшую позиции потоков. Слева — старые данные (красный), справа — прямой эфир (зелёный). До 16 получателей, самые отставшие сверху.</string>
<string name="stream_tx_total">Буфер отправки: %1$.1f КиБ · ACK / 1 с: %2$.1f КиБ (%3$.2f Мбит/с)</string>
<string name="stream_title">Трансляции</string>
<string name="stream_channel_title">Трансляция: %1$s</string>
@@ -47,6 +47,7 @@
<string name="stream_bitrate_auto">Video bitrate: Auto</string>
<string name="stream_bitrate_value">%1$.2f Mbps</string>
<string name="stream_tx_queues">Video queue: %1$.1f KiB · Unacknowledged: %2$.1f KiB</string>
<string name="stream_buffer_indicator">Send buffer: %1$d receivers. Each line spans the slowest to fastest substream. Left: oldest retained data (red); right: live (green). Up to 16 receivers, slowest first.</string>
<string name="stream_tx_total">Send buffer: %1$.1f KiB · ACK / 1 s: %2$.1f KiB (%3$.2f Mbps)</string>
<string name="stream_title">Live streams</string>
<string name="stream_channel_title">Stream: %1$s</string>
@@ -18,6 +18,19 @@
enum { STREAM_AUDIO_CHANNELS = 2, STREAM_AUDIO_FRAMES = 960, STREAM_AUDIO_SAMPLES = 1920 };
/* Kotlin использует тот же фильтр video и log hook, что и native-диагностика. */
JNIEXPORT void JNICALL Java_com_utun_chat_data_NativeLib_videoLog(JNIEnv *env, jobject obj, jboolean trace, jstring message) {
(void)obj;
debug_level_t level = trace ? DEBUG_LEVEL_TRACE : DEBUG_LEVEL_DEBUG;
if (!debug_should_output(level, DEBUG_CATEGORY_VIDEO)) return;
if (!message) { DEBUG_ERROR(DEBUG_CATEGORY_VIDEO, "video log missing message"); return; }
const char *text = (*env)->GetStringUTFChars(env, message, NULL);
if (!text) { DEBUG_ERROR(DEBUG_CATEGORY_VIDEO, "video log string access failed"); return; }
if (trace) { DEBUG_TRACE(DEBUG_CATEGORY_VIDEO, "%s", text); }
else { DEBUG_DEBUG(DEBUG_CATEGORY_VIDEO, "%s", text); }
(*env)->ReleaseStringUTFChars(env, message, text);
}
struct android_stream_opus {
opus_codec_encoder_t *encoder;
opus_codec_decoder_t *decoder;
@@ -71,10 +84,17 @@ JNIEXPORT jboolean JNICALL Java_com_utun_chat_data_NativeLib_nativeStreamSetAudi
JNIEXPORT jlongArray JNICALL Java_com_utun_chat_data_NativeLib_nativeStreamTxStats(JNIEnv *env, jobject obj, jlong id) {
(void)obj; struct android_stream_tx_stats s;
if (android_stream_tx_snapshot((uint64_t)id, &s) < 0) return NULL;
jlong values[] = {s.sample_us, s.video_bytes, s.audio_bytes, s.stream_bytes, s.transport_bytes,
s.inflight_bytes, s.acked_1s, s.total_bytes, s.connections};
jlongArray result = (*env)->NewLongArray(env, 9);
if (result) (*env)->SetLongArrayRegion(env, result, 0, 9, values);
jlong values[13 + STREAM_BUFFER_POSITION_PEERS * 4] = {s.sample_us, s.video_bytes, s.audio_bytes, s.stream_bytes, s.transport_bytes,
s.inflight_bytes, s.acked_1s, s.total_bytes, s.connections, s.buffer.first, s.buffer.head, s.buffer.total, s.buffer.count};
for (unsigned i = 0; i < s.buffer.count; i++) {
values[13 + i * 4] = s.buffer.peers[i].peer;
values[14 + i * 4] = s.buffer.peers[i].worst;
values[15 + i * 4] = s.buffer.peers[i].best;
values[16 + i * 4] = s.buffer.peers[i].subflows;
}
jsize count = 13 + s.buffer.count * 4;
jlongArray result = (*env)->NewLongArray(env, count);
if (result) (*env)->SetLongArrayRegion(env, result, 0, count, values);
else DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "JNI TX stats allocation failed");
return result;
}
@@ -205,9 +225,10 @@ JNIEXPORT jlongArray JNICALL Java_com_utun_chat_data_NativeLib_nativeStreamPlayb
if (stream_playback_clock_update(&p->clock, now, p->position, ready, guard, pending, starved, &p->state) < 0) return NULL;
jlong values[] = {p->position, p->state.reserve_us, p->state.minimum_us, p->state.tempo_milli, p->state.waiting,
p->audio ? stream_playback_audio_needs_input(p->audio) : 0,
p->audio ? (jlong)stream_playback_audio_written(p->audio) : 0};
jlongArray output = (*env)->NewLongArray(env, 7);
if (output) (*env)->SetLongArrayRegion(env, output, 0, 7, values);
p->audio ? (jlong)stream_playback_audio_written(p->audio) : 0,
p->state.guard_us, p->state.reason, p->state.correction_us, p->state.pending_gain_us, (jlong)p->clock.pauses};
jlongArray output = (*env)->NewLongArray(env, 12);
if (output) (*env)->SetLongArrayRegion(env, output, 0, 12, values);
else DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "JNI playback clock allocation failed");
return output;
}
@@ -104,7 +104,8 @@ int android_stream_tx_snapshot(uint64_t id, struct android_stream_tx_stats *out)
struct stream_publisher_status status;
int result = stream_publisher_snapshot(transmitter, id, &status);
if (!result) *out = (struct android_stream_tx_stats){status.id, status.sample_us, status.video_bytes, status.audio_bytes,
status.stream_bytes, status.transport_bytes, status.inflight_bytes, status.acked_1s, status.total_bytes, status.connections};
status.stream_bytes, status.transport_bytes, status.inflight_bytes, status.acked_1s, status.total_bytes,
status.connections, status.buffer};
pthread_mutex_unlock(&media_lock); return result;
}
@@ -12,6 +12,7 @@
#include <stddef.h>
#include <stdint.h>
#include "../../../src/stream/stream_buffer.h"
struct UTUN_INSTANCE;
struct stream_media_format;
@@ -21,6 +22,7 @@ struct android_stream_signal;
struct android_stream_tx_stats {
uint64_t id, sample_us, video_bytes, audio_bytes, stream_bytes, transport_bytes, inflight_bytes, acked_1s, total_bytes;
unsigned connections;
struct stream_buffer_positions buffer;
};
/* Снимок обновляется в uasync каждые 100ms; getter копирует его под bridge mutex. */
int android_stream_tx_snapshot(uint64_t id, struct android_stream_tx_stats *stats);
+6 -6
View File
@@ -208,7 +208,7 @@ bool LiveVideoEncoder::open(const LiveVideoConfig& config, Backend backend) {
params += ":ref=1:";
if (name == "libx264") params += "level=" + level;
else params += "level-idc=" + level + ":high-tier=0:hrd=0:info=0:scenecut=0";
DEBUG_DEBUG(DEBUG_CATEGORY_VIDEO, "live encoder level name=%s contract=%d parameter=%s",
DEBUG_TRACE(DEBUG_CATEGORY_VIDEO, "live encoder level name=%s contract=%d parameter=%s",
name.constData(), config.level, level.constData());
}
av_dict_set(&options, name == "libx264" ? "x264-params" : "x265-params", params.constData(), 0);
@@ -279,7 +279,7 @@ bool LiveVideoEncoder::open(const LiveVideoConfig& config, Backend backend) {
}
if (result < 0) {
m->error = ffError(result);
DEBUG_DEBUG(DEBUG_CATEGORY_VIDEO, "desktop encoder unavailable name=%s size=%dx%d error=%s",
DEBUG_TRACE(DEBUG_CATEGORY_VIDEO, "desktop encoder unavailable name=%s size=%dx%d error=%s",
name.constData(), config.width, config.height, qPrintable(m->error)); break;
}
if (!pass) {
@@ -308,7 +308,7 @@ bool LiveVideoEncoder::open(const LiveVideoConfig& config, Backend backend) {
name.constData(), (unsigned long long)pts, attempt);
return false;
}
if (constrained) DEBUG_DEBUG(DEBUG_CATEGORY_VIDEO,
if (constrained) DEBUG_TRACE(DEBUG_CATEGORY_VIDEO,
"call GPU probe name=%s pts=%llu pipeline_frames=%d time_base=%d/%d",
name.constData(), (unsigned long long)pts, attempt, c->time_base.num, c->time_base.den);
return true;
@@ -492,7 +492,7 @@ bool LiveVideoEncoder::encode(const QImage& image, uint64_t pts, bool key, QList
}
encoded.pts = m->timestamps.front().second; m->timestamps.pop_front();
}
if (key && (encoded.key || m->forcedKeys)) DEBUG_DEBUG(DEBUG_CATEGORY_VIDEO,
if (key && (encoded.key || m->forcedKeys)) DEBUG_TRACE(DEBUG_CATEGORY_VIDEO,
"desktop requested KEY name=%s input_PTS=%llu output_PTS=%llu key=%d flags=%x bytes=%d",
qPrintable(m->name), (unsigned long long)pts, (unsigned long long)encoded.pts, encoded.key, packet->flags, packet->size);
if (encoded.key) {
@@ -539,7 +539,7 @@ bool LiveVideoEncoder::encode(const QImage& image, uint64_t pts, bool key, QList
m->slowReported = false;
m->prepareMax = m->submitMax = m->drainMax = m->keyMax = m->deltaMax = 0;
}
if (key && output.isEmpty()) DEBUG_DEBUG(DEBUG_CATEGORY_VIDEO, "desktop requested KEY buffered name=%s PTS=%llu",
if (key && output.isEmpty()) DEBUG_TRACE(DEBUG_CATEGORY_VIDEO, "desktop requested KEY buffered name=%s PTS=%llu",
qPrintable(m->name), (unsigned long long)pts);
if (finish && !m->timestamps.empty()) {
m->error = "Encoder did not drain all accepted inputs";
@@ -582,7 +582,7 @@ bool LiveVideoEncoder::setBitrate(int bitrate) {
CODEC_LOG(DEBUG_LEVEL_INFO, DEBUG_CATEGORY_VIDEO, "desktop encoder bitrate updated in place name=libx264 bitrate=%d", bitrate); return true;
}
#endif
DEBUG_DEBUG(DEBUG_CATEGORY_VIDEO, "desktop bitrate needs encoder replacement name=%s", qPrintable(m->name)); return false;
DEBUG_TRACE(DEBUG_CATEGORY_VIDEO, "desktop bitrate needs encoder replacement name=%s", qPrintable(m->name)); return false;
}
bool LiveVideoEncoder::canSetMaxQp() const {
+1 -1
View File
@@ -1255,7 +1255,7 @@ void MessageList::prepareVideoThumbnail(const QString& path) {
QString output = path + ".jpg";
int rc = video_extract_thumbnail(path.toUtf8().constData(), output.toUtf8().constData(), 320);
if (rc) DEBUG_WARN(DEBUG_CATEGORY_VIDEO, "thumbnail failed path=%s error=%d", qPrintable(path), rc);
else DEBUG_DEBUG(DEBUG_CATEGORY_VIDEO, "thumbnail ready path=%s", qPrintable(output));
else DEBUG_TRACE(DEBUG_CATEGORY_VIDEO, "thumbnail ready path=%s", qPrintable(output));
QMetaObject::invokeMethod(dispatcher, [self, path]() {
if (!self) return;
self->m_thumbnailJobs.remove(path);
+2 -2
View File
@@ -88,7 +88,7 @@ void StreamPortalCapture::request(const QString& method, QVariantList args, QVar
options["handle_token"] = requestToken; args << options;
auto message = QDBusMessage::createMethodCall(service, path, screenCast, method); message.setArguments(args);
if (m->state == Impl::Starting) m->timer.stop(); else m->timer.start(30000); // Выбор пользователя не ограничен таймаутом.
DEBUG_DEBUG(DEBUG_CATEGORY_VIDEO, "Wayland capture request=%s path=%s", qPrintable(method), qPrintable(m->request));
DEBUG_TRACE(DEBUG_CATEGORY_VIDEO, "Wayland capture request=%s path=%s", qPrintable(method), qPrintable(m->request));
auto* watcher = new QDBusPendingCallWatcher(bus.asyncCall(message), this);
connect(watcher, &QDBusPendingCallWatcher::finished, this, [this, watcher, expected = m->request] {
QDBusPendingReply<QDBusObjectPath> reply = *watcher; watcher->deleteLater();
@@ -103,7 +103,7 @@ void StreamPortalCapture::response(uint code, const QVariantMap& results, const
auto bus = QDBusConnection::sessionBus();
bus.disconnect(service, m->request, requestInterface, "Response", this, SLOT(response(uint,QVariantMap,QDBusMessage)));
m->request.clear(); m->timer.stop();
DEBUG_DEBUG(DEBUG_CATEGORY_VIDEO, "Wayland capture response stage=%d code=%u", m->state, code);
DEBUG_TRACE(DEBUG_CATEGORY_VIDEO, "Wayland capture response stage=%d code=%u", m->state, code);
if (code) { fail(code == 1 ? "Screen selection cancelled" : QString("ScreenCast request failed: %1").arg(code)); return; }
if (m->state == Impl::Creating) {
if (results.value("session_handle").toString() != m->session) { fail("Unexpected ScreenCast session path"); return; }
+6 -6
View File
@@ -183,7 +183,7 @@ StreamPublishEngine::StreamPublishEngine(QObject* parent) : QObject(parent), m(s
m->previewDispatchMax = std::max(m->previewDispatchMax, monotonicUs() - dispatchAt); m->previewUpdates++;
}
if (tickAt - m->previewReportAt >= 1000000) {
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG,
DEBUG_TRACE(DEBUG_CATEGORY_VIDEO,
"desktop preview GUI id=%llu source_generation=%llu interval_us=%lld updates=%u tick_gap_max_us=%lld dispatch_max_us=%lld",
(unsigned long long)m->id, (unsigned long long)m->sourceGeneration, (long long)(tickAt - m->previewReportAt),
m->previewUpdates, (long long)m->previewGapMax, (long long)m->previewDispatchMax);
@@ -302,7 +302,7 @@ QList<StreamVideoSource> StreamPublishEngine::sources() {
StreamVideoSource source{StreamVideoSource::Camera, camera.description(), camera.id(), {}};
for (const auto& format : camera.videoFormats()) {
stream_video_add_camera_modes(source.modes, format.resolution(), format.minFrameRate(), format.maxFrameRate());
DEBUG_DEBUG(DEBUG_CATEGORY_VIDEO, "desktop camera capability device=%s size=%dx%d pixel=%d fps=%.3f..%.3f",
DEBUG_TRACE(DEBUG_CATEGORY_VIDEO, "desktop camera capability device=%s size=%dx%d pixel=%d fps=%.3f..%.3f",
camera.id().constData(), format.resolution().width(), format.resolution().height(), format.pixelFormat(),
format.minFrameRate(), format.maxFrameRate());
}
@@ -318,7 +318,7 @@ QList<StreamVideoSource> StreamPublishEngine::sources() {
QSize size = screen->size() * screen->devicePixelRatio();
sources.push_back({StreamVideoSource::Screen, QStringLiteral("%1: %2").arg(++number).arg(screen->name()),
screenId(screen), {}, size, {}, screen});
DEBUG_DEBUG(DEBUG_CATEGORY_VIDEO, "desktop screen capability source=%s logical=%dx%d scale=%.3f pixels=%dx%d",
DEBUG_TRACE(DEBUG_CATEGORY_VIDEO, "desktop screen capability source=%s logical=%dx%d scale=%.3f pixels=%dx%d",
qPrintable(screen->name()), screen->size().width(), screen->size().height(), screen->devicePixelRatio(), size.width(), size.height());
}
}
@@ -772,7 +772,7 @@ uint64_t StreamPublishEngine::softwareEncodingRequest() const {
void StreamPublishEngine::answerSoftwareEncoding(uint64_t generation, bool allow) {
{ std::lock_guard<std::mutex> guard(m->lock);
if (!m->running || !generation || m->softwareRequest != generation || m->generation != generation) {
DEBUG_DEBUG(DEBUG_CATEGORY_VIDEO, "desktop CPU response ignored generation=%llu", (unsigned long long)generation); return;
DEBUG_TRACE(DEBUG_CATEGORY_VIDEO, "desktop CPU response ignored generation=%llu", (unsigned long long)generation); return;
}
if (allow) { m->softwareCodec = m->wanted.video.codec; m->softwareRequest = 0; m->started = monotonicUs(); } }
DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "desktop CPU encoding decision generation=%llu allow=%d", (unsigned long long)generation, allow);
@@ -1060,7 +1060,7 @@ void StreamPublishEngine::videoLoop(StreamPublishConfig initial) {
(unsigned long long)qp.bitrate(), config.video.bitrate);
}
if (now - report >= 1000000) {
DEBUG_DEBUG(DEBUG_CATEGORY_VIDEO, "desktop capture generation=%llu encoded=%u skipped=%u frame_age=%lldus encoder=%s",
DEBUG_TRACE(DEBUG_CATEGORY_VIDEO, "desktop capture generation=%llu encoded=%u skipped=%u frame_age=%lldus encoder=%s",
(unsigned long long)generation, count, skipped, (long long)(screen ? 0 : now - m->origin - capture), qPrintable(encoder->name()));
report = now; count = skipped = 0;
}
@@ -1090,7 +1090,7 @@ void StreamPublishEngine::tick() {
if (bitrateElapsed >= 1000000) {
const uint64_t encodedBytes = m->encodedBytes.load();
m->videoBitrate = (encodedBytes - m->bitrateBytes) * 8000000 / bitrateElapsed;
DEBUG_DEBUG(DEBUG_CATEGORY_VIDEO, "desktop video bitrate id=%llu actual_bps=%llu target_bps=%d bytes=%llu interval_us=%lld "
DEBUG_TRACE(DEBUG_CATEGORY_VIDEO, "desktop video bitrate id=%llu actual_bps=%llu target_bps=%d bytes=%llu interval_us=%lld "
"last_key_bytes=%llu queue_bytes=%llu need_key=%d",
(unsigned long long)m->id, (unsigned long long)m->videoBitrate, config().video.bitrate,
(unsigned long long)(encodedBytes - m->bitrateBytes), (long long)bitrateElapsed,
+62 -2
View File
@@ -19,6 +19,7 @@
#include <QScopedValueRollback>
#include <QCloseEvent>
#include <QShortcut>
#include <algorithm>
#ifdef HAVE_STREAM_CAPTURE
#include <QVideoWidget>
#include <QVideoSink>
@@ -30,6 +31,60 @@ constexpr int MaximumManualBitrate = 12000000;
QString text(const char* value) { return QCoreApplication::translate("ChatGui", value); }
QString bytes(uint64_t value) { return QString::number(value / 1024.0, 'f', 1) + " KiB"; }
}
/* Горизонталь — сохранённое окно источника, одна линия объединяет субпотоки получателя. */
class StreamBufferIndicator : public QWidget {
public:
explicit StreamBufferIndicator(QWidget* parent) : QWidget(parent) {
setObjectName(QStringLiteral("streamBufferIndicator"));
setPositions({});
}
void setPositions(const stream_buffer_positions& positions) {
m_positions = positions;
QString description = text("Send buffer: %1 receivers. Left: oldest data; right: live. Slowest receivers first.")
.arg(positions.total);
for (unsigned i = 0; i < positions.count; i++) {
const auto& peer = positions.peers[i];
description += QStringLiteral("\n%1: %2–%3% (%4)").arg(peer.peer, 0, 16)
.arg(fraction(peer.worst) * 100, 0, 'f', 1).arg(fraction(peer.best) * 100, 0, 'f', 1).arg(peer.subflows);
}
setToolTip(description); setAccessibleName(description); update();
}
protected:
void paintEvent(QPaintEvent*) override {
QPainter painter(this); painter.setRenderHint(QPainter::Antialiasing);
QColor background = palette().window().color();
painter.setPen(Qt::NoPen); painter.setBrush(background.lightness() < 128 ? background.lighter(135) : background.darker(106));
painter.drawRoundedRect(rect(), 10, 10);
QRectF area = QRectF(rect()).adjusted(6, 6, -6, -6);
if (!m_positions.count) {
QColor color = palette().windowText().color(); color.setAlphaF(0.3);
painter.setPen(QPen(color, 2)); painter.drawLine(QPointF(area.left(), area.center().y()), QPointF(area.right(), area.center().y()));
return;
}
const QColor colors[] = {QColor(225, 91, 85, 160), QColor(225, 173, 54, 160), QColor(64, 184, 120, 160)};
double pitch = area.height() / m_positions.count, thickness = std::min(4.0, pitch * 0.65);
for (unsigned i = 0; i < m_positions.count; i++) {
double y = area.top() + pitch * (i + 0.5);
for (unsigned zone = 0; zone < 3; zone++)
painter.fillRect(QRectF(area.left() + area.width() * zone / 3, y - thickness / 2, area.width() / 3, thickness), colors[zone]);
double worst = area.left() + area.width() * fraction(m_positions.peers[i].worst);
double best = area.left() + area.width() * fraction(m_positions.peers[i].best);
painter.setPen(QPen(palette().windowText(), thickness)); painter.drawLine(QPointF(worst, y), QPointF(best, y));
painter.setPen(QPen(palette().windowText(), 1));
painter.drawLine(QPointF(worst, y - thickness), QPointF(worst, y + thickness));
painter.drawLine(QPointF(best, y - thickness), QPointF(best, y + thickness));
}
}
private:
stream_buffer_positions m_positions{};
double fraction(uint64_t next) const {
if (m_positions.head <= m_positions.first) return 1;
next = std::clamp(next, m_positions.first, m_positions.head);
return double(next - m_positions.first) / double(m_positions.head - m_positions.first);
}
};
StreamPublishWindow::StreamPublishWindow(StreamPublishEngine* engine, uint64_t group, QWidget* parent, const QString& saved)
: QWidget(parent, Qt::Window), m_engine(engine), m_group(group) {
if (!saved.isEmpty()) {
@@ -42,6 +97,10 @@ StreamPublishWindow::StreamPublishWindow(StreamPublishEngine* engine, uint64_t g
auto* escape = new QShortcut(QKeySequence(Qt::Key_Escape), this); escape->setContext(Qt::WidgetWithChildrenShortcut);
connect(escape, &QShortcut::activated, this, &QWidget::close);
auto* layout = new QVBoxLayout(this);
auto* controls = new QHBoxLayout; controls->addStretch();
m_start = new QPushButton(text("Start stream"), this); m_start->setMinimumHeight(40);
m_buffer = new StreamBufferIndicator(this); m_buffer->setFixedSize(std::max(160, m_start->sizeHint().width() * 2), 40);
controls->addWidget(m_buffer); controls->addWidget(m_start); layout->addLayout(controls);
#ifdef HAVE_STREAM_CAPTURE
auto* preview = new QVideoWidget(this); m_preview = preview;
preview->setMinimumHeight(160); preview->setAspectRatioMode(Qt::KeepAspectRatio);
@@ -107,7 +166,6 @@ StreamPublishWindow::StreamPublishWindow(StreamPublishEngine* engine, uint64_t g
m_status = new QLabel(this); m_status->setWordWrap(true); layout->addWidget(m_status);
m_stats = new QLabel(this); m_stats->setObjectName(QStringLiteral("streamPublishStats"));
m_stats->setWordWrap(true); m_stats->setMinimumHeight(66); layout->addWidget(m_stats);
m_start = new QPushButton(text("Start stream"), this); layout->addWidget(m_start);
connect(m_start, &QPushButton::clicked, this, [this] {
if (!m_engine->active()) m_status->setProperty("streamError", QString());
if (m_engine->active()) m_engine->stop(); else m_engine->start(m_group, selection()); synchronize();
@@ -162,6 +220,7 @@ StreamPublishWindow::StreamPublishWindow(StreamPublishEngine* engine, uint64_t g
});
connect(engine, &StreamPublishEngine::audioChannelsChanged, this, [this](int channels) { m_inputChannels->setText(channels == 2 ? text("Stereo") : text("Mono")); });
connect(engine, &StreamPublishEngine::telemetryChanged, this, [this](const stream_publisher_status& stats) {
m_buffer->setPositions(stats.buffer);
m_stats->setText(text("Current video bitrate: %1 Mbit/s").arg(m_engine->videoBitrate() / 1000000.0, 0, 'f', 2) + '\n' +
(m_engine->videoMaxQp() >= 0 ? text("Current QP ceiling: %1").arg(m_engine->videoMaxQp()) + '\n' : QString()) +
text("Video queue: %1 · Inflight: %2 · Send buffer: %3\nACK, all connections, last second: %4")
@@ -302,7 +361,7 @@ void StreamPublishWindow::refreshModes(const StreamPublishConfig& previous, bool
"desktop video mode adjusted source=%s requested=%dx%d/%d selected=%dx%d/%d",
qPrintable(source.name), wanted.size.width(), wanted.size.height(), wanted.fps, mode.size.width(), mode.size.height(), mode.fps);
m_mode->setToolTip(source.kind == StreamVideoSource::Camera ? text("Frame rate is limited by the camera and the selected stream mode") : QString());
DEBUG_DEBUG(DEBUG_CATEGORY_VIDEO, "desktop video modes source=%s count=%lld selected=%dx%d/%d native=%d",
DEBUG_TRACE(DEBUG_CATEGORY_VIDEO, "desktop video modes source=%s count=%lld selected=%dx%d/%d native=%d",
qPrintable(source.name), (long long)m_modes.size(), mode.size.width(), mode.size.height(), mode.fps,
m_mode->currentData(Qt::UserRole + 1).toBool());
}
@@ -343,6 +402,7 @@ void StreamPublishWindow::apply() {
}
void StreamPublishWindow::synchronize() {
QScopedValueRollback<bool> updating(m_updating, true); bool active = m_engine->active();
if (!active) m_buffer->setPositions({});
m_start->setText(active ? text("Stop stream") : text("Start stream"));
m_start->setEnabled((active && m_engine->running()) || (!active && StreamPublishEngine::supported()
&& m_source->currentIndex() >= 0 && m_mode->currentIndex() >= 0 && m_codec->count()));
@@ -10,6 +10,7 @@ class QCheckBox;
class QButtonGroup;
class QSpinBox;
class QCloseEvent;
class StreamBufferIndicator;
class StreamPublishWindow : public QWidget {
Q_OBJECT
public:
@@ -30,6 +31,7 @@ private:
QCheckBox* m_adaptiveQp;
QLabel *m_status, *m_stats, *m_inputChannels;
QPushButton* m_start;
StreamBufferIndicator* m_buffer;
QButtonGroup* m_rotation;
QWidget* m_preview;
bool m_updating = false;
+1 -1
View File
@@ -24,7 +24,7 @@ StreamScreenCursor::StreamScreenCursor(QScreen* screen) : m_screen(screen), m_or
}
StreamScreenCursor::~StreamScreenCursor() {
DEBUG_DEBUG(DEBUG_CATEGORY_VIDEO, "desktop cursor closed ready=%d drawn=%d", m_ready, m_reported);
DEBUG_TRACE(DEBUG_CATEGORY_VIDEO, "desktop cursor closed ready=%d drawn=%d", m_ready, m_reported);
}
bool StreamScreenCursor::draw(QImage& frame) {
@@ -19,7 +19,7 @@ StreamScreenCursor::StreamScreenCursor(QScreen* screen) {
}
StreamScreenCursor::~StreamScreenCursor() {
DEBUG_DEBUG(DEBUG_CATEGORY_VIDEO, "desktop cursor closed ready=%d drawn=%d backend=Win32", m_ready, m_reported);
DEBUG_TRACE(DEBUG_CATEGORY_VIDEO, "desktop cursor closed ready=%d drawn=%d backend=Win32", m_ready, m_reported);
}
/* DrawIconEx работает с фоном кадра: это сохраняет и alpha, и AND/XOR системных курсоров. */
+39 -5
View File
@@ -71,7 +71,7 @@ struct StreamPlayerWindow::Playback {
stream_media_status media{};
int64_t position = -1, clockAt = 0, ready = -1, audioDacAt = 0, deviceLead = 0, lastShownPts = -1;
int64_t decoderLead = 0, framePeriod = 33333, lastInputPts = -1, lastDecodedPts = -1;
uint64_t audioFrameStart = 0, shown = 0, underruns = 0, endFrames = UINT64_MAX;
uint64_t audioFrameStart = 0, shown = 0, discarded = 0, underruns = 0, endFrames = UINT64_MAX;
bool started = false, presenting = false, videoReady = false, ended = false, videoEnded = false;
bool activeAudio = false, audioEnded = false, resync = false;
int64_t audioFedAt = 0, audioFedPts = -1, workerAt = 0, guiAt = 0, guiGapMax = 0;
@@ -194,6 +194,7 @@ void StreamPlayerWindow::tick() {
p.guiAt = now;
int64_t clock = p.clock(now);
while (p.started && !p.videos.empty() && p.videos.front().pts <= clock) {
if (!image.isNull()) p.discarded++;
p.videoBytes -= static_cast<size_t>(p.videos.front().image.sizeInBytes());
p.lastShownPts = p.videos.front().pts; image = std::move(p.videos.front().image); p.videos.pop_front();
}
@@ -252,6 +253,8 @@ void StreamPlayerWindow::decode() {
bool opened = false;
uint64_t reportedGaps = 0, observedGaps = 0;
int64_t deadline = nowUs() + 10000000, reportAt = nowUs(), gapReportAt = 0;
int64_t playReportAt = reportAt;
uint64_t shownBase = 0, discardedBase = 0, underrunBase = 0, silenceBase = 0;
uint64_t audioPackets = 0, videoPackets = 0, batches = 0, batchMax = 0;
int64_t feedMax = 0, pullMax = 0, workMax = 0;
auto fail = [&](const QString& reason, bool retry = true) {
@@ -342,7 +345,7 @@ void StreamPlayerWindow::decode() {
else { p.position = std::max(p.position, position); p.clockAt = now; tempo = p.timing.tempo_milli; }
}
if (began - reportAt >= 1000000) {
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG,
DEBUG_TRACE(DEBUG_CATEGORY_VIDEO,
"desktop viewer work id=%llu interval_us=%lld audio=%llu video=%llu batches=%llu batch_max=%llu "
"feed_max_us=%lld pull_max_us=%lld work_max_us=%lld PCM_frames=%u tempo=%d clock_us=%lld ready_us=%lld "
"reserve_us=%lld underruns=%llu silence_frames=%llu videoQ=%zu video_input=%zu video_busy=%d "
@@ -356,6 +359,37 @@ void StreamPlayerWindow::decode() {
reportAt = began; audioPackets = videoPackets = batches = batchMax = 0;
feedMax = pullMax = workMax = 0; p.guiGapMax = 0;
}
if (began - playReportAt >= 500000) {
if (debug_should_output(DEBUG_LEVEL_DEBUG, DEBUG_CATEGORY_VIDEO)) {
auto ms = [](int64_t us, bool valid = true) {
return valid ? QByteArray::number(us / 1000.0, 'f', 1) : QByteArray("na");
};
bool timing = p.started && p.adaptation.initialized;
DEBUG_DEBUG(DEBUG_CATEGORY_VIDEO,
"stream PLAY group=%llx source=%llx epoch=%llx id=%llu dt_ms=%lld generation=%llu "
"TIME{position_ms=%s ready_ms=%s reserve_ms=%s minimum_ms=%s guard_ms=%s tempo=%d reason=%s "
"correction_ms=%s pending_gain_ms=%s} "
"VIDEO{shown=%llu drop=%llu pts_ms=%s av_ms=%s queue=%zu input=%zu decoder_lead_ms=%s wait_key=%d} "
"AUDIO{enabled=%d pcm_ms=%.1f device_ms=%s underruns=%llu silence_frames=%llu} waiting=%d ended=%d",
(unsigned long long)p.group, (unsigned long long)p.source, (unsigned long long)p.epoch,
(unsigned long long)p.id, (long long)((began - playReportAt) / 1000), (unsigned long long)p.media.generation,
ms(p.position, p.started).constData(), ms(p.ready, p.ready >= 0).constData(),
ms(p.ready - p.position, p.started && p.ready >= 0).constData(), ms(p.timing.minimum_us, timing).constData(),
ms(p.timing.guard_us, timing).constData(), p.adaptation.tempo_milli ? p.adaptation.tempo_milli : 1000,
timing ? stream_playback_reason_name(p.timing.reason) : p.started ? "key" : "opening",
ms(p.timing.correction_us, timing).constData(), ms(p.timing.pending_gain_us, timing).constData(),
(unsigned long long)(p.shown - shownBase), (unsigned long long)(p.discarded - discardedBase),
ms(p.lastShownPts, p.lastShownPts >= 0).constData(),
ms(p.lastShownPts - position, p.lastShownPts >= 0 && position >= 0).constData(),
p.videos.size(), p.videoInput.size(), ms(p.decoderLead, p.videoReady).constData(), !p.videoReady,
p.activeAudio, p.sound ? stream_playback_audio_buffered(p.sound) / 48.0 : 0,
ms(p.deviceLead, p.activeAudio && p.audioDacAt).constData(),
(unsigned long long)(p.underruns - underrunBase),
(unsigned long long)(p.audioGap.silentFrames - silenceBase), timing && p.timing.waiting, p.ended);
}
playReportAt = began; shownBase = p.shown; discardedBase = p.discarded;
underrunBase = p.underruns; silenceBase = p.audioGap.silentFrames;
}
}
if (p.sound) stream_playback_audio_set_tempo(p.sound, tempo);
auto urgent = std::find_if(records.begin(), records.end(), [](const StreamRecord& r) {
@@ -464,7 +498,7 @@ void StreamPlayerWindow::decodeVideo() {
if (current != generation) {
avcodec_free_context(&decoder); av_packet_unref(packet); revision = 0; eosSent = false;
generation = current; progressAt = nowUs();
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "desktop video generation id=%llu generation=%llu",
DEBUG_TRACE(DEBUG_CATEGORY_VIDEO, "desktop video generation id=%llu generation=%llu",
(unsigned long long)p.id, (unsigned long long)generation);
}
if (videoDone) {
@@ -503,7 +537,7 @@ void StreamPlayerWindow::decodeVideo() {
(unsigned long long)p.id, (long long)pts, (long long)(nowUs() - p.createdAt),
(long long)(nowUs() - p.firstInputAt));
}
if (pts < p.clock(nowUs()) - 100000) { ++discarded; continue; }
if (pts < p.clock(nowUs()) - 100000) { ++discarded; ++p.discarded; continue; }
}
int64_t convertAt = nowUs();
converter = sws_getCachedContext(converter, frame->width, frame->height, (AVPixelFormat)frame->format,
@@ -620,7 +654,7 @@ void StreamPlayerWindow::decodeVideo() {
}
int64_t now = nowUs();
if (now - reportAt >= 1000000) {
DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG,
DEBUG_TRACE(DEBUG_CATEGORY_VIDEO,
"desktop video worker id=%llu interval_us=%lld decoded=%llu discarded=%llu codec_us=%lld/%lld convert_us=%lld/%lld",
(unsigned long long)p.id, (long long)(now - reportAt), (unsigned long long)decoded,
(unsigned long long)discarded, (long long)codecTotal, (long long)codecMax, (long long)convertTotal, (long long)convertMax);
+2 -2
View File
@@ -226,7 +226,7 @@ void Worker::run() {
DEBUG_ERROR(DEBUG_CATEGORY_VIDEO, "VideoEngine: seek to %lldms failed ret=%d",
(long long)targetMs, sret);
else
DEBUG_DEBUG(DEBUG_CATEGORY_VIDEO, "VideoEngine: seek to %lldms", (long long)targetMs);
DEBUG_TRACE(DEBUG_CATEGORY_VIDEO, "VideoEngine: seek to %lldms", (long long)targetMs);
avcodec_flush_buffers(vdec);
gen = epoch->loadRelaxed();
eof = false;
@@ -439,7 +439,7 @@ void VideoPlayerEngine::onAudioDecoded(std::shared_ptr<std::vector<int16_t>> pcm
m_audioSampleRate = 0;
DEBUG_WARN(DEBUG_CATEGORY_VIDEO, "VideoEngine: audio not playing, fallback to system clock");
} else {
DEBUG_DEBUG(DEBUG_CATEGORY_VIDEO, "VideoEngine: audio started %dHz %dch", sampleRate, channels, pcm);
DEBUG_TRACE(DEBUG_CATEGORY_VIDEO, "VideoEngine: audio started %dHz %dch", sampleRate, channels, pcm);
}
} else {
m_hasAudio = false;
+6 -6
View File
@@ -33,7 +33,7 @@ void VideoRenderWidget::setFrame(const QImage& img) {
m_frame = img;
if (m_zoomEnabled && changed) {
finishDrag(); m_scale = 0; m_offset = {}; constrainView("frame");
DEBUG_DEBUG(DEBUG_CATEGORY_VIDEO, "VideoView %p frame=%dx%d reset to fit scale=%.4f DPR=%.3f",
DEBUG_TRACE(DEBUG_CATEGORY_VIDEO, "VideoView %p frame=%dx%d reset to fit scale=%.4f DPR=%.3f",
static_cast<void*>(this), img.width(), img.height(), frameScale(), devicePixelRatioF());
}
update();
@@ -72,7 +72,7 @@ void VideoRenderWidget::constrainView(const char* reason) {
void VideoRenderWidget::finishDrag() {
if (!m_panning) return;
m_panning = false; setCursor(Qt::OpenHandCursor);
DEBUG_DEBUG(DEBUG_CATEGORY_VIDEO, "VideoView %p drag end offset=(%.2f,%.2f)",
DEBUG_TRACE(DEBUG_CATEGORY_VIDEO, "VideoView %p drag end offset=(%.2f,%.2f)",
static_cast<void*>(this), m_offset.x(), m_offset.y());
}
@@ -88,7 +88,7 @@ bool VideoRenderWidget::event(QEvent* event) {
#endif
) {
constrainView("screen"); update();
DEBUG_DEBUG(DEBUG_CATEGORY_VIDEO, "VideoView %p screen changed DPR=%.3f scale=%.4f",
DEBUG_TRACE(DEBUG_CATEGORY_VIDEO, "VideoView %p screen changed DPR=%.3f scale=%.4f",
static_cast<void*>(this), devicePixelRatioF(), frameScale());
}
return handled;
@@ -98,7 +98,7 @@ void VideoRenderWidget::resizeEvent(QResizeEvent* event) {
QWidget::resizeEvent(event);
if (!m_zoomEnabled) return;
constrainView("resize");
DEBUG_DEBUG(DEBUG_CATEGORY_VIDEO, "VideoView %p resize=%dx%d scale=%.4f offset=(%.2f,%.2f)",
DEBUG_TRACE(DEBUG_CATEGORY_VIDEO, "VideoView %p resize=%dx%d scale=%.4f offset=(%.2f,%.2f)",
static_cast<void*>(this), width(), height(), frameScale(), m_offset.x(), m_offset.y());
}
@@ -122,7 +122,7 @@ void VideoRenderWidget::wheelEvent(QWheelEvent* event) {
QPointF relative = position - QPointF(width() / 2.0, height() / 2.0);
m_offset += relative * (1 / next - 1 / previous);
constrainView("zoom"); update(); event->accept();
DEBUG_DEBUG(DEBUG_CATEGORY_VIDEO, "VideoView %p zoom delta=%d scale=%.4f->%.4f physical=%.4f anchor=(%.1f,%.1f) offset=(%.2f,%.2f)",
DEBUG_TRACE(DEBUG_CATEGORY_VIDEO, "VideoView %p zoom delta=%d scale=%.4f->%.4f physical=%.4f anchor=(%.1f,%.1f) offset=(%.2f,%.2f)",
static_cast<void*>(this), delta, previous, next, next * devicePixelRatioF(), position.x(), position.y(),
m_offset.x(), m_offset.y());
}
@@ -139,7 +139,7 @@ void VideoRenderWidget::mousePressEvent(QMouseEvent* event) {
m_lastMouse = event->localPos();
#endif
m_panning = true; setCursor(Qt::ClosedHandCursor); event->accept();
DEBUG_DEBUG(DEBUG_CATEGORY_VIDEO, "VideoView %p drag begin scale=%.4f offset=(%.2f,%.2f)",
DEBUG_TRACE(DEBUG_CATEGORY_VIDEO, "VideoView %p drag begin scale=%.4f offset=(%.2f,%.2f)",
static_cast<void*>(this), frameScale(), m_offset.x(), m_offset.y());
}