From 576c79657bb5b144d223847dc8bc1a4371ab9c95 Mon Sep 17 00:00:00 2001 From: evgeny Date: Sun, 11 Oct 2026 14:04:45 +0200 Subject: [PATCH] Implement stream supplier probe cycles, FEC-aware selection and overlap migration --- src/stream/stream_optimizer.c | 357 ++++++++++++++++++++++++++++++++++ src/stream/stream_priv.h | 3 +- 2 files changed, 359 insertions(+), 1 deletion(-) create mode 100644 src/stream/stream_optimizer.c diff --git a/src/stream/stream_optimizer.c b/src/stream/stream_optimizer.c new file mode 100644 index 00000000..9a0189c7 --- /dev/null +++ b/src/stream/stream_optimizer.c @@ -0,0 +1,357 @@ +#include + +#include "../../lib/debug_config.h" +#include "../../lib/mem.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 + +struct stream_offer { + struct stream_receive_info info; + uint64_t retry_ms; + unsigned failures; +}; +struct stream_supplier { + struct stream_supplier *next; + uint64_t node, cycle, retry_ms; + struct STREAM_RECEIVE *probe; + struct stream_offer *offers; + unsigned probe_sub, failures; + int online, compatible, source; +}; +struct stream_selection { + struct STREAM_RECEIVE *current, *replacement; + uint64_t opened_ms, replacement_ms, better_ms, better_node; +}; +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; + uint64_t cycle, cycle_ms, next_probe_ms, roster_ms, progress_ms, next_seq; + int started, progress_valid; +}; + +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->compatible = node_conn_direct_compatible(o->inst, s->node); + } + 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->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->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) { + 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_TRANSPORT || info.reason == STREAM_REASON_MEMBER || info.reason == STREAM_REASON_PROTOCOL) { + s->failures++; s->retry_ms = now + backoff(s->failures); + stream_receive_close(s->probe); s->probe = NULL; continue; + } + } else { s->offers[s->probe_sub].info = info; s->failures = 0; s->retry_ms = 0; } + 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 budget = o->started ? 1 : STREAM_PROBE_BURST; + if (!o->started) { o->started = 1; o->cycle = o->info.cycles = 1; o->cycle_ms = now; } + else if (now < o->next_probe_ms) return; + unsigned pending = 0, active = 0; + for (struct stream_supplier *s = o->suppliers; s; s = s->next) { + if (s->probe) active++; + if (s->compatible && s->cycle != o->cycle) pending++; + } + if (!pending && !active && now - o->cycle_ms >= STREAM_PROBE_CYCLE_MS) { + o->cycle++; o->info.cycles++; o->cycle_ms = now; + DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream optimizer cycle=%llu", (unsigned long long)o->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->cycle || s->probe || now < s->retry_ms) continue; + if (!chosen || s->source > chosen->source || (s->source == chosen->source && s->node < chosen->node)) chosen = s; + } + if (!chosen) break; + probe_start(o, h, chosen, now); + } + o->next_probe_ms = now + STREAM_PROBE_PERIOD_MS; +} + +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; +} + +static double problems(const struct stream_metric_snapshot *m, int valid) { + if (!valid) return 0; + double fraction = m->blocks ? (double)m->late_blocks / (double)m->blocks : 0; + return (m->stopped || m->lag_ms > STREAM_LATE_MS ? 2 : 0) + fraction + (m->stops ? 0.01 : 0); +} + +static uint64_t total_rtt(const struct stream_receive_info *info) { + return info->rtt_valid && info->path_rtt_valid ? (uint64_t)info->rtt_ms + info->path_rtt_ms : UINT64_MAX; +} + +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 && (seq < info->first_available || seq > info->last_available))) 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; + double score = problems(&info->supplier_metrics, info->supplier_metrics_valid); + double old_score = problems(&previous->supplier_metrics, previous->supplier_metrics_valid); + uint64_t rtt = total_rtt(info), old_rtt = total_rtt(previous); + if (score < old_score || (score == old_score && (rtt < old_rtt + || (rtt == old_rtt && (shared_path(o, info, sub) < shared_path(o, previous, sub) + || (shared_path(o, info, sub) == shared_path(o, previous, sub) + && (info->path.count < previous->path.count || (info->path.count == previous->path.count && s->node < best->node)))))))) + best = s; + } + } + return best; +} + +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]; offer->failures++; offer->retry_ms = now + 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; + double before = problems(¤t->metrics, current->metrics_valid); + double after = problems(&offer->supplier_metrics, offer->supplier_metrics_valid); + if (after + 0.10 < before) return 1; + if (after > before) return 0; + uint64_t rtt = total_rtt(offer), old = total_rtt(current); + if (old != UINT64_MAX && rtt < old && old - rtt >= o->settings.rtt_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 + && (rtt == UINT64_MAX || old == UINT64_MAX || rtt <= old + o->settings.rtt_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->info.automatic = 1; *out = o; return STREAM_OK; +} + +void stream_optimizer_destroy(struct stream_optimizer *o) { + if (!o) return; + for (unsigned sub = 0; sub < o->descriptor.n; sub++) { + stream_receive_close(o->selected[sub].current); stream_receive_close(o->selected[sub].replacement); + } + 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->rtt_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_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; + if (local.finished) { + for (struct stream_supplier *s = o->suppliers; s; s = s->next) { stream_receive_close(s->probe); s->probe = NULL; } + o->info.probes_active = 0; return; + } + 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; + 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) received++; busy++; } + } + 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_STREAM, "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); + } + } + if (!o->progress_valid || o->next_seq != local.next) { + o->progress_valid = 1; o->next_seq = local.next; o->progress_ms = now; + } + if (local.anchored && now - o->progress_ms >= STREAM_MIGRATION_TIMEOUT_MS && received < (unsigned)(o->descriptor.n - o->descriptor.r)) { + int newer = 0; + for (struct stream_supplier *s = o->suppliers; s; s = s->next) + if (s->offers) for (unsigned sub = 0; sub < o->descriptor.n; sub++) { + struct stream_receive_info *offer = &s->offers[sub].info; + if (offer->info_ms && now - offer->info_ms < STREAM_OFFER_TTL_MS && offer->last_available > local.next) newer = 1; + } + if (newer) { + for (unsigned sub = 0; sub < o->descriptor.n; sub++) { + struct stream_selection *pick = &o->selected[sub]; stream_receive_close(pick->current); stream_receive_close(pick->replacement); + memset(pick, 0, sizeof(*pick)); + } + 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); + } + } + } + int migrating = 0; + for (unsigned sub = 0; sub < o->descriptor.n; sub++) if (o->selected[sub].replacement) migrating = 1; + 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->current) stream_receive_get_info(pick->current, &info); + uint64_t seq = local.anchored ? local.next : UINT64_MAX; + 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; busy++; + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream optimizer selected peer=%llx sub=%u index=%llu", + (unsigned long long)s->node, sub, (unsigned long long)seq); + } + 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; + if (stream_receive_open(h, s->node, (uint8_t)sub, seq, &pick->replacement) == STREAM_OK) { + pick->replacement_ms = now; migrating = 1; + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream optimizer overlap sub=%u old=%llx new=%llx index=%llu", + 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; + } + for (unsigned sub = 0; sub < o->descriptor.n; sub++) { + if (o->selected[sub].current) o->info.subflows++; + if (o->selected[sub].replacement) o->info.overlaps++; + } + if (o->info.subflows != o->descriptor.n) o->info.protected_by_fec = 0; +} diff --git a/src/stream/stream_priv.h b/src/stream/stream_priv.h index 1ee0131f..e3368495 100644 --- a/src/stream/stream_priv.h +++ b/src/stream/stream_priv.h @@ -10,12 +10,13 @@ struct stream_local { struct stream_descriptor descriptor; struct stream_buffer *buffer; - uint64_t start, last; + uint64_t start, last, next; int anchored, publisher, finished; const uint8_t *end; }; /* Заимствованные буфер/END действуют только до следующего callback/мутации сервиса. */ struct stream_network *stream_network_get(struct UTUN_INSTANCE *inst); +void stream_optimizers_tick(struct UTUN_INSTANCE *inst, uint64_t now_ms); int stream_local_get(struct UTUN_INSTANCE *inst, const struct stream_descriptor *key, struct stream_local *local); 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);