diff options
Diffstat (limited to 'log_db')
| -rw-r--r-- | log_db/src/common.rs | 93 | ||||
| -rw-r--r-- | log_db/src/engine.rs | 173 | ||||
| -rw-r--r-- | log_db/src/lib.rs | 74 | ||||
| -rw-r--r-- | log_db/src/lock.rs | 100 | ||||
| -rw-r--r-- | log_db/src/log_reader_forward.rs | 8 |
5 files changed, 241 insertions, 207 deletions
diff --git a/log_db/src/common.rs b/log_db/src/common.rs index 6b7cac8..cbf6ce4 100644 --- a/log_db/src/common.rs +++ b/log_db/src/common.rs @@ -7,7 +7,6 @@ use std::fs::{self, metadata, File}; use std::io::{self, Read, Seek, SeekFrom, Write}; use std::ops::{Bound, RangeBounds}; use std::path::{Path, PathBuf}; -use std::thread; use thiserror::Error; use uuid::Uuid; @@ -20,12 +19,14 @@ use std::os::unix::fs::MetadataExt; use std::os::windows::fs::MetadataExt; pub const ACTIVE_SYMLINK_FILENAME: &str = "active"; +pub const LOCK_FILENAME: &str = "lock"; +pub const EXCL_LOCK_REQ_FILENAME: &str = "excl_lock_req"; +pub const INITIALIZED_FILENAME: &str = "initialized"; + pub const METADATA_FILE_HEADER_SIZE: usize = 24; pub const METADATA_ROW_LENGTH: usize = 16; -pub const EXCL_LOCK_REQUEST_FILENAME: &str = "excl_lock_req"; -pub const INIT_LOCK_FILENAME: &str = "init_lock"; pub const DEFAULT_READ_BUF_SIZE: usize = 1024 * 1024; // 1 MB -pub const TEST_RESOURCES_DIR: &str = "tests/resources"; +pub const LOCK_WAIT_MAX_MS: u64 = 1000; // Serialized value tags pub const B_NULL: u8 = 0x0; @@ -470,7 +471,10 @@ pub fn create_segment_metadata_file( let metadata_filename = format!("metadata.{}", new_num); let metadata_path = data_dir_path.join(metadata_filename); - let mut metadata_file = APPEND_MODE.clone().create(true).open(&metadata_path)?; + let mut metadata_file = fs::OpenOptions::new() + .create(true) + .append(true) + .open(&metadata_path)?; let metadata_header = MetadataHeader { version: 1, @@ -529,7 +533,6 @@ pub fn create_segment_data_file(data_dir_path: &Path) -> DBResult<(Uuid, PathBuf let new_segment_path = data_dir_path.join(uuid.to_string()); fs::OpenOptions::new() .create(true) - .write(true) .append(true) .open(&new_segment_path)?; @@ -650,84 +653,6 @@ pub fn ensure_active_metadata_is_valid( } } -const LOCK_WAIT_MAX_MS: u64 = 1000; - -pub fn is_exclusive_lock_requested(data_dir: &Path) -> DBResult<bool> { - let lock_request_path = data_dir.join(EXCL_LOCK_REQUEST_FILENAME); - let lock_request_file = fs::OpenOptions::new() - .create(true) - .write(true) // When requesting a lock, we need to have either read or write permissions - .open(&lock_request_path)?; - - // Attempt to acquire a shared lock on the lock request file - // If the file is already locked, return false - match lock_request_file.try_lock_shared() { - Err(e) => { - if e.kind() == lock_contended_error().kind() { - return Ok(true); - } - return Err(DBError::IOError(e)); - } - - Ok(_) => { - // Check that the exclusive lock request file is still the same as the one we opened - if !is_file_same_as_path(&lock_request_file, &lock_request_path)? { - // The lock request file has been removed - return Err(DBError::ConsistencyError( - "Lock request file was removed while checking for exclusive lock".to_owned(), - )); - } - - lock_request_file.unlock()?; - return Ok(false); - } - } -} - -pub fn request_shared_lock(data_dir: &Path, file: &mut fs::File) -> DBResult<()> { - let mut timeout = 5; - loop { - if is_exclusive_lock_requested(data_dir)? { - debug!( - "Exclusive lock requested, waiting for {}ms before requesting a shared lock again", - timeout - ); - thread::sleep(std::time::Duration::from_millis(timeout)); - timeout *= 2; - - if timeout > LOCK_WAIT_MAX_MS { - return Err(DBError::LockRequestError( - "Acquisition of shared lock timed out after {LOCK_WAIT_MAX_MS}".to_owned(), - )); - } - } else { - file.lock_shared()?; - return Ok(()); - } - } -} - -pub fn request_exclusive_lock(data_dir: &Path, file: &mut fs::File) -> DBResult<()> { - // Create a lock on the exclusive lock request file to signal to readers that they should wait - let lock_request_path = data_dir.join(EXCL_LOCK_REQUEST_FILENAME); - let lock_request_file = fs::OpenOptions::new() - .create(true) - .write(true) // When requesting a lock, we need to have either read or write permissions - .open(&lock_request_path)?; - - // Attempt to acquire an exclusive lock on the lock request file - // This will block until the lock is acquired - lock_request_file.lock_exclusive()?; - - // Acquire an exclusive lock on the segment files - file.lock_exclusive()?; - - // Unlock the request file - lock_request_file.unlock()?; - - Ok(()) -} - pub struct OwnedBounds<T> { start: Bound<T>, end: Bound<T>, diff --git a/log_db/src/engine.rs b/log_db/src/engine.rs index 2858bab..8ad4590 100644 --- a/log_db/src/engine.rs +++ b/log_db/src/engine.rs @@ -2,13 +2,15 @@ use super::*; pub struct Engine<R: Recordable> { pub config: Config<R>, + pub lock_manager: LockManager, - data_dir: PathBuf, - active_metadata_file: fs::File, - active_data_file: fs::File, + data_dir_path: PathBuf, primary_key_index: usize, refresh_next_logkey: LogKey, + active_metadata_file: fs::File, + active_data_file: fs::File, + // TODO: these could be made private. Currently they are public for testing in lib.rs. pub primary_memtable: PrimaryMemtable, pub secondary_memtables: Vec<SecondaryMemtable>, @@ -17,14 +19,12 @@ pub struct Engine<R: Recordable> { impl<R: Recordable> Engine<R> { pub fn initialize(config: Config<R>) -> DBResult<Engine<R>> { info!("Initializing DB..."); - // If data_dir does not exist or is empty, create it and any necessary files - // After creation, the directory should always be in a complete state - // without missing files. - // A tempdir-move strategy is used to achieve one-phase commit. + // If data_dir does not exist or is empty, create it and any necessary files. + // After creation, the directory should always be in a complete state without missing files. // Ensure the data directory exists - let data_dir = Path::new(&config.data_dir).to_path_buf(); - match fs::create_dir(&data_dir) { + let data_dir_path = Path::new(&config.data_dir).to_path_buf(); + match fs::create_dir(&data_dir_path) { Ok(_) => {} Err(e) => { if e.kind() != io::ErrorKind::AlreadyExists { @@ -33,38 +33,43 @@ impl<R: Recordable> Engine<R> { } } - // Create an initialize lock file to prevent multiple concurrent initializations - let init_lock_file = fs::OpenOptions::new() - .create(true) - .write(true) - .open(&data_dir.join(INIT_LOCK_FILENAME))?; - - init_lock_file.lock_exclusive()?; + // Create the lock file first to prevent multiple concurrent initializations + let mut lock_manager = LockManager::new(data_dir_path.clone())?; + lock_manager.lock_exclusive()?; // We have acquired the lock, check if the data directory is in a complete state // If not, initialize it, otherwise skip. - if !fs::exists(data_dir.join(ACTIVE_SYMLINK_FILENAME))? { - let (segment_uuid, _) = create_segment_data_file(&data_dir)?; - let (segment_num, _) = create_segment_metadata_file(&data_dir, &segment_uuid)?; - set_active_segment(&data_dir, segment_num)?; + if !fs::exists(data_dir_path.join(INITIALIZED_FILENAME))? { + // Delete all files except the lock files to ensure a clean state + for entry in fs::read_dir(&data_dir_path)? { + let entry = entry?; + let path = entry.path(); + if path.is_file() + && path.file_name().unwrap() != LOCK_FILENAME + && path.file_name().unwrap() != EXCL_LOCK_REQ_FILENAME + { + fs::remove_file(&path)?; + } + } - // Create the exclusive lock request file - fs::OpenOptions::new() - .create(true) - .write(true) - .open(data_dir.join(EXCL_LOCK_REQUEST_FILENAME))?; + // Create the initial segment files + let (segment_uuid, _) = create_segment_data_file(&data_dir_path)?; + let (segment_num, _) = create_segment_metadata_file(&data_dir_path, &segment_uuid)?; + set_active_segment(&data_dir_path, segment_num)?; + + // Create the initialized file to indicate that the directory is in a complete state + fs::File::create(data_dir_path.join(INITIALIZED_FILENAME))?; } - init_lock_file.unlock()?; + lock_manager.unlock()?; // Calculate the index of the primary value in a record let primary_key_index = config .fields .iter() .position(|(field, _)| field == &config.primary_key) - .ok_or(io::Error::new( - io::ErrorKind::InvalidInput, - "Primary key not found in schema after initialize", + .ok_or(DBError::ValidationError( + "Primary key not found in schema after initialize".to_owned(), ))?; // Join primary key and secondary keys vec into a single vec @@ -105,12 +110,13 @@ impl<R: Recordable> Engine<R> { let mut engine = Engine::<R> { config, - data_dir, - active_metadata_file, - active_data_file, + lock_manager, + data_dir_path, primary_key_index, primary_memtable, secondary_memtables, + active_metadata_file, + active_data_file, refresh_next_logkey: LogKey::new(1, 0), }; @@ -123,16 +129,16 @@ impl<R: Recordable> Engine<R> { } pub fn refresh_indexes(&mut self) -> DBResult<()> { - let active_symlink_path = self.data_dir.join(ACTIVE_SYMLINK_FILENAME); + let active_symlink_path = self.data_dir_path.join(ACTIVE_SYMLINK_FILENAME); let active_target = fs::read_link(active_symlink_path)?; - let active_metadata_path = self.data_dir.join(active_target); + let active_metadata_path = self.data_dir_path.join(active_target); let to_segnum = parse_segment_number(&active_metadata_path)?; let from_segnum = self.refresh_next_logkey.segment_num(); let mut from_index = self.refresh_next_logkey.index(); for segnum in from_segnum..=to_segnum { - let metadata_path = self.data_dir.join(metadata_filename(segnum)); + let metadata_path = self.data_dir_path.join(metadata_filename(segnum)); let mut metadata_file = READ_MODE.open(&metadata_path)?; let metadata_len = metadata_file.seek(SeekFrom::End(0))?; @@ -144,12 +150,10 @@ impl<R: Recordable> Engine<R> { ))); } - request_shared_lock(&self.data_dir, &mut metadata_file)?; - let metadata_header = read_metadata_header(&mut metadata_file)?; validate_metadata_header(&metadata_header)?; - let data_path = self.data_dir.join(metadata_header.uuid.to_string()); + let data_path = self.data_dir_path.join(metadata_header.uuid.to_string()); let data_file = READ_MODE.open(data_path)?; for ForwardLogReaderItem { record, index } in @@ -219,26 +223,23 @@ impl<R: Recordable> Engine<R> { } pub fn batch_upsert_records(&mut self, records: impl Iterator<Item = Record>) -> DBResult<()> { - debug!("Opening file in append mode and acquiring exclusive lock..."); - - // Acquire an exclusive lock for writing - request_exclusive_lock(&self.data_dir, &mut self.active_metadata_file)?; + debug!("Opening file in append mode..."); if !self.ensure_metadata_file_is_active()? - || !ensure_active_metadata_is_valid(&self.data_dir, &mut self.active_metadata_file)? + || !ensure_active_metadata_is_valid( + &self.data_dir_path, + &mut self.active_metadata_file, + )? { // The log file has been rotated, so we must try again - self.active_metadata_file.unlock()?; return self.batch_upsert_records(records); } - self.active_data_file.lock_exclusive()?; - - let active_symlink_path = self.data_dir.join(ACTIVE_SYMLINK_FILENAME); + let active_symlink_path = self.data_dir_path.join(ACTIVE_SYMLINK_FILENAME); let active_target = fs::read_link(active_symlink_path)?; let segment_num = parse_segment_number(&active_target)?; - debug!("Exclusive lock acquired, appending to log file"); + debug!("Appending to log file"); let mut serialized_data: Vec<u8> = vec![]; let mut serialized_metadata: Vec<u8> = vec![]; @@ -284,11 +285,7 @@ impl<R: Recordable> Engine<R> { self.active_metadata_file.sync_all()?; } - debug!("Records appended to log file, releasing locks"); - - // Manually release the locks because the file handles are left open - self.active_data_file.unlock()?; - self.active_metadata_file.unlock()?; + debug!("Records appended to log file"); for (log_key, record) in pending_memtable_insertions { self.insert_record_to_memtables(log_key, record); @@ -403,18 +400,14 @@ impl<R: Recordable> Engine<R> { for (segment_num, mut segment_indexes) in log_keys_map { segment_indexes.sort_unstable(); - let metadata_path = &self.data_dir.join(metadata_filename(segment_num)); + let metadata_path = &self.data_dir_path.join(metadata_filename(segment_num)); let mut metadata_file = READ_MODE.open(&metadata_path)?; - request_shared_lock(&self.data_dir, &mut metadata_file)?; - let metadata_header = read_metadata_header(&mut metadata_file)?; - let data_path = &self.data_dir.join(metadata_header.uuid.to_string()); + let data_path = &self.data_dir_path.join(metadata_header.uuid.to_string()); let mut data_file = READ_MODE.open(&data_path)?; - data_file.lock_shared()?; - let header_size = METADATA_FILE_HEADER_SIZE as i64; let row_length = METADATA_ROW_LENGTH as i64; let mut current_metadata_offset = header_size; @@ -439,9 +432,6 @@ impl<R: Recordable> Engine<R> { current_metadata_offset = new_metadata_offset + row_length; } - - metadata_file.unlock()?; - data_file.unlock()?; } Ok(records) @@ -510,21 +500,19 @@ impl<R: Recordable> Engine<R> { /// If the segment has been rotated, the handle will be closed and reopened. /// Returns `false` if the file has been rotated and the handle has been reopened, `true` otherwise. fn ensure_metadata_file_is_active(&mut self) -> DBResult<bool> { - let active_target = fs::read_link(&self.data_dir.join(ACTIVE_SYMLINK_FILENAME))?; - let active_metadata_path = &self.data_dir.join(active_target); + let active_target = fs::read_link(&self.data_dir_path.join(ACTIVE_SYMLINK_FILENAME))?; + let active_metadata_path = &self.data_dir_path.join(active_target); let correct = is_file_same_as_path(&self.active_metadata_file, &active_metadata_path)?; if !correct { debug!("Metadata file has been rotated. Reopening..."); let mut metadata_file = APPEND_MODE.open(&active_metadata_path)?; - request_shared_lock(&self.data_dir, &mut metadata_file)?; - let metadata_header = read_metadata_header(&mut self.active_metadata_file)?; validate_metadata_header(&metadata_header)?; - let data_file_path = &self.data_dir.join(metadata_header.uuid.to_string()); + let data_file_path = &self.data_dir_path.join(metadata_header.uuid.to_string()); self.active_metadata_file = metadata_file; self.active_data_file = APPEND_MODE.open(&data_file_path)?; @@ -536,9 +524,8 @@ impl<R: Recordable> Engine<R> { } pub fn delete_by_field(&mut self, field: &R::Field, value: &Value) -> DBResult<Vec<Record>> { - let value_batch = std::iter::once(value); let recs: Vec<Record> = self - .batch_find_by_records(field, value_batch)? + .batch_find_by_records(field, std::iter::once(value))? .into_iter() .map(|(_, mut rec)| { rec.tombstone = true; @@ -546,9 +533,6 @@ impl<R: Recordable> Engine<R> { }) .collect(); - request_exclusive_lock(&self.data_dir, &mut self.active_metadata_file)?; - self.active_data_file.lock_exclusive()?; - for record in &recs { let record_serialized = record.serialize(); @@ -584,36 +568,28 @@ impl<R: Recordable> Engine<R> { self.remove_record_from_memtables(&record); } - self.active_metadata_file.unlock()?; - self.active_data_file.unlock()?; - debug!("Records deleted"); Ok(recs) } pub fn do_maintenance_tasks(&mut self) -> DBResult<()> { - request_exclusive_lock(&self.data_dir, &mut self.active_metadata_file)?; - - ensure_active_metadata_is_valid(&self.data_dir, &mut self.active_metadata_file)?; + ensure_active_metadata_is_valid(&self.data_dir_path, &mut self.active_metadata_file)?; let metadata_size = self.active_metadata_file.seek(SeekFrom::End(0))?; if metadata_size >= self.config.segment_size as u64 { self.rotate_and_compact()?; } - self.active_metadata_file.unlock()?; - Ok(()) } fn rotate_and_compact(&mut self) -> DBResult<()> { debug!("Active log size exceeds threshold, starting rotation and compaction..."); - self.active_data_file.lock_shared()?; let original_data_len = self.active_data_file.seek(SeekFrom::End(0))?; - let active_target = fs::read_link(&self.data_dir.join(ACTIVE_SYMLINK_FILENAME))?; + let active_target = fs::read_link(&self.data_dir_path.join(ACTIVE_SYMLINK_FILENAME))?; let active_num = parse_segment_number(&active_target)?; debug!("Reading segment data into a BTreeMap"); @@ -633,8 +609,6 @@ impl<R: Recordable> Engine<R> { }) .collect(); - self.active_data_file.unlock()?; - for (pk, record) in forward_read_items.iter() { pk_to_item_map.insert(pk, record); } @@ -647,7 +621,7 @@ impl<R: Recordable> Engine<R> { // Create a new log data file and write it debug!("Opening new data file and writing compacted data"); - let (new_data_uuid, new_data_path) = create_segment_data_file(&self.data_dir)?; + let (new_data_uuid, new_data_path) = create_segment_data_file(&self.data_dir_path)?; let mut new_data_file = APPEND_MODE.open(&new_data_path)?; let mut pk_to_data_map = BTreeMap::new(); @@ -701,15 +675,15 @@ impl<R: Recordable> Engine<R> { temp_metadata_file.sync_all()?; debug!("Moving temporary files to their final locations"); - let new_data_path = &self.data_dir.join(new_data_uuid.to_string()); - let active_metadata_path = &self.data_dir.join(metadata_filename(active_num)); // overwrite active + let new_data_path = &self.data_dir_path.join(new_data_uuid.to_string()); + let active_metadata_path = &self.data_dir_path.join(metadata_filename(active_num)); // overwrite active fs::rename(&temp_metadata_path, &active_metadata_path)?; debug!("Compaction complete, creating new segment"); let new_segment_num = active_num + 1; - let new_metadata_path = self.data_dir.join(metadata_filename(new_segment_num)); + let new_metadata_path = self.data_dir_path.join(metadata_filename(new_segment_num)); let mut new_metadata_file = APPEND_MODE.clone().create(true).open(&new_metadata_path)?; let new_metadata_header = MetadataHeader { @@ -719,15 +693,11 @@ impl<R: Recordable> Engine<R> { new_metadata_file.write_all(&new_metadata_header.serialize())?; - set_active_segment(&self.data_dir, new_segment_num)?; - - // Old active metadata file should lose lock by RAII, or by - // the manual unlock call in the do_maintenance_tasks method. + set_active_segment(&self.data_dir_path, new_segment_num)?; self.active_metadata_file = APPEND_MODE.open(&new_metadata_path)?; self.active_data_file = APPEND_MODE.open(&new_data_path)?; - // The new active log file is not locked by this client so it cannot be touched. debug!( "Active log file {} rotated and compacted, new segment: {}", active_num, new_segment_num @@ -744,4 +714,21 @@ impl<R: Recordable> Engine<R> { .find(|(f, _)| f == field) .map(|(_, t)| t) } + + pub fn with_exclusive_lock<T>( + &mut self, + f: impl FnOnce(&mut Self) -> DBResult<T>, + ) -> DBResult<T> { + self.lock_manager.lock_exclusive()?; + let result = f(self)?; + self.lock_manager.unlock()?; + Ok(result) + } + + pub fn with_shared_lock<T>(&mut self, f: impl FnOnce(&mut Self) -> DBResult<T>) -> DBResult<T> { + self.lock_manager.lock_shared()?; + let result = f(self)?; + self.lock_manager.unlock()?; + Ok(result) + } } diff --git a/log_db/src/lib.rs b/log_db/src/lib.rs index f7a7e47..72e5f81 100644 --- a/log_db/src/lib.rs +++ b/log_db/src/lib.rs @@ -1,7 +1,7 @@ #[macro_use] extern crate log; -use fs2::FileExt; +use fs2::{lock_contended_error, FileExt}; use std::collections::BTreeMap; use std::fmt::Debug; use std::fmt::Display; @@ -10,11 +10,13 @@ use std::io::{self, Read, Seek, SeekFrom, Write}; use std::marker::PhantomData; use std::ops::*; use std::path::{Path, PathBuf}; +use std::thread; #[macro_use] mod common; mod config; mod engine; +mod lock; mod log_reader_forward; mod memtable_primary; mod memtable_secondary; @@ -27,6 +29,7 @@ pub use record::Recordable; use common::*; use config::*; use engine::*; +use lock::*; use log_reader_forward::*; use memtable_primary::PrimaryMemtable; use memtable_secondary::SecondaryMemtable; @@ -56,7 +59,11 @@ impl<R: Recordable> DB<R> { record.validate(&self.engine.config.fields)?; debug!("Record is valid"); - self.engine.batch_upsert_records(std::iter::once(record)) + self.engine.with_exclusive_lock(move |engine| { + engine.batch_upsert_records(std::iter::once(record)) + })?; + + Ok(()) } /// Insert a batch of records into the database. If the primary key value for a record already exists, @@ -73,20 +80,26 @@ impl<R: Recordable> DB<R> { } debug!("Records are valid"); - self.engine.batch_upsert_records(records.into_iter()) + self.engine + .with_exclusive_lock(move |engine| engine.batch_upsert_records(records.into_iter()))?; + + Ok(()) } /// Get a record by its primary index value. /// E.g. `db.get(Value::Int(10))`. pub fn get(&mut self, value: &Value) -> DBResult<Option<R>> { - let value_batch = std::iter::once(value); - let records = self - .engine - // TODO: This clone is only here to appease the borrow checker - .batch_find_by_records(&self.engine.config.primary_key.clone(), value_batch)?; - assert!(records.len() <= 1); + let recs = self.engine.with_shared_lock(|engine| { + engine.batch_find_by_records( + // TODO: This clone is only here to appease the borrow checker + &engine.config.primary_key.clone(), + std::iter::once(value), + ) + })?; + + assert!(recs.len() <= 1); - Ok(records + Ok(recs .into_iter() .next() .map(|(_, rec)| R::from_record(rec.values))) @@ -95,10 +108,11 @@ impl<R: Recordable> DB<R> { /// Get a collection of records based on a field value. /// Indexes will be used if they are applicable. pub fn find_by(&mut self, field: &R::Field, value: &Value) -> DBResult<Vec<R>> { - let value_batch = std::iter::once(value); - Ok(self - .engine - .batch_find_by_records(field, value_batch)? + let recs = self.engine.with_shared_lock(|engine| { + engine.batch_find_by_records(field, std::iter::once(value)) + })?; + + Ok(recs .into_iter() .map(|(_, rec)| R::from_record(rec.values)) .collect()) @@ -113,9 +127,11 @@ impl<R: Recordable> DB<R> { field: &R::Field, values: &[Value], ) -> DBResult<Vec<(usize, R)>> { - Ok(self + let recs = self .engine - .batch_find_by_records(field, values.iter())? + .with_shared_lock(|engine| engine.batch_find_by_records(field, values.iter()))?; + + Ok(recs .into_iter() .map(|(tag, rec)| (tag, R::from_record(rec.values))) .collect()) @@ -126,9 +142,11 @@ impl<R: Recordable> DB<R> { field: &R::Field, range: B, ) -> DBResult<Vec<R>> { - Ok(self + let recs = self .engine - .range_by_records(field, range)? + .with_shared_lock(|engine| engine.range_by_records(field, range))?; + + Ok(recs .into_iter() .map(|rec| R::from_record(rec.values)) .collect()) @@ -141,7 +159,9 @@ impl<R: Recordable> DB<R> { /// Deletion is done by marking the record as a tombstone. The record will still be present in the log file, /// but will be ignored by reads. Upon compaction, tombstoned records will be removed. pub fn delete_by(&mut self, field: &R::Field, value: &Value) -> DBResult<Vec<R>> { - let recs = self.engine.delete_by_field(field, value)?; + let recs = self + .engine + .with_exclusive_lock(|engine| engine.delete_by_field(field, value))?; Ok(recs .into_iter() @@ -151,10 +171,12 @@ impl<R: Recordable> DB<R> { /// Delete record by primary key. pub fn delete(&mut self, pk: &Value) -> DBResult<Option<R>> { - let recs = self - .engine - // TODO: This clone is only here to appease the borrow checker - .delete_by_field(&self.engine.config.primary_key.clone(), pk)?; + let recs = self.engine.with_exclusive_lock(|engine| { + engine + // TODO: This clone is only here to appease the borrow checker + .delete_by_field(&engine.config.primary_key.clone(), pk) + })?; + assert!(recs.len() <= 1); Ok(recs @@ -171,13 +193,15 @@ impl<R: Recordable> DB<R> { /// You may call this function in a separate thread or process to avoid blocking the main thread. /// However, the database will be exclusively locked, so all writes and reads will be blocked during the tasks. pub fn do_maintenance_tasks(&mut self) -> DBResult<()> { - self.engine.do_maintenance_tasks() + self.engine + .with_exclusive_lock(|engine| engine.do_maintenance_tasks()) } /// Refresh the in-memory indexes from the log files. /// This needs to only be called if the read consistency is set to `ReadConsistency::Eventual`. pub fn refresh_indexes(&mut self) -> DBResult<()> { - self.engine.refresh_indexes() + self.engine + .with_exclusive_lock(|engine| engine.refresh_indexes()) } } diff --git a/log_db/src/lock.rs b/log_db/src/lock.rs new file mode 100644 index 0000000..01961f9 --- /dev/null +++ b/log_db/src/lock.rs @@ -0,0 +1,100 @@ +use super::*; + +pub struct LockManager { + lock_file: fs::File, + excl_lock_file: fs::File, + + state: LockState, +} + +#[derive(Debug, PartialEq, Eq)] +enum LockState { + NotLocked, + Shared, + Exclusive, + ManualExclusive, +} + +impl LockManager { + pub fn new(data_dir_path: PathBuf) -> DBResult<LockManager> { + let lock_file = fs::File::create(data_dir_path.join(LOCK_FILENAME))?; + let excl_lock_file = fs::File::create(data_dir_path.join(EXCL_LOCK_REQ_FILENAME))?; + + Ok(LockManager { + lock_file, + excl_lock_file, + state: LockState::NotLocked, + }) + } + + fn is_exclusive_lock_requested(&self) -> DBResult<bool> { + // Attempt to acquire a shared lock on the lock request file + // If the file is already locked, return false + match self.excl_lock_file.try_lock_shared() { + Err(e) => { + if e.kind() == lock_contended_error().kind() { + return Ok(true); + } + return Err(DBError::IOError(e)); + } + + Ok(_) => { + self.excl_lock_file.unlock()?; + return Ok(false); + } + } + } + + pub fn lock_shared(&mut self) -> DBResult<()> { + if self.state == LockState::Shared { + return Ok(()); + } + + let mut timeout = 5; + loop { + if self.is_exclusive_lock_requested()? { + debug!( + "Exclusive lock requested, waiting for {}ms before requesting a shared lock again", + timeout + ); + thread::sleep(std::time::Duration::from_millis(timeout)); + timeout *= 2; + + if timeout > LOCK_WAIT_MAX_MS { + return Err(DBError::LockRequestError( + "Acquisition of shared lock timed out after {LOCK_WAIT_MAX_MS}".to_owned(), + )); + } + } else { + self.lock_file.lock_shared()?; + self.state = LockState::Shared; + return Ok(()); + } + } + } + + pub fn lock_exclusive(&mut self) -> DBResult<()> { + if self.state == LockState::Exclusive || self.state == LockState::ManualExclusive { + return Ok(()); + } + + // Create a lock on the exclusive lock request file to signal to readers that they should wait + // This will block until the lock is acquired + self.excl_lock_file.lock_exclusive()?; + + // Acquire an exclusive lock on the actual lock files + self.lock_file.lock_exclusive()?; + self.state = LockState::Exclusive; + + // Unlock the request file + self.excl_lock_file.unlock()?; + + Ok(()) + } + + pub fn unlock(&mut self) -> DBResult<()> { + self.lock_file.unlock()?; + self.state = LockState::NotLocked; + Ok(()) + } +} diff --git a/log_db/src/log_reader_forward.rs b/log_db/src/log_reader_forward.rs index fc824b2..a1979b3 100644 --- a/log_db/src/log_reader_forward.rs +++ b/log_db/src/log_reader_forward.rs @@ -1,7 +1,4 @@ -use super::common::*; -use super::record::*; -use std::fs::{self}; -use std::io::{self, Read, Seek}; +use super::*; pub struct ForwardLogReader { metadata_reader: io::BufReader<fs::File>, @@ -96,7 +93,8 @@ impl Iterator for ForwardLogReader { #[cfg(test)] mod tests { use super::*; - use std::path::Path; + + const TEST_RESOURCES_DIR: &str = "tests/resources"; #[test] fn test_forward_log_reader_fixture_db1() { |