alioth/src/virtio/dev/vsock/uds_vsock.rs - a0deb75

uds_vsock.rs86.82%

// Copyright 2025 Google LLC
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     https://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
 
use std::collections::HashMap;
use std::fmt::Debug;
use std::fs;
use std::io::{BufRead, BufReader, BufWriter, ErrorKind, IoSlice, IoSliceMut, Read, Write};
use std::mem::size_of_val;
use std::num::Wrapping;
use std::os::fd::AsRawFd;
use std::os::unix::net::{UnixListener, UnixStream};
use std::path::Path;
use std::sync::Arc;
use std::sync::mpsc::Receiver;
use std::thread::JoinHandle;
 
use crate::ffi;
use crate::hv::IoeventFd;
use crate::mem::mapped::RamBus;
use crate::sync::notifier::Notifier;
use crate::virtio::dev::vsock::{
    ShutdownFlag, VSOCK_CID_HOST, VsockConfig, VsockFeature, VsockHeader, VsockOp, VsockType,
    VsockVirtq,
};
use crate::virtio::dev::{DevParam, Virtio, WakeEvent};
use crate::virtio::queue::{DescChain, Queue, QueueReg, Status, VirtQueue};
use crate::virtio::worker::mio::{ActiveMio, Mio, VirtioMio};
use crate::virtio::{DeviceId, FEATURE_BUILT_IN, IrqSender, Result, error};
 
use mio::event::Event;
use mio::unix::SourceFd;
use mio::{Interest, Registry, Token};
use serde::Deserialize;
use serde_aco::Help;
use zerocopy::{FromBytes, IntoBytes};
 
const HEADER_SIZE: usize = size_of::<VsockHeader>();
const SOCKET_TYPE: VsockType = VsockType::STREAM;
 
#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Help)]
pub struct UdsVsockParam {
    /// Vsock context id.
    pub cid: u32,
    /// Host-side Unix domain socket path.
    pub path: Box<Path>,
}
 
impl DevParam for UdsVsockParam {
    type Device = UdsVsock;
 
    fn build(self, name: impl Into<Arc<str>>) -> Result<UdsVsock> {3x
        UdsVsock::new(self, name)3x
    }3x
}
 
#[derive(Debug)]
pub struct UdsVsock {
    name: Arc<str>,
    config: Arc<VsockConfig>,
    path: Box<Path>,
    listener: UnixListener,
    connections: HashMap<(u32, u32), Connection>,
    ports: HashMap<Token, (u32, u32)>,
    sockets: HashMap<Token, UnixStream>,
    host_ports: HashMap<u32, u32>,
    next_port: u32,
}
 
fn get_buf_size(stream: &UnixStream) -> Result<usize> {6x
    let mut buf_size = 0i32;6x
    let mut arg_size = size_of_val(&buf_size) as libc::socklen_t;6x
    ffi!(unsafe {6x
        libc::getsockopt(6x
            stream.as_raw_fd(),6x
            libc::SOL_SOCKET,
            libc::SO_SNDBUF,
            &mut buf_size as *mut _ as _,6x
            &mut arg_size,6x
        )
    })?;
    Ok(buf_size as usize)6x
}6x
 
impl UdsVsock {
    fn allocate_port(&mut self) -> Option<u32> {3x
        let mut count: u64 = 0;3x
        while self.host_ports.contains_key(&self.next_port) && count < u32::MAX as u64 {3x
            self.next_port = self.next_port.wrapping_add(1);
            count += 1;
        }
        if count == u32::MAX as u64 {3x
            None
        } else {
            Some(self.next_port)3x
        }
    }3x
 
    fn create_socket(&mut self, registry: &Registry) -> Result<()> {3x
        let (stream, _) = self.listener.accept()?;3x
        stream.set_nonblocking(true)?;3x
        let token = Token(stream.as_raw_fd() as usize);3x
        registry.register(3x
            &mut SourceFd(&stream.as_raw_fd()),3x
            token,3x
            Interest::READABLE,
        )?;
        self.sockets.insert(token, stream);3x
        Ok(())3x
    }3x
 
