diff --git a/src/transport_layer/etcp.c b/src/transport_layer/etcp.c index 93ddded2..c9715703 100644 --- a/src/transport_layer/etcp.c +++ b/src/transport_layer/etcp.c @@ -605,6 +605,7 @@ void etcp_conn_ready(struct ETCP_CONN* conn) { etcp_cbk_fire(conn, ETCP_CBK_EVENT_REINIT); etcp_fire_conn_status(conn, ETCP_CONN_STATUS_REINIT); } + if (conn->normalizer && !conn->close_requested) queue_resume_callback(conn->normalizer->input); etcp_conn_wake_sender(conn); } @@ -848,7 +849,7 @@ static void input_queue_cb(struct ll_queue* q, void* arg) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] cannot allocate INFLIGHT_PACKET (pool=%p etcp=%p)", etcp->log_name, etcp->inflight_pool, etcp); queue_dgram_free(&in_pkt->ll); queue_entry_free((struct ll_entry*)in_pkt); - queue_resume_callback(q); + etcp_conn_fatal_reinit(etcp, "TX inflight OOM"); return; } @@ -871,7 +872,7 @@ static void input_queue_cb(struct ll_queue* q, void* arg) { // При провале queue_data_put_with_index сам освобождает entry+dgram, повторный free дал бы double-free. if (queue_data_put_with_index(etcp->input_send_q, &p->ll) != 0) { DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] failed to add packet seq=%u to input_send_q", etcp->log_name, p->seq); - DEBUG_TRACE(DEBUG_CATEGORY_ETCP, "[%s] EXIT (queue put failed)", etcp->log_name); + etcp_conn_fatal_reinit(etcp, "TX inflight queue failure"); return; } etcp->unacked_bytes += len; @@ -1132,11 +1133,12 @@ void etcp_output_try_assembly(struct ETCP_CONN* etcp) { // etcp->log_name, etcp, etcp->last_delivered_id, queue_entry_count(etcp->recv_q)); uint32_t next_expected_id = etcp->last_delivered_id + 1; + uint32_t reset = etcp->reset_count; int delivered_count = 0; uint32_t delivered_bytes = 0; // Look for contiguous packets starting from next_expected_id - while (1) { + while (!etcp->close_requested && !etcp->reinit_pending && etcp->reset_count == reset) { struct ETCP_FRAGMENT* rx_pkt = (struct ETCP_FRAGMENT*)queue_find_data_by_index(etcp->recv_q, &next_expected_id); if (!rx_pkt) { // No more contiguous packets found @@ -1165,7 +1167,7 @@ void etcp_output_try_assembly(struct ETCP_CONN* etcp) { } // pn_unpacker_cb (вызывается из put синхронно) мог ре-энтрантно сделать reinit/close: // при reinit last_delivered_id уже сброшен в 0 и recv_q очищен, при close state==2. - if (etcp->state == 2 || etcp->reinit_pending) break; + if (etcp->close_requested || etcp->state == 2 || etcp->reinit_pending || etcp->reset_count != reset) break; delivered_bytes += pkt_len; delivered_count++; @@ -1230,7 +1232,7 @@ void etcp_conn_input(struct ETCP_DGRAM* pkt) { while (len >= 1) { // Ре-энтрантный reinit/close (из pn_unpacker_cb/etcp_int_recv) — дальше разбирать пакет нельзя. - if (etcp->state == 2 || etcp->reinit_pending) break; + if (etcp->close_requested || etcp->state == 2 || etcp->reinit_pending) break; uint8_t type = data[0]; // Process sections as per protocol.txt diff --git a/src/transport_layer/etcp_session.c b/src/transport_layer/etcp_session.c index d9e5d4e7..e0885384 100644 --- a/src/transport_layer/etcp_session.c +++ b/src/transport_layer/etcp_session.c @@ -10,6 +10,7 @@ static uint64_t read_id(const uint8_t* p) { uint64_t v; memcpy(&v,p,8); return be64toh(v); } static void write_id(uint8_t* p, uint64_t v) { v = htobe64(v); memcpy(p,&v,8); } +static void session_schedule(struct ETCP_CONN* c, int delay_tb); void etcp_session_encode(uint8_t* data, uint8_t code, uint64_t sender, uint64_t target) { data[0] = code; write_id(data+1,sender); write_id(data+9,target); } @@ -47,20 +48,25 @@ static void session_tick(void* arg) { } if (c->session_required) session_send(c,ETCP_SESSION_HELLO,c->peer_reset_id,0); if (c->session_candidate) session_send(c,ETCP_SESSION_CHALLENGE,c->session_candidate,c->session_challenge); - c->session_timer = uasync_set_timeout(c->instance->ua,SESSION_RETRY_TB,c,session_tick,"etcp_session"); + session_schedule(c, SESSION_RETRY_TB); } -static void session_schedule(struct ETCP_CONN* c) { - if (c->session_timer) return; - c->session_started = get_time_tb(); - c->session_timer = uasync_set_timeout(c->instance->ua,1,c,session_tick,"etcp_session"); +static void session_schedule(struct ETCP_CONN* c, int delay_tb) { + if (c->session_timer || c->close_requested) return; + if (delay_tb == 1) c->session_started = get_time_tb(); + c->session_timer = uasync_set_timeout(c->instance->ua, delay_tb, c, session_tick, "etcp_session"); + if (!c->session_timer) { + DEBUG_ERROR(DEBUG_CATEGORY_ETCP, "[%s] session timer OOM delay_tb=%d required=%u; closing", + c->log_name, delay_tb, c->session_required); + c->callbacks_running++; etcp_connection_close(c); c->callbacks_running--; + } } void etcp_session_start(struct ETCP_CONN* c) { etcp_session_cancel(c); c->session_required = 1; DEBUG_INFO(DEBUG_CATEGORY_ETCP,"[%s] session negotiation started local=%016llx peer=%016llx",c->log_name, (unsigned long long)c->reset_id,(unsigned long long)c->peer_reset_id); - session_schedule(c); + session_schedule(c, 1); } void etcp_session_observe(struct ETCP_CONN* c, uint64_t peer_epoch) { @@ -74,7 +80,7 @@ void etcp_session_observe(struct ETCP_CONN* c, uint64_t peer_epoch) { } } while (!c->session_challenge); } - session_schedule(c); + session_schedule(c, 1); } int etcp_session_receive(struct ETCP_CONN* c, const uint8_t* data, size_t len) {