From 4914348557abc45caa33d6f4fcdd5a1a0857850a Mon Sep 17 00:00:00 2001 From: Evgeny Date: Sun, 5 Jul 2026 12:57:41 +0300 Subject: [PATCH] =?UTF-8?q?etcp=5Frouter:=20=D1=82=D1=80=D0=B0=D0=BD=D0=B7?= =?UTF-8?q?=D0=B8=D1=82=D0=BD=D1=8B=D0=B5=20=D0=BE=D1=87=D0=B5=D1=80=D0=B5?= =?UTF-8?q?=D0=B4=D0=B8=20per=20(src,dst)=20=D1=81=20backpressure=20+=20?= =?UTF-8?q?=D1=83=D0=BD=D0=B8=D1=84=D0=B8=D0=BA=D0=B0=D1=86=D0=B8=D1=8F=20?= =?UTF-8?q?etcp=5Fsend?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - TRANSIT_QUEUE: отдельная очередь на каждую пару src+dst узлов - ll_queue с хеш-индексом для быстрого поиска transit-очередей - backpressure через waiter на send_input_q (threshold=0, один пакет за вызов, round-robin) - Очередь создаётся при первом пакете который не получается отправить напрямую, удаляется при опустошении - Унифицирован etcp_send: единый путь queue_data_put(send_input_q) для UDP (normalizer->input) и STCP (tx_queue) - Убран STCP-ветвления из router_forward_transit/drain_cb - Исправлено перекрытие ll.data и tq->q в TRANSIT_QUEUE (добавлены явные поля src_node_id/dst_node_id) - Фикс лика dgram в tx_queue_cb/client_tx_queue_cb (добавлен queue_dgram_free) - transit_queues живут внутри ETCP_CONN, инициализируются лениво, очищаются в etcp_connection_close (до pn_deinit) и stcp_link_close --- src/control_server.c | 15 ++ src/etcp.c | 9 +- src/etcp.h | 2 + src/etcp_api.c | 10 +- src/etcp_router.c | 264 +++++++++++++++++++++++++++++-- src/etcp_router.h | 47 +++++- src/pkt_normalizer.c | 2 + src/stcp_client.c | 1 + src/stcp_link.c | 22 +++ src/stcp_server.c | 1 + tools/etcpmon/etcpmon_gui.c | 20 ++- tools/etcpmon/etcpmon_gui.h | 8 + tools/etcpmon/etcpmon_protocol.h | 4 + 13 files changed, 379 insertions(+), 26 deletions(-) diff --git a/src/control_server.c b/src/control_server.c index 9c2acc8e..4c15291e 100644 --- a/src/control_server.c +++ b/src/control_server.c @@ -1196,6 +1196,7 @@ static void send_metrics(struct control_server* server, struct control_client* c /* Fill router congestion metrics — aggregate over all router_conns */ memset(&rsp->router, 0, sizeof(rsp->router)); if (instance->router_conns) { + uint32_t rtt_cnt = 0; struct ll_entry* re = instance->router_conns->head; while (re) { struct ETCP_ROUTER_CONN* rc = (struct ETCP_ROUTER_CONN*)re; @@ -1211,8 +1212,22 @@ static void send_metrics(struct control_server* server, struct control_client* c rsp->router.oob_dropped += rc->c_oob_dropped; rsp->router.stale_ack += rc->c_stale_ack; rsp->router.sign_fail += rc->c_sign_fail; + if (rc->rtt > 0) { + if (rc->rtt > rsp->router.rtt_last) rsp->router.rtt_last = rc->rtt; + rsp->router.rtt_avg += rc->rtt; + rtt_cnt++; + } + { + uint16_t j = (uint16_t)(rc->rtt_jitter / 32768); + if (j > rsp->router.jitter) rsp->router.jitter = j; + } + if (rc->minrtt_window_count > 0) { + if (rsp->router.minrtt == 0 || rc->minrtt < rsp->router.minrtt) + rsp->router.minrtt = rc->minrtt; + } re = re->next; } + if (rtt_cnt > 0) rsp->router.rtt_avg /= rtt_cnt; rsp->router.total_conns = (uint32_t)queue_entry_count(instance->router_conns); } diff --git a/src/etcp.c b/src/etcp.c index 1de4cb01..d2b6abd2 100644 --- a/src/etcp.c +++ b/src/etcp.c @@ -3,6 +3,7 @@ #include "etcp.h" #include "etcp_debug.h" #include "etcp_loadbalancer.h" +#include "etcp_router.h" #include "routing.h" #include "route_bgp.h" #include "route_ping.h" @@ -202,6 +203,7 @@ struct ETCP_CONN* etcp_connection_create(struct UTUN_INSTANCE* instance, char* n etcp->input_queue = queue_new(instance->ua, 0, 0, 0, "ETCP input"); // No hash for input_queue queue_set_threshold(etcp->input_queue, 0, 0); // Backpressure: ждать полного освобождения etcp->output_queue = queue_new(instance->ua, 0, 0, 0, "ETCP output"); // No hash for output_queue + etcp->transit_queues = queue_new(instance->ua, TRANSIT_QUEUE_HASH, 0, 16, "transit_q_reg"); // Hash for transit queues etcp->input_send_q = queue_new(instance->ua, INFLIGHT_INITIAL_HASH_SIZE, 0, 4, "input_send_q"); // Hash for send_q etcp->input_wait_ack = queue_new(instance->ua, INFLIGHT_INITIAL_HASH_SIZE, 0, 4, "input_wait_ack"); // Hash for wait_ack etcp->recv_q = queue_new(instance->ua, INFLIGHT_INITIAL_HASH_SIZE, 0, 4, "recv_q"); // Hash for recv_q @@ -217,7 +219,8 @@ struct ETCP_CONN* etcp_connection_create(struct UTUN_INSTANCE* instance, char* n // Initialize log_name with local node_id (peer will be updated later when known) snprintf(etcp->log_name, sizeof(etcp->log_name), "%04X->???? [%s]", (uint16_t)instance->node_id, etcp->name); - if (!etcp->input_queue || !etcp->output_queue || !etcp->input_send_q || !etcp->recv_q || !etcp->ack_q || + if (!etcp->input_queue || !etcp->output_queue || !etcp->transit_queues || + !etcp->input_send_q || !etcp->recv_q || !etcp->ack_q || !etcp->input_wait_ack || !etcp->inflight_pool || !etcp->io_pool) { DEBUG_ERROR(DEBUG_CATEGORY_CONNECTION, "etcp_connection_create: error - closing - input:%p output:%p send:%p wait:%x pool:%p", etcp->input_queue, etcp->output_queue, etcp->input_send_q, etcp->input_wait_ack, etcp->inflight_pool); @@ -240,6 +243,7 @@ struct ETCP_CONN* etcp_connection_create(struct UTUN_INSTANCE* instance, char* n etcp_connection_close(etcp); return NULL; } + etcp->send_input_q = etcp->normalizer->input; // Set input queue callback queue_set_callback(etcp->input_queue, input_queue_cb, etcp); @@ -298,6 +302,9 @@ void etcp_connection_close(struct ETCP_CONN* etcp) { // NOTE: route_bgp_remove_conn is already called via down_cbk in etcp_on_down() above // DO NOT call it again here to avoid double-processing and ref_count corruption + // Очистить транзитные очереди (до pn_deinit/stcp_link_close — waiter'ы на normalizer->input/tx_queue) + etcp_router_transit_queues_destroy(etcp); + // Deinitialize packet normalizer (this will call routing_del_conn) if (etcp->normalizer) { pn_deinit((struct PKTNORM*)etcp->normalizer); diff --git a/src/etcp.h b/src/etcp.h index a8389bef..4b5f7ebc 100644 --- a/src/etcp.h +++ b/src/etcp.h @@ -130,6 +130,8 @@ struct ETCP_CONN { void (*link_ready_for_send_fn)(struct ETCP_CONN*);// функцию которую должен вызвать драйвер линка при готовности линка принимать данные struct ll_queue* output_queue; // Assembled outgoing packets (storage: ETCP_FRAGMENT / rx_pool) + struct ll_queue* transit_queues; // hash по (src_node_id:8, dst_node_id:8) — транзитные очереди + struct ll_queue* send_input_q; // единая входная очередь отправки (normalizer->input или tx_queue) // IDs and state diff --git a/src/etcp_api.c b/src/etcp_api.c index 62b3922d..28dcaba2 100644 --- a/src/etcp_api.c +++ b/src/etcp_api.c @@ -6,7 +6,6 @@ #include "etcp.h" #include "pkt_normalizer.h" #include "utun_instance.h" -#include "stcp_link.h" #include "../lib/debug_config.h" #include @@ -45,13 +44,8 @@ int etcp_unbind(struct UTUN_INSTANCE* inst, uint8_t id) { } int etcp_send(struct ETCP_CONN* conn, struct ll_entry* entry) { - if (!conn || !entry) return -1; - // TCP (stcp_link) — send directly - if (conn->transport_link) - return stcp_link_send((struct stcp_link *)conn->transport_link, entry->dgram, entry->len); - // UDP — через normalizer - if (!conn->normalizer || !conn->normalizer->input) return -1; - return queue_data_put(conn->normalizer->input, entry); + if (!conn || !entry || !conn->send_input_q) return -1; + return queue_data_put(conn->send_input_q, entry); } void etcp_int_recv(struct ll_queue* queue, void* arg) { diff --git a/src/etcp_router.c b/src/etcp_router.c index 58fb44c0..9fb34a0d 100644 --- a/src/etcp_router.c +++ b/src/etcp_router.c @@ -41,7 +41,7 @@ static void router_close_and_notify(struct ETCP_ROUTER_CONN* rconn); static void router_close_finalize(void* arg); static void free_entry(struct ll_entry* e) { queue_dgram_free(e); queue_entry_free(e); } static int router_verify_signature(struct UTUN_INSTANCE* inst, struct ll_entry* entry, struct SVC_ROUTE_HDR* hdr, size_t* pl_len); -static void router_handle_ack(struct UTUN_INSTANCE* inst, struct ETCP_ROUTER_CONN* rconn, uint32_t seq, uint64_t src_node_id); +static void router_handle_ack(struct UTUN_INSTANCE* inst, struct ETCP_ROUTER_CONN* rconn, uint32_t seq, uint64_t src_node_id, uint16_t ack_ts); static void router_forward_transit(struct UTUN_INSTANCE* inst, struct ll_entry* entry, struct SVC_ROUTE_HDR* hdr); static int router_check_peer_restart(struct UTUN_INSTANCE* inst, struct ETCP_ROUTER_CONN** prconn, struct SVC_ROUTE_HDR* hdr); static void router_handle_data_packet(struct ETCP_ROUTER_CONN* rconn, struct ETCP_CONN* conn, uint32_t seq, const uint8_t* pl, size_t pl_len); @@ -49,6 +49,87 @@ static void router_retransmit_one(struct ETCP_ROUTER_CONN* rconn, struct ROUTER_ static void router_retrans_schedule(struct ETCP_ROUTER_CONN* rconn); static void router_retrans_timer_cb(void* arg); static void router_incoming_q_cb(struct ll_queue* q, void* arg); +static void router_track_inflight_state(struct ETCP_ROUTER_CONN* rconn); +static uint32_t router_effective_max_inflight(struct ETCP_ROUTER_CONN* rconn); +static void router_update_inflight_limit(struct ETCP_ROUTER_CONN* rconn); +static void router_update_minrtt(struct ETCP_ROUTER_CONN* rconn); + +// ==================================================================== +// Транзитные очереди — per (src, dst) pair, backpressure через waiter на send_input_q + +static struct TRANSIT_QUEUE* transit_queue_find(struct ETCP_CONN* conn, + uint64_t src_node_id, uint64_t dst_node_id) { + if (!conn->transit_queues) return NULL; + uint8_t key[16]; + memcpy(key, &src_node_id, 8); + memcpy(key + 8, &dst_node_id, 8); + return (struct TRANSIT_QUEUE*)queue_find_data_by_index(conn->transit_queues, key); +} + +static struct TRANSIT_QUEUE* transit_queue_get_or_create(struct ETCP_CONN* conn, + uint64_t src_node_id, uint64_t dst_node_id) { + if (!conn->transit_queues) { + struct UASYNC* ua = conn->instance ? conn->instance->ua : NULL; + if (!ua) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "transit_queue_get_or_create: no ua"); return NULL; } + conn->transit_queues = queue_new(ua, TRANSIT_QUEUE_HASH, 0, 16, "transit_q_reg"); + if (!conn->transit_queues) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "transit_queue_get_or_create: queue_new failed"); return NULL; } + } + struct TRANSIT_QUEUE* tq = transit_queue_find(conn, src_node_id, dst_node_id); + if (tq) return tq; + size_t data_size = sizeof(struct TRANSIT_QUEUE) - sizeof(struct ll_entry); + tq = (struct TRANSIT_QUEUE*)queue_entry_new(data_size); + if (!tq) { + DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "transit_queue_get_or_create: queue_entry_new failed"); + return NULL; + } + memset(&tq->waiter, 0, sizeof(tq->waiter)); + tq->conn = conn; + tq->q = queue_new(conn->instance->ua, 0, 0, 0, "transit_pair_q"); + if (!tq->q) { queue_entry_free(&tq->ll); return NULL; } + memcpy(tq->ll.data, &src_node_id, 8); + memcpy(tq->ll.data + 8, &dst_node_id, 8); + queue_data_put_with_index(conn->transit_queues, &tq->ll); + DEBUG_INFO(DEBUG_CATEGORY_ETCPROUTE, "transit_q: new src=%016llx dst=%016llx", + (unsigned long long)src_node_id, (unsigned long long)dst_node_id); + return tq; +} + +static void transit_queue_destroy(struct ETCP_CONN* conn, struct TRANSIT_QUEUE* tq) { + if (conn->send_input_q) queue_waiter_cancel(conn->send_input_q, &tq->waiter); + struct ll_entry* e; + while ((e = queue_data_get(tq->q)) != NULL) free_entry(e); + queue_free(tq->q); + queue_remove_data(conn->transit_queues, &tq->ll); + queue_entry_free(&tq->ll); +} + +static void transit_queue_drain_cb(struct ll_queue* q, void* arg) { + struct TRANSIT_QUEUE* tq = (struct TRANSIT_QUEUE*)arg; + struct ETCP_CONN* conn = tq->conn; + struct ll_entry* e = queue_data_get(tq->q); + if (!e) { transit_queue_destroy(conn, tq); return; } + int ret = etcp_send(conn, e); + if (ret != 0) free_entry(e); + if (queue_entry_count(tq->q) > 0) + queue_waiter_wait(conn->send_input_q, &tq->waiter, transit_queue_drain_cb, tq); + else + transit_queue_destroy(conn, tq); +} + +void etcp_router_transit_queues_destroy(struct ETCP_CONN* conn) { + if (!conn || !conn->transit_queues) return; + struct ll_entry* entry; + while ((entry = queue_data_get(conn->transit_queues)) != NULL) { + struct TRANSIT_QUEUE* tq = (struct TRANSIT_QUEUE*)entry; + if (conn->send_input_q) queue_waiter_cancel(conn->send_input_q, &tq->waiter); + struct ll_entry* e; + while ((e = queue_data_get(tq->q)) != NULL) free_entry(e); + queue_free(tq->q); + queue_entry_free(&tq->ll); + } + queue_free(conn->transit_queues); + conn->transit_queues = NULL; +} // ==================================================================== // Управление ROUTER_CONN @@ -93,6 +174,7 @@ static int router_send_one_flags(struct ETCP_ROUTER_CONN* rconn, const uint8_t* if (is_signed) f |= ROUTER_FLAG_SIGNED; hdr->flags = f; } + hdr->timestamp = get_current_timestamp(); if (pl_len > 0) memcpy(dgram + SVC_ROUTE_HDR_SIZE, payload, pl_len); struct ETCP_CONN* conn = NULL; @@ -170,6 +252,7 @@ static int router_send_one_flags(struct ETCP_ROUTER_CONN* rconn, const uint8_t* if (!rconn->retrans_timer) router_retrans_schedule(rconn); } } + if (!flag_bits) router_track_inflight_state(rconn); } return ret; } @@ -186,6 +269,7 @@ static void router_send_to(struct UTUN_INSTANCE* inst, uint64_t dst, uint8_t svc hdr->seq = 0; hdr->flags = flag_bits; hdr->svc_id = svc_id; + hdr->timestamp = get_current_timestamp(); struct ll_entry* entry = queue_entry_new(0); if (!entry) { u_free(hdr); return; } entry->dgram = (uint8_t*)hdr; @@ -276,14 +360,14 @@ static void router_drain_send_q(struct ETCP_ROUTER_CONN* rconn) { if (sq > 0) DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "SEND_Q_DRAIN: svc_id=%u send_q=%d inflight=%d", rconn->svc_id, sq, (int32_t)(rconn->tx_seq - rconn->tx_acked)); while (1) { - if ((int32_t)(rconn->tx_seq - rconn->tx_acked) >= ROUTER_MAX_INFLIGHT) break; + if ((int32_t)(rconn->tx_seq - rconn->tx_acked) >= (int32_t)router_effective_max_inflight(rconn)) break; struct ll_entry* e = queue_data_get(rconn->send_q); if (!e) { rconn->send_blocked = 0; break; } u_check(e, "drain:after_get", "router"); int s_err = router_send_one(rconn, e->dgram, e->len); u_check(e, "drain:after_send", "router"); queue_dgram_free(e); queue_entry_free(e); - if (s_err != 0 && (int32_t)(rconn->tx_seq - rconn->tx_acked) >= ROUTER_MAX_INFLIGHT) break; + if (s_err != 0 && (int32_t)(rconn->tx_seq - rconn->tx_acked) >= (int32_t)router_effective_max_inflight(rconn)) break; } if (rconn->send_blocked && !rconn->send_resume_timer) rconn->send_resume_timer = uasync_set_timeout(rconn->inst->ua, @@ -333,6 +417,7 @@ static void router_retransmit_one(struct ETCP_ROUTER_CONN* rconn, struct ROUTER_ hdr->seq = inf->seq; hdr->svc_id = rconn->svc_id; hdr->flags = (rconn->sess_id << ROUTER_SESS_ID_SHIFT); + hdr->timestamp = get_current_timestamp(); struct ETCP_CONN* conn = NULL; struct ROUTE_BGP* bgp = inst->bgp; @@ -439,6 +524,90 @@ static int router_enqueue_send(struct ETCP_ROUTER_CONN* rconn, const uint8_t* pa return 0; } +// ==================================================================== +// minRTT helpers +// ==================================================================== + +static uint32_t router_effective_max_inflight(struct ETCP_ROUTER_CONN* rconn) { + return rconn->inflight_limit; +} + +static void router_update_inflight_limit(struct ETCP_ROUTER_CONN* rconn) { + uint32_t new_limit = rconn->minrtt_probe ? MINRTT_PROBE_MAX_INFLIGHT : rconn->max_inflight; + if (new_limit == rconn->inflight_limit) return; + uint32_t old = rconn->inflight_limit; + rconn->inflight_limit = new_limit; + uint32_t inflight = rconn->tx_seq - rconn->tx_acked; + + if (new_limit < old && inflight >= new_limit) { + if (!rconn->send_blocked) rconn->send_blocked = 1; + } + if (new_limit > old && (rconn->send_blocked || queue_entry_count(rconn->send_q) > 0)) { + router_drain_send_q(rconn); + } +} + +void router_set_max_inflight(struct ETCP_ROUTER_CONN* rconn, uint16_t new_max) { + if (!rconn) return; + if (new_max < MINRTT_PROBE_MAX_INFLIGHT) new_max = MINRTT_PROBE_MAX_INFLIGHT; + rconn->max_inflight = new_max; + if (!rconn->minrtt_probe) router_update_inflight_limit(rconn); + router_track_inflight_state(rconn); +} + +static void router_track_inflight_state(struct ETCP_ROUTER_CONN* rconn) { + uint32_t inflight = rconn->tx_seq - rconn->tx_acked; + uint32_t threshold = rconn->max_inflight / 2; + uint8_t is_loaded = (inflight >= threshold); + if (is_loaded && !rconn->inflight_was_loaded) { + rconn->last_inflight_loaded_tb = get_time_tb(); + rconn->inflight_was_loaded = 1; + } else if (!is_loaded && rconn->inflight_was_loaded) { + rconn->last_inflight_unloaded_tb = get_time_tb(); + rconn->inflight_was_loaded = 0; + } +} + +static void router_update_minrtt(struct ETCP_ROUTER_CONN* rconn) { + uint64_t now = get_time_tb(); + uint16_t w_idx = rconn->minrtt_window_idx; + uint8_t w_cnt = rconn->minrtt_window_count; + + // Probe check: oldest entry freshness + uint8_t prev_probe = rconn->minrtt_probe; + if (w_cnt > 0) { + uint8_t oldest = (w_idx - w_cnt + MINRTT_WINDOW_SIZE) % MINRTT_WINDOW_SIZE; + uint64_t oldest_age = now - rconn->minrtt_window_tb[oldest]; + if (w_cnt >= MINRTT_WINDOW_SIZE && oldest_age < MINRTT_PROBE_TIMEOUT_TB) { + rconn->minrtt_probe = 0; + } else if (oldest_age >= MINRTT_PROBE_TIMEOUT_TB) { + rconn->minrtt_probe = 1; + } + } + if (prev_probe != rconn->minrtt_probe) router_update_inflight_limit(rconn); + + // Update minRTT if channel is unloaded long enough + if (!rconn->inflight_was_loaded) { + uint64_t unloaded_dur = now - rconn->last_inflight_unloaded_tb; + uint64_t min_dur = (uint64_t)rconn->rtt * 2; + if (min_dur < MINRTT_UNLOADED_MIN_TB) min_dur = MINRTT_UNLOADED_MIN_TB; + if (unloaded_dur >= min_dur) { + rconn->minrtt_window[w_idx] = rconn->rtt; + rconn->minrtt_window_tb[w_idx] = now; + rconn->minrtt_window_idx = (w_idx + 1) % MINRTT_WINDOW_SIZE; + if (w_cnt < MINRTT_WINDOW_SIZE) w_cnt++; + rconn->minrtt_window_count = w_cnt; + uint32_t sum = 0, cnt = 0; + for (int i = 0; i < MINRTT_WINDOW_SIZE; i++) { + if (rconn->minrtt_window[i] > 0) { sum += rconn->minrtt_window[i]; cnt++; } + } + rconn->minrtt = cnt > 0 ? (uint16_t)(sum / cnt) : MINRTT_DEFAULT_TB; + DEBUG_DEBUG(DEBUG_CATEGORY_ETCPROUTE, "MINRTT_UPD: svc_id=%u minrtt=%u (rtt=%u cnt=%u probe=%d unloaded=%lu)", + rconn->svc_id, rconn->minrtt, rconn->rtt, cnt, rconn->minrtt_probe, (unsigned long)unloaded_dur); + } + } +} + // ==================================================================== // ACK timers // ==================================================================== @@ -464,6 +633,13 @@ static void router_send_ack(struct ETCP_ROUTER_CONN* rconn) { hdr->seq = rconn->rx_seq; hdr->svc_id = rconn->svc_id; hdr->flags = (rconn->sess_id << ROUTER_SESS_ID_SHIFT); + if (rconn->last_recv_updated) { + uint64_t now = get_time_tb(); + hdr->timestamp = rconn->last_recv_pkt_ts + (uint32_t)(now - rconn->last_recv_pkt_local_tb); + rconn->last_recv_updated = 0; + } else { + hdr->timestamp = 0; + } struct ll_entry* entry = queue_entry_new(0); if (!entry) { u_free(hdr); return; } @@ -653,9 +829,16 @@ static int router_verify_signature(struct UTUN_INSTANCE* inst, struct ll_entry* } static void router_handle_ack(struct UTUN_INSTANCE* inst, struct ETCP_ROUTER_CONN* rconn, - uint32_t seq, uint64_t src_node_id) { + uint32_t seq, uint64_t src_node_id, uint16_t ack_ts) { (void)src_node_id; if (!rconn) return; + if (ack_ts != 0) { + uint16_t new_rtt = get_current_timestamp() - ack_ts; + int d_rtt = (int)new_rtt - (int)rconn->rtt; + if (d_rtt < 0) d_rtt = -d_rtt; + rconn->rtt_jitter += ((int32_t)(d_rtt * 65536 - rconn->rtt_jitter)) / 32; + rconn->rtt = new_rtt; + } if ((int32_t)(seq - rconn->tx_acked) >= 0) { uint32_t old_acked = rconn->tx_acked; rconn->tx_acked = seq; @@ -671,6 +854,8 @@ static void router_handle_ack(struct UTUN_INSTANCE* inst, struct ETCP_ROUTER_CON } rconn->last_ack_changed_tb = get_time_tb(); rconn->no_ack_count = 0; + router_track_inflight_state(rconn); + router_update_minrtt(rconn); if (queue_entry_count(rconn->inflight_q) == 0 && rconn->retrans_timer) { uasync_cancel_timeout(rconn->inst->ua, rconn->retrans_timer); rconn->retrans_timer = NULL; @@ -696,13 +881,27 @@ static void router_forward_transit(struct UTUN_INSTANCE* inst, struct ll_entry* (unsigned long long)hdr->dst_node_id, hdr->svc_id); free_entry(entry); return; } - DEBUG_TRACE(DEBUG_CATEGORY_ETCPROUTE, "etcp_router: forwarding svc_id=%u → %016llx via %s", - hdr->svc_id, (unsigned long long)hdr->dst_node_id, next->log_name); - int fwd_err = etcp_send(next, entry); - if (fwd_err != 0) { - free_entry(entry); - DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "etcp_router: transit etcp_send failed ret=%d", fwd_err); + struct TRANSIT_QUEUE* tq = transit_queue_find(next, hdr->src_node_id, hdr->dst_node_id); + + if (!tq && next->send_input_q && queue_entry_count(next->send_input_q) == 0) { + DEBUG_TRACE(DEBUG_CATEGORY_ETCPROUTE, "etcp_router: forwarding svc_id=%u → %016llx via %s", + hdr->svc_id, (unsigned long long)hdr->dst_node_id, next->log_name); + int ret = etcp_send(next, entry); + if (ret != 0) free_entry(entry); + return; } + + if (!tq) { + tq = transit_queue_get_or_create(next, hdr->src_node_id, hdr->dst_node_id); + if (!tq) { free_entry(entry); return; } + } + int was_empty = (queue_entry_count(tq->q) == 0); + queue_data_put(tq->q, entry); + DEBUG_TRACE(DEBUG_CATEGORY_ETCPROUTE, "etcp_router: queued transit svc_id=%u src=%016llx dst=%016llx q=%d", + hdr->svc_id, (unsigned long long)hdr->src_node_id, + (unsigned long long)hdr->dst_node_id, queue_entry_count(tq->q)); + if (was_empty && next->send_input_q) + queue_waiter_wait(next->send_input_q, &tq->waiter, transit_queue_drain_cb, tq); } static int router_check_peer_restart(struct UTUN_INSTANCE* inst, struct ETCP_ROUTER_CONN** prconn, @@ -739,7 +938,7 @@ static void router_handle_data_packet(struct ETCP_ROUTER_CONN* rconn, struct ETC {// SEQ out of bounds int32_t d = (int32_t)(seq - rconn->rx_seq); - if (d > ROUTER_MAX_INFLIGHT || d < -ROUTER_MAX_INFLIGHT) { + if (d > (int32_t)rconn->max_inflight || d < -(int32_t)rconn->max_inflight) { DEBUG_ERROR(DEBUG_CATEGORY_ETCPROUTE, "router: seq=%u out of bounds, rx_seq=%u (d=%d), dropping", seq, rconn->rx_seq, d); rconn->c_oob_dropped++; return; @@ -794,7 +993,7 @@ static void etcp_router_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) if (rconn) router_close_and_notify(rconn); free_entry(entry); return; } - if (pl_len == 0) { router_handle_ack(inst, rconn, hdr->seq, hdr->src_node_id); free_entry(entry); return; } + if (pl_len == 0) { router_handle_ack(inst, rconn, hdr->seq, hdr->src_node_id, hdr->timestamp); free_entry(entry); return; } if (!rconn) { rconn = etcp_router_conn_get(inst, hdr->src_node_id, hdr->svc_id); @@ -803,6 +1002,9 @@ static void etcp_router_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry) } if (router_check_peer_restart(inst, &rconn, hdr) != 0) { free_entry(entry); return; } + rconn->last_recv_pkt_ts = hdr->timestamp; + rconn->last_recv_pkt_local_tb = get_time_tb(); + rconn->last_recv_updated = 1; router_handle_data_packet(rconn, conn, hdr->seq, pl, pl_len); free_entry(entry); } @@ -905,7 +1107,7 @@ int etcp_route_send(struct UTUN_INSTANCE* inst, uint64_t dst_node_id, struct ll_ struct ETCP_ROUTER_CONN* rconn = etcp_router_conn_get(inst, dst_node_id, svc_id); if (!rconn) { queue_dgram_free(entry); queue_entry_free(entry); return -1; } - if ((int32_t)(rconn->tx_seq - rconn->tx_acked) >= ROUTER_MAX_INFLIGHT) { + if ((int32_t)(rconn->tx_seq - rconn->tx_acked) >= (int32_t)router_effective_max_inflight(rconn)) { int eq_ret = router_enqueue_send(rconn, entry->dgram + 1, payload_len, force); queue_dgram_free(entry); queue_entry_free(entry); return eq_ret; @@ -1017,6 +1219,22 @@ void etcp_router_conn_restart(struct UTUN_INSTANCE* inst, uint64_t remote_node_i rconn->c_stale_ack = 0; rconn->c_sign_fail = 0; rconn->last_ack_changed_tb = 0; + rconn->rtt = 0; + rconn->rtt_jitter = 0; + rconn->max_inflight = ROUTER_MAX_INFLIGHT; + rconn->inflight_limit = ROUTER_MAX_INFLIGHT; + rconn->minrtt = MINRTT_DEFAULT_TB; + memset(rconn->minrtt_window, 0, sizeof(rconn->minrtt_window)); + memset(rconn->minrtt_window_tb, 0, sizeof(rconn->minrtt_window_tb)); + rconn->minrtt_window_idx = 0; + rconn->minrtt_window_count = 0; + rconn->minrtt_probe = 0; + rconn->last_inflight_loaded_tb = 0; + rconn->last_inflight_unloaded_tb = 0; + rconn->inflight_was_loaded = 0; + rconn->last_recv_pkt_ts = 0; + rconn->last_recv_pkt_local_tb = 0; + rconn->last_recv_updated = 0; rconn->no_route = 0; rconn->no_ack_count = 0; rconn->closed = 0; @@ -1055,6 +1273,22 @@ struct ETCP_ROUTER_CONN* etcp_router_conn_get(struct UTUN_INSTANCE* inst, rconn->remote_node_id = remote_node_id; rconn->svc_id = svc_id; rconn->last_dgram_ts = get_current_timestamp(); + rconn->rtt = 0; + rconn->rtt_jitter = 0; + rconn->max_inflight = ROUTER_MAX_INFLIGHT; + rconn->inflight_limit = ROUTER_MAX_INFLIGHT; + rconn->minrtt = MINRTT_DEFAULT_TB; + memset(rconn->minrtt_window, 0, sizeof(rconn->minrtt_window)); + memset(rconn->minrtt_window_tb, 0, sizeof(rconn->minrtt_window_tb)); + rconn->minrtt_window_idx = 0; + rconn->minrtt_window_count = 0; + rconn->minrtt_probe = 0; + rconn->last_inflight_loaded_tb = 0; + rconn->last_inflight_unloaded_tb = 0; + rconn->inflight_was_loaded = 0; + rconn->last_recv_pkt_ts = 0; + rconn->last_recv_pkt_local_tb = 0; + rconn->last_recv_updated = 0; rconn->tx_seq = 0; rconn->rx_seq = 0; rconn->tx_acked = 0; @@ -1111,7 +1345,7 @@ int etcp_router_conn_send(struct ETCP_ROUTER_CONN* rconn, (void*)rconn, (const void*)data, len); return -1; } - if ((int32_t)(rconn->tx_seq - rconn->tx_acked) >= ROUTER_MAX_INFLIGHT) { + if ((int32_t)(rconn->tx_seq - rconn->tx_acked) >= (int32_t)router_effective_max_inflight(rconn)) { return router_enqueue_send(rconn, data, len, 0); } return router_send_one(rconn, data, len); @@ -1124,7 +1358,7 @@ int etcp_router_conn_send_signed(struct ETCP_ROUTER_CONN* rconn, (void*)rconn, (const void*)data, len); return -1; } - if ((int32_t)(rconn->tx_seq - rconn->tx_acked) >= ROUTER_MAX_INFLIGHT) { + if ((int32_t)(rconn->tx_seq - rconn->tx_acked) >= (int32_t)router_effective_max_inflight(rconn)) { DEBUG_WARN(DEBUG_CATEGORY_ETCPROUTE, "router_conn_send_signed: inflight full svc_id=%u", rconn->svc_id); return -1; } diff --git a/src/etcp_router.h b/src/etcp_router.h index 7c3da3f8..11068ca6 100644 --- a/src/etcp_router.h +++ b/src/etcp_router.h @@ -12,6 +12,8 @@ #include #include "etcp_api.h" +#define TRANSIT_QUEUE_HASH 256 + #pragma pack(push, 1) struct SVC_ROUTE_HDR { uint8_t cmd; // ETCP_ID_SVC_ROUTE (0x03) / ETCP_RT_ID_SVC_ROUTE @@ -20,9 +22,10 @@ struct SVC_ROUTE_HDR { uint32_t seq; // data: tx_seq; ACK: rx_seq (ожидаемый seq) uint8_t svc_id; // идентификатор сервиса uint8_t flags; // bit7=START, bit6=RST, bit5-4=sess_id + uint16_t timestamp; // часы отправителя (0.1ms), для RTT; ACK: эхо последнего data-пакета }; #pragma pack(pop) -#define SVC_ROUTE_HDR_SIZE sizeof(struct SVC_ROUTE_HDR) // 23 +#define SVC_ROUTE_HDR_SIZE sizeof(struct SVC_ROUTE_HDR) // 25 #define SVC_ROUTE_MAX_BINDINGS 256 // Биты в flags @@ -33,12 +36,41 @@ struct SVC_ROUTE_HDR { #define ROUTER_SESS_ID_SHIFT 4 #define ROUTER_SESS_ID_MASK 0x30 +// Очередь транзитных пакетов для пары (src_node_id, dst_node_id) +// Создаётся когда не получается отправить пакет в ETCP напрямую, +// удаляется когда очередь становится пустой. +// ll.data[0..15] = src:8 + dst:8 — совпадает с полями ниже по памяти +struct TRANSIT_QUEUE { + struct ll_entry ll; // data[0..15] = src_node_id:8 + dst_node_id:8 (16-byte hash index) + uint64_t src_node_id; // = data[0..7] + uint64_t dst_node_id; // = data[8..15] + struct ll_queue* q; // FIFO транзитных пакетов (ll_entry с dgram), начиная с data[16] + struct queue_waiter_handle waiter; // backpressure waiter на bp-очередь (normalizer->input или tx_queue) + struct ETCP_CONN* conn; // next_hop (для etcp_send в drain_cb) +}; + // Состояние одного логического подключения (remote_node_id + svc_id) struct ETCP_ROUTER_CONN { struct ll_entry ll; // data[0..7]=remote_node_id, data[8]=svc_id — хеш-индекс uint64_t remote_node_id; // = data[0..7] uint8_t svc_id; // = data[8] uint16_t last_dgram_ts; // последний timestamp датаграммы (rx или tx, timebase 0.1ms) + uint16_t rtt; // измеренный RTT (0.1ms) + int32_t rtt_jitter; // EMA jitter, fixed-point *65536 (как в ETCP) + uint16_t max_inflight; // настраиваемый рабочий максимум (default=ROUTER_MAX_INFLIGHT) + uint16_t inflight_limit; // текущий эффективный лимит (=max_inflight или MINRTT_PROBE_MAX_INFLIGHT) + uint16_t minrtt; // скользящее среднее minRTT (0.1ms), default=100 (10ms) + uint16_t minrtt_window[10]; // окно 10 последних замеров minRTT + uint64_t minrtt_window_tb[10];// таймстемпы замеров (timebase 0.1ms) + uint8_t minrtt_window_idx; // cyclic index 0..9 + uint8_t minrtt_window_count; // активных записей в окне (0..10) + uint8_t minrtt_probe; // режим измерения (inflight capped at 4) + uint64_t last_inflight_loaded_tb; // когда inflight >= threshold + uint64_t last_inflight_unloaded_tb; // когда inflight < threshold + uint8_t inflight_was_loaded; // 1=канал загружен + uint16_t last_recv_pkt_ts; // timestamp пира из последнего data-пакета + uint64_t last_recv_pkt_local_tb; // наше время приёма последнего data-пакета (0.1ms timebase) + uint8_t last_recv_updated; // флаг: есть данные для эха в ACK uint32_t tx_seq; // следующий seq для отправки uint32_t rx_seq; // ожидаемый seq для сборки (next expected) @@ -100,6 +132,13 @@ struct ETCP_ROUTER_CONN { #define ROUTER_NO_ROUTE_RETRY_TB 200 // 20ms проверка появления маршрута #define ROUTER_NO_ACK_MAX_RETRANS 17 // 17 × 300ms ≈ 5s таймаут без ACK +// minRTT measurement +#define MINRTT_UNLOADED_MIN_TB 500 // 50ms минимальное время разгрузки канала +#define MINRTT_PROBE_TIMEOUT_TB 100000 // 10 сек без свежих замеров → probe +#define MINRTT_PROBE_MAX_INFLIGHT 4 // лимит inflight в режиме probe +#define MINRTT_WINDOW_SIZE 10 +#define MINRTT_DEFAULT_TB 100 // default minRTT = 10ms (заменяется первым RTT) + // Inflight запись — копия отправленного пакета для возможного ретрансмита struct ROUTER_INFLIGHT { struct ll_entry ll; // индекс по seq (4 байта, offset 0) @@ -144,6 +183,9 @@ int etcp_router_conn_send_signed(struct ETCP_ROUTER_CONN* rconn, // Закрыть seq-подключение void etcp_router_conn_close(struct ETCP_ROUTER_CONN* rconn); +// Установить рабочий max_inflight, пересчитывает inflight_limit и state machine +void router_set_max_inflight(struct ETCP_ROUTER_CONN* rconn, uint16_t new_max); + // Асинхронное закрытие: коллбэк вызывается после освобождения памяти rconn void etcp_router_conn_close_async(struct ETCP_ROUTER_CONN* rconn, void (*on_close_done)(void* arg), @@ -174,4 +216,7 @@ void etcp_router_cancel_send_ready(struct UTUN_INSTANCE* inst, uint64_t node_id, // Очищает send_q/recv_q, уведомляет сервис через cb(NULL, entry) с remote_node_id. void etcp_router_conn_restart(struct UTUN_INSTANCE* inst, uint64_t remote_node_id, uint8_t svc_id); +// Удалить все транзитные очереди соединения (вызывается при закрытии ETCP_CONN) +void etcp_router_transit_queues_destroy(struct ETCP_CONN* conn); + #endif // ETCP_ROUTER_H diff --git a/src/pkt_normalizer.c b/src/pkt_normalizer.c index 4e36ab01..cabc7f18 100644 --- a/src/pkt_normalizer.c +++ b/src/pkt_normalizer.c @@ -53,6 +53,8 @@ struct PKTNORM* pn_init(struct ETCP_CONN* etcp) { pn_deinit(pn); return NULL; } + queue_set_threshold(pn->input, 0, 0); // транзитные очереди: ждать полного освобождения + queue_set_waiter_defer(pn->input, 1); // defer чтобы избежать рекурсии drain_cb pn->output = queue_new(pn->ua, 0, 0, 0, "pn_output"); // No hash needed if (!pn->output) { diff --git a/src/stcp_client.c b/src/stcp_client.c index 90bff59a..48b030a9 100644 --- a/src/stcp_client.c +++ b/src/stcp_client.c @@ -110,6 +110,7 @@ static void client_tx_queue_cb(struct ll_queue *q, void *arg) { if (buf && !encrypt_and_crc(c, e->dgram, e->len, buf, &need)) client_try_send(c, buf, need); else if (buf) u_free(buf); } + queue_dgram_free(e); queue_entry_free(e); queue_resume_callback(q); } diff --git a/src/stcp_link.c b/src/stcp_link.c index b6f1b441..84cde258 100644 --- a/src/stcp_link.c +++ b/src/stcp_link.c @@ -5,6 +5,7 @@ #include "stcp_client.h" #include "secure_channel.h" #include "etcp.h" +#include "etcp_router.h" #include "utun_instance.h" #include "../lib/debug_config.h" #include "../lib/ll_queue.h" @@ -73,6 +74,9 @@ static void server_accept_cb(struct stcp_conn *conn, void *arg) { queue_set_callback(rx, link_rx_cb, link); stcp_conn_set_rx_queue(conn, rx); link->tx_queue = queue_new(conn->ua, 0, 0, 0, "stx"); + queue_set_threshold(link->tx_queue, 0, 0); + queue_set_waiter_defer(link->tx_queue, 1); + link->etcp_conn.send_input_q = link->tx_queue; stcp_conn_set_tx_queue(conn, link->tx_queue); DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "stcp_link: server accepted connection"); @@ -94,6 +98,9 @@ static void client_ready_cb(struct stcp_conn *conn, void *arg) { queue_set_callback(rx, link_rx_cb, link); stcp_conn_set_rx_queue(conn, rx); link->tx_queue = queue_new(conn->ua, 0, 0, 0, "ctx"); + queue_set_threshold(link->tx_queue, 0, 0); + queue_set_waiter_defer(link->tx_queue, 1); + link->etcp_conn.send_input_q = link->tx_queue; stcp_conn_set_tx_queue(conn, link->tx_queue); DEBUG_INFO(DEBUG_CATEGORY_SOCKET, "stcp_link: client connected"); @@ -160,6 +167,21 @@ struct stcp_link *stcp_link_connect(struct stcp_link_config *cfg) { void stcp_link_close(struct stcp_link *link) { if (!link) return; + struct ll_queue* tq_reg = link->etcp_conn.transit_queues; + if (tq_reg) { + struct ll_entry* te; + while ((te = queue_data_get(tq_reg)) != NULL) { + struct TRANSIT_QUEUE* tq = (struct TRANSIT_QUEUE*)te; + if (link->etcp_conn.send_input_q) queue_waiter_cancel(link->etcp_conn.send_input_q, &tq->waiter); + struct ll_entry* pe; + while ((pe = queue_data_get(tq->q)) != NULL) { queue_dgram_free(pe); queue_entry_free(pe); } + queue_free(tq->q); + queue_entry_free(&tq->ll); + } + queue_free(tq_reg); + link->etcp_conn.transit_queues = NULL; + link->etcp_conn.send_input_q = NULL; + } if (link->conn) { if (link->conn->rx_queue) queue_free(link->conn->rx_queue); if (link->tx_queue) queue_free(link->tx_queue); diff --git a/src/stcp_server.c b/src/stcp_server.c index 5c58d2a3..5cd5980e 100644 --- a/src/stcp_server.c +++ b/src/stcp_server.c @@ -156,6 +156,7 @@ static void tx_queue_cb(struct ll_queue *q, void *arg) { struct ll_entry *e = queue_data_get(q); if (!e) { queue_resume_callback(q); return; } if (e->dgram) stcp_conn_send_message(c, e->dgram, e->len); + queue_dgram_free(e); queue_entry_free(e); queue_resume_callback(q); } diff --git a/tools/etcpmon/etcpmon_gui.c b/tools/etcpmon/etcpmon_gui.c index 257426e1..697d225a 100644 --- a/tools/etcpmon/etcpmon_gui.c +++ b/tools/etcpmon/etcpmon_gui.c @@ -759,7 +759,7 @@ static void CreateControls(struct etcpmon_app* app) { int lw = 85, ew = 70; /* label width, edit width */ CreateWindowExA(0, "BUTTON", "Router Congestion", WS_CHILD | WS_VISIBLE | BS_GROUPBOX, - 10, ry, WINDOW_WIDTH - 20, 140, hWnd, (HMENU)IDC_GB_ROUTER, hInst, NULL); + 10, ry, WINDOW_WIDTH - 20, 160, hWnd, (HMENU)IDC_GB_ROUTER, hInst, NULL); /* Row 1 */ int ry1 = ry + 22; CreateWindowExA(0, "STATIC", "Conns:", WS_CHILD | WS_VISIBLE, r_col1, ry1, 50, 16, hWnd, (HMENU)IDC_STATIC, hInst, NULL); @@ -794,6 +794,16 @@ static void CreateControls(struct etcpmon_app* app) { app->hEditRtrDupDrop = CreateWindowExA(WS_EX_CLIENTEDGE, "EDIT", "", WS_CHILD | WS_VISIBLE | ES_READONLY | ES_CENTER, r_col1 + lw, ry4 - 2, ew, 18, hWnd, (HMENU)IDC_EDIT_RTR_DUP_DROP, hInst, NULL); CreateWindowExA(0, "STATIC", "OOB Drop:", WS_CHILD | WS_VISIBLE, r_col2, ry4, lw, 16, hWnd, (HMENU)IDC_STATIC, hInst, NULL); app->hEditRtrOobDrop = CreateWindowExA(WS_EX_CLIENTEDGE, "EDIT", "", WS_CHILD | WS_VISIBLE | ES_READONLY | ES_CENTER, r_col2 + lw, ry4 - 2, ew, 18, hWnd, (HMENU)IDC_EDIT_RTR_OOB_DROP, hInst, NULL); + /* Row 5: RTT / Jitter */ + int ry5 = ry + 122; + CreateWindowExA(0, "STATIC", "RTT:", WS_CHILD | WS_VISIBLE, r_col1, ry5, 40, 16, hWnd, (HMENU)IDC_STATIC, hInst, NULL); + app->hEditRtrRtt = CreateWindowExA(WS_EX_CLIENTEDGE, "EDIT", "", WS_CHILD | WS_VISIBLE | ES_READONLY | ES_CENTER, r_col1 + 45, ry5 - 2, 70, 18, hWnd, (HMENU)IDC_EDIT_RTR_RTT, hInst, NULL); + CreateWindowExA(0, "STATIC", "Avg RTT:", WS_CHILD | WS_VISIBLE, r_col2, ry5, 65, 16, hWnd, (HMENU)IDC_STATIC, hInst, NULL); + app->hEditRtrRttAvg = CreateWindowExA(WS_EX_CLIENTEDGE, "EDIT", "", WS_CHILD | WS_VISIBLE | ES_READONLY | ES_CENTER, r_col2 + 65, ry5 - 2, 70, 18, hWnd, (HMENU)IDC_EDIT_RTR_RTT_AVG, hInst, NULL); + CreateWindowExA(0, "STATIC", "Jitter:", WS_CHILD | WS_VISIBLE, r_col3, ry5, 55, 16, hWnd, (HMENU)IDC_STATIC, hInst, NULL); + app->hEditRtrJitter = CreateWindowExA(WS_EX_CLIENTEDGE, "EDIT", "", WS_CHILD | WS_VISIBLE | ES_READONLY | ES_CENTER, r_col3 + 55, ry5 - 2, 70, 18, hWnd, (HMENU)IDC_EDIT_RTR_JITTER, hInst, NULL); + CreateWindowExA(0, "STATIC", "MinRTT:", WS_CHILD | WS_VISIBLE, r_col4, ry5, 60, 16, hWnd, (HMENU)IDC_STATIC, hInst, NULL); + app->hEditRtrMinrtt = CreateWindowExA(WS_EX_CLIENTEDGE, "EDIT", "", WS_CHILD | WS_VISIBLE | ES_READONLY | ES_CENTER, r_col4 + 60, ry5 - 2, 70, 18, hWnd, (HMENU)IDC_EDIT_RTR_MINRTT, hInst, NULL); } /* ======================================================================== @@ -1659,6 +1669,10 @@ void etcpmon_gui_update_metrics(struct etcpmon_app* app, UpdateEditIfChanged(hMain, IDC_EDIT_RTR_OOB_DROP, "%u", metrics->router.oob_dropped); UpdateEditIfChanged(hMain, IDC_EDIT_RTR_STALE_ACK, "%u", metrics->router.stale_ack); UpdateEditIfChanged(hMain, IDC_EDIT_RTR_SIGN_FAIL, "%u", metrics->router.sign_fail); + UpdateEditIfChanged(hMain, IDC_EDIT_RTR_RTT, "%.1f ms", metrics->router.rtt_last/10.f); + UpdateEditIfChanged(hMain, IDC_EDIT_RTR_RTT_AVG, "%.1f ms", metrics->router.rtt_avg/10.f); + UpdateEditIfChanged(hMain, IDC_EDIT_RTR_JITTER, "%.1f ms", metrics->router.jitter/10.f); + UpdateEditIfChanged(hMain, IDC_EDIT_RTR_MINRTT, "%.1f ms", metrics->router.minrtt/10.f); /* Queue Metrics */ UpdateEditIfChanged(hMain, IDC_EDIT_Q_IN_Q_BYTES, "%u", metrics->etcp.input_queue_bytes); @@ -1844,6 +1858,10 @@ void etcpmon_gui_clear_metrics(struct etcpmon_app* app) { SetDlgItemTextA(app->hWndMain, IDC_EDIT_RTR_OOB_DROP, ""); SetDlgItemTextA(app->hWndMain, IDC_EDIT_RTR_STALE_ACK, ""); SetDlgItemTextA(app->hWndMain, IDC_EDIT_RTR_SIGN_FAIL, ""); + SetDlgItemTextA(app->hWndMain, IDC_EDIT_RTR_RTT, ""); + SetDlgItemTextA(app->hWndMain, IDC_EDIT_RTR_RTT_AVG, ""); + SetDlgItemTextA(app->hWndMain, IDC_EDIT_RTR_JITTER, ""); + SetDlgItemTextA(app->hWndMain, IDC_EDIT_RTR_MINRTT, ""); SetDlgItemTextA(app->hWndMain, IDC_EDIT_Q_IN_Q_BYTES, ""); SetDlgItemTextA(app->hWndMain, IDC_EDIT_Q_IN_Q_PKTS, ""); diff --git a/tools/etcpmon/etcpmon_gui.h b/tools/etcpmon/etcpmon_gui.h index cc1f42e1..1a9df3c0 100644 --- a/tools/etcpmon/etcpmon_gui.h +++ b/tools/etcpmon/etcpmon_gui.h @@ -96,6 +96,10 @@ extern "C" { #define IDC_EDIT_RTR_OOB_DROP 361 #define IDC_EDIT_RTR_STALE_ACK 362 #define IDC_EDIT_RTR_SIGN_FAIL 363 +#define IDC_EDIT_RTR_RTT 364 +#define IDC_EDIT_RTR_RTT_AVG 365 +#define IDC_EDIT_RTR_JITTER 366 +#define IDC_EDIT_RTR_MINRTT 367 /* Link list control ID */ #define IDC_LIST_LINKS 400 @@ -282,6 +286,10 @@ struct etcpmon_app { HWND hEditRtrOobDrop; HWND hEditRtrStaleAck; HWND hEditRtrSignFail; + HWND hEditRtrRtt; + HWND hEditRtrRttAvg; + HWND hEditRtrJitter; + HWND hEditRtrMinrtt; /* Links list */ HWND hListLinks; diff --git a/tools/etcpmon/etcpmon_protocol.h b/tools/etcpmon/etcpmon_protocol.h index fae0f711..9f3b02f7 100644 --- a/tools/etcpmon/etcpmon_protocol.h +++ b/tools/etcpmon/etcpmon_protocol.h @@ -313,6 +313,10 @@ struct etcpmon_router_metrics { uint32_t oob_dropped; uint32_t stale_ack; uint32_t sign_fail; + uint32_t rtt_last; /* max RTT across all rconns (0.1ms) */ + uint32_t rtt_avg; /* avg RTT across all rconns with rtt>0 (0.1ms) */ + uint32_t jitter; /* max jitter across all rconns (0.1ms) */ + uint32_t minrtt; /* min minRTT across all rconns (0.1ms) */ }; struct etcpmon_rsp_metrics {