tyler/proxmox-backup
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

move CachedChunkReader to pbs-datastore

Browse Source 
this was actually still missing from the previous commit

Signed-off-by: Wolfgang Bumiller <w.bumiller@proxmox.com>
This commit is contained in:
Wolfgang Bumiller
2021-07-21 14:20:00 +02:00
parent eb5e0ae65a
commit ef4df211ab
5 changed files with 7 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

191
src/backup/cached_chunk_reader.rs
View File

@ -1,191 +0,0 @@
//! An async and concurrency safe data reader backed by a local LRU cache.
use std::future::Future;
use std::io::SeekFrom;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
use anyhow::Error;
use futures::ready;
use tokio::io::{AsyncRead, AsyncSeek, ReadBuf};
use proxmox::io_format_err;
use proxmox::sys::error::io_err_other;
use pbs_datastore::index::IndexFile;
use pbs_datastore::read_chunk::AsyncReadChunk;
use pbs_tools::async_lru_cache::{AsyncCacher, AsyncLruCache};
struct AsyncChunkCacher<T> {
reader: Arc<T>,
}
impl<T: AsyncReadChunk + Send + Sync + 'static> AsyncCacher<[u8; 32], Arc<Vec<u8>>>
for AsyncChunkCacher<T>
{
fn fetch(
&self,
key: [u8; 32],
) -> Box<dyn Future<Output = Result<Option<Arc<Vec<u8>>>, Error>> + Send> {
let reader = Arc::clone(&self.reader);
Box::new(async move {
AsyncReadChunk::read_chunk(reader.as_ref(), &key)
.await
.map(|x| Some(Arc::new(x)))
})
}
}
/// Allows arbitrary data reads from an Index via an AsyncReadChunk implementation, using an LRU
/// cache internally to cache chunks and provide support for multiple concurrent reads (potentially
/// to the same chunk).
pub struct CachedChunkReader<I: IndexFile, R: AsyncReadChunk + Send + Sync + 'static> {
cache: Arc<AsyncLruCache<[u8; 32], Arc<Vec<u8>>>>,
cacher: AsyncChunkCacher<R>,
index: I,
}
impl<I: IndexFile, R: AsyncReadChunk + Send + Sync + 'static> CachedChunkReader<I, R> {
/// Create a new reader with a local LRU cache containing 'capacity' chunks.
pub fn new(reader: R, index: I, capacity: usize) -> Self {
let cache = Arc::new(AsyncLruCache::new(capacity));
Self::new_with_cache(reader, index, cache)
}
/// Create a new reader with a custom LRU cache. Use this to share a cache between multiple
/// readers.
pub fn new_with_cache(
reader: R,
index: I,
cache: Arc<AsyncLruCache<[u8; 32], Arc<Vec<u8>>>>,
) -> Self {
Self {
cache,
cacher: AsyncChunkCacher {
reader: Arc::new(reader),
},
index,
}
}
/// Read data at a given byte offset into a variable size buffer. Returns the amount of bytes
/// read, which will always be the size of the buffer except when reaching EOF.
pub async fn read_at(&self, buf: &mut [u8], offset: u64) -> Result<usize, Error> {
let size = buf.len();
let mut read: usize = 0;
while read < size {
let cur_offset = offset + read as u64;
if let Some(chunk) = self.index.chunk_from_offset(cur_offset) {
// chunk indices retrieved from chunk_from_offset always resolve to Some(_)
let info = self.index.chunk_info(chunk.0).unwrap();
// will never be None, see AsyncChunkCacher
let data = self.cache.access(info.digest, &self.cacher).await?.unwrap();
let want_bytes = ((info.range.end - cur_offset) as usize).min(size - read);
let slice = &mut buf[read..(read + want_bytes)];
let intra_chunk = chunk.1 as usize;
slice.copy_from_slice(&data[intra_chunk..(intra_chunk + want_bytes)]);
read += want_bytes;
} else {
// EOF
break;
}
}
Ok(read)
}
}
impl<I: IndexFile + Send + Sync + 'static, R: AsyncReadChunk + Send + Sync + 'static>
CachedChunkReader<I, R>
{
/// Returns a SeekableCachedChunkReader based on this instance, which implements AsyncSeek and
/// AsyncRead for use in interfaces which require that. Direct use of read_at is preferred
/// otherwise.
pub fn seekable(self) -> SeekableCachedChunkReader<I, R> {
SeekableCachedChunkReader {
index_bytes: self.index.index_bytes(),
reader: Arc::new(self),
position: 0,
read_future: None,
}
}
}
pub struct SeekableCachedChunkReader<
I: IndexFile + Send + Sync + 'static,
R: AsyncReadChunk + Send + Sync + 'static,
> {
reader: Arc<CachedChunkReader<I, R>>,
index_bytes: u64,
position: u64,
read_future: Option<Pin<Box<dyn Future<Output = Result<(Vec<u8>, usize), Error>> + Send>>>,
}
impl<I, R> AsyncSeek for SeekableCachedChunkReader<I, R>
where
I: IndexFile + Send + Sync + 'static,
R: AsyncReadChunk + Send + Sync + 'static,
{
fn start_seek(self: Pin<&mut Self>, pos: SeekFrom) -> tokio::io::Result<()> {
let this = Pin::get_mut(self);
let seek_to_pos = match pos {
SeekFrom::Start(offset) => offset as i64,
SeekFrom::End(offset) => this.index_bytes as i64 + offset,
SeekFrom::Current(offset) => this.position as i64 + offset,
};
if seek_to_pos < 0 {
return Err(io_format_err!("cannot seek to negative values"));
} else if seek_to_pos > this.index_bytes as i64 {
this.position = this.index_bytes;
} else {
this.position = seek_to_pos as u64;
}
Ok(())
}
fn poll_complete(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<tokio::io::Result<u64>> {
Poll::Ready(Ok(self.position))
}
}
impl<I, R> AsyncRead for SeekableCachedChunkReader<I, R>
where
I: IndexFile + Send + Sync + 'static,
R: AsyncReadChunk + Send + Sync + 'static,
{
fn poll_read(
self: Pin<&mut Self>,
cx: &mut Context,
buf: &mut ReadBuf,
) -> Poll<tokio::io::Result<()>> {
let this = Pin::get_mut(self);
let offset = this.position;
let wanted = buf.capacity();
let reader = Arc::clone(&this.reader);
let fut = this.read_future.get_or_insert_with(|| {
Box::pin(async move {
let mut read_buf = vec![0u8; wanted];
let read = reader.read_at(&mut read_buf[..wanted], offset).await?;
Ok((read_buf, read))
})
});
let ret = match ready!(fut.as_mut().poll(cx)) {
Ok((read_buf, read)) => {
buf.put_slice(&read_buf[..read]);
this.position += read as u64;
Ok(())
}
Err(err) => Err(io_err_other(err)),
};
// future completed, drop
this.read_future = None;
Poll::Ready(ret)
}
}

4
src/backup/mod.rs
View File

@ -75,6 +75,7 @@ pub use pbs_datastore::prune::*;
pub use pbs_datastore::store_progress::StoreProgress;
pub use pbs_datastore::cached_chunk_reader::*;
pub use pbs_datastore::dynamic_index::*;
pub use pbs_datastore::fixed_index;
pub use pbs_datastore::fixed_index::*;
@ -91,9 +92,6 @@ pub use datastore::*;
mod verify;
pub use verify::*;
mod cached_chunk_reader;
pub use cached_chunk_reader::*;
pub struct BackupLockGuard(std::fs::File);
/// Open or create a lock file owned by user "backup" and lock it.

3
src/bin/proxmox_backup_client/mount.rs
View File

@ -20,12 +20,11 @@ use proxmox::tools::fd::Fd;
use pbs_datastore::{BackupDir, BackupGroup, CryptConfig, load_and_decrypt_key};
use pbs_datastore::index::IndexFile;
use pbs_datastore::dynamic_index::BufferedDynamicReader;
use pbs_datastore::cached_chunk_reader::CachedChunkReader;
use pbs_client::tools::key_source::get_encryption_key_password;
use pbs_client::{BackupReader, RemoteChunkReader};
use pbs_tools::json::required_string_param;
use proxmox_backup::backup::CachedChunkReader;
use crate::{
REPO_URL_SCHEMA,
extract_repository_from_value,