diff --git a/pymodbus/transaction/transaction.py b/pymodbus/transaction/transaction.py index 765fccf70..f5f002579 100644 --- a/pymodbus/transaction/transaction.py +++ b/pymodbus/transaction/transaction.py @@ -65,6 +65,7 @@ def __init__( self._lock = asyncio.Lock() self.low_level_send = self.send self.response_future: asyncio.Future = asyncio.Future() + self.cancelled_requests: dict[int, tuple[asyncio.Future, float]] = {} self.last_pdu: ModbusPDU | None = None self.last_addr: tuple | None = None @@ -181,13 +182,13 @@ def sync_execute(self, no_response_expected: bool, request: ModbusPDU) -> Modbus Log.error(txt) raise self._io_exception_from_request(txt, request) - async def execute( + async def execute( # noqa: C901 self, no_response_expected: bool, request: ModbusPDU ) -> ModbusPDU: """Execute requests asynchronously. - REMARK: this method is identical to sync_execute, apart from the lock and try/except. - any changes in either method MUST be mirrored !!! + REMARK: this method is identical to sync_execute, apart from the lock, try/except + and the wait after a cancel. any changes in either method MUST be mirrored !!! """ if not self.transport: Log.warning("Not connected, trying to connect!") @@ -196,6 +197,9 @@ async def execute( "Client cannot connect (automatic retry continuing) !!" ) async with self._lock: + if cancelled := self.cancelled_requests.pop(request.dev_id, None): + late_response, until = cancelled + await asyncio.wait([late_response], timeout=until - monotonic()) request.transaction_id = self.getNextTID() count_retries = 0 while count_retries <= self.retries: @@ -228,6 +232,18 @@ async def execute( return response except asyncio.exceptions.TimeoutError: count_retries += 1 + except asyncio.exceptions.CancelledError: + # RTU/ASCII responses carry no transaction id, so the next + # request to this device first waits for the response to + # this one. handleFrame skips a response from another + # device, so requests to other devices are not delayed. + if not request.transaction_id: + self.response_future = asyncio.Future() + self.cancelled_requests[request.dev_id] = ( + self.response_future, + monotonic() + self.comm_params.timeout_connect, + ) + raise if self.count_until_disconnect < 0: if self.comm_params.comm_type != CommType.SERIAL: self.connection_lost(asyncio.TimeoutError("Server not responding")) diff --git a/test/transaction/test_transaction.py b/test/transaction/test_transaction.py index 11dcacab0..cb4e1a855 100755 --- a/test/transaction/test_transaction.py +++ b/test/transaction/test_transaction.py @@ -11,6 +11,10 @@ from pymodbus.framer import FramerRTU, FramerSocket, FramerType from pymodbus.pdu import DecodePDU, ExceptionResponse from pymodbus.pdu.bit_message import ReadCoilsRequest, ReadCoilsResponse +from pymodbus.pdu.register_message import ( + ReadHoldingRegistersRequest, + ReadHoldingRegistersResponse, +) from pymodbus.transaction import TransactionManager from pymodbus.transport import CommType @@ -294,6 +298,81 @@ async def test_transaction_execute(self, use_clc, scenario): # noqa: C901 assert exc_info.value.dev_id == request.dev_id assert exc_info.value.transaction_id == request.transaction_id + @pytest.mark.parametrize("late_response", [True, False]) + async def test_transaction_execute_cancel(self, use_clc, late_response): + """Test next request does not get the response of a cancelled request.""" + transact = TransactionManager( + use_clc, + FramerRTU(DecodePDU(False)), + 0, + False, + None, + None, + None, + ) + transport = mock.Mock() + transact.connection_made(transport) + transact.comm_params.timeout_connect = 0.5 + resp = asyncio.create_task( + transact.execute(False, ReadHoldingRegistersRequest(dev_id=1, count=4)) + ) + await asyncio.sleep(0.1) + resp.cancel() + with pytest.raises(asyncio.CancelledError): + await resp + resp = asyncio.create_task( + transact.execute(False, ReadHoldingRegistersRequest(dev_id=1, count=5)) + ) + await asyncio.sleep(0.1) + if late_response: + late = ReadHoldingRegistersResponse(dev_id=1, registers=[1, 2, 3, 4]) + transact.data_received(transact.framer.buildFrame(late)) + while transport.write.call_count < 2: + await asyncio.sleep(0.1) + response = ReadHoldingRegistersResponse(dev_id=1, registers=[1, 2, 3, 4, 5]) + transact.data_received(transact.framer.buildFrame(response)) + assert (await resp).registers == response.registers + + @pytest.mark.parametrize(("framer", "dev_id"), [(FramerRTU, 2), (FramerSocket, 1)]) + async def test_transaction_execute_cancel_no_wait(self, use_clc, framer, dev_id): + """Test no wait after a cancel for a request that cannot get its response. + + That is a request to another device, or any request with a transaction id. + """ + transact = TransactionManager( + use_clc, + framer(DecodePDU(False)), + 0, + False, + None, + None, + None, + ) + transport = mock.Mock() + transact.connection_made(transport) + transact.comm_params.timeout_connect = 5 + resp = asyncio.create_task( + transact.execute(False, ReadHoldingRegistersRequest(dev_id=1, count=4)) + ) + await asyncio.sleep(0.1) + resp.cancel() + with pytest.raises(asyncio.CancelledError): + await resp + resp = asyncio.create_task( + transact.execute(False, ReadHoldingRegistersRequest(dev_id=dev_id, count=5)) + ) + await asyncio.sleep(0.1) + assert transport.write.call_count == 2 + late = ReadHoldingRegistersResponse( + dev_id=1, transaction_id=1, registers=[1, 2, 3, 4] + ) + transact.data_received(transact.framer.buildFrame(late)) + response = ReadHoldingRegistersResponse( + dev_id=dev_id, transaction_id=2, registers=[1, 2, 3, 4, 5] + ) + transact.data_received(transact.framer.buildFrame(response)) + assert (await resp).registers == response.registers + async def test_transaction_receiver(self, use_clc): """Test tracers in disconnect.""" transact = TransactionManager(