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

use new Mmap helper for dynamic index 

Signed-off-by: Wolfgang Bumiller <w.bumiller@proxmox.com>
This commit is contained in:
Wolfgang Bumiller 2020-06-12 13:57:56 +02:00
parent 3136792c95
commit 57e50fb906
2 changed files with 46 additions and 94 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

139
src/backup/dynamic_index.rs
View File

@ -1,4 +1,3 @@
use std::convert::TryInto;
use std::fs::File; use std::fs::File;
use std::io::{BufWriter, Seek, SeekFrom, Write}; use std::io::{BufWriter, Seek, SeekFrom, Write};
use std::ops::Range; use std::ops::Range;
@ -11,6 +10,7 @@ use anyhow::{bail, format_err, Error};
use proxmox::tools::io::ReadExt; use proxmox::tools::io::ReadExt;
use proxmox::tools::uuid::Uuid; use proxmox::tools::uuid::Uuid;
use proxmox::tools::vec; use proxmox::tools::vec;
use proxmox::tools::mmap::Mmap;
use super::chunk_stat::ChunkStat; use super::chunk_stat::ChunkStat;
use super::chunk_store::ChunkStore; use super::chunk_store::ChunkStore;
@ -38,34 +38,27 @@ proxmox::static_assert_size!(DynamicIndexHeader, 4096);
// pub data: DynamicIndexHeaderData, // pub data: DynamicIndexHeaderData,
// } // }
#[derive(Clone, Debug)]
#[repr(C)]
pub struct DynamicEntry {
end: u64,
digest: [u8; 32],
}
pub struct DynamicIndexReader { pub struct DynamicIndexReader {
_file: File, _file: File,
pub size: usize, pub size: usize,
index: *const u8, index: Mmap<DynamicEntry>,
index_entries: usize,
pub uuid: [u8; 16], pub uuid: [u8; 16],
pub ctime: u64, pub ctime: u64,
pub index_csum: [u8; 32], pub index_csum: [u8; 32],
} }
// `index` is mmap()ed which cannot be thread-local so should be sendable
// FIXME: Introduce an mmap wrapper type for this?
unsafe impl Send for DynamicIndexReader {}
unsafe impl Sync for DynamicIndexReader {}
impl Drop for DynamicIndexReader {
fn drop(&mut self) {
if let Err(err) = self.unmap() {
eprintln!("Unable to unmap dynamic index - {}", err);
}
}
}
impl DynamicIndexReader { impl DynamicIndexReader {
pub fn open(path: &Path) -> Result<Self, Error> { pub fn open(path: &Path) -> Result<Self, Error> {
File::open(path) File::open(path)
.map_err(Error::from) .map_err(Error::from)
.and_then(|file| Self::new(file)) .and_then(Self::new)
.map_err(|err| format_err!("Unable to open dynamic index {:?} - {}", path, err)) .map_err(|err| format_err!("Unable to open dynamic index {:?} - {}", path, err))
} }
@ -76,6 +69,7 @@ impl DynamicIndexReader {
bail!("unable to get shared lock - {}", err); bail!("unable to get shared lock - {}", err);
} }
// FIXME: This is NOT OUR job! Check the callers of this method and remove this!
file.seek(SeekFrom::Start(0))?; file.seek(SeekFrom::Start(0))?;
let header_size = std::mem::size_of::<DynamicIndexHeader>(); let header_size = std::mem::size_of::<DynamicIndexHeader>();
@ -95,126 +89,85 @@ impl DynamicIndexReader {
let size = stat.st_size as usize; let size = stat.st_size as usize;
let index_size = size - header_size; let index_size = size - header_size;
if (index_size % 40) != 0 { let index_count = index_size / 40;
if index_count * 40 != index_size {
bail!("got unexpected file size"); bail!("got unexpected file size");
} }
let data = unsafe { let index = unsafe {
nix::sys::mman::mmap( Mmap::map_fd(
std::ptr::null_mut(), rawfd,
index_size, header_size as u64,
index_count,
nix::sys::mman::ProtFlags::PROT_READ, nix::sys::mman::ProtFlags::PROT_READ,
nix::sys::mman::MapFlags::MAP_PRIVATE, nix::sys::mman::MapFlags::MAP_PRIVATE,
rawfd, )?
header_size as i64, };
)
}? as *const u8;
Ok(Self { Ok(Self {
_file: file, _file: file,
size, size,
index: data, index,
index_entries: index_size / 40,
ctime, ctime,
uuid: header.uuid, uuid: header.uuid,
index_csum: header.index_csum, index_csum: header.index_csum,
}) })
} }
fn unmap(&mut self) -> Result<(), Error> {
if self.index == std::ptr::null_mut() {
return Ok(());
}
if let Err(err) = unsafe {
nix::sys::mman::munmap(self.index as *mut std::ffi::c_void, self.index_entries * 40)
} {
bail!("unmap dynamic index failed - {}", err);
}
self.index = std::ptr::null_mut();
Ok(())
}
#[allow(clippy::cast_ptr_alignment)] #[allow(clippy::cast_ptr_alignment)]
pub fn chunk_info(&self, pos: usize) -> Result<ChunkReadInfo, Error> { pub fn chunk_info(&self, pos: usize) -> Result<ChunkReadInfo, Error> {
if pos >= self.index_entries { if pos >= self.index.len() {
bail!("chunk index out of range"); bail!("chunk index out of range");
} }
let start = if pos == 0 { let start = if pos == 0 {
0 0
} else { } else {
unsafe { *(self.index.add((pos - 1) * 40) as *const u64) } u64::from_le(self.index[pos - 1].end)
}; };
let end = unsafe { *(self.index.add(pos * 40) as *const u64) }; let end = u64::from_le(self.index[pos].end);
let mut digest = std::mem::MaybeUninit::<[u8; 32]>::uninit();
unsafe {
std::ptr::copy_nonoverlapping(
self.index.add(pos * 40 + 8),
(*digest.as_mut_ptr()).as_mut_ptr(),
32,
);
}
Ok(ChunkReadInfo { Ok(ChunkReadInfo {
range: start..end, range: start..end,
digest: unsafe { digest.assume_init() }, digest: self.index[pos].digest.clone(),
}) })
} }
#[inline] #[inline]
#[allow(clippy::cast_ptr_alignment)] #[allow(clippy::cast_ptr_alignment)]
fn chunk_end(&self, pos: usize) -> u64 { fn chunk_end(&self, pos: usize) -> u64 {
if pos >= self.index_entries { if pos >= self.index.len() {
panic!("chunk index out of range"); panic!("chunk index out of range");
} }
unsafe { *(self.index.add(pos * 40) as *const u64) } u64::from_le(self.index[pos].end)
} }
#[inline] #[inline]
fn chunk_digest(&self, pos: usize) -> &[u8; 32] { fn chunk_digest(&self, pos: usize) -> &[u8; 32] {
if pos >= self.index_entries { if pos >= self.index.len() {
panic!("chunk index out of range"); panic!("chunk index out of range");
} }
let slice = unsafe { std::slice::from_raw_parts(self.index.add(pos * 40 + 8), 32) }; &self.index[pos].digest
slice.try_into().unwrap()
} }
/// Compute checksum and data size /// Compute checksum and data size
pub fn compute_csum(&self) -> ([u8; 32], u64) { pub fn compute_csum(&self) -> ([u8; 32], u64) {
let mut csum = openssl::sha::Sha256::new(); let mut csum = openssl::sha::Sha256::new();
let mut chunk_end = 0; for entry in &self.index {
for pos in 0..self.index_entries { csum.update(&entry.end.to_ne_bytes());
chunk_end = self.chunk_end(pos); csum.update(&entry.digest);
let digest = self.chunk_digest(pos);
csum.update(&chunk_end.to_le_bytes());
csum.update(digest);
} }
let csum = csum.finish(); let csum = csum.finish();
(csum, chunk_end) (
csum,
self.index
.last()
.map(|entry| entry.end)
.unwrap_or(0))
} }
/* // TODO: can we use std::slice::binary_search with Mmap now?
pub fn dump_pxar(&self, mut writer: Box<dyn Write>) -> Result<(), Error> {
for pos in 0..self.index_entries {
let _end = self.chunk_end(pos);
let digest = self.chunk_digest(pos);
//println!("Dump {:08x}", end );
let chunk = self.store.read_chunk(digest)?;
// fimxe: handle encrypted chunks
let data = chunk.decode(None)?;
writer.write_all(&data)?;
}
Ok(())
}
*/
fn binary_search( fn binary_search(
&self, &self,
start_idx: usize, start_idx: usize,
@ -243,11 +196,11 @@ impl DynamicIndexReader {
impl IndexFile for DynamicIndexReader { impl IndexFile for DynamicIndexReader {
fn index_count(&self) -> usize { fn index_count(&self) -> usize {
self.index_entries self.index.len()
} }
fn index_digest(&self, pos: usize) -> Option<&[u8; 32]> { fn index_digest(&self, pos: usize) -> Option<&[u8; 32]> {
if pos >= self.index_entries { if pos >= self.index.len() {
None None
} else { } else {
Some(unsafe { std::mem::transmute(self.chunk_digest(pos).as_ptr()) }) Some(unsafe { std::mem::transmute(self.chunk_digest(pos).as_ptr()) })
@ -255,10 +208,10 @@ impl IndexFile for DynamicIndexReader {
} }
fn index_bytes(&self) -> u64 { fn index_bytes(&self) -> u64 {
if self.index_entries == 0 { if self.index.is_empty() {
0 0
} else { } else {
self.chunk_end((self.index_entries - 1) as usize) self.chunk_end(self.index.len() - 1)
} }
} }
} }
@ -309,7 +262,7 @@ impl<'a, S: ReadChunk> crate::tools::lru_cache::Cacher<usize, CachedChunk> for C
impl<S: ReadChunk> BufferedDynamicReader<S> { impl<S: ReadChunk> BufferedDynamicReader<S> {
pub fn new(index: DynamicIndexReader, store: S) -> Self { pub fn new(index: DynamicIndexReader, store: S) -> Self {
let archive_size = index.chunk_end(index.index_entries - 1); let archive_size = index.index_bytes();
Self { Self {
store, store,
index, index,
@ -359,7 +312,7 @@ impl<S: ReadChunk> crate::tools::BufferedRead for BufferedDynamicReader<S> {
// optimization for sequential read // optimization for sequential read
if buffer_len > 0 if buffer_len > 0
&& ((self.buffered_chunk_idx + 1) < index.index_entries) && ((self.buffered_chunk_idx + 1) < index.index.len())
&& (offset >= (self.buffered_chunk_start + (self.read_buffer.len() as u64))) && (offset >= (self.buffered_chunk_start + (self.read_buffer.len() as u64)))
{ {
let next_idx = self.buffered_chunk_idx + 1; let next_idx = self.buffered_chunk_idx + 1;
@ -375,7 +328,7 @@ impl<S: ReadChunk> crate::tools::BufferedRead for BufferedDynamicReader<S> {
|| (offset < self.buffered_chunk_start) || (offset < self.buffered_chunk_start)
|| (offset >= (self.buffered_chunk_start + (self.read_buffer.len() as u64))) || (offset >= (self.buffered_chunk_start + (self.read_buffer.len() as u64)))
{ {
let end_idx = index.index_entries - 1; let end_idx = index.index.len() - 1;
let end = index.chunk_end(end_idx); let end = index.chunk_end(end_idx);
let idx = index.binary_search(0, 0, end_idx, end, offset)?; let idx = index.binary_search(0, 0, end_idx, end, offset)?;
self.buffer_chunk(idx)?; self.buffer_chunk(idx)?;

1
src/tools.rs
View File

@ -622,4 +622,3 @@ pub fn epoch_now_f64() -> Result<f64, SystemTimeError> {
pub fn epoch_now_u64() -> Result<u64, SystemTimeError> { pub fn epoch_now_u64() -> Result<u64, SystemTimeError> {
Ok(epoch_now()?.as_secs()) Ok(epoch_now()?.as_secs())
} }