lib/bup/hashsplit.py

   1 import math
   2 from bup import _helpers
   3 from bup.helpers import *
   4
   5 BLOB_LWM = 8192*2
   6 BLOB_MAX = BLOB_LWM*2
   7 BLOB_HWM = 1024*1024
   8 MAX_PER_TREE = 256
   9 progress_callback = None
  10 max_pack_size = 1000*1000*1000  # larger packs will slow down pruning
  11 max_pack_objects = 200*1000  # cache memory usage is about 83 bytes per object
  12 fanout = 16
  13
  14 class Buf:
  15     def __init__(self):
  16         self.data = ''
  17         self.start = 0
  18
  19     def put(self, s):
  20         if s:
  21             self.data = buffer(self.data, self.start) + s
  22             self.start = 0
  23
  24     def peek(self, count):
  25         return buffer(self.data, self.start, count)
  26
  27     def eat(self, count):
  28         self.start += count
  29
  30     def get(self, count):
  31         v = buffer(self.data, self.start, count)
  32         self.start += count
  33         return v
  34
  35     def used(self):
  36         return len(self.data) - self.start
  37
  38
  39 def splitbuf(buf):
  40     b = buf.peek(buf.used())
  41     (ofs, bits) = _helpers.splitbuf(b)
  42     if ofs:
  43         buf.eat(ofs)
  44         return (buffer(b, 0, ofs), bits)
  45     return (None, 0)
  46
  47
  48 def blobiter(files, progress=None):
  49     for filenum,f in enumerate(files):
  50         ofs = 0
  51         b = ''
  52         while 1:
  53             if progress:
  54                 progress(filenum, len(b))
  55             fadvise_done(f, max(0, ofs - 1024*1024))
  56             b = f.read(BLOB_HWM)
  57             ofs += len(b)
  58             if not b:
  59                 fadvise_done(f, ofs)
  60                 break
  61             yield b
  62
  63
  64 def drainbuf(buf, finalize):
  65     while 1:
  66         (blob, bits) = splitbuf(buf)
  67         if blob:
  68             yield (blob, bits)
  69         else:
  70             break
  71     if buf.used() > BLOB_MAX:
  72         # limit max blob size
  73         yield (buf.get(buf.used()), 0)
  74     elif finalize and buf.used():
  75         yield (buf.get(buf.used()), 0)
  76
  77
  78 def _hashsplit_iter(files, progress):
  79     assert(BLOB_HWM > BLOB_MAX)
  80     buf = Buf()
  81     fi = blobiter(files, progress)
  82     while 1:
  83         for i in drainbuf(buf, finalize=False):
  84             yield i
  85         while buf.used() < BLOB_HWM:
  86             bnew = next(fi)
  87             if not bnew:
  88                 # eof
  89                 for i in drainbuf(buf, finalize=True):
  90                     yield i
  91                 return
  92             buf.put(bnew)
  93
  94
  95 def _hashsplit_iter_keep_boundaries(files, progress):
  96     for real_filenum,f in enumerate(files):
  97         if progress:
  98             def prog(filenum, nbytes):
  99                 # the inner _hashsplit_iter doesn't know the real file count,
 100                 # so we'll replace it here.
 101                 return progress(real_filenum, nbytes)
 102         else:
 103             prog = None
 104         for i in _hashsplit_iter([f], progress=prog):
 105             yield i
 106
 107
 108 def hashsplit_iter(files, keep_boundaries, progress):
 109     if keep_boundaries:
 110         return _hashsplit_iter_keep_boundaries(files, progress)
 111     else:
 112         return _hashsplit_iter(files, progress)
 113
 114
 115 total_split = 0
 116 def _split_to_blobs(w, files, keep_boundaries, progress):
 117     global total_split
 118     for (blob, bits) in hashsplit_iter(files, keep_boundaries, progress):
 119         sha = w.new_blob(blob)
 120         total_split += len(blob)
 121         if w.outbytes >= max_pack_size or w.count >= max_pack_objects:
 122             w.breakpoint()
 123         if progress_callback:
 124             progress_callback(len(blob))
 125         yield (sha, len(blob), bits)
 126
 127
 128 def _make_shalist(l):
 129     ofs = 0
 130     shalist = []
 131     for (mode, sha, size) in l:
 132         shalist.append((mode, '%016x' % ofs, sha))
 133         ofs += size
 134     total = ofs
 135     return (shalist, total)
 136
 137
 138 def _squish(w, stacks, n):
 139     i = 0
 140     while i<n or len(stacks[i]) > MAX_PER_TREE:
 141         while len(stacks) <= i+1:
 142             stacks.append([])
 143         if len(stacks[i]) == 1:
 144             stacks[i+1] += stacks[i]
 145         elif stacks[i]:
 146             (shalist, size) = _make_shalist(stacks[i])
 147             tree = w.new_tree(shalist)
 148             stacks[i+1].append(('40000', tree, size))
 149         stacks[i] = []
 150         i += 1
 151
 152
 153 def split_to_shalist(w, files, keep_boundaries, progress=None):
 154     sl = _split_to_blobs(w, files, keep_boundaries, progress)
 155     if not fanout:
 156         shal = []
 157         for (sha,size,bits) in sl:
 158             shal.append(('100644', sha, size))
 159         return _make_shalist(shal)[0]
 160     else:
 161         base_bits = _helpers.blobbits()
 162         fanout_bits = int(math.log(fanout, 2))
 163         def bits_to_idx(n):
 164             assert(n >= base_bits)
 165             return (n - base_bits)/fanout_bits
 166         stacks = [[]]
 167         for (sha,size,bits) in sl:
 168             assert(bits <= 32)
 169             stacks[0].append(('100644', sha, size))
 170             if bits > base_bits:
 171                 _squish(w, stacks, bits_to_idx(bits))
 172         #log('stacks: %r\n' % [len(i) for i in stacks])
 173         _squish(w, stacks, len(stacks)-1)
 174         #log('stacks: %r\n' % [len(i) for i in stacks])
 175         return _make_shalist(stacks[-1])[0]
 176
 177
 178 def split_to_blob_or_tree(w, files, keep_boundaries):
 179     shalist = list(split_to_shalist(w, files, keep_boundaries))
 180     if len(shalist) == 1:
 181         return (shalist[0][0], shalist[0][2])
 182     elif len(shalist) == 0:
 183         return ('100644', w.new_blob(''))
 184     else:
 185         return ('40000', w.new_tree(shalist))
 186
 187
 188 def open_noatime(name):
 189     fd = _helpers.open_noatime(name)
 190     try:
 191         return os.fdopen(fd, 'rb', 1024*1024)
 192     except:
 193         try:
 194             os.close(fd)
 195         except:
 196             pass
 197         raise
 198
 199
 200 def fadvise_done(f, ofs):
 201     assert(ofs >= 0)
 202     if ofs > 0 and hasattr(f, 'fileno'):
 203         _helpers.fadvise_done(f.fileno(), ofs)