From ce06fe242bdb67d862ca0a2f207422fb59bfd740 Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 27 Oct 2026 13:38:40 +0100 Subject: [PATCH] Queue monitor TCP commands and validate matched responses --- tools/etcpmon/etcpmon_client.c | 639 ++++++++++++++------------------- tools/etcpmon/etcpmon_client.h | 123 ++++--- tools/etcpmon/etcpmon_gui.c | 4 +- 3 files changed, 344 insertions(+), 422 deletions(-) diff --git a/tools/etcpmon/etcpmon_client.c b/tools/etcpmon/etcpmon_client.c index 690a1071d..218f3a541 100644 --- a/tools/etcpmon/etcpmon_client.c +++ b/tools/etcpmon/etcpmon_client.c @@ -18,6 +18,116 @@ static struct etcpmon_pending_request* find_pending_request(struct etcpmon_client* client, uint8_t seq_id); +/* Кадр живёт до полной отправки; offset не меняет исходные байты и порядок FIFO. */ +struct etcpmon_tx_frame { + struct etcpmon_tx_frame* next; + uint16_t size; + uint16_t offset; + uint8_t seq_id; + uint8_t bytes[]; +}; + +/* Единственная ветка аварийного отключения: лог, освобождение транспорта и уведомления владельца. */ +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); + fflush(log); + etcpmon_client_disconnect(client); + if (client->on_disconnected) client->on_disconnected(client->user_data); + if (client->on_error) client->on_error(reason, client->user_data); + return -1; +} + +/* Неблокирующая досылка: за вызов не более одного максимального кадра по объёму. */ +static int flush_outgoing(struct etcpmon_client* client) { + size_t budget = ETCPMON_MAX_MSG_SIZE; + while (client->fully_connected && client->tx_head && budget) { + struct etcpmon_tx_frame* frame = client->tx_head; + int size = frame->size - frame->offset; + if ((size_t)size > budget) size = (int)budget; + int sent = send(client->sock, (const char*)frame->bytes + frame->offset, size, 0); + if (sent == SOCKET_ERROR) { + 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); + fflush(client->log_file); + } + client->tx_blocked = 1; + return 0; + } + 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); + fflush(client->log_file); + } + client->tx_blocked = 0; + client->tx_progress_at = GetTickCount(); + frame->offset += sent; + budget -= sent; + if (frame->offset != frame->size) continue; + struct etcpmon_pending_request* req = find_pending_request(client, frame->seq_id); + if (req) { req->sent = 1; req->sent_at = client->tx_progress_at; } + if (client->log_file) { + fprintf(client->log_file, "[DEBUG] TX complete: type=0x%02X seq=%u size=%u\n", + frame->bytes[2], frame->seq_id, frame->size); + fflush(client->log_file); + } + client->tx_head = frame->next; + if (!client->tx_head) client->tx_tail = NULL; + client->tx_bytes -= frame->size; + client->tx_count--; + free(frame); + } + return 0; +} + +/* Принимает req во владение, если передан; публикация кадра и pending выполняется до первого send. */ +static int send_message(struct etcpmon_client* client, uint8_t type, const void* payload, + uint16_t payload_size, struct etcpmon_pending_request* req) { + size_t size = sizeof(struct etcpmon_msg_header) + (size_t)payload_size; + 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); + 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); + 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); + fflush(log); free(req); + return -1; + } + frame->next = NULL; frame->size = (uint16_t)size; frame->offset = 0; frame->seq_id = req ? req->seq_id : 0; + etcpmon_build_header((void*)frame->bytes, payload_size, type, frame->seq_id); + if (payload_size) memcpy(frame->bytes + sizeof(struct etcpmon_msg_header), payload, payload_size); + if (req) { + if (client->pending_tail) client->pending_tail->next = req; + else client->pending_head = req; + client->pending_tail = req; + } + if (client->tx_tail) client->tx_tail->next = frame; + 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); + fflush(client->log_file); + } + return flush_outgoing(client); +} + /* Get current timestamp in milliseconds (Unix epoch) */ uint64_t get_timestamp_ms(void) { FILETIME ft; @@ -140,7 +250,8 @@ int etcpmon_client_connect(struct etcpmon_client* client, const char* addr, uint /* Set non-blocking mode */ u_long nonblock = 1; - ioctlsocket(client->sock, FIONBIO, &nonblock); + if (ioctlsocket(client->sock, FIONBIO, &nonblock) == SOCKET_ERROR) + return client_fail(client, "Failed to enable nonblocking TCP", WSAGetLastError()); /* Setup address */ struct sockaddr_in server_addr; @@ -188,6 +299,7 @@ int etcpmon_client_connect(struct etcpmon_client* client, const char* addr, uint client->server_port = port; client->connected = 1; client->fully_connected = 0; + client->connect_started_at = GetTickCount(); if (client->log_file) { fprintf(client->log_file, "%llu: [LOG] Socket created, connection in progress\n", @@ -202,20 +314,6 @@ void etcpmon_client_disconnect(struct etcpmon_client* client) { if (!client) return; if (client->sock != INVALID_SOCKET) { - /* Send disconnect command */ - if (client->connected) { - struct etcpmon_msg_header hdr; - etcpmon_build_header(&hdr, 0, ETCPMON_CMD_DISCONNECT, 0); - - if (client->log_file) { - log_hex_data(client->log_file, "TX", (const uint8_t*)&hdr, sizeof(hdr)); - fprintf(client->log_file, "%llu: [LOG] Sending CMD_DISCONNECT\n", - (unsigned long long)get_timestamp_ms()); - } - - send(client->sock, (const char*)&hdr, sizeof(hdr), 0); - } - if (client->log_file) { fprintf(client->log_file, "%llu: [LOG] Connection closed\n", (unsigned long long)get_timestamp_ms()); @@ -228,6 +326,13 @@ void etcpmon_client_disconnect(struct etcpmon_client* client) { client->connected = 0; client->fully_connected = 0; + while (client->tx_head) { + struct etcpmon_tx_frame* frame = client->tx_head; + client->tx_head = frame->next; + free(frame); + } + client->tx_tail = NULL; client->tx_count = 0; client->tx_bytes = 0; + client->tx_blocked = 0; client->tx_progress_at = 0; client->connect_started_at = 0; client->recv_len = 0; client->selected_peer_id = 0; client->last_metrics_request_ms = 0; @@ -262,60 +367,12 @@ int etcpmon_client_is_connected(const struct etcpmon_client* client) { } int etcpmon_client_request_list(struct etcpmon_client* client) { - if (!client || client->sock == INVALID_SOCKET) return -1; - - struct etcpmon_msg_header hdr; - etcpmon_build_header(&hdr, 0, ETCPMON_CMD_LIST_CONN, 0); /* seq=0 = legacy режим */ - - if (client->log_file) { - log_hex_data(client->log_file, "TX", (const uint8_t*)&hdr, sizeof(hdr)); - fprintf(client->log_file, "%llu: [LOG] CMD_LIST_CONN (seq=0)\n", - (unsigned long long)get_timestamp_ms()); - fflush(client->log_file); - } - - int sent = send(client->sock, (const char*)&hdr, sizeof(hdr), 0); - if (sent != sizeof(hdr)) { - if (client->log_file) { - fprintf(client->log_file, "%llu: [ERROR] Failed to send CMD_LIST_CONN: %d\n", - (unsigned long long)get_timestamp_ms(), WSAGetLastError()); - fflush(client->log_file); - } - return -1; - } - - return 0; + return send_message(client, ETCPMON_CMD_LIST_CONN, NULL, 0, NULL); } int etcpmon_client_select_connection(struct etcpmon_client* client, uint64_t peer_node_id) { - if (!client || client->sock == INVALID_SOCKET) return -1; - - /* SELECT is special - it's not a request/response, it just sets state - * So we send with seq_id=0 (broadcast) */ - uint8_t buffer[sizeof(struct etcpmon_msg_header) + sizeof(struct etcpmon_cmd_select)]; - struct etcpmon_msg_header* hdr = (struct etcpmon_msg_header*)buffer; - struct etcpmon_cmd_select* cmd = (struct etcpmon_cmd_select*)(buffer + sizeof(*hdr)); - - etcpmon_build_header(hdr, sizeof(*cmd), ETCPMON_CMD_SELECT_CONN, 0); /* seq_id=0 for SELECT */ - cmd->peer_node_id = peer_node_id; - - if (client->log_file) { - log_hex_data(client->log_file, "TX", buffer, sizeof(buffer)); - fprintf(client->log_file, "%llu: [LOG] CMD_SELECT_CONN peer_id=%016llX\n", - (unsigned long long)get_timestamp_ms(), (unsigned long long)peer_node_id); - fflush(client->log_file); - } - - int sent = send(client->sock, (const char*)buffer, sizeof(buffer), 0); - if (sent != sizeof(buffer)) { - if (client->log_file) { - fprintf(client->log_file, "%llu: [ERROR] Failed to send CMD_SELECT_CONN: %d\n", - (unsigned long long)get_timestamp_ms(), WSAGetLastError()); - fflush(client->log_file); - } - return -1; - } - + struct etcpmon_cmd_select command = {peer_node_id}; + if (send_message(client, ETCPMON_CMD_SELECT_CONN, &command, sizeof(command), NULL) < 0) return -1; client->selected_peer_id = peer_node_id; client->last_metrics_request_ms = 0; memset(&client->history, 0, sizeof(client->history)); @@ -323,55 +380,11 @@ int etcpmon_client_select_connection(struct etcpmon_client* client, uint64_t pee } int etcpmon_client_request_socket_list(struct etcpmon_client* client) { - if (!client || client->sock == INVALID_SOCKET) return -1; - - struct etcpmon_msg_header hdr; - etcpmon_build_header(&hdr, 0, ETCPMON_CMD_LIST_SOCKETS, 0); - - if (client->log_file) { - log_hex_data(client->log_file, "TX", (const uint8_t*)&hdr, sizeof(hdr)); - fprintf(client->log_file, "%llu: [LOG] CMD_LIST_SOCKETS (seq=0)\n", - (unsigned long long)get_timestamp_ms()); - fflush(client->log_file); - } - - int sent = send(client->sock, (const char*)&hdr, sizeof(hdr), 0); - if (sent != sizeof(hdr)) { - if (client->log_file) { - fprintf(client->log_file, "%llu: [ERROR] Failed to send CMD_LIST_SOCKETS: %d\n", - (unsigned long long)get_timestamp_ms(), WSAGetLastError()); - fflush(client->log_file); - } - return -1; - } - - return 0; + return send_message(client, ETCPMON_CMD_LIST_SOCKETS, NULL, 0, NULL); } int etcpmon_client_request_metrics(struct etcpmon_client* client) { - if (!client || client->sock == INVALID_SOCKET) return -1; - - struct etcpmon_msg_header hdr; - etcpmon_build_header(&hdr, 0, ETCPMON_CMD_GET_METRICS, 0); - - if (client->log_file) { - log_hex_data(client->log_file, "TX", (const uint8_t*)&hdr, sizeof(hdr)); - fprintf(client->log_file, "%llu: [LOG] CMD_GET_METRICS\n", - (unsigned long long)get_timestamp_ms()); - fflush(client->log_file); - } - - int sent = send(client->sock, (const char*)&hdr, sizeof(hdr), 0); - if (sent != sizeof(hdr)) { - if (client->log_file) { - fprintf(client->log_file, "%llu: [ERROR] Failed to send CMD_GET_METRICS: %d\n", - (unsigned long long)get_timestamp_ms(), WSAGetLastError()); - fflush(client->log_file); - } - return -1; - } - - return 0; + return send_message(client, ETCPMON_CMD_GET_METRICS, NULL, 0, NULL); } static void action_auto_callback(void* arg, uint8_t msg_type, const uint8_t* payload, uint16_t payload_size) { @@ -393,93 +406,33 @@ int etcpmon_client_send_action(struct etcpmon_client* client, const char* action /* Send request with callback - queued version */ int etcpmon_client_send_request(struct etcpmon_client* client, - uint8_t type, - const void* payload, uint16_t payload_size, - void (*callback)(void* arg, uint8_t msg_type, const uint8_t* payload, uint16_t payload_size), - void* arg) { - if (!client || client->sock == INVALID_SOCKET || !callback) return -1; - if (payload_size > ETCPMON_MAX_MSG_SIZE - sizeof(struct etcpmon_msg_header) || (payload_size && !payload)) { - if (client->log_file) { - fprintf(client->log_file, "%llu: [ERROR] Invalid request payload: type=0x%02X size=%u\n", - (unsigned long long)get_timestamp_ms(), type, payload_size); - fflush(client->log_file); - } + uint8_t type, const void* payload, uint16_t payload_size, + void (*callback)(void*, uint8_t, const uint8_t*, uint16_t), void* arg) { + if (!client || !callback) { + fprintf(stderr, "[ERROR] Invalid request owner/callback: type=0x%02X\n", type); return -1; } - - /* Не переиспользуем seq_id ещё ожидающего ответа запроса. */ int attempts = 0; while (find_pending_request(client, client->next_seq_id) && attempts++ < 255) { if (++client->next_seq_id == 0) client->next_seq_id = 1; } if (attempts >= 255) { - if (client->log_file) { - fprintf(client->log_file, "%llu: [ERROR] All request sequence IDs are in use\n", - (unsigned long long)get_timestamp_ms()); - fflush(client->log_file); - } + FILE* log = client->log_file ? client->log_file : stderr; + fprintf(log, "[ERROR] All request sequence IDs are in use\n"); + fflush(log); return -1; } - - /* Allocate pending request */ - struct etcpmon_pending_request* req = (struct etcpmon_pending_request*)malloc(sizeof(*req)); + struct etcpmon_pending_request* req = calloc(1, sizeof(*req)); if (!req) { - if (client->log_file) { - fprintf(client->log_file, "%llu: [ERROR] Failed to allocate pending request\n", - (unsigned long long)get_timestamp_ms()); - fflush(client->log_file); - } + FILE* log = client->log_file ? client->log_file : stderr; + fprintf(log, "[ERROR] Pending request allocation failed: type=0x%02X\n", type); + fflush(log); return -1; } - - /* Assign sequence ID */ req->seq_id = client->next_seq_id++; - if (client->next_seq_id == 0) client->next_seq_id = 1; /* Skip 0 (broadcast) */ - req->callback = callback; - req->arg = arg; - req->type = type; - req->sent_at = GetTickCount(); - req->next = NULL; - - /* Build message with seq_id */ - uint8_t buffer[ETCPMON_MAX_MSG_SIZE]; - - struct etcpmon_msg_header* hdr = (struct etcpmon_msg_header*)buffer; - etcpmon_build_header(hdr, payload_size, type, req->seq_id); - - if (payload && payload_size > 0) { - memcpy(buffer + sizeof(*hdr), payload, payload_size); - } - - uint16_t total_size = sizeof(*hdr) + payload_size; - - if (client->log_file) { - log_hex_data(client->log_file, "TX", buffer, total_size); - fprintf(client->log_file, "%llu: [LOG] CMD 0x%02X seq=%d queued\n", - (unsigned long long)get_timestamp_ms(), type, req->seq_id); - fflush(client->log_file); - } - - int sent = send(client->sock, (const char*)buffer, total_size, 0); - if (sent != total_size) { - free(req); - if (client->log_file) { - fprintf(client->log_file, "%llu: [ERROR] Failed to send request: type=0x%02X sent=%d/%u error=%d\n", - (unsigned long long)get_timestamp_ms(), type, sent, total_size, sent == SOCKET_ERROR ? WSAGetLastError() : 0); - fflush(client->log_file); - } - /* После частичной отправки следующий пакет уже нельзя писать в этот поток. */ - etcpmon_client_disconnect(client); - if (client->on_disconnected) client->on_disconnected(client->user_data); - return -1; - } - - /* В очередь попадает только полностью отправленный запрос. */ - if (client->pending_tail) client->pending_tail->next = req; - else client->pending_head = req; - client->pending_tail = req; - - return 0; + if (!client->next_seq_id) client->next_seq_id = 1; + req->callback = callback; req->arg = arg; req->type = type; + return send_message(client, type, payload, payload_size, req); } static void handle_response(struct etcpmon_client* client, uint8_t msg_type, @@ -733,19 +686,78 @@ static void remove_pending_request(struct etcpmon_client* client, struct etcpmon } /* Handle response - find matching request and call callback */ +/* Полный payload проверяется до снятия pending и любых пользовательских callbacks. */ +static int valid_response(uint8_t type, const uint8_t* payload, uint16_t size) { + switch (type) { + case ETCPMON_RSP_CONN_LIST: + return size >= 1 && payload[0] <= ETCPMON_MAX_CONNECTIONS && + size == 1 + payload[0] * sizeof(struct etcpmon_conn_info); + case ETCPMON_RSP_SOCKET_LIST: + return size >= 1 && payload[0] <= ETCPMON_MAX_CONNECTIONS && + size == 1 + payload[0] * sizeof(struct etcpmon_socket_info); + case ETCPMON_RSP_METRICS: { + if (size < sizeof(struct etcpmon_rsp_metrics)) return 0; + const struct etcpmon_rsp_metrics* metrics = (const void*)payload; + return metrics->etcp.links_count <= ETCPMON_MAX_LINKS && + size == sizeof(*metrics) + metrics->etcp.links_count * sizeof(struct etcpmon_link_metrics); + } + case ETCPMON_RSP_ERROR: + case ETCPMON_RSP_ACTION_RESULT: + return size >= 2 && memchr(payload + 1, 0, size - 1) != NULL; + case ETCPMON_RSP_DEBUG_CONFIG: { + const uint8_t* end = size > 2 ? memchr(payload + 2, 0, size - 2) : NULL; + if (!end || payload[0] > 5 || payload[1] > ETCPMON_MAX_DEBUG_CATEGORIES || + (size_t)(end - payload) + 1 + payload[1] != size) return 0; + unsigned names = end == payload + 2 ? 0 : 1; + for (const uint8_t* p = payload + 2; p < end; p++) if (*p == ',') names++; + return names == payload[1]; + } + case ETCPMON_RSP_NODE_INFO: return size >= 2 && payload[0] == 1 && payload[1] == 4; + case ETCPMON_RSP_NODE_INFO_END: return size == 0; + case ETCPMON_RSP_NODE_REMOVED: return size == sizeof(uint64_t); + default: return 0; + } +} + +/* Тип успешного ответа; RSP_ERROR завершает любой запрос. */ +static uint8_t expected_response(uint8_t type) { + switch (type) { + case ETCPMON_CMD_LIST_CONN: return ETCPMON_RSP_CONN_LIST; + case ETCPMON_CMD_GET_METRICS: return ETCPMON_RSP_METRICS; + case ETCPMON_CMD_LIST_SOCKETS: return ETCPMON_RSP_SOCKET_LIST; + case ETCPMON_CMD_ACTION: return ETCPMON_RSP_ACTION_RESULT; + case ETCPMON_CMD_GET_DEBUG_CONFIG: return ETCPMON_RSP_DEBUG_CONFIG; + case ETCPMON_CMD_SUBSCRIBE_NODES: return ETCPMON_RSP_NODE_INFO_END; + default: return 0; + } +} + static void handle_response_with_callback(struct etcpmon_client* client, uint8_t msg_type, const uint8_t* payload, uint16_t payload_size) { uint8_t seq_id = client->last_response_seq_id; + if (!valid_response(msg_type, payload, payload_size)) { + if (client->log_file) fprintf(client->log_file, "[ERROR] Invalid response: type=0x%02X seq=%u size=%u\n", + msg_type, seq_id, payload_size); + client_fail(client, "Invalid response payload", 0); + return; + } /* Find pending request */ struct etcpmon_pending_request* req = find_pending_request(client, seq_id); if (req) { + if (!req->sent || (msg_type != ETCPMON_RSP_ERROR && msg_type != expected_response(req->type))) { + if (client->log_file) fprintf(client->log_file, "[ERROR] Response mismatch: command=0x%02X response=0x%02X seq=%u sent=%u\n", + req->type, msg_type, seq_id, req->sent); + client_fail(client, "Response does not match request", 0); + return; + } /* Callback может отправить новый запрос или отключить клиент. */ void (*callback)(void*, uint8_t, const uint8_t*, uint16_t) = req->callback; void* arg = req->arg; remove_pending_request(client, req); if (callback) callback(arg, msg_type, payload, payload_size); + if (!client->connected) return; if (client->log_file) { fprintf(client->log_file, "%llu: [LOG] Response handled, seq=%d removed from queue\n", @@ -753,6 +765,13 @@ static void handle_response_with_callback(struct etcpmon_client* client, uint8_t fflush(client->log_file); } } else { + if (seq_id) { + if (client->log_file) { + fprintf(client->log_file, "[WARN] Ignoring response without pending: type=0x%02X seq=%u\n", msg_type, seq_id); + fflush(client->log_file); + } + return; + } /* No pending request - maybe broadcast/unsolicited or stale */ if (client->log_file) { fprintf(client->log_file, "%llu: [LOG] Response seq=%d - no pending request (broadcast?)\n", @@ -781,178 +800,79 @@ static void metrics_auto_callback(void* arg, uint8_t msg_type, const uint8_t* pa int etcpmon_client_process(struct etcpmon_client* client) { if (!client || client->sock == INVALID_SOCKET) return -1; - - /* Проверка завершения non-blocking connect */ - if (client->connected && !client->fully_connected) { + if (!client->fully_connected) { + if (GetTickCount() - client->connect_started_at >= REQUEST_TIMEOUT_MS) + return client_fail(client, "Connect timeout", WSAETIMEDOUT); fd_set writable, failed; - FD_ZERO(&writable); - FD_ZERO(&failed); - FD_SET(client->sock, &writable); - FD_SET(client->sock, &failed); + FD_ZERO(&writable); FD_ZERO(&failed); + FD_SET(client->sock, &writable); FD_SET(client->sock, &failed); struct timeval timeout = {0, 0}; int ready = select(0, NULL, &writable, &failed, &timeout); - if (ready == 0) return 0; - int error = 0; - int len = sizeof(error); - if (ready > 0 && getsockopt(client->sock, SOL_SOCKET, SO_ERROR, (char*)&error, &len) == 0) { - if (error == 0) { - client->fully_connected = 1; - if (client->log_file) { - fprintf(client->log_file, "%llu: [LOG] Connection established\n", - (unsigned long long)get_timestamp_ms()); - fflush(client->log_file); - } - if (client->on_connected) { - client->on_connected(client->user_data); - } - return 0; - } else if (error == WSAEINPROGRESS || error == WSAEALREADY) { - return 0; - } else { - if (client->log_file) { - fprintf(client->log_file, "%llu: [ERROR] Connect failed: %d\n", - (unsigned long long)get_timestamp_ms(), error); - fflush(client->log_file); - } - etcpmon_client_disconnect(client); - if (client->on_disconnected) client->on_disconnected(client->user_data); - if (client->on_error) { - char msg[256]; - snprintf(msg, sizeof(msg), "Connect failed: %d", error); - client->on_error(msg, client->user_data); - } - return -1; - } - } + if (!ready) return 0; + if (ready == SOCKET_ERROR) return client_fail(client, "Connect select failed", WSAGetLastError()); + int error = 0, len = sizeof(error); + if (getsockopt(client->sock, SOL_SOCKET, SO_ERROR, (char*)&error, &len) == SOCKET_ERROR) + return client_fail(client, "Connect SO_ERROR failed", WSAGetLastError()); + if (error == WSAEINPROGRESS || error == WSAEALREADY) return 0; + if (error) return client_fail(client, "Connect failed", error); + client->fully_connected = 1; if (client->log_file) { - fprintf(client->log_file, "%llu: [ERROR] Failed to check connect completion: %d\n", - (unsigned long long)get_timestamp_ms(), WSAGetLastError()); + fprintf(client->log_file, "%llu: [INFO] Connection established to %s:%u\n", + (unsigned long long)get_timestamp_ms(), client->server_addr, client->server_port); fflush(client->log_file); } - etcpmon_client_disconnect(client); - if (client->on_disconnected) client->on_disconnected(client->user_data); - return -1; + if (client->on_connected) client->on_connected(client->user_data); + return client->connected ? 0 : -1; + } + if (flush_outgoing(client) < 0) return -1; + int space = ETCPMON_MAX_MSG_SIZE - client->recv_len; + if (space <= 0) return client_fail(client, "Receive buffer full without a complete frame", 0); + uint8_t* input = client->recv_buffer + client->recv_len; + int received = recv(client->sock, (char*)input, space, 0); + if (received == SOCKET_ERROR && WSAGetLastError() != WSAEWOULDBLOCK) + return client_fail(client, "TCP recv failed", WSAGetLastError()); + if (!received) return client_fail(client, "Connection closed by server", 0); + if (received > 0) { + if (client->log_file) log_hex_data(client->log_file, "RX", input, received); + client->recv_len += received; + } + int processed = 0; + while (client->recv_len >= sizeof(struct etcpmon_msg_header)) { + struct etcpmon_msg_header* header = (void*)client->recv_buffer; + if (etcpmon_validate_header(header) < 0) return client_fail(client, "Invalid TCP frame header", 0); + uint16_t size = header->size; + if (client->recv_len < size) break; + client->last_response_seq_id = header->seq_id; + if (client->log_file) { + fprintf(client->log_file, "[DEBUG] RX frame: type=0x%02X seq=%u size=%u\n", header->type, header->seq_id, size); + fflush(client->log_file); + } + handle_response_with_callback(client, header->type, client->recv_buffer + sizeof(*header), size - sizeof(*header)); + if (!client->connected) return -1; + client->recv_len -= size; + memmove(client->recv_buffer, client->recv_buffer + size, client->recv_len); + processed++; } - - /* Один запрос метрик в полёте; зависший сервер не раздувает очередь. */ - int metrics_pending = 0; DWORD tick = GetTickCount(); + if (client->tx_head && tick - client->tx_progress_at >= REQUEST_TIMEOUT_MS) + return client_fail(client, "TCP TX progress timeout", WSAETIMEDOUT); + int metrics_pending = 0; for (struct etcpmon_pending_request* req = client->pending_head; req; req = req->next) { - if (tick - req->sent_at >= REQUEST_TIMEOUT_MS) { - if (client->log_file) { - fprintf(client->log_file, "%llu: [ERROR] Response timeout: type=0x%02X seq=%u elapsed=%lu ms\n", - (unsigned long long)get_timestamp_ms(), req->type, req->seq_id, (unsigned long)(tick - req->sent_at)); - fflush(client->log_file); - } - etcpmon_client_disconnect(client); - if (client->on_disconnected) client->on_disconnected(client->user_data); - return -1; + if (req->sent && tick - req->sent_at >= REQUEST_TIMEOUT_MS) { + if (client->log_file) fprintf(client->log_file, "[ERROR] Response timeout: type=0x%02X seq=%u elapsed=%lu ms\n", + req->type, req->seq_id, (unsigned long)(tick - req->sent_at)); + return client_fail(client, "Response timeout", WSAETIMEDOUT); } if (req->type == ETCPMON_CMD_GET_METRICS) metrics_pending = 1; } - - /* Авто-запрос метрик каждые 100 мс (асинхронно) */ - if (client->fully_connected && client->selected_peer_id != 0 && !metrics_pending) { + if (client->selected_peer_id && !metrics_pending) { uint64_t now = GetTickCount64(); if (now - client->last_metrics_request_ms >= ETCPMON_UPDATE_INTERVAL_MS) { - if (etcpmon_client_send_request(client, ETCPMON_CMD_GET_METRICS, NULL, 0, metrics_auto_callback, NULL) != 0) { - etcpmon_client_disconnect(client); - if (client->on_disconnected) client->on_disconnected(client->user_data); - return -1; - } + if (etcpmon_client_send_request(client, ETCPMON_CMD_GET_METRICS, NULL, 0, metrics_auto_callback, NULL) < 0) + return client->connected ? client_fail(client, "Failed to queue metrics request", 0) : -1; client->last_metrics_request_ms = now; } } - - /* Приём данных */ - uint8_t* buf = client->recv_buffer + client->recv_len; - int buf_space = ETCPMON_MAX_MSG_SIZE - client->recv_len; - if (buf_space <= 0) { - client->recv_len = 0; /* защита от переполнения */ - buf = client->recv_buffer; - buf_space = ETCPMON_MAX_MSG_SIZE; - } - - int received = recv(client->sock, (char*)buf, buf_space, 0); - - if (received == SOCKET_ERROR) { - int err = WSAGetLastError(); - if (err == WSAEWOULDBLOCK) return 0; - if (client->log_file) { - fprintf(client->log_file, "%llu: [ERROR] recv failed: %d\n", - (unsigned long long)get_timestamp_ms(), err); - fflush(client->log_file); - } - etcpmon_client_disconnect(client); - if (client->on_disconnected) client->on_disconnected(client->user_data); - return -1; - } - - if (received == 0) { - if (client->log_file) { - fprintf(client->log_file, "%llu: [LOG] Connection closed by server\n", - (unsigned long long)get_timestamp_ms()); - fflush(client->log_file); - } - etcpmon_client_disconnect(client); - if (client->on_disconnected) client->on_disconnected(client->user_data); - return -1; - } - - if (client->log_file) { - fprintf(client->log_file, "%llu: [LOG] Received %d bytes (offset=%d)\n", - (unsigned long long)get_timestamp_ms(), received, client->recv_len); - log_hex_data(client->log_file, "RX", buf, received); - } - - client->recv_len += received; - - /* === ОСНОВНОЙ ПАРСЕР (ИСПРАВЛЕННЫЙ) === */ - int processed = 0; - while (client->recv_len >= sizeof(struct etcpmon_msg_header)) { - struct etcpmon_msg_header* hdr = (struct etcpmon_msg_header*)client->recv_buffer; - - /* Валидация */ - if (etcpmon_validate_header(hdr) != 0) { - if (client->log_file) { - fprintf(client->log_file, "%llu: [ERROR] Invalid header, skipping byte %02X\n", - (unsigned long long)get_timestamp_ms(), client->recv_buffer[0]); - fflush(client->log_file); - } - memmove(client->recv_buffer, client->recv_buffer + 1, client->recv_len - 1); - client->recv_len--; - continue; - } - - /* hdr->size — это ПОЛНЫЙ размер сообщения (header + payload) */ - uint16_t total_size = hdr->size; - if (client->recv_len < total_size) { - break; /* ждём остаток */ - } - - uint16_t payload_size = total_size - sizeof(*hdr); - uint8_t* payload = client->recv_buffer + sizeof(*hdr); - - client->last_response_seq_id = hdr->seq_id; - - if (client->log_file) { - fprintf(client->log_file, "%llu: [LOG] Parsed message type=0x%02X total=%d payload=%d seq=%d\n", - (unsigned long long)get_timestamp_ms(), hdr->type, total_size, payload_size, hdr->seq_id); - fflush(client->log_file); - } - - handle_response_with_callback(client, hdr->type, payload, payload_size); - if (!client->connected) return -1; - processed++; - - /* Удаляем обработанное сообщение */ - if (client->recv_len > total_size) { - memmove(client->recv_buffer, client->recv_buffer + total_size, - client->recv_len - total_size); - } - client->recv_len -= total_size; - } - return processed; } @@ -1007,39 +927,16 @@ int etcpmon_client_request_debug_config(struct etcpmon_client* client) { int etcpmon_client_send_debug_config(struct etcpmon_client* client, uint8_t global_level, uint8_t count, const uint8_t* levels) { - if (!client || client->sock == INVALID_SOCKET || !levels || count == 0 || count > ETCPMON_MAX_DEBUG_CATEGORIES) return -1; - - uint8_t payload[2 + ETCPMON_MAX_DEBUG_CATEGORIES]; - payload[0] = global_level; - payload[1] = count; - memcpy(payload + 2, levels, count); - uint16_t payload_size = 2 + count; - - uint8_t buffer[sizeof(struct etcpmon_msg_header) + sizeof(payload)]; - struct etcpmon_msg_header* hdr = (struct etcpmon_msg_header*)buffer; - etcpmon_build_header(hdr, payload_size, ETCPMON_CMD_SET_DEBUG_CONFIG, 0); - memcpy(buffer + sizeof(*hdr), payload, payload_size); - - if (client->log_file) { - fprintf(client->log_file, "%llu: [LOG] CMD_SET_DEBUG_CONFIG global=%d count=%d\n", - (unsigned long long)get_timestamp_ms(), global_level, count); - fflush(client->log_file); - } - - int total_size = sizeof(*hdr) + payload_size; - int sent = send(client->sock, (const char*)buffer, total_size, 0); - if (sent != total_size) { - if (client->log_file) { - fprintf(client->log_file, "[ERROR] Failed to send debug config: sent=%d/%d error=%d\n", - sent, total_size, sent == SOCKET_ERROR ? WSAGetLastError() : 0); - fflush(client->log_file); - } - etcpmon_client_disconnect(client); - if (client->on_disconnected) client->on_disconnected(client->user_data); + if (!client || !levels || !count || count > ETCPMON_MAX_DEBUG_CATEGORIES || global_level > 5) { + fprintf(stderr, "[ERROR] Invalid debug config: global=%u count=%u\n", global_level, count); return -1; } + uint8_t payload[2 + ETCPMON_MAX_DEBUG_CATEGORIES]; + payload[0] = global_level; payload[1] = count; + memcpy(payload + 2, levels, count); + if (send_message(client, ETCPMON_CMD_SET_DEBUG_CONFIG, payload, 2 + count, NULL) < 0) return -1; client->debug_global_level = global_level; - for (uint8_t i = 0; i < count; i++) client->debug_category_levels[i] = levels[i]; + memcpy(client->debug_category_levels, levels, count); return 0; } diff --git a/tools/etcpmon/etcpmon_client.h b/tools/etcpmon/etcpmon_client.h index 6b0df368e..d6bc24f7b 100644 --- a/tools/etcpmon/etcpmon_client.h +++ b/tools/etcpmon/etcpmon_client.h @@ -1,7 +1,10 @@ /* - * etcpmon_client.h - ETCP Monitor Client Network Layer - * - * Handles TCP connection to utun control server + * TCP-клиент control_server: init → connect → регулярный process → cleanup. + * Все вызовы и callbacks выполняются в одном потоке; Winsock запускает/останавливает вызывающий код. + * connect принимает числовой IPv4. connected включает CONNECTING, fully_connected означает готовый TCP. + * process завершает connect, принимает кадры и сам запрашивает метрики выбранного пира раз в 100 мс. + * Списки и debug config запрашивает вызывающая сторона. Отдельного потока и очереди TX здесь нет: + * send выполняется сразу; pending хранит только ожидания ответов (до 255, таймаут 5 секунд). */ #ifndef ETCPMON_CLIENT_H @@ -18,6 +21,10 @@ extern "C" { /* Graph history size (samples) */ #define GRAPH_HISTORY_SIZE 860 +#define ETCPMON_TX_MAX_BYTES (2u * ETCPMON_MAX_MSG_SIZE) +#define ETCPMON_TX_MAX_FRAMES 256 + +struct etcpmon_tx_frame; /* Graph metrics types */ typedef enum { @@ -32,35 +39,42 @@ typedef enum { GRAPH_METRIC_COUNT } graph_metric_type_t; -/* Metrics history buffer */ +/* Кольцевая история первого линка: объёмы в KiB, скорости в kbit/s, minRTT в мс, mode безразмерный. */ struct metrics_history { float values[GRAPH_METRIC_COUNT][GRAPH_HISTORY_SIZE]; float min_val[GRAPH_METRIC_COUNT]; float max_val[GRAPH_METRIC_COUNT]; DWORD last_minmax_update; - int head; - int count; + int head; /* Индекс следующей записи, 0..GRAPH_HISTORY_SIZE-1 */ + int count; /* Число заполненных сэмплов, не более GRAPH_HISTORY_SIZE */ uint32_t last_retrans; uint32_t last_acks; uint64_t last_bytes_sent; }; -/* Client state */ +/* Клиент владеет сокетом, логом, кешами и pending. Не копировать активную структуру. */ struct etcpmon_client { - SOCKET sock; - int connected; - int fully_connected; + SOCKET sock; /* INVALID_SOCKET после init/disconnect */ + int connected; /* 1 с начала неблокирующего connect */ + int fully_connected; /* 1 после успешной проверки select/SO_ERROR в process */ + DWORD connect_started_at; /* Монотонные мс начала connect; срок 5 секунд */ + struct etcpmon_tx_frame* tx_head; /* FIFO исходящих кадров; client владеет памятью */ + struct etcpmon_tx_frame* tx_tail; + size_t tx_bytes; /* Сумма полных размеров кадров, включая частично отправленную голову */ + unsigned tx_count; /* Не более ETCPMON_TX_MAX_FRAMES */ + DWORD tx_progress_at; /* Начало ожидания или последний положительный send */ + int tx_blocked; /* 1 после WSAEWOULDBLOCK, для логирования переходов */ /* Receive buffer */ uint8_t recv_buffer[ETCPMON_MAX_MSG_SIZE]; - uint16_t recv_len; + uint16_t recv_len; /* Неразобранный хвост TCP-потока; границы recv не являются границами кадров */ /* Server address */ char server_addr[64]; uint16_t server_port; /* Selected connection */ - uint64_t selected_peer_id; + uint64_t selected_peer_id; /* 0 запрещает автоматический запрос метрик */ /* Connection info cache */ struct etcpmon_conn_info* conn_list; @@ -81,7 +95,7 @@ struct etcpmon_client { /* Metrics history for graph */ struct metrics_history history; - /* Pending requests queue with callbacks */ + /* Отправленные запросы с ненулевым seq_id; disconnect отменяет их без вызова callbacks. */ struct etcpmon_pending_request* pending_head; struct etcpmon_pending_request* pending_tail; uint8_t next_seq_id; @@ -96,7 +110,7 @@ struct etcpmon_client { char debug_category_names[512]; uint8_t debug_config_dirty; - /* Callbacks */ + /* Callbacks вызываются синхронно в потоке вызывающего API; аргументы заимствованы, владение не передаётся. */ void (*on_connected)(void* user_data); void (*on_disconnected)(void* user_data); void (*on_conn_list)(struct etcpmon_conn_info* list, uint8_t count, void* user_data); @@ -108,7 +122,7 @@ struct etcpmon_client { void (*on_action_result)(const char* text, void* user_data); void* user_data; - uint64_t last_metrics_request_ms; + uint64_t last_metrics_request_ms; /* GetTickCount64, монотонные мс; 0 после выбора/отключения */ }; /* Pending request with callback */ @@ -116,66 +130,69 @@ struct etcpmon_pending_request { uint8_t seq_id; /* Sequence ID */ uint8_t type; /* Тип команды для ограничения опроса */ DWORD sent_at; /* Монотонное время отправки */ + uint8_t sent; /* 1 после полной отправки; только тогда идёт таймаут ответа */ void (*callback)(void* arg, uint8_t msg_type, const uint8_t* payload, uint16_t payload_size); void* arg; /* User argument for callback */ struct etcpmon_pending_request* next; }; -/* Initialize client structure */ +/* Обнулить неактивную caller-owned структуру client; сокет/лог не открывает. */ void etcpmon_client_init(struct etcpmon_client* client); -/* Cleanup client resources */ +/* Отключить client и освободить его кеши, pending и лог; сам client и Winsock не освобождает. */ void etcpmon_client_cleanup(struct etcpmon_client* client); -/* Connect to server - * Returns 0 on success, -1 on error - */ +/* Начать connect к числовому IPv4 addr:port. 0 = попытка начата, -1 = немедленная ошибка. + * Завершение проверяет process; on_connected означает успех. Собственного таймаута connect нет. + * Уже открытое соединение закрывается. Лог etcpmon.log создаётся при первой попытке в текущем каталоге. */ int etcpmon_client_connect(struct etcpmon_client* client, const char* addr, uint16_t port); -/* Disconnect from server */ +/* Закрыть сокет client, сбросить выбор/историю/кеши и отменить pending без callbacks. + * Лог и кеш debug config сохраняются. Сам disconnect не вызывает on_disconnected. */ void etcpmon_client_disconnect(struct etcpmon_client* client); -/* Check if connected */ +/* Вернуть connected (также 1 во время CONNECTING), для NULL — 0. */ int etcpmon_client_is_connected(const struct etcpmon_client* client); -/* Request connection list - * Returns 0 on success, -1 on error - */ +/* Отправить LIST_CONN с seq=0 без pending/таймаута. 0 = кадр отправлен целиком, -1 = ошибка. + * Для request_list/request_socket_list/request_metrics/select_connection ошибка send + * не закрывает сокет: вызывающая сторона должна отключить client перед следующей отправкой. */ int etcpmon_client_request_list(struct etcpmon_client* client); -/* Request local socket list - * Returns 0 on success, -1 on error - */ +/* Отправить LIST_SOCKETS с seq=0 без pending/таймаута; возврат и ошибки как у request_list. */ int etcpmon_client_request_socket_list(struct etcpmon_client* client); -/* Select connection to monitor - * Returns 0 on success, -1 on error - */ +/* Отправить SELECT_CONN с seq=0, без подтверждения; peer_node_id=0 снимает выбор. + * После полной отправки меняет локальный selected_peer_id и очищает историю. + * 0 = отправлено, -1 = ошибка; существование пира проверит сервер при запросе метрик. */ int etcpmon_client_select_connection(struct etcpmon_client* client, uint64_t peer_node_id); -/* Request metrics for selected connection - * Returns 0 on success, -1 on error - */ +/* Разовый GET_METRICS с seq=0, без pending/таймаута. Обычно достаточно автоопроса в process. + * 0 = отправлено, -1 = ошибка; этот вызов не участвует в ограничении автоопроса. */ int etcpmon_client_request_metrics(struct etcpmon_client* client); -/* Send request with callback - * callback will be called when response is received - * arg will be passed to callback - * Returns 0 on success (queued), -1 on error - */ +/* Немедленно отправить type/payload с уникальным seq=1..255; payload копируется до возврата. + * 0 = весь кадр передан send и ожидание ответа добавлено в pending; -1 = ошибка. + * При short send/WSAEWOULDBLOCK закрывает соединение и вызывает on_disconnected; TX не откладывается. + * Обязательный callback(arg, type, payload, size) вызывается по совпадению seq до общих callbacks; + * тип и payload до него не проверяются. Payload действителен только во время callback. + * callback может отправлять запросы или отключить клиент; рекурсивно вызывать process нельзя. + * Таймаут 5 с проверяется в process; при отмене/таймауте callback не вызывается. */ int etcpmon_client_send_request(struct etcpmon_client* client, uint8_t type, const void* payload, uint16_t payload_size, void (*callback)(void* arg, uint8_t msg_type, const uint8_t* payload, uint16_t payload_size), void* arg); -/* Process incoming data (non-blocking) - * Should be called periodically (e.g., from window message loop) - * Returns 1 if data was processed, 0 if no data, -1 on error - */ +/* Неблокирующий шаг в потоке владельца: connect, таймауты, автоопрос, одно recv и разбор полных кадров. + * Возвращает число разобранных кадров (может быть >1), 0 = полных кадров нет, -1 = ошибка/отключение. + * На автоопрос метрик одновременно допускается один pending. Вызывать регулярно, чаще 100 мс. + * EOF/ошибка recv/таймаут pending закрывают соединение и вызывают on_disconnected. */ int etcpmon_client_process(struct etcpmon_client* client); -/* Set callbacks */ +/* Установить необязательные общие callbacks и caller-owned user_data. + * Списки/метрики принадлежат client до следующей замены кеша/отключения; + * строки ошибок/Action действительны только во время callback. */ void etcpmon_client_set_callbacks(struct etcpmon_client* client, void (*on_connected)(void*), void (*on_disconnected)(void*), @@ -188,22 +205,30 @@ void etcpmon_client_set_callbacks(struct etcpmon_client* client, void (*on_action_result)(const char*, void*), void* user_data); -/* Get connection name by peer_id */ +/* Заимствованное имя peer_id в кеше client до его замены/отключения; NULL, если не найдено. */ const char* etcpmon_client_get_conn_name(struct etcpmon_client* client, uint64_t peer_id); -/* Get metrics history */ +/* Заимствованная встроенная история client; NULL для NULL client, отдельно не освобождать. */ struct metrics_history* etcpmon_client_get_history(struct etcpmon_client* client); -/* Add metrics to history */ +/* Добавить сэмпл из client->last_links[0]; metrics должен быть ненулевым. + * Вызывается парсером метрик ровно один раз на принятый снимок; GUI повторно не добавляет. */ void etcpmon_client_add_to_history(struct etcpmon_client* client, struct etcpmon_rsp_metrics* metrics); -/* Send action command to server */ +/* Отправить action (до 31 байта + NUL) как запрос с pending; возврат как у send_request. + * Результат передаётся on_action_result. Сейчас сервер отвечает только на "nodes"; + * "nat"/неизвестная команда оставляют pending, что через 5 с отключит клиент. */ int etcpmon_client_send_action(struct etcpmon_client* client, const char* action); +/* Запрос debug config с pending; 0 = отправлено, -1 = ошибка. Кеш обновляет process. */ int etcpmon_client_request_debug_config(struct etcpmon_client* client); +/* Отправить SET_DEBUG_CONFIG с seq=0 без подтверждения: count=1..64, levels[count]. + * 0 = отправлено и локальные уровни обновлены; -1 = ошибка (send также отключает клиент). + * Категории следуют порядку имён из ответа GET_DEBUG_CONFIG; GUI поддерживает уровни 0..5. */ int etcpmon_client_send_debug_config(struct etcpmon_client* client, uint8_t global_level, uint8_t count, const uint8_t* levels); -uint64_t get_timestamp_ms(void); // Add this declaration +/* Wall-clock мс Unix epoch для лога; для таймаутов не используется. */ +uint64_t get_timestamp_ms(void); #ifdef __cplusplus } diff --git a/tools/etcpmon/etcpmon_gui.c b/tools/etcpmon/etcpmon_gui.c index fec845c9e..ad60d3b15 100644 --- a/tools/etcpmon/etcpmon_gui.c +++ b/tools/etcpmon/etcpmon_gui.c @@ -1168,8 +1168,8 @@ static void OnAction(struct etcpmon_app* app) { action[sizeof(action) - 1] = '\0'; if (strlen(action) > 0) { - etcpmon_client_send_action(app->client, action); - SetWindowTextA(app->hEditAction, ""); + if (etcpmon_client_send_action(app->client, action) == 0) SetWindowTextA(app->hEditAction, ""); + else etcpmon_gui_set_status(app, "Failed to queue Action"); } }