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

Test bounded monitor TCP sends and graph reconnect behavior

Browse Source
This commit is contained in:
evgeny committed 2026-10-27 13:47:41 +01:00
1 parent 488d8a601f
commit ae594e1e1b
2 files changed
+118 -11
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tools/etcpmon/etcpmon_client.c
+12 -11
View File
@@ -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);
tools/etcpmon/test_etcpmon.c
+106
View File
@@ -1,8 +1,13 @@
/* Регрессии опроса и выбора connections на настоящих Winsock/WinAPI. */
#include <assert.h>
#include <winsock2.h>
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);