diff --git a/src/stream/stream_optimizer.c b/src/stream/stream_optimizer.c index fa4f196c..22e555ed 100644 --- a/src/stream/stream_optimizer.c +++ b/src/stream/stream_optimizer.c @@ -2,6 +2,7 @@ #include "../../lib/debug_config.h" #include "../../lib/mem.h" +#include "../../lib/u_async.h" #include "../utun_instance.h" #include "../transport_layer/node_conn_direct.h" #include "stream_optimizer.h" @@ -12,6 +13,7 @@ #define STREAM_PROBE_CYCLE_MS 30000 #define STREAM_OFFER_TTL_MS 60000 #define STREAM_MIGRATION_TIMEOUT_MS 5000 +#define STREAM_BOOTSTRAP_RELAY_MS 100 #define STREAM_PROBLEM_PENALTY_MS 1000 #define STREAM_HOP_PENALTY_MS 3 @@ -44,10 +46,18 @@ struct stream_optimizer { struct stream_selection_settings settings; struct stream_selection_info info; struct stream_probe_schedule schedule; + void *startup_timer; + uint64_t started_ms, startup_peer; uint64_t roster_ms, progress_ms, next_seq, diagnostic_ms, deficient_ms, relay_poll_ms; - int progress_valid, draining; + int progress_valid, draining, startup_complete; }; +/* Точный срок гонки не зависит от фазы общего 100-мс таймера; selector остаётся отложенным. */ +static void startup_timeout(void *arg) { + struct stream_optimizer *o = arg; o->startup_timer = NULL; + stream_network_wake(stream_network_get(o->inst)); +} + 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; @@ -221,12 +231,17 @@ static struct stream_supplier *supplier_choose(struct stream_optimizer *o, unsig 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; + if (!o->startup_complete && o->startup_peer && s->node != o->startup_peer) 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 && info->first_available != UINT64_MAX && seq < info->first_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; + if (!o->startup_complete && seq == UINT64_MAX) { + if (info->info_order < previous->info_order) best = s; + continue; + } 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; @@ -240,6 +255,18 @@ static struct stream_supplier *supplier_choose(struct stream_optimizer *o, unsig return best; } +/* Первый пригодный INFO среди всех субпотоков выигрывает гонку; порядок приёма не зависит от порядка списка пиров. */ +static struct stream_supplier *startup_choose(struct stream_optimizer *o, uint64_t now, unsigned *subblock) { + struct stream_supplier *first = NULL; uint64_t order = UINT64_MAX; + for (unsigned sub = 0; sub < o->descriptor.n; sub++) { + struct stream_supplier *s = supplier_choose(o, sub, UINT64_MAX, 0, now); + if (s && s->offers[sub].info.info_order < order) { + first = s; order = s->offers[sub].info.info_order; *subblock = sub; + } + } + return first; +} + 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); @@ -273,11 +300,18 @@ int stream_optimizer_create(struct UTUN_INSTANCE *inst, const struct stream_desc 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->started_ms = get_time_tb() / 10; + o->startup_timer = uasync_set_timeout(inst->ua, STREAM_BOOTSTRAP_RELAY_MS * 10, o, startup_timeout, "stream_bootstrap"); + if (!o->startup_timer) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream bootstrap timer allocation failed"); + u_free(o->selected); u_free(o); return STREAM_ERR_MEMORY; + } o->info.automatic = 1; *out = o; return STREAM_OK; } void stream_optimizer_destroy(struct stream_optimizer *o) { if (!o) return; + if (o->startup_timer) uasync_cancel_timeout(o->inst->ua, o->startup_timer); for (unsigned sub = 0; sub < o->descriptor.n; sub++) { stream_receive_close(o->selected[sub].current); stream_receive_close(o->selected[sub].replacement); } @@ -316,6 +350,13 @@ void stream_optimizer_refresh(struct stream_optimizer *o, uint64_t peer) { 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.decoded && !o->startup_complete) { + o->startup_complete = 1; + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream bootstrap complete group=%llx source=%llx epoch=%llx peer=%llx elapsed_ms=%llu start=%llu", + (unsigned long long)o->descriptor.group_id, (unsigned long long)o->descriptor.source_node_id, + (unsigned long long)o->descriptor.source_epoch, (unsigned long long)o->startup_peer, + (unsigned long long)(now - o->started_ms), (unsigned long long)local.start); + } if (local.finished) { /* END одного поставщика задаёт границу источника; остальные выдачи ещё могут требовать восстановления. */ if (local.last == UINT64_MAX || (local.anchored && local.next > local.last)) { @@ -385,6 +426,21 @@ void stream_optimizer_tick(struct stream_optimizer *o, struct STREAM_HANDLE *h, } int migrating = 0; for (unsigned sub = 0; sub < o->descriptor.n; sub++) if (o->selected[sub].replacement) migrating = 1; + if (!local.anchored && !busy && !o->startup_complete) { + o->startup_peer = 0; + unsigned sub = 0; struct stream_supplier *s = startup_choose(o, now, &sub); + if (s && stream_receive_open(h, s->node, (uint8_t)sub, UINT64_MAX, &o->selected[sub].current) == STREAM_OK) { + o->startup_peer = s->node; o->selected[sub].opened = 1; o->selected[sub].opened_ms = now; busy++; + if (o->startup_timer) { uasync_cancel_timeout(o->inst->ua, o->startup_timer); o->startup_timer = NULL; } + for (struct stream_supplier *other = o->suppliers; other; other = other->next) + if (other != s && other->probe) { stream_receive_close(other->probe); other->probe = NULL; } + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream bootstrap winner group=%llx source=%llx epoch=%llx peer=%llx sub=%u" + " order=%llu elapsed_ms=%llu relay=%d", + (unsigned long long)o->descriptor.group_id, (unsigned long long)o->descriptor.source_node_id, + (unsigned long long)o->descriptor.source_epoch, (unsigned long long)s->node, sub, + (unsigned long long)s->offers[sub].info.info_order, (unsigned long long)(now - o->started_ms), s->relay); + } + } 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; @@ -476,9 +532,11 @@ void stream_optimizer_tick(struct stream_optimizer *o, struct STREAM_HANDLE *h, int deficient = active < required || (viable < required && newer); if (!deficient) o->deficient_ms = 0; else if (!o->deficient_ms) o->deficient_ms = now; - if (now >= o->relay_poll_ms) { + if (!o->startup_complete || now >= o->relay_poll_ms) { o->relay_poll_ms = now + STREAM_PROBE_PERIOD_MS; int needed = deficient && now - o->deficient_ms >= STREAM_MIGRATION_TIMEOUT_MS; + if (!o->startup_complete && !o->startup_peer && now - o->started_ms >= STREAM_BOOTSTRAP_RELAY_MS) needed = 1; + if (!o->startup_complete && o->startup_peer && stream_relay_confirmed(o->inst, &o->descriptor, o->startup_peer)) needed = 1; stream_relay_demand(o->inst, &o->descriptor, needed, now); } /* Пока нет первого блока, раз в секунду показываем причину ожидания каждого поставщика. */ diff --git a/tests/test_stream_service.c b/tests/test_stream_service.c index ece5ae71..7cece716 100644 --- a/tests/test_stream_service.c +++ b/tests/test_stream_service.c @@ -54,10 +54,17 @@ static int automatic_drop_relay; static int automatic_drop_tail; static etcp_recv_fn automatic_handlers[4]; static uint8_t automatic_request[STREAM_WIRE_HEADER + 1]; -static etcp_recv_fn relay_raw_handlers[3], relay_route_handler; +static etcp_recv_fn relay_raw_handlers[4], relay_route_handler; static unsigned relay_data[3], relay_ends[3], relay_requests_seen; static int relay_drop_source; static uint8_t relay_request_wire[STREAM_WIRE_HEADER]; +static struct { + int hold_info, hold_request, check_starts; + uint8_t info[STREAM_WIRE_HEADER + STREAM_INFO_SIZE + STREAM_PATH_MAX * 8]; + size_t info_length; + unsigned source_subscribes, helper_subscribes, accepted; + uint64_t request_tb, start; +} relay_race; static struct { unsigned data, reinits, responses; int hold_info; @@ -1009,7 +1016,27 @@ static void relay_role(unsigned member, int enabled, unsigned version) { /* Прямой stream-вход источника блокируем только у зрителей; Merkle/BGP и получение суперузла остаются настоящими. */ static void relay_test_raw(struct ETCP_CONN *conn, struct ll_entry *entry) { - unsigned index = 0; while (index < 3 && conn->instance != nodes[index]) index++; CHECK(index < 3); + unsigned index = 0; while (index < 4 && conn->instance != nodes[index]) index++; CHECK(index < 4); + if (entry->len >= STREAM_WIRE_HEADER) { + unsigned command = entry->dgram[2]; + if (command == STREAM_CMD_SUBSCRIBE && conn->peer_node_id == nodes[2]->node_id) { + if (index == 3) relay_race.source_subscribes++; + if (index == 1) relay_race.helper_subscribes++; + } + if (index == 2 && conn->peer_node_id == nodes[3]->node_id && command == STREAM_CMD_INFO_RESP && relay_race.hold_info) { + if (!relay_race.info_length) { + CHECK(entry->len <= sizeof(relay_race.info)); relay_race.info_length = entry->len; + memcpy(relay_race.info, entry->dgram, entry->len); + } + queue_dgram_free(entry); queue_entry_free(entry); return; + } + if (relay_race.check_starts && index == 2 && conn->peer_node_id == nodes[1]->node_id && command == STREAM_CMD_ACCEPT) { + struct stream_message m; CHECK(stream_wire_parse(entry->dgram, entry->len, &m) == STREAM_OK && m.length >= 16); + uint64_t start = stream_read_be(m.payload + 8, 8); + if (!relay_race.accepted) relay_race.start = start; + CHECK(start == relay_race.start && m.subblock < 8); relay_race.accepted |= 1u << m.subblock; + } + } if (relay_drop_source && index != 1 && conn->peer_node_id == nodes[3]->node_id) { queue_dgram_free(entry); queue_entry_free(entry); return; } @@ -1021,6 +1048,8 @@ static void relay_test_routed(struct ETCP_CONN *conn, struct ll_entry *entry) { if (entry->len >= ROUTER_SVC_PAYLOAD_OFF + STREAM_WIRE_HEADER && entry->dgram[ROUTER_SVC_PAYLOAD_OFF + 2] == STREAM_CMD_RELAY_REQUEST) { memcpy(relay_request_wire, entry->dgram + ROUTER_SVC_PAYLOAD_OFF, STREAM_WIRE_HEADER); relay_requests_seen++; + if (!relay_race.request_tb) relay_race.request_tb = get_time_tb(); + if (relay_race.hold_request) { queue_dgram_free(entry); queue_entry_free(entry); return; } } relay_route_handler(conn, entry); } @@ -1041,6 +1070,15 @@ static void relay_test_control(const struct stream_message *message) { CHECK(etcp_route_send(nodes[2], GROUP2, nodes[1]->node_id, entry, 0, ROUTE_CRYPTO_SIGN | ROUTE_CRYPTO_ENCRYPT) == 0); } +/* Запоздалый настоящий INFO подаётся тому же raw dispatcher; старый probe nonce не должен менять победителя. */ +static void relay_release_info(void) { + CHECK(relay_race.info_length); + struct ETCP_CONN *conn = instance_find_conn(nodes[2], nodes[3]->node_id); CHECK(conn && conn->links_up); + struct ll_entry *entry = ll_alloc_lldgram((uint16_t)relay_race.info_length); CHECK(entry); + memcpy(entry->dgram, relay_race.info, relay_race.info_length); entry->len = relay_race.info_length; + relay_raw_handlers[2](conn, entry); relay_race.info_length = 0; relay_race.hold_info = 0; +} + /* Все проверки помощи используют подписанное управление и настоящий raw ETCP, не прямой ingest. */ static void test_supernode_relay(void) { int route_level = g_debug_config.category_levels[DEBUG_CATEGORY_ETCPROUTE]; @@ -1055,14 +1093,14 @@ static void test_supernode_relay(void) { CHECK(member_sync_start(nodes[i], nodes[3]->node_id, "710002", NULL, NULL) == 0); } relay_role(1, 1, 100); - for (unsigned i = 0; i < 3; i++) { + for (unsigned i = 0; i < 4; i++) { relay_raw_handlers[i] = nodes[i]->api_bindings.callbacks[ETCP_ID_STREAM]; CHECK(etcp_bind(nodes[i], ETCP_ID_STREAM, relay_test_raw) == 0); - stream_set_event_callback(nodes[i], relay_test_event, NULL); + if (i < 3) stream_set_event_callback(nodes[i], relay_test_event, NULL); } relay_route_handler = nodes[1]->router_bindings.callbacks[ETCP_RT_ID_STREAM]; CHECK(etcp_router_bind(nodes[1], ETCP_RT_ID_STREAM, relay_test_routed) == 0); - struct stream_parameters parameters = {.n = 2, .r = 1, .block_size = STREAM_RS_DEFAULT_SIZE}; + struct stream_parameters parameters = {.n = 8, .r = 2, .block_size = STREAM_RS_DEFAULT_SIZE}; struct STREAM_HANDLE *publisher = NULL, *viewers[3] = {0}; CHECK(stream_publish_open(nodes[3], GROUP2, ¶meters, &publisher) == STREAM_OK); struct stream_info origin; CHECK(stream_get_info(publisher, &origin) == STREAM_OK); @@ -1074,11 +1112,45 @@ static void test_supernode_relay(void) { CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_WAIT && !relay_requests_seen); stream_close(viewers[2]); viewers[2] = NULL; test_clock_run_for(ua, 10000); + /* Помощь начинается через 100 мс, но прямой INFO выигрывает до ответа помощника и отменяет его запрос. */ + relay_race.hold_info = relay_race.hold_request = 1; + uint64_t started = get_time_tb(); + CHECK(stream_subscribe_open(nodes[2], GROUP2, nodes[3]->node_id, &viewers[2]) == STREAM_OK); + test_clock_run_for(ua, 900); CHECK(!relay_requests_seen); + deadline = get_time_tb() + 10000; + while (!relay_requests_seen && get_time_tb() < deadline) test_clock_poll(ua, 100); + CHECK(relay_requests_seen && relay_race.info_length); + CHECK(relay_race.request_tb - started >= 990 && relay_race.request_tb - started < 1500); + relay_release_info(); run_until(&relay_data[2], 2); test_clock_run_for(ua, 2000); + CHECK(relay_race.source_subscribes && !relay_race.helper_subscribes); + CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_WAIT); + struct stream_message cancelled; CHECK(stream_wire_parse(relay_request_wire, sizeof(relay_request_wire), &cancelled) == STREAM_OK); + relay_race.hold_request = 0; relay_test_control(&cancelled); test_clock_run_for(ua, 2000); + CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_WAIT); + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] bootstrap direct wins after helper request: request_delay_ms=%llu, late REQUEST cancelled", + (unsigned long long)((relay_race.request_tb - started) / 10)); + stream_close(viewers[2]); viewers[2] = NULL; test_clock_run_for(ua, 1000); + /* Закрытие reader до дедлайна отменяет и таймер, и объединённую работу сети. */ + unsigned requests_before = relay_requests_seen; + relay_race.hold_info = 1; + CHECK(stream_subscribe_open(nodes[2], GROUP2, nodes[3]->node_id, &viewers[2]) == STREAM_OK); + test_clock_run_for(ua, 500); stream_close(viewers[2]); viewers[2] = NULL; + test_clock_run_for(ua, 2000); CHECK(relay_requests_seen == requests_before); + memset(&relay_race, 0, sizeof(relay_race)); relay_race.hold_info = relay_race.check_starts = 1; relay_drop_source = 1; + started = get_time_tb(); CHECK(stream_subscribe_open(nodes[2], GROUP2, nodes[3]->node_id, &viewers[2]) == STREAM_OK); deadline = get_time_tb() + 300000; - while (relay_data[2] < 2 && get_time_tb() < deadline) test_clock_poll(ua, 100); - CHECK(relay_data[2] == 2 && relay_requests_seen && !relay_data[1]); + while (relay_data[2] < 3 && get_time_tb() < deadline) test_clock_poll(ua, 100); + CHECK(relay_data[2] == 3 && relay_requests_seen && !relay_data[1]); + CHECK(get_time_tb() - started < 10000 && relay_race.request_tb - started >= 990); + test_clock_run_for(ua, 1000); CHECK(relay_race.accepted == 255); + CHECK(relay_race.helper_subscribes && !relay_race.source_subscribes); + relay_release_info(); test_clock_run_for(ua, 1000); + CHECK(!relay_race.source_subscribes && relay_data[2] == 3); + relay_race.check_starts = 0; + DEBUG_INFO(DEBUG_CATEGORY_STREAM, "[PASS] bootstrap helper wins: elapsed_ms=%llu, n=8 r=2, common_start=%llu, stale direct INFO ignored", + (unsigned long long)((get_time_tb() - started) / 10), (unsigned long long)relay_race.start); CHECK(stream_local_get(nodes[1], &origin.descriptor, &local) == STREAM_OK && local.relay_only); /* Поздний обычный подписчик также удерживает раздачу, даже без собственного запроса помощи. */ CHECK(stream_subscribe_open(nodes[0], GROUP2, nodes[3]->node_id, &viewers[0]) == STREAM_OK); @@ -1163,9 +1235,9 @@ static void test_supernode_relay(void) { stream_close(viewers[2]); stream_close(publisher); relay_drop_source = 0; relay_role(1, 0, 101); CHECK(etcp_router_bind(nodes[1], ETCP_RT_ID_STREAM, relay_route_handler) == 0); - for (unsigned i = 0; i < 3; i++) { + for (unsigned i = 0; i < 4; i++) { CHECK(etcp_bind(nodes[i], ETCP_ID_STREAM, relay_raw_handlers[i]) == 0); - stream_set_event_callback(nodes[i], i < 2 ? on_stream : NULL, NULL); + if (i < 3) stream_set_event_callback(nodes[i], i < 2 ? on_stream : NULL, NULL); } for (unsigned i = 0; i < 3; i++) topo_group_peer_close(connections[i]); debug_set_category_level(DEBUG_CATEGORY_ETCPROUTE, route_level);