workspace: move crates under crates

crates/cpu-worker/src/jobs/img_copy.rs

@@ -0,0 +1,60 @@
+use {
+    crate::{AsyncCpuWork, CpuWork},
+    jay_geometry::Rect,
+    std::ptr,
+};
+
+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>> {
+        jay_tracy::zone!("ImgCopyWork");
+        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
+    }
+}

crates/cpu-worker/src/jobs/read_write.rs

@@ -0,0 +1,145 @@
+use {
+    crate::{AsyncCpuWork, CompletedWork, CpuWork, WorkCompletion},
+    jay_async_engine::{AsyncEngine, SpawnedFuture},
+    jay_io_uring::{IoUring, IoUringError, IoUringTaskId},
+    std::{
+        any::Any,
+        ptr,
+        rc::Rc,
+        slice,
+        sync::{
+            Arc,
+            atomic::{AtomicBool, AtomicU64, Ordering::Relaxed},
+        },
+    },
+    thiserror::Error,
+    uapi::{Fd, c},
+};
+
+#[derive(Debug, Error)]
+pub enum ReadWriteJobError {
+    #[error("An io_uring error occurred")]
+    IoUring(#[source] IoUringError),
+    #[error("The job was cancelled")]
+    Cancelled,
+    #[error("Tried to operate outside the bounds of the file descriptor")]
+    OutOfBounds,
+}
+
+pub struct ReadWriteWork {
+    cancel: Arc<CancelState>,
+    config: Option<Box<ReadWriteWorkConfig>>,
+}
+
+unsafe impl Send for ReadWriteWork {}
+
+impl ReadWriteWork {
+    pub unsafe fn new() -> Self {
+        let cancel = Arc::new(CancelState::default());
+        ReadWriteWork {
+            cancel: cancel.clone(),
+            config: Some(Box::new(ReadWriteWorkConfig {
+                fd: -1,
+                offset: 0,
+                ptr: ptr::null_mut(),
+                len: 0,
+                write: false,
+                cancel,
+                result: None,
+            })),
+        }
+    }
+
+    pub fn config(&mut self) -> &mut ReadWriteWorkConfig {
+        self.config.as_mut().unwrap()
+    }
+}
+
+pub struct ReadWriteWorkConfig {
+    pub fd: c::c_int,
+    pub offset: usize,
+    pub ptr: *mut u8,
+    pub len: usize,
+    pub write: bool,
+    pub result: Option<Result<(), ReadWriteJobError>>,
+    cancel: Arc<CancelState>,
+}
+
+#[derive(Default)]
+struct CancelState {
+    cancelled: AtomicBool,
+    cancel_id: AtomicU64,
+}
+
+impl CpuWork for ReadWriteWork {
+    fn run(&mut self) -> Option<Box<dyn AsyncCpuWork>> {
+        self.cancel.cancelled.store(false, Relaxed);
+        self.cancel.cancel_id.store(0, Relaxed);
+        self.config.take().map(|b| b as _)
+    }
+
+    fn cancel_async(&mut self, ring: &Rc<IoUring>) {
+        self.cancel.cancelled.store(true, Relaxed);
+        let id = self.cancel.cancel_id.load(Relaxed);
+        if id != 0 {
+            ring.cancel(IoUringTaskId::from_raw(id));
+        }
+    }
+
+    fn async_work_done(&mut self, work: Box<dyn AsyncCpuWork>) {
+        let work = (work as Box<dyn Any>).downcast().unwrap();
+        self.config = Some(work);
+    }
+}
+
+impl AsyncCpuWork for ReadWriteWorkConfig {
+    fn run(
+        mut self: Box<Self>,
+        eng: &Rc<AsyncEngine>,
+        ring: &Rc<IoUring>,
+        completion: WorkCompletion,
+    ) -> SpawnedFuture<CompletedWork> {
+        let ring = ring.clone();
+        eng.spawn("shm read/write", async move {
+            let res = loop {
+                if self.cancel.cancelled.load(Relaxed) {
+                    break Err(ReadWriteJobError::Cancelled);
+                }
+                if self.len == 0 {
+                    break Ok(());
+                };
+                let res = if self.write {
+                    ring.write_no_cancel(
+                        Fd::new(self.fd),
+                        self.offset,
+                        unsafe { slice::from_raw_parts(self.ptr, self.len) },
+                        None,
+                        |id| self.cancel.cancel_id.store(id.raw(), Relaxed),
+                    )
+                    .await
+                } else {
+                    ring.read_no_cancel(
+                        Fd::new(self.fd),
+                        self.offset,
+                        unsafe { slice::from_raw_parts_mut(self.ptr, self.len) },
+                        |id| self.cancel.cancel_id.store(id.raw(), Relaxed),
+                    )
+                    .await
+                };
+                match res {
+                    Ok(0) => break Err(ReadWriteJobError::OutOfBounds),
+                    Ok(n) => {
+                        self.len -= n;
+                        self.offset += n;
+                        unsafe {
+                            self.ptr = self.ptr.add(n);
+                        }
+                    }
+                    Err(e) => break Err(ReadWriteJobError::IoUring(e)),
+                }
+            };
+            self.result = Some(res);
+            completion.complete(self)
+        })
+    }
+}