ThreadSafeBus implementation is ignored from inside the library

[deronek]

Describe the bug

ThreadSafeBus implementation of send method is not used when it's called from inside the library. This means that for all send_periodic implementation, ThreadSafeBus is completely ignored and send method of the concrete bus is used directly.

To Reproduce

Use send_periodic in ThreadSafeBus to observe that thread safe implementation of send function is not used. See "Traceback and logs" for code example.

Expected behavior

send_periodic implementation should use ThreadSafeBus implementation of send. According to the implementation, it is definitely expected. https://github.com/hardbyte/python-can/blob/dc0ae68acb28f01a503e084718cfa2acf39d1e16/can/thread_safe_bus.py#L61-L62

Below the dummy lock is created: https://github.com/hardbyte/python-can/blob/dc0ae68acb28f01a503e084718cfa2acf39d1e16/can/thread_safe_bus.py#L41-L43

And here this lock is "used" - actually it's doing nothing, since the send implementation should be synchronized in ThreadSafeBus. https://github.com/hardbyte/python-can/blob/dc0ae68acb28f01a503e084718cfa2acf39d1e16/can/broadcastmanager.py#L299-L300

Additional context

This behavior was replicated using hardware vector bus. As this issue is critical for our use case, we used a following workaround - inject the send lock in place of the dummy periodic lock after ThreadSafeBus was initialized:

bus.__wrapped__._lock_send_periodic = bus._lock_send

OS and version: Windows 10 Enterprise 10.0.19045 Build 19045 Python version: Python 3.11.0 (main, Oct 24 2022, 18:26:48) [MSC v.1933 64 bit (AMD64)] on win32 python-can version: 4.2.2 python-can interface/s (if applicable): virtual, vector

Traceback and logs

Simple code example, in which frames are sent both periodically and manually:

import time
import can

bus = can.ThreadSafeBus(
    interface='virtual',
    receive_own_messages=True
)

msg_periodic = can.Message(
    arbitration_id=0x123, data=[1, 2, 3, 4, 5, 6], is_extended_id=False
)
msg_manual = can.Message(
    arbitration_id=0x234, data=[7, 8, 9], is_extended_id=False
)

# periodic send
task = bus.send_periodic(msg_periodic, 0.2)
assert isinstance(task, can.CyclicSendTaskABC)

# manual send
bus.send(msg_manual)

# receive
start_time = time.time()
while time.time() - start_time < 1:
    rec_msg = bus.recv(1)
    print(rec_msg)
task.stop()

To observe the bug, print statements are injected into ThreadSafeBus:

    def recv(
        self, timeout=None, *args, **kwargs
    ):  # pylint: disable=keyword-arg-before-vararg
        print(f'Using thread safe recv')
        with self._lock_recv:
            return self.__wrapped__.recv(timeout=timeout, *args, **kwargs)

    def send(
        self, msg, timeout=None, *args, **kwargs
    ):  # pylint: disable=keyword-arg-before-vararg
        print(f'Using thread safe send {msg=}')
        with self._lock_send:
            return self.__wrapped__.send(msg, timeout=timeout, *args, **kwargs)

Executing above gives following output:

Using thread safe send msg=can.Message(timestamp=0.0, arbitration_id=0x234, is_extended_id=False, dlc=3, data=[0x7, 0x8, 0x9])
Using thread safe recv
Timestamp: 1687888509.605106        ID: 0123    S Tx                DL:  6    01 02 03 04 05 06
Using thread safe recv
Timestamp: 1687888509.605106        ID: 0234    S Tx                DL:  3    07 08 09
Using thread safe recv
Timestamp: 1687888509.805578        ID: 0123    S Tx                DL:  6    01 02 03 04 05 06
Using thread safe recv
Timestamp: 1687888510.005523        ID: 0123    S Tx                DL:  6    01 02 03 04 05 06
Using thread safe recv
Timestamp: 1687888510.206290        ID: 0123    S Tx                DL:  6    01 02 03 04 05 06
Using thread safe recv
Timestamp: 1687888510.406522        ID: 0123    S Tx                DL:  6    01 02 03 04 05 06
Using thread safe recv
Timestamp: 1687888510.606262        ID: 0123    S Tx                DL:  6    01 02 03 04 05 06

Manual call of bus.send worked correctly and ThreadSafeBus implementation was used. However, periodically sent frames ignored thread-safe implementation of send (no print), but the frames were sent on the bus.

[zariiii9003]

Seems plausible. Would you like to create a PR?

[deronek]

I tried to find a clean solution and was unable to do so. I presume wrapt.ObjectProxy is not correct for this use case, since something is definitely unexpected about how Python resolves the method calls. The alternative must work correctly with existing Bus implementation and how it creates the instances.

Debugging my example step by step:

• on the line task = bus.send_periodic(msg_periodic, 0.2):

>>> bus.recv
<bound method ThreadSafeBus.recv of <ThreadSafeBus at 0x2484be0f550 for VirtualBus at 0x2484bfe98d0>>
>>> bus.send
<bound method ThreadSafeBus.send of <ThreadSafeBus at 0x2484be0f550 for VirtualBus at 0x2484bfe98d0>>

• stepping into this method, on the first line of send_periodic inside the BusABC class:

>>> self.recv
<bound method BusABC.recv of <can.interfaces.virtual.VirtualBus object at 0x000002484BFE98D0>>
>>> self.send
<bound method VirtualBus.send of <can.interfaces.virtual.VirtualBus object at 0x000002484BFE98D0>>

[nerdielol]

Is this still a known issue? If it is, I would love to work to resolve this.