use std::{future::Future, sync::Arc};
use wram::{AlignedBuf, BufferPool, SectorRange};
use crate::{
chunk::segment_shift,
error::{Error, Result},
};
pub trait Device: Send + Sync + 'static {
fn sector_size(&self) -> usize;
fn segment_size(&self) -> Option<u64>;
#[inline]
fn direct_io(&self) -> bool {
cfg!(target_os = "linux")
}
#[inline]
fn start_segment(&self) -> u32 {
0
}
#[inline]
fn capacity(&self) -> Option<u64> {
None
}
#[inline]
fn end_segment(&self) -> Option<u32> {
None
}
fn pool(&self) -> &Arc<BufferPool>;
fn write_aligned(
&self,
offset: u64,
buf: AlignedBuf,
) -> impl Future<Output = (Result<usize>, AlignedBuf)>;
fn read_aligned(
&self,
offset: u64,
buf: AlignedBuf,
) -> impl Future<Output = (Result<usize>, AlignedBuf)>;
fn read_raw(
&self,
offset: u64,
buf: AlignedBuf,
) -> impl Future<Output = (Result<usize>, AlignedBuf)>;
fn read_range(&self, offset: u64, len: usize) -> impl Future<Output = Result<AlignedBuf>> {
self.read_range_pooled(offset, len, self.pool())
}
fn read_range_pooled(
&self,
offset: u64,
len: usize,
pool: &Arc<BufferPool>,
) -> impl Future<Output = Result<AlignedBuf>> {
async move {
if len == 0 {
return AlignedBuf::new(0, self.sector_size()).map_err(Error::from);
}
if !self.direct_io() {
let buf = pool.get_with_policy(len, false)?;
let (res, mut buf) = self.read_raw(offset, buf).await;
let bytes_read = res?;
if bytes_read < len {
return Err(Error::UnexpectedEof {
expected: len,
actual: bytes_read,
});
}
buf.set_len(len)?;
return Ok(buf);
}
let sector_size = self.sector_size();
let range = SectorRange::calculate(offset, len, sector_size)?;
let buf = pool.get_with_policy(range.aligned_len, false)?;
let (res, mut buf) = self.read_aligned(range.aligned_offset, buf).await;
let bytes_read = res?;
let sub = range.sub_range(len);
if bytes_read < sub.end {
return Err(Error::UnexpectedEof {
expected: len,
actual: bytes_read.saturating_sub(range.internal_offset),
});
}
if range.internal_offset == 0 && range.aligned_len == len {
buf.set_len(len)?;
return Ok(buf);
}
pool.get_from_slice(&buf[sub]).map_err(Error::from)
}
}
fn sync(&self) -> impl Future<Output = Result<()>>;
fn sync_data(&self) -> impl Future<Output = Result<()>> {
self.sync()
}
#[inline]
fn get_file_size(&self, _segment_id: u32) -> Result<u64> {
Ok(0)
}
fn remove_segment(&self, _segment_id: u32) -> impl Future<Output = Result<()>> {
async move { Ok(()) }
}
#[inline]
fn reset(&self) {}
fn truncate_until_segment(&self, segment_id: u32) -> impl Future<Output = Result<()>>;
fn truncate_until_address(&self, to_address: u64) -> impl Future<Output = Result<()>> {
async move {
if let Some(seg_size) = self.segment_size() {
let to_seg_u64 = if seg_size.is_power_of_two() {
to_address >> segment_shift(seg_size)
} else {
to_address / seg_size
};
let to_seg = u32::try_from(to_seg_u64).unwrap_or(u32::MAX);
self.truncate_until_segment(to_seg).await
} else {
Ok(())
}
}
}
#[inline]
fn recover(&self) -> Result<()> {
Ok(())
}
}
pub use Device as StorageDevice;