diff --git a/src/transport_layer/etcp_connections.c b/src/transport_layer/etcp_connections.c index be56f309..2b4f65b2 100644 --- a/src/transport_layer/etcp_connections.c +++ b/src/transport_layer/etcp_connections.c @@ -1205,14 +1205,14 @@ void etcp_link_update_inflight_lim(struct ETCP_LINK* link, uint32_t new_lim) { etcp_conn_on_inflight_lim_changed(link->etcp); } -// Обнуляет last_link для всех inflight-записей ссылающихся на dead_link. -// Вызывается перед u_free(link) чтобы etcp_ack_recv не упал на use-after-free. +// Убрать курсор и бит слота до удаления линка из списка: callbacks смогут безопасно переиспользовать его local_link_id. static void etcp_inflight_nullify_link(struct ll_queue* q, struct ETCP_LINK* dead_link) { if (!q) return; struct ll_entry* entry = q->head; while (entry) { struct INFLIGHT_PACKET* inf = (struct INFLIGHT_PACKET*)entry; if (inf->last_link == dead_link) inf->last_link = NULL; + inf->used_links[dead_link->local_link_id >> 3] &= (uint8_t)~(1u << (dead_link->local_link_id & 7)); entry = entry->next; } } @@ -1278,6 +1278,9 @@ void etcp_link_close(struct ETCP_LINK* link) { link->keepalive_timer = NULL; } + etcp_inflight_nullify_link(etcp->input_wait_ack, link); + etcp_inflight_nullify_link(etcp->input_send_q, link); + // универсальное удаление из односвязного списка struct ETCP_LINK **pp = &link->etcp->links; while (*pp) { @@ -1294,8 +1297,6 @@ void etcp_link_close(struct ETCP_LINK* link) { DEBUG_INFO(DEBUG_CATEGORY_CONNECTION, "[%s] Link %d closed: addr=%s rcvd=%zub ack=%llub infl=%ub/%upkt", link->etcp->log_name, link->local_link_id, sockaddr_storage_to_str(&link->remote_addr).str, link->total_decrypted, (unsigned long long)link->acked_bytes, link->inflight_bytes, link->inflight_packets); - etcp_inflight_nullify_link(link->etcp->input_wait_ack, link); - etcp_inflight_nullify_link(link->etcp->input_send_q, link); etcp_fire_link_status_cbk(link, old_state, old_status); if (!etcp->close_requested) etcp_on_link_down(etcp, NULL); callring_put(CR_LINK_CLOSE_FREE, (uintptr_t)link, 0); diff --git a/src/transport_layer/etcp_tx.c b/src/transport_layer/etcp_tx.c index 494e7f64..b6242ad3 100644 --- a/src/transport_layer/etcp_tx.c +++ b/src/transport_layer/etcp_tx.c @@ -5,6 +5,7 @@ #include "../lib/debug_config.h" #include #include +#include static void retransmit_cb(void* arg); @@ -60,6 +61,15 @@ static void attempt_retire(struct ETCP_LINK* link, struct ETCP_TX_ATTEMPT* attem } link->inflight_bytes -= attempt->ll.len; link->inflight_packets--; conn->active_attempts--; if (packet) packet->active_attempts--; + if (loss) { + if (link->data_rto_streak < ETCP_TX_STALL_THRESHOLD) link->data_rto_streak++; + if (!link->tx_stalled && link->data_rto_streak == ETCP_TX_STALL_THRESHOLD) { + link->tx_stalled = 1; link->tx_stalled_since_us = get_time_us(); + DEBUG_WARN(DEBUG_CATEGORY_ETCP, "[%s] DATA path stalled link=%u consecutive_RTO=%u seq=%u flight=%u", + conn->log_name, link->local_link_id, link->data_rto_streak, attempt->seq, link->inflight_bytes); + etcp_conn_wake_sender(conn); + } + } if (loss && link->bbr) { uint64_t lost = (uint64_t)link->bbr_loss_since_ack + attempt->ll.len; link->bbr_loss_since_ack = lost > INT_MAX ? INT_MAX : (uint32_t)lost; @@ -129,6 +139,9 @@ void etcp_tx_link_reset(struct ETCP_LINK* link, int retry) { link->shaper_load_time_tb = get_time_tb(); link->shaper_sub_nanotime = 0; link->bbr_pacing_rate = link->bandwidth = 0; link->data_rtt_us = link->data_jitter_us = link->bbr_loss_since_ack = 0; + if (link->tx_stalled) DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] clear stalled path on reset link=%u", + link->etcp->log_name, link->local_link_id); + link->data_rto_streak = link->tx_stalled = 0; link->tx_stalled_since_us = 0; link->ack_reply_pending = 0; link->delivered_bytes = link->acked_bytes = link->acked_packets = 0; link->rx_data_bytes = link->rx_data_time_us = 0; @@ -209,6 +222,13 @@ void etcp_tx_complete(struct ETCP_DGRAM* dgram, int success) { if (packet->last_attempt_id && packet->last_link) packet->last_link->total_retransmissions++; packet->send_hist[packet->send_count % 8] = link->local_link_id; if (packet->send_count != UINT32_MAX) packet->send_count++; + if (etcp_packet_link_used(packet, link->local_link_id)) { + memset(packet->used_links, 0, sizeof(packet->used_links)); + DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] retry starts new path round seq=%u link=%u", conn->log_name, + packet->seq, link->local_link_id); + } + packet->used_links[link->local_link_id >> 3] |= 1u << (link->local_link_id & 7); + conn->last_rr_link = link; packet->last_link = link; packet->last_timestamp = now / 100; packet->last_attempt_id = attempt->attempt_id; queue_remove_data(conn->input_send_q, &packet->ll); packet->state = INFLIGHT_STATE_WAIT_ACK; if (queue_data_put_with_index(conn->input_wait_ack, &packet->ll) < 0) { @@ -233,6 +253,11 @@ void etcp_tx_ack(struct ETCP_LINK* link, uint32_t seq, uint64_t attempt_id, uint } if (attempt) { struct ETCP_TX_ATTEMPT sample = *attempt; + if (!link->tx_stalled || sample.sent_at_us >= link->tx_stalled_since_us) { + if (link->tx_stalled) DEBUG_INFO(DEBUG_CATEGORY_ETCP, "[%s] DATA path recovered link=%u seq=%u attempt=%u", + conn->log_name, link->local_link_id, seq, (unsigned)(uint16_t)attempt_id); + link->data_rto_streak = link->tx_stalled = 0; link->tx_stalled_since_us = 0; + } int rx_valid = peer_rx_bytes && peer_rx_time_us; if (!!peer_rx_bytes != !!peer_rx_time_us || (rx_valid && link->peer_rx_data_time_us && ((peer_rx_bytes > link->peer_rx_data_bytes && peer_rx_time_us < link->peer_rx_data_time_us) ||