diff --git a/alioth/src/virtio/dev/vsock/uds_vsock.rs b/alioth/src/virtio/dev/vsock/uds_vsock.rs index 913a20eb..456482dc 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; @@ -71,11 +71,29 @@ 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, +} + +/// 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; @@ -106,22 +124,46 @@ 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, - )?; - self.sockets.insert(token, stream); + // 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(()) + } + + 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 +171,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 +219,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, @@ -254,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), }; @@ -455,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: {}", @@ -554,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 { @@ -652,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; @@ -673,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 { @@ -686,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!( @@ -706,7 +819,6 @@ impl UdsVsock { self.name, hdr.len ); - conn.writer.flush()?; Ok(()) } } @@ -837,8 +949,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 +999,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..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,47 +427,145 @@ 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); + + t.shutdown(); +} + +#[test] +fn vsock_partial_conn_request_test() { + let ram_bus = Arc::new(fixture_ram_bus()); + let ram = ram_bus.lock_layout(); + let mut t = VsockTest::new(&ram_bus, &ram); + + 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 + // 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(); + + 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); + + t.shutdown(); +} + +#[test] +fn vsock_simultaneous_conn_test() { + let ram_bus = Arc::new(fixture_ram_bus()); + let ram = ram_bus.lock_layout(); + let mut t = VsockTest::new(&ram_bus, &ram); + + 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 = t.connect(); + stream + .write_all(format!("CONNECT {port}\n").as_bytes()) + .unwrap(); + streams.push(stream); + } + + let mut requests = Vec::new(); + for _ in GUEST_PORTS { + 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 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); + + t.shutdown(); +} + +#[test] +fn vsock_conn_request_eof_test() { + let ram_bus = Arc::new(fixture_ram_bus()); + let ram = ram_bus.lock_layout(); + let mut t = VsockTest::new(&ram_bus, &ram); + + // A client that connects and disconnects without saying anything. + drop(t.connect()); + + // A client that disconnects in the middle of a connection request. + 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. + const H2G_GUEST_PORT: u32 = 1025; + let (h2g_stream, h2g_host_port) = t.request_conn(H2G_GUEST_PORT); + t.accept_conn(&h2g_stream, H2G_GUEST_PORT, h2g_host_port); + + t.shutdown(); +} + +#[test] +fn vsock_conn_request_close_test() { + let ram_bus = Arc::new(fixture_ram_bus()); + let ram = ram_bus.lock_layout(); + 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 = t.add_rx_desc(); + const H2G_GUEST_PORT: u32 = 1025; + let mut h2g_stream = t.connect(); + h2g_stream + .write_all(format!("CONNECT {H2G_GUEST_PORT}\n").as_bytes()) + .unwrap(); + drop(h2g_stream); + + 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 = 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(); }