X-Git-Url: http://git.vrable.net/?a=blobdiff_plain;f=store.cc;h=8f2c8b328c2d7cacea3ca4c3683993532d1a7980;hb=def20364a3596d7b1fa4a07f3d3ee056cfff2d1e;hp=5df4b7797d4a377b21f8ba5995d276938c7fc354;hpb=38c66f088ed65d2f42264c92add6e0b33eac2bfc;p=cumulus.git diff --git a/store.cc b/store.cc index 5df4b77..8f2c8b3 100644 --- a/store.cc +++ b/store.cc @@ -1,145 +1,297 @@ /* LBS: An LFS-inspired filesystem backup system - * Copyright (C) 2006 Michael Vrable + * Copyright (C) 2007 Michael Vrable * * Backup data is stored in a collection of objects, which are grouped together - * into segments for storage purposes. This file provides interfaces for - * reading and writing objects and segments. */ + * into segments for storage purposes. This implementation of the object store + * is built on top of libtar, and represents segments as TAR files and objects + * as files within them. */ #include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include #include "store.h" +#include "ref.h" +using std::max; +using std::list; +using std::map; +using std::set; using std::string; -void OutputStream::write_u8(uint8_t val) +/* Default filter program is bzip2 */ +const char *filter_program = "bzip2 -c"; +const char *filter_extension = ".bz2"; + +static void cloexec(int fd) { - write(&val, 1); + long flags = fcntl(fd, F_GETFD); + + if (flags < 0) + return; + + fcntl(fd, F_SETFD, flags | FD_CLOEXEC); } -void OutputStream::write_u16(uint16_t val) +Tarfile::Tarfile(const string &path, const string &segment) + : size(0), + segment_name(segment) { - unsigned char buf[2]; + real_fd = open(path.c_str(), O_WRONLY | O_CREAT, 0600); + if (real_fd < 0) + throw IOException("Error opening output file"); - buf[0] = val & 0xff; - buf[1] = (val >> 8) & 0xff; - write(buf, 2); + filter_fd = spawn_filter(real_fd); + + if (tar_fdopen(&t, filter_fd, (char *)path.c_str(), NULL, + O_WRONLY | O_CREAT, 0666, TAR_VERBOSE | TAR_GNU) == -1) + throw IOException("Error opening Tarfile"); } -void OutputStream::write_u32(uint32_t val) +Tarfile::~Tarfile() { - unsigned char buf[4]; + /* Close the tar file... */ + tar_append_eof(t); + + if (tar_close(t) != 0) + throw IOException("Error closing Tarfile"); + + /* ...and wait for filter process to finish. */ + int status; + waitpid(filter_pid, &status, 0); - buf[0] = val & 0xff; - buf[1] = (val >> 8) & 0xff; - buf[2] = (val >> 16) & 0xff; - buf[3] = (val >> 24) & 0xff; - write(buf, 4); + if (!WIFEXITED(status) || WEXITSTATUS(status) != 0) { + throw IOException("Filter process error"); + } + + close(real_fd); } -void OutputStream::write_u64(uint64_t val) +/* Launch a child process which can act as a filter (compress, encrypt, etc.) + * on the TAR output. The file descriptor to which output should be written + * must be specified; the return value is the file descriptor which will be + * attached to the standard input of the filter program. */ +int Tarfile::spawn_filter(int fd_out) { - unsigned char buf[8]; + int fds[2]; + + /* Create a pipe for communicating with the filter process. */ + if (pipe(fds) < 0) { + throw IOException("Unable to create pipe for filter"); + } + + /* Create a child process which can exec() the filter program. */ + filter_pid = fork(); + if (filter_pid < 0) + throw IOException("Unable to fork filter process"); + + if (filter_pid > 0) { + /* Parent process */ + close(fds[0]); + cloexec(fds[1]); + } else { + /* Child process. Rearrange file descriptors. stdin is fds[0], stdout + * is fd_out, stderr is unchanged. */ + close(fds[1]); - buf[0] = val & 0xff; - buf[1] = (val >> 8) & 0xff; - buf[2] = (val >> 16) & 0xff; - buf[3] = (val >> 24) & 0xff; - buf[4] = (val >> 32) & 0xff; - buf[5] = (val >> 40) & 0xff; - buf[6] = (val >> 48) & 0xff; - buf[7] = (val >> 56) & 0xff; - write(buf, 8); + if (dup2(fds[0], 0) < 0) + exit(1); + close(fds[0]); + + if (dup2(fd_out, 1) < 0) + exit(1); + close(fd_out); + + /* Exec the filter program. */ + execlp("/bin/sh", "/bin/sh", "-c", filter_program, NULL); + + /* Should not reach here except for error cases. */ + fprintf(stderr, "Could not exec filter: %m\n"); + exit(1); + } + + return fds[1]; } -/* Writes an integer to an output stream using a variable-sized representation: - * seven bits are written at a time (little-endian), and the eigth bit of each - * byte is set if more data follows. */ -void OutputStream::write_varint(uint64_t val) +void Tarfile::write_object(int id, const char *data, size_t len) { - do { - uint8_t remainder = (val & 0x7f); - val >>= 7; - if (val) - remainder |= 0x80; - write_u8(remainder); - } while (val); + char buf[64]; + sprintf(buf, "%08x", id); + string path = segment_name + "/" + buf; + + internal_write_object(path, data, len); } -/* Write an arbitrary string by first writing out the length, followed by the - * data itself. */ -void OutputStream::write_string(const string &s) +void Tarfile::internal_write_object(const string &path, + const char *data, size_t len) { - size_t len = s.length(); - write_varint(len); - write(s.data(), len); + memset(&t->th_buf, 0, sizeof(struct tar_header)); + + th_set_type(t, S_IFREG | 0600); + th_set_user(t, 0); + th_set_group(t, 0); + th_set_mode(t, 0600); + th_set_size(t, len); + th_set_mtime(t, time(NULL)); + th_set_path(t, const_cast(path.c_str())); + th_finish(t); + + if (th_write(t) != 0) + throw IOException("Error writing tar header"); + + size += T_BLOCKSIZE; + + if (len == 0) + return; + + size_t blocks = (len + T_BLOCKSIZE - 1) / T_BLOCKSIZE; + size_t padding = blocks * T_BLOCKSIZE - len; + + for (size_t i = 0; i < blocks - 1; i++) { + if (tar_block_write(t, &data[i * T_BLOCKSIZE]) == -1) + throw IOException("Error writing tar block"); + } + + char block[T_BLOCKSIZE]; + memset(block, 0, sizeof(block)); + memcpy(block, &data[T_BLOCKSIZE * (blocks - 1)], T_BLOCKSIZE - padding); + if (tar_block_write(t, block) == -1) + throw IOException("Error writing final tar block"); + + size += blocks * T_BLOCKSIZE; +} + +/* Estimate the size based on the size of the actual output file on disk. + * However, it might be the case that the filter program is buffering all its + * data, and might potentially not write a single byte until we have closed + * our end of the pipe. If we don't do so until we see data written, we have + * a problem. So, arbitrarily pick an upper bound on the compression ratio + * that the filter will achieve (128:1), and return a size estimate which is + * the larger of a) bytes actually seen written to disk, and b) input + * bytes/128. */ +size_t Tarfile::size_estimate() +{ + struct stat statbuf; + + if (fstat(real_fd, &statbuf) == 0) + return max((int64_t)statbuf.st_size, (int64_t)(size / 128)); + + /* Couldn't stat the file on disk, so just return the actual number of + * bytes, before compression. */ + return size; } -void OutputStream::write_dictionary(const dictionary &d) +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) { - size_t size = d.size(); - size_t written = 0; + struct segment_info *segment; + + // Find the segment into which the object should be written, looking up by + // group. If no segment exists yet, create one. + if (segments.find(group) == segments.end()) { + segment = new segment_info; + + segment->name = generate_uuid(); - write_varint(size); + string filename = path + "/" + segment->name + ".tar"; + filename += filter_extension; + segment->file = new Tarfile(filename, segment->name); - for (dictionary::const_iterator i = d.begin(); i != d.end(); ++i) { - write_string(i->first); - write_string(i->second); - written++; + segment->count = 0; + + segments[group] = segment; + } else { + segment = segments[group]; } - assert(written == size); + int id = segment->count; + char id_buf[64]; + sprintf(id_buf, "%08x", id); + + segment->file->write_object(id, data, len); + segment->count++; + + group_sizes[group] += len; + + ObjectReference ref(segment->name, id_buf); + + // If this segment meets or exceeds the size target, close it so that + // future objects will go into a new segment. + if (segment->file->size_estimate() >= SEGMENT_SIZE) + close_segment(group); + + return ref; } -StringOutputStream::StringOutputStream() - : buf(std::ios_base::out) +void TarSegmentStore::sync() { + while (!segments.empty()) + close_segment(segments.begin()->first); } -void StringOutputStream::write(const void *data, size_t len) +void TarSegmentStore::dump_stats() { - buf.write((const char *)data, len); - if (!buf.good()) - throw IOException("error writing to StringOutputStream"); + printf("Data written:\n"); + for (map::iterator i = group_sizes.begin(); + i != group_sizes.end(); ++i) { + printf(" %s: %lld\n", i->first.c_str(), i->second); + } } -FileOutputStream::FileOutputStream(FILE *file) +void TarSegmentStore::close_segment(const string &group) { - f = file; + struct segment_info *segment = segments[group]; + + delete segment->file; + segments.erase(segments.find(group)); + delete segment; } -FileOutputStream::~FileOutputStream() +string TarSegmentStore::object_reference_to_segment(const string &object) { - fclose(f); + return object; } -void FileOutputStream::write(const void *data, size_t len) +LbsObject::LbsObject() + : group(""), data(NULL), data_len(0), written(false) { - size_t res; - - res = fwrite(data, 1, len, f); - if (res != len) { - throw IOException("write error"); - } } -/* Utility functions, for encoding data types to strings. */ -string encode_u16(uint16_t val) +LbsObject::~LbsObject() { - StringOutputStream s; - s.write_u16(val); - return s.contents(); } -string encode_u32(uint32_t val) +void LbsObject::write(TarSegmentStore *store) { - StringOutputStream s; - s.write_u32(val); - return s.contents(); + assert(data != NULL); + assert(!written); + + ref = store->write_object(data, data_len, group); + written = true; } -string encode_u64(uint64_t val) +void LbsObject::checksum() { - StringOutputStream s; - s.write_u64(val); - return s.contents(); + assert(written); + + SHA1Checksum hash; + hash.process(data, data_len); + ref.set_checksum(hash.checksum_str()); }