Use integral scores to select stream suppliers

doc/stream.txt

@@ -261,7 +261,7 @@ DOWN закрывает выдачи соответствующей сессии
 Дедупликация — по (stream key, batch_seq, subblock), независимо от handles.
 При перекрытии качество поставщиков учитывается отдельно, но блок потребителю выдаётся один раз.
 При явном отказе разрешено одновременно восстанавливать несколько субпотоков.
-Точная задержка обычного переноса и гистерезис задаются настройками; небольшое колебание RTT не запускает перестройку.
+Точная задержка обычного переноса и гистерезис задаются настройками; небольшое колебание оценки не запускает перестройку.
 
 9. Пути и защита от циклов
 
@@ -336,8 +336,12 @@ INFO_RESP также сообщает число свободных исходя
 Используются route_connectivity, sock_match/NCD и метрики существующих соединений; второй адресный probe-движок не создаётся.
 Прямая доступность — UNKNOWN/REACHABLE/UNREACHABLE с временем замера, а не вечный boolean.
 
-Порядок выбора: доступный субпоток/индекс и отсутствие цикла; затем проблемы доставки и RTT;
-далее устойчивость набора поставщиков, совпадение промежуточных узлов и число хопов.
+После проверки доступного субпотока/индекса и отсутствия цикла минимизируем интегральную оценку в миллисекундах:
+  score_ms = RTT_цепочки_до_поставщика + 2 * RTT_до_поставщика + 3 * hop_count + problem_penalty_ms
+hop_count включает последний переход от поставщика к получателю; для прямого получения у автора равен 1.
+problem_penalty_ms = min(1000, доля_поздних_блоков * 1000 + (были_остановки ? 10 : 0)), с округлением вниз до мс.
+Текущая остановка или lag > 500 мс дают штраф 1000 мс; неизвестные метрики дают 0, неизвестный RTT — бесконечную оценку.
+При равенстве оценок выбираем меньше общих узлов с остальными путями, затем меньший ID поставщика.
 Стремимся получить n субпотоков, хотя k достаточно для decode.
 При возможности не более r субпотоков от одного поставщика: потеря его при получении всех n оставляет не менее k.
 При r=0 такой защиты нет; при недостатке кандидатов ограничение ослабляется с отражением устойчивости в статусе.
@@ -607,10 +611,17 @@ Stop/destroy выполняются по правилам ядра, вне callb
     RTT цепочки относится к этому пути; смена фактического пути DATA делает старое измерение неизвестным,
     новая проба обновляет RTT соответствующей активной выдачи. Невыбранные снимки истекают через 60с.
     Первый поставщик запрашивается с конца; остальные стартуют с его проверенного N, даже если k уже позволили выдать пачку.
-    Поиск учитывает доступную старую позицию, запрет LOOP, проблемы/RTT, лимит r при наличии альтернатив и общие узлы пути.
+    Поиск учитывает доступную старую позицию, запрет LOOP и лимит r при наличии альтернатив.
+    Оценка поставщика в мс: RTT цепочки до поставщика + 2 * RTT до поставщика + 3 * hops + штраф проблем (0..1000 мс).
+    hops = path.count: путь INFO содержит автора/поставщика, последний переход к получателю добавляет один hop.
+    Штраф: доля поздних блоков за 60с * 1000 мс + 10 мс при остановках в истории, округление вниз, максимум 1000 мс;
+    текущая остановка или lag > 500 мс сразу дают 1000 мс. Неизвестные метрики не дают штрафа, неизвестный RTT даёт UINT64_MAX.
+    Меньшая оценка лучше; при равенстве выбираются меньше общих узлов путей, затем меньший ID поставщика.
     Более новый индекс повторно проверяется свежим INFO/SUBSCRIBE: старое объявление last не является вечным верхним пределом live-потока.
     Обычно переносится один субпоток; новая выдача перекрывает старую и заменяет её только после проверенного DATA и догоняющего cursor.
-    По умолчанию: подтверждение улучшения 2с, удержание поставщика 5с, RTT-гистерезис 20мс; параметры доступны через stream_selection_configure.
+    По умолчанию: подтверждение улучшения 2с, удержание поставщика 5с, гистерезис оценки 20мс; параметры доступны через stream_selection_configure.
+    Перенос сравнивает ту же оценку по фактическим метрикам текущей выдачи и INFO альтернативы;
+    распределение для FEC допускает ухудшение оценки не более гистерезиса. Логи выбора/переноса показывают оценку и её составляющие.
     Новая выдача без достаточного продвижения отменяется через 5с, старая остаётся. CLOSED/DOWN восстанавливают несколько отсутствующих субпотоков.
     При отсутствии decode-прогресса 5с, недостатке работающих субпотоков и наличии более новой истории возможен явный restart с конца;
     пропущенный диапазон передаётся потребителю GAP, кольцо/downstream сохраняются. Сам по себе простой источника restart не вызывает.

src/stream/stream.h

@@ -49,7 +49,7 @@ struct stream_candidate {
     uint8_t source;
 };
 
