X-Git-Url: http://git.vrable.net/?a=blobdiff_plain;f=cleaner%2Fcleaner;h=4cda7c7f95ec70dd2742d3442625cb2a52f19954;hb=05844fe26f931ffab5f4b320d7554589c3a41618;hp=91037696bbccbb052a30e6af47aaae7b48ca090a;hpb=25f5cec66069a4aebdfc64993590f8c855b2487a;p=bluesky.git diff --git a/cleaner/cleaner b/cleaner/cleaner index 9103769..4cda7c7 100755 --- a/cleaner/cleaner +++ b/cleaner/cleaner @@ -8,7 +8,7 @@ # Copyright (C) 2010 The Regents of the University of California # Written by Michael Vrable -import base64, os, re, struct, sys +import base64, os, re, struct, sys, time import boto from boto.s3.key import Key @@ -19,13 +19,26 @@ HEADER_MAGIC1 = 'AgI-' # Unencrypted data HEADER_MAGIC2 = 'AgI=' # Encrypted data HEADER_SIZE = struct.calcsize(HEADER_FORMAT) +CHECKPOINT_MAGIC = struct.pack(' 0: fp.seek(offset) - if legnth is None: + if length is None: return fp.read() else: return fp.read(length) @@ -60,13 +74,6 @@ class FileBackend: def delete(self, filename): os.unlink(os.path.join(self.path, filename)) - def loc_to_name(self, location): - return "log-%08d-%08d" % (location) - - def name_to_loc(self, name): - m = re.match(r"^log-(\d+)-(\d+)$", name) - if m: return (int(m.group(1)), int(m.group(2))) - def retry_wrap(method): def wrapped(self, *args, **kwargs): for retries in range(3): @@ -75,10 +82,11 @@ def retry_wrap(method): except: print >>sys.stderr, "S3 operation failed, retrying..." self.connect() + time.sleep(1.0) return method(self, *args, **kwargs) return wrapped -class S3Backend: +class S3Backend(Backend): """An interface to BlueSky where the log segments are on in Amazon S3.""" def __init__(self, bucket, path='', cachedir="."): @@ -92,9 +100,10 @@ class S3Backend: self.conn = boto.connect_s3(is_secure=False) self.bucket = self.conn.get_bucket(self.bucket_name) - def list(self): + def list(self, directory=0): files = [] - for k in self.bucket.list(self.path + 'log-'): + prefix = "log-%08d-" % (directory,) + for k in self.bucket.list(self.path + prefix): files.append((k.key, k.size)) return files @@ -138,12 +147,63 @@ class S3Backend: if filename in self.cache: del self.cache[filename] - def loc_to_name(self, location): - return "log-%08d-%08d" % (location) +class SimpleBackend(Backend): + """An interface to the simple BlueSky test network server.""" - def name_to_loc(self, name): - m = re.match(r"^log-(\d+)-(\d+)$", name) - if m: return (int(m.group(1)), int(m.group(2))) + def __init__(self, server=('localhost', 12345), cachedir="."): + self.bucket_name = bucket + self.server_address = server + self.cachedir = cachedir + self.cache = {} + + def _get_socket(self): + return socket.create_connection(self.server_address).makefile() + + def list(self, directory=0): + files = [] + prefix = "log-%08d-" % (directory,) + for k in self.bucket.list(self.path + prefix): + files.append((k.key, k.size)) + return files + + def read(self, filename, offset=0, length=None): + if filename in self.cache: + fp = open(os.path.join(self.cachedir, filename), 'rb') + if offset > 0: + fp.seek(offset) + if length is None: + return fp.read() + else: + return fp.read(length) + else: + f = self._get_socket() + f.write("GET %s %d %d\n" % (filename, 0, 0)) + f.flush() + datalen = int(f.readline()) + if datalen < 0: + raise RuntimeError + data = f.read(datalen) + fp = open(os.path.join(self.cachedir, filename), 'wb') + fp.write(data) + fp.close() + self.cache[filename] = True + if offset > 0: + data = data[offset:] + if length is not None: + data = data[0:length] + return data + + def write(self, filename, data): + f = self._get_socket() + f.write("PUT %s %d %d\n" % (filename, len(data))) + f.write(data) + f.flush() + result = int(f.readline()) + if filename in self.cache: + del self.cache[filename] + + def delete(self, filename): + pass class LogItem: """In-memory representation of a single item stored in a log file.""" @@ -207,7 +267,8 @@ class LogDirectory: self.backend = backend self.dir_num = dir self.seq_num = 0 - for logname in backend.list(): + for logname in backend.list(dir): + print "Old log file:", logname loc = backend.name_to_loc(logname[0]) if loc is not None and loc[0] == dir: self.seq_num = max(self.seq_num, loc[1] + 1) @@ -240,7 +301,7 @@ class UtilizationTracker: def __init__(self, backend): self.segments = {} - for (segment, size) in backend.list(): + for (segment, size) in backend.list(0) + backend.list(1): self.segments[segment] = [size, 0] def add_item(self, item): @@ -326,8 +387,8 @@ def parse_log(data, location=None): if item is not None: yield item offset += size -def load_checkpoint_record(backend): - for (log, size) in reversed(backend.list()): +def load_checkpoint_record(backend, directory=0): + for (log, size) in reversed(backend.list(directory)): for item in reversed(list(parse_log(backend.read(log), log))): print item if item.type == ITEM_TYPE.CHECKPOINT: @@ -342,6 +403,7 @@ class InodeMap: This will also build up information about segment utilization.""" + self.version_vector = {} self.checkpoint_record = checkpoint_record util = UtilizationTracker(backend) @@ -349,9 +411,25 @@ class InodeMap: inodes = {} self.obsolete_segments = set() + data = checkpoint_record.data + if not data.startswith(CHECKPOINT_MAGIC): + raise ValueError, "Invalid checkpoint record!" + data = data[len(CHECKPOINT_MAGIC):] + (vvlen,) = struct.unpack_from(" 0: + if (float(u[1]) / u[0] < 0.6) and u[1] > 0: print "Should clean segment", s loc = backend.name_to_loc(s) if s: inode_map.obsolete_segments.add(loc) @@ -482,13 +566,15 @@ def run_cleaner(backend, inode_map, log, repack_inodes=False): if __name__ == '__main__': backend = S3Backend("mvrable-bluesky", cachedir=".") + #backend = FileBackend(".") chkpt = load_checkpoint_record(backend) print backend.list() imap = InodeMap() imap.build(backend, chkpt) print chkpt - log_dir = LogDirectory(backend, 0) + log_dir = LogDirectory(backend, 1) run_cleaner(backend, imap, log_dir) + print "Version vector:", imap.version_vector imap.write(backend, log_dir) log_dir.close_all()