src/backup/chunk_store.rs

   1 use failure::*;
   2 use std::path::{Path, PathBuf};
   3 use std::io::Write;
   4
   5 use crypto::digest::Digest;
   6 use crypto::sha2::Sha512Trunc256;
   7 use std::sync::Mutex;
   8
   9 use std::fs::{File, OpenOptions};
  10 use nix::fcntl::{flock, FlockArg};
  11 use std::os::unix::io::AsRawFd;
  12
  13 pub struct ChunkStore {
  14     base: PathBuf,
  15     chunk_dir: PathBuf,
  16     hasher: Sha512Trunc256,
  17     mutex: Mutex<bool>,
  18     lockfile: File,
  19 }
  20
  21 const HEX_CHARS: &'static [u8; 16] = b"0123456789abcdef";
  22
  23 fn u256_to_hex(digest: &[u8; 32]) -> String {
  24
  25     let mut buf = Vec::<u8>::with_capacity(64);
  26
  27     for i in 0..32 {
  28         buf.push(HEX_CHARS[(digest[i] >> 4) as usize]);
  29         buf.push(HEX_CHARS[(digest[i] & 0xf) as usize]);
  30     }
  31
  32     unsafe { String::from_utf8_unchecked(buf) }
  33 }
  34
  35 fn u256_to_prefix(digest: &[u8; 32]) -> PathBuf {
  36
  37     let mut buf = Vec::<u8>::with_capacity(3+1+2+1);
  38
  39     buf.push(HEX_CHARS[(digest[0] as usize) >> 4]);
  40     buf.push(HEX_CHARS[(digest[0] as usize) &0xf]);
  41     buf.push(HEX_CHARS[(digest[1] as usize) >> 4]);
  42     buf.push('/' as u8);
  43
  44     buf.push(HEX_CHARS[(digest[1] as usize) & 0xf]);
  45     buf.push(HEX_CHARS[(digest[2] as usize) >> 4]);
  46     buf.push('/' as u8);
  47
  48     let path = unsafe { String::from_utf8_unchecked(buf)};
  49
  50     path.into()
  51 }
  52
  53 impl ChunkStore {
  54
  55     fn chunk_dir<P: AsRef<Path>>(path: P) -> PathBuf {
  56
  57         let mut chunk_dir: PathBuf = PathBuf::from(path.as_ref());
  58         chunk_dir.push(".chunks");
  59
  60         chunk_dir
  61     }
  62
  63     pub fn create<P: Into<PathBuf>>(path: P) -> Result<Self, Error> {
  64
  65         let base: PathBuf = path.into();
  66         let chunk_dir = Self::chunk_dir(&base);
  67
  68         if let Err(err) = std::fs::create_dir(&base) {
  69             bail!("unable to create chunk store {:?} - {}", base, err);
  70         }
  71
  72         if let Err(err) = std::fs::create_dir(&chunk_dir) {
  73             bail!("unable to create chunk store subdir {:?} - {}", chunk_dir, err);
  74         }
  75
  76         // create 4096 subdir
  77         for i in 0..4096 {
  78             let mut l1path = base.clone();
  79             l1path.push(format!("{:03x}",i));
  80             if let Err(err) = std::fs::create_dir(&l1path) {
  81                 bail!("unable to create chunk subdir {:?} - {}", l1path, err);
  82             }
  83         }
  84
  85         Self::open(base)
  86     }
  87
  88     pub fn open<P: Into<PathBuf>>(path: P) -> Result<Self, Error> {
  89
  90         let base: PathBuf = path.into();
  91         let chunk_dir = Self::chunk_dir(&base);
  92
  93         let metadata = match std::fs::metadata(&chunk_dir) {
  94             Ok(data) => data,
  95             Err(err) => bail!("unable to open chunk store {:?} - {}", chunk_dir, err),
  96         };
  97
  98         let mut lockfile_path = base.clone();
  99         lockfile_path.push(".lock");
 100
 101         let lockfile = match OpenOptions::new()
 102             .create(true)
 103             .append(true)
 104             .open(&lockfile_path) {
 105                 Ok(file) => file,
 106                 Err(err) => bail!("unable to open chunk store lock file {:?} - {}",
 107                                   lockfile_path, err),
 108             };
 109
 110         let fd = lockfile.as_raw_fd();
 111
 112         let now = std::time::SystemTime::now();
 113         let timeout = 10;
 114         let mut print_msg = true;
 115         loop {
 116             match flock(fd, FlockArg::LockExclusiveNonblock) {
 117                 Ok(_) => break,
 118                 Err(_) => {
 119                     if print_msg {
 120                         print_msg = false;
 121                         eprintln!("trying to aquire lock...");
 122                     }
 123                 }
 124             }
 125
 126             match now.elapsed() {
 127                 Ok(elapsed) => {
 128                     if elapsed.as_secs() >= timeout {
 129                         bail!("unable to aquire chunk store lock {:?} - got timeout",
 130                               lockfile_path);
 131                     }
 132                 }
 133                 Err(err) => {
 134                     bail!("unable to aquire chunk store lock {:?} - clock problems - {}",
 135                           lockfile_path, err);
 136                 }
 137             }
 138             std::thread::sleep_ms(100);
 139         }
 140
 141         Ok(ChunkStore {
 142             base,
 143             chunk_dir,
 144             hasher: Sha512Trunc256::new(),
 145             lockfile,
 146             mutex: Mutex::new(false)
 147         })
 148     }
 149
 150     pub fn insert_chunk(&mut self, chunk: &[u8]) -> Result<([u8; 32]), Error> {
 151
 152         self.hasher.reset();
 153         self.hasher.input(chunk);
 154
 155         let mut digest = [0u8; 32];
 156         self.hasher.result(&mut digest);
 157         println!("DIGEST {}", u256_to_hex(&digest));
 158
 159         let mut chunk_path = self.base.clone();
 160         let prefix = u256_to_prefix(&digest);
 161         chunk_path.push(&prefix);
 162         let digest_str = u256_to_hex(&digest);
 163         chunk_path.push(&digest_str);
 164
 165         let lock = self.mutex.lock();
 166
 167         if let Ok(metadata) = std::fs::metadata(&chunk_path) {
 168             if metadata.is_file() {
 169                  return Ok(digest);
 170             } else {
 171                 bail!("Got unexpected file type for chunk {}", digest_str);
 172             }
 173         }
 174
 175         let mut chunk_dir = self.base.clone();
 176         chunk_dir.push(&prefix);
 177
 178         if let Err(_) = std::fs::create_dir(&chunk_dir) { /* ignore */ }
 179
 180         let mut tmp_path = chunk_path.clone();
 181         tmp_path.set_extension("tmp");
 182         let mut f = std::fs::File::create(&tmp_path)?;
 183         f.write_all(chunk)?;
 184
 185         if let Err(err) = std::fs::rename(&tmp_path, &chunk_path) {
 186             if let Err(_) = std::fs::remove_file(&tmp_path)  { /* ignore */ }
 187             bail!("Atomic rename failed for chunk {} - {}", digest_str, err);
 188         }
 189
 190         println!("PATH {:?}", chunk_path);
 191
 192         drop(lock);
 193
 194         Ok(digest)
 195     }
 196
 197 }
 198
 199
 200 #[test]
 201 fn test_chunk_store1() {
 202
 203     if let Err(_e) = std::fs::remove_dir_all(".testdir") { /* ignore */ }
 204
 205     let chunk_store = ChunkStore::open(".testdir");
 206     assert!(chunk_store.is_err());
 207
 208     let mut chunk_store = ChunkStore::create(".testdir").unwrap();
 209     chunk_store.insert_chunk(&[0u8, 1u8]).unwrap();
 210     chunk_store.insert_chunk(&[0u8, 1u8]).unwrap();
 211
 212
 213     let chunk_store = ChunkStore::create(".testdir");
 214     assert!(chunk_store.is_err());
 215
 216
 217 }