lib/bup/hashsplit.py

   1 import math
   2 from bup import _helpers
   3 from bup.helpers import *
   4
   5 BLOB_MAX = 8192*4   # 8192 is the "typical" blob size for bupsplit
   6 BLOB_READ_SIZE = 1024*1024
   7 MAX_PER_TREE = 256
   8 progress_callback = None
   9 fanout = 16
  10
  11 GIT_MODE_FILE = 0100644
  12 GIT_MODE_TREE = 040000
  13 assert(GIT_MODE_TREE != 40000)  # 0xxx should be treated as octal
  14
  15 # The purpose of this type of buffer is to avoid copying on peek(), get(),
  16 # and eat().  We do copy the buffer contents on put(), but that should
  17 # be ok if we always only put() large amounts of data at a time.
  18 class Buf:
  19     def __init__(self):
  20         self.data = ''
  21         self.start = 0
  22
  23     def put(self, s):
  24         if s:
  25             self.data = buffer(self.data, self.start) + s
  26             self.start = 0
  27
  28     def peek(self, count):
  29         return buffer(self.data, self.start, count)
  30
  31     def eat(self, count):
  32         self.start += count
  33
  34     def get(self, count):
  35         v = buffer(self.data, self.start, count)
  36         self.start += count
  37         return v
  38
  39     def used(self):
  40         return len(self.data) - self.start
  41
  42
  43 def readfile_iter(files, progress=None):
  44     for filenum,f in enumerate(files):
  45         ofs = 0
  46         b = ''
  47         while 1:
  48             if progress:
  49                 progress(filenum, len(b))
  50             fadvise_done(f, max(0, ofs - 1024*1024))
  51             b = f.read(BLOB_READ_SIZE)
  52             ofs += len(b)
  53             if not b:
  54                 fadvise_done(f, ofs)
  55                 break
  56             yield b
  57
  58
  59 def _splitbuf(buf, basebits, fanbits):
  60     while 1:
  61         b = buf.peek(buf.used())
  62         (ofs, bits) = _helpers.splitbuf(b)
  63         if ofs > BLOB_MAX:
  64             ofs = BLOB_MAX
  65         if ofs:
  66             buf.eat(ofs)
  67             level = (bits-basebits)//fanbits  # integer division
  68             yield buffer(b, 0, ofs), level
  69         else:
  70             break
  71     while buf.used() >= BLOB_MAX:
  72         # limit max blob size
  73         yield buf.get(BLOB_MAX), 0
  74
  75
  76 def _hashsplit_iter(files, progress):
  77     assert(BLOB_READ_SIZE > BLOB_MAX)
  78     basebits = _helpers.blobbits()
  79     fanbits = int(math.log(fanout or 128, 2))
  80     buf = Buf()
  81     for inblock in readfile_iter(files, progress):
  82         buf.put(inblock)
  83         for buf_and_level in _splitbuf(buf, basebits, fanbits):
  84             yield buf_and_level
  85     if buf.used():
  86         yield buf.get(buf.used()), 0
  87
  88
  89 def _hashsplit_iter_keep_boundaries(files, progress):
  90     for real_filenum,f in enumerate(files):
  91         if progress:
  92             def prog(filenum, nbytes):
  93                 # the inner _hashsplit_iter doesn't know the real file count,
  94                 # so we'll replace it here.
  95                 return progress(real_filenum, nbytes)
  96         else:
  97             prog = None
  98         for buf_and_level in _hashsplit_iter([f], progress=prog):
  99             yield buf_and_level
 100
 101
 102 def hashsplit_iter(files, keep_boundaries, progress):
 103     if keep_boundaries:
 104         return _hashsplit_iter_keep_boundaries(files, progress)
 105     else:
 106         return _hashsplit_iter(files, progress)
 107
 108
 109 total_split = 0
 110 def split_to_blobs(makeblob, files, keep_boundaries, progress):
 111     global total_split
 112     for (blob, level) in hashsplit_iter(files, keep_boundaries, progress):
 113         sha = makeblob(blob)
 114         total_split += len(blob)
 115         if progress_callback:
 116             progress_callback(len(blob))
 117         yield (sha, len(blob), level)
 118
 119
 120 def _make_shalist(l):
 121     ofs = 0
 122     shalist = []
 123     for (mode, sha, size) in l:
 124         shalist.append((mode, '%016x' % ofs, sha))
 125         ofs += size
 126     total = ofs
 127     return (shalist, total)
 128
 129
 130 def _squish(maketree, stacks, n):
 131     i = 0
 132     while i<n or len(stacks[i]) > MAX_PER_TREE:
 133         while len(stacks) <= i+1:
 134             stacks.append([])
 135         if len(stacks[i]) == 1:
 136             stacks[i+1] += stacks[i]
 137         elif stacks[i]:
 138             (shalist, size) = _make_shalist(stacks[i])
 139             tree = maketree(shalist)
 140             stacks[i+1].append((GIT_MODE_TREE, tree, size))
 141         stacks[i] = []
 142         i += 1
 143
 144
 145 def split_to_shalist(makeblob, maketree, files,
 146                      keep_boundaries, progress=None):
 147     sl = split_to_blobs(makeblob, files, keep_boundaries, progress)
 148     assert(fanout != 0)
 149     if not fanout:
 150         shal = []
 151         for (sha,size,level) in sl:
 152             shal.append((GIT_MODE_FILE, sha, size))
 153         return _make_shalist(shal)[0]
 154     else:
 155         stacks = [[]]
 156         for (sha,size,level) in sl:
 157             stacks[0].append((GIT_MODE_FILE, sha, size))
 158             if level:
 159                 _squish(maketree, stacks, level)
 160         #log('stacks: %r\n' % [len(i) for i in stacks])
 161         _squish(maketree, stacks, len(stacks)-1)
 162         #log('stacks: %r\n' % [len(i) for i in stacks])
 163         return _make_shalist(stacks[-1])[0]
 164
 165
 166 def split_to_blob_or_tree(makeblob, maketree, files, keep_boundaries):
 167     shalist = list(split_to_shalist(makeblob, maketree,
 168                                     files, keep_boundaries))
 169     if len(shalist) == 1:
 170         return (shalist[0][0], shalist[0][2])
 171     elif len(shalist) == 0:
 172         return (GIT_MODE_FILE, makeblob(''))
 173     else:
 174         return (GIT_MODE_TREE, maketree(shalist))
 175
 176
 177 def open_noatime(name):
 178     fd = _helpers.open_noatime(name)
 179     try:
 180         return os.fdopen(fd, 'rb', 1024*1024)
 181     except:
 182         try:
 183             os.close(fd)
 184         except:
 185             pass
 186         raise
 187
 188
 189 def fadvise_done(f, ofs):
 190     assert(ofs >= 0)
 191     if ofs > 0 and hasattr(f, 'fileno'):
 192         _helpers.fadvise_done(f.fileno(), ofs)