Terminate worker threads when the event loop is closed (#324)

.github/workflows/test-and-release.yml

@@ -64,7 +64,7 @@ jobs:
       - name: Run build and test
         run: |
           nox --non-interactive --error-on-missing-interpreters --session test pristine --python ${{ matrix.python }}
-          nox --non-interactive --session demo check_style docs
+          nox --non-interactive --no-error-on-missing-interpreters --session demo check_style docs
         env:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
           SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}

docs/conf.py

@@ -11,6 +11,7 @@
 # add these directories to sys.path here. If the directory is relative to the
 # documentation root, use os.path.abspath to make it absolute.
 import os
+import re
 import sys
 import pathlib
 import inspect
@@ -213,6 +214,7 @@ def report_exception(exc: Exception) -> None:
         "sphinx-apidoc",
         "-o",
         str(APIDOC_GENERATED_ROOT),
+        "-d1",  # Set :maxdepth:
         "--force",
         "--follow-links",
         "--separate",

docs/pages/api.rst

@@ -22,6 +22,7 @@ Submodules
 ----------
 
 .. toctree::
+   :maxdepth: 3
 
    /api/pycyphal.dsdl
    /api/pycyphal.application

docs/requirements.txt

@@ -5,6 +5,6 @@
 # This file is meant to be used from the project root directory.
 
 .[transport-can-pythoncan,transport-serial,transport-udp]
-sphinx           == 4.3.*
-sphinx_rtd_theme == 1.0.*
+sphinx           ~= 7.2.6
+sphinx_rtd_theme ~= 2.0.0
 sphinx-computron ~= 1.0

noxfile.py

@@ -163,7 +163,7 @@ def demo(session):
         return 0
 
     session.install("-e", f".[{','.join(EXTRAS_REQUIRE.keys())}]")
-    session.install("yakut ~= 0.11")
+    session.install("yakut ~= 0.13")
 
     demo_dir = ROOT_DIR / "demo"
     tmp_dir = Path(session.create_tmp()).resolve()
@@ -203,11 +203,11 @@ def pristine(session):
 
 @nox.session(reuse_venv=True)
 def check_style(session):
-    session.install("black == 23.*")
+    session.install("black ~= 23.12")
     session.run("black", "--check", ".")
 
 
-@nox.session()
+@nox.session(python=PYTHONS[-1])
 def docs(session):
     try:
         session.run("dot", "-V", silent=True, external=True)
@@ -233,7 +233,7 @@ def docs(session):
     session.log(f"DOCUMENTATION BUILD OUTPUT: file://{out_dir}/index.html")
 
     session.cd(ROOT_DIR)
-    session.install("doc8 ~= 0.11")
+    session.install("doc8 ~= 1.1")
     if is_latest_python(session):
         session.run("doc8", "docs", *map(str, ROOT_DIR.glob("*.rst")))
 

pycyphal/_version.py

@@ -1 +1 @@
-__version__ = "1.17.3"
+__version__ = "1.17.4"

pycyphal/transport/can/media/pythoncan/_pythoncan.py

