tokio-uring: temporarily include tokio-uring into common module

Tempoarily include tokio-uring into dbs-fuse, so we could publish
dbs-fuse. We will switch to upstream crate once our enhancements
have been merged and new version is available.

Signed-off-by: wllenyj <[email protected]>
Signed-off-by: Jiang Liu <[email protected]>

Author: wllenyj

Cargo.toml

@@ -26,7 +26,7 @@ log = "0.4.6"
 mio = { version = "0.8", features = ["os-poll", "os-ext"]}
 nix = "0.24"
 lazy_static = "1.4"
-tokio-uring = { version = "0.3.0", optional = true }
+tokio = { version = "1", optional = true }
 vmm-sys-util = { version = "0.9", optional = true }
 vm-memory = { version = "0.8", features = ["backend-mmap"] }
 virtio-queue = { version = "0.4", optional = true }
@@ -37,16 +37,22 @@ core-foundation-sys = { version = ">=0.8", optional = true }
 
 [target.'cfg(target_os = "linux")'.dependencies]
 caps = { version = "0.3", optional = true }
+tokio-uring = "0.3.0"
+io-uring = { version = "0.5.2", features = ["unstable"] }
+socket2 = { version = "0.4.4", features = ["all"] }
+scoped-tls = "1.0.0"
+slab = "0.4.6"
 
 [dev-dependencies]
 futures = { version = "0.3", features = ["thread-pool"]}
 stderrlog = "0.5"
 vmm-sys-util = "0.9"
 vm-memory = { version = "0.8", features = ["backend-mmap", "backend-bitmap"] }
+tokio-test = "0.4.2"
 
 [features]
 default = ["fusedev"]
-async-io = ["async-trait", "futures", "tokio-uring"]
+async-io = ["async-trait", "futures", "tokio/fs", "tokio/net", "tokio/sync", "tokio/rt"]
 fusedev = ["vmm-sys-util", "caps", "core-foundation-sys"]
 virtiofs = ["virtio-queue", "caps"]
 vhost-user-fs = ["virtiofs", "vhost", "caps"]

README.md

