From 08f7cb176b2dc8904334303021946d19884d67d0 Mon Sep 17 00:00:00 2001 From: Changyuan Lyu Date: Sat, 12 Sep 2026 23:47:44 -0700 Subject: [PATCH 1/4] fix(vsock): handle partial connection requests A `CONNECT ` line from a host client can arrive in several pieces. For example `writeln!()` emits the literal, the port and the newline in three separate writes. Since the accepted socket is non-blocking, `read_line()` then fails with `EAGAIN`, and the error was propagated out of the mio worker loop, tearing down the whole device and closing every connection. Keep the buffered reader and the bytes received so far in the socket map and resume reading once the socket is readable again. Drop and deregister the socket when the client hangs up mid-request or sends an unparsable port. This also fixes flaky failures of vsock_host_close_test, where the `writeln!()` of the test itself raced with the device. Assisted-by: Antigravity:Claude-Opus-5 Signed-off-by: Changyuan Lyu --- alioth/src/virtio/dev/vsock/uds_vsock.rs | 68 +++++++++++++++---- alioth/src/virtio/dev/vsock/uds_vsock_test.rs | 68 +++++++++++++++++++ 2 files changed, 122 insertions(+), 14 deletions(-) diff --git a/alioth/src/virtio/dev/vsock/uds_vsock.rs b/alioth/src/virtio/dev/vsock/uds_vsock.rs index 913a20eb..00c5b05a 100644 --- a/alioth/src/virtio/dev/vsock/uds_vsock.rs +++ b/alioth/src/virtio/dev/vsock/uds_vsock.rs @@ -71,11 +71,20 @@ pub struct UdsVsock { listener: UnixListener, connections: HashMap<(u32, u32), Connection>, ports: HashMap, - sockets: HashMap, + sockets: HashMap, host_ports: HashMap, next_port: u32, } +/// An accepted socket whose `CONNECT` request line has not been fully +/// received yet. +#[derive(Debug)] +struct PendingConn { + reader: BufReader, + /// Bytes of the request line received so far. + msg: String, +} + fn get_buf_size(stream: &UnixStream) -> Result { let mut buf_size = 0i32; let mut arg_size = size_of_val(&buf_size) as libc::socklen_t; @@ -114,14 +123,24 @@ impl UdsVsock { token, Interest::READABLE, )?; - self.sockets.insert(token, stream); + let pending = PendingConn { + reader: BufReader::new(stream), + msg: String::new(), + }; + self.sockets.insert(token, pending); + Ok(()) + } + + fn drop_socket(&self, socket: &UnixStream, registry: &Registry) -> Result<()> { + registry.deregister(&mut SourceFd(&socket.as_raw_fd()))?; Ok(()) } fn handle_conn_request<'m, Q, S>( &mut self, token: Token, - socket: UnixStream, + mut pending: PendingConn, + registry: &Registry, rx_q: &mut Queue<'_, 'm, Q>, irq_sender: &S, ) -> Result<()> @@ -129,19 +148,39 @@ impl UdsVsock { Q: VirtQueue<'m>, S: IrqSender, { - let mut msg = String::new(); - let writer = socket.try_clone()?; - let mut reader = BufReader::new(socket); + // The socket is non-blocking, so a request line can arrive in pieces. + // Keep what has been received so far and wait for the next event + // instead of tearing down the connection. + match pending.reader.read_line(&mut pending.msg) { + Ok(_) => {} + Err(e) if e.kind() == ErrorKind::WouldBlock => { + self.sockets.insert(token, pending); + return Ok(()); + } + Err(e) => return Err(e.into()), + } + if !pending.msg.ends_with('\n') { + if pending.msg.is_empty() { + log::debug!("{}: socket closed before any request", self.name); + } else { + log::warn!( + "{}: socket closed mid-request: {:?}", + self.name, + pending.msg + ); + } + return self.drop_socket(pending.reader.get_ref(), registry); + } + let writer = pending.reader.get_ref().try_clone()?; let buf_size = get_buf_size(&writer)?; - reader.read_line(&mut msg)?; - let port_str = msg.trim_start_matches("CONNECT ").trim_end(); + let port_str = pending.msg.trim_start_matches("CONNECT ").trim_end(); let Ok(port) = port_str.parse::() else { log::error!("{}: failed to parse port {port_str}", self.name); - return Ok(()); + return self.drop_socket(pending.reader.get_ref(), registry); }; let Some(host_port) = self.allocate_port() else { log::error!("{}: failed to allocate port", self.name); - return Ok(()); + return self.drop_socket(pending.reader.get_ref(), registry); }; let hdr = VsockHeader { src_cid: VSOCK_CID_HOST, @@ -157,7 +196,7 @@ impl UdsVsock { self.respond(&hdr, irq_sender, rx_q)?; let conn = Connection { state: ConnState::Requested, - reader, + reader: pending.reader, writer: BufWriter::new(writer), buf_alloc: buf_size as u32, eof: false, @@ -837,8 +876,8 @@ impl VirtioMio for UdsVsock { }; if token.0 == self.listener.as_raw_fd() as usize { self.create_socket(registry) - } else if let Some(socket) = self.sockets.remove(&token) { - self.handle_conn_request(token, socket, rx_q, irq_sender) + } else if let Some(pending) = self.sockets.remove(&token) { + self.handle_conn_request(token, pending, registry, rx_q, irq_sender) } else if let Some(port_pair) = self.ports.get(&token) { let (host_port, guest_port) = port_pair.to_owned(); self.process_rx_data(host_port, guest_port, registry, rx_q, irq_sender) @@ -887,7 +926,8 @@ impl VirtioMio for UdsVsock { log::error!("{}: failed to deregister socket: {err}", self.name); } } - for (_, socket) in self.sockets.drain() { + for (_, pending) in self.sockets.drain() { + let socket = pending.reader.into_inner(); if let Err(err) = registry.deregister(&mut SourceFd(&socket.as_raw_fd())) { log::error!("{}: failed to deregister socket: {err}", self.name); } diff --git a/alioth/src/virtio/dev/vsock/uds_vsock_test.rs b/alioth/src/virtio/dev/vsock/uds_vsock_test.rs index ee503253..72399de0 100644 --- a/alioth/src/virtio/dev/vsock/uds_vsock_test.rs +++ b/alioth/src/virtio/dev/vsock/uds_vsock_test.rs @@ -644,3 +644,71 @@ fn vsock_host_close_no_desc_test() { notifier.notify().unwrap(); handle.join().unwrap(); } + +#[test] +fn vsock_partial_conn_request_test() { + let ram_bus = Arc::new(fixture_ram_bus()); + let ram = ram_bus.lock_layout(); + let regs: Arc<[QueueReg]> = Arc::from(fixture_queues(3)); + let reg_rx = ®s[VsockVirtq::RX.raw() as usize]; + let mut rx_q = GuestQueue::new( + SplitQueue::new(reg_rx, &ram, false).unwrap().unwrap(), + reg_rx, + ); + + let temp_dir = TempDir::new().unwrap(); + let sock_path = temp_dir.path().join("vsock.sock"); + + const GUEST_CID: u32 = 3; + let param = UdsVsockSpec { + cid: GUEST_CID, + path: sock_path.clone().into(), + }; + let dev = param.build("vsock").unwrap(); + + let (tx, rx) = flume::unbounded(); + let (handle, notifier) = dev.spawn_worker(rx, ram_bus.clone(), regs).unwrap(); + let (irq_tx, irq_rx) = flume::unbounded(); + let irq_sender = Arc::new(FakeIrqSender { q_tx: irq_tx }); + let start_param = StartParam { + feature: VirtioFeature::VERSION_1.bits(), + irq_sender, + notifiers: Option::>::None, + }; + tx.send(WakeEvent::Start { param: start_param }).unwrap(); + + let rx_buf_addr = DATA_ADDR; + + let mut h2g_stream = UnixStream::connect(&sock_path).unwrap(); + let buf_id = rx_q.add_desc(&[], &[(rx_buf_addr, 4096)]); + + // A connection request can be split over multiple writes, e.g. as done by + // `writeln!()`. The device must wait for the complete line instead of + // dropping the connection. + const H2G_GUEST_PORT: u32 = 1025; + h2g_stream.write_all(b"CONNECT ").unwrap(); + thread::sleep(Duration::from_millis(50)); + h2g_stream + .write_all(format!("{H2G_GUEST_PORT}\n").as_bytes()) + .unwrap(); + + assert_eq!( + irq_rx.recv_timeout(Duration::from_secs(1)).unwrap(), + VsockVirtq::RX.raw() + ); + let used = rx_q.get_used().unwrap(); + assert_eq!(used.id, buf_id); + assert_eq!(used.len as usize, size_of::()); + + let mut hdr = VsockHeader::new_zeroed(); + ram.read(rx_buf_addr, hdr.as_mut_bytes()).unwrap(); + assert_eq!(hdr.src_cid, VSOCK_CID_HOST); + assert_eq!(hdr.dst_cid, GUEST_CID); + assert_eq!(hdr.dst_port, H2G_GUEST_PORT); + assert_eq!(hdr.op, VsockOp::REQUEST); + assert_eq!(hdr.type_, VsockType::STREAM); + + tx.send(WakeEvent::Shutdown).unwrap(); + notifier.notify().unwrap(); + handle.join().unwrap(); +} From ed37accb5acd1fd54c954e34a5c8de031b65842b Mon Sep 17 00:00:00 2001 From: Changyuan Lyu Date: Sat, 12 Sep 2026 23:49:49 -0700 Subject: [PATCH 2/4] fix(vsock): drain the listener backlog on each event The listener is registered with mio, which uses edge-triggered epoll, but only one connection was accepted per event. When a client connects while another connection is still queued, no new edge is reported and the queued client is never served. In practice every connection was then served one client behind: the pending connection is only accepted once the next client connects. Accept in a loop until `EAGAIN`. While at it, keep the device alive when a client aborts a connection before it is accepted. Assisted-by: Antigravity:Claude-Opus-5 Signed-off-by: Changyuan Lyu --- alioth/src/virtio/dev/vsock/uds_vsock.rs | 51 ++++-- alioth/src/virtio/dev/vsock/uds_vsock_test.rs | 165 ++++++++++++++++++ 2 files changed, 202 insertions(+), 14 deletions(-) diff --git a/alioth/src/virtio/dev/vsock/uds_vsock.rs b/alioth/src/virtio/dev/vsock/uds_vsock.rs index 00c5b05a..f8879dca 100644 --- a/alioth/src/virtio/dev/vsock/uds_vsock.rs +++ b/alioth/src/virtio/dev/vsock/uds_vsock.rs @@ -15,7 +15,7 @@ use std::collections::HashMap; use std::fmt::Debug; use std::fs; -use std::io::{BufRead, BufReader, BufWriter, ErrorKind, IoSlice, IoSliceMut, Read, Write}; +use std::io::{self, BufRead, BufReader, BufWriter, ErrorKind, IoSlice, IoSliceMut, Read, Write}; use std::mem::size_of_val; use std::num::Wrapping; use std::os::fd::AsRawFd; @@ -85,6 +85,15 @@ struct PendingConn { msg: String, } +/// Returns true if `e` means the host side of a connection is gone, which is +/// a normal event that must only affect that single connection. +fn is_conn_lost(e: &io::Error) -> bool { + matches!( + e.kind(), + ErrorKind::BrokenPipe | ErrorKind::ConnectionReset | ErrorKind::ConnectionAborted + ) +} + fn get_buf_size(stream: &UnixStream) -> Result { let mut buf_size = 0i32; let mut arg_size = size_of_val(&buf_size) as libc::socklen_t; @@ -115,19 +124,33 @@ impl UdsVsock { } fn create_socket(&mut self, registry: &Registry) -> Result<()> { - let (stream, _) = self.listener.accept()?; - stream.set_nonblocking(true)?; - let token = Token(stream.as_raw_fd() as usize); - registry.register( - &mut SourceFd(&stream.as_raw_fd()), - token, - Interest::READABLE, - )?; - let pending = PendingConn { - reader: BufReader::new(stream), - msg: String::new(), - }; - self.sockets.insert(token, pending); + // The listener is registered edge-triggered, so drain the backlog. + // Otherwise a connection that arrives while another one is pending + // stalls until yet another client shows up. + loop { + let stream = match self.listener.accept() { + Ok((stream, _)) => stream, + Err(e) if e.kind() == ErrorKind::WouldBlock => break, + Err(e) if e.kind() == ErrorKind::Interrupted => continue, + Err(e) if is_conn_lost(&e) => { + log::debug!("{}: aborted connection: {e:?}", self.name); + continue; + } + Err(e) => return Err(e.into()), + }; + stream.set_nonblocking(true)?; + let token = Token(stream.as_raw_fd() as usize); + registry.register( + &mut SourceFd(&stream.as_raw_fd()), + token, + Interest::READABLE, + )?; + let pending = PendingConn { + reader: BufReader::new(stream), + msg: String::new(), + }; + self.sockets.insert(token, pending); + } Ok(()) } diff --git a/alioth/src/virtio/dev/vsock/uds_vsock_test.rs b/alioth/src/virtio/dev/vsock/uds_vsock_test.rs index 72399de0..3fe5a411 100644 --- a/alioth/src/virtio/dev/vsock/uds_vsock_test.rs +++ b/alioth/src/virtio/dev/vsock/uds_vsock_test.rs @@ -712,3 +712,168 @@ fn vsock_partial_conn_request_test() { notifier.notify().unwrap(); handle.join().unwrap(); } + +#[test] +fn vsock_simultaneous_conn_test() { + let ram_bus = Arc::new(fixture_ram_bus()); + let ram = ram_bus.lock_layout(); + let regs: Arc<[QueueReg]> = Arc::from(fixture_queues(3)); + let reg_rx = ®s[VsockVirtq::RX.raw() as usize]; + let mut rx_q = GuestQueue::new( + SplitQueue::new(reg_rx, &ram, false).unwrap().unwrap(), + reg_rx, + ); + + let temp_dir = TempDir::new().unwrap(); + let sock_path = temp_dir.path().join("vsock.sock"); + + const GUEST_CID: u32 = 3; + let param = UdsVsockSpec { + cid: GUEST_CID, + path: sock_path.clone().into(), + }; + let dev = param.build("vsock").unwrap(); + + let (tx, rx) = flume::unbounded(); + let (handle, notifier) = dev.spawn_worker(rx, ram_bus.clone(), regs).unwrap(); + let (irq_tx, irq_rx) = flume::unbounded(); + let irq_sender = Arc::new(FakeIrqSender { q_tx: irq_tx }); + let start_param = StartParam { + feature: VirtioFeature::VERSION_1.bits(), + irq_sender, + notifiers: Option::>::None, + }; + tx.send(WakeEvent::Start { param: start_param }).unwrap(); + + let buf_addrs = [DATA_ADDR, DATA_ADDR + 2048]; + let buf_ids = buf_addrs.map(|addr| rx_q.add_desc(&[], &[(addr, 2048)])); + + // Two clients connecting back to back pile up in the listener backlog, + // and both must be served. + const GUEST_PORTS: [u32; 2] = [1025, 1026]; + let mut streams = Vec::new(); + for port in GUEST_PORTS { + let mut stream = UnixStream::connect(&sock_path).unwrap(); + stream + .write_all(format!("CONNECT {port}\n").as_bytes()) + .unwrap(); + streams.push(stream); + } + + let mut requests = Vec::new(); + for _ in GUEST_PORTS { + irq_rx.recv_timeout(Duration::from_secs(1)).unwrap(); + let used = rx_q.get_used().unwrap(); + assert_eq!(used.len as usize, size_of::()); + let index = buf_ids.iter().position(|id| *id == used.id).unwrap(); + let mut hdr = VsockHeader::new_zeroed(); + ram.read(buf_addrs[index], hdr.as_mut_bytes()).unwrap(); + assert_eq!(hdr.op, VsockOp::REQUEST); + requests.push(hdr.dst_port); + } + requests.sort_unstable(); + assert_eq!(requests, GUEST_PORTS); + + tx.send(WakeEvent::Shutdown).unwrap(); + notifier.notify().unwrap(); + handle.join().unwrap(); +} + +#[test] +fn vsock_conn_request_eof_test() { + let ram_bus = Arc::new(fixture_ram_bus()); + let ram = ram_bus.lock_layout(); + let regs: Arc<[QueueReg]> = Arc::from(fixture_queues(3)); + let reg_tx = ®s[VsockVirtq::TX.raw() as usize]; + let reg_rx = ®s[VsockVirtq::RX.raw() as usize]; + let mut rx_q = GuestQueue::new( + SplitQueue::new(reg_rx, &ram, false).unwrap().unwrap(), + reg_rx, + ); + let mut tx_q = GuestQueue::new( + SplitQueue::new(reg_tx, &ram, false).unwrap().unwrap(), + reg_tx, + ); + + let temp_dir = TempDir::new().unwrap(); + let sock_path = temp_dir.path().join("vsock.sock"); + + const GUEST_CID: u32 = 3; + let param = UdsVsockSpec { + cid: GUEST_CID, + path: sock_path.clone().into(), + }; + let dev = param.build("vsock").unwrap(); + + let (tx, rx) = flume::unbounded(); + let (handle, notifier) = dev.spawn_worker(rx, ram_bus.clone(), regs).unwrap(); + let (irq_tx, irq_rx) = flume::unbounded(); + let irq_sender = Arc::new(FakeIrqSender { q_tx: irq_tx }); + let start_param = StartParam { + feature: VirtioFeature::VERSION_1.bits(), + irq_sender, + notifiers: Option::>::None, + }; + tx.send(WakeEvent::Start { param: start_param }).unwrap(); + + let rx_buf_addr = DATA_ADDR; + let tx_buf_addr = DATA_ADDR + 4096; + + // A client that connects and disconnects without saying anything. + drop(UnixStream::connect(&sock_path).unwrap()); + + // A client that disconnects in the middle of a connection request. + let mut partial = UnixStream::connect(&sock_path).unwrap(); + partial.write_all(b"CONNECT ").unwrap(); + thread::sleep(Duration::from_millis(50)); + drop(partial); + thread::sleep(Duration::from_millis(50)); + + // Neither takes down the device: a well-behaved client still works. + let mut h2g_stream = UnixStream::connect(&sock_path).unwrap(); + let buf_id = rx_q.add_desc(&[], &[(rx_buf_addr, 4096)]); + const H2G_GUEST_PORT: u32 = 1025; + writeln!(h2g_stream, "CONNECT {H2G_GUEST_PORT}").unwrap(); + assert_eq!( + irq_rx.recv_timeout(Duration::from_secs(1)).unwrap(), + VsockVirtq::RX.raw() + ); + let used = rx_q.get_used().unwrap(); + assert_eq!(used.id, buf_id); + assert_eq!(used.len as usize, size_of::()); + + let mut hdr = VsockHeader::new_zeroed(); + ram.read(rx_buf_addr, hdr.as_mut_bytes()).unwrap(); + assert_eq!(hdr.op, VsockOp::REQUEST); + assert_eq!(hdr.dst_port, H2G_GUEST_PORT); + let h2g_host_port = hdr.src_port; + + let resp_hdr = VsockHeader { + src_cid: GUEST_CID, + dst_cid: VSOCK_CID_HOST, + src_port: H2G_GUEST_PORT, + dst_port: h2g_host_port, + op: VsockOp::RESPONSE, + type_: VsockType::STREAM, + ..Default::default() + }; + send_to_tx( + &resp_hdr, + &[], + &ram, + tx_buf_addr, + &mut tx_q, + &tx, + ¬ifier, + &irq_rx, + false, + ); + let mut reader = BufReader::new(&h2g_stream); + let mut line = String::new(); + reader.read_line(&mut line).unwrap(); + assert_eq!(line, format!("OK {h2g_host_port}\n")); + + tx.send(WakeEvent::Shutdown).unwrap(); + notifier.notify().unwrap(); + handle.join().unwrap(); +} From c0823787c37999a7ed7a60d162fa7db1ee2c7a86 Mon Sep 17 00:00:00 2001 From: Changyuan Lyu Date: Sat, 12 Sep 2026 23:51:52 -0700 Subject: [PATCH 3/4] fix(vsock): reset a connection when the host peer hangs up Writing to a host socket whose client already closed fails with EPIPE. The error was propagated out of the mio worker loop, so a single client hanging up at the wrong moment took down the whole device together with all other connections. Two paths are affected: - the "OK " acknowledgement sent when the guest accepts a connection that the client abandoned in the meantime, - forwarding guest data of an established connection. Mark the connection as EOF in both cases and let process_rx_data() send an RST to the guest, exactly like a client that closes while idle. If the RX queue has no descriptor at that point, flush_rx_data() delivers the RST once the guest provides one. Also downgrade the log for host data arriving before the guest accepts a connection: it stays buffered in the socket and is not an error. Assisted-by: Antigravity:Claude-Opus-5 Signed-off-by: Changyuan Lyu --- alioth/src/virtio/dev/vsock/uds_vsock.rs | 90 +++++++++++++---- alioth/src/virtio/dev/vsock/uds_vsock_test.rs | 97 +++++++++++++++++++ 2 files changed, 167 insertions(+), 20 deletions(-) diff --git a/alioth/src/virtio/dev/vsock/uds_vsock.rs b/alioth/src/virtio/dev/vsock/uds_vsock.rs index f8879dca..456482dc 100644 --- a/alioth/src/virtio/dev/vsock/uds_vsock.rs +++ b/alioth/src/virtio/dev/vsock/uds_vsock.rs @@ -316,8 +316,20 @@ impl UdsVsock { ); return Ok(()); }; - writeln!(conn.writer, "OK {host_port}")?; - conn.writer.flush()?; + let acked = writeln!(conn.writer, "OK {host_port}").and_then(|_| conn.writer.flush()); + match acked { + Ok(()) => {} + // The host hung up before the guest accepted the connection. + // `process_rx_data()` below turns this into an RST for the guest. + Err(e) if is_conn_lost(&e) => { + log::debug!( + "{}: host:{host_port} -> vm:{guest_port}: host closed before accept", + self.name + ); + conn.eof = true; + } + Err(e) => return Err(e.into()), + } conn.state = ConnState::Established { fwd_cnt: Wrapping(0), }; @@ -517,7 +529,7 @@ impl UdsVsock { VsockOp::REQUEST => self.handle_tx_request(hdr, registry, irq_sender, rx_q), VsockOp::RESPONSE => self.handle_tx_response(hdr, registry, rx_q, irq_sender), VsockOp::RST => self.handle_tx_rst(hdr, registry), - VsockOp::RW => self.transfer_tx_data(hdr, body, readable), + VsockOp::RW => self.transfer_tx_data(hdr, body, readable, registry, rx_q, irq_sender), VsockOp::CREDIT_UPDATE => { log::info!( "{name}: CREDIT_UPDATE: fwd_cnt: {}, buf_alloc: {}", @@ -616,7 +628,13 @@ impl UdsVsock { return Ok(()); } let ConnState::Established { fwd_cnt } = conn.state else { - log::error!("{}: unexpected state {:?}", self.name, conn.state); + // Data can arrive before the guest accepts the connection. It + // stays buffered in the socket until then. + log::debug!( + "{}: host:{host_port} -> vm:{guest_port}: not ready, state {:?}", + self.name, + conn.state + ); return Ok(()); }; let mut hdr = VsockHeader { @@ -714,17 +732,24 @@ impl UdsVsock { Ok(()) } - fn transfer_tx_data( + fn transfer_tx_data<'m, Q, S>( &mut self, hdr: &VsockHeader, body: &[u8], buffers: &[IoSlice], - ) -> Result<()> { + registry: &Registry, + rx_q: &mut Queue<'_, 'm, Q>, + irq_sender: &S, + ) -> Result<()> + where + Q: VirtQueue<'m>, + S: IrqSender, + { fn copy_to_conn( buf: &[u8], conn: &mut BufWriter, remain: &mut usize, - ) -> Result<()> { + ) -> io::Result<()> { if let Some(b) = buf.get(..*remain) { conn.write_all(b)?; *remain = 0; @@ -735,6 +760,28 @@ impl UdsVsock { Ok(()) } + /// Writes up to `len` bytes of `body` and `buffers` to `conn`, + /// returning the number of bytes that were not covered by the input. + fn write_to_conn( + conn: &mut BufWriter, + body: &[u8], + buffers: &[IoSlice], + len: usize, + ) -> io::Result { + let mut remain = len; + if !body.is_empty() { + copy_to_conn(body, conn, &mut remain)?; + } + for buf in buffers { + if remain == 0 { + break; + } + copy_to_conn(buf, conn, &mut remain)?; + } + conn.flush()?; + Ok(remain) + } + let host_port = hdr.dst_port; let guest_port = hdr.src_port; let Some(conn) = self.connections.get_mut(&(host_port, guest_port)) else { @@ -748,19 +795,23 @@ impl UdsVsock { log::warn!("{}: invalid connection state {:?}", self.name, conn.state); return Ok(()); }; - let mut remain = hdr.len as usize; - if !body.is_empty() { - copy_to_conn(body, &mut conn.writer, &mut remain)?; - } - for buf in buffers { - if remain == 0 { - break; + match write_to_conn(&mut conn.writer, body, buffers, hdr.len as usize) { + Ok(0) => {} + Ok(remain) => { + log::error!("{}: missing {remain} bytes", self.name); + return error::InvalidBuffer.fail(); } - copy_to_conn(buf, &mut conn.writer, &mut remain)?; - } - if remain != 0 { - log::error!("{}: missing {remain} bytes", self.name); - return error::InvalidBuffer.fail(); + // The host hung up. Reset this connection only, the rest of the + // device keeps running. + Err(e) if is_conn_lost(&e) => { + log::debug!( + "{}: vm:{guest_port} -> host:{host_port}: host closed", + self.name + ); + conn.eof = true; + return self.process_rx_data(host_port, guest_port, registry, rx_q, irq_sender); + } + Err(e) => return Err(e.into()), } *fwd_cnt += hdr.len; log::trace!( @@ -768,7 +819,6 @@ impl UdsVsock { self.name, hdr.len ); - conn.writer.flush()?; Ok(()) } } diff --git a/alioth/src/virtio/dev/vsock/uds_vsock_test.rs b/alioth/src/virtio/dev/vsock/uds_vsock_test.rs index 3fe5a411..04a540d4 100644 --- a/alioth/src/virtio/dev/vsock/uds_vsock_test.rs +++ b/alioth/src/virtio/dev/vsock/uds_vsock_test.rs @@ -877,3 +877,100 @@ fn vsock_conn_request_eof_test() { notifier.notify().unwrap(); handle.join().unwrap(); } + +#[test] +fn vsock_conn_request_close_test() { + let ram_bus = Arc::new(fixture_ram_bus()); + let ram = ram_bus.lock_layout(); + let regs: Arc<[QueueReg]> = Arc::from(fixture_queues(3)); + let reg_tx = ®s[VsockVirtq::TX.raw() as usize]; + let reg_rx = ®s[VsockVirtq::RX.raw() as usize]; + let mut rx_q = GuestQueue::new( + SplitQueue::new(reg_rx, &ram, false).unwrap().unwrap(), + reg_rx, + ); + let mut tx_q = GuestQueue::new( + SplitQueue::new(reg_tx, &ram, false).unwrap().unwrap(), + reg_tx, + ); + + let temp_dir = TempDir::new().unwrap(); + let sock_path = temp_dir.path().join("vsock.sock"); + + const GUEST_CID: u32 = 3; + let param = UdsVsockSpec { + cid: GUEST_CID, + path: sock_path.clone().into(), + }; + let dev = param.build("vsock").unwrap(); + + let (tx, rx) = flume::unbounded(); + let (handle, notifier) = dev.spawn_worker(rx, ram_bus.clone(), regs).unwrap(); + let (irq_tx, irq_rx) = flume::unbounded(); + let irq_sender = Arc::new(FakeIrqSender { q_tx: irq_tx }); + let start_param = StartParam { + feature: VirtioFeature::VERSION_1.bits(), + irq_sender, + notifiers: Option::>::None, + }; + tx.send(WakeEvent::Start { param: start_param }).unwrap(); + + let rx_buf_addr = DATA_ADDR; + let tx_buf_addr = DATA_ADDR + 4096; + + // A client that sends a complete request and hangs up before the guest + // accepts the connection. + let buf_id = rx_q.add_desc(&[], &[(rx_buf_addr, 4096)]); + const H2G_GUEST_PORT: u32 = 1025; + let mut h2g_stream = UnixStream::connect(&sock_path).unwrap(); + h2g_stream + .write_all(format!("CONNECT {H2G_GUEST_PORT}\n").as_bytes()) + .unwrap(); + drop(h2g_stream); + + assert_eq!( + irq_rx.recv_timeout(Duration::from_secs(1)).unwrap(), + VsockVirtq::RX.raw() + ); + let used = rx_q.get_used().unwrap(); + assert_eq!(used.id, buf_id); + let mut hdr = VsockHeader::new_zeroed(); + ram.read(rx_buf_addr, hdr.as_mut_bytes()).unwrap(); + assert_eq!(hdr.op, VsockOp::REQUEST); + let h2g_host_port = hdr.src_port; + + // The guest accepts, but the host side is already gone. The device must + // report a reset instead of failing. + let rst_buf_id = rx_q.add_desc(&[], &[(rx_buf_addr, 4096)]); + let resp_hdr = VsockHeader { + src_cid: GUEST_CID, + dst_cid: VSOCK_CID_HOST, + src_port: H2G_GUEST_PORT, + dst_port: h2g_host_port, + op: VsockOp::RESPONSE, + type_: VsockType::STREAM, + ..Default::default() + }; + send_to_tx( + &resp_hdr, + &[], + &ram, + tx_buf_addr, + &mut tx_q, + &tx, + ¬ifier, + &irq_rx, + true, + ); + let used = rx_q.get_used().unwrap(); + assert_eq!(used.id, rst_buf_id); + assert_eq!(used.len as usize, size_of::()); + ram.read(rx_buf_addr, hdr.as_mut_bytes()).unwrap(); + assert_eq!(hdr.op, VsockOp::RST); + assert_eq!(hdr.src_port, h2g_host_port); + assert_eq!(hdr.dst_port, H2G_GUEST_PORT); + + tx.send(WakeEvent::Shutdown).unwrap(); + notifier.notify().unwrap(); + handle.join().unwrap(); +} From 3d5d005cc6a1f38057304cc6d5e6bb7643e257c6 Mon Sep 17 00:00:00 2001 From: Changyuan Lyu Date: Sun, 13 Sep 2026 00:36:50 -0700 Subject: [PATCH 4/4] test(vsock): add a fixture for the uds-vsock tests Every test repeated the same ~35 lines to set up guest memory, the queues, the device and the worker, and then open-coded the interrupt, descriptor and connection handshake steps. Add a VsockTest fixture that owns the running worker, the queues and the host socket path, with helpers for the recurring steps: offering RX descriptors, waiting for RX messages, sending TX messages, connecting a host client and completing the CONNECT handshake. Move the device metadata assertions of vsock_conn_test into a separate vsock_dev_test, as they need no worker. No change in coverage, the file shrinks from 977 to 571 lines. Assisted-by: Antigravity:Claude-Opus-5 Signed-off-by: Changyuan Lyu --- alioth/src/virtio/dev/vsock/uds_vsock_test.rs | 1005 +++++------------ 1 file changed, 300 insertions(+), 705 deletions(-) diff --git a/alioth/src/virtio/dev/vsock/uds_vsock_test.rs b/alioth/src/virtio/dev/vsock/uds_vsock_test.rs index 04a540d4..328bc4cd 100644 --- a/alioth/src/virtio/dev/vsock/uds_vsock_test.rs +++ b/alioth/src/virtio/dev/vsock/uds_vsock_test.rs @@ -15,8 +15,9 @@ use std::io::{BufRead, BufReader, ErrorKind, Read, Write}; use std::mem::size_of; use std::os::unix::net::{UnixListener, UnixStream}; +use std::path::PathBuf; use std::sync::Arc; -use std::thread; +use std::thread::{self, JoinHandle}; use std::time::Duration; use assert_matches::assert_matches; @@ -25,7 +26,7 @@ use tempfile::TempDir; use zerocopy::{FromBytes, FromZeros, IntoBytes}; use crate::mem::emulated::{Action, Mmio}; -use crate::mem::mapped::Ram; +use crate::mem::mapped::{Ram, RamBus}; use crate::sync::notifier::Notifier; use crate::virtio::dev::vsock::{ ShutdownFlag, UdsVsockSpec, VSOCK_CID_HOST, VsockConfig, VsockFeature, VsockHeader, VsockOp, @@ -34,10 +35,19 @@ use crate::virtio::dev::vsock::{ use crate::virtio::dev::{DevSpec, StartParam, Virtio, WakeEvent}; use crate::virtio::queue::QueueReg; use crate::virtio::queue::split::SplitQueue; -use crate::virtio::queue::tests::{GuestQueue, VirtQueueGuest}; +use crate::virtio::queue::tests::{GuestQueue, UsedDesc}; use crate::virtio::tests::{DATA_ADDR, FakeIrqSender, fixture_queues, fixture_ram_bus}; use crate::virtio::{DeviceId, FEATURE_BUILT_IN, VirtioFeature}; +const GUEST_CID: u32 = 3; +const HDR_SIZE: usize = size_of::(); +/// Guest memory used for RX descriptors. +const RX_ADDR: u64 = DATA_ADDR; +const RX_LEN: u32 = 4096; +/// Guest memory used for TX descriptors. +const TX_ADDR: u64 = DATA_ADDR + 4096; +const TIMEOUT: Duration = Duration::from_secs(1); + #[test] fn vsock_config_test() { let config = VsockConfig { @@ -49,80 +59,14 @@ fn vsock_config_test() { assert_matches!(config.write(0, 8, 0), Ok(Action::None)); } -#[allow(clippy::too_many_arguments)] -fn send_to_tx<'m, Q>( - hdr: &VsockHeader, - data: &[u8], - ram: &'m Ram, - buf_addr: u64, - q: &mut GuestQueue<'m, Q>, - tx: &Sender>, - notifier: &Notifier, - irq_rx: &Receiver, - expect_rx: bool, -) where - Q: VirtQueueGuest<'m>, -{ - let hdr_addr = buf_addr; - let data_addr = hdr_addr + size_of::() as u64; - let hdr_buf = hdr.as_bytes(); - ram.write(hdr_addr, hdr_buf).unwrap(); - if !data.is_empty() { - ram.write(data_addr, data).unwrap(); - } - let buf_id = q.add_desc( - &[ - (hdr_addr, size_of::() as u32), - (data_addr, data.len() as u32), - ], - &[], - ); - tx.send(WakeEvent::Notify { - q_index: VsockVirtq::TX.raw(), - }) - .unwrap(); - notifier.notify().unwrap(); - if expect_rx { - assert_eq!( - irq_rx.recv_timeout(Duration::from_secs(1)).unwrap(), - VsockVirtq::RX.raw() - ); - } - assert_eq!( - irq_rx.recv_timeout(Duration::from_secs(1)).unwrap(), - VsockVirtq::TX.raw() - ); - let used = q.get_used().unwrap(); - assert_eq!(used.id, buf_id); - assert_eq!(used.len, 0); -} - #[test] -fn vsock_conn_test() { - let ram_bus = Arc::new(fixture_ram_bus()); - let ram = ram_bus.lock_layout(); - let regs: Arc<[QueueReg]> = Arc::from(fixture_queues(3)); - let reg_tx = ®s[VsockVirtq::TX.raw() as usize]; - let reg_rx = ®s[VsockVirtq::RX.raw() as usize]; - let mut rx_q = GuestQueue::new( - SplitQueue::new(reg_rx, &ram, false).unwrap().unwrap(), - reg_rx, - ); - let mut tx_q = GuestQueue::new( - SplitQueue::new(reg_tx, &ram, false).unwrap().unwrap(), - reg_tx, - ); - +fn vsock_dev_test() { let temp_dir = TempDir::new().unwrap(); - let sock_path = temp_dir.path().join("vsock.sock"); - - const GUEST_CID: u32 = 3; let param = UdsVsockSpec { cid: GUEST_CID, - path: sock_path.clone().into(), + path: temp_dir.path().join("vsock.sock").into(), }; let dev = param.build("vsock").unwrap(); - assert_matches!(dev.id(), DeviceId::SOCKET); assert_eq!(dev.name(), "vsock"); assert_eq!(dev.num_queues(), 3); @@ -131,105 +75,239 @@ fn vsock_conn_test() { dev.feature(), VsockFeature::STREAM.bits() | FEATURE_BUILT_IN ); +} - let (tx, rx) = flume::unbounded(); - let (handle, notifier) = dev.spawn_worker(rx, ram_bus.clone(), regs).unwrap(); - let (irq_tx, irq_rx) = flume::unbounded(); - let irq_sender = Arc::new(FakeIrqSender { q_tx: irq_tx }); - let start_param = StartParam { - feature: VirtioFeature::VERSION_1.bits(), - irq_sender, - notifiers: Option::>::None, - }; - tx.send(WakeEvent::Start { param: start_param }).unwrap(); +/// Builds the header of a message sent by the guest to the host. +fn guest_hdr(op: VsockOp, guest_port: u32, host_port: u32) -> VsockHeader { + VsockHeader { + src_cid: GUEST_CID, + dst_cid: VSOCK_CID_HOST, + src_port: guest_port, + dst_port: host_port, + op, + type_: VsockType::STREAM, + ..Default::default() + } +} - let rx_buf_addr = DATA_ADDR; - let tx_buf_addr = DATA_ADDR + 4096; +/// A running `UdsVsock` worker together with the guest queues and the host +/// socket path, so that a test can act as both the guest and a host client. +struct VsockTest<'m> { + ram: &'m Ram, + sock_path: PathBuf, + rx_q: GuestQueue<'m, SplitQueue<'m>>, + tx_q: GuestQueue<'m, SplitQueue<'m>>, + tx: Sender>, + irq_rx: Receiver, + notifier: Arc, + handle: JoinHandle<()>, + _temp_dir: TempDir, +} - // 0. Setup connection - // 0.1 host-initiated connection - let mut h2g_stream = UnixStream::connect(&sock_path).unwrap(); - h2g_stream.set_nonblocking(true).unwrap(); +impl<'m> VsockTest<'m> { + /// Starts a device worker listening on a socket in a temporary directory. + fn new(ram_bus: &Arc, ram: &'m Ram) -> Self { + let regs: Arc<[QueueReg]> = Arc::from(fixture_queues(3)); + let reg_rx = ®s[VsockVirtq::RX.raw() as usize]; + let reg_tx = ®s[VsockVirtq::TX.raw() as usize]; + let rx_q = GuestQueue::new( + SplitQueue::new(reg_rx, ram, false).unwrap().unwrap(), + reg_rx, + ); + let tx_q = GuestQueue::new( + SplitQueue::new(reg_tx, ram, false).unwrap().unwrap(), + reg_tx, + ); - let buf_id = rx_q.add_desc(&[], &[(rx_buf_addr, 4096)]); - const H2G_GUEST_PORT: u32 = 1025; - writeln!(h2g_stream, "CONNECT {H2G_GUEST_PORT}").unwrap(); - assert_eq!( - irq_rx.recv_timeout(Duration::from_secs(1)).unwrap(), - VsockVirtq::RX.raw() - ); - let used = rx_q.get_used().unwrap(); - assert_eq!(used.id, buf_id); - assert_eq!(used.len as usize, size_of::()); + let temp_dir = TempDir::new().unwrap(); + let sock_path = temp_dir.path().join("vsock.sock"); + let param = UdsVsockSpec { + cid: GUEST_CID, + path: sock_path.clone().into(), + }; + let dev = param.build("vsock").unwrap(); + + let (tx, rx) = flume::unbounded(); + let (handle, notifier) = dev.spawn_worker(rx, ram_bus.clone(), regs).unwrap(); + let (irq_tx, irq_rx) = flume::unbounded(); + let start_param = StartParam { + feature: VirtioFeature::VERSION_1.bits(), + irq_sender: Arc::new(FakeIrqSender { q_tx: irq_tx }), + notifiers: Option::>::None, + }; + tx.send(WakeEvent::Start { param: start_param }).unwrap(); + + VsockTest { + ram, + sock_path, + rx_q, + tx_q, + tx, + irq_rx, + notifier, + handle, + _temp_dir: temp_dir, + } + } - let mut hdr = VsockHeader::new_zeroed(); - ram.read(rx_buf_addr, hdr.as_mut_bytes()).unwrap(); - assert_eq!(hdr.src_cid, VSOCK_CID_HOST); - assert_eq!(hdr.dst_cid, GUEST_CID); - assert_eq!(hdr.dst_port, H2G_GUEST_PORT); - assert_eq!(hdr.op, VsockOp::REQUEST); - assert_eq!(hdr.type_, VsockType::STREAM); + /// Offers `bufs` as one writable descriptor chain on the RX queue. + fn add_rx_chain(&mut self, bufs: &[(u64, u32)]) -> u16 { + self.rx_q.add_desc(&[], bufs) + } - let h2g_host_port = hdr.src_port; - let resp_hdr = VsockHeader { - src_cid: GUEST_CID, - dst_cid: VSOCK_CID_HOST, - src_port: H2G_GUEST_PORT, - dst_port: h2g_host_port, - op: VsockOp::RESPONSE, - type_: VsockType::STREAM, - ..Default::default() - }; - send_to_tx( - &resp_hdr, - &[], - &ram, - tx_buf_addr, - &mut tx_q, - &tx, - ¬ifier, - &irq_rx, - false, - ); - let mut reader = BufReader::new(&h2g_stream); - let mut line = String::new(); - reader.read_line(&mut line).unwrap(); - assert_eq!(line, format!("OK {h2g_host_port}\n")); + /// Offers a single writable RX buffer of `len` bytes at `addr`. + fn add_rx_desc_at(&mut self, addr: u64, len: u32) -> u16 { + self.add_rx_chain(&[(addr, len)]) + } + + /// Offers a single writable RX buffer covering the whole RX area. + fn add_rx_desc(&mut self) -> u16 { + self.add_rx_desc_at(RX_ADDR, RX_LEN) + } + + /// Tells the device that RX descriptors became available. + fn notify_rx(&self) { + self.tx + .send(WakeEvent::Notify { + q_index: VsockVirtq::RX.raw(), + }) + .unwrap(); + self.notifier.notify().unwrap(); + } + + fn assert_no_irq(&self) { + assert_eq!(self.irq_rx.try_recv(), Err(TryRecvError::Empty)); + } + + /// Waits for the device to signal the RX queue. + fn wait_rx_irq(&self) { + assert_eq!( + self.irq_rx.recv_timeout(TIMEOUT).unwrap(), + VsockVirtq::RX.raw() + ); + } + + /// Waits for the device to signal the RX queue and returns the descriptor + /// it used. + fn wait_rx_used(&mut self) -> UsedDesc { + self.wait_rx_irq(); + self.rx_q.get_used().unwrap() + } + + /// Takes a header-only message the device already placed in `buf_id`. + fn take_rx_hdr(&mut self, buf_id: u16) -> VsockHeader { + let used = self.rx_q.get_used().unwrap(); + assert_eq!(used.id, buf_id); + assert_eq!(used.len as usize, HDR_SIZE); + self.read_hdr(RX_ADDR) + } + + /// Waits for a header-only message in `buf_id` and returns it. + fn wait_rx_hdr(&mut self, buf_id: u16) -> VsockHeader { + self.wait_rx_irq(); + self.take_rx_hdr(buf_id) + } + + fn read_hdr(&self, addr: u64) -> VsockHeader { + let mut hdr = VsockHeader::new_zeroed(); + self.ram.read(addr, hdr.as_mut_bytes()).unwrap(); + hdr + } + + /// Sends `hdr` followed by `data` to the device on the TX queue. + /// `expect_rx` tells whether the device answers on the RX queue while + /// handling the message. + fn send_to_tx(&mut self, hdr: &VsockHeader, data: &[u8], expect_rx: bool) { + let data_addr = TX_ADDR + HDR_SIZE as u64; + self.ram.write(TX_ADDR, hdr.as_bytes()).unwrap(); + if !data.is_empty() { + self.ram.write(data_addr, data).unwrap(); + } + let buf_id = self.tx_q.add_desc( + &[(TX_ADDR, HDR_SIZE as u32), (data_addr, data.len() as u32)], + &[], + ); + self.tx + .send(WakeEvent::Notify { + q_index: VsockVirtq::TX.raw(), + }) + .unwrap(); + self.notifier.notify().unwrap(); + if expect_rx { + self.wait_rx_irq(); + } + assert_eq!( + self.irq_rx.recv_timeout(TIMEOUT).unwrap(), + VsockVirtq::TX.raw() + ); + let used = self.tx_q.get_used().unwrap(); + assert_eq!(used.id, buf_id); + assert_eq!(used.len, 0); + } + + /// Connects a host client to the device socket. + fn connect(&self) -> UnixStream { + let stream = UnixStream::connect(&self.sock_path).unwrap(); + stream.set_nonblocking(true).unwrap(); + stream + } + + /// Connects a host client asking for `guest_port` and consumes the + /// REQUEST the device sends to the guest. Returns the client socket and + /// the host port assigned by the device. + fn request_conn(&mut self, guest_port: u32) -> (UnixStream, u32) { + let buf_id = self.add_rx_desc(); + let mut stream = self.connect(); + writeln!(stream, "CONNECT {guest_port}").unwrap(); + let hdr = self.wait_rx_hdr(buf_id); + assert_eq!(hdr.src_cid, VSOCK_CID_HOST); + assert_eq!(hdr.dst_cid, GUEST_CID); + assert_eq!(hdr.dst_port, guest_port); + assert_eq!(hdr.op, VsockOp::REQUEST); + assert_eq!(hdr.type_, VsockType::STREAM); + (stream, hdr.src_port) + } + + /// Accepts a host-initiated connection on behalf of the guest and checks + /// the acknowledgement the client receives. + fn accept_conn(&mut self, stream: &UnixStream, guest_port: u32, host_port: u32) { + let resp_hdr = guest_hdr(VsockOp::RESPONSE, guest_port, host_port); + self.send_to_tx(&resp_hdr, &[], false); + let mut line = String::new(); + BufReader::new(stream).read_line(&mut line).unwrap(); + assert_eq!(line, format!("OK {host_port}\n")); + } + + /// Stops the worker and waits for it to exit. + fn shutdown(self) { + self.tx.send(WakeEvent::Shutdown).unwrap(); + self.notifier.notify().unwrap(); + self.handle.join().unwrap(); + } +} + +#[test] +fn vsock_conn_test() { + let ram_bus = Arc::new(fixture_ram_bus()); + let ram = ram_bus.lock_layout(); + let mut t = VsockTest::new(&ram_bus, &ram); + + // 0. Setup connection + // 0.1 host-initiated connection + const H2G_GUEST_PORT: u32 = 1025; + let (mut h2g_stream, h2g_host_port) = t.request_conn(H2G_GUEST_PORT); + t.accept_conn(&h2g_stream, H2G_GUEST_PORT, h2g_host_port); // 0.2 guest-initiated connection const G2H_HOST_PORT: u32 = 8706; const G2H_GUEST_PORT: u32 = 8707; - let listener_path = format!("{}_{G2H_HOST_PORT}", sock_path.to_string_lossy()); + let listener_path = format!("{}_{G2H_HOST_PORT}", t.sock_path.to_string_lossy()); let listener = UnixListener::bind(&listener_path).unwrap(); listener.set_nonblocking(true).unwrap(); - let rx_buf_id = rx_q.add_desc(&[], &[(rx_buf_addr, 4096)]); - let request_hdr = VsockHeader { - src_cid: GUEST_CID, - dst_cid: VSOCK_CID_HOST, - src_port: G2H_GUEST_PORT, - dst_port: G2H_HOST_PORT, - op: VsockOp::REQUEST, - len: 0, - type_: VsockType::STREAM, - ..Default::default() - }; - send_to_tx( - &request_hdr, - &[], - &ram, - tx_buf_addr, - &mut tx_q, - &tx, - ¬ifier, - &irq_rx, - true, - ); - let used = rx_q.get_used().unwrap(); - assert_eq!(used.id, rx_buf_id); - assert_eq!(used.len as usize, size_of::()); - - let mut hdr = VsockHeader::new_zeroed(); - ram.read(rx_buf_addr, hdr.as_mut_bytes()).unwrap(); + let rx_buf_id = t.add_rx_desc(); + let request_hdr = guest_hdr(VsockOp::REQUEST, G2H_GUEST_PORT, G2H_HOST_PORT); + t.send_to_tx(&request_hdr, &[], true); + let hdr = t.take_rx_hdr(rx_buf_id); assert_eq!(hdr.src_cid, VSOCK_CID_HOST); assert_eq!(hdr.dst_cid, GUEST_CID); assert_eq!(hdr.src_port, G2H_HOST_PORT); @@ -242,35 +320,20 @@ fn vsock_conn_test() { // 1. Host to Guest via guest-initiated connection let h2g_data = "hello from host"; - let buf_id = rx_q.add_desc( - &[], - &[ - (rx_buf_addr, 32), - (rx_buf_addr + 32, 32), - (rx_buf_addr + 64, 32), - ], - ); - tx.send(WakeEvent::Notify { - q_index: VsockVirtq::RX.raw(), - }) - .unwrap(); - notifier.notify().unwrap(); - assert_eq!(irq_rx.try_recv(), Err(TryRecvError::Empty)); + let buf_id = t.add_rx_chain(&[(RX_ADDR, 32), (RX_ADDR + 32, 32), (RX_ADDR + 64, 32)]); + t.notify_rx(); + t.assert_no_irq(); g2h_stream.write_all(h2g_data.as_bytes()).unwrap(); g2h_stream.flush().unwrap(); - assert_eq!( - irq_rx.recv_timeout(Duration::from_secs(1)).unwrap(), - VsockVirtq::RX.raw() - ); - let used = rx_q.get_used().unwrap(); + let used = t.wait_rx_used(); assert_eq!(used.id, buf_id); - let total_len = size_of::() + h2g_data.len(); + let total_len = HDR_SIZE + h2g_data.len(); assert_eq!(used.len, total_len as u32); let mut h2g_buf = vec![0; total_len]; - ram.read(rx_buf_addr, &mut h2g_buf).unwrap(); - let (h2g_hdr_buf, h2g_data_buf) = h2g_buf.split_at(size_of::()); + ram.read(RX_ADDR, &mut h2g_buf).unwrap(); + let (h2g_hdr_buf, h2g_data_buf) = h2g_buf.split_at(HDR_SIZE); let h2g_hdr = VsockHeader::read_from_bytes(h2g_hdr_buf).unwrap(); assert_eq!(h2g_hdr.src_port, G2H_HOST_PORT); assert_eq!(h2g_hdr.dst_port, G2H_GUEST_PORT); @@ -281,26 +344,10 @@ fn vsock_conn_test() { // 2. Guest to Host via host-initiated connection let g2h_data = "hello from guest"; let g2h_hdr = VsockHeader { - src_cid: GUEST_CID, - dst_cid: VSOCK_CID_HOST, - src_port: H2G_GUEST_PORT, - dst_port: h2g_host_port, - op: VsockOp::RW, len: g2h_data.len() as u32, - type_: VsockType::STREAM, - ..Default::default() + ..guest_hdr(VsockOp::RW, H2G_GUEST_PORT, h2g_host_port) }; - send_to_tx( - &g2h_hdr, - g2h_data.as_bytes(), - &ram, - tx_buf_addr, - &mut tx_q, - &tx, - ¬ifier, - &irq_rx, - false, - ); + t.send_to_tx(&g2h_hdr, g2h_data.as_bytes(), false); let mut g2h_read_buf = vec![0; g2h_data.len()]; let _ = h2g_stream.read(&mut g2h_read_buf).unwrap(); assert_eq!(String::from_utf8_lossy(&g2h_read_buf), g2h_data); @@ -308,190 +355,46 @@ fn vsock_conn_test() { // 3. Shutdown host-initiated connection // 3.1 Send ShutdownFlag::RECEIVE let shutdown_hdr = VsockHeader { - src_cid: GUEST_CID, - dst_cid: VSOCK_CID_HOST, - src_port: H2G_GUEST_PORT, - dst_port: h2g_host_port, - op: VsockOp::SHUTDOWN, - len: 0, - type_: VsockType::STREAM, flags: ShutdownFlag::RECEIVE.bits(), - ..Default::default() + ..guest_hdr(VsockOp::SHUTDOWN, H2G_GUEST_PORT, h2g_host_port) }; - send_to_tx( - &shutdown_hdr, - &[], - &ram, - tx_buf_addr, - &mut tx_q, - &tx, - ¬ifier, - &irq_rx, - false, - ); + t.send_to_tx(&shutdown_hdr, &[], false); let mut buf = [0u8; 8]; assert_matches!(h2g_stream.read(&mut buf), Err(e) if e.kind() == ErrorKind::WouldBlock); // 3.2 Send ShutdownFlag::SEND let shutdown_hdr = VsockHeader { - src_cid: GUEST_CID, - dst_cid: VSOCK_CID_HOST, - src_port: H2G_GUEST_PORT, - dst_port: h2g_host_port, - op: VsockOp::SHUTDOWN, - len: 0, - type_: VsockType::STREAM, flags: ShutdownFlag::SEND.bits(), - ..Default::default() + ..guest_hdr(VsockOp::SHUTDOWN, H2G_GUEST_PORT, h2g_host_port) }; - send_to_tx( - &shutdown_hdr, - &[], - &ram, - tx_buf_addr, - &mut tx_q, - &tx, - ¬ifier, - &irq_rx, - false, - ); + t.send_to_tx(&shutdown_hdr, &[], false); assert_matches!(h2g_stream.read(&mut buf), Ok(0)); // 4. Reset guest-initiated connection - let reset_hdr = VsockHeader { - src_cid: GUEST_CID, - dst_cid: VSOCK_CID_HOST, - src_port: G2H_GUEST_PORT, - dst_port: G2H_HOST_PORT, - op: VsockOp::RST, - len: 0, - type_: VsockType::STREAM, - flags: 0, - ..Default::default() - }; - send_to_tx( - &reset_hdr, - &[], - &ram, - tx_buf_addr, - &mut tx_q, - &tx, - ¬ifier, - &irq_rx, - false, - ); + let reset_hdr = guest_hdr(VsockOp::RST, G2H_GUEST_PORT, G2H_HOST_PORT); + t.send_to_tx(&reset_hdr, &[], false); assert_matches!(g2h_stream.read(&mut buf), Ok(0)); - tx.send(WakeEvent::Shutdown).unwrap(); - notifier.notify().unwrap(); - handle.join().unwrap(); + t.shutdown(); } #[test] fn vsock_host_close_test() { let ram_bus = Arc::new(fixture_ram_bus()); let ram = ram_bus.lock_layout(); - let regs: Arc<[QueueReg]> = Arc::from(fixture_queues(3)); - let reg_tx = ®s[VsockVirtq::TX.raw() as usize]; - let reg_rx = ®s[VsockVirtq::RX.raw() as usize]; - let mut rx_q = GuestQueue::new( - SplitQueue::new(reg_rx, &ram, false).unwrap().unwrap(), - reg_rx, - ); - let mut tx_q = GuestQueue::new( - SplitQueue::new(reg_tx, &ram, false).unwrap().unwrap(), - reg_tx, - ); - - let temp_dir = TempDir::new().unwrap(); - let sock_path = temp_dir.path().join("vsock.sock"); - - const GUEST_CID: u32 = 3; - let param = UdsVsockSpec { - cid: GUEST_CID, - path: sock_path.clone().into(), - }; - let dev = param.build("vsock").unwrap(); - - let (tx, rx) = flume::unbounded(); - let (handle, notifier) = dev.spawn_worker(rx, ram_bus.clone(), regs).unwrap(); - let (irq_tx, irq_rx) = flume::unbounded(); - let irq_sender = Arc::new(FakeIrqSender { q_tx: irq_tx }); - let start_param = StartParam { - feature: VirtioFeature::VERSION_1.bits(), - irq_sender, - notifiers: Option::>::None, - }; - tx.send(WakeEvent::Start { param: start_param }).unwrap(); - - let rx_buf_addr = DATA_ADDR; - let tx_buf_addr = DATA_ADDR + 4096; + let mut t = VsockTest::new(&ram_bus, &ram); // Establish a host-initiated connection - let mut h2g_stream = UnixStream::connect(&sock_path).unwrap(); - h2g_stream.set_nonblocking(true).unwrap(); - - let buf_id = rx_q.add_desc(&[], &[(rx_buf_addr, 4096)]); const H2G_GUEST_PORT: u32 = 1025; - writeln!(h2g_stream, "CONNECT {H2G_GUEST_PORT}").unwrap(); - assert_eq!( - irq_rx.recv_timeout(Duration::from_secs(1)).unwrap(), - VsockVirtq::RX.raw() - ); - let used = rx_q.get_used().unwrap(); - assert_eq!(used.id, buf_id); - assert_eq!(used.len as usize, size_of::()); - - let mut hdr = VsockHeader::new_zeroed(); - ram.read(rx_buf_addr, hdr.as_mut_bytes()).unwrap(); - assert_eq!(hdr.op, VsockOp::REQUEST); - let h2g_host_port = hdr.src_port; - - let resp_hdr = VsockHeader { - src_cid: GUEST_CID, - dst_cid: VSOCK_CID_HOST, - src_port: H2G_GUEST_PORT, - dst_port: h2g_host_port, - op: VsockOp::RESPONSE, - type_: VsockType::STREAM, - ..Default::default() - }; - send_to_tx( - &resp_hdr, - &[], - &ram, - tx_buf_addr, - &mut tx_q, - &tx, - ¬ifier, - &irq_rx, - false, - ); - let mut reader = BufReader::new(&h2g_stream); - let mut line = String::new(); - reader.read_line(&mut line).unwrap(); - assert_eq!(line, format!("OK {h2g_host_port}\n")); + let (h2g_stream, h2g_host_port) = t.request_conn(H2G_GUEST_PORT); + t.accept_conn(&h2g_stream, H2G_GUEST_PORT, h2g_host_port); // Provide RX descriptor first, then close host socket - let rx_buf_id = rx_q.add_desc(&[], &[(rx_buf_addr, 4096)]); - tx.send(WakeEvent::Notify { - q_index: VsockVirtq::RX.raw(), - }) - .unwrap(); - notifier.notify().unwrap(); - + let rx_buf_id = t.add_rx_desc(); + t.notify_rx(); drop(h2g_stream); // EOF to alioth // Verify guest receives RST - assert_eq!( - irq_rx.recv_timeout(Duration::from_secs(1)).unwrap(), - VsockVirtq::RX.raw() - ); - let used = rx_q.get_used().unwrap(); - assert_eq!(used.id, rx_buf_id); - assert_eq!(used.len as usize, size_of::()); - - let mut hdr = VsockHeader::new_zeroed(); - ram.read(rx_buf_addr, hdr.as_mut_bytes()).unwrap(); + let hdr = t.wait_rx_hdr(rx_buf_id); assert_eq!(hdr.src_cid, VSOCK_CID_HOST); assert_eq!(hdr.dst_cid, GUEST_CID); assert_eq!(hdr.src_port, h2g_host_port); @@ -499,95 +402,19 @@ fn vsock_host_close_test() { assert_eq!(hdr.op, VsockOp::RST); assert_eq!(hdr.type_, VsockType::STREAM); - tx.send(WakeEvent::Shutdown).unwrap(); - notifier.notify().unwrap(); - handle.join().unwrap(); + t.shutdown(); } #[test] fn vsock_host_close_no_desc_test() { let ram_bus = Arc::new(fixture_ram_bus()); let ram = ram_bus.lock_layout(); - let regs: Arc<[QueueReg]> = Arc::from(fixture_queues(3)); - let reg_tx = ®s[VsockVirtq::TX.raw() as usize]; - let reg_rx = ®s[VsockVirtq::RX.raw() as usize]; - let mut rx_q = GuestQueue::new( - SplitQueue::new(reg_rx, &ram, false).unwrap().unwrap(), - reg_rx, - ); - let mut tx_q = GuestQueue::new( - SplitQueue::new(reg_tx, &ram, false).unwrap().unwrap(), - reg_tx, - ); - - let temp_dir = TempDir::new().unwrap(); - let sock_path = temp_dir.path().join("vsock.sock"); - - const GUEST_CID: u32 = 3; - let param = UdsVsockSpec { - cid: GUEST_CID, - path: sock_path.clone().into(), - }; - let dev = param.build("vsock").unwrap(); - - let (tx, rx) = flume::unbounded(); - let (handle, notifier) = dev.spawn_worker(rx, ram_bus.clone(), regs).unwrap(); - let (irq_tx, irq_rx) = flume::unbounded(); - let irq_sender = Arc::new(FakeIrqSender { q_tx: irq_tx }); - let start_param = StartParam { - feature: VirtioFeature::VERSION_1.bits(), - irq_sender, - notifiers: Option::>::None, - }; - tx.send(WakeEvent::Start { param: start_param }).unwrap(); - - let rx_buf_addr = DATA_ADDR; - let tx_buf_addr = DATA_ADDR + 4096; + let mut t = VsockTest::new(&ram_bus, &ram); // Establish a host-initiated connection - let mut h2g_stream = UnixStream::connect(&sock_path).unwrap(); - h2g_stream.set_nonblocking(true).unwrap(); - - let buf_id = rx_q.add_desc(&[], &[(rx_buf_addr, 4096)]); const H2G_GUEST_PORT: u32 = 1025; - writeln!(h2g_stream, "CONNECT {H2G_GUEST_PORT}").unwrap(); - assert_eq!( - irq_rx.recv_timeout(Duration::from_secs(1)).unwrap(), - VsockVirtq::RX.raw() - ); - let used = rx_q.get_used().unwrap(); - assert_eq!(used.id, buf_id); - assert_eq!(used.len as usize, size_of::()); - - let mut hdr = VsockHeader::new_zeroed(); - ram.read(rx_buf_addr, hdr.as_mut_bytes()).unwrap(); - assert_eq!(hdr.op, VsockOp::REQUEST); - let h2g_host_port = hdr.src_port; - - let resp_hdr = VsockHeader { - src_cid: GUEST_CID, - dst_cid: VSOCK_CID_HOST, - src_port: H2G_GUEST_PORT, - dst_port: h2g_host_port, - op: VsockOp::RESPONSE, - type_: VsockType::STREAM, - ..Default::default() - }; - send_to_tx( - &resp_hdr, - &[], - &ram, - tx_buf_addr, - &mut tx_q, - &tx, - ¬ifier, - &irq_rx, - false, - ); - let mut reader = BufReader::new(&h2g_stream); - let mut line = String::new(); - reader.read_line(&mut line).unwrap(); - assert_eq!(line, format!("OK {h2g_host_port}\n")); + let (mut h2g_stream, h2g_host_port) = t.request_conn(H2G_GUEST_PORT); + t.accept_conn(&h2g_stream, H2G_GUEST_PORT, h2g_host_port); // Write data and close the host socket WITHOUT providing an RX descriptor. // The data must be delivered before the guest sees the final RST. @@ -600,87 +427,35 @@ fn vsock_host_close_no_desc_test() { thread::sleep(Duration::from_millis(50)); // The first descriptor drains the host data. - let data_buf_id = rx_q.add_desc(&[], &[(rx_buf_addr, 4096)]); - tx.send(WakeEvent::Notify { - q_index: VsockVirtq::RX.raw(), - }) - .unwrap(); - notifier.notify().unwrap(); - - assert_eq!( - irq_rx.recv_timeout(Duration::from_secs(1)).unwrap(), - VsockVirtq::RX.raw() - ); - let used = rx_q.get_used().unwrap(); + let data_buf_id = t.add_rx_desc(); + t.notify_rx(); + let used = t.wait_rx_used(); assert_eq!(used.id, data_buf_id); - assert_eq!(used.len as usize, size_of::() + DATA.len()); - let mut hdr = VsockHeader::new_zeroed(); - ram.read(rx_buf_addr, hdr.as_mut_bytes()).unwrap(); + assert_eq!(used.len as usize, HDR_SIZE + DATA.len()); + let hdr = t.read_hdr(RX_ADDR); assert_eq!(hdr.op, VsockOp::RW); assert_eq!(hdr.len as usize, DATA.len()); let mut data = vec![0; DATA.len()]; - ram.read(rx_buf_addr + size_of::() as u64, &mut data) - .unwrap(); + ram.read(RX_ADDR + HDR_SIZE as u64, &mut data).unwrap(); assert_eq!(data, DATA); // The next descriptor observes EOF and receives the final RST. - let rst_buf_id = rx_q.add_desc(&[], &[(rx_buf_addr, 4096)]); - tx.send(WakeEvent::Notify { - q_index: VsockVirtq::RX.raw(), - }) - .unwrap(); - notifier.notify().unwrap(); - assert_eq!( - irq_rx.recv_timeout(Duration::from_secs(1)).unwrap(), - VsockVirtq::RX.raw() - ); - let used = rx_q.get_used().unwrap(); - assert_eq!(used.id, rst_buf_id); - assert_eq!(used.len as usize, size_of::()); - ram.read(rx_buf_addr, hdr.as_mut_bytes()).unwrap(); + let rst_buf_id = t.add_rx_desc(); + t.notify_rx(); + let hdr = t.wait_rx_hdr(rst_buf_id); assert_eq!(hdr.op, VsockOp::RST); - tx.send(WakeEvent::Shutdown).unwrap(); - notifier.notify().unwrap(); - handle.join().unwrap(); + t.shutdown(); } #[test] fn vsock_partial_conn_request_test() { let ram_bus = Arc::new(fixture_ram_bus()); let ram = ram_bus.lock_layout(); - let regs: Arc<[QueueReg]> = Arc::from(fixture_queues(3)); - let reg_rx = ®s[VsockVirtq::RX.raw() as usize]; - let mut rx_q = GuestQueue::new( - SplitQueue::new(reg_rx, &ram, false).unwrap().unwrap(), - reg_rx, - ); + let mut t = VsockTest::new(&ram_bus, &ram); - let temp_dir = TempDir::new().unwrap(); - let sock_path = temp_dir.path().join("vsock.sock"); - - const GUEST_CID: u32 = 3; - let param = UdsVsockSpec { - cid: GUEST_CID, - path: sock_path.clone().into(), - }; - let dev = param.build("vsock").unwrap(); - - let (tx, rx) = flume::unbounded(); - let (handle, notifier) = dev.spawn_worker(rx, ram_bus.clone(), regs).unwrap(); - let (irq_tx, irq_rx) = flume::unbounded(); - let irq_sender = Arc::new(FakeIrqSender { q_tx: irq_tx }); - let start_param = StartParam { - feature: VirtioFeature::VERSION_1.bits(), - irq_sender, - notifiers: Option::>::None, - }; - tx.send(WakeEvent::Start { param: start_param }).unwrap(); - - let rx_buf_addr = DATA_ADDR; - - let mut h2g_stream = UnixStream::connect(&sock_path).unwrap(); - let buf_id = rx_q.add_desc(&[], &[(rx_buf_addr, 4096)]); + let mut h2g_stream = t.connect(); + let buf_id = t.add_rx_desc(); // A connection request can be split over multiple writes, e.g. as done by // `writeln!()`. The device must wait for the complete line instead of @@ -692,68 +467,31 @@ fn vsock_partial_conn_request_test() { .write_all(format!("{H2G_GUEST_PORT}\n").as_bytes()) .unwrap(); - assert_eq!( - irq_rx.recv_timeout(Duration::from_secs(1)).unwrap(), - VsockVirtq::RX.raw() - ); - let used = rx_q.get_used().unwrap(); - assert_eq!(used.id, buf_id); - assert_eq!(used.len as usize, size_of::()); - - let mut hdr = VsockHeader::new_zeroed(); - ram.read(rx_buf_addr, hdr.as_mut_bytes()).unwrap(); + let hdr = t.wait_rx_hdr(buf_id); assert_eq!(hdr.src_cid, VSOCK_CID_HOST); assert_eq!(hdr.dst_cid, GUEST_CID); assert_eq!(hdr.dst_port, H2G_GUEST_PORT); assert_eq!(hdr.op, VsockOp::REQUEST); assert_eq!(hdr.type_, VsockType::STREAM); - tx.send(WakeEvent::Shutdown).unwrap(); - notifier.notify().unwrap(); - handle.join().unwrap(); + t.shutdown(); } #[test] fn vsock_simultaneous_conn_test() { let ram_bus = Arc::new(fixture_ram_bus()); let ram = ram_bus.lock_layout(); - let regs: Arc<[QueueReg]> = Arc::from(fixture_queues(3)); - let reg_rx = ®s[VsockVirtq::RX.raw() as usize]; - let mut rx_q = GuestQueue::new( - SplitQueue::new(reg_rx, &ram, false).unwrap().unwrap(), - reg_rx, - ); - - let temp_dir = TempDir::new().unwrap(); - let sock_path = temp_dir.path().join("vsock.sock"); + let mut t = VsockTest::new(&ram_bus, &ram); - const GUEST_CID: u32 = 3; - let param = UdsVsockSpec { - cid: GUEST_CID, - path: sock_path.clone().into(), - }; - let dev = param.build("vsock").unwrap(); - - let (tx, rx) = flume::unbounded(); - let (handle, notifier) = dev.spawn_worker(rx, ram_bus.clone(), regs).unwrap(); - let (irq_tx, irq_rx) = flume::unbounded(); - let irq_sender = Arc::new(FakeIrqSender { q_tx: irq_tx }); - let start_param = StartParam { - feature: VirtioFeature::VERSION_1.bits(), - irq_sender, - notifiers: Option::>::None, - }; - tx.send(WakeEvent::Start { param: start_param }).unwrap(); - - let buf_addrs = [DATA_ADDR, DATA_ADDR + 2048]; - let buf_ids = buf_addrs.map(|addr| rx_q.add_desc(&[], &[(addr, 2048)])); + let buf_addrs = [RX_ADDR, RX_ADDR + 2048]; + let buf_ids = buf_addrs.map(|addr| t.add_rx_desc_at(addr, 2048)); // Two clients connecting back to back pile up in the listener backlog, // and both must be served. const GUEST_PORTS: [u32; 2] = [1025, 1026]; let mut streams = Vec::new(); for port in GUEST_PORTS { - let mut stream = UnixStream::connect(&sock_path).unwrap(); + let mut stream = t.connect(); stream .write_all(format!("CONNECT {port}\n").as_bytes()) .unwrap(); @@ -762,215 +500,72 @@ fn vsock_simultaneous_conn_test() { let mut requests = Vec::new(); for _ in GUEST_PORTS { - irq_rx.recv_timeout(Duration::from_secs(1)).unwrap(); - let used = rx_q.get_used().unwrap(); - assert_eq!(used.len as usize, size_of::()); + let used = t.wait_rx_used(); + assert_eq!(used.len as usize, HDR_SIZE); let index = buf_ids.iter().position(|id| *id == used.id).unwrap(); - let mut hdr = VsockHeader::new_zeroed(); - ram.read(buf_addrs[index], hdr.as_mut_bytes()).unwrap(); + let hdr = t.read_hdr(buf_addrs[index]); assert_eq!(hdr.op, VsockOp::REQUEST); requests.push(hdr.dst_port); } requests.sort_unstable(); assert_eq!(requests, GUEST_PORTS); - tx.send(WakeEvent::Shutdown).unwrap(); - notifier.notify().unwrap(); - handle.join().unwrap(); + t.shutdown(); } #[test] fn vsock_conn_request_eof_test() { let ram_bus = Arc::new(fixture_ram_bus()); let ram = ram_bus.lock_layout(); - let regs: Arc<[QueueReg]> = Arc::from(fixture_queues(3)); - let reg_tx = ®s[VsockVirtq::TX.raw() as usize]; - let reg_rx = ®s[VsockVirtq::RX.raw() as usize]; - let mut rx_q = GuestQueue::new( - SplitQueue::new(reg_rx, &ram, false).unwrap().unwrap(), - reg_rx, - ); - let mut tx_q = GuestQueue::new( - SplitQueue::new(reg_tx, &ram, false).unwrap().unwrap(), - reg_tx, - ); - - let temp_dir = TempDir::new().unwrap(); - let sock_path = temp_dir.path().join("vsock.sock"); - - const GUEST_CID: u32 = 3; - let param = UdsVsockSpec { - cid: GUEST_CID, - path: sock_path.clone().into(), - }; - let dev = param.build("vsock").unwrap(); - - let (tx, rx) = flume::unbounded(); - let (handle, notifier) = dev.spawn_worker(rx, ram_bus.clone(), regs).unwrap(); - let (irq_tx, irq_rx) = flume::unbounded(); - let irq_sender = Arc::new(FakeIrqSender { q_tx: irq_tx }); - let start_param = StartParam { - feature: VirtioFeature::VERSION_1.bits(), - irq_sender, - notifiers: Option::>::None, - }; - tx.send(WakeEvent::Start { param: start_param }).unwrap(); - - let rx_buf_addr = DATA_ADDR; - let tx_buf_addr = DATA_ADDR + 4096; + let mut t = VsockTest::new(&ram_bus, &ram); // A client that connects and disconnects without saying anything. - drop(UnixStream::connect(&sock_path).unwrap()); + drop(t.connect()); // A client that disconnects in the middle of a connection request. - let mut partial = UnixStream::connect(&sock_path).unwrap(); + let mut partial = t.connect(); partial.write_all(b"CONNECT ").unwrap(); thread::sleep(Duration::from_millis(50)); drop(partial); thread::sleep(Duration::from_millis(50)); // Neither takes down the device: a well-behaved client still works. - let mut h2g_stream = UnixStream::connect(&sock_path).unwrap(); - let buf_id = rx_q.add_desc(&[], &[(rx_buf_addr, 4096)]); const H2G_GUEST_PORT: u32 = 1025; - writeln!(h2g_stream, "CONNECT {H2G_GUEST_PORT}").unwrap(); - assert_eq!( - irq_rx.recv_timeout(Duration::from_secs(1)).unwrap(), - VsockVirtq::RX.raw() - ); - let used = rx_q.get_used().unwrap(); - assert_eq!(used.id, buf_id); - assert_eq!(used.len as usize, size_of::()); + let (h2g_stream, h2g_host_port) = t.request_conn(H2G_GUEST_PORT); + t.accept_conn(&h2g_stream, H2G_GUEST_PORT, h2g_host_port); - let mut hdr = VsockHeader::new_zeroed(); - ram.read(rx_buf_addr, hdr.as_mut_bytes()).unwrap(); - assert_eq!(hdr.op, VsockOp::REQUEST); - assert_eq!(hdr.dst_port, H2G_GUEST_PORT); - let h2g_host_port = hdr.src_port; - - let resp_hdr = VsockHeader { - src_cid: GUEST_CID, - dst_cid: VSOCK_CID_HOST, - src_port: H2G_GUEST_PORT, - dst_port: h2g_host_port, - op: VsockOp::RESPONSE, - type_: VsockType::STREAM, - ..Default::default() - }; - send_to_tx( - &resp_hdr, - &[], - &ram, - tx_buf_addr, - &mut tx_q, - &tx, - ¬ifier, - &irq_rx, - false, - ); - let mut reader = BufReader::new(&h2g_stream); - let mut line = String::new(); - reader.read_line(&mut line).unwrap(); - assert_eq!(line, format!("OK {h2g_host_port}\n")); - - tx.send(WakeEvent::Shutdown).unwrap(); - notifier.notify().unwrap(); - handle.join().unwrap(); + t.shutdown(); } #[test] fn vsock_conn_request_close_test() { let ram_bus = Arc::new(fixture_ram_bus()); let ram = ram_bus.lock_layout(); - let regs: Arc<[QueueReg]> = Arc::from(fixture_queues(3)); - let reg_tx = ®s[VsockVirtq::TX.raw() as usize]; - let reg_rx = ®s[VsockVirtq::RX.raw() as usize]; - let mut rx_q = GuestQueue::new( - SplitQueue::new(reg_rx, &ram, false).unwrap().unwrap(), - reg_rx, - ); - let mut tx_q = GuestQueue::new( - SplitQueue::new(reg_tx, &ram, false).unwrap().unwrap(), - reg_tx, - ); - - let temp_dir = TempDir::new().unwrap(); - let sock_path = temp_dir.path().join("vsock.sock"); - - const GUEST_CID: u32 = 3; - let param = UdsVsockSpec { - cid: GUEST_CID, - path: sock_path.clone().into(), - }; - let dev = param.build("vsock").unwrap(); - - let (tx, rx) = flume::unbounded(); - let (handle, notifier) = dev.spawn_worker(rx, ram_bus.clone(), regs).unwrap(); - let (irq_tx, irq_rx) = flume::unbounded(); - let irq_sender = Arc::new(FakeIrqSender { q_tx: irq_tx }); - let start_param = StartParam { - feature: VirtioFeature::VERSION_1.bits(), - irq_sender, - notifiers: Option::>::None, - }; - tx.send(WakeEvent::Start { param: start_param }).unwrap(); - - let rx_buf_addr = DATA_ADDR; - let tx_buf_addr = DATA_ADDR + 4096; + let mut t = VsockTest::new(&ram_bus, &ram); // A client that sends a complete request and hangs up before the guest // accepts the connection. - let buf_id = rx_q.add_desc(&[], &[(rx_buf_addr, 4096)]); + let buf_id = t.add_rx_desc(); const H2G_GUEST_PORT: u32 = 1025; - let mut h2g_stream = UnixStream::connect(&sock_path).unwrap(); + let mut h2g_stream = t.connect(); h2g_stream .write_all(format!("CONNECT {H2G_GUEST_PORT}\n").as_bytes()) .unwrap(); drop(h2g_stream); - assert_eq!( - irq_rx.recv_timeout(Duration::from_secs(1)).unwrap(), - VsockVirtq::RX.raw() - ); - let used = rx_q.get_used().unwrap(); - assert_eq!(used.id, buf_id); - let mut hdr = VsockHeader::new_zeroed(); - ram.read(rx_buf_addr, hdr.as_mut_bytes()).unwrap(); + let hdr = t.wait_rx_hdr(buf_id); assert_eq!(hdr.op, VsockOp::REQUEST); let h2g_host_port = hdr.src_port; // The guest accepts, but the host side is already gone. The device must // report a reset instead of failing. - let rst_buf_id = rx_q.add_desc(&[], &[(rx_buf_addr, 4096)]); - let resp_hdr = VsockHeader { - src_cid: GUEST_CID, - dst_cid: VSOCK_CID_HOST, - src_port: H2G_GUEST_PORT, - dst_port: h2g_host_port, - op: VsockOp::RESPONSE, - type_: VsockType::STREAM, - ..Default::default() - }; - send_to_tx( - &resp_hdr, - &[], - &ram, - tx_buf_addr, - &mut tx_q, - &tx, - ¬ifier, - &irq_rx, - true, - ); - let used = rx_q.get_used().unwrap(); - assert_eq!(used.id, rst_buf_id); - assert_eq!(used.len as usize, size_of::()); - ram.read(rx_buf_addr, hdr.as_mut_bytes()).unwrap(); + let rst_buf_id = t.add_rx_desc(); + let resp_hdr = guest_hdr(VsockOp::RESPONSE, H2G_GUEST_PORT, h2g_host_port); + t.send_to_tx(&resp_hdr, &[], true); + let hdr = t.take_rx_hdr(rst_buf_id); assert_eq!(hdr.op, VsockOp::RST); assert_eq!(hdr.src_port, h2g_host_port); assert_eq!(hdr.dst_port, H2G_GUEST_PORT); - tx.send(WakeEvent::Shutdown).unwrap(); - notifier.notify().unwrap(); - handle.join().unwrap(); + t.shutdown(); }