[Cyphal/CAN] PythonCAN: Remove send executor; improve development documentation on type checking (#293)

Co-authored-by: Pavel Kirienko <pavel.kirienko@gmail.com>

Author: silverv

.gitmodules

@@ -1,3 +1,3 @@
 [submodule "public_regulated_data_types_for_testing"]
 path = demo/public_regulated_data_types
-url = https://github.com/UAVCAN/public_regulated_data_types
+url = https://github.com/OpenCyphal/public_regulated_data_types

CHANGELOG.rst

@@ -3,6 +3,11 @@
 Changelog
 =========
 
+v1.15
+-----
+
+- Made PythonCAN support better.
+
 v1.14
 -----
 

CONTRIBUTING.rst

@@ -197,6 +197,49 @@ To abort on first error::
 
  nox -x -- -x
 
+Running MyPy
+.........................
+
+Warning, this might be obsolete.
+
+Sometimes it is useful to run MyPy directly, for instance, to check the types without waiting for a very long time
+for the tests to finish.
+Here's how to do it on Windows::
+
+ .nox\test-3-10\Scripts\activate
+ pip install mypy
+ mypy --strict pycyphal tests .nox\test-3-10\tmp\.compiled
+
+
+Running pylint
+.........................
+
+Warning, this might be obsolete.
+
+Sometimes it is useful to run pylint directly, for instance, to check the code quality without waiting
+for a very long time for the tests to finish.
+
+Here's how to do it on Windows::
+
+ .nox\test-3-10\Scripts\activate
+ pip install pylint
+ pylint pycyphal tests .nox\test-3-10\tmp\.compiled
+
+
+Running black
+.........................
+
+Warning, this might be obsolete.
+
+Sometimes it is useful to run black directly, for instance, to check the code formatting
+without waiting for a very long time for the tests to finish.
+It is better, however, to configure the IDE to invoke Black automatically on save.
+
+Here's how to do it on Windows::
+
+ pip install black
+ black pycyphal tests .nox\test-3-10\tmp\.compiled
+
 
 Running a subset of tests
 .........................

demo/public_regulated_data_types

@@ -33,3 +33,23 @@ I am getting ``ModuleNotFoundError: No module named 'uavcan'``. Do I need to ins
 Imports fail with ``AttributeError: module 'uavcan...' has no attribute '...'``. What am I doing wrong?
  Remove the legacy library: ``pip uninstall -y uavcan``.
  Read the :ref:`installation` guide for details.
+
+
+I am experiencing poor SLCAN read/write performance on Windows. What can I do?
+ Increasing the process priority to REALTIME
+ (available if the application has administrator privileges) will help.
+ Without administrator privileges, the HIGH priority set by this code,
+ will still help with delays in SLCAN performance.
+ Here's an example::
+
+ if sys.platform.startswith("win"):
+ import ctypes, psutil
+
+ # Reconfigure the system timer to run at a higher resolution. This is desirable for the real-time tests.
+ t = ctypes.c_ulong()
+ ctypes.WinDLL("NTDLL.DLL").NtSetTimerResolution(5000, 1, ctypes.byref(t))
+ p = psutil.Process(os.getpid())
+ p.nice(psutil.REALTIME_PRIORITY_CLASS)
+ elif sys.platform.startswith("linux"):
+ p = psutil.Process(os.getpid())
+ p.nice(-20)

docs/pages/faq.rst

@@ -1 +1 @@
-__version__ = "1.14.4"
+__version__ = "1.15.0"

pycyphal/_version.py

@@ -3,15 +3,15 @@
 # Author: Alex Kiselev <a.kiselev@volz-servos.com>, Pavel Kirienko <pavel@opencyphal.org>
 
 from __future__ import annotations
+import queue
 import time
 import typing
 import asyncio
 import logging
 import threading
-import functools
+from functools import partial
 import dataclasses
 import collections
-import concurrent.futures
 import warnings
 
 import can
@@ -22,6 +22,14 @@
 _logger = logging.getLogger(__name__)
 
 
+@dataclasses.dataclass(frozen=True)
+class _TxItem:
+ msg: can.Message
+ timeout: float
+ future: asyncio.Future[None]
+ loop: asyncio.AbstractEventLoop
+
+
 @dataclasses.dataclass(frozen=True)
 class PythonCANBusOptions:
  hardware_loopback: bool = False
@@ -188,7 +196,9 @@ def __init__(
  self._closed = False
  self._maybe_thread: typing.Optional[threading.Thread] = None
  self._rx_handler: typing.Optional[Media.ReceivedFramesHandler] = None
- self._background_executor = concurrent.futures.ThreadPoolExecutor(max_workers=1)
+ # This is for communication with a thread that handles the call to _bus.send
+ self._tx_queue: queue.Queue[_TxItem | None] = queue.Queue()
+ self._tx_thread = threading.Thread(target=self.transmit_thread_worker, daemon=True)
 
  params: typing.Union[_FDInterfaceParameters, _ClassicInterfaceParameters]
  if self._is_fd:
@@ -231,6 +241,7 @@ def is_fd(self) -> bool:
  return self._is_fd
 
  def start(self, handler: Media.ReceivedFramesHandler, no_automatic_retransmission: bool) -> None:
+ self._tx_thread.start()
  if self._maybe_thread is None:
  self._rx_handler = handler
  self._maybe_thread = threading.Thread(
@@ -254,6 +265,24 @@ def configure_acceptance_filters(self, configuration: typing.Sequence[FilterConf
  _logger.debug("%s: Acceptance filters activated: %s", self, ", ".join(map(str, configuration)))
  self._bus.set_filters(filters)
 
+ def transmit_thread_worker(self) -> None:
+ try:
+ while not self._closed:
+ tx = self._tx_queue.get(block=True)
+ if self._closed or tx is None:
+ break
+ try:
+ self._bus.send(tx.msg, tx.timeout)
+ tx.loop.call_soon_threadsafe(partial(tx.future.set_result, None))
+ except Exception as ex:
+ tx.loop.call_soon_threadsafe(partial(tx.future.set_exception, ex))
+ except Exception as ex:
+ _logger.critical(
+ "Unhandled exception in transmit thread, transmission thread stopped and transmission is no longer possible: %s",
+ ex,
+ exc_info=True,
+ )
+
  async def send(self, frames: typing.Iterable[Envelope], monotonic_deadline: float) -> int:
  num_sent = 0
  loopback: typing.List[typing.Tuple[Timestamp, Envelope]] = []
@@ -269,10 +298,16 @@ async def send(self, frames: typing.Iterable[Envelope], monotonic_deadline: floa
  )
  try:
  desired_timeout = monotonic_deadline - loop.time()
- await loop.run_in_executor(
- self._background_executor,
- functools.partial(self._bus.send, message, timeout=max(desired_timeout, 0)),
+ received_future: asyncio.Future[None] = asyncio.Future()
+ self._tx_queue.put_nowait(
+ _TxItem(
+ message,
+ max(desired_timeout, 0),
+ received_future,
+ asyncio.get_running_loop(),
+ )
  )
+ await received_future
  except (asyncio.TimeoutError, can.CanError): # CanError is also used to report timeouts (weird).
  break
  else:
@@ -287,6 +322,8 @@ async def send(self, frames: typing.Iterable[Envelope], monotonic_deadline: floa
  def close(self) -> None:
  self._closed = True
  try:
+ self._tx_queue.put(None)
+ self._tx_thread.join(timeout=self._MAXIMAL_TIMEOUT_SEC * 10)
  if self._maybe_thread is not None:
  self._maybe_thread.join(timeout=self._MAXIMAL_TIMEOUT_SEC * 10)
  self._maybe_thread = None