From 2cc9834e4c2fe28f828539d95c4ae421c3dc9d0b Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 13 Oct 2026 01:14:38 +0200 Subject: [PATCH] Use integral scores to select stream suppliers --- doc/stream.txt | 21 +++++++--- src/stream/stream.h | 4 +- src/stream/stream_optimizer.c | 79 +++++++++++++++++++++-------------- src/stream/stream_optimizer.h | 3 ++ tests/test_stream_core.c | 37 ++++++++++++++++ 5 files changed, 105 insertions(+), 39 deletions(-) diff --git a/doc/stream.txt b/doc/stream.txt index afd1ebcc..7497c760 100644 --- a/doc/stream.txt +++ b/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 не вызывает. diff --git a/src/stream/stream.h b/src/stream/stream.h index 3f6da37b..4b563625 100644 --- a/src/stream/stream.h +++ b/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); diff --git a/src/stream/stream_optimizer.c b/src/stream/stream_optimizer.c index b54e51bc..3a82e0de 100644 --- a/src/stream/stream_optimizer.c +++ b/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(¤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; + uint64_t before = stream_supplier_score(current, ¤t->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; diff --git a/src/stream/stream_optimizer.h b/src/stream/stream_optimizer.h index ce297a1d..317cd501 100644 --- a/src/stream/stream_optimizer.h +++ b/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; }; diff --git a/tests/test_stream_core.c b/tests/test_stream_core.c index bdcc1d62..c9d59db2 100644 --- a/tests/test_stream_core.c +++ b/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);