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

Test: Added test_pkt_normalizer_etcp for testing pkt_normalizer with ETCP

Browse Source 
- Created test_pkt_normalizer_etcp.c based on test_etcp_100_packets
- Tests bidirectional transfer of 100 packets (10-10000 bytes) via normalizer
- Fixed memory management bugs in pkt_normalizer.c:
  * Fixed double-free in pn_buf_renew()
  * Added pn_send_to_etcp() to properly create ETCP_FRAGMENT
  * Fixed memory freeing in pn_unpacker_cb()
- Added test to Makefile.am
This commit is contained in:
Evgeny committed 2026-02-01 22:31:25 +03:00
1 parent dd4644776d
commit add785c8a1
4 files changed
+623 -271
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/pkt_normalizer.c
+197 -243
View File
@@ -6,6 +6,7 @@
#include <stdlib.h>
#include <string.h>
#include <stdio.h> // For debugging (can be removed if not needed)
#include "debug_config.h" // Assuming this for DEBUG_ERROR
// Internal helper to convert void* data to struct ll_entry*
static inline struct ll_entry* data_to_entry(void* data) {
@@ -15,8 +16,10 @@ static inline struct ll_entry* data_to_entry(void* data) {
// Forward declarations
static void packer_cb(struct ll_queue* q, void* arg);
static void flush_cb(void* arg);
static void input_ready_cb(struct ll_queue* q, void* arg);
static void pn_flush_cb(void* arg);
static void etcp_input_ready_cb(struct ll_queue* q, void* arg);
static void pn_unpacker_cb(struct ll_queue* q, void* arg);
static void pn_send_to_etcp(struct PKTNORM* pn, struct ll_entry* entry);
// Initialization
struct PKTNORM* pn_init(struct ETCP_CONN* etcp) {
@@ -27,31 +30,22 @@ struct PKTNORM* pn_init(struct ETCP_CONN* etcp) {
pn->etcp = etcp;
pn->ua = etcp->instance->ua;
pn->pkt_size = etcp->mtu; // Use MTU as fixed packet size (adjust if headers need subtraction)
pn->frag_size = etcp->mtu - 100; // Use MTU as fixed packet size (adjust if headers need subtraction)
pn->tx_wait_time = 10;
pn->input = queue_new(pn->ua, 0); // No hash needed
pn->output = queue_new(pn->ua, 0); // No hash needed
pn->pending = queue_new(pn->ua, 0);
pn->send_pending = queue_new(pn->ua, 0);
if (!pn->input || !pn->output || !pn->pending || !pn->send_pending) {
if (!pn->input || !pn->output) {
pn_pair_deinit(pn);
return NULL;
}
pn->in_buf = calloc(1, pn->pkt_size);
pn->cap = pn->pkt_size; // For packer buffer (fixed size)
pn->len = 0; // Current length in packer buffer
pn->out_buf = calloc(1, pn->pkt_size * 10); // Arbitrary large buffer for assembly
pn->cap = pn->pkt_size * 10; // Initial capacity for unpacker (can reallocate if needed)
pn->len = 0;
pn->total_len = 0;
pn->in_fragment = 0;
// Set callback for automatic processing when items are added to input
queue_set_callback(pn->input, packer_cb, pn);
queue_set_callback(etcp->output_queue, pn_unpacker_cb, pn);
pn->sndpart = NULL;
pn->recvpart = NULL;
pn->flush_timer = NULL;
return pn;
@@ -65,6 +59,10 @@ void pn_pair_deinit(struct PKTNORM* pn) {
if (pn->input) {
void* data;
while ((data = queue_data_get(pn->input)) != NULL) {
struct ll_entry* entry = data_to_entry(data);
if (entry->dgram) {
free(entry->dgram);
}
queue_data_free(data);
}
queue_free(pn->input);
@@ -72,284 +70,240 @@ void pn_pair_deinit(struct PKTNORM* pn) {
if (pn->output) {
void* data;
while ((data = queue_data_get(pn->output)) != NULL) {
struct ll_entry* entry = data_to_entry(data);
if (entry->dgram) {
free(entry->dgram);
}
queue_data_free(data);
}
queue_free(pn->output);
}
if (pn->pending) {
void* data;
while ((data = queue_data_get(pn->pending)) != NULL) {
queue_data_free(data);
}
queue_free(pn->pending);
}
if (pn->send_pending) {
void* data;
while ((data = queue_data_get(pn->send_pending)) != NULL) {
struct ETCP_FRAGMENT* frag = data_to_entry(data);
if (frag->ll.dgram) memory_pool_free(pn->etcp->instance->data_pool, frag->ll.dgram);
memory_pool_free(pn->etcp->rx_pool, frag);
}
queue_free(pn->send_pending);
}
if (pn->flush_timer) {
uasync_cancel_timeout(pn->ua, pn->flush_timer);
}
free(pn->in_buf);
free(pn->out_buf);
if (pn->sndpart) {
ll_free_dgram(pn->sndpart);
queue_data_free(pn->sndpart);
}
if (pn->recvpart) {
ll_free_dgram(pn->recvpart);
queue_data_free(pn->recvpart);
}
free(pn);
}
// Reset unpacker state
void pn_unpacker_reset_state(struct PKTNORM* pn) {
if (!pn) return;
pn->len = 0;
pn->total_len = 0;
pn->in_fragment = 0;
if (pn->recvpart) {
ll_free_dgram(pn->recvpart);
queue_data_free(pn->recvpart);
pn->recvpart = NULL;
}
}
// Send data to packer (copies and adds to input queue or pending, triggering callback)
void pn_packer_send(struct PKTNORM* pn, uint8_t* data, uint16_t len) {
if (!pn || !data || len == 0) return;
void* entry_data = queue_data_new(len);
if (!entry_data) return;
struct ll_entry* entry = data_to_entry(entry_data);
memcpy(entry->data, data, len);
struct ll_entry* entry = ll_alloc_lldgram(len);
memcpy(entry->dgram, data, len);
entry->len = len;
entry->size = len;
entry->dgram_pool = NULL;
// To minimize delay, add to input only if empty, else to pending and set waiter if not set
if (queue_entry_count(pn->input) == 0 && pn->input->waiter.callback == NULL) {
queue_data_put(pn->input, entry_data, 0);
} else {
queue_data_put(pn->pending, entry_data, 0);
if (pn->input->waiter.callback == NULL) {
queue_wait_threshold(pn->input, 0, 0, input_ready_cb, pn);
}
queue_data_put(pn->input, entry, 0);
// Cancel flush timer if active
if (pn->flush_timer) {
uasync_cancel_timeout(pn->ua, pn->flush_timer);
pn->flush_timer = NULL;
}
}
// Internal: Callback when input queue becomes empty, move next from pending
static void input_ready_cb(struct ll_queue* q, void* arg) {
struct PKTNORM* pn = (struct PKTNORM*)arg;
void* data = queue_data_get(pn->pending);
if (data) {
queue_data_put(q, data, 0); // This will trigger packer_cb async if needed
if (queue_entry_count(pn->pending) > 0) {
queue_wait_threshold(q, 0, 0, input_ready_cb, pn);
}
}
}
static void etcp_input_ready_cb(struct ll_queue* q, void* arg) {
struct PKTNORM* pn = (struct PKTNORM*)arg;
void* data = queue_data_get(pn->send_pending);
if (data) {
queue_data_put(q, data, 0);
if (queue_entry_count(pn->send_pending) > 0) {
queue_wait_threshold(q, 0, 0, etcp_input_ready_cb, pn);
}
}
}
// Internal: Send a fixed-size chunk to etcp->input_queue (or send_pending)
static void send_chunk(struct PKTNORM* pn, uint8_t* chunk_data, uint16_t chunk_len) {
// Alloc ETCP_FRAGMENT for the chunk (as payload for etcp)
struct ETCP_FRAGMENT* frag = memory_pool_alloc(pn->etcp->rx_pool);
if (!frag) return;
frag->ll.dgram = memory_pool_alloc(pn->etcp->instance->data_pool);
if (!frag->ll.dgram) {
memory_pool_free(pn->etcp->rx_pool, frag);
return;
}
memcpy(frag->ll.dgram, chunk_data, chunk_len);
frag->ll.size = chunk_len;
frag->ll.len = chunk_len;
frag->seq = 0; // Not used for input
frag->timestamp = 0;
struct ll_queue* eq = pn->etcp->input_queue;
// Add to etcp input_queue only if empty, else to send_pending and set waiter
if (queue_entry_count(eq) == 0 && eq->waiter.callback == NULL) {
queue_data_put(eq, frag, 0);
} else {
queue_data_put(pn->send_pending, frag, 0);
if (eq->waiter.callback == NULL) {
queue_wait_threshold(eq, 0, 0, etcp_input_ready_cb, pn);
}
}
}
// Internal: Packer callback (aggregates small, fragments large, sends chunks)
// Internal: Packer callback
static void packer_cb(struct ll_queue* q, void* arg) {
struct PKTNORM* pn = (struct PKTNORM*)arg;
if (!pn) return;
pn->len = 0; // Start with empty buffer for this batch
queue_wait_threshold(pn->etcp->input_queue, 0, 0, etcp_input_ready_cb, pn);
}
while (q->head) {
void* data = queue_data_get(q);
struct ll_entry* entry = data_to_entry(data);
uint16_t item_len = entry->len;
// Helper to send sndpart to ETCP as ETCP_FRAGMENT
static void pn_send_to_etcp(struct PKTNORM* pn, struct ll_entry* entry) {
if (!pn || !entry || entry->len == 0) return;
// Allocate data
uint8_t* packet_data = malloc(entry->len);
if (!packet_data) return;
memcpy(packet_data, entry->dgram, entry->len);
// Allocate ETCP_FRAGMENT from rx_pool
struct ETCP_FRAGMENT* frag = memory_pool_alloc(pn->etcp->rx_pool);
if (!frag) {
free(packet_data);
return;
}
frag->seq = 0;
frag->timestamp = 0;
frag->ll.dgram = packet_data;
frag->ll.size = entry->len;
frag->ll.len = entry->len;
frag->ll.memlen = entry->len;
frag->ll.dgram_pool = NULL;
queue_data_put(pn->etcp->input_queue, frag, 0);
}
if (item_len + 2 > pn->pkt_size) {
// Large item: fragment (must be alone in buffer)
if (pn->len > 0) {
// Flush current aggregated small items first
memset(pn->in_buf + pn->len, 0, pn->pkt_size - pn->len); // Pad
send_chunk(pn, pn->in_buf, pn->pkt_size);
pn->len = 0;
// Internal: Renew sndpart buffer
static void pn_buf_renew(struct PKTNORM* pn) {
if (pn->sndpart) {
int remain = pn->frag_size - pn->sndpart->len;
if (remain < 3) {
if (pn->sndpart->len > 0) {
// Transfer ownership to ETCP queue
pn_send_to_etcp(pn, pn->sndpart);
}
// Fragment the large item
uint8_t* ptr = entry->data;
uint32_t remaining = item_len;
// First fragment
*(uint16_t*)(pn->in_buf) = 0xffff;
*(uint32_t*)(pn->in_buf + 2) = remaining;
uint16_t chunk = pn->pkt_size - 6;
memcpy(pn->in_buf + 6, ptr, chunk);
send_chunk(pn, pn->in_buf, pn->pkt_size);
ptr += chunk;
remaining -= chunk;
// Continuation fragments
while (remaining > 0) {
chunk = (remaining < pn->pkt_size - 2) ? remaining : pn->pkt_size - 2;
*(uint16_t*)(pn->in_buf) = 0x0000; // Continuation marker
memcpy(pn->in_buf + 2, ptr, chunk);
memset(pn->in_buf + 2 + chunk, 0, pn->pkt_size - 2 - chunk); // Pad
send_chunk(pn, pn->in_buf, pn->pkt_size);
ptr += chunk;
remaining -= chunk;
}
} else {
// Small item: try to aggregate
if (pn->len + item_len + 2 > pn->pkt_size) {
// Flush current buffer
memset(pn->in_buf + pn->len, 0, pn->pkt_size - pn->len); // Pad
send_chunk(pn, pn->in_buf, pn->pkt_size);
pn->len = 0;
}
// Add to buffer
*(uint16_t*)(pn->in_buf + pn->len) = item_len;
memcpy(pn->in_buf + pn->len + 2, entry->data, item_len);
pn->len += item_len + 2;
// Free the ll_entry (but not dgram - it's copied in pn_send_to_etcp)
ll_free_dgram(pn->sndpart);
queue_data_free(pn->sndpart);
pn->sndpart = NULL;
}
}
if (!pn->sndpart) {
pn->sndpart = ll_alloc_lldgram(pn->frag_size);
if (pn->sndpart) {
pn->sndpart->len = 0;
}
}
}
queue_data_free(data); // Free the entry
// Internal: Process input when etcp->input_queue is ready (empty)
static void etcp_input_ready_cb(struct ll_queue* q, void* arg) {
struct PKTNORM* pn = (struct PKTNORM*)arg;
if (!pn) return;
void* data = queue_data_get(pn->input);
if (!data) {
queue_resume_callback(pn->input);
return;
}
// If remaining in buffer, set timeout to flush if queues empty
if (pn->len > 0) {
if (pn->flush_timer) {
uasync_cancel_timeout(pn->ua, pn->flush_timer);
struct ll_entry* in_dgram = data_to_entry(data);
uint16_t ptr = 0;
while (ptr < in_dgram->len) {
pn_buf_renew(pn);
if (!pn->sndpart) break; // Allocation failed
int remain = pn->frag_size - pn->sndpart->len;
if (remain < 3) {
DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "etcp_normalizer: part size error, remain=%d", remain);
break;
}
pn->flush_timer = uasync_set_timeout(pn->ua, 10, pn, flush_cb); // 1ms = 10 * 0.1ms
if (ptr == 0) {
pn->sndpart->dgram[pn->sndpart->len++] = in_dgram->len & 0xFF;
pn->sndpart->dgram[pn->sndpart->len++] = (in_dgram->len >> 8) & 0xFF;
remain -= 2;
}
int n = remain;
int rem = in_dgram->len - ptr;
if (n > rem) n = rem;
memcpy(pn->sndpart->dgram + pn->sndpart->len, in_dgram->dgram + ptr, n);
pn->sndpart->len += n;
ptr += n;
pn_buf_renew(pn);
}
queue_resume_callback(q);
ll_free_dgram(in_dgram);
queue_data_free(data);
// Cancel flush timer if active
if (pn->flush_timer) {
uasync_cancel_timeout(pn->ua, pn->flush_timer);
pn->flush_timer = NULL;
}
// Set flush timer if no more input
if (queue_entry_count(pn->input) == 0) {
pn->flush_timer = uasync_set_timeout(pn->ua, pn->tx_wait_time, pn, pn_flush_cb);
}
queue_resume_callback(pn->input);
}
// Internal: Flush callback on timeout
static void flush_cb(void* arg) {
static void pn_flush_cb(void* arg) {
struct PKTNORM* pn = (struct PKTNORM*)arg;
if (!pn) return;
pn->flush_timer = NULL;
if (pn->len > 0 && queue_entry_count(pn->input) == 0 && queue_entry_count(pn->pending) == 0) {
memset(pn->in_buf + pn->len, 0, pn->pkt_size - pn->len); // Pad
send_chunk(pn, pn->in_buf, pn->pkt_size);
pn->len = 0;
if (pn->sndpart && pn->sndpart->len > 0) {
pn_send_to_etcp(pn, pn->sndpart);
// Free the ll_entry after sending
ll_free_dgram(pn->sndpart);
queue_data_free(pn->sndpart);
pn->sndpart = NULL; // Will alloc new when needed
}
}
// Internal: Add assembled data to etcp->output_queue as ETCP_FRAGMENT
static void add_to_output(struct PKTNORM* pn, uint8_t* app_data, uint16_t app_len) {
struct ETCP_FRAGMENT* frag = memory_pool_alloc(pn->etcp->rx_pool);
if (!frag) return;
// 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;
frag->ll.dgram = memory_pool_alloc(pn->etcp->instance->data_pool);
if (!frag->ll.dgram) {
while (1) {
void* data = queue_data_get(pn->etcp->output_queue);
if (!data) break;
struct ETCP_FRAGMENT* frag = (struct ETCP_FRAGMENT*)data; // Since data is ll_entry*
uint8_t* payload = frag->ll.dgram;
uint16_t len = frag->ll.len;
uint16_t ptr = 0;
while (ptr < len) {
if (!pn->recvpart) {
// Need length header for new packet
if (len - ptr < 2) {
// Incomplete header, reset
pn_unpacker_reset_state(pn);
break;
}
uint16_t pkt_len = payload[ptr] | (payload[ptr + 1] << 8);
ptr += 2;
pn->recvpart = ll_alloc_lldgram(pkt_len);
if (!pn->recvpart) {
break;
}
pn->recvpart->len = 0;
}
uint16_t rem = pn->recvpart->memlen - pn->recvpart->len;
uint16_t avail = len - ptr;
uint16_t cp = (rem < avail) ? rem : avail;
memcpy(pn->recvpart->dgram + pn->recvpart->len, payload + ptr, cp);
pn->recvpart->len += cp;
ptr += cp;
if (pn->recvpart->len == pn->recvpart->memlen) {
queue_data_put(pn->output, pn->recvpart, 0);
pn->recvpart = NULL;
}
}
// Free the fragment - dgram was malloc'd in pn_send_to_etcp
free(frag->ll.dgram);
memory_pool_free(pn->etcp->rx_pool, frag);
return;
}
memcpy(frag->ll.dgram, app_data, app_len);
frag->ll.size = app_len;
frag->ll.len = app_len;
frag->seq = 0;
frag->timestamp = 0;
queue_data_put(pn->etcp->output_queue, frag, 0);
}
// Internal: Process incoming fixed-size payload chunk from etcp
void pn_unpacker_input(struct PKTNORM* pn, uint8_t* data, uint16_t len) {
if (!pn || len != pn->pkt_size) return;
uint16_t header = *(uint16_t*)data;
if (pn->in_fragment) {
// Expect continuation
if (header != 0x0000) {
// Error: invalid continuation
pn_unpacker_reset_state(pn);
return;
}
uint16_t chunk = pn->pkt_size - 2;
if (pn->len + chunk > pn->cap) {
pn->cap *= 2;
pn->out_buf = realloc(pn->out_buf, pn->cap);
}
memcpy(pn->out_buf + pn->len, data + 2, chunk);
pn->len += chunk;
if (pn->len >= pn->total_len) {
// Assembly complete: add to output_queue
add_to_output(pn, pn->out_buf, pn->total_len);
pn_unpacker_reset_state(pn);
}
} else {
if (header == 0xffff) {
// Start of fragment
pn->total_len = *(uint32_t*)(data + 2);
uint16_t chunk = pn->pkt_size - 6;
if (pn->total_len <= chunk) {
// Invalid or degenerate case
pn_unpacker_reset_state(pn);
return;
}
memcpy(pn->out_buf, data + 6, chunk);
pn->len = chunk;
pn->in_fragment = 1;
} else {
// Normal aggregated packet: process multiple items (ignore pad)
uint8_t* ptr = data;
size_t processed = 0;
while (processed < pn->pkt_size) {
header = *(uint16_t*)ptr;
if (header == 0 || header == 0xffff || header == 0x0000) break; // End (pad) or error
if (processed + 2 + header > pn->pkt_size) break;
add_to_output(pn, ptr + 2, header);
ptr += 2 + header;
processed += 2 + header;
}
}
}
queue_resume_callback(q);
}
src/pkt_normalizer.h
+12 -28
View File
@@ -8,31 +8,24 @@
// Структура для packer
struct PKTNORM {
// public:
// public:
struct ll_queue* input; // Входная очередь в packer (через нее отправляем пакеты)
struct ll_queue* output; // Выходная очередь из unpacker (через нее принимаем пакеты)
uint16_t tx_wait_time;
// private:
struct ETCP_CONN* etcp;
uint16_t pkt_size;
// packer:
// private:
uasync_t* ua; // uasync instance
uint8_t* in_buf; // Буфер для упаковки
size_t len; // Текущая длина в буфере
size_t cap; // Емкость буфера (MAX_PACKET_SIZE)
struct ETCP_CONN* etcp;
// packer:
uint16_t frag_size; // размер фрагмента на которые разбивать
struct ll_entry* sndpart; // блок ожидающий досборки
void* flush_timer; // For timeout flush
struct ll_queue* pending; // For not filling input
// unpacker:
uint8_t* out_buf; // Буфер для сборки фрагментов
// size_t len; // Текущая накопленная длина
size_t total_len; // Ожидаемая общая длина (для фрагментов)
// size_t cap; // Емкость буфера
int in_fragment; // Флаг: идет сборка фрагмента (1) или нет (0)
struct ll_queue* send_pending; // For not filling etcp->input_queue
struct ll_entry* pending; // Partial processed input entry
uint16_t pending_in_ptr; // Pointer in pending entry
// unpacker:
struct ll_entry* recvpart; // блок ожидающий заполнение
};
// Инициализация пары
@@ -47,13 +40,4 @@ void pn_unpacker_reset_state(struct PKTNORM* pn);
// создаёт malloc data, копирует, помещает в input.
void pn_packer_send(struct PKTNORM* pn, uint8_t* data, uint16_t len);
/* Как работает:
Формат отправки в etcp: 2 байта размер, далее данные (порезанные на куски и отправленные через etcp)
собирает по возможности полные пакеты с размером pkt_size.
+ timeout: неполный пакет отправляется по таймауту 1ms (если очереди пустые)
+ входящая очередь etcp не должне наполняться для минимизации задержки - новый пакет отправляем только когда очередь пустая
*/
#endif // PKT_NORMALIZER_H