moving more code to pbs-datastore

prune and fixed/dynamic index

Signed-off-by: Wolfgang Bumiller <w.bumiller@proxmox.com>

pbs-datastore/src/dynamic_index.rs

@@ -0,0 +1,508 @@
+use std::fs::File;
+use std::io::{BufWriter, Seek, SeekFrom, Write};
+use std::os::unix::io::AsRawFd;
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
+
+use anyhow::{bail, format_err, Error};
+
+use proxmox::tools::io::ReadExt;
+use proxmox::tools::uuid::Uuid;
+use proxmox::tools::mmap::Mmap;
+
+use pbs_tools::process_locker::ProcessLockSharedGuard;
+
+use crate::Chunker;
+use crate::chunk_stat::ChunkStat;
+use crate::chunk_store::ChunkStore;
+use crate::data_blob::{DataBlob, DataChunkBuilder};
+use crate::file_formats;
+use crate::index::{IndexFile, ChunkReadInfo};
+
+/// Header format definition for dynamic index files (`.dixd`)
+#[repr(C)]
+pub struct DynamicIndexHeader {
+    pub magic: [u8; 8],
+    pub uuid: [u8; 16],
+    pub ctime: i64,
+    /// Sha256 over the index ``SHA256(offset1||digest1||offset2||digest2||...)``
+    pub index_csum: [u8; 32],
+    reserved: [u8; 4032], // overall size is one page (4096 bytes)
+}
+proxmox::static_assert_size!(DynamicIndexHeader, 4096);
+// TODO: Once non-Copy unions are stabilized, use:
+// union DynamicIndexHeader {
+//     reserved: [u8; 4096],
+//     pub data: DynamicIndexHeaderData,
+// }
+
+impl DynamicIndexHeader {
+    /// Convenience method to allocate a zero-initialized header struct.
+    pub fn zeroed() -> Box<Self> {
+        unsafe {
+            Box::from_raw(std::alloc::alloc_zeroed(std::alloc::Layout::new::<Self>()) as *mut Self)
+        }
+    }
+
+    pub fn as_bytes(&self) -> &[u8] {
+        unsafe {
+            std::slice::from_raw_parts(
+                self as *const Self as *const u8,
+                std::mem::size_of::<Self>(),
+            )
+        }
+    }
+}
+
+#[derive(Clone, Debug)]
+#[repr(C)]
+pub struct DynamicEntry {
+    end_le: u64,
+    digest: [u8; 32],
+}
+
+impl DynamicEntry {
+    #[inline]
+    pub fn end(&self) -> u64 {
+        u64::from_le(self.end_le)
+    }
+}
+
+pub struct DynamicIndexReader {
+    _file: File,
+    pub size: usize,
+    index: Mmap<DynamicEntry>,
+    pub uuid: [u8; 16],
+    pub ctime: i64,
+    pub index_csum: [u8; 32],
+}
+
+impl DynamicIndexReader {
+    pub fn open(path: &Path) -> Result<Self, Error> {
+        File::open(path)
+            .map_err(Error::from)
+            .and_then(Self::new)
+            .map_err(|err| format_err!("Unable to open dynamic index {:?} - {}", path, err))
+    }
+
+    pub fn index(&self) -> &[DynamicEntry] {
+        &self.index
+    }
+
+    pub fn new(mut file: std::fs::File) -> Result<Self, Error> {
+        // FIXME: This is NOT OUR job! Check the callers of this method and remove this!
+        file.seek(SeekFrom::Start(0))?;
+
+        let header_size = std::mem::size_of::<DynamicIndexHeader>();
+
+        let rawfd = file.as_raw_fd();
+        let stat = match nix::sys::stat::fstat(rawfd) {
+            Ok(stat) => stat,
+            Err(err) => bail!("fstat failed - {}", err),
+        };
+
+        let size = stat.st_size as usize;
+
+        if size < header_size {
+            bail!("index too small ({})", stat.st_size);
+        }
+
+        let header: Box<DynamicIndexHeader> = unsafe { file.read_host_value_boxed()? };
+
+        if header.magic != file_formats::DYNAMIC_SIZED_CHUNK_INDEX_1_0 {
+            bail!("got unknown magic number");
+        }
+
+        let ctime = proxmox::tools::time::epoch_i64();
+
+        let index_size = stat.st_size as usize - header_size;
+        let index_count = index_size / 40;
+        if index_count * 40 != index_size {
+            bail!("got unexpected file size");
+        }
+
+        let index = unsafe {
+            Mmap::map_fd(
+                rawfd,
+                header_size as u64,
+                index_count,
+                nix::sys::mman::ProtFlags::PROT_READ,
+                nix::sys::mman::MapFlags::MAP_PRIVATE,
+            )?
+        };
+
+        Ok(Self {
+            _file: file,
+            size,
+            index,
+            ctime,
+            uuid: header.uuid,
+            index_csum: header.index_csum,
+        })
+    }
+
+    #[inline]
+    #[allow(clippy::cast_ptr_alignment)]
+    pub fn chunk_end(&self, pos: usize) -> u64 {
+        if pos >= self.index.len() {
+            panic!("chunk index out of range");
+        }
+        self.index[pos].end()
+    }
+
+    #[inline]
+    fn chunk_digest(&self, pos: usize) -> &[u8; 32] {
+        if pos >= self.index.len() {
+            panic!("chunk index out of range");
+        }
+        &self.index[pos].digest
+    }
+
+    pub fn binary_search(
+        &self,
+        start_idx: usize,
+        start: u64,
+        end_idx: usize,
+        end: u64,
+        offset: u64,
+    ) -> Result<usize, Error> {
+        if (offset >= end) || (offset < start) {
+            bail!("offset out of range");
+        }
+
+        if end_idx == start_idx {
+            return Ok(start_idx); // found
+        }
+        let middle_idx = (start_idx + end_idx) / 2;
+        let middle_end = self.chunk_end(middle_idx);
+
+        if offset < middle_end {
+            self.binary_search(start_idx, start, middle_idx, middle_end, offset)
+        } else {
+            self.binary_search(middle_idx + 1, middle_end, end_idx, end, offset)
+        }
+    }
+}
+
+impl IndexFile for DynamicIndexReader {
+    fn index_count(&self) -> usize {
+        self.index.len()
+    }
+
+    fn index_digest(&self, pos: usize) -> Option<&[u8; 32]> {
+        if pos >= self.index.len() {
+            None
+        } else {
+            Some(unsafe { &*(self.chunk_digest(pos).as_ptr() as *const [u8; 32]) })
+        }
+    }
+
+    fn index_bytes(&self) -> u64 {
+        if self.index.is_empty() {
+            0
+        } else {
+            self.chunk_end(self.index.len() - 1)
+        }
+    }
+
+    fn compute_csum(&self) -> ([u8; 32], u64) {
+        let mut csum = openssl::sha::Sha256::new();
+        let mut chunk_end = 0;
+        for pos in 0..self.index_count() {
+            let info = self.chunk_info(pos).unwrap();
+            chunk_end = info.range.end;
+            csum.update(&chunk_end.to_le_bytes());
+            csum.update(&info.digest);
+        }
+        let csum = csum.finish();
+        (csum, chunk_end)
+    }
+
+    fn chunk_info(&self, pos: usize) -> Option<ChunkReadInfo> {
+        if pos >= self.index.len() {
+            return None;
+        }
+        let start = if pos == 0 { 0 } else { self.index[pos - 1].end() };
+
+        let end = self.index[pos].end();
+
+        Some(ChunkReadInfo {
+            range: start..end,
+            digest: self.index[pos].digest,
+        })
+    }
+
+    fn index_ctime(&self) -> i64 {
+        self.ctime
+    }
+
+    fn index_size(&self) -> usize {
+        self.size as usize
+    }
+
+    fn chunk_from_offset(&self, offset: u64) -> Option<(usize, u64)> {
+        let end_idx = self.index.len() - 1;
+        let end = self.chunk_end(end_idx);
+        let found_idx = self.binary_search(0, 0, end_idx, end, offset);
+        let found_idx = match found_idx {
+            Ok(i) => i,
+            Err(_) => return None
+        };
+
+        let found_start = if found_idx == 0 {
+            0
+        } else {
+            self.chunk_end(found_idx - 1)
+        };
+
+        Some((found_idx, offset - found_start))
+    }
+}
+
+/// Create dynamic index files (`.dixd`)
+pub struct DynamicIndexWriter {
+    store: Arc<ChunkStore>,
+    _lock: ProcessLockSharedGuard,
+    writer: BufWriter<File>,
+    closed: bool,
+    filename: PathBuf,
+    tmp_filename: PathBuf,
+    csum: Option<openssl::sha::Sha256>,
+    pub uuid: [u8; 16],
+    pub ctime: i64,
+}
+
+impl Drop for DynamicIndexWriter {
+    fn drop(&mut self) {
+        let _ = std::fs::remove_file(&self.tmp_filename); // ignore errors
+    }
+}
+
+impl DynamicIndexWriter {
+    pub fn create(store: Arc<ChunkStore>, path: &Path) -> Result<Self, Error> {
+        let shared_lock = store.try_shared_lock()?;
+
+        let full_path = store.relative_path(path);
+        let mut tmp_path = full_path.clone();
+        tmp_path.set_extension("tmp_didx");
+
+        let file = std::fs::OpenOptions::new()
+            .create(true)
+            .truncate(true)
+            .read(true)
+            .write(true)
+            .open(&tmp_path)?;
+
+        let mut writer = BufWriter::with_capacity(1024 * 1024, file);
+
+        let ctime = proxmox::tools::time::epoch_i64();
+
+        let uuid = Uuid::generate();
+
+        let mut header = DynamicIndexHeader::zeroed();
+        header.magic = file_formats::DYNAMIC_SIZED_CHUNK_INDEX_1_0;
+        header.ctime = i64::to_le(ctime);
+        header.uuid = *uuid.as_bytes();
+        // header.index_csum = [0u8; 32];
+        writer.write_all(header.as_bytes())?;
+
+        let csum = Some(openssl::sha::Sha256::new());
+
+        Ok(Self {
+            store,
+            _lock: shared_lock,
+            writer,
+            closed: false,
+            filename: full_path,
+            tmp_filename: tmp_path,
+            ctime,
+            uuid: *uuid.as_bytes(),
+            csum,
+        })
+    }
+
+    // fixme: use add_chunk instead?
+    pub fn insert_chunk(&self, chunk: &DataBlob, digest: &[u8; 32]) -> Result<(bool, u64), Error> {
+        self.store.insert_chunk(chunk, digest)
+    }
+
+    pub fn close(&mut self) -> Result<[u8; 32], Error> {
+        if self.closed {
+            bail!(
+                "cannot close already closed archive index file {:?}",
+                self.filename
+            );
+        }
+
+        self.closed = true;
+
+        self.writer.flush()?;
+
+        let csum_offset = proxmox::offsetof!(DynamicIndexHeader, index_csum);
+        self.writer.seek(SeekFrom::Start(csum_offset as u64))?;
+
+        let csum = self.csum.take().unwrap();
+        let index_csum = csum.finish();
+
+        self.writer.write_all(&index_csum)?;
+        self.writer.flush()?;
+
+        if let Err(err) = std::fs::rename(&self.tmp_filename, &self.filename) {
+            bail!("Atomic rename file {:?} failed - {}", self.filename, err);
+        }
+
+        Ok(index_csum)
+    }
+
+    // fixme: rename to add_digest
+    pub fn add_chunk(&mut self, offset: u64, digest: &[u8; 32]) -> Result<(), Error> {
+        if self.closed {
+            bail!(
+                "cannot write to closed dynamic index file {:?}",
+                self.filename
+            );
+        }
+
+        let offset_le: &[u8; 8] = unsafe { &std::mem::transmute::<u64, [u8; 8]>(offset.to_le()) };
+
+        if let Some(ref mut csum) = self.csum {
+            csum.update(offset_le);
+            csum.update(digest);
+        }
+
+        self.writer.write_all(offset_le)?;
+        self.writer.write_all(digest)?;
+        Ok(())
+    }
+}
+
+/// Writer which splits a binary stream into dynamic sized chunks
+///
+/// And store the resulting chunk list into the index file.
+pub struct DynamicChunkWriter {
+    index: DynamicIndexWriter,
+    closed: bool,
+    chunker: Chunker,
+    stat: ChunkStat,
+    chunk_offset: usize,
+    last_chunk: usize,
+    chunk_buffer: Vec<u8>,
+}
+
+impl DynamicChunkWriter {
+    pub fn new(index: DynamicIndexWriter, chunk_size: usize) -> Self {
+        Self {
+            index,
+            closed: false,
+            chunker: Chunker::new(chunk_size),
+            stat: ChunkStat::new(0),
+            chunk_offset: 0,
+            last_chunk: 0,
+            chunk_buffer: Vec::with_capacity(chunk_size * 4),
+        }
+    }
+
+    pub fn stat(&self) -> &ChunkStat {
+        &self.stat
+    }
+
+    pub fn close(&mut self) -> Result<(), Error> {
+        if self.closed {
+            return Ok(());
+        }
+
+        self.closed = true;
+
+        self.write_chunk_buffer()?;
+
+        self.index.close()?;
+
+        self.stat.size = self.chunk_offset as u64;
+
+        // add size of index file
+        self.stat.size +=
+            (self.stat.chunk_count * 40 + std::mem::size_of::<DynamicIndexHeader>()) as u64;
+
+        Ok(())
+    }
+
+    fn write_chunk_buffer(&mut self) -> Result<(), Error> {
+        let chunk_size = self.chunk_buffer.len();
+
+        if chunk_size == 0 {
+            return Ok(());
+        }
+
+        let expected_chunk_size = self.chunk_offset - self.last_chunk;
+        if expected_chunk_size != self.chunk_buffer.len() {
+            bail!("wrong chunk size {} != {}", expected_chunk_size, chunk_size);
+        }
+
+        self.stat.chunk_count += 1;
+
+        self.last_chunk = self.chunk_offset;
+
+        let (chunk, digest) = DataChunkBuilder::new(&self.chunk_buffer)
+            .compress(true)
+            .build()?;
+
+        match self.index.insert_chunk(&chunk, &digest) {
+            Ok((is_duplicate, compressed_size)) => {
+                self.stat.compressed_size += compressed_size;
+                if is_duplicate {
+                    self.stat.duplicate_chunks += 1;
+                } else {
+                    self.stat.disk_size += compressed_size;
+                }
+
+                println!(
+                    "ADD CHUNK {:016x} {} {}% {} {}",
+                    self.chunk_offset,
+                    chunk_size,
+                    (compressed_size * 100) / (chunk_size as u64),
+                    is_duplicate,
+                    proxmox::tools::digest_to_hex(&digest)
+                );
+                self.index.add_chunk(self.chunk_offset as u64, &digest)?;
+                self.chunk_buffer.truncate(0);
+                Ok(())
+            }
+            Err(err) => {
+                self.chunk_buffer.truncate(0);
+                Err(err)
+            }
+        }
+    }
+}
+
+impl Write for DynamicChunkWriter {
+    fn write(&mut self, data: &[u8]) -> std::result::Result<usize, std::io::Error> {
+        let chunker = &mut self.chunker;
+
+        let pos = chunker.scan(data);
+
+        if pos > 0 {
+            self.chunk_buffer.extend_from_slice(&data[0..pos]);
+            self.chunk_offset += pos;
+
+            if let Err(err) = self.write_chunk_buffer() {
+                return Err(std::io::Error::new(
+                    std::io::ErrorKind::Other,
+                    err.to_string(),
+                ));
+            }
+            Ok(pos)
+        } else {
+            self.chunk_offset += data.len();
+            self.chunk_buffer.extend_from_slice(data);
+            Ok(data.len())
+        }
+    }
+
+    fn flush(&mut self) -> std::result::Result<(), std::io::Error> {
+        Err(std::io::Error::new(
+            std::io::ErrorKind::Other,
+            "please use close() instead of flush()",
+        ))
+    }
+}

