X-Git-Url: http://git.vrable.net/?p=cumulus.git;a=blobdiff_plain;f=store.cc;h=772f9c504493229792d2b9f6aae242dfbdcdc4c4;hp=720d3bb7da8a6abd4e7e42be35c0bb7c733660e5;hb=3d780590edec4583eb3ef0ca16120afd0f7451f9;hpb=0dfc70e01ddb7d2bce0db03d5364c0bd3a2bb308 diff --git a/store.cc b/store.cc index 720d3bb..772f9c5 100644 --- a/store.cc +++ b/store.cc @@ -1,7 +1,6 @@ -/* Cumulus: Smart Filesystem Backup to Dumb Servers - * - * Copyright (C) 2008 The Regents of the University of California - * Written by Michael Vrable +/* Cumulus: Efficient Filesystem Backup to the Cloud + * Copyright (C) 2008-2009 The Cumulus Developers + * See the AUTHORS file for a list of contributors. * * This program is free software; you can redistribute it and/or modify * it under the terms of the GNU General Public License as published by @@ -25,6 +24,7 @@ #include #include #include +#include #include #include #include @@ -40,8 +40,11 @@ #include #include +#include "hash.h" +#include "localdb.h" #include "store.h" #include "ref.h" +#include "util.h" using std::max; using std::list; @@ -54,16 +57,6 @@ using std::string; const char *filter_program = "bzip2 -c"; const char *filter_extension = ".bz2"; -static void cloexec(int fd) -{ - long flags = fcntl(fd, F_GETFD); - - if (flags < 0) - return; - - fcntl(fd, F_SETFD, flags | FD_CLOEXEC); -} - Tarfile::Tarfile(RemoteFile *file, const string &segment) : size(0), segment_name(segment) @@ -85,14 +78,14 @@ Tarfile::~Tarfile() tar_write(buf, TAR_BLOCK_SIZE); if (close(filter_fd) != 0) - throw IOException("Error closing Tarfile"); + fatal("Error closing Tarfile"); /* ...and wait for filter process to finish. */ int status; waitpid(filter_pid, &status, 0); if (!WIFEXITED(status) || WEXITSTATUS(status) != 0) { - throw IOException("Filter process error"); + fatal("Filter process error"); } close(real_fd); @@ -109,13 +102,13 @@ int spawn_filter(int fd_out, const char *program, pid_t *filter_pid) /* Create a pipe for communicating with the filter process. */ if (pipe(fds) < 0) { - throw IOException("Unable to create pipe for filter"); + fatal("Unable to create pipe for filter"); } /* Create a child process which can exec() the filter program. */ pid = fork(); if (pid < 0) - throw IOException("Unable to fork filter process"); + fatal("Unable to fork filter process"); if (pid > 0) { /* Parent process */ @@ -158,7 +151,7 @@ void Tarfile::tar_write(const char *data, size_t len) if (errno == EINTR) continue; fprintf(stderr, "Write error: %m\n"); - throw IOException("Write error"); + fatal("Write error"); } len -= res; @@ -180,7 +173,7 @@ void Tarfile::write_object(int id, const char *data, size_t len) sprintf(header.mode, "%07o", 0600); sprintf(header.uid, "%07o", 0); sprintf(header.gid, "%07o", 0); - sprintf(header.size, "%011o", len); + sprintf(header.size, "%011o", (int)len); sprintf(header.mtime, "%011o", (int)time(NULL)); header.typeflag = '0'; strcpy(header.magic, "ustar "); @@ -234,7 +227,9 @@ static const size_t SEGMENT_SIZE = 4 * 1024 * 1024; static map > group_sizes; ObjectReference TarSegmentStore::write_object(const char *data, size_t len, - const std::string &group) + const std::string &group, + const std::string &checksum, + double age) { struct segment_info *segment; @@ -248,7 +243,7 @@ ObjectReference TarSegmentStore::write_object(const char *data, size_t len, segment->basename = segment->name + ".tar"; segment->basename += filter_extension; segment->count = 0; - segment->size = 0; + segment->data_size = 0; segment->rf = remote->alloc_file(segment->basename, "segments"); segment->file = new Tarfile(segment->rf, segment->name); @@ -263,11 +258,16 @@ ObjectReference TarSegmentStore::write_object(const char *data, size_t len, segment->file->write_object(id, data, len); segment->count++; - segment->size += len; + segment->data_size += len; group_sizes[group].first += len; ObjectReference ref(segment->name, id_buf); + ref.set_range(0, len, true); + if (checksum.size() > 0) + ref.set_checksum(checksum); + if (db != NULL) + db->StoreObject(ref, age); // If this segment meets or exceeds the size target, close it so that // future objects will go into a new segment. @@ -289,7 +289,7 @@ void TarSegmentStore::dump_stats() for (map >::iterator i = group_sizes.begin(); i != group_sizes.end(); ++i) { printf(" %s: %lld (%lld compressed)\n", i->first.c_str(), - i->second.first, i->second.second); + (long long)i->second.first, (long long)i->second.second); } } @@ -300,17 +300,21 @@ void TarSegmentStore::close_segment(const string &group) delete segment->file; if (db != NULL) { + struct stat stat_buf; + int disk_size = 0; + if (stat(segment->rf->get_local_path().c_str(), &stat_buf) == 0) { + disk_size = stat_buf.st_size; + group_sizes[segment->group].second += disk_size; + } + SHA1Checksum segment_checksum; + string checksum; if (segment_checksum.process_file(segment->rf->get_local_path().c_str())) { - string checksum = segment_checksum.checksum_str(); - db->SetSegmentChecksum(segment->name, segment->basename, checksum, - segment->size); + checksum = segment_checksum.checksum_str(); } - struct stat stat_buf; - if (stat(segment->rf->get_local_path().c_str(), &stat_buf) == 0) { - group_sizes[segment->group].second += stat_buf.st_size; - } + db->SetSegmentChecksum(segment->name, segment->basename, checksum, + segment->data_size, disk_size); } segment->rf->send(); @@ -325,7 +329,7 @@ string TarSegmentStore::object_reference_to_segment(const string &object) } LbsObject::LbsObject() - : group(""), data(NULL), data_len(0), written(false) + : group(""), age(0.0), data(NULL), data_len(0), written(false) { } @@ -333,20 +337,26 @@ LbsObject::~LbsObject() { } -void LbsObject::write(TarSegmentStore *store) +void LbsObject::set_data(const char *d, size_t len, const char *checksum) { - assert(data != NULL); - assert(!written); + data = d; + data_len = len; - ref = store->write_object(data, data_len, group); - written = true; + if (checksum != NULL) { + this->checksum = checksum; + } else { + Hash *hash = Hash::New(); + hash->update(data, data_len); + this->checksum = hash->digest_str(); + delete hash; + } } -void LbsObject::checksum() +void LbsObject::write(TarSegmentStore *store) { - assert(written); + assert(data != NULL); + assert(!written); - SHA1Checksum hash; - hash.process(data, data_len); - ref.set_checksum(hash.checksum_str()); + ref = store->write_object(data, data_len, group, checksum, age); + written = true; }