diff --git a/src/transport_layer/etcp.h b/src/transport_layer/etcp.h index a4e9dd2a..032bda23 100644 --- a/src/transport_layer/etcp.h +++ b/src/transport_layer/etcp.h @@ -163,6 +163,7 @@ struct ETCP_CONN { // IDs and state uint32_t next_tx_id; // ID для добавления в очередь отправки (с этим id будет добавлен следующий пакет) uint64_t next_attempt_id; // Не переиспользуется при reset/пересоздании линков; исчерпание закрывает соединение + uint32_t active_attempts; // Все физические записи, включая резерв send; максимум 2*MAX_INFLIGHT_SIZE uint32_t last_rx_id; // Last received ID uint32_t last_delivered_id; // Last delivered to output_queue diff --git a/src/transport_layer/etcp_tx.c b/src/transport_layer/etcp_tx.c index ff6f5438..975f7291 100644 --- a/src/transport_layer/etcp_tx.c +++ b/src/transport_layer/etcp_tx.c @@ -51,13 +51,14 @@ static void retransmit_arm(struct ETCP_LINK* link) { static void attempt_retire(struct ETCP_LINK* link, struct ETCP_TX_ATTEMPT* attempt, int loss, int retry) { struct ETCP_CONN* conn = link->etcp; struct INFLIGHT_PACKET* packet = logical_find(conn, attempt->seq); - if (link->inflight_bytes < attempt->ll.len || !link->inflight_packets || (packet && !packet->active_attempts)) { + if (!conn->active_attempts || link->inflight_bytes < attempt->ll.len || !link->inflight_packets || + (packet && !packet->active_attempts)) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] attempt accounting mismatch link=%u id=%llu seq=%u flight=%u/%u len=%u logical=%u", conn->log_name, link->local_link_id, (unsigned long long)attempt->attempt_id, attempt->seq, link->inflight_bytes, link->inflight_packets, attempt->ll.len, packet ? packet->active_attempts : 0); abort(); } - link->inflight_bytes -= attempt->ll.len; link->inflight_packets--; + link->inflight_bytes -= attempt->ll.len; link->inflight_packets--; conn->active_attempts--; if (packet) packet->active_attempts--; if (loss && link->bbr) { uint64_t lost = (uint64_t)link->bbr_loss_since_ack + attempt->ll.len; @@ -101,7 +102,11 @@ static void retransmit_cb(void* arg) { int etcp_tx_link_init(struct ETCP_LINK* link) { link->attempts = queue_new(link->etcp->instance->ua, 128, 0, 8, "ETCP attempts"); link->ack_q = queue_new(link->etcp->instance->ua, 128, 0, 8, "ETCP path ACK"); - if (link->attempts && link->ack_q) return 0; + if (link->attempts && link->ack_q) { + queue_set_size_limit(link->attempts, MAX_INFLIGHT_SIZE); + queue_set_size_limit(link->ack_q, MAX_INFLIGHT_SIZE); + return 0; + } DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] cannot initialize link attempt/ACK queues", link->etcp->log_name); if (link->attempts) queue_free(link->attempts); if (link->ack_q) queue_free(link->ack_q); @@ -142,6 +147,11 @@ void etcp_tx_link_destroy(struct ETCP_LINK* link) { struct ETCP_TX_ATTEMPT* etcp_tx_prepare(struct ETCP_LINK* link, struct INFLIGHT_PACKET* packet) { struct ETCP_CONN* conn = link->etcp; + if (conn->active_attempts >= 2 * MAX_INFLIGHT_SIZE || link->attempts->count >= MAX_INFLIGHT_SIZE) { + DEBUG_WARN(DEBUG_CATEGORY_ETCP, "[%s] transmission attempt budget exhausted: total=%u path=%u count=%d", + conn->log_name, conn->active_attempts, link->local_link_id, link->attempts->count); + return NULL; + } if (!conn->next_attempt_id || conn->next_attempt_id == UINT64_MAX) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] transmission attempt ID exhausted", conn->log_name); conn->callbacks_running++; etcp_connection_close(conn); conn->callbacks_running--; return NULL; @@ -152,7 +162,7 @@ struct ETCP_TX_ATTEMPT* etcp_tx_prepare(struct ETCP_LINK* link, struct INFLIGHT_ if (queue_data_put_with_index(link->attempts, &attempt->ll) < 0) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] cannot index transmission attempt", conn->log_name); return NULL; } - link->inflight_bytes += packet->ll.len; link->inflight_packets++; packet->active_attempts++; + link->inflight_bytes += packet->ll.len; link->inflight_packets++; packet->active_attempts++; conn->active_attempts++; return attempt; }