fix(socketcan): support file descriptors above 1023

can/interfaces/socketcan/socketcan.py

@@ -9,6 +9,7 @@
 import ctypes.util
 import errno
 import logging
+import math
 import select
 import socket
 import struct
@@ -46,6 +47,12 @@
 RECEIVED_ANCILLARY_BUFFER_SIZE = (
     CMSG_SPACE(RECEIVED_TIMESTAMP_STRUCT.size) if CMSG_SPACE_available else 0
 )
+MAX_POLL_TIMEOUT_MS = 2_147_483_647
+
+
+def _poll_timeout_ms(timeout: float) -> int:
+    """Convert seconds to a timeout accepted by ``poll()``."""
+    return min(math.ceil(timeout * 1000), MAX_POLL_TIMEOUT_MS)
 
 
 # Setup BCM struct
@@ -814,10 +821,19 @@ def shutdown(self) -> None:
         self.socket.close()
 
     def _recv_internal(self, timeout: float | None) -> tuple[Message | None, bool]:
+        if timeout is not None and timeout < 0:
+            raise ValueError("timeout must not be negative")
+
         try:
-            # get all sockets that are ready (can be a list with a single value
-            # being self.socket or an empty list if self.socket is not ready)
-            ready_receive_sockets, _, _ = select.select([self.socket], [], [], timeout)
+            poller = select.poll()
+            poller.register(self.socket, select.POLLIN)
+            time_left = timeout
+            while True:
+                timeout_ms = None if time_left is None else _poll_timeout_ms(time_left)
+                ready_receive_sockets = poller.poll(timeout_ms)
+                if ready_receive_sockets or timeout_ms != MAX_POLL_TIMEOUT_MS:
+                    break
+                time_left -= MAX_POLL_TIMEOUT_MS / 1000
         except OSError as error:
             # something bad happened (e.g. the interface went down)
             raise can.CanOperationError(
@@ -850,26 +866,32 @@ def send(self, msg: Message, timeout: float | None = None) -> None:
         logger_tx = log.getChild("tx")
         logger_tx.debug("sending: %s", msg)
 
-        started = time.time()
+        started = time.monotonic()
         # If no timeout is given, poll for availability
         if timeout is None:
             timeout = 0
         time_left = timeout
         data = build_can_frame(msg)
+        poller = select.poll()
+        poller.register(self.socket, select.POLLOUT)
 
         while time_left >= 0:
             # Wait for write availability
-            ready = select.select([], [self.socket], [], time_left)[1]
+            timeout_ms = _poll_timeout_ms(time_left)
+            ready = poller.poll(timeout_ms)
             if not ready:
-                # Timeout
-                break
+                if timeout_ms != MAX_POLL_TIMEOUT_MS:
+                    # Timeout
+                    break
+                time_left = timeout - (time.monotonic() - started)
+                continue
             channel = str(msg.channel) if msg.channel else None
             sent = self._send_once(data, channel)
             if sent == len(data):
                 return
             # Not all data were sent, try again with remaining data
             data = data[sent:]
-            time_left = timeout - (time.time() - started)
+            time_left = timeout - (time.monotonic() - started)
 
         raise can.CanOperationError("Transmit buffer full")
 

doc/changelog.d/2053.fixed.rst

@@ -0,0 +1 @@
+Fix SocketCAN send and receive operations for socket file descriptors greater than 1023.

test/test_socketcan.py

@@ -5,11 +5,20 @@
 """
 
 import ctypes
+import os
+import socket
 import struct
 import sys
 import unittest
 import warnings
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
+
+try:
+    import fcntl
+    import resource
+except ImportError:
+    fcntl = None
+    resource = None
 
 import can
 from can.interfaces.socketcan.constants import (
@@ -26,6 +35,7 @@
     build_bcm_transmit_header,
     build_bcm_tx_delete_header,
     build_bcm_update_header,
+    build_can_frame,
 )
 
 from .config import IS_LINUX, IS_PYPY, TEST_INTERFACE_SOCKETCAN
@@ -391,5 +401,89 @@ def test_pypy_socketcan_support(self):
                 )
 
 
+@unittest.skipUnless(IS_LINUX, "socketcan is only available on Linux")
+class SocketCANHighFdTest(unittest.TestCase):
+    def setUp(self):
+        assert fcntl is not None
+        assert resource is not None
+
+        soft_limit, _ = resource.getrlimit(resource.RLIMIT_NOFILE)
+        if soft_limit != resource.RLIM_INFINITY and soft_limit <= 1024:
+            self.skipTest("RLIMIT_NOFILE does not permit file descriptor 1024")
+
+        patcher_create = patch("can.interfaces.socketcan.socketcan.create_socket")
+        patcher_bind = patch("can.interfaces.socketcan.socketcan.bind_socket")
+
+        self.mock_create_socket = patcher_create.start()
+        self.addCleanup(patcher_create.stop)
+        self.mock_bind_socket = patcher_bind.start()
+        self.addCleanup(patcher_bind.stop)
+
+        poll_socket, peer_socket = socket.socketpair()
+        self.addCleanup(poll_socket.close)
+        self.addCleanup(peer_socket.close)
+        self.peer_socket = peer_socket
+        self.high_fd = fcntl.fcntl(poll_socket.fileno(), fcntl.F_DUPFD_CLOEXEC, 1024)
+        self.addCleanup(os.close, self.high_fd)
+
+        self.mock_socket = MagicMock()
+        self.mock_socket.fileno.return_value = self.high_fd
+        self.mock_create_socket.return_value = self.mock_socket
+
+        self.bus = can.Bus(interface="socketcan", channel="can0")
+        self.addCleanup(self.bus.shutdown)
+
+    def test_send_high_fd(self):
+        msg = can.Message(arbitration_id=0x123, data=range(8))
+        frame_data = build_can_frame(msg)
+        self.mock_socket.send.return_value = len(frame_data)
+
+        self.bus.send(msg)
+
+        self.mock_socket.send.assert_called_once_with(frame_data)
+
+    @patch("can.interfaces.socketcan.socketcan.capture_message")
+    def test_recv_high_fd(self, mock_capture):
+        expected_msg = can.Message(
+            arbitration_id=0x123,
+            data=range(8),
+            channel="can0",
+            timestamp=1000.0,
+        )
+        mock_capture.return_value = expected_msg
+        self.peer_socket.send(b"x")
+
+        msg = self.bus.recv(timeout=1.0)
+
+        self.assertIsNotNone(msg)
+        self.assertEqual(msg.arbitration_id, 0x123)
+        self.assertEqual(msg.data, bytearray(range(8)))
+        mock_capture.assert_called_once_with(self.mock_socket, False)
+
+    @patch("can.interfaces.socketcan.socketcan.select.poll")
+    def test_recv_rejects_negative_timeout(self, mock_poll):
+        mock_poll.return_value.poll.return_value = []
+
+        with self.assertRaisesRegex(ValueError, "timeout must not be negative"):
+            self.bus.recv(timeout=-1.0)
+
+        mock_poll.return_value.poll.assert_not_called()
+
+    @patch("can.interfaces.socketcan.socketcan.select.poll")
+    def test_send_caps_large_finite_poll_timeout(self, mock_poll):
+        max_poll_timeout_ms = 2_147_483_647
+        mock_poll.return_value.poll.return_value = []
+        msg = can.Message(arbitration_id=0x123, data=range(8))
+
+        with patch(
+            "can.interfaces.socketcan.socketcan.time.monotonic",
+            side_effect=[0.0, max_poll_timeout_ms / 1000 + 2.0],
+        ):
+            with self.assertRaisesRegex(can.CanOperationError, "Transmit buffer full"):
+                self.bus.send(msg, timeout=max_poll_timeout_ms / 1000 + 1.0)
+
+        mock_poll.return_value.poll.assert_called_once_with(max_poll_timeout_ms)
+
+
 if __name__ == "__main__":
     unittest.main()