fix cleanup at shutdown

app tests in ngapp create many apps -> need clean shutdown

Author: ChrLackner

webgpu/canvas.py

@@ -147,10 +147,12 @@ def __init__(self, device, canvas, multisample_count=4):
         self.update_html_canvas(canvas)
 
     def __del__(self):
-        if self._resize_observer is not None:
-            self._resize_observer.disconnect()
-        if self._intersection_observer is not None:
-            self._intersection_observer.disconnect()
+        disconnect = getattr(self._resize_observer, "disconnect", None)
+        if callable(disconnect):
+            disconnect()
+        disconnect = getattr(self._intersection_observer, "disconnect", None)
+        if callable(disconnect):
+            disconnect()
 
     def update_html_canvas(self, html_canvas):
         """Reconfigure the canvas with the current HTML canvas element. This is necessary when the HTML canvas element changes, disappears (e.g. when switching a tab) and appears again."""

webgpu/link/base.py

@@ -475,11 +475,13 @@ class LinkBaseAsync(LinkBase):
     _callback_loop: asyncio.AbstractEventLoop
     _callback_queue: asyncio.Queue
     _callback_thread: threading.Thread
+    _callback_task: asyncio.Task | None
 
     def __init__(self):
         super().__init__()
         self._send_loop = asyncio.new_event_loop()
         self._callback_loop = asyncio.new_event_loop()
+        self._callback_task = None
         self._callback_thread = threading.Thread(target=self._start_callback_thread, daemon=True)
         self._callback_thread.start()
 
@@ -514,9 +516,11 @@ def _send_data(self, metadata, data, key=None):
             event = threading.Event()
             self._requests[request_id] = event, key
 
+        coro = self._send_async(data)
         try:
-            asyncio.run_coroutine_threadsafe(self._send_async(data), self._send_loop)
+            asyncio.run_coroutine_threadsafe(coro, self._send_loop)
         except RuntimeError:
+            coro.close()
             # Event loop is closed — connection is dead, clean up and bail.
             if event:
                 self._requests.pop(request_id, None)
@@ -547,7 +551,15 @@ async def handle_callbacks():
             self._callback_loop = asyncio.new_event_loop()
             asyncio.set_event_loop(self._callback_loop)
             self._callback_queue = asyncio.Queue()
-            self._callback_loop.create_task(handle_callbacks())
-            self._callback_loop.run_forever()
+            self._callback_task = self._callback_loop.create_task(handle_callbacks())
+            try:
+                self._callback_loop.run_forever()
+            finally:
+                if self._callback_task is not None and not self._callback_task.done():
+                    self._callback_task.cancel()
+                    self._callback_loop.run_until_complete(
+                        asyncio.gather(self._callback_task, return_exceptions=True)
+                    )
+                self._callback_loop.close()
         except Exception as e:
             print("exception in _start_callback_thread", e)

webgpu/link/websocket.py

@@ -22,7 +22,7 @@ def __init__(self):
         self._start_handling_messages = threading.Event()
         self._send_loop = asyncio.new_event_loop()
 
-        self._websocket_thread = threading.Thread(target=self._connect)
+        self._websocket_thread = threading.Thread(target=self._connect, daemon=True)
         self._websocket_thread.start()
 
     def wait_for_server_running(self):
@@ -89,7 +89,7 @@ async def _websocket_handler(self, websocket, path=""):
             self._connection = websocket
             self._event_is_connected.set()
             async for message in websocket:
-                thread = threading.Thread(target=self._on_message, args=(message,))
+                thread = threading.Thread(target=self._on_message, args=(message,), daemon=True)
                 thread.start()
         finally:
             self._connection = None
@@ -118,15 +118,37 @@ async def start_websocket():
             self._send_loop.run_until_complete(start_websocket())
         except Exception as e:
             print("exception in _start_websocket_server", e)
+        finally:
+            pending = [
+                task for task in asyncio.all_tasks(self._send_loop)
+                if not task.done()
+            ]
+            for task in pending:
+                task.cancel()
+            if pending:
+                self._send_loop.run_until_complete(
+                    asyncio.gather(*pending, return_exceptions=True)
+                )
+            self._send_loop.run_until_complete(self._send_loop.shutdown_asyncgens())
+            self._send_loop.run_until_complete(self._send_loop.shutdown_default_executor())
+            self._send_loop.close()
 
     def stop(self):
-        self._send_loop.call_soon_threadsafe(self._stop.set_result, None)
+        try:
+            if not self._stop.done():
+                self._send_loop.call_soon_threadsafe(self._stop.set_result, None)
+        except RuntimeError:
+            pass  # Event loop already closed
+        if threading.current_thread() is not self._websocket_thread:
+            self._websocket_thread.join(timeout=2)
 
         # Stop the callback event loop so the _callback_thread exits.
         try:
             self._callback_loop.call_soon_threadsafe(self._callback_loop.stop)
         except RuntimeError:
             pass  # Event loop already closed
+        if threading.current_thread() is not self._callback_thread:
+            self._callback_thread.join(timeout=1)
 
         # Unblock any threads stuck waiting for websocket RPC responses.
         for rid, val in list(self._requests.items()):

webgpu/scene.py

@@ -97,6 +97,8 @@ def init(self, canvas):
             self.options.timestamp = time.time()
             self.options.update_buffers()
             for obj in self.render_objects:
+                if not obj.active:
+                    continue
                 try:
                     obj._update_and_create_render_pipeline(self.options)
                 except Exception as e: