camelliakv/storage/journal/

nvm_buffer.rs

use std::cmp;
use std::io::{self, Read, Seek, SeekFrom, Write};
use std::ptr;

use crate::block::{AlignedBytes, BlockSize};
use crate::nvm::NonVolatileMemory;
use crate::{ErrorKind, Result};

/// Buffer for journal region.
///
/// While filling the alignment constraint of the internal implementation for `NonVolatileMemory` 
/// (i.e., aligned with the storage block boundary) and purpose of improving the journal region.
#[derive(Debug)]
pub struct JournalNvmBuffer<N: NonVolatileMemory> {
    // An internal NVM instance used to actually persist data in a journal region.
    inner: N,

    // Current read/write position.
    position: u64,

    // Write buffer
    //
    // Write request issued from the journal region 
    // Keep in this buffer on the memory until one of the following conditions are met,
    // Not reflected in internal NVM:
    // - The `sync` method was called:
    //   - Journal region periodically call this method
    // - When a read request is issued for a region overlapping the coverage range of write buffer:
    //   - Flush the contents of write buffer, processing read command after synchronizes the internal NVM
    // - When a write request is issued for a region that does not overlap the write range of write buffer:
    //   - Data structures in current write buffer cannot have gaps (i. e., Multiple not continuous subregions)
    //   - Therefore, once after the contents of old buffer are refreshed, allocate a buffer for processing write request. 
    //
    // The write request issued from journal region,
    // It also aligns to internal NVM block boundary.
    write_buf: AlignedBytes,

    // T field for holding the position of `write_buf` on internal NVM.
    //
    // It is similar to the `position` field, pointing to the position on internal NVM 
    // `position` is the value of the read and seek operations.
    // `write_buf_offset` continues to use fixed value until the contents of the write buffer are flushed.
    write_buf_offset: u64,

    // Used to determine whether or not data is stored in the write buffer.
    //
    // If the data is written in the write buffer, it is set to `true`
    // After the data in the buffer is flushed to the internal NVM, it is set to `false`
    maybe_dirty: bool,

    // Read buffer
    //
    // The read request issued by the journal region 
    // that used to align the block boundary of the internal NVM.
    read_buf: AlignedBytes,
}
impl<N: NonVolatileMemory> JournalNvmBuffer<N> {
    /// Create a new `JournalNvmBuffer` instance.
    ///
    /// It actually use ` NVM ` for reading and writing.
    ///
    /// It is not necessary for user to care about `nvm` that it is aligned with 
    /// the block boundary required by `nvm`, because `JournalNvmBuffer` ensures that.
    ///
    /// Therefore, it should be noted that overwritten data from the seek point to 
    /// the next block boundary is not included in search.
    pub fn new(nvm: N) -> Self {
        let block_size = nvm.block_size();
        JournalNvmBuffer {
            inner: nvm,
            position: 0,
            maybe_dirty: false,
            write_buf_offset: 0,
            write_buf: AlignedBytes::new(0, block_size),
            read_buf: AlignedBytes::new(0, block_size),
        }
    }

    #[cfg(test)]
    pub fn nvm(&self) -> &N {
        &self.inner
    }

    fn is_dirty_area(&self, offset: u64, length: usize) -> bool {
        if !self.maybe_dirty || length == 0 || self.write_buf.is_empty() {
            return false;
        }
        if self.write_buf_offset < offset {
            let buf_end = self.write_buf_offset + self.write_buf.len() as u64;
            offset < buf_end
        } else {
            let end = offset + length as u64;
            self.write_buf_offset < end
        }
    }

    fn flush_write_buf(&mut self) -> Result<()> {
        if self.write_buf.is_empty() || !self.maybe_dirty {
            return Ok(());
        }

        track_io!(self.inner.seek(SeekFrom::Start(self.write_buf_offset)))?;
        track_io!(self.inner.write(&self.write_buf))?;
        if self.write_buf.len() > self.block_size().as_u16() as usize {
            // This if case leaves information about trailing alignment bytes (= new_len) in the buffer.
            // write_buf_offset is perform by write_buf.len() - new_len(= drop_len).
            //
            // write_buf_offset  can clear write_buf according to the write_buf.len() written successfully. 
            // Because it only can be written by block length, 
            // so NVM must be read once in the next write to obtain the entire block.
            // To avoid this read, it takes the current implementation.
            let new_len = self.block_size().as_u16() as usize;
            let drop_len = self.write_buf.len() - new_len;
            unsafe {
                // This nonoverlappingness is guranteed by the callers.
                ptr::copy(
                    self.write_buf.as_ptr().add(drop_len), // src
                    self.write_buf.as_mut_ptr(),           // dst
                    new_len,
                );
            }
            self.write_buf.truncate(new_len);

            self.write_buf_offset += drop_len as u64;
        }
        self.maybe_dirty = false;
        Ok(())
    }