    fn handle_conn_request<'m, Q, S>(3x
        &mut self,3x
        token: Token,3x
        socket: UnixStream,3x
        rx_q: &mut Queue<'_, 'm, Q>,3x
        irq_sender: &S,3x
    ) -> Result<()>3x
    where3x
        Q: VirtQueue<'m>,3x
        S: IrqSender,3x
    {
        let mut msg = String::new();3x
        let writer = socket.try_clone()?;3x
        let mut reader = BufReader::new(socket);3x
        let buf_size = get_buf_size(&writer)?;3x
        reader.read_line(&mut msg)?;3x
        let port_str = msg.trim_start_matches("CONNECT ").trim_end();3x
        let Ok(port) = port_str.parse::<u32>() else {3x
            log::error!("{}: failed to parse port {port_str}", self.name);
            return Ok(());
        };
        let Some(host_port) = self.allocate_port() else {3x
            log::error!("{}: failed to allocate port", self.name);
            return Ok(());
        };
        let hdr = VsockHeader {3x
            src_cid: VSOCK_CID_HOST,3x
            dst_cid: self.config.guest_cid,3x
            src_port: host_port,3x
            dst_port: port,3x
            type_: SOCKET_TYPE,3x
            op: VsockOp::REQUEST,3x
            fwd_cnt: Wrapping(0),3x
            buf_alloc: buf_size as u32,3x
            ..Default::default()3x
        };3x
        self.respond(&hdr, irq_sender, rx_q)?;3x
        let conn = Connection {3x
            state: ConnState::Requested,3x
            reader,3x
            writer: BufWriter::new(writer),3x
            buf_alloc: buf_size as u32,3x
        };3x
        self.connections.insert((host_port, port), conn);3x
        let count = self.host_ports.entry(host_port).or_default();3x
        *count += 1;3x
        log::trace!(3x
            "{}: host:{host_port}: count incremented to {count}",
            self.name
        );
        self.ports.insert(token, (host_port, port));3x
        log::trace!("{}: host:{host_port} -> vm:{port}: requested", self.name);3x
        Ok(())3x
    }3x
 
    fn respond_rst<'m, Q, S>(3x
        &self,3x
        hdr: &VsockHeader,3x
        irq_sender: &S,3x
        rx_q: &mut Queue<'_, 'm, Q>,3x
    ) -> Result<()>3x
    where3x
        Q: VirtQueue<'m>,3x
        S: IrqSender,3x
    {
        let resp = VsockHeader {3x
            src_cid: VSOCK_CID_HOST,3x
            dst_cid: self.config.guest_cid,3x
            src_port: hdr.dst_port,3x
            dst_port: hdr.src_port,3x
            type_: hdr.type_,3x
            op: VsockOp::RST,3x
            ..Default::default()3x
        };3x
        self.respond(&resp, irq_sender, rx_q)3x
    }3x
 
    fn respond<'m, Q, S>(9x
        &self,9x
        hdr: &VsockHeader,9x
        irq_sender: &S,9x
        rx_q: &mut Queue<'_, 'm, Q>,9x
    ) -> Result<()>9x
    where9x
        Q: VirtQueue<'m>,9x
        S: IrqSender,9x
    {
        let mut hdr_buf = hdr.as_bytes();9x
        rx_q.handle_desc(VsockVirtq::RX.raw(), irq_sender, |desc| {9x
            if hdr_buf.is_empty() {6x
                return Ok(Status::Break);
            }6x
            let c = hdr_buf.read_vectored(&mut desc.writable)? as u32;6x
            Ok(Status::Done { len: c })6x
        })?;6x
        if !hdr_buf.is_empty() {9x
            log::error!(3x
                "{}: queue RX: no enough writable buffers for {:?}",
                self.name,
                hdr.op
            );
            return error::InvalidBuffer.fail();3x
        }6x
        Ok(())6x
    }9x
 
    fn handle_tx_response<'m, Q, S>(3x
        &mut self,3x
        hdr: &VsockHeader,3x
        rx_q: &mut Queue<'_, 'm, Q>,3x
        irq_sender: &S,3x
    ) -> Result<()>3x
    where3x
        Q: VirtQueue<'m>,3x
        S: IrqSender,3x
    {
        let host_port = hdr.dst_port;3x
        let guest_port = hdr.src_port;3x
        let Some(conn) = self.connections.get_mut(&(host_port, guest_port)) else {3x
            log::warn!(
                "{}: vm:{guest_port} -> host:{host_port}: unknown connection",
                self.name
            );
            return Ok(());
        };
        if conn.state != ConnState::Requested {3x
            log::error!(
                "{}: vm:{guest_port} -> host:{host_port}: found {:?}, expect {:?}",
                self.name,
                conn.state,
                ConnState::Requested
            );
            return Ok(());
        };3x
        writeln!(conn.writer, "OK {host_port}")?;3x
        conn.writer.flush()?;3x
        conn.state = ConnState::Established {3x
            fwd_cnt: Wrapping(0),3x
        };3x
        log::trace!(3x
            "{}: host:{host_port} -> vm:{guest_port}: established",
            self.name
        );
        self.transfer_rx_data(host_port, guest_port, rx_q, irq_sender)3x
    }3x
 
    fn remove_conn(&mut self, host_port: u32, guest_port: u32, registry: &Registry) -> Result<()> {6x
        let Some(conn) = self.connections.remove(&(host_port, guest_port)) else {6x
            log::warn!(
                "{}: vm:{guest_port} -> host:{host_port}: unknown connection",
                self.name
            );
            return Ok(());
        };
        let reader = conn.reader.into_inner();6x
        let token = Token(reader.as_raw_fd() as usize);6x
        self.ports.remove(&token);6x
        if let Some(count) = self.host_ports.get_mut(&host_port) {6x
            if *count == 1 {6x
                self.host_ports.remove(&host_port);6x
                log::trace!("{}: host:{host_port}: free port", self.name);6x
            } else {
                *count -= 1;
                log::trace!(
                    "{}: host:{host_port}: count decremented to {count}",
                    self.name
                );
            }
        } else {
            log::error!(
                "{}: vm:{guest_port} -> host:{host_port}: unknown host port",
                self.name
            );
        }
        registry.deregister(&mut SourceFd(&reader.as_raw_fd()))?;6x
        Ok(())6x
    }6x
 
    fn handle_tx_rst(&mut self, hdr: &VsockHeader, registry: &Registry) -> Result<()> {3x
        let host_port = hdr.dst_port;3x
        let guest_port = hdr.src_port;3x
        self.remove_conn(host_port, guest_port, registry)?;3x
        log::trace!("{}: vm:{guest_port} -> host:{host_port}: reset", self.name);3x
        Ok(())3x
    }3x
 
    fn handle_tx_shutdown<'m, Q, S>(6x
        &mut self,6x
        hdr: &VsockHeader,6x
        registry: &Registry,6x
        irq_sender: &S,6x
        rx_q: &mut Queue<'_, 'm, Q>,6x
    ) -> Result<()>6x
    where6x
        Q: VirtQueue<'m>,6x
        S: IrqSender,6x
    {
        let host_port = hdr.dst_port;6x
        let guest_port = hdr.src_port;6x
        let Some(conn) = self.connections.get_mut(&(host_port, guest_port)) else {6x
            log::warn!(
                "{}: vm:{guest_port} -> host:{host_port}: unknown connection",
                self.name
            );
            return Ok(());
        };
        let mut flags = if let ConnState::Shutdown { flags } = conn.state {6x
            flags3x
        } else {
            ShutdownFlag::empty()3x
        };
        flags |= ShutdownFlag::from_bits_truncate(hdr.flags);6x
        if flags != ShutdownFlag::all() {6x
            conn.state = ConnState::Shutdown { flags };3x
            log::trace!(3x
                "{}: vm:{guest_port} -> host:{host_port}: {flags:?}",
                self.name
            );
        } else {
            if let Err(e) = self.respond_rst(hdr, irq_sender, rx_q) {3x
                log::error!("{}: failed to respond to shutdown: {e:?}", self.name);3x
            }
            self.remove_conn(host_port, guest_port, registry)?;3x
            log::trace!(3x
                "{}: vm:{guest_port} -> host:{host_port}: shutdown",
                self.name
            );
        }
        Ok(())6x
    }6x
 
    fn handle_tx_request<'m, Q, S>(3x
        &mut self,3x
        hdr: &VsockHeader,3x
        registry: &Registry,3x
        irq_sender: &S,3x
        rx_q: &mut Queue<'_, 'm, Q>,3x
    ) -> Result<()>3x
    where3x
        Q: VirtQueue<'m>,3x
        S: IrqSender,3x
    {
        let host_port = hdr.dst_port;3x
        let guest_port = hdr.src_port;3x
        let port_socket = format!("{}_{host_port}", self.path.to_string_lossy());3x
        let reader = match UnixStream::connect(&port_socket) {3x
            Ok(reader) => reader,3x
            Err(e) => {
                log::error!("{}: failed to connect to {port_socket}: {e:?}", self.name);
                return self.respond_rst(hdr, irq_sender, rx_q);
            }
        };
        reader.set_nonblocking(true)?;3x
        let writer = reader.try_clone()?;3x
        let token = Token(reader.as_raw_fd() as usize);3x
        registry.register(3x
            &mut SourceFd(&reader.as_raw_fd()),3x
            token,3x
            Interest::READABLE,
        )?;
        let buf_size = get_buf_size(&writer)?;3x
        let conn = Connection {3x
            reader: BufReader::new(reader),3x
            writer: BufWriter::new(writer),3x
            buf_alloc: buf_size as u32,3x
            state: ConnState::Established {3x
                fwd_cnt: Wrapping(0),3x
            },3x
        };3x
        let resp = VsockHeader {3x
            src_cid: VSOCK_CID_HOST,3x
            dst_cid: self.config.guest_cid,3x
            src_port: host_port,3x
            dst_port: guest_port,3x
            type_: hdr.type_,3x
            op: VsockOp::RESPONSE,3x
            fwd_cnt: Wrapping(0),3x
            buf_alloc: buf_size as u32,3x
            ..Default::default()3x
        };3x
        self.respond(&resp, irq_sender, rx_q)?;3x
        self.connections.insert((host_port, guest_port), conn);3x
        let count = self.host_ports.entry(host_port).or_default();3x
        *count += 1;3x
        log::trace!(3x
            "{}: host:{host_port}: count incremented to {count}",
            self.name
        );
        self.ports.insert(token, (host_port, guest_port));3x
        log::trace!(3x
            "{}: vm:{guest_port} -> host:{host_port}: established",
            self.name
        );
        Ok(())3x
    }3x
 
    fn handle_tx_desc<'m, Q, S>(18x
        &mut self,18x
        desc: &mut DescChain,18x
        registry: &Registry,18x
        irq_sender: &S,18x
        rx_q: &mut Queue<'_, 'm, Q>,18x
    ) -> Result<()>18x
    where18x
        Q: VirtQueue<'m>,18x
        S: IrqSender,18x
    {
        let name = &*self.name;18x
        let [buf, readable @ ..] = desc.readable.as_slice() else {18x
            return error::InvalidBuffer.fail();
        };
        let Some((header, body)) = buf.split_first_chunk::<HEADER_SIZE>() else {18x
            return error::InvalidBuffer.fail();
        };
        let Ok(hdr) = VsockHeader::ref_from_bytes(header) else {18x
            return error::InvalidBuffer.fail();
        };
        if hdr.src_cid != self.config.guest_cid || hdr.dst_cid != VSOCK_CID_HOST {18x
            log::warn!(
                "{name}: invalid CID pair: {} -> {}",
                hdr.src_cid,
                hdr.dst_cid
            );
        }18x
        log::trace!(18x
            "{name}: vm:{} -> host:{}: {:?}",
            hdr.src_port,
            hdr.dst_port,
            hdr.op
        );
        match hdr.op {18x
            VsockOp::REQUEST => self.handle_tx_request(hdr, registry, irq_sender, rx_q),3x
            VsockOp::RESPONSE => self.handle_tx_response(hdr, rx_q, irq_sender),3x
            VsockOp::RST => self.handle_tx_rst(hdr, registry),3x
            VsockOp::RW => self.transfer_tx_data(hdr, body, readable),3x
            VsockOp::CREDIT_UPDATE => {
                log::info!(
                    "{name}: CREDIT_UPDATE: fwd_cnt: {}, buf_alloc: {}",
                    hdr.fwd_cnt,
                    hdr.buf_alloc
                );
                Ok(())
            }
            VsockOp::SHUTDOWN => self.handle_tx_shutdown(hdr, registry, irq_sender, rx_q),6x
            _ => {
                log::error!("{name}: unsupported operation: {:?}", hdr.op);
                Ok(())
            }
        }
    }18x
 
    fn handle_tx<'m, Q, S, E>(18x
        &mut self,18x
        active_mio: &mut ActiveMio<'_, '_, 'm, Q, S, E>,18x
    ) -> Result<()>18x
    where18x
        Q: VirtQueue<'m>,18x
        S: IrqSender,18x
        E: IoeventFd,18x
    {
        let [Some(rx_q), Some(tx_q), ..] = active_mio.queues else {18x
            let tx_index = VsockVirtq::TX.raw();
            return error::InvalidQueueIndex { index: tx_index }.fail();
        };
        let irq_sender = active_mio.irq_sender;18x
        let registry = active_mio.poll.registry();18x
        tx_q.handle_desc(VsockVirtq::TX.raw(), irq_sender, |desc| {18x
            self.handle_tx_desc(desc, registry, irq_sender, rx_q)?;18x
            Ok(Status::Done { len: 0 })18x
        })18x
    }18x
 
    fn transfer_rx_data<'m, Q, S>(6x
        &mut self,6x
        host_port: u32,6x
        guest_port: u32,6x
        rx_q: &mut Queue<'_, 'm, Q>,6x
        irq_sender: &S,6x
    ) -> Result<()>6x
    where6x
        Q: VirtQueue<'m>,6x
        S: IrqSender,6x
    {
        fn copy_to_rx(3x
            hdr: &mut VsockHeader,3x
            conn: &mut BufReader<UnixStream>,3x
            buffers: &mut [IoSliceMut],3x
        ) -> Result<usize> {3x
            let mut nskip = 0;3x
            let mut nread = 0;3x
            for buf in buffers.iter_mut() {9x
                let r = if HEADER_SIZE > nskip {9x
                    let Some((_, data)) = buf.split_at_mut_checked(HEADER_SIZE - nskip) else {6x
                        nskip += buf.len();3x
                        continue;3x
                    };
                    nskip = HEADER_SIZE;3x
                    if data.is_empty() {3x
                        continue;
                    }3x
                    conn.read(data)3x
                } else {
                    conn.read(buf)3x
                };
                let n = match r {3x
                    Ok(0) => break,
                    Ok(n) => n,3x
                    Err(e) if e.kind() == ErrorKind::WouldBlock => break,3x
                    Err(e) => Err(e)?,
                };
                nread += n;3x
            }
            if nskip != HEADER_SIZE {3x
                return error::InvalidBuffer.fail();
            }3x
            hdr.len = nread as u32;3x
            let mut hdr_buf = hdr.as_bytes();3x
            let _ = hdr_buf.read_vectored(buffers);3x
            Ok(nread)3x
        }3x
 
        let rx_idx = VsockVirtq::RX.raw();6x
        let Some(conn) = self.connections.get_mut(&(host_port, guest_port)) else {6x
            log::warn!(
                "{}: vm:{guest_port} -> host:{host_port}: unknown connection",
                self.name
            );
            return Ok(());
        };
        let ConnState::Established { fwd_cnt } = conn.state else {6x
            log::error!("{}: unexpected state {:?}", self.name, conn.state);
            return Ok(());
        };
        let mut hdr = VsockHeader {6x
            src_cid: VSOCK_CID_HOST,6x
            dst_cid: self.config.guest_cid,6x
            src_port: host_port,6x
            dst_port: guest_port,6x
            type_: SOCKET_TYPE,6x
            op: VsockOp::RW,6x
            fwd_cnt,6x
            buf_alloc: conn.buf_alloc,6x
            ..Default::default()6x
        };6x
        rx_q.handle_desc(rx_idx, irq_sender, |desc| {6x
            let nread = copy_to_rx(&mut hdr, &mut conn.reader, &mut desc.writable)? as u32;3x
            if nread == 0 {3x
                return Ok(Status::Break);
            }3x
            log::trace!(3x
                "{}: host:{host_port} -> vm:{guest_port}: transfered {nread} bytes",
                self.name
            );
            Ok(Status::Done {3x
                len: nread + HEADER_SIZE as u32,3x
            })3x
        })?;3x
        Ok(())6x
    }6x
 
    fn transfer_tx_data(3x
        &mut self,3x
        hdr: &VsockHeader,3x
        body: &[u8],3x
        buffers: &[IoSlice],3x
    ) -> Result<()> {3x
        fn copy_to_conn(3x
            buf: &[u8],3x
            conn: &mut BufWriter<UnixStream>,3x
            remain: &mut usize,3x
        ) -> Result<()> {3x
            if let Some(b) = buf.get(..*remain) {3x
                conn.write_all(b)?;3x
                *remain = 0;3x
            } else {
                conn.write_all(buf)?;
                *remain -= buf.len();
            }
            Ok(())3x
        }3x
 
        let host_port = hdr.dst_port;3x
        let guest_port = hdr.src_port;3x
        let Some(conn) = self.connections.get_mut(&(host_port, guest_port)) else {3x
            log::warn!(
                "{}: vm:{guest_port} -> host:{host_port}: unknown connection",
                self.name
            );
            return Ok(());
        };
        let ConnState::Established { fwd_cnt } = &mut conn.state else {3x
            log::warn!("{}: invalid connection state {:?}", self.name, conn.state);
            return Ok(());
        };
        let mut remain = hdr.len as usize;3x
        if !body.is_empty() {3x
            copy_to_conn(body, &mut conn.writer, &mut remain)?;
        }3x
        for buf in buffers {3x
            if remain == 0 {3x
                break;
            }3x
            copy_to_conn(buf, &mut conn.writer, &mut remain)?;3x
        }
        if remain != 0 {3x
            log::error!("{}: missing {remain} bytes", self.name);
            return error::InvalidBuffer.fail();
        }3x
        *fwd_cnt += hdr.len;3x
        log::trace!(3x
            "{}: vm:{guest_port} -> host:{host_port}: transferred {} bytes",
            self.name,
            hdr.len
        );
        conn.writer.flush()?;3x
        Ok(())3x
    }3x
}
 