@@ -349,7 +349,7 @@ def _invoke_rx_handler(self, frs: typing.List[typing.Tuple[Timestamp, Envelope]]
             _logger.exception("%s unhandled exception in the receive handler: %s; lost frames: %s", self, exc, frs)
 
     def _thread_function(self, loop: asyncio.AbstractEventLoop) -> None:
-        while not self._closed:
+        while not self._closed and not loop.is_closed():
             try:
                 batch = self._read_batch()
                 if batch:

pycyphal/transport/can/media/socketcan/_socketcan.py

@@ -180,7 +180,7 @@ def handler_wrapper(frs: typing.Sequence[typing.Tuple[Timestamp, Envelope]]) ->
             except Exception as exc:
                 _logger.exception("%s: Unhandled exception in the receive handler: %s; lost frames: %s", self, exc, frs)
 
-        while not self._closed:
+        while not self._closed and not loop.is_closed():
             try:
                 (
                     read_ready,

pycyphal/transport/can/media/socketcand/_socketcand.py

@@ -229,7 +229,7 @@ def _invoke_rx_handler(self, frs: typing.List[typing.Tuple[Timestamp, Envelope]]
             _logger.exception("%s unhandled exception in the receive handler: %s; lost frames: %s", self, exc, frs)
 
     def _thread_function(self, loop: asyncio.AbstractEventLoop) -> None:
-        while not self._closed:
+        while not self._closed and not loop.is_closed():
             try:
                 batch = self._read_batch()
                 if batch:

pycyphal/transport/udp/_session/_input.py

@@ -146,7 +146,7 @@ def _put_into_queue(self, ts: pycyphal.transport.Timestamp, frame: typing.Option
         self._frame_queue.put_nowait((ts, frame))
 
     def _reader_thread(self, loop: asyncio.AbstractEventLoop) -> None:
-        while not self._closed and self._socket.fileno() >= 0:
+        while not self._closed and self._socket.fileno() >= 0 and not loop.is_closed():
             try:
                 # TODO: add a dedicated socket for aborting the select call
                 # when self.close() is invoked to avoid blocking on
@@ -175,6 +175,9 @@ def _reader_thread(self, loop: asyncio.AbstractEventLoop) -> None:
                     except asyncio.QueueFull:
                         # TODO: make the queue capacity configurable
                         _logger.error("%s: Frame queue is full", self)
+                    except RuntimeError as ex:  # Event loop is closed.
+                        _logger.critical("%s: Stopping because: %s", self, ex, exc_info=True)
+                        break
             except Exception as ex:
                 _logger.exception("%s: Exception while consuming UDP frames: %s", self, ex)
 

tests/application/plug_and_play.py

@@ -58,33 +58,33 @@ async def _unittest_slow_plug_and_play_centralized(
     allocator.register_node(43, None)  # Does not overwrite
 
     use_v2 = mtu > cln_a._MTU_THRESHOLD  # pylint: disable=protected-access
-    await asyncio.sleep(2.0)
+    await asyncio.sleep(3.0)
     assert cln_a.get_result() == (44 if use_v2 else 125)
 
     # Another request.
     cln_b = Allocatee(trans_client, _uid("aabbccddeeff00112233445566778899"))
     assert cln_b.get_result() is None
-    await asyncio.sleep(2.0)
+    await asyncio.sleep(3.0)
     assert cln_b.get_result() == (125 if use_v2 else 124)
 
     # Re-request A and make sure we get the same response.
     cln_a = Allocatee(trans_client, _uid("00112233445566778899aabbccddeeff"), 42)
     assert cln_a.get_result() is None
-    await asyncio.sleep(2.0)
+    await asyncio.sleep(3.0)
     assert cln_a.get_result() == (44 if use_v2 else 125)
 
     # C should be served from the manually added entries above if we're on v2, otherwise new allocation.
     cln_c = Allocatee(trans_client, _uid("00000000000000000000000000000003"))
     assert cln_c.get_result() is None
-    await asyncio.sleep(2.0)
+    await asyncio.sleep(3.0)
     assert cln_c.get_result() == (43 if use_v2 else 122)  # 123 is used by the allocator itself, so we get 122.
 
     # Modify the entry we just created to ensure the pseudo-UID is not overwritten.
     # https://github.com/OpenCyphal/pycyphal/issues/160
     allocator.register_node(122, _uid("00000000000000000000000000000122"))
     cln_c = Allocatee(trans_client, _uid("00000000000000000000000000000003"))  # Same pseudo-UID
     assert cln_c.get_result() is None
-    await asyncio.sleep(2.0)
+    await asyncio.sleep(3.0)
     # We shall get the same response but the reasons are different depending on the message version used:
     # - v1 will return the same allocation because we're using the same pseudo-UID hash.
     # - v2 will return the same allocation because entry 43 is still stored with its old UID, 122 got a new UID.