From b315c8c63321c0378bb1dbed79fd99b8988908bf Mon Sep 17 00:00:00 2001 From: Klaudiusz Staniek Date: Wed, 3 Sep 2025 17:01:54 +0200 Subject: [PATCH 1/6] fix(tx): correct _tx_loop to pop _TxItem and send frames contiguously - pop `_TxItem` (not raw bytes) and check `item is None` - avoid variable shadowing; use `frame_bytes` in send loop - on writer=None: requeue whole item, clear `_connected`, yield - on write error: requeue remaining frames as one atomic item - preserves non-interleaving batches and prevents tight re-loops --- caneth/client.py | 205 +++++++++++++++++++++++++++++++++-------------- 1 file changed, 147 insertions(+), 58 deletions(-) diff --git a/caneth/client.py b/caneth/client.py index 2cd9c23..bf9b6fd 100644 --- a/caneth/client.py +++ b/caneth/client.py @@ -32,6 +32,15 @@ # ----------------------------- Data structures --------------------------------- +@dataclass(slots=True) +class _TxItem: + """One atomic unit for transmission.""" + + frames: list[bytes] # one or more 13-byte encoded frames + atomic: bool # if True, must not be interleaved + can_id: int | None = None # optional metadata (for logging) + + @dataclass(slots=True) class CANFrame: """ @@ -150,7 +159,7 @@ def __init__( self.reconnect_cap = float(reconnect_cap) # Outgoing buffered frames (already encoded as 13-byte chunks) - self._tx_buf: deque[bytes] = deque() + self._tx_buf: deque[_TxItem] = deque() self._tx_cv: asyncio.Condition = asyncio.Condition() self._send_buffer_limit = int(send_buffer_limit) self._drop_oldest_on_full = bool(drop_oldest_on_full) @@ -376,55 +385,58 @@ async def send( wait_for_space: bool = False, ) -> None: """ - Enqueue a frame for sending. The TX loop will flush it when connected. - - Args: - can_id: Integer CAN ID. - data: Byte-like payload (0..8 bytes). Iterable[int] accepted. - extended: Force extended or standard. If None, inferred as (can_id > 0x7FF). - rtr: Remote frame flag. - wait_for_space: If True and buffer is full and `drop_oldest_on_full=False`, - this call will block until space is available. + Enqueue a single frame (non-atomic item). TX loop will flush it when connected. + For non-interleaved sequences, use send_batch(...) or the atomic(...) context. """ - # If the client was explicitly closed, sending is an error. + # Explicit close => sending is an error if self._closed.is_set(): raise RuntimeError("Client is closed") - payload = bytes(data) if not isinstance(data, bytes | bytearray) else bytes(data) + # Normalize payload + payload = data if isinstance(data, bytes | bytearray) else bytes(data) if len(payload) > 8: raise ValueError("data length must be <= 8") + if extended is None: extended = can_id > 0x7FF - frame = CANFrame(can_id=int(can_id), data=payload, extended=bool(extended), rtr=bool(rtr), dlc=len(payload)) - raw = frame.to_bytes() - async with self._tx_cv: - # Fast path: space available - if len(self._tx_buf) < self._send_buffer_limit: - self._tx_buf.append(raw) - self._tx_cv.notify() - return + raw = self._encode_frame(can_id=int(can_id), data=payload, extended=bool(extended), rtr=bool(rtr)) - # Buffer full - if self._drop_oldest_on_full: - # Drop oldest, enqueue newest - _ = self._tx_buf.popleft() - self._tx_buf.append(raw) - self._tx_cv.notify() - return + # Non-atomic item with a single frame (keeps backward compatibility) + item = _TxItem(frames=[raw], atomic=False, can_id=int(can_id)) + await self._enqueue_item(item, wait_for_space=wait_for_space) - if not wait_for_space: - # Respect limit; drop this frame and log a warning - self.log.warning("TX buffer full; dropping frame id=0x%X", can_id) - return + async def send_batch( + self, + can_id: int, + data: Iterable[bytes | bytearray | Iterable[int]], + *, + extended: bool | None = None, + rtr: bool = False, + wait_for_space: bool = False, + ) -> None: + """ + Enqueue several frames for the same CAN ID as an *atomic* batch. + They will be sent back-to-back without interleaving. + The batch remains atomic across reconnects (remaining frames re-queued together). + """ + if self._closed.is_set(): + raise RuntimeError("Client is closed") - # Back-pressure: wait for space - while len(self._tx_buf) >= self._send_buffer_limit and not self._closed.is_set(): - await self._tx_cv.wait() - if self._closed.is_set(): - return - self._tx_buf.append(raw) - self._tx_cv.notify() + data_list: list[bytes] = [] + for d in data: + b = bytes(d) if not isinstance(d, bytes | bytearray) else bytes(d) + if len(b) > 8: + raise ValueError("data length must be <= 8") + data_list.append(b) + + if extended is None: + extended = can_id > 0x7FF + + frames = [self._encode_frame(can_id, b, extended=extended, rtr=rtr) for b in data_list] + if not frames: + return + await self._enqueue_item(_TxItem(frames=frames, atomic=True, can_id=int(can_id)), wait_for_space=wait_for_space) # ---------------------------- Internals -------------------------------- @@ -530,60 +542,65 @@ async def _read_loop(self) -> None: async def _tx_loop(self) -> None: """ - Background transmitter: flush buffered frames when connected. + Background transmitter: flush buffered items when connected. - - Waits for `_connected` before attempting writes. - - On write error, re-queues the frame at the head and waits for reconnection. + - Pops one _TxItem at a time. + - Sends all frames within the item back-to-back (preserving atomic batches). + - On write error mid-item, re-queues the remaining frames as a single atomic item. + - If writer disappears mid-loop, re-queues the whole item, clears _connected, and yields. """ while not self._closed.is_set(): - # Wait until we have a connection + # Wait for a connection await self._connected.wait() if self._closed.is_set(): break - # Drain buffer while connected try: while self._connected.is_set() and not self._closed.is_set(): - raw: bytes | None = None + item: _TxItem | None = None async with self._tx_cv: if self._tx_buf: - raw = self._tx_buf.popleft() + item = self._tx_buf.popleft() # notify space for potential waiters self._tx_cv.notify() else: - # Wait for new data or disconnection/close + # Wait for new data or periodic re-check (to notice disconnect/close) try: await asyncio.wait_for(self._tx_cv.wait(), timeout=self._TX_WAIT_TIMEOUT) except asyncio.TimeoutError: - # Timeout occurred; no new data, continue loop continue - if raw is None: + if item is None: continue writer = self._writer if writer is None: - # Lost connection detected here; make state consistent. + # Lost connection; put the whole item back and hand off to reconnect. async with self._tx_cv: - self._tx_buf.appendleft(raw) + self._tx_buf.appendleft(item) self._tx_cv.notify() self._connected.clear() # Yield to avoid a tight loop and let reconnect/teardown tasks run. - await asyncio.sleep(0) # avoid potential tight re-loop + await asyncio.sleep(0) # cooperative yield to avoid a tight re-loop break + # Send all frames in this item contiguously try: - writer.write(raw) - await writer.drain() + for idx, frame_bytes in enumerate(item.frames): # noqa B007 + writer.write(frame_bytes) + await writer.drain() except Exception as e: self.log.warning("Write error: %s; will retry after reconnect", e) - # Re-queue at head to preserve order - async with self._tx_cv: - self._tx_buf.appendleft(raw) - self._tx_cv.notify() - # Force reconnect handling by clearing the flag + # Re-queue remaining frames atomically (preserve non-interleaving) + remaining = item.frames[idx:] if "idx" in locals() else item.frames + if remaining: + async with self._tx_cv: + self._tx_buf.appendleft(_TxItem(frames=remaining, atomic=True, can_id=item.can_id)) + self._tx_cv.notify() + # Hand control to reconnect manager self._connected.clear() break + except asyncio.CancelledError: break except Exception: @@ -648,3 +665,75 @@ async def _dispatch(self, frame: CANFrame) -> None: await res except Exception: self.log.exception("Error in wait_for callback") + + def _encode_frame(self, can_id: int, data: bytes, *, extended: bool, rtr: bool) -> bytes: + frame = CANFrame(can_id=int(can_id), data=data, extended=bool(extended), rtr=bool(rtr), dlc=len(data)) + return frame.to_bytes() + + async def _enqueue_item(self, item: _TxItem, *, wait_for_space: bool) -> None: + async with self._tx_cv: + if len(self._tx_buf) < self._send_buffer_limit: + self._tx_buf.append(item) + self._tx_cv.notify() + return + + if self._drop_oldest_on_full: + _ = self._tx_buf.popleft() # drop oldest item + self._tx_buf.append(item) + self._tx_cv.notify() + return + + if not wait_for_space: + self.log.warning("TX buffer full; dropping item (can_id=%s, frames=%d)", item.can_id, len(item.frames)) + return + + # Back-pressure + while len(self._tx_buf) >= self._send_buffer_limit and not self._closed.is_set(): + await self._tx_cv.wait() + if not self._closed.is_set(): + self._tx_buf.append(item) + self._tx_cv.notify() + + +class _AtomicSender: + def __init__( + self, client: WaveShareCANClient, can_id: int, *, extended: bool | None, rtr: bool, wait_for_space: bool + ): + self._client = client + self._can_id = int(can_id) + self._extended = extended + self._rtr = rtr + self._wait = wait_for_space + self._datas: list[bytes] = [] + + async def send(self, data: bytes | bytearray | Iterable[int]) -> None: + b = bytes(data) if not isinstance(data, bytes | bytearray) else bytes(data) + if len(b) > 8: + raise ValueError("data length must be <= 8") + self._datas.append(b) + + async def __aenter__(self) -> _AtomicSender: + return self + + async def __aexit__(self, exc_type, exc, tb) -> None: + if exc is None and self._datas: + await self._client.send_batch( + self._can_id, + self._datas, + extended=self._extended, + rtr=self._rtr, + wait_for_space=self._wait, + ) + + +def atomic( + self, can_id: int, *, extended: bool | None = None, rtr: bool = False, wait_for_space: bool = False +) -> _AtomicSender: + """ + Usage: + async with client.atomic(0x123) as a: + await a.send(b"\x01") + await a.send(b"\x02") + Both frames are sent contiguously (no interleaving). + """ + return _AtomicSender(self, can_id, extended=extended, rtr=rtr, wait_for_space=wait_for_space) From 76bde45fac328102b83f6d7b230d186cc0ed0664 Mon Sep 17 00:00:00 2001 From: Klaudiusz Staniek Date: Wed, 3 Sep 2025 17:21:34 +0200 Subject: [PATCH 2/6] fix: atomic context manager fix --- caneth/client.py | 25 ++++---- tests/test_atomic_context.py | 119 +++++++++++++++++++++++++++++++++++ 2 files changed, 131 insertions(+), 13 deletions(-) create mode 100644 tests/test_atomic_context.py diff --git a/caneth/client.py b/caneth/client.py index bf9b6fd..11fe0b9 100644 --- a/caneth/client.py +++ b/caneth/client.py @@ -694,6 +694,18 @@ async def _enqueue_item(self, item: _TxItem, *, wait_for_space: bool) -> None: self._tx_buf.append(item) self._tx_cv.notify() + def atomic( + self, can_id: int, *, extended: bool | None = None, rtr: bool = False, wait_for_space: bool = False + ) -> _AtomicSender: + """ + Usage: + async with client.atomic(0x123) as a: + await a.send(b"\x01") + await a.send(b"\x02") + Both frames are sent contiguously (no interleaving). + """ + return _AtomicSender(self, can_id, extended=extended, rtr=rtr, wait_for_space=wait_for_space) + class _AtomicSender: def __init__( @@ -724,16 +736,3 @@ async def __aexit__(self, exc_type, exc, tb) -> None: rtr=self._rtr, wait_for_space=self._wait, ) - - -def atomic( - self, can_id: int, *, extended: bool | None = None, rtr: bool = False, wait_for_space: bool = False -) -> _AtomicSender: - """ - Usage: - async with client.atomic(0x123) as a: - await a.send(b"\x01") - await a.send(b"\x02") - Both frames are sent contiguously (no interleaving). - """ - return _AtomicSender(self, can_id, extended=extended, rtr=rtr, wait_for_space=wait_for_space) diff --git a/tests/test_atomic_context.py b/tests/test_atomic_context.py new file mode 100644 index 0000000..e574c78 --- /dev/null +++ b/tests/test_atomic_context.py @@ -0,0 +1,119 @@ +import asyncio + +import pytest +from caneth.client import WaveShareCANClient + +from .conftest import build_frame + +pytestmark = pytest.mark.asyncio + + +async def test_atomic_context_no_interleave(ws_server): + """ + Frames added within `async with client.atomic(can_id)` must be sent + back-to-back (no interleaving), relative to other frames. + """ + host, port, state = ws_server + + client = WaveShareCANClient( + host, + port, + name="atomic-no-interleave", + send_buffer_limit=16, + drop_oldest_on_full=True, + ) + + # Enqueue something BEFORE the batch + await client.send(0x100, b"\xa0") + + # Start & connect + await client.start() + await client.wait_connected(timeout=2.0) + await state.wait_client_connected(timeout=2.0) + + # Create an atomic batch for a different CAN ID + async with client.atomic(0x222) as a: + await a.send(b"\x01") + await a.send(b"\x02") + await a.send(b"\x03") + + # Enqueue something AFTER the batch + await client.send(0x100, b"\xa1") + + # Expect order: A0, [batch 01, 02, 03 contiguous], A1 + got1 = await state.recv(timeout=2.0) + got2 = await state.recv(timeout=2.0) + got3 = await state.recv(timeout=2.0) + got4 = await state.recv(timeout=2.0) + got5 = await state.recv(timeout=2.0) + + assert got1 == build_frame(0x100, b"\xa0", extended=False) + + # Batch frames contiguous, same CAN ID + assert got2 == build_frame(0x222, b"\x01", extended=False) + assert got3 == build_frame(0x222, b"\x02", extended=False) + assert got4 == build_frame(0x222, b"\x03", extended=False) + + assert got5 == build_frame(0x100, b"\xa1", extended=False) + + # No more frames + with pytest.raises(asyncio.TimeoutError): + await state.recv(timeout=0.2) + + await client.close() + + +async def test_atomic_context_mid_batch_disconnect(ws_server): + """ + If the connection drops mid-batch, the remaining frames of that atomic + batch must be re-queued together and delivered contiguously after reconnect. + """ + host, port, state = ws_server + + client = WaveShareCANClient( + host, + port, + name="atomic-mid-drop", + send_buffer_limit=16, + drop_oldest_on_full=True, + ) + + await client.start() + await client.wait_connected(timeout=2.0) + await state.wait_client_connected(timeout=2.0) + + # Enqueue an atomic batch of three frames + async with client.atomic(0x333) as a: + await a.send(b"\x10") + await a.send(b"\x11") + await a.send(b"\x12") + + # Receive the first frame of the batch + first = await state.recv(timeout=2.0) + assert first == build_frame(0x333, b"\x10", extended=False) + + # Simulate mid-batch disconnect: close the client's writer transport. + # This will cause the next write/drain to fail and trigger the TX loop's + # requeue of the remaining frames as a single atomic item. + w = client._writer # type: ignore[attr-defined] + if w is not None: + try: + w.close() + await w.wait_closed() + except Exception: + pass + + # The client should reconnect automatically + await state.wait_client_connected(timeout=2.0) + + # Remaining two frames must arrive contiguously and in order + rem1 = await state.recv(timeout=2.0) + rem2 = await state.recv(timeout=2.0) + assert rem1 == build_frame(0x333, b"\x11", extended=False) + assert rem2 == build_frame(0x333, b"\x12", extended=False) + + # No extras + with pytest.raises(asyncio.TimeoutError): + await state.recv(timeout=0.2) + + await client.close() From 768cd2c83eed9e71dd921dc2b31412a4946de5b0 Mon Sep 17 00:00:00 2001 From: Klaudiusz Staniek Date: Wed, 3 Sep 2025 17:47:54 +0200 Subject: [PATCH 3/6] fix: Use an explicit counter --- caneth/client.py | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) diff --git a/caneth/client.py b/caneth/client.py index 11fe0b9..763d2c6 100644 --- a/caneth/client.py +++ b/caneth/client.py @@ -585,20 +585,23 @@ async def _tx_loop(self) -> None: break # Send all frames in this item contiguously + sent = 0 # how many frames were fully sent+drained try: - for idx, frame_bytes in enumerate(item.frames): # noqa B007 + for idx, frame_bytes in enumerate(item.frames): writer.write(frame_bytes) await writer.drain() + sent = idx + 1 # we finished this one successfully + except (asyncio.CancelledError, GeneratorExit): + raise # don't swallow cancellations except Exception as e: self.log.warning("Write error: %s; will retry after reconnect", e) - # Re-queue remaining frames atomically (preserve non-interleaving) - remaining = item.frames[idx:] if "idx" in locals() else item.frames + # Re-queue the remaining frames atomically (starting from the first unsent) + remaining = item.frames[sent:] if remaining: async with self._tx_cv: self._tx_buf.appendleft(_TxItem(frames=remaining, atomic=True, can_id=item.can_id)) self._tx_cv.notify() - # Hand control to reconnect manager - self._connected.clear() + self._connected.clear() # hand control to reconnect manager break except asyncio.CancelledError: From 7597ec260eac6fc9c98c078165b3188c2f8a824d Mon Sep 17 00:00:00 2001 From: Klaudiusz Staniek Date: Wed, 3 Sep 2025 18:00:29 +0200 Subject: [PATCH 4/6] fix: remaining frames keep atomic setting --- caneth/client.py | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/caneth/client.py b/caneth/client.py index 763d2c6..a454b46 100644 --- a/caneth/client.py +++ b/caneth/client.py @@ -587,8 +587,8 @@ async def _tx_loop(self) -> None: # Send all frames in this item contiguously sent = 0 # how many frames were fully sent+drained try: - for idx, frame_bytes in enumerate(item.frames): - writer.write(frame_bytes) + for idx, frame_data in enumerate(item.frames): + writer.write(frame_data) await writer.drain() sent = idx + 1 # we finished this one successfully except (asyncio.CancelledError, GeneratorExit): @@ -599,7 +599,9 @@ async def _tx_loop(self) -> None: remaining = item.frames[sent:] if remaining: async with self._tx_cv: - self._tx_buf.appendleft(_TxItem(frames=remaining, atomic=True, can_id=item.can_id)) + self._tx_buf.appendleft( + _TxItem(frames=remaining, atomic=item.atomic, can_id=item.can_id) + ) self._tx_cv.notify() self._connected.clear() # hand control to reconnect manager break From a2da4fed5e927a6735a5706a63712431341df2a8 Mon Sep 17 00:00:00 2001 From: Klaudiusz Staniek Date: Wed, 3 Sep 2025 18:03:22 +0200 Subject: [PATCH 5/6] Update caneth/client.py Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --- caneth/client.py | 27 ++++++++++++++++++++++++--- 1 file changed, 24 insertions(+), 3 deletions(-) diff --git a/caneth/client.py b/caneth/client.py index a454b46..ce8128a 100644 --- a/caneth/client.py +++ b/caneth/client.py @@ -34,13 +34,34 @@ @dataclass(slots=True) class _TxItem: - """One atomic unit for transmission.""" + """ + One atomic unit for transmission. + + The `atomic` field controls frame ordering and interleaving guarantees: + + - If `atomic=True`, all frames in `frames` are sent together, without interleaving + with frames from other transmissions. This guarantees that the sequence of frames + is preserved, even across reconnections. If a reconnection occurs before all frames + are sent, the entire atomic group will be retransmitted, ensuring ordering and atomicity. + + - If `atomic=False`, frames may be interleaved with other transmissions. There is no + guarantee that the frames will be sent together or in order relative to other frames. + During reconnection, only unsent frames will be retransmitted, and ordering may not + be preserved. + + Use `atomic=True` when frame ordering and atomic delivery are required (e.g., for + multi-frame transactions or protocols that require strict sequencing). Use `atomic=False` + for independent frames where ordering and grouping are not critical. + + Attributes: + frames: List of one or more 13-byte encoded frames to transmit. + atomic: If True, frames are sent as an atomic group; if False, frames may be interleaved. + can_id: Optional CAN ID metadata (for logging). + """ frames: list[bytes] # one or more 13-byte encoded frames atomic: bool # if True, must not be interleaved can_id: int | None = None # optional metadata (for logging) - - @dataclass(slots=True) class CANFrame: """ From 6b61c650111374874a080f51665960b0b529a74c Mon Sep 17 00:00:00 2001 From: Klaudiusz Staniek Date: Wed, 3 Sep 2025 18:04:33 +0200 Subject: [PATCH 6/6] fix: formatting --- caneth/client.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/caneth/client.py b/caneth/client.py index ce8128a..e84782c 100644 --- a/caneth/client.py +++ b/caneth/client.py @@ -62,6 +62,8 @@ class _TxItem: frames: list[bytes] # one or more 13-byte encoded frames atomic: bool # if True, must not be interleaved can_id: int | None = None # optional metadata (for logging) + + @dataclass(slots=True) class CANFrame: """