Unlock the LOCK files on drop (#379)

 

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

Cargo.toml

@@ -3,7 +3,7 @@ name = "raft-engine"
 version = "0.4.2"
 authors = ["The TiKV Project Developers"]
 edition = "2018"
-rust-version = "1.75.0"
+rust-version = "1.85.0"
 description = "A persistent storage engine for Multi-Raft logs"
 readme = "README.md"
 repository = "https://github.com/tikv/raft-engine"

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/log_file.rs

@@ -107,13 +107,12 @@ impl<F: FileSystem> LogFileWriter<F> {
             }
             self.capacity += alloc;
         }
-        self.writer.write_all(buf).map_err(|e| {
+        self.writer.write_all(buf).inspect_err(|_| {
             self.writer
                 .seek(SeekFrom::Start(self.written as u64))
                 .unwrap_or_else(|e| {
                     panic!("failed to reseek after write failure: {}", e);
                 });
-            e
         })?;
         self.written = new_written;
         Ok(())

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

@@ -228,7 +228,7 @@ impl<A: AllocatorTrait> MemTable<A> {
         }
 
         if let Some(g) = rhs.atomic_group.take() {
-            assert!(self.atomic_group.map_or(true, |(_, end)| end <= g.0));
+            assert!(self.atomic_group.is_none_or(|(_, end)| end <= g.0));
             self.atomic_group = Some(g);
         }
 
@@ -545,7 +545,7 @@ impl<A: AllocatorTrait> MemTable<A> {
     }
 
     pub fn apply_rewrite_atomic_group(&mut self, start: FileSeq, end: FileSeq) {
-        assert!(self.atomic_group.map_or(true, |(_, b)| b <= start));
+        assert!(self.atomic_group.is_none_or(|(_, b)| b <= start));
         self.atomic_group = Some((start, end));
     }
 
@@ -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());
     }