diff --git a/src/control_server.c b/src/control_server.c index 1073fcf4..62c66e49 100644 --- a/src/control_server.c +++ b/src/control_server.c @@ -377,7 +377,6 @@ static void accept_callback(socket_t fd, void* arg) { #endif return; } - DEBUG_WARN(DEBUG_CATEGORY_CONTROL, "Accept..."); /* Check max clients */ if (server->client_count >= server->max_clients) { @@ -429,7 +428,6 @@ static void accept_callback(socket_t fd, void* arg) { } client->subscribed_nodes = 0; client->write_registered = 0; - client->idle_timer = uasync_set_timeout(server->ua, 300000, client, control_client_idle_cb, "ctrl_idle"); /* Register with uasync */ client->socket_id = uasync_add_socket_t(server->ua, client_fd, @@ -446,6 +444,7 @@ static void accept_callback(socket_t fd, void* arg) { fflush(server->log_file); } DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to register client socket with uasync"); + queue_free(client->output_queue); u_free(client); #ifdef _WIN32 closesocket(client_fd); @@ -460,6 +459,13 @@ static void accept_callback(socket_t fd, void* arg) { server->clients = client; server->client_count++; + client->idle_timer = uasync_set_timeout(server->ua, 300000, client, control_client_idle_cb, "ctrl_idle"); + if (!client->idle_timer) { + DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to arm control client idle timer fd=%d", (int)client_fd); + close_client(server, client); + return; + } + const char* addr_str; if (client_addr.ss_family == AF_INET) { struct sockaddr_in* sin = (struct sockaddr_in*)&client_addr; @@ -587,7 +593,7 @@ static void client_write_callback(socket_t fd, void* arg) { struct ll_entry* entry = q->head; #ifdef _WIN32 - int sent = send(fd, (const char*)entry->dgram, entry->len, 0); + int sent = send(fd, (const char*)entry->dgram + client->output_offset, entry->len - client->output_offset, 0); if (sent == SOCKET_ERROR) { int err = WSAGetLastError(); if (err != WSAEWOULDBLOCK) { @@ -598,7 +604,7 @@ static void client_write_callback(socket_t fd, void* arg) { return; } #else - ssize_t sent = send(fd, entry->dgram, entry->len, 0); + ssize_t sent = send(fd, entry->dgram + client->output_offset, entry->len - client->output_offset, 0); if (sent < 0) { if (errno != EAGAIN && errno != EWOULDBLOCK) { DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Client send error: %s, closing", strerror(errno)); @@ -608,10 +614,21 @@ static void client_write_callback(socket_t fd, void* arg) { return; } #endif - if ((size_t)sent < entry->len) { entry->dgram += sent; entry->len -= (uint16_t)sent; q->total_bytes -= (size_t)sent; return; } + if (sent == 0) { + DEBUG_WARN(DEBUG_CATEGORY_CONTROL, "Control client send made no progress fd=%d, closing", (int)fd); + close_client(server, client); + return; + } + client->output_offset += (uint16_t)sent; + if (client->output_offset < entry->len) { + DEBUG_DEBUG(DEBUG_CATEGORY_CONTROL, "Control partial send fd=%d offset=%u size=%u queued_bytes=%zu", + (int)fd, client->output_offset, entry->len, queue_total_bytes(q)); + return; + } struct ll_entry* e = queue_data_get(q); queue_dgram_free(e); queue_entry_free(e); + client->output_offset = 0; } uasync_set_socket_write(server->ua, client->socket_id, 0); @@ -631,14 +648,22 @@ static void client_except_callback(socket_t fd, void* arg) { (void)fd; } +/* Принимает владение буфером, включая ветки ошибки постановки в очередь. */ static int control_client_send(struct control_server* server, struct control_client* client, uint8_t* dgram, uint16_t len) { (void)server; - if (!client || !client->output_queue || !dgram || len == 0) return -1; + if (!client || !client->output_queue || !dgram || len == 0 || len > ETCPMON_MAX_MSG_SIZE) { + DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Invalid control output message client=%p size=%u", (void*)client, len); + u_free(dgram); + return -1; + } struct ll_entry* entry = queue_entry_new(0); - if (!entry) { DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to allocate output queue entry"); return -1; } + if (!entry) { + DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to allocate output queue entry size=%u", len); + u_free(dgram); + return -1; + } entry->dgram = dgram; entry->len = len; - entry->int_len = len; queue_data_put(client->output_queue, entry); if (!client->write_registered) { uasync_set_socket_write(client->server->ua, client->socket_id, 1); @@ -657,6 +682,10 @@ static void control_client_idle_cb(void* arg) { static void close_client(struct control_server* server, struct control_client* client) { if (!server || !client) return; + + DEBUG_DEBUG(DEBUG_CATEGORY_CONTROL, "Closing control client fd=%d queued_packets=%d queued_bytes=%zu output_offset=%u timer=%d", + (int)client->fd, queue_entry_count(client->output_queue), queue_total_bytes(client->output_queue), + client->output_offset, client->idle_timer != NULL); if (client->idle_timer) { uasync_cancel_timeout(server->ua, client->idle_timer); client->idle_timer = NULL; } @@ -680,7 +709,15 @@ static void close_client(struct control_server* server, struct control_client* c } #endif - if (client->output_queue) { queue_free(client->output_queue); client->output_queue = NULL; } + if (client->output_queue) { + struct ll_entry* entry; + while ((entry = queue_data_get(client->output_queue))) { + queue_dgram_free(entry); + queue_entry_free(entry); + } + queue_free(client->output_queue); + client->output_queue = NULL; + } /* Remove from list */ struct control_client** curr = &server->clients; @@ -822,25 +859,29 @@ static void handle_client_data(struct control_server* server, struct control_cli send_debug_config(server, client, req_seq); break; - case ETCPMON_CMD_SET_DEBUG_CONFIG: - if (payload_size >= 2) { - uint8_t global_lvl = payload[0]; - uint8_t cat_count = payload[1]; - if (cat_count > ETCPMON_MAX_DEBUG_CATEGORIES) cat_count = ETCPMON_MAX_DEBUG_CATEGORIES; - if (global_lvl <= DEBUG_LEVEL_TRACE) debug_set_level((debug_level_t)global_lvl); - const uint8_t* levels = payload + 2; - for (uint8_t i = 0; i < cat_count && i + 1 < DEBUG_CATEGORY_COUNT; i++) { - if (levels[i] <= DEBUG_LEVEL_TRACE) - debug_set_category_level((debug_category_t)(i + 1), (debug_level_t)levels[i]); - } - DEBUG_INFO(DEBUG_CATEGORY_CONTROL, "Debug config applied: global=%d categories=%d", global_lvl, cat_count); - if (server->log_file) { - fprintf(server->log_file, "%llu: [LOG] Debug config set: global=%d categories=%d\n", - (unsigned long long)get_timestamp_ms(), global_lvl, cat_count); - fflush(server->log_file); - } + case ETCPMON_CMD_SET_DEBUG_CONFIG: { + if (payload_size < 2 || payload[1] > ETCPMON_MAX_DEBUG_CATEGORIES || payload_size != 2u + payload[1]) { + DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Bad SET_DEBUG_CONFIG payload size=%u categories=%d", + payload_size, payload_size >= 2 ? payload[1] : -1); + close_client(server, client); + return; + } + uint8_t global_lvl = payload[0]; + uint8_t cat_count = payload[1]; + if (global_lvl <= DEBUG_LEVEL_TRACE) debug_set_level((debug_level_t)global_lvl); + const uint8_t* levels = payload + 2; + for (uint8_t i = 0; i < cat_count && i + 1 < DEBUG_CATEGORY_COUNT; i++) { + if (levels[i] <= DEBUG_LEVEL_TRACE) + debug_set_category_level((debug_category_t)(i + 1), (debug_level_t)levels[i]); + } + DEBUG_INFO(DEBUG_CATEGORY_CONTROL, "Debug config applied: global=%d categories=%d", global_lvl, cat_count); + if (server->log_file) { + fprintf(server->log_file, "%llu: [LOG] Debug config set: global=%d categories=%d\n", + (unsigned long long)get_timestamp_ms(), global_lvl, cat_count); + fflush(server->log_file); } break; + } case ETCPMON_CMD_SUBSCRIBE_NODES: client->subscribed_nodes = 1; @@ -871,6 +912,11 @@ static void handle_client_data(struct control_server* server, struct control_cli /* Reset idle timer on each valid command */ if (client->idle_timer) { uasync_cancel_timeout(server->ua, client->idle_timer); } client->idle_timer = uasync_set_timeout(server->ua, 300000, client, control_client_idle_cb, "ctrl_idle"); + if (!client->idle_timer) { + DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to reset control client idle timer fd=%d", (int)client->fd); + close_client(server, client); + return; + } /* Remove processed message from buffer */ uint16_t msg_size = hdr->size; @@ -957,12 +1003,18 @@ static void send_socket_list(struct control_server* server, struct control_clien uint8_t count = 0; struct ETCP_SOCKET* sock = instance->etcp_sockets; while (sock && count < ETCPMON_MAX_CONNECTIONS) { - if (!sock->probe_only) count++; + if (!sock->probe_only) { + if (ETCPMON_SOCKET_LIST_SIZE(count + 1u) > ETCPMON_MAX_MSG_SIZE) { + DEBUG_WARN(DEBUG_CATEGORY_CONTROL, "Socket list exceeds protocol limit: count>%u next_size=%zu max_size=%u", + count, ETCPMON_SOCKET_LIST_SIZE(count + 1u), ETCPMON_MAX_MSG_SIZE); + send_error(client, ETCPMON_ERR_SERVER_BUSY, "Socket list exceeds protocol limit", seq_id); + return; + } + count++; + } sock = sock->next; } - if (count > ETCPMON_MAX_CONNECTIONS) count = ETCPMON_MAX_CONNECTIONS; - uint16_t rsp_size = ETCPMON_SOCKET_LIST_SIZE(count); uint8_t* buffer = (uint8_t*)u_malloc(rsp_size); if (!buffer) { @@ -1241,6 +1293,7 @@ static void send_metrics(struct control_server* server, struct control_client* c struct etcpmon_link_metrics* link_info = (struct etcpmon_link_metrics*)(buffer + sizeof(*hdr) + sizeof(*rsp)); link = conn->links; for (uint8_t i = 0; i < links_count && link; i++) { + memset(&link_info[i], 0, sizeof(link_info[i])); link_info[i].local_link_id = link->local_link_id; link_info[i].status = link->link_status; link_info[i].encrypt_errors = (uint32_t)link->encrypt_errors; @@ -1278,8 +1331,6 @@ static void send_metrics(struct control_server* server, struct control_client* c uint32_t max_bw = link->bbr->bw_hi[0] > link->bbr->bw_hi[1] ? link->bbr->bw_hi[0] : link->bbr->bw_hi[1]; link_info[i].bbr_bw_hi = max_bw ? (uint32_t)((uint64_t)max_bw * link->mtu * 1000000ULL / 16777216ULL) : 0; link_info[i].bbr_bw_lo = link->bbr->bw_lo != ~0U ? (uint32_t)((uint64_t)link->bbr->bw_lo * link->mtu * 1000000ULL / 16777216ULL) : 0; - } else { - memset(&link_info[i].bbr_mode, 0, 11 * 4 + 2 + 2); // zero out all BBR fields } link = link->next; } diff --git a/src/control_server.h b/src/control_server.h index c12b375d..858580ca 100644 --- a/src/control_server.h +++ b/src/control_server.h @@ -1,7 +1,8 @@ /* * control_server.h - Control Socket Server for ETCP Monitoring * - * Provides TCP control interface for ETCP connection monitoring + * TCP-мониторинг ETCP: init → обработка команд в uasync-потоке → shutdown. + * Метрики отправляются по запросу; доступ — по IPv4 control_allow, idle timeout — 30 секунд. */ #ifndef CONTROL_SERVER_H @@ -40,6 +41,7 @@ struct control_client { uint64_t selected_peer_id; /* 0 = none selected */ struct ll_queue* output_queue; /* Output queue for async sends */ + uint16_t output_offset; /* Отправленные байты головы очереди; dgram/len не меняются */ void* idle_timer; /* Client idle timeout */ uint8_t subscribed_nodes; /* 1 = subscribed to node updates */ uint8_t write_registered; /* 1 = write events registered in uasync */ @@ -97,10 +99,8 @@ int control_server_init(struct control_server* server, */ void control_server_shutdown(struct control_server* server); -/* Process periodic updates - * - * Should be called periodically (e.g., from main loop) to send - * metrics updates to clients that have selected connections +/* Разобрать накопленные команды всех клиентов в uasync-потоке. + * Чтение сокета также запускает разбор; метрики отправляются только по CMD_GET_METRICS. * * Parameters: * server - Server state