Using async changes thread for non-async callbacks which is unexpected

[sveinse]

Using the can library with async enabled Notifier change the thread which all callbacks is called from. Without async, the notifier callbacks are run in the thread of the can rx thread. With async, it runs the callbacks via loop.call_soon_threadsafe() which cause the notifier callbacks to be run in the main async thread instead. This changes behavior for all callbacks, not just async callbacks, which is unexpected. E.g. if using a non-async can protocol stack plus another async protocol stack together might create problems due to the change of callback threads.

https://github.com/hardbyte/python-can/blob/5d62394006f42d1bf98e159fe9bb08d10e47e6eb/can/notifier.py#L111-L119

I propose a fix that prevents the async call_soon_threadsafe in Notifier._rx_thread() with the introduction of a setting run_message_reception_in_eventloop. Then the user can chose to opt out of running the regular non-async callbacks in the asyncio main thread. The default should be True so we don't change the current behavior.

def _rx_thread(self, bus: BusABC) -> None:
    # determine message handling callable early, not inside while loop
    if self._loop and self.run_message_reception_in_eventloop:              # <--- NEW
        handle_message: Callable[[Message], Any] = functools.partial(
            self._loop.call_soon_threadsafe,
            self._on_message_received,  # type: ignore[arg-type]
        )
    else:
        handle_message = self._on_message_received

A change is required in Notifier._on_message_received(). This function might be called in either threads, so calling the asyncio coroutine must be made thread safe:

def _on_message_received(self, msg: Message) -> None:
    for callback in self.listeners:
        res = callback(msg)
        if res and self._loop and asyncio.iscoroutine(res):
            # Schedule coroutine
            # In __init__: self._tasks: set[asyncio.Task] = set()
            task = asyncio.run_coroutine_threadsafe(res, self._loop)

When looking at this, #1938 and #1912 should be addressed in the same round.

[zariiii9003]

This changes behavior for all callbacks, not just async callbacks, which is unexpected.

I disagree. If i run an event loop, i wouldn't want a library to make callbacks from random threads. That would introduce potential race conditions, which is not desirable in asyncio.

Btw.: self._loop.add_reader(reader, self._on_message_available, bus) also does not call back from another thread. You only consider the threading fallback.

[sveinse]

The current behavior implies that applications using python-can cannot switch arbitrarily from non-async to async, even when not using coroutines.

Allow me to present a use case: I'm a maintainer for the python canopen package and I'm working on getting async support. The current implementation of canopen use callbacks with blocking locks to synchronize data with the main thread. I had hoped to avoid rewriting the entire callback backend of the package by reusing the same synchronous callbacks.

However if python-can is invoked with loop it will change the synchronous callbacks to be called from the event loop thread as discussed. This means that these synchronous callbacks now call blocking locks in the main event loop thread, which is an anti-pattern. The current workaround I'm using is to call python-can without loop, even when the calling package is run via asyncio. This way the old behavior of being called from a separate thread is maintained.

After some thought I think I agree with not implementing my proposal. Say that someone wants to use canopen next to another protocol for 29-bits CAN messages (which is realistic). If one library depends on callbacks being called from a separate thread, and another needs them called from the main loop thread, the proposal is incomplete.