diff --git a/src/stream/stream_optimizer.c b/src/stream/stream_optimizer.c index 68f39e68..91e8a76b 100644 --- a/src/stream/stream_optimizer.c +++ b/src/stream/stream_optimizer.c @@ -38,10 +38,20 @@ struct stream_optimizer { 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; + struct stream_probe_schedule schedule; + uint64_t roster_ms, progress_ms, next_seq; + int progress_valid; }; +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; @@ -78,7 +88,7 @@ static void roster_update(struct stream_optimizer *o, struct STREAM_HANDLE *h, u /* Один 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; + 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); @@ -89,7 +99,7 @@ static int probe_start(struct stream_optimizer *o, struct STREAM_HANDLE *h, stru } 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)s->node, (unsigned long long)o->schedule.cycle, (unsigned long long)o->info.probes_started, s->source); return 1; } @@ -118,29 +128,26 @@ static void probes_poll(struct stream_optimizer *o, struct STREAM_HANDLE *h, uin } 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 (now < s->retry_ms) s->cycle = o->cycle; - 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); + 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->cycle || s->probe || now < s->retry_ms) continue; + if (!s->compatible || s->cycle == o->schedule.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) { diff --git a/src/stream/stream_optimizer.h b/src/stream/stream_optimizer.h index 324672d8..ce297a1d 100644 --- a/src/stream/stream_optimizer.h +++ b/src/stream/stream_optimizer.h @@ -6,6 +6,10 @@ #include "stream.h" struct stream_optimizer; +/* Только монотонное расписание; pending/active — незавершённый обход/пробы текущего цикла. + * Возвращает допустимое число новых узлов, не адресов/субпотоков: 32 один раз, затем 0/1. */ +struct stream_probe_schedule { uint64_t cycle, started_ms, next_ms; }; +unsigned stream_probe_schedule_tick(struct stream_probe_schedule *schedule, uint64_t now_ms, unsigned pending, unsigned active); int stream_optimizer_create(struct UTUN_INSTANCE *inst, const struct stream_descriptor *descriptor, struct stream_optimizer **out); void stream_optimizer_destroy(struct stream_optimizer *optimizer); void stream_optimizer_tick(struct stream_optimizer *optimizer, struct STREAM_HANDLE *subscription, uint64_t now_ms);