diff options
| -rw-r--r-- | Cargo.lock | 10 | ||||
| -rw-r--r-- | Cargo.toml | 1 | ||||
| -rw-r--r-- | src/common.rs | 1 | ||||
| -rw-r--r-- | src/lib.rs | 1 | ||||
| -rw-r--r-- | src/log_reader.rs | 352 | ||||
| -rw-r--r-- | tests/integration.rs | 107 |
6 files changed, 354 insertions, 118 deletions
@@ -291,7 +291,6 @@ dependencies = [ "fs2", "log", "priority-queue", - "rev_buf_reader", "serial_test", "tempfile", ] @@ -411,15 +410,6 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2b15c43186be67a4fd63bee50d0303afffcef381492ebe2c5d87f324e1b8815c" [[package]] -name = "rev_buf_reader" -version = "0.3.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8c0f2e47e00e29920959826e2e1784728a3780d1a784247be5257258cc75f910" -dependencies = [ - "memchr", -] - -[[package]] name = "rustix" version = "0.38.37" source = "registry+https://github.com/rust-lang/crates.io-index" @@ -10,7 +10,6 @@ crate-type = ["lib"] fs2 = "0.4.3" log = "0.4.22" priority-queue = "2.1.1" -rev_buf_reader = "0.3.0" [dev-dependencies] ctor = "0.2.8" diff --git a/src/common.rs b/src/common.rs index c4fce39..9ea43aa 100644 --- a/src/common.rs +++ b/src/common.rs @@ -15,6 +15,7 @@ pub const EXCL_LOCK_REQUEST_FILENAME: &str = "excl_lock_req"; pub const DEFAULT_READ_BUF_SIZE: usize = 1024 * 1024; // 1 MB pub const FIELD_SEPARATOR: u8 = b'\x1C'; pub const ESCAPE_CHARACTER: u8 = b'\x1D'; +pub const TEST_RESOURCES_DIR: &str = "tests/resources"; // Special sequences. Note: these must have the same length! // Since the log is read both forwards and backwards, we must have a signal @@ -1,6 +1,5 @@ #[macro_use] extern crate log; -extern crate rev_buf_reader; mod common; mod log_reader; diff --git a/src/log_reader.rs b/src/log_reader.rs index d2b364b..d07326a 100644 --- a/src/log_reader.rs +++ b/src/log_reader.rs @@ -1,7 +1,7 @@ use super::common::*; -use rev_buf_reader::RevBufReader; use std::fs::{self}; use std::io::{self, BufRead, Read, Seek, SeekFrom}; +use std::path::Path; /// There are three special sequences that need to be handled: /// Here: SC = escape char, FS = field separator. @@ -18,83 +18,355 @@ fn validate_special(buf: &[u8]) -> Option<SpecialSequence> { } pub struct ReverseLogReader<'a> { - rev_reader: RevBufReader<&'a mut fs::File>, + file: &'a mut fs::File, + /// The internal buffer used to read from the file. + /// It is populated with the last INTERNAL_BUF_SIZE bytes read from the file + /// and is used to read records in reverse order + internal_buf: Vec<u8>, + /// The current position in the internal buffer. It is decremented as bytes are read + /// from the buffer. When a read is requested and the internal position is 0, the buffer + /// is populated with the next (= closer to the start of the file) INTERNAL_BUF_SIZE bytes from the file. + /// Note: This is the index of the next byte to be read from the internal buffer + 1 + internal_pos: usize, + consumed_record_sep: bool, } +const INTERNAL_BUF_SIZE: usize = 4096; impl<'a> ReverseLogReader<'a> { pub fn new(file: &mut fs::File) -> Result<ReverseLogReader, io::Error> { - let rev_reader = RevBufReader::new(file); - Ok(ReverseLogReader { rev_reader }) + file.seek(SeekFrom::End(0))?; + Ok(ReverseLogReader { + file, + internal_buf: vec![0; INTERNAL_BUF_SIZE], + internal_pos: 0, + consumed_record_sep: false, + }) } - fn read_record(&mut self) -> Result<Option<Record>, io::Error> { - if self.rev_reader.stream_position()? == 0 { + pub fn new_with_size( + file: &mut fs::File, + internal_buf_size: usize, + ) -> Result<ReverseLogReader, io::Error> { + file.seek(SeekFrom::End(0))?; + Ok(ReverseLogReader { + file, + internal_buf: vec![0; internal_buf_size], + internal_pos: 0, + consumed_record_sep: false, + }) + } + + pub fn read_record(&mut self) -> Result<Option<Record>, io::Error> { + if self.file.stream_position()? == 0 && self.internal_pos == 0 { return Ok(None); } - // Check that the record starts with the record separator - if self.read_special_sequence()? != SpecialSequence::RecordSeparator { - return Err(io::Error::new( - io::ErrorKind::InvalidData, - "Record candidate does not end with record separator", - )); + if !self.consumed_record_sep { + // Check that record ends with a record separator + match self.read_special_sequence()? { + SpecialSequence::RecordSeparator => {} + _ => { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "Record does not end in record separator", + )); + } + } } + self.consumed_record_sep = false; - // The buffer that stores all the bytes of the record read so far in reverse order. - let mut result_buf: Vec<u8> = Vec::new(); - // The buffer that stores the bytes read from the file. - let mut read_buf: Vec<u8> = Vec::new(); - + let mut result_buf: Vec<u8> = vec![]; loop { - read_buf.clear(); - self.rev_reader - .read_until(ESCAPE_CHARACTER, &mut read_buf)?; + let mut read_buf = vec![]; + let read = self.read_until(ESCAPE_CHARACTER, &mut read_buf)?; - result_buf.extend(read_buf.iter().rev()); + result_buf.extend(&read_buf); - if self.rev_reader.stream_position()? == 0 { - // If we've reached the beginning of the file, we've read the entire record. + if self.file.stream_position()? == 0 && self.internal_pos == 0 { + // We read until the start of the file, we are done break; } - // Otherwise, we must have encountered an escape character. match self.read_special_sequence()? { - SpecialSequence::RecordSeparator => { - // The record is complete, so we can break out of the loop. - // Move the cursor back to the beginning of the special sequence. - self.rev_reader.seek_relative(SEQ_RECORD_SEP.len() as i64)?; - break; + SpecialSequence::LiteralEscape => { + result_buf.push(ESCAPE_CHARACTER); } SpecialSequence::LiteralFieldSeparator => { - // The field separator is escaped, so we need to add it to the result buffer. result_buf.push(FIELD_SEPARATOR); } - SpecialSequence::LiteralEscape => { - // The escape character is escaped, so we need to add it to the result buffer. - result_buf.push(ESCAPE_CHARACTER); + SpecialSequence::RecordSeparator => { + self.consumed_record_sep = true; + break; } } } result_buf.reverse(); - let record = Record::deserialize(&result_buf); - Ok(Some(record)) + Ok(Some(Record::deserialize(&result_buf))) + } + + /// Read exactly `buf.len()` bytes from the file, return an error if the file is exhausted. + /// The bytes are returned in start -> end order. + fn read_exact(&mut self, buf: &mut [u8]) -> Result<usize, io::Error> { + let mut read = 0; + while read < buf.len() { + if self.internal_pos == 0 { + let populated_n = self.populate_internal_buf()?; + if populated_n == 0 { + return Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + "Unexpected end of file", + )); + } + } + 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; + } + Ok(read) + } + + fn populate_internal_buf(&mut self) -> Result<usize, io::Error> { + let current_seek_pos = self.file.stream_position()? as usize; + + // Seek back by the size of the internal buffer or to the beginning of the file + let seek_length = if current_seek_pos > self.internal_buf.len() { + self.internal_pos = self.internal_buf.len(); + self.internal_buf.len() + } else { + self.internal_buf = vec![0; current_seek_pos as usize]; + self.internal_pos = current_seek_pos as usize; + current_seek_pos + }; + + self.file.seek_relative(-(seek_length as i64))?; + self.file.read_exact(&mut self.internal_buf)?; + self.file.seek_relative(-(seek_length as i64))?; + + Ok(seek_length as usize) + } + + /// Iterate over the internal buffer with internal_pos as the index. + /// If the byte is found or the file has been exhausted, we return the number of bytes read. + /// The `buf` parameter is used to store the bytes read from the internal buffer, excluding the found byte, + /// in reverse order. + fn read_until(&mut self, byte: u8, buf: &mut Vec<u8>) -> Result<usize, io::Error> { + // TODO: optimize this + let mut read = 0; + loop { + // If we reach internal_pos == 0, we need to populate the internal buffer. + if self.internal_pos == 0 { + let populated_n = self.populate_internal_buf()?; + if populated_n == 0 { + return Ok(read); + } + } + + while self.internal_pos > 0 { + let index = self.internal_pos - 1; + if self.internal_buf[index] == byte { + return Ok(read); + } + buf.push(self.internal_buf[index]); + read += 1; + self.internal_pos -= 1; + } + } } fn read_special_sequence(&mut self) -> Result<SpecialSequence, io::Error> { let mut special_buf: Vec<u8> = vec![0; SEQ_RECORD_SEP.len()]; - self.rev_reader.read_exact(&mut special_buf)?; + self.read_exact(&mut special_buf)?; match validate_special(&special_buf.as_slice()) { Some(special) => Ok(special), None => Err(io::Error::new( io::ErrorKind::InvalidData, - "Not a special sequence", + format!("Not a special sequence: {:?}", special_buf), )), } } } +#[cfg(test)] +mod reverse_reader_tests { + use super::*; + use std::io::Write; + + #[test] + fn test_read_until_found() { + let mut file = tempfile::tempfile().unwrap(); + file.write_all(b"hello,world").unwrap(); + let mut reader = ReverseLogReader::new(&mut file).unwrap(); + let mut buf = vec![]; + assert_eq!(reader.read_until(b',', &mut buf).unwrap(), 5); + assert_eq!(buf, b"dlrow"); + } + + #[test] + fn test_read_until_not_found() { + let mut file = tempfile::tempfile().unwrap(); + file.write_all(b"hello,world").unwrap(); + let mut reader = ReverseLogReader::new(&mut file).unwrap(); + let mut buf = vec![]; + let read = reader.read_until(b'!', &mut buf).unwrap(); + assert_eq!(buf, b"dlrow,olleh"); + assert_eq!(read, 11); + } + + #[test] + fn test_read_special_sequence() { + let mut file = tempfile::tempfile().unwrap(); + let mut buf = vec![]; + buf.extend(SEQ_RECORD_SEP); + buf.extend(SEQ_LIT_ESCAPE); + buf.extend(SEQ_LIT_FIELD_SEP); + // Note: written in start -> end order, read end -> start + file.write_all(&buf).unwrap(); + + let mut reader = ReverseLogReader::new(&mut file).unwrap(); + assert_eq!( + reader.read_special_sequence().unwrap(), + SpecialSequence::LiteralFieldSeparator + ); + assert_eq!( + reader.read_special_sequence().unwrap(), + SpecialSequence::LiteralEscape + ); + assert_eq!( + reader.read_special_sequence().unwrap(), + SpecialSequence::RecordSeparator + ); + } + + #[test] + fn test_populate_internal_buf() { + let mut file = tempfile::tempfile().unwrap(); + file.write_all(b"hello,world").unwrap(); + let mut reader = ReverseLogReader::new_with_size(&mut file, 3).unwrap(); + + reader.populate_internal_buf().unwrap(); + assert_eq!( + String::from_utf8(reader.internal_buf.clone()).unwrap(), + "rld".to_string() + ); + + reader.populate_internal_buf().unwrap(); + assert_eq!( + String::from_utf8(reader.internal_buf.clone()).unwrap(), + ",wo".to_string() + ); + + reader.populate_internal_buf().unwrap(); + assert_eq!( + String::from_utf8(reader.internal_buf.clone()).unwrap(), + "llo".to_string() + ); + + reader.populate_internal_buf().unwrap(); + assert_eq!( + String::from_utf8(reader.internal_buf.clone()).unwrap(), + "he".to_string() + ); + } + + #[test] + fn test_reverse_log_reader_fixture_db1() { + let db_path = Path::new(TEST_RESOURCES_DIR).join("test_db1"); + let mut file = fs::OpenOptions::new() + .read(true) + .open(&db_path) + .expect("Failed to open file"); + let mut reverse_log_reader = ReverseLogReader::new(&mut file).unwrap(); + + // There are two records in the log with "schema": Int, Null + + let last_record = reverse_log_reader + .next() + .expect("Failed to read the last record"); + assert!(match last_record.values.as_slice() { + [RecordValue::Int(10), RecordValue::Null] => true, + _ => false, + }); + + let first_record = reverse_log_reader + .next() + .expect("Failed to read the first record"); + assert!(match first_record.values.as_slice() { + // Note: the int value is equal to the escape byte + [RecordValue::Int(0x1D), RecordValue::Null] => true, + _ => false, + }); + + assert!(reverse_log_reader.next().is_none()); + } + + #[test] + fn test_forward_log_reader_fixture_db1() { + let db_path = Path::new(TEST_RESOURCES_DIR).join("test_db1"); + let mut file = fs::OpenOptions::new() + .read(true) + .open(&db_path) + .expect("Failed to open file"); + let mut forward_log_reader = ForwardLogReader::new(&mut file); + + // There are two records in the log with "schema": Int, Null + + let first_record = forward_log_reader + .next() + .expect("Failed to read the first record"); + assert!(match first_record.values.as_slice() { + [RecordValue::Int(0x1D), RecordValue::Null] => true, + _ => false, + }); + + let last_record = forward_log_reader + .next() + .expect("Failed to read the last record"); + assert!(match last_record.values.as_slice() { + [RecordValue::Int(10), RecordValue::Null] => true, + _ => false, + }); + + assert!(forward_log_reader.next().is_none()); + } + + #[test] + fn test_read_exact() { + let mut file = tempfile::tempfile().unwrap(); + file.write_all(b"hello,world").unwrap(); + let mut reader = ReverseLogReader::new(&mut file).unwrap(); + let mut buf = vec![0; 3]; + + let read = reader.read_exact(&mut buf).unwrap(); + assert_eq!(buf, b"rld"); + assert_eq!(read, 3); + + let read = reader.read_exact(&mut buf).unwrap(); + assert_eq!(buf, b",wo"); + assert_eq!(read, 3); + + let read = reader.read_exact(&mut buf).unwrap(); + assert_eq!(buf, b"llo"); + assert_eq!(read, 3); + + assert!(reader.read_exact(&mut buf).unwrap_err().kind() == io::ErrorKind::UnexpectedEof); + } + + #[test] + fn test_read_exact_insufficient_bytes() { + let mut file = tempfile::tempfile().unwrap(); + file.write_all(b"hello").unwrap(); + 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"); + } +} + impl Iterator for ReverseLogReader<'_> { type Item = Record; @@ -125,11 +397,11 @@ impl<'a> ForwardLogReader<'a> { // Try reading a byte from the file. // If we've reached the end of the file, return None. - let mut peek_buf = vec![0, 1]; + let mut peek_buf = vec![0]; match self.reader.read_exact(&mut peek_buf) { Ok(_) => { - // Go back one byte (not sure why you need to seek by -2 here?) - self.reader.seek_relative(-2)?; + // Go back one byte + self.reader.seek_relative(-1)?; } Err(ref e) if e.kind() == io::ErrorKind::UnexpectedEof => { return Ok(None); diff --git a/tests/integration.rs b/tests/integration.rs index fdca59f..2bebc6d 100644 --- a/tests/integration.rs +++ b/tests/integration.rs @@ -3,17 +3,15 @@ extern crate tempfile; use ctor::ctor; use env_logger; +use log::debug; use log_db; -use log_db::{ForwardLogReader, Record, RecordFieldType, RecordValue, ReverseLogReader, DB}; -use serial_test::serial; +use log_db::{Record, RecordFieldType, RecordValue, DB, TEST_RESOURCES_DIR}; use std::fs; use std::path::Path; use std::thread; use std::time::Duration; use tempfile::tempdir; -const TEST_RESOURCES_DIR: &str = "tests/resources"; - #[ctor] fn init_logger() { let _ = env_logger::builder().is_test(true).try_init(); @@ -212,67 +210,6 @@ fn test_upsert_fails_on_invalid_value_type() { } #[test] -fn test_reverse_log_reader_fixture_db1() { - let db_path = Path::new(TEST_RESOURCES_DIR).join("test_db1"); - let mut file = fs::OpenOptions::new() - .read(true) - .open(&db_path) - .expect("Failed to open file"); - let mut reverse_log_reader = ReverseLogReader::new(&mut file).unwrap(); - - // There are two records in the log with "schema": Int, Null - - let last_record = reverse_log_reader - .next() - .expect("Failed to read the last record"); - assert!(match last_record.values.as_slice() { - [RecordValue::Int(10), RecordValue::Null] => true, - _ => false, - }); - - let first_record = reverse_log_reader - .next() - .expect("Failed to read the first record"); - assert!(match first_record.values.as_slice() { - // Note: the int value is equal to the escape byte - [RecordValue::Int(0x1D), RecordValue::Null] => true, - _ => false, - }); - - assert!(reverse_log_reader.next().is_none()); -} - -#[test] -fn test_forward_log_reader_fixture_db1() { - let db_path = Path::new(TEST_RESOURCES_DIR).join("test_db1"); - let mut file = fs::OpenOptions::new() - .read(true) - .open(&db_path) - .expect("Failed to open file"); - let mut forward_log_reader = ForwardLogReader::new(&mut file); - - // There are two records in the log with "schema": Int, Null - - let first_record = forward_log_reader - .next() - .expect("Failed to read the first record"); - assert!(match first_record.values.as_slice() { - [RecordValue::Int(0x1D), RecordValue::Null] => true, - _ => false, - }); - - let last_record = forward_log_reader - .next() - .expect("Failed to read the last record"); - assert!(match last_record.values.as_slice() { - [RecordValue::Int(10), RecordValue::Null] => true, - _ => false, - }); - - assert!(forward_log_reader.next().is_none()); -} - -#[test] fn test_upsert_and_get_from_secondary_memtable() { let data_dir = tmp_dir(); let mut db = DB::configure() @@ -413,7 +350,7 @@ fn test_multiple_writing_threads() { fn test_one_writer_and_multiple_reading_threads() { let data_dir = tmp_dir(); let mut threads = vec![]; - let threads_n = 20; + let threads_n = 100; // Add readers that poll for the records for i in 0..threads_n { @@ -469,3 +406,41 @@ fn test_one_writer_and_multiple_reading_threads() { thread.join().expect("Failed to join thread"); } } + +#[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) + .memtable_capacity(0) // disable memtables + .fields(&vec![ + (Field::Id, RecordFieldType::Int), + (Field::Data, RecordFieldType::Bytes), + ]) + .primary_key(Field::Id) + .initialize() + .expect("Failed to initialize DB instance"); + + let record = Record { + values: vec![ + RecordValue::Int(1), + RecordValue::Bytes(vec![0x1A, 0x1B, 0x1C, 0x1D]), + ], + }; + + db.upsert(&record).expect("Failed to upsert record"); + + let found = db + .get(&RecordValue::Int(1)) + .expect("Failed to get record") + .expect("Record not found"); + + let received = match &found.values[1] { + RecordValue::Bytes(bytes) => bytes, + _ => panic!("Unexpected record value type"), + }; + + assert_eq!(received, &vec![0x1A, 0x1B, 0x1C, 0x1D]); +} |