libkrun X11 pipewire support

ffi/src/include/rutabaga_gfx_ffi.h

@@ -92,6 +92,8 @@ extern "C" {
  * Rutabaga channel types
  */
 #define RUTABAGA_CHANNEL_TYPE_WAYLAND 1
+#define RUTABAGA_CHANNEL_TYPE_PIPEWIRE 0x10
+#define RUTABAGA_CHANNEL_TYPE_X11 0x11
 
 /**
  * Rutabaga WSI

ffi/src/tests/virtgpu_cross_domain_protocol.h

@@ -19,6 +19,8 @@
 // Channel types (must match rutabaga channel types)
 #define CROSS_DOMAIN_CHANNEL_TYPE_WAYLAND 0x0001
 #define CROSS_DOMAIN_CHANNEL_TYPE_CAMERA 0x0002
+#define CROSS_DOMAIN_CHANNEL_TYPE_PIPEWIRE 0x0010
+#define CROSS_DOMAIN_CHANNEL_TYPE_X11 0x0011
 
 // The maximum number of identifiers (value based on wp_linux_dmabuf)
 #define CROSS_DOMAIN_MAX_IDENTIFIERS 4

src/cross_domain/atomic_memory_sentinel_manager.rs

@@ -0,0 +1,270 @@
+// Copyright 2026 Red Hat, Inc.
+// Copyright 2021 The ChromiumOS Authors
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+//! Manager for atomic memory sentinel instances.
+//!
+//! This module encapsulates the management of AtomicMemorySentinelThread instances,
+//! providing a clean interface for creating, signaling, and destroying memory watchers.
+
+use std::collections::BTreeMap as Map;
+use std::mem::size_of;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::Arc;
+use std::thread;
+
+use log::info;
+use mesa3d_util::AtomicMemorySentinel;
+use mesa3d_util::Event;
+use mesa3d_util::MemoryMapping;
+use mesa3d_util::MesaError;
+use mesa3d_util::OwnedDescriptor;
+
+use crate::rutabaga_core::VirtioFsKey;
+use crate::rutabaga_core::VirtioFsTable;
+use crate::rutabaga_utils::RutabagaError;
+use crate::rutabaga_utils::RutabagaResult;
+use crate::rutabaga_utils::RUTABAGA_MAP_ACCESS_RW;
+use crate::rutabaga_utils::RUTABAGA_MAP_CACHE_CACHED;
+
+use super::cross_domain_protocol::CrossDomainSignalAtomicMemorySentinel;
+
+/// Thread worker for monitoring atomic memory changes.
+struct AtomicMemorySentinelThread {
+    sentinel: Arc<AtomicMemorySentinel>,
+    shutdown: Arc<AtomicBool>,
+    sender: Event,
+    initial_value: u32,
+}
+
+impl AtomicMemorySentinelThread {
+    fn new(
+        sentinel: Arc<AtomicMemorySentinel>,
+        shutdown: Arc<AtomicBool>,
+        sender: Event,
+        initial_value: u32,
+    ) -> Self {
+        Self {
+            sentinel,
+            shutdown,
+            sender,
+            initial_value,
+        }
+    }
+
+    fn run(mut self) {
+        // The goal of this code is to ensure that the other side observes at least
+        // the latest wake event along with the value that the futex had when that
+        // wake event was signaled.
+
+        // The initial value is passed in from the futex creation, and therefore
+        // was loaded synchronously with the cross domain operation that created
+        // the futex, so it cannot have an associated wake event yet.
+        let mut val = self.initial_value;
+        let _ = self.sender.signal();
+
+        loop {
+            // This returns when the futex is woken up OR if the value has changed.
+            self.sentinel.wait(val);
+            // Load the new value, which the other side is guaranteed to observe.
+            val = self.sentinel.load();
+
+            // If this wake was triggered by the shutdown code below, just bail.
+            // If the shutdown command is issued after this point, then it will
+            // change the futex value, which will disagree with the one we read
+            // above, so we will still not block in futex wait.
+            if self.shutdown.load(Ordering::SeqCst) {
+                // Signal the futex to process the shutdown and remove it from
+                // the waiter table
+                let _ = self.sender.signal();
+                break;
+            }
+
+            // Signal the other side after the load. If another change occurs and
+            // another wake is signaled here, we will miss the wake, but the
+            // disagreeing value will cause futex wait to return early.
+            if self.sender.signal().is_err() {
+                break;
+            }
+        }
+    }
+}
+
+/// Private struct containing a single memory watcher instance
+struct SentinelInstance {
+    sentinel: Arc<AtomicMemorySentinel>,
+    watcher_thread: Option<thread::JoinHandle<()>>,
+    shutdown: Arc<AtomicBool>,
+    evt: Event,
+}
+
+impl SentinelInstance {
+    fn shutdown(&mut self) {
+        self.shutdown.store(true, Ordering::SeqCst);
+        self.sentinel.wake_all();
+        if let Some(thread) = self.watcher_thread.take() {
+            let _ = thread.join();
+        }
+    }
+
+    fn is_shutdown(&self) -> bool {
+        self.shutdown.load(Ordering::Relaxed)
+    }
+}
+
+impl Drop for SentinelInstance {
+    fn drop(&mut self) {
+        if !self.is_shutdown() {
+            info!("Atomic memory sentinel worker dropped without shutdown");
+            self.shutdown();
+        }
+    }
+}
+
+/// Manager for all atomic memory sentinel watchers
+pub struct AtomicMemorySentinelManager {
+    watchers: Map<u32, SentinelInstance>,
+    virtiofs_table: Option<VirtioFsTable>,
+}
+
+impl AtomicMemorySentinelManager {
+    /// Creates a new AtomicMemorySentinelManager
+    pub fn new(virtiofs_table: Option<VirtioFsTable>) -> Self {
+        Self {
+            watchers: Map::new(),
+            virtiofs_table,
+        }
+    }
+
+    /// Creates a new memory watcher and returns its ID and event descriptor
+    pub fn create_watcher(&mut self, id: u32, fs_id: u64, handle: u64) -> RutabagaResult<Event> {
+        if self.watchers.contains_key(&id) {
+            return Err(RutabagaError::AlreadyInUse);
+        }
+
+        let virtiofs_table = self
+            .virtiofs_table
+            .as_ref()
+            .ok_or(RutabagaError::InvalidCrossDomainItemId)?;
+        let virtiofs = virtiofs_table.lock().unwrap();
+
+        let key = VirtioFsKey { fs_id, handle };
+        let file = virtiofs
+            .get(&key)
+            .ok_or(RutabagaError::InvalidCrossDomainItemId)?
+            .try_clone()
+            .map_err(MesaError::IoError)?;
+
+        let handle = OwnedDescriptor::from(file);
+        let mapping = MemoryMapping::from_safe_descriptor(
+            handle,
+            size_of::<u32>(),
+            RUTABAGA_MAP_ACCESS_RW | RUTABAGA_MAP_CACHE_CACHED,
+        )?;
+
+        let sentinel = Arc::new(AtomicMemorySentinel::new(mapping)?);
+        let initial_value = sentinel.load();
+
+        let memory_watcher_evt = Event::new()?;
+        let evt_for_waitctx = memory_watcher_evt.try_clone()?;
+
+        let shutdown = Arc::new(AtomicBool::new(false));
+        let thread_shutdown = shutdown.clone();
+        let thread_sentinel = sentinel.clone();
+
+        // Spawn the watcher thread
+        let watcher_thread = thread::Builder::new()
+            .name(format!("atomic_memory_sentinel_{}", id))
+            .spawn(move || {
+                AtomicMemorySentinelThread::new(
+                    thread_sentinel,
+                    thread_shutdown,
+                    memory_watcher_evt,
+                    initial_value,
+                )
+                .run();
+            })
+            .map_err(MesaError::IoError)?;
+
+        self.watchers.insert(
+            id,
+            SentinelInstance {
+                sentinel,
+                watcher_thread: Some(watcher_thread),
+                shutdown,
+                evt: evt_for_waitctx.try_clone()?,
+            },
+        );
+
+        Ok(evt_for_waitctx)
+    }
+
+    /// Handles an event for a memory watcher, returns the command to write to the ring
+    pub fn handle_event(
+        &self,
+        id: u32,
+    ) -> RutabagaResult<Option<CrossDomainSignalAtomicMemorySentinel>> {
+        if let Some(watcher) = self.watchers.get(&id) {
+            if watcher.is_shutdown() {
+                Ok(None)
+            } else {
+                watcher.evt.wait()?;
+
+                let mut cmd_memory_watcher: CrossDomainSignalAtomicMemorySentinel =
+                    Default::default();
+                cmd_memory_watcher.hdr.cmd =
+                    super::cross_domain_protocol::CROSS_DOMAIN_CMD_SIGNAL_ATOMIC_MEMORY_SENTINEL;
+                cmd_memory_watcher.id = id;
+                Ok(Some(cmd_memory_watcher))
+            }
+        } else {
+            Err(RutabagaError::InvalidCrossDomainItemId)
+        }
+    }
+
+    /// Signals a memory watcher
+    pub fn signal_watcher(&self, id: u32) -> RutabagaResult<()> {
+        if let Some(worker) = self.watchers.get(&id) {
+            worker.sentinel.signal()?;
+            Ok(())
+        } else {
+            Err(RutabagaError::InvalidCrossDomainItemId)
+        }
+    }
+
+    /// Destroys a memory watcher
+    pub fn destroy_watcher(&mut self, id: u32) -> RutabagaResult<()> {
+        self.watchers
+            .get_mut(&id)
+            .ok_or(RutabagaError::InvalidCrossDomainItemId)?
+            .shutdown();
+        Ok(())
+    }
+
+    /// Checks if a watcher is shutdown (for cleanup in handle_fence)
+    pub fn is_shutdown(&self, id: u32) -> bool {
+        self.watchers
+            .get(&id)
+            .map(|w| w.is_shutdown())
+            .unwrap_or(true)
+    }
+
+    /// Removes a watcher from the map (after shutdown)
+    pub fn remove_watcher(&mut self, id: u32) {
+        self.watchers.remove(&id);
+    }
+
+    /// Gets the event descriptor for a watcher
+    pub fn get_event(&self, id: u32) -> Option<&Event> {
+        self.watchers.get(&id).map(|w| &w.evt)
+    }
+}
+
+impl Drop for AtomicMemorySentinelManager {
+    fn drop(&mut self) {
+        for (_, mut watcher) in std::mem::take(&mut self.watchers) {
+            watcher.shutdown();
+        }
+    }
+}

src/cross_domain/cross_domain_protocol.rs

@@ -19,10 +19,16 @@ pub const CROSS_DOMAIN_CMD_SEND: u8 = 4;
 pub const CROSS_DOMAIN_CMD_RECEIVE: u8 = 5;
 pub const CROSS_DOMAIN_CMD_READ: u8 = 6;
 pub const CROSS_DOMAIN_CMD_WRITE: u8 = 7;
+pub const CROSS_DOMAIN_CMD_CREATE_ATOMIC_MEMORY_SENTINEL: u8 = 8;
+pub const CROSS_DOMAIN_CMD_SIGNAL_ATOMIC_MEMORY_SENTINEL: u8 = 9;
+pub const CROSS_DOMAIN_CMD_DESTROY_ATOMIC_MEMORY_SENTINEL: u8 = 10;
+pub const CROSS_DOMAIN_CMD_READ_CREATE_EVENT: u8 = 11;
 
 /// Channel types (must match rutabaga channel types)
 pub const CROSS_DOMAIN_CHANNEL_TYPE_WAYLAND: u32 = 0x0001;
 pub const CROSS_DOMAIN_CHANNEL_TYPE_CAMERA: u32 = 0x0002;
+pub const CROSS_DOMAIN_CHANNEL_TYPE_PIPEWIRE: u32 = 0x0010;
+pub const CROSS_DOMAIN_CHANNEL_TYPE_X11: u32 = 0x0011;
 
 /// The maximum number of identifiers
 pub const CROSS_DOMAIN_MAX_IDENTIFIERS: usize = 28;
@@ -38,6 +44,8 @@ pub const CROSS_DOMAIN_ID_TYPE_READ_PIPE: u32 = 3;
 /// The host receives the write end of the pipe over the host Wayland socket.
 pub const CROSS_DOMAIN_ID_TYPE_WRITE_PIPE: u32 = 4;
 
+pub const CROSS_DOMAIN_ID_TYPE_EVENT: u32 = 5;
+
 /// No ring
 pub const CROSS_DOMAIN_RING_NONE: u32 = 0xffffffff;
 /// A ring for metadata queries.
@@ -47,6 +55,8 @@ pub const CROSS_DOMAIN_CHANNEL_RING: u32 = 1;
 
 /// Read pipe IDs start at this value.
 pub const CROSS_DOMAIN_PIPE_READ_START: u32 = 0x80000000;
+/// Memory watcher IDs start at this value.
+pub const CROSS_DOMAIN_ATOMIC_MEMORY_SENTINEL_START: u32 = 0x40000000;
 
 #[repr(C)]
 #[derive(Copy, Clone, Debug, Default, FromBytes, IntoBytes, Immutable)]
@@ -80,7 +90,7 @@ pub struct CrossDomainHeader {
 }
 
 #[repr(C)]
-#[derive(Copy, Clone, Debug, Default, FromBytes, IntoBytes, Immutable)]
+#[derive(Copy, Clone, Default, FromBytes, IntoBytes, Immutable)]
 pub struct CrossDomainInit {
     pub hdr: CrossDomainHeader,
     pub query_ring_id: u32,
@@ -120,3 +130,40 @@ pub struct CrossDomainReadWrite {
     pub pad: u32,
     // Data of size "opaque data size follows"
 }
+
+#[repr(C)]
+#[derive(Copy, Clone, Default, FromBytes, IntoBytes, Immutable)]
+pub struct CrossDomainCreateAtomicMemorySentinel {
+    pub hdr: CrossDomainHeader,
+    /// VirtioFS filesystem ID - identifies which virtio-fs instance
+    pub fs_id: u64,
+    /// VirtioFS file handle - identifies the file within the filesystem to map and watch
+    pub handle: u64,
+    /// Memory watcher ID - unique ID for this watcher, used in signal/destroy commands
+    pub id: u32,
+    pub pad: u32,
+}
+
+#[repr(C)]
+#[derive(Copy, Clone, Default, FromBytes, IntoBytes, Immutable)]
+pub struct CrossDomainSignalAtomicMemorySentinel {
+    pub hdr: CrossDomainHeader,
+    pub id: u32,
+    pub pad: u32,
+}
+
+#[repr(C)]
+#[derive(Copy, Clone, Default, FromBytes, IntoBytes, Immutable)]
+pub struct CrossDomainDestroyAtomicMemorySentinel {
+    pub hdr: CrossDomainHeader,
+    pub id: u32,
+    pub pad: u32,
+}
+
+#[repr(C)]
+#[derive(Copy, Clone, Default, FromBytes, IntoBytes, Immutable)]
+pub struct CrossDomainCreateEvent {
+    pub hdr: CrossDomainHeader,
+    pub id: u32,
+    pub pad: u32,
+}