This commit is contained in:
Evgeny committed 2026-02-02 00:46:12 +03:00
1 parent 9a5db10939
commit e9ebab4037
7 files changed
+64 -9

No files matched your search

+19 -5
View File
@@ -109,13 +109,15 @@ void pn_unpacker_reset_state(struct PKTNORM* pn) {
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;
queue_data_put(pn->input, entry, 0);
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) {
@@ -128,6 +130,7 @@ 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");
queue_wait_threshold(pn->etcp->input_queue, 0, 0, etcp_input_ready_cb, pn);
}
@@ -136,6 +139,7 @@ static void packer_cb(struct ll_queue* q, void* arg) {
static void pn_send_to_etcp(struct PKTNORM* pn) {
if (!pn || !pn->sndpart.data || !pn->sndpart.len==0) return;
DEBUG_INFO(DEBUG_CATEGORY_ETCP, "pn_packer: pn_send_to_etcp");
// Allocate ETCP_FRAGMENT from rx_pool
struct ETCP_FRAGMENT* frag = queue_entry_new_from_pool(pn->etcp->rx_pool);
if (!frag) {// drop data
@@ -151,6 +155,10 @@ static void pn_send_to_etcp(struct PKTNORM* pn) {
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
@@ -172,6 +180,8 @@ 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");
void* data = queue_data_get(pn->input);
if (!data) {
queue_resume_callback(pn->input);
@@ -187,8 +197,9 @@ static void etcp_input_ready_cb(struct ll_queue* q, void* arg) {
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_send_to_etcp(pn);
continue; // Продолжить с новым буфером
}
if (ptr == 0) {
@@ -204,7 +215,10 @@ static void etcp_input_ready_cb(struct ll_queue* q, void* arg) {
pn->sndpart.len += n;
ptr += n;
pn_buf_renew(pn);
// Проверить, не заполнился ли буфер
if (pn->sndpart.len >= pn->frag_size - 2) {
pn_send_to_etcp(pn);
}
}
queue_dgram_free(in_dgram);