    fn check_overflow(&self, write_len: usize) -> Result<()> {
        let next_position = self.position() + write_len as u64;
        track_assert!(
            next_position <= self.capacity(),
            ErrorKind::InconsistentState,
            "self.position={}, write_len={}, self.len={}",
            self.position(),
            write_len,
            self.capacity()
        );
        Ok(())
    }
}
impl<N: NonVolatileMemory> NonVolatileMemory for JournalNvmBuffer<N> {
    fn sync(&mut self) -> Result<()> {
        track!(self.flush_write_buf())?;
        self.inner.sync()
    }

    fn position(&self) -> u64 {
        self.position
    }

    fn capacity(&self) -> u64 {
        self.inner.capacity()
    }

    fn block_size(&self) -> BlockSize {
        self.inner.block_size()
    }

    fn split(self, _: u64) -> Result<(Self, Self)> {
        unreachable!()
    }
}
impl<N: NonVolatileMemory> Drop for JournalNvmBuffer<N> {
    fn drop(&mut self) {
        let _ = self.sync();
    }
}
impl<N: NonVolatileMemory> Seek for JournalNvmBuffer<N> {
    fn seek(&mut self, pos: SeekFrom) -> io::Result<u64> {
        let offset = track!(self.convert_to_offset(pos))?;
        self.position = offset;
        Ok(offset)
    }
}
impl<N: NonVolatileMemory> Read for JournalNvmBuffer<N> {
    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        if self.is_dirty_area(self.position, buf.len()) {
            track!(self.flush_write_buf())?;
        }

        let aligned_start = self.block_size().floor_align(self.position);
        let aligned_end = self
            .block_size()
            .ceil_align(self.position + buf.len() as u64);

        self.read_buf
            .aligned_resize((aligned_end - aligned_start) as usize);
        self.inner.seek(SeekFrom::Start(aligned_start))?;
        let inner_read_size = self.inner.read(&mut self.read_buf)?;

        let start = (self.position - aligned_start) as usize;
        let end = cmp::min(inner_read_size, start + buf.len());
        let read_size = end - start;
        (&mut buf[..read_size]).copy_from_slice(&self.read_buf[start..end]);
        self.position += read_size as u64;
        Ok(read_size)
    }
}
impl<N: NonVolatileMemory> Write for JournalNvmBuffer<N> {
    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
        track!(self.check_overflow(buf.len()))?;

