diff --git a/guest/test_udp_to_sink.py b/guest/test_udp_to_sink.py new file mode 100644 index 0000000..a618d0c --- /dev/null +++ b/guest/test_udp_to_sink.py @@ -0,0 +1,85 @@ +#!/usr/bin/env python3 +"""Regression tests for the live UDP -> pw-play relay.""" + +import socket +import unittest + +import udp_to_sink as bridge + + +class FakeSocket: + def __init__(self, events): + self.events = list(events) + self.recv_calls = 0 + + def recvfrom(self, _size): + self.recv_calls += 1 + event = self.events.pop(0) + if event == "timeout": + raise socket.timeout + return event, ("127.0.0.1", 4010) + + +class RelayTests(unittest.TestCase): + def test_backpressure_drops_frames_but_keeps_draining_udp(self): + sock = FakeSocket([b"x" * 1400 for _ in range(100)]) + + def blocked(_fd, _data): + raise BlockingIOError + + misses = 0 + for _ in range(100): + misses, written = bridge._relay_once(sock, 1, misses, blocked) + self.assertFalse(written) + self.assertEqual(100, sock.recv_calls) + self.assertEqual(0, misses) + + def test_eagain_does_not_escape_or_crash(self): + def blocked(_fd, _data): + raise BlockingIOError + + self.assertFalse(bridge._write_live(1, b"pcm", blocked)) + + def test_protocol_oversize_is_dropped_without_partial_write(self): + calls = [] + self.assertFalse(bridge._write_live( + 1, b"x" * (bridge.PIPE_BUF + 1), + lambda fd, data: calls.append((fd, data)))) + self.assertEqual([], calls) + + def test_current_packet_is_written_intact(self): + calls = [] + frame = b"x" * 1400 + + def capture(fd, data): + calls.append((fd, data)) + return len(data) + + self.assertTrue(bridge._write_live(1, frame, capture)) + self.assertEqual([(1, frame)], calls) + + def test_partial_write_is_not_reported_as_success(self): + self.assertFalse(bridge._write_live( + 1, b"pcm", lambda _fd, _data: 1)) + + def test_silence_keepalive_starts_after_existing_debounce(self): + sock = FakeSocket(["timeout"] * bridge.MISS_LIMIT) + writes = [] + misses = 0 + + def capture(fd, data): + writes.append((fd, data)) + return len(data) + + for _ in range(bridge.MISS_LIMIT - 1): + misses, written = bridge._relay_once( + sock, 1, misses, capture) + self.assertFalse(written) + misses, written = bridge._relay_once( + sock, 1, misses, capture) + self.assertTrue(written) + self.assertEqual([(1, bridge.SILENCE)], writes) + + +if __name__ == "__main__": + unittest.main() diff --git a/guest/udp_to_sink.py b/guest/udp_to_sink.py index c5f1ae9..65f0d2b 100644 --- a/guest/udp_to_sink.py +++ b/guest/udp_to_sink.py @@ -17,8 +17,10 @@ LATENCY (2026-07-10): two mechanisms silently added delay on top of the 250ms. (1) The stdin pipe to pw-play is a 64KB Linux pipe = a hidden 341ms of audio; - it is now shrunk to 16KB (~85ms) via F_SETPIPE_SZ, and the Popen is - unbuffered so Python adds no batching of its own. + it is now shrunk to 16KB (~85ms) via F_SETPIPE_SZ, Popen is unbuffered, + and writes are nonblocking. If the real-time sink is full, the current + frame is dropped so the receiver keeps draining UDP instead of retaining + stale PCM in the large socket receive buffer. (2) The old gap-filler injected silence on EVERY 20ms socket timeout. When packets were merely LATE (BLE airtime burst, VM scheduling) the late data still arrived after the silence -- net queued bytes grew, playback consumes @@ -52,37 +54,66 @@ GAP = 0.02 SILENCE = b"\x00" * int(48000 * 2 * 2 * GAP) # 20ms -> matches the timeout below MISS_LIMIT = 5 # inject silence only after 5*GAP = 100ms of true no-data +PIPE_BUF = 4096 # Linux atomic pipe-write limit; sender packets are <=1400B -sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) -sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 1 << 20) -sock.bind(("0.0.0.0", 4010)) -sock.settimeout(GAP) -while True: - player = subprocess.Popen( - ["pw-play", "--format=s16", "--rate=48000", "--channels=2", - f"--latency={LATENCY_MS}ms", "-"], - stdin=subprocess.PIPE, bufsize=0, env=env) +def _write_live(pipe_fd, data, writer=os.write): + """Write one current PCM frame or drop it if the sink is backpressured. + + OpenSpan sender datagrams (<=1400B) and SILENCE (3840B) fit PIPE_BUF, so + nonblocking writes are atomic: the whole frame is accepted or EAGAIN. + Keeping the UDP receiver live is more important than replaying stale PCM. + """ + if len(data) > PIPE_BUF: + return False try: - fcntl.fcntl(player.stdin.fileno(), F_SETPIPE_SZ, PIPE_BYTES) - except OSError: - pass # kernel refuses -> default 64KB pipe; higher latency, still works - misses = 0 + return writer(pipe_fd, data) == len(data) + except BlockingIOError: + return False + + +def _relay_once(sock, pipe_fd, misses, writer=os.write): + """Drain one UDP frame (or keepalive timeout) without blocking on sink.""" try: - while player.poll() is None: - try: - data, _ = sock.recvfrom(8192) - misses = 0 - except socket.timeout: - misses += 1 - if misses < MISS_LIMIT: - continue # brief jitter: let the 250ms cushion ride it - data = SILENCE # real pause: keep the A2DP stream alive - player.stdin.write(data) - except (BrokenPipeError, OSError): - pass - finally: + data, _ = sock.recvfrom(8192) + misses = 0 + except socket.timeout: + misses += 1 + if misses < MISS_LIMIT: + return misses, False + data = SILENCE # real pause: keep the A2DP stream alive + return misses, _write_live(pipe_fd, data, writer) + + +def main(): + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 1 << 20) + sock.bind(("0.0.0.0", 4010)) + sock.settimeout(GAP) + + while True: + player = subprocess.Popen( + ["pw-play", "--format=s16", "--rate=48000", "--channels=2", + f"--latency={LATENCY_MS}ms", "-"], + stdin=subprocess.PIPE, bufsize=0, env=env) try: - player.terminate() - except Exception: + pipe_fd = player.stdin.fileno() + try: + fcntl.fcntl(pipe_fd, F_SETPIPE_SZ, PIPE_BYTES) + except OSError: + pass # kernel refuses -> default pipe; still nonblocking + os.set_blocking(pipe_fd, False) + misses = 0 + while player.poll() is None: + misses, _ = _relay_once(sock, pipe_fd, misses) + except (BrokenPipeError, OSError): pass + finally: + try: + player.terminate() + except Exception: + pass + + +if __name__ == "__main__": + main()