Implement per-path attempt accounting and isolated ACK delivery samples
This commit is contained in:
evgeny committed
1 parent
b68e0ac3db
commit
7bc2ac5a46
3 files changed
+261
No files matched your search
@@ -79,6 +79,7 @@ utun_CORE_SOURCES = \
|
||||
transport_layer/stun_nat.h \
|
||||
transport_layer/etcp_keepalive.c \
|
||||
transport_layer/etcp_bbr.c \
|
||||
transport_layer/etcp_tx.c \
|
||||
transport_layer/etcp_loadbalancer.c \
|
||||
transport_layer/etcp_debug.c \
|
||||
transport_layer/etcp_dump.c \
|
||||
@@ -265,6 +266,7 @@ libutun_a_SOURCES = \
|
||||
transport_layer/stun_nat.h \
|
||||
transport_layer/etcp_keepalive.c \
|
||||
transport_layer/etcp_bbr.c \
|
||||
transport_layer/etcp_tx.c \
|
||||
transport_layer/etcp_loadbalancer.c \
|
||||
transport_layer/etcp_debug.c \
|
||||
transport_layer/etcp_dump.c \
|
||||
|
||||
@@ -252,6 +252,7 @@ struct ETCP_LINK {
|
||||
struct ll_queue* attempts; // Физические попытки, индекс attempt_id:u64; ll.len учитывается в окне
|
||||
struct ll_queue* ack_q; // ACK только DATA этого линка, индекс attempt_id:u64
|
||||
void* retrans_timer; // Срок старейшей принятой транспортом попытки этого канала
|
||||
uint64_t retrans_deadline_us; // Уже запланированная проверка: более ранний таймер можно сохранить
|
||||
uint32_t data_rtt_us; // Сглаженный RTT DATA->ACK этого пути; 0 до первого измерения
|
||||
uint32_t data_jitter_us; // Сглаженное абсолютное отклонение собственного DATA RTT
|
||||
|
||||
|
||||
@@ -0,0 +1,258 @@
|
||||
#include "etcp.h"
|
||||
#include "etcp_bbr.h"
|
||||
#include "etcp_loadbalancer.h"
|
||||
#include "../utun_instance.h"
|
||||
#include "../lib/debug_config.h"
|
||||
#include <limits.h>
|
||||
#include <stdlib.h>
|
||||
|
||||
static void retransmit_cb(void* arg);
|
||||
|
||||
static struct INFLIGHT_PACKET* logical_find(struct ETCP_CONN* conn, uint32_t seq) {
|
||||
struct INFLIGHT_PACKET* packet = (struct INFLIGHT_PACKET*)queue_find_data_by_index(conn->input_wait_ack, &seq);
|
||||
if (!packet) packet = (struct INFLIGHT_PACKET*)queue_find_data_by_index(conn->input_send_q, &seq);
|
||||
return packet;
|
||||
}
|
||||
|
||||
uint32_t etcp_tx_rto_us(const struct ETCP_LINK* link) {
|
||||
uint64_t rtt = link->data_rtt_us ? link->data_rtt_us : (uint64_t)link->rtt_last * 100;
|
||||
uint64_t rto = rtt ? 2 * (rtt + link->data_jitter_us) : 1000000;
|
||||
if (rto < 10000) rto = 10000;
|
||||
if (rto > 1000000) rto = 1000000;
|
||||
return (uint32_t)rto;
|
||||
}
|
||||
|
||||
static void retransmit_arm(struct ETCP_LINK* link) {
|
||||
if (!link->attempts || link->etcp->close_requested || link->etcp->state == 2) return;
|
||||
struct ETCP_TX_ATTEMPT* first = (struct ETCP_TX_ATTEMPT*)link->attempts->head;
|
||||
if (!first || !first->sent_at_us) {
|
||||
if (link->retrans_timer) {
|
||||
uasync_cancel_timeout(link->etcp->instance->ua, link->retrans_timer); link->retrans_timer = NULL;
|
||||
}
|
||||
return;
|
||||
}
|
||||
uint64_t deadline = first->sent_at_us + etcp_tx_rto_us(link);
|
||||
if (link->retrans_timer && link->retrans_deadline_us <= deadline) return;
|
||||
if (link->retrans_timer) {
|
||||
uasync_cancel_timeout(link->etcp->instance->ua, link->retrans_timer); link->retrans_timer = NULL;
|
||||
}
|
||||
uint64_t now = get_time_us(), delay = deadline > now ? (deadline - now + 99) / 100 : 0;
|
||||
link->retrans_deadline_us = deadline;
|
||||
link->retrans_timer = uasync_set_timeout(link->etcp->instance->ua, (int)delay, link, retransmit_cb, "etcp_attempt_rto");
|
||||
if (!link->retrans_timer) {
|
||||
struct ETCP_CONN* conn = link->etcp;
|
||||
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] cannot schedule attempt RTO link=%u delay_tb=%llu",
|
||||
conn->log_name, link->local_link_id, (unsigned long long)delay);
|
||||
conn->callbacks_running++; etcp_connection_close(conn); conn->callbacks_running--;
|
||||
}
|
||||
}
|
||||
|
||||
/* Удаление физической записи не удаляет payload и не подтверждает его логически. */
|
||||
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)) {
|
||||
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--;
|
||||
if (packet) packet->active_attempts--;
|
||||
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;
|
||||
bbr_note_loss(link->bbr);
|
||||
}
|
||||
DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] retire link=%u attempt=%llu seq=%u reason=%s flight=%u/%u logical_active=%u",
|
||||
conn->log_name, link->local_link_id, (unsigned long long)attempt->attempt_id, attempt->seq,
|
||||
loss ? "RTO" : retry ? "link_close" : "ACK/rollback", link->inflight_bytes, link->inflight_packets,
|
||||
packet ? packet->active_attempts : 0);
|
||||
queue_remove_data(link->attempts, &attempt->ll); queue_entry_free(&attempt->ll);
|
||||
if (retry && packet && !packet->active_attempts && packet->state == INFLIGHT_STATE_WAIT_ACK && !conn->close_requested) {
|
||||
queue_remove_data(conn->input_wait_ack, &packet->ll);
|
||||
packet->state = INFLIGHT_STATE_WAIT_SEND;
|
||||
if (packet->retrans_req_count != UINT32_MAX) packet->retrans_req_count++;
|
||||
if (queue_data_put_with_index(conn->input_send_q, &packet->ll) < 0) {
|
||||
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] failed to queue retry after retiring attempt", conn->log_name);
|
||||
conn->callbacks_running++; etcp_connection_close(conn); conn->callbacks_running--;
|
||||
}
|
||||
conn->retransmissions_count++;
|
||||
etcp_conn_wake_sender(conn);
|
||||
}
|
||||
}
|
||||
|
||||
static void retransmit_cb(void* arg) {
|
||||
struct ETCP_LINK* link = arg;
|
||||
link->retrans_timer = NULL;
|
||||
uint64_t now = get_time_us();
|
||||
uint32_t rto = etcp_tx_rto_us(link);
|
||||
while (link->attempts->head) {
|
||||
struct ETCP_TX_ATTEMPT* attempt = (struct ETCP_TX_ATTEMPT*)link->attempts->head;
|
||||
if (!attempt->sent_at_us || now - attempt->sent_at_us < rto) break;
|
||||
DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] RTO link=%u attempt=%llu seq=%u age_us=%llu RTO_us=%u RTT_us=%u jitter_us=%u",
|
||||
link->etcp->log_name, link->local_link_id, (unsigned long long)attempt->attempt_id, attempt->seq,
|
||||
(unsigned long long)(now - attempt->sent_at_us), rto, link->data_rtt_us, link->data_jitter_us);
|
||||
attempt_retire(link, attempt, 1, 1);
|
||||
}
|
||||
retransmit_arm(link);
|
||||
}
|
||||
|
||||
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;
|
||||
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);
|
||||
link->attempts = link->ack_q = NULL;
|
||||
return -1;
|
||||
}
|
||||
|
||||
void etcp_tx_link_reset(struct ETCP_LINK* link, int retry) {
|
||||
if (link->retrans_timer) {
|
||||
uasync_cancel_timeout(link->etcp->instance->ua, link->retrans_timer); link->retrans_timer = NULL;
|
||||
}
|
||||
while (link->attempts && link->attempts->head)
|
||||
attempt_retire(link, (struct ETCP_TX_ATTEMPT*)link->attempts->head, 0, retry);
|
||||
while (link->ack_q && link->ack_q->head) queue_entry_free(queue_data_get(link->ack_q));
|
||||
if (link->shaper_timer) {
|
||||
uasync_cancel_timeout(link->etcp->instance->ua, link->shaper_timer); link->shaper_timer = NULL;
|
||||
}
|
||||
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;
|
||||
link->delivered_bytes = link->acked_bytes = link->acked_packets = 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) {
|
||||
uint32_t budget = link->bbr->queue_budget_us;
|
||||
bbr_init(link->bbr); link->bbr->queue_budget_us = budget;
|
||||
etcp_link_update_inflight_lim(link, link->bbr->init_cwnd * link->etcp->normalizer->frag_size);
|
||||
}
|
||||
}
|
||||
|
||||
void etcp_tx_link_destroy(struct ETCP_LINK* link) {
|
||||
etcp_tx_link_reset(link, !link->etcp->close_requested && link->etcp->state != 2);
|
||||
if (link->attempts) queue_free(link->attempts);
|
||||
if (link->ack_q) queue_free(link->ack_q);
|
||||
link->attempts = link->ack_q = NULL;
|
||||
}
|
||||
|
||||
struct ETCP_TX_ATTEMPT* etcp_tx_prepare(struct ETCP_LINK* link, struct INFLIGHT_PACKET* packet) {
|
||||
struct ETCP_CONN* conn = link->etcp;
|
||||
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;
|
||||
}
|
||||
struct ETCP_TX_ATTEMPT* attempt = (struct ETCP_TX_ATTEMPT*)queue_entry_new_from_pool(conn->attempt_pool);
|
||||
if (!attempt) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] cannot allocate transmission attempt", conn->log_name); return NULL; }
|
||||
attempt->attempt_id = conn->next_attempt_id++; attempt->seq = packet->seq; attempt->ll.len = packet->ll.len;
|
||||
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++;
|
||||
return attempt;
|
||||
}
|
||||
|
||||
void etcp_tx_complete(struct ETCP_DGRAM* dgram, int success) {
|
||||
struct ETCP_TX_ATTEMPT* attempt = dgram->attempt;
|
||||
if (!attempt) return;
|
||||
dgram->attempt = NULL;
|
||||
struct ETCP_LINK* link = dgram->link;
|
||||
struct ETCP_CONN* conn = link->etcp;
|
||||
struct INFLIGHT_PACKET* packet = logical_find(conn, attempt->seq);
|
||||
if (!success) { attempt_retire(link, attempt, 0, 0); return; }
|
||||
if (!packet || packet->state != INFLIGHT_STATE_WAIT_SEND) {
|
||||
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] committed attempt has no ready logical packet seq=%u", conn->log_name, attempt->seq);
|
||||
abort();
|
||||
}
|
||||
uint64_t now = get_time_us();
|
||||
if (link->inflight_packets == 1) { link->first_tx_time_us = now; link->last_ack_time_us = now; }
|
||||
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;
|
||||
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));
|
||||
if (link->bbr && link->bbr->mode == BBR_PROBE_RTT) attempt->is_app_limited = 1;
|
||||
DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] sent seq=%u attempt=%llu path=%u previous=%u bytes=%u flight=%u/%u",
|
||||
conn->log_name, packet->seq, (unsigned long long)attempt->attempt_id, link->local_link_id,
|
||||
packet->last_link ? packet->last_link->local_link_id : 0, packet->ll.len, link->inflight_bytes, link->inflight_packets);
|
||||
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++;
|
||||
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) {
|
||||
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] cannot commit logical transmission", conn->log_name);
|
||||
conn->callbacks_running++; etcp_connection_close(conn); conn->callbacks_running--; return;
|
||||
}
|
||||
retransmit_arm(link);
|
||||
}
|
||||
|
||||
void etcp_tx_ack(struct ETCP_LINK* link, uint32_t seq, uint64_t attempt_id, uint32_t receiver_delay_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)) {
|
||||
DEBUG_WARN(DEBUG_CATEGORY_ETCP, "[%s] invalid attempt ACK path=%u id=%llu seq=%u expected=%u committed=%d",
|
||||
conn->log_name, link->local_link_id, (unsigned long long)attempt_id, seq, attempt->seq, !!attempt->sent_at_us);
|
||||
return;
|
||||
}
|
||||
if (attempt) {
|
||||
struct ETCP_TX_ATTEMPT sample = *attempt;
|
||||
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;
|
||||
uint32_t rtt = receiver_delay_us != UINT32_MAX && receiver_delay_us < elapsed ?
|
||||
(elapsed - receiver_delay_us > UINT32_MAX ? UINT32_MAX : (uint32_t)(elapsed - receiver_delay_us)) : 0;
|
||||
if (rtt) {
|
||||
if (!link->data_rtt_us) { link->data_rtt_us = rtt; link->data_jitter_us = rtt / 2; }
|
||||
else {
|
||||
int64_t delta = (int64_t)rtt - link->data_rtt_us;
|
||||
uint32_t deviation = delta < 0 ? (uint32_t)-delta : (uint32_t)delta;
|
||||
link->data_jitter_us += ((int64_t)deviation - link->data_jitter_us) / 4;
|
||||
link->data_rtt_us += delta / 8;
|
||||
}
|
||||
}
|
||||
link->delivered_bytes += sample.ll.len;
|
||||
uint64_t delivered = link->delivered_bytes - sample.delivered_at_send;
|
||||
uint64_t ack_elapsed = now - sample.delivered_time_us, send_elapsed = sample.sent_at_us - sample.first_tx_us;
|
||||
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,
|
||||
.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
|
||||
};
|
||||
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) {
|
||||
uint32_t cwnd = link->inflight_lim_bytes, pacing = link->bbr_pacing_rate;
|
||||
bbr_main(link->bbr, &rs, &cwnd, &pacing, conn->normalizer->frag_size, link->inflight_bytes, limited);
|
||||
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,
|
||||
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;
|
||||
retransmit_arm(link);
|
||||
} else DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "[%s] old/foreign ACK seq=%u attempt=%llu path=%u: logical only",
|
||||
conn->log_name, seq, (unsigned long long)attempt_id, link->local_link_id);
|
||||
etcp_ack_recv(conn, seq);
|
||||
etcp_conn_wake_sender(conn);
|
||||
}
|
||||
|
||||
int etcp_tx_queue_ack(struct ETCP_LINK* link, uint32_t seq, uint64_t attempt_id) {
|
||||
if (!attempt_id) { DEBUG_WARN(DEBUG_CATEGORY_ETCP, "[%s] DATA attempt_id=0 rejected", link->etcp->log_name); return -1; }
|
||||
if (queue_find_data_by_index(link->ack_q, &attempt_id)) return 0;
|
||||
struct ACK_PACKET* ack = (struct ACK_PACKET*)queue_entry_new_from_pool(link->etcp->instance->ack_pool);
|
||||
if (!ack) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] cannot allocate path ACK", link->etcp->log_name); return -1; }
|
||||
ack->attempt_id = attempt_id; ack->seq = seq; ack->received_at_us = get_time_us();
|
||||
if (queue_data_put_with_index(link->ack_q, &ack->ll) < 0) {
|
||||
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] cannot index path ACK", link->etcp->log_name); return -1;
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
Reference in new issue
Block a user