diff options
| -rw-r--r-- | src/lib.rs | 161 | ||||
| -rw-r--r-- | tests/integration.rs | 65 |
2 files changed, 208 insertions, 18 deletions
@@ -17,6 +17,7 @@ use secondary_memtable::SecondaryMemtable; use std::fmt::Debug; use std::fs::{self}; use std::io::{self, Write}; +use std::os::unix::fs::MetadataExt; use std::path::{Path, PathBuf}; use std::thread; @@ -396,7 +397,7 @@ impl<Field: Eq + Clone + Debug> DB<Field> { // Open the file and acquire a shared lock for reading let mut file = fs::OpenOptions::new().read(true).open(&self.log_path)?; - self.request_shared_lock(&self.config.data_dir, &mut file)?; + self.request_shared_lock(&mut file)?; if !is_file_same_as_path(&file, &self.log_path)? { // The log file has been rotated, so we must try again @@ -406,20 +407,49 @@ impl<Field: Eq + Clone + Debug> DB<Field> { return self.get(query_key_original); } - debug!("Lock acquired, searching log file for record"); + debug!("Lock acquired, searching log files for record"); - let mut reverse_log_reader = ReverseLogReader::new(&mut file)?; - let result = reverse_log_reader.find(|record| { - let record_key = record.values[self.primary_key_index] - .as_indexable() - .expect("A non-indexable value was stored at key index"); - record_key == query_key - }); + let segment_numbers = self.segment_numbers()?; + let mut result: Option<Record> = None; + for &n in &segment_numbers { + if n == 0 { + debug!("Searching the active log file..."); + result = ReverseLogReader::new(&mut file)?.find(|record| { + let record_key = record.values[self.primary_key_index] + .as_indexable() + .expect("A non-indexable value was stored at key index"); + record_key == query_key + }); - file.unlock()?; - debug!("Record search complete, lock released"); + debug!("Active log file searched, releasing shared lock..."); + file.unlock()?; + } else { + debug!("Locking and searching rotated log segment file {}...", n); + let path = Path::new(&self.config.data_dir) + .join(ACTIVE_LOG_FILENAME) + .with_extension(n.to_string()); + + let mut segm_file = fs::OpenOptions::new().read(true).open(&path)?; + self.request_shared_lock(&mut segm_file)?; + result = ReverseLogReader::new(&mut segm_file)?.find(|record| { + let record_key = record.values[self.primary_key_index] + .as_indexable() + .expect("A non-indexable value was stored at key index"); + record_key == query_key + }); + + debug!("Segment file searched, releasing shared lock..."); + segm_file.unlock()?; + }; - let result_value = match result { + if result.is_some() { + break; + } + } + + debug!("Record search complete"); + + let result_value = match &result { Some(record) => record, None => { debug!("No record found for key {:?}", query_key); @@ -435,7 +465,7 @@ impl<Field: Eq + Clone + Debug> DB<Field> { debug!("Updating secondary memtables"); self.update_secondary_indexes(&result_value); - Ok(Some(result_value)) + Ok(result) } /// Get a collection of records based on a field value. @@ -646,11 +676,11 @@ impl<Field: Eq + Clone + Debug> DB<Field> { } } - fn request_shared_lock(&self, data_dir: &str, file: &mut fs::File) -> Result<(), io::Error> { + fn request_shared_lock(&self, file: &mut fs::File) -> Result<(), io::Error> { const SHARED_LOCK_WAIT_MAX_MS: u64 = 100; let mut timeout = 5; loop { - if self.is_exclusive_lock_requested(data_dir)? { + if self.is_exclusive_lock_requested(&self.config.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 = std::cmp::min(timeout * 2, SHARED_LOCK_WAIT_MAX_MS); @@ -660,4 +690,105 @@ impl<Field: Eq + Clone + Debug> DB<Field> { } } } + + /// Check if there are any pending tasks and do them. Tasks include: + /// - Rotating the active log file if it has reached capacity and compacting it. + /// + /// This function should be called periodically to ensure that the database remains in an optimal state. + /// Note that this function is synchronous and may block for a relatively long time. + /// 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 will be blocked during the tasks. + pub fn do_maintenance_tasks(&mut self) -> Result<(), io::Error> { + let active_log_path = Path::new(&self.config.data_dir).join(ACTIVE_LOG_FILENAME); + let active_log_md = fs::metadata(&active_log_path)?; + + if active_log_md.size() >= self.config.segment_size as u64 { + // Rotate the active log file + + debug!("Starting rotation, requesting exclusive lock..."); + self.request_exclusive_lock()?; + + debug!("Exclusive lock acquired, rotating active log file..."); + let next_segment_number = self.next_segment_number()?; + let next_segment_path = + &active_log_path.with_extension(next_segment_number.to_string()); + + debug!("Renaming active log file to {:?}", &next_segment_path); + fs::rename(&active_log_path, &next_segment_path)?; + + // Create a new active log file + self.log_file = fs::OpenOptions::new() + .create(true) + .write(true) + .append(true) + .open(&active_log_path)?; + + // The new active log file is not locked by this client so it cannot be touched. + // Compact the rotated segment. + + debug!("Active log file rotated"); + + // TODO compaction of rotated segments + } + + Ok(()) + } + + fn next_segment_number(&self) -> Result<u64, io::Error> { + match self.segment_numbers()?.iter().max() { + Some(greatest) => Ok(greatest + 1), + None => Ok(1), + } + } + + /// Query the filesystem to get the numbers of existing segments + /// in the intended reading order: first the active log (signaled with 0), + /// then the segments from the greatest ordinal (newest) to the least (oldest). + /// E.g. `vec![0, 4, 3, 2, 1]`. + fn segment_numbers(&self) -> Result<Vec<u64>, io::Error> { + // TODO: optimize the vecs out of here + let files = fs::read_dir(&self.config.data_dir)?; + let mut nums: Vec<u64> = files + .filter_map(|f| { + let f_path = match f { + Ok(f) => f.path(), + Err(_) => return None, + }; + + if !f_path.is_file() { + return None; + } + + let name = &f_path + .with_extension("") + .file_name() + .expect("File did not have a name?") + .to_str() + .expect("Failed to convert file name to string") + .to_string(); + + if name != ACTIVE_LOG_FILENAME { + return None; + } + + let ext = match f_path.extension() { + Some(ext) => ext, + None => return None, + }; + + let ext_num = ext + .to_str() + .expect("Extension was not a valid UTF-8 string") + .parse::<u64>() + .expect("Extension was not a valid number"); + + Some(ext_num) + }) + .collect(); + + nums.sort(); + nums.push(0); + nums.reverse(); + Ok(nums) + } } diff --git a/tests/integration.rs b/tests/integration.rs index cb3af56..e6be74f 100644 --- a/tests/integration.rs +++ b/tests/integration.rs @@ -4,9 +4,11 @@ extern crate tempfile; use ctor::ctor; use env_logger; use log::debug; -use log_db::{self, RecordField}; +use log_db::{ + self, ForwardLogReader, RecordField, ACTIVE_LOG_FILENAME, SEQ_LIT_ESCAPE, SEQ_RECORD_SEP, +}; use log_db::{Record, RecordValue, DB, TEST_RESOURCES_DIR}; -use std::fs; +use std::fs::{self, OpenOptions}; use std::path::Path; use std::thread; use std::time::Duration; @@ -460,7 +462,6 @@ fn test_one_writer_and_multiple_reading_threads() { #[test] fn test_literal_escape_is_escaped() { let data_dir = tmp_dir(); - debug!("data_dir: {:?}", data_dir); let mut db = DB::configure() .data_dir(&data_dir) @@ -494,3 +495,61 @@ fn test_literal_escape_is_escaped() { assert_eq!(received, &vec![0x1A, 0x1B, 0x1C, 0x1D]); } + +#[test] +fn test_log_is_rotated_when_capacity_reached() { + let data_dir = tmp_dir(); + + let record = Record { + values: vec![RecordValue::Int(1), RecordValue::Bytes(vec![1, 2, 3, 4])], + }; + let record_len = &record.serialize().len() + SEQ_RECORD_SEP.len(); + + let mut db = DB::configure() + .data_dir(&data_dir) + .memtable_capacity(0) // disable memtables + .segment_size(10 * record_len) // small log segment size + .fields(&vec![ + (Field::Id, RecordField::int()), + (Field::Data, RecordField::bytes()), + ]) + .primary_key(Field::Id) + .initialize() + .expect("Failed to initialize DB instance"); + + // Insert more records than fits the capacity + for _ in 0..25 { + db.upsert(&record).expect("Failed to upsert record"); + + db.do_maintenance_tasks() + .expect("Failed to do maintenance tasks"); + } + + // Check that the rotated segments exist + assert!(Path::new(&data_dir) + .join(ACTIVE_LOG_FILENAME) + .with_extension("1") + .exists()); + + assert!(Path::new(&data_dir) + .join(ACTIVE_LOG_FILENAME) + .with_extension("2") + .exists()); + + assert!(!Path::new(&data_dir) + .join(ACTIVE_LOG_FILENAME) + .with_extension("3") + .exists()); + + // Check that the active file only contains two rows + let mut file = OpenOptions::new() + .read(true) + .open(Path::new(&data_dir).join(ACTIVE_LOG_FILENAME)) + .expect("File could not be opened"); + let records_in_active_log = ForwardLogReader::new(&mut file).count(); + assert_eq!(records_in_active_log, 5); + + // Look for nonexistant record to scan all segment files + let found = db.get(&RecordValue::Int(2)).expect("Failed to get record"); + assert!(found.is_none()); +} |