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

mon fix

congestion
jeka 6 months ago
parent
0a44320245
commit
c0c366bd83
1 changed files with 52 additions and 93 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 145
      tools/etcpmon/etcpmon_client.c

145
tools/etcpmon/etcpmon_client.c
Unescape View File

@ -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) { int etcpmon_client_process(struct etcpmon_client* client) {
if (!client || client->sock == INVALID_SOCKET) return -1; if (!client || client->sock == INVALID_SOCKET) return -1;
// Проверка завершения non-blocking connect /* Проверка завершения non-blocking connect */
if (client->connected && !client->fully_connected) { if (client->connected && !client->fully_connected) {
int error = 0; int error = 0;
int len = sizeof(error); int len = sizeof(error);
if (getsockopt(client->sock, SOL_SOCKET, SO_ERROR, (char*)&error, &len) == 0) { if (getsockopt(client->sock, SOL_SOCKET, SO_ERROR, (char*)&error, &len) == 0) {
if (error == 0) { if (error == 0) {
// Connect успешно завершён
client->fully_connected = 1; client->fully_connected = 1;
if (client->log_file) { 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()); (unsigned long long)get_timestamp_ms());
fflush(client->log_file); fflush(client->log_file);
} }
if (client->on_connected) { if (client->on_connected) {
client->on_connected(client->user_data); // Здесь вызов request_list client->on_connected(client->user_data);
} }
return 0;// выходим, не лезем в recv/send в этом же тике return 0;
// Продолжаем к recv, т.к. подключены
} else if (error == WSAEINPROGRESS || error == WSAEALREADY) { } else if (error == WSAEINPROGRESS || error == WSAEALREADY) {
// Подключение ещё в процессе, ничего не делаем, ждём следующий вызов process return 0;
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 (оно провалится с ошибкой)
} else { } else {
// Подключение провалилось по другой причине
if (client->log_file) { 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); (unsigned long long)get_timestamp_ms(), error);
fflush(client->log_file); fflush(client->log_file);
} }
etcpmon_client_disconnect(client); etcpmon_client_disconnect(client);
if (client->on_disconnected) { if (client->on_disconnected) client->on_disconnected(client->user_data);
client->on_disconnected(client->user_data);
}
if (client->on_error) { if (client->on_error) {
char msg[256]; char msg[256];
snprintf(msg, sizeof(msg), "Connect failed: %d", error); snprintf(msg, sizeof(msg), "Connect failed: %d", error);
@ -686,139 +675,109 @@ int etcpmon_client_process(struct etcpmon_client* client) {
} }
return -1; 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 мс (новый асинхронный способ) === */ /* Авто-запрос метрик каждые 100 мс (асинхронно) */
if (client->fully_connected && client->selected_peer_id != 0 && client->sock != INVALID_SOCKET) { if (client->fully_connected && client->selected_peer_id != 0) {
uint64_t now = get_timestamp_ms(); uint64_t now = get_timestamp_ms();
if (now - client->last_metrics_request_ms >= ETCPMON_UPDATE_INTERVAL_MS) { if (now - client->last_metrics_request_ms >= ETCPMON_UPDATE_INTERVAL_MS) {
/* Используем новый асинхронный путь с коллбэком */ etcpmon_client_send_request(client, ETCPMON_CMD_GET_METRICS, NULL, 0,
etcpmon_client_send_request(client, metrics_auto_callback, NULL);
ETCPMON_CMD_GET_METRICS,
NULL, 0,
metrics_auto_callback,
NULL);
client->last_metrics_request_ms = now; 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; uint8_t* buf = client->recv_buffer + client->recv_len;
int buf_space = ETCPMON_MAX_MSG_SIZE - client->recv_len; int buf_space = ETCPMON_MAX_MSG_SIZE - client->recv_len;
if (buf_space <= 0) { if (buf_space <= 0) {
client->recv_len = 0; client->recv_len = 0; /* защита от переполнения */
buf = client->recv_buffer; buf = client->recv_buffer;
buf_space = ETCPMON_MAX_MSG_SIZE; buf_space = ETCPMON_MAX_MSG_SIZE;
} }
int received = recv(client->sock, (char*)buf, buf_space, 0); int received = recv(client->sock, (char*)buf, buf_space, 0);
if (received == SOCKET_ERROR) { if (received == SOCKET_ERROR) {
int err = WSAGetLastError(); int err = WSAGetLastError();
if (err == WSAEWOULDBLOCK) { if (err == WSAEWOULDBLOCK) return 0;
return 0; /* No data available */
}
/* Connection error */
if (client->log_file) { 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); (unsigned long long)get_timestamp_ms(), err);
fflush(client->log_file); fflush(client->log_file);
} }
etcpmon_client_disconnect(client); etcpmon_client_disconnect(client);
if (client->on_disconnected) { if (client->on_disconnected) client->on_disconnected(client->user_data);
client->on_disconnected(client->user_data);
}
return -1; return -1;
} }
if (received == 0) { if (received == 0) {
/* Connection closed by server */
if (client->log_file) { 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()); (unsigned long long)get_timestamp_ms());
fflush(client->log_file); fflush(client->log_file);
} }
etcpmon_client_disconnect(client); etcpmon_client_disconnect(client);
if (client->on_disconnected) { if (client->on_disconnected) client->on_disconnected(client->user_data);
client->on_disconnected(client->user_data);
}
return -1; return -1;
} }
/* Log received data */
if (client->log_file) { 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); log_hex_data(client->log_file, "RX", buf, received);
} }
client->recv_len += received; client->recv_len += received;
/* Process complete messages */ /* === ОСНОВНОЙ ПАРСЕР (ИСПРАВЛЕННЫЙ) === */
int processed = 0; int processed = 0;
while (client->recv_len >= sizeof(struct etcpmon_msg_header)) { while (client->recv_len >= sizeof(struct etcpmon_msg_header)) {
struct etcpmon_msg_header* hdr = (struct etcpmon_msg_header*)client->recv_buffer; struct etcpmon_msg_header* hdr = (struct etcpmon_msg_header*)client->recv_buffer;
/* Валидация */
if (etcpmon_validate_header(hdr) != 0) { if (etcpmon_validate_header(hdr) != 0) {
/* Invalid header, skip byte */
if (client->log_file) { if (client->log_file) {
fprintf(client->log_file, "%llu: [ERROR] Invalid header, skipping byte\n", fprintf(client->log_file, "%llu: [ERROR] Invalid header, skipping byte %02X\n",
(unsigned long long)get_timestamp_ms()); (unsigned long long)get_timestamp_ms(), client->recv_buffer[0]);
fflush(client->log_file); fflush(client->log_file);
} }
memmove(client->recv_buffer, client->recv_buffer + 1, client->recv_len - 1); memmove(client->recv_buffer, client->recv_buffer + 1, client->recv_len - 1);
client->recv_len--; client->recv_len--;
continue; continue;
} }
if (client->recv_len < hdr->size) { /* hdr->size — это ПОЛНЫЙ размер сообщения (header + payload) */
break; /* Wait for more data */ 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); 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; client->last_response_seq_id = hdr->seq_id;
if (client->log_file) { if (client->log_file) {
fprintf(client->log_file, "%llu: [LOG] Received message type=0x%02X size=%d seq=%d\n", 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, hdr->size, hdr->seq_id); (unsigned long long)get_timestamp_ms(), hdr->type, total_size, payload_size, hdr->seq_id);
fflush(client->log_file); fflush(client->log_file);
} }
/* Use callback-based handler */
handle_response_with_callback(client, hdr->type, payload, payload_size); handle_response_with_callback(client, hdr->type, payload, payload_size);
processed++; processed++;
/* Remove processed message */ /* Удаляем обработанное сообщение */
if (client->recv_len > hdr->size) { if (client->recv_len > total_size) {
memmove(client->recv_buffer, client->recv_buffer + hdr->size, memmove(client->recv_buffer, client->recv_buffer + total_size,
client->recv_len - hdr->size); client->recv_len - total_size);
} }
client->recv_len -= hdr->size; client->recv_len -= total_size;
} }
return processed; return processed;
} }

Write Preview
Loading…
Cancel
Save