From c0c366bd83972a728556868d021dff58047b97a4 Mon Sep 17 00:00:00 2001 From: jeka Date: Tue, 14 Apr 2026 23:40:14 +0300 Subject: [PATCH] mon fix --- tools/etcpmon/etcpmon_client.c | 145 ++++++++++++--------------------- 1 file changed, 52 insertions(+), 93 deletions(-) diff --git a/tools/etcpmon/etcpmon_client.c b/tools/etcpmon/etcpmon_client.c index 60c1ff5f..d49a35fa 100644 --- a/tools/etcpmon/etcpmon_client.c +++ b/tools/etcpmon/etcpmon_client.c @@ -642,43 +642,32 @@ 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 + /* Проверка завершения non-blocking connect */ if (client->connected && !client->fully_connected) { int error = 0; int len = sizeof(error); if (getsockopt(client->sock, SOL_SOCKET, SO_ERROR, (char*)&error, &len) == 0) { if (error == 0) { - // Connect успешно завершён client->fully_connected = 1; if (client->log_file) { - fprintf(client->log_file, "%llu: [LOG] Connection established\n", + 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); // Здесь вызов request_list + client->on_connected(client->user_data); } - return 0;// выходим, не лезем в recv/send в этом же тике - // Продолжаем к recv, т.к. подключены + return 0; } else if (error == WSAEINPROGRESS || error == WSAEALREADY) { - // Подключение ещё в процессе, ничего не делаем, ждём следующий вызов process - if (client->log_file) { - fprintf(client->log_file, "%llu: [LOG] Connect in progress (polling)...\n", - (unsigned long long)get_timestamp_ms()); - fflush(client->log_file); - } - return 0; // Важно: выходим, чтобы не идти в recv (оно провалится с ошибкой) + return 0; } else { - // Подключение провалилось по другой причине if (client->log_file) { - fprintf(client->log_file, "%llu: [ERROR] Connect failed: %d\n", + 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_disconnected) client->on_disconnected(client->user_data); if (client->on_error) { char msg[256]; snprintf(msg, sizeof(msg), "Connect failed: %d", error); @@ -686,139 +675,109 @@ int etcpmon_client_process(struct etcpmon_client* client) { } return -1; } - } else { - // getsockopt провалился (редко, но на всякий случай) - if (client->log_file) { - fprintf(client->log_file, "%llu: [ERROR] getsockopt failed: %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; } + etcpmon_client_disconnect(client); + return -1; } - /* === Авто-перезапрос метрик каждые 100 мс (новый асинхронный способ) === */ - if (client->fully_connected && client->selected_peer_id != 0 && client->sock != INVALID_SOCKET) { + /* Авто-запрос метрик каждые 100 мс (асинхронно) */ + if (client->fully_connected && client->selected_peer_id != 0) { uint64_t now = get_timestamp_ms(); 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); - + etcpmon_client_send_request(client, ETCPMON_CMD_GET_METRICS, NULL, 0, + metrics_auto_callback, NULL); client->last_metrics_request_ms = now; - - if (client->log_file) { - fprintf(client->log_file, "%llu: [LOG] Auto-request METRICS (async, seq=%d)\n", - (unsigned long long)now, client->next_seq_id - 1); - fflush(client->log_file); - } } } - // Если подключение в процессе, мы уже вышли выше. Здесь recv только для установленного соединения + /* Приём данных */ 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; + 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; /* No data available */ - } - /* Connection error */ + if (err == WSAEWOULDBLOCK) return 0; if (client->log_file) { - fprintf(client->log_file, "%llu: [ERROR] recv failed: %d\n", + 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); - } + if (client->on_disconnected) client->on_disconnected(client->user_data); return -1; } - + if (received == 0) { - /* Connection closed by server */ if (client->log_file) { - fprintf(client->log_file, "%llu: [LOG] Connection closed by server (recv returned 0)\n", + 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); - } + if (client->on_disconnected) client->on_disconnected(client->user_data); return -1; } - - /* Log received data */ + 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; - - /* Process complete messages */ + + /* === ОСНОВНОЙ ПАРСЕР (ИСПРАВЛЕННЫЙ) === */ 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) { - /* Invalid header, skip byte */ if (client->log_file) { - fprintf(client->log_file, "%llu: [ERROR] Invalid header, skipping byte\n", - (unsigned long long)get_timestamp_ms()); + 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; } - - if (client->recv_len < hdr->size) { - break; /* Wait for more data */ + + /* hdr->size — это ПОЛНЫЙ размер сообщения (header + payload) */ + uint16_t total_size = hdr->size; + if (client->recv_len < total_size) { + break; /* ждём остаток */ } - - /* Handle message */ + + uint16_t payload_size = total_size - sizeof(*hdr); uint8_t* payload = client->recv_buffer + sizeof(*hdr); - uint16_t payload_size = hdr->size - sizeof(*hdr); - - /* Store seq_id for matching with pending requests */ + client->last_response_seq_id = hdr->seq_id; - + if (client->log_file) { - fprintf(client->log_file, "%llu: [LOG] Received message type=0x%02X size=%d seq=%d\n", - (unsigned long long)get_timestamp_ms(), hdr->type, hdr->size, hdr->seq_id); + 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); } - - /* Use callback-based handler */ + handle_response_with_callback(client, hdr->type, payload, payload_size); processed++; - - /* Remove processed message */ - if (client->recv_len > hdr->size) { - memmove(client->recv_buffer, client->recv_buffer + hdr->size, - client->recv_len - hdr->size); + + /* Удаляем обработанное сообщение */ + if (client->recv_len > total_size) { + memmove(client->recv_buffer, client->recv_buffer + total_size, + client->recv_len - total_size); } - client->recv_len -= hdr->size; + client->recv_len -= total_size; } - + return processed; }