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

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
nodeinfo-routing-update
Evgeny 8 months ago
parent
5840e77e1e
commit
f509161701
2 changed files with 185 additions and 118 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 265
      src/pkt_normalizer.c
  2. 38
      tests/test_pkt_normalizer_standalone.c

265
src/pkt_normalizer.c
Unescape View File

@ -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);
}

38
tests/test_pkt_normalizer_standalone.c
Unescape View File

@ -5,6 +5,7 @@
#include <time.h>
#include <unistd.h>
#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;
}
}

Write Preview
Loading…
Cancel
Save