diff options
| -rw-r--r-- | log_db/Cargo.toml | 2 | ||||
| -rw-r--r-- | log_db/benches/benchmark.rs | 41 | ||||
| -rw-r--r-- | log_db/src/lib.rs | 47 | ||||
| -rw-r--r-- | log_db/tests/integration.rs | 18 |
4 files changed, 103 insertions, 5 deletions
diff --git a/log_db/Cargo.toml b/log_db/Cargo.toml index ca77753..050abd9 100644 --- a/log_db/Cargo.toml +++ b/log_db/Cargo.toml @@ -10,6 +10,7 @@ crate-type = ["lib"] fs2 = "0.4.3" log = "0.4.22" priority-queue = "2.1.1" +tempfile = "3.13.0" [dev-dependencies] ctor = "0.2.8" @@ -17,7 +18,6 @@ env_logger = "0.11.5" serial_test = "3.1.1" criterion = { version = "0.5", features = ["html_reports"] } rand = "0.8.5" -tempfile = "3.13.0" [[bench]] name = "benchmark" diff --git a/log_db/benches/benchmark.rs b/log_db/benches/benchmark.rs index 5174bdb..8f5b56e 100644 --- a/log_db/benches/benchmark.rs +++ b/log_db/benches/benchmark.rs @@ -44,6 +44,46 @@ pub fn upsert_various_initial_sizes(c: &mut Criterion) { } } +pub fn upsert_various_initial_sizes_compacted(c: &mut Criterion) { + let mut group = c.benchmark_group("upsert_various_initial_sizes_compacted"); + + for size in [100, 1000, 10000, 100_000, 1_000_000, 10_000_000] { + let data_dir_obj = tempfile::tempdir().expect("Failed to get tmpdir"); + let data_dir = &data_dir_obj + .path() + .to_str() + .expect("Failed to convert tmpdir path to str"); + + let sample_record = random_record(0, 1); + let record_length = sample_record.serialize().len() + SEQ_RECORD_SEP.len(); + + let mut db = DB::configure() + .data_dir(&data_dir) + .fields(vec![ + (Field::Id, RecordField::int()), + (Field::Name, RecordField::string()), + (Field::Data, RecordField::bytes()), + ]) + .segment_size(1000 * record_length) + .primary_key(Field::Id) + .initialize() + .expect("Failed to initialize DB"); + + prefill_db(&mut db, size).expect("Failed to prefill DB"); + db.do_maintenance_tasks() + .expect("Failed to do maintenance tasks"); + + group.bench_with_input(BenchmarkId::from_parameter(size), &size, |b, &_size| { + b.iter(|| { + let record = random_record(0, size as i64 + 1); + let _ = db.upsert(black_box(&record)); + db.do_maintenance_tasks() + .expect("Failed to do maintenance tasks"); + }); + }); + } +} + pub fn upsert_write_durability(c: &mut Criterion) { let mut group = c.benchmark_group("upsert_write_durability"); @@ -214,6 +254,7 @@ fn reverse_read_file_with_various_buffer_sizes(c: &mut Criterion) { criterion_group!( benches, upsert_various_initial_sizes, + upsert_various_initial_sizes_compacted, upsert_write_durability, get_from_disk_various_initial_sizes, get_various_memtable_capacities, diff --git a/log_db/src/lib.rs b/log_db/src/lib.rs index 2e77e19..d704c7d 100644 --- a/log_db/src/lib.rs +++ b/log_db/src/lib.rs @@ -14,12 +14,14 @@ use fs2::FileExt; use primary_memtable::PrimaryMemtable; pub use reverse_log_reader::ReverseLogReader; use secondary_memtable::SecondaryMemtable; +use std::collections::BTreeMap; 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; +use tempfile; pub struct ConfigBuilder<Field: Eq + Clone + Debug> { data_dir: Option<String>, @@ -731,11 +733,14 @@ impl<Field: Eq + Clone + Debug> DB<Field> { .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 + // Compact the rotated segment without a lock. + // Since the rotated segment and the compacted segment based on it will be + // a) read-only, and b) identical in effective content, there is no need to lock it. + self.compact_segment(&next_segment_path)?; + + debug!("Segment compacted"); } Ok(()) @@ -798,4 +803,40 @@ impl<Field: Eq + Clone + Debug> DB<Field> { nums.reverse(); Ok(nums) } + + fn compact_segment(&self, path: &Path) -> Result<(), io::Error> { + debug!("Opening segment file {:?} for compaction", path); + let mut segment_file = fs::OpenOptions::new().read(true).open(path)?; + + debug!("Reading segment data into a BTreeMap"); + let mut map = BTreeMap::new(); + let forward_log_reader = ForwardLogReader::new(&mut segment_file); + for entry in forward_log_reader { + let primary_key = entry.values[self.primary_key_index] + .as_indexable() + .expect("Primary key was not indexable"); + map.insert(primary_key, entry); + } + + debug!("Opening temporary file for writing compacted data"); + let temp_file = tempfile::NamedTempFile::new()?; + let temp_path = temp_file.as_ref(); + + let mut temp_file = fs::OpenOptions::new() + .create(true) + .append(true) + .open(temp_path)?; + + debug!("Writing compacted data to temporary file"); + for entry in map.values() { + let mut serialized = entry.serialize(); + serialized.extend(SEQ_RECORD_SEP); + temp_file.write_all(&serialized)?; + } + + debug!("Moving temporary file to replace segment file"); + fs::rename(&temp_path, path)?; + + Ok(()) + } } diff --git a/log_db/tests/integration.rs b/log_db/tests/integration.rs index 76ed853..3e712e1 100644 --- a/log_db/tests/integration.rs +++ b/log_db/tests/integration.rs @@ -532,12 +532,13 @@ fn test_log_is_rotated_when_capacity_reached() { .with_extension("2") .exists()); + // 3rd segment should not exist (note negation) assert!(!Path::new(&data_dir) .join(ACTIVE_LOG_FILENAME) .with_extension("3") .exists()); - // Check that the active file only contains two rows + // Check that the active file only contains five rows let mut file = OpenOptions::new() .read(true) .open(Path::new(&data_dir).join(ACTIVE_LOG_FILENAME)) @@ -545,6 +546,21 @@ fn test_log_is_rotated_when_capacity_reached() { let records_in_active_log = ForwardLogReader::new(&mut file).count(); assert_eq!(records_in_active_log, 5); + // Check that each rotated file contains only 1 record + // because of compaction + for i in &[1, 2] { + let mut file = OpenOptions::new() + .read(true) + .open( + Path::new(&data_dir) + .join(ACTIVE_LOG_FILENAME) + .with_extension(i.to_string()), + ) + .expect("File could not be opened"); + let records_in_rotated_log = ForwardLogReader::new(&mut file).count(); + assert_eq!(records_in_rotated_log, 1); + } + // 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()); |