block: Enable request submission in batch when using io_uring

Implement the batch submission function for raw disk, default it is
enabled. After parsing the requests this method is
called for better IO latency and bandwidth.

Signed-off-by: Bo Chen <bchen@crusoe.ai>
Signed-off-by: Muminul Islam <muislam@microsoft.com>

block/src/async_io.rs

@@ -79,6 +79,9 @@ pub enum AsyncIoError {
     /// Failed synchronizing file.
     #[error("Failed synchronizing file")]
     Fsync(#[source] std::io::Error),
+    /// Failed submitting batch requests.
+    #[error("Failed submitting batch requests: {0}")]
+    SubmitBatchRequests(#[source] std::io::Error),
 }
 
 pub type AsyncIoResult<T> = std::result::Result<T, AsyncIoError>;

block/src/raw_async.rs

@@ -12,7 +12,7 @@ use vmm_sys_util::eventfd::EventFd;
 use crate::async_io::{
     AsyncIo, AsyncIoError, AsyncIoResult, BorrowedDiskFd, DiskFile, DiskFileError, DiskFileResult,
 };
-use crate::DiskTopology;
+use crate::{BatchRequest, DiskTopology, RequestType};
 
 pub struct RawFileDisk {
     file: File,
@@ -168,4 +168,77 @@ impl AsyncIo for RawFileAsync {
             .next()
             .map(|entry| (entry.user_data(), entry.result()))
     }
+
+    fn batch_requests_enabled(&self) -> bool {
+        true
+    }
+
+    fn submit_batch_requests(&mut self, batch_request: &[BatchRequest]) -> AsyncIoResult<()> {
+        if !self.batch_requests_enabled() {
+            return Ok(());
+        }
+
+        let (submitter, mut sq, _) = self.io_uring.split();
+        let mut submitted = false;
+
+        for req in batch_request {
+            match req.request_type {
+                RequestType::In => {
+                    // SAFETY: we know the file descriptor is valid and we
+                    // relied on vm-memory to provide the buffer address.
+                    unsafe {
+                        sq.push(
+                            &opcode::Readv::new(
+                                types::Fd(self.fd),
+                                req.iovecs.as_ptr(),
+                                req.iovecs.len() as u32,
+                            )
+                            .offset(req.offset as u64)
+                            .build()
+                            .user_data(req.user_data),
+                        )
+                        .map_err(|_| {
+                            AsyncIoError::ReadVectored(Error::other("Submission queue is full"))
+                        })?
+                    };
+                    submitted = true;
+                }
+                RequestType::Out => {
+                    // SAFETY: we know the file descriptor is valid and we
+                    // relied on vm-memory to provide the buffer address.
+                    unsafe {
+                        sq.push(
+                            &opcode::Writev::new(
+                                types::Fd(self.fd),
+                                req.iovecs.as_ptr(),
+                                req.iovecs.len() as u32,
+                            )
+                            .offset(req.offset as u64)
+                            .build()
+                            .user_data(req.user_data),
+                        )
+                        .map_err(|_| {
+                            AsyncIoError::WriteVectored(Error::other("Submission queue is full"))
+                        })?
+                    };
+                    submitted = true;
+                }
+                _ => {
+                    unreachable!("Unexpected batch request type: {:?}", req.request_type)
+                }
+            }
+        }
+
+        // Only submit if we actually queued something
+        if submitted {
+            // Update the submission queue and submit new operations to the
+            // io_uring instance.
+            sq.sync();
+            submitter
+                .submit()
+                .map_err(AsyncIoError::SubmitBatchRequests)?;
+        }
+
+        Ok(())
+    }
 }