Modify API in layer 5-disk in a queue manner

[lucassong-mh]

API in layer 5-disk, means the API of the crate, means the user (FSes) would invoke, need to receive queued block I/O requests.

TODO: what does the API look like

[cqs21]

I wrote a draft according to the reference: asterinas/asterinas#432

/// Memory buffer for Bio read/write.
trait BioBuf {
    /// Returns the number of blocks of the buffer.
    fn nblocks(&self) -> usize;

    /// Returns the immutable slice of the buffer.
    fn as_slice(&self) -> &[u8];

    /// Returns the mutable slice of the buffer.
    fn as_mut_slice(&mut self) -> &mut [u8];
}

/// The types of Bio.
enum BioType {
    /// Read blocks from device.
    Read,

    /// Write blocks to device.
    Write,

    /// Flush the volatile write cache, if supported.
    Flush,

    /// Discard blocks, if supported.
    Discard,

    /// Undefined types.
    Undefined,
}

/// The flags of Bio.
enum BioFlags {
    /// No special flags.
    Normal,

    /// Request is sync (sync write or read).
    Sync,

    /// Don't touch this for merging.
    NoMerge,
}

/// The states of Bio.
enum BioState {
    /// The Bio is initialized.
    Init,

    /// The Bio is submitted to device.
    Submitted,

    /// The Bio ends successfully.
    Done,

    /// The Bio has still not completed.
    InComplete,

    /// The Bio request encounters some error.
    Error,

    /// Requeue the Bio (e.g., remapped to another device).
    Requeue,
}

/// The callback function when a device try to end the Bio.
type BioEndIo = Option<Box<dyn FnOnce(&Bio) -> BioState>>;

/// The main unit of I/O for the block device.
struct Bio {
    device: Weak<dyn RequestQueue>,
    buffer: Vec<Arc<dyn BioBuf>>,
    start_block: BlockId,
    total_blocks: usize,
    complete_blocks: usize,
    type_: BioType,
    flags: BioFlags,
    state: BioState,
    endio: BioEndIo,
}

impl Bio {
    fn new(
        device: Weak<dyn RequestQueue>,
        type_: BioType,
        start_block: BlockId,
        buffer: Vec<Arc<dyn BioBuf>>,
    ) -> Self {
       todo!()
    }

    fn submit(&self) -> Result<()> {
        todo!()
    }

    fn set_dev(&mut self, device: Weak<dyn RequestQueue>) {
        self.device = device;
    }

    fn set_flags(&mut self, flags: BioFlags) {
        self.flags = flags;
    }

    fn set_endio(&mut self, endio: BioEndIo) {
        self.endio = endio;
    }
}

/// The queue operations of a block device.
trait RequestQueue {
    fn submit(&self, bio: Bio) -> Result;
    fn submit_sync(&self, bio: Bio) -> Result;
    fn try_merge(&self) -> Result;
    fn flush(&self) -> Result;
}

/// The request of `Bio`s with same `BioType` and contiguous `BlockId`.
struct Request {
    device: Weak<dyn RequestQueue>,
    type_: BioType,
    bios: Vec<Bio>,
    region: Range<BlockId>,
    total_blocks: usize,
}

/// A block device, supporting the `RequestQueue`.
struct BlockDevice {
    reqs: Vec<Request>,
}

impl RequestQueue for BlockDevice {
    fn submit(&self, bio: Bio) -> Result {
        todo!()
    }
    fn submit_sync(&self, bio: Bio) -> Result {
        todo!()
    }
    fn try_merge(&self) -> Result {
        todo!()
    }
    fn flush(&self) -> Result {
        todo!()
    }
}

[lucassong-mh]

New design and implementation written in bio.rs and sworndisk.rs:

bio.rs

//! Block I/O (BIO).

/// A queue for managing block I/O requests (`BioReq`).
/// It provides a concurrency-safe way to store and manage
/// block I/O requests that need to be processed by a block device.
pub struct BioReqQueue {
    queue: Mutex<VecDeque<BioReq>>,
    num_reqs: AtomicUsize,
}

/// A block I/O request.
pub struct BioReq {
    type_: BioType,
    addr: BlockId,
    nblocks: u32,
    bufs: Mutex<Vec<BlockBuf>>,
    status: Mutex<BioStatus>,
    on_complete: Option<BioReqOnCompleteFn>,
    ext: Mutex<AnyMap>,
}

/// The type of a block request.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum BioType {
    /// A read request.
    Read,
    /// A write request.
    Write,
    /// A sync request.
    Sync,
}

/// A response from a block device.
pub type BioResp = Result<()>;

/// The type of the callback function invoked upon the completion of
/// a block I/O request.
pub type BioReqOnCompleteFn = fn(/* req = */ &BioReq, /* resp = */ &BioResp);

/// The status describing a block I/O request.
#[derive(Clone, Debug)]
enum BioStatus {
    Init,
    Submitted,
    Completed(BioResp),
}

/// A builder for `BioReq`.
pub struct BioReqBuilder {
    type_: BioType,
    addr: Option<BlockId>,
    bufs: Option<Vec<BlockBuf>>,
    on_complete: Option<BioReqOnCompleteFn>,
    ext: Option<AnyMap>,
}

