MT#55283 websocket: prevent HTTP disconnect deadlock by deferring writable callbacks (#2197)

Replace spinning for websocket_service_lock with a short mutex-protected
writable queue. All lws_callback_on_writable() calls are now marshalled
onto the LWS service thread and only run after lws_service() returns,
instead of being issued directly from worker threads while the service
thread may be holding the same lock. The callback queue mutex is
released before calling into libwebsockets, and the amount of queued
work drained per service cycle is bounded to avoid starving other
connections under a connection-abort storm. Pending callbacks are
dropped on teardown before the LWS session storage is freed.

Add an HTTP disconnect-storm regression test (two worker threads
hammering /metrics with aborted connections while a liveness /ping
request keeps proving the service thread is not stuck) to
t/auto-daemon-tests-websocket.py, plus a standalone stress script. The
stress test waits for the listening socket to come up before firing,
avoiding a FileNotFoundError race on CI.

Closes #2206

Change-Id: I81e03c1d7f1a0b5650aa719d00d60be3d3d0213b
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

daemon/websocket.c

@@ -42,6 +42,7 @@ struct websocket_conn {
 	// multithreaded message processing
 	mutex_t lock;
 	unsigned int jobs;
+	bool writable_queued; // protected by websocket_callback_lock
 	websocket_message_q messages;
 	cond_t cond;
 	janus_sessions_ht janus_sessions;
@@ -62,6 +63,9 @@ struct websocket_ng_buf {
 static GQueue websocket_vhost_configs;
 static struct lws_context *websocket_context;
 static GThreadPool *websocket_threads;
+// Workers only enqueue writable notifications. The LWS service thread
+// consumes this queue after lws_service() returns.
+static GQueue websocket_pending_writes;
 static mutex_t websocket_callback_lock = MUTEX_STATIC_INIT;
 static mutex_t websocket_service_lock = MUTEX_STATIC_INIT;
 
@@ -134,68 +138,52 @@ static void __websocket_write_raw(struct websocket_conn *wc, const char *msg, si
 	t_queue_push_tail(&wc->output_q, websocket_output_new());
 }
 
-// called after all writes are done - non-locking
-// may intermittently release wc->lock
+// Called with wc->lock held, after appending response data. Do not wait
+// for the LWS service lock here: the service thread can be closing another
+// connection and waiting for a queued worker to finish (issue #2197).
 static void __websocket_write_done(struct websocket_conn *wc) {
-	while (true) {
-		// If this connection is already closed, don't trigger
-		// a wakeup of the service thread and don't request a
-		// writeable callback. We can't write anyway if the
-		// connection is already closed. Prevents a deadlock
-		// between acquiring the service lock here and the
-		// websocket_conn_cleanup called from within the service
-		// loop.
-		if (!wc->wsi)
-			return;
-
-		// Sadly lws_callback_on_writable() doesn't do any internal
-		// locking, therefore we must protect it against a concurrently
-		// running lws_service(), as well as against other threads
-		// invoking lws_callback_on_writable().
-		//
-		// Acquire the callback lock first, which is normally unlocked,
-		// then wake up the service thread and try to break out of
-		// lws_service().  The service thread holds the service lock
-		// while lws_service() is executing and releases it as soon as
-		// lws_service() is done. We therefore try to acquire the
-		// service lock here, which blocks us until lws_service() is
-		// actually done. At this point the service thread will try to
-		// acquire the callback lock, which is still held by us here,
-		// and so the service thread will block until we are done
-		// calling lws_callback_on_writable(). Finally we release both
-		// locks, which allows the service thread to resume
-		// lws_service().
-		//
-		// The suggested approach of using
-		// LWS_CALLBACK_EVENT_WAIT_CANCELLED together with a queue and
-		// then calling lws_callback_on_writable() from the service
-		// thread is not usable as libwebsockets 2.0 doesn't support
-		// LWS_CALLBACK_EVENT_WAIT_CANCELLED.
+	if (!wc->wsi)
+		return;
 
-		mutex_lock(&websocket_callback_lock);
-		lws_cancel_service(websocket_context);
-
-		if (mutex_trylock(&websocket_service_lock)) {
-			// Wrong lock order: service_lock should be the outer lock
-			// and wc->lock should be the inner lock. The connection may
-			// be in the process of being closed (websocket_conn_cleanup).
-			// Release wc->lock and try again.
-			mutex_unlock(&websocket_callback_lock);
-			mutex_unlock(&wc->lock);
-			sched_yield();
-			mutex_lock(&wc->lock);
-			continue;
-		}
-		lws_callback_on_writable(wc->wsi);
+	mutex_lock(&websocket_callback_lock);
+	if (!wc->writable_queued) {
+		wc->writable_queued = true;
+		g_queue_push_tail(&websocket_pending_writes, wc);
+	}
+	mutex_unlock(&websocket_callback_lock);
 
-		mutex_unlock(&websocket_service_lock);
+	// This can be called safely from a worker; only the service thread
+	// invokes lws_callback_on_writable() while not inside lws_service().
+	lws_cancel_service(websocket_context);
+}
+
+// Executed only by the LWS service thread, outside lws_service().
+// Caller holds the service lock; callback_lock only protects the queue.
+static void websocket_flush_write_notifications(void) {
+	// Only the service thread can free an LWS connection, so a popped
+	// connection remains live until its writable event is scheduled.
+	// Do not hold the queue lock inside libwebsockets, as callbacks can
+	// re-enter the HTTP/WS code.
+	mutex_lock(&websocket_callback_lock);
+	unsigned int pending = websocket_pending_writes.length;
+	mutex_unlock(&websocket_callback_lock);
+
+	// Process a snapshot to avoid starving lws_service() if producers
+	// continuously queue new callbacks.
+	while (pending--) {
+		mutex_lock(&websocket_callback_lock);
+		struct websocket_conn *wc = g_queue_pop_head(&websocket_pending_writes);
+		if (wc)
+			wc->writable_queued = false;
 		mutex_unlock(&websocket_callback_lock);
 
-		break;
+		if (!wc)
+			break;
+		if (wc->wsi)
+			lws_callback_on_writable(wc->wsi);
 	}
 }
 
-
 // adds data to output buffer (can be null) and triggers specified response: http or binary websocket - non-locking
 void __websocket_write_http_len(struct websocket_conn *wc, const char *msg, size_t len) {
 	__websocket_write_raw(wc, msg, len, LWS_WRITE_HTTP);
@@ -801,6 +789,14 @@ static void websocket_conn_cleanup(struct websocket_conn *wc) {
 	mutex_lock(&wc->lock);
 
 	wc->wsi = NULL;
+	// Drop any queued callback before LWS frees its session storage.
+	// Lock order is wc->lock, then websocket_callback_lock everywhere.
+	mutex_lock(&websocket_callback_lock);
+	if (wc->writable_queued) {
+		g_queue_remove(&websocket_pending_writes, wc);
+		wc->writable_queued = false;
+	}
+	mutex_unlock(&websocket_callback_lock);
 	while (wc->jobs)
 		cond_wait(&wc->cond, &wc->lock);
 
@@ -1327,10 +1323,10 @@ static void websocket_loop(void *p) {
 
 		mutex_lock(&websocket_service_lock);
 		lws_service(websocket_context, 100);
+		// Flush only after LWS has exited its callbacks and is not
+		// holding the connection cleanup path.
+		websocket_flush_write_notifications();
 		mutex_unlock(&websocket_service_lock);
-
-		mutex_lock(&websocket_callback_lock);
-		mutex_unlock(&websocket_callback_lock);
 	}
 
 	websocket_cleanup();

t/auto-daemon-tests-websocket.py

@@ -93,6 +93,48 @@ async def testOJanus(self, msg, conn_num=0):
     await self._ws[conn_num].send(json.dumps(msg))
 
 
+class TestHTTPDisconnectStress(unittest.TestCase):
+    """Exercise the worker/service-lock cycle reported in issue #2197."""
+
+    def testAbortStormDoesNotBlockServiceThread(self):
+        import concurrent.futures
+        import http.client
+        import struct
+        import time
+
+        for _ in range(300):
+            try:
+                with socket.create_connection(("127.0.0.1", 9191), timeout=3):
+                    break
+            except (FileNotFoundError, ConnectionRefusedError):
+                time.sleep(0.1)
+
+        def abort_metrics():
+            try:
+                with socket.create_connection(("127.0.0.1", 9191), timeout=3) as sock:
+                    sock.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER,
+                                    struct.pack("ii", 1, 0))
+                    sock.sendall(b"GET /metrics HTTP/1.1\r\n"
+                                 b"Host: localhost\r\n"
+                                 b"Connection: close\r\n\r\n")
+            except (ConnectionResetError, ConnectionAbortedError, BrokenPipeError):
+                pass
+
+        with concurrent.futures.ThreadPoolExecutor(max_workers=32) as pool:
+            for _ in range(12):
+                list(pool.map(lambda _: abort_metrics(), range(32)))
+                # Failure mode: the LWS service thread waits on a queued
+                # worker while all pool workers wait for its service lock.
+                conn = http.client.HTTPConnection("127.0.0.1", 9191, timeout=5)
+                try:
+                    conn.request("GET", "/ping", headers={"Connection": "close"})
+                    response = conn.getresponse()
+                    self.assertEqual(response.status, 200)
+                    self.assertEqual(response.read(), b"pong\n")
+                finally:
+                    conn.close()
+
+
 class TestWSEcho(unittest.TestCase):
     @classmethod
     def setUpClass(cls):
@@ -3599,6 +3641,7 @@ if __name__ == "__main__":
                 "7",
                 "-E",
                 "--listen-http=127.0.0.1:9191",
+                "--http-threads=2",
                 "--janus-secret=dfgdfgdvgLyATjHPvckg",
                 "--delete-delay=0",
             ],

t/websocket-disconnect-stress.py

@@ -0,0 +1,81 @@
+#!/usr/bin/env python3
+"""Regression stress test for issue #2197.
+
+Start rtpengine separately with --listen-http=127.0.0.1:2225 and a
+small --http-threads value, then run this script. The test deliberately
+closes concurrent /metrics connections before the server finishes.
+It also verifies that /ping remains responsive throughout the storm.
+"""
+
+import argparse
+from concurrent.futures import ThreadPoolExecutor
+import http.client
+import socket
+import struct
+import time
+
+
+def ping(host, port, timeout):
+    connection = http.client.HTTPConnection(host, port, timeout=timeout)
+    try:
+        connection.request("GET", "/ping", headers={"Connection": "close"})
+        response = connection.getresponse()
+        body = response.read()
+        if response.status != 200 or body != b"pong\n":
+            raise RuntimeError("unexpected /ping response: %s %r" %
+                               (response.status, body))
+    finally:
+        connection.close()
+
+
+def abort_metrics(host, port, timeout):
+    with socket.create_connection((host, port), timeout=timeout) as sock:
+        sock.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER,
+                        struct.pack("ii", 1, 0))
+        sock.sendall(b"GET /metrics HTTP/1.1\r\n"
+                     b"Host: localhost\r\nConnection: close\r\n\r\n")
+        # Abort without consuming the response.
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--host", default="127.0.0.1")
+    ap.add_argument("--port", type=int, default=2225)
+    ap.add_argument("--workers", type=int, default=32)
+    ap.add_argument("--rounds", type=int, default=50)
+    ap.add_argument("--timeout", type=float, default=2.0)
+    args = ap.parse_args()
+    ping(args.host, args.port, args.timeout)
+
+    failures = 0
+    started = time.monotonic()
+    with ThreadPoolExecutor(max_workers=args.workers) as pool:
+        for round_number in range(args.rounds):
+            futures = [pool.submit(abort_metrics, args.host, args.port,
+                                   args.timeout)
+                       for _ in range(args.workers)]
+            for future in futures:
+                try:
+                    future.result(timeout=args.timeout + 2)
+                except OSError:
+                    # A reset during the storm is expected and harmless.
+                    pass
+                except Exception as exc:
+                    failures += 1
+                    print("request error: %s" % exc)
+            # The LWS service thread must continue to make progress.
+            ping(args.host, args.port, args.timeout)
+            if (round_number + 1) % 10 == 0:
+                print("completed %d/%d rounds" %
+                      (round_number + 1, args.rounds))
+
+    ping(args.host, args.port, args.timeout)
+    if failures:
+        raise SystemExit("%d unexpected worker errors" % failures)
+    print("PASS: /ping remained responsive over %d aborted requests "
+          "(%.1f sec)" % (args.workers * args.rounds,
+                          time.monotonic() - started))
+
+
+if __name__ == "__main__":
+    main()