From e59bc9857bfecce07cb08217d56682864c4d347a Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 20 Oct 2026 15:01:27 +0200 Subject: [PATCH] Test multicast routing, overload and descriptor lifetime --- tests/Makefile.am | 5 + tests/test_multicast.c | 433 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 438 insertions(+) create mode 100644 tests/test_multicast.c diff --git a/tests/Makefile.am b/tests/Makefile.am index a9e0b6dc..0953f4c1 100644 --- a/tests/Makefile.am +++ b/tests/Makefile.am @@ -54,6 +54,7 @@ check_PROGRAMS = \ test_media_delivery_download \ test_transport \ test_etcp_router_unit \ + test_multicast \ test_group_ownership \ test_merkle_protocol \ test_ncd_config \ @@ -402,6 +403,10 @@ test_etcp_router_faults_LDFLAGS = $(test_clock_ldflags) test_etcp_router_unit_SOURCES = test_etcp_router_unit.c test_etcp_router_unit_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS) +test_multicast_SOURCES = test_multicast.c $(test_clock_sources) +test_multicast_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS) +test_multicast_LDFLAGS = $(test_clock_ldflags) + test_etcp_router_reconnect_SOURCES = test_etcp_router_reconnect.c test_etcp_router_reconnect_LDADD = $(LIBUTUN) $(CRYPTO_LIBS) $(COMMON_LIBS) diff --git a/tests/test_multicast.c b/tests/test_multicast.c new file mode 100644 index 00000000..718b1628 --- /dev/null +++ b/tests/test_multicast.c @@ -0,0 +1,433 @@ +/* Настоящие multicast/router, BGP-маршруты и uasync; ETCP-пакеты передаются в памяти. + * Управляемые очереди позволяют проверять точную границу, независимые ветки и смену пути. */ +#include +#include +#include +#include "../lib/debug_config.h" +#include "../lib/mem.h" +#include "etcp.h" +#include "etcp_router.h" +#include "multicast.h" +#include "pkt_normalizer.h" +#include "topo_group.h" +#include "utun_instance.h" +#include "test_clock.h" + +#define CHECK(x) do { if (!(x)) { \ + DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "FAIL %s:%d: %s", __func__, __LINE__, #x); abort(); \ +} } while (0) +#define A 0 +#define B 1 +#define C 2 +#define D 3 +#define GROUPS 2 + +struct wire { + struct ETCP_CONN conn; + struct wire* reverse; + struct ll_entry* blocker; + unsigned packets, max_recipients; +}; + +struct fixture { + struct UASYNC* ua; + struct UTUN_INSTANCE* inst[4]; + struct TOPO_GROUP* group[4][GROUPS]; + struct wire wire[4][4]; +}; + +struct received { + unsigned count; + size_t len; + uint8_t prefix[16]; + int unsubscribe; +}; + +struct losses { + unsigned count; + struct multicast_loss item[256]; + uint64_t ids[256][8]; + int close; +}; + +static void release(struct ll_entry* e) { + if (e) { queue_dgram_free(e); queue_entry_free(e); } +} + +static void clear_queue(struct ll_queue* q) { + if (!q) return; + struct ll_entry* e; + while ((e = queue_data_get(q))) release(e); + queue_free(q); +} + +static void receive(struct multicast_subscription* sub, const uint8_t* payload, size_t len, void* arg) { + struct received* r = arg; + r->count++; r->len = len; + memset(r->prefix, 0, sizeof(r->prefix)); + if (len) memcpy(r->prefix, payload, len < 16 ? len : 16); + if (r->unsubscribe) multicast_unsubscribe(sub); +} + +static void loss(struct UTUN_INSTANCE* inst, const struct multicast_loss* event, void* arg) { + struct losses* l = arg; + CHECK(l->count < 256); + unsigned i = l->count++; + l->item[i] = *event; + unsigned n = event->recipient_count < 8 ? event->recipient_count : 8; + memcpy(l->ids[i], event->recipients, n * sizeof(uint64_t)); + l->item[i].recipients = l->ids[i]; + if (l->close) multicast_close(inst, event->descriptor); +} + +static void wire_send(struct ll_queue* q, void* arg) { + struct wire* w = arg; + struct ll_entry* e = queue_data_get(q); + if (e) { + CHECK(e != w->blocker && e->len > 0); + if (e->dgram[0] == ETCP_ID_MULTICAST) { + w->packets++; + unsigned n = e->dgram[11] | (unsigned)e->dgram[12] << 8; + if (n > w->max_recipients) w->max_recipients = n; + } + struct ETCP_CONN* peer = &w->reverse->conn; + etcp_recv_fn cb = peer->instance->api_bindings.callbacks[e->dgram[0]]; + CHECK(cb); + cb(peer, e); + } + queue_resume_callback(q); +} + +static void block(struct wire* w) { + CHECK(!w->blocker && w->conn.send_input_q->count == 0); + queue_set_callback(w->conn.send_input_q, NULL, NULL); + w->blocker = ll_alloc_lldgram(1); CHECK(w->blocker); + w->blocker->len = 1; w->blocker->dgram[0] = 0xff; + CHECK(queue_data_put(w->conn.send_input_q, w->blocker) == 0); +} + +static void unblock(struct wire* w) { + CHECK(w->blocker); + CHECK(queue_data_get(w->conn.send_input_q) == w->blocker); + release(w->blocker); w->blocker = NULL; + queue_set_callback(w->conn.send_input_q, wire_send, w); + queue_resume_callback(w->conn.send_input_q); +} + +static void route(struct fixture* f, int from, int group, uint64_t dst, int via) { + struct TOPO_GROUP* g = f->group[from][group]; + struct TOPO_GROUP_NODE* node = topo_node_find_by_id(g, dst); + if (!node) { + node = (struct TOPO_GROUP_NODE*)queue_entry_new(sizeof(*node) - sizeof(struct ll_entry)); CHECK(node); + node->node_id = dst; + CHECK(queue_data_put_with_index(g->nodes, &node->ll) == 0); + } + topo_nodeq_clear_paths(node); + if (via >= 0) { + uint64_t hop = via + 1; + CHECK(f->wire[from][via].conn.send_input_q); + CHECK(topo_group_add_path(node, &f->wire[from][via].conn, &hop, 1, 0) == 0); + } +} + +static struct fixture* setup(void) { + struct fixture* f = u_calloc(1, sizeof(*f)); CHECK(f); + f->ua = uasync_create(); CHECK(f->ua); + for (int i = 0; i < 4; i++) { + struct UTUN_INSTANCE* inst = f->inst[i] = u_calloc(1, sizeof(*inst)); CHECK(inst); + inst->node_id = i + 1; inst->ua = f->ua; + inst->connections = queue_new(f->ua, 16, 0, 8, "mc_test_connections"); CHECK(inst->connections); + inst->topo_groups = u_calloc(1, sizeof(*inst->topo_groups)); CHECK(inst->topo_groups); + inst->topo_groups->instance = inst; + inst->topo_groups->group_list = queue_new(f->ua, 16, 0, 8, "mc_test_groups"); CHECK(inst->topo_groups->group_list); + for (int j = 0; j < GROUPS; j++) { + struct TOPO_GROUP* g = (struct TOPO_GROUP*)queue_entry_new(sizeof(*g) - sizeof(struct ll_entry)); CHECK(g); + g->instance = inst; g->group_id = j ? 0x55 : TOPO_GROUP_UTUN; + g->nodes = queue_new(f->ua, 256, 0, 8, "mc_test_nodes"); CHECK(g->nodes); + g->senders_list = queue_new(f->ua, 0, 0, 0, "mc_test_peers"); CHECK(g->senders_list); + CHECK(queue_data_put_with_index(inst->topo_groups->group_list, &g->ll) == 0); + f->group[i][j] = g; + } + CHECK(etcp_router_init(inst) == 0); + CHECK(multicast_init(inst) == 0); + } + const int edges[][2] = {{A, B}, {B, C}, {B, D}, {A, D}, {D, C}}; + for (unsigned i = 0; i < sizeof(edges) / sizeof(edges[0]); i++) { + for (int dir = 0; dir < 2; dir++) { + int a = edges[i][dir], b = edges[i][1 - dir]; + struct wire* w = &f->wire[a][b]; + w->reverse = &f->wire[b][a]; w->conn.instance = f->inst[a]; + w->conn.peer_node_id = b + 1; w->conn.state = 1; w->conn.links_up = 1; + w->conn.send_input_q = queue_new(f->ua, 0, 0, 0, "mc_test_wire"); CHECK(w->conn.send_input_q); + queue_set_callback_defer(w->conn.send_input_q, 1); + queue_set_callback(w->conn.send_input_q, wire_send, w); + struct ll_entry* e = queue_entry_new(sizeof(struct conn_queue_entry)); CHECK(e); + struct conn_queue_entry* ce = (struct conn_queue_entry*)e->data; + ce->peer_node_id = b + 1; ce->conn = &w->conn; + CHECK(queue_data_put_with_index(f->inst[a]->connections, e) == 0); + for (int g = 0; g < GROUPS; g++) { + e = queue_entry_new(sizeof(struct TOPO_GROUP_CONN_ITEM)); CHECK(e); + struct TOPO_GROUP_CONN_ITEM* peer = (struct TOPO_GROUP_CONN_ITEM*)e->data; + peer->group = f->group[a][g]; peer->node_id = b + 1; peer->conn = &w->conn; peer->accepted = 1; + CHECK(queue_data_put(f->group[a][g]->senders_list, e) == 0); + } + } + } + const int next[4][4] = {{-1, B, B, B}, {A, -1, C, D}, {B, B, -1, D}, {B, B, C, -1}}; + for (int a = 0; a < 4; a++) for (int b = 0; b < 4; b++) if (a != b) + for (int g = 0; g < GROUPS; g++) route(f, a, g, b + 1, next[a][b]); + return f; +} + +static void pump(struct fixture* f) { test_clock_run_for(f->ua, 10); } + +static struct multicast_stats stats(struct fixture* f, int node) { + struct multicast_stats s; CHECK(multicast_get_stats(f->inst[node], &s) == 0); return s; +} + +static void teardown(struct fixture* f) { + for (int i = 0; i < 4; i++) for (int j = 0; j < 4; j++) + if (f->wire[i][j].conn.send_input_q) queue_set_callback(f->wire[i][j].conn.send_input_q, NULL, NULL); + for (int i = 0; i < 4; i++) multicast_destroy(f->inst[i]); + for (int i = 0; i < 4; i++) etcp_router_destroy(f->inst[i]); + for (int i = 0; i < 4; i++) for (int j = 0; j < 4; j++) etcp_router_transit_queues_destroy(&f->wire[i][j].conn); + uasync_drain_immediate(f->ua); + for (int i = 0; i < 4; i++) { + for (int j = 0; j < 4; j++) clear_queue(f->wire[i][j].conn.send_input_q); + for (int g = 0; g < GROUPS; g++) { + struct TOPO_GROUP* group = f->group[i][g]; + for (struct ll_entry* e = group->nodes->head; e; e = e->next) topo_nodeq_clear_paths((struct TOPO_GROUP_NODE*)e); + clear_queue(group->nodes); clear_queue(group->senders_list); + } + clear_queue(f->inst[i]->topo_groups->group_list); + u_free(f->inst[i]->topo_groups); + clear_queue(f->inst[i]->connections); + CHECK(!f->inst[i]->conn_status_cbks); + u_free(f->inst[i]); + } + CHECK(f->ua->timer_alloc_count == f->ua->timer_free_count); + uasync_destroy(f->ua); u_free(f); +} + +static void test_delivery(void) { + DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "multicast: tree splitting, copying, subscriptions and source isolation"); + struct fixture* f = setup(); + struct losses l = {0}; + int fd = multicast_open(f->group[A][0], loss, &l); CHECK(fd == 1); + struct received rx[4] = {{0}}; + struct multicast_subscription* sub[4]; + for (int i = 0; i < 4; i++) { sub[i] = multicast_subscribe(f->group[i][0], 1, fd, receive, &rx[i]); CHECK(sub[i]); } + CHECK(!multicast_subscribe(f->group[C][0], 1, fd, receive, &rx[C])); + uint64_t ids[] = {4, 3, 2, 1, 3}; + char data[] = "copied"; + CHECK(multicast_send(f->inst[A], fd, ids, 5, data, sizeof(data)) == 0); + CHECK(rx[A].count == 0); memset(data, 'x', sizeof(data)); memset(ids, 0, sizeof(ids)); + pump(f); + for (int i = 0; i < 4; i++) CHECK(rx[i].count == 1 && memcmp(rx[i].prefix, "copied", 7) == 0); + CHECK(f->wire[A][B].packets == 1 && f->wire[A][B].max_recipients == 3); + CHECK(f->wire[B][C].packets == 1 && f->wire[B][D].packets == 1); + multicast_unsubscribe(sub[B]); sub[B] = NULL; + uint64_t bcd[] = {2, 3, 4}; + CHECK(multicast_send(f->inst[A], fd, bcd, 3, NULL, 0) == 0); + pump(f); CHECK(rx[B].count == 1 && rx[C].count == 2 && rx[C].len == 0 && rx[D].count == 2); + CHECK(stats(f, B).unsubscribed == 1 && l.count == 0); + int other = multicast_open(f->group[D][0], loss, &l); CHECK(other == fd); + CHECK(multicast_send(f->inst[D], other, bcd + 1, 1, "other source", 12) == 0); + pump(f); CHECK(rx[C].count == 2 && stats(f, C).unsubscribed == 1); + for (int i = 0; i < 4; i++) multicast_unsubscribe(sub[i]); + teardown(f); +} + +static void test_lifetime(void) { + DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "multicast: deferred callbacks, close/unsubscribe, group isolation"); + struct fixture* f = setup(); + struct losses l = {0}; + int fd = multicast_open(f->group[A][0], loss, &l); + int other = multicast_open(f->group[A][1], loss, &l); CHECK(fd == 1 && other == 2); + uint64_t self = 1, remote = 2; + struct received rx = {0}, rx_other = {0}; + struct multicast_subscription* sub = multicast_subscribe(f->group[A][0], 1, fd, receive, &rx); CHECK(sub); + CHECK(multicast_send(f->inst[A], fd, &self, 1, "pending", 7) == 0); + multicast_unsubscribe(sub); + sub = multicast_subscribe(f->group[A][0], 1, fd, receive, &rx); CHECK(sub); + pump(f); CHECK(rx.count == 0); + rx.unsubscribe = 1; + CHECK(multicast_send(f->inst[A], fd, &self, 1, "first", 5) == 0); + CHECK(multicast_send(f->inst[A], fd, &self, 1, "second", 6) == 0); + pump(f); CHECK(rx.count == 1 && stats(f, A).delivery_bytes == 0); + sub = multicast_subscribe(f->group[A][0], 1, fd, receive, &rx); CHECK(sub); + CHECK(multicast_send(f->inst[A], fd, &self, 1, "close", 5) == 0); + multicast_close(f->inst[A], fd); + pump(f); CHECK(rx.count == 1 && stats(f, A).delivery_bytes == 0); + multicast_unsubscribe(sub); + fd = multicast_open(f->group[A][0], loss, &l); CHECK(fd == 3); + sub = multicast_subscribe(f->group[B][0], 1, fd, receive, &rx); CHECK(sub); + struct multicast_subscription* sub_other = multicast_subscribe(f->group[B][1], 1, other, receive, &rx_other); CHECK(sub_other); + block(&f->wire[A][B]); + CHECK(multicast_send(f->inst[A], fd, &remote, 1, "cancel", 6) == 0); + CHECK(multicast_send(f->inst[A], other, &remote, 1, "keep", 4) == 0); + pump(f); CHECK(stats(f, A).queues == 1); + multicast_close_group(f->inst[A], f->group[A][0]->group_id); + unblock(&f->wire[A][B]); pump(f); + CHECK(rx.count == 1 && rx_other.count == 1 && memcmp(rx_other.prefix, "keep", 4) == 0); + CHECK(stats(f, A).descriptors == 1); + multicast_close_group(f->inst[B], f->group[B][0]->group_id); + multicast_unsubscribe(sub); + /* Handle остаётся пригодным для unsubscribe после уничтожения ядра. */ + teardown(f); multicast_unsubscribe(sub_other); +} + +static void test_overload(void) { + DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "multicast: exact 256 KiB boundary, healthy branch and timer-only tail report"); + struct fixture* f = setup(); + struct losses l = {0}; + int fd = multicast_open(f->group[A][0], loss, &l); CHECK(fd > 0); + struct received c = {0}, d = {0}; + struct multicast_subscription* sc = multicast_subscribe(f->group[C][0], 1, fd, receive, &c); CHECK(sc); + struct multicast_subscription* sd = multicast_subscribe(f->group[D][0], 1, fd, receive, &d); CHECK(sd); + block(&f->wire[B][C]); + uint64_t both[] = {3, 4}, one = 3; + uint8_t payload[8146]; memset(payload, 0xab, sizeof(payload)); + for (int i = 0; i < 32; i++) { + CHECK(multicast_send(f->inst[A], fd, both, 2, payload, sizeof(payload)) == 0); + pump(f); + } + CHECK(stats(f, B).queued_bytes == MULTICAST_QUEUE_BYTES && stats(f, B).overload_packets == 0); + CHECK(c.count == 0 && d.count == 32); + CHECK(multicast_send(f->inst[A], fd, &one, 1, "one", 3) == 0); + CHECK(l.count == 0); pump(f); + CHECK(l.count == 1); + CHECK(l.item[0].src == 2 && l.item[0].dst == 3 && l.item[0].source_node_id == 1 && l.item[0].descriptor == fd); + CHECK(l.item[0].packets == 1 && l.item[0].recipient_count == 1 && l.ids[0][0] == 3); + CHECK(l.item[0].prefix_len == 3 && memcmp(l.item[0].prefix, "one", 3) == 0); + for (int i = 3; i < 16; i++) CHECK(l.item[0].prefix[i] == 0); + CHECK(multicast_send(f->inst[A], fd, &one, 1, "two", 3) == 0); + CHECK(multicast_send(f->inst[A], fd, &one, 1, "three", 5) == 0); + route(f, A, 0, 5, B); route(f, B, 0, 5, C); + uint64_t different[] = {3, 5}; + CHECK(multicast_send(f->inst[A], fd, different, 2, "set", 3) == 0); + pump(f); CHECK(l.count == 1); + test_clock_run_for(f->ua, 2900); CHECK(l.count == 1); + test_clock_run_for(f->ua, 150); CHECK(l.count == 3); + CHECK(l.item[1].packets == 2 && l.item[1].recipient_count == 1 && memcmp(l.item[1].prefix, "two", 3) == 0); + CHECK(l.item[2].packets == 1 && l.item[2].recipient_count == 2 && l.ids[2][0] == 3 && l.ids[2][1] == 5); + CHECK(stats(f, B).overload_packets == 4 && stats(f, B).queued_bytes == MULTICAST_QUEUE_BYTES); + unblock(&f->wire[B][C]); pump(f); + CHECK(c.count == 32 && stats(f, B).queued_bytes == 0); + test_clock_run_for(f->ua, 3100); CHECK(stats(f, B).report_bytes == 0); + multicast_unsubscribe(sc); multicast_unsubscribe(sd); teardown(f); +} + +static void test_report_budget(void) { + DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "multicast: bounded 16 KiB reports and closing a descriptor inside its loss callback"); + struct fixture* f = setup(); + struct losses l = {0}; + int fd = multicast_open(f->group[A][0], loss, &l); + block(&f->wire[A][B]); + uint64_t dst = 2; + uint8_t payload[8154] = {0}; + for (int i = 0; i < 32; i++) CHECK(multicast_send(f->inst[A], fd, &dst, 1, payload, sizeof(payload)) == 0); + CHECK(stats(f, A).queued_bytes == MULTICAST_QUEUE_BYTES); + for (int i = 0; i < 200; i++) { + int descriptor = multicast_open(f->group[A][0], loss, &l); CHECK(descriptor > 0); + CHECK(multicast_send(f->inst[A], descriptor, &dst, 1, "loss", 4) == 0); + CHECK(stats(f, A).report_bytes <= MULTICAST_REPORT_BYTES); + } + CHECK(stats(f, A).reports_evicted > 0 && l.count == 0); + pump(f); CHECK(l.count > 0 && l.count < 200); + CHECK(stats(f, A).queues == 1); + /* Два готовых уведомления одного FD; close в первом отменяет второе. */ + struct losses close_loss = { .close = 1 }; + int closing = multicast_open(f->group[A][0], loss, &close_loss); + uint64_t missing[] = {900, 901}; + CHECK(multicast_send(f->inst[A], closing, missing, 1, "first", 5) == 0); + CHECK(multicast_send(f->inst[A], closing, missing + 1, 1, "second", 6) == 0); + pump(f); CHECK(close_loss.count == 1 && close_loss.item[0].reason == MULTICAST_NO_ROUTE); + multicast_close(f->inst[A], fd); CHECK(stats(f, A).queued_bytes == 0); + teardown(f); +} + +static void test_reroute(void) { + DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "multicast: route change behind blocked head and ETCP DELETE with pending waiter"); + struct fixture* f = setup(); + struct losses l = {0}; struct received c = {0}, d = {0}; + int fd = multicast_open(f->group[A][0], loss, &l); + struct multicast_subscription* sc = multicast_subscribe(f->group[C][0], 1, fd, receive, &c); CHECK(sc); + struct multicast_subscription* sd = multicast_subscribe(f->group[D][0], 1, fd, receive, &d); CHECK(sd); + uint64_t b = 2, target = 3; + block(&f->wire[A][B]); + CHECK(multicast_send(f->inst[A], fd, &b, 1, "blocked head", 12) == 0); + CHECK(multicast_send(f->inst[A], fd, &target, 1, "reroute", 7) == 0); + pump(f); + route(f, A, 0, 3, D); multicast_routes_changed(f->inst[A]); pump(f); + CHECK(c.count == 1 && memcmp(c.prefix, "reroute", 7) == 0 && f->wire[A][B].packets == 0); + route(f, A, 0, 2, D); + struct wire* old = &f->wire[A][B]; + old->conn.links_up = 0; old->conn.state = 2; + etcp_fire_conn_status(&old->conn, ETCP_CONN_STATUS_DELETE); + clear_queue(old->conn.send_input_q); old->conn.send_input_q = NULL; old->blocker = NULL; + pump(f); CHECK(stats(f, A).queued_bytes == 0 && l.count == 0); + CHECK(stats(f, B).unsubscribed == 1); + multicast_unsubscribe(sc); multicast_unsubscribe(sd); teardown(f); +} + +static void put_le(uint8_t* p, uint64_t value, unsigned n) { + for (unsigned i = 0; i < n; i++) p[i] = value >> (i * 8); +} + +static struct ll_entry* data_packet(uint16_t fd, const uint64_t* ids, unsigned count, const uint64_t* path, unsigned hops) { + unsigned len = MULTICAST_HEADER_SIZE + count * 8 + hops * 8 + 1; + struct ll_entry* e = ll_alloc_lldgram(len); CHECK(e); e->len = len; + e->dgram[0] = ETCP_ID_MULTICAST; put_le(e->dgram + 1, TOPO_GROUP_UTUN, 8); + put_le(e->dgram + 9, fd, 2); put_le(e->dgram + 11, count, 2); put_le(e->dgram + 13, 1, 8); + for (unsigned i = 0; i < count; i++) put_le(e->dgram + MULTICAST_HEADER_SIZE + i * 8, ids[i], 8); + for (unsigned i = 0; i < hops; i++) put_le(e->dgram + MULTICAST_HEADER_SIZE + count * 8 + i * 8, path[i], 8); + e->dgram[len - 1] = hops; return e; +} + +static void test_validation(void) { + DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "multicast: malformed wire, loop prevention and argument rejection without partial send"); + struct fixture* f = setup(); + struct losses l = {0}; struct received r = {0}; + int fd = multicast_open(f->group[A][0], loss, &l); + struct multicast_subscription* sub = multicast_subscribe(f->group[B][0], 1, fd, receive, &r); CHECK(sub); + uint64_t ids[] = {2, 2}, path[] = {1, 2, 4}; + etcp_recv_fn recv = f->inst[B]->api_bindings.callbacks[ETCP_ID_MULTICAST]; + recv(&f->wire[B][A].conn, data_packet(fd, ids, 2, path, 1)); /* повтор адресата */ + recv(&f->wire[B][A].conn, data_packet(fd, ids, 1, path, 2)); /* последний hop не равен пиру */ + recv(&f->wire[B][D].conn, data_packet(fd, ids, 1, path, 3)); /* повторное посещение B */ + struct ll_entry* e = data_packet(fd, ids, 1, path, 1); e->len--; + recv(&f->wire[B][A].conn, e); /* обрезанный trailer */ + CHECK(stats(f, B).invalid_packets == 3 && stats(f, B).loop_packets == 1); + uint8_t payload[PKTNORM_MAX_DGRAM_SIZE] = {0}; + CHECK(multicast_send(f->inst[A], fd, ids, 1, payload, sizeof(payload)) < 0); + ids[1] = 0; + CHECK(multicast_send(f->inst[A], fd, ids, 2, NULL, 0) < 0); + CHECK(multicast_send(f->inst[A], 0, ids, 1, NULL, 0) < 0); + pump(f); CHECK(r.count == 0 && l.count == 0 && f->wire[A][B].packets == 0); + multicast_unsubscribe(sub); teardown(f); +} + +static void test_descriptor_wrap(void) { + DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "multicast: sequential 16-bit IDs, exhaustion, wrap and skipping occupied IDs"); + struct fixture* f = setup(); + struct losses l = {0}; + for (int i = 1; i <= UINT16_MAX; i++) CHECK(multicast_open(f->group[A][i % GROUPS], loss, &l) == i); + CHECK(multicast_open(f->group[A][0], loss, &l) < 0); + multicast_close(f->inst[A], 2); multicast_close(f->inst[A], 65000); + CHECK(multicast_open(f->group[A][0], loss, &l) == 2); + CHECK(multicast_open(f->group[A][0], loss, &l) == 65000); + CHECK(stats(f, A).descriptors == UINT16_MAX); + teardown(f); +} + +int main(void) { + debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN); debug_set_categories(DEBUG_CATEGORY_ALL); + debug_set_category_level(DEBUG_CATEGORY_GENERAL, DEBUG_LEVEL_INFO); + test_clock_init(1000); + test_delivery(); test_lifetime(); test_overload(); test_report_budget(); + test_reroute(); test_validation(); test_descriptor_wrap(); + test_clock_finish(); + printf("PASS multicast: delivery, lifetime, overload, reports, reroute, validation, descriptor wrap\n"); + return 0; +}