Browse Source
- Fixed all incompatible pointer type warnings in src/etcp.c - Fixed warnings in src/pkt_normalizer.c - Fixed warnings in tests/test_etcp_simple_traffic.c - Fixed warnings in tests/test_etcp_100_packets.c - Fixed warnings in tests/test_ll_queue.c - Fixed DEBUG_CATEGORY_ALL overflow warning in debug_config.h - Fixed DEBUG_CATEGORY_LL_QUEUE redefinition warning in test_ll_queue.c - Fixed write() unused result warning in test_u_async_comprehensive.cnodeinfo-routing-update
Evgeny
2 months ago
23 changed files with 514 additions and 226 deletions
@ -0,0 +1,297 @@
|
||||
// pkt_normalizer.c - Implementation of packet normalizer for ETCP |
||||
#include "pkt_normalizer.h" |
||||
#include "etcp.h" // For ETCP_CONN and related structures |
||||
#include "ll_queue.h" // For queue operations |
||||
#include "u_async.h" // For UASYNC |
||||
#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) { |
||||
if (!data) return NULL; |
||||
return (struct ll_entry*)data; |
||||
} |
||||
|
||||
// Forward declarations |
||||
static void packer_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); |
||||
|
||||
// Initialization |
||||
struct PKTNORM* pn_init(struct ETCP_CONN* etcp) { |
||||
if (!etcp) return NULL; |
||||
|
||||
struct PKTNORM* pn = calloc(1, sizeof(struct PKTNORM)); |
||||
if (!pn) return NULL; |
||||
|
||||
pn->etcp = etcp; |
||||
pn->ua = etcp->instance->ua; |
||||
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 |
||||
|
||||
if (!pn->input || !pn->output) { |
||||
pn_pair_deinit(pn); |
||||
return NULL; |
||||
} |
||||
|
||||
queue_set_callback(pn->input, packer_cb, pn); |
||||
queue_set_callback(etcp->output_queue, pn_unpacker_cb, pn); |
||||
|
||||
pn->sndpart.dgram = NULL; |
||||
pn->sndpart.len = 0; |
||||
pn->recvpart = NULL; |
||||
pn->flush_timer = NULL; |
||||
|
||||
return pn; |
||||
} |
||||
|
||||
// Deinitialization |
||||
void pn_pair_deinit(struct PKTNORM* pn) { |
||||
if (!pn) return; |
||||
|
||||
// Drain and free queues |
||||
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_entry_free(data); |
||||
} |
||||
queue_free(pn->input); |
||||
} |
||||
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_entry_free(data); |
||||
} |
||||
queue_free(pn->output); |
||||
} |
||||
|
||||
if (pn->flush_timer) { |
||||
uasync_cancel_timeout(pn->ua, pn->flush_timer); |
||||
} |
||||
|
||||
if (pn->sndpart.dgram) { |
||||
queue_entry_free(&pn->sndpart); |
||||
} |
||||
if (pn->recvpart) { |
||||
queue_dgram_free(pn->recvpart); |
||||
queue_entry_free(pn->recvpart); |
||||
} |
||||
|
||||
free(pn); |
||||
} |
||||
|
||||
// Reset unpacker state |
||||
void pn_unpacker_reset_state(struct PKTNORM* pn) { |
||||
if (!pn) return; |
||||
if (pn->recvpart) { |
||||
queue_dgram_free(pn->recvpart); |
||||
queue_entry_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; |
||||
|
||||
DEBUG_INFO(DEBUG_CATEGORY_ETCP, "pn_packer_send: pn=%p, len=%d", pn, len); |
||||
|
||||
struct ll_entry* entry = ll_alloc_lldgram(len); |
||||
memcpy(entry->dgram, data, len); |
||||
entry->len = len; |
||||
entry->dgram_pool = NULL; |
||||
|
||||
int ret = queue_data_put(pn->input, entry, 0); |
||||
DEBUG_INFO(DEBUG_CATEGORY_ETCP, "pn_packer_send: queue_data_put returned %d, input count=%d", ret, queue_entry_count(pn->input)); |
||||
|
||||
// Cancel flush timer if active |
||||
if (pn->flush_timer) { |
||||
uasync_cancel_timeout(pn->ua, pn->flush_timer); |
||||
pn->flush_timer = NULL; |
||||
} |
||||
} |
||||
|
||||
// Internal: Packer callback |
||||
static void packer_cb(struct ll_queue* q, void* arg) { |
||||
struct PKTNORM* pn = (struct PKTNORM*)arg; |
||||
if (!pn) return; |
||||
|
||||
queue_wait_threshold(pn->etcp->input_queue, 0, 0, etcp_input_ready_cb, pn); |
||||
} |
||||
|
||||
// Helper to send sndpart to ETCP as ETCP_FRAGMENT |
||||
static void pn_send_to_etcp(struct PKTNORM* pn) { |
||||
if (!pn || !pn->sndpart.data || !pn->sndpart.len==0) return; |
||||
|
||||
// Allocate ETCP_FRAGMENT from rx_pool |
||||
struct ETCP_FRAGMENT* frag = queue_entry_new_from_pool(pn->etcp->rx_pool); |
||||
if (!frag) {// drop data |
||||
pn->sndpart.len = 0; |
||||
return; |
||||
} |
||||
|
||||
frag->seq = 0; |
||||
frag->timestamp = 0; |
||||
frag->ll.dgram = pn->sndpart.dgram; |
||||
frag->ll.len = pn->sndpart.len; |
||||
frag->ll.memlen = pn->sndpart.memlen; |
||||
|
||||
queue_data_put(pn->etcp->input_queue, frag, 0); |
||||
queue_entry_free(&pn->sndpart); |
||||
// Сбросить структуру после освобождения |
||||
pn->sndpart.dgram = NULL; |
||||
pn->sndpart.len = 0; |
||||
pn->sndpart.memlen = 0; |
||||
} |
||||
|
||||
// Internal: Renew sndpart buffer |
||||
static void pn_buf_renew(struct PKTNORM* pn) { |
||||
if (pn->sndpart.dgram) { |
||||
int remain = pn->frag_size - pn->sndpart.len; |
||||
if (remain < 3) pn_send_to_etcp(pn); |
||||
} |
||||
if (!pn->sndpart.dgram) { |
||||
pn->sndpart.len=0; |
||||
pn->sndpart.dgram_pool = pn->etcp->instance->data_pool; |
||||
pn->sndpart.memlen=pn->etcp->instance->data_pool->object_size;//pn->frag_size; |
||||
pn->sndpart.dgram = memory_pool_alloc(pn->etcp->instance->data_pool); |
||||
} |
||||
} |
||||
|
||||
// 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; |
||||
} |
||||
|
||||
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.dgram) break; // Allocation failed |
||||
|
||||
int remain = pn->frag_size - pn->sndpart.len; |
||||
if (remain < 3) { |
||||
// Буфер почти полон, отправить его |
||||
pn_send_to_etcp(pn); |
||||
continue; // Продолжить с новым буфером |
||||
} |
||||
|
||||
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; |
||||
|
||||
// Проверить, не заполнился ли буфер |
||||
if (pn->sndpart.len >= pn->frag_size - 2) { |
||||
pn_send_to_etcp(pn); |
||||
} |
||||
} |
||||
|
||||
queue_dgram_free(in_dgram); |
||||
queue_entry_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 pn_flush_cb(void* arg) { |
||||
struct PKTNORM* pn = (struct PKTNORM*)arg; |
||||
if (!pn) return; |
||||
|
||||
pn->flush_timer = NULL; |
||||
pn_send_to_etcp(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; |
||||
|
||||
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); |
||||
} |
||||
|
||||
queue_resume_callback(q); |
||||
} |
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Loading…
Reference in new issue