debug: 增加终端日志测试连通性

This commit is contained in:
2026-03-30 14:58:55 +08:00
parent 7a8bc4ea33
commit 9cfd2815b8

View File

@@ -36,6 +36,7 @@ typedef struct kcp_socket_debug_state {
uint32_t next_tx_id; uint32_t next_tx_id;
kcp_packet_debug_pending_t *pending_head; kcp_packet_debug_pending_t *pending_head;
atomic_int closed; atomic_int closed;
atomic_int last_send_errno;
} kcp_socket_debug_state_t; } kcp_socket_debug_state_t;
typedef struct kcp_session_entry kcp_session_entry_t; typedef struct kcp_session_entry kcp_session_entry_t;
@@ -806,17 +807,23 @@ static int kcp_socket_send_packet(kcp_socket_debug_state_t *state, const struct
uint32_t tx_id = 0; uint32_t tx_id = 0;
ssize_t rc; ssize_t rc;
if (state->logger != NULL && kcp_socket_debug_reserve_tx(state, remote_addr, remote_addr_len, packet, packet_len, &tx_id) != 0) { if (state->logger != NULL && kcp_socket_debug_reserve_tx(state, remote_addr, remote_addr_len, packet, packet_len, &tx_id) != 0) {
atomic_store(&state->last_send_errno, errno != 0 ? errno : EIO);
return -1; return -1;
} }
pthread_mutex_lock(&state->write_mu); pthread_mutex_lock(&state->write_mu);
rc = sendto(state->fd, packet, packet_len, 0, (const struct sockaddr *) remote_addr, remote_addr_len); rc = sendto(state->fd, packet, packet_len, 0, (const struct sockaddr *) remote_addr, remote_addr_len);
pthread_mutex_unlock(&state->write_mu); pthread_mutex_unlock(&state->write_mu);
if (rc < 0 || (size_t) rc != packet_len) { if (rc < 0 || (size_t) rc != packet_len) {
if (rc >= 0 && (size_t) rc != packet_len && errno == 0) {
errno = EIO;
}
atomic_store(&state->last_send_errno, errno != 0 ? errno : EIO);
if (state->logger != NULL) { if (state->logger != NULL) {
kcp_socket_debug_rollback_tx(state, tx_id); kcp_socket_debug_rollback_tx(state, tx_id);
} }
return -1; return -1;
} }
atomic_store(&state->last_send_errno, 0);
return 0; return 0;
} }
@@ -1109,10 +1116,13 @@ static void *kcp_update_thread_main(void *arg) {
} }
static int kcp_conn_start_stats_thread(kcp_conn_t *conn) { static int kcp_conn_start_stats_thread(kcp_conn_t *conn) {
int thread_rc;
if (conn == NULL || conn->stats_logger == NULL || conn->stats_thread_started) { if (conn == NULL || conn->stats_logger == NULL || conn->stats_thread_started) {
return 0; return 0;
} }
if (pthread_create(&conn->stats_thread, NULL, kcp_stats_thread_main, conn) != 0) { thread_rc = pthread_create(&conn->stats_thread, NULL, kcp_stats_thread_main, conn);
if (thread_rc != 0) {
errno = thread_rc;
return -1; return -1;
} }
conn->stats_thread_started = 1; conn->stats_thread_started = 1;
@@ -1122,8 +1132,10 @@ static int kcp_conn_start_stats_thread(kcp_conn_t *conn) {
static kcp_conn_t *kcp_conn_alloc_common(int fd, const struct sockaddr_storage *remote_addr, socklen_t remote_addr_len, kcp_socket_debug_state_t *sock_state, latency_logger_t *logger, const char *node_role, const char *node_id, kcp_session_stats_logger_t *stats_logger, int stats_interval_ms) { static kcp_conn_t *kcp_conn_alloc_common(int fd, const struct sockaddr_storage *remote_addr, socklen_t remote_addr_len, kcp_socket_debug_state_t *sock_state, latency_logger_t *logger, const char *node_role, const char *node_id, kcp_session_stats_logger_t *stats_logger, int stats_interval_ms) {
kcp_conn_t *conn = (kcp_conn_t *) calloc(1, sizeof(*conn)); kcp_conn_t *conn = (kcp_conn_t *) calloc(1, sizeof(*conn));
uint32_t conv; uint32_t conv;
int thread_rc;
if (conn == NULL) { if (conn == NULL) {
errno = ENOMEM;
return NULL; return NULL;
} }
conn->fd = fd; conn->fd = fd;
@@ -1149,6 +1161,7 @@ static kcp_conn_t *kcp_conn_alloc_common(int fd, const struct sockaddr_storage *
} }
conn->kcp = ikcp_create(conv, conn); conn->kcp = ikcp_create(conv, conn);
if (conn->kcp == NULL) { if (conn->kcp == NULL) {
errno = ENOMEM;
protocol_frame_decoder_destroy(&conn->decoder); protocol_frame_decoder_destroy(&conn->decoder);
pthread_cond_destroy(&conn->rx_cond); pthread_cond_destroy(&conn->rx_cond);
pthread_mutex_destroy(&conn->kcp_mu); pthread_mutex_destroy(&conn->kcp_mu);
@@ -1180,7 +1193,9 @@ static kcp_conn_t *kcp_conn_alloc_common(int fd, const struct sockaddr_storage *
free(conn); free(conn);
return NULL; return NULL;
} }
if (pthread_create(&conn->update_thread, NULL, kcp_update_thread_main, conn) != 0) { thread_rc = pthread_create(&conn->update_thread, NULL, kcp_update_thread_main, conn);
if (thread_rc != 0) {
errno = thread_rc;
if (conn->stats_thread_started) { if (conn->stats_thread_started) {
atomic_store(&conn->closed, 1); atomic_store(&conn->closed, 1);
pthread_join(conn->stats_thread, NULL); pthread_join(conn->stats_thread, NULL);
@@ -1212,6 +1227,7 @@ kcp_conn_t *kcp_conn_dial(const char *server_addr, const char *bind_ip, const ch
} }
sock_state = (kcp_socket_debug_state_t *) calloc(1, sizeof(*sock_state)); sock_state = (kcp_socket_debug_state_t *) calloc(1, sizeof(*sock_state));
if (sock_state == NULL) { if (sock_state == NULL) {
errno = ENOMEM;
close(fd); close(fd);
return NULL; return NULL;
} }
@@ -1524,6 +1540,7 @@ int kcp_conn_configure_runtime(kcp_conn_t *conn, latency_logger_t *logger, const
int kcp_conn_send(kcp_conn_t *conn, const message_t *msg) { int kcp_conn_send(kcp_conn_t *conn, const message_t *msg) {
uint8_t *frame = NULL; uint8_t *frame = NULL;
size_t frame_len = 0; size_t frame_len = 0;
int send_errno = 0;
if (conn == NULL || msg == NULL) { if (conn == NULL || msg == NULL) {
errno = EINVAL; errno = EINVAL;
return -1; return -1;
@@ -1535,14 +1552,22 @@ int kcp_conn_send(kcp_conn_t *conn, const message_t *msg) {
kcp_log_session_snapshot(conn, "send_handoff_begin"); kcp_log_session_snapshot(conn, "send_handoff_begin");
kcp_process_sampler_request_sample(conn->process_sampler, "send_handoff_begin"); kcp_process_sampler_request_sample(conn->process_sampler, "send_handoff_begin");
pthread_mutex_lock(&conn->kcp_mu); pthread_mutex_lock(&conn->kcp_mu);
atomic_store(&conn->sock_state->last_send_errno, 0);
conn->kcp->current = omni_now_millis32(); conn->kcp->current = omni_now_millis32();
if (ikcp_send(conn->kcp, (const char *) frame, (int) frame_len) != 0) { if (ikcp_send(conn->kcp, (const char *) frame, (int) frame_len) != 0) {
pthread_mutex_unlock(&conn->kcp_mu); pthread_mutex_unlock(&conn->kcp_mu);
errno = EIO;
free(frame); free(frame);
return -1; return -1;
} }
ikcp_flush(conn->kcp); ikcp_flush(conn->kcp);
send_errno = atomic_load(&conn->sock_state->last_send_errno);
pthread_mutex_unlock(&conn->kcp_mu); pthread_mutex_unlock(&conn->kcp_mu);
if (send_errno != 0) {
errno = send_errno;
free(frame);
return -1;
}
kcp_log_session_snapshot(conn, "send_handoff_end"); kcp_log_session_snapshot(conn, "send_handoff_end");
kcp_process_sampler_request_sample(conn->process_sampler, "send_handoff_end"); kcp_process_sampler_request_sample(conn->process_sampler, "send_handoff_end");
latencylog_log_message_event(conn->logger, conn->node_role, conn->node_id, EVENT_SEND_HANDOFF_END, msg); latencylog_log_message_event(conn->logger, conn->node_role, conn->node_id, EVENT_SEND_HANDOFF_END, msg);