Feature: Additional thread safe bus instance wrapper

can/__init__.py

@@ -2,7 +2,7 @@
 # coding: utf-8
 
 """
-can is an object-orient Controller Area Network interface module.
+can is an object-oriented Controller Area Network interface module.
 """
 
 from __future__ import absolute_import
@@ -17,27 +17,31 @@
 
 
 class CanError(IOError):
+    """
+    Indicates an error related to CAN messages or busses.
+    """
     pass
 
-from can.listener import Listener, BufferedReader, RedirectReader
+from .listener import Listener, BufferedReader, RedirectReader
 
-from can.io import Logger, Printer, LogReader
-from can.io import ASCWriter, ASCReader
-from can.io import BLFReader, BLFWriter
-from can.io import CanutilsLogReader, CanutilsLogWriter
-from can.io import CSVWriter, CSVReader
-from can.io import SqliteWriter, SqliteReader
+from .io import Logger, Printer, LogReader
+from .io import ASCWriter, ASCReader
+from .io import BLFReader, BLFWriter
+from .io import CanutilsLogReader, CanutilsLogWriter
+from .io import CSVWriter, CSVReader
+from .io import SqliteWriter, SqliteReader
 
-from can.util import set_logging_level
+from .util import set_logging_level
 
-from can.message import Message
-from can.bus import BusABC
-from can.notifier import Notifier
-from can.interfaces import VALID_INTERFACES
+from .message import Message
+from .bus import BusABC
+from .notifier import Notifier
+from .interfaces import VALID_INTERFACES
 from . import interface
 from .interface import Bus
+from .thread_safe_bus import ThreadSafeBus
 
-from can.broadcastmanager import send_periodic, \
+from .broadcastmanager import send_periodic, \
     CyclicSendTaskABC, \
     LimitedDurationCyclicSendTaskABC, \
     ModifiableCyclicTaskABC, \

can/bus.py

@@ -100,10 +100,10 @@ def send_periodic(self, msg, period, duration=None):
             least *duration* seconds.
 
         """
-        if not hasattr(self, "_lock"):
+        if not hasattr(self, "_lock_send_periodic"):
             # Create send lock for this bus
-            self._lock = threading.Lock()
-        return ThreadBasedCyclicSendTask(self, self._lock, msg, period, duration)
+            self._lock_send_periodic = threading.Lock()
+        return ThreadBasedCyclicSendTask(self, self._lock_send_periodic, msg, period, duration)
 
     def __iter__(self):
         """Allow iteration on messages as they are received.

can/interface.py

@@ -93,7 +93,7 @@ def __new__(cls, other, channel=None, *args, **kwargs):
                 )
             )
 
-        return cls(channel, **kwargs)
+        return cls(channel, *args, **kwargs)
 
 
 class CyclicSendTask(CyclicSendTaskABC):

can/interfaces/socketcan/socketcan_ctypes.py

@@ -60,7 +60,7 @@ def __init__(self,
         log.debug("Result of createSocket was %d", self.socket)
 
         # Add any socket options such as can frame filters
-        if 'can_filters' in kwargs and len(kwargs['can_filters']) > 0:
+        if 'can_filters' in kwargs:
             log.debug("Creating a filtered can bus")
             self.set_filters(kwargs['can_filters'])
 

can/thread_safe_bus.py

@@ -0,0 +1,76 @@
+#!/usr/bin/env python
+# coding: utf-8
+
+"""
+"""
+
+from __future__ import print_function, absolute_import
+
+from threading import RLock
+
+from can import Bus, BusABC
+
+
+class NullContextManager(object):
+    """
+    A context manager that does nothing at all.
+    """
+    def __init__(self, resource=None):
+        self.resource = resource
+    def __enter__(self):
+        return self.resource
+    def __exit__(self, *args):
+        pass
+
+
+class ThreadSafeBus():
+    """
+    Contains a thread safe :class:`~can.BusABC` implementation that
+    wraps around an existing interface instance. All methods of that
+    base class are now safe to be called from multiple threads.
+
+    This approach assumes that both :meth:`~can.BusABC.send` and
+    :meth:`~can.BusABC.recv` of the underlying bus instance can be
+    called simultaneously.
+
+    Use this as a drop in replacement for :class:`can.BusABC`.
+    """
+
+    def __init__(self, *args, **kwargs):
+        # create the underlying bus
+        setattr(self, '_bus', Bus(*args, **kwargs))
+
+        # init a lock for sending and one for receiving
+        setattr(self, '_lock_send', RLock())
+        setattr(self, '_lock_recv', RLock())
+
+        # now the send periodic does not need a lock anymore, but the
+        # implementation still requires a context manager to be able
+        # to be called
+        self._bus._lock_send_periodic = NullContextManager()
+
+    def __getattribute__(self, name):
+        return getattr(self._bus, name)
+
+    def __setattr__(self, name, value):
+        setattr(self._bus, name, value)
+
+    def recv(self, timeout=None, *args, **kwargs):
+        with self._lock_recv:
+            return self._bus.recv(timeout=timeout, *args, **kwargs)
+
+    def send(self, msg, timeout=None, *args, **kwargs):
+        with self._lock_send:
+            return self._bus.send(msg, timeout=timeout, *args, **kwargs)
+
+    def set_filters(self, can_filters=None, *args, **kwargs):
+        with self._lock_recv:
+            return self._bus.set_filters(can_filters=can_filters, *args, **kwargs)
+
+    def flush_tx_buffer(self, *args, **kwargs):
+        with self._lock_send:
+            return self._bus.flush_tx_buffer(*args, **kwargs)
+
+    def shutdown(self, *args, **kwargs):
+        with self._lock_send, self._lock_recv:
+            return self._bus.shutdown(*args, **kwargs)