86 lines
2.4 KiB
Rust
86 lines
2.4 KiB
Rust
use std::fs::{File, OpenOptions};
|
|
use std::io::{Read, Write};
|
|
use std::path::Path;
|
|
use std::sync::Arc;
|
|
|
|
use anyhow::{bail, Context, Result};
|
|
use bytes::{Buf, BufMut};
|
|
use parking_lot::{Mutex, MutexGuard};
|
|
use serde::{Deserialize, Serialize};
|
|
|
|
use crate::compact::CompactionTask;
|
|
|
|
pub struct Manifest {
|
|
file: Arc<Mutex<File>>,
|
|
}
|
|
|
|
#[derive(Serialize, Deserialize)]
|
|
pub enum ManifestRecord {
|
|
Flush(usize),
|
|
NewMemtable(usize),
|
|
Compaction(CompactionTask, Vec<usize>),
|
|
}
|
|
|
|
impl Manifest {
|
|
pub fn create(path: impl AsRef<Path>) -> Result<Self> {
|
|
Ok(Self {
|
|
file: Arc::new(Mutex::new(
|
|
OpenOptions::new()
|
|
.read(true)
|
|
.create_new(true)
|
|
.write(true)
|
|
.open(path)
|
|
.context("failed to create manifest")?,
|
|
)),
|
|
})
|
|
}
|
|
|
|
pub fn recover(path: impl AsRef<Path>) -> Result<(Self, Vec<ManifestRecord>)> {
|
|
let mut file = OpenOptions::new()
|
|
.read(true)
|
|
.append(true)
|
|
.open(path)
|
|
.context("failed to recover manifest")?;
|
|
let mut buf = Vec::new();
|
|
file.read_to_end(&mut buf)?;
|
|
let mut buf_ptr = buf.as_slice();
|
|
let mut records = Vec::new();
|
|
while buf_ptr.has_remaining() {
|
|
let len = buf_ptr.get_u64();
|
|
let slice = &buf_ptr[..len as usize];
|
|
let json = serde_json::from_slice::<ManifestRecord>(slice)?;
|
|
buf_ptr.advance(len as usize);
|
|
let checksum = buf_ptr.get_u32();
|
|
if checksum != crc32fast::hash(slice) {
|
|
bail!("checksum mismatched!");
|
|
}
|
|
records.push(json);
|
|
}
|
|
Ok((
|
|
Self {
|
|
file: Arc::new(Mutex::new(file)),
|
|
},
|
|
records,
|
|
))
|
|
}
|
|
|
|
pub fn add_record(
|
|
&self,
|
|
_state_lock_observer: &MutexGuard<()>,
|
|
record: ManifestRecord,
|
|
) -> Result<()> {
|
|
self.add_record_when_init(record)
|
|
}
|
|
|
|
pub fn add_record_when_init(&self, record: ManifestRecord) -> Result<()> {
|
|
let mut file = self.file.lock();
|
|
let mut buf = serde_json::to_vec(&record)?;
|
|
let hash = crc32fast::hash(&buf);
|
|
file.write_all(&(buf.len() as u64).to_be_bytes())?;
|
|
buf.put_u32(hash);
|
|
file.write_all(&buf)?;
|
|
file.sync_all()?;
|
|
Ok(())
|
|
}
|
|
}
|