diff --git a/src/gateway/http.c b/src/gateway/http.c index 44f3333..f806930 100644 --- a/src/gateway/http.c +++ b/src/gateway/http.c @@ -163,6 +163,9 @@ void http_stop(void) g_ctx->thread = 0; } if (g_ctx->lws_ctx) { + /* Drop the pointer before destroy so a late send cannot + * lws_cancel_service a freed context. */ + ws_set_context(NULL); lws_context_destroy(g_ctx->lws_ctx); g_ctx->lws_ctx = NULL; } diff --git a/src/gateway/http_lws.c b/src/gateway/http_lws.c index 86ecb79..9794007 100644 --- a/src/gateway/http_lws.c +++ b/src/gateway/http_lws.c @@ -488,6 +488,10 @@ int ws_callback(struct lws *wsi, enum lws_callback_reasons reason, void *user, if (conn_id > 0) ws_unregister_conn(conn_id); break; } + case LWS_CALLBACK_EVENT_WAIT_CANCELLED: + /* wsi here is not a client socket. Mark real conns from our table. */ + ws_on_service_cancel(); + break; default: break; } diff --git a/src/gateway/ws.c b/src/gateway/ws.c index c1848f5..4aa0a1e 100644 --- a/src/gateway/ws.c +++ b/src/gateway/ws.c @@ -8,7 +8,31 @@ #ifdef SHELLCLAW_WS_TEST struct lws; struct lws_context; -static void lws_callback_on_writable(struct lws *wsi) { (void)wsi; } +static int g_ws_test_writable_count; +static int g_ws_test_cancel_count; +static void lws_callback_on_writable(struct lws *wsi) +{ + (void)wsi; + g_ws_test_writable_count++; +} +static void lws_cancel_service(struct lws_context *ctx) +{ + (void)ctx; + g_ws_test_cancel_count++; +} +int ws_test_writable_count(void) +{ + return g_ws_test_writable_count; +} +int ws_test_cancel_count(void) +{ + return g_ws_test_cancel_count; +} +void ws_test_reset_wake_counts(void) +{ + g_ws_test_writable_count = 0; + g_ws_test_cancel_count = 0; +} #else #include "gateway/lws_compat.h" #include @@ -59,7 +83,48 @@ static struct lws_context *g_lws_ctx; void ws_set_context(void *ctx) { + pthread_mutex_lock(&g_mutex); g_lws_ctx = (struct lws_context *)ctx; + pthread_mutex_unlock(&g_mutex); +} + +/* Cross-thread wake. lws_cancel_service is the call that is safe off the + * service thread; the writable mark happens in ws_on_service_cancel. */ +static void ws_wake_service(void) +{ + struct lws_context *ctx; + + pthread_mutex_lock(&g_mutex); + ctx = g_lws_ctx; + pthread_mutex_unlock(&g_mutex); + if (ctx) + lws_cancel_service(ctx); +} + +void ws_on_service_cancel(void) +{ + struct lws *wake[MAX_CONNECTIONS]; + int n = 0; + int i; + + pthread_mutex_lock(&g_mutex); + for (i = 0; i < MAX_CONNECTIONS; i++) { + ws_msg_t *msg; + int conn_id; + + if (!g_conns[i].wsi) + continue; + conn_id = g_conns[i].conn_id; + for (msg = g_outgoing_head; msg; msg = msg->next) { + if (msg->conn_id != conn_id) + continue; + wake[n++] = (struct lws *)g_conns[i].wsi; + break; + } + } + pthread_mutex_unlock(&g_mutex); + for (i = 0; i < n; i++) + lws_callback_on_writable(wake[i]); } int ws_next_conn_id(void) @@ -250,15 +315,14 @@ int ws_send_to(const char *session_id, const char *text) } pthread_mutex_unlock(&g_mutex); if (!wsi) return 0; - lws_callback_on_writable((struct lws *)wsi); + ws_wake_service(); return 0; } void ws_broadcast_text(const char *text) { int i; - int wake_count = 0; - ws_conn_t wake_list[MAX_CONNECTIONS]; + int queued = 0; if (!ws_text_len_ok(text)) return; pthread_mutex_lock(&g_mutex); for (i = 0; i < MAX_CONNECTIONS; i++) { @@ -284,16 +348,11 @@ void ws_broadcast_text(const char *text) g_outgoing_tail = m; } g_outgoing_count++; - if (wake_count < MAX_CONNECTIONS) { - wake_list[wake_count] = g_conns[i].wsi; - wake_count++; - } + queued = 1; } pthread_mutex_unlock(&g_mutex); - for (i = 0; i < wake_count; i++) { - if (wake_list[i]) - lws_callback_on_writable((struct lws *)wake_list[i]); - } + if (queued) + ws_wake_service(); } void ws_shutdown_signal(void) diff --git a/src/gateway/ws.h b/src/gateway/ws.h index 9c781e9..1d4b152 100644 --- a/src/gateway/ws.h +++ b/src/gateway/ws.h @@ -97,6 +97,17 @@ int ws_dequeue_outgoing(int conn_id, char *buf, size_t buf_size, size_t *len_out */ int ws_has_pending_outgoing(int conn_id); +/** + * Ask the service thread to flush queued WebChat frames. + * Call only from the libwebsockets thread, in + * LWS_CALLBACK_EVENT_WAIT_CANCELLED. lws_callback_on_writable() is not + * safe from the agent thread: libwebsockets 4.3 updates the poll set + * without the per-thread lock. + * + * Example: `case LWS_CALLBACK_EVENT_WAIT_CANCELLED: ws_on_service_cancel(); break;` + */ +void ws_on_service_cancel(void); + /** * Get next connection ID for new connections. * diff --git a/tests/test_ws.c b/tests/test_ws.c index af53973..ebedee4 100644 --- a/tests/test_ws.c +++ b/tests/test_ws.c @@ -268,6 +268,68 @@ static int test_pop_incoming_invalid_args(void) return 0; } +/* Counters live in the SHELLCLAW_WS_TEST build of ws.c. */ +extern int ws_test_writable_count(void); +extern int ws_test_cancel_count(void); +extern void ws_test_reset_wake_counts(void); + +/** + * Agent replies run on the main thread while lws_service runs on the gateway + * thread. The wake must be lws_cancel_service, and only the service-thread + * hook may call lws_callback_on_writable. + */ +static int test_send_wakes_service_thread(void) +{ + char buf[32]; + size_t len; + + ws_cleanup(); + ws_test_reset_wake_counts(); + ws_set_context((void *)(intptr_t)1); + ASSERT(ws_register_conn(11, (ws_conn_t)(intptr_t)11) == 0); + ASSERT(ws_send_to("webchat:11", "hello") == 0); + ASSERT(ws_test_cancel_count() == 1); + ASSERT(ws_test_writable_count() == 0); + ws_on_service_cancel(); + ASSERT(ws_test_writable_count() == 1); + ASSERT(ws_dequeue_outgoing(11, buf, sizeof(buf), &len) == 1); + ASSERT(strcmp(buf, "hello") == 0); + ws_on_service_cancel(); + ASSERT(ws_test_writable_count() == 1); + ws_set_context(NULL); + ws_cleanup(); + return 0; +} + +static int test_broadcast_wakes_once(void) +{ + ws_cleanup(); + ws_test_reset_wake_counts(); + ws_set_context((void *)(intptr_t)1); + ASSERT(ws_register_conn(12, (ws_conn_t)(intptr_t)12) == 0); + ASSERT(ws_register_conn(13, (ws_conn_t)(intptr_t)13) == 0); + ws_broadcast_text("status"); + ASSERT(ws_test_cancel_count() == 1); + ASSERT(ws_test_writable_count() == 0); + ws_on_service_cancel(); + ASSERT(ws_test_writable_count() == 2); + ws_set_context(NULL); + ws_cleanup(); + return 0; +} + +static int test_send_without_context_does_not_touch_lws(void) +{ + ws_cleanup(); + ws_test_reset_wake_counts(); + ASSERT(ws_register_conn(14, (ws_conn_t)(intptr_t)14) == 0); + ASSERT(ws_send_to("webchat:14", "queued") == 0); + ASSERT(ws_test_cancel_count() == 0); + ASSERT(ws_test_writable_count() == 0); + ws_cleanup(); + return 0; +} + static int test_shutdown_stops_pop(void) { char session[32]; @@ -297,6 +359,9 @@ int main(void) RUN(test_push_incoming_rejects_empty()); RUN(test_pop_incoming_invalid_args()); RUN(test_shutdown_stops_pop()); + RUN(test_send_wakes_service_thread()); + RUN(test_broadcast_wakes_once()); + RUN(test_send_without_context_does_not_touch_lws()); printf("test_ws: all tests passed\n"); return 0; }