diff --git a/cortex/export/headless.py b/cortex/export/headless.py index d3e8f2337..a8eadb458 100644 --- a/cortex/export/headless.py +++ b/cortex/export/headless.py @@ -342,38 +342,64 @@ def headless_viewer( pw_thread = _PlaywrightThread() handle = None + # ------------------------------------------------------------------ + # 3. Begin waiting for the WebSocket "connect" message on a *daemon* + # thread *before* navigating, so we cannot miss it even if the browser + # connects before page.goto() returns. + # + # A daemon thread (rather than a ThreadPoolExecutor) is essential here: + # server.get_client() blocks on a threading.Event with no timeout, so if + # the browser never sends "connect" the getter parks forever. An + # executor's worker is non-daemon and concurrent.futures joins every + # worker at interpreter exit -- a parked getter would then wedge the + # whole process at shutdown (observed as a multi-minute CI hang after + # the tests have already finished). Daemon threads are never joined at + # exit, so a stuck getter can never hold the process hostage. + # ------------------------------------------------------------------ + connect_result: dict[str, Any] = {} + + def _await_client() -> None: + try: + connect_result["handle"] = server.get_client() + except BaseException as exc: # noqa: BLE001 - reported to main thread + connect_result["error"] = exc + + client_thread = threading.Thread( + target=_await_client, name="headless-await-client", daemon=True + ) + try: - # ------------------------------------------------------------------ - # 3. Begin waiting for the WebSocket "connect" message in a thread - # *before* navigating, so we cannot miss it even if the browser - # connects before page.goto() returns. - # ------------------------------------------------------------------ - pool = concurrent.futures.ThreadPoolExecutor(max_workers=1) - fut = pool.submit(server.get_client) + client_thread.start() try: # Launch the browser and navigate. python_interface.js runs on # load and sends "connect" over WebSocket, which unblocks # server.get_client(). pw_thread.start(url, timeout=timeout) - # Retrieve the handle; it should already be ready by this point, - # but the timeout guard surfaces hung state clearly. - handle = fut.result(timeout=timeout) + # Wait (bounded) for the getter to return; the timeout guard + # surfaces hung state clearly instead of blocking indefinitely. + client_thread.join(timeout=timeout) + if "error" in connect_result: + raise connect_result["error"] + if "handle" not in connect_result: + raise TimeoutError( + f"No WebSocket 'connect' received within {timeout:.0f}s" + ) + handle = connect_result["handle"] except Exception as e: - fut.cancel() browser_errors = pw_thread.browser_errors detail = ( "\nBrowser errors:\n" + "\n".join(browser_errors) if browser_errors else "\nNo browser errors were captured." ) - pool.shutdown(wait=False, cancel_futures=True) + # Unblock the (daemon) getter if it is still parked on the connect + # event so it exits promptly rather than lingering across sessions. + server.connect.set() raise RuntimeError( f"Failed to establish WebSocket connection with headless browser at {url} " f"within {timeout:.0f} seconds. {detail}" ) from e - else: - pool.shutdown(wait=False) assert not isinstance(handle, list) # type narrowing to JSMixer handle.server = server diff --git a/cortex/webgl/serve.py b/cortex/webgl/serve.py index 354c665d0..cab2d0226 100644 --- a/cortex/webgl/serve.py +++ b/cortex/webgl/serve.py @@ -28,6 +28,7 @@ import tornado.ioloop import tornado.httpserver from tornado import websocket +from tornado.netutil import bind_sockets from tornado.web import HTTPError cwd = os.path.split(os.path.abspath(__file__))[0] @@ -310,9 +311,32 @@ def __init__( (r"/wsconnect/", ClientSocket, dict(parent=self)), (r"/(.*)", tornado.web.StaticFileHandler, dict(path=cwd)), ] - self.port = port + # Bind the listening socket(s) here, synchronously, in the calling + # thread -- rather than picking a random port and calling + # ``server.listen(port)`` later on the server thread. This fixes two + # long-standing sources of intermittent headless/CI hangs: + # * ``port=0`` asks the OS for a guaranteed-free ephemeral port, so + # we never collide with a lingering server (the old code used + # ``random.randint`` and hoped for the best). + # * a bind failure now raises *here*, visibly, in the caller. The old + # ``listen()`` ran on the server thread, so an + # "Address already in use" died silently and left callers to hang + # for the full timeout waiting to connect to a server that had + # never actually come up. + # The socket is bound immediately, so the OS accepts and queues client + # connections in the backlog even before the IOLoop starts serving -- + # eliminating the connect race too. + self._sockets = bind_sockets(port if port is not None else 0) + # When port==0 the OS assigns the port; read the real value back so + # callers can build a correct URL. + self.port = self._sockets[0].getsockname()[1] self.response: Queue[Union[str, bytes]] = Queue() self.connect = threading.Event() + # Set by run() once self.server and self.ioloop exist and the server is + # about to serve. stop() waits on this so an early stop() (called before + # the server thread has finished starting up) neither hits missing + # attributes nor gets silently lost. + self._ready = threading.Event() self.sockets: list[websocket.WebSocketHandler] = [] @property @@ -333,13 +357,34 @@ def run(self): self.server = tornado.httpserver.HTTPServer(application, io_loop=ioloop) else: self.server = tornado.httpserver.HTTPServer(application) - self.server.listen(self.port) + # The socket(s) were already bound in __init__; just start serving on + # them. This cannot raise "Address already in use" on the server thread. + self.server.add_sockets(self._sockets) + # self.server and self.ioloop are now set; signal that stop() is safe. + self._ready.set() ioloop.start() def stop(self): print("Stopping server") - self.server.stop() - self.ioloop.stop() + # stop() may race a just-started server: run() sets self.ioloop and + # self.server on the server thread, so a stop() that arrives first would + # hit missing attributes -- or, worse, do nothing and let run() go on to + # serve forever. Wait for the server to finish coming up before stopping + # it. + if self.is_alive() and self._ready.wait(timeout=5): + self.server.stop() # also closes the sockets it now owns + self.ioloop.stop() + else: + # The server thread never brought the server up (start() was never + # called, or it died during startup), so nothing owns the sockets + # bound in __init__. Close them here so we do not leak the + # listening port. is_alive() short-circuits the wait when start() + # was never called. + for sock in self._sockets: + try: + sock.close() + except OSError: + pass def send(self, **kwargs: Any) -> Union[list[JSON], list[None]]: msg = json.dumps(kwargs, cls=NPEncode, ensure_ascii=False) diff --git a/cortex/webgl/view.py b/cortex/webgl/view.py index d32ad7916..bfd15249f 100644 --- a/cortex/webgl/view.py +++ b/cortex/webgl/view.py @@ -930,7 +930,10 @@ def get_local_client(self): return JSMixer(self.srvsend, "window.viewer") if port is None: - port = random.randint(1024, 65536) + # Let the OS assign a guaranteed-free ephemeral port (WebApp binds + # port 0 and reads the real port back). This avoids the random-port + # collisions that made headless/CI runs intermittently hang. + port = 0 server = WebApp([(r'/ctm/(.*)', CTMHandler), (r'/data/(.*)', DataHandler),