src/backup/dynamic_index.rs

   1 use std::fs::File;
   2 use std::io::{BufWriter, Seek, SeekFrom, Write};
   3 use std::ops::Range;
   4 use std::os::unix::io::AsRawFd;
   5 use std::path::{Path, PathBuf};
   6 use std::sync::Arc;
   7
   8 use anyhow::{bail, format_err, Error};
   9
  10 use proxmox::tools::io::ReadExt;
  11 use proxmox::tools::uuid::Uuid;
  12 use proxmox::tools::vec;
  13 use proxmox::tools::mmap::Mmap;
  14
  15 use super::chunk_stat::ChunkStat;
  16 use super::chunk_store::ChunkStore;
  17 use super::index::ChunkReadInfo;
  18 use super::read_chunk::ReadChunk;
  19 use super::Chunker;
  20 use super::IndexFile;
  21 use super::{DataBlob, DataChunkBuilder};
  22 use crate::tools::{self, epoch_now_u64};
  23
  24 /// Header format definition for dynamic index files (`.dixd`)
  25 #[repr(C)]
  26 pub struct DynamicIndexHeader {
  27     pub magic: [u8; 8],
  28     pub uuid: [u8; 16],
  29     pub ctime: u64,
  30     /// Sha256 over the index ``SHA256(offset1||digest1||offset2||digest2||...)``
  31     pub index_csum: [u8; 32],
  32     reserved: [u8; 4032], // overall size is one page (4096 bytes)
  33 }
  34 proxmox::static_assert_size!(DynamicIndexHeader, 4096);
  35 // TODO: Once non-Copy unions are stabilized, use:
  36 // union DynamicIndexHeader {
  37 //     reserved: [u8; 4096],
  38 //     pub data: DynamicIndexHeaderData,
  39 // }
  40
  41 #[derive(Clone, Debug)]
  42 #[repr(C)]
  43 pub struct DynamicEntry {
  44     end_le: u64,
  45     digest: [u8; 32],
  46 }
  47
  48 impl DynamicEntry {
  49     #[inline]
  50     pub fn end(&self) -> u64 {
  51         u64::from_le(self.end_le)
  52     }
  53 }
  54
  55 pub struct DynamicIndexReader {
  56     _file: File,
  57     pub size: usize,
  58     index: Mmap<DynamicEntry>,
  59     pub uuid: [u8; 16],
  60     pub ctime: u64,
  61     pub index_csum: [u8; 32],
  62 }
  63
  64 impl DynamicIndexReader {
  65     pub fn open(path: &Path) -> Result<Self, Error> {
  66         File::open(path)
  67             .map_err(Error::from)
  68             .and_then(Self::new)
  69             .map_err(|err| format_err!("Unable to open dynamic index {:?} - {}", path, err))
  70     }
  71
  72     pub fn new(mut file: std::fs::File) -> Result<Self, Error> {
  73         if let Err(err) =
  74             nix::fcntl::flock(file.as_raw_fd(), nix::fcntl::FlockArg::LockSharedNonblock)
  75         {
  76             bail!("unable to get shared lock - {}", err);
  77         }
  78
  79         // FIXME: This is NOT OUR job! Check the callers of this method and remove this!
  80         file.seek(SeekFrom::Start(0))?;
  81
  82         let header_size = std::mem::size_of::<DynamicIndexHeader>();
  83
  84         let header: Box<DynamicIndexHeader> = unsafe { file.read_host_value_boxed()? };
  85
  86         if header.magic != super::DYNAMIC_SIZED_CHUNK_INDEX_1_0 {
  87             bail!("got unknown magic number");
  88         }
  89
  90         let ctime = u64::from_le(header.ctime);
  91
  92         let rawfd = file.as_raw_fd();
  93
  94         let stat = nix::sys::stat::fstat(rawfd)?;
  95
  96         let size = stat.st_size as usize;
  97
  98         let index_size = size - header_size;
  99         let index_count = index_size / 40;
 100         if index_count * 40 != index_size {
 101             bail!("got unexpected file size");
 102         }
 103
 104         let index = unsafe {
 105             Mmap::map_fd(
 106                 rawfd,
 107                 header_size as u64,
 108                 index_count,
 109                 nix::sys::mman::ProtFlags::PROT_READ,
 110                 nix::sys::mman::MapFlags::MAP_PRIVATE,
 111             )?
 112         };
 113
 114         Ok(Self {
 115             _file: file,
 116             size,
 117             index,
 118             ctime,
 119             uuid: header.uuid,
 120             index_csum: header.index_csum,
 121         })
 122     }
 123
 124     #[allow(clippy::cast_ptr_alignment)]
 125     pub fn chunk_info(&self, pos: usize) -> Result<ChunkReadInfo, Error> {
 126         if pos >= self.index.len() {
 127             bail!("chunk index out of range");
 128         }
 129         let start = if pos == 0 {
 130             0
 131         } else {
 132             self.index[pos - 1].end()
 133         };
 134
 135         let end = self.index[pos].end();
 136
 137         Ok(ChunkReadInfo {
 138             range: start..end,
 139             digest: self.index[pos].digest.clone(),
 140         })
 141     }
 142
 143     #[inline]
 144     #[allow(clippy::cast_ptr_alignment)]
 145     fn chunk_end(&self, pos: usize) -> u64 {
 146         if pos >= self.index.len() {
 147             panic!("chunk index out of range");
 148         }
 149         self.index[pos].end()
 150     }
 151
 152     #[inline]
 153     fn chunk_digest(&self, pos: usize) -> &[u8; 32] {
 154         if pos >= self.index.len() {
 155             panic!("chunk index out of range");
 156         }
 157         &self.index[pos].digest
 158     }
 159
 160     /// Compute checksum and data size
 161     pub fn compute_csum(&self) -> ([u8; 32], u64) {
 162         let mut csum = openssl::sha::Sha256::new();
 163         for entry in &self.index {
 164             csum.update(&entry.end_le.to_ne_bytes());
 165             csum.update(&entry.digest);
 166         }
 167         let csum = csum.finish();
 168
 169         (
 170             csum,
 171             self.index
 172                 .last()
 173                 .map(|entry| entry.end())
 174                 .unwrap_or(0)
 175         )
 176     }
 177
 178     // TODO: can we use std::slice::binary_search with Mmap now?
 179     fn binary_search(
 180         &self,
 181         start_idx: usize,
 182         start: u64,
 183         end_idx: usize,
 184         end: u64,
 185         offset: u64,
 186     ) -> Result<usize, Error> {
 187         if (offset >= end) || (offset < start) {
 188             bail!("offset out of range");
 189         }
 190
 191         if end_idx == start_idx {
 192             return Ok(start_idx); // found
 193         }
 194         let middle_idx = (start_idx + end_idx) / 2;
 195         let middle_end = self.chunk_end(middle_idx);
 196
 197         if offset < middle_end {
 198             self.binary_search(start_idx, start, middle_idx, middle_end, offset)
 199         } else {
 200             self.binary_search(middle_idx + 1, middle_end, end_idx, end, offset)
 201         }
 202     }
 203 }
 204
 205 impl IndexFile for DynamicIndexReader {
 206     fn index_count(&self) -> usize {
 207         self.index.len()
 208     }
 209
 210     fn index_digest(&self, pos: usize) -> Option<&[u8; 32]> {
 211         if pos >= self.index.len() {
 212             None
 213         } else {
 214             Some(unsafe { std::mem::transmute(self.chunk_digest(pos).as_ptr()) })
 215         }
 216     }
 217
 218     fn index_bytes(&self) -> u64 {
 219         if self.index.is_empty() {
 220             0
 221         } else {
 222             self.chunk_end(self.index.len() - 1)
 223         }
 224     }
 225 }
 226
 227 struct CachedChunk {
 228     range: Range<u64>,
 229     data: Vec<u8>,
 230 }
 231
 232 impl CachedChunk {
 233     /// Perform sanity checks on the range and data size:
 234     pub fn new(range: Range<u64>, data: Vec<u8>) -> Result<Self, Error> {
 235         if data.len() as u64 != range.end - range.start {
 236             bail!(
 237                 "read chunk with wrong size ({} != {})",
 238                 data.len(),
 239                 range.end - range.start,
 240             );
 241         }
 242         Ok(Self { range, data })
 243     }
 244 }
 245
 246 pub struct BufferedDynamicReader<S> {
 247     store: S,
 248     index: DynamicIndexReader,
 249     archive_size: u64,
 250     read_buffer: Vec<u8>,
 251     buffered_chunk_idx: usize,
 252     buffered_chunk_start: u64,
 253     read_offset: u64,
 254     lru_cache: crate::tools::lru_cache::LruCache<usize, CachedChunk>,
 255 }
 256
 257 struct ChunkCacher<'a, S> {
 258     store: &'a mut S,
 259     index: &'a DynamicIndexReader,
 260 }
 261
 262 impl<'a, S: ReadChunk> crate::tools::lru_cache::Cacher<usize, CachedChunk> for ChunkCacher<'a, S> {
 263     fn fetch(&mut self, index: usize) -> Result<Option<CachedChunk>, Error> {
 264         let info = self.index.chunk_info(index)?;
 265         let range = info.range;
 266         let data = self.store.read_chunk(&info.digest)?;
 267         CachedChunk::new(range, data).map(Some)
 268     }
 269 }
 270
 271 impl<S: ReadChunk> BufferedDynamicReader<S> {
 272     pub fn new(index: DynamicIndexReader, store: S) -> Self {
 273         let archive_size = index.index_bytes();
 274         Self {
 275             store,
 276             index,
 277             archive_size,
 278             read_buffer: Vec::with_capacity(1024 * 1024),
 279             buffered_chunk_idx: 0,
 280             buffered_chunk_start: 0,
 281             read_offset: 0,
 282             lru_cache: crate::tools::lru_cache::LruCache::new(32),
 283         }
 284     }
 285
 286     pub fn archive_size(&self) -> u64 {
 287         self.archive_size
 288     }
 289
 290     fn buffer_chunk(&mut self, idx: usize) -> Result<(), Error> {
 291         //let (start, end, data) = self.lru_cache.access(
 292         let cached_chunk = self.lru_cache.access(
 293             idx,
 294             &mut ChunkCacher {
 295                 store: &mut self.store,
 296                 index: &self.index,
 297             },
 298         )?.ok_or_else(|| format_err!("chunk not found by cacher"))?;
 299
 300         // fixme: avoid copy
 301         self.read_buffer.clear();
 302         self.read_buffer.extend_from_slice(&cached_chunk.data);
 303
 304         self.buffered_chunk_idx = idx;
 305
 306         self.buffered_chunk_start = cached_chunk.range.start;
 307         //println!("BUFFER {} {}",  self.buffered_chunk_start, end);
 308         Ok(())
 309     }
 310 }
 311
 312 impl<S: ReadChunk> crate::tools::BufferedRead for BufferedDynamicReader<S> {
 313     fn buffered_read(&mut self, offset: u64) -> Result<&[u8], Error> {
 314         if offset == self.archive_size {
 315             return Ok(&self.read_buffer[0..0]);
 316         }
 317
 318         let buffer_len = self.read_buffer.len();
 319         let index = &self.index;
 320
 321         // optimization for sequential read
 322         if buffer_len > 0
 323             && ((self.buffered_chunk_idx + 1) < index.index.len())
 324             && (offset >= (self.buffered_chunk_start + (self.read_buffer.len() as u64)))
 325         {
 326             let next_idx = self.buffered_chunk_idx + 1;
 327             let next_end = index.chunk_end(next_idx);
 328             if offset < next_end {
 329                 self.buffer_chunk(next_idx)?;
 330                 let buffer_offset = (offset - self.buffered_chunk_start) as usize;
 331                 return Ok(&self.read_buffer[buffer_offset..]);
 332             }
 333         }
 334
 335         if (buffer_len == 0)
 336             || (offset < self.buffered_chunk_start)
 337             || (offset >= (self.buffered_chunk_start + (self.read_buffer.len() as u64)))
 338         {
 339             let end_idx = index.index.len() - 1;
 340             let end = index.chunk_end(end_idx);
 341             let idx = index.binary_search(0, 0, end_idx, end, offset)?;
 342             self.buffer_chunk(idx)?;
 343         }
 344
 345         let buffer_offset = (offset - self.buffered_chunk_start) as usize;
 346         Ok(&self.read_buffer[buffer_offset..])
 347     }
 348 }
 349
 350 impl<S: ReadChunk> std::io::Read for BufferedDynamicReader<S> {
 351     fn read(&mut self, buf: &mut [u8]) -> Result<usize, std::io::Error> {
 352         use crate::tools::BufferedRead;
 353         use std::io::{Error, ErrorKind};
 354
 355         let data = match self.buffered_read(self.read_offset) {
 356             Ok(v) => v,
 357             Err(err) => return Err(Error::new(ErrorKind::Other, err.to_string())),
 358         };
 359
 360         let n = if data.len() > buf.len() {
 361             buf.len()
 362         } else {
 363             data.len()
 364         };
 365
 366         buf[0..n].copy_from_slice(&data[0..n]);
 367
 368         self.read_offset += n as u64;
 369
 370         Ok(n)
 371     }
 372 }
 373
 374 impl<S: ReadChunk> std::io::Seek for BufferedDynamicReader<S> {
 375     fn seek(&mut self, pos: SeekFrom) -> Result<u64, std::io::Error> {
 376         let new_offset = match pos {
 377             SeekFrom::Start(start_offset) => start_offset as i64,
 378             SeekFrom::End(end_offset) => (self.archive_size as i64) + end_offset,
 379             SeekFrom::Current(offset) => (self.read_offset as i64) + offset,
 380         };
 381
 382         use std::io::{Error, ErrorKind};
 383         if (new_offset < 0) || (new_offset > (self.archive_size as i64)) {
 384             return Err(Error::new(
 385                 ErrorKind::Other,
 386                 format!(
 387                     "seek is out of range {} ([0..{}])",
 388                     new_offset, self.archive_size
 389                 ),
 390             ));
 391         }
 392         self.read_offset = new_offset as u64;
 393
 394         Ok(self.read_offset)
 395     }
 396 }
 397
 398 /// Create dynamic index files (`.dixd`)
 399 pub struct DynamicIndexWriter {
 400     store: Arc<ChunkStore>,
 401     _lock: tools::ProcessLockSharedGuard,
 402     writer: BufWriter<File>,
 403     closed: bool,
 404     filename: PathBuf,
 405     tmp_filename: PathBuf,
 406     csum: Option<openssl::sha::Sha256>,
 407     pub uuid: [u8; 16],
 408     pub ctime: u64,
 409 }
 410
 411 impl Drop for DynamicIndexWriter {
 412     fn drop(&mut self) {
 413         let _ = std::fs::remove_file(&self.tmp_filename); // ignore errors
 414     }
 415 }
 416
 417 impl DynamicIndexWriter {
 418     pub fn create(store: Arc<ChunkStore>, path: &Path) -> Result<Self, Error> {
 419         let shared_lock = store.try_shared_lock()?;
 420
 421         let full_path = store.relative_path(path);
 422         let mut tmp_path = full_path.clone();
 423         tmp_path.set_extension("tmp_didx");
 424
 425         let file = std::fs::OpenOptions::new()
 426             .create(true)
 427             .truncate(true)
 428             .read(true)
 429             .write(true)
 430             .open(&tmp_path)?;
 431
 432         let mut writer = BufWriter::with_capacity(1024 * 1024, file);
 433
 434         let header_size = std::mem::size_of::<DynamicIndexHeader>();
 435
 436         // todo: use static assertion when available in rust
 437         if header_size != 4096 {
 438             panic!("got unexpected header size");
 439         }
 440
 441         let ctime = epoch_now_u64()?;
 442
 443         let uuid = Uuid::generate();
 444
 445         let mut buffer = vec::zeroed(header_size);
 446         let header = crate::tools::map_struct_mut::<DynamicIndexHeader>(&mut buffer)?;
 447
 448         header.magic = super::DYNAMIC_SIZED_CHUNK_INDEX_1_0;
 449         header.ctime = u64::to_le(ctime);
 450         header.uuid = *uuid.as_bytes();
 451
 452         header.index_csum = [0u8; 32];
 453
 454         writer.write_all(&buffer)?;
 455
 456         let csum = Some(openssl::sha::Sha256::new());
 457
 458         Ok(Self {
 459             store,
 460             _lock: shared_lock,
 461             writer,
 462             closed: false,
 463             filename: full_path,
 464             tmp_filename: tmp_path,
 465             ctime,
 466             uuid: *uuid.as_bytes(),
 467             csum,
 468         })
 469     }
 470
 471     // fixme: use add_chunk instead?
 472     pub fn insert_chunk(&self, chunk: &DataBlob, digest: &[u8; 32]) -> Result<(bool, u64), Error> {
 473         self.store.insert_chunk(chunk, digest)
 474     }
 475
 476     pub fn close(&mut self) -> Result<[u8; 32], Error> {
 477         if self.closed {
 478             bail!(
 479                 "cannot close already closed archive index file {:?}",
 480                 self.filename
 481             );
 482         }
 483
 484         self.closed = true;
 485
 486         self.writer.flush()?;
 487
 488         let csum_offset = proxmox::offsetof!(DynamicIndexHeader, index_csum);
 489         self.writer.seek(SeekFrom::Start(csum_offset as u64))?;
 490
 491         let csum = self.csum.take().unwrap();
 492         let index_csum = csum.finish();
 493
 494         self.writer.write_all(&index_csum)?;
 495         self.writer.flush()?;
 496
 497         if let Err(err) = std::fs::rename(&self.tmp_filename, &self.filename) {
 498             bail!("Atomic rename file {:?} failed - {}", self.filename, err);
 499         }
 500
 501         Ok(index_csum)
 502     }
 503
 504     // fixme: rename to add_digest
 505     pub fn add_chunk(&mut self, offset: u64, digest: &[u8; 32]) -> Result<(), Error> {
 506         if self.closed {
 507             bail!(
 508                 "cannot write to closed dynamic index file {:?}",
 509                 self.filename
 510             );
 511         }
 512
 513         let offset_le: &[u8; 8] = unsafe { &std::mem::transmute::<u64, [u8; 8]>(offset.to_le()) };
 514
 515         if let Some(ref mut csum) = self.csum {
 516             csum.update(offset_le);
 517             csum.update(digest);
 518         }
 519
 520         self.writer.write_all(offset_le)?;
 521         self.writer.write_all(digest)?;
 522         Ok(())
 523     }
 524 }
 525
 526 /// Writer which splits a binary stream into dynamic sized chunks
 527 ///
 528 /// And store the resulting chunk list into the index file.
 529 pub struct DynamicChunkWriter {
 530     index: DynamicIndexWriter,
 531     closed: bool,
 532     chunker: Chunker,
 533     stat: ChunkStat,
 534     chunk_offset: usize,
 535     last_chunk: usize,
 536     chunk_buffer: Vec<u8>,
 537 }
 538
 539 impl DynamicChunkWriter {
 540     pub fn new(index: DynamicIndexWriter, chunk_size: usize) -> Self {
 541         Self {
 542             index,
 543             closed: false,
 544             chunker: Chunker::new(chunk_size),
 545             stat: ChunkStat::new(0),
 546             chunk_offset: 0,
 547             last_chunk: 0,
 548             chunk_buffer: Vec::with_capacity(chunk_size * 4),
 549         }
 550     }
 551
 552     pub fn stat(&self) -> &ChunkStat {
 553         &self.stat
 554     }
 555
 556     pub fn close(&mut self) -> Result<(), Error> {
 557         if self.closed {
 558             return Ok(());
 559         }
 560
 561         self.closed = true;
 562
 563         self.write_chunk_buffer()?;
 564
 565         self.index.close()?;
 566
 567         self.stat.size = self.chunk_offset as u64;
 568
 569         // add size of index file
 570         self.stat.size +=
 571             (self.stat.chunk_count * 40 + std::mem::size_of::<DynamicIndexHeader>()) as u64;
 572
 573         Ok(())
 574     }
 575
 576     fn write_chunk_buffer(&mut self) -> Result<(), Error> {
 577         let chunk_size = self.chunk_buffer.len();
 578
 579         if chunk_size == 0 {
 580             return Ok(());
 581         }
 582
 583         let expected_chunk_size = self.chunk_offset - self.last_chunk;
 584         if expected_chunk_size != self.chunk_buffer.len() {
 585             bail!("wrong chunk size {} != {}", expected_chunk_size, chunk_size);
 586         }
 587
 588         self.stat.chunk_count += 1;
 589
 590         self.last_chunk = self.chunk_offset;
 591
 592         let (chunk, digest) = DataChunkBuilder::new(&self.chunk_buffer)
 593             .compress(true)
 594             .build()?;
 595
 596         match self.index.insert_chunk(&chunk, &digest) {
 597             Ok((is_duplicate, compressed_size)) => {
 598                 self.stat.compressed_size += compressed_size;
 599                 if is_duplicate {
 600                     self.stat.duplicate_chunks += 1;
 601                 } else {
 602                     self.stat.disk_size += compressed_size;
 603                 }
 604
 605                 println!(
 606                     "ADD CHUNK {:016x} {} {}% {} {}",
 607                     self.chunk_offset,
 608                     chunk_size,
 609                     (compressed_size * 100) / (chunk_size as u64),
 610                     is_duplicate,
 611                     proxmox::tools::digest_to_hex(&digest)
 612                 );
 613                 self.index.add_chunk(self.chunk_offset as u64, &digest)?;
 614                 self.chunk_buffer.truncate(0);
 615                 Ok(())
 616             }
 617             Err(err) => {
 618                 self.chunk_buffer.truncate(0);
 619                 Err(err)
 620             }
 621         }
 622     }
 623 }
 624
 625 impl Write for DynamicChunkWriter {
 626     fn write(&mut self, data: &[u8]) -> std::result::Result<usize, std::io::Error> {
 627         let chunker = &mut self.chunker;
 628
 629         let pos = chunker.scan(data);
 630
 631         if pos > 0 {
 632             self.chunk_buffer.extend_from_slice(&data[0..pos]);
 633             self.chunk_offset += pos;
 634
 635             if let Err(err) = self.write_chunk_buffer() {
 636                 return Err(std::io::Error::new(
 637                     std::io::ErrorKind::Other,
 638                     err.to_string(),
 639                 ));
 640             }
 641             Ok(pos)
 642         } else {
 643             self.chunk_offset += data.len();
 644             self.chunk_buffer.extend_from_slice(data);
 645             Ok(data.len())
 646         }
 647     }
 648
 649     fn flush(&mut self) -> std::result::Result<(), std::io::Error> {
 650         Err(std::io::Error::new(
 651             std::io::ErrorKind::Other,
 652             "please use close() instead of flush()",
 653         ))
 654     }
 655 }