From de43ef47e42cb1c581706291d195502170ddd10c Mon Sep 17 00:00:00 2001 From: davidmonterocrespo24 Date: Mon, 18 May 2026 06:51:23 +0200 Subject: [PATCH] fix(pi3): pipe chardev for proto channel + Phase 2 protocol mux MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit QEMU 10's virtserialport on a socket chardev (server=on,wait=off) silently drops guest→host bytes. Reproduced cleanly: writes from inside the guest to /dev/vportp succeed (no errno) but the connected client socket receives 0 bytes. Same bug whether the client is a single recv loop, multiple threads, TCP or UNIX socket, or whether QEMU runs as server vs client. virtconsole on the same socket works fine — only virtserialport is broken. Workaround: use `pipe` chardev (a pair of named FIFOs created beforehand by qemu_manager). guest→host through .out flows reliably in QEMU 10 — verified with manual test: 'echo PIPE_TEST > /dev/vport1p1' in the guest produces 'PIPE_TEST\n' immediately on the host side. Changes: - qemu_manager._boot: allocate a temp basename, mkfifo .in + .out, pass to QEMU as 'pipe,path='. - qemu_manager._connect_gpio: open both FIFOs O_RDWR | O_NONBLOCK on host side (O_RDWR keeps the FIFOs open even when guest hasn't opened its side yet), wire .out into asyncio via loop.add_reader. - qemu_manager._reply_gpio / _send_gpio: write to .in fd via os.write. - qemu_manager._handle_gpio_line: extended Phase 1 GPIO-only parser into a full Phase 2 mux: GPIO/GPIO_SETUP/GPIO_IN/PWM_*/I2C/SPI/UART with appropriate replies. - qemu_manager._shutdown: close FDs + unlink the FIFOs. - manifest.json: bump raspberry-pi-3-virt rootfs to 2026.05+phase2-shims (the new rootfs ships the velxio shim Python modules under /usr/lib/velxio-shims/). --- .../app/services/boot_images/manifest.json | 8 +- backend/app/services/qemu_manager.py | 304 ++++++++++++++---- 2 files changed, 252 insertions(+), 60 deletions(-) diff --git a/backend/app/services/boot_images/manifest.json b/backend/app/services/boot_images/manifest.json index 340219dd..b5061c92 100644 --- a/backend/app/services/boot_images/manifest.json +++ b/backend/app/services/boot_images/manifest.json @@ -22,13 +22,13 @@ { "name": "velxio-pi-rootfs-arm64.ext4", "asset_id": "velxio-pi-rootfs-arm64-zst", - "sha256": "d09eec3bbc023dc4798d70de87048b25d2db81e5530fe0fbe0853f745bdaf910", + "sha256": "f0ce53b933543dc071ddbf6a789f7045e603590000159951d886cf67c17c91ba", "size_bytes": 629145600, - "version": "2026.05", + "version": "2026.05+phase2-shims", "compressed": { "encoding": "zstd", - "sha256": "de55525e8475eea17fe7f4c26491154a998f7379ed82a651f0d10550cb9c997f", - "size_bytes": 24725464 + "sha256": "f93b7e97466dc97f01b98f987a60b0205558d74fd51c852771d4a7d2aa8106ce", + "size_bytes": 24733652 } } ] diff --git a/backend/app/services/qemu_manager.py b/backend/app/services/qemu_manager.py index 112bcef0..e7f3753e 100644 --- a/backend/app/services/qemu_manager.py +++ b/backend/app/services/qemu_manager.py @@ -77,9 +77,13 @@ class PiInstance: self.process: subprocess.Popen | None = None self.overlay_path: str | None = None self.serial_port: int = 0 # virtio-console TCP port (/dev/hvc0) - self.gpio_port: int = 0 # virtio-serial protocol TCP port (/dev/vport0p2) + self.gpio_port: int = 0 # legacy field (kept for log compat) + self.proto_pipe_base: str | None = None # pipe chardev FIFO basename self._serial_writer: asyncio.StreamWriter | None = None self._gpio_writer: asyncio.StreamWriter | None = None + # File descriptors for the proto pipe pair (host side). + self._proto_in_fd: int | None = None # we write here → guest reads + self._proto_out_fd: int | None = None # we read here ← guest writes self._tasks: list[asyncio.Task] = [] self.running = False @@ -156,9 +160,28 @@ class QemuManager: initramfs_path: Path = images[PI3_INITRAMFS_NAME] rootfs_base: Path = images[PI3_ROOTFS_NAME] - # Allocate TCP ports for the two chardevs. + # Allocate transport endpoints for the two chardevs. + # + # User console: TCP socket — virtconsole on top of a socket + # works bidirectionally (proven by Phase 1). + # + # Protocol channel: pipe (FIFO pair on disk) — `virtserialport` + # on a socket chardev has a guest→host flow bug in QEMU 10 + # (data is silently dropped). Pipe chardev sidesteps that; + # see project/pi-emulation/decisions.md D9. inst.serial_port = _find_free_port() - inst.gpio_port = _find_free_port() + inst.gpio_port = _find_free_port() # kept for legacy log fields + + # Make a fresh FIFO pair per session. QEMU's pipe chardev + # appends ".in" (host writes / guest reads) and ".out" + # (guest writes / host reads) to the path. + inst.proto_pipe_base = tempfile.mktemp(prefix="velxio-pi-proto-") + for suffix in (".in", ".out"): + path = inst.proto_pipe_base + suffix + try: + os.mkfifo(path, 0o600) + except FileExistsError: + pass # Create overlay qcow2 backed by the velxio rootfs ext4. Each # session gets its own writable layer; reads cascade down to @@ -218,12 +241,14 @@ class QemuManager: f'server=on,wait=off', '-device', 'virtio-serial-pci,id=virtio-serial0', '-device', 'virtconsole,chardev=cons', - # Protocol channel: second virtserialport on the SAME - # virtio-serial controller. Inside the guest this is - # /dev/vport0p2. Phase 2 hooks this up to pi_protocol_mux - # for GPIO/I2C/SPI/UART/PWM. - '-chardev', f'socket,id=proto,host=127.0.0.1,port={inst.gpio_port},' - f'server=on,wait=off', + # Protocol channel: pipe (FIFO pair) instead of a socket + # chardev. virtserialport on socket chardev has a known + # guest→host flow bug in QEMU 10 (see decisions.md D9). + # Pipe creates .in (host→guest) + .out + # (guest→host) as named FIFOs that QEMU opens lazily. + # Inside the guest this still appears as /dev/vportp + # with the name "velxio-protocol". + '-chardev', f'pipe,id=proto,path={inst.proto_pipe_base}', '-device', 'virtserialport,chardev=proto,name=velxio-protocol', # Kernel cmdline: # console=hvc0 — the virtio-console is the user terminal. @@ -315,66 +340,215 @@ class QemuManager: # mux while keeping this connect/read/write plumbing identical. async def _connect_gpio(self, inst: PiInstance) -> None: - for attempt in range(10): - try: - reader, writer = await asyncio.open_connection( - '127.0.0.1', inst.gpio_port, - ) - inst._gpio_writer = writer - logger.info('%s: protocol channel connected on port %d', - inst.client_id, inst.gpio_port) - await self._read_gpio(inst, reader) - return - except (ConnectionRefusedError, OSError): - await asyncio.sleep(1.0 * (attempt + 1)) - logger.warning('%s: protocol channel connection failed', - inst.client_id) + """Open the proto pipe pair. QEMU's `pipe` chardev creates + two FIFOs: .in (we write → guest reads via /dev/vport) + and .out (guest writes → we read). - async def _read_gpio(self, inst: PiInstance, - reader: asyncio.StreamReader) -> None: - """Parse text-protocol lines from the Pi shim layer. - - Phase 1 understands GPIO only (existing protocol). Phase 2 - extends this to dispatch I2C/SPI/UART/PWM as well via - pi_protocol_mux. + We open both ends non-blocking so opening the not-yet-written + side doesn't deadlock, then register the read FIFO with the + asyncio loop so incoming bytes wake the line parser. """ - linebuf = b'' - while inst.running: - try: - chunk = await asyncio.wait_for(reader.read(256), timeout=0.1) - if not chunk: - break - linebuf += chunk - while b'\n' in linebuf: - line, linebuf = linebuf.split(b'\n', 1) - await self._handle_gpio_line( - inst, line.decode('ascii', 'ignore').strip(), - ) - except asyncio.TimeoutError: - continue - except Exception as e: - logger.warning('%s protocol read: %s', inst.client_id, e) + if not inst.proto_pipe_base: + return + in_path = inst.proto_pipe_base + '.in' + out_path = inst.proto_pipe_base + '.out' + + # Wait for the FIFOs to exist (created by _boot before this + # task is spawned). If QEMU failed to start they may never + # appear — bail after a few seconds. + for _ in range(50): + if os.path.exists(in_path) and os.path.exists(out_path): break + await asyncio.sleep(0.1) + else: + logger.warning('%s: proto pipe FIFOs missing; aborting', + inst.client_id) + return + + try: + # O_RDWR keeps both ends opened on the host side so the + # FIFOs never reach an EOF state, even if QEMU briefly + # disconnects (e.g. between booting and the guest opening + # /dev/vport). Non-blocking so we can integrate with + # the asyncio loop via add_reader/add_writer. + inst._proto_in_fd = os.open(in_path, + os.O_RDWR | os.O_NONBLOCK) + inst._proto_out_fd = os.open(out_path, + os.O_RDWR | os.O_NONBLOCK) + except OSError as exc: + logger.warning('%s: open proto pipes: %s', + inst.client_id, exc) + return + + logger.info('%s: protocol channel opened (pipe %s.{in,out})', + inst.client_id, inst.proto_pipe_base) + + loop = asyncio.get_running_loop() + # Use add_reader to integrate the readable FIFO with the loop. + linebuf = bytearray() + + def _on_readable() -> None: + fd = inst._proto_out_fd + if fd is None: + return + try: + data = os.read(fd, 4096) + except BlockingIOError: + return + except OSError: + return + if not data: + return + linebuf.extend(data) + while b'\n' in linebuf: + line, _, rest = linebuf.partition(b'\n') + linebuf[:] = rest + asyncio.create_task(self._handle_gpio_line( + inst, line.decode('ascii', 'ignore').strip(), + )) + + loop.add_reader(inst._proto_out_fd, _on_readable) + # Keep this coroutine alive while inst is running so the loop + # doesn't garbage-collect the reader registration. + while inst.running: + await asyncio.sleep(1.0) + try: + loop.remove_reader(inst._proto_out_fd) + except Exception: + pass async def _handle_gpio_line(self, inst: PiInstance, line: str) -> None: - # Expected: "GPIO <0|1>" + """Dispatch a single text-protocol line from the Pi shim layer. + + Phase 2 supports the velxio Pi shim wire format: + GPIO <0|1> → gpio_change event + GPIO_SETUP → noted (no canvas effect) + GPIO_IN → reply VAL <0|1> + PWM_START → gpio_pwm event + PWM_CHANGE → gpio_pwm event + PWM_STOP → gpio_pwm stop + I2C / SPI / UART … → reply I2C_ERR / SPI_DATA empty + (Phase 2.5 wires real bridges) + """ parts = line.split() - if len(parts) == 3 and parts[0] == 'GPIO': + if not parts: + return + op = parts[0] + + if op == 'GPIO' and len(parts) == 3: try: pin = int(parts[1]) state = int(parts[2]) - await inst.emit('gpio_change', {'pin': pin, 'state': state}) + await inst.emit('gpio_change', + {'pin': pin, 'state': state}) except ValueError: pass + return - async def _send_gpio(self, inst: PiInstance, pin: int, state: bool) -> None: - if inst._gpio_writer: - msg = f'SET {pin} {1 if state else 0}\n'.encode() - inst._gpio_writer.write(msg) + if op == 'GPIO_SETUP' and len(parts) >= 3: + # Setup is informational; emit so the frontend can show + # pin direction badges if it wants. No canvas-level wiring + # needed today. try: - await inst._gpio_writer.drain() - except Exception as e: - logger.warning('%s protocol send: %s', inst.client_id, e) + pin = int(parts[1]) + direction = parts[2] + await inst.emit( + 'gpio_setup', + {'pin': pin, 'direction': direction, + 'pull': parts[3] if len(parts) > 3 else 'pud_off'}, + ) + except ValueError: + pass + return + + if op == 'GPIO_IN' and len(parts) == 2: + # Reply with the last known state of the pin. For Phase 2 + # we just echo 0 — the canvas-side input wiring fans in + # through SET commands which the shim caches on the guest. + # When canvas-driven inputs land in Phase 2.5 this will + # query the gpio event bus' last-state map. + try: + pin = int(parts[1]) + await self._reply_gpio(inst, f'VAL {pin} 0') + except ValueError: + pass + return + + if op in ('PWM_START', 'PWM_CHANGE') and len(parts) == 4: + try: + pin = int(parts[1]) + freq = float(parts[2]) + duty = float(parts[3]) + await inst.emit('gpio_pwm', + {'pin': pin, 'frequency': freq, + 'duty_cycle': duty, + 'event': 'start' if op == 'PWM_START' else 'change'}) + except ValueError: + pass + return + + if op == 'PWM_STOP' and len(parts) == 2: + try: + pin = int(parts[1]) + await inst.emit('gpio_pwm', + {'pin': pin, 'event': 'stop'}) + except ValueError: + pass + return + + # I2C / SPI / UART — Phase 2.5 will wire these to real canvas + # bridges. For Phase 2 we send an immediate "no slave" reply + # so user code gets an exception path instead of hanging. + if op == 'I2C' and len(parts) >= 4: + sub = parts[3] + if sub in ('R', 'RR'): + bus = parts[1] + addr = parts[2] + await self._reply_gpio(inst, f'I2C_ERR {bus} {addr} no-slave') + return + + if op == 'SPI' and len(parts) >= 4 and parts[3] == 'X': + bus = parts[1] + cs = parts[2] + # Reply with zero bytes of the same length so user code + # gets a deterministic empty xfer. + try: + req_hex = parts[4] if len(parts) > 4 else '' + length = len(bytes.fromhex(req_hex)) + except ValueError: + length = 0 + await self._reply_gpio(inst, + f'SPI_DATA {bus} {cs} {"00" * length}') + return + + if op == 'UART' and len(parts) >= 3 and parts[2] == 'RX_REQ': + port = parts[1] + await self._reply_gpio(inst, f'UART_RX {port}') + return + + # Unknown — log at debug level (not a hot path) + logger.debug('pi-protocol: unhandled line: %r', line) + + async def _reply_gpio(self, inst: PiInstance, line: str) -> None: + """Send a reply frame back to the guest over the protocol + chardev (pipe-backed). Used for any op that the shim layer + waits on.""" + if inst._proto_in_fd is None: + return + try: + os.write(inst._proto_in_fd, (line + '\n').encode('ascii')) + except OSError as e: + logger.warning('%s protocol reply: %s', inst.client_id, e) + + async def _send_gpio(self, inst: PiInstance, pin: int, + state: bool) -> None: + if inst._proto_in_fd is None: + return + msg = f'SET {pin} {1 if state else 0}\n'.encode() + try: + os.write(inst._proto_in_fd, msg) + except OSError as e: + logger.warning('%s protocol send: %s', inst.client_id, e) # ── QEMU stderr watcher ─────────────────────────────────────────────────── @@ -412,6 +586,24 @@ class QemuManager: pass inst._gpio_writer = None + # Close the proto pipe FDs and unlink the FIFOs. + for attr in ('_proto_in_fd', '_proto_out_fd'): + fd = getattr(inst, attr, None) + if fd is not None: + try: + os.close(fd) + except OSError: + pass + setattr(inst, attr, None) + if inst.proto_pipe_base: + for suffix in ('.in', '.out'): + p = inst.proto_pipe_base + suffix + try: + os.unlink(p) + except OSError: + pass + inst.proto_pipe_base = None + if inst._serial_writer: try: inst._serial_writer.close()