aboutsummaryrefslogtreecommitdiffstats
path: root/log_db/src
diff options
context:
space:
mode:
Diffstat (limited to 'log_db/src')
-rw-r--r--log_db/src/common.rs57
-rw-r--r--log_db/src/lib.rs52
-rw-r--r--log_db/src/record.rs2
3 files changed, 49 insertions, 62 deletions
diff --git a/log_db/src/common.rs b/log_db/src/common.rs
index a6fd1d2..5539689 100644
--- a/log_db/src/common.rs
+++ b/log_db/src/common.rs
@@ -42,10 +42,12 @@ pub fn metadata_filename(num: u16) -> String {
format!("metadata.{}", num)
}
+pub type DBResult<A> = Result<A, DBError>;
+
#[derive(Debug, Error)]
pub enum DBError {
#[error("lock request failed: {0}")]
- LockRequestError(#[from] LockRequestError),
+ LockRequestError(String),
#[error("validation failed: {0}")]
ValidationError(String),
#[error("consistency check failed: {0}")]
@@ -432,7 +434,7 @@ pub fn get_secondary_memtable_index_by_field<Field: Eq>(
sks.iter().position(|schema_field| schema_field == field)
}
-pub fn is_file_same_as_path(file: &File, path: &PathBuf) -> io::Result<bool> {
+pub fn is_file_same_as_path(file: &File, path: &PathBuf) -> DBResult<bool> {
// Get the metadata for the open file handle
let file_metadata = file.metadata()?;
@@ -468,7 +470,7 @@ pub fn symlink(original: &Path, link: &Path) -> io::Result<()> {
}
/// Set the active segment to the segment with the given ordinal number.
-pub fn set_active_segment(data_dir_path: &Path, segment_num: u16) -> Result<(), io::Error> {
+pub fn set_active_segment(data_dir_path: &Path, segment_num: u16) -> DBResult<()> {
let tmp_uuid = Uuid::new_v4();
let tmp_filename = format!("active_{}", tmp_uuid.to_string());
let tmp_path = data_dir_path.join(tmp_filename);
@@ -489,7 +491,7 @@ pub fn set_active_segment(data_dir_path: &Path, segment_num: u16) -> Result<(),
pub fn create_segment_metadata_file(
data_dir_path: &Path,
data_file_uuid: &Uuid,
-) -> Result<(u16, PathBuf), io::Error> {
+) -> DBResult<(u16, PathBuf)> {
let current_greatest_num = greatest_segment_number(data_dir_path)?;
let new_num = current_greatest_num + 1;
@@ -514,7 +516,7 @@ pub fn create_segment_metadata_file(
}
/// Parse the segment number from a metadata file path
-pub fn parse_segment_number(metadata_path: &Path) -> Result<u16, io::Error> {
+pub fn parse_segment_number(metadata_path: &Path) -> DBResult<u16> {
let filename = metadata_path
.file_name()
.expect("No filename in symlink")
@@ -529,17 +531,14 @@ pub fn parse_segment_number(metadata_path: &Path) -> Result<u16, io::Error> {
.parse::<u16>();
segment_number.map_err(|_| {
- io::Error::new(
- io::ErrorKind::InvalidData,
- "Failed to parse segment number from filename",
- )
+ DBError::ValidationError("Failed to parse segment number from filename".to_owned())
})
}
/// Get the number of the segment with the greatest ordinal.
/// This is the newest segment, i.e. the one that is pointed to by the `active` symlink.
/// If there are no segments yet, returns 0.
-pub fn greatest_segment_number(data_dir_path: &Path) -> Result<u16, io::Error> {
+pub fn greatest_segment_number(data_dir_path: &Path) -> DBResult<u16> {
let active_symlink = data_dir_path.join(ACTIVE_SYMLINK_FILENAME);
if !fs::exists(&active_symlink)? {
@@ -553,7 +552,7 @@ pub fn greatest_segment_number(data_dir_path: &Path) -> Result<u16, io::Error> {
/// Create a new segment data file and return its UUID.
/// A data file contains the segment data, tightly packed without separators.
/// An accompanying metadata file is required to interpret the data.
-pub fn create_segment_data_file(data_dir_path: &Path) -> Result<(Uuid, PathBuf), io::Error> {
+pub fn create_segment_data_file(data_dir_path: &Path) -> DBResult<(Uuid, PathBuf)> {
let uuid = Uuid::new_v4();
let new_segment_path = data_dir_path.join(uuid.to_string());
fs::OpenOptions::new()
@@ -567,7 +566,7 @@ pub fn create_segment_data_file(data_dir_path: &Path) -> Result<(Uuid, PathBuf),
/// Reads the metadata header from the metadata file.
/// Leaves the file seek head at the beginning of the records, after the header.
-pub fn read_metadata_header(metadata_file: &mut fs::File) -> Result<MetadataHeader, io::Error> {
+pub fn read_metadata_header(metadata_file: &mut fs::File) -> DBResult<MetadataHeader> {
metadata_file.seek(SeekFrom::Start(0))?;
let mut buf = [0u8; METADATA_FILE_HEADER_SIZE];
metadata_file.read_exact(&mut buf)?;
@@ -576,7 +575,7 @@ pub fn read_metadata_header(metadata_file: &mut fs::File) -> Result<MetadataHead
Ok(header)
}
-pub fn validate_metadata_header(header: &MetadataHeader) -> Result<(), DBError> {
+pub fn validate_metadata_header(header: &MetadataHeader) -> DBResult<()> {
if header.version != 1 {
return Err(DBError::ValidationError(
"Unsupported metadata file version".to_owned(),
@@ -592,9 +591,7 @@ pub enum IsMetadatafileValidResult {
TruncateToSize(u64),
}
-pub fn is_metadata_file_valid(
- metadata_file: &mut fs::File,
-) -> Result<IsMetadatafileValidResult, io::Error> {
+pub fn is_metadata_file_valid(metadata_file: &mut fs::File) -> DBResult<IsMetadatafileValidResult> {
let size = metadata_file.seek(SeekFrom::End(0))? as usize;
if size < METADATA_FILE_HEADER_SIZE {
@@ -626,7 +623,7 @@ pub fn is_metadata_file_valid(
pub fn ensure_active_metadata_is_valid(
data_dir: &Path,
metadata_file: &mut fs::File,
-) -> Result<bool, io::Error> {
+) -> DBResult<bool> {
let current_len = metadata_file.seek(SeekFrom::End(0))? as usize;
match is_metadata_file_valid(metadata_file)? {
@@ -683,18 +680,8 @@ pub fn ensure_active_metadata_is_valid(
const LOCK_WAIT_MAX_MS: u64 = 1000;
-#[derive(Error, Debug)]
-pub enum LockRequestError {
- #[error("lock request file was removed unexpectedly")]
- LockRequestFileRemoved,
- #[error("timed out while waiting for a lock, max wait time: {0} ms")]
- TimedOut(u64),
- #[error("unexpected IO error: {0}")]
- IOError(#[from] io::Error),
-}
-
-pub fn is_exclusive_lock_requested(data_dir: &Path) -> Result<bool, LockRequestError> {
let lock_request_path = data_dir.join(EXCL_LOCK_REQUEST_FILENAME);
+pub fn is_exclusive_lock_requested(data_dir: &Path) -> DBResult<bool> {
let lock_request_file = fs::OpenOptions::new()
.create(true)
.write(true) // When requesting a lock, we need to have either read or write permissions
@@ -707,14 +694,16 @@ pub fn is_exclusive_lock_requested(data_dir: &Path) -> Result<bool, LockRequestE
if e.kind() == lock_contended_error().kind() {
return Ok(true);
}
- return Err(LockRequestError::IOError(e));
+ return Err(DBError::IOError(e));
}
Ok(_) => {
// Check that the exclusive lock request file is still the same as the one we opened
if !is_file_same_as_path(&lock_request_file, &lock_request_path)? {
// The lock request file has been removed
- return Err(LockRequestError::LockRequestFileRemoved);
+ return Err(DBError::ConsistencyError(
+ "Lock request file was removed while checking for exclusive lock".to_owned(),
+ ));
}
lock_request_file.unlock()?;
@@ -723,7 +712,7 @@ pub fn is_exclusive_lock_requested(data_dir: &Path) -> Result<bool, LockRequestE
}
}
-pub fn request_shared_lock(data_dir: &Path, file: &mut fs::File) -> Result<(), LockRequestError> {
+pub fn request_shared_lock(data_dir: &Path, file: &mut fs::File) -> DBResult<()> {
let mut timeout = 5;
loop {
if is_exclusive_lock_requested(data_dir)? {
@@ -735,7 +724,9 @@ pub fn request_shared_lock(data_dir: &Path, file: &mut fs::File) -> Result<(), L
timeout *= 2;
if timeout > LOCK_WAIT_MAX_MS {
- return Err(LockRequestError::TimedOut(LOCK_WAIT_MAX_MS));
+ return Err(DBError::LockRequestError(
+ "Acquisition of shared lock timed out after {LOCK_WAIT_MAX_MS}".to_owned(),
+ ));
}
} else {
file.lock_shared()?;
@@ -744,7 +735,7 @@ pub fn request_shared_lock(data_dir: &Path, file: &mut fs::File) -> Result<(), L
}
}
-pub fn request_exclusive_lock(data_dir: &Path, file: &mut fs::File) -> Result<(), io::Error> {
+pub fn request_exclusive_lock(data_dir: &Path, file: &mut fs::File) -> DBResult<()> {
// Create a lock on the exclusive lock request file to signal to readers that they should wait
let lock_request_path = data_dir.join(EXCL_LOCK_REQUEST_FILENAME);
let lock_request_file = fs::OpenOptions::new()
diff --git a/log_db/src/lib.rs b/log_db/src/lib.rs
index 50e5a65..97ea28f 100644
--- a/log_db/src/lib.rs
+++ b/log_db/src/lib.rs
@@ -75,7 +75,7 @@ impl<R: Recordable> ConfigBuilder<R> {
self
}
- pub fn initialize(&self) -> Result<DB<R>, DBError> {
+ pub fn initialize(&self) -> DBResult<DB<R>> {
let config = Config {
fields: R::schema(),
primary_key: R::primary_key(),
@@ -125,7 +125,7 @@ impl<R: Recordable> DB<R> {
ConfigBuilder::new()
}
- fn initialize(config: Config<R>) -> Result<DB<R>, DBError> {
+ fn initialize(config: Config<R>) -> DBResult<DB<R>> {
info!("Initializing DB...");
// If data_dir does not exist or is empty, create it and any necessary files
// After creation, the directory should always be in a complete state
@@ -234,7 +234,7 @@ impl<R: Recordable> DB<R> {
/// Refresh the in-memory indexes from the log files.
/// This needs to only be called if the read consistency is set to `ReadConsistency::Eventual`.
- pub fn refresh_indexes(&mut self) -> Result<(), DBError> {
+ pub fn refresh_indexes(&mut self) -> DBResult<()> {
let active_symlink_path = self.data_dir.join(ACTIVE_SYMLINK_FILENAME);
let active_target = fs::read_link(active_symlink_path)?;
let active_metadata_path = self.data_dir.join(active_target);
@@ -256,8 +256,7 @@ impl<R: Recordable> DB<R> {
)));
}
- request_shared_lock(&self.data_dir, &mut metadata_file)
- .map_err(|lre| DBError::LockRequestError(lre))?;
+ request_shared_lock(&self.data_dir, &mut metadata_file)?;
let metadata_header = read_metadata_header(&mut metadata_file)?;
validate_metadata_header(&metadata_header)?;
@@ -333,7 +332,7 @@ impl<R: Recordable> DB<R> {
/// Insert a record into the database. If the primary key value already exists,
/// the existing record will be replaced by the supplied one.
- pub fn upsert(&mut self, recordable: R) -> Result<(), DBError> {
+ pub fn upsert(&mut self, recordable: R) -> DBResult<()> {
let record = Record::from(&recordable.into_record());
debug!("Upserting record: {:?}", record);
@@ -345,7 +344,7 @@ impl<R: Recordable> DB<R> {
/// Insert a batch of records into the database. If the primary key value for a record already exists,
/// the existing record will be replaced by the supplied one. Records are inserted in the order they are given.
- pub fn batch_upsert(&mut self, recordables: Vec<R>) -> Result<(), DBError> {
+ pub fn batch_upsert(&mut self, recordables: Vec<R>) -> DBResult<()> {
let records = recordables
.into_iter()
.map(|r| Record::from(&r.into_record()))
@@ -360,10 +359,7 @@ impl<R: Recordable> DB<R> {
self.batch_upsert_records(records.into_iter())
}
- fn batch_upsert_records(
- &mut self,
- records: impl Iterator<Item = Record>,
- ) -> Result<(), DBError> {
+ fn batch_upsert_records(&mut self, records: impl Iterator<Item = Record>) -> DBResult<()> {
debug!("Opening file in append mode and acquiring exclusive lock...");
// Acquire an exclusive lock for writing
@@ -444,7 +440,7 @@ impl<R: Recordable> DB<R> {
/// Get a record by its primary index value.
/// E.g. `db.get(Value::Int(10))`.
- pub fn get(&mut self, value: &Value) -> Result<Option<R>, DBError> {
+ pub fn get(&mut self, value: &Value) -> DBResult<Option<R>> {
let value_batch = std::iter::once(value);
let records = self.batch_find_by_records(&self.config.primary_key.clone(), value_batch)?;
assert!(records.len() <= 1);
@@ -457,7 +453,7 @@ impl<R: Recordable> DB<R> {
/// Get a collection of records based on a field value.
/// Indexes will be used if they are applicable.
- pub fn find_by(&mut self, field: &R::Field, value: &Value) -> Result<Vec<R>, DBError> {
+ pub fn find_by(&mut self, field: &R::Field, value: &Value) -> DBResult<Vec<R>> {
let value_batch = std::iter::once(value);
Ok(self
.batch_find_by_records(field, value_batch)?
@@ -474,7 +470,7 @@ impl<R: Recordable> DB<R> {
&mut self,
field: &R::Field,
values: &[Value],
- ) -> Result<Vec<(usize, R)>, DBError> {
+ ) -> DBResult<Vec<(usize, R)>> {
Ok(self
.batch_find_by_records(field, values.iter())?
.into_iter()
@@ -486,7 +482,7 @@ impl<R: Recordable> DB<R> {
&mut self,
field: &R::Field,
values: impl Iterator<Item = &'a Value>,
- ) -> Result<Vec<(usize, Record)>, DBError> {
+ ) -> DBResult<Vec<(usize, Record)>> {
let field_type = self.get_field_type(field).ok_or(DBError::ValidationError(
"Field not found in schema".to_owned(),
))?;
@@ -504,7 +500,7 @@ impl<R: Recordable> DB<R> {
)))
}
})
- .collect::<Result<Vec<IndexableValue>, DBError>>()?;
+ .collect::<DBResult<Vec<IndexableValue>>>()?;
// Otherwise, continue with querying secondary indexes.
debug!(
@@ -546,7 +542,7 @@ impl<R: Recordable> DB<R> {
Ok(log_keys)
}
})
- .collect::<Result<Vec<Vec<&LogKey>>, DBError>>()?;
+ .collect::<DBResult<Vec<Vec<&LogKey>>>>()?;
debug!("Found log keys in memtable: {:?}", log_key_batches);
@@ -570,7 +566,7 @@ impl<R: Recordable> DB<R> {
fn read_tagged_log_keys<'a>(
&self,
log_keys: impl Iterator<Item = (usize, &'a LogKey)>,
- ) -> Result<Vec<(usize, Record)>, DBError> {
+ ) -> DBResult<Vec<(usize, Record)>> {
let mut records = vec![];
let mut log_keys_map = BTreeMap::new();
@@ -636,7 +632,7 @@ impl<R: Recordable> DB<R> {
&mut self,
field: &R::Field,
range: B,
- ) -> Result<Vec<R>, DBError> {
+ ) -> DBResult<Vec<R>> {
Ok(self
.range_by_records(field, range)?
.into_iter()
@@ -648,12 +644,12 @@ impl<R: Recordable> DB<R> {
&mut self,
field: &R::Field,
range: B,
- ) -> Result<Vec<Record>, DBError> {
+ ) -> DBResult<Vec<Record>> {
fn range_bound_to_indexable(
bound: Bound<&Value>,
field_type: &ValueType,
- ) -> Result<Bound<IndexableValue>, DBError> {
- fn convert(value: &Value, field_type: &ValueType) -> Result<IndexableValue, DBError> {
+ ) -> DBResult<Bound<IndexableValue>> {
+ fn convert(value: &Value, field_type: &ValueType) -> DBResult<IndexableValue> {
if !type_check(&value, field_type) {
return Err(DBError::ValidationError(format!(
"Queried value does not match type: {:?}",
@@ -706,7 +702,7 @@ impl<R: Recordable> DB<R> {
/// Ensures that the `self.metadata_file` and `self.data_file` handles are still pointing to the correct files.
/// If the segment has been rotated, the handle will be closed and reopened.
/// Returns `false` if the file has been rotated and the handle has been reopened, `true` otherwise.
- fn ensure_metadata_file_is_active(&mut self) -> Result<bool, DBError> {
+ fn ensure_metadata_file_is_active(&mut self) -> DBResult<bool> {
let active_target = fs::read_link(&self.data_dir.join(ACTIVE_SYMLINK_FILENAME))?;
let active_metadata_path = &self.data_dir.join(active_target);
@@ -738,7 +734,7 @@ impl<R: Recordable> DB<R> {
///
/// Deletion is done by marking the record as a tombstone. The record will still be present in the log file,
/// but will be ignored by reads. Upon compaction, tombstoned records will be removed.
- pub fn delete_by(&mut self, field: &R::Field, value: &Value) -> Result<Vec<R>, DBError> {
+ pub fn delete_by(&mut self, field: &R::Field, value: &Value) -> DBResult<Vec<R>> {
let recs = self.delete_by_field(field, value)?;
Ok(recs
@@ -748,7 +744,7 @@ impl<R: Recordable> DB<R> {
}
/// Delete record by primary key.
- pub fn delete(&mut self, pk: &Value) -> Result<Option<R>, DBError> {
+ pub fn delete(&mut self, pk: &Value) -> DBResult<Option<R>> {
let recs = self.delete_by_field(&self.config.primary_key.clone(), pk)?;
assert!(recs.len() <= 1);
@@ -758,7 +754,7 @@ impl<R: Recordable> DB<R> {
.map(|rec| R::from_record(rec.values)))
}
- fn delete_by_field(&mut self, field: &R::Field, value: &Value) -> Result<Vec<Record>, DBError> {
+ fn delete_by_field(&mut self, field: &R::Field, value: &Value) -> DBResult<Vec<Record>> {
let value_batch = std::iter::once(value);
let recs: Vec<Record> = self
.batch_find_by_records(field, value_batch)?
@@ -822,7 +818,7 @@ impl<R: Recordable> DB<R> {
/// 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 and reads will be blocked during the tasks.
- pub fn do_maintenance_tasks(&mut self) -> Result<(), DBError> {
+ pub fn do_maintenance_tasks(&mut self) -> DBResult<()> {
request_exclusive_lock(&self.data_dir, &mut self.active_metadata_file)?;
ensure_active_metadata_is_valid(&self.data_dir, &mut self.active_metadata_file)?;
@@ -837,7 +833,7 @@ impl<R: Recordable> DB<R> {
Ok(())
}
- fn rotate_and_compact(&mut self) -> Result<(), io::Error> {
+ fn rotate_and_compact(&mut self) -> DBResult<()> {
debug!("Active log size exceeds threshold, starting rotation and compaction...");
self.active_data_file.lock_shared()?;
diff --git a/log_db/src/record.rs b/log_db/src/record.rs
index 4a30902..cbb1faa 100644
--- a/log_db/src/record.rs
+++ b/log_db/src/record.rs
@@ -49,7 +49,7 @@ impl Record {
&self.values[index]
}
- pub fn validate<Field: Eq>(&self, schema: &Vec<(Field, ValueType)>) -> Result<(), DBError> {
+ pub fn validate<Field: Eq>(&self, schema: &Vec<(Field, ValueType)>) -> DBResult<()> {
// Validate the record length
if self.values.len() != schema.len() {
return Err(DBError::ValidationError(format!(

Want to contribute? Host your changes somewhere and send me the link: jan (at) sydanviita.fi. Also, check out my website.


forge.jan.systems — powered by cgit