diff --git a/daemon/websocket.c b/daemon/websocket.c index 95e00e7a4..797ad6a26 100644 --- a/daemon/websocket.c +++ b/daemon/websocket.c @@ -41,6 +41,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; @@ -61,6 +62,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; @@ -133,68 +137,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); @@ -780,6 +768,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); @@ -1302,10 +1298,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(); diff --git a/t/auto-daemon-tests-websocket.py b/t/auto-daemon-tests-websocket.py index 3bdd57192..6b783ddab 100644 --- a/t/auto-daemon-tests-websocket.py +++ b/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", ], diff --git a/t/websocket-disconnect-stress.py b/t/websocket-disconnect-stress.py new file mode 100644 index 000000000..4bb35b669 --- /dev/null +++ b/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()