@@ -93,3 +93,10 @@ impl FuseServer {
 This project is licensed under
 - [Apache License](http://www.apache.org/licenses/LICENSE-2.0), Version 2.0
 - [BSD-3-Clause License](https://opensource.org/licenses/BSD-3-Clause)
+
+Source code under [src/tokio-uring] is temporarily copied from [tokio-uring](https://github.com/tokio-rs/tokio-uring)
+with modifications, which is licensed under [MIT](https://github.com/tokio-rs/tokio-uring/blob/master/LICENSE).
+
+We will use `crate.io` directly instead of this temporary copy when the pendding PRs is merged.
+https://github.com/tokio-rs/tokio-uring/pull/87
+https://github.com/tokio-rs/tokio-uring/pull/88

src/common/mod.rs

@@ -6,3 +6,10 @@
 
 pub mod file_buf;
 pub mod file_traits;
+
+// Temporarily include all source code tokio-uring.
+// Will switch to upstream once our enhancement have been merged and new version available.
+#[cfg(all(feature = "async-io", target_os = "linux"))]
+pub mod tokio_uring;
+#[cfg(all(feature = "async-io", target_os = "linux"))]
+pub(crate) use self::tokio_uring::{buf, driver, fs, future, BufResult};

src/common/tokio_uring/buf/io_buf.rs

@@ -0,0 +1,162 @@
+use crate::buf::Slice;
+
+use std::ops;
+
+/// An `io-uring` compatible buffer.
+///
+/// The `IoBuf` trait is implemented by buffer types that can be passed to
+/// io-uring operations. Users will not need to use this trait directly, except
+/// for the [`slice`] method.
+///
+/// # Slicing
+///
+/// Because buffers are passed by ownership to the runtime, Rust's slice API
+/// (`&buf[..]`) cannot be used. Instead, `tokio-uring` provides an owned slice
+/// API: [`slice()`]. The method takes ownership fo the buffer and returns a
+/// `Slice<Self>` type that tracks the requested offset.
+///
+/// # Implementation notes
+///
+/// Buffers passed to `io-uring` operations must reference a stable memory
+/// region. While the runtime holds ownership to a buffer, the pointer returned
+/// by `stable_ptr` must remain valid even if the `IoBuf` value is moved.
+///
+/// [`slice()`]: IoBuf::slice
+pub unsafe trait IoBuf: Unpin + 'static {
+    /// Returns a raw pointer to the vector’s buffer.
+    ///
+    /// This method is to be used by the `tokio-uring` runtime and it is not
+    /// expected for users to call it directly.
+    ///
+    /// The implementation must ensure that, while the `tokio-uring` runtime
+    /// owns the value, the pointer returned by `stable_ptr` **does not**
+    /// change.
+    fn stable_ptr(&self) -> *const u8;
+
+    /// Number of initialized bytes.
+    ///
+    /// This method is to be used by the `tokio-uring` runtime and it is not
+    /// expected for users to call it directly.
+    ///
+    /// For `Vec`, this is identical to `len()`.
+    fn bytes_init(&self) -> usize;
+
+    /// Total size of the buffer, including uninitialized memory, if any.
+    ///
+    /// This method is to be used by the `tokio-uring` runtime and it is not
+    /// expected for users to call it directly.
+    ///
+    /// For `Vec`, this is identical to `capacity()`.
+    fn bytes_total(&self) -> usize;
+
+    /// Returns a view of the buffer with the specified range.
+    ///
+    /// This method is similar to Rust's slicing (`&buf[..]`), but takes
+    /// ownership of the buffer.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use tokio_uring::buf::IoBuf;
+    ///
+    /// let buf = b"hello world".to_vec();
+    /// buf.slice(5..10);
+    /// ```
+    fn slice(self, range: impl ops::RangeBounds<usize>) -> Slice<Self>
+    where
+        Self: Sized,
+    {
+        use core::ops::Bound;
+
+        let begin = match range.start_bound() {
+            Bound::Included(&n) => n,
+            Bound::Excluded(&n) => n + 1,
+            Bound::Unbounded => 0,
+        };
+
+        assert!(begin < self.bytes_total());
+
+        let end = match range.end_bound() {
+            Bound::Included(&n) => n.checked_add(1).expect("out of range"),
+            Bound::Excluded(&n) => n,
+            Bound::Unbounded => self.bytes_total(),
+        };
+
+        assert!(end <= self.bytes_total());
+        assert!(begin <= self.bytes_init());
+
+        Slice::new(self, begin, end)
+    }
+}
+
+unsafe impl IoBuf for Vec<u8> {
+    fn stable_ptr(&self) -> *const u8 {
+        self.as_ptr()
+    }
+
+    fn bytes_init(&self) -> usize {
+        self.len()
+    }
+
+    fn bytes_total(&self) -> usize {
+        self.capacity()
+    }
+}
+
+unsafe impl IoBuf for &'static [u8] {
+    fn stable_ptr(&self) -> *const u8 {
+        self.as_ptr()
+    }
+
+    fn bytes_init(&self) -> usize {
+        <[u8]>::len(self)
+    }
+
+    fn bytes_total(&self) -> usize {
+        self.bytes_init()
+    }
+}
+
+unsafe impl IoBuf for &'static str {
+    fn stable_ptr(&self) -> *const u8 {
+        self.as_ptr()
+    }
+
+    fn bytes_init(&self) -> usize {
+        <str>::len(self)
+    }
+
+    fn bytes_total(&self) -> usize {
+        self.bytes_init()
+    }
+}
+
+#[cfg(feature = "bytes")]
+unsafe impl IoBuf for bytes::Bytes {
+    fn stable_ptr(&self) -> *const u8 {
+        self.as_ptr()
+    }
+
+    fn bytes_init(&self) -> usize {
+        self.len()
+    }
+
+    fn bytes_total(&self) -> usize {
+        self.len()
+    }
+}
+
+#[cfg(feature = "bytes")]
+unsafe impl IoBuf for bytes::BytesMut {
+    fn stable_ptr(&self) -> *const u8 {
+        self.as_ptr()
+    }
+
+    fn bytes_init(&self) -> usize {
+        self.len()
+    }
+
+    fn bytes_total(&self) -> usize {
+        self.capacity()
+    }
+}

src/common/tokio_uring/buf/io_buf_mut.rs

@@ -0,0 +1,59 @@
+use crate::buf::IoBuf;
+
+/// A mutable`io-uring` compatible buffer.
+///
+/// The `IoBufMut` trait is implemented by buffer types that can be passed to
+/// io-uring operations. Users will not need to use this trait directly.
+///
+/// # Implementation notes
+///
+/// Buffers passed to `io-uring` operations must reference a stable memory
+/// region. While the runtime holds ownership to a buffer, the pointer returned
+/// by `stable_mut_ptr` must remain valid even if the `IoBufMut` value is moved.
+pub unsafe trait IoBufMut: IoBuf {
+    /// Returns a raw mutable pointer to the vector’s buffer.
+    ///
+    /// This method is to be used by the `tokio-uring` runtime and it is not
+    /// expected for users to call it directly.
+    ///
+    /// The implementation must ensure that, while the `tokio-uring` runtime
+    /// owns the value, the pointer returned by `stable_mut_ptr` **does not**
+    /// change.
+    fn stable_mut_ptr(&mut self) -> *mut u8;
+
+    /// Updates the number of initialized bytes.
+    ///
+    /// The specified `pos` becomes the new value returned by
+    /// `IoBuf::bytes_init`.
+    ///
+    /// # Safety
+    ///
+    /// The caller must ensure that all bytes starting at `stable_mut_ptr()` up
+    /// to `pos` are initialized and owned by the buffer.
+    unsafe fn set_init(&mut self, pos: usize);
+}
+
+unsafe impl IoBufMut for Vec<u8> {
+    fn stable_mut_ptr(&mut self) -> *mut u8 {
+        self.as_mut_ptr()
+    }
+
+    unsafe fn set_init(&mut self, init_len: usize) {
+        if self.len() < init_len {
+            self.set_len(init_len);
+        }
+    }
+}
+
+#[cfg(feature = "bytes")]
+unsafe impl IoBufMut for bytes::BytesMut {
+    fn stable_mut_ptr(&mut self) -> *mut u8 {
+        self.as_mut_ptr()
+    }
+
+    unsafe fn set_init(&mut self, init_len: usize) {
+        if self.len() < init_len {
+            self.set_len(init_len);
+        }
+    }
+}

src/common/tokio_uring/buf/mod.rs

@@ -0,0 +1,26 @@
+//! Utilities for working with buffers.
+//!
+//! `io-uring` APIs require passing ownership of buffers to the runtime. The
+//! crate defines [`IoBuf`] and [`IoBufMut`] traits which are implemented by buffer
+//! types that respect the `io-uring` contract.
+
+mod io_buf;
+pub use io_buf::IoBuf;
+
+mod io_buf_mut;
+pub use io_buf_mut::IoBufMut;
+
+mod slice;
+pub use slice::Slice;
+
+pub(crate) fn deref(buf: &impl IoBuf) -> &[u8] {
+    // Safety: the `IoBuf` trait is marked as unsafe and is expected to be
+    // implemented correctly.
+    unsafe { std::slice::from_raw_parts(buf.stable_ptr(), buf.bytes_init()) }
+}
+
+pub(crate) fn deref_mut(buf: &mut impl IoBufMut) -> &mut [u8] {
+    // Safety: the `IoBufMut` trait is marked as unsafe and is expected to be
+    // implemented correct.
+    unsafe { std::slice::from_raw_parts_mut(buf.stable_mut_ptr(), buf.bytes_init()) }
+}