Isolate monotonic probe scheduling policy for precise boundary checks

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) {

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);