subfile.cc

   1 /* Cumulus: Smart Filesystem Backup to Dumb Servers
   2  *
   3  * Copyright (C) 2008  The Regents of the University of California
   4  * Written by Michael Vrable <mvrable@cs.ucsd.edu>
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License as published by
   8  * the Free Software Foundation; either version 2 of the License, or
   9  * (at your option) any later version.
  10  *
  11  * This program is distributed in the hope that it will be useful,
  12  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14  * GNU General Public License for more details.
  15  *
  16  * You should have received a copy of the GNU General Public License along
  17  * with this program; if not, write to the Free Software Foundation, Inc.,
  18  * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  19  */
  20
  21 /* Allow for sub-file incremental backups: if only a portion of a file changes,
  22  * allow the new data to be written out, and the old data to simply be
  23  * referenced from the new metadata log. */
  24
  25 #include <stdlib.h>
  26 #include <assert.h>
  27 #include <arpa/inet.h>
  28
  29 #include "subfile.h"
  30 #include "chunk.h"
  31 #include "sha1.h"
  32
  33 using std::list;
  34 using std::map;
  35 using std::set;
  36 using std::string;
  37 using std::vector;
  38 using std::pair;
  39 using std::make_pair;
  40
  41 Subfile::Subfile(LocalDb *localdb)
  42     : db(localdb), checksums_loaded(false), new_block_summary_valid(false)
  43 {
  44 }
  45
  46 Subfile::~Subfile()
  47 {
  48     for (size_t i = 0; i < block_list.size(); i++) {
  49         delete[] block_list[i].chunks;
  50     }
  51
  52     free_analysis();
  53 }
  54
  55 void Subfile::free_analysis()
  56 {
  57     if (new_block_summary_valid)
  58         delete[] new_block_summary.chunks;
  59
  60     new_block_summary_valid = false;
  61 }
  62
  63 void Subfile::load_old_blocks(const list<ObjectReference> &blocks)
  64 {
  65     for (list<ObjectReference>::const_iterator i = blocks.begin();
  66          i != blocks.end(); ++i) {
  67         if (!i->is_normal())
  68             continue;
  69
  70         ObjectReference base = i->base();
  71         if (old_blocks.find(base) == old_blocks.end()) {
  72             old_blocks.insert(base);
  73             if (checksums_loaded)
  74                 index_chunks(base);
  75         }
  76     }
  77 }
  78
  79 /* Actually load chunk signatures from the database, and index them in memory.
  80  * This should only be called once per segment. */
  81 void Subfile::index_chunks(ObjectReference ref)
  82 {
  83     string refstr = ref.to_string();
  84
  85     if (!db->IsAvailable(ref))
  86         return;
  87
  88     /* Look for checksums for this block in the database.  They may not exist,
  89      * in which case simply return without error. */
  90     char *packed_sigs;
  91     size_t len;
  92     string algorithm;
  93     if (!db->LoadChunkSignatures(ref, (void **)&packed_sigs, &len, &algorithm))
  94         return;
  95     if (algorithm != get_algorithm()) {
  96         free(packed_sigs);
  97         return;
  98     }
  99
 100     int block_id = block_list.size();
 101
 102     block_summary summary;
 103     summary.ref = ref.base();
 104     summary.num_chunks = len / (2 + HASH_SIZE);
 105     summary.chunks = new chunk_info[summary.num_chunks];
 106
 107     int block_start = 0;
 108     for (int i = 0; i < summary.num_chunks; i++) {
 109         char *packed_info = &packed_sigs[i * (2 + HASH_SIZE)];
 110         memcpy(summary.chunks[i].hash, &packed_info[2], HASH_SIZE);
 111
 112         uint16_t chunk_len;
 113         memcpy(&chunk_len, &packed_info[0], 2);
 114         summary.chunks[i].len = ntohs(chunk_len);
 115         summary.chunks[i].offset = block_start;
 116         block_start += summary.chunks[i].len;
 117
 118         chunk_index[string(summary.chunks[i].hash, HASH_SIZE)]
 119             = make_pair(block_id, i);
 120     }
 121
 122     block_list.push_back(summary);
 123     free(packed_sigs);
 124 }
 125
 126 /* Signatures can be loaded lazily; this method should be called before any
 127  * actual access to the chunk signatures is required, to ensure the data has
 128  * actually been loaded. */
 129 void Subfile::ensure_signatures_loaded()
 130 {
 131     if (checksums_loaded)
 132         return;
 133
 134     for (set<ObjectReference>::iterator i = old_blocks.begin();
 135          i != old_blocks.end(); ++i) {
 136         index_chunks(*i);
 137     }
 138
 139     checksums_loaded = true;
 140 }
 141
 142 void Subfile::analyze_new_block(const char *buf, size_t len)
 143 {
 144     analyzed_buf = buf;
 145     analyzed_len = len;
 146     int max_chunks = chunk_compute_max_num_breaks(len);
 147
 148     free_analysis();
 149
 150     size_t *breakpoints = new size_t[max_chunks];
 151     int num_breakpoints = chunk_compute_breaks(buf, len, breakpoints);
 152
 153     if (num_breakpoints == 0) {
 154         delete[] breakpoints;
 155         return;
 156     }
 157
 158     new_block_summary.num_chunks = num_breakpoints;
 159     new_block_summary.chunks = new chunk_info[num_breakpoints];
 160
 161     int block_start = 0;
 162     for (int i = 0; i < num_breakpoints; i++) {
 163         new_block_summary.chunks[i].offset = block_start;
 164         new_block_summary.chunks[i].len = breakpoints[i] - block_start + 1;
 165         block_start = breakpoints[i] + 1;
 166
 167         SHA1Checksum hash;
 168         hash.process(&buf[new_block_summary.chunks[i].offset],
 169                      new_block_summary.chunks[i].len);
 170         assert(hash.checksum_size() == (size_t)HASH_SIZE);
 171         memcpy(new_block_summary.chunks[i].hash, hash.checksum(), HASH_SIZE);
 172     }
 173
 174     new_block_summary_valid = true;
 175     delete[] breakpoints;
 176 }
 177
 178 void Subfile::store_block_signatures(ObjectReference ref, block_summary summary)
 179 {
 180     int n = summary.num_chunks;
 181     char *packed = (char *)malloc(n * (2 + HASH_SIZE));
 182
 183     for (int i = 0; i < n; i++) {
 184         assert(summary.chunks[i].len >= 0 && summary.chunks[i].len <= 0xffff);
 185         uint16_t len = htons(summary.chunks[i].len);
 186         char *packed_info = &packed[i * (2 + HASH_SIZE)];
 187         memcpy(&packed_info[0], &len, 2);
 188         memcpy(&packed_info[2], summary.chunks[i].hash, HASH_SIZE);
 189     }
 190
 191     db->StoreChunkSignatures(ref, packed, n * (2 + HASH_SIZE),
 192                              get_algorithm());
 193
 194     free(packed);
 195 }
 196
 197 /* Compute an incremental representation of the most recent block analyzed. */
 198 enum subfile_item_type { SUBFILE_COPY, SUBFILE_NEW };
 199
 200 struct subfile_item {
 201     subfile_item_type type;
 202
 203     // For type SUBFILE_COPY
 204     ObjectReference ref;
 205
 206     // For type SUBFILE_NEW
 207     int src_offset, dst_offset;
 208     int len;
 209     char hash[Subfile::HASH_SIZE];
 210 };
 211
 212 /* Compute an incremental representation of the data last analyzed.  A list of
 213  * references will be returned corresponding to the data.  If new data must be
 214  * written out to the backup, it will be written out via the LbsObject
 215  * provided, to the provided TarSegmentStore. */
 216 list<ObjectReference> Subfile::create_incremental(TarSegmentStore *tss,
 217                                                   LbsObject *o,
 218                                                   double block_age)
 219 {
 220     assert(new_block_summary_valid);
 221     bool matched_old = false;
 222     size_t new_data = 0;
 223
 224     list<subfile_item> items;
 225     list<ObjectReference> refs;
 226
 227     ensure_signatures_loaded();
 228
 229     assert(new_block_summary.num_chunks > 0);
 230
 231     for (int i = 0; i < new_block_summary.num_chunks; i++) {
 232         map<string, pair<int, int> >::iterator m
 233             = chunk_index.find(string(new_block_summary.chunks[i].hash,
 234                                       HASH_SIZE));
 235
 236         struct subfile_item item;
 237         if (m == chunk_index.end()) {
 238             item.type = SUBFILE_NEW;
 239             item.src_offset = new_block_summary.chunks[i].offset;
 240             item.dst_offset = new_data;
 241             item.len = new_block_summary.chunks[i].len;
 242             memcpy(item.hash, new_block_summary.chunks[i].hash, HASH_SIZE);
 243             new_data += item.len;
 244         } else {
 245             struct chunk_info &old_chunk
 246                 = block_list[m->second.first].chunks[m->second.second];
 247             item.type = SUBFILE_COPY;
 248             item.ref = block_list[m->second.first].ref;
 249             item.ref.set_range(old_chunk.offset, old_chunk.len);
 250             matched_old = true;
 251         }
 252
 253         items.push_back(item);
 254     }
 255
 256     // No data was matched.  The entire block can be written out as is into a
 257     // new object, and the new_block_summary used to save chunk signatures.
 258     if (!matched_old) {
 259         SHA1Checksum block_hash;
 260         block_hash.process(analyzed_buf, analyzed_len);
 261         string block_csum = block_hash.checksum_str();
 262
 263         o->set_data(analyzed_buf, analyzed_len);
 264         o->write(tss);
 265         ObjectReference ref = o->get_ref();
 266         db->StoreObject(ref, block_csum, analyzed_len, block_age);
 267         if (analyzed_len >= 16384)
 268             store_block_signatures(ref, new_block_summary);
 269         refs.push_back(ref);
 270         delete o;
 271         return refs;
 272     }
 273
 274     // Otherwise, construct a new block containing all literal data needed (if
 275     // any exists), write it out, and construct the appropriate list of
 276     // references.
 277     list<subfile_item>::iterator i;
 278     if (new_data > 0) {
 279         char *literal_buf = new char[new_data];
 280         for (i = items.begin(); i != items.end(); ++i) {
 281             if (i->type == SUBFILE_NEW) {
 282                 memcpy(&literal_buf[i->dst_offset],
 283                        &analyzed_buf[i->src_offset], i->len);
 284             }
 285         }
 286
 287         SHA1Checksum block_hash;
 288         block_hash.process(literal_buf, new_data);
 289         string block_csum = block_hash.checksum_str();
 290
 291         o->set_group("data");
 292         o->set_data(literal_buf, new_data);
 293         o->write(tss);
 294         ObjectReference ref = o->get_ref();
 295         for (i = items.begin(); i != items.end(); ++i) {
 296             if (i->type == SUBFILE_NEW) {
 297                 i->ref = ref;
 298                 i->ref.set_range(i->dst_offset, i->len);
 299             }
 300         }
 301
 302         db->StoreObject(ref, block_csum, new_data, 0.0);
 303
 304         block_summary summary;
 305         summary.ref = ref;
 306         summary.num_chunks = 0;
 307         summary.chunks = new chunk_info[items.size()];
 308         for (i = items.begin(); i != items.end(); ++i) {
 309             if (i->type == SUBFILE_NEW) {
 310                 chunk_info &info = summary.chunks[summary.num_chunks];
 311                 memcpy(info.hash, i->hash, HASH_SIZE);
 312                 info.offset = i->dst_offset;
 313                 info.len = i->len;
 314                 summary.num_chunks++;
 315             }
 316         }
 317
 318         store_block_signatures(ref, summary);
 319
 320         delete[] summary.chunks;
 321         delete[] literal_buf;
 322     }
 323
 324     delete o;
 325
 326     ObjectReference ref;
 327     for (i = items.begin(); i != items.end(); ++i) {
 328         string refstr = i->ref.to_string();
 329         if (!ref.merge(i->ref)) {
 330             refs.push_back(ref);
 331             ref = i->ref;
 332         }
 333     }
 334     assert(!ref.is_null());
 335     refs.push_back(ref);
 336
 337     return refs;
 338 }