jeka/utun2
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Implement multicast descriptors and receiver subscriptions

Browse Source
This commit is contained in:
evgeny committed 2026-10-20 14:53:23 +02:00
1 parent e416d69212
commit 4a1a65c3ba
2 files changed
+254 -32
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/routing_layer/multicast.c
+251 -30
View File
@@ -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;) {
src/routing_layer/multicast.h
+3 -2
View File
@@ -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; /* активные входящие подписки */