pbs-datastore/src/fixed_index.rs

   1 use std::fs::File;
   2 use std::io::Write;
   3 use std::os::unix::io::AsRawFd;
   4 use std::path::{Path, PathBuf};
   5 use std::sync::Arc;
   6 use std::io::{Seek, SeekFrom};
   7
   8 use anyhow::{bail, format_err, Error};
   9
  10 use pbs_tools::process_locker::ProcessLockSharedGuard;
  11
  12 use proxmox_io::ReadExt;
  13 use proxmox_uuid::Uuid;
  14
  15 use crate::chunk_stat::ChunkStat;
  16 use crate::chunk_store::ChunkStore;
  17 use crate::data_blob::ChunkInfo;
  18 use crate::file_formats;
  19 use crate::index::{ChunkReadInfo, IndexFile};
  20
  21 /// Header format definition for fixed index files (`.fidx`)
  22 #[repr(C)]
  23 pub struct FixedIndexHeader {
  24     pub magic: [u8; 8],
  25     pub uuid: [u8; 16],
  26     pub ctime: i64,
  27     /// Sha256 over the index ``SHA256(digest1||digest2||...)``
  28     pub index_csum: [u8; 32],
  29     pub size: u64,
  30     pub chunk_size: u64,
  31     reserved: [u8; 4016], // overall size is one page (4096 bytes)
  32 }
  33 proxmox_lang::static_assert_size!(FixedIndexHeader, 4096);
  34
  35 // split image into fixed size chunks
  36
  37 pub struct FixedIndexReader {
  38     _file: File,
  39     pub chunk_size: usize,
  40     pub size: u64,
  41     index_length: usize,
  42     index: *mut u8,
  43     pub uuid: [u8; 16],
  44     pub ctime: i64,
  45     pub index_csum: [u8; 32],
  46 }
  47
  48 // `index` is mmap()ed which cannot be thread-local so should be sendable
  49 unsafe impl Send for FixedIndexReader {}
  50 unsafe impl Sync for FixedIndexReader {}
  51
  52 impl Drop for FixedIndexReader {
  53     fn drop(&mut self) {
  54         if let Err(err) = self.unmap() {
  55             eprintln!("Unable to unmap file - {}", err);
  56         }
  57     }
  58 }
  59
  60 impl FixedIndexReader {
  61     pub fn open(path: &Path) -> Result<Self, Error> {
  62         File::open(path)
  63             .map_err(Error::from)
  64             .and_then(Self::new)
  65             .map_err(|err| format_err!("Unable to open fixed index {:?} - {}", path, err))
  66     }
  67
  68     pub fn new(mut file: std::fs::File) -> Result<Self, Error> {
  69         file.seek(SeekFrom::Start(0))?;
  70
  71         let header_size = std::mem::size_of::<FixedIndexHeader>();
  72
  73         let stat = match nix::sys::stat::fstat(file.as_raw_fd()) {
  74             Ok(stat) => stat,
  75             Err(err) => bail!("fstat failed - {}", err),
  76         };
  77
  78         let size = stat.st_size as usize;
  79
  80         if size < header_size {
  81             bail!("index too small ({})", stat.st_size);
  82         }
  83
  84         let header: Box<FixedIndexHeader> = unsafe { file.read_host_value_boxed()? };
  85
  86         if header.magic != file_formats::FIXED_SIZED_CHUNK_INDEX_1_0 {
  87             bail!("got unknown magic number");
  88         }
  89
  90         let size = u64::from_le(header.size);
  91         let ctime = i64::from_le(header.ctime);
  92         let chunk_size = u64::from_le(header.chunk_size);
  93
  94         let index_length = ((size + chunk_size - 1) / chunk_size) as usize;
  95         let index_size = index_length * 32;
  96
  97         let expected_index_size = (stat.st_size as usize) - header_size;
  98         if index_size != expected_index_size {
  99             bail!(
 100                 "got unexpected file size ({} != {})",
 101                 index_size,
 102                 expected_index_size
 103             );
 104         }
 105
 106         let data = unsafe {
 107             nix::sys::mman::mmap(
 108                 std::ptr::null_mut(),
 109                 index_size,
 110                 nix::sys::mman::ProtFlags::PROT_READ,
 111                 nix::sys::mman::MapFlags::MAP_PRIVATE,
 112                 file.as_raw_fd(),
 113                 header_size as i64,
 114             )
 115         }? as *mut u8;
 116
 117         Ok(Self {
 118             _file: file,
 119             chunk_size: chunk_size as usize,
 120             size,
 121             index_length,
 122             index: data,
 123             ctime,
 124             uuid: header.uuid,
 125             index_csum: header.index_csum,
 126         })
 127     }
 128
 129     fn unmap(&mut self) -> Result<(), Error> {
 130         if self.index.is_null() {
 131             return Ok(());
 132         }
 133
 134         let index_size = self.index_length * 32;
 135
 136         if let Err(err) =
 137             unsafe { nix::sys::mman::munmap(self.index as *mut std::ffi::c_void, index_size) }
 138         {
 139             bail!("unmap file failed - {}", err);
 140         }
 141
 142         self.index = std::ptr::null_mut();
 143
 144         Ok(())
 145     }
 146
 147     pub fn print_info(&self) {
 148         println!("Size: {}", self.size);
 149         println!("ChunkSize: {}", self.chunk_size);
 150
 151         let mut ctime_str = self.ctime.to_string();
 152         if let Ok(s) = proxmox_time::strftime_local("%c", self.ctime) {
 153             ctime_str = s;
 154         }
 155
 156         println!("CTime: {}", ctime_str);
 157         println!("UUID: {:?}", self.uuid);
 158     }
 159 }
 160
 161 impl IndexFile for FixedIndexReader {
 162     fn index_count(&self) -> usize {
 163         self.index_length
 164     }
 165
 166     fn index_digest(&self, pos: usize) -> Option<&[u8; 32]> {
 167         if pos >= self.index_length {
 168             None
 169         } else {
 170             Some(unsafe { &*(self.index.add(pos * 32) as *const [u8; 32]) })
 171         }
 172     }
 173
 174     fn index_bytes(&self) -> u64 {
 175         self.size
 176     }
 177
 178     fn chunk_info(&self, pos: usize) -> Option<ChunkReadInfo> {
 179         if pos >= self.index_length {
 180             return None;
 181         }
 182
 183         let start = (pos * self.chunk_size) as u64;
 184         let mut end = start + self.chunk_size as u64;
 185
 186         if end > self.size {
 187             end = self.size;
 188         }
 189
 190         let digest = self.index_digest(pos).unwrap();
 191         Some(ChunkReadInfo {
 192             range: start..end,
 193             digest: *digest,
 194         })
 195     }
 196
 197     fn index_ctime(&self) -> i64 {
 198         self.ctime
 199     }
 200
 201     fn index_size(&self) -> usize {
 202         self.size as usize
 203     }
 204
 205     fn compute_csum(&self) -> ([u8; 32], u64) {
 206         let mut csum = openssl::sha::Sha256::new();
 207         let mut chunk_end = 0;
 208         for pos in 0..self.index_count() {
 209             let info = self.chunk_info(pos).unwrap();
 210             chunk_end = info.range.end;
 211             csum.update(&info.digest);
 212         }
 213         let csum = csum.finish();
 214
 215         (csum, chunk_end)
 216     }
 217
 218     fn chunk_from_offset(&self, offset: u64) -> Option<(usize, u64)> {
 219         if offset >= self.size {
 220             return None;
 221         }
 222
 223         Some((
 224             (offset / self.chunk_size as u64) as usize,
 225             offset & (self.chunk_size - 1) as u64, // fast modulo, valid for 2^x chunk_size
 226         ))
 227     }
 228 }
 229
 230 pub struct FixedIndexWriter {
 231     store: Arc<ChunkStore>,
 232     file: File,
 233     _lock: ProcessLockSharedGuard,
 234     filename: PathBuf,
 235     tmp_filename: PathBuf,
 236     chunk_size: usize,
 237     size: usize,
 238     index_length: usize,
 239     index: *mut u8,
 240     pub uuid: [u8; 16],
 241     pub ctime: i64,
 242 }
 243
 244 // `index` is mmap()ed which cannot be thread-local so should be sendable
 245 unsafe impl Send for FixedIndexWriter {}
 246
 247 impl Drop for FixedIndexWriter {
 248     fn drop(&mut self) {
 249         let _ = std::fs::remove_file(&self.tmp_filename); // ignore errors
 250         if let Err(err) = self.unmap() {
 251             eprintln!("Unable to unmap file {:?} - {}", self.tmp_filename, err);
 252         }
 253     }
 254 }
 255
 256 impl FixedIndexWriter {
 257     #[allow(clippy::cast_ptr_alignment)]
 258     pub fn create(
 259         store: Arc<ChunkStore>,
 260         path: &Path,
 261         size: usize,
 262         chunk_size: usize,
 263     ) -> Result<Self, Error> {
 264         let shared_lock = store.try_shared_lock()?;
 265
 266         let full_path = store.relative_path(path);
 267         let mut tmp_path = full_path.clone();
 268         tmp_path.set_extension("tmp_fidx");
 269
 270         let mut file = std::fs::OpenOptions::new()
 271             .create(true)
 272             .truncate(true)
 273             .read(true)
 274             .write(true)
 275             .open(&tmp_path)?;
 276
 277         let header_size = std::mem::size_of::<FixedIndexHeader>();
 278
 279         // todo: use static assertion when available in rust
 280         if header_size != 4096 {
 281             panic!("got unexpected header size");
 282         }
 283
 284         let ctime = proxmox_time::epoch_i64();
 285
 286         let uuid = Uuid::generate();
 287
 288         let buffer = vec![0u8; header_size];
 289         let header = unsafe { &mut *(buffer.as_ptr() as *mut FixedIndexHeader) };
 290
 291         header.magic = file_formats::FIXED_SIZED_CHUNK_INDEX_1_0;
 292         header.ctime = i64::to_le(ctime);
 293         header.size = u64::to_le(size as u64);
 294         header.chunk_size = u64::to_le(chunk_size as u64);
 295         header.uuid = *uuid.as_bytes();
 296
 297         header.index_csum = [0u8; 32];
 298
 299         file.write_all(&buffer)?;
 300
 301         let index_length = (size + chunk_size - 1) / chunk_size;
 302         let index_size = index_length * 32;
 303         nix::unistd::ftruncate(file.as_raw_fd(), (header_size + index_size) as i64)?;
 304
 305         let data = unsafe {
 306             nix::sys::mman::mmap(
 307                 std::ptr::null_mut(),
 308                 index_size,
 309                 nix::sys::mman::ProtFlags::PROT_READ | nix::sys::mman::ProtFlags::PROT_WRITE,
 310                 nix::sys::mman::MapFlags::MAP_SHARED,
 311                 file.as_raw_fd(),
 312                 header_size as i64,
 313             )
 314         }? as *mut u8;
 315
 316         Ok(Self {
 317             store,
 318             file,
 319             _lock: shared_lock,
 320             filename: full_path,
 321             tmp_filename: tmp_path,
 322             chunk_size,
 323             size,
 324             index_length,
 325             index: data,
 326             ctime,
 327             uuid: *uuid.as_bytes(),
 328         })
 329     }
 330
 331     pub fn index_length(&self) -> usize {
 332         self.index_length
 333     }
 334
 335     fn unmap(&mut self) -> Result<(), Error> {
 336         if self.index.is_null() {
 337             return Ok(());
 338         }
 339
 340         let index_size = self.index_length * 32;
 341
 342         if let Err(err) =
 343             unsafe { nix::sys::mman::munmap(self.index as *mut std::ffi::c_void, index_size) }
 344         {
 345             bail!("unmap file {:?} failed - {}", self.tmp_filename, err);
 346         }
 347
 348         self.index = std::ptr::null_mut();
 349
 350         Ok(())
 351     }
 352
 353     pub fn close(&mut self) -> Result<[u8; 32], Error> {
 354         if self.index.is_null() {
 355             bail!("cannot close already closed index file.");
 356         }
 357
 358         let index_size = self.index_length * 32;
 359         let data = unsafe { std::slice::from_raw_parts(self.index, index_size) };
 360         let index_csum = openssl::sha::sha256(data);
 361
 362         self.unmap()?;
 363
 364         let csum_offset = proxmox_lang::offsetof!(FixedIndexHeader, index_csum);
 365         self.file.seek(SeekFrom::Start(csum_offset as u64))?;
 366         self.file.write_all(&index_csum)?;
 367         self.file.flush()?;
 368
 369         if let Err(err) = std::fs::rename(&self.tmp_filename, &self.filename) {
 370             bail!("Atomic rename file {:?} failed - {}", self.filename, err);
 371         }
 372
 373         Ok(index_csum)
 374     }
 375
 376     pub fn check_chunk_alignment(&self, offset: usize, chunk_len: usize) -> Result<usize, Error> {
 377         if offset < chunk_len {
 378             bail!("got chunk with small offset ({} < {}", offset, chunk_len);
 379         }
 380
 381         let pos = offset - chunk_len;
 382
 383         if offset > self.size {
 384             bail!("chunk data exceeds size ({} >= {})", offset, self.size);
 385         }
 386
 387         // last chunk can be smaller
 388         if ((offset != self.size) && (chunk_len != self.chunk_size))
 389             || (chunk_len > self.chunk_size)
 390             || (chunk_len == 0)
 391         {
 392             bail!(
 393                 "chunk with unexpected length ({} != {}",
 394                 chunk_len,
 395                 self.chunk_size
 396             );
 397         }
 398
 399         if pos & (self.chunk_size - 1) != 0 {
 400             bail!("got unaligned chunk (pos = {})", pos);
 401         }
 402
 403         Ok(pos / self.chunk_size)
 404     }
 405
 406     // Note: We want to add data out of order, so do not assume any order here.
 407     pub fn add_chunk(&mut self, chunk_info: &ChunkInfo, stat: &mut ChunkStat) -> Result<(), Error> {
 408         let chunk_len = chunk_info.chunk_len as usize;
 409         let offset = chunk_info.offset as usize; // end of chunk
 410
 411         let idx = self.check_chunk_alignment(offset, chunk_len)?;
 412
 413         let (is_duplicate, compressed_size) = self
 414             .store
 415             .insert_chunk(&chunk_info.chunk, &chunk_info.digest)?;
 416
 417         stat.chunk_count += 1;
 418         stat.compressed_size += compressed_size;
 419
 420         let digest = &chunk_info.digest;
 421
 422         println!(
 423             "ADD CHUNK {} {} {}% {} {}",
 424             idx,
 425             chunk_len,
 426             (compressed_size * 100) / (chunk_len as u64),
 427             is_duplicate,
 428             proxmox::tools::digest_to_hex(digest)
 429         );
 430
 431         if is_duplicate {
 432             stat.duplicate_chunks += 1;
 433         } else {
 434             stat.disk_size += compressed_size;
 435         }
 436
 437         self.add_digest(idx, digest)
 438     }
 439
 440     pub fn add_digest(&mut self, index: usize, digest: &[u8; 32]) -> Result<(), Error> {
 441         if index >= self.index_length {
 442             bail!(
 443                 "add digest failed - index out of range ({} >= {})",
 444                 index,
 445                 self.index_length
 446             );
 447         }
 448
 449         if self.index.is_null() {
 450             bail!("cannot write to closed index file.");
 451         }
 452
 453         let index_pos = index * 32;
 454         unsafe {
 455             let dst = self.index.add(index_pos);
 456             dst.copy_from_nonoverlapping(digest.as_ptr(), 32);
 457         }
 458
 459         Ok(())
 460     }
 461
 462     pub fn clone_data_from(&mut self, reader: &FixedIndexReader) -> Result<(), Error> {
 463         if self.index_length != reader.index_count() {
 464             bail!("clone_data_from failed - index sizes not equal");
 465         }
 466
 467         for i in 0..self.index_length {
 468             self.add_digest(i, reader.index_digest(i).unwrap())?;
 469         }
 470
 471         Ok(())
 472     }
 473 }