diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 84ef93f754..1d87ad948d 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -157,16 +157,26 @@ jobs: run: | QEMU_ACCEL=kvm CARGO_BUILD="$PWD/.github/tools/github_actions_run_cargo build" \ ./dev_tools/run_optee_on_vm_userland.sh -u -t concurrent-sessions -t ta-death -t hello-ta + - name: Enable vhost-vsock + # For the vsock test of run_linux_on_vm_userland.sh, which + # REQUIRE_VSOCK makes fail rather than skip without it. + run: | + sudo modprobe vhost_vsock + echo 'KERNEL=="vhost-vsock", MODE="0666", OPTIONS+="static_node=vhost-vsock"' | sudo tee /etc/udev/rules.d/99-vhost-vsock.rules + sudo udevadm control --reload-rules + sudo udevadm trigger --name-match=vhost-vsock + sudo udevadm settle + test -r /dev/vhost-vsock -a -w /dev/vhost-vsock - name: Run static and dynamic Linux programs in a QEMU VM with stacked runners (KVM) # Also builds the kernel and userland runners. run: | - QEMU_ACCEL=kvm CARGO_BUILD="$PWD/.github/tools/github_actions_run_cargo build" \ + REQUIRE_VSOCK=1 QEMU_ACCEL=kvm CARGO_BUILD="$PWD/.github/tools/github_actions_run_cargo build" \ ./dev_tools/run_linux_on_vm_userland.sh - name: Run Linux programs in a QEMU VM with stacked runners on 2 CPUs (KVM) # Processes run on every CPU; the suite also checks that the other CPU # ran some. run: | - QEMU_SMP=2 QEMU_ACCEL=kvm CARGO_BUILD="$PWD/.github/tools/github_actions_run_cargo build" \ + REQUIRE_VSOCK=1 QEMU_SMP=2 QEMU_ACCEL=kvm CARGO_BUILD="$PWD/.github/tools/github_actions_run_cargo build" \ ./dev_tools/run_linux_on_vm_userland.sh - name: Run OP-TEE TAs in a QEMU VM with stacked runners on 2 CPUs (KVM) run: | diff --git a/dev_tools/qemu_test_lib.sh b/dev_tools/qemu_test_lib.sh index 7e98cbe827..9a367ee6d2 100644 --- a/dev_tools/qemu_test_lib.sh +++ b/dev_tools/qemu_test_lib.sh @@ -53,7 +53,21 @@ default_tas() { } WORK=$(mktemp -d) -trap 'rm -rf "$WORK"' EXIT +# run_vm's background jobs still running (`timeout`, which passes the signal +# on to QEMU, and the host client): stopped and reaped on exit. +background=() +cleanup() { + local pid + set +e + for pid in "${background[@]}"; do + # A client's processes, then the subshell running it. + pkill -P "$pid" 2>/dev/null + kill "$pid" 2>/dev/null + done + [[ ${#background[@]} -eq 0 ]] || wait "${background[@]}" 2>/dev/null + rm -rf "$WORK" +} +trap cleanup EXIT pass=0 fail=0 @@ -63,37 +77,85 @@ skipped=0 # run_vm NAME KERNEL PAYLOAD [QEMU_ARG]...: boots KERNEL with PAYLOAD as its # initrd and records the outcome; with $verbose set, prints passing logs too. -# If $check names a function, a guest pass also needs `$check NAME` to -# succeed; it prints why not otherwise. +# If $client names a function, `$client NAME` runs alongside as the host +# client, and a pass needs it to exit 0: a failing client stops the VM, and +# one still running 5s after the VM ends is stopped. If $check names a +# function, a guest pass also needs `$check NAME` to succeed; it prints why +# not otherwise. run_vm() { - local name=$1 kernel=$2 payload=$3 log="$WORK/$1.log" status why + local name=$1 kernel=$2 payload=$3 log="$WORK/$1.log" status why qemu cpid + local client_status=0 stopped=0 shift 3 set +e - # shellcheck disable=SC2086 # QEMU is a command line - timeout "$TIMEOUT" $QEMU \ - -machine q35 -accel "$QEMU_ACCEL" -cpu "$CPU" \ - -m "$QEMU_MEM" -smp "$QEMU_SMP" -nic none \ - -kernel "$kernel" -initrd "$payload" -append "litebox.log=${LITEBOX_LOG:-info}" \ - -serial stdio -display none -no-reboot \ - -device isa-debug-exit,iobase=0xf4,iosize=0x04 \ - "$@" \ - "$log" 2>&1 - status=$? + # shellcheck disable=SC2206 # QEMU is a command line + local vm=(timeout "$TIMEOUT" $QEMU + -machine q35 -accel "$QEMU_ACCEL" -cpu "$CPU" + -m "$QEMU_MEM" -smp "$QEMU_SMP" -nic none + -kernel "$kernel" -initrd "$payload" -append "litebox.log=${LITEBOX_LOG:-info}" + -serial stdio -display none -no-reboot + -device "isa-debug-exit,iobase=0xf4,iosize=0x04" + "$@") + if [[ -z ${client:-} ]]; then + "${vm[@]}" "$log" 2>&1 + status=$? + else + "${vm[@]}" "$log" 2>&1 & + qemu=$! + background=("$qemu") + $client "$name" >"$WORK/$name.client" 2>&1 & + cpid=$! + background+=("$cpid") + # Until either ends (before bash 5.1, `wait -n` takes no PIDs). + while kill -0 "$qemu" 2>/dev/null && kill -0 "$cpid" 2>/dev/null; do + wait -n "$qemu" "$cpid" + done + if kill -0 "$qemu" 2>/dev/null; then + background=("$qemu") + wait "$cpid" + client_status=$? + if [[ $client_status -ne 0 ]]; then + kill "$qemu" 2>/dev/null && stopped=1 + fi + else + background=("$cpid") + for _ in {1..50}; do + kill -0 "$cpid" 2>/dev/null || break + sleep 0.1 + done + if kill -0 "$cpid" 2>/dev/null; then + pkill -P "$cpid" 2>/dev/null || kill "$cpid" 2>/dev/null + fi + wait "$cpid" + client_status=$? + fi + wait "$qemu" + status=$? + background=() + fi set -e # isa-debug-exit: QEMU exits with (value << 1) | 1; 33 = pass. - if [[ $status -eq 33 ]] && { [[ -z ${check:-} ]] || why=$($check "$name"); }; then + if [[ $status -eq 33 && $client_status -eq 0 ]] && + { [[ -z ${check:-} ]] || why=$($check "$name"); }; then echo "PASS $name" pass=$((pass + 1)) - [[ ${verbose:-0} -eq 0 ]] || cat "$log" + if [[ ${verbose:-0} -ne 0 ]]; then + cat "$log" + [[ -z ${client:-} ]] || cat "$WORK/$name.client" + fi else case $status in 33) ;; 65) why="guest reported failure" ;; 124) why="timed out after ${TIMEOUT}s" ;; - *) why="exit status $status" ;; + *) if [[ $stopped -ne 0 ]]; then why="stopped"; else why="exit status $status"; fi ;; esac + [[ $client_status -eq 0 ]] || why="${why:+$why, }host client exit status $client_status" echo "FAIL $name ($why)" sed 's/^/ | /' "$log" | tail -40 + if [[ -n ${client:-} ]]; then + echo " host client:" + sed 's/^/ > /' "$WORK/$name.client" | tail -20 + fi fail=$((fail + 1)) failed+=("$name") fi diff --git a/dev_tools/run_linux_on_vm_userland.sh b/dev_tools/run_linux_on_vm_userland.sh index 8076419b56..74e1174e17 100755 --- a/dev_tools/run_linux_on_vm_userland.sh +++ b/dev_tools/run_linux_on_vm_userland.sh @@ -37,6 +37,12 @@ # A fresh host directory per test, with greeting.txt, is mounted read-only at # /mnt/host over virtio-9p (its mount tag). # +# A test with a host client, .host.py, also gets a vhost-vsock device +# with guest CID $VSOCK_CID, and runs `python3 .host.py CID CONSOLE` +# alongside the VM (CONSOLE: the console output's file), which must exit 0 +# too. Without a usable /dev/vhost-vsock, or python3 AF_VSOCK, it is skipped, +# or, with REQUIRE_VSOCK set, fails. +# # With QEMU_SMP above 1, the CPUs but the boot CPU must have run programs too: # the kernel prints the slices each CPU ran, and the run fails if, over the # passing tests with concurrent runs, the other CPUs ran none (a smoke check, @@ -60,6 +66,10 @@ # QEMU_SMP guest CPUs, as QEMU's -smp (default: 1) # TIMEOUT per-run timeout in seconds (default: 120, or 600 with tcg) # LITEBOX_LOG guest log level, e.g. debug (default: info) +# REQUIRE_VSOCK if set, tests with a host client fail where they cannot run, +# instead of being skipped (CI sets it) +# VSOCK_CID guest CID of tests with a host client (default: 3 plus +# this script's PID, unique among running scripts) # CARGO_BUILD command that builds the runners, run in their directories # (default: "cargo build"; CI uses # ".github/tools/github_actions_run_cargo build") @@ -74,6 +84,11 @@ TESTS_DIR="$KERNEL_DIR/tests/linux" HOSTED_TESTS_DIR="$REPO_DIR/litebox_runner_linux_userland/tests" SHARED=(fork_threads_parent.c fork_interrupt_parent.c fork_event_parent.c fork_epoll_parent.c fork_shared_futex_parent.c) +VSOCK_CID=${VSOCK_CID:-$((3 + $$))} +if [[ ! $VSOCK_CID =~ ^[1-9][0-9]{0,9}$ ]] || ((VSOCK_CID < 3 || VSOCK_CID > 4294967294)); then + echo "error: VSOCK_CID '$VSOCK_CID' must be a decimal integer from 3 to 4294967294" >&2 + exit 2 +fi # shellcheck source=dev_tools/qemu_test_lib.sh source "$SCRIPT_DIR/qemu_test_lib.sh" @@ -173,6 +188,21 @@ share_args() { -device virtio-9p-pci,fsdev=fs0,mount_tag=/mnt/host,disable-legacy=on } +# Why host clients cannot run here, if they cannot. +host_client_unusable() { + if [[ ! -r /dev/vhost-vsock || ! -w /dev/vhost-vsock ]]; then + echo "/dev/vhost-vsock is not readable and writable" + elif ! python3 -c 'import socket; socket.socket(socket.AF_VSOCK, socket.SOCK_STREAM).close()' \ + 2>/dev/null; then + echo "python3 cannot open AF_VSOCK sockets" + fi +} + +# Test $1's host client: see the header. +run_host_client() { + TIMEOUT=$TIMEOUT python3 "$TESTS_DIR/$1.host.py" "$VSOCK_CID" "$WORK/$1.stdout" +} + # The kernel checks the output; this only shows it. show_console() { if [[ ${verbose:-0} -ne 0 ]]; then @@ -236,6 +266,23 @@ for t in "${tests[@]}"; do skipped=$((skipped + 1)) continue fi + client= + vsock=() + if [[ -f "$TESTS_DIR/$t.host.py" ]]; then + why=$(host_client_unusable) + if [[ -n $why && -n ${REQUIRE_VSOCK:-} ]]; then + echo "FAIL $t ($why)" + fail=$((fail + 1)) + failed+=("$t") + continue + elif [[ -n $why ]]; then + echo "SKIP $t ($why)" + skipped=$((skipped + 1)) + continue + fi + client=run_host_client + vsock=(-device "vhost-vsock-pci,guest-cid=$VSOCK_CID,disable-legacy=on") + fi mkdir -p "$WORK/$t" cp "$USERLAND" "$WORK/$t/runner.elf" cp "$WORK/rootfs.tar" "$WORK/$t/rootfs.tar" @@ -253,7 +300,8 @@ for t in "${tests[@]}"; do feeder=$! fi # shellcheck disable=SC2046 # these print separate arguments - run_vm "$t" "$KERNEL" "$WORK/$t.tar" $(console_args "$t") $(network_args) $(share_args "$t") + run_vm "$t" "$KERNEL" "$WORK/$t.tar" $(console_args "$t") $(network_args) $(share_args "$t") \ + "${vsock[@]}" if [[ -n $feeder ]]; then kill "$feeder" 2>/dev/null || true wait "$feeder" 2>/dev/null || true diff --git a/litebox_broker_vm_kernel/Cargo.toml b/litebox_broker_vm_kernel/Cargo.toml index a6da0478be..5ac510792b 100644 --- a/litebox_broker_vm_kernel/Cargo.toml +++ b/litebox_broker_vm_kernel/Cargo.toml @@ -18,5 +18,8 @@ litebox_platform_vm_kernel = { path = "../litebox_platform_vm_kernel", version = log = { version = "0.4", default-features = false } spin = { version = "0.10.0", default-features = false, features = ["spin_mutex"] } +[dev-dependencies] +litebox_broker_core = { path = "../litebox_broker_core", version = "0.1.0", features = ["test-support"] } + [lints] workspace = true diff --git a/litebox_broker_vm_kernel/src/lib.rs b/litebox_broker_vm_kernel/src/lib.rs index eed911aab6..0c2b3bc59d 100644 --- a/litebox_broker_vm_kernel/src/lib.rs +++ b/litebox_broker_vm_kernel/src/lib.rs @@ -14,9 +14,11 @@ //! Timers are on the kernel clock. IPv4 TCP and UDP //! sockets are `litebox_broker_smoltcp`'s, which mediates their traffic: //! between runner processes directly, and, with the runner's [`Network`] -//! device ([`Config::network`]), to the outside on smoltcp. Readiness (e.g., of -//! standard input, a timer's expiry, or sockets) goes to the association's -//! notification ring. +//! device ([`Config::network`]), to the outside on smoltcp. With the runner's +//! socket device ([`Config::vsock`]), the [`vsock`] forwarder connects the +//! host's `AF_VSOCK` connections to guests' Unix stream sockets. Readiness +//! (e.g., of standard input, a timer's expiry, or sockets) goes to the +//! association's notification ring. //! //! As in the userland broker, requests are served apart from the runner's //! calls ([`Association::serve`]), and a wait blocks only its process: the @@ -60,6 +62,7 @@ mod readiness; pub mod share; pub mod stdio; mod timers; +pub mod vsock; use alloc::boxed::Box; use alloc::sync::Arc; @@ -131,6 +134,10 @@ pub struct Config { pub shares: Vec, /// The network sockets reach beyond the guest, if any. pub network: Option<&'static dyn Network>, + /// The socket device the host's `AF_VSOCK` connections arrive on, if + /// any (see [`vsock`]); ignored without + /// [`Self::allow_process_duplication`] (programs). + pub vsock: Option<&'static dyn vsock::Vsock>, /// Whether processes may start child processes (process duplication), as /// the userland broker's `allow_process_duplication`; the kernel then /// creates their runner processes (see [`PendingRunnerAssociation`]). Roots @@ -149,6 +156,8 @@ pub struct Broker { sockets: Arc, /// The virtual init, with [`Config::allow_process_duplication`]. init: Option>, + /// With [`Config::vsock`]. + vsock: Option>, } impl Broker { @@ -192,6 +201,18 @@ impl Broker { .expect("a running process can change its place in the tree"); init }); + // Only programs' guests listen on vsock; after init, which must be + // process 1. + if config.vsock.is_some() && init.is_none() { + log::warn!("vsock device ignored without process duplication: host connections hang"); + } + let platform = config.platform; + let vsock = config.vsock.filter(|_| init.is_some()).and_then(|device| { + vsock::Forwarder::new(&core, device, Box::new(move || platform.now())) + .inspect_err(|error| log::warn!("no vsock forwarder: {error}")) + .ok() + .map(Arc::new) + }); Self { core, platform: config.platform, @@ -200,6 +221,7 @@ impl Broker { timers, sockets, init, + vsock, } } @@ -294,17 +316,24 @@ impl Broker { } } - /// Looks at what changes outside requests: standard input, timers, and - /// the network. + /// Looks at what changes outside requests: standard input, timers, the + /// network, and the socket device. fn poll_sources(&self) { self.stdio.poll(); self.timers.fire(); self.sockets.poll(); + if let Some(vsock) = &self.vsock { + vsock.poll(); + } } - /// When the network stack must next run, if ever. - fn next_network_deadline(&self) -> Option { - self.sockets.next_deadline().map(timers::instant) + /// When the network stack, or the vsock forwarder, must next run, if + /// ever. + fn next_sources_deadline(&self) -> Option { + timers::earliest( + self.sockets.next_deadline().map(timers::instant), + self.vsock.as_ref().and_then(|vsock| vsock.next_deadline()), + ) } } @@ -457,15 +486,15 @@ fn frame(bytes: &[u8]) -> Result { impl Association { /// When, on the kernel clock, [`Self::poll_events`] next has something /// to do: the earliest armed broker timer's deadline, or the network - /// stack's (e.g., to retransmit). + /// stack's (e.g., to retransmit), or the vsock forwarder's. pub fn next_timer_deadline(&self) -> Option { timers::earliest( self.broker.timers.next_deadline(), - self.broker.next_network_deadline(), + self.broker.next_sources_deadline(), ) } - /// Looks at the devices (standard input, the network) and fires the + /// Looks at the devices (standard input, the network, vsock) and fires the /// broker timers that are due, publishing readiness, as /// [`RingFutex::Wait`] would; for a process that is running instead /// (e.g., an interrupt stopped it). A notification ring this corrupts @@ -789,14 +818,14 @@ impl Association { if wait.deadline.is_none() && !wait.notifications && !self.submitted.get() { return Err(EnterError::Status(Status::Stalled)); } - // Only notifications see timers; the network stack runs at its - // deadlines, whatever the wait. + // Only notifications see timers; the network stack and the vsock + // forwarder run at their deadlines, whatever the wait. let wake = if wait.notifications { timers::earliest(wait.deadline, self.broker.timers.next_deadline()) } else { wait.deadline }; - let wake = timers::earliest(wake, self.broker.next_network_deadline()); + let wake = timers::earliest(wake, self.broker.next_sources_deadline()); Ok(WaitState::Blocked { wake }) }) .unwrap_or_else(|error| WaitState::Ended(Err(error))) diff --git a/litebox_broker_vm_kernel/src/vsock.rs b/litebox_broker_vm_kernel/src/vsock.rs new file mode 100644 index 0000000000..ee3a63ea4b --- /dev/null +++ b/litebox_broker_vm_kernel/src/vsock.rs @@ -0,0 +1,1537 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT license. + +//! The vsock forwarder: the host uses `AF_VSOCK`, guests Unix sockets. A host +//! connection to guest port `P`, from the VM's socket device ([`Vsock`], +//! [`crate::Config::vsock`]), becomes a broker Unix stream socket connected, +//! without waiting, to the guest listener on the abstract name +//! `litebox-vsock:

