X-Git-Url: http://git.vrable.net/?a=blobdiff_plain;f=cleaner%2Fcleaner;h=7b50c73e7a1113317f822b963739b66ca254f2de;hb=8ff0fd08d6e1cc97cdb7e94b7cd97dc28c29e674;hp=675819bad7845f016454f9a6c70a98d40c9258dc;hpb=4a22b094a5dbd2aabc101ee5a6579b577b6e614c;p=bluesky.git diff --git a/cleaner/cleaner b/cleaner/cleaner index 675819b..7b50c73 100755 --- a/cleaner/cleaner +++ b/cleaner/cleaner @@ -7,6 +7,30 @@ # # Copyright (C) 2010 The Regents of the University of California # Written by Michael Vrable +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions +# are met: +# 1. Redistributions of source code must retain the above copyright +# notice, this list of conditions and the following disclaimer. +# 2. Redistributions in binary form must reproduce the above copyright +# notice, this list of conditions and the following disclaimer in the +# documentation and/or other materials provided with the distribution. +# 3. Neither the name of the University nor the names of its contributors +# may be used to endorse or promote products derived from this software +# without specific prior written permission. +# +# THIS SOFTWARE IS PROVIDED BY THE REGENTS AND CONTRIBUTORS ``AS IS'' AND +# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE +# ARE DISCLAIMED. IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE LIABLE +# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL +# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS +# OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) +# HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT +# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY +# OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF +# SUCH DAMAGE. import base64, os, re, struct, sys, time import boto @@ -21,13 +45,35 @@ HEADER_SIZE = struct.calcsize(HEADER_FORMAT) CHECKPOINT_MAGIC = struct.pack('>sys.stderr, "S3 operation failed, retrying..." + print >>sys.stderr, " %s %s %s" % (method, args, kwargs) 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="."): @@ -90,7 +130,12 @@ class S3Backend: self.path = path self.cachedir = cachedir self.cache = {} + for f in os.listdir(cachedir): + self.cache[f] = True + #print "Initial cache contents:", list(self.cache.keys()) self.connect() + self.stats_get = [0, 0] + self.stats_put = [0, 0] def connect(self): self.conn = boto.connect_s3(is_secure=False) @@ -121,6 +166,8 @@ class S3Backend: fp.write(data) fp.close() self.cache[filename] = True + self.stats_get[0] += 1 + self.stats_get[1] += len(data) if offset > 0: data = data[offset:] if length is not None: @@ -132,6 +179,8 @@ class S3Backend: k = Key(self.bucket) k.key = self.path + filename k.set_contents_from_string(data) + self.stats_put[0] += 1 + self.stats_put[1] += len(data) if filename in self.cache: del self.cache[filename] @@ -143,12 +192,70 @@ class S3Backend: if filename in self.cache: del self.cache[filename] - def loc_to_name(self, location): - return "log-%08d-%08d" % (location) + def dump_stats(self): + print "S3 statistics:" + print "GET: %d ops / %d bytes" % tuple(self.stats_get) + print "PUT: %d ops / %d bytes" % tuple(self.stats_put) + benchlog_write("s3_get: %d", self.stats_get[1]) + benchlog_write("s3_put: %d", self.stats_put[1]) - 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))) +class SimpleBackend(Backend): + """An interface to the simple BlueSky test network server.""" + + 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.""" @@ -213,7 +320,7 @@ class LogDirectory: self.dir_num = dir self.seq_num = 0 for logname in backend.list(dir): - print "Old log file:", logname + #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) @@ -372,12 +479,12 @@ class InodeMap: data = data[self.vvsize:] - print "Inode map:" + #print "Inode map:" for i in range(len(data) // 16): (start, end) = 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) @@ -491,7 +619,7 @@ def run_cleaner(backend, inode_map, log, repack_inodes=False): dirty_inode_data = set() for s in inode_map.obsolete_segments: filename = backend.loc_to_name(s) - print "Scanning", filename, "for live data" + #print "Scanning", filename, "for live data" for item in parse_log(backend.read(filename), filename): if item.type in (ITEM_TYPE.DATA, ITEM_TYPE.INODE): if item.inum != 0: @@ -500,26 +628,40 @@ def run_cleaner(backend, inode_map, log, repack_inodes=False): dirty_inodes.add(item.inum) if item.inum not in dirty_inode_data: for b in inode.links: - if s == b[1][0:2]: + if b[1] is not None and s == b[1][0:2]: dirty_inode_data.add(item.inum) break - print "Inodes to rewrite:", dirty_inodes - print "Inodes with data to rewrite:", dirty_inode_data + #print "Inodes to rewrite:", dirty_inodes + #print "Inodes with data to rewrite:", dirty_inode_data for i in sorted(dirty_inodes.union(dirty_inode_data)): rewrite_inode(backend, inode_map, i, log, i in dirty_inode_data) if __name__ == '__main__': - backend = S3Backend("mvrable-bluesky", cachedir=".") + benchlog = open('cleaner.log', 'a') + benchlog_write("*** START CLEANER RUN ***") + start_time = time.time() + backend = S3Backend("mvrable-bluesky-west", cachedir="/tmp/bluesky-cache") #backend = FileBackend(".") chkpt = load_checkpoint_record(backend) - print backend.list() + #print backend.list() + log_dir = LogDirectory(backend, 1) imap = InodeMap() imap.build(backend, chkpt) print chkpt - log_dir = LogDirectory(backend, 1) + print "Version vector:", imap.version_vector + print "Last cleaner log file:", log_dir.seq_num - 1 + if imap.version_vector.get(1, -1) != log_dir.seq_num - 1: + print "Proxy hasn't updated to latest cleaner segment yet!" + benchlog_write("waiting for proxy...") + sys.exit(0) + run_cleaner(backend, imap, log_dir) print "Version vector:", imap.version_vector imap.write(backend, log_dir) log_dir.close_all() + end_time = time.time() + backend.dump_stats() + benchlog_write("running_time: %s", end_time - start_time) + benchlog_write("")