concurrent iou

lib.rs

use std::io;
use std::io::IoSlice;
use std::io::IoSliceMut;
use std::ops::Deref;
use std::ptr::drop_in_place;
use std::sync::Mutex;

use iou::{CompletionQueue, CompletionQueueEvent, IoUring, Registrar, SubmissionQueue};

struct Ring<'a> {
    ring: Option<Mutex<*mut IoUring>>,
    sq: Option<Mutex<SubmissionQueue<'a>>>,
    cq: Option<Mutex<CompletionQueue<'a>>>,
    reg: Option<Mutex<Registrar<'a>>>,
}

impl<'a> Ring<'a> {
    fn new(entries: u32) -> io::Result<Self> {
        let ring = Box::new(IoUring::new(entries)?);

        let ring: *mut IoUring = Box::leak(ring);

        let (sq, cq, reg) = unsafe { (&mut *ring).queues() };

        Ok(Self {
            ring: Some(Mutex::new(ring)),
            sq: Some(Mutex::new(sq)),
            cq: Some(Mutex::new(cq)),
            reg: Some(Mutex::new(reg)),
        })
    }

    fn set_write_submit_event(&self, fd: i32, bufs: &[IoSlice]) -> Option<()> {
        let mut guard = self.sq.as_ref().unwrap().lock().unwrap();

        let mut submission_queue_event = guard.next_sqe()?;

        unsafe { submission_queue_event.prep_write_vectored(fd, bufs, 0); }

        Some(())
    }

    fn set_read_submit_event(&self, fd: i32, bufs: &mut [IoSliceMut]) -> Option<()> {
        let mut guard = self.sq.as_ref().unwrap().lock().unwrap();

        let mut submission_queue_event = guard.next_sqe()?;

        unsafe { submission_queue_event.prep_read_vectored(fd, bufs, 0); }

        Some(())
    }

    fn submit_and_wait(&self) -> io::Result<CompletionQueueEvent<'_>> {
        let guard = self.ring.as_ref().unwrap().lock().unwrap();

        let ring = *guard.deref();

        let ring = unsafe { &mut *ring };

        ring.submit_sqes()?;

        ring.wait_for_cqe()
    }
}

unsafe impl Send for Ring<'_> {}

unsafe impl Sync for Ring<'_> {}

impl Drop for Ring<'_> {
    fn drop(&mut self) {
        drop(self.sq.take().unwrap());
        drop(self.cq.take().unwrap());
        drop(self.reg.take().unwrap());

        let ring = self.ring.take().unwrap().into_inner().unwrap();

        unsafe { drop_in_place(ring); }
    }
}