From ce90cff4d4b90b26238397276549ef0af78880f7 Mon Sep 17 00:00:00 2001 From: Sankalp Thakur Date: Tue, 27 Oct 2026 00:09:41 +0400 Subject: [PATCH 1/2] Discard the late response of a cancelled request When execute() is cancelled while it waits, the response can still arrive after the next request is sent. RTU and ASCII responses carry no transaction id, so the next request returned it as its own. On cancel, keep a fresh response future, and let the next execute() wait for it before it sends, at most timeout_connect after the cancel. Socket framers are unchanged: handleFrame already skips a response with an old transaction id. --- pymodbus/transaction/transaction.py | 20 +++++++++++--- test/transaction/test_transaction.py | 39 ++++++++++++++++++++++++++++ 2 files changed, 56 insertions(+), 3 deletions(-) diff --git a/pymodbus/transaction/transaction.py b/pymodbus/transaction/transaction.py index 765fccf70..f8e4b1447 100644 --- a/pymodbus/transaction/transaction.py +++ b/pymodbus/transaction/transaction.py @@ -65,6 +65,8 @@ def __init__( self._lock = asyncio.Lock() self.low_level_send = self.send self.response_future: asyncio.Future = asyncio.Future() + self.pending_response = False + self.pending_until = 0.0 self.last_pdu: ModbusPDU | None = None self.last_addr: tuple | None = None @@ -181,13 +183,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 +198,11 @@ async def execute( "Client cannot connect (automatic retry continuing) !!" ) async with self._lock: + if self.pending_response: + await asyncio.wait( + [self.response_future], timeout=self.pending_until - monotonic() + ) + self.pending_response = False request.transaction_id = self.getNextTID() count_retries = 0 while count_retries <= self.retries: @@ -228,6 +235,13 @@ 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 first waits for the response to this one. + self.response_future = asyncio.Future() + self.pending_response = not request.transaction_id + self.pending_until = 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..7cbad470b 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,41 @@ 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 + async def test_transaction_receiver(self, use_clc): """Test tracers in disconnect.""" transact = TransactionManager( From bbf30db1fa2f5f634c2e1e26d42e0f2d93496b5e Mon Sep 17 00:00:00 2001 From: Sankalp Thakur Date: Mon, 2 Nov 2026 11:27:04 +0400 Subject: [PATCH 2/2] Delay only the next request to the device of a cancelled request A request to another device cannot get the late response, because handleFrame skips a frame from another device. So the wait after a cancel is now kept per device, and a request with a transaction id is never delayed. --- pymodbus/transaction/transaction.py | 24 +++++++++-------- test/transaction/test_transaction.py | 40 ++++++++++++++++++++++++++++ 2 files changed, 53 insertions(+), 11 deletions(-) diff --git a/pymodbus/transaction/transaction.py b/pymodbus/transaction/transaction.py index f8e4b1447..f5f002579 100644 --- a/pymodbus/transaction/transaction.py +++ b/pymodbus/transaction/transaction.py @@ -65,8 +65,7 @@ def __init__( self._lock = asyncio.Lock() self.low_level_send = self.send self.response_future: asyncio.Future = asyncio.Future() - self.pending_response = False - self.pending_until = 0.0 + self.cancelled_requests: dict[int, tuple[asyncio.Future, float]] = {} self.last_pdu: ModbusPDU | None = None self.last_addr: tuple | None = None @@ -198,11 +197,9 @@ async def execute( # noqa: C901 "Client cannot connect (automatic retry continuing) !!" ) async with self._lock: - if self.pending_response: - await asyncio.wait( - [self.response_future], timeout=self.pending_until - monotonic() - ) - self.pending_response = False + 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: @@ -237,10 +234,15 @@ async def execute( # noqa: C901 count_retries += 1 except asyncio.exceptions.CancelledError: # RTU/ASCII responses carry no transaction id, so the next - # request first waits for the response to this one. - self.response_future = asyncio.Future() - self.pending_response = not request.transaction_id - self.pending_until = monotonic() + self.comm_params.timeout_connect + # 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: diff --git a/test/transaction/test_transaction.py b/test/transaction/test_transaction.py index 7cbad470b..cb4e1a855 100755 --- a/test/transaction/test_transaction.py +++ b/test/transaction/test_transaction.py @@ -333,6 +333,46 @@ async def test_transaction_execute_cancel(self, use_clc, late_response): 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(