diff --git a/can/interfaces/socketcan/socketcan.py b/can/interfaces/socketcan/socketcan.py index 6dc856cbf..572b2ba12 100644 --- a/can/interfaces/socketcan/socketcan.py +++ b/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") diff --git a/doc/changelog.d/2053.fixed.rst b/doc/changelog.d/2053.fixed.rst new file mode 100644 index 000000000..8003a9744 --- /dev/null +++ b/doc/changelog.d/2053.fixed.rst @@ -0,0 +1 @@ +Fix SocketCAN send and receive operations for socket file descriptors greater than 1023. diff --git a/test/test_socketcan.py b/test/test_socketcan.py index 9d042f425..2810aea3a 100644 --- a/test/test_socketcan.py +++ b/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()