From 34e346463d3b0603a6737db6a9b7b9e728fd1cdb Mon Sep 17 00:00:00 2001 From: evgeny Date: Thu, 22 Oct 2026 22:38:20 +0200 Subject: [PATCH] Cover initial UP, pacing and bidirectional tight windows --- tests/test_etcp_windows.c | 41 +++++++++++++++++++++--- tests/test_stcp_traffic.c | 65 ++++++++++++++++----------------------- 2 files changed, 64 insertions(+), 42 deletions(-) diff --git a/tests/test_etcp_windows.c b/tests/test_etcp_windows.c index a8dfeacc..5434287b 100644 --- a/tests/test_etcp_windows.c +++ b/tests/test_etcp_windows.c @@ -221,11 +221,11 @@ static void removed_link_and_bbr(void) { ack(1); CHECK(!conn->unacked_bytes); b->bbr = u_calloc(1, sizeof(*b->bbr)); CHECK(b->bbr); bbr_init(b->bbr); b->inflight_lim_bytes = 8192; - for (int i = 0; i < 8; i++) enqueue(1024); - for (int i = 0; i < 8; i++) { + for (int i = 0; i < 7; i++) enqueue(1024); + for (int i = 0; i < 7; i++) { struct ETCP_DGRAM* pkt = request(); CHECK(pkt); memory_pool_free(inst.pkt_pool, pkt); } - enqueue(100); CHECK(!request() && b->cwnd_limited); + enqueue(1025); CHECK(!request() && b->cwnd_limited && b->inflight_bytes < b->inflight_lim_bytes); // Первый ACK относится к отправке при наличии ожидающих данных: она не app_limited. struct INFLIGHT_PACKET* first = (struct INFLIGHT_PACKET*)conn->input_wait_ack->head; CHECK(!first->is_app_limited); @@ -233,6 +233,39 @@ static void removed_link_and_bbr(void) { teardown(); } +static void shaper_ready(void* arg) { + struct ETCP_LINK* link = arg; + link->shaper_timer = NULL; + loadbalancer_link_ready(link); +} + +static void pacing_reset_close(void) { + setup(); b->link_status = 0; + a->shaper_timer = uasync_set_timeout(ua, 20, a, shaper_ready, "test_shaper"); CHECK(a->shaper_timer); + enqueue(100); CHECK(!request()); + wait_sends(1); ack_all(); + enqueue(100); CHECK(conn->tx_timer); + etcp_conn_reinit_id(conn, "queued sender reset", conn->reset_id); + CHECK(!conn->tx_timer && !conn->unacked_bytes && conn->reinit_pending); + enqueue(100); CHECK(conn->tx_timer); + CHECK(etcp_conn_ref_take(conn) == 0); + etcp_connection_close(conn); CHECK(conn->state == 2 && !conn->tx_timer); + etcp_conn_ref_free(conn); + teardown(); +} + +/* Отправитель не превращает pending transport в UP до первого INIT/подтверждения сессии. */ +static void initial_readiness(void) { + setup(); conn->state = 0; conn->initialized = conn->links_up = conn->tx_state = 0; + conn->session_required = 1; enqueue(100); + test_clock_run_for(ua, 20); + CHECK(!conn->tx_state && !sends && !conn->links_up); + conn->link_ready_for_send_fn(conn); CHECK(!conn->links_up); + conn->session_required = 0; etcp_conn_ready(conn); wait_sends(1); + CHECK(conn->state == 1 && conn->links_up && conn->initialized); + ack_all(); teardown(); +} + static void config_limits(void) { const char* values[] = {"0", "8191", "-1", "2147483648", "999999999999999999999", "abc", "8192x", "8192", "65536"}; for (unsigned i = 0; i < sizeof(values) / sizeof(values[0]); i++) { @@ -249,7 +282,7 @@ int main(void) { test_clock_init(10); size_t baseline = u_get_allocated_count(); void (*cases[])(void) = {selection, boundaries_ack_burst, late_ack_and_migration, lost_ack_and_shrink, - grow_and_session, allocation_and_send_failure, removed_link_and_bbr, config_limits}; + grow_and_session, allocation_and_send_failure, removed_link_and_bbr, pacing_reset_close, initial_readiness, config_limits}; for (unsigned i = 0; i < sizeof(cases) / sizeof(cases[0]); i++) { cases[i](); CHECK(u_get_allocated_count() == baseline); DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "PASS ETCP window scenario %u", i); diff --git a/tests/test_stcp_traffic.c b/tests/test_stcp_traffic.c index 552ccc07..44b67782 100644 --- a/tests/test_stcp_traffic.c +++ b/tests/test_stcp_traffic.c @@ -28,8 +28,7 @@ static int test_failed = 0; } while(0) struct rx_ctx { - uint8_t *buf; - size_t len, cap; + size_t len; int pkts; int done; size_t expected_total; @@ -41,32 +40,22 @@ static struct ETCP_CONN *srv_conn, *cli_conn; static struct NODE_CONN_DIRECT *cli_owner; static int srv_send_ready, cli_send_ready; -static void srv_recv_cb(struct ETCP_CONN *conn, struct ll_entry *entry) { - (void)conn; - if (!entry->dgram || entry->len < 5) { queue_dgram_free(entry); queue_entry_free(entry); return; } - srv_rx.pkts++; - size_t pay = entry->len - 5; - size_t need = srv_rx.len + pay; - if (need > srv_rx.cap) { srv_rx.cap = need + 65536; srv_rx.buf = u_realloc(srv_rx.buf, srv_rx.cap); } - memcpy(srv_rx.buf + srv_rx.len, entry->dgram + 5, pay); - srv_rx.len += pay; - queue_dgram_free(entry); - queue_entry_free(entry); - if (srv_rx.len >= srv_rx.expected_total) srv_rx.done = 1; -} - -static void cli_recv_cb(struct ETCP_CONN *conn, struct ll_entry *entry) { - (void)conn; - if (!entry->dgram || entry->len < 5) { queue_dgram_free(entry); queue_entry_free(entry); return; } - cli_rx.pkts++; - size_t pay = entry->len - 5; - size_t need = cli_rx.len + pay; - if (need > cli_rx.cap) { cli_rx.cap = need + 65536; cli_rx.buf = u_realloc(cli_rx.buf, cli_rx.cap); } - memcpy(cli_rx.buf + cli_rx.len, entry->dgram + 5, pay); - cli_rx.len += pay; - queue_dgram_free(entry); - queue_entry_free(entry); - if (cli_rx.len >= cli_rx.expected_total) cli_rx.done = 1; +/* Проверяем порядок, отсутствие дублей и каждый байт, включая трафик сразу после NCD UP. */ +static void receive_packet(struct ETCP_CONN* conn, struct ll_entry* entry) { + struct rx_ctx* rx = conn->instance == srv_inst ? &srv_rx : &cli_rx; + unsigned side = rx == &srv_rx ? 1 : 2, seq = 0; + int valid = entry->dgram && entry->len >= 5; + if (valid) { + for (unsigned i = 0; i < 4; i++) seq |= (unsigned)entry->dgram[i + 1] << (8 * i); + valid = entry->dgram[0] == side && seq == (unsigned)rx->pkts; + for (unsigned i = 5; valid && i < entry->len; i++) + valid = entry->dgram[i] == (uint8_t)(i * 7 + seq * 13 + side); + } + if (!valid) { + DEBUG_ERROR(DEBUG_CATEGORY_GENERAL, "Invalid STCP payload side=%u seq=%u expected=%d len=%u", side, seq, rx->pkts, entry->len); + test_failed = 1; + } else { rx->pkts++; rx->len += entry->len - 5; rx->done = rx->len == rx->expected_total; } + queue_dgram_free(entry); queue_entry_free(entry); } static void connect_cb(struct NODE_CONN_DIRECT* h, enum ncd_event event, void* arg) { @@ -101,7 +90,9 @@ static int send_etcp_packet(struct ETCP_CONN *conn, int side, int seq, size_t pa if (!e->dgram) { queue_entry_free(e); return -1; } memcpy(e->dgram, buf, tot); e->len = (uint16_t)tot; - return etcp_send(conn, e); + int rc = etcp_send(conn, e); + if (rc < 0) { queue_dgram_free(e); queue_entry_free(e); } + return rc; } static char *build_server_config(int port) { @@ -109,6 +100,7 @@ static char *build_server_config(int port) { snprintf(buf, sizeof(buf), "[global]\n" "my_node_id=0xAAAAAAAAAAAAAAAA\n" + "bbr_max_cwnd=8192\n" "my_private_key=38240cb82199e504686507f11f6eaa4f740fde6f0c425c495e49a523019a5d68\n" "my_public_key=ce8871f07fa056c636d297115f231b08c29cdf94e0d440fce83a07c34416d36a\n" "\n" @@ -125,6 +117,7 @@ static char *build_client_config(int port) { snprintf(buf, sizeof(buf), "[global]\n" "my_node_id=0xBBBBBBBBBBBBBBBB\n" + "bbr_max_cwnd=8192\n" "my_private_key=704f2e012c8fa8768130cb0f988a997dccb628372bc5ceccacc78dcbfec5916f\n" "my_public_key=b3193173def895bd0fcea6f86af077c7d77216f10395275f627ac18242ec0f01\n" "\n" @@ -156,8 +149,8 @@ int main(void) { memset(&srv_rx, 0, sizeof(srv_rx)); srv_rx.expected_total = TRAFFIC_MB; memset(&cli_rx, 0, sizeof(cli_rx)); cli_rx.expected_total = TRAFFIC_MB; - TASSERT(etcp_bind(srv_inst, 1, srv_recv_cb) >= 0); - TASSERT(etcp_bind(cli_inst, 2, cli_recv_cb) >= 0); + TASSERT(etcp_bind(srv_inst, 1, receive_packet) >= 0); + TASSERT(etcp_bind(cli_inst, 2, receive_packet) >= 0); struct TOPO_GROUP_NODE *srv_node = topo_groups_get_default(srv_inst->topo_groups) ? topo_groups_get_default(srv_inst->topo_groups)->local_node : NULL; TASSERT(srv_node); @@ -197,14 +190,14 @@ int main(void) { if (!srv_done && remaining_s > 0) { size_t sz = PKT_MIN + (rand() % (PKT_MAX - PKT_MIN + 1)); if (sz > remaining_s) sz = remaining_s; - if (send_etcp_packet(srv_conn, 2, seq_s++, sz) == 0) remaining_s -= sz; + if (send_etcp_packet(srv_conn, 2, seq_s, sz) == 0) { remaining_s -= sz; seq_s++; } } if (remaining_s == 0 && seq_s > 0) srv_done = 1; if (!cli_done && remaining_c > 0) { size_t sz = PKT_MIN + (rand() % (PKT_MAX - PKT_MIN + 1)); if (sz > remaining_c) sz = remaining_c; - if (send_etcp_packet(cli_conn, 1, seq_c++, sz) == 0) remaining_c -= sz; + if (send_etcp_packet(cli_conn, 1, seq_c, sz) == 0) { remaining_c -= sz; seq_c++; } } if (remaining_c == 0 && seq_c > 0) cli_done = 1; @@ -217,11 +210,7 @@ int main(void) { DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "srv_rx: %zu bytes, %d pkts, done=%d", srv_rx.len, srv_rx.pkts, srv_rx.done); DEBUG_INFO(DEBUG_CATEGORY_GENERAL, "cli_rx: %zu bytes, %d pkts, done=%d", cli_rx.len, cli_rx.pkts, cli_rx.done); - TASSERT(srv_rx.len >= TRAFFIC_MB); - TASSERT(cli_rx.len >= TRAFFIC_MB); - - if (srv_rx.buf) u_free(srv_rx.buf); - if (cli_rx.buf) u_free(cli_rx.buf); + TASSERT(!test_failed && srv_rx.len == TRAFFIC_MB && cli_rx.len == TRAFFIC_MB); node_conn_direct_close(cli_owner); cli_owner = NULL; srv_inst->running = 0; cli_inst->running = 0;