Merge pull request #257 from mahkoh/jorth/transfer-thread

vulkan: implement async shm uploads

src/allocator.rs

@@ -13,7 +13,7 @@ use {
 
 #[derive(Debug, Error)]
 #[error(transparent)]
-pub struct AllocatorError(#[from] pub Box<dyn Error>);
+pub struct AllocatorError(#[from] pub Box<dyn Error + Send>);
 
 bitflags! {
     BufferUsage: u32;

src/cli/input.rs

@@ -353,8 +353,10 @@ impl Input {
     async fn handle_keymap(&self, input: JayInputId) -> Vec<u8> {
         let data = Rc::new(RefCell::new(Vec::new()));
         jay_input::Keymap::handle(&self.tc, input, data.clone(), |d, map| {
-            let mem = Rc::new(ClientMem::new(map.keymap.raw(), map.keymap_len as _, true).unwrap())
-                .offset(0);
+            let mem = Rc::new(
+                ClientMem::new(&map.keymap, map.keymap_len as _, true, None, None).unwrap(),
+            )
+            .offset(0);
             mem.read(d.borrow_mut().deref_mut()).unwrap();
         });
         self.tc.round_trip().await;

src/clientmem.rs

@@ -1,14 +1,21 @@
 use {
-    crate::utils::vec_ext::VecExt,
+    crate::{
+        client::Client,
+        cpu_worker::{AsyncCpuWork, CpuJob, CpuWork, CpuWorker},
+        utils::vec_ext::VecExt,
+    },
     std::{
         cell::Cell,
-        mem::MaybeUninit,
+        mem::{ManuallyDrop, MaybeUninit},
         ptr,
         rc::Rc,
         sync::atomic::{compiler_fence, Ordering},
     },
     thiserror::Error,
-    uapi::{c, c::raise},
+    uapi::{
+        c::{self, raise},
+        OwnedFd,
+    },
 };
 
 #[derive(Debug, Error)]
@@ -22,27 +29,45 @@ pub enum ClientMemError {
 }
 
 pub struct ClientMem {
+    fd: ManuallyDrop<Rc<OwnedFd>>,
     failed: Cell<bool>,
     sigbus_impossible: bool,
     data: *const [Cell<u8>],
+    cpu: Option<Rc<CpuWorker>>,
 }
 
 #[derive(Clone)]
 pub struct ClientMemOffset {
     mem: Rc<ClientMem>,
+    offset: usize,
     data: *const [Cell<u8>],
 }
 
 impl ClientMem {
-    pub fn new(fd: i32, len: usize, read_only: bool) -> Result<Self, ClientMemError> {
+    pub fn new(
+        fd: &Rc<OwnedFd>,
+        len: usize,
+        read_only: bool,
+        client: Option<&Client>,
+        cpu: Option<&Rc<CpuWorker>>,
+    ) -> Result<Self, ClientMemError> {
         let mut sigbus_impossible = false;
-        if let Ok(seals) = uapi::fcntl_get_seals(fd) {
+        if let Ok(seals) = uapi::fcntl_get_seals(fd.raw()) {
             if seals & c::F_SEAL_SHRINK != 0 {
-                if let Ok(stat) = uapi::fstat(fd) {
+                if let Ok(stat) = uapi::fstat(fd.raw()) {
                     sigbus_impossible = stat.st_size as u64 >= len as u64;
                 }
             }
         }
+        if !sigbus_impossible {
+            if let Some(client) = client {
+                log::debug!(
+                    "Client {} ({}) has created a shm buffer that might cause SIGBUS",
+                    client.pid_info.comm,
+                    client.id,
+                );
+            }
+        }
         let data = if len == 0 {
             &mut [][..]
         } else {
@@ -51,17 +76,19 @@ impl ClientMem {
                 false => c::PROT_READ | c::PROT_WRITE,
             };
             unsafe {
-                let data = c::mmap64(ptr::null_mut(), len, prot, c::MAP_SHARED, fd, 0);
+                let data = c::mmap64(ptr::null_mut(), len, prot, c::MAP_SHARED, fd.raw(), 0);
                 if data == c::MAP_FAILED {
                     return Err(ClientMemError::MmapFailed(uapi::Errno::default().into()));
                 }
                 std::slice::from_raw_parts_mut(data as *mut Cell<u8>, len)
             }
         };
         Ok(Self {
+            fd: ManuallyDrop::new(fd.clone()),
             failed: Cell::new(false),
             sigbus_impossible,
             data,
+            cpu: cpu.cloned(),
         })
     }
 
@@ -73,12 +100,33 @@ impl ClientMem {
         let mem = unsafe { &*self.data };
         ClientMemOffset {
             mem: self.clone(),
+            offset,
             data: &mem[offset..],
         }
     }
+
+    pub fn fd(&self) -> &Rc<OwnedFd> {
+        &self.fd
+    }
+
+    pub fn sigbus_impossible(&self) -> bool {
+        self.sigbus_impossible
+    }
 }
 
 impl ClientMemOffset {
+    pub fn pool(&self) -> &ClientMem {
+        &self.mem
+    }
+
+    pub fn offset(&self) -> usize {
+        self.offset
+    }
+
+    pub fn ptr(&self) -> *const [Cell<u8>] {
+        self.data
+    }
+
     pub fn access<T, F: FnOnce(&[Cell<u8>]) -> T>(&self, f: F) -> Result<T, ClientMemError> {
         unsafe {
             if self.mem.sigbus_impossible {
@@ -114,8 +162,17 @@ impl ClientMemOffset {
 
 impl Drop for ClientMem {
     fn drop(&mut self) {
-        unsafe {
-            c::munmap(self.data as _, self.len());
+        let fd = unsafe { ManuallyDrop::take(&mut self.fd) };
+        if let Some(cpu) = &self.cpu {
+            let pending = cpu.submit(Box::new(CloseMemWork {
+                fd: Rc::try_unwrap(fd).ok(),
+                data: self.data,
+            }));
+            pending.detach();
+        } else {
+            unsafe {
+                c::munmap(self.data as _, self.len());
+            }
         }
     }
 }
@@ -178,3 +235,30 @@ pub fn init() -> Result<(), ClientMemError> {
         }
     }
 }
+
+struct CloseMemWork {
+    fd: Option<OwnedFd>,
+    data: *const [Cell<u8>],
+}
+
+unsafe impl Send for CloseMemWork {}
+
+impl CpuJob for CloseMemWork {
+    fn work(&mut self) -> &mut dyn CpuWork {
+        self
+    }
+
+    fn completed(self: Box<Self>) {
+        // nothing
+    }
+}
+
+impl CpuWork for CloseMemWork {
+    fn run(&mut self) -> Option<Box<dyn AsyncCpuWork>> {
+        self.fd.take();
+        unsafe {
+            c::munmap(self.data as _, self.data.len());
+        }
+        None
+    }
+}

src/compositor.rs

@@ -13,6 +13,7 @@ use {
         client::{ClientId, Clients},
         clientmem::{self, ClientMemError},
         config::ConfigProxy,
+        cpu_worker::{CpuWorker, CpuWorkerError},
         damage::{visualize_damage, DamageVisualizer},
         dbus::Dbus,
         ei::ei_client::EiClients,
@@ -107,6 +108,8 @@ pub enum CompositorError {
     WheelError(#[from] WheelError),
     #[error("Could not create an io-uring")]
     IoUringError(#[from] IoUringError),
+    #[error("Could not create cpu worker")]
+    CpuWorkerError(#[from] CpuWorkerError),
 }
 
 pub const WAYLAND_DISPLAY: &str = "WAYLAND_DISPLAY";
@@ -143,6 +146,7 @@ fn start_compositor2(
     let node_ids = NodeIds::default();
     let scales = RefCounted::default();
     scales.add(Scale::from_int(1));
+    let cpu_worker = Rc::new(CpuWorker::new(&ring, &engine)?);
     let state = Rc::new(State {
         xkb_ctx,
         backend: CloneCell::new(Rc::new(DummyBackend)),
@@ -258,6 +262,7 @@ fn start_compositor2(
         enable_ei_acceptor: Default::default(),
         ei_clients: EiClients::new(),
         slow_ei_clients: Default::default(),
+        cpu_worker,
     });
     state.tracker.register(ClientId::from_raw(0));
     create_dummy_output(&state);