pbs-datastore/src/fixed_index.rs

@@ -0,0 +1,473 @@
+use std::fs::File;
+use std::io::Write;
+use std::os::unix::io::AsRawFd;
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
+use std::io::{Seek, SeekFrom};
+
+use anyhow::{bail, format_err, Error};
+
+use pbs_tools::process_locker::ProcessLockSharedGuard;
+
+use proxmox::tools::io::ReadExt;
+use proxmox::tools::Uuid;
+
+use crate::chunk_stat::ChunkStat;
+use crate::chunk_store::ChunkStore;
+use crate::data_blob::ChunkInfo;
+use crate::file_formats;
+use crate::index::{ChunkReadInfo, IndexFile};
+
+/// Header format definition for fixed index files (`.fidx`)
+#[repr(C)]
+pub struct FixedIndexHeader {
+    pub magic: [u8; 8],
+    pub uuid: [u8; 16],
+    pub ctime: i64,
+    /// Sha256 over the index ``SHA256(digest1||digest2||...)``
+    pub index_csum: [u8; 32],
+    pub size: u64,
+    pub chunk_size: u64,
+    reserved: [u8; 4016], // overall size is one page (4096 bytes)
+}
+proxmox::static_assert_size!(FixedIndexHeader, 4096);
+
+// split image into fixed size chunks
+
+pub struct FixedIndexReader {
+    _file: File,
+    pub chunk_size: usize,
+    pub size: u64,
+    index_length: usize,
+    index: *mut u8,
+    pub uuid: [u8; 16],
+    pub ctime: i64,
+    pub index_csum: [u8; 32],
+}
+
+// `index` is mmap()ed which cannot be thread-local so should be sendable
+unsafe impl Send for FixedIndexReader {}
+unsafe impl Sync for FixedIndexReader {}
+
+impl Drop for FixedIndexReader {
+    fn drop(&mut self) {
+        if let Err(err) = self.unmap() {
+            eprintln!("Unable to unmap file - {}", err);
+        }
+    }
+}
+
+impl FixedIndexReader {
+    pub fn open(path: &Path) -> Result<Self, Error> {
+        File::open(path)
+            .map_err(Error::from)
+            .and_then(Self::new)
+            .map_err(|err| format_err!("Unable to open fixed index {:?} - {}", path, err))
+    }
+
+    pub fn new(mut file: std::fs::File) -> Result<Self, Error> {
+        file.seek(SeekFrom::Start(0))?;
+
+        let header_size = std::mem::size_of::<FixedIndexHeader>();
+
+        let stat = match nix::sys::stat::fstat(file.as_raw_fd()) {
+            Ok(stat) => stat,
+            Err(err) => bail!("fstat failed - {}", err),
+        };
+
+        let size = stat.st_size as usize;
+
+        if size < header_size {
+            bail!("index too small ({})", stat.st_size);
+        }
+
+        let header: Box<FixedIndexHeader> = unsafe { file.read_host_value_boxed()? };
+
+        if header.magic != file_formats::FIXED_SIZED_CHUNK_INDEX_1_0 {
+            bail!("got unknown magic number");
+        }
+
+        let size = u64::from_le(header.size);
+        let ctime = i64::from_le(header.ctime);
+        let chunk_size = u64::from_le(header.chunk_size);
+
+        let index_length = ((size + chunk_size - 1) / chunk_size) as usize;
+        let index_size = index_length * 32;
+
+        let expected_index_size = (stat.st_size as usize) - header_size;
+        if index_size != expected_index_size {
+            bail!(
+                "got unexpected file size ({} != {})",
+                index_size,
+                expected_index_size
+            );
+        }
+
+        let data = unsafe {
+            nix::sys::mman::mmap(
+                std::ptr::null_mut(),
+                index_size,
+                nix::sys::mman::ProtFlags::PROT_READ,
+                nix::sys::mman::MapFlags::MAP_PRIVATE,
+                file.as_raw_fd(),
+                header_size as i64,
+            )
+        }? as *mut u8;
+
+        Ok(Self {
+            _file: file,
+            chunk_size: chunk_size as usize,
+            size,
+            index_length,
+            index: data,
+            ctime,
+            uuid: header.uuid,
+            index_csum: header.index_csum,
+        })
+    }
+
+    fn unmap(&mut self) -> Result<(), Error> {
+        if self.index.is_null() {
+            return Ok(());
+        }
+
+        let index_size = self.index_length * 32;
+
+        if let Err(err) =
+            unsafe { nix::sys::mman::munmap(self.index as *mut std::ffi::c_void, index_size) }
+        {
+            bail!("unmap file failed - {}", err);
+        }
+
+        self.index = std::ptr::null_mut();
+
+        Ok(())
+    }
+
+    pub fn print_info(&self) {
+        println!("Size: {}", self.size);
+        println!("ChunkSize: {}", self.chunk_size);
+
+        let mut ctime_str = self.ctime.to_string();
+        if let Ok(s) = proxmox::tools::time::strftime_local("%c", self.ctime) {
+            ctime_str = s;
+        }
+
+        println!("CTime: {}", ctime_str);
+        println!("UUID: {:?}", self.uuid);
+    }
+}
+
+impl IndexFile for FixedIndexReader {
+    fn index_count(&self) -> usize {
+        self.index_length
+    }
+
+    fn index_digest(&self, pos: usize) -> Option<&[u8; 32]> {
+        if pos >= self.index_length {
+            None
+        } else {
+            Some(unsafe { &*(self.index.add(pos * 32) as *const [u8; 32]) })
+        }
+    }
+
+    fn index_bytes(&self) -> u64 {
+        self.size
+    }
+
+    fn chunk_info(&self, pos: usize) -> Option<ChunkReadInfo> {
+        if pos >= self.index_length {
+            return None;
+        }
+
+        let start = (pos * self.chunk_size) as u64;
+        let mut end = start + self.chunk_size as u64;
+
+        if end > self.size {
+            end = self.size;
+        }
+
+        let digest = self.index_digest(pos).unwrap();
+        Some(ChunkReadInfo {
+            range: start..end,
+            digest: *digest,
+        })
+    }
+
+    fn index_ctime(&self) -> i64 {
+        self.ctime
+    }
+
+    fn index_size(&self) -> usize {
+        self.size as usize
+    }
+
+    fn compute_csum(&self) -> ([u8; 32], u64) {
+        let mut csum = openssl::sha::Sha256::new();
+        let mut chunk_end = 0;
+        for pos in 0..self.index_count() {
+            let info = self.chunk_info(pos).unwrap();
+            chunk_end = info.range.end;
+            csum.update(&info.digest);
+        }
+        let csum = csum.finish();
+
+        (csum, chunk_end)
+    }
+
+    fn chunk_from_offset(&self, offset: u64) -> Option<(usize, u64)> {
+        if offset >= self.size {
+            return None;
+        }
+
+        Some((
+            (offset / self.chunk_size as u64) as usize,
+            offset & (self.chunk_size - 1) as u64, // fast modulo, valid for 2^x chunk_size
+        ))
+    }
+}
+
+pub struct FixedIndexWriter {
+    store: Arc<ChunkStore>,
+    file: File,
+    _lock: ProcessLockSharedGuard,
+    filename: PathBuf,
+    tmp_filename: PathBuf,
+    chunk_size: usize,
+    size: usize,
+    index_length: usize,
+    index: *mut u8,
+    pub uuid: [u8; 16],
+    pub ctime: i64,
+}
+
+// `index` is mmap()ed which cannot be thread-local so should be sendable
+unsafe impl Send for FixedIndexWriter {}
+
+impl Drop for FixedIndexWriter {
+    fn drop(&mut self) {
+        let _ = std::fs::remove_file(&self.tmp_filename); // ignore errors
+        if let Err(err) = self.unmap() {
+            eprintln!("Unable to unmap file {:?} - {}", self.tmp_filename, err);
+        }
+    }
+}
+
+impl FixedIndexWriter {
+    #[allow(clippy::cast_ptr_alignment)]
+    pub fn create(
+        store: Arc<ChunkStore>,
+        path: &Path,
+        size: usize,
+        chunk_size: usize,
+    ) -> Result<Self, Error> {
+        let shared_lock = store.try_shared_lock()?;
+
+        let full_path = store.relative_path(path);
+        let mut tmp_path = full_path.clone();
+        tmp_path.set_extension("tmp_fidx");
+
+        let mut file = std::fs::OpenOptions::new()
+            .create(true)
+            .truncate(true)
+            .read(true)
+            .write(true)
+            .open(&tmp_path)?;
+
+        let header_size = std::mem::size_of::<FixedIndexHeader>();
+
+        // todo: use static assertion when available in rust
+        if header_size != 4096 {
+            panic!("got unexpected header size");
+        }
+
+        let ctime = proxmox::tools::time::epoch_i64();
+
+        let uuid = Uuid::generate();
+
+        let buffer = vec![0u8; header_size];
+        let header = unsafe { &mut *(buffer.as_ptr() as *mut FixedIndexHeader) };
+
+        header.magic = file_formats::FIXED_SIZED_CHUNK_INDEX_1_0;
+        header.ctime = i64::to_le(ctime);
+        header.size = u64::to_le(size as u64);
+        header.chunk_size = u64::to_le(chunk_size as u64);
+        header.uuid = *uuid.as_bytes();
+
+        header.index_csum = [0u8; 32];
+
+        file.write_all(&buffer)?;
+
+        let index_length = (size + chunk_size - 1) / chunk_size;
+        let index_size = index_length * 32;
+        nix::unistd::ftruncate(file.as_raw_fd(), (header_size + index_size) as i64)?;
+
+        let data = unsafe {
+            nix::sys::mman::mmap(
+                std::ptr::null_mut(),
+                index_size,
+                nix::sys::mman::ProtFlags::PROT_READ | nix::sys::mman::ProtFlags::PROT_WRITE,
+                nix::sys::mman::MapFlags::MAP_SHARED,
+                file.as_raw_fd(),
+                header_size as i64,
+            )
+        }? as *mut u8;
+
+        Ok(Self {
+            store,
+            file,
+            _lock: shared_lock,
+            filename: full_path,
+            tmp_filename: tmp_path,
+            chunk_size,
+            size,
+            index_length,
+            index: data,
+            ctime,
+            uuid: *uuid.as_bytes(),
+        })
+    }
+
+    pub fn index_length(&self) -> usize {
+        self.index_length
+    }
+
+    fn unmap(&mut self) -> Result<(), Error> {
+        if self.index.is_null() {
+            return Ok(());
+        }
+
+        let index_size = self.index_length * 32;
+
+        if let Err(err) =
+            unsafe { nix::sys::mman::munmap(self.index as *mut std::ffi::c_void, index_size) }
+        {
+            bail!("unmap file {:?} failed - {}", self.tmp_filename, err);
+        }
+
+        self.index = std::ptr::null_mut();
+
+        Ok(())
+    }
+
+    pub fn close(&mut self) -> Result<[u8; 32], Error> {
+        if self.index.is_null() {
+            bail!("cannot close already closed index file.");
+        }
+
+        let index_size = self.index_length * 32;
+        let data = unsafe { std::slice::from_raw_parts(self.index, index_size) };
+        let index_csum = openssl::sha::sha256(data);
+
+        self.unmap()?;
+
+        let csum_offset = proxmox::offsetof!(FixedIndexHeader, index_csum);
+        self.file.seek(SeekFrom::Start(csum_offset as u64))?;
+        self.file.write_all(&index_csum)?;
+        self.file.flush()?;
+
+        if let Err(err) = std::fs::rename(&self.tmp_filename, &self.filename) {
+            bail!("Atomic rename file {:?} failed - {}", self.filename, err);
+        }
+
+        Ok(index_csum)
+    }
+
+    pub fn check_chunk_alignment(&self, offset: usize, chunk_len: usize) -> Result<usize, Error> {
+        if offset < chunk_len {
+            bail!("got chunk with small offset ({} < {}", offset, chunk_len);
+        }
+
+        let pos = offset - chunk_len;
+
+        if offset > self.size {
+            bail!("chunk data exceeds size ({} >= {})", offset, self.size);
+        }
+
+        // last chunk can be smaller
+        if ((offset != self.size) && (chunk_len != self.chunk_size))
+            || (chunk_len > self.chunk_size)
+            || (chunk_len == 0)
+        {
+            bail!(
+                "chunk with unexpected length ({} != {}",
+                chunk_len,
+                self.chunk_size
+            );
+        }
+
+        if pos & (self.chunk_size - 1) != 0 {
+            bail!("got unaligned chunk (pos = {})", pos);
+        }
+
+        Ok(pos / self.chunk_size)
+    }
+
+    // Note: We want to add data out of order, so do not assume any order here.
+    pub fn add_chunk(&mut self, chunk_info: &ChunkInfo, stat: &mut ChunkStat) -> Result<(), Error> {
+        let chunk_len = chunk_info.chunk_len as usize;
+        let offset = chunk_info.offset as usize; // end of chunk
+
+        let idx = self.check_chunk_alignment(offset, chunk_len)?;
+
+        let (is_duplicate, compressed_size) = self
+            .store
+            .insert_chunk(&chunk_info.chunk, &chunk_info.digest)?;
+
+        stat.chunk_count += 1;
+        stat.compressed_size += compressed_size;
+
+        let digest = &chunk_info.digest;
+
+        println!(
+            "ADD CHUNK {} {} {}% {} {}",
+            idx,
+            chunk_len,
+            (compressed_size * 100) / (chunk_len as u64),
+            is_duplicate,
+            proxmox::tools::digest_to_hex(digest)
+        );
+
+        if is_duplicate {
+            stat.duplicate_chunks += 1;
+        } else {
+            stat.disk_size += compressed_size;
+        }
+
+        self.add_digest(idx, digest)
+    }
+
+    pub fn add_digest(&mut self, index: usize, digest: &[u8; 32]) -> Result<(), Error> {
+        if index >= self.index_length {
+            bail!(
+                "add digest failed - index out of range ({} >= {})",
+                index,
+                self.index_length
+            );
+        }
+
+        if self.index.is_null() {
+            bail!("cannot write to closed index file.");
+        }
+
+        let index_pos = index * 32;
+        unsafe {
+            let dst = self.index.add(index_pos);
+            dst.copy_from_nonoverlapping(digest.as_ptr(), 32);
+        }
+
+        Ok(())
+    }
+
+    pub fn clone_data_from(&mut self, reader: &FixedIndexReader) -> Result<(), Error> {
+        if self.index_length != reader.index_count() {
+            bail!("clone_data_from failed - index sizes not equal");
+        }
+
+        for i in 0..self.index_length {
+            self.add_digest(i, reader.index_digest(i).unwrap())?;
+        }
+
+        Ok(())
+    }
+}

