lib/bup/hashsplit.py

   1 import math, os
   2
   3 from bup import _helpers, helpers
   4 from bup.helpers import sc_page_size
   5
   6 _fmincore = getattr(helpers, 'fmincore', None)
   7
   8 BLOB_MAX = 8192*4   # 8192 is the "typical" blob size for bupsplit
   9 BLOB_READ_SIZE = 1024*1024
  10 MAX_PER_TREE = 256
  11 progress_callback = None
  12 fanout = 16
  13
  14 GIT_MODE_FILE = 0100644
  15 GIT_MODE_TREE = 040000
  16 GIT_MODE_SYMLINK = 0120000
  17 assert(GIT_MODE_TREE != 40000)  # 0xxx should be treated as octal
  18
  19 # The purpose of this type of buffer is to avoid copying on peek(), get(),
  20 # and eat().  We do copy the buffer contents on put(), but that should
  21 # be ok if we always only put() large amounts of data at a time.
  22 class Buf:
  23     def __init__(self):
  24         self.data = ''
  25         self.start = 0
  26
  27     def put(self, s):
  28         if s:
  29             self.data = buffer(self.data, self.start) + s
  30             self.start = 0
  31
  32     def peek(self, count):
  33         return buffer(self.data, self.start, count)
  34
  35     def eat(self, count):
  36         self.start += count
  37
  38     def get(self, count):
  39         v = buffer(self.data, self.start, count)
  40         self.start += count
  41         return v
  42
  43     def used(self):
  44         return len(self.data) - self.start
  45
  46
  47 def _fadvise_pages_done(fd, first_page, count):
  48     assert(first_page >= 0)
  49     assert(count >= 0)
  50     if count > 0:
  51         _helpers.fadvise_done(fd,
  52                               first_page * sc_page_size,
  53                               count * sc_page_size)
  54
  55
  56 def _nonresident_page_regions(status_bytes, incore_mask, max_region_len=None):
  57     """Return (start_page, count) pairs in ascending start_page order for
  58     each contiguous region of nonresident pages indicated by the
  59     mincore() status_bytes.  Limit the number of pages in each region
  60     to max_region_len."""
  61     assert(max_region_len is None or max_region_len > 0)
  62     start = None
  63     for i, x in enumerate(status_bytes):
  64         in_core = x & incore_mask
  65         if start is None:
  66             if not in_core:
  67                 start = i
  68         else:
  69             count = i - start
  70             if in_core:
  71                 yield (start, count)
  72                 start = None
  73             elif max_region_len and count >= max_region_len:
  74                 yield (start, count)
  75                 start = i
  76     if start is not None:
  77         yield (start, len(status_bytes) - start)
  78
  79
  80 def _uncache_ours_upto(fd, offset, first_region, remaining_regions):
  81     """Uncache the pages of fd indicated by first_region and
  82     remaining_regions that are before offset, where each region is a
  83     (start_page, count) pair.  The final region must have a start_page
  84     of None."""
  85     rstart, rlen = first_region
  86     while rstart is not None and (rstart + rlen) * sc_page_size <= offset:
  87         _fadvise_pages_done(fd, rstart, rlen)
  88         rstart, rlen = next(remaining_regions, (None, None))
  89     return (rstart, rlen)
  90
  91
  92 def readfile_iter(files, progress=None):
  93     for filenum,f in enumerate(files):
  94         ofs = 0
  95         b = ''
  96         fd = rpr = rstart = rlen = None
  97         if _fmincore and hasattr(f, 'fileno'):
  98             fd = f.fileno()
  99             max_chunk = max(1, (8 * 1024 * 1024) / sc_page_size)
 100             rpr = _nonresident_page_regions(_fmincore(fd),
 101                                             helpers.MINCORE_INCORE, max_chunk)
 102             rstart, rlen = next(rpr, (None, None))
 103         while 1:
 104             if progress:
 105                 progress(filenum, len(b))
 106             b = f.read(BLOB_READ_SIZE)
 107             ofs += len(b)
 108             if rpr:
 109                 rstart, rlen = _uncache_ours_upto(fd, ofs, (rstart, rlen), rpr)
 110             if not b:
 111                 break
 112             yield b
 113         if rpr:
 114             rstart, rlen = _uncache_ours_upto(fd, ofs, (rstart, rlen), rpr)
 115
 116
 117 def _splitbuf(buf, basebits, fanbits):
 118     while 1:
 119         b = buf.peek(buf.used())
 120         (ofs, bits) = _helpers.splitbuf(b)
 121         if ofs:
 122             if ofs > BLOB_MAX:
 123                 ofs = BLOB_MAX
 124                 level = 0
 125             else:
 126                 level = (bits-basebits)//fanbits  # integer division
 127             buf.eat(ofs)
 128             yield buffer(b, 0, ofs), level
 129         else:
 130             break
 131     while buf.used() >= BLOB_MAX:
 132         # limit max blob size
 133         yield buf.get(BLOB_MAX), 0
 134
 135
 136 def _hashsplit_iter(files, progress):
 137     assert(BLOB_READ_SIZE > BLOB_MAX)
 138     basebits = _helpers.blobbits()
 139     fanbits = int(math.log(fanout or 128, 2))
 140     buf = Buf()
 141     for inblock in readfile_iter(files, progress):
 142         buf.put(inblock)
 143         for buf_and_level in _splitbuf(buf, basebits, fanbits):
 144             yield buf_and_level
 145     if buf.used():
 146         yield buf.get(buf.used()), 0
 147
 148
 149 def _hashsplit_iter_keep_boundaries(files, progress):
 150     for real_filenum,f in enumerate(files):
 151         if progress:
 152             def prog(filenum, nbytes):
 153                 # the inner _hashsplit_iter doesn't know the real file count,
 154                 # so we'll replace it here.
 155                 return progress(real_filenum, nbytes)
 156         else:
 157             prog = None
 158         for buf_and_level in _hashsplit_iter([f], progress=prog):
 159             yield buf_and_level
 160
 161
 162 def hashsplit_iter(files, keep_boundaries, progress):
 163     if keep_boundaries:
 164         return _hashsplit_iter_keep_boundaries(files, progress)
 165     else:
 166         return _hashsplit_iter(files, progress)
 167
 168
 169 total_split = 0
 170 def split_to_blobs(makeblob, files, keep_boundaries, progress):
 171     global total_split
 172     for (blob, level) in hashsplit_iter(files, keep_boundaries, progress):
 173         sha = makeblob(blob)
 174         total_split += len(blob)
 175         if progress_callback:
 176             progress_callback(len(blob))
 177         yield (sha, len(blob), level)
 178
 179
 180 def _make_shalist(l):
 181     ofs = 0
 182     l = list(l)
 183     total = sum(size for mode,sha,size, in l)
 184     vlen = len('%x' % total)
 185     shalist = []
 186     for (mode, sha, size) in l:
 187         shalist.append((mode, '%0*x' % (vlen,ofs), sha))
 188         ofs += size
 189     assert(ofs == total)
 190     return (shalist, total)
 191
 192
 193 def _squish(maketree, stacks, n):
 194     i = 0
 195     while i < n or len(stacks[i]) >= MAX_PER_TREE:
 196         while len(stacks) <= i+1:
 197             stacks.append([])
 198         if len(stacks[i]) == 1:
 199             stacks[i+1] += stacks[i]
 200         elif stacks[i]:
 201             (shalist, size) = _make_shalist(stacks[i])
 202             tree = maketree(shalist)
 203             stacks[i+1].append((GIT_MODE_TREE, tree, size))
 204         stacks[i] = []
 205         i += 1
 206
 207
 208 def split_to_shalist(makeblob, maketree, files,
 209                      keep_boundaries, progress=None):
 210     sl = split_to_blobs(makeblob, files, keep_boundaries, progress)
 211     assert(fanout != 0)
 212     if not fanout:
 213         shal = []
 214         for (sha,size,level) in sl:
 215             shal.append((GIT_MODE_FILE, sha, size))
 216         return _make_shalist(shal)[0]
 217     else:
 218         stacks = [[]]
 219         for (sha,size,level) in sl:
 220             stacks[0].append((GIT_MODE_FILE, sha, size))
 221             _squish(maketree, stacks, level)
 222         #log('stacks: %r\n' % [len(i) for i in stacks])
 223         _squish(maketree, stacks, len(stacks)-1)
 224         #log('stacks: %r\n' % [len(i) for i in stacks])
 225         return _make_shalist(stacks[-1])[0]
 226
 227
 228 def split_to_blob_or_tree(makeblob, maketree, files,
 229                           keep_boundaries, progress=None):
 230     shalist = list(split_to_shalist(makeblob, maketree,
 231                                     files, keep_boundaries, progress))
 232     if len(shalist) == 1:
 233         return (shalist[0][0], shalist[0][2])
 234     elif len(shalist) == 0:
 235         return (GIT_MODE_FILE, makeblob(''))
 236     else:
 237         return (GIT_MODE_TREE, maketree(shalist))
 238
 239
 240 def open_noatime(name):
 241     fd = _helpers.open_noatime(name)
 242     try:
 243         return os.fdopen(fd, 'rb', 1024*1024)
 244     except:
 245         try:
 246             os.close(fd)
 247         except:
 248             pass
 249         raise