impl Drop for UdsVsock {
    fn drop(&mut self) {3x
        let Ok(addr) = self.listener.local_addr() else {3x
            return;
        };
        let Some(path) = addr.as_pathname() else {3x
            return;
        };
        if let Err(e) = fs::remove_file(path) {3x
            log::error!("{}: error removing {path:?}: {e:?}", self.name);
        }3x
    }3x
}
 
#[derive(Debug, PartialEq, Eq)]
pub enum ConnState {
    Requested,
    Established { fwd_cnt: Wrapping<u32> },
    Shutdown { flags: ShutdownFlag },
}
 
#[derive(Debug)]
pub struct Connection {
    state: ConnState,
    reader: BufReader<UnixStream>,
    writer: BufWriter<UnixStream>,
    buf_alloc: u32,
}
 
impl UdsVsock {
    fn new(param: UdsVsockParam, name: impl Into<Arc<str>>) -> Result<Self> {3x
        let name = name.into();3x
        let listener = UnixListener::bind(&param.path)?;3x
        listener.set_nonblocking(true)?;3x
        let vsock = UdsVsock {3x
            name,3x
            path: param.path,3x
            config: Arc::new(VsockConfig {3x
                guest_cid: param.cid,3x
                ..Default::default()3x
            }),3x
            listener,3x
            connections: HashMap::new(),3x
            sockets: HashMap::new(),3x
            ports: HashMap::new(),3x
            host_ports: HashMap::new(),3x
            next_port: 1024,3x
        };3x
        Ok(vsock)3x
    }3x
}
 
