Relax restriction of arbitration ID uniqueness for SocketCAN (#785)

Socketcan bcm now supports sending periodic messages with the same arbitration ID.
Updates documentation and tests for socketcan interface.

Closes #721

Author: tysonite

can/interfaces/socketcan/socketcan.py

@@ -14,6 +14,7 @@
 import socket
 import struct
 import time
+import threading
 import errno
 
 log = logging.getLogger(__name__)
@@ -162,7 +163,7 @@ def build_can_frame(msg: Message) -> bytes:
         __u8    data[CANFD_MAX_DLEN] __attribute__((aligned(8)));
     };
     """
-    can_id = _add_flags_to_can_id(msg)
+    can_id = _compose_arbitration_id(msg)
     flags = 0
     if msg.bitrate_switch:
         flags |= CANFD_BRS
@@ -286,7 +287,7 @@ def send_bcm(bcm_socket: socket.socket, data: bytes) -> int:
             raise e
 
 
-def _add_flags_to_can_id(message: Message) -> int:
+def _compose_arbitration_id(message: Message) -> int:
     can_id = message.arbitration_id
     if message.is_extended_id:
         log.debug("sending an extended id type message")
@@ -297,7 +298,6 @@ def _add_flags_to_can_id(message: Message) -> int:
     if message.is_error_frame:
         log.debug("sending error frame")
         can_id |= CAN_ERR_FLAG
-
     return can_id
 
 
@@ -310,18 +310,22 @@ class CyclicSendTask(
         - setting of a task duration
         - modifying the data
         - stopping then subsequent restarting of the task
-
     """
 
     def __init__(
         self,
         bcm_socket: socket.socket,
+        task_id: int,
         messages: Union[Sequence[Message], Message],
         period: float,
         duration: Optional[float] = None,
     ):
-        """
+        """Construct and :meth:`~start` a task.
+
         :param bcm_socket: An open BCM socket on the desired CAN channel.
+        :param task_id:
+            The identifier used to uniquely reference particular cyclic send task
+            within Linux BCM.
         :param messages:
             The messages to be sent periodically.
         :param period:
