read entries using AIO

src/engine.rs

@@ -308,11 +308,45 @@ where
                 vec.push(read_entry_from_file::<M, _>(self.pipe_log.as_ref(), i)?);
             }
             ENGINE_READ_ENTRY_COUNT_HISTOGRAM.observe(ents_idx.len() as f64);
+
             return Ok(ents_idx.len());
         }
         Ok(0)
     }
 
+    pub fn fetch_entries_to_aio<M: Message + MessageExt<Entry = M>>(
+        &self,
+        region_id: u64,
+        begin: u64,
+        end: u64,
+        max_size: Option<usize>,
+        vec: &mut Vec<M::Entry>,
+    ) -> Result<usize> {
+        let _t = StopWatch::new(&*ENGINE_READ_ENTRY_DURATION_HISTOGRAM);
+        if let Some(memtable) = self.memtables.get(region_id) {
+            let length = (end - begin) as usize;
+            let mut ents_idx: Vec<EntryIndex> = Vec::with_capacity(length);
+            memtable
+                .read()
+                .fetch_entries_to(begin, end, max_size, &mut ents_idx)?;
+
+            let mut blocks: Vec<FileBlockHandle> = Vec::new();
+            for (t, i) in ents_idx.iter().enumerate() {
+                if t == 0 || (i.entries.unwrap() != ents_idx[t - 1].entries.unwrap()) {
+                    blocks.push(i.entries.unwrap());
+                }
+            }
+            let bytes = self.pipe_log.async_read_bytes(blocks).unwrap();
+            parse_entries_from_bytes::<M>(bytes, &mut ents_idx, vec);
+
+            ENGINE_READ_ENTRY_COUNT_HISTOGRAM.observe(ents_idx.len() as f64);
+
+            return Ok(ents_idx.len());
+        }
+
+        Ok(0)
+    }
+
     pub fn first_index(&self, region_id: u64) -> Option<u64> {
         if let Some(memtable) = self.memtables.get(region_id) {
             return memtable.read().first_index();
@@ -544,7 +578,36 @@ impl BlockCache {
 thread_local! {
     static BLOCK_CACHE: BlockCache = BlockCache::new();
 }
-
+pub(crate) fn parse_entries_from_bytes<M: MessageExt>(
+    bytes: Vec<Vec<u8>>,
+    ents_idx: &mut [EntryIndex],
+    vec: &mut Vec<M::Entry>,
+) {
+    let mut decode_buf = vec![];
+    let mut seq: i32 = -1;
+    for (t, idx) in ents_idx.iter().enumerate() {
+        decode_buf =
+            match t == 0 || ents_idx[t - 1].entries.unwrap() != ents_idx[t].entries.unwrap() {
+                true => {
+                    seq += 1;
+                    bytes[seq as usize].to_vec()
+                }
+                false => decode_buf,
+            };
+        vec.push(
+            parse_from_bytes(
+                &LogBatch::decode_entries_block(
+                    &decode_buf,
+                    idx.entries.unwrap(),
+                    idx.compression_type,
+                )
+                .unwrap()
+                    [idx.entry_offset as usize..(idx.entry_offset + idx.entry_len) as usize],
+            )
+            .unwrap(),
+        );
+    }
+}
 pub(crate) fn read_entry_from_file<M, P>(pipe_log: &P, idx: &EntryIndex) -> Result<M::Entry>
 where
     M: MessageExt,
@@ -683,6 +746,41 @@ mod tests {
                 reader(e.index, entry_index.entries.unwrap().id.queue, &e.data);
             }
         }
+        fn scan_entries_aio<FR: Fn(u64, LogQueue, &[u8])>(
+            &self,
+            rid: u64,
+            start: u64,
+            end: u64,
+            reader: FR,
+        ) {
+            let mut entries = Vec::new();
+            self.fetch_entries_to_aio::<Entry>(
+                rid,
+                self.first_index(rid).unwrap(),
+                self.last_index(rid).unwrap() + 1,
+                None,
+                &mut entries,
+            )
+            .unwrap();
+            assert_eq!(entries.len(), (end - start) as usize);
+            assert_eq!(entries.first().unwrap().index, start);
+            assert_eq!(
+                entries.last().unwrap().index,
+                self.decode_last_index(rid).unwrap()
+            );
+            assert_eq!(entries.last().unwrap().index + 1, end);
+            for e in entries.iter() {
+                let entry_index = self
+                    .memtables
+                    .get(rid)
+                    .unwrap()
+                    .read()
+                    .get_entry(e.index)
+                    .unwrap();
+                assert_eq!(&self.get_entry::<Entry>(rid, e.index).unwrap().unwrap(), e);
+                reader(e.index, entry_index.entries.unwrap().id.queue, &e.data);
+            }
+        }
 
         fn file_count(&self, queue: Option<LogQueue>) -> usize {
             if let Some(queue) = queue {
@@ -745,6 +843,14 @@ mod tests {
                     assert_eq!(d, &data);
                 });
             }
+            for i in 10..20 {
+                let rid = i;
+                let index = i;
+                engine.scan_entries_aio(rid, index, index + 2, |_, q, d| {
+                    assert_eq!(q, LogQueue::Append);
+                    assert_eq!(d, &data);
+                });
+            }
 
             // Recover the engine.
             let engine = engine.reopen();
@@ -1994,6 +2100,20 @@ mod tests {
         type Handle = <ObfuscatedFileSystem as FileSystem>::Handle;
         type Reader = <ObfuscatedFileSystem as FileSystem>::Reader;
         type Writer = <ObfuscatedFileSystem as FileSystem>::Writer;
+        type AsyncIoContext = <ObfuscatedFileSystem as FileSystem>::AsyncIoContext;
+
+        fn async_read(
+            &self,
+            ctx: &mut Self::AsyncIoContext,
+            handle: Arc<Self::Handle>,
+            block: &FileBlockHandle,
+        ) -> std::io::Result<()> {
+            self.inner.async_read(ctx, handle, block)
+        }
+
+        fn async_finish(&self, ctx: &mut Self::AsyncIoContext) -> std::io::Result<Vec<Vec<u8>>> {
+            self.inner.async_finish(ctx)
+        }
 
         fn create<P: AsRef<Path>>(&self, path: P) -> std::io::Result<Self::Handle> {
             let handle = self.inner.create(&path)?;
@@ -2056,6 +2176,10 @@ mod tests {
         fn new_writer(&self, h: Arc<Self::Handle>) -> std::io::Result<Self::Writer> {
             self.inner.new_writer(h)
         }
+
+        fn new_async_io_context(&self) -> std::io::Result<Self::AsyncIoContext> {
+            self.inner.new_async_io_context()
+        }
     }
 
     #[test]

src/env/default.rs

@@ -3,18 +3,23 @@
 use std::io::{Read, Result as IoResult, Seek, SeekFrom, Write};
 use std::os::unix::io::RawFd;
 use std::path::Path;
+use std::pin::Pin;
+use std::slice;
 use std::sync::Arc;
 
 use fail::fail_point;
 use log::error;
 use nix::errno::Errno;
 use nix::fcntl::{self, OFlag};
+use nix::sys::aio::{aio_suspend, Aio, AioRead};
+use nix::sys::signal::SigevNotify;
 use nix::sys::stat::Mode;
 use nix::sys::uio::{pread, pwrite};
 use nix::unistd::{close, ftruncate, lseek, Whence};
 use nix::NixPath;
 
 use crate::env::{FileSystem, Handle, WriteExt};
+use crate::pipe_log::FileBlockHandle;
 
 fn from_nix_error(e: nix::Error, custom: &'static str) -> std::io::Error {
     let kind = std::io::Error::from(e).kind();
@@ -257,12 +262,66 @@ impl WriteExt for LogFile {
     }
 }
 
+pub struct AioContext {
+    aio_vec: Vec<Pin<Box<AioRead<'static>>>>,
+    buf_vec: Vec<Vec<u8>>,
+}
+impl Default for AioContext {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+impl AioContext {
+    pub fn new() -> Self {
+        Self {
+            aio_vec: Vec::new(),
+            buf_vec: Vec::new(),
+        }
+    }
+}
+
 pub struct DefaultFileSystem;
 
 impl FileSystem for DefaultFileSystem {
     type Handle = LogFd;
     type Reader = LogFile;
     type Writer = LogFile;
+    type AsyncIoContext = AioContext;
+
+    fn async_read(
+        &self,
+        ctx: &mut Self::AsyncIoContext,
+        handle: Arc<Self::Handle>,
+        block: &FileBlockHandle,
+    ) -> IoResult<()> {
+        let buf = vec![0_u8; block.len];
+        ctx.buf_vec.push(buf);
+
+        let mut aior = Box::pin(AioRead::new(
+            handle.0,
+            block.offset as i64,
+            unsafe {
+                slice::from_raw_parts_mut(ctx.buf_vec.last_mut().unwrap().as_mut_ptr(), block.len)
+            },
+            0,
+            SigevNotify::SigevNone,
+        ));
+        aior.as_mut()
+            .submit()
+            .expect("aio read request submit failed");
+        ctx.aio_vec.push(aior);
+
+        Ok(())
+    }
+    fn async_finish(&self, ctx: &mut Self::AsyncIoContext) -> IoResult<Vec<Vec<u8>>> {
+        for seq in 0..ctx.aio_vec.len() {
+            let buf_len = ctx.buf_vec[seq].len();
+            aio_suspend(&[&*ctx.aio_vec[seq]], None).expect("aio_suspend failed");
+            assert_eq!(ctx.aio_vec[seq].as_mut().aio_return().unwrap(), buf_len);
+        }
+        let res = ctx.buf_vec.to_owned();
+        Ok(res)
+    }
 
     fn create<P: AsRef<Path>>(&self, path: P) -> IoResult<Self::Handle> {
         LogFd::create(path.as_ref())
@@ -288,4 +347,8 @@ impl FileSystem for DefaultFileSystem {
     fn new_writer(&self, handle: Arc<Self::Handle>) -> IoResult<Self::Writer> {
         Ok(LogFile::new(handle))
     }
+
+    fn new_async_io_context(&self) -> IoResult<Self::AsyncIoContext> {
+        Ok(AioContext::new())
+    }
 }

src/env/mod.rs

@@ -7,14 +7,25 @@ use std::sync::Arc;
 mod default;
 mod obfuscated;
 
+pub use default::AioContext;
 pub use default::DefaultFileSystem;
 pub use obfuscated::ObfuscatedFileSystem;
 
+use crate::pipe_log::FileBlockHandle;
 /// FileSystem
 pub trait FileSystem: Send + Sync {
     type Handle: Send + Sync + Handle;
     type Reader: Seek + Read + Send;
     type Writer: Seek + Write + Send + WriteExt;
+    type AsyncIoContext;
+
+    fn async_read(
+        &self,
+        ctx: &mut Self::AsyncIoContext,
+        handle: Arc<Self::Handle>,
+        block: &FileBlockHandle,
+    ) -> Result<()>;
+    fn async_finish(&self, ctx: &mut Self::AsyncIoContext) -> Result<Vec<Vec<u8>>>;
 
     fn create<P: AsRef<Path>>(&self, path: P) -> Result<Self::Handle>;
 
@@ -55,6 +66,8 @@ pub trait FileSystem: Send + Sync {
     fn new_reader(&self, handle: Arc<Self::Handle>) -> Result<Self::Reader>;
 
     fn new_writer(&self, handle: Arc<Self::Handle>) -> Result<Self::Writer>;
+
+    fn new_async_io_context(&self) -> Result<Self::AsyncIoContext>;
 }
 
 pub trait Handle {

src/env/obfuscated.rs

@@ -7,6 +7,7 @@ use std::sync::Arc;
 
 use crate::env::{DefaultFileSystem, FileSystem, WriteExt};
 
+use crate::pipe_log::FileBlockHandle;
 pub struct ObfuscatedReader(<DefaultFileSystem as FileSystem>::Reader);
 
 impl Read for ObfuscatedReader {
@@ -89,6 +90,30 @@ impl FileSystem for ObfuscatedFileSystem {
     type Handle = <DefaultFileSystem as FileSystem>::Handle;
     type Reader = ObfuscatedReader;
     type Writer = ObfuscatedWriter;
+    type AsyncIoContext = <DefaultFileSystem as FileSystem>::AsyncIoContext;
+
+    fn async_read(
+        &self,
+        ctx: &mut Self::AsyncIoContext,
+        handle: Arc<Self::Handle>,
+        block: &FileBlockHandle,
+    ) -> IoResult<()> {
+        self.inner.async_read(ctx, handle, block)
+    }
+
+    fn async_finish(&self, ctx: &mut Self::AsyncIoContext) -> IoResult<Vec<Vec<u8>>> {
+        let base = self.inner.async_finish(ctx).unwrap();
+        let mut res = vec![];
+        for v in base {
+            let mut temp = vec![];
+            //do obfuscation.
+            for c in v {
+                temp.push(c.wrapping_sub(1));
+            }
+            res.push(temp);
+        }
+        Ok(res)
+    }
 
     fn create<P: AsRef<Path>>(&self, path: P) -> IoResult<Self::Handle> {
         let r = self.inner.create(path);
@@ -127,4 +152,8 @@ impl FileSystem for ObfuscatedFileSystem {
     fn new_writer(&self, handle: Arc<Self::Handle>) -> IoResult<Self::Writer> {
         Ok(ObfuscatedWriter(self.inner.new_writer(handle)?))
     }
+
+    fn new_async_io_context(&self) -> IoResult<Self::AsyncIoContext> {
+        self.inner.new_async_io_context()
+    }
 }

src/file_pipe_log/log_file.rs

@@ -172,7 +172,7 @@ impl<F: FileSystem> LogFileReader<F> {
     }
 
     pub fn read(&mut self, handle: FileBlockHandle) -> Result<Vec<u8>> {
-        let mut buf = vec![0; handle.len as usize];
+        let mut buf = vec![0; handle.len];
         let size = self.read_to(handle.offset, &mut buf)?;
         buf.truncate(size);
         Ok(buf)