feat(stream_reader): optimize serial stream protocol detection. reduce CPU busy-spins
This commit is contained in:
jack
@@ -32,9 +32,35 @@ class StreamReader:
|
||||
self._thread = threading.Thread(target=self._run, daemon = True)
|
||||
self._thread.start()
|
||||
|
||||
def _read_chunk(self, min_bytes: int = 1) -> bytes:
|
||||
"""Read a chunk of available bytes from the transport.
|
||||
|
||||
Reads up to 256 bytes in one shot instead of byte-at-a-time,
|
||||
falling back to single-byte reads with a yield when nothing is
|
||||
immediately available."""
|
||||
chunk = self._transport.read(max(min_bytes, 256))
|
||||
if chunk:
|
||||
return chunk
|
||||
# Nothing available right now – yield briefly to avoid busy-spin.
|
||||
b = self._transport.read(1)
|
||||
if not b:
|
||||
time.sleep(0.005)
|
||||
return b
|
||||
|
||||
def _accumulate(self, needed: int) -> bytearray:
|
||||
"""Accumulate *needed* bytes using buffered reads."""
|
||||
buf = bytearray()
|
||||
while len(buf) < needed and not self._stop.is_set():
|
||||
chunk = self._transport.read(needed - len(buf))
|
||||
if not chunk:
|
||||
time.sleep(0.005)
|
||||
continue
|
||||
buf.extend(chunk)
|
||||
return buf
|
||||
|
||||
def _run(self) -> None:
|
||||
try:
|
||||
# Check the first few bytes to determine if the wafer is streaming binary, ASCII hex dump, or text lines.
|
||||
# Peek at the first few bytes to determine protocol.
|
||||
peek_buf = bytearray()
|
||||
while not self._stop.is_set() and len(peek_buf) < 32:
|
||||
b = self._transport.read(1)
|
||||
@@ -47,16 +73,24 @@ class StreamReader:
|
||||
if self._stop.is_set() or not peek_buf:
|
||||
return
|
||||
|
||||
detect_buf = bytes(c for c in peek_buf if c not in (10, 13, 0))
|
||||
if not detect_buf:
|
||||
detect_buf = bytes(peek_buf)
|
||||
raw = bytes(peek_buf)
|
||||
|
||||
if detect_buf.startswith(b"\xaa\x88") or (len(detect_buf) >= 2 and detect_buf[0] == 0xAA and detect_buf[1] == 0x88):
|
||||
self._run_binary(bytes(peek_buf))
|
||||
elif b"," in detect_buf:
|
||||
self._run_text_lines(bytes(peek_buf))
|
||||
else:
|
||||
self._run_ascii(bytes(peek_buf))
|
||||
# Binary sync bytes are the strongest signal.
|
||||
if raw.startswith(b"\xaa\x88"):
|
||||
self._run_binary(raw)
|
||||
return
|
||||
|
||||
# Text-lines (CSV) requires BOTH a newline AND commas in the
|
||||
# pre-newline content. This avoids misclassifying a binary
|
||||
# payload that happens to contain 0x2C as text.
|
||||
has_newline = b"\n" in raw or b"\r" in raw
|
||||
line_before_nl = raw.split(b"\n")[0].split(b"\r")[0]
|
||||
if has_newline and b"," in line_before_nl:
|
||||
self._run_text_lines(raw)
|
||||
return
|
||||
|
||||
# Default: ASCII hex dump stream.
|
||||
self._run_ascii(raw)
|
||||
finally:
|
||||
try:
|
||||
self._transport.close()
|
||||
@@ -68,20 +102,21 @@ class StreamReader:
|
||||
buf = bytearray(initial_bytes)
|
||||
|
||||
while not self._stop.is_set():
|
||||
# Find sync marker in buffer.
|
||||
idx = 0
|
||||
synced = False
|
||||
while idx < len(buf) - 1:
|
||||
if buf[idx] == 0xAA and buf[idx+1] == 0x88:
|
||||
if buf[idx] == 0xAA and buf[idx + 1] == 0x88:
|
||||
synced = True
|
||||
break
|
||||
idx += 1
|
||||
|
||||
if synced:
|
||||
buf = buf[idx:]
|
||||
# Accumulate header (5 bytes after sync).
|
||||
while len(buf) < 7 and not self._stop.is_set():
|
||||
b = self._transport.read(1)
|
||||
if b:
|
||||
buf.extend(b)
|
||||
chunk = self._read_chunk()
|
||||
buf.extend(chunk)
|
||||
|
||||
if self._stop.is_set():
|
||||
return
|
||||
@@ -89,17 +124,18 @@ class StreamReader:
|
||||
payload_len = int.from_bytes(buf[5:7], byteorder='little')
|
||||
total_packet_len = 2 + 5 + payload_len + 2
|
||||
|
||||
while len(buf) < total_packet_len and not self._stop.is_set():
|
||||
b = self._transport.read(1)
|
||||
if b:
|
||||
buf.extend(b)
|
||||
# Accumulate the rest of the packet in one buffered read.
|
||||
remaining = total_packet_len - len(buf)
|
||||
if remaining > 0:
|
||||
extra = self._accumulate(remaining)
|
||||
buf.extend(extra)
|
||||
|
||||
if self._stop.is_set():
|
||||
return
|
||||
|
||||
header = buf[2:7]
|
||||
seq = int.from_bytes(header[1:3], byteorder='little')
|
||||
payload = buf[7:7+payload_len]
|
||||
payload = buf[7:7 + payload_len]
|
||||
|
||||
try:
|
||||
frame = self._parse(payload, seq)
|
||||
@@ -112,14 +148,14 @@ class StreamReader:
|
||||
|
||||
buf = buf[total_packet_len:]
|
||||
else:
|
||||
# Resync: keep only trailing 0xAA if present, then read more.
|
||||
if len(buf) > 0 and buf[-1] == 0xAA:
|
||||
buf = bytearray([0xAA])
|
||||
else:
|
||||
buf.clear()
|
||||
|
||||
b = self._transport.read(1)
|
||||
if b:
|
||||
buf.extend(b)
|
||||
chunk = self._read_chunk()
|
||||
buf.extend(chunk)
|
||||
|
||||
def _run_ascii(self, initial_bytes: bytes) -> None:
|
||||
log.info("StreamReader: starting ASCII hex dump stream parsing")
|
||||
@@ -127,6 +163,7 @@ class StreamReader:
|
||||
valid_words = 80 if self._family_code == "X" else 244
|
||||
block_words = 256
|
||||
word_char_len = 4
|
||||
chars_needed = block_words * word_char_len # total hex chars per block
|
||||
|
||||
hex_chars = bytearray()
|
||||
if initial_bytes:
|
||||
@@ -136,44 +173,43 @@ class StreamReader:
|
||||
elif b == 0xAA:
|
||||
hex_chars.append(b)
|
||||
|
||||
def read_next_hex_char() -> str | None:
|
||||
if hex_chars:
|
||||
val = chr(hex_chars.pop(0))
|
||||
return val
|
||||
while not self._stop.is_set():
|
||||
b = self._transport.read(1)
|
||||
if not b:
|
||||
continue
|
||||
if b[0] == 0xAA:
|
||||
hex_chars.extend(b)
|
||||
return None
|
||||
c = chr(b[0])
|
||||
if c in "0123456789ABCDEFabcdef":
|
||||
return c
|
||||
return None
|
||||
|
||||
seq = 0
|
||||
while not self._stop.is_set():
|
||||
word_buffer = []
|
||||
eof = False
|
||||
for _ in range(block_words):
|
||||
word_str = ""
|
||||
for _ in range(word_char_len):
|
||||
c = read_next_hex_char()
|
||||
if c is None:
|
||||
eof = True
|
||||
# Accumulate a full block of hex characters using buffered reads.
|
||||
while len(hex_chars) < chars_needed and not self._stop.is_set():
|
||||
chunk = self._read_chunk(min_bytes=chars_needed - len(hex_chars))
|
||||
for byte in chunk:
|
||||
if byte == 0xAA:
|
||||
# Sync marker appeared – hand off to binary parser.
|
||||
hex_chars.append(byte)
|
||||
break
|
||||
word_str += c
|
||||
if eof or len(word_str) < word_char_len:
|
||||
eof = True
|
||||
break
|
||||
word_buffer.append(word_str)
|
||||
c = chr(byte)
|
||||
if c in "0123456789ABCDEFabcdef":
|
||||
hex_chars.append(c)
|
||||
else:
|
||||
continue
|
||||
break # broke out of inner loop due to 0xAA
|
||||
|
||||
if eof:
|
||||
if hex_chars and hex_chars[0] == 0xAA:
|
||||
log.info("StreamReader: Detected 0xAA, switching to binary mode.")
|
||||
self._run_binary(bytes(hex_chars))
|
||||
break
|
||||
# Check for early termination (0xAA sync marker anywhere in buffer).
|
||||
aa_idx = hex_chars.find(bytes([0xAA]))
|
||||
if aa_idx >= 0:
|
||||
log.info("StreamReader: Detected 0xAA at position %d, switching to binary mode.", aa_idx)
|
||||
self._run_binary(hex_chars[aa_idx:])
|
||||
return
|
||||
|
||||
# Parse words from the accumulated hex characters.
|
||||
word_buffer = []
|
||||
for i in range(0, min(len(hex_chars), chars_needed), word_char_len):
|
||||
word_str = hex_chars[i:i + word_char_len]
|
||||
if len(word_str) < word_char_len:
|
||||
break
|
||||
word_buffer.append(bytes(word_str))
|
||||
|
||||
if not word_buffer:
|
||||
time.sleep(0.01)
|
||||
continue
|
||||
|
||||
hex_chars = hex_chars[len(word_buffer) * word_char_len:]
|
||||
|
||||
valid_hex_words = word_buffer[:valid_words]
|
||||
values = []
|
||||
@@ -198,7 +234,8 @@ class StreamReader:
|
||||
seq = 0
|
||||
|
||||
while not self._stop.is_set():
|
||||
if b"\n" in buf:
|
||||
# Drain complete lines from the buffer.
|
||||
while b"\n" in buf:
|
||||
line_bytes, _, buf = buf.partition(b"\n")
|
||||
line = line_bytes.decode('utf-8', errors='ignore').strip()
|
||||
if line:
|
||||
@@ -210,13 +247,10 @@ class StreamReader:
|
||||
except Exception as exc:
|
||||
self.error_count += 1
|
||||
self._on_error(exc)
|
||||
continue
|
||||
|
||||
b = self._transport.read(1)
|
||||
if not b:
|
||||
time.sleep(0.005)
|
||||
continue
|
||||
buf.extend(b)
|
||||
# Read more data using buffered reads.
|
||||
chunk = self._read_chunk()
|
||||
buf.extend(chunk)
|
||||
|
||||
def stop(self) -> None:
|
||||
self._stop.set()
|
||||
|
||||
Reference in New Issue
Block a user