bottomless/src/wal.rs
Rama main · réplica de lectura del repositorio original.
use anyhow::{anyhow, Result};
use std::io::SeekFrom;
use std::path::Path;
use tokio::fs::File;
use tokio::io::{AsyncReadExt, AsyncSeekExt, AsyncWrite};
#[repr(transparent)]
#[derive(Debug, Clone, Eq, PartialEq)]
pub struct WalFrameHeader([u8; WalFrameHeader::SIZE]);
impl WalFrameHeader {
pub const SIZE: usize = 24;
/// In multi-page transactions, only the last page in the transaction contains
/// the size_after_transaction field. If it's zero, it means it's an uncommited
/// page.
pub fn is_committed(&self) -> bool {
self.size_after() != 0
}
/// Page number
pub fn pgno(&self) -> u32 {
u32::from_be_bytes([self.0[0], self.0[1], self.0[2], self.0[3]])
}
/// For commit records, the size of the database image in pages
/// after the commit. For all other records, zero.
pub fn size_after(&self) -> u32 {
u32::from_be_bytes([self.0[4], self.0[5], self.0[6], self.0[7]])
}
#[allow(dead_code)]
pub fn salt(&self) -> u64 {
u64::from_be_bytes([
self.0[8], self.0[9], self.0[10], self.0[11], self.0[12], self.0[13], self.0[14],
self.0[15],
])
}
pub fn crc(&self) -> (u32, u32) {
(
u32::from_be_bytes([self.0[16], self.0[17], self.0[18], self.0[19]]),
u32::from_be_bytes([self.0[20], self.0[21], self.0[22], self.0[23]]),
)
}
pub fn verify(&self, init_crc: (u32, u32), page_data: &[u8]) -> Result<(u32, u32)> {
let mut crc = init_crc;
crc = checksum_step(crc, &self.0[0..8]);
crc = checksum_step(crc, page_data);
let frame_crc = self.crc();
if crc == frame_crc {
Ok(crc)
} else {
Err(anyhow!(
"Frame checksum verification failed for page no. {}. Expected: {:X}-{:X}. Got: {:X}-{:X}",
self.pgno(),
frame_crc.0,
frame_crc.1,
crc.0,
crc.1,
))
}
}
}
impl From<[u8; WalFrameHeader::SIZE]> for WalFrameHeader {
fn from(value: [u8; WalFrameHeader::SIZE]) -> Self {
WalFrameHeader(value)
}
}
impl From<WalFrameHeader> for [u8; WalFrameHeader::SIZE] {
fn from(h: WalFrameHeader) -> [u8; WalFrameHeader::SIZE] {
h.0
}
}
impl AsRef<[u8]> for WalFrameHeader {
fn as_ref(&self) -> &[u8] {
self.0.as_ref()
}
}
#[repr(C, packed)]
#[derive(Debug, Clone, Eq, PartialEq)]
pub(crate) struct WalHeader {
/// Magic number. 0x377f0682 or 0x377f0683
pub magic_no: u32,
/// File format version. Currently 3007000
pub version: u32,
/// Database page size.
pub page_size: u32,
/// Checkpoint sequence number
pub checkpoint_seq_no: u32,
/// Random integer incremented with each checkpoint
pub salt_1: u32,
/// A different random integer changing with each checkpoint
pub salt_2: u32,
/// Checksum for first 24 bytes of header
pub checksum_1: u32,
pub checksum_2: u32,
}
impl WalHeader {
pub const SIZE: u64 = 32;
}
impl From<[u8; WalHeader::SIZE as usize]> for WalHeader {
fn from(v: [u8; WalHeader::SIZE as usize]) -> Self {
WalHeader {
magic_no: u32::from_be_bytes([v[0], v[1], v[2], v[3]]),
version: u32::from_be_bytes([v[4], v[5], v[6], v[7]]),
page_size: u32::from_be_bytes([v[8], v[9], v[10], v[11]]),
checkpoint_seq_no: u32::from_be_bytes([v[12], v[13], v[14], v[15]]),
salt_1: u32::from_be_bytes([v[16], v[17], v[18], v[19]]),
salt_2: u32::from_be_bytes([v[20], v[21], v[22], v[23]]),
checksum_1: u32::from_be_bytes([v[24], v[25], v[26], v[27]]),
checksum_2: u32::from_be_bytes([v[28], v[29], v[30], v[31]]),
}
}
}
#[derive(Debug)]
pub(crate) struct WalFileReader {
file: File,
header: WalHeader,
}
impl WalFileReader {
pub async fn open<P: AsRef<Path>>(fpath: P) -> Result<Option<Self>> {
let mut file = File::open(fpath).await?;
let len = file.metadata().await.map(|m| m.len()).unwrap_or(0);
if len < WalHeader::SIZE {
return Ok(None);
}
let header = {
let mut buf = [0u8; WalHeader::SIZE as usize];
file.read_exact(buf.as_mut()).await?;
WalHeader::from(buf)
};
Ok(Some(WalFileReader { file, header }))
}
/// Returns page size stored in WAL file header.
pub fn page_size(&self) -> u32 {
self.header.page_size
}
pub fn checksum(&self) -> (u32, u32) {
(self.header.checksum_1, self.header.checksum_2)
}
pub fn frame_size(&self) -> u64 {
(WalFrameHeader::SIZE as u64) + (self.page_size() as u64)
}
/// Returns an offset in a WAL file, where the data of a frame with given number starts.
pub fn offset(&self, frame_no: u32) -> u64 {
WalHeader::SIZE + ((frame_no - 1) as u64) * self.frame_size()
}
/// Returns a number of pages stored in current WAL file.
pub async fn frame_count(&self) -> u32 {
let len = self.file.metadata().await.map(|m| m.len()).unwrap_or(0);
if len < WalHeader::SIZE {
0
} else {
((len - WalHeader::SIZE) / self.frame_size()) as u32
}
}
/// Sets a file cursor at the beginning of a frame with given number.
pub async fn seek_frame(&mut self, frame_no: u32) -> Result<()> {
let offset = self.offset(frame_no);
self.file.seek(SeekFrom::Start(offset)).await?;
Ok(())
}
/// Reads a header of a WAL frame, without reading the entire page that frame is
/// responsible for.
///
/// For reading specific frame use [WalFileReader::seek_frame] before calling this method.
pub async fn read_frame_header(&mut self) -> Result<Option<WalFrameHeader>> {
let mut header = [0u8; WalFrameHeader::SIZE];
match self.file.read_exact(header.as_mut()).await {
Ok(_) => Ok(Some(WalFrameHeader::from(header))),
Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => Ok(None),
Err(e) => Err(e.into()),
}
}
pub async fn copy_frames<W>(&mut self, w: &mut W, frame_count: usize) -> Result<()>
where
W: AsyncWrite + Unpin,
{
//TODO - specialize non-compressed file cloning:
// libc::copy_file_range(wal.as_mut(), wal.offset(frame), out, 0, len)
let len = (frame_count as u64) * self.frame_size();
let h = self.file.try_clone().await?;
let mut range = h.take(len);
tokio::io::copy(&mut range, w).await?;
Ok(())
}
/// Reads a range of next consecutive frames, including headers, into given buffer.
/// Returns a number of frames read this way.
///
/// # Errors
///
/// This function will propagate any WAL file I/O errors.
/// It will return an error if provided `buf` length is not multiplication of an underlying
/// WAL frame size.
/// It will return an error if at least one frame was not fully read.
#[allow(dead_code)]
pub async fn read_frame_range(&mut self, buf: &mut [u8]) -> Result<usize> {
let frame_size = self.frame_size() as usize;
if buf.len() % frame_size != 0 {
return Err(anyhow!("Provided buffer doesn't fit full frames"));
}
let read = self.file.read_exact(buf).await?;
if read % frame_size != 0 {
Err(anyhow!("Some of the read frames where not complete"))
} else {
Ok(read / frame_size)
}
}
#[allow(dead_code)]
pub async fn next_frame(&mut self, page: &mut [u8]) -> Result<Option<WalFrameHeader>> {
debug_assert_eq!(page.len(), self.page_size() as usize);
let header = self.read_frame_header().await?;
if header.is_some() {
self.file.read_exact(page).await?;
}
Ok(header)
}
}
impl AsMut<File> for WalFileReader {
fn as_mut(&mut self) -> &mut File {
&mut self.file
}
}
/// Generate or extend an 8 byte checksum based on the data in
///the `page` and the `init` value. `page` size must be multiple of 8.
/// FIXME: these computations are performed with host endianness,
/// which is softly assumed to be little endian for majority of devices.
/// However, the only proper way to do this is to get the endianness
/// from the WAL header, as per https://www.sqlite.org/fileformat.html#checksum_algorithm
pub fn checksum_step(init: (u32, u32), page: &[u8]) -> (u32, u32) {
debug_assert_eq!(page.len() % 8, 0);
let (mut s0, mut s1) = init;
let page = unsafe { std::slice::from_raw_parts(page.as_ptr() as *const u32, page.len() / 4) };
let mut i = 0;
while i < page.len() {
s0 = s0.wrapping_add(page[i]).wrapping_add(s1);
s1 = s1.wrapping_add(page[i + 1]).wrapping_add(s0);
i += 2;
}
(s0, s1)
}
#[cfg(test)]
mod test {
use crate::wal::WalHeader;
#[test]
fn wal_header_mem_mapping() {
// copied from actual SQLite WAL file
let source = [
55, 127, 6, 130, 0, 45, 226, 24, 0, 0, 16, 0, 0, 0, 0, 0, 190, 6, 47, 124, 39, 191, 98,
92, 81, 22, 9, 209, 101, 96, 160, 157,
];
let expected = WalHeader {
magic_no: 0x377f0682,
version: 3007000,
page_size: 4096,
checkpoint_seq_no: 0,
salt_1: 3188076412,
salt_2: 666853980,
checksum_1: 1360398801,
checksum_2: 1700831389,
};
let actual = WalHeader::from(source);
assert_eq!(actual, expected);
}
}