diff --git a/src/routing_layer/multicast.c b/src/routing_layer/multicast.c new file mode 100644 index 00000000..e1abbe0c --- /dev/null +++ b/src/routing_layer/multicast.c @@ -0,0 +1,643 @@ +#include +#include +#include +#include "../lib/debug_config.h" +#include "../lib/mem.h" +#include "../lib/u_async.h" +#include "multicast.h" +#include "etcp.h" +#include "etcp_api.h" +#include "etcp_router.h" +#include "pkt_normalizer.h" +#include "topo_group.h" +#include "utun_instance.h" + +#define MC_CAT DEBUG_CATEGORY_ETCPROUTE +#define MC_MAX_RECIPIENTS ((PKTNORM_MAX_DGRAM_SIZE - MULTICAST_HEADER_SIZE - MULTICAST_PATH_RESERVE) / 8) +#define MC_REPORT_CHUNK ((PKTNORM_MAX_DGRAM_SIZE - SVC_ROUTE_HDR_SIZE - ROUTER_PATH_MAX_SIZE - MULTICAST_REPORT_HEADER_SIZE) / 8) + +struct mc_packet { + uint64_t group, source; + uint16_t descriptor, count; + const uint8_t *recipients, *payload, *visited; + size_t payload_len; + uint8_t hops; +}; + +struct mc_out { + struct mc_out* next; + struct multicast_ctx* ctx; + struct ETCP_CONN* conn; /* заимствован до DELETE; затем NULL до опустошения FIFO */ + struct ll_queue* q; + struct queue_waiter_handle waiter; +}; + +struct mc_window { + struct mc_window* next; + uint64_t group, source, dst, deadline; + uint16_t descriptor; + int dirty; /* есть потери после первого немедленного отчёта */ +}; + +struct mc_report { + struct mc_report* next; + struct multicast_ctx* ctx; + struct multicast_loss loss; + uint64_t due, retry; /* неизменный конец окна / следующая попытка передачи */ + uint16_t cursor; /* уже переданные router получатели длинного отчёта */ + struct queue_waiter_handle waiter; + uint64_t ids[]; +}; + +struct multicast_ctx { + struct UTUN_INSTANCE* inst; + struct mc_out* outputs; + struct mc_window* windows; + struct mc_report* reports; + void *report_timer, *route_wake; + uint64_t eviction_warn; + struct multicast_stats stats; + multicast_recv_fn recv; + multicast_loss_fn loss; + void* arg; +}; + +struct mc_target { struct ETCP_CONN* conn; uint64_t id; }; + +static void mc_report_tick(void* arg); +static void mc_report_schedule(struct multicast_ctx* ctx); +static void mc_forward(struct multicast_ctx* ctx, struct ll_entry* e, struct mc_out* previous); +static void mc_out_drain(struct ll_queue* q, void* arg); + +static uint64_t mc_read(const uint8_t* p, unsigned bytes) { + uint64_t v = 0; + for (unsigned i = 0; i < bytes; i++) v |= (uint64_t)p[i] << (i * 8); + return v; +} + +static void mc_write(uint8_t* p, uint64_t v, unsigned bytes) { + for (unsigned i = 0; i < bytes; i++) p[i] = (uint8_t)(v >> (i * 8)); +} + +static void mc_free(struct ll_entry* e) { queue_dgram_free(e); queue_entry_free(e); } + +/* Разбирает только границы; полный контроль идентичностей выполняется на входе. */ +static int mc_parse(const struct ll_entry* e, struct mc_packet* p) { + if (!e || !e->dgram || e->len < MULTICAST_HEADER_SIZE + 9 || e->len > PKTNORM_MAX_DGRAM_SIZE) return -1; + const uint8_t* w = e->dgram; + p->group = mc_read(w + 1, 8); p->descriptor = mc_read(w + 9, 2); + p->count = mc_read(w + 11, 2); p->source = mc_read(w + 13, 8); p->hops = w[e->len - 1]; + size_t head = MULTICAST_HEADER_SIZE + (size_t)p->count * 8, tail = (size_t)p->hops * 8 + 1; + if (w[0] != ETCP_ID_MULTICAST || !p->source || !p->count || p->count > MC_MAX_RECIPIENTS || + !p->hops || p->hops > MULTICAST_MAX_VISITED || head + tail > e->len) return -1; + p->recipients = w + MULTICAST_HEADER_SIZE; p->payload = w + head; + p->payload_len = e->len - head - tail; p->visited = w + e->len - tail; + if (head + p->payload_len + MULTICAST_PATH_RESERVE > PKTNORM_MAX_DGRAM_SIZE) return -1; + return 0; +} + +static int mc_has_visited(const struct mc_packet* p, uint64_t id) { + for (unsigned i = 0; i < p->hops; i++) if (mc_read(p->visited + i * 8, 8) == id) return 1; + return 0; +} + +/* Проверка канонического списка не требует выделений и не допускает повторной доставки. */ +static int mc_ids_valid(const uint8_t* ids, uint16_t count) { + uint64_t previous = 0; + for (unsigned i = 0; i < count; i++) { + uint64_t id = mc_read(ids + i * 8, 8); + if (id <= previous) return 0; + previous = id; + } + return count != 0; +} + +/* Собирает ветку с тем же payload и уже дополненным путём. */ +static struct ll_entry* mc_build(const struct mc_packet* p, const struct mc_target* targets, unsigned count) { + size_t len = MULTICAST_HEADER_SIZE + count * 8 + p->payload_len + p->hops * 8 + 1; + struct ll_entry* e = ll_alloc_lldgram((uint16_t)len); + if (!e) { DEBUG_ERROR(MC_CAT, "multicast packet allocation failed bytes=%zu recipients=%u", len, count); return NULL; } + uint8_t* w = e->dgram; + w[0] = ETCP_ID_MULTICAST; mc_write(w + 1, p->group, 8); mc_write(w + 9, p->descriptor, 2); + mc_write(w + 11, count, 2); mc_write(w + 13, p->source, 8); + for (unsigned i = 0; i < count; i++) mc_write(w + MULTICAST_HEADER_SIZE + i * 8, targets[i].id, 8); + size_t off = MULTICAST_HEADER_SIZE + count * 8; + if (p->payload_len) memcpy(w + off, p->payload, p->payload_len); + memcpy(w + off + p->payload_len, p->visited, p->hops * 8); w[len - 1] = p->hops; + return e; +} + +/* Запись всегда удаляется вместе с waiter, в том числе уже запланированным. */ +static void mc_report_remove(struct multicast_ctx* ctx, struct mc_report* r) { + struct mc_report** pp = &ctx->reports; + while (*pp != r) pp = &(*pp)->next; + *pp = r->next; + etcp_router_cancel_send_ready(ctx->inst, r->loss.group_id, r->loss.source_node_id, ETCP_RT_ID_MULTICAST_REPORT, &r->waiter); + ctx->stats.report_bytes -= sizeof(*r) + (size_t)r->loss.recipient_count * 8; + u_free(r); +} + +static void mc_window_remove(struct multicast_ctx* ctx, struct mc_window** pp) { + struct mc_window* w = *pp; + *pp = w->next; ctx->stats.report_bytes -= sizeof(*w); u_free(w); +} + +/* Общий жёсткий бюджет включает сами записи, их ID и состояние троттлинга. */ +static int mc_report_room(struct multicast_ctx* ctx, size_t bytes) { + if (bytes > MULTICAST_REPORT_BYTES) { + DEBUG_ERROR(MC_CAT, "multicast report exceeds budget bytes=%zu limit=%u", bytes, MULTICAST_REPORT_BYTES); return -1; + } + while (ctx->stats.report_bytes + bytes > MULTICAST_REPORT_BYTES) { + if (ctx->reports) { + ctx->stats.reports_evicted++; + mc_report_remove(ctx, ctx->reports); + } else if (ctx->windows) mc_window_remove(ctx, &ctx->windows); + else break; + uint64_t now = get_time_tb(); + if (now >= ctx->eviction_warn) { + DEBUG_WARN(MC_CAT, "multicast report budget exhausted bytes=%zu need=%zu evicted=%llu", + ctx->stats.report_bytes, bytes, (unsigned long long)ctx->stats.reports_evicted); + ctx->eviction_warn = now + MULTICAST_REPORT_INTERVAL_TB; + } + } + return 0; +} + +/* Сохраняет потерю до освобождения пакета. Удалённому источнику сообщается только OVERLOAD. */ +static void mc_loss(struct multicast_ctx* ctx, const struct mc_packet* p, uint64_t dst, + enum multicast_loss_reason reason, const struct mc_target* targets, unsigned count) { + if (reason == MULTICAST_OVERLOAD) ctx->stats.overload_packets++; + else if (reason == MULTICAST_NO_ROUTE) ctx->stats.no_route_packets++; + else if (reason == MULTICAST_PATH_LOOP) ctx->stats.loop_packets++; + else ctx->stats.internal_errors++; + if (reason != MULTICAST_OVERLOAD) { + DEBUG_WARN(MC_CAT, "multicast drop group=%016llx source=%016llx descriptor=%u dst=%016llx recipients=%u reason=%d", + (unsigned long long)p->group, (unsigned long long)p->source, p->descriptor, + (unsigned long long)dst, count, reason); + if (p->source != ctx->inst->node_id) return; + } + uint64_t now = get_time_tb(), due = now; + struct mc_window* window = NULL; + if (reason == MULTICAST_OVERLOAD) { + for (window = ctx->windows; window; window = window->next) + if (window->group == p->group && window->source == p->source && + window->descriptor == p->descriptor && window->dst == dst) break; + if (window) { + /* Просроченное окно сначала завершит таймер; поздний пакет относится к новому окну. */ + if (window->deadline <= now) window->deadline = now + MULTICAST_REPORT_INTERVAL_TB; + due = window->deadline; window->dirty = 1; + } + } + for (struct mc_report* r = ctx->reports; r; r = r->next) { + if (r->due != due || r->cursor || r->loss.group_id != p->group || r->loss.source_node_id != p->source || + r->loss.dst != dst || r->loss.descriptor != p->descriptor || r->loss.reason != reason || + r->loss.recipient_count != count || r->loss.packets == UINT32_MAX) continue; + unsigned i = 0; + while (i < count && r->ids[i] == targets[i].id) i++; + if (i == count) { r->loss.packets++; mc_report_schedule(ctx); return; } + } + size_t bytes = sizeof(struct mc_report) + count * 8; + int new_window = reason == MULTICAST_OVERLOAD && !window; + if (mc_report_room(ctx, bytes + (new_window ? sizeof(struct mc_window) : 0)) < 0) return; + /* room может вытеснить окно; после него старый указатель больше не используется. */ + if (new_window) { + window = u_calloc(1, sizeof(*window)); + if (!window) { DEBUG_ERROR(MC_CAT, "multicast throttle allocation failed"); return; } + window->group = p->group; window->source = p->source; window->dst = dst; window->descriptor = p->descriptor; + window->deadline = now + MULTICAST_REPORT_INTERVAL_TB; + struct mc_window** tail = &ctx->windows; + while (*tail) tail = &(*tail)->next; + *tail = window; ctx->stats.report_bytes += sizeof(*window); + } + struct mc_report* r = u_calloc(1, bytes); + if (!r) { DEBUG_ERROR(MC_CAT, "multicast loss allocation failed bytes=%zu", bytes); return; } + r->ctx = ctx; r->due = due; r->retry = due; + r->loss = (struct multicast_loss){ .group_id = p->group, .source_node_id = p->source, + .src = ctx->inst->node_id, .dst = dst, .descriptor = p->descriptor, .recipient_count = count, + .recipients = r->ids, .packets = 1, .reason = reason, .prefix_len = p->payload_len < 16 ? p->payload_len : 16 }; + if (r->loss.prefix_len) memcpy(r->loss.prefix, p->payload, r->loss.prefix_len); + for (unsigned i = 0; i < count; i++) r->ids[i] = targets[i].id; + struct mc_report** tail = &ctx->reports; + while (*tail) tail = &(*tail)->next; + *tail = r; ctx->stats.report_bytes += bytes; + mc_report_schedule(ctx); +} + +/* Освобождение router-очереди только будит общий таймер; нет рекурсивной отправки. */ +static void mc_report_ready(struct ll_queue* q, void* arg) { + (void)q; + struct mc_report* r = arg; + r->retry = get_time_tb(); mc_report_schedule(r->ctx); +} + +/* Возвращает 1 после передачи всех частей, 0 при backpressure, -1 при ошибке отправки. */ +static int mc_report_send(struct multicast_ctx* ctx, struct mc_report* r) { + if (r->loss.source_node_id == ctx->inst->node_id) { + if (ctx->loss) ctx->loss(ctx->inst, &r->loss, ctx->arg); + ctx->stats.reports_sent++; ctx->stats.reports_received++; return 1; + } + while (r->cursor < r->loss.recipient_count) { + if (!etcp_router_send_q_has_room(ctx->inst, r->loss.group_id, r->loss.source_node_id, ETCP_RT_ID_MULTICAST_REPORT)) { + etcp_router_on_send_ready(ctx->inst, r->loss.group_id, r->loss.source_node_id, + ETCP_RT_ID_MULTICAST_REPORT, &r->waiter, mc_report_ready, r); + return 0; + } + unsigned count = r->loss.recipient_count - r->cursor; + if (count > MC_REPORT_CHUNK) count = MC_REPORT_CHUNK; + struct ll_entry* e = ll_alloc_lldgram(1 + MULTICAST_REPORT_HEADER_SIZE + count * 8); + if (!e) { DEBUG_ERROR(MC_CAT, "multicast report packet allocation failed recipients=%u", count); return -1; } + uint8_t* w = e->dgram + 1; e->dgram[0] = ETCP_RT_ID_MULTICAST_REPORT; + mc_write(w, r->loss.descriptor, 2); mc_write(w + 2, r->loss.src, 8); mc_write(w + 10, r->loss.dst, 8); + mc_write(w + 18, r->loss.packets, 4); mc_write(w + 22, count, 2); w[24] = r->loss.prefix_len; + memcpy(w + 25, r->loss.prefix, 16); + for (unsigned i = 0; i < count; i++) mc_write(w + MULTICAST_REPORT_HEADER_SIZE + i * 8, r->ids[r->cursor + i], 8); + if (etcp_route_send(ctx->inst, r->loss.group_id, r->loss.source_node_id, e, 0, 0) < 0) { + DEBUG_WARN(MC_CAT, "multicast report send failed source=%016llx descriptor=%u cursor=%u", + (unsigned long long)r->loss.source_node_id, r->loss.descriptor, r->cursor); return -1; + } + r->cursor += count; ctx->stats.reports_sent++; + } + return 1; +} + +/* Фиксированные окна заканчиваются независимо от нового трафика; blocked-отчёты не смешиваются с новыми. */ +static void mc_report_tick(void* arg) { + struct multicast_ctx* ctx = arg; + ctx->report_timer = NULL; + uint64_t now = get_time_tb(); + for (struct mc_window** pp = &ctx->windows; *pp;) { + struct mc_window* w = *pp; + if (w->deadline > now) { pp = &w->next; continue; } + if (!w->dirty) { mc_window_remove(ctx, pp); continue; } + w->dirty = 0; w->deadline = now + MULTICAST_REPORT_INTERVAL_TB; pp = &w->next; + } + unsigned processed = 0; + for (;;) { + struct mc_report* r = ctx->reports; + while (r && r->retry > now) r = r->next; + if (!r || processed++ == 16) break; + /* Callback приложения может отправлять новые пакеты: не сохраняем курсор списка через него. + * Временно снимаем запись с бюджета/списка, чтобы eviction не освободил её из callback. */ + if (r->loss.source_node_id == ctx->inst->node_id) { + struct mc_report** pp = &ctx->reports; + while (*pp != r) pp = &(*pp)->next; + *pp = r->next; ctx->stats.report_bytes -= sizeof(*r) + r->loss.recipient_count * 8; + mc_report_send(ctx, r); u_free(r); continue; + } + r->retry = now + MULTICAST_REPORT_INTERVAL_TB; + int result = mc_report_send(ctx, r); + if (result == 1) { + DEBUG_WARN(MC_CAT, "multicast overload report group=%016llx source=%016llx descriptor=%u link=%016llx->%016llx recipients=%u packets=%u", + (unsigned long long)r->loss.group_id, (unsigned long long)r->loss.source_node_id, r->loss.descriptor, + (unsigned long long)r->loss.src, (unsigned long long)r->loss.dst, r->loss.recipient_count, r->loss.packets); + mc_report_remove(ctx, r); + } + } + mc_report_schedule(ctx); +} + +/* Единственный таймер выбирает ближайшее окно или повтор, в покое таймера нет. */ +static void mc_report_schedule(struct multicast_ctx* ctx) { + uint64_t due = UINT64_MAX, now = get_time_tb(); + for (struct mc_window* w = ctx->windows; w; w = w->next) if (w->deadline < due) due = w->deadline; + for (struct mc_report* r = ctx->reports; r; r = r->next) if (r->retry < due) due = r->retry; + if (ctx->report_timer) { uasync_cancel_timeout(ctx->inst->ua, ctx->report_timer); ctx->report_timer = NULL; } + if (due == UINT64_MAX) return; + int delay = due <= now ? 0 : (due - now > INT_MAX ? INT_MAX : (int)(due - now)); + ctx->report_timer = uasync_set_timeout(ctx->inst->ua, delay, ctx, mc_report_tick, "multicast_reports"); + if (!ctx->report_timer) DEBUG_ERROR(MC_CAT, "multicast report timer allocation failed pending_bytes=%zu", ctx->stats.report_bytes); +} + +/* Удаление FIFO только после отмены waiter; callbacks очереди отменяет queue_free. */ +static void mc_out_destroy(struct multicast_ctx* ctx, struct mc_out* o) { + if (o->conn) queue_waiter_cancel(o->conn->send_input_q, &o->waiter); + struct mc_out** pp = &ctx->outputs; + while (*pp != o) pp = &(*pp)->next; + *pp = o->next; + DEBUG_DEBUG(MC_CAT, "multicast queue destroy peer=%016llx packets=%d bytes=%zu", + (unsigned long long)(o->conn ? o->conn->peer_node_id : 0), o->q->count, o->q->total_bytes); + queue_set_callback(o->q, NULL, NULL); + struct ll_entry* e; + while ((e = queue_data_get(o->q))) mc_free(e); + queue_free(o->q); u_free(o); +} + +/* Единственная FIFO на транспорт, без групповых или descriptor-подочередей. */ +static struct mc_out* mc_out_get(struct multicast_ctx* ctx, struct ETCP_CONN* conn) { + for (struct mc_out* o = ctx->outputs; o; o = o->next) if (o->conn == conn) return o; + struct mc_out* o = u_calloc(1, sizeof(*o)); + if (!o) { DEBUG_ERROR(MC_CAT, "multicast output allocation failed"); return NULL; } + o->q = queue_new(ctx->inst->ua, 0, 0, 0, "multicast_out"); + if (!o->q) { DEBUG_ERROR(MC_CAT, "multicast FIFO allocation failed"); u_free(o); return NULL; } + o->ctx = ctx; o->conn = conn; o->next = ctx->outputs; ctx->outputs = o; + queue_set_callback_defer(o->q, 1); queue_set_callback(o->q, mc_out_drain, o); + DEBUG_DEBUG(MC_CAT, "multicast queue created peer=%016llx limit=%u", (unsigned long long)conn->peer_node_id, MULTICAST_QUEUE_BYTES); + return o; +} + +/* Сортировка по next hop, внутри ветки — по ID: канонический wire сохраняется после split. */ +static int mc_target_compare(const void* a, const void* b) { + const struct mc_target *x = a, *y = b; + uintptr_t xc = (uintptr_t)x->conn, yc = (uintptr_t)y->conn; + if (xc != yc) return xc < yc ? -1 : 1; + return (x->id > y->id) - (x->id < y->id); +} + +/* Выбирает маршруты один раз на обработку, изолирует потери каждой исходящей ветки. */ +static void mc_forward(struct multicast_ctx* ctx, struct ll_entry* e, struct mc_out* previous) { + struct mc_packet p; + if (mc_parse(e, &p) < 0) { + ctx->stats.invalid_packets++; DEBUG_ERROR(MC_CAT, "multicast invalid internal packet len=%u", e->len); mc_free(e); return; + } + struct TOPO_GROUP* group = topo_groups_find(ctx->inst->topo_groups, p.group); + if (!group || group->stopping) { + DEBUG_WARN(MC_CAT, "multicast group unavailable group=%016llx", (unsigned long long)p.group); mc_free(e); return; + } + struct mc_target targets[MC_MAX_RECIPIENTS]; + unsigned count = 0; + for (unsigned i = 0; i < p.count; i++) { + uint64_t id = mc_read(p.recipients + i * 8, 8); + if (id == ctx->inst->node_id) { + ctx->stats.delivered++; + if (ctx->recv) ctx->recv(ctx->inst, p.group, p.source, p.descriptor, p.payload, p.payload_len, ctx->arg); + else DEBUG_WARN(MC_CAT, "multicast no receiver group=%016llx descriptor=%u", (unsigned long long)p.group, p.descriptor); + continue; + } + targets[count++] = (struct mc_target){ topo_group_find_conn_for_node(group, id), id }; + } + qsort(targets, count, sizeof(*targets), mc_target_compare); + for (unsigned start = 0; start < count;) { + unsigned end = start + 1; + struct ETCP_CONN* conn = targets[start].conn; + while (end < count && targets[end].conn == conn) end++; + unsigned n = end - start; + if (!conn) mc_loss(ctx, &p, 0, MULTICAST_NO_ROUTE, targets + start, n); + else if (mc_has_visited(&p, conn->peer_node_id)) + mc_loss(ctx, &p, conn->peer_node_id, MULTICAST_PATH_LOOP, targets + start, n); + else { + struct mc_out* o = mc_out_get(ctx, conn); + size_t size = MULTICAST_HEADER_SIZE + n * 8 + p.payload_len + p.hops * 8 + 1; + if (!o) mc_loss(ctx, &p, conn->peer_node_id, MULTICAST_INTERNAL_ERROR, targets + start, n); + else if (o->q->total_bytes + size > MULTICAST_QUEUE_BYTES) { + DEBUG_TRACE(MC_CAT, "multicast overflow source=%016llx descriptor=%u peer=%016llx bytes=%zu packet=%zu recipients=%u", + (unsigned long long)p.source, p.descriptor, (unsigned long long)conn->peer_node_id, + o->q->total_bytes, size, n); + mc_loss(ctx, &p, conn->peer_node_id, MULTICAST_OVERLOAD, targets + start, n); + } else { + struct ll_entry* branch = mc_build(&p, targets + start, n); + if (!branch) mc_loss(ctx, &p, conn->peer_node_id, MULTICAST_INTERNAL_ERROR, targets + start, n); + else { + DEBUG_TRACE(MC_CAT, "multicast split source=%016llx descriptor=%u peer=%016llx recipients=%u bytes=%u", + (unsigned long long)p.source, p.descriptor, (unsigned long long)conn->peer_node_id, n, branch->len); + if (o == previous) queue_data_put_first(o->q, branch); + else queue_data_put(o->q, branch); + } + } + } + start = end; + } + mc_free(e); +} + +/* Проверяет каждый адрес: изменение части назначений требует повторного разделения. */ +static int mc_same_route(struct mc_out* o, const struct ll_entry* e) { + struct mc_packet p; + if (!o->conn || o->conn->state == 2 || o->conn->close_requested || !o->conn->links_up || mc_parse(e, &p) < 0) return 0; + struct TOPO_GROUP* group = topo_groups_find(o->ctx->inst->topo_groups, p.group); + if (!group || group->stopping) return 0; + for (unsigned i = 0; i < p.count; i++) + if (topo_group_find_conn_for_node(group, mc_read(p.recipients + i * 8, 8)) != o->conn) return 0; + return 1; +} + +static void mc_out_ready(struct ll_queue* q, void* arg) { + (void)q; + struct mc_out* o = arg; + queue_resume_callback(o->q); +} + +/* Автозабор FIFO: один пакет на свободную send_input_q, затем снова round-robin waiter. */ +static void mc_out_drain(struct ll_queue* q, void* arg) { + struct mc_out* o = arg; + struct multicast_ctx* ctx = o->ctx; + struct ll_entry* e = queue_data_get(q); + if (!e) { queue_resume_callback(q); return; } + if (!mc_same_route(o, e)) mc_forward(ctx, e, o); + else if (o->conn->send_input_q->count) { + queue_data_put_first(q, e); + if (queue_waiter_wait(o->conn->send_input_q, &o->waiter, mc_out_ready, o) >= 0) return; + DEBUG_ERROR(MC_CAT, "multicast send waiter allocation failed peer=%016llx", (unsigned long long)o->conn->peer_node_id); + e = queue_data_get(q); mc_free(e); ctx->stats.internal_errors++; + } else if (etcp_send(o->conn, e) < 0) { + DEBUG_ERROR(MC_CAT, "multicast transport send failed peer=%016llx", (unsigned long long)o->conn->peer_node_id); + mc_free(e); ctx->stats.internal_errors++; + } else ctx->stats.sent++; + queue_resume_callback(q); + if (!o->conn && !q->count) mc_out_destroy(ctx, o); +} + +/* Отмена изменённых назначений идёт через remove, обычный забор FIFO остаётся в drain callback. */ +static void mc_recheck(void* arg) { + struct multicast_ctx* ctx = arg; + ctx->route_wake = NULL; + for (struct mc_out* o = ctx->outputs; o;) { + struct mc_out* next = o->next; + for (struct ll_entry* e = o->q->head; e;) { + struct ll_entry* following = e->next; + if (!mc_same_route(o, e)) { + queue_remove_data(o->q, e); mc_forward(ctx, e, o); + } + e = following; + } + if (o->conn) queue_waiter_cancel(o->conn->send_input_q, &o->waiter); + queue_resume_callback(o->q); + if (!o->conn && !o->q->count) mc_out_destroy(ctx, o); + o = next; + } +} + +void multicast_routes_changed(struct UTUN_INSTANCE* inst) { + struct multicast_ctx* ctx = inst ? inst->multicast : NULL; + if (!ctx || !ctx->outputs || ctx->route_wake) return; + ctx->route_wake = uasync_call_soon(inst->ua, ctx, mc_recheck); + if (!ctx->route_wake) DEBUG_ERROR(MC_CAT, "multicast route wake allocation failed"); +} + +/* DELETE приходит до уничтожения send_input_q: waiter отменяется, указатель conn забывается. */ +static void mc_conn_status(struct ETCP_CONN* conn, int status, void* arg) { + struct multicast_ctx* ctx = arg; + if (status == ETCP_CONN_STATUS_NEW) return; + if (status == ETCP_CONN_STATUS_DELETE) { + for (struct mc_out* o = ctx->outputs; o; o = o->next) if (o->conn == conn) { + queue_waiter_cancel(conn->send_input_q, &o->waiter); o->conn = NULL; + } + } + multicast_routes_changed(ctx->inst); +} + +/* Приём DATA: проверяет путь до локальной доставки и добавляет self один раз до очередей. */ +static void mc_recv(struct ETCP_CONN* from, struct ll_entry* e) { + struct multicast_ctx* ctx = from && from->instance ? from->instance->multicast : NULL; + struct mc_packet p; + if (!ctx || mc_parse(e, &p) < 0) goto invalid; + struct TOPO_GROUP* group = topo_groups_find(ctx->inst->topo_groups, p.group); + if (!group || group->stopping || !mc_ids_valid(p.recipients, p.count)) goto invalid; + int member = 0; + for (struct ll_entry* se = group->senders_list ? group->senders_list->head : NULL; se; se = se->next) { + struct TOPO_GROUP_CONN_ITEM* peer = (struct TOPO_GROUP_CONN_ITEM*)se->data; + if (peer->conn == from && peer->accepted && !peer->transport_down) { member = 1; break; } + } + if (!member || mc_read(p.visited, 8) != p.source || mc_read(p.visited + (p.hops - 1) * 8, 8) != from->peer_node_id) + goto invalid; + for (unsigned i = 0; i < p.hops; i++) { + uint64_t id = mc_read(p.visited + i * 8, 8); + if (!id) goto invalid; + for (unsigned j = 0; j < i; j++) if (mc_read(p.visited + j * 8, 8) == id) goto invalid; + } + if (mc_has_visited(&p, ctx->inst->node_id) || p.hops == MULTICAST_MAX_VISITED) { + ctx->stats.loop_packets++; + DEBUG_WARN(MC_CAT, "multicast loop/path limit source=%016llx peer=%016llx hops=%u", + (unsigned long long)p.source, (unsigned long long)from->peer_node_id, p.hops); + mc_free(e); return; + } + struct ll_entry* expanded = ll_alloc_lldgram(e->len + 8); + if (!expanded) { DEBUG_ERROR(MC_CAT, "multicast path allocation failed"); ctx->stats.internal_errors++; mc_free(e); return; } + memcpy(expanded->dgram, e->dgram, e->len - 1); + mc_write(expanded->dgram + e->len - 1, ctx->inst->node_id, 8); expanded->dgram[expanded->len - 1] = p.hops + 1; + mc_free(e); mc_forward(ctx, expanded, NULL); return; +invalid: + if (ctx) ctx->stats.invalid_packets++; + DEBUG_WARN(MC_CAT, "multicast invalid data peer=%016llx len=%u", + (unsigned long long)(from ? from->peer_node_id : 0), e ? e->len : 0); + if (e) mc_free(e); +} + +/* Router уже доставил пакет адресу ядра; проверяем связность src и каноничность отчёта. */ +static void mc_report_recv(struct ETCP_CONN* conn, struct ll_entry* e) { + struct multicast_ctx* ctx = conn && conn->instance ? conn->instance->multicast : NULL; + if (!ctx || !e || !e->dgram || e->len < ROUTER_SVC_HDR_SIZE) goto invalid; + if (e->len == ROUTER_SVC_HDR_SIZE) { mc_free(e); return; } /* штатный router CLOSE */ + if (e->len < ROUTER_SVC_HDR_SIZE + MULTICAST_REPORT_HEADER_SIZE) goto invalid; + uint64_t source, dst, group; + memcpy(&source, e->dgram + ROUTER_SVC_SRC_OFF, 8); memcpy(&dst, e->dgram + ROUTER_SVC_DST_OFF, 8); + memcpy(&group, e->dgram + ROUTER_SVC_GROUP_OFF, 8); + const uint8_t* w = e->dgram + ROUTER_SVC_HDR_SIZE; + unsigned count = mc_read(w + 22, 2); + if (dst != ctx->inst->node_id || mc_read(w + 2, 8) != source || !source || !mc_read(w + 10, 8) || + !mc_read(w + 18, 4) || w[24] > 16 || count > MC_REPORT_CHUNK || + e->len != ROUTER_SVC_HDR_SIZE + MULTICAST_REPORT_HEADER_SIZE + count * 8 || + !mc_ids_valid(w + MULTICAST_REPORT_HEADER_SIZE, count)) goto invalid; + struct TOPO_GROUP* g = topo_groups_find(ctx->inst->topo_groups, group); + if (!g || g->stopping) goto invalid; + uint64_t ids[MC_REPORT_CHUNK]; + for (unsigned i = 0; i < count; i++) ids[i] = mc_read(w + MULTICAST_REPORT_HEADER_SIZE + i * 8, 8); + struct multicast_loss loss = { .group_id = group, .source_node_id = dst, .src = source, + .dst = mc_read(w + 10, 8), .descriptor = mc_read(w, 2), .recipient_count = count, + .recipients = ids, .packets = mc_read(w + 18, 4), .reason = MULTICAST_OVERLOAD, .prefix_len = w[24] }; + memcpy(loss.prefix, w + 25, 16); ctx->stats.reports_received++; + DEBUG_WARN(MC_CAT, "multicast overload received group=%016llx descriptor=%u link=%016llx->%016llx recipients=%u packets=%u", + (unsigned long long)group, loss.descriptor, (unsigned long long)loss.src, + (unsigned long long)loss.dst, count, loss.packets); + if (ctx->loss) ctx->loss(ctx->inst, &loss, ctx->arg); + mc_free(e); return; +invalid: + if (ctx) ctx->stats.invalid_packets++; + DEBUG_WARN(MC_CAT, "multicast malformed report len=%u", e ? e->len : 0); + if (e) mc_free(e); +} + +int multicast_init(struct UTUN_INSTANCE* inst) { + if (!inst || !inst->ua || inst->multicast || inst->api_bindings.callbacks[ETCP_ID_MULTICAST] || + inst->router_bindings.callbacks[ETCP_RT_ID_MULTICAST_REPORT]) { + DEBUG_ERROR(MC_CAT, "multicast init invalid instance or occupied binding"); return -1; + } + struct multicast_ctx* ctx = u_calloc(1, sizeof(*ctx)); + if (!ctx) { DEBUG_ERROR(MC_CAT, "multicast context allocation failed"); return -1; } + ctx->inst = inst; inst->multicast = ctx; + etcp_add_conn_status_cbk(inst, mc_conn_status, ctx); + if (!inst->conn_status_cbks || inst->conn_status_cbks->fn != mc_conn_status || inst->conn_status_cbks->arg != ctx) { + DEBUG_ERROR(MC_CAT, "multicast status subscription allocation failed"); u_free(ctx); inst->multicast = NULL; return -1; + } + etcp_bind(inst, ETCP_ID_MULTICAST, mc_recv); etcp_router_bind(inst, ETCP_RT_ID_MULTICAST_REPORT, mc_report_recv); + DEBUG_INFO(MC_CAT, "multicast initialized data_limit=%u reports_limit=%u interval_ms=300", + MULTICAST_QUEUE_BYTES, MULTICAST_REPORT_BYTES); + return 0; +} + +void multicast_destroy(struct UTUN_INSTANCE* inst) { + struct multicast_ctx* ctx = inst ? inst->multicast : NULL; + if (!ctx) return; + DEBUG_INFO(MC_CAT, "multicast destroy sent=%llu delivered=%llu overload=%llu reports_evicted=%llu report_bytes=%zu", + (unsigned long long)ctx->stats.sent, (unsigned long long)ctx->stats.delivered, + (unsigned long long)ctx->stats.overload_packets, (unsigned long long)ctx->stats.reports_evicted, ctx->stats.report_bytes); + etcp_unbind(inst, ETCP_ID_MULTICAST); etcp_router_unbind(inst, ETCP_RT_ID_MULTICAST_REPORT); + etcp_remove_conn_status_cbk(inst, mc_conn_status, ctx); + if (ctx->route_wake) uasync_call_soon_cancel(inst->ua, ctx->route_wake); + if (ctx->report_timer) uasync_cancel_timeout(inst->ua, ctx->report_timer); + while (ctx->outputs) mc_out_destroy(ctx, ctx->outputs); + while (ctx->reports) mc_report_remove(ctx, ctx->reports); + while (ctx->windows) mc_window_remove(ctx, &ctx->windows); + inst->multicast = NULL; u_free(ctx); +} + +void multicast_set_callbacks(struct UTUN_INSTANCE* inst, multicast_recv_fn recv, multicast_loss_fn loss, void* arg) { + if (!inst || !inst->multicast) { DEBUG_ERROR(MC_CAT, "multicast callbacks before init"); return; } + inst->multicast->recv = recv; inst->multicast->loss = loss; inst->multicast->arg = arg; +} + +int multicast_send(struct TOPO_GROUP* group, uint16_t descriptor, const uint64_t* recipients, + uint16_t count, const void* payload, size_t len) { + if (!group || group->stopping || !group->instance || !group->instance->multicast || !recipients || + !count || count > MC_MAX_RECIPIENTS || (!payload && len) || + len > PKTNORM_MAX_DGRAM_SIZE - MULTICAST_HEADER_SIZE - MULTICAST_PATH_RESERVE - (size_t)count * 8) { + DEBUG_ERROR(MC_CAT, "multicast send invalid arguments count=%u len=%zu", count, len); return -1; + } + struct mc_target targets[MC_MAX_RECIPIENTS]; + for (unsigned i = 0; i < count; i++) { + if (!recipients[i]) { DEBUG_ERROR(MC_CAT, "multicast zero recipient index=%u", i); return -1; } + targets[i] = (struct mc_target){ NULL, recipients[i] }; + } + qsort(targets, count, sizeof(*targets), mc_target_compare); + unsigned unique = 0; + for (unsigned i = 0; i < count; i++) if (!unique || targets[i].id != targets[unique - 1].id) targets[unique++] = targets[i]; + uint8_t visited[8]; mc_write(visited, group->instance->node_id, 8); + struct mc_packet p = { .group = group->group_id, .source = group->instance->node_id, .descriptor = descriptor, + .payload = payload, .payload_len = len, .visited = visited, .hops = 1 }; + struct ll_entry* e = mc_build(&p, targets, unique); + if (!e) return -1; + mc_forward(group->instance->multicast, e, NULL); return 0; +} + +int multicast_get_stats(struct UTUN_INSTANCE* inst, struct multicast_stats* out) { + if (!inst || !inst->multicast || !out) { DEBUG_ERROR(MC_CAT, "multicast stats invalid arguments"); return -1; } + *out = inst->multicast->stats; + for (struct mc_out* o = inst->multicast->outputs; o; o = o->next) { out->queues++; out->queued_bytes += o->q->total_bytes; } + return 0; +} + +void multicast_close_group(struct UTUN_INSTANCE* inst, uint64_t group_id) { + struct multicast_ctx* ctx = inst ? inst->multicast : NULL; + if (!ctx) return; + unsigned dropped = 0; + for (struct mc_out* o = ctx->outputs; o; o = o->next) { + for (struct ll_entry* e = o->q->head; e;) { + struct ll_entry* next = e->next; + if (mc_read(e->dgram + 1, 8) == group_id) { queue_remove_data(o->q, e); mc_free(e); dropped++; } + e = next; + } + if (o->conn) queue_waiter_cancel(o->conn->send_input_q, &o->waiter); + queue_resume_callback(o->q); + } + for (struct mc_report* r = ctx->reports; r;) { + struct mc_report* next = r->next; + if (r->loss.group_id == group_id) mc_report_remove(ctx, r); + r = next; + } + for (struct mc_window** pp = &ctx->windows; *pp;) { + if ((*pp)->group == group_id) mc_window_remove(ctx, pp); + else pp = &(*pp)->next; + } + DEBUG_INFO(MC_CAT, "multicast group closed group=%016llx discarded=%u reports_bytes=%zu", + (unsigned long long)group_id, dropped, ctx->stats.report_bytes); + mc_report_schedule(ctx); +} diff --git a/src/routing_layer/multicast.h b/src/routing_layer/multicast.h new file mode 100644 index 00000000..b6bf467f --- /dev/null +++ b/src/routing_layer/multicast.h @@ -0,0 +1,95 @@ +/* multicast — адресная рассылка по маршрутам topo-группы, с разделением по next hop. + * send копирует данные; descriptor задаёт приложение источника, это не номер пакета. + * Одна FIFO на ETCP-соединение обслуживает все группы/источники/дескрипторы. + * При переполнении теряется новая ветка; источник получает OVERLOAD через router. + * Все вызовы и callbacks — uasync-поток. В callbacks допустима отправка, но нельзя + * уничтожать ядро/группу/модуль. Данные callbacks заимствованы только на время вызова. + * + * Wire DATA, все числа little-endian: + * cmd:1 | group:8 | descriptor:2 | count:2 | source:8 | recipients:8*count | + * payload | visited:8*hops | hops:1. + * Получатели строго возрастают. Источник начинает visited собой, каждый транзит + * добавляет себя ровно один раз. Первый visited=source, последний=ETCP peer. + * Полный резерв visited учитывается до отправки, payload на транзите не меняется. + * + * Wire OVERLOAD — payload отдельного router-сервиса: + * descriptor:2 | src:8 | dst:8 | packets:4 | count:2 | prefix_len:1 | + * prefix:16 | recipients:8*count. + * src/dst — концы перегруженного участка; router dst — источник DATA. + * packets относится к КАЖДОМУ указанному получателю. Части длинного отчёта имеют + * непересекающиеся списки: складывать их packets как число исходных пакетов нельзя. + * Отчёт сохраняет префикс первого пакета; разные наборы получателей не смешиваются. + */ +#ifndef UTUN_MULTICAST_H +#define UTUN_MULTICAST_H + +#include +#include + +#ifdef __cplusplus +extern "C" { +#endif + +struct UTUN_INSTANCE; +struct TOPO_GROUP; +struct multicast_ctx; + +#define MULTICAST_QUEUE_BYTES (256u * 1024u) +#define MULTICAST_REPORT_BYTES (16u * 1024u) +#define MULTICAST_REPORT_INTERVAL_TB 3000 +#define MULTICAST_MAX_VISITED 64 +#define MULTICAST_HEADER_SIZE 21 +#define MULTICAST_PATH_RESERVE (8 * MULTICAST_MAX_VISITED + 1) +#define MULTICAST_REPORT_HEADER_SIZE 41 + +enum multicast_loss_reason { + MULTICAST_OVERLOAD = 1, /* очередь выбранного исходящего соединения заполнена */ + MULTICAST_NO_ROUTE, /* только локальное уведомление источнику */ + MULTICAST_PATH_LOOP, /* локальное уведомление; транзит логирует потерю */ + MULTICAST_INTERNAL_ERROR /* выделение памяти / отказ транспорта */ +}; + +struct multicast_loss { + uint64_t group_id; /* группа исходной рассылки */ + uint64_t source_node_id; /* источник исходного DATA */ + uint64_t src, dst; /* узел потери и выбранный next hop; dst=0 без маршрута */ + uint16_t descriptor; /* идентификатор приложения источника */ + uint16_t recipient_count; /* число ID в recipients */ + const uint64_t* recipients; /* заимствованный массив в host byte order */ + uint32_t packets; /* потери для каждого получателя этого списка */ + enum multicast_loss_reason reason; + uint8_t prefix_len; /* 0..16 значимых байт */ + uint8_t prefix[16]; /* начало payload первого потерянного пакета */ +}; + +typedef void (*multicast_recv_fn)(struct UTUN_INSTANCE* inst, uint64_t group_id, uint64_t source, + uint16_t descriptor, const uint8_t* payload, size_t len, void* arg); +typedef void (*multicast_loss_fn)(struct UTUN_INSTANCE* inst, const struct multicast_loss* loss, void* arg); + +struct multicast_stats { + uint64_t sent, delivered, overload_packets, no_route_packets, loop_packets, invalid_packets, internal_errors; + uint64_t reports_sent, reports_received, reports_evicted; /* отправленные части / принятые / вытесненные записи */ + size_t queued_bytes, report_bytes; /* данные в FIFO / память отчётов и окон, без уже переданного транспорту */ + unsigned queues; /* лениво созданные FIFO, в том числе временно опустевшие */ +}; + +/* Инициализация после etcp_router_init; destroy до router/ETCP. 0/-1; повторный init — ошибка. */ +int multicast_init(struct UTUN_INSTANCE* inst); +void multicast_destroy(struct UTUN_INSTANCE* inst); +/* Один приёмник и обработчик потерь на ядро; NULL отключает соответствующий callback. */ +void multicast_set_callbacks(struct UTUN_INSTANCE* inst, multicast_recv_fn recv, multicast_loss_fn loss, void* arg); +/* count>0, ID ненулевые; повторы удаляются. len может быть 0. 0=принято к обработке, + * -1=неверный аргумент/размер или отказ начального выделения. Частичные потери — через loss. + * Доставка best effort: повторов/сквозных ACK нет; порядок при смене маршрутов не гарантируется. */ +int multicast_send(struct TOPO_GROUP* group, uint16_t descriptor, const uint64_t* recipients, + uint16_t count, const void* payload, size_t len); +/* Снимок диагностики; 0/-1, не создаёт состояния. */ +int multicast_get_stats(struct UTUN_INSTANCE* inst, struct multicast_stats* out); +/* Внутренние hooks топологии: отложенный пересмотр маршрутов / отмена данных удаляемой группы. */ +void multicast_routes_changed(struct UTUN_INSTANCE* inst); +void multicast_close_group(struct UTUN_INSTANCE* inst, uint64_t group_id); + +#ifdef __cplusplus +} +#endif +#endif