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

Recover streams after five seconds and start from complete keys

Browse Source 
Preserve queued media under backpressure, advertise retained keyframe ranges in stream wire v6, and reopen stalled subflows without resetting playback.

Cover large fragmented keys, sustained slow delivery, missing subflows, buffer eviction and stale bootstrap snapshots in integration regressions.
This commit is contained in:
evgeny committed 2026-10-25 12:03:49 +01:00
1 parent 4a785c2598
commit c3cf86b9e0
15 files changed
+461 -160
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

doc/stream.txt
+62 -28
View File
@@ -412,9 +412,9 @@ Rate закрываемых выдач фиксируется по оценке
Если очередь опустела, запрос завершается сразу. По истечении 2 секунд отправитель завершает запрос без слепых закрытий.
Далее снова измеряется рост; автоматический выбор учитывает освобождаемую нагрузку.
При глобальном отставании сокращение выбирается по всем пирам модуля.
При возрасте ожидающих DATA >1 секунды все выдачи одной точной эпохи одному получателю переходят на общий следующий seq.
Устаревшие задания учитываются как X; GAP сохраняет самую раннюю границу нескольких пропусков до его отправки.
Принятые ETCP пакеты, кольцо и подписки сохраняются. Завершённый источник дочитывается без такого пропуска.
Возраст ожидающих DATA используется для диагностики перегрузки, но не разрешает пропуск данных.
Backpressure сохраняет курсоры и непрерывную цепочку кадров. GAP обозначает фактическое вытеснение/утрату диапазона.
Принятые ETCP пакеты и подписки сохраняются. Завершённый источник дочитывается до подписанного END.
13. Статусы через опции мембера и Merkle
@@ -602,8 +602,8 @@ Stop/destroy выполняются по правилам ядра, вне callb
Глобальный бюджет включается только для backlog нескольких пиров вместе с CPU/опозданием callbacks;
CPU отдельно не отключает выдачи, уже запрошенное сокращение повторно не учитывается.
Control FIFO ограничена 128 кодограммами по 1094 байта, возрастом 4с и отменой неподтверждённых запросов.
С 25.10 DATA старше 1с пропускаются согласованно для всех частей одной эпохи одному получателю, независимо от OVERLOAD.
Пропуск учитывается X и передаётся явным GAP, кольцо не стирается; после END ожидающие DATA дочитываются.
Удаление DATA только по возрасту отменено: оно разрезало KEY и вызывало постоянные resync при недостаточной полосе.
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-метрики.
@@ -671,20 +671,24 @@ Stop/destroy выполняются по правилам ядра, вне callb
Все stream-числа на wire — big-endian, без packed-структур и зависимости от host byte order.
Версия stream wire — 4; обратная совместимость не поддерживается.
Заголовок raw (52 байта, version=5): cmd:1, version:1, command:1, subblock:1,
Заголовок raw (52 байта, version=6): cmd:1, version:1, command:1, subblock:1,
group_id:8, source_node_id:8, source_epoch:8, sender_epoch:8, receiver_epoch:8, handle:8.
INFO_REQ/RESP согласуют sender/receiver epochs; receiver_epoch=0 используется INFO_REQ для обнаружения поколения и REVERSE_REQ.
Подписанный routed payload содержит тот же заголовок; его group_id сверяется с реальной группой router,
sender берётся из end-to-end router metadata. DATA через routed service отвергается.
INFO_REQ: probe_nonce:8, не резервирует номер SUBSCRIBE.
INFO_REQ: probe_nonce:8, want_media:1 (0/1), не резервирует номер SUBSCRIBE.
want_media=1 просит поставщика подготовить общий медиаиндекс; поиск KEY выполняется отложенно с ограниченным бюджетом.
INFO_RESP/ACCEPT: probe_nonce:8 (0 в ACCEPT), actual_seq:8, first_available:8, last_available:8, n:1, r:1, block_size:2,
metric_flags:1, lag_ms:8, source_rate:8, receive_rate:8, blocks:8, late_blocks:8, stops:8,
offer_flags:1, free_slots:2, cpu_basis_points:2, queue_bytes:8, progress_age_ms:8, path_rtt_ms:4,
path_count:1, node_ids:8*path_count — фиксированная часть 111 байт. В INFO actual_seq — последний доступный, в ACCEPT — старт.
key_seq:8, key_last_seq:8, path_count:1, node_ids:8*path_count — фиксированная часть 127 байт. В INFO actual_seq — последний доступный, в ACCEPT — старт.
metric_flags: bit0 — измерение известно, bit1 — source_rate известен, bit2 — receive_rate известен, bit3 — stopped.
offer_flags: bit0 — принимает новые выдачи, bit1 — RTT цепочки известен, bit2 — возраст прогресса известен,
bit3 — CPU известен, bit4 — кольцо непусто для этого субпотока. CPU — 0..10000 (сотые доли процента).
bit3 — CPU известен, bit4 — кольцо непусто для этого субпотока, bit5 — полный KEY сохранён для выдачи этой части.
При bit5=0 обе позиции KEY равны UINT64_MAX; при bit5=1 first_available <= key_seq <= key_last_seq <= last_available.
Каждая часть диапазона KEY проверяется на наличие; свежий незавершённый KEY не заменяет предыдущий полный.
CPU — 0..10000 (сотые доли процента).
Пустой источник: first/last=UINT64_MAX, bit4=0, INFO actual_seq=UINT64_MAX, ACCEPT actual_seq=0, путь {source}.
Курсор last+1 допустим только в ACCEPT непустого кольца; DATA/END проверяются по точной позиции и подписи источника.
SUBSCRIBE: requested_seq:8; REJECT: reason:1, для SUBSCRIBE дополнительно first_available:8,last_available:8,
@@ -1331,25 +1335,55 @@ Android — 96x48dp слева от Start/Stop в шапке, desktop — ряд
Показываются до 16 наиболее отставших получателей, порядок при равенстве — ID. Позиции вне окна прижимаются к краям.
Индикатор отражает чтение кольца для отправки; неподтверждённые транспортные данные учитываются отдельно в inflight.
Восстановление после смены Wi-Fi и временного застоя — 25.10.2026
Восстановление доставки и старт с полного KEY — 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с.
Отправитель не пропускает очередь по времени: 1-секундный catchup удалён. Возраст и рост очереди остаются метриками.
Последняя полная проверенная RS-пачка обновляет progress_ms общей сессии. Приход отдельных частей/дубликатов,
INFO/RENEW и уже объявленные GAP не являются таким прогрессом. Отсутствие полной пачки 5с запускает перезапрос, в том числе до первого DATA;
следующая попытка не раньше чем через 5с. Сохраняются сессия, принятые части, consumer/media cursors и decoder references.
Переоткрываются лишь зависшие выдачи, свежие INFO обновляют диапазоны и точки KEY поставщиков.
Счётчик selection.restarts означает попытки перезапроса, а не сбросы медиабуфера.
Автор публикует все n частей вместе: один проверенный INFO разрешает параллельное открытие частей,
каждая DATA-выдача всё равно проходит собственное INFO/ACCEPT. Реле проверяются отдельно для каждой части.
Повтор после транспортного сбоя уже работавшей выдачи — через 250мс, без ожидания 30-секундного цикла проб.
При отсутствии прогресса 1с restart возможен только при нехватке k работающих частей и известном более новом seq;
неизвестная граница UINT64_MAX и обычная пауза автора не являются свидетельством новых данных.
INFO и ACCEPT сообщают last_available и [key_seq,key_last_seq] последнего полного сохранённого KEY.
Медиаиндекс обновляет KEY только после всех фрагментов; позднее создание индекса ищет начало назад с бюджетом
и индексирует вперёд. Снимок использует уже известную точку, без повторного сканирования всего 32-МБ кольца.
INFO_REQ.want_media позволяет реле подготовить индекс без локального плеера; частичный relay не объявляет непроверенный KEY.
Начальная media-подписка выбирает KEY из INFO и открывает части параллельно с одной явной позиции.
Обычный непрозрачный stream сохраняет прежний старт с последней доступной пачки.
Проверка 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.
При восстановлении сначала запрашивается текущая недостающая пачка. Если свежие диапазоны/отказы поставщиков
и локальные части не позволяют собрать k, выбирается полный доступный KEY не ниже уже пройденной границы.
Отправитель не заменяет согласованную позицию самостоятельно; вытеснение между INFO и SUBSCRIBE возвращает EXPIRED.
Данные старых handles не примешиваются к новой попытке. Пропущенный диапазон выдаётся одним явным GAP;
новая цепочка начинает декодирование с KEY, уже выданные DATA/PTS не повторяются.
Desktop/Android не закрывают соединение по отдельному 10-секундному ожиданию KEY: отсутствие DATA обслуживает ядро.
Ошибка декодера, смена эпохи/членства, остановка сервиса и подписанный END сохраняют собственную обработку.
Новый KEY может занимать несколько пачек; продвижение этих пачек предотвращает ложный перезапрос во время сборки кадра.
Автор публикует все n частей вместе: один INFO разрешает параллельный выбор, каждое открытие проходит собственный INFO/ACCEPT.
Реле проверяются отдельно для каждой части. Подтверждённая поломка уже работавшего транспорта допускает повтор через 250мс.
Диагностика video:
- stream source offers ready — first/last, key/key_last, valid;
- stream recovery request — epoch, недостающий next, idle_ms, число переоткрытых выдач, preserve_media=1;
- stream recovery resumed — продвижение после попытки без сброса медиа;
- stream recovery KEY — пройденная граница, выбранный KEY и причина недоступности прежнего диапазона;
- stream GAP — точные first/next и направление; stream NET/PLAY — сводки каждые >=500мс.
Регрессии test_stream_service --recovery-only используют настоящий CHAT/ETCP, подписи и RS 8+0.
Медиатест создаёт 50 кадров/с по 6000 байт, KEY 55000 байт на нескольких пачках раз в 50 кадров.
Проверяются порядок, возрастающий PTS, длины и каждый байт после штатного media cursor; H.264-декодер здесь не запускается.
Сценарии: блокировка отправки 850мс и 2600мс без GAP; REINIT с задержкой stream-управления 100мс;
10с выдачи 60КБ/с без потери KEY; остановка 5600мс с одной попыткой после 5с; поздний viewer с общим KEY-стартом;
настоящее вытеснение 32-МБ source-кольца с переходом на KEY; зависание одной части при продолжающейся доставке семи других;
вытеснение KEY между INFO и SUBSCRIBE ещё до первого DATA с обновлением стартовой точки через 5с.
После коротких задержек сохраняются все кадры и состояние media; при фактической утрате диапазона требуется явный GAP.
test_stream_media отдельно проверяет пустой индекс, неполный KEY, сохранение предыдущего KEY, завершение, вытеснение и GAP.
Предыдущая проверка 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.
Новая реализация использует stream wire v6: обновляются отправитель, получатели и задействованные реле.
Обратная совместимость v5 не поддерживается. Desktop и Android собраны из utun2.
src/stream/stream.c
+25 -7
View File
@@ -34,7 +34,7 @@ struct stream_session {
void *delivery;
uint64_t next_seq;
uint64_t start_seq, last_seq, passed_before;
uint64_t restart_from, restart_last;
uint64_t restart_from, restart_last, progress_ms;
uint8_t end[STREAM_END_SIZE];
int end_valid;
unsigned owners, relay_owners, holds;
@@ -412,6 +412,7 @@ static void session_deliver(void *arg) {
int result = s->next_seq == UINT64_MAX ? STREAM_WAIT
: stream_buffer_decode(s->buffer, s->next_seq, s->output, stream_input_size(&s->descriptor), &length);
if (result == STREAM_OK) {
s->progress_ms = get_time_tb() / 10;
if (!s->decoded) {
s->decoded = 1;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream first batch reconstructed group=%llx source=%llx epoch=%llx seq=%llu k=%u signature=verified",
@@ -840,10 +841,8 @@ int stream_get_info(const struct STREAM_HANDLE *h, struct stream_info *info) {
return STREAM_OK;
}
/* Первый frontend создаёт индекс точной эпохи, остальные владельцы используют его без копии RS-кольца. */
int stream_media_enable(struct STREAM_HANDLE *h, unsigned tracks) {
if (!stream_handle_active(h)) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "media enable: detached handle"); return STREAM_ERR_STATE; }
struct stream_session *s = h->session;
/* Индекс общий у локальных readers и сетевых поставщиков; его построение никогда не блокирует INFO. */
static int session_media_prepare(struct stream_session *s) {
if (!s->media) {
int result = stream_media_store_create(&s->media, s->buffer, &s->descriptor);
if (result != STREAM_OK) return result;
@@ -853,8 +852,26 @@ int stream_media_enable(struct STREAM_HANDLE *h, unsigned tracks) {
s->media_seek = s->next_seq - 1; s->media_seeking = 1;
}
}
session_media_schedule(s); return STREAM_OK;
}
int stream_local_media_prepare(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key) {
struct stream_local local;
if (stream_local_get(inst, key, &local) != STREAM_OK) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "media supplier unavailable"); return STREAM_ERR_STATE;
}
struct stream_session *s = session_find(inst->stream, key->group_id, key->source_node_id);
return session_media_prepare(s);
}
/* Первый frontend создаёт индекс точной эпохи, остальные владельцы используют его без копии RS-кольца. */
int stream_media_enable(struct STREAM_HANDLE *h, unsigned tracks) {
if (!stream_handle_active(h)) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "media enable: detached handle"); return STREAM_ERR_STATE; }
struct stream_session *s = h->session;
int result = session_media_prepare(s);
if (result != STREAM_OK) return result;
if (tracks && !h->media_cursor) {
int result = stream_media_cursor_create(&h->media_cursor, s->media, tracks);
result = stream_media_cursor_create(&h->media_cursor, s->media, tracks);
if (result != STREAM_OK) return result;
session_radio_update(s);
}
@@ -1057,7 +1074,8 @@ int stream_local_get(struct UTUN_INSTANCE *inst, const struct stream_descriptor
if (!key || !local || !inst || !inst->stream || inst->stream->stopping) return STREAM_WAIT;
struct stream_session *s = session_find(inst->stream, key->group_id, key->source_node_id);
if (!s || s->descriptor.source_epoch != key->source_epoch) return STREAM_WAIT;
*local = (struct stream_local){.descriptor = s->descriptor, .buffer = s->buffer, .start = s->start_seq, .last = s->last_seq, .next = s->next_seq,
*local = (struct stream_local){.descriptor = s->descriptor, .buffer = s->buffer, .media = s->media,
.start = s->start_seq, .last = s->last_seq, .next = s->next_seq, .progress_ms = s->progress_ms,
.anchored = s->anchored, .decoded = s->decoded, .publisher = s->source != NULL, .finished = s->end_valid,
.relay_only = !s->source && s->owners == s->relay_owners, .end = s->end_valid ? s->end : NULL};
return STREAM_OK;
src/stream/stream.h
+1 -1
View File
@@ -58,7 +58,7 @@ struct stream_candidate {
struct stream_selection_settings { uint32_t migration_confirm_ms, minimum_hold_ms, score_hysteresis_ms; };
struct stream_selection_info {
uint64_t probes_started, cycles, migrations, failures, restarts;
uint64_t probes_started, cycles, migrations, failures, restarts; // restarts — попытки перезапроса после 5с без полной пачки
unsigned candidates, probes_active, subflows, overlaps, providers;
int automatic, protected_by_fec;
};
src/stream/stream_load.h
-1
View File
@@ -16,7 +16,6 @@ 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 {
src/stream/stream_media_buffer.c
+19 -4
View File
@@ -21,7 +21,7 @@ struct stream_media_store {
int configured, waiting_key, finished, failed;
struct stream_media_format format;
struct stream_media_video pending;
uint64_t pending_seq, key_seq, last_frame_id;
uint64_t pending_seq, key_seq, key_last_seq, last_frame_id;
uint32_t pending_used;
int64_t video_us, audio_us, key_us, last_audio_us;
uint64_t pressure_report_us, pending_started_us, key_completed_us;
@@ -111,7 +111,7 @@ int stream_media_store_create(struct stream_media_store **out, struct stream_buf
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "media index allocation failed slots=%u", s->capacity); u_free(s); return STREAM_ERR_MEMORY;
}
s->buffer = buffer; s->descriptor = *d; s->generation = 1; s->waiting_key = 1;
s->video_us = s->audio_us = s->key_us = s->last_audio_us = -1; s->key_seq = UINT64_MAX;
s->video_us = s->audio_us = s->key_us = s->last_audio_us = -1; s->key_seq = s->key_last_seq = UINT64_MAX;
*out = s;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "media index created source=%llx records=%u bytes=%zu shared_RS=1",
(unsigned long long)d->source_node_id, s->capacity, s->capacity * sizeof(*s->positions));
@@ -133,7 +133,7 @@ static int media_discontinuity(struct stream_media_store *s) {
s->failed = 1; return STREAM_ERR_FORMAT;
}
s->generation++; s->pending_used = 0; s->waiting_key = 1;
s->video_us = s->audio_us = s->key_us = s->last_audio_us = -1; s->key_seq = UINT64_MAX;
s->video_us = s->audio_us = s->key_us = s->last_audio_us = -1; s->key_seq = s->key_last_seq = UINT64_MAX;
return STREAM_OK;
}
@@ -213,7 +213,8 @@ int stream_media_store_feed(struct stream_media_store *s, uint64_t seq, const ui
if (s->pending_used != s->pending.frame_size) continue;
s->video_us = (int64_t)v->pts_us; s->last_frame_id = v->frame_id; s->waiting_key = 0; s->pending_used = 0;
if (v->flags & STREAM_MEDIA_KEY) {
s->key_seq = s->pending_seq; s->key_us = (int64_t)v->pts_us; s->key_completed_us = get_time_us();
s->key_seq = s->pending_seq; s->key_last_seq = seq;
s->key_us = (int64_t)v->pts_us; s->key_completed_us = get_time_us();
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM,
"media KEY indexed source=%llx epoch=%llx pts=%lld seq=%llu..%llu bytes=%u assemble_us=%llu",
(unsigned long long)s->descriptor.source_node_id, (unsigned long long)s->descriptor.source_epoch,
@@ -239,6 +240,20 @@ void stream_media_store_finish(struct stream_media_store *s) {
s->finished = 1;
}
/* Индекс уже нашёл последний полный KEY при сборке: незавершённый следующий кадр его не заменяет. */
int stream_media_store_recovery(struct stream_media_store *s, struct stream_media_recovery *out) {
if (!s || !out || s->failed) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "media recovery snapshot unavailable"); return STREAM_ERR_STATE;
}
stream_media_store_trim(s);
struct stream_media_recovery position = {.last_seq = UINT64_MAX, .key_seq = UINT64_MAX, .key_last_seq = UINT64_MAX};
if (s->count) position.last_seq = media_at(s, s->count - 1)->last;
if (s->key_seq != UINT64_MAX && media_find(s, s->key_seq) && media_find(s, s->key_last_seq)) {
position.key_seq = s->key_seq; position.key_last_seq = s->key_last_seq; position.key_valid = 1;
}
*out = position; return STREAM_OK;
}
/* Курсор хранит только собственные рабочие буферы, не копию накопленного медиапотока. */
int stream_media_cursor_create(struct stream_media_cursor **out, struct stream_media_store *s, unsigned tracks) {
if (!out || *out || !s || !(tracks & STREAM_MEDIA_TRACK_VIDEO) || (tracks & ~3u)) {
src/stream/stream_media_buffer.h
+8
View File
@@ -37,6 +37,12 @@ struct stream_media_status {
uint64_t generation;
int initialized, audio_enabled, finished, video_ended, audio_ended;
};
/* Снимок индекса одной эпохи: last_seq — последняя проиндексированная пачка,
* [key_seq, key_last_seq] — последний полный сохранённый KEY. Нет KEY: key_valid=0, позиции UINT64_MAX. */
struct stream_media_recovery {
uint64_t last_seq, key_seq, key_last_seq;
int key_valid;
};
int stream_media_store_create(struct stream_media_store **out, struct stream_buffer *buffer, const struct stream_descriptor *descriptor);
void stream_media_store_destroy(struct stream_media_store *store);
@@ -44,6 +50,8 @@ int stream_media_store_feed(struct stream_media_store *store, uint64_t seq, cons
void stream_media_store_gap(struct stream_media_store *store, uint64_t first, uint64_t last);
void stream_media_store_finish(struct stream_media_store *store);
void stream_media_store_trim(struct stream_media_store *store);
/* Снимок не читает/не копирует кадры и не закрепляет кольцо; STREAM_OK либо ошибка аргументов/store. */
int stream_media_store_recovery(struct stream_media_store *store, struct stream_media_recovery *position);
int stream_media_cursor_create(struct stream_media_cursor **out, struct stream_media_store *store, unsigned tracks);
void stream_media_cursor_destroy(struct stream_media_cursor *cursor);
void stream_media_cursor_reset(struct stream_media_cursor *cursor);
src/stream/stream_network.c
+33 -45
View File
@@ -72,8 +72,7 @@ struct STREAM_RECEIVE {
struct stream_metrics metrics, offered_metrics;
struct stream_load load;
uint64_t admitted_ms, pending_highest, probe, probe_sent_ms;
uint64_t gap_from;
int gap_pending, account_failed;
int account_failed;
};
struct stream_peer {
struct stream_peer *next;
@@ -526,13 +525,19 @@ static int parse_info(const struct stream_message *m, const struct stream_descri
return STREAM_ERR_FORMAT;
}
const uint8_t *offer = m->payload + STREAM_INFO_OFFER_OFF;
if ((offer[0] & ~31u) || !!(offer[0] & 16) == empty || stream_read_be(offer + 3, 2) > 10000) {
uint64_t key = stream_read_be(m->payload + STREAM_INFO_KEY_OFF, 8);
uint64_t key_last = stream_read_be(m->payload + STREAM_INFO_KEY_OFF + 8, 8);
int key_valid = !!(offer[0] & 32);
if ((offer[0] & ~63u) || !!(offer[0] & 16) == empty || stream_read_be(offer + 3, 2) > 10000
|| (key_valid && (empty || key > key_last || key_last == UINT64_MAX || key < first || key_last > last))
|| (!key_valid && (key != UINT64_MAX || key_last != UINT64_MAX))) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream invalid offer flags=%u cpu=%llu", offer[0],
(unsigned long long)stream_read_be(offer + 3, 2)); return STREAM_ERR_FORMAT;
}
*start = stream_read_be(m->payload + 8, 8);
info->path = path; info->first_available = stream_read_be(m->payload + 16, 8);
info->last_available = stream_read_be(m->payload + 24, 8); info->info_ms = get_time_tb() / 10;
info->key_seq = key; info->key_last_seq = key_last; info->key_valid = key_valid;
info->accepting = (offer[0] & 1) != 0; info->path_rtt_valid = (offer[0] & 2) != 0;
info->progress_valid = (offer[0] & 4) != 0; info->cpu_valid = (offer[0] & 8) != 0;
info->free_slots = (uint16_t)stream_read_be(offer + 1, 2); info->cpu_basis_points = (uint16_t)stream_read_be(offer + 3, 2);
@@ -548,6 +553,22 @@ static void offer_payload(struct stream_peer *p, struct stream_local *local, uin
unsigned slots = STREAM_PEER_FLOW_LIMIT - count;
if (slots > STREAM_FLOW_LIMIT - net->flow_count) slots = STREAM_FLOW_LIMIT - net->flow_count;
offer[0] = !p->pressure.active && !p->cooling ? 1 : 0;
stream_write_be(payload + STREAM_INFO_KEY_OFF, UINT64_MAX, 8);
stream_write_be(payload + STREAM_INFO_KEY_OFF + 8, UINT64_MAX, 8);
struct stream_media_recovery position;
if (local->media && stream_media_store_recovery(local->media, &position) == STREAM_OK && position.key_valid
&& position.key_seq >= stream_read_be(payload + 16, 8)
&& position.key_last_seq <= stream_read_be(payload + 24, 8)) {
int retained = 1;
for (uint64_t seq = position.key_seq; seq <= position.key_last_seq; seq++) {
const uint8_t *wire; size_t length; uint64_t actual;
if (stream_buffer_read(local->buffer, seq, sub, &wire, &length, &actual) != STREAM_OK) { retained = 0; break; }
}
if (retained) {
offer[0] |= 32; stream_write_be(payload + STREAM_INFO_KEY_OFF, position.key_seq, 8);
stream_write_be(payload + STREAM_INFO_KEY_OFF + 8, position.key_last_seq, 8);
}
}
if (stream_read_be(payload + 16, 8) != UINT64_MAX) offer[0] |= 16;
stream_write_be(offer + 1, slots, 2);
double cpu = net->cpu.valid ? net->cpu.average_percent : 0;
@@ -611,13 +632,6 @@ static int downstream_send(struct STREAM_RECEIVE *f) {
struct stream_peer *p = f->peer; struct stream_local local;
if (stream_local_get(p->network->inst, &f->descriptor, &local) != STREAM_OK) { flow_close(f, STREAM_REASON_STOPPED, 1); return 0; }
struct stream_message m = {.descriptor = f->descriptor, .handle = f->info.handle, .subblock = f->subblock};
if (f->gap_pending) {
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;
int result = f->info.next == UINT64_MAX ? STREAM_WAIT
: stream_buffer_read(local.buffer, f->info.next, f->subblock, &wire, &length, &actual);
@@ -668,10 +682,12 @@ static int downstream_send(struct STREAM_RECEIVE *f) {
static void prepare_requests(struct stream_peer *p) {
for (struct STREAM_RECEIVE *f = p->flows; f; f = f->next) {
if (!f->outgoing || f->info.state >= STREAM_RECEIVE_REQUESTED) continue;
struct stream_local local;
if (stream_local_get(p->network->inst, &f->descriptor, &local) != STREAM_OK) { flow_close(f, STREAM_REASON_STOPPED, 0); continue; }
struct stream_message m = {.descriptor = f->descriptor, .subblock = f->subblock};
if (!f->info_sent) {
m.command = STREAM_CMD_INFO_REQ;
uint8_t probe[8];
uint8_t probe[9] = {0}; probe[8] = local.media != NULL;
if (!f->probe) f->probe = new_epoch();
if (!f->probe) { flow_close(f, STREAM_REASON_PROTOCOL, 0); continue; }
stream_write_be(probe, f->probe, 8);
@@ -679,8 +695,6 @@ static void prepare_requests(struct stream_peer *p) {
}
if (!f->info_ready || f->info.probe_only || !p->other_epoch) continue;
uint64_t requested = f->requested;
struct stream_local local;
if (stream_local_get(p->network->inst, &f->descriptor, &local) != STREAM_OK) { flow_close(f, STREAM_REASON_STOPPED, 0); continue; }
if (requested == UINT64_MAX) {
if (local.anchored) requested = local.start;
else if (!f->bootstrap) continue;
@@ -841,11 +855,15 @@ static void dispatch(struct stream_network *net, uint64_t sender, const uint8_t
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream routed DATA/control forbidden command=%u", m.command); return;
}
if (m.command == STREAM_CMD_INFO_REQ && !routed) {
if (m.handle || m.length != 8 || !stream_read_be(m.payload, 8) || m.subblock >= local.descriptor.n) {
if (m.handle || m.length != 9 || !stream_read_be(m.payload, 8) || m.payload[8] > 1 || m.subblock >= local.descriptor.n) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream INFO_REQ malformed peer=%llx bytes=%zu handle=%llu sub=%u n=%u",
(unsigned long long)sender, m.length, (unsigned long long)m.handle, m.subblock, local.descriptor.n); return;
}
if (!authorized(net, sender, &m, &local)) return;
if (m.payload[8]) {
if (stream_local_media_prepare(net->inst, &local.descriptor) != STREAM_OK
|| stream_local_get(net->inst, &m.descriptor, &local) != STREAM_OK) return;
}
if (!p) p = peer_open(net, sender);
if (!p || (m.receiver_epoch && m.receiver_epoch != p->own_epoch)) return;
if (p->other_epoch && p->other_epoch != m.sender_epoch) {
@@ -1314,7 +1332,7 @@ static int receive_open(struct STREAM_HANDLE *subscription, uint64_t node, uint8
if (f->outgoing && !f->info.probe_only && f->info.state != STREAM_RECEIVE_CLOSED && same_stream(&f->descriptor, &d)) bootstrap = 0;
struct STREAM_RECEIVE *f = flow_new(p, &d, sub, 1);
if (!f) return STREAM_ERR_MEMORY;
f->owned = 1; f->requested = seq; f->bootstrap = bootstrap; f->info.probe_only = probe_only; *out = f; peer_schedule(p);
f->owned = 1; f->requested = f->info.start = seq; f->bootstrap = bootstrap; f->info.probe_only = probe_only; *out = f; peer_schedule(p);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream receive opened peer=%llx group=%llx source=%llx epoch=%llx sub=%u index=%llu"
" bootstrap=%d probe_only=%d transport_up=%d timeout_ms=%u",
(unsigned long long)node, (unsigned long long)d.group_id, (unsigned long long)d.source_node_id,
@@ -1874,34 +1892,6 @@ malformed:
(unsigned long long)p->node_id, m->length);
}
/* Один общий 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мс. */
static void pressure_tick(struct stream_network *net, uint64_t now) {
unsigned troubled = 0; uint64_t maximum_age = 0;
@@ -1981,8 +1971,6 @@ 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
&& flow_age(f, now) > STREAM_LOAD_CATCHUP_MS) expire_pending(f);
}
p->busy--; peer_sweep(p);
}
src/stream/stream_network.h
+2 -1
View File
@@ -27,10 +27,11 @@ struct stream_receive_info {
int metrics_valid, supplier_metrics_valid, rtt_valid, close_requested;
struct stream_path path;
uint64_t first_available, last_available, info_ms, info_order, supplier_queue_bytes, progress_age_ms;
uint64_t key_seq, key_last_seq; // Включительный диапазон последнего полного KEY, если key_valid=1
uint32_t path_rtt_ms;
uint16_t free_slots, cpu_basis_points;
uint8_t subblock;
int info_ready, probe_only, accepting, path_rtt_valid, progress_valid, cpu_valid;
int info_ready, probe_only, accepting, path_rtt_valid, progress_valid, cpu_valid, key_valid;
};
/* INFO без выделения remote handle и без DATA; close обязателен также после detach.
* После ready можно тем же NCD-владением последовательно опросить остальные субпотоки. */
src/stream/stream_optimizer.c
+116 -33
View File
@@ -14,7 +14,7 @@
#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_RECOVERY_STALL_MS 5000
#define STREAM_BOOTSTRAP_RELAY_MS 100
#define STREAM_AVAILABILITY_RETRY_MS 100
#define STREAM_AVAILABILITY_WAIT_MS 5000
@@ -25,8 +25,9 @@ static uint64_t problem_penalty(const struct stream_metric_snapshot *metrics, in
struct stream_offer {
struct stream_receive_info info;
uint64_t retry_ms;
uint64_t retry_ms, missing_seq;
unsigned failures;
int missing_valid; // Отказ POSITION/EXPIRED точной позиции до следующей попытки восстановления
};
struct stream_supplier {
struct stream_supplier *next;
@@ -35,7 +36,7 @@ struct stream_supplier {
struct STREAM_RECEIVE *probe;
struct stream_offer *offers;
unsigned probe_sub, failures;
int online, compatible, source, relay;
int online, compatible, source, relay, refresh;
};
struct stream_selection {
struct STREAM_RECEIVE *current, *replacement;
@@ -52,8 +53,8 @@ struct stream_optimizer {
struct stream_probe_schedule schedule;
void *startup_timer;
uint64_t started_ms, startup_peer;
uint64_t roster_ms, progress_ms, next_seq, diagnostic_ms, deficient_ms, relay_poll_ms;
int progress_valid, draining, startup_complete;
uint64_t roster_ms, recovery_ms, start_seq, diagnostic_ms, deficient_ms, relay_poll_ms;
int media, recovering, draining, startup_complete;
};
/* Точный срок гонки не зависит от фазы общего 100-мс таймера; selector остаётся отложенным. */
@@ -135,6 +136,8 @@ static int probe_start(struct stream_optimizer *o, struct STREAM_HANDLE *h, stru
}
static void probes_poll(struct stream_optimizer *o, struct STREAM_HANDLE *h, uint64_t now) {
struct stream_local local;
if (stream_local_get(o->inst, &o->descriptor, &local) != STREAM_OK) return;
for (struct stream_supplier *s = o->suppliers; s; s = s->next) {
if (!s->probe) continue;
struct stream_receive_info info; stream_receive_get_info(s->probe, &info);
@@ -164,6 +167,11 @@ 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;
if (local.media && ((!local.anchored && !info.key_valid && info.last_available != UINT64_MAX)
|| (o->recovering && !info.accepting))) {
if (!s->availability_until_ms) s->availability_until_ms = now + STREAM_AVAILABILITY_WAIT_MS;
if (now < s->availability_until_ms) s->availability_retry_ms = now + STREAM_AVAILABILITY_RETRY_MS;
}
/* Автор создаёт все n частей одной пачки и вытесняет их вместе. Его диапазон/путь общий;
* новые DATA-запросы всё равно проходят собственный INFO/ACCEPT, но открываются параллельно. */
if (s->source) {
@@ -171,9 +179,10 @@ static void probes_poll(struct stream_optimizer *o, struct STREAM_HANDLE *h, uin
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",
DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "stream source offers ready peer=%llx parts=%u range=%llu..%llu key=%llu..%llu valid=%d parallel=1",
(unsigned long long)s->node, o->descriptor.n,
(unsigned long long)info.first_available, (unsigned long long)info.last_available);
(unsigned long long)info.first_available, (unsigned long long)info.last_available,
(unsigned long long)info.key_seq, (unsigned long long)info.key_last_seq, info.key_valid);
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"
@@ -280,7 +289,9 @@ static struct stream_supplier *supplier_choose(struct stream_optimizer *o, unsig
|| (s->probe && s->probe_sub == sub)) continue;
struct stream_offer *offer = &s->offers[sub]; struct stream_receive_info *info = &offer->info;
if (now < offer->retry_ms || !info->info_ms || now - info->info_ms > STREAM_OFFER_TTL_MS || !info->accepting || !info->free_slots
|| (seq != UINT64_MAX && info->first_available != UINT64_MAX && seq < info->first_available)) continue;
|| (seq != UINT64_MAX && info->first_available != UINT64_MAX && seq < info->first_available)
|| (offer->missing_valid && seq == offer->missing_seq)) continue;
if (seq == UINT64_MAX && o->media && info->last_available != UINT64_MAX && !info->key_valid) continue;
if (!pass && o->descriptor.r && usage(o, s->node, sub) >= o->descriptor.r) continue;
if (!best) { best = s; continue; }
struct stream_receive_info *previous = &best->offers[sub].info;
@@ -318,6 +329,10 @@ static void receive_failed(struct stream_optimizer *o, struct STREAM_RECEIVE **h
struct stream_supplier *s = supplier_find(o, info.peer);
if (s && s->offers) {
struct stream_offer *offer = &s->offers[sub];
s->refresh = 1;
if ((info.reason == STREAM_REASON_POSITION || info.reason == STREAM_REASON_EXPIRED) && info.start != UINT64_MAX) {
offer->missing_valid = 1; offer->missing_seq = info.start;
}
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; }
@@ -350,7 +365,7 @@ int stream_optimizer_create(struct UTUN_INSTANCE *inst, const struct stream_desc
if (o) o->selected = u_calloc(d->n, sizeof(*o->selected));
if (!o || !o->selected) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream optimizer allocation failed"); if (o) u_free(o); return STREAM_ERR_MEMORY; }
o->inst = inst; o->descriptor = *d; o->settings = (struct stream_selection_settings){2000, 5000, 20};
o->started_ms = get_time_tb() / 10;
o->started_ms = get_time_tb() / 10; o->start_seq = UINT64_MAX;
o->startup_timer = uasync_set_timeout(inst->ua, STREAM_BOOTSTRAP_RELAY_MS * 10, o, startup_timeout, "stream_bootstrap");
if (!o->startup_timer) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream bootstrap timer allocation failed");
@@ -415,9 +430,67 @@ void stream_optimizer_refresh(struct stream_optimizer *o, uint64_t peer) {
(unsigned long long)o->descriptor.source_node_id, (unsigned long long)peer);
}
/* Перезапрос не меняет consumer/media cursor: уже принятые части и decoder references остаются пригодны. */
static void recovery_request(struct stream_optimizer *o, const struct stream_local *local, uint64_t now) {
unsigned closed = 0;
for (unsigned sub = 0; sub < o->descriptor.n; sub++) {
struct stream_selection *pick = &o->selected[sub];
struct STREAM_RECEIVE **handles[] = {&pick->current, &pick->replacement};
for (unsigned i = 0; i < 2; i++) if (*handles[i]) {
struct stream_receive_info info; stream_receive_get_info(*handles[i], &info);
if (!local->anchored || info.next <= local->next) {
struct stream_supplier *s = supplier_find(o, info.peer); if (s) s->refresh = 1;
stream_receive_close(*handles[i]); *handles[i] = NULL; closed++;
}
}
}
for (struct stream_supplier *s = o->suppliers; s; s = s->next) {
/* Обновляем автора/зависших поставщиков. Остальной обход и уже начатые INFO сохраняют свои сроки:
* перезапуск всего списка каждые 5с никогда не дал бы дойти до дальних кандидатов. */
if (!s->source && !s->refresh) continue;
s->refresh = 0; s->cycle = s->retry_ms = s->availability_retry_ms = s->availability_until_ms = 0;
if (s->offers) for (unsigned sub = 0; sub < o->descriptor.n; sub++) {
s->offers[sub].info.info_ms = 0; s->offers[sub].retry_ms = 0; s->offers[sub].missing_valid = 0;
}
}
if (!local->anchored) o->start_seq = UINT64_MAX;
o->schedule.next_ms = 0; o->recovering = 1; o->recovery_ms = now; o->info.restarts++;
DEBUG_WARN(DEBUG_CATEGORY_VIDEO, "stream recovery request source=%llx epoch=%llx next=%llu idle_ms=%llu closed=%u preserve_media=1",
(unsigned long long)o->descriptor.source_node_id, (unsigned long long)o->descriptor.source_epoch,
(unsigned long long)local->next, (unsigned long long)(now - (local->progress_ms ? local->progress_ms : o->started_ms)), closed);
}
/* KEY нужен лишь если свежие диапазоны уже не позволяют собрать k частей недостающей пачки.
* Позиция одна для всех выдач; неподтверждённый INFO не является медиа, подпись/KEY проверяются обычным ingest. */
static int recovery_key(struct stream_optimizer *o, const struct stream_local *local, uint64_t now, uint64_t *start) {
uint8_t available[RS_MAX_BLOCKS] = {0}; unsigned count = 0;
int expired = 0; uint64_t key = UINT64_MAX;
for (unsigned sub = 0; sub < o->descriptor.n; sub++) {
const uint8_t *wire; size_t length; uint64_t actual;
if (stream_buffer_read(local->buffer, local->next, sub, &wire, &length, &actual) == STREAM_OK) available[sub] = 1;
}
for (struct stream_supplier *s = o->suppliers; s; s = s->next) if (s->offers && s->online && s->compatible) {
for (unsigned sub = 0; sub < o->descriptor.n; sub++) {
struct stream_offer *offer = &s->offers[sub];
struct stream_receive_info *info = &offer->info;
if (!info->info_ms || info->info_ms < o->recovery_ms || now - info->info_ms > STREAM_OFFER_TTL_MS
|| info->first_available == UINT64_MAX) continue;
int missing = offer->missing_valid && offer->missing_seq == local->next;
if (info->first_available <= local->next && info->last_available >= local->next && !missing) available[sub] = 1;
if (info->first_available > local->next || missing) expired = 1;
if (info->accepting && info->key_valid && info->key_seq >= local->next
&& (key == UINT64_MAX || info->key_seq > key)) key = info->key_seq;
}
}
for (unsigned sub = 0; sub < o->descriptor.n; sub++) count += available[sub];
if (!expired || count >= (unsigned)(o->descriptor.n - o->descriptor.r) || key == UINT64_MAX) return 0;
*start = key; return 1;
}
void stream_optimizer_tick(struct stream_optimizer *o, struct STREAM_HANDLE *h, uint64_t now) {
struct stream_local local;
if (stream_local_get(o->inst, &o->descriptor, &local) != STREAM_OK) return;
o->media = local.media != NULL;
if (local.decoded && !o->startup_complete) {
o->startup_complete = 1;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream bootstrap complete group=%llx source=%llx epoch=%llx peer=%llx elapsed_ms=%llu start=%llu",
@@ -444,14 +517,13 @@ void stream_optimizer_tick(struct stream_optimizer *o, struct STREAM_HANDLE *h,
}
if (now >= o->roster_ms) roster_update(o, h, now);
probes_poll(o, h, now); probes_schedule(o, h, now);
unsigned busy = 0, received = 0;
unsigned busy = 0;
for (unsigned sub = 0; sub < o->descriptor.n; sub++) {
struct stream_selection *pick = &o->selected[sub]; struct stream_receive_info info;
if (pick->current) {
stream_receive_get_info(pick->current, &info);
if (info.state == STREAM_RECEIVE_CLOSED || !supplier_find(o, info.peer)) receive_failed(o, &pick->current, sub, now);
else {
if (info.blocks && info.metrics_valid && !info.metrics.stopped) received++;
busy++;
struct stream_supplier *supplier = supplier_find(o, info.peer);
if (supplier->offers && info.metrics_valid) {
@@ -476,45 +548,56 @@ void stream_optimizer_tick(struct stream_optimizer *o, struct STREAM_HANDLE *h,
(unsigned long long)replacement.blocks, (unsigned long long)replacement.next);
}
}
if (!o->progress_valid || o->next_seq != local.next) {
o->progress_valid = 1; o->next_seq = local.next; o->progress_ms = now;
uint64_t progress = local.progress_ms ? local.progress_ms : o->started_ms;
if (o->recovering && local.progress_ms > o->recovery_ms) {
o->recovering = 0;
DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "stream recovery resumed source=%llx next=%llu elapsed_ms=%llu preserve_media=1",
(unsigned long long)o->descriptor.source_node_id, (unsigned long long)local.next,
(unsigned long long)(now - o->recovery_ms));
}
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 != UINT64_MAX && offer->last_available > local.next) newer = 1;
}
if (newer) {
selections_close(o);
if (stream_local_restart(o->inst, &o->descriptor) == STREAM_OK) {
o->info.restarts++; o->progress_ms = now; busy = 0; stream_local_get(o->inst, &o->descriptor, &local);
}
if ((local.anchored || local.media) && now - progress >= STREAM_RECOVERY_STALL_MS
&& now - o->recovery_ms >= STREAM_RECOVERY_STALL_MS) {
recovery_request(o, &local, now);
busy = 0;
for (unsigned sub = 0; sub < o->descriptor.n; sub++) busy += o->selected[sub].current != NULL;
}
uint64_t key;
if (o->recovering && local.media && local.anchored && recovery_key(o, &local, now, &key)) {
selections_close(o);
if (stream_local_restart(o->inst, &o->descriptor) == STREAM_OK) {
o->start_seq = key; busy = 0; o->recovering = 0;
DEBUG_WARN(DEBUG_CATEGORY_VIDEO, "stream recovery KEY source=%llx floor=%llu key=%llu reason=unavailable",
(unsigned long long)o->descriptor.source_node_id, (unsigned long long)local.next, (unsigned long long)key);
stream_local_get(o->inst, &o->descriptor, &local);
}
}
int migrating = 0;
for (unsigned sub = 0; sub < o->descriptor.n; sub++) if (o->selected[sub].replacement) migrating = 1;
if (!local.anchored && !busy && !o->startup_complete) {
if (!local.anchored && !busy) {
o->startup_peer = 0;
unsigned sub = 0; struct stream_supplier *s = startup_choose(o, now, &sub);
if (s && stream_receive_open(h, s->node, (uint8_t)sub, UINT64_MAX, &o->selected[sub].current) == STREAM_OK) {
unsigned sub = 0;
struct stream_supplier *s = o->start_seq == UINT64_MAX ? startup_choose(o, now, &sub)
: supplier_choose(o, sub, o->start_seq, 0, now);
if (s && o->media && s->offers[sub].info.key_valid && o->start_seq == UINT64_MAX)
o->start_seq = s->offers[sub].info.key_seq;
if (s && stream_receive_open(h, s->node, (uint8_t)sub, o->start_seq, &o->selected[sub].current) == STREAM_OK) {
o->startup_peer = s->node; o->selected[sub].opened = 1; o->selected[sub].opened_ms = now; busy++;
if (o->startup_timer) { uasync_cancel_timeout(o->inst->ua, o->startup_timer); o->startup_timer = NULL; }
/* Остальные INFO нужны для недостающих частей: победитель определяет только общий DATA старт. */
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream bootstrap winner group=%llx source=%llx epoch=%llx peer=%llx sub=%u"
" order=%llu elapsed_ms=%llu relay=%d",
" order=%llu elapsed_ms=%llu relay=%d start=%llu head=%llu key=%llu..%llu",
(unsigned long long)o->descriptor.group_id, (unsigned long long)o->descriptor.source_node_id,
(unsigned long long)o->descriptor.source_epoch, (unsigned long long)s->node, sub,
(unsigned long long)s->offers[sub].info.info_order, (unsigned long long)(now - o->started_ms), s->relay);
(unsigned long long)s->offers[sub].info.info_order, (unsigned long long)(now - o->started_ms), s->relay,
(unsigned long long)o->start_seq, (unsigned long long)s->offers[sub].info.last_available,
(unsigned long long)s->offers[sub].info.key_seq, (unsigned long long)s->offers[sub].info.key_last_seq);
}
}
for (unsigned sub = 0; sub < o->descriptor.n; sub++) {
struct stream_selection *pick = &o->selected[sub]; struct stream_receive_info info = {0};
if (pick->replacement || (!local.anchored && busy)) continue;
if (pick->replacement || (!local.anchored && busy && o->start_seq == UINT64_MAX)) continue;
if (pick->current) stream_receive_get_info(pick->current, &info);
uint64_t seq = local.anchored ? (pick->opened ? local.next : local.start) : UINT64_MAX;
uint64_t seq = local.anchored ? (pick->opened ? local.next : local.start) : o->start_seq;
struct stream_supplier *s = supplier_choose(o, sub, seq, info.peer, now);
if (!s) { pick->better_ms = pick->better_node = 0; continue; }
if (!pick->current) {
src/stream/stream_priv.h
+4
View File
@@ -10,7 +10,9 @@
struct stream_local {
struct stream_descriptor descriptor;
struct stream_buffer *buffer;
struct stream_media_store *media; // Заимствованный индекс, NULL для непрозрачного байтового потока
uint64_t start, last, next;
uint64_t progress_ms; // Монотонное время последней полной проверенной пачки, не прихода отдельной RS-части
int anchored, decoded, publisher, finished, relay_only;
const uint8_t *end;
};
@@ -18,6 +20,8 @@ struct stream_local {
struct stream_network *stream_network_get(struct UTUN_INSTANCE *inst);
void stream_optimizers_tick(struct UTUN_INSTANCE *inst, uint64_t now_ms);
int stream_local_get(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key, struct stream_local *local);
/* По запросу media-получателя создать общий индекс поставщика; поиск истории выполняется отложенно с бюджетом. */
int stream_local_media_prepare(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key);
int stream_handle_context(struct STREAM_HANDLE *handle, struct UTUN_INSTANCE **inst, struct stream_descriptor *descriptor);
int stream_ingest_path(struct UTUN_INSTANCE *inst, const uint8_t *wire, size_t length, const struct stream_path *path);
int stream_ingest_end(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key, const uint8_t *wire, size_t length);
src/stream/stream_wire.h
+4 -3
View File
@@ -10,15 +10,16 @@
extern "C" {
#endif
#define STREAM_WIRE_VERSION 5
#define STREAM_WIRE_VERSION 6
#define STREAM_WIRE_HEADER 52
#define STREAM_PATH_MAX 32
#define STREAM_END_SIZE 106
#define STREAM_METRIC_WIRE_SIZE 49
#define STREAM_INFO_METRIC_OFF 36
#define STREAM_INFO_OFFER_OFF (STREAM_INFO_METRIC_OFF + STREAM_METRIC_WIRE_SIZE)
#define STREAM_INFO_PATH_OFF (STREAM_INFO_OFFER_OFF + 25)
#define STREAM_INFO_SIZE (STREAM_INFO_PATH_OFF + 1) /* probe:8, границы/FEC:28, метрики:49, предложение:25, count:1 */
#define STREAM_INFO_KEY_OFF (STREAM_INFO_OFFER_OFF + 25)
#define STREAM_INFO_PATH_OFF (STREAM_INFO_KEY_OFF + 16)
#define STREAM_INFO_SIZE (STREAM_INFO_PATH_OFF + 1) /* probe:8, границы/FEC:28, метрики:49, предложение:25, KEY:16, count:1 */
#define STREAM_OVERLOAD_MAX 64
#define STREAM_OVERLOAD_HEADER 18
tests/test_stream_media.c
+35 -1
View File
@@ -263,6 +263,40 @@ static void test_cursor_key_recovery(void) {
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "PASS cursor recovery: GAP, fragment discontinuity and expiry resume only from a complete KEY");
}
/* Объявлять можно только целый сохранённый KEY; незавершённый новый кадр не заменяет предыдущую точку старта. */
static void test_recovery_positions(void) {
struct media_fixture f; media_fixture_open(&f);
struct stream_media_recovery position;
CHECK(stream_media_store_recovery(f.store, &position) == STREAM_OK && !position.key_valid && position.last_seq == UINT64_MAX);
struct stream_media_block b = {.format = {64, 64, 0, STREAM_MEDIA_H264, 0, 1}, .video_count = 1};
b.video[0] = (struct stream_media_video){1, 0, 12, 0, 6, STREAM_MEDIA_KEY, expected_video};
CHECK(media_fixture_feed(&f, &b) == STREAM_OK);
CHECK(stream_media_store_recovery(f.store, &position) == STREAM_OK && !position.key_valid && position.last_seq == 0);
b.video[0].offset = 6; b.video[0].data = expected_video + 6;
CHECK(media_fixture_feed(&f, &b) == STREAM_OK);
CHECK(stream_media_store_recovery(f.store, &position) == STREAM_OK && position.key_valid);
CHECK(position.key_seq == 0 && position.key_last_seq == 1 && position.last_seq == 1);
b.video[0] = (struct stream_media_video){2, 40000, 12, 0, 6, STREAM_MEDIA_KEY, expected_video};
CHECK(media_fixture_feed(&f, &b) == STREAM_OK);
CHECK(stream_media_store_recovery(f.store, &position) == STREAM_OK && position.key_valid);
CHECK(position.key_seq == 0 && position.key_last_seq == 1 && position.last_seq == 2);
b.video[0].offset = 6; b.video[0].data = expected_video + 6;
CHECK(media_fixture_feed(&f, &b) == STREAM_OK);
CHECK(stream_media_store_recovery(f.store, &position) == STREAM_OK && position.key_seq == 2 && position.key_last_seq == 3);
for (unsigned i = 0; i < 48; i++) {
b.video[0] = (struct stream_media_video){3 + i, 80000 + i * 40000, 12, 0, 12, 0, expected_video};
CHECK(media_fixture_feed(&f, &b) == STREAM_OK);
}
CHECK(stream_media_store_recovery(f.store, &position) == STREAM_OK && !position.key_valid && position.last_seq == f.seq - 1);
b.video[0] = (struct stream_media_video){100, 4000000, 12, 0, 12, STREAM_MEDIA_KEY, expected_video};
CHECK(media_fixture_feed(&f, &b) == STREAM_OK);
CHECK(stream_media_store_recovery(f.store, &position) == STREAM_OK && position.key_valid && position.key_seq == f.seq - 1);
stream_media_store_gap(f.store, f.seq, f.seq);
CHECK(stream_media_store_recovery(f.store, &position) == STREAM_OK && !position.key_valid && position.last_seq == f.seq);
media_fixture_close(&f);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "PASS recovery positions: complete/fragmented KEY, retained previous KEY, eviction and GAP");
}
int main(int argc, char **argv) {
debug_config_init(); debug_set_level(DEBUG_LEVEL_INFO);
/* Усечения проверяются массово; ожидаемые отказы не загромождают диагностический лог. */
@@ -276,7 +310,7 @@ int main(int argc, char **argv) {
CHECK(argc == 1);
test_media(); test_batches(); test_stereo(); test_codecs();
debug_set_category_level(DEBUG_CATEGORY_STREAM, DEBUG_LEVEL_INFO);
test_revision_transition(); test_cursor_key_recovery(); test_format_conflicts();
test_revision_transition(); test_cursor_key_recovery(); test_format_conflicts(); test_recovery_positions();
}
CHECK(u_get_allocated_count() == before);
debug_set_category_level(DEBUG_CATEGORY_STREAM, DEBUG_LEVEL_INFO);
tests/test_stream_service.c
+145 -25
View File
@@ -812,48 +812,80 @@ struct recovery_test {
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;
struct ll_queue *pump_queue;
queue_callback_fn pump_callback;
void *pump_arg;
uint64_t pump_due_us, common_start;
unsigned pump_Bps, accepted, other_parts;
unsigned delay_ms, overloads, overload_closed, media_gaps, expired_rejections;
int paused, check_start, hold_part, hold_info;
};
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) {
int target = stream_wire_parse(entry->dgram, entry->len, &message) == STREAM_OK
&& message.descriptor.source_epoch == recovery->descriptor.source_epoch;
if (target) {
if (message.command == STREAM_CMD_OVERLOAD) recovery->overloads++;
if (message.command == STREAM_CMD_REJECT && message.length && message.payload[0] == STREAM_REASON_EXPIRED)
recovery->expired_rejections++;
if (message.command == STREAM_CMD_CLOSED && message.length && message.payload[0] == STREAM_REASON_OVERLOAD)
recovery->overload_closed++;
if (message.command == STREAM_CMD_ACCEPT && recovery->check_start) {
CHECK(stream_read_be(message.payload + 8, 8) == recovery->common_start);
if (++recovery->accepted == recovery->descriptor.n) recovery->check_start = 0;
}
if (message.command == STREAM_CMD_DATA && message.subblock) recovery->other_parts++;
}
if (!recovery->delay_ms && !recovery->head) { recovery->receive(conn, entry); return; }
int hold = target && ((recovery->hold_part && message.command == STREAM_CMD_DATA && !message.subblock)
|| (recovery->hold_info && message.command == STREAM_CMD_INFO_RESP));
if ((recovery->hold_part || recovery->hold_info) && !hold) { recovery->receive(conn, entry); return; }
if (!hold && !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 (hold) pending->due_ms = UINT64_MAX;
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. */
/* Полный access unit разбивается как в publisher: KEY 55КБ занимает несколько последовательных RS-пачек. */
static void recovery_publish(struct recovery_test *r, uint64_t now_us) {
uint8_t video[55000] = {0}, wire[16384]; size_t length = 0;
unsigned frame_size = r->sent % 50 ? 6000 : sizeof(video);
memset(video + 4, (uint8_t)r->sent, frame_size - 4);
struct stream_media_block block = {.format = {320, 180, 0, STREAM_MEDIA_H264, 0, 1}, .video_count = 1};
size_t fragment = stream_publish_capacity(r->publisher) - STREAM_MEDIA_HEADER_SIZE - STREAM_MEDIA_VIDEO_SIZE;
for (unsigned offset = 0; offset < frame_size;) {
unsigned take = frame_size - offset < fragment ? frame_size - offset : fragment;
block.video[0] = (struct stream_media_video){r->sent + 1, now_us - r->origin_us, frame_size, offset,
take, r->sent % 50 ? 0 : STREAM_MEDIA_KEY, video + offset};
CHECK(stream_media_encode(&block, wire, sizeof(wire), &length) == STREAM_OK);
CHECK(stream_publish_block(r->publisher, wire, length) == STREAM_OK); offset += take;
}
r->sent++; r->next_frame_us = now_us + 20000;
}
/* 50 кадров/с, KEY 55КБ на нескольких RS-пачках раз в секунду, n=8/r=0; payload проверяется после настоящего 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;
if (r->pump_queue && r->pump_queue->head && now_us >= r->pump_due_us) {
r->pump_due_us = now_us + (uint64_t)r->pump_queue->head->len * 1000000 / r->pump_Bps;
r->pump_callback(r->pump_queue, r->pump_arg);
}
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;
recovery_publish(r, now_us);
}
if (!r->viewer) { test_clock_poll(ua, 50); continue; }
for (unsigned work = 0; work < 32; work++) {
struct stream_media_packet packet;
int result = stream_media_read(r->viewer, -1, 1, -1, &packet);
@@ -861,7 +893,8 @@ static void recovery_run(struct recovery_test *r, unsigned duration_ms) {
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(result == STREAM_OK && packet.kind == STREAM_MEDIA_READ_VIDEO);
CHECK(packet.video.length == ((packet.video.frame_id - 1) % 50 ? 6000 : 55000));
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;
@@ -876,23 +909,64 @@ static void recovery_run(struct recovery_test *r, unsigned duration_ms) {
}
}
/* Сначала дожидаемся индекса всей опубликованной истории; fixed sleep зависит от нагрузки общего test runner. */
static struct stream_media_recovery recovery_position(struct recovery_test *r) {
struct stream_local local; struct stream_media_recovery position; struct stream_buffer_stats stats;
CHECK(stream_local_get(nodes[0], &r->descriptor, &local) == STREAM_OK && local.media);
CHECK(stream_buffer_get_stats(local.buffer, &stats) == STREAM_OK && stats.blocks);
uint64_t deadline = get_time_tb() + 100000;
do {
CHECK(stream_media_store_recovery(local.media, &position) == STREAM_OK);
if (position.last_seq == stats.highest_seq) break;
CHECK(get_time_tb() < deadline); test_clock_poll(ua, 100);
} while (1);
DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "recovery fixture indexed head=%llu key=%llu..%llu valid=%d frames=%llu",
(unsigned long long)position.last_seq, (unsigned long long)position.key_seq,
(unsigned long long)position.key_last_seq, position.key_valid, (unsigned long long)r->sent);
CHECK(position.key_valid); return position;
}
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++) {
for (unsigned scenario = 0; scenario < 9; scenario++) {
int wifi = scenario == 1;
struct recovery_test r = {0}; recovery = &r;
CHECK(stream_publish_open(nodes[0], GROUP1, NULL, &r.publisher) == STREAM_OK);
CHECK(stream_media_enable(r.publisher, 0) == 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);
r.origin_us = get_time_us(); r.next_frame_us = r.origin_us;
if (scenario == 5 || scenario == 8) {
recovery_run(&r, 2500);
/* Явный P после KEY: 2,5с под нагрузкой могут закончиться ровно на KEY. */
do { recovery_publish(&r, get_time_us()); } while ((r.sent - 1) % 50 == 0);
r.paused = 1;
struct stream_media_recovery position = recovery_position(&r);
CHECK(position.key_last_seq > position.key_seq && position.last_seq > position.key_last_seq);
r.check_start = scenario == 5; r.common_start = position.key_seq; r.hold_info = scenario == 8;
}
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);
if (scenario == 8) {
recovery_run(&r, 100); CHECK(r.head && !r.frames);
for (unsigned frame = 0; frame < 6000; frame++) {
recovery_publish(&r, get_time_us());
if (!(frame % 16)) test_clock_poll(ua, 0);
}
struct stream_media_recovery position = recovery_position(&r);
CHECK(position.key_seq > r.common_start);
r.common_start = position.key_seq; r.check_start = 1; r.hold_info = 0;
for (struct recovery_pending *pending = r.head; pending; pending = pending->next) pending->due_ms = get_time_tb() / 10;
recovery_run(&r, 6500); CHECK(r.accepted == 8 && r.frames && r.expired_rejections);
}
r.delay_ms = wifi ? 100 : 0;
recovery_run(&r, 3000); CHECK(r.frames >= (scenario == 5 || scenario == 8 ? 1 : 50));
if (scenario == 5) CHECK(r.accepted == 8 && !r.media_gaps);
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;
@@ -900,7 +974,43 @@ static void test_recovery_scenarios(void) {
if (wifi) {
etcp_conn_reinit(conn, "test Wi-Fi network switch");
recovery_run(&r, 3500);
} else {
} else if (scenario == 3) {
queue_set_callback(queue, NULL, NULL);
r.pump_queue = queue; r.pump_callback = callback; r.pump_arg = arg; r.pump_Bps = 60000;
recovery_run(&r, 10000); r.pump_queue = NULL;
struct stream_selection_info selection; CHECK(stream_selection_get_info(r.viewer, &selection) == STREAM_OK);
CHECK(!r.media_gaps && !selection.restarts && r.maximum_gap_ms < 3000);
queue_set_callback(queue, callback, arg); queue_resume_callback(queue); recovery_run(&r, 3500);
} else if (scenario == 4) {
queue_set_callback(queue, NULL, NULL); recovery_run(&r, 4900);
struct stream_selection_info selection; CHECK(stream_selection_get_info(r.viewer, &selection) == STREAM_OK);
CHECK(!selection.restarts && !r.media_gaps);
recovery_run(&r, 700);
CHECK(stream_selection_get_info(r.viewer, &selection) == STREAM_OK && selection.restarts == 1);
queue_set_callback(queue, callback, arg); queue_resume_callback(queue); recovery_run(&r, 3500);
} else if (scenario == 7) {
unsigned before = r.other_parts;
r.hold_part = 1; recovery_run(&r, 4900);
struct stream_selection_info selection; CHECK(stream_selection_get_info(r.viewer, &selection) == STREAM_OK);
CHECK(!selection.restarts && !r.media_gaps && r.other_parts > before + 100);
recovery_run(&r, 700);
CHECK(stream_selection_get_info(r.viewer, &selection) == STREAM_OK && selection.restarts == 1);
r.hold_part = 0;
for (struct recovery_pending *pending = r.head; pending; pending = pending->next) pending->due_ms = get_time_tb() / 10;
recovery_run(&r, 3500);
} else if (scenario == 6) {
queue_set_callback(queue, NULL, NULL);
/* Реальное вытеснение 32-МБ source-кольца; транспорт не доставляет старую позицию. */
for (unsigned frame = 0; frame < 6000; frame++) {
recovery_publish(&r, get_time_us());
if (!(frame % 16)) test_clock_poll(ua, 0);
}
r.paused = 1; recovery_run(&r, 6000);
struct stream_media_recovery position = recovery_position(&r);
r.common_start = position.key_seq; r.check_start = 1;
queue_set_callback(queue, callback, arg); queue_resume_callback(queue); recovery_run(&r, 3500);
CHECK(r.accepted == 8 && r.media_gaps);
} else if (scenario != 5 && scenario != 8) {
/* Очередь растёт 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);
@@ -909,16 +1019,25 @@ static void test_recovery_scenarios(void) {
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",
wifi ? "wifi-switch-control-delay100ms" : scenario == 2 ? "stalled-wifi-2600ms" : scenario == 3 ? "sustained-60KBps"
: scenario == 4 ? "stalled-wifi-5600ms" : scenario == 5 ? "late-viewer-complete-KEY"
: scenario == 6 ? "expired-source-KEY" : scenario == 7 ? "one-part-stalled-5600ms"
: scenario == 8 ? "KEY-evicted-after-INFO" : "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);
failed += r.overload_closed != 0 || r.maximum_gap_ms > (wifi ? 1800 : scenario == 2 ? 4000 : scenario == 4 || scenario == 7 ? 8000
: scenario == 5 || scenario == 8 ? 5000 : scenario == 6 ? 16000 : scenario == 3 ? 3000 : 1800)
|| (scenario != 5 && scenario != 6 && scenario != 8 && get_time_tb() / 10 - r.last_frame_ms >= 300);
CHECK(scenario == 6 ? r.media_gaps > 0 : r.media_gaps == 0);
CHECK(scenario == 6 || scenario == 8 ? selection.restarts >= 1 : selection.restarts == (scenario == 4 || scenario == 7 ? 1 : 0));
/* Пауза автора не является поломкой пути и не должна перезапускать приёмник. */
if (scenario == 8) {
r.paused = 0; recovery_run(&r, 500);
CHECK(stream_selection_get_info(r.viewer, &selection) == STREAM_OK);
}
uint64_t restarts = selection.restarts;
r.paused = 1; recovery_run(&r, 2000);
r.paused = 1; recovery_run(&r, scenario >= 5 ? 500 : 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);
@@ -928,7 +1047,8 @@ static void test_recovery_scenarios(void) {
}
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");
DEBUG_INFO(DEBUG_CATEGORY_STREAM,
"[PASS] recovery: large KEY, common start, transient/sustained congestion and exact 5s re-request without media reset");
}
/* Настоящий send_input_q намеренно держим непустым: короткий глитч не режет, устойчивый рост вызывает OVERLOAD. */
tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamPlayerEngine.kt
+3 -6
View File
@@ -90,7 +90,6 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv
var video: StreamVideoPlayback? = null
var waitingKey = true
var audioEnabled = false
var opened = false
var ended = false
var videoInputEnded = false
var audioInputEnded = false
@@ -113,7 +112,6 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv
var underrunBase = 0L
var silenceBase = 0L
var workMaxUs = 0L
var deadline = System.nanoTime() + 10_000_000_000
var clockState: LongArray? = null
fun observeStatus(meta: LongArray?) {
@@ -139,7 +137,7 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv
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
resetVideo()
if (request) check(NativeLib.streamResync(id)) { "Media resync refused" }
}
try {
@@ -180,7 +178,6 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv
1 -> {
if (revision != videoRevision) {
check(revision > videoRevision); resetVideo(); videoRevision = revision
deadline = System.nanoTime() + 10_000_000_000
}
val key = flags and 1 != 0
if (!waitingKey || key) {
@@ -203,7 +200,7 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv
if (result == -2) resetTimeline(true) else check(result >= 0) { "Opus feed failed=$result" }
}
3 -> { LogManager.addLog("INFO", "stream", "playback GAP id=$id"); resetTimeline(false) }
4 -> opened = true
4 -> Unit
5, 7, 8 -> {
if (kind != 8 && !videoInputEnded) { videoInputEnded = true; video.finish() }
if (kind != 7 && !audioInputEnded) { audioInputEnded = true; playback?.finish() }
@@ -224,7 +221,7 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv
}
}
if (!running) break
if (!ended && (!opened || waitingKey) && System.nanoTime() > deadline) error("Viewer open timeout")
// Ядро перезапрашивает DATA после 5с без прогресса; большой KEY может собираться дольше.
val decoded = video.state
if (!playbackStarted && decoded.firstOutputPts != null) playbackStarted = true
if (playbackStarted && playback != null && ready >= 0) {
tools/chatgui/src/streamplayer_window.cpp
+4 -5
View File
@@ -250,9 +250,8 @@ void StreamPlayerWindow::decode() {
p.failed = p.complete = true; emit statusChanged(text("Live video playback requires FFmpeg"));
#else
std::deque<StreamRecord> records;
bool opened = false;
uint64_t reportedGaps = 0, observedGaps = 0;
int64_t deadline = nowUs() + 10000000, reportAt = nowUs(), gapReportAt = 0;
int64_t 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;
@@ -280,7 +279,7 @@ void StreamPlayerWindow::decode() {
stream_playback_audio_destroy(previous); p.notify();
if (request && !gui_stream_resync(p.id)) DEBUG_WARN(DEBUG_CATEGORY_STREAM, "desktop playback resync refused id=%llu",
(unsigned long long)p.id);
deadline = nowUs() + 10000000; emit statusChanged(text("Waiting for a key frame…"));
emit statusChanged(text("Waiting for a key frame…"));
};
auto observe = [&] {
stream_media_status status{};
@@ -415,13 +414,13 @@ void StreamPlayerWindow::decode() {
pullMax = std::max(pullMax, nowUs() - at); workMax = std::max(workMax, nowUs() - began);
if (received) { ++batches; batchMax = std::max<uint64_t>(batchMax, records.size() - previous); continue; }
if (!gui_stream_ready()) { fail(text("Chat service is stopped")); break; }
if ((!opened || !p.started) && nowUs() > deadline) { fail(text("Stream connection timed out")); break; }
/* Отсутствие DATA восстанавливает ядро через 5с; сборка большого KEY не является ошибкой соединения. */
std::unique_lock<std::mutex> guard(p.wakeLock);
p.wake.wait_for(guard, std::chrono::milliseconds(50), [&] { return !p.running || p.failed || p.wakeVersion != version; });
continue;
}
StreamRecord record = std::move(*next); records.erase(next);
if (record.kind == StreamRecordKind::Opened) { opened = true; continue; }
if (record.kind == StreamRecordKind::Opened) continue;
if (record.kind == StreamRecordKind::Gap) { reset(false); continue; }
if (record.kind == StreamRecordKind::Closed) {
const char* reason = record.reason == STREAM_CLOSE_SOURCE_CHANGED ? "The stream restarted. Select a source again."