diff options
| -rw-r--r-- | Cargo.toml | 2 | ||||
| -rw-r--r-- | benches/benchmark.rs | 4 | ||||
| -rw-r--r-- | src/common.rs | 11 | ||||
| -rw-r--r-- | src/lib.rs | 2 | ||||
| -rw-r--r-- | src/reverse_log_reader.rs | 35 | ||||
| -rw-r--r-- | tests/integration.rs | 33 | ||||
| -rw-r--r-- | tests/resources/test_db2 | bin | 39 -> 46243 bytes | |||
| -rw-r--r-- | tests/resources/test_db3 | bin | 0 -> 46243 bytes |
8 files changed, 66 insertions, 21 deletions
@@ -15,9 +15,9 @@ priority-queue = "2.1.1" ctor = "0.2.8" env_logger = "0.11.5" serial_test = "3.1.1" -tempfile = "3.13.0" criterion = { version = "0.5", features = ["html_reports"] } rand = "0.8.5" +tempfile = "3.13.0" [[bench]] name = "benchmark" diff --git a/benches/benchmark.rs b/benches/benchmark.rs index 0328ac5..b9ecbb3 100644 --- a/benches/benchmark.rs +++ b/benches/benchmark.rs @@ -34,8 +34,8 @@ pub fn upsert_benchmark(c: &mut Criterion) { prefill_db_with_n_records(&mut db, size).expect("Failed to prefill DB"); group.bench_with_input(BenchmarkId::from_parameter(size), &size, |b, &_size| { - let record = random_record(); b.iter(|| { + let record = random_record(); let _ = db.upsert(black_box(&record)); }); }); @@ -64,8 +64,8 @@ pub fn get_benchmark(c: &mut Criterion) { prefill_db_with_n_records(&mut db, size).expect("Failed to prefill DB"); group.bench_with_input(BenchmarkId::from_parameter(size), &size, |b, &_size| { - let id = random_int(); b.iter(|| { + let id = random_int(); let _ = db.get(black_box(&RecordValue::Int(id))); }); }); diff --git a/src/common.rs b/src/common.rs index 33e99e7..3d31263 100644 --- a/src/common.rs +++ b/src/common.rs @@ -72,15 +72,16 @@ pub enum MemtableEvictPolicy { #[derive(Debug, Clone, Eq, PartialEq)] pub enum WriteDurability { /// Changes are written to an application-level write buffer without flushing to the OS write buffer or syncing to disk. + /// The buffered writer will batch writes to the OS buffer for maximum performance. /// Offers the lowest durability guarantees but is very fast. - AsyncWrite, - /// Changes are written to the OS write buffer but not synced to disk. - /// Offers better durability guarantees than AsyncWrite but is slower. + Async, + /// Changes are written to the OS write buffer but not immediately synced to disk. + /// Offers better durability guarantees than Async but is slower. /// This is generally recommended. Most OSes will sync the write buffer to disk within a few seconds. Flush, - /// Changes are written to the OS write buffer and synced to disk. + /// Changes are written to the OS write buffer and synced to disk immediately. /// Offers the best durability guarantees but is the slowest. - SyncWrite, + FlushSync, } #[derive(Debug, Clone, Ord, PartialOrd, Eq, PartialEq, Hash)] @@ -322,7 +322,7 @@ impl<Field: Eq + Clone + Debug> DB<Field> { if self.config.write_durability == WriteDurability::Flush { self.log_file.flush()?; } - if self.config.write_durability == WriteDurability::SyncWrite { + if self.config.write_durability == WriteDurability::FlushSync { self.log_file.flush()?; self.log_file.sync_all()?; } diff --git a/src/reverse_log_reader.rs b/src/reverse_log_reader.rs index d8021ba..8a2126f 100644 --- a/src/reverse_log_reader.rs +++ b/src/reverse_log_reader.rs @@ -93,6 +93,7 @@ impl<'a> ReverseLogReader<'a> { /// Read exactly `buf.len()` bytes from the file, return an error if the file is exhausted. /// The bytes are returned in start -> end order. + /// If an error is returned, the contents of `buf` are in an undefined state. fn read_exact(&mut self, buf: &mut [u8]) -> Result<usize, io::Error> { let mut read = 0; while read < buf.len() { @@ -105,13 +106,15 @@ impl<'a> ReverseLogReader<'a> { )); } } - let bytes_to_read = std::cmp::min(buf.len() - read, self.internal_pos); - buf[read..read + bytes_to_read].copy_from_slice( - &self.internal_buf[self.internal_pos - bytes_to_read..self.internal_pos], - ); - self.internal_pos -= bytes_to_read; - read += bytes_to_read; + + let end = buf.len() - read; + let n = std::cmp::min(self.internal_pos, end); + buf[end - n..end] + .copy_from_slice(&self.internal_buf[self.internal_pos - n..self.internal_pos]); + read += n; + self.internal_pos -= n; } + Ok(read) } @@ -169,10 +172,17 @@ impl<'a> ReverseLogReader<'a> { match validate_special(&special_buf.as_slice()) { Some(special) => Ok(special), - None => Err(io::Error::new( - io::ErrorKind::InvalidData, - format!("Not a special sequence: {:?}", special_buf), - )), + None => { + let pos = self.file.stream_position().unwrap() + self.internal_pos as u64; + + Err(io::Error::new( + io::ErrorKind::InvalidData, + format!( + "Not a special sequence: {:?} at pos: {:x}", + special_buf, pos, + ), + )) + } } } } @@ -320,7 +330,6 @@ mod reverse_reader_tests { let mut reader = ReverseLogReader::new(&mut file).unwrap(); let mut buf = vec![0; 10]; assert!(reader.read_exact(&mut buf).unwrap_err().kind() == io::ErrorKind::UnexpectedEof); - assert_eq!(String::from_utf8(buf[..5].to_vec()).unwrap(), "hello"); } } @@ -331,7 +340,9 @@ impl Iterator for ReverseLogReader<'_> { match self.read_record() { Ok(Some(record)) => Some(record), Ok(None) => None, - Err(err) => panic!("Error reading record: {:?}", err), + Err(err) => { + panic!("Error reading record: {:?}", err,) + } } } } diff --git a/tests/integration.rs b/tests/integration.rs index 2bebc6d..fc1d0cc 100644 --- a/tests/integration.rs +++ b/tests/integration.rs @@ -299,6 +299,39 @@ fn test_initialize_and_read_from_primary_memtable_fixture_db2() { } #[test] +fn test_initialize_without_memtables_fixture_db3() { + let data_dir = tmp_dir(); + // Copy the fixture DB to the test data directory + fs::create_dir_all(&data_dir).expect("Failed to create the test data directory"); + fs::copy( + &Path::new(TEST_RESOURCES_DIR).join("test_db3"), + &Path::new(&data_dir).join("db"), + ) + .expect("Failed to copy the fixture DB"); + + let mut db = DB::configure() + .data_dir(&data_dir) + .fields(&vec![ + (Field::Id, RecordFieldType::Int), + (Field::Name, RecordFieldType::String), + (Field::Data, RecordFieldType::Bytes), + ]) + .memtable_capacity(0) + .primary_key(Field::Id) + .initialize() + .expect("Failed to initialize DB instance"); + + let result = db.get(&RecordValue::Int(1)).unwrap().unwrap(); + + // Check that the IDs match + let expected = RecordValue::Int(1); + assert!(match (&result.values[0], &expected) { + (RecordValue::Int(a), RecordValue::Int(b)) => a == b, + _ => false, + }); +} + +#[test] fn test_multiple_writing_threads() { let data_dir = tmp_dir(); let mut threads = vec![]; diff --git a/tests/resources/test_db2 b/tests/resources/test_db2 Binary files differindex 2448640..156659f 100644 --- a/tests/resources/test_db2 +++ b/tests/resources/test_db2 diff --git a/tests/resources/test_db3 b/tests/resources/test_db3 Binary files differnew file mode 100644 index 0000000..156659f --- /dev/null +++ b/tests/resources/test_db3 |