X-Git-Url: http://git.vrable.net/?a=blobdiff_plain;f=cleaner%2Fcleaner;h=b148b95f04c584d37055eb4d1b635c9eb67be4e1;hb=afc95f0ab2900a311802028f1b811867fa3c0c31;hp=91037696bbccbb052a30e6af47aaae7b48ca090a;hpb=25f5cec66069a4aebdfc64993590f8c855b2487a;p=bluesky.git

diff --git a/cleaner/cleaner b/cleaner/cleaner
index 9103769..b148b95 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 <mvrable@cs.ucsd.edu>
 
-import base64, os, re, struct, sys
+import base64, os, re, struct, sys, time
 import boto
 from boto.s3.key import Key
 
@@ -19,6 +19,8 @@ HEADER_MAGIC1 = 'AgI-'          # Unencrypted data
 HEADER_MAGIC2 = 'AgI='          # Encrypted data
 HEADER_SIZE = struct.calcsize(HEADER_FORMAT)
 
+CHECKPOINT_MAGIC = struct.pack('<Q', 0x7ad7dafb42a498b4)
+
 class ITEM_TYPE:
     DATA = '1'
     INODE = '2'
@@ -34,10 +36,11 @@ class FileBackend:
     def __init__(self, path):
         self.path = path
 
-    def list(self):
+    def list(self, directory=0):
         """Return a listing of all log segments and their sizes."""
 
-        files = [f for f in os.listdir(self.path) if f.startswith('log-')]
+        prefix = "log-%08d-" % (directory,)
+        files = [f for f in os.listdir(self.path) if f.startswith(prefix)]
         files.sort()
 
         return [(f, os.stat(os.path.join(self.path, f)).st_size)
@@ -47,7 +50,7 @@ class FileBackend:
         fp = open(os.path.join(self.path, filename), 'rb')
         if offset > 0:
             fp.seek(offset)
-        if legnth is None:
+        if length is None:
             return fp.read()
         else:
             return fp.read(length)
@@ -75,6 +78,7 @@ 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
 
@@ -92,9 +96,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
 
@@ -207,7 +212,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 +246,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 +332,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 +348,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 +356,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("<I", data, 0)
+        self.vvsize = 4 + 8*vvlen
+        for i in range(vvlen):
+            (v1, v2) = struct.unpack_from("<II", data, 4 + 8*i)
+            self.version_vector[v1] = v2
+        print self.version_vector
+        self.version_vector[checkpoint_record.location[0]] \
+            = checkpoint_record.location[1]
+        print self.version_vector
+
+        data = data[self.vvsize:]
+
         print "Inode map:"
-        for i in range(len(checkpoint_record.data) // 16):
-            (start, end) = struct.unpack_from("<QQ", checkpoint_record.data, 16*i)
+        for i in range(len(data) // 16):
+            (start, end) = struct.unpack_from("<QQ", data, 16*i)
             imap = load_item(backend, checkpoint_record.links[i][1])
             util.add_item(imap)
             print "[%d, %d]: %s" % (start, end, imap)
@@ -371,8 +394,9 @@ class InodeMap:
         for (s, u) in sorted(util.segments.items()):
             print "%s: %s %s" % (s, u, float(u[1]) / u[0])
             if u[1] == 0:
-                print "Deleting..."
-                backend.delete(s)
+                # print "Deleting..."
+                # backend.delete(s)
+                pass
 
         self.inodes = inodes
         self.util = util
@@ -388,13 +412,18 @@ class InodeMap:
         new_checkpoint.id = LogItem.random_id()
         new_checkpoint.inum = 0
         new_checkpoint.type = ITEM_TYPE.CHECKPOINT
-        new_checkpoint.data = ""
+        new_checkpoint.data = CHECKPOINT_MAGIC
         new_checkpoint.links = []
 
-        for i in range(len(self.checkpoint_record.data) // 16):
-            (start, end) = struct.unpack_from("<QQ", self.checkpoint_record.data, 16*i)
+        new_checkpoint.data += struct.pack('<I', len(self.version_vector))
+        for d in sorted(self.version_vector):
+            new_checkpoint.data += struct.pack('<II', d, self.version_vector[d])
+
+        data = self.checkpoint_record.data[self.vvsize:]
+        for i in range(len(data) // 16):
+            (start, end) = struct.unpack_from("<QQ", data, 16*i)
 
-            new_checkpoint.data += self.checkpoint_record.data[16*i : 16*i + 16]
+            new_checkpoint.data += data[16*i : 16*i + 16]
 
             # Case 1: No inodes in this range of the old inode map have
             # changed.  Simply emit a new pointer to the same inode map block.
@@ -481,14 +510,16 @@ def run_cleaner(backend, inode_map, log, repack_inodes=False):
         rewrite_inode(backend, inode_map, i, log, i in dirty_inode_data)
 
 if __name__ == '__main__':
-    backend = S3Backend("mvrable-bluesky", cachedir=".")
+    #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()