jeka/utun2
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Files
3a6ef25ca6f1ad145a575fc4c41b633cb8483af2
utun2/src/stream/stream_optimizer.c
T
evgeny c3cf86b9e0 Recover streams after five seconds and start from complete keys 
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.
2026-10-25 12:03:49 +01:00

713 lines
45 KiB
C
Raw Blame History

#include <string.h>
#include "../../lib/debug_config.h"
#include "../../lib/mem.h"
#include "../../lib/u_async.h"
#include "../utun_instance.h"
#include "../transport_layer/node_conn_direct.h"
#include "stream_optimizer.h"
#include "stream_priv.h"
#define STREAM_PROBE_BURST 32
#define STREAM_PROBE_PERIOD_MS 1000
#define STREAM_PROBE_CYCLE_MS 30000
#define STREAM_OFFER_TTL_MS 60000
#define STREAM_MIGRATION_TIMEOUT_MS 5000
#define STREAM_TRANSPORT_RETRY_MS 250
#define STREAM_RECOVERY_STALL_MS 5000
#define STREAM_BOOTSTRAP_RELAY_MS 100
#define STREAM_AVAILABILITY_RETRY_MS 100
#define STREAM_AVAILABILITY_WAIT_MS 5000
#define STREAM_PROBLEM_PENALTY_MS 1000
#define STREAM_HOP_PENALTY_MS 3
static uint64_t problem_penalty(const struct stream_metric_snapshot *metrics, int valid);
struct stream_offer {
struct stream_receive_info info;
uint64_t retry_ms, missing_seq;
unsigned failures;
int missing_valid; // Отказ POSITION/EXPIRED точной позиции до следующей попытки восстановления
};
struct stream_supplier {
struct stream_supplier *next;
uint64_t node, cycle, retry_ms, availability_retry_ms, availability_until_ms;
uint64_t local_epoch, peer_epoch;
struct STREAM_RECEIVE *probe;
struct stream_offer *offers;
unsigned probe_sub, failures;
int online, compatible, source, relay, refresh;
};
struct stream_selection {
struct STREAM_RECEIVE *current, *replacement;
uint64_t opened_ms, replacement_ms, better_ms, better_node;
int opened;
};
struct stream_optimizer {
struct UTUN_INSTANCE *inst;
struct stream_descriptor descriptor;
struct stream_supplier *suppliers;
struct stream_selection *selected;
struct stream_selection_settings settings;
struct stream_selection_info info;
struct stream_probe_schedule schedule;
void *startup_timer;
uint64_t started_ms, startup_peer;
uint64_t roster_ms, recovery_ms, start_seq, diagnostic_ms, deficient_ms, relay_poll_ms;
int media, recovering, draining, startup_complete;
};
/* Точный срок гонки не зависит от фазы общего 100-мс таймера; selector остаётся отложенным. */
static void startup_timeout(void *arg) {
struct stream_optimizer *o = arg; o->startup_timer = NULL;
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream bootstrap relay deadline group=%llx source=%llx epoch=%llx elapsed_ms=%llu delay_ms=%u",
(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)(get_time_tb() / 10 - o->started_ms),
STREAM_BOOTSTRAP_RELAY_MS);
stream_network_wake(stream_network_get(o->inst));
}
unsigned stream_probe_schedule_tick(struct stream_probe_schedule *s, uint64_t now, unsigned pending, unsigned active) {
if (!s) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream probe schedule missing"); return 0; }
unsigned budget = 1;
if (!s->cycle) { s->cycle = 1; s->started_ms = now; budget = STREAM_PROBE_BURST; }
else if (now < s->next_ms) return 0;
else if (!pending && !active && now - s->started_ms >= STREAM_PROBE_CYCLE_MS) { s->cycle++; s->started_ms = now; }
s->next_ms = now + STREAM_PROBE_PERIOD_MS; return budget;
}
static struct stream_supplier *supplier_find(struct stream_optimizer *o, uint64_t node) {
for (struct stream_supplier *s = o->suppliers; s; s = s->next) if (s->node == node) return s;
return NULL;
}
static uint64_t backoff(unsigned failures) {
return (uint64_t)STREAM_PROBE_CYCLE_MS << (failures > 4 ? 4 : failures - 1);
}
static void roster_update(struct stream_optimizer *o, struct STREAM_HANDLE *h, uint64_t now) {
struct stream_candidate *candidates = NULL; size_t count = 0;
if (stream_get_candidates(h, &candidates, &count) != STREAM_OK) return;
for (struct stream_supplier *s = o->suppliers; s; s = s->next) s->online = 0;
for (size_t i = 0; i < count; i++) {
struct stream_supplier *s = supplier_find(o, candidates[i].node_id);
if (!s) {
s = u_calloc(1, sizeof(*s));
if (!s) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream optimizer candidate allocation failed"); continue; }
s->node = candidates[i].node_id; s->next = o->suppliers; o->suppliers = s;
}
s->online = 1; s->source = candidates[i].source;
s->relay = stream_relay_confirmed(o->inst, &o->descriptor, s->node);
int compatible = node_conn_direct_compatible(o->inst, s->node);
if (compatible && !s->compatible && s->cycle != o->schedule.cycle && now >= s->retry_ms) {
o->schedule.next_ms = 0;
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream optimizer new supplier wakes probe group=%llx source=%llx peer=%llx",
(unsigned long long)o->descriptor.group_id, (unsigned long long)o->descriptor.source_node_id,
(unsigned long long)s->node);
}
s->compatible = compatible;
}
u_free(candidates);
struct stream_supplier **link = &o->suppliers;
while (*link) {
struct stream_supplier *s = *link;
if (s->online) { link = &s->next; continue; }
if (s->probe) stream_receive_close(s->probe);
*link = s->next; u_free(s->offers); u_free(s);
}
o->roster_ms = now + STREAM_PROBE_PERIOD_MS;
}
/* Один NCD owner пира переиспользуется всеми пробами/выдачами, INFO не открывает DATA. */
static int probe_start(struct stream_optimizer *o, struct STREAM_HANDLE *h, struct stream_supplier *s, uint64_t now) {
s->cycle = o->schedule.cycle; s->probe_sub = 0;
if (!s->offers) s->offers = u_calloc(o->descriptor.n, sizeof(*s->offers));
if (!s->offers) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream optimizer offers allocation failed"); return 0; }
int result = stream_probe_open(h, s->node, 0, &s->probe);
if (result != STREAM_OK) {
s->failures++; s->retry_ms = now + backoff(s->failures);
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream optimizer probe refused peer=%llx result=%d retry_ms=%llu",
(unsigned long long)s->node, result, (unsigned long long)s->retry_ms); return 0;
}
o->info.probes_started++;
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream optimizer probe peer=%llx cycle=%llu ordinal=%llu source=%d",
(unsigned long long)s->node, (unsigned long long)o->schedule.cycle,
(unsigned long long)o->info.probes_started, s->source); return 1;
}
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);
if (!info.info_ready && info.state != STREAM_RECEIVE_CLOSED) continue;
if (info.state == STREAM_RECEIVE_CLOSED) {
s->offers[s->probe_sub].info.info_ms = 0;
if (info.reason == STREAM_REASON_POSITION) {
if (!s->availability_until_ms) s->availability_until_ms = now + STREAM_AVAILABILITY_WAIT_MS;
if (now < s->availability_until_ms && !s->availability_retry_ms) {
s->availability_retry_ms = now + STREAM_AVAILABILITY_RETRY_MS;
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream supplier buffer pending source=%llx peer=%llx sub=%u retry_ms=%u",
(unsigned long long)o->descriptor.source_node_id, (unsigned long long)s->node,
s->probe_sub, STREAM_AVAILABILITY_RETRY_MS);
}
}
if (info.reason == STREAM_REASON_TRANSPORT || info.reason == STREAM_REASON_MEMBER || info.reason == STREAM_REASON_PROTOCOL) {
s->failures++;
s->retry_ms = now + (info.reason == STREAM_REASON_TRANSPORT && s->local_epoch
? STREAM_TRANSPORT_RETRY_MS : backoff(s->failures));
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream optimizer probe closed source=%llx peer=%llx sub=%u reason=%d"
" info_ready=%d local_epoch=%llx peer_epoch=%llx failures=%u retry_in_ms=%llu cycle=%llu",
(unsigned long long)o->descriptor.source_node_id, (unsigned long long)s->node, s->probe_sub, info.reason,
info.info_ready, (unsigned long long)info.local_epoch, (unsigned long long)info.peer_epoch, s->failures,
(unsigned long long)(s->retry_ms - now), (unsigned long long)s->cycle);
stream_receive_close(s->probe); s->probe = NULL; continue;
}
} 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) {
for (unsigned sub = 0; sub < o->descriptor.n; sub++) {
s->offers[sub].info = info; s->offers[sub].info.subblock = sub;
s->offers[sub].retry_ms = 0; s->offers[sub].failures = 0;
}
DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "stream source offers ready peer=%llx parts=%u range=%llu..%llu 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.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"
" rtt=%u/%d chain=%u/%d slots=%u accepting=%d lag=%llu stopped=%d age=%llu/%d queue=%llu penalty_ms=%llu score_ms=%llu",
(unsigned long long)s->node, s->probe_sub, (unsigned long long)info.first_available,
(unsigned long long)info.last_available, info.path.count, info.rtt_ms, info.rtt_valid,
info.path_rtt_ms, info.path_rtt_valid, info.free_slots, info.accepting,
(unsigned long long)info.supplier_metrics.lag_ms, info.supplier_metrics.stopped,
(unsigned long long)info.progress_age_ms, info.progress_valid, (unsigned long long)info.supplier_queue_bytes,
(unsigned long long)problem_penalty(&info.supplier_metrics, info.supplier_metrics_valid),
(unsigned long long)stream_supplier_score(&info, &info.supplier_metrics, info.supplier_metrics_valid));
}
unsigned next = s->probe_sub + 1;
if (next == o->descriptor.n) { stream_receive_close(s->probe); s->probe = NULL; continue; }
if (info.state == STREAM_RECEIVE_CLOSED) {
stream_receive_close(s->probe); s->probe = NULL;
if (stream_probe_open(h, s->node, (uint8_t)next, &s->probe) != STREAM_OK) {
s->failures++; s->retry_ms = now + backoff(s->failures); continue;
}
} else if (stream_probe_next(s->probe, (uint8_t)next) != STREAM_OK) continue;
s->probe_sub = next;
}
}
static void probes_schedule(struct stream_optimizer *o, struct STREAM_HANDLE *h, uint64_t now) {
unsigned pending = 0, active = 0;
for (struct stream_supplier *s = o->suppliers; s; s = s->next) {
if (s->retry_ms && now >= s->retry_ms) {
s->retry_ms = 0; s->cycle = 0; o->schedule.next_ms = 0;
}
if (!s->probe && s->availability_retry_ms && now >= s->availability_retry_ms) {
s->availability_retry_ms = 0;
if (s->compatible && now < s->availability_until_ms && now >= s->retry_ms) {
s->cycle = 0; o->schedule.next_ms = 0;
}
}
if (s->probe) active++;
if (now < s->retry_ms) s->cycle = o->schedule.cycle;
if (s->compatible && s->cycle != o->schedule.cycle) pending++;
}
uint64_t previous = o->schedule.cycle;
unsigned budget = stream_probe_schedule_tick(&o->schedule, now, pending, active);
o->info.cycles = o->schedule.cycle;
if (o->schedule.cycle != previous) DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream optimizer cycle=%llu",
(unsigned long long)o->schedule.cycle);
for (unsigned i = 0; i < budget; i++) {
struct stream_supplier *chosen = NULL;
for (struct stream_supplier *s = o->suppliers; s; s = s->next) {
if (!s->compatible || s->cycle == o->schedule.cycle || s->probe || now < s->retry_ms) continue;
if (!chosen || s->relay > chosen->relay || (s->relay == chosen->relay && (s->source > chosen->source
|| (s->source == chosen->source && s->node < chosen->node)))) chosen = s;
}
if (!chosen) break;
probe_start(o, h, chosen, now);
}
}
static unsigned usage(struct stream_optimizer *o, uint64_t node, unsigned except) {
unsigned count = 0;
for (unsigned sub = 0; sub < o->descriptor.n; sub++) {
struct stream_receive_info info;
if (sub == except || !o->selected[sub].current) continue;
stream_receive_get_info(o->selected[sub].current, &info);
if (info.state != STREAM_RECEIVE_CLOSED && info.peer == node) count++;
}
return count;
}
/* Доля поздних блоков задаёт штраф в мс; текущая остановка/lag дают максимум, история остановок добавляет 10 мс. */
static uint64_t problem_penalty(const struct stream_metric_snapshot *m, int valid) {
if (!valid) return 0;
if (m->stopped || m->lag_ms > STREAM_LATE_MS) return STREAM_PROBLEM_PENALTY_MS;
double fraction = m->blocks ? (double)m->late_blocks / (double)m->blocks : 0;
double penalty = fraction * STREAM_PROBLEM_PENALTY_MS + (m->stops ? 10 : 0);
return penalty >= STREAM_PROBLEM_PENALTY_MS ? STREAM_PROBLEM_PENALTY_MS : (uint64_t)penalty;
}
/* path.count включает автора и поставщика: с последним переходом к нам это число hops от автора до получателя. */
uint64_t stream_supplier_score(const struct stream_receive_info *info, const struct stream_metric_snapshot *metrics, int valid) {
if (!info || !metrics) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream supplier score invalid arguments"); return UINT64_MAX; }
if (!info->rtt_valid || !info->path_rtt_valid) return UINT64_MAX;
return (uint64_t)info->path_rtt_ms + 2 * (uint64_t)info->rtt_ms + STREAM_HOP_PENALTY_MS * info->path.count
+ problem_penalty(metrics, valid);
}
static unsigned shared_path(struct stream_optimizer *o, const struct stream_receive_info *offer, unsigned sub) {
unsigned shared = 0;
for (unsigned other = 0; other < o->descriptor.n; other++) {
struct stream_receive_info info;
if (other == sub || !o->selected[other].current) continue;
stream_receive_get_info(o->selected[other].current, &info);
for (unsigned i = 1; i < offer->path.count; i++)
for (unsigned j = 1; j < info.path.count; j++) if (offer->path.nodes[i] == info.path.nodes[j]) shared++;
}
return shared;
}
/* Доступность позиции/путь уже проверены INFO. Ограничение r ослабляется только при отсутствии подходящей альтернативы. */
static struct stream_supplier *supplier_choose(struct stream_optimizer *o, unsigned sub, uint64_t seq, uint64_t exclude, uint64_t now) {
struct stream_supplier *best = NULL;
for (unsigned pass = 0; pass < 2 && !best; pass++) {
for (struct stream_supplier *s = o->suppliers; s; s = s->next) {
if (!s->online || !s->compatible || s->node == exclude || !s->offers || now < s->retry_ms
|| (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)
|| (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;
if (!o->startup_complete && seq == UINT64_MAX) {
if (info->info_order < previous->info_order) best = s;
continue;
}
uint64_t score = stream_supplier_score(info, &info->supplier_metrics, info->supplier_metrics_valid);
uint64_t old_score = stream_supplier_score(previous, &previous->supplier_metrics, previous->supplier_metrics_valid);
int better = score < old_score;
if (score == old_score) {
unsigned shared = shared_path(o, info, sub), old_shared = shared_path(o, previous, sub);
better = shared < old_shared || (shared == old_shared && s->node < best->node);
}
if (better) best = s;
}
}
return best;
}
/* Первый пригодный INFO среди всех субпотоков выигрывает гонку; порядок приёма не зависит от порядка списка пиров. */
static struct stream_supplier *startup_choose(struct stream_optimizer *o, uint64_t now, unsigned *subblock) {
struct stream_supplier *first = NULL; uint64_t order = UINT64_MAX;
for (unsigned sub = 0; sub < o->descriptor.n; sub++) {
struct stream_supplier *s = supplier_choose(o, sub, UINT64_MAX, 0, now);
if (s && s->offers[sub].info.info_order < order) {
first = s; order = s->offers[sub].info.info_order; *subblock = sub;
}
}
return first;
}
static void receive_failed(struct stream_optimizer *o, struct STREAM_RECEIVE **handle, unsigned sub, uint64_t now) {
struct stream_receive_info info; stream_receive_get_info(*handle, &info);
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; }
else {
offer->failures++;
offer->retry_ms = now + (info.reason == STREAM_REASON_TRANSPORT && info.blocks
? STREAM_TRANSPORT_RETRY_MS : backoff(offer->failures));
}
if (info.reason == STREAM_REASON_OVERLOAD) s->retry_ms = offer->retry_ms;
}
o->info.failures++;
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream optimizer supplier failed peer=%llx sub=%u reason=%d next=%llu",
(unsigned long long)info.peer, sub, info.reason, (unsigned long long)info.next);
stream_receive_close(*handle); *handle = NULL;
}
static int improved(struct stream_optimizer *o, unsigned sub, struct stream_supplier *s, const struct stream_receive_info *current) {
struct stream_receive_info *offer = &s->offers[sub].info;
uint64_t before = stream_supplier_score(current, &current->metrics, current->metrics_valid);
uint64_t after = stream_supplier_score(offer, &offer->supplier_metrics, offer->supplier_metrics_valid);
if (after < before && (before == UINT64_MAX || before - after >= o->settings.score_hysteresis_ms)) return 1;
return o->descriptor.r && usage(o, current->peer, sub) >= o->descriptor.r
&& usage(o, s->node, sub) < o->descriptor.r
&& (after <= before || after - before <= o->settings.score_hysteresis_ms);
}
int stream_optimizer_create(struct UTUN_INSTANCE *inst, const struct stream_descriptor *d, struct stream_optimizer **out) {
if (!inst || !d || !out || *out) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream optimizer create invalid arguments"); return STREAM_ERR_ARGUMENT; }
struct stream_optimizer *o = u_calloc(1, sizeof(*o));
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->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");
u_free(o->selected); u_free(o); return STREAM_ERR_MEMORY;
}
o->info.automatic = 1; *out = o; return STREAM_OK;
}
/* Освобождает основные и заменяющие выдачи; повторный вызов безопасен, статистика истории сохраняется. */
static unsigned selections_close(struct stream_optimizer *o) {
unsigned closed = 0;
for (unsigned sub = 0; sub < o->descriptor.n; sub++) {
struct stream_selection *pick = &o->selected[sub];
closed += (pick->current != NULL) + (pick->replacement != NULL);
stream_receive_close(pick->current); stream_receive_close(pick->replacement);
memset(pick, 0, sizeof(*pick));
}
o->info.subflows = o->info.overlaps = o->info.providers = o->info.protected_by_fec = 0;
return closed;
}
void stream_optimizer_destroy(struct stream_optimizer *o) {
if (!o) return;
if (o->startup_timer) uasync_cancel_timeout(o->inst->ua, o->startup_timer);
selections_close(o);
while (o->suppliers) {
struct stream_supplier *s = o->suppliers; o->suppliers = s->next;
stream_receive_close(s->probe); u_free(s->offers); u_free(s);
}
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream optimizer stopped source=%llx probes=%llu migrations=%llu failures=%llu restarts=%llu",
(unsigned long long)o->descriptor.source_node_id, (unsigned long long)o->info.probes_started,
(unsigned long long)o->info.migrations, (unsigned long long)o->info.failures, (unsigned long long)o->info.restarts);
u_free(o->selected); u_free(o);
}
int stream_optimizer_configure(struct stream_optimizer *o, const struct stream_selection_settings *settings) {
if (!o || !settings || settings->migration_confirm_ms > 60000 || settings->minimum_hold_ms > 60000
|| settings->score_hysteresis_ms > 60000) {
DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream optimizer settings invalid"); return STREAM_ERR_ARGUMENT;
}
o->settings = *settings; return STREAM_OK;
}
void stream_optimizer_info(struct stream_optimizer *o, struct stream_selection_info *info) { *info = o->info; }
/* Несколько изменений до одного прохода объединяются; только новый доступный поставщик ускоряет расписание пробы. */
void stream_optimizer_roster_changed(struct stream_optimizer *o, uint64_t peer) {
if (o->roster_ms) DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream optimizer roster invalidated group=%llx source=%llx peer=%llx",
(unsigned long long)o->descriptor.group_id, (unsigned long long)o->descriptor.source_node_id, (unsigned long long)peer);
o->roster_ms = 0;
}
/* READY отменяет ожидание старого цикла/ошибки только для подтверждённого поставщика. */
void stream_optimizer_refresh(struct stream_optimizer *o, uint64_t peer) {
o->roster_ms = o->schedule.next_ms = 0;
struct stream_supplier *s = supplier_find(o, peer);
if (s) {
s->cycle = s->retry_ms = 0; s->failures = 0;
if (s->offers) for (unsigned sub = 0; sub < o->descriptor.n; sub++) s->offers[sub].retry_ms = 0;
}
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream optimizer relay wake source=%llx peer=%llx",
(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",
(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)o->startup_peer,
(unsigned long long)(now - o->started_ms), (unsigned long long)local.start);
}
if (local.finished) {
/* END одного поставщика задаёт границу источника; остальные выдачи ещё могут требовать восстановления. */
if (local.last == UINT64_MAX || (local.anchored && local.next > local.last)) {
stream_relay_demand(o->inst, &o->descriptor, 0, now);
unsigned closed = selections_close(o);
for (struct stream_supplier *s = o->suppliers; s; s = s->next) { stream_receive_close(s->probe); s->probe = NULL; }
if (o->startup_timer) { uasync_cancel_timeout(o->inst->ua, o->startup_timer); o->startup_timer = NULL; }
if (closed || o->draining) DEBUG_INFO(DEBUG_CATEGORY_STREAM,
"stream optimizer END drained source=%llx next=%llu last=%llu released=%u probes=%u",
(unsigned long long)o->descriptor.source_node_id, (unsigned long long)local.next,
(unsigned long long)local.last, closed, o->info.probes_active);
o->draining = 0; o->info.probes_active = 0; return;
}
if (!o->draining) DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream optimizer draining END source=%llx next=%llu last=%llu",
(unsigned long long)o->descriptor.source_node_id, (unsigned long long)local.next, (unsigned long long)local.last);
o->draining = 1;
}
if (now >= o->roster_ms) roster_update(o, h, now);
probes_poll(o, h, now); probes_schedule(o, h, now);
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 {
busy++;
struct stream_supplier *supplier = supplier_find(o, info.peer);
if (supplier->offers && info.metrics_valid) {
supplier->offers[sub].info.supplier_metrics = info.metrics;
supplier->offers[sub].info.supplier_metrics_valid = 1;
supplier->offers[sub].info.last_available = info.next - 1;
}
}
}
if (!pick->replacement) continue;
struct stream_receive_info replacement; stream_receive_get_info(pick->replacement, &replacement);
if (replacement.state == STREAM_RECEIVE_CLOSED || !supplier_find(o, replacement.peer)
|| now - pick->replacement_ms >= STREAM_MIGRATION_TIMEOUT_MS) {
receive_failed(o, &pick->replacement, sub, now); continue;
}
if (replacement.blocks && (!pick->current || replacement.next >= info.next)) {
uint64_t previous = pick->current ? info.peer : 0;
stream_receive_close(pick->current); pick->current = pick->replacement; pick->replacement = NULL;
pick->opened_ms = now; pick->better_ms = pick->better_node = 0; o->info.migrations++;
DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "stream optimizer moved sub=%u from=%llx to=%llx verified_blocks=%llu next=%llu",
sub, (unsigned long long)previous, (unsigned long long)replacement.peer,
(unsigned long long)replacement.blocks, (unsigned long long)replacement.next);
}
}
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 || 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_peer = 0;
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 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)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 && 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) : 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) {
if (stream_receive_open(h, s->node, (uint8_t)sub, seq, &pick->current) == STREAM_OK) {
pick->opened_ms = now; pick->opened = 1; busy++;
struct stream_receive_info *offer = &s->offers[sub].info;
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream optimizer selected peer=%llx sub=%u index=%llu score_ms=%llu"
" rtt_ms=%u chain_ms=%u hops=%u penalty_ms=%llu",
(unsigned long long)s->node, sub, (unsigned long long)seq,
(unsigned long long)stream_supplier_score(offer, &offer->supplier_metrics, offer->supplier_metrics_valid),
offer->rtt_ms, offer->path_rtt_ms, offer->path.count,
(unsigned long long)problem_penalty(&offer->supplier_metrics, offer->supplier_metrics_valid));
}
continue;
}
if (info.state != STREAM_RECEIVE_ACTIVE || info.close_requested || migrating
|| now - pick->opened_ms < o->settings.minimum_hold_ms || !improved(o, sub, s, &info)) {
pick->better_ms = pick->better_node = 0; continue;
}
if (pick->better_node != s->node) { pick->better_node = s->node; pick->better_ms = now; }
if (now - pick->better_ms < o->settings.migration_confirm_ms) continue;
struct stream_receive_info *offer = &s->offers[sub].info;
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream optimizer improvement sub=%u old=%llx new=%llx old_score_ms=%llu new_score_ms=%llu"
" old_penalty_ms=%llu new_penalty_ms=%llu old_rtt_ms=%u/%d new_rtt_ms=%u/%d old_chain_ms=%u/%d new_chain_ms=%u/%d"
" old_hops=%u new_hops=%u old_usage=%u new_usage=%u confirmed_ms=%llu",
sub, (unsigned long long)info.peer, (unsigned long long)s->node,
(unsigned long long)stream_supplier_score(&info, &info.metrics, info.metrics_valid),
(unsigned long long)stream_supplier_score(offer, &offer->supplier_metrics, offer->supplier_metrics_valid),
(unsigned long long)problem_penalty(&info.metrics, info.metrics_valid),
(unsigned long long)problem_penalty(&offer->supplier_metrics, offer->supplier_metrics_valid),
info.rtt_ms, info.rtt_valid, offer->rtt_ms, offer->rtt_valid,
info.path_rtt_ms, info.path_rtt_valid, offer->path_rtt_ms, offer->path_rtt_valid, info.path.count, offer->path.count,
usage(o, info.peer, sub), usage(o, s->node, sub), (unsigned long long)(now - pick->better_ms));
if (stream_receive_open(h, s->node, (uint8_t)sub, seq, &pick->replacement) == STREAM_OK) {
pick->replacement_ms = now; migrating = 1;
DEBUG_INFO(DEBUG_CATEGORY_VIDEO, "stream optimizer overlap sub=%u old=%llx new=%llx index=%llu",
sub, (unsigned long long)info.peer, (unsigned long long)s->node, (unsigned long long)seq);
}
}
o->info.candidates = o->info.probes_active = o->info.subflows = o->info.overlaps = o->info.providers = 0;
o->info.protected_by_fec = o->descriptor.r != 0;
for (struct stream_supplier *s = o->suppliers; s; s = s->next) {
o->info.candidates++; if (s->probe) o->info.probes_active++;
unsigned used = usage(o, s->node, o->descriptor.n);
if (used) o->info.providers++;
if (used > o->descriptor.r) o->info.protected_by_fec = 0;
if (used || s->probe || !s->offers) continue;
uint64_t keep_until = 0;
for (unsigned sub = 0; sub < o->descriptor.n; sub++) {
struct stream_offer *offer = &s->offers[sub];
uint64_t until = offer->info.info_ms + STREAM_OFFER_TTL_MS;
if (until < offer->retry_ms) until = offer->retry_ms;
if (until > keep_until) keep_until = until;
}
if (now >= keep_until) { u_free(s->offers); s->offers = NULL; }
}
for (unsigned sub = 0; sub < o->descriptor.n; sub++) {
if (o->selected[sub].current) {
struct stream_receive_info info; stream_receive_get_info(o->selected[sub].current, &info);
o->info.subflows++;
if (info.state != STREAM_RECEIVE_ACTIVE || !info.blocks || !info.metrics_valid || info.metrics.stopped || info.close_requested)
o->info.protected_by_fec = 0;
}
if (o->selected[sub].replacement) o->info.overlaps++;
}
if (o->info.subflows != o->descriptor.n) o->info.protected_by_fec = 0;
/* Помощь нужна для получения k субпотоков, а не для идеальной защиты n/r; пауза источника не является отказом пути. */
unsigned active = 0, viable = 0; int newer = 0;
for (unsigned sub = 0; sub < o->descriptor.n; sub++) {
struct STREAM_RECEIVE *current = o->selected[sub].current;
if (!current) continue;
struct stream_receive_info info; stream_receive_get_info(current, &info);
if (info.state != STREAM_RECEIVE_ACTIVE || info.close_requested) continue;
active++;
if ((info.blocks && info.metrics_valid && !info.metrics.stopped) || (!info.blocks && info.last_available == UINT64_MAX)) viable++;
}
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;
}
unsigned required = (unsigned)(o->descriptor.n - o->descriptor.r);
int deficient = active < required || (viable < required && newer);
if (!deficient) o->deficient_ms = 0;
else if (!o->deficient_ms) o->deficient_ms = now;
if (!o->startup_complete || now >= o->relay_poll_ms) {
o->relay_poll_ms = now + STREAM_PROBE_PERIOD_MS;
int needed = deficient && now - o->deficient_ms >= STREAM_MIGRATION_TIMEOUT_MS;
if (!o->startup_complete && !o->startup_peer && now - o->started_ms >= STREAM_BOOTSTRAP_RELAY_MS) needed = 1;
if (!o->startup_complete && o->startup_peer && stream_relay_confirmed(o->inst, &o->descriptor, o->startup_peer)) needed = 1;
stream_relay_demand(o->inst, &o->descriptor, needed, now);
}
/* Пока нет первого блока, раз в секунду показываем причину ожидания каждого поставщика. */
if (!local.anchored && now >= o->diagnostic_ms) {
o->diagnostic_ms = now + STREAM_PROBE_PERIOD_MS;
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream bootstrap waits group=%llx source=%llx epoch=%llx candidates=%u probes=%u"
" subflows=%u cycle=%llu cycle_age_ms=%llu",
(unsigned long long)o->descriptor.group_id, (unsigned long long)o->descriptor.source_node_id,
(unsigned long long)o->descriptor.source_epoch, o->info.candidates, o->info.probes_active, o->info.subflows,
(unsigned long long)o->schedule.cycle, (unsigned long long)(now - o->schedule.started_ms));
for (struct stream_supplier *s = o->suppliers; s; s = s->next) {
struct stream_receive_info probe = {0};
if (s->probe) stream_receive_get_info(s->probe, &probe);
DEBUG_TRACE(DEBUG_CATEGORY_STREAM, "stream bootstrap supplier source=%llx peer=%llx online=%d compatible=%d source_peer=%d"
" cycle=%llu failures=%u retry_in_ms=%llu probe=%d sub=%u state=%d info_ready=%d reason=%d",
(unsigned long long)o->descriptor.source_node_id, (unsigned long long)s->node, s->online, s->compatible, s->source,
(unsigned long long)s->cycle, s->failures, (unsigned long long)(s->retry_ms > now ? s->retry_ms - now : 0),
s->probe != NULL, s->probe_sub, probe.state, probe.info_ready, probe.reason);
}
}
}
Reference in new issue View Git Blame Copy Permalink