diff --git a/src/control_server.c b/src/control_server.c index be9df858d..948bbc3ad 100644 --- a/src/control_server.c +++ b/src/control_server.c @@ -91,9 +91,12 @@ static void accept_callback(socket_t fd, void* arg); static void client_read_callback(socket_t fd, void* arg); static void client_write_callback(socket_t fd, void* arg); static void client_except_callback(socket_t fd, void* arg); -static void handle_client_data(struct control_server* server, struct control_client* client); +static int handle_client_data(struct control_server* server, struct control_client* client); static void close_client(struct control_server* server, struct control_client* client); static void control_client_idle_cb(void* arg); +static void control_output_ready(struct ll_queue* queue, void* arg); +static int control_wait_output(struct control_client* client); +static int control_client_arm_timer(struct control_client* client, int tx); static void send_conn_list(struct control_server* server, struct control_client* client, uint8_t seq_id); static void send_socket_list(struct control_server* server, struct control_client* client, uint8_t seq_id); @@ -430,6 +433,9 @@ static void accept_callback(socket_t fd, void* arg) { } client->subscribed_nodes = 0; client->write_registered = 0; + queue_set_size_limit(client->output_queue, 1); + queue_set_threshold(client->output_queue, 0, 0); + queue_set_waiter_defer(client->output_queue, 1); /* Register with uasync */ client->socket_id = uasync_add_socket_t(server->ua, client_fd, @@ -490,6 +496,7 @@ static void accept_callback(socket_t fd, void* arg) { static void client_read_callback(socket_t fd, void* arg) { struct control_client* client = (struct control_client*)arg; struct control_server* server = client->server; + if (client->read_paused) return; if (client->recv_len >= ETCPMON_MAX_MSG_SIZE) { DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Control recv buffer overflow (pre-check)"); @@ -586,57 +593,87 @@ static void client_read_callback(socket_t fd, void* arg) { } } +/* Один таймер: ожидание команд в покое, отсутствие положительного send во время TX. */ +static int control_client_arm_timer(struct control_client* client, int tx) { + if (client->idle_timer) uasync_cancel_timeout(client->server->ua, client->idle_timer); + client->timer_is_tx = tx != 0; + client->idle_timer = uasync_set_timeout(client->server->ua, 300000, client, control_client_idle_cb, tx ? "ctrl_tx" : "ctrl_idle"); + if (!client->idle_timer) { + DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Control timer failed fd=%d mode=%s", (int)client->fd, tx ? "TX" : "idle"); + client->failed = 1; + return -1; + } + return 0; +} + +/* Останавливает входных producers до пустой очереди. Waiter всегда отложен, повторных регистраций нет. */ +static int control_wait_output(struct control_client* client) { + if (!client->read_paused) { + if (uasync_set_socket_read(client->server->ua, client->socket_id, 0) < 0) { + DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to pause control reads fd=%d", (int)client->fd); + client->failed = 1; + return -1; + } + client->read_paused = 1; + DEBUG_DEBUG(DEBUG_CATEGORY_CONTROL, "Control reads paused fd=%d queued=%zu snapshot=%zu/%zu dirty=%zu", + (int)client->fd, queue_total_bytes(client->output_queue), client->snapshot_index, + client->snapshot_count, client->dirty_count); + } + if (!client->timer_is_tx && control_client_arm_timer(client, 1) < 0) return -1; + if (!client->output_waiter.internal && !client->output_waiter.call_soon_id && + queue_waiter_wait(client->output_queue, &client->output_waiter, control_output_ready, client) < 0) { + DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to wait for control output fd=%d", (int)client->fd); + client->failed = 1; + return -1; + } + return 0; +} + static void client_write_callback(socket_t fd, void* arg) { - struct control_client* client = (struct control_client*)arg; + struct control_client* client = arg; struct control_server* server = client->server; - struct ll_queue* q = client->output_queue; - - while (q->head) { - struct ll_entry* entry = q->head; - + struct ll_queue* queue = client->output_queue; + struct ll_entry* entry = queue->head; + if (client->failed) { close_client(server, client); return; } + if (entry) { #ifdef _WIN32 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) { - DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Client send error: %d, closing", err); - close_client(server, client); - return; - } - return; - } + int error = WSAGetLastError(); + if (error == WSAEWOULDBLOCK) return; + DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Control send failed fd=%d error=%d", (int)fd, error); #else 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)); - close_client(server, client); - return; - } + if (errno == EAGAIN || errno == EWOULDBLOCK) return; + DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Control send failed fd=%d: %s", (int)fd, strerror(errno)); +#endif + close_client(server, client); return; } -#endif - if (sent == 0) { - DEBUG_WARN(DEBUG_CATEGORY_CONTROL, "Control client send made no progress fd=%d, closing", (int)fd); + if (!sent) { + DEBUG_WARN(DEBUG_CATEGORY_CONTROL, "Control send made no progress fd=%d", (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)); + DEBUG_DEBUG(DEBUG_CATEGORY_CONTROL, "Control partial send fd=%d offset=%u/%u", + (int)fd, client->output_offset, entry->len); + if (control_client_arm_timer(client, 1) < 0) close_client(server, client); return; } - - struct ll_entry* e = queue_data_get(q); - queue_dgram_free(e); queue_entry_free(e); + entry = queue_data_get(queue); + queue_dgram_free(entry); queue_entry_free(entry); client->output_offset = 0; } - - uasync_set_socket_write(server->ua, client->socket_id, 0); + if (uasync_set_socket_write(server->ua, client->socket_id, 0) < 0) { + DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to disable control writes fd=%d", (int)fd); + close_client(server, client); + return; + } client->write_registered = 0; - if (q->count > 128) DEBUG_WARN(DEBUG_CATEGORY_CONTROL, "ctrl_out: large output queue count=%d fd=%d", q->count, (int)client->fd); - (void)fd; + if (control_client_arm_timer(client, client->snapshot_active || client->dirty_count) < 0) close_client(server, client); } static void client_except_callback(socket_t fd, void* arg) { @@ -652,33 +689,48 @@ static void client_except_callback(socket_t fd, void* arg) { /* Принимает владение буфером, включая ветки ошибки постановки в очередь. */ 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 || len > ETCPMON_MAX_MSG_SIZE) { - DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Invalid control output message client=%p size=%u", (void*)client, len); + if (!client || !client->output_queue || !dgram || !len || len > ETCPMON_MAX_MSG_SIZE) { + DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Invalid control output client=%p size=%u", (void*)client, len); + if (client) client->failed = 1; u_free(dgram); return -1; } + if (client->output_queue->head || client->failed) { + DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Control producer violated TX backpressure fd=%d queued=%zu incoming=%u", + (int)client->fd, queue_total_bytes(client->output_queue), len); + client->failed = 1; 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 size=%u", len); - u_free(dgram); + DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to allocate control output entry fd=%d size=%u", (int)client->fd, len); + client->failed = 1; u_free(dgram); + return -1; + } + entry->dgram = dgram; entry->len = len; + if (queue_data_put(client->output_queue, entry) < 0) { + DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to enqueue control output fd=%d size=%u", (int)client->fd, len); + client->failed = 1; return -1; } - entry->dgram = dgram; - entry->len = len; - queue_data_put(client->output_queue, entry); if (!client->write_registered) { - uasync_set_socket_write(client->server->ua, client->socket_id, 1); + if (uasync_set_socket_write(server->ua, client->socket_id, 1) < 0) { + DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to enable control writes fd=%d", (int)client->fd); + client->failed = 1; + return -1; + } client->write_registered = 1; } - return 0; + return control_wait_output(client); } static void control_client_idle_cb(void* arg) { - struct control_client* client = (struct control_client*)arg; + struct control_client* client = arg; if (!client || !client->server) return; client->idle_timer = NULL; - DEBUG_WARN(DEBUG_CATEGORY_CONTROL, "Control client idle timeout fd=%d, closing", client->fd); + DEBUG_WARN(DEBUG_CATEGORY_CONTROL, "Control %s timeout fd=%d queued_bytes=%zu snapshot=%zu/%zu dirty=%zu", + client->timer_is_tx ? "TX progress" : "idle", (int)client->fd, queue_total_bytes(client->output_queue), + client->snapshot_index, client->snapshot_count, client->dirty_count); close_client(client->server, client); } @@ -690,6 +742,8 @@ static void close_client(struct control_server* server, struct control_client* c client->output_offset, client->idle_timer != NULL); if (client->idle_timer) { uasync_cancel_timeout(server->ua, client->idle_timer); client->idle_timer = NULL; } + if (client->output_queue) queue_waiter_cancel(client->output_queue, &client->output_waiter); + u_free(client->snapshot_ids); client->snapshot_ids = NULL; /* Log disconnection */ if (server->log_file) { @@ -742,19 +796,24 @@ static void close_client(struct control_server* server, struct control_client* c * Message Handling * ============================================================================ */ -static void handle_client_data(struct control_server* server, struct control_client* client) { +static int handle_client_data(struct control_server* server, struct control_client* client) { + if (client->failed) { close_client(server, client); return -1; } + if (client->output_queue->head || client->snapshot_active || client->dirty_count) { + if (control_wait_output(client) < 0) { close_client(server, client); return -1; } + return 0; + } while (client->recv_len >= sizeof(struct etcpmon_msg_header)) { struct etcpmon_msg_header* hdr = (struct etcpmon_msg_header*)client->recv_buffer; if (hdr->size == 0 || hdr->size > ETCPMON_MAX_MSG_SIZE) { DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Invalid message size from client: %u", hdr->size); close_client(server, client); - return; + return -1; } if (hdr->type < ETCPMON_CMD_LIST_CONN || hdr->type > ETCPMON_CMD_SUBSCRIBE_NODES) { DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Invalid command type from client: 0x%02X", hdr->type); close_client(server, client); - return; + return -1; } /* Validate header */ @@ -766,7 +825,7 @@ static void handle_client_data(struct control_server* server, struct control_cli } DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Invalid message header from client"); close_client(server, client); - return; + return -1; } /* Check if full message received */ @@ -779,6 +838,14 @@ static void handle_client_data(struct control_server* server, struct control_cli uint16_t payload_size = hdr->size - sizeof(struct etcpmon_msg_header); uint8_t req_seq = hdr->seq_id; + if ((hdr->type == ETCPMON_CMD_SELECT_CONN && payload_size != sizeof(struct etcpmon_cmd_select)) || + (hdr->type != ETCPMON_CMD_SELECT_CONN && hdr->type != ETCPMON_CMD_ACTION && + hdr->type != ETCPMON_CMD_SET_DEBUG_CONFIG && payload_size)) { + DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Invalid command payload fd=%d type=0x%02X size=%u", + (int)client->fd, hdr->type, payload_size); + close_client(server, client); + return -1; + } if (server->log_file) { fprintf(server->log_file, "%llu: [LOG] Received command type=0x%02X seq=%d size=%d\n", @@ -805,7 +872,7 @@ static void handle_client_data(struct control_server* server, struct control_cli } else { DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Bad SELECT_CONN payload size %u", payload_size); close_client(server, client); - return; + return -1; } break; @@ -825,7 +892,7 @@ static void handle_client_data(struct control_server* server, struct control_cli case ETCPMON_CMD_ACTION: if (handle_action(server, client, payload, payload_size, req_seq) < 0) { close_client(server, client); - return; + return -1; } break; @@ -838,7 +905,7 @@ static void handle_client_data(struct control_server* server, struct control_cli 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; + return -1; } uint8_t global_lvl = payload[0]; uint8_t cat_count = payload[1]; @@ -858,13 +925,6 @@ static void handle_client_data(struct control_server* server, struct control_cli } case ETCPMON_CMD_SUBSCRIBE_NODES: - client->subscribed_nodes = 1; - if (server->log_file) { - fprintf(server->log_file, "%llu: [LOG] Client subscribed to node updates\n", - (unsigned long long)get_timestamp_ms()); - fflush(server->log_file); - } - DEBUG_INFO(DEBUG_CATEGORY_CONTROL, "Client subscribed to node updates"); control_server_send_node_list(server, client, req_seq); break; @@ -875,22 +935,15 @@ static void handle_client_data(struct control_server* server, struct control_cli fflush(server->log_file); } close_client(server, client); - return; + return -1; default: DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Unknown command from client: 0x%02X", hdr->type); close_client(server, client); - return; + return -1; } - /* 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; - } + if (client->failed) { close_client(server, client); return -1; } /* Remove processed message from buffer */ uint16_t msg_size = hdr->size; @@ -899,7 +952,13 @@ static void handle_client_data(struct control_server* server, struct control_cli client->recv_len - msg_size); } client->recv_len -= msg_size; + if (client->output_queue->head || client->snapshot_active || client->dirty_count) { + if (control_wait_output(client) < 0) { close_client(server, client); return -1; } + return 0; + } + if (control_client_arm_timer(client, 0) < 0) { close_client(server, client); return -1; } } + return 0; } /* ============================================================================ @@ -973,6 +1032,7 @@ static void send_conn_list(struct control_server* server, struct control_client* uint8_t* buffer = (uint8_t*)u_malloc(rsp_size); if (!buffer) { DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to allocate connection list buffer"); + client->failed = 1; return; } @@ -1012,7 +1072,11 @@ static void send_socket_list(struct control_server* server, struct control_clien if (!instance || !instance->etcp_sockets) { uint16_t rsp_size = ETCPMON_SOCKET_LIST_SIZE(0); uint8_t* buffer = (uint8_t*)u_malloc(rsp_size); - if (!buffer) return; + if (!buffer) { + DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to allocate empty socket list fd=%d", (int)client->fd); + client->failed = 1; + return; + } struct etcpmon_msg_header* hdr = (struct etcpmon_msg_header*)buffer; etcpmon_build_header(hdr, sizeof(struct etcpmon_rsp_socket_list), @@ -1044,6 +1108,7 @@ static void send_socket_list(struct control_server* server, struct control_clien uint8_t* buffer = (uint8_t*)u_malloc(rsp_size); if (!buffer) { DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to allocate socket list buffer"); + client->failed = 1; return; } @@ -1135,6 +1200,7 @@ static void send_metrics(struct control_server* server, struct control_client* c uint8_t* buffer = (uint8_t*)u_malloc(rsp_size); if (!buffer) { DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to allocate metrics buffer"); + client->failed = 1; return; } @@ -1387,7 +1453,11 @@ static void send_error(struct control_client* client, uint8_t error_code, const uint16_t rsp_size = ETCPMON_ERROR_SIZE(msg_len); uint8_t* buffer = (uint8_t*)u_malloc(rsp_size); - if (!buffer) return; + if (!buffer) { + DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to allocate control error fd=%d seq=%u", (int)client->fd, seq_id); + client->failed = 1; + return; + } struct etcpmon_msg_header* hdr = (struct etcpmon_msg_header*)buffer; etcpmon_build_header(hdr, sizeof(struct etcpmon_rsp_error) + msg_len + 1, @@ -1428,7 +1498,11 @@ static void send_debug_config(struct control_server* server, struct control_clie uint16_t payload_size = 2 + names_len + cat_count; uint16_t rsp_size = sizeof(struct etcpmon_msg_header) + payload_size; uint8_t* buffer = (uint8_t*)u_malloc(rsp_size); - if (!buffer) { DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to allocate debug config buffer"); return; } + if (!buffer) { + DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to allocate debug config buffer"); + client->failed = 1; + return; + } struct etcpmon_msg_header* hdr = (struct etcpmon_msg_header*)buffer; etcpmon_build_header(hdr, payload_size, ETCPMON_RSP_DEBUG_CONFIG, seq_id); @@ -1468,16 +1542,29 @@ static struct ETCP_CONN* find_connection_by_peer_id(struct UTUN_INSTANCE* instan static void send_single_node_info(struct control_server* server, struct control_client* client, struct TOPO_GROUP_NODE* nq) { if (!server || !nq) return; struct TOPO_NODE* ni = topo_node_registry_find(server->instance->topo_groups, nq->node_id); - if (!ni) return; - uint8_t buf[8192]; + if (!ni) { + DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Node snapshot has no registry entry fd=%d node=%016llx", (int)client->fd, + (unsigned long long)nq->node_id); + client->failed = 1; + return; + } + uint8_t buf[TOPO_NODE_WIRE_MAX_SIZE + 2]; int ser_len = topo_node_serialize(ni, nq, TOPO_GROUP_UTUN, TOPO_FLAG_SEND_SUBNETS, buf + 2, sizeof(buf) - 2, 0); - if (ser_len < 0) { DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to serialize node info"); return; } + if (ser_len < 0) { + DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to serialize node info node=%016llx", (unsigned long long)nq->node_id); + client->failed = 1; + return; + } buf[0] = CTRL_BGP_CMD; buf[1] = CTRL_BGP_SUBCMD; size_t data_size = 2 + (size_t)ser_len; size_t np_size = sizeof(struct etcpmon_msg_header) + data_size; uint8_t* buffer = (uint8_t*)u_malloc(np_size); - if (!buffer) { DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to allocate node info buffer"); return; } + if (!buffer) { + DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to allocate node info buffer"); + client->failed = 1; + return; + } struct etcpmon_msg_header* hdr = (struct etcpmon_msg_header*)buffer; etcpmon_build_header(hdr, (uint16_t)data_size, ETCPMON_RSP_NODE_INFO, 0); @@ -1488,61 +1575,148 @@ static void send_single_node_info(struct control_server* server, struct control_ static void control_server_send_node_list(struct control_server* server, struct control_client* client, uint8_t seq_id) { struct TOPO_GROUP* group = topo_groups_get_default(server->instance->topo_groups); - if (!group) { send_error(client, ETCPMON_ERR_SERVER_BUSY, "BGP not available", seq_id); return; } - - /* Send local node first */ - if (group->local_node) send_single_node_info(server, client, group->local_node); - - /* Send all known remote nodes */ - struct ll_entry* e = group->nodes ? group->nodes->head : NULL; - while (e) { - struct TOPO_GROUP_NODE* nq = (struct TOPO_GROUP_NODE*)e; - send_single_node_info(server, client, nq); - e = e->next; + if (!group) { + DEBUG_WARN(DEBUG_CATEGORY_CONTROL, "Node subscription unavailable fd=%d seq=%u", (int)client->fd, seq_id); + send_error(client, ETCPMON_ERR_SERVER_BUSY, "BGP not available", seq_id); + return; } + size_t count = (group->nodes ? queue_entry_count(group->nodes) : 0) + (group->local_node != NULL); + if (count > UINT32_MAX / sizeof(uint64_t)) { + DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Node snapshot too large fd=%d nodes=%zu", (int)client->fd, count); + client->failed = 1; + return; + } + uint64_t* ids = count ? u_malloc(count * sizeof(*ids)) : NULL; + if (count && !ids) { + DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Node snapshot allocation failed fd=%d nodes=%zu", (int)client->fd, count); + client->failed = 1; + return; + } + size_t used = 0; + if (group->local_node) ids[used++] = group->local_node->node_id; + for (struct ll_entry* e = group->nodes ? group->nodes->head : NULL; e; e = e->next) { + if (used >= count) { + DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Node snapshot count mismatch fd=%d count=%zu", (int)client->fd, count); + u_free(ids); client->failed = 1; + return; + } + ids[used++] = ((struct TOPO_GROUP_NODE*)e)->node_id; + } + u_free(client->snapshot_ids); + client->snapshot_ids = ids; client->snapshot_count = used; client->snapshot_index = 0; + client->snapshot_active = 1; client->snapshot_seq = seq_id; client->subscribed_nodes = 1; + DEBUG_INFO(DEBUG_CATEGORY_CONTROL, "Node snapshot started fd=%d seq=%u nodes=%zu id_bytes=%zu", + (int)client->fd, seq_id, used, used * sizeof(*ids)); +} - /* Send end-of-list marker */ - uint8_t* buffer = (uint8_t*)u_malloc(sizeof(struct etcpmon_msg_header)); - if (!buffer) return; - struct etcpmon_msg_header* hdr = (struct etcpmon_msg_header*)buffer; - etcpmon_build_header(hdr, 0, ETCPMON_RSP_NODE_INFO_END, seq_id); - control_client_send(server, client, buffer, sizeof(struct etcpmon_msg_header)); +/* Продолжение producer после опустошения TX; за один вызов создаётся максимум один кадр. */ +static void control_output_ready(struct ll_queue* queue, void* arg) { + struct control_client* client = arg; + struct control_server* server = client->server; + if (client->failed) { close_client(server, client); return; } + if (queue->head) { + if (control_wait_output(client) < 0) close_client(server, client); + return; + } + struct TOPO_GROUP* group = topo_groups_get_default(server->instance->topo_groups); + if ((client->snapshot_active || client->dirty_count) && !group) { + DEBUG_WARN(DEBUG_CATEGORY_CONTROL, "Subscribed UTUN group disappeared fd=%d", (int)client->fd); + close_client(server, client); + return; + } + while (client->snapshot_active && client->snapshot_index < client->snapshot_count && !queue->head && !client->failed) { + uint64_t id = client->snapshot_ids[client->snapshot_index++]; + struct TOPO_GROUP_NODE* node = group->local_node && group->local_node->node_id == id ? + group->local_node : topo_node_find_by_id(group, id); + if (node) send_single_node_info(server, client, node); + else DEBUG_DEBUG(DEBUG_CATEGORY_CONTROL, "Snapshot skips removed node fd=%d node=%016llx", (int)client->fd, + (unsigned long long)id); + } + if (client->snapshot_active && client->snapshot_index == client->snapshot_count && !queue->head && !client->failed) { + uint8_t* buffer = u_malloc(sizeof(struct etcpmon_msg_header)); + if (!buffer) { + DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to allocate node list end fd=%d", (int)client->fd); + client->failed = 1; + } else { + etcpmon_build_header((void*)buffer, 0, ETCPMON_RSP_NODE_INFO_END, client->snapshot_seq); + DEBUG_INFO(DEBUG_CATEGORY_CONTROL, "Node snapshot queued fd=%d seq=%u nodes=%zu", + (int)client->fd, client->snapshot_seq, client->snapshot_count); + client->snapshot_active = 0; + u_free(client->snapshot_ids); client->snapshot_ids = NULL; + client->snapshot_count = 0; client->snapshot_index = 0; + control_client_send(server, client, buffer, sizeof(struct etcpmon_msg_header)); + } + } + while (!client->snapshot_active && client->dirty_count && !queue->head && !client->failed) { + uint64_t id = client->dirty_ids[0]; + int removed = client->dirty_removed[0]; + client->dirty_count--; + memmove(client->dirty_ids, client->dirty_ids + 1, client->dirty_count * sizeof(*client->dirty_ids)); + memmove(client->dirty_removed, client->dirty_removed + 1, client->dirty_count * sizeof(*client->dirty_removed)); + struct TOPO_GROUP_NODE* node = group->local_node && group->local_node->node_id == id ? + group->local_node : topo_node_find_by_id(group, id); + if (!removed && node) send_single_node_info(server, client, node); + else { + size_t size = sizeof(struct etcpmon_msg_header) + sizeof(id); + uint8_t* buffer = u_malloc(size); + if (!buffer) { + DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to allocate node removal fd=%d", (int)client->fd); + client->failed = 1; + break; + } + etcpmon_build_header((void*)buffer, sizeof(id), ETCPMON_RSP_NODE_REMOVED, 0); + memcpy(buffer + sizeof(struct etcpmon_msg_header), &id, sizeof(id)); + control_client_send(server, client, buffer, (uint16_t)size); + } + } + if (client->failed) { close_client(server, client); return; } + if (queue->head) return; + if (handle_client_data(server, client) < 0) return; + if (queue->head || client->snapshot_active || client->dirty_count) return; + if (client->read_paused) { + if (uasync_set_socket_read(server->ua, client->socket_id, 1) < 0) { + DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to resume control reads fd=%d", (int)client->fd); + close_client(server, client); + return; + } + client->read_paused = 0; + DEBUG_DEBUG(DEBUG_CATEGORY_CONTROL, "Control reads resumed fd=%d buffered=%u", (int)client->fd, client->recv_len); + } + if (client->timer_is_tx && control_client_arm_timer(client, 0) < 0) close_client(server, client); +} - DEBUG_INFO(DEBUG_CATEGORY_CONTROL, "Sent node list to client (seq=%d)", seq_id); +/* Сохраняет только последнее событие узла. Ограниченный набор не растёт от повторов одного ID. */ +static int control_mark_node(struct control_client* client, uint64_t id, int removed) { + size_t index = 0; + while (index < client->dirty_count && client->dirty_ids[index] != id) index++; + if (index == CONTROL_NODE_DIRTY_MAX) { + DEBUG_WARN(DEBUG_CATEGORY_CONTROL, "Control subscriber overwhelmed fd=%d dirty_nodes=%zu queued_bytes=%zu", + (int)client->fd, client->dirty_count, queue_total_bytes(client->output_queue)); + return -1; + } + if (index == client->dirty_count) client->dirty_count++; + client->dirty_ids[index] = id; client->dirty_removed[index] = removed != 0; + return control_wait_output(client); } void control_server_notify_node_change(struct control_server* server, struct TOPO_GROUP_NODE* node) { if (!server || !node) return; struct control_client* client = server->clients; while (client) { - if (client->subscribed_nodes) send_single_node_info(server, client, node); - client = client->next; + struct control_client* next = client->next; + if (client->subscribed_nodes && control_mark_node(client, node->node_id, 0) < 0) close_client(server, client); + client = next; } } void control_server_notify_node_removed(struct control_server* server, uint64_t node_id) { if (!server) return; - - uint8_t* buffer = (uint8_t*)u_malloc(sizeof(struct etcpmon_msg_header) + sizeof(uint64_t)); - if (!buffer) return; - - struct etcpmon_msg_header* hdr = (struct etcpmon_msg_header*)buffer; - etcpmon_build_header(hdr, sizeof(uint64_t), ETCPMON_RSP_NODE_REMOVED, 0); - *(uint64_t*)(buffer + sizeof(*hdr)) = node_id; - struct control_client* client = server->clients; while (client) { - if (client->subscribed_nodes) { - /* Every client needs its own copy since ownership transfers to the queue */ - uint8_t* copy = (uint8_t*)u_malloc(sizeof(struct etcpmon_msg_header) + sizeof(uint64_t)); - if (copy) { - memcpy(copy, buffer, sizeof(struct etcpmon_msg_header) + sizeof(uint64_t)); - control_client_send(server, client, copy, sizeof(struct etcpmon_msg_header) + sizeof(uint64_t)); - } - } - client = client->next; + struct control_client* next = client->next; + if (client->subscribed_nodes && control_mark_node(client, node_id, 1) < 0) close_client(server, client); + client = next; } - u_free(buffer); } /* ============================================================================ diff --git a/src/control_server.h b/src/control_server.h index 858580ca9..f4d408cd6 100644 --- a/src/control_server.h +++ b/src/control_server.h @@ -2,7 +2,9 @@ * control_server.h - Control Socket Server for ETCP Monitoring * * TCP-мониторинг ETCP: init → обработка команд в uasync-потоке → shutdown. - * Метрики отправляются по запросу; доступ — по IPv4 control_allow, idle timeout — 30 секунд. + * Метрики отправляются по запросу; доступ — по IPv4 [control] allow, idle timeout — 30 секунд без команд. + * Шифрования/аутентификации протокола нет. Общий wire-контракт — etcpmon_protocol.h. + * Client владеет своим сокетом, таймером и очередью TX; очередь сейчас без лимита/backpressure. */ #ifndef CONTROL_SERVER_H @@ -10,6 +12,7 @@ #include "../tools/etcpmon/etcpmon_protocol.h" #include "../lib/socket_compat.h" +#include "../lib/ll_queue.h" #include #include @@ -26,6 +29,8 @@ struct ll_queue; /* Forward declaration */ struct control_server; +#define CONTROL_NODE_DIRTY_MAX 256 + /* Client connection state */ struct control_client { struct control_client* next; /* Linked list */ @@ -40,11 +45,23 @@ struct control_client { /* Selected connection */ uint64_t selected_peer_id; /* 0 = none selected */ - struct ll_queue* output_queue; /* Output queue for async sends */ + struct ll_queue* output_queue; /* Владеет TX-буферами; сейчас без ограничения объёма/числа сообщений */ 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 */ + uint8_t read_paused; /* Чтение/разбор ждут пустой TX-очереди */ + uint8_t failed; /* Ошибка producer; владелец callback закроет client после возврата */ + uint8_t timer_is_tx; /* Таймер контролирует прогресс TX, иначе отсутствие команд */ + struct queue_waiter_handle output_waiter; /* Отложенное продолжение; отменить до освобождения client */ + uint64_t* snapshot_ids; /* ID начального списка; никаких заимствованных TOPO_GROUP_NODE* */ + size_t snapshot_count; + size_t snapshot_index; /* Следующий ID для сериализации */ + uint8_t snapshot_active; /* 1 до постановки NODE_INFO_END */ + uint8_t snapshot_seq; /* seq запроса для NODE_INFO_END */ + uint64_t dirty_ids[CONTROL_NODE_DIRTY_MAX]; /* Последние изменения объединяются по ID */ + uint8_t dirty_removed[CONTROL_NODE_DIRTY_MAX]; /* Последнее событие: 1 = удаление, 0 = обновление */ + size_t dirty_count; /* При переполнении подписчик отключается, события не теряются молча */ }; /* Control server state */ @@ -79,10 +96,11 @@ struct control_server { * ua - uasync context for event handling * instance - Parent utun instance * bind_addr - Address to bind (from config control_sock) - * max_clients - Maximum concurrent client connections + * max_clients - Maximum concurrent client connections; 0 selects 8 * * Returns: - * 0 on success, -1 on error + * 0 on success, -1 on error. Server copies bind_addr; caller retains ua/instance/server. + * With valid arguments, shutdown also cleans resources left by a failed initialization. */ int control_server_init(struct control_server* server, struct UASYNC* ua, @@ -100,7 +118,8 @@ int control_server_init(struct control_server* server, void control_server_shutdown(struct control_server* server); /* Разобрать накопленные команды всех клиентов в uasync-потоке. - * Чтение сокета также запускает разбор; метрики отправляются только по CMD_GET_METRICS. + * Чтение сокета само запускает разбор; отдельный периодический вызов не требуется. + * Функция лишь повторно разбирает уже принятые данные, не читает сокеты и не рассылает метрики по таймеру. * * Parameters: * server - Server state @@ -117,7 +136,11 @@ void control_server_process_updates(struct control_server* server); */ uint32_t control_server_get_client_count(const struct control_server* server); +/* В uasync-потоке поставить подписанным клиентам снимок node (заимствован на время вызова). + * NODE_INFO содержит 01 04 + topo_node_serialize для UTUN с подсетями; seq=0. + * Неизвестный в registry узел не отправляется. Буферы ответов принадлежат очередям клиентов. */ void control_server_notify_node_change(struct control_server* server, struct TOPO_GROUP_NODE* node); +/* В uasync-потоке поставить подписанным клиентам NODE_REMOVED с node_id и seq=0. */ void control_server_notify_node_removed(struct control_server* server, uint64_t node_id); #ifdef __cplusplus diff --git a/src/routing_layer/topo_group.c b/src/routing_layer/topo_group.c index 0526ba396..885926b0e 100644 --- a/src/routing_layer/topo_group.c +++ b/src/routing_layer/topo_group.c @@ -1200,9 +1200,11 @@ static int topo_group_routes_changed(struct TOPO_GROUP* group, struct TOPO_GROUP topo_fire_nodeinfo_cbk(group->instance, group, node); if (!node->paths) { event = TOPO_NODE_EVENT_REMOVE; - if (group->instance->control_srv) control_server_notify_node_removed(group->instance->control_srv, id); + if (group->group_id == TOPO_GROUP_UTUN && group->instance->control_srv) + control_server_notify_node_removed(group->instance->control_srv, id); topo_nodeq_remove_node(group, node); - } else if (group->instance->control_srv) control_server_notify_node_change(group->instance->control_srv, node); + } else if (group->group_id == TOPO_GROUP_UTUN && group->instance->control_srv) + control_server_notify_node_change(group->instance->control_srv, node); for (struct topo_node_cbk_entry* c = group->node_cbks; c; c = c->next) c->fn(group, id, event, c->arg); topo_group_publish_route(group, id); topo_recovery_changed(group);