jeka/utun2
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Carry receiver progress only in ACKs with attempt confirmations

Browse Source
This commit is contained in:
evgeny committed 2026-10-23 16:23:20 +02:00
1 parent 5586528555
commit ea4a92125b
2 files changed
+38 -5
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/transport_layer/etcp.c
+24 -4
View File
@@ -1064,15 +1064,22 @@ struct ETCP_DGRAM* etcp_request_pkt(struct ETCP_CONN* conn) {
data[0] = ETCP_SECTION_ACK; data[1] = 0;
write_le(data + 2, conn->last_delivered_id, 4); write_le(data + 6, conn->rx_dup_count, 2);
int ptr = ETCP_ACK_BASE_SIZE, remaining = max_enc - ptr - payload_size - dgram->ack_request;
while (remaining >= ETCP_ACK_ENTRY_SIZE && data[1] != UINT8_MAX) {
int ack_space = remaining - ETCP_ACK_RX_POINT_SIZE;
while (ack_space >= ETCP_ACK_ENTRY_SIZE && data[1] != UINT8_MAX) {
struct ACK_PACKET* ack = (struct ACK_PACKET*)queue_data_get(link->ack_q);
if (!ack) break;
write_le(data + ptr, ack->seq, 4); write_le(data + ptr + 4, ack->attempt_id, 8);
uint64_t delay = get_time_us() - ack->received_at_us;
write_le(data + ptr + 12, delay >= UINT32_MAX ? UINT32_MAX : delay, 4);
ptr += ETCP_ACK_ENTRY_SIZE; remaining -= ETCP_ACK_ENTRY_SIZE; data[1]++;
ptr += ETCP_ACK_ENTRY_SIZE; remaining -= ETCP_ACK_ENTRY_SIZE; ack_space -= ETCP_ACK_ENTRY_SIZE; data[1]++;
queue_entry_free(&ack->ll);
}
if (data[1]) {
write_le(data + ptr, link->rx_data_bytes, 8); write_le(data + ptr + 8, link->rx_data_time_us, 8);
ptr += ETCP_ACK_RX_POINT_SIZE; remaining -= ETCP_ACK_RX_POINT_SIZE;
DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] ACK point path=%u count=%u RX=%llu/%llu", conn->log_name,
link->local_link_id, data[1], (unsigned long long)link->rx_data_bytes, (unsigned long long)link->rx_data_time_us);
}
if (dgram->ack_request) data[ptr++] = ETCP_SECTION_ACK_REQUEST;
if (link->burst_resp_pending && !link->burst_active && remaining >= MEAS_RESP_SECTION_SIZE) {
data[ptr++] = ETCP_SECTION_MEAS_RESP;
@@ -1239,7 +1246,7 @@ void etcp_conn_input(struct ETCP_DGRAM* pkt) {
switch (type) {
case ETCP_SECTION_ACK: {
if (len < ETCP_ACK_BASE_SIZE) { DEBUG_WARN(DEBUG_CATEGORY_ETCP, "[%s] short ACK len=%u", etcp->log_name, len); len = 0; break; }
unsigned count = data[1], section_len = ETCP_ACK_BASE_SIZE + count * ETCP_ACK_ENTRY_SIZE;
unsigned count = data[1], section_len = etcp_ack_section_size(data[1]);
if (section_len > len) {
DEBUG_WARN(DEBUG_CATEGORY_ETCP, "[%s] ACK exceeds packet count=%u len=%u", etcp->log_name, count, len);
len = 0; break;
@@ -1247,10 +1254,14 @@ void etcp_conn_input(struct ETCP_DGRAM* pkt) {
uint32_t till = (uint32_t)read_le(data + 2, 4), last_tx = etcp->next_tx_id - 1;
if ((int32_t)(till - last_tx) > 0) till = last_tx;
uint16_t duplicates = (uint16_t)read_le(data + 6, 2);
unsigned point_offset = ETCP_ACK_BASE_SIZE + count * ETCP_ACK_ENTRY_SIZE;
uint64_t rx_bytes = count ? read_le(data + point_offset, 8) : 0;
uint64_t rx_time = count ? read_le(data + point_offset + 8, 8) : 0;
etcp->tx_dup_count += (int16_t)(duplicates - (uint16_t)etcp->tx_dup_count);
for (unsigned i = 0; i < count; i++) {
const uint8_t* entry = data + ETCP_ACK_BASE_SIZE + i * ETCP_ACK_ENTRY_SIZE;
etcp_tx_ack(pkt->link, (uint32_t)read_le(entry, 4), read_le(entry + 4, 8), (uint32_t)read_le(entry + 12, 4));
etcp_tx_ack(pkt->link, (uint32_t)read_le(entry, 4), read_le(entry + 4, 8),
(uint32_t)read_le(entry + 12, 4), rx_bytes, rx_time);
if (etcp->close_requested || etcp->reinit_pending) break;
}
uint32_t advanced = 0;
@@ -1345,6 +1356,12 @@ void etcp_conn_input(struct ETCP_DGRAM* pkt) {
etcp->log_name, seq, etcp->last_delivered_id, payload_len);
len = 0; break;
}
if (pkt->link->rx_data_bytes > UINT64_MAX - payload_len) {
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] RX byte counter exhausted path=%u seq=%u", etcp->log_name,
pkt->link->local_link_id, seq);
etcp->callbacks_running++; etcp_connection_close(etcp); etcp->callbacks_running--;
len = 0; break;
}
int duplicate = distance <= 0 || queue_find_data_by_index(etcp->recv_q, &seq) != NULL;
if (!duplicate) {
struct ETCP_FRAGMENT* fragment = (struct ETCP_FRAGMENT*)queue_entry_new_from_pool(etcp->io_pool);
@@ -1364,6 +1381,9 @@ void etcp_conn_input(struct ETCP_DGRAM* pkt) {
len = 0; break;
}
} else etcp->rx_dup_count++;
if (payload_len) {
pkt->link->rx_data_bytes += payload_len; pkt->link->rx_data_time_us = get_time_us();
}
etcp->got_initial_pkt = 1;
if (etcp_tx_queue_ack(pkt->link, seq, attempt_id) == 0) ack_schedule(etcp);
DEBUG_DEBUG(DEBUG_CATEGORY_ETCP, "[%s] accepted DATA seq=%u attempt=%llu path=%u bytes=%u duplicate=%d",
src/transport_layer/etcp_debug.c
+14 -1
View File
@@ -47,7 +47,20 @@ void etcp_dump_pkt_sections(struct ETCP_DGRAM* pkt, struct ETCP_LINK* link, int
((uint32_t)data[offset + 4] << 16) | ((uint32_t)data[offset + 5] << 24);
uint16_t ack_rx_dup = data[offset + 6] | (data[offset + 7] << 8);
pos += snprintf(buf + pos, sizeof(buf) - pos, " ack=%u/%d dup=%u", ack_last_delivered, ack_count, ack_rx_dup);
offset += ETCP_ACK_BASE_SIZE + ack_count * ETCP_ACK_ENTRY_SIZE;
unsigned section_len = etcp_ack_section_size((uint8_t)ack_count);
if (offset + section_len > len) {
DEBUG_WARN(DEBUG_CATEGORY_DUMP, "Short ACK section: count=%d remaining=%u required=%u",
ack_count, len-offset, section_len);
offset = len; break;
}
if (ack_count) {
const uint8_t* point = data + offset + ETCP_ACK_BASE_SIZE + ack_count * ETCP_ACK_ENTRY_SIZE;
uint64_t bytes = 0, time = 0;
for (unsigned i = 0; i < 8; i++) { bytes |= (uint64_t)point[i] << (8*i); time |= (uint64_t)point[i+8] << (8*i); }
pos += snprintf(buf + pos, sizeof(buf) - pos, " rx=%llu/%lluus",
(unsigned long long)bytes, (unsigned long long)time);
}
offset += section_len;
break;
}
case ETCP_SECTION_TIMESTAMP: {