aboutsummaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
authorDeterminant <[email protected]>2020-06-10 15:56:05 -0400
committerDeterminant <[email protected]>2020-06-10 15:56:05 -0400
commit617c515ec1834501d4835076de3adcfb495d8ef9 (patch)
tree56e12428154cdb5c94de4a414d3e0966b054e70d /src
parent415578aa48a12cac01d763b7c08384dbcd46911b (diff)
add basic error handling
Diffstat (limited to 'src')
-rw-r--r--src/wal.rs79
1 files changed, 42 insertions, 37 deletions
diff --git a/src/wal.rs b/src/wal.rs
index b5b3e91..bcc7a19 100644
--- a/src/wal.rs
+++ b/src/wal.rs
@@ -60,25 +60,25 @@ pub trait WALFile {
/// `allocate/truncate` invocation should be visible if ordered earlier (should be guaranteed
/// by most OS). Additionally, the final write caused by each invocation of this function
/// should be _atomic_ (the entire single write should be all or nothing).
- fn write(&self, offset: WALPos, data: WALBytes);
+ fn write(&self, offset: WALPos, data: WALBytes) -> Result<(), ()>;
/// Read data with offset.
- fn read(&self, offset: WALPos, length: usize) -> Option<WALBytes>;
+ fn read(&self, offset: WALPos, length: usize) -> Result<Option<WALBytes>, ()>;
}
pub trait WALStore {
type FileNameIter: Iterator<Item = String>;
/// Open a file given the filename, create the file if not exists when `touch` is `true`.
- fn open_file(&mut self, filename: &str, touch: bool) -> Option<Box<dyn WALFile>>;
+ fn open_file(&mut self, filename: &str, touch: bool) -> Result<Box<dyn WALFile>, ()>;
/// Unlink a file given the filename.
fn remove_file(&mut self, filename: &str) -> Result<(), ()>;
/// Enumerate all WAL filenames. It should include all WAL files that are previously opened
/// (created) but not removed. The list could be unordered.
- fn enumerate_files(&self) -> Self::FileNameIter;
- /// Apply the payload during recovery. This notifies the application should redo the given
- /// operation to ensure its state is consistent (the major goal of having a WAL). We assume
- /// the application applies the payload by the _order_ of this callback invocation.
- fn apply_payload(&mut self, payload: WALBytes);
+ fn enumerate_files(&self) -> Result<Self::FileNameIter, ()>;
+ /// Apply the payload during recovery. An invocation of the callback waits the application for
+ /// redoing the given operation to ensure its state is consistent. We assume the necessary
+ /// changes by the payload has already been persistent when the callback returns.
+ fn apply_payload(&mut self, payload: WALBytes) -> Result<(), ()>;
}
/// The middle layer that manages WAL file handles and invokes public trait functions to actually
@@ -108,15 +108,15 @@ impl<F: WALStore> WALFilePool<F> {
format!("{:08x}.log", fid)
}
- fn get_file(&mut self, fid: u64, touch: bool) -> &'static dyn WALFile {
+ fn get_file(&mut self, fid: u64, touch: bool) -> Result<&'static dyn WALFile, ()> {
let h = match self.handles.get(&fid) {
Some(h) => &**h,
None => {
- self.handles.put(fid, self.store.open_file(&Self::get_fname(fid), touch).unwrap());
+ self.handles.put(fid, self.store.open_file(&Self::get_fname(fid), touch)?);
&**self.handles.get(&fid).unwrap()
}
};
- unsafe {&*(h as *const dyn WALFile)}
+ Ok(unsafe {&*(h as *const dyn WALFile)})
}
fn get_fid(&mut self, fname: &str) -> WALFileId {
@@ -124,17 +124,20 @@ impl<F: WALStore> WALFilePool<F> {
}
// TODO: evict stale handles
- fn write(&mut self, writes: Vec<(WALPos, WALBytes)>) {
+ fn write(&mut self, writes: Vec<(WALPos, WALBytes)>) -> Result<(), ()> {
// pre-allocate the file space
let mut fid = writes[0].0 >> self.file_nbit;
let mut alloc_start = writes[0].0 & (self.file_size - 1);
let mut alloc_end = alloc_start + writes[0].1.len() as u64;
- let mut h = self.get_file(fid, true);
- for (off, w) in &writes[1..] {
+ let files = writes.iter().map(|(off, _)|
+ self.get_file((*off) >> self.file_nbit, true)).collect::<Result<Vec<&dyn WALFile>, ()>>()?;
+ // prepare file handles
+ let mut last_h = files[0];
+ for ((off, w), h) in writes[1..].iter().zip(files[1..].iter()) {
let next_fid = off >> self.file_nbit;
if next_fid != fid {
- h.allocate(alloc_start, (alloc_end - alloc_start) as usize).unwrap();
- h = self.get_file(next_fid, true);
+ last_h.allocate(alloc_start, (alloc_end - alloc_start) as usize)?;
+ last_h = *h;
alloc_start = 0;
alloc_end = alloc_start + w.len() as u64;
fid = next_fid;
@@ -142,10 +145,11 @@ impl<F: WALStore> WALFilePool<F> {
alloc_end += w.len() as u64;
}
}
- h.allocate(alloc_start, (alloc_end - alloc_start) as usize).unwrap();
+ last_h.allocate(alloc_start, (alloc_end - alloc_start) as usize)?;
for (off, w) in writes.into_iter() {
- self.get_file(off >> self.file_nbit, true).write(off & (self.file_size - 1), w);
+ self.get_file(off >> self.file_nbit, true)?.write(off & (self.file_size - 1), w)?;
}
+ Ok(())
}
fn remove_file(&mut self, fid: u64) -> Result<(), ()> {
@@ -186,7 +190,7 @@ impl<F: WALStore> WALWriter<F> {
/// Submit a sequence of records to WAL; WALStore/WALFile callbacks are invoked before the
/// function returns. The caller then has the knowledge of WAL writes so it should defer
/// actual data writes after WAL writes.
- pub fn grow<T: AsRef<[WALBytes]>>(&mut self, records: T) -> Box<[WALRingId]> {
+ pub fn grow<T: AsRef<[WALBytes]>>(&mut self, records: T) -> Result<Box<[WALRingId]>, ()> {
let mut res = Vec::new();
let mut writes = Vec::new();
let msize = std::mem::size_of::<WALRingBlob>() as u32;
@@ -257,13 +261,13 @@ impl<F: WALStore> WALWriter<F> {
.to_vec().into_boxed_slice()));
self.state.next += (bbuff_cur - bbuff_start) as u64;
}
- self.file_pool.write(writes);
- res.into_boxed_slice()
+ self.file_pool.write(writes)?;
+ Ok(res.into_boxed_slice())
}
/// Inform the WALWriter that data writes (specified by a slice of (offset, length) tuples) are
/// complete so that it could automatically remove obsolete WAL files.
- pub fn peel<T: AsRef<[WALRingId]>>(&mut self, records: T) {
+ pub fn peel<T: AsRef<[WALRingId]>>(&mut self, records: T) -> Result<(), ()> {
let msize = std::mem::size_of::<WALRingBlob>() as u64;
let block_size = self.block_size as u64;
for rec in records.as_ref() {
@@ -283,9 +287,10 @@ impl<F: WALStore> WALWriter<F> {
}
let next_fid = self.next_complete >> self.state.file_nbit;
for fid in orig_fid..next_fid {
- self.file_pool.remove_file(fid).unwrap();
+ self.file_pool.remove_file(fid)?;
}
self.state.first_fid = next_fid;
+ Ok(())
}
}
@@ -301,20 +306,20 @@ impl<F: WALStore> WALLoader<F> {
WALLoader{ file_pool, filename_fmt }
}
- pub fn recover(mut self) -> WALWriter<F> {
+ pub fn recover(mut self) -> Result<WALWriter<F>, ()> {
let block_size = 1 << self.file_pool.block_nbit;
let msize = std::mem::size_of::<WALRingBlob>() as u32;
let mut logfiles: Vec<String> = self.file_pool.store
- .enumerate_files()
+ .enumerate_files()?
.filter(|f| self.filename_fmt.is_match(f)).collect();
// TODO: check for missing logfiles
logfiles.sort();
let mut chunks = None;
for fname in logfiles.iter() {
let fid = self.file_pool.get_fid(fname);
- let f = self.file_pool.get_file(fid, false);
+ let f = self.file_pool.get_file(fid, false)?;
let mut off = 0;
- while let Some(header_raw) = f.read(off, msize as usize) {
+ while let Some(header_raw) = f.read(off, msize as usize)? {
if block_size - (off & (block_size - 1)) <= msize as u64 { break }
off += msize as u64;
let header = unsafe {
@@ -323,21 +328,21 @@ impl<F: WALStore> WALLoader<F> {
match header.rtype {
WALRingType::Full => {
assert!(chunks.is_none());
- let payload = f.read(off, rsize as usize).unwrap();
+ let payload = f.read(off, rsize as usize)?.ok_or(())?;
off += rsize as u64;
- self.file_pool.store.apply_payload(payload);
+ self.file_pool.store.apply_payload(payload)?;
},
WALRingType::First => {
assert!(chunks.is_none());
- chunks = Some(vec![f.read(off, rsize as usize).unwrap()]);
+ chunks = Some(vec![f.read(off, rsize as usize)?.ok_or(())?]);
off += rsize as u64;
},
WALRingType::Middle => {
- chunks.as_mut().unwrap().push(f.read(off, rsize as usize).unwrap());
+ chunks.as_mut().unwrap().push(f.read(off, rsize as usize)?.ok_or(())?);
off += rsize as u64;
},
WALRingType::Last => {
- chunks.as_mut().unwrap().push(f.read(off, rsize as usize).unwrap());
+ chunks.as_mut().unwrap().push(f.read(off, rsize as usize)?.ok_or(())?);
off += rsize as u64;
let _chunks = chunks.take().unwrap();
@@ -348,19 +353,19 @@ impl<F: WALStore> WALLoader<F> {
ps[..c.len()].copy_from_slice(&*c);
ps = &mut ps[c.len()..];
}
- self.file_pool.store.apply_payload(payload.into_boxed_slice());
+ self.file_pool.store.apply_payload(payload.into_boxed_slice())?;
},
WALRingType::Null => break,
}
}
- f.truncate(0).unwrap();
- self.file_pool.remove_file(fid).unwrap();
+ f.truncate(0)?;
+ self.file_pool.remove_file(fid)?;
}
self.file_pool.reset();
- WALWriter::new(WALState {
+ Ok(WALWriter::new(WALState {
first_fid: 0,
next: 0,
file_nbit: self.file_pool.file_nbit,
- }, self.file_pool)
+ }, self.file_pool))
}
}