From ae594e1e1b4eea0ba601f208eeee8c1f04700a7c Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 27 Oct 2026 13:47:41 +0100 Subject: [PATCH] Test bounded monitor TCP sends and graph reconnect behavior --- tools/etcpmon/etcpmon_client.c | 23 +++---- tools/etcpmon/test_etcpmon.c | 106 +++++++++++++++++++++++++++++++++ 2 files changed, 118 insertions(+), 11 deletions(-) diff --git a/tools/etcpmon/etcpmon_client.c b/tools/etcpmon/etcpmon_client.c index 218f3a54..14b8e02e 100644 --- a/tools/etcpmon/etcpmon_client.c +++ b/tools/etcpmon/etcpmon_client.c @@ -30,8 +30,8 @@ struct etcpmon_tx_frame { /* Единственная ветка аварийного отключения: лог, освобождение транспорта и уведомления владельца. */ static int client_fail(struct etcpmon_client* client, const char* reason, int error) { FILE* log = client->log_file ? client->log_file : stderr; - fprintf(log, "%llu: [ERROR] %s: error=%d queued_frames=%u queued_bytes=%zu\n", - (unsigned long long)get_timestamp_ms(), reason, error, client->tx_count, client->tx_bytes); + fprintf(log, "%llu: [ERROR] %s: error=%d queued_frames=%u queued_bytes=%u\n", + (unsigned long long)get_timestamp_ms(), reason, error, client->tx_count, (unsigned)client->tx_bytes); fflush(log); etcpmon_client_disconnect(client); if (client->on_disconnected) client->on_disconnected(client->user_data); @@ -51,8 +51,8 @@ static int flush_outgoing(struct etcpmon_client* client) { int error = WSAGetLastError(); if (error != WSAEWOULDBLOCK) return client_fail(client, "TCP send failed", error); if (!client->tx_blocked && client->log_file) { - fprintf(client->log_file, "[DEBUG] TX waiting: frames=%u bytes=%zu offset=%u/%u\n", - client->tx_count, client->tx_bytes, frame->offset, frame->size); + fprintf(client->log_file, "[DEBUG] TX waiting: frames=%u bytes=%u offset=%u/%u\n", + client->tx_count, (unsigned)client->tx_bytes, frame->offset, frame->size); fflush(client->log_file); } client->tx_blocked = 1; @@ -60,7 +60,7 @@ static int flush_outgoing(struct etcpmon_client* client) { } if (!sent) return client_fail(client, "TCP send made no progress", 0); if (client->tx_blocked && client->log_file) { - fprintf(client->log_file, "[DEBUG] TX resumed: sent=%d frames=%u bytes=%zu\n", sent, client->tx_count, client->tx_bytes); + fprintf(client->log_file, "[DEBUG] TX resumed: sent=%d frames=%u bytes=%u\n", sent, client->tx_count, (unsigned)client->tx_bytes); fflush(client->log_file); } client->tx_blocked = 0; @@ -91,21 +91,22 @@ static int send_message(struct etcpmon_client* client, uint8_t type, const void* if (!client || !client->connected || client->sock == INVALID_SOCKET || size > ETCPMON_MAX_MSG_SIZE || (payload_size && !payload)) { FILE* log = client && client->log_file ? client->log_file : stderr; - fprintf(log, "[ERROR] Invalid outgoing command: type=0x%02X size=%zu connected=%d\n", - type, size, client ? client->connected : 0); + fprintf(log, "[ERROR] Invalid outgoing command: type=0x%02X size=%u connected=%d\n", + type, (unsigned)size, client ? client->connected : 0); fflush(log); free(req); return -1; } if (client->tx_count >= ETCPMON_TX_MAX_FRAMES || size > ETCPMON_TX_MAX_BYTES - client->tx_bytes) { FILE* log = client->log_file ? client->log_file : stderr; - fprintf(log, "[ERROR] TX queue full: type=0x%02X size=%zu frames=%u bytes=%zu\n", type, size, client->tx_count, client->tx_bytes); + fprintf(log, "[ERROR] TX queue full: type=0x%02X size=%u frames=%u bytes=%u\n", + type, (unsigned)size, client->tx_count, (unsigned)client->tx_bytes); fflush(log); free(req); return -1; } struct etcpmon_tx_frame* frame = malloc(sizeof(*frame) + size); if (!frame) { FILE* log = client->log_file ? client->log_file : stderr; - fprintf(log, "[ERROR] TX allocation failed: type=0x%02X size=%zu\n", type, size); + fprintf(log, "[ERROR] TX allocation failed: type=0x%02X size=%u\n", type, (unsigned)size); fflush(log); free(req); return -1; } @@ -121,8 +122,8 @@ static int send_message(struct etcpmon_client* client, uint8_t type, const void* else { client->tx_head = frame; client->tx_progress_at = GetTickCount(); } client->tx_tail = frame; client->tx_bytes += size; client->tx_count++; if (client->log_file) { - fprintf(client->log_file, "[DEBUG] TX queued: type=0x%02X seq=%u size=%zu frames=%u bytes=%zu\n", - type, frame->seq_id, size, client->tx_count, client->tx_bytes); + fprintf(client->log_file, "[DEBUG] TX queued: type=0x%02X seq=%u size=%u frames=%u bytes=%u\n", + type, frame->seq_id, (unsigned)size, client->tx_count, (unsigned)client->tx_bytes); fflush(client->log_file); } return flush_outgoing(client); diff --git a/tools/etcpmon/test_etcpmon.c b/tools/etcpmon/test_etcpmon.c index 96b34096..8d1ae80f 100644 --- a/tools/etcpmon/test_etcpmon.c +++ b/tools/etcpmon/test_etcpmon.c @@ -1,8 +1,13 @@ /* Регрессии опроса и выбора connections на настоящих Winsock/WinAPI. */ #include +#include + +static int WSAAPI test_send(SOCKET fd, const char* bytes, int size, int flags); /* Проверяем также внутренние callbacks без расширения публичного API. */ +#define send test_send #include "etcpmon_client.c" +#undef send #include "etcpmon_gui.c" static SOCKET listener; @@ -10,6 +15,18 @@ static SOCKET server_sock; static struct etcpmon_client client; static struct etcpmon_app app; static int callbacks; +static SOCKET limited_socket = INVALID_SOCKET; +static int send_allowance, send_chunk; + +/* Настоящий TCP с управляемой границей short send и WSAEWOULDBLOCK. */ +static int WSAAPI test_send(SOCKET fd, const char* bytes, int size, int flags) { + if (fd == limited_socket) { + if (!send_allowance) { WSASetLastError(WSAEWOULDBLOCK); return SOCKET_ERROR; } + if (send_allowance > 0) send_allowance--; + if (size > send_chunk) size = send_chunk; + } + return send(fd, bytes, size, flags); +} static void request_callback(void* arg, uint8_t type, const uint8_t* payload, uint16_t size) { (void)arg; (void)type; (void)payload; (void)size; @@ -86,6 +103,88 @@ static void deliver_list(struct etcpmon_conn_info* list, uint8_t count) { assert(SendMessage(app.hListConnections, LB_GETCOUNT, 0, 0) == count); } +/* Повторный вывод тех же величин после очистки и отсутствие старого сэмпла при исчезновении линка. */ +static void test_graph_values(void) { + struct etcpmon_client graph_client; + etcpmon_client_init(&graph_client); + struct etcpmon_client* previous = app.client; + app.client = &graph_client; + struct etcpmon_link_metrics link = {.inflight_lim_bytes = 64000, .bbr_bw_hi = 900000, .bbr_min_rtt_us = 12300}; + struct etcpmon_rsp_metrics metrics = {0}; + graph_client.last_links = &link; graph_client.last_links_count = 1; + etcpmon_client_add_to_history(&graph_client, &metrics); + for (int pass = 0; pass < 2; pass++) { + for (int i = 0; i < GRAPH_METRIC_COUNT; i++) SetWindowTextA(app.hChannelValue[i], ""); + UpdateChannelValues(&app, 0); + char text[64]; + GetWindowTextA(app.hChannelValue[GRAPH_METRIC_INFLIGHT_LIM], text, sizeof(text)); + assert(!strcmp(text, "62.5 KiB")); + GetWindowTextA(app.hChannelValue[GRAPH_METRIC_BW_HI], text, sizeof(text)); + assert(!strcmp(text, "878.9 KiB/s")); + GetWindowTextA(app.hChannelValue[GRAPH_METRIC_MIN_RTT], text, sizeof(text)); + assert(!strcmp(text, "12.3 ms")); + } + graph_client.history.head = 0; graph_client.last_links_count = 0; + etcpmon_client_add_to_history(&graph_client, &metrics); + for (int i = 0; i < GRAPH_METRIC_COUNT; i++) assert(graph_client.history.values[i][0] == 0); + app.client = previous; + puts("PASS graph: bytes/rates, same values after clear and no stale values without links"); +} + +/* Очередь сохраняет wire-порядок и не запускает response timeout до полной отправки. */ +static void test_client_transport(void) { + closesocket(server_sock); connect_client(); + limited_socket = client.sock; send_chunk = 2; send_allowance = 1; + struct etcpmon_cmd_action action = {"nat"}; + assert(etcpmon_client_send_request(&client, ETCPMON_CMD_ACTION, &action, sizeof(action), request_callback, NULL) == 0); + assert(client.tx_count == 1 && client.tx_head->offset == 2 && !client.pending_head->sent); + uint8_t seq = client.pending_head->seq_id; + client.pending_head->sent_at = GetTickCount() - REQUEST_TIMEOUT_MS; + assert(etcpmon_client_process(&client) == 0 && client.connected); + assert(etcpmon_client_request_list(&client) == 0 && client.tx_count == 2); + limited_socket = INVALID_SOCKET; + assert(etcpmon_client_process(&client) >= 0 && !client.tx_head && client.pending_head->sent); + uint8_t expected[sizeof(struct etcpmon_msg_header) * 2 + sizeof(action)], actual[sizeof(expected)]; + etcpmon_build_header((void*)expected, sizeof(action), ETCPMON_CMD_ACTION, seq); + memcpy(expected + sizeof(struct etcpmon_msg_header), &action, sizeof(action)); + etcpmon_build_header((void*)(expected + sizeof(struct etcpmon_msg_header) + sizeof(action)), 0, ETCPMON_CMD_LIST_CONN, 0); + size_t used = 0; + for (int i = 0; i < 1000 && used < sizeof(actual); i++) { + int got = recv(server_sock, (char*)actual + used, sizeof(actual) - used, 0); + if (got == SOCKET_ERROR) assert(WSAGetLastError() == WSAEWOULDBLOCK); + else { assert(got > 0); used += got; } + Sleep(1); + } + assert(used == sizeof(actual) && !memcmp(expected, actual, used)); + uint8_t error[] = {ETCPMON_ERR_INVALID_CMD, 'n','o',0}; + client.last_response_seq_id = seq; + int before = callbacks; + handle_response_with_callback(&client, ETCPMON_RSP_ERROR, error, sizeof(error)); + assert(client.connected && !client.pending_head && callbacks == before + 1); + + limited_socket = client.sock; send_allowance = 0; + uint8_t payload[ETCPMON_MAX_MSG_SIZE - sizeof(struct etcpmon_msg_header)] = {0}; + for (int i = 0; i < 2; i++) + assert(etcpmon_client_send_request(&client, ETCPMON_CMD_ACTION, payload, sizeof(payload), request_callback, NULL) == 0); + assert(client.tx_bytes == ETCPMON_TX_MAX_BYTES && client.tx_count == 2); + assert(etcpmon_client_send_request(&client, ETCPMON_CMD_ACTION, NULL, 0, request_callback, NULL) == -1); + assert(client.connected && client.tx_count == 2); + client.tx_progress_at = GetTickCount() - REQUEST_TIMEOUT_MS; + assert(etcpmon_client_process(&client) == -1 && !client.connected && !client.tx_head && !client.pending_head); + limited_socket = INVALID_SOCKET; + for (int malformed = 0; malformed < 2; malformed++) { + closesocket(server_sock); connect_client(); + assert(etcpmon_client_send_request(&client, ETCPMON_CMD_ACTION, &action, sizeof(action), request_callback, NULL) == 0); + client.last_response_seq_id = client.pending_head->seq_id; + before = callbacks; + const uint8_t bad[] = {0, 'x'}; + handle_response_with_callback(&client, malformed ? ETCPMON_RSP_ACTION_RESULT : ETCPMON_RSP_CONN_LIST, + bad, malformed ? sizeof(bad) : 1); + assert(!client.connected && !client.pending_head && callbacks == before); + } + puts("PASS client TX: short send/EWOULDBLOCK, exact FIFO, queue bound, progress deadline and response validation"); +} + int main(void) { WSADATA data; assert(WSAStartup(MAKEWORD(2, 2), &data) == 0); @@ -108,6 +207,11 @@ int main(void) { app.hListSockets = CreateWindowExA(0, "LISTBOX", "", WS_CHILD, 0, 100, 400, 100, app.hWndMain, (HMENU)IDC_LIST_SOCKETS, GetModuleHandle(NULL), NULL); assert(app.hListConnections && app.hListSockets); + for (int i = 0; i < GRAPH_METRIC_COUNT; i++) { + app.hChannelValue[i] = CreateWindowExA(0, "EDIT", "", WS_CHILD, 0, 0, 100, 20, + app.hWndMain, (HMENU)(UINT_PTR)(IDC_CH_VALUE_0 + i), GetModuleHandle(NULL), NULL); + assert(app.hChannelValue[i]); + } etcpmon_client_set_callbacks(&client, on_connected, on_disconnected, on_conn_list, on_socket_list, on_metrics, on_error, NULL, &app); connect_client(); @@ -231,6 +335,8 @@ int main(void) { assert(etcpmon_client_send_request(&client, ETCPMON_CMD_ACTION, NULL, 0, request_callback, NULL) == -1); assert(!client.connected && client.pending_head == NULL && client.pending_tail == NULL && app.need_initial_request); + test_graph_values(); + test_client_transport(); etcpmon_client_cleanup(&client); closesocket(server_sock); closesocket(listener);