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

Queue monitor TCP commands and validate matched responses

Browse Source
This commit is contained in:
evgeny committed 2026-10-27 13:38:40 +01:00
1 parent 2d12dfb975
commit ce06fe242b
3 files changed
+344 -422
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tools/etcpmon/etcpmon_client.c
+268 -371
View File
@@ -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;
}
tools/etcpmon/etcpmon_client.h
+74 -49
View File
@@ -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
}
tools/etcpmon/etcpmon_gui.c
+2 -2
View File
@@ -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");
}
}