` (`P` in decimal, e.g., `@litebox-vsock:1024` for the +//! Kata agent's `unix://@litebox-vsock:1024`); the request is refused if that +//! fails (nothing listens there, or its backlog is full). The guest sees an +//! unnamed peer. The forwarder adds no broker operation and does not change +//! the broker core: it is a user of the broker core's Unix sockets. +//! +//! Ends map as on Linux: +//! +//! - The host's `SHUTDOWN` of sending: the Unix socket's write side is shut +//! down, after the host's data. +//! - The guest's end of data: a `SHUTDOWN` of sending, after the guest's data. +//! - The guest's socket receiving no more (closed, or shut down receiving): a +//! `SHUTDOWN` of receiving, so the host's sends fail, and what it sent +//! meanwhile is dropped; the guest's data still goes to the host. +//! - The host's `SHUTDOWN` of receiving: the guest's sends fail. +//! - The host's reset: the Unix socket is closed, after the host's data. +//! +//! A connection ended both ways is closed on the device, which keeps it until +//! the host resets it. From when the guest's socket can neither send nor +//! receive, the connection has [`CLOSE_TIMEOUT`] to deliver the guest's data +//! and be reset by the host; then it is forgotten (reset), whatever is left. +//! The broker core does not tell the guest's close from its socket being shut +//! down both ways, so the timeout also starts when the guest shuts down both +//! ways, or shuts down sending after the host did, while it keeps its socket +//! open: then, unlike on Linux, a host that takes none of the guest's data in +//! time loses what is left of it. +//! +//! The host and guests can make more work than a poll should do: each poll +//! answers at most `MAX_REQUESTS_PER_POLL` requests and moves at most +//! `MAX_BYTES_PER_FLOW` bytes per direction of a connection and +//! `MAX_BYTES_PER_POLL` in all; the next poll is then due `BUSY_INTERVAL` +//! later. A send that fails for want of broker memory is retried after +//! `POLL_INTERVAL`, as freeing memory does not notify. +//! +//! The forwarder's process has no runner, parent or children, and never +//! exits. It exists only with the virtual init, and is created right after +//! it, so it is process 2 and does not take the broker's terminal, which the +//! first root process does; guests can signal it, to no effect, as the +//! virtual init. Guest port names are not reserved: a guest can squat one; +//! the VM, not a guest, is the isolation boundary. +//! +//! Locks: the forwarder's, then the device's (each [`Vsock`] call) or the +//! broker core's, never both. A CPU that finds another polling does not wait +//! for it. The forwarder's sockets' readiness only notifies every CPU +//! ([`cpu::notify_all`]), without taking the forwarder's lock. + +use alloc::boxed::Box; +use alloc::collections::BTreeMap; +use alloc::sync::Arc; +use alloc::vec::Vec; +use core::ops::Bound; +use core::sync::atomic::{AtomicBool, Ordering}; +use core::time::Duration; +use litebox_broker_core::readiness::ReadinessSink; +use litebox_broker_core::unix_socket::{self, Received}; +use litebox_broker_core::{BrokerCore, BrokerError, BrokerProcess, CallerCredential}; +use litebox_broker_protocol::ObjectHandle; +use litebox_broker_protocol::fs::{FileOpenFlags, FileUser}; +use litebox_broker_protocol::readiness::ReadinessFlags; +use litebox_broker_protocol::socket::{ShutdownMode, SocketType}; +use litebox_broker_protocol::unix_socket::{MAX_UNIX_SOCKET_TRANSFER_SIZE, UnixSocketAddress}; +use litebox_hal::virtio::vsock::ConnectionId; +use litebox_platform::time::Instant as _; +use litebox_platform_vm_kernel::{Instant, cpu}; + +/// The VM's socket device, as the forwarder drives it (see +/// [`VirtioVsock`](litebox_hal::virtio::vsock::VirtioVsock)). Only +/// [`Self::take_request`], [`Self::read`], [`Self::write`], +/// [`Self::is_closed`] and [`Self::forget`] first take what the device +/// returned; the others take nothing new, so a request just taken is still +/// one when it is accepted. +pub trait Vsock: Sync { + /// The next connection the host requested, to accept or refuse. + fn take_request(&self) -> Option; + + /// Accepts a requested connection; false if it is not one. + fn accept(&self, id: ConnectionId) -> bool; + + /// Refuses a requested connection. + fn refuse(&self, id: ConnectionId); + + /// Copies received data into `output`: how much, or `None` at the end of + /// the stream. + fn read(&self, id: ConnectionId, output: &mut [u8]) -> Option; + + /// Queues as much of `bytes` as the host takes now: how much, or `None` + /// if the connection takes no more. + fn write(&self, id: ConnectionId, bytes: &[u8]) -> Option; + + /// Shuts down sending, after what was written. + fn shutdown(&self, id: ConnectionId); + + /// Shuts down receiving: the host's sends fail. Data already received + /// can still be read; data the host sent before it knew is dropped. + fn shutdown_receiving(&self, id: ConnectionId); + + /// Shuts the connection down both ways, after what was written; it is + /// closed once the host resets it. + fn close(&self, id: ConnectionId); + + /// Whether the connection is gone and its data read, or unknown. + fn is_closed(&self, id: ConnectionId) -> bool; + + /// Ends the connection, resetting it unless it is gone, and frees its + /// slot. + fn forget(&self, id: ConnectionId); + + /// Takes whether the calls since the last of these took something from + /// the device, or stopped at a budget with more left, which it does not + /// wake the broker for again (e.g., data for a connection whose turn in + /// the poll is over). + fn take_backlog(&self) -> bool; + + /// Whether the device wakes the broker as it changes; else the forwarder + /// polls it. + fn interrupts(&self) -> bool; +} + +/// How long a connection closed both ways waits for the host's reset before +/// it is forgotten (reset) anyway, as Linux's `VSOCK_CLOSE_TIMEOUT`. +pub const CLOSE_TIMEOUT: Duration = Duration::from_secs(8); + +/// How often a device that does not interrupt is polled, and a send the +/// broker had no memory for retried (freeing memory does not notify). +const POLL_INTERVAL: Duration = Duration::from_millis(10); + +/// How soon a poll that left work follows the last: soon, but not at once, +/// as a deadline that is due interrupts every running process. +const BUSY_INTERVAL: Duration = Duration::from_millis(1); + +/// Bytes buffered per direction of a connection, between the device and the +/// Unix socket. +const CHUNK: usize = 16 * 1024; + +// So a Unix socket transfer of a chunk is not refused. +const _: () = assert!(CHUNK <= MAX_UNIX_SOCKET_TRANSFER_SIZE as usize); + +/// Bytes moved (or dropped) per direction of a connection per poll. +const MAX_BYTES_PER_FLOW: usize = 64 * 1024; + +/// Bytes moved (or dropped) per poll, in all; the rest waits for the next +/// poll, which starts with the connections this one did not get to. +const MAX_BYTES_PER_POLL: usize = 256 * 1024; + +/// Host requests answered per poll; the rest wait for the next poll. +const MAX_REQUESTS_PER_POLL: usize = 8; + +/// The forwarder's Unix sockets act as this user, which abstract names do +/// not check. +const USER: FileUser = FileUser::ROOT; + +/// Bytes on their way from one end to the other: `bytes[start..end]`. The +/// buffer is allocated (and zeroed) once, on the first refill, as polls run on +/// every broker event. +struct Pending { + bytes: Vec, + start: usize, + end: usize, +} + +impl Pending { + fn new() -> Self { + Self { + bytes: Vec::new(), + start: 0, + end: 0, + } + } + + fn is_empty(&self) -> bool { + self.start == self.end + } + + fn rest(&self) -> &[u8] { + &self.bytes[self.start..self.end] + } + + fn consume(&mut self, len: usize) { + self.start += len; + if self.is_empty() { + (self.start, self.end) = (0, 0); + } + } + + /// Refills the empty buffer with up to `max` (at most [`CHUNK`]) bytes + /// from `fill`, which returns how many it wrote; `None`, without calling + /// it, if the buffer cannot be allocated (the host's traffic must not + /// abort the kernel). + fn refill( + &mut self, + max: usize, + fill: impl FnOnce(&mut [u8]) -> T, + len: impl FnOnce(&T) -> usize, + ) -> Option { + debug_assert!(self.is_empty()); + if self.bytes.len() < CHUNK { + self.bytes + .try_reserve_exact(CHUNK - self.bytes.len()) + .ok()?; + self.bytes.resize(CHUNK, 0); + } + let max = max.min(CHUNK); + let result = fill(&mut self.bytes[..max]); + (self.start, self.end) = (0, len(&result).min(max)); + Some(result) + } +} + +/// A direction of a connection. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum Flow { + Open, + /// Its source sends no more; what it sent is still to be delivered. + Ending, + /// What it sent is delivered, and its end passed on, or there is no one + /// to deliver to (what its source still sends is dropped). + Ended, +} + +struct Connection { + socket: ObjectHandle, + /// From the host, not yet sent to the Unix socket. + to_guest: Pending, + /// Host to guest: its end shuts down the Unix socket's write side. + inbound: Flow, + /// From the Unix socket, not yet written to the device. + to_host: Pending, + /// Guest to host: its end shuts down the device connection's sending + /// side. + outbound: Flow, + /// When the connection is forgotten, whatever is left: [`CLOSE_TIMEOUT`] + /// after the guest's socket could neither send nor receive, or this end + /// closed the Unix socket. + deadline: Option, + /// Closed both ways on the device, waiting for the host's reset until + /// the deadline; the Unix socket is closed. + closing: bool, +} + +impl Connection { + fn is_due(&self, now: Instant) -> bool { + self.deadline.is_some_and(|deadline| now >= deadline) + } +} + +/// Whether a connection goes on after a step. +enum Step { + Continue, + /// Over on both ends, the Unix socket closed: forget the device + /// connection. + Release, +} + +/// What a send to the guest did. +#[derive(Debug, PartialEq, Eq)] +enum Sent { + /// It sent this many bytes (zero only of none). + Bytes(usize), + /// Nothing until the guest makes room, which notifies. + Full, + /// Nothing for want of broker memory (e.g., the guest's quota of queued + /// bytes), which frees without notice. + Stalled, + /// The guest's socket receives no more: it is closed, or shut down + /// receiving. + Gone, +} + +/// A poll's budget, and what it left. +struct Pass { + now: Instant, + /// Bytes it may still move (or drop), in all. + bytes: usize, + /// It stopped at a budget with more to do. + busy: bool, + /// Work waits for memory, which frees without notice. + stalled: bool, +} + +impl Pass { + /// What a direction of a connection may move. + fn flow_limit(&self) -> usize { + self.bytes.min(MAX_BYTES_PER_FLOW) + } + + /// Counts `moved` bytes against the budget; `limit` was the direction's. + fn moved(&mut self, moved: usize, limit: usize) { + self.bytes = self.bytes.saturating_sub(moved); + if moved >= limit { + self.busy = true; + } + } +} + +struct State { + connections: BTreeMap, + /// The last poll stopped at a budget with more to do, or left the + /// device with a backlog. + busy: bool, + /// Work waits for memory, which frees without notice. + stalled: bool, + /// The connection the last poll's byte budget stopped after: the next + /// starts after it, so each connection gets its turn. + stopped_after: Option, + last_poll: Option, +} + +/// See the [module](self) docs. +pub(crate) struct Forwarder { + process: Arc, + sink: Arc, + device: &'static dyn Vsock, + /// The kernel clock. + now: Box Instant + Send + Sync>, + state: spin::Mutex, + /// A poll was asked for, and not yet started: one that found another CPU + /// polling leaves it to that one, which polls again, or else to + /// [`Self::next_deadline`], which reports a poll due. + repoll: AtomicBool, +} + +impl Forwarder { + /// The forwarder for `device`, with its broker process, on the clock + /// `now`. + /// + /// # Errors + /// + /// The process cannot be created. + pub(crate) fn new( + core: &BrokerCore, + device: &'static dyn Vsock, + now: Box Instant + Send + Sync>, + ) -> Result { + let process = core.create_process(CallerCredential::HostGuaranteed, None)?; + process.complete_start()?; + Ok(Self { + process, + sink: Arc::new(Sink), + device, + now, + state: spin::Mutex::new(State { + connections: BTreeMap::new(), + busy: false, + stalled: false, + stopped_after: None, + last_poll: None, + }), + repoll: AtomicBool::new(false), + }) + } + + /// Takes the host's requests, and moves what each connection can, a + /// bounded amount. Never waits for another CPU's poll. + pub(crate) fn poll(&self) { + self.repoll.store(true, Ordering::SeqCst); + // Once more for a CPU that asked meanwhile; then, the deadline. + for _ in 0..2 { + let Some(mut state) = self.state.try_lock() else { + return; + }; + if !self.repoll.swap(false, Ordering::SeqCst) { + return; + } + self.poll_locked(&mut state); + drop(state); + if !self.repoll.load(Ordering::SeqCst) { + return; + } + } + } + + /// When the forwarder must next poll, if before the device changes: soon + /// if a poll is asked for or left work, at the earliest close timeout, + /// or, for a device that does not interrupt (or work waiting for memory), + /// its poll interval. + pub(crate) fn next_deadline(&self) -> Option { + let soon = || Some(after((self.now)(), BUSY_INTERVAL)); + if self.repoll.load(Ordering::SeqCst) { + return soon(); + } + // Another CPU polls, and then has its own deadline. + let Some(state) = self.state.try_lock() else { + return soon(); + }; + let last_poll = state.last_poll?; + if state.busy { + return Some(after(last_poll, BUSY_INTERVAL)); + } + let poll = + (state.stalled || !self.device.interrupts()).then(|| after(last_poll, POLL_INTERVAL)); + state + .connections + .values() + .filter_map(|connection| connection.deadline) + .chain(poll) + .min() + } + + fn poll_locked(&self, state: &mut State) { + let mut pass = Pass { + now: (self.now)(), + bytes: MAX_BYTES_PER_POLL, + busy: false, + stalled: false, + }; + state.last_poll = Some(pass.now); + let mut requests = 0; + while requests < MAX_REQUESTS_PER_POLL + && let Some(id) = self.device.take_request() + { + requests += 1; + match self.open(id) { + // The device takes nothing new before `accept`, so this is + // still a request, and the guest never gets a connection + // the host gave up on. + Some(socket) if self.device.accept(id) => { + state.connections.insert( + id, + Connection { + socket, + to_guest: Pending::new(), + inbound: Flow::Open, + to_host: Pending::new(), + outbound: Flow::Open, + deadline: None, + closing: false, + }, + ); + } + Some(socket) => self.close_socket(socket), + None => self.device.refuse(id), + } + } + // More may wait. + pass.busy |= requests == MAX_REQUESTS_PER_POLL; + // From after where the last poll stopped, around once; the table only + // shrinks after the loop. + let start = state.stopped_after.take(); + let mut cursor = start; + let mut wrapped = start.is_none(); + let mut released = Vec::new(); + loop { + let next = match cursor { + Some(after) => state + .connections + .range((Bound::Excluded(after), Bound::Unbounded)) + .next(), + None => state.connections.iter().next(), + } + .map(|(&id, _)| id); + let id = match next { + Some(id) if !(wrapped && start.is_some_and(|start| id > start)) => id, + None if !wrapped => { + (cursor, wrapped) = (None, true); + continue; + } + _ => break, + }; + cursor = Some(id); + if pass.bytes == 0 { + pass.busy = true; + break; + } + let connection = state + .connections + .get_mut(&id) + .expect("connections are removed only below"); + if let Step::Release = self.step(id, connection, &mut pass) { + released.push(id); + } + if pass.bytes == 0 { + state.stopped_after = Some(id); + } + } + for id in released { + self.device.forget(id); + state.connections.remove(&id); + } + state.busy = pass.busy || self.device.take_backlog(); + state.stalled = pass.stalled; + } + + /// A Unix socket connected to the listener on `id`'s guest port; `None` + /// if it cannot be. + fn open(&self, id: ConnectionId) -> Option { + let socket = match unix_socket::create( + &self.process, + SocketType::Stream, + FileOpenFlags::NONBLOCKING, + &self.sink, + ) { + Ok(socket) => socket, + Err(error) => { + log::debug!("vsock: refusing {id:?}: no socket: {error:?}"); + return None; + } + }; + let guest = UnixSocketAddress::Abstract(guest_name(id.guest_port)); + match unix_socket::connect(&self.process, socket, &guest, USER) { + Ok(Ok(())) => Some(socket), + result => { + log::debug!("vsock: refusing {id:?}: {result:?}"); + self.close_socket(socket); + None + } + } + } + + fn close_socket(&self, socket: ObjectHandle) { + if let Err(error) = self.process.close_object_reference(socket) { + log::debug!("vsock: closing a socket failed: {error:?}"); + } + } + + /// Moves what `connection` can, both ways, and ends what is over. + fn step(&self, id: ConnectionId, connection: &mut Connection, pass: &mut Pass) -> Step { + if connection.closing { + // What the host sends after the close resets it, and goes + // nowhere. + let limit = pass.flow_limit(); + let dropped = self.discard(id, limit); + pass.moved(dropped, limit); + return if self.device.is_closed(id) || connection.is_due(pass.now) { + Step::Release + } else { + Step::Continue + }; + } + // The guest's socket can neither send nor receive (see the module + // docs), whether or not its data is delivered. + if connection.deadline.is_none() + && self + .process + .check_readiness(connection.socket) + .is_ok_and(|readiness| readiness.contains(ReadinessFlags::CLOSED)) + { + connection.deadline = Some(after(pass.now, CLOSE_TIMEOUT)); + } + if connection.is_due(pass.now) { + // What is left of the guest's is dropped. + self.close_socket(connection.socket); + return Step::Release; + } + self.to_guest(id, connection, pass); + self.to_host(id, connection, pass); + if connection.to_guest.is_empty() && self.device.is_closed(id) { + // The host reset it (or shut it down both ways), and what it + // sent is delivered. + self.close_socket(connection.socket); + return Step::Release; + } + if connection.outbound == Flow::Ending && connection.to_host.is_empty() { + self.device.shutdown(id); + connection.outbound = Flow::Ended; + } + // Probes whether the guest's socket receives no more; not once this + // end shut down its write side, which fails sends too. + if connection.inbound == Flow::Open + && connection.to_guest.is_empty() + && self.send(connection.socket, &[]) == Sent::Gone + { + self.receives_no_more(id, connection); + } + if connection.outbound == Flow::Ended && connection.inbound == Flow::Ended { + self.device.close(id); + self.close_socket(connection.socket); + connection.closing = true; + connection + .deadline + .get_or_insert(after(pass.now, CLOSE_TIMEOUT)); + } + Step::Continue + } + + /// The guest's socket receives no more (it is closed, or shut down + /// receiving): the host is told, so its sends fail, as on Linux, and + /// what it sent meanwhile is dropped. + fn receives_no_more(&self, id: ConnectionId, connection: &mut Connection) { + connection.to_guest = Pending::new(); + connection.inbound = Flow::Ended; + self.device.shutdown_receiving(id); + } + + /// Moves the host's data to the Unix socket, and its end. Once the + /// guest's socket receives no more, see [`Self::receives_no_more`]; the + /// guest's data still goes to the host. + fn to_guest(&self, id: ConnectionId, connection: &mut Connection, pass: &mut Pass) { + let limit = pass.flow_limit(); + let mut moved = 0; + loop { + if !connection.to_guest.is_empty() { + if moved >= limit { + break; + } + let rest = connection.to_guest.rest(); + match self.send(connection.socket, &rest[..rest.len().min(limit - moved)]) { + Sent::Bytes(sent) => { + connection.to_guest.consume(sent); + moved += sent; + if !connection.to_guest.is_empty() { + continue; + } + } + Sent::Full => break, + Sent::Stalled => { + pass.stalled = true; + break; + } + Sent::Gone => self.receives_no_more(id, connection), + } + } + match connection.inbound { + Flow::Open => {} + Flow::Ending => { + self.shut_down_socket(connection.socket, ShutdownMode::Write); + connection.inbound = Flow::Ended; + break; + } + Flow::Ended => { + moved += self.discard(id, limit - moved); + break; + } + } + if moved >= limit { + break; + } + match connection.to_guest.refill( + limit - moved, + |buffer| self.device.read(id, buffer), + |read| read.unwrap_or(0), + ) { + None => { + pass.stalled = true; + break; + } + Some(Some(0)) => break, + Some(Some(_)) => {} + Some(None) => connection.inbound = Flow::Ending, + } + } + pass.moved(moved, limit); + } + + /// Moves the guest's data to the device, and notes its end. + fn to_host(&self, id: ConnectionId, connection: &mut Connection, pass: &mut Pass) { + let limit = pass.flow_limit(); + let mut moved = 0; + loop { + if !connection.to_host.is_empty() { + if moved >= limit { + break; + } + let rest = connection.to_host.rest(); + let Some(written) = self + .device + .write(id, &rest[..rest.len().min(limit - moved)]) + else { + self.host_receives_no_more(connection); + break; + }; + connection.to_host.consume(written); + moved += written; + if !connection.to_host.is_empty() { + // Until the host gives credit, or the device returns + // buffers, which interrupts; or the next poll. + break; + } + } + if connection.outbound != Flow::Open || moved >= limit { + break; + } + // So the guest's sends fail as soon as the host's `SHUTDOWN` of + // receiving arrives, not after its next send. + if self.device.write(id, &[]).is_none() { + self.host_receives_no_more(connection); + break; + } + let received = connection.to_host.refill( + limit - moved, + |mut buffer| { + unix_socket::receive( + &self.process, + connection.socket, + &mut buffer, + false, + false, + true, + ) + }, + |received| match received { + Ok(Ok(Received { received, .. })) => *received, + _ => 0, + }, + ); + match received { + None => { + pass.stalled = true; + break; + } + Some(Ok(Ok(Received { received: 0, .. }))) => connection.outbound = Flow::Ending, + Some(Ok(Ok(_))) => {} + // Until the guest sends, which notifies. + Some(Err(BrokerError::NonBlockingWouldBlock | BrokerError::WouldBlock)) => break, + Some(Ok(Err(_)) | Err(_)) => connection.outbound = Flow::Ending, + } + } + pass.moved(moved, limit); + } + + /// The host receives no more (or the connection is gone): the guest's + /// sends fail from now on, what it sent is dropped, and no `SHUTDOWN` + /// follows. + fn host_receives_no_more(&self, connection: &mut Connection) { + connection.to_host = Pending::new(); + self.shut_down_socket(connection.socket, ShutdownMode::Read); + connection.outbound = Flow::Ended; + } + + fn shut_down_socket(&self, socket: ObjectHandle, mode: ShutdownMode) { + match unix_socket::shutdown(&self.process, socket, mode) { + Ok(Ok(())) => {} + Ok(Err(error)) => log::debug!("vsock: shutting down a socket failed: {error:?}"), + Err(error) => log::debug!("vsock: shutting down a socket failed: {error:?}"), + } + } + + /// Reads and drops up to `limit` bytes the host sent on `id`: how many. + fn discard(&self, id: ConnectionId, limit: usize) -> usize { + let mut buffer = [0; 4096]; + let mut dropped = 0; + while dropped < limit { + let len = buffer.len().min(limit - dropped); + match self.device.read(id, &mut buffer[..len]) { + Some(read) if read != 0 => dropped += read, + _ => break, + } + } + dropped + } + + /// Sends from the start of `bytes` to the guest. + fn send(&self, socket: ObjectHandle, bytes: &[u8]) -> Sent { + match unix_socket::send(&self.process, socket, None, &bytes, USER) { + Ok(Ok(sent)) => Sent::Bytes(sent), + Err(BrokerError::NonBlockingWouldBlock | BrokerError::WouldBlock) => Sent::Full, + Err(BrokerError::ResourceExhausted | BrokerError::OutOfMemory) => Sent::Stalled, + Ok(Err(_)) | Err(_) => Sent::Gone, + } + } +} + +/// The abstract name (without its leading NUL) guests listen on for +/// connections to vsock port `port`. +fn guest_name(port: u32) -> Vec { + alloc::format!("litebox-vsock:{port}").into_bytes() +} + +/// `duration` after `instant`, or the end of time. +fn after(instant: Instant, duration: Duration) -> Instant { + instant + .checked_add(duration) + .unwrap_or(Instant::from_nanos(u64::MAX)) +} + +/// The forwarder's sockets' readiness: a change (e.g., the guest sent data, +/// or made room) has every CPU poll again, which polls the forwarder. +struct Sink; + +impl ReadinessSink for Sink { + fn max_tracked_objects(&self) -> usize { + usize::MAX + } + + fn publish( + &self, + _handle: ObjectHandle, + _readiness: ReadinessFlags, + ) -> litebox_broker_core::Result<()> { + cpu::notify_all(); + Ok(()) + } + + fn republish( + &self, + _handle: ObjectHandle, + _readiness: ReadinessFlags, + ) -> litebox_broker_core::Result<()> { + cpu::notify_all(); + Ok(()) + } + + fn retire(&self, _handle: ObjectHandle) {} +} + +#[cfg(test)] +mod tests { + use super::*; + use alloc::collections::VecDeque; + use alloc::vec; + use core::sync::atomic::AtomicU64; + use litebox_broker_core::test_support::TestBrokerCoreBuilder; + use litebox_broker_core::{BrokerCoreLimits, ObjectRights, PolicyEngine}; + use litebox_broker_protocol::fs::FileMode; + use litebox_broker_protocol::unix_socket::{UNIX_SOCKET_BUFFER_SIZE, UnixSocketError}; + + const PORT: u32 = 1024; + + const fn id(guest_port: u32, host_port: u32) -> ConnectionId { + ConnectionId { + guest_port, + host_port, + } + } + + /// A connection to [`PORT`]. + const fn on(host_port: u32) -> ConnectionId { + id(PORT, host_port) + } + + /// The host's end of a connection, and the device's state of it. + #[derive(Default)] + #[expect( + clippy::struct_excessive_bools, + reason = "independent ends of a connection, set by the tests" + )] + struct HostConnection { + live: bool, + /// Reset by either end. + gone: bool, + /// Sent by the host, not yet read. + sent: VecDeque, + /// The host sends no more. + sent_all: bool, + /// The host receives no more. + receives_no_more: bool, + received: Vec, + /// How much more the host takes. + credit: usize, + /// The forwarder shut down sending. + shut_down: bool, + /// The forwarder shut down receiving. + shut_down_receiving: bool, + /// The forwarder closed it. + closed: bool, + } + + /// What the device returns on a connection. + enum Returned { + Data(Vec), + Credit(usize), + } + + #[derive(Default)] + struct HostState { + requests: VecDeque, + /// Requests the device takes only on the next read, of any + /// connection, as the driver may. + late_requests: VecDeque, + /// What the device takes only as the forwarder checks whether a + /// connection is closed, after its turn in a poll, as the driver may. + returned_after_turn: Vec<(ConnectionId, Returned)>, + connections: BTreeMap, + refused: Vec, + /// Forgotten, and whether they were reset then (not gone). + forgotten: Vec<(ConnectionId, bool)>, + /// The device took something since the last `take_backlog`, as the + /// driver latches it: data read or written, or a control packet sent. + took: bool, + } + + /// The host, behind a device that takes what it is given at once, but + /// for what is late. + #[derive(Default)] + struct Host(spin::Mutex); + + impl Host { + fn connect(&self, id: ConnectionId) { + self.0.lock().requests.push_back(id); + self.add(id); + } + + /// As [`Self::connect`], with the request taken on the next read. + fn connect_late(&self, id: ConnectionId) { + self.0.lock().late_requests.push_back(id); + self.add(id); + } + + fn add(&self, id: ConnectionId) { + self.0.lock().connections.insert( + id, + HostConnection { + credit: usize::MAX, + ..HostConnection::default() + }, + ); + } + + /// The device's transport reset. + fn reset_all(&self) { + let mut state = self.0.lock(); + let requests: Vec<_> = state.requests.drain(..).collect(); + for id in requests { + state.connections.remove(&id); + } + for connection in state.connections.values_mut() { + connection.gone = true; + } + } + + fn with(&self, id: ConnectionId, f: impl FnOnce(&mut HostConnection) -> T) -> T { + f(self.0.lock().connections.get_mut(&id).unwrap()) + } + + /// As [`Self::with`], for a control packet the forwarder sends. + fn control(&self, id: ConnectionId, f: impl FnOnce(&mut HostConnection) -> T) -> T { + let mut state = self.0.lock(); + state.took = true; + f(state.connections.get_mut(&id).unwrap()) + } + + fn send(&self, id: ConnectionId, bytes: &[u8]) { + self.with(id, |connection| connection.sent.extend(bytes)); + } + + fn take_received(&self, id: ConnectionId) -> Vec { + self.with(id, |connection| core::mem::take(&mut connection.received)) + } + + /// The device takes `returned` on `id` after its turn in the next + /// poll. + fn return_after_turn(&self, id: ConnectionId, returned: Returned) { + self.0.lock().returned_after_turn.push((id, returned)); + } + } + + impl Vsock for Host { + fn take_request(&self) -> Option { + self.0.lock().requests.pop_front() + } + + fn accept(&self, id: ConnectionId) -> bool { + self.control(id, |connection| { + connection.live = !connection.gone; + connection.live + }) + } + + fn refuse(&self, id: ConnectionId) { + let mut state = self.0.lock(); + state.took = true; + state.connections.remove(&id); + state.refused.push(id); + } + + fn read(&self, id: ConnectionId, output: &mut [u8]) -> Option { + let mut state = self.0.lock(); + let state = &mut *state; + state.took |= !state.late_requests.is_empty(); + state.requests.extend(state.late_requests.drain(..)); + let connection = state.connections.get_mut(&id)?; + if connection.sent.is_empty() { + let ended = connection.gone + || connection.sent_all + || connection.closed + || connection.shut_down_receiving; + return (!ended).then_some(0); + } + let len = output.len().min(connection.sent.len()); + for (byte, sent) in output.iter_mut().zip(connection.sent.drain(..len)) { + *byte = sent; + } + state.took |= len != 0; + Some(len) + } + + fn write(&self, id: ConnectionId, bytes: &[u8]) -> Option { + let mut state = self.0.lock(); + let state = &mut *state; + let connection = state.connections.get_mut(&id)?; + if !connection.live + || connection.gone + || connection.receives_no_more + || connection.shut_down + { + return None; + } + let len = bytes.len().min(connection.credit); + connection.received.extend_from_slice(&bytes[..len]); + connection.credit -= len; + state.took |= len != 0; + Some(len) + } + + fn shutdown(&self, id: ConnectionId) { + self.control(id, |connection| connection.shut_down = true); + } + + fn shutdown_receiving(&self, id: ConnectionId) { + self.control(id, |connection| connection.shut_down_receiving = true); + } + + fn close(&self, id: ConnectionId) { + self.control(id, |connection| { + connection.shut_down = true; + connection.closed = true; + }); + } + + fn is_closed(&self, id: ConnectionId) -> bool { + let mut state = self.0.lock(); + let state = &mut *state; + for (returned_on, returned) in state.returned_after_turn.drain(..) { + state.took = true; + let Some(connection) = state.connections.get_mut(&returned_on) else { + continue; + }; + match returned { + Returned::Data(bytes) => connection.sent.extend(bytes), + Returned::Credit(credit) => connection.credit = credit, + } + } + state + .connections + .get(&id) + .is_none_or(|connection| connection.gone && connection.sent.is_empty()) + } + + fn forget(&self, id: ConnectionId) { + let mut state = self.0.lock(); + state.took = true; + let gone = state + .connections + .remove(&id) + .is_none_or(|connection| connection.gone); + state.forgotten.push((id, !gone)); + } + + fn take_backlog(&self) -> bool { + core::mem::take(&mut self.0.lock().took) + } + + fn interrupts(&self) -> bool { + true + } + } + + fn guest_socket(guest: &BrokerProcess) -> ObjectHandle { + let sink: Arc = Arc::new(Sink); + unix_socket::create(guest, SocketType::Stream, FileOpenFlags::NONBLOCKING, &sink).unwrap() + } + + /// A listener on guest port `port`. + fn listen(guest: &BrokerProcess, port: u32, backlog: u32) -> ObjectHandle { + let listener = guest_socket(guest); + let name = UnixSocketAddress::Abstract(guest_name(port)); + unix_socket::bind(guest, listener, &name, USER, FileMode::empty()) + .unwrap() + .unwrap(); + unix_socket::listen(guest, listener, backlog) + .unwrap() + .unwrap(); + listener + } + + /// A forwarder, the host behind its device, and a guest process with a + /// listener on [`PORT`]. + struct Fixture { + forwarder: Forwarder, + host: &'static Host, + guest: Arc, + listener: ObjectHandle, + clock: Arc, + } + + impl Fixture { + fn new() -> Self { + Self::with_limits(BrokerCoreLimits::DEFAULT) + } + + fn with_limits(limits: BrokerCoreLimits) -> Self { + // As the VM broker's. + let core = TestBrokerCoreBuilder::new(PolicyEngine::with_host_guaranteed_rights( + ObjectRights::all(), + )) + .with_limits(limits) + .build() + .unwrap(); + let guest = core + .create_process(CallerCredential::HostGuaranteed, None) + .unwrap(); + let host: &'static Host = Box::leak(Box::default()); + let clock = Arc::new(AtomicU64::new(1)); + let now = Arc::clone(&clock); + let forwarder = Forwarder::new( + &core, + host, + Box::new(move || Instant::from_nanos(now.load(Ordering::SeqCst))), + ) + .unwrap(); + let listener = listen(&guest, PORT, 64); + Self { + forwarder, + host, + guest, + listener, + clock, + } + } + + fn now(&self) -> Instant { + Instant::from_nanos(self.clock.load(Ordering::SeqCst)) + } + + fn advance(&self, duration: Duration) { + let nanos = u64::try_from(duration.as_nanos()).unwrap(); + self.clock.fetch_add(nanos, Ordering::SeqCst); + } + + fn poll(&self) { + self.forwarder.poll(); + } + + fn assert_due_soon(&self) { + assert_eq!( + self.forwarder.next_deadline(), + Some(after(self.now(), BUSY_INTERVAL)) + ); + } + + /// The poll due soon after one the device made progress in, which + /// then finds nothing new. + fn trailing_poll(&self) { + self.assert_due_soon(); + self.poll(); + } + + /// The host connects from `host_port`, and the guest accepts it. + fn connect(&self, host_port: u32) -> ObjectHandle { + self.host.connect(on(host_port)); + self.poll(); + self.accept() + } + + fn accept(&self) -> ObjectHandle { + let sink: Arc = Arc::new(Sink); + unix_socket::accept( + &self.guest, + self.listener, + FileOpenFlags::NONBLOCKING, + &sink, + ) + .unwrap() + .unwrap() + } + + fn send(&self, socket: ObjectHandle, bytes: &[u8]) -> Result { + unix_socket::send(&self.guest, socket, None, &bytes, USER).unwrap() + } + + /// What the guest can receive now; `None` at the end of the stream. + fn receive(&self, socket: ObjectHandle) -> Option> { + let mut received = Vec::new(); + loop { + let mut buffer = vec![0; CHUNK]; + match unix_socket::receive(&self.guest, socket, &mut buffer, false, false, true) { + Ok(Ok(Received { received: 0, .. })) => { + return (!received.is_empty()).then_some(received); + } + Ok(Ok(Received { received: len, .. })) => { + received.extend_from_slice(&buffer[..len]); + } + Err(BrokerError::NonBlockingWouldBlock) => return Some(received), + other => panic!("receive: {other:?}"), + } + } + } + + fn close(&self, socket: ObjectHandle) { + self.guest.close_object_reference(socket).unwrap(); + } + + fn forgotten(&self) -> Vec<(ConnectionId, bool)> { + core::mem::take(&mut self.host.0.lock().forgotten) + } + } + + #[test] + fn requests_nothing_listens_for_are_refused() { + let f = Fixture::new(); + f.host.connect(id(PORT + 1, 5)); + f.host.connect(on(6)); + f.poll(); + assert_eq!(f.host.0.lock().refused, [id(PORT + 1, 5)]); + f.accept(); + assert!(f.host.with(on(6), |connection| connection.live)); + } + + #[test] + fn requests_beyond_a_full_backlog_are_refused() { + let f = Fixture::new(); + // Two connections wait at most. + listen(&f.guest, PORT + 1, 1); + for host_port in 0..4 { + f.host.connect(id(PORT + 1, host_port)); + } + f.poll(); + assert_eq!(f.host.0.lock().refused, [id(PORT + 1, 2), id(PORT + 1, 3)]); + for host_port in 0..2 { + assert!( + f.host + .with(id(PORT + 1, host_port), |connection| connection.live) + ); + } + } + + #[test] + fn data_and_ends_pass_both_ways() { + let f = Fixture::new(); + let socket = f.connect(7); + f.host.send(on(7), b"ping"); + f.host.with(on(7), |connection| connection.sent_all = true); + f.poll(); + assert_eq!(f.receive(socket), Some(b"ping".to_vec())); + assert_eq!(f.receive(socket), None); + // After the host's end, the guest still sends. + assert_eq!(f.send(socket, b"pong"), Ok(4)); + f.close(socket); + f.poll(); + assert_eq!(f.host.take_received(on(7)), b"pong"); + // Closed both ways, until the host resets it. + assert!(f.host.with(on(7), |connection| connection.closed)); + assert_eq!(f.forgotten(), []); + f.host.with(on(7), |connection| connection.gone = true); + f.poll(); + assert_eq!(f.forgotten(), [(on(7), false)]); + } + + #[test] + fn host_data_after_the_guest_closes_does_not_lose_the_guests() { + let f = Fixture::new(); + let socket = f.connect(7); + f.host.send(on(7), b"request"); + f.poll(); + assert_eq!(f.receive(socket), Some(b"request".to_vec())); + // The host takes the reply slowly. + f.host.with(on(7), |connection| connection.credit = 1000); + let reply: Vec = (0..100_000u32).map(|i| i.to_le_bytes()[0]).collect(); + assert_eq!(f.send(socket, &reply), Ok(reply.len())); + f.close(socket); + f.host.send(on(7), b"more"); + for _ in 0..200 { + f.poll(); + f.host.with(on(7), |connection| connection.credit = 1000); + } + // What the host sent is dropped, what the guest sent arrived, and + // the end is a close, not a reset. + assert_eq!(f.host.take_received(on(7)), reply); + f.host.with(on(7), |connection| { + assert!(connection.sent.is_empty()); + assert!(connection.shut_down_receiving); + assert!(connection.closed && !connection.gone); + }); + assert_eq!(f.forgotten(), []); + } + + #[test] + fn guests_shutting_down_receiving_still_send() { + let f = Fixture::new(); + let socket = f.connect(7); + unix_socket::shutdown(&f.guest, socket, ShutdownMode::Read) + .unwrap() + .unwrap(); + f.host.send(on(7), b"dropped"); + f.poll(); + assert_eq!(f.send(socket, b"still"), Ok(5)); + f.poll(); + assert_eq!(f.host.take_received(on(7)), b"still"); + f.host.with(on(7), |connection| { + assert!(connection.sent.is_empty()); + assert!(connection.shut_down_receiving && !connection.shut_down); + }); + assert_eq!(f.forgotten(), []); + f.trailing_poll(); + // Not a close: no deadline. + assert_eq!(f.forwarder.next_deadline(), None); + } + + #[test] + fn hosts_shutting_down_receiving_fail_guest_sends() { + let f = Fixture::new(); + let socket = f.connect(7); + f.host + .with(on(7), |connection| connection.receives_no_more = true); + // Before the guest sends anything. + f.poll(); + assert_eq!(f.send(socket, b"fails"), Err(UnixSocketError::BrokenPipe)); + // The host's data still arrives. + f.host.send(on(7), b"data"); + f.poll(); + assert_eq!(f.receive(socket), Some(b"data".to_vec())); + } + + /// The guest sends a reply the host gives no credit for, and closes its + /// socket; the close's timeout is due at the returned instant. + fn close_with_data_the_host_takes_no_credit_for() -> (Fixture, Instant) { + let f = Fixture::new(); + let socket = f.connect(7); + f.host.with(on(7), |connection| connection.credit = 0); + assert_eq!(f.send(socket, b"reply"), Ok(5)); + f.poll(); + f.close(socket); + f.poll(); + // The host is told before the data is delivered. + f.host.with(on(7), |connection| { + assert!(connection.shut_down_receiving && !connection.shut_down); + }); + f.trailing_poll(); + let timeout = after(f.now(), CLOSE_TIMEOUT); + assert_eq!(f.forwarder.next_deadline(), Some(timeout)); + f.advance(CLOSE_TIMEOUT.checked_sub(Duration::from_nanos(1)).unwrap()); + f.poll(); + assert_eq!(f.forgotten(), []); + (f, timeout) + } + + #[test] + fn closes_time_out_while_the_host_gives_no_credit() { + let (f, _) = close_with_data_the_host_takes_no_credit_for(); + assert_eq!(f.host.take_received(on(7)), b""); + f.advance(Duration::from_nanos(1)); + f.poll(); + assert_eq!(f.forgotten(), [(on(7), true)]); + f.trailing_poll(); + assert_eq!(f.forwarder.next_deadline(), None); + } + + #[test] + fn closes_deliver_everything_if_the_host_gives_credit_in_time() { + let (f, timeout) = close_with_data_the_host_takes_no_credit_for(); + f.host + .with(on(7), |connection| connection.credit = usize::MAX); + f.poll(); + assert_eq!(f.host.take_received(on(7)), b"reply"); + assert!(f.host.with(on(7), |connection| connection.closed)); + f.trailing_poll(); + // The close's time still bounds the wait for the host's reset. + assert_eq!(f.forwarder.next_deadline(), Some(timeout)); + f.advance(Duration::from_nanos(1)); + f.poll(); + assert_eq!(f.forgotten(), [(on(7), true)]); + } + + #[test] + fn sends_the_broker_has_no_memory_for_are_retried() { + let mut limits = BrokerCoreLimits::DEFAULT; + limits.max_unix_socket_bytes_per_process = 1000; + let f = Fixture::with_limits(limits); + let socket = f.connect(7); + f.host.send(on(7), &[1; 600]); + f.poll(); + f.host.send(on(7), &[2; 600]); + f.poll(); + f.trailing_poll(); + // Not the guest's end: retried later. + assert_eq!( + f.forwarder.next_deadline(), + Some(after(f.now(), POLL_INTERVAL)) + ); + assert_eq!(f.receive(socket), Some(vec![1; 600])); + f.poll(); + assert_eq!(f.receive(socket), Some(vec![2; 600])); + assert!( + f.host + .with(on(7), |connection| !connection.gone && !connection.closed) + ); + assert_eq!(f.forgotten(), []); + } + + #[test] + fn requests_are_answered_a_few_per_poll() { + let f = Fixture::new(); + for host_port in 0..20 { + f.host.connect(id(PORT + 1, host_port)); + } + f.poll(); + assert_eq!(f.host.0.lock().refused.len(), MAX_REQUESTS_PER_POLL); + f.assert_due_soon(); + f.poll(); + f.poll(); + assert_eq!(f.host.0.lock().refused.len(), 20); + f.poll(); + assert_eq!(f.forwarder.next_deadline(), None); + } + + #[test] + fn bytes_are_moved_a_bounded_amount_per_poll_each_connection_in_turn() { + let f = Fixture::new(); + let count = MAX_BYTES_PER_POLL / MAX_BYTES_PER_FLOW + 1; + let ports = 0..u32::try_from(count).unwrap(); + let sockets: Vec<_> = ports.clone().map(|port| f.connect(port)).collect(); + for port in ports { + f.host.send(on(port), &vec![1; 2 * MAX_BYTES_PER_FLOW]); + } + let received = |f: &Fixture| -> Vec { + sockets + .iter() + .map(|&socket| f.receive(socket).unwrap().len()) + .collect() + }; + f.poll(); + let mut expected = vec![MAX_BYTES_PER_FLOW; count]; + expected[count - 1] = 0; + assert_eq!(received(&f), expected); + f.assert_due_soon(); + // The next starts with the one left out. + f.poll(); + let mut expected = vec![MAX_BYTES_PER_FLOW; count]; + expected[count - 1 - 1] = 0; + assert_eq!(received(&f), expected); + } + + #[test] + fn byte_limits_are_never_exceeded() { + let f = Fixture::new(); + let count = MAX_BYTES_PER_POLL / MAX_BYTES_PER_FLOW + 1; + let sockets: Vec<_> = (0..u32::try_from(count).unwrap()) + .map(|port| f.connect(port)) + .collect(); + // So the last's share of the poll's budget is no multiple of a chunk. + f.host.send(on(0), &[1; 1000]); + for port in 1..u32::try_from(count).unwrap() { + f.host.send(on(port), &vec![1; 2 * MAX_BYTES_PER_FLOW]); + } + f.poll(); + let received: Vec<_> = sockets + .iter() + .map(|&socket| f.receive(socket).unwrap().len()) + .collect(); + assert!(received.iter().all(|&len| len <= MAX_BYTES_PER_FLOW)); + assert_eq!(received.iter().sum::(), MAX_BYTES_PER_POLL); + } + + #[test] + fn requests_taken_while_moving_data_are_answered_soon() { + let f = Fixture::new(); + let _socket = f.connect(7); + f.poll(); + assert_eq!(f.forwarder.next_deadline(), None); + // Taken while reading connection 7, after requests were answered. + f.host.connect_late(on(8)); + f.poll(); + f.assert_due_soon(); + f.poll(); + f.accept(); + assert!(f.host.with(on(8), |connection| connection.live)); + } + + #[test] + fn data_and_credit_taken_after_a_connections_turn_are_used_soon() { + let f = Fixture::new(); + let socket = f.connect(7); + f.host.with(on(7), |connection| connection.credit = 0); + assert_eq!(f.send(socket, b"reply"), Ok(5)); + f.poll(); + assert_eq!(f.forwarder.next_deadline(), None); + f.host + .return_after_turn(on(7), Returned::Data(b"late".to_vec())); + f.host + .return_after_turn(on(7), Returned::Credit(usize::MAX)); + f.poll(); + assert_eq!(f.receive(socket), Some(Vec::new())); + assert_eq!(f.host.take_received(on(7)), b""); + f.assert_due_soon(); + f.poll(); + assert_eq!(f.receive(socket), Some(b"late".to_vec())); + assert_eq!(f.host.take_received(on(7)), b"reply"); + f.trailing_poll(); + assert_eq!(f.forwarder.next_deadline(), None); + } + + #[test] + fn stalled_connections_leave_no_polls_due() { + let f = Fixture::new(); + let socket = f.connect(7); + // The host gives no credit, and sends more than the guest's socket + // holds. + f.host.with(on(7), |connection| connection.credit = 0); + assert_eq!(f.send(socket, b"reply"), Ok(5)); + let capacity = usize::try_from(UNIX_SOCKET_BUFFER_SIZE).unwrap(); + f.host.send(on(7), &vec![1; 2 * capacity]); + let unread = || f.host.with(on(7), |connection| connection.sent.len()); + // Until the guest's socket is full. + loop { + let before = unread(); + f.poll(); + if unread() == before { + break; + } + } + assert_ne!(unread(), 0); + let mut polls = 0; + while f.forwarder.next_deadline().is_some() { + assert!(polls < 2, "still due after {polls} polls"); + f.poll(); + polls += 1; + } + for _ in 0..5 { + f.poll(); + assert_eq!(f.forwarder.next_deadline(), None); + } + assert_eq!(f.host.take_received(on(7)), b""); + } + + #[test] + fn a_transport_reset_ends_live_connections_after_their_data() { + let f = Fixture::new(); + let first = f.connect(7); + let second = f.connect(8); + f.host.send(on(7), b"last"); + f.host.reset_all(); + f.poll(); + assert_eq!(f.receive(first), Some(b"last".to_vec())); + assert_eq!(f.receive(first), None); + assert_eq!(f.receive(second), None); + assert_eq!(f.send(second, b"lost"), Err(UnixSocketError::BrokenPipe)); + assert_eq!(f.forgotten(), [(on(7), false), (on(8), false)]); + f.trailing_poll(); + assert_eq!(f.forwarder.next_deadline(), None); + } +} diff --git a/litebox_runner_vm_kernel/src/devices.rs b/litebox_runner_vm_kernel/src/devices.rs index f2322fbcf2..f46dbd6da1 100644 --- a/litebox_runner_vm_kernel/src/devices.rs +++ b/litebox_runner_vm_kernel/src/devices.rs @@ -7,6 +7,8 @@ //! - [`Console`]: the VM's console device, else output to the serial port //! and no input. //! - [`Network`]: the VM's network device, if any. +//! - [`Vsock`]: the VM's socket device (virtio-vsock), if any, for the +//! broker's vsock forwarder. //! - [`Events`]: the devices' interrupt, and the local APIC's timer for //! deadlines. //! - [`ProcessTimer`]: the same timer, while a process runs, for its @@ -36,9 +38,11 @@ use litebox_broker_smoltcp::Network; use litebox_broker_vm_kernel::Events; use litebox_broker_vm_kernel::share::{Share, Transport, TransportError}; use litebox_broker_vm_kernel::stdio::Console; +use litebox_broker_vm_kernel::vsock::Vsock; use litebox_hal::clock::Clock; -use litebox_hal::devices::{ConsoleDevice, NetworkDevice, NinePDevice}; +use litebox_hal::devices::{ConsoleDevice, NetworkDevice, NinePDevice, VsockDevice}; use litebox_hal::interrupt::{LocalApic, SPURIOUS_VECTORS, StartupIpi}; +use litebox_hal::virtio::vsock::ConnectionId; use litebox_platform::time::{Instant as _, TimeProvider as _}; use litebox_platform_vm_kernel::cpu::{self, CpuId, InterruptController, Ipi, MAX_CPUS}; use litebox_platform_vm_kernel::{Instant, VmKernel, worker}; @@ -88,6 +92,9 @@ pub struct Devices { network: Option>, /// Whether the network device interrupts; else it is polled. network_interrupts: bool, + vsock: Option>, + /// Whether the socket device interrupts; else it is polled. + vsock_interrupts: bool, /// Per CPU, what [`ProcessTimer::arm`] last programmed into its timer /// (see [`armed`]), so it can skip rewriting it: under KVM each write is /// a VM exit, and it rearms before every resume of a process. @@ -128,6 +135,17 @@ impl Devices { Some(device) => litebox_util_log::info!(device:% = device; "network"), None => litebox_util_log::info!("no network device: sockets stay in the guest"), } + let vsock = match VsockDevice::probe(platform, apic.target(DEVICE_VECTOR)) { + Ok(device) => device, + Err(error) => { + litebox_util_log::warn!(error:% = error; "vsock device unusable"); + None + } + }; + match &vsock { + Some(device) => litebox_util_log::info!(device:% = device; "vsock"), + None => litebox_util_log::info!("no vsock device: no host AF_VSOCK connections"), + } let devices: &'static Self = Box::leak(Box::new(Self { platform, apic, @@ -135,6 +153,8 @@ impl Devices { console_interrupts: console.interrupts(), network_interrupts: network.as_ref().is_some_and(NetworkDevice::interrupts), network: network.map(Mutex::new), + vsock_interrupts: vsock.as_ref().is_some_and(VsockDevice::interrupts), + vsock: vsock.map(Mutex::new), armed: [const { AtomicU64::new(ARMED_UNKNOWN) }; MAX_CPUS], device_interrupted: AtomicU64::new(0), console: Mutex::new(console), @@ -157,6 +177,18 @@ impl Devices { .lock() } + /// The socket device, if any, for the broker's vsock forwarder. + pub fn vsock(&'static self) -> Option<&'static dyn Vsock> { + self.vsock.is_some().then_some(self) + } + + fn vsock_device(&self) -> spin::MutexGuard<'_, VsockDevice> { + self.vsock + .as_ref() + .expect("the broker has a vsock forwarder only with a device") + .lock() + } + /// The VM's virtio-9p devices, brought up, as shares mounted at their /// tags (see [`mount_path`]). Devices that cannot be driven, or whose tag /// is not such a path or repeats another's, are skipped. Call once. @@ -288,7 +320,11 @@ unsafe impl InterruptController for Devices { impl Events for Devices { fn wait(&self, deadline: Option) -> bool { - if deadline.is_none() && !self.console_interrupts && !self.network_interrupts { + if deadline.is_none() + && !self.console_interrupts + && !self.network_interrupts + && !self.vsock_interrupts + { return false; } self.halt(deadline); @@ -392,6 +428,57 @@ impl Network for Devices { } } +/// Each call holds the device's lock only for itself. +impl Vsock for Devices { + fn take_request(&self) -> Option { + self.vsock_device().take_request() + } + + fn accept(&self, id: ConnectionId) -> bool { + self.vsock_device().accept(id) + } + + fn refuse(&self, id: ConnectionId) { + self.vsock_device().refuse(id); + } + + fn read(&self, id: ConnectionId, output: &mut [u8]) -> Option { + self.vsock_device().read(id, output) + } + + fn write(&self, id: ConnectionId, bytes: &[u8]) -> Option { + self.vsock_device().write(id, bytes) + } + + fn shutdown(&self, id: ConnectionId) { + self.vsock_device().shutdown(id); + } + + fn shutdown_receiving(&self, id: ConnectionId) { + self.vsock_device().shutdown_receiving(id); + } + + fn close(&self, id: ConnectionId) { + self.vsock_device().close(id); + } + + fn is_closed(&self, id: ConnectionId) -> bool { + self.vsock_device().is_closed(id) + } + + fn forget(&self, id: ConnectionId) { + self.vsock_device().forget(id); + } + + fn take_backlog(&self) -> bool { + self.vsock_device().take_backlog() + } + + fn interrupts(&self) -> bool { + self.vsock_interrupts + } +} + /// A share's virtio-9p device: a request at a time, waiting for its response /// in [`Devices::halt`], not [`Devices::block`]: the 9P client holds its lock /// across the round trip, and while the worker is suspended, the kernel may diff --git a/litebox_runner_vm_kernel/src/main.rs b/litebox_runner_vm_kernel/src/main.rs index 8c2d5379f7..21891c1297 100644 --- a/litebox_runner_vm_kernel/src/main.rs +++ b/litebox_runner_vm_kernel/src/main.rs @@ -13,8 +13,9 @@ //! QEMU exits with 33 if all tests pass, 65 otherwise. Runner processes' //! standard streams are the VM's console device, if there is one, and their //! sockets reach the VM's network device, if there is one; programs also see -//! the host directories of virtio-9p devices, read-only, at their mount tags -//! ([`devices`]). +//! the host directories of virtio-9p devices, read-only, at their mount tags, +//! and, as Unix connections on `@litebox-vsock:`, the host's `AF_VSOCK` +//! connections through the VM's socket device, if there is one ([`devices`]). //! //! Every CPU runs the processes ([`litebox_shim_vm_kernel::sched`]); the boot //! CPU also runs the client and service, the others only the processes @@ -62,6 +63,7 @@ fn kernel_main(kernel: Kernel) -> ! { root: payload.file("rootfs.tar"), shares, network: devices.network(), + vsock: devices.vsock(), allow_process_duplication, }) }; diff --git a/litebox_runner_vm_kernel/tests/linux/vsock.c b/litebox_runner_vm_kernel/tests/linux/vsock.c new file mode 100644 index 0000000000..5ded8fba3d --- /dev/null +++ b/litebox_runner_vm_kernel/tests/linux/vsock.c @@ -0,0 +1,231 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT license. + +// The host's AF_VSOCK connections to vsock port PORT, which the broker's vsock +// forwarder passes on to the guest's Unix socket listening on the abstract +// name `litebox-vsock:PORT`; vsock.host.py makes them. +// +// vsock PORT Listens on `@litebox-vsock:PORT`; prints `ready`; then +// serves, in order: +// - two connections at once: reads each to EOF (the +// host's half-close), then echoes it and closes; +// - CHURN connections in turn, more than the device +// holds at once, ending as: the host half-closes and +// this echoes; this writes and closes first; the host +// writes and closes at once; +// - AFTER_CLOSE connections in turn: reads a request, +// writes REPLY bytes and closes, while the host still +// sends, which must not lose the reply; +// - two connections in turn: reads BULK (then SLOW) +// bytes, writes as many, shuts down writing, and waits +// for the host's ACK and EOF before closing. The host +// reads the second slowly. + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#define CHECK(cond) \ + do { \ + if (!(cond)) { \ + printf("FAIL %s:%d: %s (errno %d: %s)\n", __FILE__, __LINE__, \ + #cond, errno, strerror(errno)); \ + exit(1); \ + } \ + } while (0) + +// As vsock.host.py's. +#define BULK (2 * 1024 * 1024) +#define SLOW (256 * 1024) +#define CHURN 40 +#define AFTER_CLOSE 3 +#define REQUEST "request!" +#define REPLY (64 * 1024) +#define ACK "ack" + +static unsigned char pattern(size_t i) { + return (unsigned char)(i * 7 + i / 4096); +} + +// Reads `fd` to EOF into `buffer`, NUL-terminated. +static void read_to_eof(int fd, char *buffer, size_t size) { + size_t got = 0; + for (;;) { + CHECK(got < size - 1); + ssize_t n = read(fd, buffer + got, size - 1 - got); + CHECK(n >= 0); + if (n == 0) { + break; + } + got += (size_t)n; + } + buffer[got] = '\0'; +} + +// Reads exactly `len` bytes. +static void read_exactly(int fd, unsigned char *buffer, size_t len) { + size_t got = 0; + while (got < len) { + ssize_t n = read(fd, buffer + got, len - got); + CHECK(n > 0); + got += (size_t)n; + } +} + +static void write_all(int fd, const unsigned char *buffer, size_t len) { + size_t done = 0; + while (done < len) { + ssize_t n = write(fd, buffer + done, len - done); + CHECK(n > 0); + done += (size_t)n; + } +} + +static int accept_host(int listener) { + int fd = accept(listener, NULL, NULL); + CHECK(fd >= 0); + return fd; +} + +// Two connections at once, each echoed once the host half-closes it. +static void serve_concurrent(int listener) { + int fds[2]; + unsigned char data[2][4096]; + size_t got[2] = {0, 0}; + int open[2] = {1, 1}; + for (int i = 0; i < 2; i++) { + fds[i] = accept_host(listener); + } + while (open[0] || open[1]) { + struct pollfd polls[2]; + for (int i = 0; i < 2; i++) { + polls[i] = (struct pollfd){.fd = open[i] ? fds[i] : -1, .events = POLLIN}; + } + CHECK(poll(polls, 2, -1) > 0); + for (int i = 0; i < 2; i++) { + if (!(polls[i].revents & (POLLIN | POLLHUP | POLLERR | POLLNVAL))) { + continue; + } + CHECK(got[i] < sizeof(data[i])); + ssize_t n = read(fds[i], data[i] + got[i], sizeof(data[i]) - got[i]); + CHECK(n >= 0); + if (n == 0) { + open[i] = 0; + } + got[i] += (size_t)n; + } + } + for (int i = 0; i < 2; i++) { + CHECK(got[i] > 0); + // After the host's EOF, writes still reach it. + write_all(fds[i], data[i], got[i]); + CHECK(close(fds[i]) == 0); + } +} + +// One connection after another, each ending one of three ways (see the +// header). +static void serve_churn(int listener) { + char buffer[64], expected[64]; + for (int i = 0; i < CHURN; i++) { + int fd = accept_host(listener); + switch (i % 3) { + case 0: + read_to_eof(fd, buffer, sizeof(buffer)); + snprintf(expected, sizeof(expected), "ping %d", i); + CHECK(strcmp(buffer, expected) == 0); + write_all(fd, (const unsigned char *)buffer, strlen(buffer)); + break; + case 1: + snprintf(buffer, sizeof(buffer), "hello %d", i); + write_all(fd, (const unsigned char *)buffer, strlen(buffer)); + break; + default: + read_to_eof(fd, buffer, sizeof(buffer)); + snprintf(expected, sizeof(expected), "abort %d", i); + CHECK(strcmp(buffer, expected) == 0); + break; + } + CHECK(close(fd) == 0); + } +} + +// Replies to a request and closes at once, while the host still sends. +static void serve_after_close(int listener) { + static unsigned char buffer[REPLY]; + for (int i = 0; i < AFTER_CLOSE; i++) { + int fd = accept_host(listener); + read_exactly(fd, buffer, strlen(REQUEST)); + CHECK(memcmp(buffer, REQUEST, strlen(REQUEST)) == 0); + for (size_t j = 0; j < REPLY; j++) { + buffer[j] = pattern(j); + } + write_all(fd, buffer, REPLY); + CHECK(close(fd) == 0); + } +} + +// `size` bytes each way, through the device's and the sockets' flow control. +static void serve_bulk(int listener, size_t size) { + static unsigned char buffer[64 * 1024]; + int fd = accept_host(listener); + for (size_t done = 0; done < size;) { + size_t len = size - done < sizeof(buffer) ? size - done : sizeof(buffer); + ssize_t n = read(fd, buffer, len); + CHECK(n > 0); + for (ssize_t i = 0; i < n; i++) { + CHECK(buffer[i] == pattern(done + (size_t)i)); + } + done += (size_t)n; + } + for (size_t done = 0; done < size; done += sizeof(buffer)) { + size_t len = size - done < sizeof(buffer) ? size - done : sizeof(buffer); + for (size_t i = 0; i < len; i++) { + buffer[i] = pattern(done + i); + } + write_all(fd, buffer, len); + } + CHECK(shutdown(fd, SHUT_WR) == 0); + // The host acknowledges once it has read everything, then ends. + char ack[sizeof(ACK) + 1]; + read_to_eof(fd, ack, sizeof(ack)); + CHECK(strcmp(ack, ACK) == 0); + CHECK(close(fd) == 0); +} + +static int serve(const char *port) { + int listener = socket(AF_UNIX, SOCK_STREAM, 0); + CHECK(listener >= 0); + // Abstract: a NUL, then the name, which the address length bounds. + struct sockaddr_un address = {.sun_family = AF_UNIX}; + int name = snprintf(address.sun_path + 1, sizeof(address.sun_path) - 1, + "litebox-vsock:%s", port); + CHECK(name > 0 && (size_t)name < sizeof(address.sun_path) - 1); + socklen_t len = (socklen_t)(offsetof(struct sockaddr_un, sun_path) + 1 + (size_t)name); + CHECK(bind(listener, (struct sockaddr *)&address, len) == 0); + CHECK(listen(listener, 8) == 0); + printf("ready\n"); + fflush(stdout); + serve_concurrent(listener); + serve_churn(listener); + serve_after_close(listener); + serve_bulk(listener, BULK); + serve_bulk(listener, SLOW); + CHECK(close(listener) == 0); + printf("served\n"); + return 0; +} + +int main(int argc, char **argv) { + if (argc != 2) { + fprintf(stderr, "usage: vsock PORT\n"); + return 2; + } + return serve(argv[1]); +} diff --git a/litebox_runner_vm_kernel/tests/linux/vsock.host.py b/litebox_runner_vm_kernel/tests/linux/vsock.host.py new file mode 100755 index 0000000000..d0bacd3846 --- /dev/null +++ b/litebox_runner_vm_kernel/tests/linux/vsock.host.py @@ -0,0 +1,179 @@ +#!/usr/bin/env python3 + +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. + +"""Host client of the vsock test: usage `vsock.host.py CID CONSOLE`. + +Waits for the guest's `ready` line in the console output file CONSOLE, then +connects to guest CID, port 1024, for the scenarios vsock.c serves, in its +order, with a refused connection to a port nothing listens on after the +first; exits 0 if all of them went as expected. +""" + +import os +import socket +import sys +import time + +PORT = 1024 +UNUSED_PORT = 1025 +# As vsock.c's. +BULK = 2 * 1024 * 1024 +SLOW = 256 * 1024 +CHURN = 40 +AFTER_CLOSE = 3 +REQUEST = b"request!" +REPLY = 64 * 1024 +ACK = b"ack" +# The receive buffer, and the pause between receives, of slow readers. +SLOW_BUFFER = 4096 +SLOW_PAUSE = 0.001 +# Each step's; the run script's TIMEOUT bounds the whole. +STEP_TIMEOUT = 60 + + +def pattern(length): + return bytes((i * 7 + i // 4096) & 0xFF for i in range(length)) + + +def wait_ready(console, deadline): + while time.monotonic() < deadline: + try: + with open(console, "rb") as f: + if b"ready" in f.read().splitlines(): + return + except FileNotFoundError: + pass + time.sleep(0.1) + raise TimeoutError("the guest never printed `ready`") + + +def connect(cid, port, slow=False): + s = socket.socket(socket.AF_VSOCK, socket.SOCK_STREAM) + s.settimeout(STEP_TIMEOUT) + if slow: + # The credit it gives the guest. + s.setsockopt(socket.AF_VSOCK, socket.SO_VM_SOCKETS_BUFFER_SIZE, SLOW_BUFFER) + s.connect((cid, port)) + return s + + +def read_to_eof(s, slow=False): + chunks = [] + while True: + if slow: + time.sleep(SLOW_PAUSE) + chunk = s.recv(SLOW_BUFFER if slow else 65536) + if not chunk: + return b"".join(chunks) + chunks.append(chunk) + + +def check_data(received, expected): + if len(received) != len(expected): + raise AssertionError(f"received {len(received)} bytes, expected {len(expected)}") + if received != expected: + first = next(i for i in range(len(expected)) if received[i] != expected[i]) + raise AssertionError(f"received data differs at byte {first}") + + +def concurrent(cid): + sockets = [connect(cid, PORT) for _ in range(2)] + messages = [f"hello from connection {i}\n".encode() * (i + 1) for i in range(2)] + # Both open before either sends. + for s, message in reversed(list(zip(sockets, messages))): + s.sendall(message) + s.shutdown(socket.SHUT_WR) + for s, message in zip(sockets, messages): + echo = read_to_eof(s) + if echo != message: + raise AssertionError(f"echo {echo!r}, expected {message!r}") + s.close() + + +def refused(cid): + try: + connect(cid, UNUSED_PORT).close() + except ConnectionResetError: + return + raise AssertionError(f"connected to port {UNUSED_PORT}, where nothing listens") + + +def churn(cid): + for i in range(CHURN): + s = connect(cid, PORT) + kind = i % 3 + if kind == 0: + message = f"ping {i}".encode() + s.sendall(message) + s.shutdown(socket.SHUT_WR) + expected = message + elif kind == 1: + expected = f"hello {i}".encode() + else: + s.sendall(f"abort {i}".encode()) + s.close() + continue + received = read_to_eof(s) + if received != expected: + raise AssertionError(f"connection {i}: {received!r}, expected {expected!r}") + s.close() + + +def send_until_broken_pipe(s): + deadline = time.monotonic() + STEP_TIMEOUT + while time.monotonic() < deadline: + try: + s.send(b"more") + except BrokenPipeError: + return + time.sleep(0.01) + raise AssertionError("sends after the guest closed did not fail") + + +def after_close(cid): + for _ in range(AFTER_CLOSE): + s = connect(cid, PORT, slow=True) + s.sendall(REQUEST) + # The guest replies and closes meanwhile; little of the reply fits + # in this end's buffer, so the rest waits in the closing guest. + time.sleep(0.2) + send_until_broken_pipe(s) + check_data(read_to_eof(s, slow=True), pattern(REPLY)) + s.close() + + +def bulk(cid, size, slow): + s = connect(cid, PORT, slow) + data = pattern(size) + s.sendall(data) + check_data(read_to_eof(s, slow), data) + # The guest waits for this before it ends. + s.sendall(ACK) + s.close() + + +def main(): + if len(sys.argv) != 3: + print(__doc__, file=sys.stderr) + return 2 + cid, console = int(sys.argv[1]), sys.argv[2] + # The run's, as the run script passes it. + timeout = int(os.environ.get("TIMEOUT", "120")) + wait_ready(console, time.monotonic() + timeout) + try: + concurrent(cid) + refused(cid) + churn(cid) + after_close(cid) + bulk(cid, BULK, slow=False) + bulk(cid, SLOW, slow=True) + except (AssertionError, OSError) as error: + print(f"FAIL: {error!r}") + return 1 + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/litebox_runner_vm_kernel/tests/linux/vsock.json b/litebox_runner_vm_kernel/tests/linux/vsock.json new file mode 100644 index 0000000000..a427ad2f4d --- /dev/null +++ b/litebox_runner_vm_kernel/tests/linux/vsock.json @@ -0,0 +1,3 @@ +[ + { "argv": ["/vsock", "1024"], "expect_stdout": "ready\nserved\n", "expect_stderr": "" } +]