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

Implement stream supplier probe cycles, FEC-aware selection and overlap migration

Browse Source
This commit is contained in:
evgeny committed 2026-10-12 00:01:39 +02:00
1 parent 3db2130b56
commit 576c79657b
2 files changed
+359 -1
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/stream/stream_optimizer.c
+357
View File
@@ -0,0 +1,357 @@
#include <string.h>
#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(&current->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;
}
src/stream/stream_priv.h
+2 -1
View File
@@ -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);