Inline protocol parser for improved throughput

[caspervonb]

Replace the modular parse() function with chunk-based inline parsing in _read_loop. Reading 64KB chunks and parsing multiple messages per buffer avoids function call overhead on the hot path.

Handler signatures now accept bytes instead of str to defer decoding until necessary.

Subscriber throughput improves ~63% (295k to 480k msg/sec with pub/sub on CPython).

[Copilot]

Pull request overview

This PR replaces the modular parse() function with chunk-based inline parsing in the _read_loop method to improve subscriber throughput by ~63% (295k to 480k msg/sec). The changes introduce a 64KB buffer-based approach that parses multiple messages per buffer to reduce function call overhead.

Key changes:

• Inline protocol parsing using 64KB chunks instead of calling parse() for each message
• Handler signatures now accept bytes instead of str for subject, SID, and reply fields to defer decoding
• Manual buffer management with rollback logic for incomplete messages

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Missing error handling for decode() calls on subject and reply. Both can raise UnicodeDecodeError.

Add error handling:

try:
    message = Message(
        subject=subject.decode(),
        data=payload,
        reply=reply.decode() if reply else None,
        headers=Headers(headers) if headers else None,
        status=status,
    )
except UnicodeDecodeError as e:
    logger.error("Invalid UTF-8 in HMSG fields: %s", e)
    return

[Copilot]

When an unknown protocol is encountered, the code skips to the next line and continues parsing. However, if CRLF is not found in the remaining buffer, the incomplete variable is set to data[pos:], but the next iteration will attempt to parse this data as a complete message starting from position 0. This could lead to repeatedly trying to parse invalid protocol lines.

Consider logging the unknown protocol before skipping:

# Unknown protocol, skip to next line
logger.warning("Unknown protocol at position %d: %r", pos, data[pos:min(pos+20, data_len)])
crlf = data.find(CRLF, pos)
if crlf == -1:
    incomplete = data[pos:]
    break
pos = crlf + 2

Suggested change

	crlf = data.find(CRLF, pos)
	if crlf == -1:
	incomplete = data[pos:]
	logger.warning("Unknown protocol at position %d: %r", pos, data[pos:min(pos+20, data_len)])
	crlf = data.find(CRLF, pos)
	if crlf == -1:
	# Discard invalid data to avoid infinite loop
	logger.error("No CRLF found after unknown protocol at position %d, discarding %d bytes", pos, data_len - pos)
	incomplete = b""

[Copilot]

The position rollback calculation appears incorrect. When the subject parsing fails, the code rolls back by len(subject) + 5, but at this point subject hasn't been assigned yet (the space1 wasn't found). This will cause an UnboundLocalError.

The rollback should simply be:

if space1 == -1:
    incomplete = data[pos - 4:]
    break

[Copilot]

Missing error handling for decode() call. If the subject contains invalid UTF-8, this will raise UnicodeDecodeError.

Add error handling:

try:
    subject_str = subject.decode()
except UnicodeDecodeError:
    logger.error("Invalid UTF-8 in subject for slow consumer warning")
    subject_str = repr(subject)

Suggested change

	subject_str = subject.decode()
	try:
	subject_str = subject.decode()
	except UnicodeDecodeError:
	logger.error("Invalid UTF-8 in subject for slow consumer warning")
	subject_str = repr(subject)

[Copilot]

The new inline parsing logic lacks dedicated unit tests. The existing tests in test_protocol.py test the old modular parse() function which is no longer used in the read loop. The inline parser should have comprehensive tests covering:

• Incomplete messages spanning multiple chunks
• All message types (MSG, HMSG, PING, PONG, INFO, +OK, -ERR)
• Invalid payload sizes and error conditions
• Malformed messages and protocol violations
• Position rollback scenarios
• Buffer boundary conditions

Consider adding integration tests that exercise the inline parser through the _read_loop or extracting the parsing logic into testable functions.

[Copilot]

Potential crash due to missing error handling. The decode() call can raise UnicodeDecodeError if the SID contains invalid UTF-8 bytes. This would crash the read loop.

Add error handling:

try:
    sid_str = sid.decode()
except UnicodeDecodeError:
    logger.error("Invalid UTF-8 in SID: %r", sid)
    return

Suggested change

	sid_str = sid.decode()
	try:
	sid_str = sid.decode()
	except UnicodeDecodeError:
	logger.error("Invalid UTF-8 in SID: %r", sid)
	return

[Copilot]

Similar rollback calculation issue as lines 460-462 and 470-472. The formula pos - len(subject) - len(sid) - len(header_rest) - 10 is fragile and may not accurately restore to the message start. A miscalculation could cause parsing errors or skip messages.

Use a saved starting position instead:

msg_start_pos = pos - 4  # Save position before "MSG "
# ... parsing logic ...
if payload_end + 2 > data_len:
    incomplete = data[msg_start_pos:]
    break

[Copilot]

Magic number 0x4D (ASCII 'M') is used without explanation. While there's a comment, consider defining these as constants at the top of the function for better maintainability:

MSG_CHAR = 0x4D  # 'M'
PING_PONG_CHAR = 0x50  # 'P'
HMSG_CHAR = 0x48  # 'H'
INFO_CHAR = 0x49  # 'I'
OK_CHAR = 0x2B  # '+'
ERR_CHAR = 0x2D  # '-'
PING_I_CHAR = 0x49  # 'I'
PONG_O_CHAR = 0x4F  # 'O'

[Copilot]

Missing error handling for decode() call. If the subject contains invalid UTF-8, this will raise UnicodeDecodeError.

Add error handling:

try:
    subject_str = subject.decode()
except UnicodeDecodeError:
    logger.error("Invalid UTF-8 in subject for slow consumer warning")
    subject_str = repr(subject)

Suggested change

	subject_str = subject.decode()
	try:
	subject_str = subject.decode()
	except UnicodeDecodeError:
	logger.error("Invalid UTF-8 in subject for slow consumer warning")
	subject_str = repr(subject)

[Copilot]

Similar to line 784, missing error handling for decode(). If the SID contains invalid UTF-8, this will crash.

Add error handling:

try:
    sid_str = sid.decode()
except UnicodeDecodeError:
    logger.error("Invalid UTF-8 in SID: %r", sid)
    return

Suggested change

	sid_str = sid.decode()
	try:
	sid_str = sid.decode()
	except UnicodeDecodeError:
	logger.error("Invalid UTF-8 in SID: %r", sid)
	return