pbs-datastore/src/lib.rs

@@ -195,9 +195,13 @@ pub mod file_formats;
 pub mod index;
 pub mod key_derivation;
 pub mod manifest;
+pub mod prune;
 pub mod read_chunk;
 pub mod task;
 
+pub mod dynamic_index;
+pub mod fixed_index;
+
 pub use backup_info::{BackupDir, BackupGroup, BackupInfo};
 pub use checksum_reader::ChecksumReader;
 pub use checksum_writer::ChecksumWriter;

pbs-datastore/src/prune.rs

@@ -0,0 +1,237 @@
+use std::collections::{HashMap, HashSet};
+use std::path::PathBuf;
+
+use anyhow::{Error};
+
+use super::BackupInfo;
+
+enum PruneMark { Keep, KeepPartial, Remove }
+
+fn mark_selections<F: Fn(&BackupInfo) -> Result<String, Error>> (
+    mark: &mut HashMap<PathBuf, PruneMark>,
+    list: &[BackupInfo],
+    keep: usize,
+    select_id: F,
+) -> Result<(), Error> {
+
+    let mut include_hash = HashSet::new();
+
+    let mut already_included = HashSet::new();
+    for info in list {
+        let backup_id = info.backup_dir.relative_path();
+        if let Some(PruneMark::Keep) = mark.get(&backup_id) {
+            let sel_id: String = select_id(&info)?;
+            already_included.insert(sel_id);
+        }
+    }
+
+    for info in list {
+        let backup_id = info.backup_dir.relative_path();
+        if mark.get(&backup_id).is_some() { continue; }
+        let sel_id: String = select_id(&info)?;
+
+        if already_included.contains(&sel_id) { continue; }
+
+        if !include_hash.contains(&sel_id) {
+            if include_hash.len() >= keep { break; }
+            include_hash.insert(sel_id);
+            mark.insert(backup_id, PruneMark::Keep);
+        } else {
+            mark.insert(backup_id, PruneMark::Remove);
+        }
+    }
+
+    Ok(())
+}
+
+fn remove_incomplete_snapshots(
+    mark: &mut HashMap<PathBuf, PruneMark>,
+    list: &[BackupInfo],
+) {
+
+    let mut keep_unfinished = true;
+    for info in list.iter() {
+        // backup is considered unfinished if there is no manifest
+        if info.is_finished() {
+            // There is a new finished backup, so there is no need
+            // to keep older unfinished backups.
+            keep_unfinished = false;
+        } else {
+            let backup_id = info.backup_dir.relative_path();
+            if keep_unfinished { // keep first unfinished
+                mark.insert(backup_id, PruneMark::KeepPartial);
+            } else {
+                mark.insert(backup_id, PruneMark::Remove);
+            }
+            keep_unfinished = false;
+        }
+    }
+}
+
+#[derive(Default)]
+pub struct PruneOptions {
+    pub keep_last: Option<u64>,
+    pub keep_hourly: Option<u64>,
+    pub keep_daily: Option<u64>,
+    pub keep_weekly: Option<u64>,
+    pub keep_monthly: Option<u64>,
+    pub keep_yearly: Option<u64>,
+}
+
+impl PruneOptions {
+
+    pub fn new() -> Self {
+        Self {
+            keep_last: None,
+            keep_hourly: None,
+            keep_daily: None,
+            keep_weekly: None,
+            keep_monthly: None,
+            keep_yearly: None,
+        }
+    }
+
+    pub fn keep_hourly(mut self, value: Option<u64>) -> Self {
+        self.keep_hourly = value;
+        self
+    }
+
+    pub fn keep_last(mut self, value: Option<u64>) -> Self {
+        self.keep_last = value;
+        self
+    }
+
+    pub fn keep_daily(mut self, value: Option<u64>) -> Self {
+        self.keep_daily = value;
+        self
+    }
+
+    pub fn keep_weekly(mut self, value: Option<u64>) -> Self {
+        self.keep_weekly = value;
+        self
+    }
+
+    pub fn keep_monthly(mut self, value: Option<u64>) -> Self {
+        self.keep_monthly = value;
+        self
+    }
+
+    pub fn keep_yearly(mut self, value: Option<u64>) -> Self {
+        self.keep_yearly = value;
+        self
+    }
+
+    pub fn keeps_something(&self) -> bool {
+        let mut keep_something = false;
+        if let Some(count) = self.keep_last { if count > 0 { keep_something = true; } }
+        if let Some(count) = self.keep_hourly { if count > 0 { keep_something = true; } }
+        if let Some(count) = self.keep_daily { if count > 0 { keep_something = true; } }
+        if let Some(count) = self.keep_weekly { if count > 0 { keep_something = true; } }
+        if let Some(count) = self.keep_monthly { if count > 0 { keep_something = true; } }
+        if let Some(count) = self.keep_yearly { if count > 0 { keep_something = true; } }
+        keep_something
+    }
+
+    pub fn cli_options_string(&self) -> String {
+        let mut opts = Vec::new();
+
+        if let Some(count) = self.keep_last {
+            if count > 0 {
+                opts.push(format!("--keep-last {}", count));
+            }
+        }
+        if let Some(count) = self.keep_hourly {
+            if count > 0 {
+                opts.push(format!("--keep-hourly {}", count));
+            }
+        }
+        if let Some(count) = self.keep_daily {
+            if count > 0 {
+                opts.push(format!("--keep-daily {}", count));
+            }
+        }
+        if let Some(count) = self.keep_weekly {
+            if count > 0 {
+                opts.push(format!("--keep-weekly {}", count));
+            }
+        }
+        if let Some(count) = self.keep_monthly {
+            if count > 0 {
+                opts.push(format!("--keep-monthly {}", count));
+            }
+        }
+        if let Some(count) = self.keep_yearly {
+            if count > 0 {
+                opts.push(format!("--keep-yearly {}", count));
+            }
+        }
+
+        opts.join(" ")
+    }
+}
+
+pub fn compute_prune_info(
+    mut list: Vec<BackupInfo>,
+    options: &PruneOptions,
+) -> Result<Vec<(BackupInfo, bool)>, Error> {
+
+    let mut mark = HashMap::new();
+
+    BackupInfo::sort_list(&mut list, false);
+
+    remove_incomplete_snapshots(&mut mark, &list);
+
+    if let Some(keep_last) = options.keep_last {
+        mark_selections(&mut mark, &list, keep_last as usize, |info| {
+            Ok(info.backup_dir.backup_time_string().to_owned())
+        })?;
+    }
+
+    use proxmox::tools::time::strftime_local;
+
+    if let Some(keep_hourly) = options.keep_hourly {
+        mark_selections(&mut mark, &list, keep_hourly as usize, |info| {
+            strftime_local("%Y/%m/%d/%H", info.backup_dir.backup_time())
+        })?;
+    }
+
+    if let Some(keep_daily) = options.keep_daily {
+        mark_selections(&mut mark, &list, keep_daily as usize, |info| {
+            strftime_local("%Y/%m/%d", info.backup_dir.backup_time())
+        })?;
+    }
+
+    if let Some(keep_weekly) = options.keep_weekly {
+        mark_selections(&mut mark, &list, keep_weekly as usize, |info| {
+            // Note: Use iso-week year/week here. This year number
+            // might not match the calendar year number.
+            strftime_local("%G/%V", info.backup_dir.backup_time())
+        })?;
+    }
+
+    if let Some(keep_monthly) = options.keep_monthly {
+        mark_selections(&mut mark, &list, keep_monthly as usize, |info| {
+            strftime_local("%Y/%m", info.backup_dir.backup_time())
+        })?;
+    }
+
+    if let Some(keep_yearly) = options.keep_yearly {
+        mark_selections(&mut mark, &list, keep_yearly as usize, |info| {
+            strftime_local("%Y", info.backup_dir.backup_time())
+        })?;
+    }
+
+    let prune_info: Vec<(BackupInfo, bool)> = list.into_iter()
+        .map(|info| {
+            let backup_id = info.backup_dir.relative_path();
+            let keep = match mark.get(&backup_id) {
+                Some(PruneMark::Keep) => true,
+                Some(PruneMark::KeepPartial) => true,
+               _ => false,
+            };
+            (info, keep)
+        })
+        .collect();
+
+    Ok(prune_info)
+}