Release dir locks on engine closing (#381)

 

Signed-off-by: Yang Zhang <yang.zhang@pingcap.com>

Author: v01dstar

.github/workflows/rust.yml

@@ -30,7 +30,7 @@ jobs:
         uses: actions-rs/toolchain@v1
         with:
           profile: minimal
-          toolchain: nightly-2023-12-31
+          toolchain: nightly-2025-04-03
           override: true
           components: rustfmt, clippy, rust-src
       - uses: Swatinem/rust-cache@v1
@@ -65,7 +65,7 @@ jobs:
         uses: actions-rs/toolchain@v1
         with:
           profile: minimal
-          toolchain: 1.75.0
+          toolchain: 1.85.0
           override: true
           components: rustfmt, clippy, rust-src
       - uses: Swatinem/rust-cache@v1
@@ -92,7 +92,7 @@ jobs:
         uses: actions-rs/toolchain@v1
         with:
           profile: minimal
-          toolchain: nightly-2023-12-31
+          toolchain: nightly-2025-04-03
           override: true
           components: llvm-tools-preview
       - uses: Swatinem/rust-cache@v1

src/codec.rs

@@ -360,7 +360,6 @@ mod tests {
     use super::*;
     use protobuf::CodedOutputStream;
     use std::io::ErrorKind;
-    use std::{f32, f64, i16, i32, i64, u16, u32, u64};
 
     const U16_TESTS: &[u16] = &[
         i16::MIN as u16,

src/env/default.rs

@@ -62,7 +62,7 @@ impl Read for LogFile {
 impl Seek for LogFile {
     fn seek(&mut self, pos: SeekFrom) -> IoResult<u64> {
         fail_point!("log_file::seek::err", |_| {
-            Err(std::io::Error::new(std::io::ErrorKind::Other, "fp"))
+            Err(std::io::Error::other("fp"))
         });
         match pos {
             SeekFrom::Start(offset) => self.offset = offset as usize,

src/file_pipe_log/pipe.rs

@@ -7,6 +7,7 @@ use std::sync::Arc;
 
 use crossbeam::utils::CachePadded;
 use fail::fail_point;
+use fs2::FileExt;
 use log::error;
 use parking_lot::{Mutex, MutexGuard, RwLock};
 
@@ -464,7 +465,17 @@ impl<F: FileSystem> SinglePipe<F> {
 pub struct DualPipes<F: FileSystem> {
     pipes: [SinglePipe<F>; 2],
 
-    _dir_locks: Vec<StdFile>,
+    dir_locks: Vec<StdFile>,
+}
+
+impl<F: FileSystem> Drop for DualPipes<F> {
+    fn drop(&mut self) {
+        for lock in &self.dir_locks {
+            if let Err(e) = FileExt::unlock(lock) {
+                error!("error while unlocking directory: {e}");
+            }
+        }
+    }
 }
 
 impl<F: FileSystem> DualPipes<F> {
@@ -481,7 +492,7 @@ impl<F: FileSystem> DualPipes<F> {
 
         Ok(Self {
             pipes: [appender, rewriter],
-            _dir_locks: dir_locks,
+            dir_locks,
         })
     }
 
@@ -727,4 +738,21 @@ mod tests {
             assert_eq!(pipe_log.read_bytes(handle).unwrap(), content(i + 1));
         }
     }
+
+    #[test]
+    fn test_release_on_drop() {
+        let dir = Builder::new()
+            .prefix("test_release_on_drop")
+            .tempdir()
+            .unwrap();
+        let path = dir.path().to_str().unwrap();
+        let cfg = Config {
+            dir: path.to_owned(),
+            target_file_size: ReadableSize(1),
+            ..Default::default()
+        };
+        let pipe_log = new_test_pipes(&cfg).unwrap();
+        drop(pipe_log);
+        assert!(new_test_pipes(&cfg).is_ok());
+    }
 }

src/file_pipe_log/pipe_builder.rs

@@ -379,7 +379,7 @@ impl<F: FileSystem> DualPipesBuilder<F> {
     fn recover_queue_imp<M: ReplayMachine, FA: Factory<M>>(
         file_system: Arc<F>,
         recovery_cfg: RecoveryConfig,
-        files: &mut Vec<File<F>>,
+        files: &mut [File<F>],
         machine_factory: &FA,
     ) -> Result<M> {
         if recovery_cfg.concurrency == 0 || files.is_empty() {
@@ -390,7 +390,7 @@ impl<F: FileSystem> DualPipesBuilder<F> {
         let recovery_mode = recovery_cfg.mode;
         let recovery_read_block_size = recovery_cfg.read_block_size as usize;
 
-        let max_chunk_size = std::cmp::max((files.len() + concurrency - 1) / concurrency, 1);
+        let max_chunk_size = files.len().div_ceil(concurrency);
         let chunks = files.par_chunks_mut(max_chunk_size);
         let chunk_count = chunks.len();
         debug_assert!(chunk_count <= concurrency);

src/log_batch.rs

@@ -2,9 +2,9 @@
 
 use std::fmt::Debug;
 use std::io::BufRead;
+use std::mem;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::Arc;
-use std::{mem, u64};
 
 use byteorder::{BigEndian, LittleEndian, ReadBytesExt, WriteBytesExt};
 use log::error;
@@ -53,7 +53,7 @@ pub enum CompressionType {
 impl CompressionType {
     pub fn from_u8(t: u8) -> Result<Self> {
         if t <= CompressionType::Lz4 as u8 {
-            Ok(unsafe { mem::transmute(t) })
+            Ok(unsafe { mem::transmute::<u8, Self>(t) })
         } else {
             Err(Error::Corruption(format!(
                 "Unrecognized compression type: {t}"
@@ -168,7 +168,7 @@ pub enum OpType {
 impl OpType {
     pub fn from_u8(t: u8) -> Result<Self> {
         if t <= OpType::Del as u8 {
-            Ok(unsafe { mem::transmute(t) })
+            Ok(unsafe { mem::transmute::<u8, Self>(t) })
         } else {
             Err(Error::Corruption(format!("Unrecognized op type: {t}")))
         }

src/memtable.rs

@@ -763,7 +763,7 @@ impl<A: AllocatorTrait> MemTable<A> {
         debug_assert!(count > 0);
         self.entry_indexes
             .get(count - 1)
-            .map_or(false, |ei| ei.entries.unwrap().id.seq <= gate.seq)
+            .is_some_and(|ei| ei.entries.unwrap().id.seq <= gate.seq)
     }
 
     /// Returns the region ID.

src/metrics.rs

@@ -122,7 +122,7 @@ pub trait TimeMetric {
     }
 }
 
-impl<'a> TimeMetric for &'a Histogram {
+impl TimeMetric for &Histogram {
     fn observe(&self, duration: Duration) {
         Histogram::observe(self, duration.as_secs_f64());
     }

src/test_util.rs

@@ -63,7 +63,7 @@ where
 }
 
 pub struct PanicGuard {
-    prev_hook: *mut (dyn Fn(&panic::PanicInfo<'_>) + Sync + Send + 'static),
+    prev_hook: *mut (dyn Fn(&panic::PanicHookInfo<'_>) + Sync + Send + 'static),
 }
 
 struct PointerHolder<T: ?Sized>(*mut T);

src/util.rs

@@ -148,7 +148,7 @@ impl<'de> Deserialize<'de> for ReadableSize {
     {
         struct SizeVisitor;
 
-        impl<'de> Visitor<'de> for SizeVisitor {
+        impl Visitor<'_> for SizeVisitor {
             type Value = ReadableSize;
 
             fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
@@ -221,7 +221,7 @@ pub fn unhash_u64(mut i: u64) -> u64 {
 
 pub mod lz4 {
     use crate::{Error, Result};
-    use std::{i32, ptr};
+    use std::ptr;
 
     pub const DEFAULT_LZ4_COMPRESSION_LEVEL: usize = 1;
 
@@ -330,7 +330,7 @@ pub trait Factory<Target>: Send + Sync {
 /// ```
 #[inline]
 pub fn round_up(offset: usize, alignment: usize) -> usize {
-    (offset + alignment - 1) / alignment * alignment
+    offset.div_ceil(alignment) * alignment
 }
 
 /// Returns an aligned `offset`.