-struct stream_selection_settings { uint32_t migration_confirm_ms, minimum_hold_ms, rtt_hysteresis_ms; };
+struct stream_selection_settings { uint32_t migration_confirm_ms, minimum_hold_ms, score_hysteresis_ms; };
 struct stream_selection_info {
     uint64_t probes_started, cycles, migrations, failures, restarts;
     unsigned candidates, probes_active, subflows, overlaps, providers;
@@ -78,7 +78,7 @@ int stream_publish_open(struct UTUN_INSTANCE *inst, uint64_t group_id, const str
 int stream_subscribe_open(struct UTUN_INSTANCE *inst, uint64_t group_id, uint64_t source_id, struct STREAM_HANDLE **out);
 /* Ручные выдачи для диагностики/явного управления; режим общий для всех owners экземпляра и не смешивается с automatic. */
 int stream_subscribe_manual_open(struct UTUN_INSTANCE *inst, uint64_t group_id, uint64_t source_id, struct STREAM_HANDLE **out);
-/* По умолчанию: подтверждение улучшения 2000 мс, удержание 5000 мс, гистерезис RTT 20 мс. */
+/* По умолчанию: подтверждение улучшения 2000 мс, удержание 5000 мс, гистерезис интегральной оценки 20 мс. */
 int stream_selection_configure(struct STREAM_HANDLE *handle, const struct stream_selection_settings *settings);
 int stream_selection_get_info(const struct STREAM_HANDLE *handle, struct stream_selection_info *info);
 void stream_close(struct STREAM_HANDLE *handle);

src/stream/stream_optimizer.c

@@ -12,6 +12,10 @@
 #define STREAM_PROBE_CYCLE_MS 30000
 #define STREAM_OFFER_TTL_MS 60000
 #define STREAM_MIGRATION_TIMEOUT_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;
@@ -124,12 +128,14 @@ static void probes_poll(struct stream_optimizer *o, struct STREAM_HANDLE *h, uin
             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;
             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",
+                " 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)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; }
@@ -177,14 +183,21 @@ static unsigned usage(struct stream_optimizer *o, uint64_t node, unsigned except
     return count;
 }
 
-static double problems(const struct stream_metric_snapshot *m, int valid) {
+/* Доля поздних блоков задаёт штраф в мс; текущая остановка/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;
-    return (m->stopped || m->lag_ms > STREAM_LATE_MS ? 2 : 0) + fraction + (m->stops ? 0.01 : 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;
 }
 
-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;
+/* 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) {
@@ -212,18 +225,12 @@ static struct stream_supplier *supplier_choose(struct stream_optimizer *o, unsig
             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);
+            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) {
-                better = rtt < old_rtt;
-                if (rtt == old_rtt) {
-                    unsigned shared = shared_path(o, info, sub), old_shared = shared_path(o, previous, sub);
-                    better = shared < old_shared;
-                    if (shared == old_shared) better = info->path.count < previous->path.count
-                        || (info->path.count == previous->path.count && s->node < best->node);
-                }
+                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;
         }
@@ -250,15 +257,12 @@ static void receive_failed(struct stream_optimizer *o, struct STREAM_RECEIVE **h
 
 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;
+    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
-           && (rtt == UINT64_MAX || old == UINT64_MAX || rtt <= old + o->settings.rtt_hysteresis_ms);
+           && 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) {
@@ -287,7 +291,7 @@ void stream_optimizer_destroy(struct stream_optimizer *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) {
+        || settings->score_hysteresis_ms > 60000) {
         DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream optimizer settings invalid"); return STREAM_ERR_ARGUMENT;
     }
     o->settings = *settings; return STREAM_OK;
@@ -376,8 +380,13 @@ void stream_optimizer_tick(struct stream_optimizer *o, struct STREAM_HANDLE *h,
         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++;
-                DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream optimizer selected peer=%llx sub=%u index=%llu",
-                           (unsigned long long)s->node, sub, (unsigned long long)seq);
+                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;
         }
@@ -387,11 +396,17 @@ void stream_optimizer_tick(struct stream_optimizer *o, struct STREAM_HANDLE *h,
         }
         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;
-        DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "stream optimizer improvement sub=%u old=%llx new=%llx old_problems=%.3f new_problems=%.3f"
-            " old_rtt=%llu new_rtt=%llu old_usage=%u new_usage=%u confirmed_ms=%llu",
-            sub, (unsigned long long)info.peer, (unsigned long long)s->node, problems(&info.metrics, info.metrics_valid),
-            problems(&s->offers[sub].info.supplier_metrics, s->offers[sub].info.supplier_metrics_valid),
-            (unsigned long long)total_rtt(&info), (unsigned long long)total_rtt(&s->offers[sub].info),
+        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;

src/stream/stream_optimizer.h

@@ -6,6 +6,9 @@
 #include "stream.h"
 
 struct stream_optimizer;
+struct stream_receive_info;
+/* Меньшая оценка лучше: chain RTT + 2 * peer RTT + 3 мс/hop + штраф 0..1000 мс; неизвестный RTT = UINT64_MAX. */
+uint64_t stream_supplier_score(const struct stream_receive_info *info, const struct stream_metric_snapshot *metrics, int valid);
 /* Только монотонное расписание; pending/active — незавершённый обход/пробы текущего цикла.
  * Возвращает допустимое число новых узлов, не адресов/субпотоков: 32 один раз, затем 0/1. */
 struct stream_probe_schedule { uint64_t cycle, started_ms, next_ms; };

tests/test_stream_core.c

@@ -7,6 +7,7 @@
 #include "../src/stream/stream_metrics.h"
 #include "../src/stream/stream_load.h"
 #include "../src/stream/stream_optimizer.h"
+#include "../src/stream/stream_network.h"
 #include "../lib/debug_config.h"
 #include "../lib/mem.h"
 #include "test_clock.h"
@@ -382,6 +383,41 @@ static void test_source_timer(void) {
     DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] source: 100ms timer, immediate full batch, padding, finish and cancellation on destroy");
 }
 
+/* Близкий сосед, цена дополнительных hops и проблем сравниваются в одних единицах, неизвестный RTT не становится нулём. */
+static void test_supplier_score(void) {
+    struct stream_receive_info near = {.rtt_ms = 10, .path_rtt_ms = 80, .rtt_valid = 1, .path_rtt_valid = 1, .path = {.count = 3}};
+    struct stream_receive_info far = {.rtt_ms = 45, .path_rtt_ms = 40, .rtt_valid = 1, .path_rtt_valid = 1, .path = {.count = 2}};
+    struct stream_metric_snapshot clean = {.blocks = 100}, late = {.blocks = 100, .late_blocks = 2};
+    CHECK(stream_supplier_score(&near, &clean, 1) == 109 && stream_supplier_score(&far, &clean, 1) == 136);
+    /* Обычная сумма RTT предпочла бы far (85 < 90); удвоенный RTT соседа предпочитает near. */
+    CHECK(stream_supplier_score(&near, &clean, 1) < stream_supplier_score(&far, &clean, 1));
+    CHECK(stream_supplier_score(&near, &late, 1) == 129);
+    CHECK(stream_supplier_score(&near, &late, 1) < stream_supplier_score(&far, &clean, 1));
+    late.late_blocks = 3;
+    CHECK(stream_supplier_score(&near, &late, 1) > stream_supplier_score(&far, &clean, 1));
+    far = near; far.rtt_ms = 9; far.path.count = 4;
+    CHECK(stream_supplier_score(&far, &clean, 1) == 110);
+    CHECK(stream_supplier_score(&near, &clean, 1) < stream_supplier_score(&far, &clean, 1));
+    late = (struct stream_metric_snapshot){.blocks = 4, .late_blocks = 1, .stops = 7};
+    CHECK(stream_supplier_score(&near, &late, 1) == 369);
+    late.lag_ms = STREAM_LATE_MS;
+    CHECK(stream_supplier_score(&near, &late, 1) == 369);
+    late.lag_ms++;
+    CHECK(stream_supplier_score(&near, &late, 1) == 1109);
+    late.lag_ms = 0; late.stopped = 1;
+    CHECK(stream_supplier_score(&near, &late, 1) == 1109);
+    late.stopped = 0; late.late_blocks = late.blocks;
+    CHECK(stream_supplier_score(&near, &late, 1) == 1109); /* История остановок не превышает лимит штрафа. */
+    CHECK(stream_supplier_score(&near, &late, 0) == 109);
+    near.rtt_valid = 0;
+    CHECK(stream_supplier_score(&near, &clean, 1) == UINT64_MAX);
+    near.rtt_valid = 1; near.path_rtt_valid = 0;
+    CHECK(stream_supplier_score(&near, &clean, 1) == UINT64_MAX);
+    near.path_rtt_valid = 1; near.rtt_ms = near.path_rtt_ms = UINT32_MAX;
+    CHECK(stream_supplier_score(&near, &late, 1) == 3 * (uint64_t)UINT32_MAX + 1009);
+    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] supplier score: nearby peer, hop cost, problem tradeoffs, 1000ms cap, unknown and wide RTT");
+}
+
 static void test_probe_schedule(void) {
     struct stream_probe_schedule schedule = {0};
     CHECK(stream_probe_schedule_tick(&schedule, 0, 40, 0) == 32 && schedule.cycle == 1);
@@ -414,6 +450,7 @@ int main(void) {
     test_wire(); CHECK(u_get_allocated_count() == allocations);
     test_load(); CHECK(u_get_allocated_count() == allocations);
     test_probe_schedule(); CHECK(u_get_allocated_count() == allocations);
+    test_supplier_score(); CHECK(u_get_allocated_count() == allocations);
     test_fec_and_padding(); CHECK(u_get_allocated_count() == allocations);
     test_ring_and_conflicts(); CHECK(u_get_allocated_count() == allocations);
     test_relative_metrics(); CHECK(u_get_allocated_count() == allocations);