import base64, os, re, struct, sys
import boto
+from boto.s3.key import Key
# The BlueSky 'struct cloudlog_header' data type.
HEADER_FORMAT = '<4sb16sQIII'
fp.write(data)
fp.close()
+ 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)))
+
+class S3Backend:
+ """An interface to BlueSky where the log segments are on in Amazon S3."""
+
+ def __init__(self, bucket, path='', cachedir="."):
+ self.conn = boto.connect_s3(is_secure=False)
+ self.bucket = self.conn.get_bucket(bucket)
+ self.path = path
+ self.cachedir = cachedir
+ self.cache = {}
+
+ def list(self):
+ files = []
+ for k in self.bucket.list(self.path + 'log-'):
+ files.append((k.key, k.size))
+ return files
+
+ def read(self, filename):
+ if filename in self.cache:
+ fp = open(os.path.join(self.cachedir, filename), 'rb')
+ return fp.read()
+ else:
+ k = Key(self.bucket)
+ k.key = self.path + filename
+ data = k.get_contents_as_string()
+ fp = open(os.path.join(self.cachedir, filename), 'wb')
+ fp.write(data)
+ fp.close()
+ self.cache[filename] = True
+ return data
+
+ def write(self, filename, data):
+ k = Key(self.bucket)
+ k.key = self.path + filename
+ k.set_contents_from_string(data)
+ if filename in self.cache:
+ del self.cache[filename]
+
+ def delete(self, filename):
+ k = Key(self.bucket)
+ k.key = self.path + filename
+ k.delete()
+ if filename in self.cache:
+ del self.cache[filename]
+
def loc_to_name(self, location):
return "log-%08d-%08d" % (location)
def load_checkpoint_record(backend):
for (log, size) in reversed(backend.list()):
for item in reversed(list(parse_log(backend.read(log), log))):
+ print item
if item.type == ITEM_TYPE.CHECKPOINT:
return item
print
print "Segment utilizations:"
for (s, u) in sorted(util.segments.items()):
- #if u[1] > 0:
print "%s: %s %s" % (s, u, float(u[1]) / u[0])
+ if u[1] == 0:
+ print "Deleting..."
+ backend.delete(s)
self.inodes = inodes
self.util = util
log.write(new_checkpoint, 2)
self.checkpoint_record = new_checkpoint
-def rewrite_inode(backend, inode_map, inum, log):
+def rewrite_inode(backend, inode_map, inum, log, copy_data=True):
inode = inode_map.inodes[inum]
- blocks = []
- for l in inode.links:
- data = load_item(backend, l[1])
- blocks.append(data)
- log.write(data, 0)
- inode.links = [(b.id, b.location) for b in blocks]
+ if copy_data:
+ blocks = []
+ for l in inode.links:
+ data = load_item(backend, l[1])
+ blocks.append(data)
+ log.write(data, 0)
+ inode.links = [(b.id, b.location) for b in blocks]
log.write(inode, 1)
inode_map.mark_updated(inum)
# Determine which segments are poorly utilized and should be cleaned. We
# need better heuristics here.
for (s, u) in sorted(inode_map.util.segments.items()):
- if float(u[1]) / u[0] < 0.99 and u[1] > 0:
+ if (float(u[1]) / u[0] < 0.6 or u[1] < 32768) and u[1] > 0:
print "Should clean segment", s
loc = backend.name_to_loc(s)
if s: inode_map.obsolete_segments.add(loc)
+ # TODO: We probably also want heuristics that will find inodes with
+ # badly-fragmented data and rewrite that to achieve better locality.
+
# Given that list of segments to clean, scan through those segments to find
# data which is still live and mark relevant inodes as needing to be
# rewritten.
dirty_inodes = set()
+ dirty_inode_data = set()
for s in inode_map.obsolete_segments:
filename = backend.loc_to_name(s)
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:
- dirty_inodes.add(item.inum)
+ inode = inode_map.inodes[item.inum]
+ if s == inode.location[0:2]:
+ dirty_inodes.add(item.inum)
+ if item.inum not in dirty_inode_data:
+ for b in inode.links:
+ if s == b[1][0:2]:
+ dirty_inode_data.add(item.inum)
+ break
print "Inodes to rewrite:", dirty_inodes
- for i in sorted(dirty_inodes):
- rewrite_inode(backend, inode_map, i, log)
+ 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 = FileBackend(".")
+ backend = S3Backend("mvrable-bluesky", cachedir=".")
chkpt = load_checkpoint_record(backend)
+ print backend.list()
imap = InodeMap()
imap.build(backend, chkpt)
print chkpt