diff --git a/src/transport_layer/etcp_connections.h b/src/transport_layer/etcp_connections.h index d76454e4..675e575c 100644 --- a/src/transport_layer/etcp_connections.h +++ b/src/transport_layer/etcp_connections.h @@ -84,6 +84,12 @@ struct ETCP_DGRAM {// пакет (незашифрованный) // Размер базового ACK: type(1) + count(1) + last_delivered_id(4) + rx_dup_count(2) #define ETCP_ACK_BASE_SIZE (2 + 4 + 2) +#define ETCP_ACK_RX_POINT_SIZE 16 // При count>0 после entries: принятые DATA-байты u64 и время их приёма u64 (LE, мкс) + +// Размер ACK-секции для count=0..255; пустой ACK не содержит точки приёма и остаётся 8-байтовым. +static inline unsigned etcp_ack_section_size(uint8_t count) { + return ETCP_ACK_BASE_SIZE + (unsigned)count * ETCP_ACK_ENTRY_SIZE + (count ? ETCP_ACK_RX_POINT_SIZE : 0); +} // INIT внутри шифрованного тела, до padding: запрос — 67 байт, ответ — 64 байта. // Каждая сторона передаёт свою эпоху reset_id; смена известной эпохи подтверждается через etcp_session. @@ -341,6 +347,10 @@ struct ETCP_LINK { uint64_t delivered_bytes; // кумулятивно delivered bytes на линке (для round detection) uint64_t last_ack_time_us; // время последней доставки/начала первого flight в микросекундах uint64_t first_tx_time_us; // начало интервала отправок, продвигается подтверждёнными попытками + uint64_t rx_data_bytes; // допустимые физические DATA-байты пути, включая копии; сбрасывается с потоком + uint64_t rx_data_time_us; // время последнего увеличения rx_data_bytes, локальные монотонные мкс; 0 — точки нет + uint64_t peer_rx_data_bytes; // последняя достоверная точка из ACK действующей попытки этого пути + uint64_t peer_rx_data_time_us; // часы получателя, не сравниваются с локальными; 0 — нет базы текущего flight uint32_t bbr_pacing_rate; // Точный pacing B/s; менять через set_pacing_rate для пересчёта долга/таймера uint32_t bbr_loss_since_ack; // счётчик потерь на линке с последнего ACK (для rs->lost) diff --git a/src/transport_layer/etcp_tx.c b/src/transport_layer/etcp_tx.c index 9c4855b2..46363028 100644 --- a/src/transport_layer/etcp_tx.c +++ b/src/transport_layer/etcp_tx.c @@ -100,6 +100,8 @@ static void retransmit_cb(void* arg) { } int etcp_tx_link_init(struct ETCP_LINK* link) { + link->rx_data_bytes = link->rx_data_time_us = 0; + link->peer_rx_data_bytes = link->peer_rx_data_time_us = 0; 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) { @@ -129,6 +131,8 @@ void etcp_tx_link_reset(struct ETCP_LINK* link, int retry) { link->data_rtt_us = link->data_jitter_us = link->bbr_loss_since_ack = 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; + link->peer_rx_data_bytes = link->peer_rx_data_time_us = 0; link->last_ack_time_us = get_time_us(); link->first_tx_time_us = 0; link->cwnd_limited = 0; if (link->bbr && link->etcp->normalizer) { @@ -184,10 +188,15 @@ void etcp_tx_complete(struct ETCP_DGRAM* dgram, int success) { abort(); } uint64_t now = get_time_us(); - if (link->inflight_packets == 1) { link->first_tx_time_us = now; link->last_ack_time_us = now; } + if (link->inflight_packets == 1) { + link->first_tx_time_us = now; link->last_ack_time_us = now; + link->peer_rx_data_bytes = link->peer_rx_data_time_us = 0; + } attempt->sent_at_us = now; attempt->delivered_at_send = link->delivered_bytes; attempt->delivered_time_us = link->last_ack_time_us; attempt->first_tx_us = link->first_tx_time_us; attempt->inflight_at_send = link->inflight_bytes; + attempt->peer_rx_bytes_at_send = link->peer_rx_data_bytes; + attempt->peer_rx_time_at_send = link->peer_rx_data_time_us; struct PKTNORM* pn = conn->normalizer; attempt->is_app_limited = !conn->input_queue->count && conn->input_send_q->count == 1 && (!pn || (!pn->input->count && !pn->pending && !pn->data_ptr)); @@ -207,7 +216,8 @@ void etcp_tx_complete(struct ETCP_DGRAM* dgram, int success) { retransmit_arm(link); } -void etcp_tx_ack(struct ETCP_LINK* link, uint32_t seq, uint64_t attempt_id, uint32_t receiver_delay_us) { +void etcp_tx_ack(struct ETCP_LINK* link, uint32_t seq, uint64_t attempt_id, uint32_t receiver_delay_us, + uint64_t peer_rx_bytes, uint64_t peer_rx_time_us) { struct ETCP_CONN* conn = link->etcp; struct ETCP_TX_ATTEMPT* attempt = (struct ETCP_TX_ATTEMPT*)queue_find_data_by_index(link->attempts, &attempt_id); if (attempt && (attempt->seq != seq || !attempt->sent_at_us)) { @@ -217,6 +227,20 @@ void etcp_tx_ack(struct ETCP_LINK* link, uint32_t seq, uint64_t attempt_id, uint } if (attempt) { struct ETCP_TX_ATTEMPT sample = *attempt; + 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) || + (peer_rx_bytes < link->peer_rx_data_bytes && peer_rx_time_us > link->peer_rx_data_time_us) || + (peer_rx_bytes == link->peer_rx_data_bytes && peer_rx_time_us != link->peer_rx_data_time_us)))) { + DEBUG_WARN(DEBUG_CATEGORY_ETCP, "[%s] inconsistent RX point path=%u attempt=%llu point=%llu/%llu previous=%llu/%llu", + conn->log_name, link->local_link_id, (unsigned long long)attempt_id, + (unsigned long long)peer_rx_bytes, (unsigned long long)peer_rx_time_us, + (unsigned long long)link->peer_rx_data_bytes, (unsigned long long)link->peer_rx_data_time_us); + rx_valid = 0; + } + if (rx_valid && peer_rx_bytes > link->peer_rx_data_bytes) { + link->peer_rx_data_bytes = peer_rx_bytes; link->peer_rx_data_time_us = peer_rx_time_us; + } int limited = link->cwnd_limited || link->inflight_bytes >= link->inflight_lim_bytes; attempt_retire(link, attempt, 0, 0); uint64_t now = get_time_us(), elapsed = now - sample.sent_at_us; @@ -237,10 +261,28 @@ void etcp_tx_ack(struct ETCP_LINK* link, uint32_t seq, uint64_t attempt_id, uint uint64_t interval = ack_elapsed > send_elapsed ? ack_elapsed : send_elapsed; struct bbr_rate_sample rs = { .delivered = delivered > UINT32_MAX ? UINT32_MAX : (uint32_t)delivered, - .interval_us = interval > UINT32_MAX ? UINT32_MAX : (uint32_t)interval, + .interval_us = interval > UINT32_MAX ? 0 : (uint32_t)interval, .rtt_us = rtt, .acked_sacked = sample.ll.len, .prior_delivered = (uint32_t)sample.delivered_at_send, .tx_in_flight = sample.inflight_at_send, .lost = (int)link->bbr_loss_since_ack, .is_app_limited = sample.is_app_limited }; + uint64_t rx_bytes = 0, rx_us = 0; + if (rs.interval_us && rx_valid && sample.peer_rx_time_at_send && + link->peer_rx_data_bytes > sample.peer_rx_bytes_at_send && + link->peer_rx_data_time_us > sample.peer_rx_time_at_send) { + rx_bytes = link->peer_rx_data_bytes - sample.peer_rx_bytes_at_send; + rx_us = link->peer_rx_data_time_us - sample.peer_rx_time_at_send; + if (rx_us > UINT32_MAX) rs.interval_us = 0; + else { + uint64_t numerator = (uint64_t)rs.delivered * rx_us; + uint64_t bound = numerator / rx_bytes + !!(numerator % rx_bytes); + if (bound > UINT32_MAX) rs.interval_us = 0; + else if (bound > rs.interval_us) rs.interval_us = (uint32_t)bound; + } + } + if (interval > UINT32_MAX || (interval && !rs.interval_us)) + DEBUG_WARN(DEBUG_CATEGORY_ETCP, "[%s] BW interval unrepresentable path=%u attempt=%llu local_us=%llu RX=%llu/%llu", + conn->log_name, link->local_link_id, (unsigned long long)attempt_id, (unsigned long long)interval, + (unsigned long long)rx_bytes, (unsigned long long)rx_us); link->last_ack_time_us = now; if (sample.sent_at_us > link->first_tx_time_us) link->first_tx_time_us = sample.sent_at_us; if (link->bbr) { @@ -249,8 +291,10 @@ void etcp_tx_ack(struct ETCP_LINK* link, uint32_t seq, uint64_t attempt_id, uint etcp_link_update_inflight_lim(link, cwnd); etcp_loadbalancer_set_pacing_rate(link, pacing); } DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] ACK seq=%u attempt=%llu path=%u RTT_us=%u delay_us=%u sample=%u/%u " - "flight=%u cwnd=%u pace=%u RTO_us=%u", conn->log_name, seq, (unsigned long long)attempt_id, link->local_link_id, - rtt, receiver_delay_us, rs.delivered, rs.interval_us, link->inflight_bytes, link->inflight_lim_bytes, + "local_interval_us=%llu RX=%llu/%llu flight=%u cwnd=%u pace=%u RTO_us=%u", + conn->log_name, seq, (unsigned long long)attempt_id, link->local_link_id, + rtt, receiver_delay_us, rs.delivered, rs.interval_us, (unsigned long long)interval, + (unsigned long long)rx_bytes, (unsigned long long)rx_us, link->inflight_bytes, link->inflight_lim_bytes, link->bbr_pacing_rate, etcp_tx_rto_us(link)); link->acked_bytes += sample.ll.len; link->acked_packets++; link->bbr_loss_since_ack = 0; link->cwnd_limited = 0; diff --git a/src/transport_layer/etcp_tx.h b/src/transport_layer/etcp_tx.h index c5688a93..729255cc 100644 --- a/src/transport_layer/etcp_tx.h +++ b/src/transport_layer/etcp_tx.h @@ -23,6 +23,8 @@ struct ETCP_TX_ATTEMPT { uint64_t delivered_at_send; // доставленные DATA-байты этого канала на момент отправки uint64_t delivered_time_us; // начало ACK-интервала оценки скорости uint64_t first_tx_us; // начало интервала отправки + uint64_t peer_rx_bytes_at_send; // точка приёма, известная при отправке этой попытки + uint64_t peer_rx_time_at_send; // время этой точки на часах получателя; 0 — новая/неизвестная база uint32_t inflight_at_send; // окно после включения этой попытки uint8_t is_app_limited; // источник не имел ожидающих данных либо идёт PROBE_RTT }; @@ -44,8 +46,12 @@ struct ETCP_TX_ATTEMPT* etcp_tx_prepare(struct ETCP_LINK* link, struct INFLIGHT_ * Вызвать ровно один раз до освобождения dgram и до следующего poll; dgram->attempt обнуляется. */ void etcp_tx_complete(struct ETCP_DGRAM* dgram, int success); /* Принятый ACK: seq подтверждает логическую доставку; attempt_id обновляет окно/BBR только своего линка. - * Неизвестный старый ID не создаёт измерение и не вычитает окно. UINT32_MAX delay — RTT недостоверен. */ -void etcp_tx_ack(struct ETCP_LINK* link, uint32_t seq, uint64_t attempt_id, uint32_t receiver_delay_us); + * Неизвестный старый ID не создаёт измерение и не вычитает окно. UINT32_MAX delay — RTT недостоверен. + * peer_rx_bytes/time_us — согласованная точка приёма пути из ACK; 0/0 — неизвестна. + * Достоверная пара ограничивает BW темпом приёма, сохраняя границы send/ACK; RTT и доставка от неё не зависят. + * Непредставимый интервал не создаёт оценку BW. Пустой flight заново начинает приёмную базу, исключая простой. */ +void etcp_tx_ack(struct ETCP_LINK* link, uint32_t seq, uint64_t attempt_id, uint32_t receiver_delay_us, + uint64_t peer_rx_bytes, uint64_t peer_rx_time_us); /* Поставить ACK уже сохранённого/доставленного DATA в очередь этого линка; 0/-1, ошибка логируется. */ int etcp_tx_queue_ack(struct ETCP_LINK* link, uint32_t seq, uint64_t attempt_id);