Improved pipewire bridge support

crates/muvm/src/guest/bridge/common.rs

@@ -329,25 +329,13 @@ impl Context {
             channel_type,
             protocol_version: CROSS_DOMAIN_PROTOCOL_VERSION,
         };
-        this.submit_cmd(&init_cmd, mem::size_of::<CrossDomainInit>(), None, None)?;
+        this.submit_cmd(&init_cmd, mem::size_of::<CrossDomainInit>(), None)?;
         this.poll_cmd()?;
         Ok(this)
     }
 
-    pub fn submit_cmd<T>(
-        &self,
-        cmd: &T,
-        cmd_size: usize,
-        ring_idx: Option<u32>,
-        ring_handle: Option<u32>,
-    ) -> Result<()> {
-        submit_cmd_raw(
-            self.fd.as_raw_fd() as c_int,
-            cmd,
-            cmd_size,
-            ring_idx,
-            ring_handle,
-        )
+    pub fn submit_cmd<T>(&self, cmd: &T, cmd_size: usize, ring_idx: Option<u32>) -> Result<()> {
+        submit_cmd_raw(self.fd.as_raw_fd() as c_int, cmd, cmd_size, ring_idx)
     }
 
     fn poll_cmd(&self) -> Result<()> {
@@ -356,18 +344,11 @@ impl Context {
             &cmd,
             mem::size_of::<CrossDomainPoll>(),
             Some(CROSS_DOMAIN_CHANNEL_RING),
-            None,
         )
     }
 }
 
-pub fn submit_cmd_raw<T>(
-    fd: c_int,
-    cmd: &T,
-    cmd_size: usize,
-    ring_idx: Option<u32>,
-    ring_handle: Option<u32>,
-) -> Result<()> {
+pub fn submit_cmd_raw<T>(fd: c_int, cmd: &T, cmd_size: usize, ring_idx: Option<u32>) -> Result<()> {
     let cmd_buf = cmd as *const T as *const u8;
     let mut exec = DrmVirtgpuExecbuffer {
         command: cmd_buf as u64,
@@ -378,18 +359,10 @@ pub fn submit_cmd_raw<T>(
         exec.ring_idx = ring_idx;
         exec.flags = VIRTGPU_EXECBUF_RING_IDX;
     }
-    let ring_handle = &ring_handle;
-    if let Some(ring_handle) = ring_handle {
-        exec.bo_handles = ring_handle as *const u32 as u64;
-        exec.num_bo_handles = 1;
-    }
     // SAFETY: `exec` and `cmd` outlive the call, and it does not modify `cmd`
     unsafe {
         drm_virtgpu_execbuffer(fd, &mut exec)?;
     }
-    if ring_handle.is_some() {
-        unimplemented!();
-    }
     Ok(())
 }
 
@@ -488,6 +461,8 @@ pub trait ProtocolHandler: Sized {
     ) -> Result<StreamSendResult<Self::ResourceFinalizer>>;
 
     fn process_vgpu_extra(this: &mut Client<Self>, cmd: u8) -> Result<()>;
+
+    fn process_fd_extra(this: &mut Client<Self>, fd: u64, events: EpollFlags) -> Result<()>;
 }
 
 pub struct Client<'a, P: ProtocolHandler> {
@@ -695,7 +670,7 @@ impl<'a, P: ProtocolHandler> Client<'a, P> {
             ring_msg.identifier_types[i] = res.identifier_type;
             ring_msg.identifier_sizes[i] = res.identifier_size;
         }
-        self.gpu_ctx.submit_cmd(&ring_msg, size, None, None)?;
+        self.gpu_ctx.submit_cmd(&ring_msg, size, None)?;
         for fin in finalizers {
             fin.finalize(self)?;
         }
@@ -913,7 +888,16 @@ impl<'a, P: ProtocolHandler> Client<'a, P> {
                 );
             }
         } else {
-            unimplemented!()
+            let close = P::process_fd_extra(self, fd, events)
+                .map_err(|e| {
+                    let srv_id = self.gpu_ctx.fd.as_raw_fd() as u64;
+                    eprintln!("Server {srv_id} disconnected with error: {e:?}");
+                    e
+                })
+                .is_err();
+            if close {
+                self.sub_poll.close();
+            }
         }
     }
 }

