Add lock to async requests, correct logging and length calc.

Author: janiversen

pymodbus/client/base.py

@@ -99,6 +99,7 @@ def __init__(
         self.state = ModbusTransactionState.IDLE
         self.last_frame_end: float | None = 0
         self.silent_interval: float = 0
+        self._lock = asyncio.Lock()
 
     # ----------------------------------------------------------------------- #
     # Client external interface
@@ -162,19 +163,21 @@ async def async_execute(self, request) -> ModbusResponse:
 
         count = 0
         while count <= self.retries:
-            req = self.build_response(request.transaction_id)
-            if not count or not self.no_resend_on_retry:
-                self.send(packet)
-            if self.broadcast_enable and not request.slave_id:
-                resp = None
-                break
-            try:
-                resp = await asyncio.wait_for(
-                    req, timeout=self.comm_params.timeout_connect
-                )
-                break
-            except asyncio.exceptions.TimeoutError:
-                count += 1
+            async with self._lock:
+                req = self.build_response(request.transaction_id)
+                if not count or not self.no_resend_on_retry:
+                    self.framer.resetFrame()
+                    self.send(packet)
+                if self.broadcast_enable and not request.slave_id:
+                    resp = None
+                    break
+                try:
+                    resp = await asyncio.wait_for(
+                        req, timeout=self.comm_params.timeout_connect
+                    )
+                    break
+                except asyncio.exceptions.TimeoutError:
+                    count += 1
         if count > self.retries:
             self.close(reconnect=True)
             raise ModbusIOException(
@@ -190,6 +193,7 @@ def callback_connected(self) -> None:
         """Call when connection is succcesfull."""
         if self.on_reconnect_callback:
             self.on_reconnect_callback()
+        self.framer.resetFrame()
 
     def callback_disconnected(self, exc: Exception | None) -> None:
         """Call when connection is lost."""

pymodbus/framer/base.py

@@ -131,8 +131,8 @@ def processIncomingPacket(self, data, callback, slave, **kwargs):
         :param kwargs:
         :raises ModbusIOException:
         """
-        Log.debug("Processing: {}", data, ":hex")
         self._buffer += data
+        Log.debug("Processing: {}", self._buffer, ":hex")
         if not isinstance(slave, (list, tuple)):
             slave = [slave]
         single = kwargs.pop("single", False)

pymodbus/framer/socket_framer.py

@@ -60,7 +60,7 @@ def decode_data(self, data):
             }
         return {}
 
-    def frameProcessIncomingPacket(self, single, callback, slave, tid=None, **kwargs):
+    def frameProcessIncomingPacket(self, single, callback, slave, tid=None, **kwargs):  # noqa: C901
         """Process new packet pattern.
 
         This takes in a new request packet, adds it to the current
@@ -74,7 +74,11 @@ def frameProcessIncomingPacket(self, single, callback, slave, tid=None, **kwargs
         """
         def check_frame(self):
             """Check and decode the next frame."""
-            if not len(self._buffer) > self._hsize:
+            if not self._buffer:
+                Log.debug("Frame check, no more data!")
+                return False
+            if not len(self._buffer) >= self._hsize +1:
+                Log.debug("Frame check failed, short frame {} >= {} !!", len(self._buffer), self._hsize+2)
                 return False
             (
                 self._header["tid"],
@@ -88,7 +92,7 @@ def check_frame(self):
                 self._header = {"tid": 0, "pid": 0, "len": 0, "uid": 0}
             elif len(self._buffer) - self._hsize + 1 >= self._header["len"]:
                 return True
-            Log.debug("Frame check failed, missing part of message!!")
+            Log.debug("Frame check failed, missing part of message len {}, MBAP len {} !!", len(self._buffer), self._header["len"])
             return False
 
         while True:

test/test_network.py

@@ -80,10 +80,10 @@ async def test_stub(self, use_port, use_cls):
         client.close()
         stub.close()
 
-    async def test_double_packet(self, use_port, use_cls):
+    async def test_parallel_requests(self, use_port, use_cls):
         """Test double packet on network."""
         old_data = b''
-        client = AsyncModbusTcpClient(NULLMODEM_HOST, port=use_port, retries=0)
+        client = AsyncModbusTcpClient(NULLMODEM_HOST, port=use_port, retries=0, timeout=30)
 
         def local_handle_data(data: bytes) -> bytes | None:
             """Handle server side for this test case."""
@@ -128,6 +128,9 @@ async def local_call(addr: int) -> bool:
         await stub.start_run()
 
         assert await client.connect()
-        await asyncio.gather(*[local_call(x) for x in range(1, 10)])
+        try:
+            await asyncio.gather(*[local_call(1) for x in range(1, 10)])
+        except Exception as exc:  # pylint: disable=broad-exception-caught
+            pytest.fail(exc)
         client.close()
         stub.close()