diff --git a/doc/stream.txt b/doc/stream.txt index 3a5a72e2..69c670ee 100644 --- a/doc/stream.txt +++ b/doc/stream.txt @@ -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. diff --git a/src/stream/stream.c b/src/stream/stream.c index 82133dbb..4236eedc 100644 --- a/src/stream/stream.c +++ b/src/stream/stream.c @@ -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; diff --git a/src/stream/stream.h b/src/stream/stream.h index 51678a7f..b433dfe2 100644 --- a/src/stream/stream.h +++ b/src/stream/stream.h @@ -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; }; diff --git a/src/stream/stream_load.h b/src/stream/stream_load.h index faef64ca..ca116381 100644 --- a/src/stream/stream_load.h +++ b/src/stream/stream_load.h @@ -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 { diff --git a/src/stream/stream_media_buffer.c b/src/stream/stream_media_buffer.c index 23b5606c..cfa08b22 100644 --- a/src/stream/stream_media_buffer.c +++ b/src/stream/stream_media_buffer.c @@ -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)) { diff --git a/src/stream/stream_media_buffer.h b/src/stream/stream_media_buffer.h index e7bd0ee0..9765036c 100644 --- a/src/stream/stream_media_buffer.h +++ b/src/stream/stream_media_buffer.h @@ -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); diff --git a/src/stream/stream_network.c b/src/stream/stream_network.c index 499f0de0..bb062ad4 100644 --- a/src/stream/stream_network.c +++ b/src/stream/stream_network.c @@ -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); } diff --git a/src/stream/stream_network.h b/src/stream/stream_network.h index e02d92b3..ec392f06 100644 --- a/src/stream/stream_network.h +++ b/src/stream/stream_network.h @@ -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-владением последовательно опросить остальные субпотоки. */ diff --git a/src/stream/stream_optimizer.c b/src/stream/stream_optimizer.c index 11ac27d3..ca62b1e6 100644 --- a/src/stream/stream_optimizer.c +++ b/src/stream/stream_optimizer.c @@ -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) { diff --git a/src/stream/stream_priv.h b/src/stream/stream_priv.h index f035962f..1bb7e3f8 100644 --- a/src/stream/stream_priv.h +++ b/src/stream/stream_priv.h @@ -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); diff --git a/src/stream/stream_wire.h b/src/stream/stream_wire.h index 347362b1..d7f1b9ab 100644 --- a/src/stream/stream_wire.h +++ b/src/stream/stream_wire.h @@ -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 diff --git a/tests/test_stream_media.c b/tests/test_stream_media.c index 4dbe4e30..99cca249 100644 --- a/tests/test_stream_media.c +++ b/tests/test_stream_media.c @@ -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); diff --git a/tests/test_stream_service.c b/tests/test_stream_service.c index 2067f999..f54b9084 100644 --- a/tests/test_stream_service.c +++ b/tests/test_stream_service.c @@ -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. */ diff --git a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamPlayerEngine.kt b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamPlayerEngine.kt index e0b91d7c..0c6bf76b 100644 --- a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamPlayerEngine.kt +++ b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamPlayerEngine.kt @@ -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) { diff --git a/tools/chatgui/src/streamplayer_window.cpp b/tools/chatgui/src/streamplayer_window.cpp index 190c2d4f..b8bb56fa 100644 --- a/tools/chatgui/src/streamplayer_window.cpp +++ b/tools/chatgui/src/streamplayer_window.cpp @@ -250,9 +250,8 @@ void StreamPlayerWindow::decode() { p.failed = p.complete = true; emit statusChanged(text("Live video playback requires FFmpeg")); #else std::deque 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(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 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."