limingjie/OmniSocketGo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 增加链路统计信息，两个链路分别显示在前端，D向A汇报D与B的信息

Browse Source
This commit is contained in:
Mock
2026-04-09 13:38:10 +08:00
parent e72f7f3fd9
commit 11e67282c7
19 changed files with 573 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

277
src/server_kcp_hub.c
View File

@@ -1,8 +1,14 @@
#include "server_kcp_hub.h"
#include "cJSON.h"
#include <stdatomic.h>
#include <unistd.h>
#define KCP_RELAY_MAX_DATAGRAM_SIZE (60 * 1024)
#define KCP_HUB_DEFAULT_TELEMETRY_INTERVAL_MS 500
#define KCP_HUB_TELEMETRY_NODE_ID "hub-telemetry"
#define KCP_HUB_DEFAULT_NODE_ID "hub"
typedef struct kcp_peer_entry {
struct kcp_peer_entry *next;
@@ -21,14 +27,29 @@ struct kcp_hub {
latency_logger_t *logger;
kcp_session_stats_logger_t *stats_logger;
int stats_interval_ms;
char telemetry_peer_id[OMNI_MAX_PEER_ID];
int telemetry_interval_ms;
pthread_t telemetry_thread;
int telemetry_thread_started;
int relay_fd;
int relay_configured;
int relay_learn_peer;
struct sockaddr_storage relay_peer_addr;
socklen_t relay_peer_addr_len;
int closed;
atomic_int closed;
};
static int kcp_hub_peer_id_has_suffix(const char *peer_id, const char *suffix);
static int kcp_hub_deliver_to_local_peer(kcp_hub_t *hub, const message_t *msg);
static int kcp_hub_peer_is_telemetry(const char *peer_id) {
return kcp_hub_peer_id_has_suffix(peer_id, "-telemetry");
}
static const char *kcp_hub_peer_node_id(const char *peer_id) {
return kcp_hub_peer_is_telemetry(peer_id) ? KCP_HUB_TELEMETRY_NODE_ID : KCP_HUB_DEFAULT_NODE_ID;
}
static void kcp_hub_unregister(kcp_hub_t *hub, const char *peer_id, kcp_conn_t *conn) {
kcp_peer_entry_t *prev = NULL;
kcp_peer_entry_t *entry;
@@ -90,9 +111,201 @@ static int kcp_hub_configure_peer_transport(kcp_conn_t *conn, const char *peer_i
kcp_conn_options_set_video_defaults(&options);
return kcp_conn_apply_options(conn, &options);
}
if (kcp_hub_peer_is_telemetry(peer_id)) {
kcp_conn_options_set_telemetry_defaults(&options);
return kcp_conn_apply_options(conn, &options);
}
return 0;
}
static int kcp_hub_add_runtime_stats_json(cJSON *object, const kcp_runtime_stats_t *stats) {
if (object == NULL || stats == NULL) {
errno = EINVAL;
return -1;
}
if (cJSON_AddNumberToObject(object, "connected", stats->connected) == NULL ||
cJSON_AddNumberToObject(object, "conv", (double) stats->conv) == NULL ||
cJSON_AddNumberToObject(object, "rto_ms", (double) stats->rto_ms) == NULL ||
cJSON_AddNumberToObject(object, "srtt_ms", (double) stats->srtt_ms) == NULL ||
cJSON_AddNumberToObject(object, "srttvar_ms", (double) stats->srttvar_ms) == NULL ||
cJSON_AddNumberToObject(object, "snd_wnd", (double) stats->snd_wnd) == NULL ||
cJSON_AddNumberToObject(object, "rmt_wnd", (double) stats->rmt_wnd) == NULL ||
cJSON_AddNumberToObject(object, "inflight", (double) stats->inflight) == NULL ||
cJSON_AddNumberToObject(object, "window_limit", (double) stats->window_limit) == NULL ||
cJSON_AddNumberToObject(object, "window_pressure_pct", stats->window_pressure_pct) == NULL ||
cJSON_AddNumberToObject(object, "snd_queue", (double) stats->snd_queue) == NULL ||
cJSON_AddNumberToObject(object, "rcv_queue", (double) stats->rcv_queue) == NULL ||
cJSON_AddNumberToObject(object, "snd_buffer", (double) stats->snd_buffer) == NULL ||
cJSON_AddNumberToObject(object, "out_segs_total", (double) stats->out_segs_total) == NULL ||
cJSON_AddNumberToObject(object, "retrans_total", (double) stats->retrans_total) == NULL ||
cJSON_AddNumberToObject(object, "fast_retrans_total", (double) stats->fast_retrans_total) == NULL ||
cJSON_AddNumberToObject(object, "lost_total", (double) stats->lost_total) == NULL ||
cJSON_AddNumberToObject(object, "repeat_total", (double) stats->repeat_total) == NULL ||
cJSON_AddNumberToObject(object, "xmit_total", (double) stats->xmit_total) == NULL) {
errno = ENOMEM;
return -1;
}
return 0;
}
static int kcp_hub_build_telemetry_payload_locked(kcp_hub_t *hub, char **out_payload) {
cJSON *root = NULL;
cJSON *sessions = NULL;
char *ts_unix_nano_text = NULL;
char *payload = NULL;
kcp_peer_entry_t *entry;
if (hub == NULL || out_payload == NULL) {
errno = EINVAL;
return -1;
}
*out_payload = NULL;
root = cJSON_CreateObject();
if (root == NULL) {
errno = ENOMEM;
return -1;
}
sessions = cJSON_AddArrayToObject(root, "sessions");
if (sessions == NULL) {
cJSON_Delete(root);
errno = ENOMEM;
return -1;
}
ts_unix_nano_text = omni_strdup_printf("%" PRId64, omni_now_unix_nano());
if (ts_unix_nano_text == NULL) {
cJSON_Delete(root);
return -1;
}
if (cJSON_AddStringToObject(root, "type", "hub_kcp_snapshot") == NULL ||
cJSON_AddStringToObject(root, "ts_unix_nano", ts_unix_nano_text) == NULL ||
cJSON_AddStringToObject(root, "node_id", KCP_HUB_DEFAULT_NODE_ID) == NULL) {
free(ts_unix_nano_text);
cJSON_Delete(root);
errno = ENOMEM;
return -1;
}
free(ts_unix_nano_text);
for (entry = hub->peers; entry != NULL; entry = entry->next) {
cJSON *session = NULL;
kcp_runtime_stats_t stats;
struct sockaddr_storage local_addr;
struct sockaddr_storage remote_addr;
socklen_t local_len = sizeof(local_addr);
socklen_t remote_len = sizeof(remote_addr);
char local_text[OMNI_MAX_ADDR_TEXT] = "";
char remote_text[OMNI_MAX_ADDR_TEXT] = "";
if (entry->conn == NULL || entry->peer_id[0] == '\0' || kcp_hub_peer_is_telemetry(entry->peer_id)) {
continue;
}
memset(&stats, 0, sizeof(stats));
kcp_conn_runtime_stats_snapshot(entry->conn, &stats);
if (kcp_conn_local_addr(entry->conn, &local_addr, &local_len) != 0) {
local_len = 0;
}
if (kcp_conn_remote_addr(entry->conn, &remote_addr, &remote_len) != 0) {
remote_len = 0;
}
if (local_len > 0) {
omni_sockaddr_to_string((const struct sockaddr *) &local_addr, local_len, local_text, sizeof(local_text));
}
if (remote_len > 0) {
omni_sockaddr_to_string((const struct sockaddr *) &remote_addr, remote_len, remote_text, sizeof(remote_text));
}
session = cJSON_CreateObject();
if (session == NULL) {
cJSON_Delete(root);
errno = ENOMEM;
return -1;
}
cJSON_AddItemToArray(sessions, session);
if (cJSON_AddStringToObject(session, "peer_id", entry->peer_id) == NULL ||
cJSON_AddStringToObject(session, "local_addr", local_text) == NULL ||
cJSON_AddStringToObject(session, "remote_addr", remote_text) == NULL ||
kcp_hub_add_runtime_stats_json(session, &stats) != 0) {
cJSON_Delete(root);
errno = ENOMEM;
return -1;
}
}
payload = cJSON_PrintUnformatted(root);
cJSON_Delete(root);
if (payload == NULL) {
errno = ENOMEM;
return -1;
}
*out_payload = payload;
return 0;
}
static int kcp_hub_push_telemetry_snapshot(kcp_hub_t *hub) {
message_t msg;
char *payload = NULL;
char telemetry_peer_id[OMNI_MAX_PEER_ID];
int rc;
if (hub == NULL) {
errno = EINVAL;
return -1;
}
pthread_rwlock_rdlock(&hub->lock);
if (hub->telemetry_peer_id[0] == '\0' || kcp_hub_find_peer(hub, hub->telemetry_peer_id) == NULL) {
pthread_rwlock_unlock(&hub->lock);
return 0;
}
snprintf(telemetry_peer_id, sizeof(telemetry_peer_id), "%s", hub->telemetry_peer_id);
rc = kcp_hub_build_telemetry_payload_locked(hub, &payload);
pthread_rwlock_unlock(&hub->lock);
if (rc != 0) {
return -1;
}
protocol_message_init(&msg);
msg.type = MSG_TYPE_TEXT;
snprintf(msg.from, sizeof(msg.from), "%s", SERVER_PEER_ID);
snprintf(msg.to, sizeof(msg.to), "%s", telemetry_peer_id);
msg.body = (uint8_t *) omni_strdup(payload == NULL ? "" : payload);
cJSON_free(payload);
if (msg.body == NULL) {
return -1;
}
msg.body_len = strlen((const char *) msg.body);
rc = kcp_hub_deliver_to_local_peer(hub, &msg);
protocol_message_clear(&msg);
if (rc != 0 && errno == ENOENT) {
return 0;
}
return rc;
}
static void *kcp_hub_telemetry_thread_main(void *arg) {
kcp_hub_t *hub = (kcp_hub_t *) arg;
while (!atomic_load(&hub->closed)) {
int interval_ms = KCP_HUB_DEFAULT_TELEMETRY_INTERVAL_MS;
pthread_rwlock_rdlock(&hub->lock);
if (hub->telemetry_interval_ms > 0) {
interval_ms = hub->telemetry_interval_ms;
}
pthread_rwlock_unlock(&hub->lock);
(void) kcp_hub_push_telemetry_snapshot(hub);
if (atomic_load(&hub->closed)) {
break;
}
usleep((useconds_t) interval_ms * 1000U);
}
return NULL;
}
static int kcp_hub_send_server_error(kcp_conn_t *conn, const char *to, const char *message) {
message_t msg;
protocol_message_init(&msg);
@@ -389,11 +602,6 @@ static int kcp_hub_register_conn(kcp_hub_t *hub, kcp_conn_t *conn, char *peer_id
pthread_rwlock_unlock(&hub->lock);
snprintf(peer_id, peer_id_len, "%s", msg.from);
if (kcp_hub_configure_peer_transport(conn, peer_id) != 0) {
kcp_hub_unregister(hub, peer_id, conn);
protocol_message_clear(&msg);
return -1;
}
protocol_message_clear(&msg);
return 0;
}
@@ -414,7 +622,9 @@ kcp_hub_t *kcp_hub_new(latency_logger_t *logger, kcp_session_stats_logger_t *sta
hub->logger = logger;
hub->stats_logger = stats_logger;
hub->stats_interval_ms = stats_interval_ms > 0 ? stats_interval_ms : KCP_DEFAULT_STATS_INTERVAL_MS;
hub->telemetry_interval_ms = KCP_HUB_DEFAULT_TELEMETRY_INTERVAL_MS;
hub->relay_fd = -1;
atomic_init(&hub->closed, 0);
return hub;
}
@@ -423,13 +633,13 @@ int kcp_hub_serve_listener(kcp_hub_t *hub, kcp_listener_t *listener) {
errno = EINVAL;
return -1;
}
while (!hub->closed) {
while (!atomic_load(&hub->closed)) {
kcp_conn_t *conn = kcp_listener_accept(listener);
kcp_session_thread_ctx_t *ctx;
pthread_t thread;
if (conn == NULL) {
if (hub->closed) {
if (atomic_load(&hub->closed)) {
return 0;
}
return -1;
@@ -455,6 +665,7 @@ int kcp_hub_serve_listener(kcp_hub_t *hub, kcp_listener_t *listener) {
int kcp_hub_serve_session(kcp_hub_t *hub, kcp_conn_t *conn) {
char peer_id[OMNI_MAX_PEER_ID];
const char *node_id;
int rc = 0;
if (hub == NULL || conn == NULL) {
@@ -462,12 +673,20 @@ int kcp_hub_serve_session(kcp_hub_t *hub, kcp_conn_t *conn) {
return -1;
}
peer_id[0] = '\0';
if (kcp_conn_configure_runtime(conn, hub->logger, OMNI_NODE_ROLE_SERVER, "hub", hub->stats_logger, hub->stats_interval_ms) != 0) {
if (kcp_hub_register_conn(hub, conn, peer_id, sizeof(peer_id)) != 0) {
kcp_conn_close(conn);
kcp_conn_free(conn);
return -1;
}
if (kcp_hub_register_conn(hub, conn, peer_id, sizeof(peer_id)) != 0) {
if (kcp_hub_configure_peer_transport(conn, peer_id) != 0) {
kcp_hub_unregister(hub, peer_id, conn);
kcp_conn_close(conn);
kcp_conn_free(conn);
return -1;
}
node_id = kcp_hub_peer_node_id(peer_id);
if (kcp_conn_configure_runtime(conn, hub->logger, OMNI_NODE_ROLE_SERVER, node_id, hub->stats_logger, hub->stats_interval_ms) != 0) {
kcp_hub_unregister(hub, peer_id, conn);
kcp_conn_close(conn);
kcp_conn_free(conn);
return -1;
@@ -512,6 +731,34 @@ int kcp_hub_set_relay(kcp_hub_t *hub, int relay_fd, const struct sockaddr *peer_
return 0;
}
int kcp_hub_set_telemetry(kcp_hub_t *hub, const char *peer_id, int interval_ms) {
int start_thread = 0;
if (hub == NULL || peer_id == NULL) {
errno = EINVAL;
return -1;
}
pthread_rwlock_wrlock(&hub->lock);
snprintf(hub->telemetry_peer_id, sizeof(hub->telemetry_peer_id), "%s", peer_id);
hub->telemetry_interval_ms = interval_ms > 0 ? interval_ms : KCP_HUB_DEFAULT_TELEMETRY_INTERVAL_MS;
if (!hub->telemetry_thread_started && hub->telemetry_peer_id[0] != '\0') {
start_thread = 1;
hub->telemetry_thread_started = 1;
}
pthread_rwlock_unlock(&hub->lock);
if (start_thread) {
if (pthread_create(&hub->telemetry_thread, NULL, kcp_hub_telemetry_thread_main, hub) != 0) {
pthread_rwlock_wrlock(&hub->lock);
hub->telemetry_thread_started = 0;
hub->telemetry_peer_id[0] = '\0';
pthread_rwlock_unlock(&hub->lock);
return -1;
}
}
return 0;
}
int kcp_hub_serve_relay(kcp_hub_t *hub) {
uint8_t buffer[KCP_RELAY_MAX_DATAGRAM_SIZE];
@@ -519,7 +766,7 @@ int kcp_hub_serve_relay(kcp_hub_t *hub) {
errno = EINVAL;
return -1;
}
while (!hub->closed) {
while (!atomic_load(&hub->closed)) {
struct sockaddr_storage source;
socklen_t source_len = sizeof(source);
ssize_t n;
@@ -537,7 +784,7 @@ int kcp_hub_serve_relay(kcp_hub_t *hub) {
n = recvfrom(relay_fd, buffer, sizeof(buffer), 0, (struct sockaddr *) &source, &source_len);
if (n < 0) {
if (hub->closed) {
if (atomic_load(&hub->closed)) {
return 0;
}
if (errno == EINTR) {
@@ -568,8 +815,7 @@ int kcp_hub_close(kcp_hub_t *hub) {
if (hub == NULL) {
return 0;
}
if (!hub->closed) {
hub->closed = 1;
if (!atomic_exchange(&hub->closed, 1)) {
if (hub->relay_fd >= 0) {
close(hub->relay_fd);
hub->relay_fd = -1;
@@ -586,6 +832,9 @@ void kcp_hub_free(kcp_hub_t *hub) {
return;
}
kcp_hub_close(hub);
if (hub->telemetry_thread_started) {
pthread_join(hub->telemetry_thread, NULL);
}
for (entry = hub->peers; entry != NULL; entry = next) {
next = entry->next;
if (entry->conn != NULL) {