vulkan: implement async shm textures

src/clientmem.rs

@@ -98,29 +98,24 @@ impl ClientMem {
         }
     }
 
-    #[expect(dead_code)]
     pub fn fd(&self) -> &Rc<OwnedFd> {
         &self.fd
     }
 
-    #[expect(dead_code)]
     pub fn sigbus_impossible(&self) -> bool {
         self.sigbus_impossible
     }
 }
 
 impl ClientMemOffset {
-    #[expect(dead_code)]
     pub fn pool(&self) -> &ClientMem {
         &self.mem
     }
 
-    #[expect(dead_code)]
     pub fn offset(&self) -> usize {
         self.offset
     }
 
-    #[expect(dead_code)]
     pub fn ptr(&self) -> *const [Cell<u8>] {
         self.data
     }

src/cpu_worker/jobs.rs

@@ -1 +1,2 @@
+pub mod img_copy;
 pub mod read_write;

src/cpu_worker/jobs/img_copy.rs

@@ -0,0 +1,62 @@
+use {
+    crate::{
+        cpu_worker::{AsyncCpuWork, CpuWork},
+        rect::Rect,
+    },
+    std::ptr,
+};
+
+#[expect(clippy::manual_non_exhaustive)]
+pub struct ImgCopyWork {
+    pub src: *mut u8,
+    pub dst: *mut u8,
+    pub width: i32,
+    pub stride: i32,
+    pub bpp: i32,
+    pub rects: Vec<Rect>,
+    _priv: (),
+}
+
+unsafe impl Send for ImgCopyWork {}
+
+impl ImgCopyWork {
+    pub unsafe fn new() -> Self {
+        Self {
+            src: ptr::null_mut(),
+            dst: ptr::null_mut(),
+            width: 0,
+            stride: 0,
+            bpp: 0,
+            rects: vec![],
+            _priv: (),
+        }
+    }
+}
+
+impl CpuWork for ImgCopyWork {
+    fn run(&mut self) -> Option<Box<dyn AsyncCpuWork>> {
+        for rect in &self.rects {
+            let mut offset = rect.y1() * self.stride + rect.x1() * self.bpp;
+            if rect.width() == self.width {
+                let offset = offset as usize;
+                let len = rect.height() * self.stride;
+                unsafe {
+                    ptr::copy_nonoverlapping(self.src.add(offset), self.dst.add(offset), len as _);
+                }
+            } else {
+                let len = rect.width() * self.bpp;
+                for _ in 0..rect.height() {
+                    unsafe {
+                        ptr::copy_nonoverlapping(
+                            self.src.add(offset as _),
+                            self.dst.add(offset as _),
+                            len as _,
+                        );
+                    }
+                    offset += self.stride;
+                }
+            }
+        }
+        None
+    }
+}

src/cpu_worker/jobs/read_write.rs

@@ -36,7 +36,6 @@ pub struct ReadWriteWork {
 unsafe impl Send for ReadWriteWork {}
 
 impl ReadWriteWork {
-    #[expect(dead_code)]
     pub unsafe fn new() -> Self {
         let cancel = Arc::new(CancelState::default());
         ReadWriteWork {
@@ -53,7 +52,6 @@ impl ReadWriteWork {
         }
     }
 
-    #[expect(dead_code)]
     pub fn config(&mut self) -> &mut ReadWriteWorkConfig {
         self.config.as_mut().unwrap()
     }

src/gfx_api.rs

@@ -538,7 +538,6 @@ pub trait ShmGfxTexture: GfxTexture {
 }
 
 pub trait AsyncShmGfxTextureCallback {
-    #[expect(dead_code)]
     fn completed(self: Rc<Self>, res: Result<(), GfxError>);
 }
 
@@ -689,7 +688,6 @@ pub fn cross_intersect_formats(
 }
 
 impl PendingShmUpload {
-    #[expect(dead_code)]
     pub fn new(cancel: Rc<dyn AsyncShmGfxTextureUploadCancellable>, id: u64) -> Self {
         Self { cancel, id }
     }

src/gfx_apis/vulkan.rs

@@ -19,7 +19,7 @@ use {
     crate::{
         allocator::{Allocator, AllocatorError},
         async_engine::AsyncEngine,
-        cpu_worker::CpuWorker,
+        cpu_worker::{jobs::read_write::ReadWriteJobError, CpuWorker},
         format::Format,
         gfx_api::{
             AsyncShmGfxTexture, GfxContext, GfxError, GfxFormat, GfxFramebuffer, GfxImage,
@@ -199,6 +199,10 @@ pub enum VulkanError {
     DupDrm(#[source] DrmError),
     #[error("Graphics context has already been dropped")]
     Defunct,
+    #[error("Could not perform an async copy to the staging buffer")]
+    AsyncCopyToStaging(#[source] ReadWriteJobError),
+    #[error("The async shm texture is busy")]
+    AsyncCopyBusy,
 }
 
 impl From<VulkanError> for GfxError {
@@ -278,19 +282,28 @@ impl GfxContext for Context {
         }
         let tex = self
             .0
-            .create_shm_texture(format, width, height, stride, data, false)?;
+            .create_shm_texture(format, width, height, stride, data, false, None)?;
         Ok(tex as _)
     }
 
     fn async_shmem_texture(
         self: Rc<Self>,
-        _format: &'static Format,
-        _width: i32,
-        _height: i32,
-        _stride: i32,
-        _cpu_worker: &Rc<CpuWorker>,
+        format: &'static Format,
+        width: i32,
+        height: i32,
+        stride: i32,
+        cpu_worker: &Rc<CpuWorker>,
     ) -> Result<Rc<dyn AsyncShmGfxTexture>, GfxError> {
-        todo!()
+        let tex = self.0.create_shm_texture(
+            format,
+            width,
+            height,
+            stride,
+            &[],
+            false,
+            Some(cpu_worker),
+        )?;
+        Ok(tex)
     }
 
     fn allocator(&self) -> Rc<dyn Allocator> {
@@ -310,7 +323,7 @@ impl GfxContext for Context {
     ) -> Result<Rc<dyn GfxFramebuffer>, GfxError> {
         let fb = self
             .0
-            .create_shm_texture(format, width, height, stride, &[], true)?;
+            .create_shm_texture(format, width, height, stride, &[], true, None)?;
         Ok(fb)
     }
 

src/gfx_apis/vulkan/image.rs

@@ -1,13 +1,17 @@
 use {
     crate::{
+        clientmem::ClientMemOffset,
         format::Format,
         gfx_api::{
-            GfxApiOpt, GfxError, GfxFramebuffer, GfxImage, GfxTexture, ShmGfxTexture, SyncFile,
+            AsyncShmGfxTexture, AsyncShmGfxTextureCallback, AsyncShmGfxTextureUploadCancellable,
+            GfxApiOpt, GfxError, GfxFramebuffer, GfxImage, GfxTexture, PendingShmUpload,
+            ShmGfxTexture, SyncFile,
         },
         gfx_apis::vulkan::{
             allocator::VulkanAllocation, device::VulkanDevice, format::VulkanModifierLimits,
             renderer::VulkanRenderer, shm_image::VulkanShmImage, VulkanError,
         },
+        rect::Region,
         theme::Color,
         utils::{clonecell::CloneCell, on_drop::OnDrop},
         video::dmabuf::{DmaBuf, PlaneVec},
@@ -53,6 +57,7 @@ pub struct VulkanImage {
     pub(super) render_view: Option<ImageView>,
     pub(super) image: Image,
     pub(super) is_undefined: Cell<bool>,
+    pub(super) contents_are_undefined: Cell<bool>,
     pub(super) ty: VulkanImageMemory,
     pub(super) render_ops: CloneCell<Vec<GfxApiOpt>>,
     pub(super) bridge: Option<VulkanFramebufferBridge>,
@@ -380,6 +385,7 @@ impl VulkanDmaBufImageTemplate {
             }),
             format: self.dmabuf.format,
             is_undefined: Cell::new(true),
+            contents_are_undefined: Cell::new(false),
             bridge,
         }))
     }
@@ -538,3 +544,58 @@ impl ShmGfxTexture for VulkanImage {
         self
     }
 }
+
+impl AsyncShmGfxTexture for VulkanImage {
+    fn async_upload(
+        self: Rc<Self>,
+        callback: Rc<dyn AsyncShmGfxTextureCallback>,
+        mem: &Rc<ClientMemOffset>,
+        damage: Region,
+    ) -> Result<Option<PendingShmUpload>, GfxError> {
+        let VulkanImageMemory::Internal(shm) = &self.ty else {
+            unreachable!();
+        };
+        let pending = shm.async_upload(&self, mem, damage, callback)?;
+        Ok(pending)
+    }
+
+    fn sync_upload(self: Rc<Self>, mem: &[Cell<u8>], damage: Region) -> Result<(), GfxError> {
+        let VulkanImageMemory::Internal(shm) = &self.ty else {
+            unreachable!();
+        };
+        if shm.async_data.as_ref().unwrap().busy.get() {
+            return Err(VulkanError::AsyncCopyBusy.into());
+        }
+        shm.upload(&self, mem, Some(damage.rects()))?;
+        Ok(())
+    }
+
+    fn compatible_with(
+        &self,
+        format: &'static Format,
+        width: i32,
+        height: i32,
+        stride: i32,
+    ) -> bool {
+        self.format == format
+            && self.width == width as u32
+            && self.height == height as u32
+            && self.stride == stride as u32
+    }
+
+    fn into_texture(self: Rc<Self>) -> Rc<dyn GfxTexture> {
+        self
+    }
+}
+
+impl AsyncShmGfxTextureUploadCancellable for VulkanImage {
+    fn cancel(&self, id: u64) {
+        let VulkanImageMemory::Internal(shm) = &self.ty else {
+            unreachable!();
+        };
+        let data = shm.async_data.as_ref().unwrap();
+        if data.callback_id.get() == id {
+            data.callback.take();
+        }
+    }
+}

src/gfx_apis/vulkan/renderer.rs

@@ -279,6 +279,9 @@ impl VulkanRenderer {
         for cmd in opts {
             if let GfxApiOpt::CopyTexture(c) = cmd {
                 let tex = c.tex.clone().into_vk(&self.device.device);
+                if tex.contents_are_undefined.get() {
+                    continue;
+                }
                 if let VulkanImageMemory::DmaBuf(_) = &tex.ty {
                     memory.sample.push(tex.clone())
                 }
@@ -458,6 +461,10 @@ impl VulkanRenderer {
                 }
                 GfxApiOpt::CopyTexture(c) => {
                     let tex = c.tex.as_vk(&self.device.device);
+                    if tex.contents_are_undefined.get() {
+                        log::warn!("Ignoring undefined texture");
+                        continue;
+                    }
                     let copy_type = match c.alpha.is_some() {
                         true => TexCopyType::Multiply,
                         false => TexCopyType::Identity,
@@ -807,6 +814,7 @@ impl VulkanRenderer {
             stride as i32,
             &[],
             true,
+            None,
         )?;
         (&*tmp_tex as &dyn GfxFramebuffer)
             .copy_texture(

src/gfx_apis/vulkan/shm_image.rs

@@ -1,7 +1,15 @@
 use {
     crate::{
+        clientmem::ClientMemOffset,
+        cpu_worker::{
+            jobs::{
+                img_copy::ImgCopyWork,
+                read_write::{ReadWriteJobError, ReadWriteWork},
+            },
+            CpuJob, CpuWork, CpuWorker,
+        },
         format::{Format, FormatShmInfo},
-        gfx_api::SyncFile,
+        gfx_api::{AsyncShmGfxTextureCallback, PendingShmUpload, SyncFile},
         gfx_apis::vulkan::{
             allocator::VulkanAllocation,
             command::VulkanCommandBuffer,
@@ -11,8 +19,8 @@ use {
             staging::VulkanStagingBuffer,
             VulkanError,
         },
-        rect::Rect,
-        utils::{errorfmt::ErrorFmt, on_drop::OnDrop},
+        rect::{Rect, Region},
+        utils::{clonecell::CloneCell, errorfmt::ErrorFmt, on_drop::OnDrop},
     },
     ash::vk::{
         AccessFlags2, BufferImageCopy2, BufferMemoryBarrier2, CommandBufferBeginInfo,
@@ -24,14 +32,32 @@ use {
     },
     gpu_alloc::UsageFlags,
     isnt::std_1::primitive::IsntSliceExt,
-    std::{cell::Cell, ptr, rc::Rc, slice},
+    std::{
+        cell::{Cell, RefCell},
+        ptr,
+        rc::Rc,
+        slice,
+    },
+    uapi::OwnedFd,
 };
 
 pub struct VulkanShmImage {
-    pub(super) _size: DeviceSize,
+    pub(super) size: DeviceSize,
     pub(super) stride: u32,
     pub(super) _allocation: VulkanAllocation,
     pub(super) shm_info: &'static FormatShmInfo,
+    pub(super) async_data: Option<VulkanShmImageAsyncData>,
+}
+
+pub struct VulkanShmImageAsyncData {
+    pub(super) busy: Cell<bool>,
+    pub(super) io_job: Cell<Option<Box<IoUploadJob>>>,
+    pub(super) copy_job: Cell<Option<Box<CopyUploadJob>>>,
+    pub(super) staging: CloneCell<Option<Rc<VulkanStagingBuffer>>>,
+    pub(super) callback: Cell<Option<Rc<dyn AsyncShmGfxTextureCallback>>>,
+    pub(super) callback_id: Cell<u64>,
+    pub(super) regions: RefCell<Vec<BufferImageCopy2<'static>>>,
+    pub(super) cpu: Rc<CpuWorker>,
 }
 
 impl VulkanShmImage {
@@ -133,6 +159,30 @@ impl VulkanShmImage {
                 ptr::copy_nonoverlapping(buf, mem, total_size as usize);
             }
         })?;
+        let Some((cmd, fence, sync_file, point)) =
+            self.submit_buffer_to_image_copy(img, &staging, cpy)?
+        else {
+            return Ok(());
+        };
+        let future = img.renderer.eng.spawn(await_upload(
+            point,
+            img.clone(),
+            cmd,
+            sync_file,
+            fence,
+            staging,
+        ));
+        img.renderer.pending_uploads.set(point, future);
+        Ok(())
+    }
+
+    fn submit_buffer_to_image_copy(
+        &self,
+        img: &Rc<VulkanImage>,
+        staging: &VulkanStagingBuffer,
+        regions: &[BufferImageCopy2],
+    ) -> Result<Option<(Rc<VulkanCommandBuffer>, Rc<VulkanFence>, SyncFile, u64)>, VulkanError>
+    {
         let memory_barrier = |sam, ssm, dam, dsm| {
             BufferMemoryBarrier2::default()
                 .buffer(staging.buffer)
@@ -185,7 +235,7 @@ impl VulkanShmImage {
             .src_buffer(staging.buffer)
             .dst_image(img.image)
             .dst_image_layout(ImageLayout::TRANSFER_DST_OPTIMAL)
-            .regions(cpy);
+            .regions(regions);
         let cmd = img.renderer.allocate_command_buffer()?;
         let dev = &img.renderer.device.device;
         let command_buffer_info = CommandBufferSubmitInfo::default().command_buffer(cmd.buffer);
@@ -210,25 +260,17 @@ impl VulkanShmImage {
             .map_err(VulkanError::Submit)?;
         }
         img.is_undefined.set(false);
+        img.contents_are_undefined.set(false);
         let release_sync_file = match release_fence.export_sync_file() {
             Ok(s) => s,
             Err(e) => {
                 log::error!("Could not export sync file from fence: {}", ErrorFmt(e));
                 img.renderer.block();
-                return Ok(());
+                return Ok(None);
             }
         };
         let point = img.renderer.allocate_point();
-        let future = img.renderer.eng.spawn(await_upload(
-            point,
-            img.clone(),
-            cmd,
-            release_sync_file,
-            release_fence,
-            staging,
-        ));
-        img.renderer.pending_uploads.set(point, future);
-        Ok(())
+        Ok(Some((cmd, release_fence, release_sync_file, point)))
     }
 }
 
@@ -252,6 +294,325 @@ async fn await_upload(
     img.renderer.pending_uploads.remove(&id);
 }
 
+impl VulkanShmImageAsyncData {
+    fn complete(&self, result: Result<(), VulkanError>) {
+        self.busy.set(false);
+        if let Some(cb) = self.callback.take() {
+            cb.completed(result.map_err(|e| e.into()));
+        }
+    }
+}
+
+impl VulkanShmImage {
+    pub fn async_upload(
+        &self,
+        img: &Rc<VulkanImage>,
+        client_mem: &Rc<ClientMemOffset>,
+        damage: Region,
+        callback: Rc<dyn AsyncShmGfxTextureCallback>,
+    ) -> Result<Option<PendingShmUpload>, VulkanError> {
+        let data = self.async_data.as_ref().unwrap();
+        let res = self.try_async_upload(img, data, client_mem, damage);
+        match res {
+            Ok(()) => {
+                let id = img.renderer.allocate_point();
+                data.callback_id.set(id);
+                data.callback.set(Some(callback));
+                Ok(Some(PendingShmUpload::new(img.clone(), id)))
+            }
+            Err(e) => Err(e),
+        }
+    }
+
+    fn try_async_upload(
+        &self,
+        img: &Rc<VulkanImage>,
+        data: &VulkanShmImageAsyncData,
+        client_mem: &Rc<ClientMemOffset>,
+        mut damage: Region,
+    ) -> Result<(), VulkanError> {
+        if data.busy.get() {
+            return Err(VulkanError::AsyncCopyBusy);
+        }
+        if self.size > client_mem.ptr().len() as u64 {
+            return Err(VulkanError::InvalidBufferSize);
+        }
+        data.busy.set(true);
+        if img.contents_are_undefined.get() {
+            damage = Region::new2(Rect::new_sized(0, 0, img.width as _, img.height as _).unwrap())
+        }
+
+        let copies = &mut *data.regions.borrow_mut();
+        copies.clear();
+
+        let mut copy = |x, y, width, height| {
+            let buffer_offset = (y as u32 * img.stride + x as u32 * self.shm_info.bpp) as u64;
+            let copy = BufferImageCopy2::default()
+                .buffer_offset(buffer_offset)
+                .image_offset(Offset3D { x, y, z: 0 })
+                .image_extent(Extent3D {
+                    width,
+                    height,
+                    depth: 1,
+                })
+                .image_subresource(ImageSubresourceLayers {
+                    aspect_mask: ImageAspectFlags::COLOR,
+                    mip_level: 0,
+                    base_array_layer: 0,
+                    layer_count: 1,
+                })
+                .buffer_image_height(img.height)
+                .buffer_row_length(img.stride / self.shm_info.bpp);
+            copies.push(copy);
+        };
+        for damage in damage.rects() {
+            let Some(damage) = Rect::new(
+                damage.x1().max(0),
+                damage.y1().max(0),
+                damage.x2().min(img.width as i32),
+                damage.y2().min(img.height as i32),
+            ) else {
+                continue;
+            };
+            if damage.is_empty() {
+                continue;
+            }
+            copy(
+                damage.x1(),
+                damage.y1(),
+                damage.width() as u32,
+                damage.height() as u32,
+            );
+        }
+
+        if let Some(staging) = data.staging.get() {
+            return self.async_upload_initiate_copy(img, data, &staging, copies, client_mem);
+        }
+
+        let img2 = img.clone();
+        let client_mem = client_mem.clone();
+        img.renderer.device.create_shm_staging(
+            &img.renderer,
+            &data.cpu,
+            self.size,
+            true,
+            false,
+            move |res| {
+                let VulkanImageMemory::Internal(shm) = &img2.ty else {
+                    unreachable!();
+                };
+                if let Err(e) = shm.async_upload_after_allocation(&img2, &client_mem, res) {
+                    shm.async_data.as_ref().unwrap().complete(Err(e));
+                }
+            },
+        )
+    }
+
+    fn async_upload_after_allocation(
+        &self,
+        img: &Rc<VulkanImage>,
+        client_mem: &Rc<ClientMemOffset>,
+        res: Result<VulkanStagingBuffer, VulkanError>,
+    ) -> Result<(), VulkanError> {
+        let staging = Rc::new(res?);
+        let data = self.async_data.as_ref().unwrap();
+        data.staging.set(Some(staging.clone()));
+        let copies = &*data.regions.borrow();
+        self.async_upload_initiate_copy(img, data, &staging, copies, client_mem)
+    }
+
+    fn async_upload_initiate_copy(
+        &self,
+        img: &Rc<VulkanImage>,
+        data: &VulkanShmImageAsyncData,
+        staging: &VulkanStagingBuffer,
+        copies: &[BufferImageCopy2],
+        client_mem: &Rc<ClientMemOffset>,
+    ) -> Result<(), VulkanError> {
+        img.renderer.check_defunct()?;
+
+        let id = img.renderer.allocate_point();
+        let pending;
+        if client_mem.pool().sigbus_impossible() {
+            let mut job = data.copy_job.take().unwrap_or_else(|| {
+                Box::new(CopyUploadJob {
+                    img: None,
+                    id,
+                    _mem: None,
+                    work: unsafe { ImgCopyWork::new() },
+                })
+            });
+            job.id = id;
+            job.img = Some(img.clone());
+            job._mem = Some(client_mem.clone());
+            job.work.src = client_mem.ptr() as _;
+            job.work.dst = staging.allocation.mem.unwrap();
+            job.work.width = img.width as _;
+            job.work.stride = img.stride as _;
+            job.work.bpp = self.shm_info.bpp as _;
+            job.work.rects.clear();
+            for copy in copies {
+                job.work.rects.push(
+                    Rect::new_sized(
+                        copy.image_offset.x as _,
+                        copy.image_offset.y as _,
+                        copy.image_extent.width as _,
+                        copy.image_extent.height as _,
+                    )
+                    .unwrap(),
+                );
+            }
+            pending = data.cpu.submit(job);
+        } else {
+            let mut min_offset = client_mem.ptr().len() as u64;
+            let mut max_offset = 0;
+            for copy in copies {
+                min_offset = min_offset.min(copy.buffer_offset);
+                let len = img.stride * (copy.image_extent.height - 1)
+                    + copy.image_extent.width * self.shm_info.bpp;
+                max_offset = max_offset.max(copy.buffer_offset + len as u64);
+            }
+            let mut job = data.io_job.take().unwrap_or_else(|| {
+                Box::new(IoUploadJob {
+                    img: None,
+                    id,
+                    _mem: None,
+                    work: unsafe { ReadWriteWork::new() },
+                    fd: None,
+                })
+            });
+            job.id = id;
+            job.img = Some(img.clone());
+            job._mem = Some(client_mem.clone());
+            job.fd = Some(client_mem.pool().fd().clone());
+            unsafe {
+                let config = job.work.config();
+                config.fd = client_mem.pool().fd().raw();
+                config.offset = client_mem.offset() + min_offset as usize;
+                config.ptr = staging.allocation.mem.unwrap().add(min_offset as _);
+                config.len = max_offset.saturating_sub(min_offset) as usize;
+                config.write = false;
+            }
+            pending = data.cpu.submit(job);
+        }
+
+        img.renderer.pending_cpu_jobs.set(id, pending);
+
+        Ok(())
+    }
+
+    fn async_upload_copy_buffer_to_image(
+        &self,
+        img: &Rc<VulkanImage>,
+        data: &VulkanShmImageAsyncData,
+        res: Result<(), ReadWriteJobError>,
+    ) -> Result<(), VulkanError> {
+        if let Err(e) = res {
+            return Err(VulkanError::AsyncCopyToStaging(e));
+        }
+        img.renderer.check_defunct()?;
+        let regions = &*data.regions.borrow();
+        let staging = data.staging.get().unwrap();
+        staging.upload(|_, _| ())?;
+        let Some((cmd, fence, sync_file, point)) =
+            self.submit_buffer_to_image_copy(img, &staging, regions)?
+        else {
+            return Ok(());
+        };
+        let future = img.renderer.eng.spawn(await_async_upload(
+            point,
+            img.clone(),
+            cmd,
+            fence,
+            sync_file,
+        ));
+        img.renderer.pending_uploads.set(point, future);
+        Ok(())
+    }
+}
+
+pub(super) struct IoUploadJob {
+    img: Option<Rc<VulkanImage>>,
+    id: u64,
+    _mem: Option<Rc<ClientMemOffset>>,
+    fd: Option<Rc<OwnedFd>>,
+    work: ReadWriteWork,
+}
+
+pub(super) struct CopyUploadJob {
+    img: Option<Rc<VulkanImage>>,
+    id: u64,
+    _mem: Option<Rc<ClientMemOffset>>,
+    work: ImgCopyWork,
+}
+
+impl CpuJob for IoUploadJob {
+    fn work(&mut self) -> &mut dyn CpuWork {
+        &mut self.work
+    }
+
+    fn completed(mut self: Box<Self>) {
+        self._mem = None;
+        self.fd = None;
+        let img = self.img.take().unwrap();
+        let res = self.work.config().result.take().unwrap();
+        complete_async_upload(&img, self.id, res, |data| data.io_job.set(Some(self)));
+    }
+}
+
+impl CpuJob for CopyUploadJob {
+    fn work(&mut self) -> &mut dyn CpuWork {
+        &mut self.work
+    }
+
+    fn completed(mut self: Box<Self>) {
+        self._mem = None;
+        let img = self.img.take().unwrap();
+        complete_async_upload(&img, self.id, Ok(()), |data| data.copy_job.set(Some(self)));
+    }
+}
+
+fn complete_async_upload(
+    img: &Rc<VulkanImage>,
+    id: u64,
+    res: Result<(), ReadWriteJobError>,
+    store: impl FnOnce(&VulkanShmImageAsyncData),
+) {
+    img.renderer.pending_cpu_jobs.remove(&id);
+    let VulkanImageMemory::Internal(shm) = &img.ty else {
+        unreachable!();
+    };
+    let data = shm.async_data.as_ref().unwrap();
+    store(data);
+    if let Err(e) = shm.async_upload_copy_buffer_to_image(img, data, res) {
+        data.complete(Err(e));
+    }
+}
+
+async fn await_async_upload(
+    id: u64,
+    img: Rc<VulkanImage>,
+    buf: Rc<VulkanCommandBuffer>,
+    _fence: Rc<VulkanFence>,
+    sync_file: SyncFile,
+) {
+    let res = img.renderer.ring.readable(&sync_file.0).await;
+    if let Err(e) = res {
+        log::error!(
+            "Could not wait for sync file to become readable: {}",
+            ErrorFmt(e)
+        );
+        img.renderer.block();
+    }
+    img.renderer.command_buffers.push(buf);
+    img.renderer.pending_uploads.remove(&id);
+    let VulkanImageMemory::Internal(shm) = &img.ty else {
+        unreachable!();
+    };
+    let data = shm.async_data.as_ref().unwrap();
+    data.complete(Ok(()));
+}
+
 impl VulkanRenderer {
     pub fn create_shm_texture(
         self: &Rc<Self>,
@@ -261,6 +622,7 @@ impl VulkanRenderer {
         stride: i32,
         data: &[Cell<u8>],
         for_download: bool,
+        cpu_worker: Option<&Rc<CpuWorker>>,
     ) -> Result<Rc<VulkanImage>, VulkanError> {
         let Some(shm_info) = &format.shm_info else {
             return Err(VulkanError::UnsupportedShmFormat(format.name));
@@ -283,7 +645,7 @@ impl VulkanRenderer {
         if width > shm.limits.max_width || height > shm.limits.max_height {
             return Err(VulkanError::ImageTooLarge);
         }
-        let size = stride.checked_mul(height).ok_or(VulkanError::ShmOverflow)?;
+        let size = stride.checked_mul(height).ok_or(VulkanError::ShmOverflow)? as u64;
         let usage = ImageUsageFlags::TRANSFER_SRC
             | match for_download {
                 true => ImageUsageFlags::COLOR_ATTACHMENT,
@@ -335,11 +697,25 @@ impl VulkanRenderer {
                 .create_image_view(&image_view_create_info, None)
         };
         let view = view.map_err(VulkanError::CreateImageView)?;
+        let mut async_data = None;
+        if let Some(cpu) = cpu_worker {
+            async_data = Some(VulkanShmImageAsyncData {
+                busy: Cell::new(false),
+                io_job: Default::default(),
+                copy_job: Default::default(),
+                staging: Default::default(),
+                callback: Default::default(),
+                callback_id: Cell::new(0),
+                regions: Default::default(),
+                cpu: cpu.clone(),
+            });
+        }
         let shm = VulkanShmImage {
-            _size: size as u64,
+            size,
             stride,
             _allocation: allocation,
             shm_info,
+            async_data,
         };
         destroy_image.forget();
         let img = Rc::new(VulkanImage {
@@ -352,6 +728,7 @@ impl VulkanRenderer {
             render_view: None,
             image,
             is_undefined: Cell::new(true),
+            contents_are_undefined: Cell::new(true),
             ty: VulkanImageMemory::Internal(shm),
             render_ops: Default::default(),
             bridge: None,

src/gfx_apis/vulkan/staging.rs

@@ -51,7 +51,6 @@ impl VulkanDevice {
         })
     }
 
-    #[expect(dead_code)]
     pub(super) fn create_shm_staging(
         self: &Rc<Self>,
         renderer: &Rc<VulkanRenderer>,