@@ -336,12 +340,12 @@ def __init__(
         super().__init__(messages, period, duration)
 
         self.bcm_socket = bcm_socket
+        self.task_id = task_id
         self._tx_setup(self.messages)
 
     def _tx_setup(self, messages: Sequence[Message]) -> None:
         # Create a low level packed frame to pass to the kernel
         body = bytearray()
-        self.can_id_with_flags = _add_flags_to_can_id(messages[0])
         self.flags = CAN_FD_FRAME if messages[0].is_fd else 0
 
         if self.duration:
@@ -353,9 +357,19 @@ def _tx_setup(self, messages: Sequence[Message]) -> None:
             ival1 = 0.0
             ival2 = self.period
 
-        # First do a TX_READ before creating a new task, and check if we get
-        # EINVAL. If so, then we are referring to a CAN message with the same
-        # ID
+        self._check_bcm_task()
+
+        header = build_bcm_transmit_header(
+            self.task_id, count, ival1, ival2, self.flags, nframes=len(messages)
+        )
+        for message in messages:
+            body += build_can_frame(message)
+        log.debug("Sending BCM command")
+        send_bcm(self.bcm_socket, header + body)
+
+    def _check_bcm_task(self):
+        # Do a TX_READ on a task ID, and check if we get EINVAL. If so,
+        # then we are referring to a CAN message with the existing ID
         check_header = build_bcm_header(
             opcode=CAN_BCM_TX_READ,
             flags=0,
@@ -364,7 +378,7 @@ def _tx_setup(self, messages: Sequence[Message]) -> None:
             ival1_usec=0,
             ival2_seconds=0,
             ival2_usec=0,
-            can_id=self.can_id_with_flags,
+            can_id=self.task_id,
             nframes=0,
         )
         try:
@@ -374,45 +388,33 @@ def _tx_setup(self, messages: Sequence[Message]) -> None:
                 raise e
         else:
             raise ValueError(
-                "A periodic Task for Arbitration ID {} has already been created".format(
-                    messages[0].arbitration_id
+                "A periodic task for Task ID {} is already in progress by SocketCAN Linux layer".format(
+                    self.task_id
                 )
             )
 
-        header = build_bcm_transmit_header(
-            self.can_id_with_flags,
-            count,
-            ival1,
-            ival2,
-            self.flags,
-            nframes=len(messages),
-        )
-        for message in messages:
-            body += build_can_frame(message)
-        log.debug("Sending BCM command")
-        send_bcm(self.bcm_socket, header + body)
-
     def stop(self) -> None:
-        """Send a TX_DELETE message to cancel this task.
+        """Stop a task by sending TX_DELETE message to Linux kernel.
 
         This will delete the entry for the transmission of the CAN-message
-        with the specified can_id CAN identifier. The message length for the command
-        TX_DELETE is {[bcm_msg_head]} (only the header).
+        with the specified :attr:`~task_id` identifier. The message length
+        for the command TX_DELETE is {[bcm_msg_head]} (only the header).
         """
         log.debug("Stopping periodic task")
 
-        stopframe = build_bcm_tx_delete_header(self.can_id_with_flags, self.flags)
+        stopframe = build_bcm_tx_delete_header(self.task_id, self.flags)
         send_bcm(self.bcm_socket, stopframe)
 
     def modify_data(self, messages: Union[Sequence[Message], Message]) -> None:
-        """Update the contents of the periodically sent messages.
+        """Update the contents of the periodically sent CAN messages by
+        sending TX_SETUP message to Linux kernel.
 
-        Note: The messages must all have the same
-        :attr:`~can.Message.arbitration_id` like the first message.
-
-        Note: The number of new cyclic messages to be sent must be equal to the
+        The number of new cyclic messages to be sent must be equal to the
         original number of messages originally specified for this task.
 
+        .. note:: The messages must all have the same
+                  :attr:`~can.Message.arbitration_id` like the first message.
+
         :param messages:
             The messages with the new :attr:`can.Message.data`.
         """
@@ -423,14 +425,22 @@ def modify_data(self, messages: Union[Sequence[Message], Message]) -> None:
 
         body = bytearray()
         header = build_bcm_update_header(
-            can_id=self.can_id_with_flags, msg_flags=self.flags, nframes=len(messages)
+            can_id=self.task_id, msg_flags=self.flags, nframes=len(messages)
         )
         for message in messages:
             body += build_can_frame(message)
         log.debug("Sending BCM command")
         send_bcm(self.bcm_socket, header + body)
 
     def start(self) -> None:
+        """Start a periodic task by sending TX_SETUP message to Linux kernel.
+
+        It verifies presence of the particular BCM task through sending TX_READ
+        message to Linux kernel prior to scheduling.
+
+        :raises ValueError:
+            If the task referenced by :attr:`~task_id` is already running.
+        """
         self._tx_setup(self.messages)
 
 
@@ -443,16 +453,17 @@ class MultiRateCyclicSendTask(CyclicSendTask):
     def __init__(
         self,
         channel: socket.socket,
+        task_id: int,
         messages: Sequence[Message],
         count: int,
         initial_period: float,
         subsequent_period: float,
     ):
-        super().__init__(channel, messages, subsequent_period)
+        super().__init__(channel, task_id, messages, subsequent_period)
 
         # Create a low level packed frame to pass to the kernel
         header = build_bcm_transmit_header(
-            self.can_id_with_flags,
+            self.task_id,
             count,
             initial_period,
             subsequent_period,
@@ -571,8 +582,10 @@ def capture_message(
 
 
 class SocketcanBus(BusABC):
-    """
-    Implements :meth:`can.BusABC._detect_available_configs`.
+    """ A SocketCAN interface to CAN.
+
+    It implements :meth:`can.BusABC._detect_available_configs` to search for
+    available interfaces.
     """
 
     def __init__(
@@ -601,6 +614,8 @@ def __init__(
         self.channel_info = "socketcan channel '%s'" % channel
         self._bcm_sockets: Dict[str, socket.socket] = {}
         self._is_filtered = False
+        self._task_id = 0
+        self._task_id_guard = threading.Lock()
 
         # set the receive_own_messages parameter
         try:
@@ -712,18 +727,26 @@ def _send_periodic_internal(
     ) -> CyclicSendTask:
         """Start sending messages at a given period on this bus.
 
-        The kernel's Broadcast Manager SocketCAN API will be used.
+        The Linux kernel's Broadcast Manager SocketCAN API is used to schedule
+        periodic sending of CAN messages. The wrapping 32-bit counter (see
+        :meth:`~_get_next_task_id()`) designated to distinguish different
+        :class:`CyclicSendTask` within BCM provides flexibility to schedule
+        CAN messages sending with the same CAN ID, but different CAN data.
 
         :param messages:
-            The messages to be sent periodically
+            The message(s) to be sent periodically.
         :param period:
             The rate in seconds at which to send the messages.
         :param duration:
             Approximate duration in seconds to continue sending messages. If
             no duration is provided, the task will continue indefinitely.
 
+        :raises ValueError:
+            If task identifier passed to :class:`CyclicSendTask` can't be used
+            to schedule new task in Linux BCM.
+
         :return:
-            A started task instance. This can be used to modify the data,
+            A :class:`CyclicSendTask` task instance. This can be used to modify the data,
             pause/resume the transmission and to stop the transmission.
 
         .. note::
@@ -732,18 +755,20 @@ def _send_periodic_internal(
             be exactly the same as the duration specified by the user. In
             general the message will be sent at the given rate until at
             least *duration* seconds.
-
         """
         msgs = LimitedDurationCyclicSendTaskABC._check_and_convert_messages(msgs)
 
         msgs_channel = str(msgs[0].channel) if msgs[0].channel else None
         bcm_socket = self._get_bcm_socket(msgs_channel or self.channel)
-        # TODO: The SocketCAN BCM interface treats all cyclic tasks sharing an
-        # Arbitration ID as the same Cyclic group. We should probably warn the
-        # user instead of overwriting the old group?
-        task = CyclicSendTask(bcm_socket, msgs, period, duration)
+        task_id = self._get_next_task_id()
+        task = CyclicSendTask(bcm_socket, task_id, msgs, period, duration)
         return task
 
+    def _get_next_task_id(self) -> int:
+        with self._task_id_guard:
+            self._task_id = (self._task_id + 1) % (2 ** 32 - 1)
+            return self._task_id
+
     def _get_bcm_socket(self, channel: str) -> socket.socket:
         if channel not in self._bcm_sockets:
             self._bcm_sockets[channel] = create_bcm_socket(self.channel)

doc/bus.rst

@@ -67,7 +67,7 @@ This thread safe version of the :class:`~can.BusABC` class can be used by multip
 Sending and receiving is locked separately to avoid unnecessary delays.
 Conflicting calls are executed by blocking until the bus is accessible.
 
-It can be used exactly like the normal :class:`~can.BusABC`:
+It can be used exactly like the normal :class:`~can.BusABC`::
 
     # 'socketcan' is only an example interface, it works with all the others too
     my_bus = can.ThreadSafeBus(interface='socketcan', channel='vcan0')