crates/muvm/src/guest/bridge/pipewire.rs

@@ -17,6 +17,7 @@ use crate::guest::bridge::common::{
 const CROSS_DOMAIN_CHANNEL_TYPE_PW: u32 = 0x10;
 const CROSS_DOMAIN_CMD_READ_EVENTFD_NEW: u8 = 11;
 const CROSS_DOMAIN_CMD_READ: u8 = 6;
+const CROSS_DOMAIN_CMD_WRITE: u8 = 7;
 
 const SPA_TYPE_STRUCT: u32 = 14;
 
@@ -149,6 +150,7 @@ impl MessageResourceFinalizer for PipeWireResourceFinalizer {
 
 struct CrossDomainEventFd {
     event_fd: EventFd,
+    resource: u32,
 }
 
 struct ClientNodeData {
@@ -172,14 +174,22 @@ struct PipeWireProtocolHandler {
 }
 
 impl PipeWireProtocolHandler {
-    fn create_guest_to_host_eventfd(this: &mut Client<Self>, node_id: u32) -> Result<OwnedFd> {
+    fn create_guest_to_host_eventfd(
+        this: &mut Client<Self>,
+        node_id: u32,
+        resource: CrossDomainResource,
+    ) -> Result<OwnedFd> {
         let efd = EventFd::from_flags(EfdFlags::EFD_NONBLOCK)?;
         let ofd = efd.as_fd().try_clone_to_owned()?;
         this.sub_poll.add(efd.as_fd(), EpollFlags::EPOLLIN);
         let raw = efd.as_raw_fd() as u64;
-        this.protocol_handler
-            .guest_to_host_eventfds
-            .insert(raw, CrossDomainEventFd { event_fd: efd });
+        this.protocol_handler.guest_to_host_eventfds.insert(
+            raw,
+            CrossDomainEventFd {
+                event_fd: efd,
+                resource: resource.identifier,
+            },
+        );
         this.protocol_handler
             .client_nodes
             .get_mut(&node_id)
@@ -208,10 +218,14 @@ impl PipeWireProtocolHandler {
             .unwrap()
             .host_to_guest
             .push(resource.identifier);
-        this.gpu_ctx.submit_cmd(&msg, msg_size, None, None)?;
-        this.protocol_handler
-            .host_to_guest_eventfds
-            .insert(resource.identifier, CrossDomainEventFd { event_fd: efd });
+        this.gpu_ctx.submit_cmd(&msg, msg_size, None)?;
+        this.protocol_handler.host_to_guest_eventfds.insert(
+            resource.identifier,
+            CrossDomainEventFd {
+                event_fd: efd,
+                resource: resource.identifier,
+            },
+        );
         Ok(ofd)
     }
 }
@@ -249,13 +263,13 @@ impl ProtocolHandler for PipeWireProtocolHandler {
                 fds.push(this.virtgpu_id_to_prime(rsc)?);
             } else if this.protocol_handler.client_nodes.contains_key(&hdr.id) {
                 if hdr.opcode == PW_OPC_CLIENT_NODE_SET_ACTIVATION {
-                    resources.pop_front().ok_or(Errno::EIO)?;
-                    fds.push(Self::create_guest_to_host_eventfd(this, hdr.id)?);
+                    let rsc = resources.pop_front().ok_or(Errno::EIO)?;
+                    fds.push(Self::create_guest_to_host_eventfd(this, hdr.id, rsc)?);
                 } else if hdr.opcode == PW_OPC_CLIENT_NODE_TRANSPORT {
                     let rsc1 = resources.pop_front().ok_or(Errno::EIO)?;
                     fds.push(Self::create_host_to_guest_eventfd(this, hdr.id, rsc1)?);
-                    resources.pop_front().ok_or(Errno::EIO)?;
-                    fds.push(Self::create_guest_to_host_eventfd(this, hdr.id)?);
+                    let rsc2 = resources.pop_front().ok_or(Errno::EIO)?;
+                    fds.push(Self::create_guest_to_host_eventfd(this, hdr.id, rsc2)?);
                 } else {
                     unimplemented!()
                 }
@@ -332,6 +346,25 @@ impl ProtocolHandler for PipeWireProtocolHandler {
             Err(Errno::ENOENT.into())
         }
     }
+
+    fn process_fd_extra(this: &mut Client<Self>, fd: u64, _: EpollFlags) -> Result<()> {
+        let efd = this
+            .protocol_handler
+            .guest_to_host_eventfds
+            .get(&fd)
+            .ok_or(Errno::ENOENT)?;
+        let msg_size = mem::size_of::<CrossDomainReadWrite<[u8; mem::size_of::<u64>()]>>();
+        let val = efd.event_fd.read()?;
+        let msg = CrossDomainReadWrite {
+            hdr: CrossDomainHeader::new(CROSS_DOMAIN_CMD_WRITE, msg_size as u16),
+            identifier: efd.resource,
+            hang_up: 0,
+            opaque_data_size: mem::size_of::<u64>() as u32,
+            pad: 0,
+            data: val.to_ne_bytes(),
+        };
+        this.gpu_ctx.submit_cmd(&msg, msg_size, None)
+    }
 }
 
 pub fn start_pwbridge() {

crates/muvm/src/guest/bridge/x11.rs

@@ -19,6 +19,7 @@ use nix::libc::{
     SYS_openat, AT_FDCWD, MAP_ANONYMOUS, MAP_FIXED, MAP_PRIVATE, MAP_SHARED, O_CLOEXEC, O_RDWR,
     PROT_READ, PROT_WRITE,
 };
+use nix::sys::epoll::EpollFlags;
 use nix::sys::mman::{mmap, munmap, MapFlags, ProtFlags};
 use nix::sys::ptrace;
 use nix::sys::signal::Signal;
@@ -121,7 +122,7 @@ impl MessageResourceFinalizer for X11ResourceFinalizer {
                 };
                 client
                     .gpu_ctx
-                    .submit_cmd(&ft_msg, ft_destroy_msg_size, None, None)?;
+                    .submit_cmd(&ft_msg, ft_destroy_msg_size, None)?;
             },
         }
         Ok(())
@@ -337,6 +338,9 @@ impl ProtocolHandler for X11ProtocolHandler {
         };
         this.protocol_handler.process_futex_signal(recv)
     }
+    fn process_fd_extra(_: &mut Client<Self>, _: u64, _: EpollFlags) -> Result<()> {
+        unreachable!()
+    }
 }
 
 impl X11ProtocolHandler {
@@ -494,8 +498,7 @@ impl X11ProtocolHandler {
             handle: handle.handle,
             pad: 0,
         };
-        this.gpu_ctx
-            .submit_cmd(&ft_msg, ft_new_msg_size, None, None)?;
+        this.gpu_ctx.submit_cmd(&ft_msg, ft_new_msg_size, None)?;
         let fd = this.gpu_ctx.fd.as_raw_fd() as c_int;
         this.protocol_handler
             .futex_watchers
@@ -597,7 +600,7 @@ impl FutexWatcherThread {
                     id: xid,
                     pad: 0,
                 };
-                common::submit_cmd_raw(fd, &ft_signal_cmd, ft_signal_msg_size, None, None).unwrap();
+                common::submit_cmd_raw(fd, &ft_signal_cmd, ft_signal_msg_size, None).unwrap();
             }
         });
         FutexWatcherThread {

crates/muvm/src/hidpipe_server.rs

@@ -104,7 +104,7 @@ struct EvdevContainer {
     names_to_fds: HashMap<String, u64>,
 }
 
-fn insert_entry<K, V>(entry: hash_map::Entry<K, V>, v: V) -> &V {
+fn insert_entry<K, V>(entry: hash_map::Entry<'_, K, V>, v: V) -> &V {
     match entry {
         hash_map::Entry::Vacant(e) => e.insert(v),
         hash_map::Entry::Occupied(mut e) => {

crates/muvm/src/net.rs

@@ -24,7 +24,7 @@ fn parse_range(r: &str) -> Result<(u32, u32)> {
 }
 
 impl PublishSpec<'_> {
-    fn parse(mut arg: &str) -> Result<PublishSpec> {
+    fn parse(mut arg: &str) -> Result<PublishSpec<'_>> {
         let mut udp = false;
         if arg.ends_with("/udp") {
             udp = true;