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         unsafe {
 367             std::ptr::copy_nonoverlapping(data.as_ptr(), buf.as_mut_ptr(), n);
 368         }
 369
 370         self.read_offset += n as u64;
 371
 372         Ok(n)
 373     }
 374 }
 375
 376 impl<S: ReadChunk> std::io::Seek for BufferedDynamicReader<S> {
 377     fn seek(&mut self, pos: SeekFrom) -> Result<u64, std::io::Error> {
 378         let new_offset = match pos {
 379             SeekFrom::Start(start_offset) => start_offset as i64,
 380             SeekFrom::End(end_offset) => (self.archive_size as i64) + end_offset,
 381             SeekFrom::Current(offset) => (self.read_offset as i64) + offset,
 382         };
 383
 384         use std::io::{Error, ErrorKind};
 385         if (new_offset < 0) || (new_offset > (self.archive_size as i64)) {
 386             return Err(Error::new(
 387                 ErrorKind::Other,
 388                 format!(
 389                     "seek is out of range {} ([0..{}])",
 390                     new_offset, self.archive_size
 391                 ),
 392             ));
 393         }
 394         self.read_offset = new_offset as u64;
 395
 396         Ok(self.read_offset)
 397     }
 398 }
 399
 400 /// Create dynamic index files (`.dixd`)
 401 pub struct DynamicIndexWriter {
 402     store: Arc<ChunkStore>,
 403     _lock: tools::ProcessLockSharedGuard,
 404     writer: BufWriter<File>,
 405     closed: bool,
 406     filename: PathBuf,
 407     tmp_filename: PathBuf,
 408     csum: Option<openssl::sha::Sha256>,
 409     pub uuid: [u8; 16],
 410     pub ctime: u64,
 411 }
 412
 413 impl Drop for DynamicIndexWriter {
 414     fn drop(&mut self) {
 415         let _ = std::fs::remove_file(&self.tmp_filename); // ignore errors
 416     }
 417 }
 418
 419 impl DynamicIndexWriter {
 420     pub fn create(store: Arc<ChunkStore>, path: &Path) -> Result<Self, Error> {
 421         let shared_lock = store.try_shared_lock()?;
 422
 423         let full_path = store.relative_path(path);
 424         let mut tmp_path = full_path.clone();
 425         tmp_path.set_extension("tmp_didx");
 426
 427         let file = std::fs::OpenOptions::new()
 428             .create(true)
 429             .truncate(true)
 430             .read(true)
 431             .write(true)
 432             .open(&tmp_path)?;
 433
 434         let mut writer = BufWriter::with_capacity(1024 * 1024, file);
 435
 436         let header_size = std::mem::size_of::<DynamicIndexHeader>();
 437
 438         // todo: use static assertion when available in rust
 439         if header_size != 4096 {
 440             panic!("got unexpected header size");
 441         }
 442
 443         let ctime = epoch_now_u64()?;
 444
 445         let uuid = Uuid::generate();
 446
 447         let mut buffer = vec::zeroed(header_size);
 448         let header = crate::tools::map_struct_mut::<DynamicIndexHeader>(&mut buffer)?;
 449
 450         header.magic = super::DYNAMIC_SIZED_CHUNK_INDEX_1_0;
 451         header.ctime = u64::to_le(ctime);
 452         header.uuid = *uuid.as_bytes();
 453
 454         header.index_csum = [0u8; 32];
 455
 456         writer.write_all(&buffer)?;
 457
 458         let csum = Some(openssl::sha::Sha256::new());
 459
 460         Ok(Self {
 461             store,
 462             _lock: shared_lock,
 463             writer,
 464             closed: false,
 465             filename: full_path,
 466             tmp_filename: tmp_path,
 467             ctime,
 468             uuid: *uuid.as_bytes(),
 469             csum,
 470         })
 471     }
 472
 473     // fixme: use add_chunk instead?
 474     pub fn insert_chunk(&self, chunk: &DataBlob, digest: &[u8; 32]) -> Result<(bool, u64), Error> {
 475         self.store.insert_chunk(chunk, digest)
 476     }
 477
 478     pub fn close(&mut self) -> Result<[u8; 32], Error> {
 479         if self.closed {
 480             bail!(
 481                 "cannot close already closed archive index file {:?}",
 482                 self.filename
 483             );
 484         }
 485
 486         self.closed = true;
 487
 488         self.writer.flush()?;
 489
 490         let csum_offset = proxmox::offsetof!(DynamicIndexHeader, index_csum);
 491         self.writer.seek(SeekFrom::Start(csum_offset as u64))?;
 492
 493         let csum = self.csum.take().unwrap();
 494         let index_csum = csum.finish();
 495
 496         self.writer.write_all(&index_csum)?;
 497         self.writer.flush()?;
 498
 499         if let Err(err) = std::fs::rename(&self.tmp_filename, &self.filename) {
 500             bail!("Atomic rename file {:?} failed - {}", self.filename, err);
 501         }
 502
 503         Ok(index_csum)
 504     }
 505
 506     // fixme: rename to add_digest
 507     pub fn add_chunk(&mut self, offset: u64, digest: &[u8; 32]) -> Result<(), Error> {
 508         if self.closed {
 509             bail!(
 510                 "cannot write to closed dynamic index file {:?}",
 511                 self.filename
 512             );
 513         }
 514
 515         let offset_le: &[u8; 8] = unsafe { &std::mem::transmute::<u64, [u8; 8]>(offset.to_le()) };
 516
 517         if let Some(ref mut csum) = self.csum {
 518             csum.update(offset_le);
 519             csum.update(digest);
 520         }
 521
 522         self.writer.write_all(offset_le)?;
 523         self.writer.write_all(digest)?;
 524         Ok(())
 525     }
 526 }
 527
 528 /// Writer which splits a binary stream into dynamic sized chunks
 529 ///
 530 /// And store the resulting chunk list into the index file.
 531 pub struct DynamicChunkWriter {
 532     index: DynamicIndexWriter,
 533     closed: bool,
 534     chunker: Chunker,
 535     stat: ChunkStat,
 536     chunk_offset: usize,
 537     last_chunk: usize,
 538     chunk_buffer: Vec<u8>,
 539 }
 540
 541 impl DynamicChunkWriter {
 542     pub fn new(index: DynamicIndexWriter, chunk_size: usize) -> Self {
 543         Self {
 544             index,
 545             closed: false,
 546             chunker: Chunker::new(chunk_size),
 547             stat: ChunkStat::new(0),
 548             chunk_offset: 0,
 549             last_chunk: 0,
 550             chunk_buffer: Vec::with_capacity(chunk_size * 4),
 551         }
 552     }
 553
 554     pub fn stat(&self) -> &ChunkStat {
 555         &self.stat
 556     }
 557
 558     pub fn close(&mut self) -> Result<(), Error> {
 559         if self.closed {
 560             return Ok(());
 561         }
 562
 563         self.closed = true;
 564
 565         self.write_chunk_buffer()?;
 566
 567         self.index.close()?;
 568
 569         self.stat.size = self.chunk_offset as u64;
 570
 571         // add size of index file
 572         self.stat.size +=
 573             (self.stat.chunk_count * 40 + std::mem::size_of::<DynamicIndexHeader>()) as u64;
 574
 575         Ok(())
 576     }
 577
 578     fn write_chunk_buffer(&mut self) -> Result<(), Error> {
 579         let chunk_size = self.chunk_buffer.len();
 580
 581         if chunk_size == 0 {
 582             return Ok(());
 583         }
 584
 585         let expected_chunk_size = self.chunk_offset - self.last_chunk;
 586         if expected_chunk_size != self.chunk_buffer.len() {
 587             bail!("wrong chunk size {} != {}", expected_chunk_size, chunk_size);
 588         }
 589
 590         self.stat.chunk_count += 1;
 591
 592         self.last_chunk = self.chunk_offset;
 593
 594         let (chunk, digest) = DataChunkBuilder::new(&self.chunk_buffer)
 595             .compress(true)
 596             .build()?;
 597
 598         match self.index.insert_chunk(&chunk, &digest) {
 599             Ok((is_duplicate, compressed_size)) => {
 600                 self.stat.compressed_size += compressed_size;
 601                 if is_duplicate {
 602                     self.stat.duplicate_chunks += 1;
 603                 } else {
 604                     self.stat.disk_size += compressed_size;
 605                 }
 606
 607                 println!(
 608                     "ADD CHUNK {:016x} {} {}% {} {}",
 609                     self.chunk_offset,
 610                     chunk_size,
 611                     (compressed_size * 100) / (chunk_size as u64),
 612                     is_duplicate,
 613                     proxmox::tools::digest_to_hex(&digest)
 614                 );
 615                 self.index.add_chunk(self.chunk_offset as u64, &digest)?;
 616                 self.chunk_buffer.truncate(0);
 617                 Ok(())
 618             }
 619             Err(err) => {
 620                 self.chunk_buffer.truncate(0);
 621                 Err(err)
 622             }
 623         }
 624     }
 625 }
 626
 627 impl Write for DynamicChunkWriter {
 628     fn write(&mut self, data: &[u8]) -> std::result::Result<usize, std::io::Error> {
 629         let chunker = &mut self.chunker;
 630
 631         let pos = chunker.scan(data);
 632
 633         if pos > 0 {
 634             self.chunk_buffer.extend_from_slice(&data[0..pos]);
 635             self.chunk_offset += pos;
 636
 637             if let Err(err) = self.write_chunk_buffer() {
 638                 return Err(std::io::Error::new(
 639                     std::io::ErrorKind::Other,
 640                     err.to_string(),
 641                 ));
 642             }
 643             Ok(pos)
 644         } else {
 645             self.chunk_offset += data.len();
 646             self.chunk_buffer.extend_from_slice(data);
 647             Ok(data.len())
 648         }
 649     }
 650
 651     fn flush(&mut self) -> std::result::Result<(), std::io::Error> {
 652         Err(std::io::Error::new(
 653             std::io::ErrorKind::Other,
 654             "please use close() instead of flush()",
 655         ))
 656     }
 657 }