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

control_server: node subscription protocol with output queue, push notifications on routing changes

Browse Source
This commit is contained in:
Evgeny committed 2026-06-25 09:28:07 +03:00
1 parent 231081dd2f
commit d8c8c1c5b0
5 files changed
+202 -848
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/control_server.c
+183 -45
View File
@@ -11,6 +11,7 @@
#include "tun_if.h"
#include "route_lib.h"
#include "route_bgp.h"
#include "route_node.h"
#include "pkt_normalizer.h"
#include "../tools/etcpmon/etcpmon_protocol.h"
#include "../lib/u_async.h"
@@ -96,8 +97,19 @@ static void send_metrics(struct control_server* server, struct control_client* c
static void send_debug_config(struct control_server* server, struct control_client* client, uint8_t seq_id);
static void send_error(struct control_client* client, uint8_t error_code, const char* msg, uint8_t seq_id);
static void control_server_send_node_list(struct control_server* server, struct control_client* client, uint8_t seq_id);
static void send_single_node_info(struct control_server* server, struct control_client* client, struct NODEINFO_Q* nq);
static int control_client_send(struct control_server* server, struct control_client* client, uint8_t* dgram, uint16_t len);
static struct ETCP_CONN* find_connection_by_peer_id(struct UTUN_INSTANCE* instance, uint64_t peer_id);
/* From route_bgp.c */
int nodeinfo_dyn_size(struct NODEINFO* node);
/* ETCP ID/subcmd constants for BGP_NODEINFO_PACKET (from route_bgp.h) */
#define CTRL_BGP_CMD 0x01
#define CTRL_BGP_SUBCMD 0x04
/* ============================================================================
* Server Initialization
* ============================================================================ */
@@ -400,13 +412,26 @@ static void accept_callback(socket_t fd, void* arg) {
client->fd = client_fd;
client->server = server; /* Store back pointer to server */
// client->connected = 1;
client->selected_peer_id = 0;
client->output_queue = queue_new(server->ua, 0, 0, 0, "ctrl_out");
if (!client->output_queue) {
DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to create output queue for client");
u_free(client);
#ifdef _WIN32
closesocket(client_fd);
#else
close(client_fd);
#endif
return;
}
client->subscribed_nodes = 0;
client->write_registered = 0;
/* Register with uasync */
client->socket_id = uasync_add_socket_t(server->ua, client_fd,
client_read_callback,
NULL, /* write callback - not needed for now */
client_write_callback,
client_except_callback,
client);
@@ -544,9 +569,44 @@ static void client_read_callback(socket_t fd, void* arg) {
}
static void client_write_callback(socket_t fd, void* arg) {
/* Not used for now - writes are immediate */
struct control_client* client = (struct control_client*)arg;
struct control_server* server = client->server;
struct ll_queue* q = client->output_queue;
while (q->head) {
struct ll_entry* entry = q->head;
#ifdef _WIN32
int sent = send(fd, (const char*)entry->dgram, entry->len, 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;
}
#else
ssize_t sent = send(fd, entry->dgram, entry->len, 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;
}
return;
}
#endif
if ((size_t)sent < entry->len) { entry->dgram += sent; entry->len -= (uint16_t)sent; q->total_bytes -= (size_t)sent; return; }
struct ll_entry* e = queue_data_get(q);
queue_dgram_free(e); queue_entry_free(e);
}
uasync_set_socket_write(server->ua, client->socket_id, 0);
client->write_registered = 0;
(void)fd;
(void)arg;
}
static void client_except_callback(socket_t fd, void* arg) {
@@ -560,6 +620,21 @@ 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;
struct ll_entry* entry = queue_entry_new(0);
if (!entry) { DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to allocate output queue entry"); 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);
client->write_registered = 1;
}
return 0;
}
static void close_client(struct control_server* server, struct control_client* client) {
if (!server || !client) return;
@@ -583,7 +658,9 @@ static void close_client(struct control_server* server, struct control_client* c
close(client->fd);
}
#endif
if (client->output_queue) { queue_free(client->output_queue); client->output_queue = NULL; }
/* Remove from list */
struct control_client** curr = &server->clients;
while (*curr) {
@@ -614,7 +691,7 @@ static void handle_client_data(struct control_server* server, struct control_cli
close_client(server, client);
return;
}
if (hdr->type < ETCPMON_CMD_LIST_CONN || hdr->type > ETCPMON_CMD_SET_DEBUG_CONFIG) {
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;
@@ -721,6 +798,17 @@ static void handle_client_data(struct control_server* server, struct control_cli
}
break;
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;
case ETCPMON_CMD_DISCONNECT:
if (server->log_file) {
fprintf(server->log_file, "%llu: [LOG] Client requested disconnect\n",
@@ -795,13 +883,7 @@ static void send_conn_list(struct control_server* server, struct control_client*
fflush(server->log_file);
}
#ifdef _WIN32
send(client->fd, (const char*)buffer, rsp_size, 0);
#else
send(client->fd, buffer, rsp_size, 0);
#endif
u_free(buffer);
control_client_send(server, client, buffer, rsp_size);
}
static void send_socket_list(struct control_server* server, struct control_client* client, uint8_t seq_id) {
@@ -819,8 +901,7 @@ static void send_socket_list(struct control_server* server, struct control_clien
struct etcpmon_rsp_socket_list* rsp = (struct etcpmon_rsp_socket_list*)(buffer + sizeof(*hdr));
rsp->count = 0;
send(client->fd, buffer, rsp_size, 0);
u_free(buffer);
control_client_send(server, client, buffer, rsp_size);
return;
}
@@ -889,17 +970,11 @@ static void send_socket_list(struct control_server* server, struct control_clien
if (server->log_file) {
log_hex_data(server->log_file, "TX", buffer, rsp_size);
fprintf(server->log_file, "%llu: [LOG] Sent RSP_SOCKET_LIST count=%d\n",
(unsigned long long)get_timestamp_ms(), count);
(unsigned long long)get_timestamp_ms(), count);
fflush(server->log_file);
}
#ifdef _WIN32
send(client->fd, (const char*)buffer, rsp_size, 0);
#else
send(client->fd, buffer, rsp_size, 0);
#endif
u_free(buffer);
control_client_send(server, client, buffer, rsp_size);
}
static void send_metrics(struct control_server* server, struct control_client* client, uint8_t seq_id) {
@@ -1136,13 +1211,7 @@ static void send_metrics(struct control_server* server, struct control_client* c
fflush(server->log_file);
}
#ifdef _WIN32
send(client->fd, (const char*)buffer, rsp_size, 0);
#else
send(client->fd, buffer, rsp_size, 0);
#endif
u_free(buffer);
control_client_send(server, client, buffer, rsp_size);
}
static void send_error(struct control_client* client, uint8_t error_code, const char* msg, uint8_t seq_id) {
@@ -1170,13 +1239,7 @@ static void send_error(struct control_client* client, uint8_t error_code, const
fflush(server->log_file);
}
#ifdef _WIN32
send(client->fd, (const char*)buffer, rsp_size, 0);
#else
send(client->fd, buffer, rsp_size, 0);
#endif
u_free(buffer);
control_client_send(server, client, buffer, rsp_size);
}
static void send_debug_config(struct control_server* server, struct control_client* client, uint8_t seq_id) {
@@ -1215,13 +1278,7 @@ static void send_debug_config(struct control_server* server, struct control_clie
fflush(server->log_file);
}
#ifdef _WIN32
send(client->fd, (const char*)buffer, rsp_size, 0);
#else
send(client->fd, buffer, rsp_size, 0);
#endif
u_free(buffer);
control_client_send(server, client, buffer, rsp_size);
}
/* ============================================================================
@@ -1240,6 +1297,87 @@ static struct ETCP_CONN* find_connection_by_peer_id(struct UTUN_INSTANCE* instan
return NULL;
}
/* ============================================================================
* Node Info Helpers (send_single_node_info, send_node_list, notify_change)
* ============================================================================ */
static void send_single_node_info(struct control_server* server, struct control_client* client, struct NODEINFO_Q* nq) {
int dyn = nodeinfo_dyn_size(&nq->node);
size_t np_size = sizeof(struct etcpmon_msg_header) + 2 + sizeof(struct NODEINFO) + dyn;
uint8_t* buffer = (uint8_t*)u_malloc(np_size);
if (!buffer) { DEBUG_ERROR(DEBUG_CATEGORY_CONTROL, "Failed to allocate node info buffer"); return; }
struct etcpmon_msg_header* hdr = (struct etcpmon_msg_header*)buffer;
etcpmon_build_header(hdr, 2 + sizeof(struct NODEINFO) + dyn, ETCPMON_RSP_NODE_INFO, 0);
uint8_t* pay = buffer + sizeof(*hdr);
pay[0] = CTRL_BGP_CMD; pay[1] = CTRL_BGP_SUBCMD;
memcpy(pay + 2, &nq->node, sizeof(struct NODEINFO));
memcpy(pay + 2 + sizeof(struct NODEINFO), (uint8_t*)&nq->node + sizeof(struct NODEINFO), dyn);
control_client_send(server, client, buffer, (uint16_t)np_size);
}
static void control_server_send_node_list(struct control_server* server, struct control_client* client, uint8_t seq_id) {
struct ROUTE_BGP* bgp = server->instance->bgp;
if (!bgp) { send_error(client, ETCPMON_ERR_SERVER_BUSY, "BGP not available", seq_id); return; }
/* Send local node first */
if (bgp->local_node) send_single_node_info(server, client, bgp->local_node);
/* Send all known remote nodes */
struct ll_entry* e = bgp->nodes ? bgp->nodes->head : NULL;
while (e) {
struct NODEINFO_Q* nq = (struct NODEINFO_Q*)e;
send_single_node_info(server, client, nq);
e = e->next;
}
/* 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));
DEBUG_INFO(DEBUG_CATEGORY_CONTROL, "Sent node list to client (seq=%d)", seq_id);
}
void control_server_notify_node_change(struct control_server* server, struct NODEINFO_Q* 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;
}
}
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;
}
u_free(buffer);
}
/* ============================================================================
* Public API Implementation
* ============================================================================ */