swaits
b15b661c2d
The 1.0 Python mux, mirroring the Rust donglora-mux 1.0 one-for-one and implementing the full §13 multi-client semantics on the DongLoRa Protocol v2 wire format.
* Structural rewrite. donglora_mux/{daemon,intercept,session,tagmap}.py replace the 601-line __init__.py monolith.
* Tag-aware dispatch. TagMapper with skip-zero allocation and per-client bulk drop on disconnect. Multiple clients can have commands in flight concurrently; the single-oneshot response slot is gone.
* Full §13.2 multi-client SET_CONFIG semantics. Tracks (client_id, modulation) lock owner and synthesizes OK(APPLIED/MINE), OK(ALREADY_MATCHED/OTHER), or OK(LOCKED_MISMATCH/OTHER).
* TX loopback fan-out (§13.4). On TX_DONE(TRANSMITTED) the mux synthesizes an RX frame with origin = LOCAL_LOOPBACK for every connected client except the transmitter.
* Async ERR(EFRAME) fan-out: tag-0 framing errors are broadcast to all clients.
* Per-client TX cache on ClientSession so the loopback fan-out does not need the sender to re-buffer.
* Socket and serial fcntl.flock to prevent multiple mux instances from fighting over the same dongle or socket path.
* 500 ms keepalive matching PROTOCOL.md §3.4 (2x margin on the 1 s inactivity timer), replacing the 5 s cadence from 0.x.
* Clean SIGINT / SIGTERM shutdown: per-client handler tasks are tracked and cancelled, so blocked reader.read() calls no longer pin the event loop open.
78 lines
2.8 KiB
Python
78 lines
2.8 KiB
Python
"""Per-client session state and bounded send queue.
|
|
|
|
Each connected client gets a :class:`ClientSession` tracking its id,
|
|
RX-interest flag, a per-client TX cache (used by the §13.4 loopback
|
|
fan-out), and a bounded asyncio queue of outgoing wire-encoded frames.
|
|
|
|
Mirrors ``mux-rs/src/session.rs``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import contextlib
|
|
import itertools
|
|
import logging
|
|
|
|
log = logging.getLogger("donglora-mux")
|
|
|
|
# Bounded send queue capacity per client. Matches mux-rs SEND_QUEUE_CAP.
|
|
SEND_QUEUE_CAP = 256
|
|
|
|
# Emit a drop-count summary after this many drops per client — a flood of
|
|
# RX while the client is slow shouldn't produce one log line per frame.
|
|
DROP_LOG_EVERY = 32
|
|
|
|
# Monotonic client-id source. Shared across every session in the process.
|
|
_next_client_id = itertools.count()
|
|
|
|
|
|
class ClientSession:
|
|
"""State carried for a single connected client.
|
|
|
|
Does NOT own the underlying stream; the daemon spawns a per-client
|
|
reader/writer pair of asyncio tasks that do I/O directly. This type
|
|
is the shared rendez-vous point those tasks enqueue frames through.
|
|
"""
|
|
|
|
__slots__ = ("_queue", "drops", "id", "rx_interested", "tx_cache")
|
|
|
|
def __init__(self) -> None:
|
|
self.id: int = next(_next_client_id)
|
|
self.rx_interested: bool = False
|
|
# Keyed by the client's upstream tag. Populated when a TX is
|
|
# forwarded to the dongle; consumed on TX_DONE(TRANSMITTED) by
|
|
# the reader so it can synthesize a loopback RX for other
|
|
# clients without asking the sender to re-buffer the payload.
|
|
self.tx_cache: dict[int, bytes] = {}
|
|
self._queue: asyncio.Queue[bytes] = asyncio.Queue(maxsize=SEND_QUEUE_CAP)
|
|
self.drops: int = 0
|
|
|
|
@property
|
|
def label(self) -> str:
|
|
return f"client-{self.id}"
|
|
|
|
async def next_outgoing(self) -> bytes:
|
|
"""Await the next frame to send. Intended for the client-writer task."""
|
|
return await self._queue.get()
|
|
|
|
def enqueue(self, frame: bytes) -> None:
|
|
"""Best-effort non-blocking enqueue of a wire frame.
|
|
|
|
Full queue → drop the frame, bump the counter, occasionally log.
|
|
This prevents one slow client from back-pressuring the whole mux.
|
|
"""
|
|
try:
|
|
self._queue.put_nowait(frame)
|
|
except asyncio.QueueFull:
|
|
self.drops += 1
|
|
if self.drops == 1 or self.drops % DROP_LOG_EVERY == 0:
|
|
log.warning("%s: send queue full, dropped %d frames so far", self.label, self.drops)
|
|
|
|
def close(self, writer: asyncio.StreamWriter | None = None) -> None:
|
|
"""Best-effort close. The daemon normally drives shutdown via
|
|
task cancellation + stream close; this is a safety net."""
|
|
if writer is not None:
|
|
with contextlib.suppress(Exception):
|
|
writer.close()
|