diff --git a/src/transport_layer/etcp.c b/src/transport_layer/etcp.c index c9715703..15ddfee2 100644 --- a/src/transport_layer/etcp.c +++ b/src/transport_layer/etcp.c @@ -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", diff --git a/src/transport_layer/etcp_debug.c b/src/transport_layer/etcp_debug.c index 3f40620f..543946ed 100644 --- a/src/transport_layer/etcp_debug.c +++ b/src/transport_layer/etcp_debug.c @@ -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: {