etcp: cyclic link_id, link close on addr change, inflight nullify; test rewrite
This commit is contained in:
1 parent
d4a170b5c7
commit
b503fa4418
6 files changed
+108
-113
No files matched your search
@@ -807,9 +807,9 @@ static void reconcile_iface(struct AUTO_SOCKET* as, uint32_t ifindex, const char
|
||||
else
|
||||
DEBUG_INFO(DEBUG_CATEGORY_AS, "[as] v4 UDP (%s) addr changed: %s→%s", ifname,
|
||||
sockaddr_storage_to_str(&old4).str, sockaddr_storage_to_str(&sock->interface_addr).str);
|
||||
create_iface_udp_socket(as, ifindex, ifname, AF_INET, new_type);
|
||||
ncd_remove_socket_links(as->instance, sock);
|
||||
etcp_socket_remove(sock);
|
||||
create_iface_udp_socket(as, ifindex, ifname, AF_INET, new_type);
|
||||
ifa->v4_type = new_type;
|
||||
{ struct ETCP_SOCKET* s = as->instance->etcp_sockets;
|
||||
while (s) { if (s->netif_index == ifindex && s->local_addr.ss_family == AF_INET) { ifa->v4_udp = s; break; } s = s->next; } }
|
||||
@@ -852,9 +852,9 @@ static void reconcile_iface(struct AUTO_SOCKET* as, uint32_t ifindex, const char
|
||||
else
|
||||
DEBUG_INFO(DEBUG_CATEGORY_AS, "[as] v6 UDP (%s) addr changed: %s→%s", ifname,
|
||||
sockaddr_storage_to_str(&old6).str, sockaddr_storage_to_str(&sock->interface_addr).str);
|
||||
create_iface_udp_socket(as, ifindex, ifname, AF_INET6, new_type);
|
||||
ncd_remove_socket_links(as->instance, sock);
|
||||
etcp_socket_remove(sock);
|
||||
create_iface_udp_socket(as, ifindex, ifname, AF_INET6, new_type);
|
||||
ifa->v6_type = new_type;
|
||||
{ struct ETCP_SOCKET* s = as->instance->etcp_sockets;
|
||||
while (s) { if (s->netif_index == ifindex && s->local_addr.ss_family == AF_INET6) { ifa->v6_udp = s; break; } s = s->next; } }
|
||||
|
||||
@@ -1112,8 +1112,10 @@ struct ETCP_DGRAM* etcp_request_pkt(struct ETCP_CONN* etcp) {
|
||||
// Always subtract from previous last_link (if any) – this handles retransmission
|
||||
if (inf_pkt->last_link) {
|
||||
inf_pkt->last_link->total_retransmissions++;
|
||||
inf_pkt->last_link->bbr_loss_since_ack += inf_pkt->ll.len;
|
||||
bbr_note_loss(inf_pkt->last_link->bbr);
|
||||
if (inf_pkt->last_link->bbr) {
|
||||
inf_pkt->last_link->bbr_loss_since_ack += inf_pkt->ll.len;
|
||||
bbr_note_loss(inf_pkt->last_link->bbr);
|
||||
}
|
||||
inf_pkt->last_link->inflight_bytes -= inf_pkt->ll.len;
|
||||
inf_pkt->last_link->inflight_packets--;
|
||||
}
|
||||
|
||||
@@ -150,6 +150,7 @@ struct ETCP_CONN {
|
||||
struct ETCP_LINK* links;
|
||||
struct ETCP_LINK* last_rr_link; // последний линк, выбранный round-robin
|
||||
uint8_t tcp_link_count; // количество UP TCP линков
|
||||
uint8_t last_assigned_link_id; // последний выданный link_id (для циклического перебора)
|
||||
|
||||
// Crypto and state
|
||||
struct secure_channel crypto_ctx;
|
||||
|
||||
@@ -565,15 +565,13 @@ int etcp_find_free_local_link_id(struct ETCP_CONN* etcp) {
|
||||
link = link->next;
|
||||
}
|
||||
|
||||
// Ищем первый свободный id
|
||||
for (int i = 0; i < 32; i++) {
|
||||
if (used_ids[i] != 0xFF) {
|
||||
// Есть свободные биты в этом байте
|
||||
for (int bit = 0; bit < 8; bit++) {
|
||||
if (!(used_ids[i] & (1 << bit))) {
|
||||
return (i << 3) + bit;
|
||||
}
|
||||
}
|
||||
// Ищем первый свободный id, начиная с last_assigned_link_id + 1 (циклический перебор)
|
||||
for (int i = 0; i < 256; i++) {
|
||||
int id = ((etcp->last_assigned_link_id + 1 + i) & 0xFF);
|
||||
if (id == 0) continue;
|
||||
if (!(used_ids[id >> 3] & (1 << (id & 7)))) {
|
||||
etcp->last_assigned_link_id = (uint8_t)id;
|
||||
return id;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -861,7 +859,13 @@ void etcp_socket_remove(struct ETCP_SOCKET* conn) {
|
||||
queue_free(conn->links_queue);
|
||||
conn->links_queue = NULL;
|
||||
}
|
||||
|
||||
|
||||
if (conn->instance) {
|
||||
struct ETCP_SOCKET** pp = &conn->instance->etcp_sockets;
|
||||
while (*pp && *pp != conn) pp = &(*pp)->next;
|
||||
if (*pp) *pp = conn->next;
|
||||
}
|
||||
|
||||
u_free(conn);
|
||||
}
|
||||
|
||||
@@ -1111,6 +1115,8 @@ void etcp_link_close(struct ETCP_LINK* link) {
|
||||
if (link->shaper_timer) uasync_cancel_timeout(link->etcp->instance->ua, link->shaper_timer);
|
||||
if (link->keepalive_timer) uasync_cancel_timeout(link->etcp->instance->ua, link->keepalive_timer);
|
||||
etcp_conn_on_inflight_lim_changed(link->etcp);
|
||||
etcp_inflight_nullify_link(link->etcp->input_wait_ack, link);
|
||||
etcp_inflight_nullify_link(link->etcp->input_send_q, link);
|
||||
u_free(link->bbr);
|
||||
u_free(link);
|
||||
return;
|
||||
@@ -2238,20 +2244,11 @@ void etcp_connections_read_callback_socket(socket_t sock, void* arg) {
|
||||
DEBUG_DEBUG(DEBUG_CATEGORY_CONNECTION, "[%s] found existing link for id=%d, socket=[%s]", conn->log_name, req->link_id, e_sock->name);
|
||||
link = existing_link;
|
||||
if (!sockaddr_equal(&link->remote_addr, &addr)) {
|
||||
DEBUG_WARN(DEBUG_CATEGORY_CONNECTION, "[%s] IP:port changed for remote_link_id=%d socket:[%s]", conn->log_name, req->link_id, e_sock->name);
|
||||
remove_link_from_queue(link); // remove from old socket queue
|
||||
link->conn = e_sock;
|
||||
memcpy(&link->remote_addr, &addr, sizeof(addr));
|
||||
if (insert_link_queue(link->conn, link) < 0) {
|
||||
DEBUG_ERROR(DEBUG_CATEGORY_CONNECTION, "Failed to reinsert link after addr change");
|
||||
goto ec_fr;
|
||||
}
|
||||
// Cancel pending NAT check and reset status
|
||||
nat_detection_cancel_for_conn(link->etcp->instance->nat_det, link->etcp);
|
||||
link->nat_check_status = NAT_CHECK_NONE;
|
||||
link->nat_type = NAT_TYPE_UNKNOWN;
|
||||
}
|
||||
|
||||
DEBUG_WARN(DEBUG_CATEGORY_CONNECTION, "[%s] IP:port changed for remote_link_id=%d socket:[%s] — closing old link, creating new",
|
||||
conn->log_name, req->link_id, e_sock->name);
|
||||
etcp_link_close(link);
|
||||
goto create_new_link;
|
||||
}
|
||||
// Link exists - reuse it for recovery
|
||||
link->remote_link_id = req->link_id;
|
||||
link->remote_socket_id = req->socket_id;
|
||||
@@ -2284,14 +2281,15 @@ void etcp_connections_read_callback_socket(socket_t sock, void* arg) {
|
||||
if (ml->init_timer) { uasync_cancel_timeout(conn->instance->ua, ml->init_timer); ml->init_timer = NULL; }
|
||||
}
|
||||
|
||||
/* Sync session_id; reset if remote is clean and we're dirty, or request reset if we're clean and remote is dirty */
|
||||
/* Sync session_id; reset only if remote session changed */
|
||||
{ int sess_changed = (conn->session_id != session_id);
|
||||
conn->session_id = session_id;
|
||||
if (req->code == ETCP_INIT_REQUEST) {
|
||||
if (conn->got_initial_pkt) etcp_conn_reinit(conn, "duplicate INIT");
|
||||
if (sess_changed || conn->got_initial_pkt) etcp_conn_reinit(conn, "duplicate INIT");
|
||||
send_reset = 0;
|
||||
} else {
|
||||
if (conn->got_initial_pkt == 0) send_reset = 1;
|
||||
}
|
||||
} }
|
||||
}
|
||||
|
||||
// Cancel existing timers
|
||||
@@ -2300,6 +2298,7 @@ void etcp_connections_read_callback_socket(socket_t sock, void* arg) {
|
||||
link->init_timer = NULL;
|
||||
}
|
||||
} else {
|
||||
create_new_link:
|
||||
DEBUG_DEBUG(DEBUG_CATEGORY_CONNECTION, "[%s] NO existing link for id=%d, socket=[%s]", conn->log_name, req->link_id, e_sock->name);
|
||||
// Create new link
|
||||
link = etcp_link_new(conn, e_sock, &addr, 1);
|
||||
@@ -2328,14 +2327,16 @@ void etcp_connections_read_callback_socket(socket_t sock, void* arg) {
|
||||
return;
|
||||
}
|
||||
|
||||
// Always sync session_id; reset if new session or data was flowing
|
||||
// Always sync session_id; reset only if remote session changed
|
||||
{ int sess_changed = (conn->session_id != session_id);
|
||||
conn->session_id = session_id;
|
||||
if (sess_changed || conn->got_initial_pkt) {
|
||||
if (sess_changed) {
|
||||
send_reset = 1;
|
||||
DEBUG_INFO(DEBUG_CATEGORY_ETCP, "[%s] REINIT new link: code=0x%02x sess=%08x→%08x got_init=%d initialized=%d links_up=%d",
|
||||
conn->log_name, code, conn->session_id, session_id, conn->got_initial_pkt, conn->initialized, conn->links_up);
|
||||
etcp_conn_reinit(conn, "session changed");
|
||||
} else if (conn->got_initial_pkt == 0) {
|
||||
send_reset = 1;
|
||||
} }
|
||||
}
|
||||
uint16_t client_ka = (req->keepalive[0]<<8) | req->keepalive[1];
|
||||
|
||||
Reference in new issue
Block a user