@ -78,9 +78,10 @@ static int router_send_one_flags(struct ETCP_ROUTER_CONN* rconn, const uint8_t*
queue_entry_free ( entry ) ; queue_dgram_free ( entry ) ; if ( ! seq_flags ) rconn - > tx_seq - - ; return - 1 ;
}
rconn - > last_dgram_ts = get_current_timestamp ( ) ;
DEBUG_TRACE ( DEBUG_CATEGORY_ETCPROUTE , " router_send: seq=%u → %016llx svc_id=%u len=%zu inflight=%d " ,
seq , ( unsigned long long ) rconn - > remote_node_id , rconn - > svc_id , pl_len ,
( int32_t ) ( rconn - > tx_seq - rconn - > tx_acked ) ) ;
DEBUG_DEBUG ( DEBUG_CATEGORY_ETCPROUTE , " ETCP_SEND: svc_id=%u seq=%u len=%zu inflight=%d → %016llx " ,
rconn - > svc_id , seq , pl_len ,
( int32_t ) ( rconn - > tx_seq - rconn - > tx_acked ) ,
( unsigned long long ) rconn - > remote_node_id ) ;
return etcp_send ( conn , entry ) ;
}
@ -149,6 +150,9 @@ static void router_close_and_notify(struct ETCP_ROUTER_CONN* rconn) {
}
static void router_drain_send_q ( struct ETCP_ROUTER_CONN * rconn ) {
int sq = queue_entry_count ( rconn - > send_q ) ;
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 ;
struct ll_entry * e = queue_data_get ( rconn - > send_q ) ;
@ -224,8 +228,8 @@ static void router_send_ack(struct ETCP_ROUTER_CONN* rconn) {
entry - > dgram = ( uint8_t * ) hdr ;
entry - > len = SVC_ROUTE_HDR_SIZE ;
DEBUG_DEBUG ( DEBUG_CATEGORY_ETCPROUTE , " router_ack: → %016llx svc_id=%u rx_seq=%u " ,
( unsigned long long ) rconn - > remote_node_id , rconn - > svc_id , rconn - > rx_seq ) ;
DEBUG_DEBUG ( DEBUG_CATEGORY_ETCPROUTE , " ACK_SEND: svc_id=%u rx_seq=%u → %016llx " ,
rconn - > svc_id , rconn - > rx_seq , ( unsigned long long ) rconn - > remote_node_id ) ;
etcp_send ( conn , entry ) ;
}
@ -325,6 +329,9 @@ static void etcp_router_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry)
size_t pl_len = entry - > len - SVC_ROUTE_HDR_SIZE ;
uint8_t * pl = entry - > dgram + SVC_ROUTE_HDR_SIZE ;
DEBUG_DEBUG ( DEBUG_CATEGORY_ETCPROUTE , " ETCP_RECV: svc_id=%u seq=%u len=%zu from %016llx " ,
hdr - > svc_id , hdr - > seq , pl_len , ( unsigned long long ) hdr - > src_node_id ) ;
if ( hdr - > dst_node_id = = inst - > node_id ) {
// ========== Мы — целевая нода ==========
struct ETCP_ROUTER_CONN * rconn = router_conn_find ( inst , hdr - > src_node_id , hdr - > svc_id ) ;
@ -340,6 +347,10 @@ static void etcp_router_recv_cb(struct ETCP_CONN* conn, struct ll_entry* entry)
if ( rconn ) {
if ( ( int32_t ) ( hdr - > seq - rconn - > tx_acked ) > = 0 ) {
rconn - > tx_acked = hdr - > seq ;
DEBUG_DEBUG ( DEBUG_CATEGORY_ETCPROUTE , " SEND_Q_ACKED: svc_id=%u ack=%u inflight=%d send_q=%d from %016llx " ,
rconn - > svc_id , hdr - > seq , ( int32_t ) ( rconn - > tx_seq - rconn - > tx_acked ) ,
queue_entry_count ( rconn - > send_q ) ,
( unsigned long long ) hdr - > src_node_id ) ;
} else {
DEBUG_WARN ( DEBUG_CATEGORY_ETCPROUTE , " router: stale ACK seq=%u tx_acked=%u from %016llx, ignoring " ,
hdr - > seq , rconn - > tx_acked , ( unsigned long long ) hdr - > src_node_id ) ;
@ -557,6 +568,8 @@ void etcp_router_consumer_ack(struct UTUN_INSTANCE* inst, uint64_t remote_node_i
struct ETCP_ROUTER_CONN * rconn = etcp_router_conn_get ( inst , remote_node_id , svc_id ) ;
if ( ! rconn ) return ;
rconn - > consumer_ack = 1 ;
DEBUG_DEBUG ( DEBUG_CATEGORY_ETCPROUTE , " CONSUMER_ACK: svc_id=%u rx_seq=%u → %016llx " ,
svc_id , rconn - > rx_seq , ( unsigned long long ) remote_node_id ) ;
router_ack_do_send ( rconn ) ;
}