diff --git a/tools/etcpmon/etcpmon_client.c b/tools/etcpmon/etcpmon_client.c index 9531f9de..d3e5e7d6 100644 --- a/tools/etcpmon/etcpmon_client.c +++ b/tools/etcpmon/etcpmon_client.c @@ -5,6 +5,7 @@ #include "etcpmon_client.h" #include "etcpmon_graph.h" #include +#include #include #include #include @@ -13,6 +14,9 @@ /* Log file name */ #define LOG_FILENAME "etcpmon.log" +#define REQUEST_TIMEOUT_MS 5000 + +static struct etcpmon_pending_request* find_pending_request(struct etcpmon_client* client, uint8_t seq_id); /* Get current timestamp in milliseconds (Unix epoch) */ uint64_t get_timestamp_ms(void) { @@ -226,6 +230,19 @@ void etcpmon_client_disconnect(struct etcpmon_client* client) { client->fully_connected = 0; client->recv_len = 0; client->selected_peer_id = 0; + client->last_metrics_request_ms = 0; + free(client->conn_list); + client->conn_list = NULL; + client->conn_count = 0; + free(client->socket_list); + client->socket_list = NULL; + client->socket_count = 0; + free(client->last_metrics); + client->last_metrics = NULL; + free(client->last_links); + client->last_links = NULL; + client->last_links_count = 0; + memset(&client->history, 0, sizeof(client->history)); /* Очистить очередь pending requests и сбросить seq_id */ struct etcpmon_pending_request* req = client->pending_head; @@ -300,6 +317,8 @@ int etcpmon_client_select_connection(struct etcpmon_client* client, uint64_t pee } client->selected_peer_id = peer_node_id; + client->last_metrics_request_ms = 0; + memset(&client->history, 0, sizeof(client->history)); return 0; } @@ -379,6 +398,28 @@ int etcpmon_client_send_request(struct etcpmon_client* client, 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); + } + 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); + } + return -1; + } /* Allocate pending request */ struct etcpmon_pending_request* req = (struct etcpmon_pending_request*)malloc(sizeof(*req)); @@ -396,22 +437,12 @@ int etcpmon_client_send_request(struct etcpmon_client* client, 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; - /* Add to queue */ - if (client->pending_tail) { - client->pending_tail->next = req; - } else { - client->pending_head = req; - } - client->pending_tail = req; - /* Build message with seq_id */ - uint8_t buffer[sizeof(struct etcpmon_msg_header) + ETCPMON_MAX_MSG_SIZE]; - if (payload_size > ETCPMON_MAX_MSG_SIZE) { - free(req); - return -1; - } + 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); @@ -431,19 +462,19 @@ int etcpmon_client_send_request(struct etcpmon_client* client, int sent = send(client->sock, (const char*)buffer, total_size, 0); if (sent != total_size) { - /* Remove from queue on error */ - if (client->pending_head == req) { - client->pending_head = req->next; - if (!client->pending_head) client->pending_tail = NULL; - } free(req); if (client->log_file) { - fprintf(client->log_file, "%llu: [ERROR] Failed to send request: %d\n", - (unsigned long long)get_timestamp_ms(), WSAGetLastError()); + 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); } return -1; } + + /* В очередь попадает только полностью отправленный запрос. */ + if (client->pending_tail) client->pending_tail->next = req; + else client->pending_head = req; + client->pending_tail = req; return 0; } @@ -452,20 +483,30 @@ static void handle_response(struct etcpmon_client* client, uint8_t msg_type, const uint8_t* payload, uint16_t payload_size) { switch (msg_type) { case ETCPMON_RSP_CONN_LIST: { - if (payload_size < sizeof(struct etcpmon_rsp_conn_list)) break; + if (payload_size < sizeof(struct etcpmon_rsp_conn_list)) { + if (client->log_file) fprintf(client->log_file, "[ERROR] Empty RSP_CONN_LIST payload\n"); + break; + } struct etcpmon_rsp_conn_list* rsp = (struct etcpmon_rsp_conn_list*)payload; uint8_t count = rsp->count; - if (payload_size < sizeof(*rsp) + count * sizeof(struct etcpmon_conn_info)) break; + if (count > ETCPMON_MAX_CONNECTIONS || payload_size != sizeof(*rsp) + count * sizeof(struct etcpmon_conn_info)) { + if (client->log_file) fprintf(client->log_file, "[ERROR] Invalid RSP_CONN_LIST: count=%u size=%u\n", count, payload_size); + break; + } /* Update cached list */ - if (client->conn_list) { - free(client->conn_list); + struct etcpmon_conn_info* list = count ? malloc(count * sizeof(*list)) : NULL; + if (count && !list) { + if (client->log_file) fprintf(client->log_file, "[ERROR] Failed to allocate connection list: count=%u\n", count); + break; } - client->conn_list = (struct etcpmon_conn_info*)malloc(count * sizeof(struct etcpmon_conn_info)); - if (client->conn_list) { - memcpy(client->conn_list, payload + sizeof(*rsp), count * sizeof(struct etcpmon_conn_info)); + { + if (count) memcpy(list, payload + sizeof(*rsp), count * sizeof(*list)); + for (uint8_t i = 0; i < count; i++) list[i].name[ETCPMON_MAX_CONN_NAME - 1] = '\0'; + free(client->conn_list); + client->conn_list = list; client->conn_count = count; if (client->log_file) { @@ -493,15 +534,25 @@ static void handle_response(struct etcpmon_client* client, uint8_t msg_type, struct etcpmon_rsp_socket_list* rsp = (struct etcpmon_rsp_socket_list*)payload; uint8_t count = rsp->count; - if (count > 16) count = 16; /* Limit to prevent buffer overflow */ - if (payload_size < sizeof(*rsp) + count * sizeof(struct etcpmon_socket_info)) break; + if (count > ETCPMON_MAX_CONNECTIONS || payload_size != sizeof(*rsp) + count * sizeof(struct etcpmon_socket_info)) { + if (client->log_file) fprintf(client->log_file, "[ERROR] Invalid RSP_SOCKET_LIST: count=%u size=%u\n", count, payload_size); + break; + } - if (client->socket_list) { - free(client->socket_list); + struct etcpmon_socket_info* list = count ? malloc(count * sizeof(*list)) : NULL; + if (count && !list) { + if (client->log_file) fprintf(client->log_file, "[ERROR] Failed to allocate socket list: count=%u\n", count); + break; } - client->socket_list = (struct etcpmon_socket_info*)malloc(count * sizeof(struct etcpmon_socket_info)); - if (client->socket_list) { - memcpy(client->socket_list, payload + sizeof(*rsp), count * sizeof(struct etcpmon_socket_info)); + { + if (count) memcpy(list, payload + sizeof(*rsp), count * sizeof(*list)); + for (uint8_t i = 0; i < count; i++) { + list[i].name[sizeof(list[i].name) - 1] = '\0'; + list[i].ip[sizeof(list[i].ip) - 1] = '\0'; + list[i].nat_ip[sizeof(list[i].nat_ip) - 1] = '\0'; + } + free(client->socket_list); + client->socket_list = list; client->socket_count = count; if (client->log_file) { @@ -522,23 +573,36 @@ static void handle_response(struct etcpmon_client* client, uint8_t msg_type, struct etcpmon_rsp_metrics* rsp = (struct etcpmon_rsp_metrics*)payload; uint8_t links_count = rsp->etcp.links_count; + + if (rsp->etcp.peer_node_id != client->selected_peer_id || client->selected_peer_id == 0) { + if (client->log_file) { + fprintf(client->log_file, "[LOG] Ignoring stale metrics: peer=%016llX selected=%016llX\n", + (unsigned long long)rsp->etcp.peer_node_id, (unsigned long long)client->selected_peer_id); + } + break; + } - if (payload_size < sizeof(*rsp) + links_count * sizeof(struct etcpmon_link_metrics)) break; + if (links_count > ETCPMON_MAX_LINKS || payload_size != sizeof(*rsp) + links_count * sizeof(struct etcpmon_link_metrics)) { + if (client->log_file) fprintf(client->log_file, "[ERROR] Invalid RSP_METRICS: links=%u size=%u\n", links_count, payload_size); + break; + } /* Update cached metrics */ - if (client->last_metrics) { - free(client->last_metrics); + struct etcpmon_rsp_metrics* metrics = malloc(sizeof(*metrics)); + struct etcpmon_link_metrics* links = links_count ? malloc(links_count * sizeof(*links)) : NULL; + if (!metrics || (links_count && !links)) { + free(metrics); + free(links); + if (client->log_file) fprintf(client->log_file, "[ERROR] Failed to allocate metrics: links=%u\n", links_count); + break; } - if (client->last_links) { + { + free(client->last_metrics); free(client->last_links); - } - - client->last_metrics = (struct etcpmon_rsp_metrics*)malloc(sizeof(*rsp)); - client->last_links = (struct etcpmon_link_metrics*)malloc(links_count * sizeof(struct etcpmon_link_metrics)); - - if (client->last_metrics && client->last_links) { + client->last_metrics = metrics; + client->last_links = links; memcpy(client->last_metrics, rsp, sizeof(*rsp)); - memcpy(client->last_links, payload + sizeof(*rsp), + if (links_count) memcpy(client->last_links, payload + sizeof(*rsp), links_count * sizeof(struct etcpmon_link_metrics)); client->last_links_count = links_count; @@ -551,12 +615,12 @@ static void handle_response(struct etcpmon_client* client, uint8_t msg_type, fflush(client->log_file); } + etcpmon_client_add_to_history(client, client->last_metrics); if (client->on_metrics) { client->on_metrics(client->last_metrics, client->last_links, links_count, client->user_data); } - etcpmon_client_add_to_history(client, client->last_metrics); } break; } @@ -638,15 +702,17 @@ static struct etcpmon_pending_request* find_pending_request(struct etcpmon_clien /* Remove pending request from queue */ static void remove_pending_request(struct etcpmon_client* client, struct etcpmon_pending_request* req) { struct etcpmon_pending_request** curr = &client->pending_head; + struct etcpmon_pending_request* prev = NULL; while (*curr) { if (*curr == req) { *curr = req->next; if (client->pending_tail == req) { - client->pending_tail = *curr; + client->pending_tail = prev; } free(req); return; } + prev = *curr; curr = &(*curr)->next; } } @@ -660,12 +726,11 @@ static void handle_response_with_callback(struct etcpmon_client* client, uint8_t struct etcpmon_pending_request* req = find_pending_request(client, seq_id); if (req) { - /* Call the callback */ - if (req->callback) { - req->callback(req->arg, msg_type, payload, payload_size); - } - /* Remove from queue */ + /* 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->log_file) { fprintf(client->log_file, "%llu: [LOG] Response handled, seq=%d removed from queue\n", @@ -704,9 +769,17 @@ int etcpmon_client_process(struct etcpmon_client* client) { /* Проверка завершения non-blocking connect */ if (client->connected && !client->fully_connected) { + fd_set writable, 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 (getsockopt(client->sock, SOL_SOCKET, SO_ERROR, (char*)&error, &len) == 0) { + 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) { @@ -736,16 +809,42 @@ int etcpmon_client_process(struct etcpmon_client* client) { return -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()); + fflush(client->log_file); + } etcpmon_client_disconnect(client); + if (client->on_disconnected) client->on_disconnected(client->user_data); return -1; } + /* Один запрос метрик в полёте; зависший сервер не раздувает очередь. */ + int metrics_pending = 0; + DWORD tick = GetTickCount(); + 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->type == ETCPMON_CMD_GET_METRICS) metrics_pending = 1; + } + /* Авто-запрос метрик каждые 100 мс (асинхронно) */ - if (client->fully_connected && client->selected_peer_id != 0) { - uint64_t now = get_timestamp_ms(); + if (client->fully_connected && client->selected_peer_id != 0 && !metrics_pending) { + uint64_t now = GetTickCount64(); if (now - client->last_metrics_request_ms >= ETCPMON_UPDATE_INTERVAL_MS) { - etcpmon_client_send_request(client, ETCPMON_CMD_GET_METRICS, NULL, 0, - metrics_auto_callback, NULL); + 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; + } client->last_metrics_request_ms = now; } } @@ -835,6 +934,7 @@ int etcpmon_client_process(struct etcpmon_client* client) { memmove(client->recv_buffer, client->recv_buffer + total_size, client->recv_len - total_size); } + if (!client->connected) return -1; client->recv_len -= total_size; } diff --git a/tools/etcpmon/etcpmon_client.h b/tools/etcpmon/etcpmon_client.h index 36e83945..6b0df368 100644 --- a/tools/etcpmon/etcpmon_client.h +++ b/tools/etcpmon/etcpmon_client.h @@ -114,6 +114,8 @@ struct etcpmon_client { /* Pending request with callback */ struct etcpmon_pending_request { uint8_t seq_id; /* Sequence ID */ + uint8_t type; /* Тип команды для ограничения опроса */ + DWORD sent_at; /* Монотонное время отправки */ 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;