From fb6cd24e9e66461f5cb97c1c30ddce91ea8ed6d6 Mon Sep 17 00:00:00 2001 From: liquidraver <504870+liquidraver@users.noreply.github.com> Date: Sun, 14 Jun 2026 21:20:02 +0200 Subject: [PATCH] linux native transport fixes --- .../adapters/transport/LinuxTCPTransport.c | 145 +++++++++++++++++- 1 file changed, 140 insertions(+), 5 deletions(-) diff --git a/zephcore/adapters/transport/LinuxTCPTransport.c b/zephcore/adapters/transport/LinuxTCPTransport.c index ceb475e..75c74c7 100644 --- a/zephcore/adapters/transport/LinuxTCPTransport.c +++ b/zephcore/adapters/transport/LinuxTCPTransport.c @@ -45,6 +45,18 @@ LOG_MODULE_REGISTER(linux_tcp_transport, LOG_LEVEL_INF); #define LISTEN_PORT CONFIG_ZEPHCORE_LINUX_TCP_PORT #define LISTEN_BACKLOG 1 +/* TX congestion control (mirrors ZephyrBLE.cpp). Push codes >= 0x80 are + * lossy event signals; protocol response frames < 0x80 are lossless. */ +#define PUSH_CODE_BASE 0x80 +#define TX_OVERFLOW_RETRY_MS 250 + +/* Bound on a single blocking send. tx_drain_work_fn runs on the system + * workqueue, so an unbounded zsock_send() against a peer that has stopped + * reading would hang the whole queue (incl. overflow_retry_work). A peer + * that can't accept one frame within this window is wedged — close it and + * let the app reconnect rather than stall the node. */ +#define TX_SEND_TIMEOUT_SEC 2 + struct frame { uint16_t len; uint8_t buf[MAX_FRAME_SIZE]; @@ -73,12 +85,44 @@ static bool listen_thread_started; static void tx_drain_work_fn(struct k_work *work); static K_WORK_DEFINE(tx_drain_work, tx_drain_work_fn); +/* TX congestion control — ported from ZephyrBLE.cpp so the TCP transport + * back-pressures instead of dropping frames when the send queue fills (e.g. + * an advert/message push colliding with a channel/contact import burst). + * + * 1. Queue full → set sticky tx_congested; senders check + * zephcore_ble_is_congested() and hold off. + * 2. Lossless protocol frame (data[0] < 0x80) → return 0 so the caller + * retries — never dropped. + * 3. Non-lossless push → stash one frame in overflow, retried every 250ms. + * 4. Drain below 1/3 (or empty) → clear congestion (hysteresis). + * 5. Disconnect → reset everything. + */ +static bool tx_congested; +static struct frame overflow_frame; +static bool overflow_pending; + +static void overflow_retry_work_fn(struct k_work *work); +static K_WORK_DELAYABLE_DEFINE(overflow_retry_work, overflow_retry_work_fn); + +static bool is_lossless_protocol_frame(const uint8_t *data, uint16_t len) +{ + if (!data || len == 0) { + return false; + } + return data[0] < PUSH_CODE_BASE; +} + static void close_client_locked(void) { if (client_fd >= 0) { zsock_close(client_fd); client_fd = -1; } + + /* Reset TX congestion state on disconnect (mirrors ZephyrBLE). */ + overflow_pending = false; + tx_congested = false; + k_work_cancel_delayable(&overflow_retry_work); } /* Write `len` bytes, retrying on partial sends. Returns 0 or -errno. */ @@ -133,6 +177,15 @@ static void tx_drain_work_fn(struct k_work *work) struct frame f; while (k_msgq_get(&ble_send_queue, &f, K_NO_WAIT) == 0) { + /* Clear congestion at low water mark (1/3) — hysteresis vs full. */ + if (tx_congested && + k_msgq_num_used_get(&ble_send_queue) <= FRAME_QUEUE_SIZE / 3) { + tx_congested = false; + LOG_INF("tx_drain: congestion cleared (queue=%u/%u)", + k_msgq_num_used_get(&ble_send_queue), + (unsigned)FRAME_QUEUE_SIZE); + } + k_mutex_lock(&sock_mu, K_FOREVER); int fd = client_fd; @@ -158,7 +211,11 @@ static void tx_drain_work_fn(struct k_work *work) k_mutex_unlock(&sock_mu); if (err != 0) { - LOG_WRN("tx send err=%d, closing client", err); + if (err == -EAGAIN || err == -EWOULDBLOCK) { + LOG_WRN("tx send timeout (peer not reading), closing client"); + } else { + LOG_WRN("tx send err=%d, closing client", err); + } k_mutex_lock(&sock_mu, K_FOREVER); close_client_locked(); active_iface = ZEPHCORE_IFACE_NONE; @@ -171,12 +228,55 @@ static void tx_drain_work_fn(struct k_work *work) } } + /* Queue fully drained — clear any residual congestion. */ + if (tx_congested && k_msgq_num_used_get(&ble_send_queue) == 0) { + tx_congested = false; + LOG_INF("tx_drain: congestion cleared (queue empty)"); + } + if (transport_cbs && transport_cbs->on_tx_idle && k_msgq_num_used_get(&ble_send_queue) == 0) { transport_cbs->on_tx_idle(); } } +/* Re-inject the stashed overflow push once the queue has room. Mirrors + * ZephyrBLE.cpp's overflow_retry_work_fn. */ +static void overflow_retry_work_fn(struct k_work *work) +{ + ARG_UNUSED(work); + + if (!overflow_pending) { + return; + } + + /* Abandon overflow if the client is gone. */ + k_mutex_lock(&sock_mu, K_FOREVER); + bool connected = (client_fd >= 0); + + k_mutex_unlock(&sock_mu); + + if (!connected) { + overflow_pending = false; + tx_congested = false; + LOG_INF("overflow cleared (disconnected)"); + return; + } + + if (k_msgq_put(&ble_send_queue, &overflow_frame, K_NO_WAIT) == 0) { + overflow_pending = false; + LOG_DBG("overflow frame queued hdr=0x%02x, kicking drain", + overflow_frame.buf[0]); + k_work_submit(&tx_drain_work); + /* Congestion flag cleared by tx_drain at low water mark. */ + } else { + /* Still full — retry at reduced rate. */ + LOG_DBG("overflow retry: queue still full, retry in %dms", + TX_OVERFLOW_RETRY_MS); + k_work_schedule(&overflow_retry_work, K_MSEC(TX_OVERFLOW_RETRY_MS)); + } +} + static void listen_thread_fn(void *a, void *b, void *c) { ARG_UNUSED(a); @@ -229,6 +329,15 @@ static void listen_thread_fn(void *a, void *b, void *c) continue; } + /* Bound blocking sends so a non-reading peer can't hang the + * system workqueue (see TX_SEND_TIMEOUT_SEC). */ + struct zsock_timeval sndtimeo = { + .tv_sec = TX_SEND_TIMEOUT_SEC, + .tv_usec = 0, + }; + (void)zsock_setsockopt(fd, ZSOCK_SOL_SOCKET, ZSOCK_SO_SNDTIMEO, + &sndtimeo, sizeof(sndtimeo)); + k_mutex_lock(&sock_mu, K_FOREVER); if (client_fd >= 0) { LOG_WRN("Second client rejected (already connected)"); @@ -349,8 +458,33 @@ size_t zephcore_ble_send(const uint8_t *data, uint16_t len) memcpy(f.buf, data, len); if (k_msgq_put(&ble_send_queue, &f, K_NO_WAIT) != 0) { - LOG_WRN("TX queue full, dropping len=%u", len); - return 0; + /* Queue full — enter congestion mode (mirrors ZephyrBLE.cpp). + * Senders hold off via zephcore_ble_is_congested(). */ + if (!tx_congested) { + LOG_WRN("TX queue full (%u/%u), entering congestion", + k_msgq_num_used_get(&ble_send_queue), + (unsigned)FRAME_QUEUE_SIZE); + tx_congested = true; + } + + /* Lossless protocol response frames are never dropped — report + * failure so the caller retries instead of clobbering overflow. */ + if (is_lossless_protocol_frame(data, len)) { + LOG_DBG("TX queue full for lossless frame hdr=0x%02x, retry later", + data[0]); + return 0; + } + + /* Non-lossless push: stash one frame in overflow, retried at + * 250ms. If overflow is occupied, drop rather than clobber it. */ + if (overflow_pending) { + LOG_WRN("overflow full, dropping push hdr=0x%02x", data[0]); + return 0; + } + overflow_frame = f; + overflow_pending = true; + k_work_schedule(&overflow_retry_work, K_MSEC(TX_OVERFLOW_RETRY_MS)); + return len; /* Accepted into overflow — will be retried. */ } k_work_submit(&tx_drain_work); @@ -389,8 +523,9 @@ bool zephcore_ble_is_connected(void) bool zephcore_ble_is_congested(void) { - return k_msgq_num_used_get(&ble_send_queue) >= - (CONFIG_ZEPHCORE_BLE_QUEUE_SIZE * 2 / 3); + /* Sticky flag (mirrors ZephyrBLE): set at full, cleared at 1/3. + * Callers also check the 2/3 high-water mark on the raw queue. */ + return tx_congested; } bool zephcore_ble_is_advertising(void)