use an enum for the backend instead of a trait with a generic

aerosouund · aerosouund · commit 9cb7f645b209 · 2025-12-26T20:27:00.000+02:00
to simplify the lifetime of file descriptors, leaving their
closing to the unix stream and the owned fd for seqpacket

Co-authored-by: aerosouund &lt;aerosound161@gmail.com&gt;
Signed-off-by: aerosouund &lt;aerosound161@gmail.com&gt;
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -12,9 +12,9 @@ and this project adheres to
 
 ### Changed
 
-- [#5595](https://github.com/firecracker-microvm/firecracker/pull/5595): Added `vsock_type` 
-field to the vsock device API to denote the type of the underlying socket. Can be `stream`
-or `seqpacket`
+- [#5595](https://github.com/firecracker-microvm/firecracker/pull/5595): Added
+  `vsock_type` field to the vsock device API to denote the type of the
+  underlying socket. Can be `stream` or `seqpacket`
 
 ### Deprecated
 
diff --git a/src/vmm/src/devices/virtio/vsock/csm/connection.rs b/src/vmm/src/devices/virtio/vsock/csm/connection.rs
@@ -140,7 +140,7 @@ pub struct VsockConnection<S: VsockConnectionBackend> {
     /// Instant when this connection should be scheduled for immediate termination, due to some
     /// timeout condition having been fulfilled.
     expiry: Option<Instant>,
-    /// Vsock type (stream or seqpacket)
+    /// The type of the underlying socket connection
     vsock_type: VsockType,
 }
 
diff --git a/src/vmm/src/devices/virtio/vsock/unix/mod.rs b/src/vmm/src/devices/virtio/vsock/unix/mod.rs
@@ -11,7 +11,8 @@ mod muxer;
 mod muxer_killq;
 mod muxer_rxq;
 mod seqpacket;
-use std::os::fd::AsRawFd as _;
+use std::io::{self, Read, Write};
+use std::os::fd::AsRawFd;
 use std::os::unix::net::UnixStream;
 use std::time::Instant;
 
@@ -57,85 +58,58 @@ pub enum VsockUnixBackendError {
     TooManyConnections,
 }
 
-type MuxerStreamConnection = super::csm::VsockConnection<UnixStream>;
-type MuxerSeqpacketConnetion = super::csm::VsockConnection<SeqpacketConn>;
-
 #[derive(Debug)]
-enum MuxerConn {
-    Stream(MuxerStreamConnection),
-    Seqpacket(MuxerSeqpacketConnetion),
+pub enum ConnBackend {
+    Stream(UnixStream),
+    Seqpacket(SeqpacketConn),
 }
-
+// can we make vsockconnection instead of being generic, hold an enum ?
 macro_rules! forward_to_inner {
     ($self:ident, $method:ident $(, $args:expr )* ) => {
         match $self {
-            MuxerConn::Stream(inner) => inner.$method($($args),*),
-            MuxerConn::Seqpacket(inner) => inner.$method($($args),*),
+            ConnBackend::Stream(inner) => inner.$method($($args),*),
+            ConnBackend::Seqpacket(inner) => inner.$method($($args),*),
         }
     };
 }
 
-impl MuxerConn {
-    fn has_pending_rx(&self) -> bool {
-        forward_to_inner!(self, has_pending_rx)
+impl Read for ConnBackend {
+    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+        forward_to_inner!(self, read, buf)
     }
+}
 
+impl AsRawFd for ConnBackend {
     fn as_raw_fd(&self) -> i32 {
         forward_to_inner!(self, as_raw_fd)
     }
+}
 
-    fn kill(&mut self) {
-        forward_to_inner!(self, kill)
-    }
-
-    fn get_polled_evset(&self) -> EventSet {
-        forward_to_inner!(self, get_polled_evset)
-    }
-
-    fn will_expire(&self) -> bool {
-        forward_to_inner!(self, will_expire)
-    }
-
-    fn has_expired(&self) -> bool {
-        forward_to_inner!(self, has_expired)
-    }
-
-    fn send_bytes_raw(&mut self, buf: &[u8]) -> Result<usize, VsockCsmError> {
-        forward_to_inner!(self, send_bytes_raw, buf)
-    }
-
-    fn state(&self) -> ConnState {
-        forward_to_inner!(self, state)
-    }
-
-    fn expiry(&self) -> Option<Instant> {
-        forward_to_inner!(self, expiry)
-    }
-
-    fn recv_pkt(&mut self, pkt: &mut VsockPacketRx) -> Result<(), VsockError> {
-        forward_to_inner!(self, recv_pkt, pkt)
-    }
-
-    fn send_pkt(&mut self, pkt: &VsockPacketTx) -> Result<(), VsockError> {
-        forward_to_inner!(self, send_pkt, pkt)
+impl ReadVolatile for ConnBackend {
+    fn read_volatile<B: vm_memory::bitmap::BitmapSlice>(
+        &mut self,
+        buf: &mut vm_memory::VolatileSlice<B>,
+    ) -> Result<usize, vm_memory::VolatileMemoryError> {
+        forward_to_inner!(self, read_volatile, buf)
     }
+}
 
-    fn notify(&mut self, evset: EventSet) {
-        forward_to_inner!(self, notify, evset)
+impl WriteVolatile for ConnBackend {
+    fn write_volatile<B: vm_memory::bitmap::BitmapSlice>(
+        &mut self,
+        buf: &vm_memory::VolatileSlice<B>,
+    ) -> Result<usize, vm_memory::VolatileMemoryError> {
+        forward_to_inner!(self, write_volatile, buf)
     }
 }
 
-#[cfg(test)]
-impl MuxerConn {
-    pub(crate) fn fwd_cnt(&self) -> std::num::Wrapping<u32> {
-        forward_to_inner!(self, fwd_cnt)
+impl Write for ConnBackend {
+    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+        forward_to_inner!(self, write, buf)
     }
-
-    pub(crate) fn insert_credit_update(&mut self) {
-        forward_to_inner!(self, insert_credit_update)
+    fn flush(&mut self) -> io::Result<()> {
+        Ok(())
     }
 }
 
-impl VsockConnectionBackend for UnixStream {}
-
-impl VsockConnectionBackend for SeqpacketConn {}
+impl VsockConnectionBackend for ConnBackend {}
diff --git a/src/vmm/src/devices/virtio/vsock/unix/muxer.rs b/src/vmm/src/devices/virtio/vsock/unix/muxer.rs
@@ -46,12 +46,12 @@ use super::super::defs::uapi;
 use super::super::{VsockBackend, VsockChannel, VsockEpollListener, VsockError};
 use super::muxer_killq::MuxerKillQ;
 use super::muxer_rxq::MuxerRxQ;
-use super::{MuxerStreamConnection, VsockUnixBackendError, defs};
-use crate::devices::virtio::vsock::csm::{VsockConnection, VsockConnectionBackend};
+use super::{VsockUnixBackendError, defs};
+use crate::devices::virtio::vsock::csm::VsockConnection;
 use crate::devices::virtio::vsock::defs::uapi::{VSOCK_TYPE_SEQPACKET, VSOCK_TYPE_STREAM};
 use crate::devices::virtio::vsock::metrics::METRICS;
 use crate::devices::virtio::vsock::packet::{VsockPacketRx, VsockPacketTx};
-use crate::devices::virtio::vsock::unix::MuxerConn;
+use crate::devices::virtio::vsock::unix::ConnBackend;
 use crate::devices::virtio::vsock::unix::seqpacket::{SeqpacketConn, SeqpacketListener, Socket};
 use crate::logger::IncMetric;
 use crate::vmm_config::vsock::VsockType;
@@ -84,7 +84,7 @@ enum EpollListener {
     HostSock,
     /// A listener interested in reading host `connect <port>` commands from a freshly
     /// connected host socket.
-    LocalStream(RawFd),
+    LocalStream(ConnBackend),
 }
 
 /// The vsock connection multiplexer.
@@ -93,7 +93,7 @@ pub struct VsockMuxer {
     /// Guest CID.
     cid: u64,
     /// A hash map used to store the active connections.
-    conn_map: HashMap<ConnMapKey, MuxerConn>,
+    conn_map: HashMap<ConnMapKey, VsockConnection<ConnBackend>>,
     /// the underlying host socket file descriptor type wrapper
     host_sock: Box<dyn Socket>,
     /// A hash map used to store epoll event listeners / handlers.
@@ -411,10 +411,7 @@ impl VsockMuxer {
                         // the guest side, we need to know the destination port. We'll read
                         // that port from a "connect" command received on this socket, so the
                         // next step is to ask to be notified the moment we can read from it.
-                        self.add_listener(
-                            stream.as_raw_fd(),
-                            EpollListener::LocalStream(stream.as_raw_fd()),
-                        )
+                        self.add_listener(stream.as_raw_fd(), EpollListener::LocalStream(stream))
                     })
                     .unwrap_or_else(|err| {
                         warn!("vsock: unable to accept local connection: {:?}", err);
@@ -424,62 +421,28 @@ impl VsockMuxer {
             // Data is ready to be read from a host-initiated connection. That would be the
             // "connect" command that we're expecting.
             Some(EpollListener::LocalStream(_)) => {
-                if let Some(EpollListener::LocalStream(fd)) = self.remove_listener(fd) {
-                    match self.vsock_type {
-                        VsockType::Stream => {
-                            // SAFETY: Safe because the fd is valid and we own it (removed from listener_map).
-                            let mut stream = unsafe { UnixStream::from_raw_fd(fd) };
-                            Self::read_local_stream_port(&mut stream)
-                                .map(|peer_port| (self.allocate_local_port(), peer_port))
-                                .and_then(|(local_port, peer_port)| {
-                                    self.add_connection(
-                                        ConnMapKey {
-                                            local_port,
-                                            peer_port,
-                                        },
-                                        MuxerConn::Stream(
-                                            VsockConnection::<UnixStream>::new_local_init(
-                                                stream,
-                                                uapi::VSOCK_HOST_CID,
-                                                self.cid,
-                                                local_port,
-                                                peer_port,
-                                                VsockType::Stream,
-                                            ),
-                                        ),
-                                    )
-                                })
-                                .unwrap_or_else(|err| {
-                                    info!("vsock: error adding local-init connection: {:?}", err);
-                                })
-                        }
-                        VsockType::Seqpacket => {
-                            let mut stream = SeqpacketConn::new(fd);
-                            Self::read_local_stream_port(&mut stream)
-                                .map(|peer_port| (self.allocate_local_port(), peer_port))
-                                .and_then(|(local_port, peer_port)| {
-                                    self.add_connection(
-                                        ConnMapKey {
-                                            local_port,
-                                            peer_port,
-                                        },
-                                        MuxerConn::Seqpacket(
-                                            VsockConnection::<SeqpacketConn>::new_local_init(
-                                                stream,
-                                                uapi::VSOCK_HOST_CID,
-                                                self.cid,
-                                                local_port,
-                                                peer_port,
-                                                VsockType::Seqpacket,
-                                            ),
-                                        ),
-                                    )
-                                })
-                                .unwrap_or_else(|err| {
-                                    info!("vsock: error adding local-init connection: {:?}", err);
-                                })
-                        }
-                    };
+                if let Some(EpollListener::LocalStream(mut stream)) = self.remove_listener(fd) {
+                    Self::read_local_stream_port(&mut stream)
+                        .map(|peer_port| (self.allocate_local_port(), peer_port))
+                        .and_then(|(local_port, peer_port)| {
+                            self.add_connection(
+                                ConnMapKey {
+                                    local_port,
+                                    peer_port,
+                                },
+                                VsockConnection::new_local_init(
+                                    stream,
+                                    uapi::VSOCK_HOST_CID,
+                                    self.cid,
+                                    local_port,
+                                    peer_port,
+                                    self.vsock_type.clone(),
+                                ),
+                            )
+                        })
+                        .unwrap_or_else(|err| {
+                            info!("vsock: error adding local-init connection: {:?}", err);
+                        });
                 }
             }
 
@@ -547,7 +510,7 @@ impl VsockMuxer {
     fn add_connection(
         &mut self,
         key: ConnMapKey,
-        conn: MuxerConn,
+        conn: VsockConnection<ConnBackend>,
     ) -> Result<(), VsockUnixBackendError> {
         // We might need to make room for this new connection, so let's sweep the kill queue
         // first.  It's fine to do this here because:
@@ -695,15 +658,15 @@ impl VsockMuxer {
                                 local_port: pkt.hdr.dst_port(),
                                 peer_port: pkt.hdr.src_port(),
                             },
-                            MuxerConn::Stream(VsockConnection::<UnixStream>::new_peer_init(
-                                stream,
+                            VsockConnection::<ConnBackend>::new_peer_init(
+                                ConnBackend::Stream(stream),
                                 uapi::VSOCK_HOST_CID,
                                 self.cid,
                                 pkt.hdr.dst_port(),
                                 pkt.hdr.src_port(),
                                 pkt.hdr.buf_alloc(),
                                 VsockType::Stream,
-                            )),
+                            ),
                         )
                     })
                     .unwrap_or_else(|_| self.enq_rst(pkt.hdr.dst_port(), pkt.hdr.src_port()));
@@ -718,15 +681,19 @@ impl VsockMuxer {
                                 local_port: pkt.hdr.dst_port(),
                                 peer_port: pkt.hdr.src_port(),
                             },
-                            MuxerConn::Seqpacket(VsockConnection::<SeqpacketConn>::new_peer_init(
-                                SeqpacketConn::new(stream.into_raw_fd()),
+                            VsockConnection::<ConnBackend>::new_peer_init(
+                                // SAFETY: There's no way this file descriptor is invalid or closed
+                                // because we only created it in the above line
+                                ConnBackend::Seqpacket(SeqpacketConn::new(unsafe {
+                                    OwnedFd::from_raw_fd(stream.into_raw_fd())
+                                })),
                                 uapi::VSOCK_HOST_CID,
                                 self.cid,
                                 pkt.hdr.dst_port(),
                                 pkt.hdr.src_port(),
                                 pkt.hdr.buf_alloc(),
                                 VsockType::Seqpacket,
-                            )),
+                            ),
                         )
                     })
                     .unwrap_or_else(|_| self.enq_rst(pkt.hdr.dst_port(), pkt.hdr.src_port()));
@@ -743,7 +710,7 @@ impl VsockMuxer {
     /// - kill the connection if an unrecoverable error occurs.
     fn apply_conn_mutation<F>(&mut self, key: ConnMapKey, mut_fn: F)
     where
-        F: FnOnce(&mut MuxerConn),
+        F: FnOnce(&mut VsockConnection<ConnBackend>),
     {
         if let Some(conn) = self.conn_map.get_mut(&key) {
             let had_rx = conn.has_pending_rx();
diff --git a/src/vmm/src/devices/virtio/vsock/unix/muxer_killq.rs b/src/vmm/src/devices/virtio/vsock/unix/muxer_killq.rs
@@ -27,10 +27,10 @@
 use std::collections::{HashMap, VecDeque};
 use std::time::Instant;
 
+use super::defs;
 use super::muxer::ConnMapKey;
-use super::{MuxerStreamConnection, defs};
-use crate::devices::virtio::vsock::csm::{VsockConnection, VsockConnectionBackend};
-use crate::devices::virtio::vsock::unix::MuxerConn;
+use crate::devices::virtio::vsock::csm::VsockConnection;
+use crate::devices::virtio::vsock::unix::ConnBackend;
 
 /// A kill queue item, holding the connection key and the scheduled time for termination.
 #[derive(Debug, Clone, Copy)]
@@ -68,7 +68,7 @@ impl MuxerKillQ {
     /// set to expire at some point in the future.
     /// Note: if more than `Self::SIZE` connections are found, the queue will be created in an
     ///       out-of-sync state, and will be discarded after it is emptied.
-    pub fn from_conn_map(conn_map: &HashMap<ConnMapKey, MuxerConn>) -> Self {
+    pub fn from_conn_map(conn_map: &HashMap<ConnMapKey, VsockConnection<ConnBackend>>) -> Self {
         let mut q_buf: Vec<MuxerKillQItem> = Vec::with_capacity(Self::SIZE);
         let mut synced = true;
         for (key, conn) in conn_map.iter() {
diff --git a/src/vmm/src/devices/virtio/vsock/unix/muxer_rxq.rs b/src/vmm/src/devices/virtio/vsock/unix/muxer_rxq.rs
@@ -20,8 +20,8 @@ use std::collections::{HashMap, VecDeque};
 use super::super::VsockChannel;
 use super::defs;
 use super::muxer::{ConnMapKey, MuxerRx};
-use crate::devices::virtio::vsock::csm::{VsockConnection, VsockConnectionBackend};
-use crate::devices::virtio::vsock::unix::MuxerConn;
+use crate::devices::virtio::vsock::csm::VsockConnection;
+use crate::devices::virtio::vsock::unix::ConnBackend;
 
 /// The muxer RX queue.
 #[derive(Debug)]
@@ -47,7 +47,7 @@ impl MuxerRxQ {
     /// Note: the resulting queue may still be desynchronized, if there are too many connections
     ///       that have pending RX data. In that case, the muxer will first drain this queue, and
     ///       then try again to build a synchronized one.
-    pub fn from_conn_map(conn_map: &HashMap<ConnMapKey, MuxerConn>) -> Self {
+    pub fn from_conn_map(conn_map: &HashMap<ConnMapKey, VsockConnection<ConnBackend>>) -> Self {
         let mut q = VecDeque::new();
         let mut synced = true;
 
diff --git a/src/vmm/src/devices/virtio/vsock/unix/seqpacket.rs b/src/vmm/src/devices/virtio/vsock/unix/seqpacket.rs
diff --git a/tests/integration_tests/functional/test_vsock.py b/tests/integration_tests/functional/test_vsock.py

Original file line number	Diff line number	Diff line change
`@@ -140,7 +140,7 @@ pub struct VsockConnection<S: VsockConnectionBackend> {`
`140`	`140`	`/// Instant when this connection should be scheduled for immediate termination, due to some`
`141`	`141`	`/// timeout condition having been fulfilled.`
`142`	`142`	`expiry: Option<Instant>,`
`143`		`- /// Vsock type (stream or seqpacket)`
	`143`	`+ /// The type of the underlying socket connection`
`144`	`144`	`vsock_type: VsockType,`
`145`	`145`	`}`
`146`	`146`