        let write_buf_start = self.write_buf_offset;
        let write_buf_end = write_buf_start + self.write_buf.len() as u64;
        if write_buf_start <= self.position && self.position <= write_buf_end {
            // The buffer is duplicated and can be insert from the middle of the buffer.
            // (i.e., Unnecessary flash buffer)
            let start = (self.position - self.write_buf_offset) as usize;
            let end = start + buf.len();
            self.write_buf.aligned_resize(end);
            (&mut self.write_buf[start..end]).copy_from_slice(buf);
            self.position += buf.len() as u64;
            self.maybe_dirty = true;
            Ok(buf.len())
        } else {
            // Because the buffer is not duplicated, the contents of the buffer are temporarily written back once.
            track!(self.flush_write_buf())?;

            if self.block_size().is_aligned(self.position) {
                self.write_buf_offset = self.position;
                self.write_buf.aligned_resize(0);
            } else {
                // Read once so that existing data preceding the seek position will not be discarded.
                let size = self.block_size().as_u16();
                self.write_buf_offset = self.block_size().floor_align(self.position);
                self.write_buf.aligned_resize(size as usize);
                self.inner.seek(SeekFrom::Start(self.write_buf_offset))?;
                self.inner.read_exact(&mut self.write_buf)?;
            }
            self.write(buf)
        }
    }

    fn flush(&mut self) -> io::Result<()> {
        track!(self.flush_write_buf())?;
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use std::io::{Read, Seek, SeekFrom, Write};
    use trackable::result::TestResult;

    use super::*;
    use crate::nvm::MemoryNvm;

    #[test]
    fn write_write_flush() -> TestResult {
        // Continuous region writes remain on buffer until `flush`.
        let mut buffer = new_buffer();
        track_io!(buffer.write_all(b"foo"))?;
        assert_eq!(&buffer.nvm().as_bytes()[0..3], &[0; 3][..]);

        track_io!(buffer.write_all(b"bar"))?;
        assert_eq!(&buffer.nvm().as_bytes()[0..3], &[0; 3][..]);
        assert_eq!(&buffer.nvm().as_bytes()[3..6], &[0; 3][..]);

        track_io!(buffer.flush())?;
        assert_eq!(&buffer.nvm().as_bytes()[0..6], b"foobar");
        Ok(())
    }

    #[test]
    fn write_seek_write_flush() -> TestResult {
        // The "continuous" determination is performed in blocks
        // (Even if the seek does not span blocks, it will not be determined as "discontinuous")
        let mut buffer = new_buffer();
        track_io!(buffer.write_all(b"foo"))?;
        assert_eq!(&buffer.nvm().as_bytes()[0..3], &[0; 3][..]);

        track_io!(buffer.seek(SeekFrom::Current(1)))?;
        track_io!(buffer.write_all(b"bar"))?;
        assert_eq!(&buffer.nvm().as_bytes()[0..3], &[0; 3][..]);
        assert_eq!(&buffer.nvm().as_bytes()[4..7], &[0; 3][..]);

        track_io!(buffer.flush())?;
        assert_eq!(&buffer.nvm().as_bytes()[0..3], b"foo");
        assert_eq!(&buffer.nvm().as_bytes()[4..7], b"bar");

        // if the target is far away, the storage is also in a continuous block.
        let mut buffer = new_buffer();
        track_io!(buffer.write_all(b"foo"))?;
        assert_eq!(&buffer.nvm().as_bytes()[0..3], &[0; 3][..]);

        track_io!(buffer.seek(SeekFrom::Start(512)))?;
        track_io!(buffer.write_all(b"bar"))?;
        assert_eq!(&buffer.nvm().as_bytes()[0..3], &[0; 3][..]);
        assert_eq!(&buffer.nvm().as_bytes()[512..515], &[0; 3][..]);

        track_io!(buffer.flush())?;
        assert_eq!(&buffer.nvm().as_bytes()[0..3], b"foo");
        assert_eq!(&buffer.nvm().as_bytes()[512..515], b"bar");

        // If the write area is overlapped.
        let mut buffer = new_buffer();
        track_io!(buffer.write_all(b"foo"))?;
        assert_eq!(&buffer.nvm().as_bytes()[0..3], &[0; 3][..]);

        track_io!(buffer.seek(SeekFrom::Current(-1)))?;
        track_io!(buffer.write_all(b"bar"))?;
        assert_eq!(&buffer.nvm().as_bytes()[0..3], &[0; 3][..]);
        assert_eq!(&buffer.nvm().as_bytes()[2..5], &[0; 3][..]);

        track_io!(buffer.flush())?;
        assert_eq!(&buffer.nvm().as_bytes()[0..5], b"fobar");
        Ok(())
    }

    #[test]
    fn write_seek_write() -> TestResult {
        // If the write target (in blocks) is no longer in block, the contents of current buffer are written back to NVM.
        let mut buffer = new_buffer();
        track_io!(buffer.write_all(b"foo"))?;
        assert_eq!(&buffer.nvm().as_bytes()[0..3], &[0; 3][..]);

        track_io!(buffer.seek(SeekFrom::Start(513)))?;
        track_io!(buffer.write_all(b"bar"))?;
        assert_eq!(&buffer.nvm().as_bytes()[0..3], b"foo");
        assert_eq!(&buffer.nvm().as_bytes()[513..516], &[0; 3][..]);
        Ok(())
    }

    #[test]
    fn write_seek_read() -> TestResult {
        // If the target is overlapped with the write buffer, the contents of buffer are written back to NVM.
        let mut buffer = new_buffer();
        track_io!(buffer.write_all(b"foo"))?;
        assert_eq!(&buffer.nvm().as_bytes()[0..3], &[0; 3][..]);

        track_io!(buffer.read_exact(&mut [0; 1][..]))?;
        assert_eq!(&buffer.nvm().as_bytes()[0..3], b"foo");

        // If the target is not overlapped with the write buffer, it will not be written back.
        let mut buffer = new_buffer();
        track_io!(buffer.write_all(b"foo"))?;
        assert_eq!(&buffer.nvm().as_bytes()[0..3], &[0; 3][..]);

        track_io!(buffer.seek(SeekFrom::Start(512)))?;
        track_io!(buffer.read_exact(&mut [0; 1][..]))?;
        assert_eq!(&buffer.nvm().as_bytes()[0..3], &[0; 3][..]);
        Ok(())
    }

    #[test]
    fn overwritten() -> TestResult {
        // Keep the data ahead of seek points.
        // (What happens to the data that is not defined to the next block boundary)
        let mut buffer = new_buffer();
        track_io!(buffer.write_all(&[b'a'; 512]))?;
        track_io!(buffer.flush())?;
        assert_eq!(&buffer.nvm().as_bytes()[0..512], &[b'a'; 512][..]);

        track_io!(buffer.seek(SeekFrom::Start(256)))?;
        track_io!(buffer.write_all(&[b'b'; 1]))?;
        track_io!(buffer.flush())?;
        assert_eq!(&buffer.nvm().as_bytes()[0..256], &[b'a'; 256][..]);
        assert_eq!(buffer.nvm().as_bytes()[256], b'b');
        Ok(())
    }

    fn new_buffer() -> JournalNvmBuffer<MemoryNvm> {
        let nvm = MemoryNvm::new(vec![0; 10 * 1024]);
        JournalNvmBuffer::new(nvm)
    }
}