From 4a1a65c3ba0e4dac0641bbf0a2dd13c3ce318dbd Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 20 Oct 2026 14:53:23 +0200 Subject: [PATCH] Implement multicast descriptors and receiver subscriptions --- src/routing_layer/multicast.c | 281 ++++++++++++++++++++++++++++++---- src/routing_layer/multicast.h | 5 +- 2 files changed, 254 insertions(+), 32 deletions(-) diff --git a/src/routing_layer/multicast.c b/src/routing_layer/multicast.c index e1abbe0c..1318b541 100644 --- a/src/routing_layer/multicast.c +++ b/src/routing_layer/multicast.c @@ -45,10 +45,36 @@ struct mc_report { struct multicast_loss loss; uint64_t due, retry; /* неизменный конец окна / следующая попытка передачи */ uint16_t cursor; /* уже переданные router получатели длинного отчёта */ + uint8_t received; /* входящий отчёт: только отложенный callback, без повторной отправки */ struct queue_waiter_handle waiter; uint64_t ids[]; }; +struct mc_descriptor { + struct ll_entry ll; + uint16_t id; /* хеш-ключ, первые два байта data[] */ + uint64_t group; + multicast_loss_fn on_loss; + void* arg; +}; + +struct multicast_subscription { + struct ll_entry ll; + uint8_t key[18]; /* group:8, source:8, descriptor:2; little-endian */ + struct multicast_ctx* ctx; /* NULL после unsubscribe или удаления группы/ядра */ + multicast_recv_fn on_data; + void* arg; + unsigned running; /* удерживает handle, если он закрыт из своего callback */ + int released; /* вызывающий уже вызвал unsubscribe */ +}; + +struct mc_delivery { + struct ll_entry ll; + struct multicast_subscription* subscription; + uint64_t source; + uint16_t descriptor; /* нужен для отмены loopback при close исходящего дескриптора */ +}; + struct multicast_ctx { struct UTUN_INSTANCE* inst; struct mc_out* outputs; @@ -57,9 +83,8 @@ struct multicast_ctx { void *report_timer, *route_wake; uint64_t eviction_warn; struct multicast_stats stats; - multicast_recv_fn recv; - multicast_loss_fn loss; - void* arg; + struct ll_queue *descriptors, *subscriptions, *deliveries; + uint16_t last_descriptor; }; struct mc_target { struct ETCP_CONN* conn; uint64_t id; }; @@ -68,6 +93,8 @@ 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 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); static uint64_t mc_read(const uint8_t* p, unsigned bytes) { uint64_t v = 0; @@ -88,7 +115,7 @@ static int mc_parse(const struct ll_entry* e, struct mc_packet* p) { 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 || + if (w[0] != ETCP_ID_MULTICAST || !p->source || !p->descriptor || !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; @@ -117,6 +144,7 @@ static struct ll_entry* mc_build(const struct mc_packet* p, const struct mc_targ 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; } + e->len = len; 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); @@ -127,6 +155,72 @@ static struct ll_entry* mc_build(const struct mc_packet* p, const struct mc_targ return e; } +/* Заимствованный дескриптор; callback может закрыть его, после callback не разыменовывать. */ +static struct mc_descriptor* mc_descriptor_find(struct multicast_ctx* ctx, uint16_t id) { + return (struct mc_descriptor*)queue_find_data_by_index(ctx->descriptors, &id); +} + +/* Доставка извлечена до callback: unsubscribe может отменить остальные и освободить arg. */ +static void mc_deliver(struct ll_queue* q, void* arg) { + struct multicast_ctx* ctx = arg; + struct mc_delivery* d = (struct mc_delivery*)queue_data_get(q); + if (!d) { queue_resume_callback(q); return; } + struct multicast_subscription* sub = d->subscription; + ctx->stats.delivery_bytes -= sizeof(*d) + d->ll.len; + sub->running++; + ctx->stats.delivered++; + sub->on_data(sub, d->ll.dgram, d->ll.len, sub->arg); + sub->running--; + if (sub->released) queue_entry_free(&sub->ll); + mc_free(&d->ll); queue_resume_callback(q); +} + +/* Подписка фиксируется в момент приёма: новая подписка не получит старые ожидающие DATA. */ +static void mc_deliver_later(struct multicast_ctx* ctx, const struct mc_packet* p) { + uint8_t key[18]; + mc_write(key, p->group, 8); mc_write(key + 8, p->source, 8); mc_write(key + 16, p->descriptor, 2); + struct multicast_subscription* sub = (struct multicast_subscription*)queue_find_data_by_index(ctx->subscriptions, key); + if (!sub) { ctx->stats.unsubscribed++; return; } + size_t bytes = sizeof(struct mc_delivery) + p->payload_len; + if (ctx->stats.delivery_bytes + bytes > MULTICAST_QUEUE_BYTES) goto failed; + struct mc_delivery* d = (struct mc_delivery*)queue_entry_new(sizeof(*d) - sizeof(struct ll_entry)); + if (!d) goto failed; + if (p->payload_len) { + d->ll.dgram = u_malloc(p->payload_len); + if (!d->ll.dgram) { queue_entry_free(&d->ll); goto failed; } + memcpy(d->ll.dgram, p->payload, p->payload_len); + } + d->ll.len = p->payload_len; d->subscription = sub; d->source = p->source; d->descriptor = p->descriptor; + ctx->stats.delivery_bytes += bytes; queue_data_put(ctx->deliveries, &d->ll); return; +failed: + DEBUG_ERROR(MC_CAT, "multicast local delivery unavailable pending=%zu need=%zu source=%016llx descriptor=%u", + ctx->stats.delivery_bytes, bytes, (unsigned long long)p->source, p->descriptor); + struct mc_target target = { NULL, ctx->inst->node_id }; + mc_loss(ctx, p, ctx->inst->node_id, MULTICAST_INTERNAL_ERROR, &target, 1); +} + +/* Удаление по handle или локальному исходящему дескриптору; активный callback уже извлечён. */ +static void mc_cancel_deliveries(struct multicast_ctx* ctx, struct multicast_subscription* sub, uint16_t descriptor) { + for (struct ll_entry* e = ctx->deliveries->head; e;) { + struct ll_entry* next = e->next; + struct mc_delivery* d = (struct mc_delivery*)e; + if ((sub && d->subscription == sub) || (descriptor && d->source == ctx->inst->node_id && d->descriptor == descriptor)) { + queue_remove_data(ctx->deliveries, e); ctx->stats.delivery_bytes -= sizeof(*d) + e->len; mc_free(e); + } + e = next; + } +} + +/* Отцепить принадлежащую приложению подписку, сохранив handle до unsubscribe. */ +static void mc_subscription_detach(struct multicast_subscription* sub) { + if (!sub->ctx) return; + struct multicast_ctx* ctx = sub->ctx; + mc_cancel_deliveries(ctx, sub, 0); + queue_remove_data(ctx->subscriptions, &sub->ll); sub->ctx = NULL; + DEBUG_DEBUG(MC_CAT, "multicast subscription detached source=%016llx descriptor=%u", + (unsigned long long)mc_read(sub->key + 8, 8), (unsigned)mc_read(sub->key + 16, 2)); +} + /* Запись всегда удаляется вместе с waiter, в том числе уже запланированным. */ static void mc_report_remove(struct multicast_ctx* ctx, struct mc_report* r) { struct mc_report** pp = &ctx->reports; @@ -189,7 +283,7 @@ static void mc_loss(struct multicast_ctx* ctx, const struct mc_packet* p, uint64 } } 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 || + if (r->received || 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; @@ -233,8 +327,13 @@ static void mc_report_ready(struct ll_queue* q, void* arg) { /* Возвращает 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; + struct mc_descriptor* d = mc_descriptor_find(ctx, r->loss.descriptor); + if (!d || d->group != r->loss.group_id) return 1; /* дескриптор уже закрыт */ + DEBUG_WARN(MC_CAT, "multicast local loss descriptor=%u dst=%016llx reason=%d recipients=%u packets=%u", + r->loss.descriptor, (unsigned long long)r->loss.dst, r->loss.reason, r->loss.recipient_count, r->loss.packets); + if (!r->received) { ctx->stats.reports_sent++; ctx->stats.reports_received++; } + d->on_loss(ctx->inst, &r->loss, d->arg); + 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)) { @@ -246,6 +345,7 @@ static int mc_report_send(struct multicast_ctx* ctx, struct mc_report* r) { 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; } + e->len = 1 + MULTICAST_REPORT_HEADER_SIZE + count * 8; 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; @@ -358,9 +458,7 @@ static void mc_forward(struct multicast_ctx* ctx, struct ll_entry* e, struct mc_ 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); + mc_deliver_later(ctx, &p); continue; } targets[count++] = (struct mc_target){ topo_group_find_conn_for_node(group, id), id }; @@ -416,6 +514,19 @@ static void mc_out_ready(struct ll_queue* q, void* arg) { queue_resume_callback(o->q); } +/* Ошибка передачи уже готовой ветки: локальный источник получает тот же loss callback. */ +static void mc_out_failed(struct mc_out* o, struct ll_entry* e) { + struct mc_packet p; + if (mc_parse(e, &p) == 0) { + struct mc_target targets[MC_MAX_RECIPIENTS]; + for (unsigned i = 0; i < p.count; i++) targets[i] = (struct mc_target){ NULL, mc_read(p.recipients + i * 8, 8) }; + mc_loss(o->ctx, &p, o->conn->peer_node_id, MULTICAST_INTERNAL_ERROR, targets, p.count); + } else { + DEBUG_ERROR(MC_CAT, "multicast invalid failed packet len=%u", e->len); o->ctx->stats.internal_errors++; + } + mc_free(e); +} + /* Автозабор FIFO: один пакет на свободную send_input_q, затем снова round-robin waiter. */ static void mc_out_drain(struct ll_queue* q, void* arg) { struct mc_out* o = arg; @@ -427,10 +538,10 @@ static void mc_out_drain(struct ll_queue* q, void* arg) { 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++; + e = queue_data_get(q); mc_out_failed(o, e); } 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++; + mc_out_failed(o, e); } else ctx->stats.sent++; queue_resume_callback(q); if (!o->conn && !q->count) mc_out_destroy(ctx, o); @@ -502,6 +613,7 @@ static void mc_recv(struct ETCP_CONN* from, struct ll_entry* e) { } 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; } + expanded->len = e->len + 8; 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; @@ -524,22 +636,28 @@ static void mc_report_recv(struct ETCP_CONN* conn, struct ll_entry* e) { 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 || + !mc_read(w, 2) || !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; + for (unsigned i = w[24]; i < 16; i++) if (w[25 + i]) 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, + struct mc_descriptor* descriptor = mc_descriptor_find(ctx, mc_read(w, 2)); + if (!descriptor || descriptor->group != group) { mc_free(e); return; } + size_t bytes = sizeof(struct mc_report) + count * 8; + if (mc_report_room(ctx, bytes) < 0) { mc_free(e); return; } + struct mc_report* r = u_calloc(1, bytes); + if (!r) { DEBUG_ERROR(MC_CAT, "multicast incoming report allocation failed bytes=%zu", bytes); mc_free(e); return; } + r->ctx = ctx; r->received = 1; r->due = r->retry = get_time_tb(); + for (unsigned i = 0; i < count; i++) r->ids[i] = mc_read(w + MULTICAST_REPORT_HEADER_SIZE + i * 8, 8); + r->loss = (struct multicast_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; + .recipients = r->ids, .packets = mc_read(w + 18, 4), .reason = MULTICAST_OVERLOAD, .prefix_len = w[24] }; + memcpy(r->loss.prefix, w + 25, 16); ctx->stats.reports_received++; + struct mc_report** tail = &ctx->reports; + while (*tail) tail = &(*tail)->next; + *tail = r; ctx->stats.report_bytes += bytes; + mc_free(e); mc_report_schedule(ctx); return; invalid: if (ctx) ctx->stats.invalid_packets++; DEBUG_WARN(MC_CAT, "multicast malformed report len=%u", e ? e->len : 0); @@ -554,9 +672,16 @@ int multicast_init(struct UTUN_INSTANCE* inst) { 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; + ctx->descriptors = queue_new(inst->ua, 256, 0, 2, "multicast_descriptors"); + ctx->subscriptions = queue_new(inst->ua, 256, 0, 18, "multicast_subscriptions"); + ctx->deliveries = queue_new(inst->ua, 0, 0, 0, "multicast_deliveries"); + if (!ctx->descriptors || !ctx->subscriptions || !ctx->deliveries) { + DEBUG_ERROR(MC_CAT, "multicast registry allocation failed"); multicast_destroy(inst); return -1; + } + queue_set_callback_defer(ctx->deliveries, 1); queue_set_callback(ctx->deliveries, mc_deliver, 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; + DEBUG_ERROR(MC_CAT, "multicast status subscription allocation failed"); multicast_destroy(inst); 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", @@ -577,17 +702,39 @@ void multicast_destroy(struct UTUN_INSTANCE* inst) { 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); + while (ctx->subscriptions && ctx->subscriptions->head) + mc_subscription_detach((struct multicast_subscription*)ctx->subscriptions->head); + struct ll_entry* e; + if (ctx->descriptors) while ((e = queue_data_get(ctx->descriptors))) queue_entry_free(e); + queue_free(ctx->deliveries); queue_free(ctx->subscriptions); queue_free(ctx->descriptors); 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; +/* Номера выдаются циклически после последнего успешного open; закрытие не меняет порядок. */ +int multicast_open(struct TOPO_GROUP* group, multicast_loss_fn on_loss, void* arg) { + if (!group || group->stopping || !group->instance || !group->instance->multicast || !on_loss) { + DEBUG_ERROR(MC_CAT, "multicast open invalid group or callback"); return -1; + } + struct multicast_ctx* ctx = group->instance->multicast; + if (ctx->descriptors->count == UINT16_MAX) { + DEBUG_ERROR(MC_CAT, "multicast descriptors exhausted count=%d", ctx->descriptors->count); return -1; + } + uint16_t id = ctx->last_descriptor; + do { id = id == UINT16_MAX ? 1 : id + 1; } while (mc_descriptor_find(ctx, id)); + struct mc_descriptor* d = (struct mc_descriptor*)queue_entry_new(sizeof(*d) - sizeof(struct ll_entry)); + if (!d) { DEBUG_ERROR(MC_CAT, "multicast descriptor allocation failed"); return -1; } + d->id = id; d->group = group->group_id; d->on_loss = on_loss; d->arg = arg; + queue_data_put_with_index(ctx->descriptors, &d->ll); ctx->last_descriptor = id; + DEBUG_DEBUG(MC_CAT, "multicast open group=%016llx descriptor=%u", (unsigned long long)d->group, id); + return id; } -int multicast_send(struct TOPO_GROUP* group, uint16_t descriptor, const uint64_t* recipients, +int multicast_send(struct UTUN_INSTANCE* inst, 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 || + struct multicast_ctx* ctx = inst ? inst->multicast : NULL; + struct mc_descriptor* d = ctx ? mc_descriptor_find(ctx, descriptor) : NULL; + struct TOPO_GROUP* group = d ? topo_groups_find(inst->topo_groups, d->group) : NULL; + if (!group || group->stopping || !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; @@ -605,12 +752,74 @@ int multicast_send(struct TOPO_GROUP* group, uint16_t descriptor, const uint64_t .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; + mc_forward(ctx, e, NULL); return 0; +} + +/* Удаляет только пакеты локального источника этого дескриптора; транзит других источников сохраняется. */ +void multicast_close(struct UTUN_INSTANCE* inst, uint16_t descriptor) { + struct multicast_ctx* ctx = inst ? inst->multicast : NULL; + struct mc_descriptor* d = ctx ? mc_descriptor_find(ctx, descriptor) : NULL; + if (!d) { DEBUG_ERROR(MC_CAT, "multicast close unknown descriptor=%u", descriptor); return; } + uint64_t group = d->group; + queue_remove_data(ctx->descriptors, &d->ll); queue_entry_free(&d->ll); + unsigned discarded = 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 + 13, 8) == inst->node_id && mc_read(e->dgram + 9, 2) == descriptor) { + queue_remove_data(o->q, e); mc_free(e); discarded++; + } + e = next; + } + if (o->conn) queue_waiter_cancel(o->conn->send_input_q, &o->waiter); + queue_resume_callback(o->q); + } + mc_cancel_deliveries(ctx, NULL, descriptor); + for (struct mc_report* r = ctx->reports; r;) { + struct mc_report* next = r->next; + if (r->loss.source_node_id == inst->node_id && r->loss.descriptor == descriptor) mc_report_remove(ctx, r); + r = next; + } + for (struct mc_window** pp = &ctx->windows; *pp;) { + if ((*pp)->source == inst->node_id && (*pp)->descriptor == descriptor) mc_window_remove(ctx, pp); + else pp = &(*pp)->next; + } + DEBUG_DEBUG(MC_CAT, "multicast close group=%016llx descriptor=%u discarded=%u", (unsigned long long)group, descriptor, discarded); + mc_report_schedule(ctx); +} + +/* Ключ принадлежит группе приёма, номер — источнику; исходящий дескриптор здесь не создаётся. */ +struct multicast_subscription* multicast_subscribe(struct TOPO_GROUP* group, uint64_t src_node, uint16_t src_descriptor, + multicast_recv_fn on_data, void* arg) { + if (!group || group->stopping || !group->instance || !group->instance->multicast || !src_node || !src_descriptor || !on_data) { + DEBUG_ERROR(MC_CAT, "multicast subscribe invalid arguments"); return NULL; + } + struct multicast_ctx* ctx = group->instance->multicast; + uint8_t key[18]; + mc_write(key, group->group_id, 8); mc_write(key + 8, src_node, 8); mc_write(key + 16, src_descriptor, 2); + if (queue_find_data_by_index(ctx->subscriptions, key)) { + DEBUG_ERROR(MC_CAT, "multicast duplicate subscription group=%016llx source=%016llx descriptor=%u", + (unsigned long long)group->group_id, (unsigned long long)src_node, src_descriptor); return NULL; + } + struct multicast_subscription* sub = (struct multicast_subscription*)queue_entry_new(sizeof(*sub) - sizeof(struct ll_entry)); + if (!sub) { DEBUG_ERROR(MC_CAT, "multicast subscription allocation failed"); return NULL; } + memcpy(sub->key, key, sizeof(key)); sub->ctx = ctx; sub->on_data = on_data; sub->arg = arg; + queue_data_put_with_index(ctx->subscriptions, &sub->ll); + DEBUG_DEBUG(MC_CAT, "multicast subscribe group=%016llx source=%016llx descriptor=%u", + (unsigned long long)group->group_id, (unsigned long long)src_node, src_descriptor); + return sub; +} + +void multicast_unsubscribe(struct multicast_subscription* sub) { + if (!sub) return; + mc_subscription_detach(sub); sub->released = 1; + if (!sub->running) queue_entry_free(&sub->ll); } 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; + out->descriptors = inst->multicast->descriptors->count; out->subscriptions = inst->multicast->subscriptions->count; for (struct mc_out* o = inst->multicast->outputs; o; o = o->next) { out->queues++; out->queued_bytes += o->q->total_bytes; } return 0; } @@ -618,6 +827,18 @@ int multicast_get_stats(struct UTUN_INSTANCE* inst, struct multicast_stats* out) void multicast_close_group(struct UTUN_INSTANCE* inst, uint64_t group_id) { struct multicast_ctx* ctx = inst ? inst->multicast : NULL; if (!ctx) return; + for (struct ll_entry* e = ctx->descriptors->head; e;) { + struct ll_entry* next = e->next; + struct mc_descriptor* d = (struct mc_descriptor*)e; + if (d->group == group_id) multicast_close(inst, d->id); + e = next; + } + for (struct ll_entry* e = ctx->subscriptions->head; e;) { + struct ll_entry* next = e->next; + struct multicast_subscription* sub = (struct multicast_subscription*)e; + if (mc_read(sub->key, 8) == group_id) mc_subscription_detach(sub); + e = next; + } unsigned dropped = 0; for (struct mc_out* o = ctx->outputs; o; o = o->next) { for (struct ll_entry* e = o->q->head; e;) { diff --git a/src/routing_layer/multicast.h b/src/routing_layer/multicast.h index f44ec18f..a0900568 100644 --- a/src/routing_layer/multicast.h +++ b/src/routing_layer/multicast.h @@ -59,8 +59,8 @@ * потери; сквозных ACK/повторов нет, порядок при смене маршрута не гарантируется. * Сам отчёт тоже может потеряться: отсутствие on_loss не подтверждает доставку. * - * Состояние API: этот заголовок описывает новый контракт; multicast.c пока - * использует прежнюю модель с общими callbacks и требует адаптации. + * Локальные callbacks DATA тоже отложены; их общая очередь ограничена 256 КиБ + * памяти. Отказ её выделения/переполнение считается MULTICAST_INTERNAL_ERROR. * Поколений нет: DATA, отчёты и подписки используют 16-битный номер дескриптора. * Новое ядро начинает выдачу с 1. После повторной выдачи номера запоздалый * сетевой пакет с тем же источником и номером неотличим от пакета нового владельца. @@ -137,6 +137,7 @@ struct multicast_stats { uint64_t reports_evicted; /* записи отчётов, вытесненные из бюджета 16 КиБ */ size_t queued_bytes; /* DATA в FIFO модуля, без уже переданного ETCP */ size_t report_bytes; /* память накопленных отчётов и окон, без переданного router */ + size_t delivery_bytes; /* память DATA, ожидающих локальные callbacks; максимум 256 КиБ */ unsigned queues; /* лениво созданные FIFO, в том числе временно опустевшие */ unsigned descriptors; /* открытые исходящие дескрипторы */ unsigned subscriptions; /* активные входящие подписки */