src/backup/fixed_index.rs

   1 use failure::*;
   2
   3 use crate::tools;
   4 use super::IndexFile;
   5 use super::chunk_stat::*;
   6 use super::chunk_store::*;
   7
   8 use std::sync::Arc;
   9 use std::io::{Read, Write};
  10 use std::fs::File;
  11 use std::path::{Path, PathBuf};
  12 use std::os::unix::io::AsRawFd;
  13 use uuid::Uuid;
  14 use chrono::{Local, TimeZone};
  15 use super::ChunkInfo;
  16
  17 /// Header format definition for fixed index files (`.fidx`)
  18 #[repr(C)]
  19 pub struct FixedIndexHeader {
  20     pub magic: [u8; 8],
  21     pub uuid: [u8; 16],
  22     pub ctime: u64,
  23     /// Sha256 over the index ``SHA256(digest1||digest2||...)``
  24     pub index_csum: [u8; 32],
  25     pub size: u64,
  26     pub chunk_size: u64,
  27     reserved: [u8; 4016], // overall size is one page (4096 bytes)
  28 }
  29
  30 // split image into fixed size chunks
  31
  32 pub struct FixedIndexReader {
  33     store: Arc<ChunkStore>,
  34     _file: File,
  35     filename: PathBuf,
  36     pub chunk_size: usize,
  37     pub size: usize,
  38     index_length: usize,
  39     index: *mut u8,
  40     pub uuid: [u8; 16],
  41     pub ctime: u64,
  42     pub index_csum: [u8; 32],
  43 }
  44
  45 // `index` is mmap()ed which cannot be thread-local so should be sendable
  46 unsafe impl Send for FixedIndexReader {}
  47
  48 impl Drop for FixedIndexReader {
  49
  50     fn drop(&mut self) {
  51         if let Err(err) = self.unmap() {
  52             eprintln!("Unable to unmap file {:?} - {}", self.filename, err);
  53         }
  54     }
  55 }
  56
  57 impl FixedIndexReader {
  58
  59     pub fn open(store: Arc<ChunkStore>, path: &Path) -> Result<Self, Error> {
  60
  61         let full_path = store.relative_path(path);
  62
  63         let mut file = File::open(&full_path)
  64             .map_err(|err| format_err!("Unable to open fixed index {:?} - {}", full_path, err))?;
  65
  66         if let Err(err) = nix::fcntl::flock(file.as_raw_fd(), nix::fcntl::FlockArg::LockSharedNonblock) {
  67             bail!("unable to get shared lock on {:?} - {}", full_path, err);
  68         }
  69
  70         let header_size = std::mem::size_of::<FixedIndexHeader>();
  71
  72         // todo: use static assertion when available in rust
  73         if header_size != 4096 { bail!("got unexpected header size for {:?}", path); }
  74
  75         let mut buffer = vec![0u8; header_size];
  76         file.read_exact(&mut buffer)?;
  77
  78         let header = unsafe { &mut * (buffer.as_ptr() as *mut FixedIndexHeader) };
  79
  80         if header.magic != super::FIXED_SIZED_CHUNK_INDEX_1_0 {
  81             bail!("got unknown magic number for {:?}", path);
  82         }
  83
  84         let size = u64::from_le(header.size) as usize;
  85         let ctime = u64::from_le(header.ctime);
  86         let chunk_size = u64::from_le(header.chunk_size) as usize;
  87
  88         let index_length = (size + chunk_size - 1)/chunk_size;
  89         let index_size = index_length*32;
  90
  91         let rawfd = file.as_raw_fd();
  92
  93         let stat = match nix::sys::stat::fstat(rawfd) {
  94             Ok(stat) => stat,
  95             Err(err) => bail!("fstat {:?} failed - {}", path, err),
  96         };
  97
  98         let expected_index_size = (stat.st_size as usize) - header_size;
  99         if index_size != expected_index_size {
 100             bail!("got unexpected file size for {:?} ({} != {})",
 101                   path, index_size, expected_index_size);
 102         }
 103
 104         let data = unsafe { nix::sys::mman::mmap(
 105             std::ptr::null_mut(),
 106             index_size,
 107             nix::sys::mman::ProtFlags::PROT_READ,
 108             nix::sys::mman::MapFlags::MAP_PRIVATE,
 109             file.as_raw_fd(),
 110             header_size as i64) }? as *mut u8;
 111
 112         Ok(Self {
 113             store,
 114             filename: full_path,
 115             _file: file,
 116             chunk_size,
 117             size,
 118             index_length,
 119             index: data,
 120             ctime,
 121             uuid: header.uuid,
 122             index_csum: header.index_csum,
 123         })
 124     }
 125
 126     fn unmap(&mut self) -> Result<(), Error> {
 127
 128         if self.index == std::ptr::null_mut() { return Ok(()); }
 129
 130         let index_size = self.index_length*32;
 131
 132         if let Err(err) = unsafe { nix::sys::mman::munmap(self.index as *mut std::ffi::c_void, index_size) } {
 133             bail!("unmap file {:?} failed - {}", self.filename, err);
 134         }
 135
 136         self.index = std::ptr::null_mut();
 137
 138         Ok(())
 139     }
 140
 141     pub fn print_info(&self) {
 142         println!("Filename: {:?}", self.filename);
 143         println!("Size: {}", self.size);
 144         println!("ChunkSize: {}", self.chunk_size);
 145         println!("CTime: {}", Local.timestamp(self.ctime as i64, 0).format("%c"));
 146         println!("UUID: {:?}", self.uuid);
 147     }
 148 }
 149
 150 impl IndexFile for FixedIndexReader {
 151     fn index_count(&self) -> usize {
 152         self.index_length
 153     }
 154
 155     fn index_digest(&self, pos: usize) -> Option<&[u8; 32]> {
 156         if pos >= self.index_length {
 157             None
 158         } else {
 159             Some(unsafe { std::mem::transmute(self.index.add(pos*32)) })
 160         }
 161     }
 162
 163     fn index_bytes(&self) -> u64 {
 164         (self.index_length * self.chunk_size) as u64
 165     }
 166 }
 167
 168 pub struct FixedIndexWriter {
 169     store: Arc<ChunkStore>,
 170     file: File,
 171     _lock: tools::ProcessLockSharedGuard,
 172     filename: PathBuf,
 173     tmp_filename: PathBuf,
 174     chunk_size: usize,
 175     size: usize,
 176     index_length: usize,
 177     index: *mut u8,
 178     pub uuid: [u8; 16],
 179     pub ctime: u64,
 180 }
 181
 182 // `index` is mmap()ed which cannot be thread-local so should be sendable
 183 unsafe impl Send for FixedIndexWriter {}
 184
 185 impl Drop for FixedIndexWriter {
 186
 187     fn drop(&mut self) {
 188         let _ = std::fs::remove_file(&self.tmp_filename); // ignore errors
 189         if let Err(err) = self.unmap() {
 190             eprintln!("Unable to unmap file {:?} - {}", self.tmp_filename, err);
 191         }
 192     }
 193 }
 194
 195 impl FixedIndexWriter {
 196
 197     pub fn create(store: Arc<ChunkStore>, path: &Path, size: usize, chunk_size: usize) -> Result<Self, Error> {
 198
 199         let shared_lock = store.try_shared_lock()?;
 200
 201         let full_path = store.relative_path(path);
 202         let mut tmp_path = full_path.clone();
 203         tmp_path.set_extension("tmp_fidx");
 204
 205         let mut file = std::fs::OpenOptions::new()
 206             .create(true).truncate(true)
 207             .read(true)
 208             .write(true)
 209             .open(&tmp_path)?;
 210
 211         let header_size = std::mem::size_of::<FixedIndexHeader>();
 212
 213         // todo: use static assertion when available in rust
 214         if header_size != 4096 { panic!("got unexpected header size"); }
 215
 216         let ctime = std::time::SystemTime::now().duration_since(
 217             std::time::SystemTime::UNIX_EPOCH)?.as_secs();
 218
 219         let uuid = Uuid::new_v4();
 220
 221         let buffer = vec![0u8; header_size];
 222         let header = unsafe { &mut * (buffer.as_ptr() as *mut FixedIndexHeader) };
 223
 224         header.magic = super::FIXED_SIZED_CHUNK_INDEX_1_0;
 225         header.ctime = u64::to_le(ctime);
 226         header.size = u64::to_le(size as u64);
 227         header.chunk_size = u64::to_le(chunk_size as u64);
 228         header.uuid = *uuid.as_bytes();
 229
 230         header.index_csum = [0u8; 32];
 231
 232         file.write_all(&buffer)?;
 233
 234         let index_length = (size + chunk_size - 1)/chunk_size;
 235         let index_size = index_length*32;
 236         nix::unistd::ftruncate(file.as_raw_fd(), (header_size + index_size) as i64)?;
 237
 238         let data = unsafe { nix::sys::mman::mmap(
 239             std::ptr::null_mut(),
 240             index_size,
 241             nix::sys::mman::ProtFlags::PROT_READ | nix::sys::mman::ProtFlags::PROT_WRITE,
 242             nix::sys::mman::MapFlags::MAP_SHARED,
 243             file.as_raw_fd(),
 244             header_size as i64) }? as *mut u8;
 245
 246         Ok(Self {
 247             store,
 248             file,
 249             _lock: shared_lock,
 250             filename: full_path,
 251             tmp_filename: tmp_path,
 252             chunk_size,
 253             size,
 254             index_length,
 255             index: data,
 256             ctime,
 257             uuid: *uuid.as_bytes(),
 258         })
 259     }
 260
 261     pub fn index_length(&self) -> usize {
 262         self.index_length
 263     }
 264
 265     fn unmap(&mut self) -> Result<(), Error> {
 266
 267         if self.index == std::ptr::null_mut() { return Ok(()); }
 268
 269         let index_size = self.index_length*32;
 270
 271         if let Err(err) = unsafe { nix::sys::mman::munmap(self.index as *mut std::ffi::c_void, index_size) } {
 272             bail!("unmap file {:?} failed - {}", self.tmp_filename, err);
 273         }
 274
 275         self.index = std::ptr::null_mut();
 276
 277         Ok(())
 278     }
 279
 280     pub fn close(&mut self)  -> Result<[u8; 32], Error> {
 281
 282         if self.index == std::ptr::null_mut() { bail!("cannot close already closed index file."); }
 283
 284         let index_size = self.index_length*32;
 285         let data = unsafe { std::slice::from_raw_parts(self.index, index_size) };
 286         let index_csum =  openssl::sha::sha256(data);
 287
 288         self.unmap()?;
 289
 290         use std::io::Seek;
 291
 292         let csum_offset = proxmox::tools::offsetof!(FixedIndexHeader, index_csum);
 293         self.file.seek(std::io::SeekFrom::Start(csum_offset as u64))?;
 294         self.file.write_all(&index_csum)?;
 295         self.file.flush()?;
 296
 297         if let Err(err) = std::fs::rename(&self.tmp_filename, &self.filename) {
 298             bail!("Atomic rename file {:?} failed - {}", self.filename, err);
 299         }
 300
 301         Ok(index_csum)
 302     }
 303
 304     // Note: We want to add data out of order, so do not assume any order here.
 305     pub fn add_chunk(&mut self, chunk_info: &ChunkInfo, stat: &mut ChunkStat) -> Result<(), Error> {
 306
 307         let chunk_len = chunk_info.chunk_len as usize;
 308         let end = chunk_info.offset as usize;
 309
 310         if end < chunk_len {
 311             bail!("got chunk with small offset ({} < {}", end, chunk_len);
 312         }
 313
 314         let pos = end - chunk_len;
 315
 316         if end > self.size {
 317             bail!("write chunk data exceeds size ({} >= {})", end, self.size);
 318         }
 319
 320         // last chunk can be smaller
 321         if ((end != self.size) && (chunk_len != self.chunk_size)) ||
 322             (chunk_len > self.chunk_size) || (chunk_len == 0) {
 323                 bail!("got chunk with wrong length ({} != {}", chunk_len, self.chunk_size);
 324             }
 325
 326         if pos & (self.chunk_size-1) != 0 { bail!("add unaligned chunk (pos = {})", pos); }
 327
 328         if (end as u64) != chunk_info.offset {
 329             bail!("got chunk with wrong offset ({} != {}", end, chunk_info.offset);
 330         }
 331
 332         let (is_duplicate, compressed_size) = self.store.insert_chunk(&chunk_info.chunk)?;
 333
 334         stat.chunk_count += 1;
 335         stat.compressed_size += compressed_size;
 336
 337         let digest = chunk_info.chunk.digest();
 338
 339         println!("ADD CHUNK {} {} {}% {} {}", pos, chunk_len,
 340                  (compressed_size*100)/(chunk_len as u64), is_duplicate, proxmox::tools::digest_to_hex(digest));
 341
 342         if is_duplicate {
 343             stat.duplicate_chunks += 1;
 344         } else {
 345             stat.disk_size += compressed_size;
 346         }
 347
 348         self.add_digest(pos / self.chunk_size, digest)
 349     }
 350
 351     pub fn add_digest(&mut self, index: usize, digest: &[u8; 32]) -> Result<(), Error> {
 352
 353         if index >= self.index_length {
 354             bail!("add digest failed - index out of range ({} >= {})", index, self.index_length);
 355         }
 356
 357         if self.index == std::ptr::null_mut() { bail!("cannot write to closed index file."); }
 358
 359         let index_pos = index*32;
 360         unsafe {
 361             let dst = self.index.add(index_pos);
 362             dst.copy_from_nonoverlapping(digest.as_ptr(), 32);
 363         }
 364
 365         Ok(())
 366     }
 367 }