diff options
Diffstat (limited to 'log_db')
| -rw-r--r-- | log_db/src/log_reader_forward.rs | 43 | ||||
| -rw-r--r-- | log_db/src/log_reader_reverse.rs | 54 |
2 files changed, 56 insertions, 41 deletions
diff --git a/log_db/src/log_reader_forward.rs b/log_db/src/log_reader_forward.rs index 0e681b9..f9f1a88 100644 --- a/log_db/src/log_reader_forward.rs +++ b/log_db/src/log_reader_forward.rs @@ -22,29 +22,36 @@ impl<'a> ForwardLogReader { } fn read_record(&mut self) -> Result<Option<Record>, io::Error> { - let mut metadata_entry_buf = vec![0; 16]; // 2x u64 - if let Err(e) = self.metadata_reader.read_exact(&mut metadata_entry_buf) { - if e.kind() == io::ErrorKind::UnexpectedEof { - return Ok(None); - } else { - return Err(e); + loop { + let mut metadata_entry_buf = vec![0; 16]; // 2x u64 + if let Err(e) = self.metadata_reader.read_exact(&mut metadata_entry_buf) { + if e.kind() == io::ErrorKind::UnexpectedEof { + return Ok(None); + } else { + return Err(e); + } } - } - // First u64 is the offset of the record in the data file, second is the length of the record - let entry_offset = u64::from_be_bytes(metadata_entry_buf[0..8].try_into().unwrap()); - let entry_length = u64::from_be_bytes(metadata_entry_buf[8..16].try_into().unwrap()); + // First u64 is the offset of the record in the data file, second is the length of the record + let entry_offset = u64::from_be_bytes(metadata_entry_buf[0..8].try_into().unwrap()); + let entry_length = u64::from_be_bytes(metadata_entry_buf[8..16].try_into().unwrap()); + + if entry_offset == 0 && entry_length == 0 { + // This is an unused entry in the metadata file, skip + continue; + } - // Use .seek_relative instead of .seek to avoid dropping the BufReader internal buffer when - // the seek distance is small - let seek_distance = entry_offset as i64 - self.data_reader.stream_position()? as i64; - self.data_reader.seek_relative(seek_distance)?; + // Use .seek_relative instead of .seek to avoid dropping the BufReader internal buffer when + // the seek distance is small + let seek_distance = entry_offset as i64 - self.data_reader.stream_position()? as i64; + self.data_reader.seek_relative(seek_distance)?; - let mut result_buf = vec![0; entry_length as usize]; - self.data_reader.read_exact(&mut result_buf)?; + let mut result_buf = vec![0; entry_length as usize]; + self.data_reader.read_exact(&mut result_buf)?; - let record = Record::deserialize(&result_buf); - Ok(Some(record)) + let record = Record::deserialize(&result_buf); + return Ok(Some(record)); + } } } diff --git a/log_db/src/log_reader_reverse.rs b/log_db/src/log_reader_reverse.rs index 0e930c8..bf093e8 100644 --- a/log_db/src/log_reader_reverse.rs +++ b/log_db/src/log_reader_reverse.rs @@ -32,35 +32,43 @@ impl ReverseLogReader { } fn read_record(&mut self) -> Result<Option<Record>, io::Error> { - assert!( - (self.metadata_pos - METADATA_FILE_HEADER_SIZE) % 16 == 0, - "metadata_pos is not aligned" - ); + loop { + assert!( + (self.metadata_pos - METADATA_FILE_HEADER_SIZE) % 16 == 0, + "metadata_pos is not aligned" + ); + // Return None if we have read the entire metadata file and + // reached the end of the header + if self.metadata_pos == METADATA_FILE_HEADER_SIZE { + return Ok(None); + } - // Return None if we have read the entire metadata file and - // reached the end of the header - if self.metadata_pos == METADATA_FILE_HEADER_SIZE { - return Ok(None); - } + self.metadata_pos -= 16; + let i = self.metadata_pos; // shorter alias - self.metadata_pos -= 16; - let i = self.metadata_pos; // shorter alias + // First u64 is the offset of the record in the data file, second is the length of the record + let entry_offset = u64::from_be_bytes(self.metadata_buf[i..i + 8].try_into().unwrap()); + let entry_length = + u64::from_be_bytes(self.metadata_buf[i + 8..i + 16].try_into().unwrap()); - // First u64 is the offset of the record in the data file, second is the length of the record - let entry_offset = u64::from_be_bytes(self.metadata_buf[i..i + 8].try_into().unwrap()); - let entry_length = u64::from_be_bytes(self.metadata_buf[i + 8..i + 16].try_into().unwrap()); + if entry_offset == 0 && entry_length == 0 { + // This is an unused entry in the metadata file, skip + continue; + } - self.data_reader.seek(io::SeekFrom::Start(entry_offset))?; - let mut result_buf = vec![0; entry_length as usize]; - self.data_reader.read_exact(&mut result_buf)?; + self.data_reader.seek(io::SeekFrom::Start(entry_offset))?; + let mut result_buf = vec![0; entry_length as usize]; + self.data_reader.read_exact(&mut result_buf)?; - let record = Record::deserialize(&result_buf); + let record = Record::deserialize(&result_buf); - assert!( - (self.metadata_pos - METADATA_FILE_HEADER_SIZE) % 16 == 0, - "metadata_pos is not aligned" - ); - Ok(Some(record)) + assert!( + (self.metadata_pos - METADATA_FILE_HEADER_SIZE) % 16 == 0, + "metadata_pos is not aligned" + ); + + return Ok(Some(record)); + } } } |