impl Virtio for UdsVsock {
    type Config = VsockConfig;
    type Feature = VsockFeature;
 
    fn id(&self) -> DeviceId {3x
        DeviceId::SOCKET3x
    }3x
 
    fn name(&self) -> &str {12x
        &self.name12x
    }12x
 
    fn num_queues(&self) -> u16 {3x
        33x
    }3x
 
    fn config(&self) -> Arc<VsockConfig> {3x
        self.config.clone()3x
    }3x
 
    fn feature(&self) -> u128 {3x
        VsockFeature::STREAM.bits() | FEATURE_BUILT_IN3x
    }3x
 
    fn spawn_worker<S, E>(3x
        self,3x
        event_rx: Receiver<WakeEvent<S, E>>,3x
        memory: Arc<RamBus>,3x
        queue_regs: Arc<[QueueReg]>,3x
    ) -> Result<(JoinHandle<()>, Arc<Notifier>)>3x
    where3x
        S: IrqSender,3x
        E: IoeventFd,3x
    {
        Mio::spawn_worker(self, event_rx, memory, queue_regs)3x
    }3x
}
 
impl VirtioMio for UdsVsock {
    fn activate<'m, Q, S, E>(3x
        &mut self,3x
        _feature: u128,3x
        active_mio: &mut ActiveMio<'_, '_, 'm, Q, S, E>,3x
    ) -> Result<()>3x
    where3x
        Q: VirtQueue<'m>,3x
        S: IrqSender,3x
        E: IoeventFd,3x
    {
        active_mio.poll.registry().register(3x
            &mut SourceFd(&self.listener.as_raw_fd()),3x
            Token(self.listener.as_raw_fd() as usize),3x
            Interest::READABLE,
        )?;
        Ok(())3x
    }3x
 
    fn handle_event<'m, Q, S, E>(9x
        &mut self,9x
        event: &Event,9x
        active_mio: &mut ActiveMio<'_, '_, 'm, Q, S, E>,9x
    ) -> Result<()>9x
    where9x
        Q: VirtQueue<'m>,9x
        S: IrqSender,9x
        E: IoeventFd,9x
    {
        let token = event.token();9x
        let registry = active_mio.poll.registry();9x
        let irq_sender = active_mio.irq_sender;9x
        let rx_index = VsockVirtq::RX.raw();9x
        let Some(Some(rx_q)) = active_mio.queues.get_mut(rx_index as usize) else {9x
            return error::InvalidQueueIndex { index: rx_index }.fail();
        };
        if token.0 == self.listener.as_raw_fd() as usize {9x
            self.create_socket(registry)3x
        } else if let Some(socket) = self.sockets.remove(&token) {6x
            self.handle_conn_request(token, socket, rx_q, irq_sender)3x
        } else if let Some(port_pair) = self.ports.get(&token) {3x
            let (host_port, guest_port) = port_pair.to_owned();3x
            self.transfer_rx_data(host_port, guest_port, rx_q, irq_sender)3x
        } else {
            log::error!("{}: invalid token: {token:#x?}", self.name);
            Ok(())
        }
    }9x
 
    fn handle_queue<'m, Q, S, E>(21x
        &mut self,21x
        index: u16,21x
        active_mio: &mut ActiveMio<'_, '_, 'm, Q, S, E>,21x
    ) -> Result<()>21x
    where21x
        Q: VirtQueue<'m>,21x
        S: IrqSender,21x
        E: IoeventFd,21x
    {
        let index = VsockVirtq::from(index);21x
        let name = &self.name;21x
        match index {21x
            VsockVirtq::TX => self.handle_tx(active_mio)?,18x
            VsockVirtq::RX => log::debug!("{name}: queue RX buffer available"),3x
            VsockVirtq::EVENT => log::debug!("{name}: queue EVENT buffer available"),
            _ => log::error!("{name}: unknown queue index {index:?}"),
        }
        Ok(())21x
    }21x
 
    fn reset(&mut self, registry: &Registry) {3x
        for (_, conn) in self.connections.drain() {3x
            let reader = conn.reader.into_inner();
            if let Err(err) = registry.deregister(&mut SourceFd(&reader.as_raw_fd())) {
                log::error!("{}: failed to deregister socket: {err}", self.name);
            }
        }
        for (_, socket) in self.sockets.drain() {3x
            if let Err(err) = registry.deregister(&mut SourceFd(&socket.as_raw_fd())) {
                log::error!("{}: failed to deregister socket: {err}", self.name);
            }
        }
        if let Err(err) = registry.deregister(&mut SourceFd(&self.listener.as_raw_fd())) {3x
            log::error!("{}: failed to deregister listener: {err}", self.name);
        }3x
        self.host_ports.clear();3x
        self.next_port = 1024;3x
    }3x
}
 
#[cfg(test)]
#[path = "uds_vsock_test.rs"]
mod tests;
 