diff --git a/src/routing_layer/multicast.c b/src/routing_layer/multicast.c index 498719b3..d76467ec 100644 --- a/src/routing_layer/multicast.c +++ b/src/routing_layer/multicast.c @@ -304,7 +304,11 @@ static void mc_loss(struct multicast_ctx* ctx, const struct mc_packet* p, uint64 *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; } + if (!r) { + DEBUG_ERROR(MC_CAT, "multicast loss allocation failed bytes=%zu", bytes); + mc_report_schedule(ctx); /* Созданное окно должно истечь даже при отказе выделения отчёта. */ + 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, diff --git a/tests/test_multicast.c b/tests/test_multicast.c index 82d7e118..83b703ee 100644 --- a/tests/test_multicast.c +++ b/tests/test_multicast.c @@ -179,7 +179,16 @@ static struct fixture* setup(void) { return f; } -static void pump(struct fixture* f) { test_clock_run_for(f->ua, 10); } +/* Дренируем готовую работу независимо от скорости CPU/Valgrind, не сдвигая будущие окна. */ +static void pump(struct fixture* f) { + for (int i = 0; i < 1024; i++) { + uasync_drain_immediate(f->ua); uasync_poll(f->ua, 0); + TimeoutEntry next; + if (!f->ua->immediate_queue_head && + (timeout_heap_peek(f->ua->timeout_heap, &next) || next.expiration > get_time_us() / 1000)) return; + } + CHECK(0 && "multicast immediate work did not settle"); +} 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; @@ -191,6 +200,7 @@ static void teardown(struct fixture* f) { 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_poll(f->ua, 0); /* Освободить помеченные отменёнными heap-таймеры. */ 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); @@ -411,6 +421,63 @@ static void test_validation(void) { multicast_unsubscribe(sub); teardown(f); } +static void test_long_report(void) { + DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "multicast: longest recipient list and loss report split into router-sized parts"); + struct fixture* f = setup(); + struct losses l = {0}; + int fd = multicast_open(f->group[A][0], loss, &l); + block(&f->wire[B][C]); + uint64_t dst = 3; + uint8_t payload[8146] = {0}; + for (int i = 0; i < 32; i++) { + CHECK(multicast_send(f->inst[A], fd, &dst, 1, payload, sizeof(payload)) == 0); pump(f); + } + CHECK(stats(f, B).queued_bytes == MULTICAST_QUEUE_BYTES); + enum { COUNT = (PKTNORM_MAX_DGRAM_SIZE - MULTICAST_HEADER_SIZE - MULTICAST_PATH_RESERVE) / 8 }; + uint64_t ids[COUNT]; + for (unsigned i = 0; i < COUNT; i++) { + ids[i] = i + 100; route(f, A, 0, ids[i], B); route(f, B, 0, ids[i], C); + } + CHECK(multicast_send(f->inst[A], fd, ids, COUNT, NULL, 0) == 0); pump(f); + CHECK(l.count == 2 && l.item[0].recipient_count + l.item[1].recipient_count == COUNT); + CHECK(l.item[0].packets == 1 && l.item[1].packets == 1 && !l.item[0].prefix_len && !l.item[1].prefix_len); + CHECK(l.ids[0][0] == 100 && l.ids[1][0] == 100 + l.item[0].recipient_count); + CHECK(stats(f, B).reports_sent == 2 && stats(f, A).reports_received == 2); + teardown(f); +} + +static void test_report_backpressure(void) { + DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "multicast: blocked report transport, eviction, restart and stale closed descriptor"); + struct fixture* f = setup(); + struct losses l = {0}; + int fd = multicast_open(f->group[A][0], loss, &l); + block(&f->wire[B][C]); block(&f->wire[B][A]); + uint64_t dst = 3; + uint8_t payload[8146] = {0}; + for (int i = 0; i < 32; i++) { + CHECK(multicast_send(f->inst[A], fd, &dst, 1, payload, sizeof(payload)) == 0); pump(f); + } + int closed = 0; + for (int i = 0; i < 200; i++) { + int descriptor = multicast_open(f->group[A][0], loss, &l); CHECK(descriptor > 0); + if (!i) closed = descriptor; + CHECK(multicast_send(f->inst[A], descriptor, &dst, 1, "congested", 9) == 0); pump(f); + CHECK(stats(f, B).report_bytes <= MULTICAST_REPORT_BYTES); + } + CHECK(l.count == 0 && stats(f, B).reports_evicted > 0); + multicast_close(f->inst[A], closed); + /* Перезапуск router очищает send_q и отменяет её waiters; таймер должен продолжить отчёты. */ + etcp_router_conn_restart(f->inst[B], f->group[B][0]->group_id, 1, ETCP_RT_ID_MULTICAST_REPORT); + unblock(&f->wire[B][A]); test_clock_run_for(f->ua, 10000); + CHECK(l.count > 0 && l.count < 200); + for (unsigned i = 0; i < l.count; i++) { + CHECK(l.item[i].descriptor != closed && l.item[i].reason == MULTICAST_OVERLOAD); + CHECK(l.item[i].src == 2 && l.item[i].dst == 3 && l.item[i].packets == 1); + } + CHECK(stats(f, B).report_bytes == 0); + 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(); @@ -434,7 +501,9 @@ int main(void) { debug_set_category_level(DEBUG_CATEGORY_TIMERS, DEBUG_LEVEL_DEBUG); } test_delivery(); test_lifetime(); test_overload(); test_report_budget(); - test_reroute(); test_validation(); test_descriptor_wrap(); + test_reroute(); test_validation(); test_long_report(); test_report_backpressure(); + debug_set_category_level(DEBUG_CATEGORY_ETCPROUTE, DEBUG_LEVEL_WARN); + test_descriptor_wrap(); test_clock_finish(); printf("PASS multicast: delivery, lifetime, overload, reports, reroute, validation, descriptor wrap\n"); return 0;