/// A buffer for block requests.
///
/// The size of `BlockBuf` is a multiple of block size.
#[derive(Debug)]
pub struct BlockBuf {
    ptr: NonNull<u8>,
    len: usize,
}

/* implementation omitted */

sworndisk.rs

//! SwornDisk as a block device.
//!
//! API: submit_bio(), submit_bio_sync(), create(), open(),
//! read(), readv(), write(), writev(), sync().
//!
//! Responsible for managing a `TxLsmTree`, whereas the TX logs (WAL and SSTs)
//! are stored; an untrusted disk storing user data, a `BlockAlloc` for managing data blocks'
//! allocation metadata. `TxLsmTree` and `BlockAlloc` are manipulated
//! based on internal transactions.

/// SwornDisk.
pub struct SwornDisk<D: BlockSet> {
    inner: Arc<DiskInner<D>>,
}

/// Inner structures of `SwornDisk`.
struct DiskInner<D: BlockSet> {
    /// Block I/O request queue.
    bio_req_queue: BioReqQueue,
    /// A `TxLsmTree` to store metadata of the logical blocks.
    logical_block_table: TxLsmTree<RecordKey, RecordValue, D>,
    /// The underlying disk where user data is stored.
    user_data_disk: D,
    /// Manage space of the data disk.
    block_validity_table: Arc<AllocBitmap>,
    /// A buffer to cache data blocks.
    data_buf: DataBuf,
    /// Root encryption key.
    root_key: Key,
    /// Whether `SwornDisk` is dropped.
    is_dropped: AtomicBool,
}

impl<D: BlockSet + 'static> SwornDisk<D> {
    /// Submit a new block I/O request to the request queue (Asynchronous).
    pub fn submit_bio(&self, bio_req: BioReq) -> BioResp {
        self.inner.bio_req_queue.enqueue(bio_req)
    }

    /// Submit a new block I/O request and wait its completion (Synchronous).
    pub fn submit_bio_sync(&self, bio_req: BioReq) -> BioResp {
        bio_req.submit();
        self.inner.handle_bio_req(&bio_req)
    }

    /// Creates a new `SwornDisk` on the given disk, with the root encryption key.
    pub fn create(disk: D, root_key: Key) -> Result<Self> {
        let new_self = /* omitted */;

        new_self.spawn_bio_req_handler();

        Ok(new_self)
    }

    /// Opens the `SwornDisk` on the given disk, with the root encryption key.
    pub fn open(disk: D, root_key: Key) -> Result<Self> {
        let opened_self = /* omitted */;

        opened_self.spawn_bio_req_handler();

        Ok(opened_self)
    }

    /// Spawn a task to continuously handle any queued block I/O requests.
    fn spawn_bio_req_handler(&self) {
        let inner = self.inner.clone();
        let _ = std::thread::spawn(move || {
            inner.handle_bio_reqs_looped();
        });
    }

    /// Extended interfaces below

    /// Read a specified number of blocks at a logical block address on the device.
    /// The block contents will be read into a single contiguous buffer.
    pub fn read(&self, lba: Lba, mut buf: BufMut) -> Result<()> {
        self.inner.read(lba, buf)
    }

    /// Read multiple blocks at a logical block address on the device.
    /// The block contents will be read into several scattered buffers.
    pub fn readv<'a>(&self, lba: Lba, bufs: &'a mut [BufMut<'a>]) -> Result<()> {
        self.inner.readv(lba, bufs)
    }

    /// Write a specified number of blocks at a logical block address on the device.
    /// The block contents reside in a single contiguous buffer.
    pub fn write(&self, mut lba: Lba, buf: BufRef) -> Result<usize> {
        self.inner.write(lba, buf)
    }

    /// Write multiple blocks at a logical block address on the device.
    /// The block contents reside in several scattered buffers.
    pub fn writev(&self, mut lba: Lba, bufs: &[BufRef]) -> Result<()> {
        self.inner.writev(lba, bufs)
    }

    /// Sync all cached data in the device to the storage medium for durability.
    pub fn sync(&self) -> Result<()> {
        self.inner.sync()
    }
}

impl<D: BlockSet + 'static> DiskInner<D> {
    /// Continuously loop and handle any pending block I/O requests.
    pub fn handle_bio_reqs_looped(&self) {
        loop {
            if self.bio_req_queue.is_empty() {
                if self.is_dropped.load(Ordering::Relaxed) {
                    break;
                }

                continue;
            }

            let bio_req = self.bio_req_queue.dequeue().unwrap();
            // TODO: Handle error
            self.handle_bio_req(&bio_req).unwrap();
        }
    }

    /// Handle one block I/O request. Mark the request completed when finished,
    /// return any error that occurs.
    pub fn handle_bio_req(&self, req: &BioReq) -> BioResp {
        let res = match req.type_() {
            BioType::Read => self.do_read(&req),
            BioType::Write => self.do_write(&req),
            BioType::Sync => self.do_sync(&req),
        };

        req.complete(res.clone());
        res
    }

    /* omitted */
}