From ef48fa2a373cdc5acc2d68402e2c5bf8cd9184ca Mon Sep 17 00:00:00 2001 From: evgeny Date: Sun, 11 Oct 2026 14:05:25 +0200 Subject: [PATCH] Enable automatic upstream selection for subscriptions with lifecycle-safe cleanup --- src/Makefile.am | 2 ++ src/stream/stream.c | 67 ++++++++++++++++++++++++++++++++----- src/stream/stream_network.c | 1 + 3 files changed, 62 insertions(+), 8 deletions(-) diff --git a/src/Makefile.am b/src/Makefile.am index 1e2a9f22..5a6cd104 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -7,6 +7,8 @@ stream_core_files = \ stream/stream.h \ stream/stream.c \ stream/stream_priv.h \ + stream/stream_optimizer.h \ + stream/stream_optimizer.c \ stream/stream_wire.h \ stream/stream_wire.c \ stream/stream_network.h \ diff --git a/src/stream/stream.c b/src/stream/stream.c index 76f6de14..b3f63252 100644 --- a/src/stream/stream.c +++ b/src/stream/stream.c @@ -10,6 +10,7 @@ #include "stream_member.h" #include "stream_source.h" #include "stream_priv.h" +#include "stream_optimizer.h" struct stream_session { struct stream_session *next; @@ -20,13 +21,16 @@ struct stream_session { struct stream_buffer *buffer; uint8_t *output; struct stream_metrics *metrics; + struct stream_optimizer *optimizer; void *delivery; uint64_t next_seq; uint64_t start_seq, last_seq, passed_before; + uint64_t restart_from, restart_last; uint8_t end[STREAM_END_SIZE]; int end_valid; unsigned owners, holds; int anchored, closing, invalidated, finished, end_delivered, reason; + int restarting, restart_gap; }; struct STREAM_HANDLE { @@ -119,6 +123,7 @@ static void session_free(struct stream_session *s) { while (*link != s) link = &(*link)->next; *link = s->next; if (s->delivery) uasync_call_soon_cancel(s->service->inst->ua, s->delivery); + stream_optimizer_destroy(s->optimizer); if (s->buffer) stream_buffer_set_observer(s->buffer, NULL, NULL); if (s->source) stream_source_destroy(s->source); else stream_buffer_destroy(s->buffer); @@ -142,6 +147,7 @@ static void session_schedule(struct stream_session *s) { static void session_close(struct stream_session *s, int reason, int update_profile, int notify) { if (s->closing) return; s->closing = 1; s->reason = reason; s->owners = 0; + stream_optimizer_destroy(s->optimizer); s->optimizer = NULL; stream_network_cancel(s->service->network, &s->descriptor); for (struct STREAM_HANDLE *h = s->service->handles; h; h = h->next) if (h->session == s) { h->session = NULL; h->reason = reason; } @@ -166,6 +172,9 @@ static void session_deliver(void *arg) { else if (s->invalidated) { DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "invalidated stream awaits deferred close source=%llx", (unsigned long long)s->descriptor.source_node_id); + } else if (s->restart_gap) { + event.type = STREAM_EVENT_GAP; event.first_seq = s->restart_from; event.last_seq = s->restart_last; + s->restart_gap = 0; emit = again = 1; } else if (s->anchored) { struct stream_buffer_stats stats; stream_buffer_get_stats(s->buffer, &stats); @@ -321,6 +330,9 @@ void stream_service_destroy(struct UTUN_INSTANCE *inst) { service->stopping = 1; for (struct stream_session *s = service->sessions; s; s = s->next) if (s->buffer) stream_buffer_set_observer(s->buffer, NULL, NULL); + for (struct stream_session *s = service->sessions; s; s = s->next) { + stream_optimizer_destroy(s->optimizer); s->optimizer = NULL; + } stream_network_destroy(service->network); service->network = NULL; if (service->registered) member_sync_remove_props_cbk(inst, member_changed, service); if (service->reconcile) uasync_call_soon_cancel(inst->ua, service->reconcile); @@ -354,12 +366,12 @@ void stream_set_event_callback(struct UTUN_INSTANCE *inst, stream_event_fn callb } /* Подготовить все ресурсы до публикации member-опций; rollback не оставляет видимую подписку. */ -static int session_open(struct UTUN_INSTANCE *inst, const struct stream_descriptor *d, const uint8_t key[32], int publisher, +static int session_open(struct UTUN_INSTANCE *inst, const struct stream_descriptor *d, const uint8_t key[32], int publisher, int automatic, struct STREAM_HANDLE **out) { struct stream_service *service = inst->stream; struct stream_session *s = session_find(service, d->group_id, d->source_node_id); int created = s == NULL; - if (s && (publisher || s->descriptor.source_epoch != d->source_epoch)) { + if (s && (publisher || s->descriptor.source_epoch != d->source_epoch || (!s->source && !!s->optimizer != automatic))) { DEBUG_WARN(DEBUG_CATEGORY_STREAM, "stream source already owned with another publication/epoch"); return STREAM_ERR_STATE; } struct STREAM_HANDLE *h = u_calloc(1, sizeof(*h)); @@ -375,6 +387,7 @@ static int session_open(struct UTUN_INSTANCE *inst, const struct stream_descript result = stream_source_create(&s->source, inst->ua, d, inst->my_ed25519_privkey, source_ready, s); if (result == STREAM_OK) s->buffer = stream_source_buffer(s->source); } else if (result == STREAM_OK) result = stream_buffer_create(&s->buffer, d, key, STREAM_BUFFER_BYTES); + if (result == STREAM_OK && automatic && !publisher) result = stream_optimizer_create(inst, d, &s->optimizer); s->next = service->sessions; service->sessions = s; if (result != STREAM_OK) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream session initialization failed result=%d", result); @@ -434,14 +447,14 @@ int stream_publish_open(struct UTUN_INSTANCE *inst, uint64_t group_id, const str } } while (!d.source_epoch); if (stream_descriptor_validate(&d) != STREAM_OK) return STREAM_ERR_ARGUMENT; - return session_open(inst, &d, inst->my_ed25519_pubkey, 1, out); + return session_open(inst, &d, inst->my_ed25519_pubkey, 1, 0, out); } -int stream_subscribe_open(struct UTUN_INSTANCE *inst, uint64_t group_id, uint64_t source_id, struct STREAM_HANDLE **out) { +static int subscribe_open(struct UTUN_INSTANCE *inst, uint64_t group_id, uint64_t source_id, int automatic, struct STREAM_HANDLE **out) { int result = stream_open_check(inst, group_id, out); if (result != STREAM_OK) return result; struct stream_session *s = session_find(inst->stream, group_id, source_id); - if (s) return session_open(inst, &s->descriptor, s->public_key, 0, out); + if (s) return session_open(inst, &s->descriptor, s->public_key, 0, automatic, out); struct stream_member_state state; uint8_t key[32]; result = stream_member_load(inst, group_id, source_id, &state, key); if (result != STREAM_OK) return result; @@ -449,7 +462,29 @@ int stream_subscribe_open(struct UTUN_INSTANCE *inst, uint64_t group_id, uint64_ DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "source not yet published source=%llx", (unsigned long long)source_id); return STREAM_WAIT; } - return session_open(inst, &state.source, key, 0, out); + return session_open(inst, &state.source, key, 0, automatic, out); +} + +int stream_subscribe_open(struct UTUN_INSTANCE *inst, uint64_t group_id, uint64_t source_id, struct STREAM_HANDLE **out) { + return subscribe_open(inst, group_id, source_id, 1, out); +} + +int stream_subscribe_manual_open(struct UTUN_INSTANCE *inst, uint64_t group_id, uint64_t source_id, struct STREAM_HANDLE **out) { + return subscribe_open(inst, group_id, source_id, 0, out); +} + +int stream_selection_configure(struct STREAM_HANDLE *h, const struct stream_selection_settings *settings) { + if (!stream_handle_active(h) || !h->session->optimizer) { + DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream selection configuration requires automatic receiver"); return STREAM_ERR_STATE; + } + return stream_optimizer_configure(h->session->optimizer, settings); +} + +int stream_selection_get_info(const struct STREAM_HANDLE *h, struct stream_selection_info *info) { + if (!stream_handle_active(h) || !info) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream selection info invalid handle/output"); return STREAM_ERR_STATE; } + memset(info, 0, sizeof(*info)); + if (h->session->optimizer) stream_optimizer_info(h->session->optimizer, info); + return STREAM_OK; } /* Publisher авторитетно останавливает трансляцию; reader снимает только собственное владение. */ @@ -554,7 +589,11 @@ int stream_ingest_path(struct UTUN_INSTANCE *inst, const uint8_t *wire, size_t l struct stream_path retained = *path; retained.nodes[retained.count++] = inst->node_id; int result = stream_buffer_put_path(s->buffer, wire, length, &retained); if (result == STREAM_OK || (result == STREAM_DUPLICATE && !s->anchored)) { - if (!s->anchored) { s->anchored = 1; s->start_seq = s->next_seq = stream_read_be(wire + STREAM_BLOCK_SEQ_OFF, 8); } + if (!s->anchored) { + s->anchored = 1; s->start_seq = s->next_seq = stream_read_be(wire + STREAM_BLOCK_SEQ_OFF, 8); + if (s->restarting && s->next_seq > s->restart_from) { s->restart_gap = 1; s->restart_last = s->next_seq - 1; } + s->restarting = 0; + } session_schedule(s); stream_network_changed(s->service->network); } @@ -568,7 +607,7 @@ int stream_local_get(struct UTUN_INSTANCE *inst, const struct stream_descriptor if (!key || !local || !inst || !inst->stream || inst->stream->stopping) return STREAM_WAIT; struct stream_session *s = session_find(inst->stream, key->group_id, key->source_node_id); if (!s || s->descriptor.source_epoch != key->source_epoch) return STREAM_WAIT; - *local = (struct stream_local){.descriptor = s->descriptor, .buffer = s->buffer, .start = s->start_seq, .last = s->last_seq, + *local = (struct stream_local){.descriptor = s->descriptor, .buffer = s->buffer, .start = s->start_seq, .last = s->last_seq, .next = s->next_seq, .anchored = s->anchored, .publisher = s->source != NULL, .finished = s->end_valid, .end = s->end_valid ? s->end : NULL}; return STREAM_OK; } @@ -624,12 +663,24 @@ int stream_local_restart(struct UTUN_INSTANCE *inst, const struct stream_descrip } struct stream_session *s = session_find(inst->stream, key->group_id, key->source_node_id); if (s->delivery) { uasync_call_soon_cancel(inst->ua, s->delivery); s->delivery = NULL; } + if (s->anchored) { s->restarting = 1; s->restart_from = s->next_seq; } s->anchored = s->end_delivered = 0; s->passed_before = s->next_seq = s->start_seq = 0; DEBUG_INFO(DEBUG_CATEGORY_STREAM, "stream receiver restart from end group=%llx source=%llx epoch=%llx retained=%u", (unsigned long long)key->group_id, (unsigned long long)key->source_node_id, (unsigned long long)key->source_epoch, local.buffer != NULL); return STREAM_OK; } +/* Заимствованный локальный handle нужен только на время tick; владельцем остаётся сама подписка. */ +void stream_optimizers_tick(struct UTUN_INSTANCE *inst, uint64_t now_ms) { + struct stream_service *service = inst ? inst->stream : NULL; + if (!service || service->stopping) return; + for (struct stream_session *s = service->sessions; s; s = s->next) { + if (!s->optimizer || s->closing || s->invalidated) continue; + struct STREAM_HANDLE subscription = {.service = service, .session = s}; + stream_optimizer_tick(s->optimizer, &subscription, now_ms); + } +} + /* Выборка идёт из live BGP-путей группы, не из nodes.online или глобальных соединений. */ int stream_get_candidates(const struct STREAM_HANDLE *h, struct stream_candidate **out, size_t *count) { if (!out || *out || !count || !stream_handle_active(h)) { diff --git a/src/stream/stream_network.c b/src/stream/stream_network.c index 49beaed2..b31ae554 100644 --- a/src/stream/stream_network.c +++ b/src/stream/stream_network.c @@ -861,6 +861,7 @@ static void network_tick(void *arg) { p = next; } pressure_tick(net, now_ms); + stream_optimizers_tick(net->inst, now_ms); net->next_tick_ms = get_time_tb() / 10 + STREAM_TICK_TB / 10; net->timer = uasync_set_timeout(net->inst->ua, STREAM_TICK_TB, net, network_tick, "stream_network"); if (!net->timer) DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "stream network timer scheduling failed");