Add some micro optimizations (#100)

This PR contains some micro optimizations: 1. use `fdatasync` instead of `fsync` 2. Fallocate without `FALLOC_FL_KEEP_SIZE` 3. Reduce useless fallocate near the end of file, so we don't need to truncate them again when rotating Up to 60% throughput improvement. Signed-off-by: tabokie <xy.tao@outlook.com>
tikv · Sep 18, 2021 · 64716c8 · 64716c8
1 parent 3a12e9b
commit 64716c8
Show file tree

Hide file tree

Showing 5 changed files with 57 additions and 25 deletions.
diff --git a/src/file_pipe_log.rs b/src/file_pipe_log.rs
@@ -132,6 +132,7 @@ struct FileToRecover<R: Seek + Read> {
 struct ActiveFile<W: Seek + Write> {
     fd: Arc<LogFd>,
     writer: W,
+
     written: usize,
     capacity: usize,
     last_sync: usize,
@@ -165,9 +166,11 @@ impl<W: Seek + Write> ActiveFile<W> {
     }
 
     fn truncate(&mut self) -> Result<()> {
-        self.fd.truncate(self.written)?;
-        self.fd.sync()?;
-        self.capacity = self.written;
+        if self.written < self.capacity {
+            self.fd.truncate(self.written)?;
+            self.fd.sync()?;
+            self.capacity = self.written;
+        }
         Ok(())
     }
 
@@ -176,13 +179,19 @@ impl<W: Seek + Write> ActiveFile<W> {
         self.written = 0;
         let mut buf = Vec::with_capacity(LOG_FILE_HEADER_LEN);
         LogFileHeader::new().encode(&mut buf)?;
-        self.write(&buf, true)
+        self.write(&buf, true, 0)
     }
 
-    fn write(&mut self, buf: &[u8], sync: bool) -> Result<()> {
+    fn write(&mut self, buf: &[u8], sync: bool, target_file_size: usize) -> Result<()> {
         if self.written + buf.len() > self.capacity {
             // Use fallocate to pre-allocate disk space for active file.
-            let alloc = std::cmp::max(self.written + buf.len() - self.capacity, FILE_ALLOCATE_SIZE);
+            let alloc = std::cmp::max(
+                self.written + buf.len() - self.capacity,
+                std::cmp::min(
+                    FILE_ALLOCATE_SIZE,
+                    target_file_size.saturating_sub(self.capacity),
+                ),
+            );
             self.fd.allocate(self.capacity, alloc)?;
             self.capacity += alloc;
         }
@@ -281,7 +290,8 @@ impl<B: FileBuilder> LogManager<B> {
 
     fn new_log_file(&mut self) -> Result<()> {
         if self.active_file_id.valid() {
-            // self.truncate_active_log()?;
+            // Necessary to truncate extra zeros from fallocate().
+            self.truncate_active_log()?;
         }
         self.active_file_id = if self.active_file_id.valid() {
             self.active_file_id.forward(1)
@@ -351,7 +361,7 @@ impl<B: FileBuilder> LogManager<B> {
             *sync = true;
         }
         let offset = self.active_file.written as u64;
-        self.active_file.write(content, *sync)?;
+        self.active_file.write(content, *sync, self.rotate_size)?;
         Ok((self.active_file_id, offset, self.active_file.fd.clone()))
     }
 
@@ -618,11 +628,6 @@ impl<B: FileBuilder> FilePipeLog<B> {
 }
 
 impl<B: FileBuilder> PipeLog for FilePipeLog<B> {
-    fn close(&self) -> Result<()> {
-        self.mut_queue(LogQueue::Rewrite).truncate_active_log()?;
-        self.mut_queue(LogQueue::Append).truncate_active_log()
-    }
-
     fn file_size(&self, queue: LogQueue, file_id: FileId) -> Result<u64> {
         self.get_queue(queue)
             .get_fd(file_id)

diff --git a/src/log_batch.rs b/src/log_batch.rs
@@ -42,8 +42,15 @@ pub enum CompressionType {
 }
 
 impl CompressionType {
-    pub fn from_u8(t: u8) -> CompressionType {
-        unsafe { mem::transmute(t) }
+    pub fn from_u8(t: u8) -> Result<Self> {
+        if t <= CompressionType::Lz4 as u8 {
+            Ok(unsafe { mem::transmute(t) })
+        } else {
+            Err(Error::Corruption(format!(
+                "Unrecognized compression type: {}",
+                t
+            )))
+        }
     }
 
     pub fn to_u8(self) -> u8 {
@@ -148,8 +155,12 @@ pub enum OpType {
 }
 
 impl OpType {
-    pub fn from_u8(t: u8) -> OpType {
-        unsafe { mem::transmute(t) }
+    pub fn from_u8(t: u8) -> Result<Self> {
+        if t <= OpType::Del as u8 {
+            Ok(unsafe { mem::transmute(t) })
+        } else {
+            Err(Error::Corruption(format!("Unrecognized op type: {}", t)))
+        }
     }
 
     pub fn to_u8(self) -> u8 {
@@ -176,7 +187,7 @@ impl KeyValue {
     }
 
     pub fn decode(buf: &mut SliceReader) -> Result<KeyValue> {
-        let op_type = OpType::from_u8(codec::read_u8(buf)?);
+        let op_type = OpType::from_u8(codec::read_u8(buf)?)?;
         let k_len = codec::decode_var_u64(buf)? as usize;
         let key = &buf[..k_len];
         buf.consume(k_len);
@@ -620,7 +631,16 @@ impl LogBatch {
 
         let len = codec::decode_u64(buf)? as usize;
         let offset = codec::decode_u64(buf)? as usize;
-        let compression_type = CompressionType::from_u8(len as u8);
+        let compression_type = CompressionType::from_u8(len as u8)?;
+        if offset > len {
+            return Err(Error::Corruption(
+                "Log item offset exceeds log batch length".to_owned(),
+            ));
+        } else if offset < LOG_BATCH_HEADER_LEN {
+            return Err(Error::Corruption(
+                "Log item offset is smaller than log batch header length".to_owned(),
+            ));
+        }
         Ok((offset, compression_type, len >> 8))
     }
 

diff --git a/src/log_file.rs b/src/log_file.rs
@@ -10,7 +10,7 @@ use nix::errno::Errno;
 use nix::fcntl::{self, OFlag};
 use nix::sys::stat::Mode;
 use nix::sys::uio::{pread, pwrite};
-use nix::unistd::{close, fsync, ftruncate, lseek, Whence};
+use nix::unistd::{close, ftruncate, lseek, Whence};
 use nix::NixPath;
 
 /// A `LogFd` is a RAII file that provides basic I/O functionality.
@@ -59,7 +59,14 @@ impl LogFd {
     }
 
     pub fn sync(&self) -> IoResult<()> {
-        fsync(self.0).map_err(|e| from_nix_error(e, "fsync"))
+        #[cfg(target_os = "linux")]
+        {
+            nix::unistd::fdatasync(self.0).map_err(|e| from_nix_error(e, "fdatasync"))
+        }
+        #[cfg(not(target_os = "linux"))]
+        {
+            nix::unistd::fsync(self.0).map_err(|e| from_nix_error(e, "fsync"))
+        }
     }
 
     pub fn read(&self, mut offset: usize, buf: &mut [u8]) -> IoResult<usize> {
@@ -112,7 +119,7 @@ impl LogFd {
         {
             fcntl::fallocate(
                 self.0,
-                fcntl::FallocateFlags::FALLOC_FL_KEEP_SIZE,
+                fcntl::FallocateFlags::empty(),
                 offset as i64,
                 size as i64,
             )

diff --git a/src/pipe_log.rs b/src/pipe_log.rs
@@ -84,9 +84,6 @@ impl FileId {
 }
 
 pub trait PipeLog: Sized {
-    /// Close the pipe log.
-    fn close(&self) -> Result<()>;
-
     fn file_size(&self, queue: LogQueue, file_id: FileId) -> Result<u64>;
 
     /// Total size of the given queue.

diff --git a/src/reader.rs b/src/reader.rs
@@ -55,6 +55,9 @@ impl<R: Read + Seek> LogItemBatchFileReader<R> {
                 LOG_BATCH_HEADER_LEN,
                 0,
             )?)?;
+            if self.valid_offset + len > self.size {
+                return Err(Error::Corruption("log batch header broken".to_owned()));
+            }
             let entries_offset = self.valid_offset + LOG_BATCH_HEADER_LEN;
             let entries_len = footer_offset - LOG_BATCH_HEADER_LEN;