From f50916170122767c95318f3d0fd3eb6a9f3e47e6 Mon Sep 17 00:00:00 2001 From: Evgeny Date: Wed, 4 Feb 2026 01:08:52 +0300 Subject: [PATCH] Fix: Packet normalizer now correctly fragments large packets - Changed frag_size from mtu-100 to data_pool->object_size (1500 bytes) - Added packet splitting for packets larger than fragment size - First fragment contains [2-byte size][data], subsequent fragments contain [data] - Unpacker correctly assembles packets from multiple fragments - packer_cb now processes all available packets in a loop instead of one at a time - Removed manual queue processing from test, now uses callbacks properly - All 17 tests pass including test_pkt_normalizer_standalone with 3000-byte packets --- src/pkt_normalizer.c | 265 +++++++++++++++---------- tests/test_pkt_normalizer_standalone.c | 38 ++-- 2 files changed, 185 insertions(+), 118 deletions(-) diff --git a/src/pkt_normalizer.c b/src/pkt_normalizer.c index 04909289..aea1b971 100644 --- a/src/pkt_normalizer.c +++ b/src/pkt_normalizer.c @@ -24,7 +24,9 @@ struct PKTNORM* pn_init(struct ETCP_CONN* etcp) { pn->etcp = etcp; pn->ua = etcp->instance->ua; - pn->frag_size = etcp->mtu - 100; // Use MTU as fixed packet size (adjust if needed) + // frag_size должен соответствовать размеру блока в data_pool + // data_pool->object_size обычно равен MTU (1500) + pn->frag_size = etcp->instance->data_pool->object_size; pn->tx_wait_time = 100; // Increased to ensure flush happens between packets pn->input = queue_new(pn->ua, 0); // No hash needed @@ -124,9 +126,14 @@ void pn_packer_send(struct PKTNORM* pn, uint8_t* data, uint16_t len) { static void packer_cb(struct ll_queue* q, void* arg) { struct PKTNORM* pn = (struct PKTNORM*)arg; if (!pn) return; - DEBUG_INFO(DEBUG_CATEGORY_ETCP, "pn_packer: packer_cb"); + + int input_count = queue_entry_count(pn->input); + DEBUG_INFO(DEBUG_CATEGORY_ETCP, "pn_packer: packer_cb called, input_count=%d", input_count); - queue_wait_threshold(pn->etcp->input_queue, 0, 0, etcp_input_ready_cb, pn); + // Process all available packets immediately + etcp_input_ready_cb(q, pn); + + DEBUG_INFO(DEBUG_CATEGORY_ETCP, "pn_packer: packer_cb finished"); } // Helper to send block to ETCP as ETCP_FRAGMENT @@ -174,59 +181,109 @@ static void etcp_input_ready_cb(struct ll_queue* q, void* arg) { struct PKTNORM* pn = (struct PKTNORM*)arg; if (!pn) return; - DEBUG_INFO(DEBUG_CATEGORY_ETCP, "pn_packer: etcp_input_ready_cb"); + int processed = 0; + int input_count = queue_entry_count(pn->input); + DEBUG_INFO(DEBUG_CATEGORY_ETCP, "pn_packer: etcp_input_ready_cb started, input_count=%d", input_count); + + // Process all available packets from pn->input + struct ll_entry* in_dgram; + int loop_count = 0; + while ((in_dgram = queue_data_get(pn->input)) != NULL) { + loop_count++; + processed++; + DEBUG_INFO(DEBUG_CATEGORY_ETCP, "pn_packer: processing packet %d, len=%d", loop_count, in_dgram->len); + // Ensure buffer is allocated + if (!pn->data) { + pn_buf_renew(pn); + if (!pn->data) { + DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "pn_packer: failed to allocate buffer"); + queue_dgram_free(in_dgram); + queue_entry_free(in_dgram); + continue; + } + } - struct ll_entry* in_dgram = queue_data_get(pn->input); - if (!in_dgram) { queue_resume_callback(pn->input); return; } + // Check if packet fits in current buffer (need 2 bytes for size header + packet data) + int space_needed = in_dgram->len + 2; + int space_available = pn->data_size - pn->data_ptr; - // Ensure buffer is allocated - if (!pn->data) { - pn_buf_renew(pn); - if (!pn->data) { - DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "pn_packer: failed to allocate buffer"); + if (space_available < space_needed && pn->data_ptr > 0) { + // Not enough space - flush current buffer first + pn_send_to_etcp(pn); + pn_buf_renew(pn); + space_available = pn->data_size - pn->data_ptr; + } + + if (space_available < space_needed) { + // Packet is too big for single fragment - need to split across multiple fragments + // First fragment: [2 bytes total_size][data_part1] + // Next fragments: [data_part2][data_part3]... + int total_size = in_dgram->len; + int bytes_copied = 0; + int is_first_fragment = 1; + + while (bytes_copied < total_size) { + if (!pn->data) { + pn_buf_renew(pn); + if (!pn->data) { + DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "pn_packer: failed to allocate buffer for split"); + break; + } + } + + int remaining = total_size - bytes_copied; + int chunk; + + if (is_first_fragment) { + // First fragment: include 2-byte size header + int space_for_data = pn->data_size - pn->data_ptr - 2; + chunk = (remaining < space_for_data) ? remaining : space_for_data; + + // Write size header + pn->data[pn->data_ptr++] = total_size & 0xFF; + pn->data[pn->data_ptr++] = (total_size >> 8) & 0xFF; + + // Copy first chunk of data + memcpy(pn->data + pn->data_ptr, in_dgram->dgram, chunk); + pn->data_ptr += chunk; + bytes_copied += chunk; + is_first_fragment = 0; + } else { + // Subsequent fragments: only data, no header + chunk = (remaining < pn->data_size - pn->data_ptr) ? remaining : pn->data_size - pn->data_ptr; + + memcpy(pn->data + pn->data_ptr, in_dgram->dgram + bytes_copied, chunk); + pn->data_ptr += chunk; + bytes_copied += chunk; + } + + // Send this fragment + pn_send_to_etcp(pn); + pn_buf_renew(pn); + } + queue_dgram_free(in_dgram); queue_entry_free(in_dgram); - queue_resume_callback(pn->input); - return; + continue; } - } - // Check if packet fits in current buffer (need 2 bytes for size header + packet data) - int space_needed = in_dgram->len + 2; - int space_available = pn->data_size - pn->data_ptr; + // Write size header (2 bytes, little-endian) + pn->data[pn->data_ptr++] = in_dgram->len & 0xFF; + pn->data[pn->data_ptr++] = (in_dgram->len >> 8) & 0xFF; - if (space_available < space_needed && pn->data_ptr > 0) { - // Not enough space - flush current buffer first - pn_send_to_etcp(pn); - pn_buf_renew(pn); - space_available = pn->data_size - pn->data_ptr; - } + // Write packet data + memcpy(pn->data + pn->data_ptr, in_dgram->dgram, in_dgram->len); + pn->data_ptr += in_dgram->len; - if (space_available < space_needed) { - // Packet is too big for buffer even when empty - DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "pn_packer: packet too big: len=%d, max=%d", - in_dgram->len, pn->data_size - 2); queue_dgram_free(in_dgram); queue_entry_free(in_dgram); - queue_resume_callback(pn->input); - return; - } - - // Write size header (2 bytes, little-endian) - pn->data[pn->data_ptr++] = in_dgram->len & 0xFF; - pn->data[pn->data_ptr++] = (in_dgram->len >> 8) & 0xFF; - // Write packet data - memcpy(pn->data + pn->data_ptr, in_dgram->dgram, in_dgram->len); - pn->data_ptr += in_dgram->len; - - queue_dgram_free(in_dgram); - queue_entry_free(in_dgram); - - // Send immediately to ensure packet is not lost - pn_send_to_etcp(pn); - pn_buf_renew(pn); + // Send immediately to ensure packet is not lost + pn_send_to_etcp(pn); + pn_buf_renew(pn); + } + DEBUG_INFO(DEBUG_CATEGORY_ETCP, "pn_packer: etcp_input_ready_cb finished, processed=%d", processed); queue_resume_callback(pn->input); } @@ -257,68 +314,76 @@ void pn_flush(struct PKTNORM* pn) { // Internal: Unpacker callback (assembles fragments into original packets) static void pn_unpacker_cb(struct ll_queue* q, void* arg) { struct PKTNORM* pn = (struct PKTNORM*)arg; - if (!pn) return; - - struct ETCP_FRAGMENT* frag = (struct ETCP_FRAGMENT*)queue_data_get(pn->etcp->output_queue); - if (!frag) return; - uint8_t* payload = frag->ll.dgram; - uint16_t len = frag->ll.len; - uint16_t ptr = 0; - - DEBUG_INFO(DEBUG_CATEGORY_ETCP, "pn_unpacker: frag len=%d, recvpart=%p, recvpart_rem=%d", - len, (void*)pn->recvpart, pn->recvpart_rem); - - while (ptr < len) { - - if (pn->recvpart_rem==0) { - if (len - ptr < 2) { - DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "pn_unpacker: incomplete header, len=%d, ptr=%d", len, ptr); - pn_unpacker_reset_state(pn); - break; + if (!pn) { + queue_resume_callback(q); + return; + } + + // Process all available fragments from output_queue + struct ETCP_FRAGMENT* frag; + while ((frag = (struct ETCP_FRAGMENT*)queue_data_get(pn->etcp->output_queue)) != NULL) { + uint8_t* payload = frag->ll.dgram; + uint16_t len = frag->ll.len; + uint16_t ptr = 0; + + DEBUG_INFO(DEBUG_CATEGORY_ETCP, "pn_unpacker: frag len=%d, recvpart=%p, recvpart_rem=%d", + len, (void*)pn->recvpart, pn->recvpart_rem); + + while (ptr < len) { + + if (pn->recvpart_rem==0) { + // New packet - read size header (2 bytes) + if (len - ptr < 2) { + DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "pn_unpacker: incomplete header, len=%d, ptr=%d", len, ptr); + pn_unpacker_reset_state(pn); + break; + } + uint16_t pkt_len = payload[ptr] | (payload[ptr + 1] << 8); + ptr += 2; + + DEBUG_INFO(DEBUG_CATEGORY_ETCP, "pn_unpacker: new packet, pkt_len=%d", pkt_len); + + if (pkt_len == 0) { + // Пустой пакет - пропускаем + continue; + } + + pn->recvpart = ll_alloc_lldgram(pkt_len); + if (!pn->recvpart) { + DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "pn_unpacker: failed to alloc recvpart"); + pn_unpacker_reset_state(pn); + break; + } + pn->recvpart->len = 0; + pn->recvpart_rem = pkt_len; // Сколько байт осталось собрать } - uint16_t pkt_len = payload[ptr] | (payload[ptr + 1] << 8); - ptr += 2; - - DEBUG_INFO(DEBUG_CATEGORY_ETCP, "pn_unpacker: new packet, pkt_len=%d", pkt_len); - if (pkt_len == 0) { - // Пустой пакет - пропускаем - continue; + // Копируем данные в recvpart + uint16_t rem = pn->recvpart_rem; + uint16_t avail = len - ptr; + uint16_t cp = (rem < avail) ? rem : avail; + + DEBUG_INFO(DEBUG_CATEGORY_ETCP, "pn_unpacker: copying cp=%d bytes (rem=%d, avail=%d)", cp, rem, avail); + + if (pn->recvpart) memcpy(pn->recvpart->dgram + pn->recvpart->len, payload + ptr, cp); + pn->recvpart->len += cp; + pn->recvpart_rem -= cp; + ptr += cp; + + // Если пакет полностью собран + if (pn->recvpart_rem == 0) { + DEBUG_INFO(DEBUG_CATEGORY_ETCP, "pn_unpacker: packet complete, len=%d", pn->recvpart->len); + if (pn->recvpart) { + queue_data_put(pn->output, pn->recvpart, 0); + pn->recvpart = NULL; // Сбросить указатель после передачи + } } - - pn->recvpart = ll_alloc_lldgram(pkt_len); - if (!pn->recvpart) { - DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "pn_unpacker: failed to alloc recvpart"); - } - pn->recvpart->len = 0; - pn->recvpart_rem = pkt_len; // Сколько байт осталось собрать } - // Копируем данные в recvpart - uint16_t rem = pn->recvpart_rem; - uint16_t avail = len - ptr; - uint16_t cp = (rem < avail) ? rem : avail; - - DEBUG_INFO(DEBUG_CATEGORY_ETCP, "pn_unpacker: copying cp=%d bytes (rem=%d, avail=%d)", cp, rem, avail); - - if (pn->recvpart) memcpy(pn->recvpart->dgram + pn->recvpart->len, payload + ptr, cp); - pn->recvpart->len += cp; - pn->recvpart_rem -= cp; - ptr += cp; - - // Если пакет полностью собран - if (pn->recvpart_rem == 0) { - DEBUG_INFO(DEBUG_CATEGORY_ETCP, "pn_unpacker: packet complete, len=%d", pn->recvpart->len); - if (pn->recvpart) { - queue_data_put(pn->output, pn->recvpart, 0); - pn->recvpart = NULL; // Сбросить указатель после передачи - } - } + // Free the fragment using ll_queue API + queue_dgram_free(&frag->ll); + queue_entry_free(&frag->ll); } - // Free the fragment using ll_queue API - queue_dgram_free(&frag->ll); - queue_entry_free(&frag->ll); - queue_resume_callback(q); } diff --git a/tests/test_pkt_normalizer_standalone.c b/tests/test_pkt_normalizer_standalone.c index 4a10822e..eebe951f 100644 --- a/tests/test_pkt_normalizer_standalone.c +++ b/tests/test_pkt_normalizer_standalone.c @@ -5,6 +5,7 @@ #include #include +#include "../lib/debug_config.h" #include "../lib/u_async.h" #include "../lib/ll_queue.h" #include "../lib/memory_pool.h" @@ -29,6 +30,8 @@ static int packets_sent = 0; static int packets_received = 0; static int packet_sizes[TOTAL_PACKETS]; static struct timespec start_time, end_time; +static int fragments_sent = 0; +static int fragments_received = 0; // Generate packet data with pattern static void generate_packet_data(int seq, uint8_t* buffer, int size) { @@ -61,12 +64,13 @@ static int verify_packet_data(uint8_t* buffer, int size, int expected_seq) { return 1; } -// Loopback callback: moves one fragment from input_queue to output_queue +// Loopback callback: moves all fragments from input_queue to output_queue static void loopback_callback(struct ll_queue* q, void* arg) { (void)arg; - struct ETCP_FRAGMENT* frag = (struct ETCP_FRAGMENT*)queue_data_get(q); - if (frag) { + struct ETCP_FRAGMENT* frag; + while ((frag = (struct ETCP_FRAGMENT*)queue_data_get(q)) != NULL) { + fragments_sent++; // Move fragment from input to output (loopback) queue_data_put(mock_etcp.output_queue, (struct ll_entry*)frag, 0); } @@ -105,6 +109,7 @@ static void check_received_packets(void) { void* data; while ((data = queue_data_get(pn->output)) != NULL) { struct ll_entry* entry = (struct ll_entry*)data; + fragments_received++; if (entry->len >= 4) { int seq = entry->dgram[0] | (entry->dgram[1] << 8); @@ -123,26 +128,13 @@ static void check_received_packets(void) { } } -// Monitor function +// Monitor function - only checks received packets, callbacks handle data flow static void monitor(void* arg) { (void)arg; if (test_completed) return; - // Trigger loopback processing manually - // In real ETCP, this would be triggered by the queue callback - if (mock_etcp.input_queue) { - void* frag_data; - while ((frag_data = queue_data_get(mock_etcp.input_queue)) != NULL) { - struct ETCP_FRAGMENT* frag = (struct ETCP_FRAGMENT*)frag_data; - // Move fragment from input to output (loopback) - if (mock_etcp.output_queue) { - queue_data_put(mock_etcp.output_queue, (struct ll_entry*)frag, 0); - } - } - } - - // Check received packets + // Check received packets - unpacker callback puts them in pn->output check_received_packets(); // Check if all packets received @@ -154,6 +146,7 @@ static void monitor(void* arg) { test_completed = 1; printf("\n=== SUCCESS: All packets received! ===\n"); printf("Sent: %d, Received: %d\n", packets_sent, packets_received); + printf("Fragments: sent=%d, received=%d\n", fragments_sent, fragments_received); printf("Duration: %.2f ms\n", duration); return; } @@ -168,6 +161,10 @@ static void test_timeout(void* arg) { if (!test_completed) { printf("\n=== TIMEOUT ===\n"); printf("Sent: %d/%d, Received: %d/%d\n", packets_sent, TOTAL_PACKETS, packets_received, TOTAL_PACKETS); + printf("Fragments: sent=%d, received=%d\n", fragments_sent, fragments_received); + printf("Input queue count: %d, Output queue count: %d\n", + queue_entry_count(mock_etcp.input_queue), + queue_entry_count(mock_etcp.output_queue)); test_completed = 2; } } @@ -254,6 +251,10 @@ int main() { printf("Total data to transfer: %d bytes (%.2f KB average per packet)\n\n", total_bytes, (float)total_bytes / TOTAL_PACKETS / 1024); + // Enable debug output + debug_set_level(DEBUG_LEVEL_INFO); + debug_enable_category(DEBUG_CATEGORY_ETCP); + // Initialize UASYNC mock_instance.ua = uasync_create(); if (!mock_instance.ua) { @@ -321,6 +322,7 @@ int main() { } else { printf("\n=== TEST FAILED ===\n"); printf("Sent: %d/%d, Received: %d/%d\n", packets_sent, TOTAL_PACKETS, packets_received, TOTAL_PACKETS); + printf("Fragments: sent=%d, received=%d\n", fragments_sent, fragments_received); return 1; } }