22
from bzrlib.lazy_import import lazy_import
23
lazy_import(globals(), """
26
24
from bzrlib import (
30
29
graph as _mod_graph,
37
from bzrlib.repofmt import pack_repo
38
from bzrlib.i18n import gettext
37
41
from bzrlib.btree_index import BTreeBuilder
38
42
from bzrlib.lru_cache import LRUSizeCache
39
from bzrlib.tsort import topo_sort
40
43
from bzrlib.versionedfile import (
42
46
AbsentContentFactory,
43
47
ChunkedContentFactory,
44
48
FulltextContentFactory,
49
VersionedFilesWithFallbacks,
48
52
# Minimum number of uncompressed bytes to try fetch at once when retrieving
49
53
# groupcompress blocks.
52
_USE_LZMA = False and (pylzma is not None)
54
56
# osutils.sha_string('')
55
57
_null_sha1 = 'da39a3ee5e6b4b0d3255bfef95601890afd80709'
135
137
self._content = ''.join(self._content_chunks)
136
138
self._content_chunks = None
137
139
if self._content is None:
138
if self._z_content is None:
140
# We join self._z_content_chunks here, because if we are
141
# decompressing, then it is *very* likely that we have a single
143
if self._z_content_chunks is None:
139
144
raise AssertionError('No content to decompress')
140
if self._z_content == '':
145
z_content = ''.join(self._z_content_chunks)
141
147
self._content = ''
142
148
elif self._compressor_name == 'lzma':
143
149
# We don't do partial lzma decomp yet
144
self._content = pylzma.decompress(self._z_content)
151
self._content = pylzma.decompress(z_content)
145
152
elif self._compressor_name == 'zlib':
146
153
# Start a zlib decompressor
147
154
if num_bytes * 4 > self._content_length * 3:
148
155
# If we are requesting more that 3/4ths of the content,
149
156
# just extract the whole thing in a single pass
150
157
num_bytes = self._content_length
151
self._content = zlib.decompress(self._z_content)
158
self._content = zlib.decompress(z_content)
153
160
self._z_content_decompressor = zlib.decompressobj()
154
161
# Seed the decompressor with the uncompressed bytes, so
155
162
# that the rest of the code is simplified
156
163
self._content = self._z_content_decompressor.decompress(
157
self._z_content, num_bytes + _ZLIB_DECOMP_WINDOW)
164
z_content, num_bytes + _ZLIB_DECOMP_WINDOW)
158
165
if not self._z_content_decompressor.unconsumed_tail:
159
166
self._z_content_decompressor = None
207
214
# XXX: Define some GCCorrupt error ?
208
215
raise AssertionError('Invalid bytes: (%d) != %d + %d' %
209
216
(len(bytes), pos, self._z_content_length))
210
self._z_content = bytes[pos:]
217
self._z_content_chunks = (bytes[pos:],)
220
def _z_content(self):
221
"""Return z_content_chunks as a simple string.
223
Meant only to be used by the test suite.
225
if self._z_content_chunks is not None:
226
return ''.join(self._z_content_chunks)
213
230
def from_bytes(cls, bytes):
269
286
self._content_length = length
270
287
self._content_chunks = content_chunks
271
288
self._content = None
272
self._z_content = None
289
self._z_content_chunks = None
274
291
def set_content(self, content):
275
292
"""Set the content of this block."""
276
293
self._content_length = len(content)
277
294
self._content = content
278
self._z_content = None
280
def _create_z_content_using_lzma(self):
281
if self._content_chunks is not None:
282
self._content = ''.join(self._content_chunks)
283
self._content_chunks = None
284
if self._content is None:
285
raise AssertionError('Nothing to compress')
286
self._z_content = pylzma.compress(self._content)
287
self._z_content_length = len(self._z_content)
289
def _create_z_content_from_chunks(self):
295
self._z_content_chunks = None
297
def _create_z_content_from_chunks(self, chunks):
290
298
compressor = zlib.compressobj(zlib.Z_DEFAULT_COMPRESSION)
291
compressed_chunks = map(compressor.compress, self._content_chunks)
299
# Peak in this point is 1 fulltext, 1 compressed text, + zlib overhead
300
# (measured peak is maybe 30MB over the above...)
301
compressed_chunks = map(compressor.compress, chunks)
292
302
compressed_chunks.append(compressor.flush())
293
self._z_content = ''.join(compressed_chunks)
294
self._z_content_length = len(self._z_content)
303
# Ignore empty chunks
304
self._z_content_chunks = [c for c in compressed_chunks if c]
305
self._z_content_length = sum(map(len, self._z_content_chunks))
296
307
def _create_z_content(self):
297
if self._z_content is not None:
300
self._create_z_content_using_lzma()
308
if self._z_content_chunks is not None:
302
310
if self._content_chunks is not None:
303
self._create_z_content_from_chunks()
305
self._z_content = zlib.compress(self._content)
306
self._z_content_length = len(self._z_content)
311
chunks = self._content_chunks
313
chunks = (self._content,)
314
self._create_z_content_from_chunks(chunks)
317
"""Create the byte stream as a series of 'chunks'"""
318
self._create_z_content()
319
header = self.GCB_HEADER
320
chunks = ['%s%d\n%d\n'
321
% (header, self._z_content_length, self._content_length),
323
chunks.extend(self._z_content_chunks)
324
total_len = sum(map(len, chunks))
325
return total_len, chunks
308
327
def to_bytes(self):
309
328
"""Encode the information into a byte stream."""
310
self._create_z_content()
312
header = self.GCB_LZ_HEADER
314
header = self.GCB_HEADER
316
'%d\n%d\n' % (self._z_content_length, self._content_length),
329
total_len, chunks = self.to_chunks()
319
330
return ''.join(chunks)
321
332
def _dump(self, include_text=False):
435
446
# Grab and cache the raw bytes for this entry
436
447
# and break the ref-cycle with _manager since we don't need it
438
self._manager._prepare_for_extract()
450
self._manager._prepare_for_extract()
451
except zlib.error as value:
452
raise errors.DecompressCorruption("zlib: " + str(value))
439
453
block = self._manager._block
440
454
self._bytes = block.extract(self.key, self._start, self._end)
441
455
# There are code paths that first extract as fulltext, and then
460
474
_full_enough_block_size = 3*1024*1024 # size at which we won't repack
461
475
_full_enough_mixed_block_size = 2*768*1024 # 1.5MB
463
def __init__(self, block):
477
def __init__(self, block, get_compressor_settings=None):
464
478
self._block = block
465
479
# We need to preserve the ordering
466
480
self._factories = []
467
481
self._last_byte = 0
482
self._get_settings = get_compressor_settings
483
self._compressor_settings = None
485
def _get_compressor_settings(self):
486
if self._compressor_settings is not None:
487
return self._compressor_settings
489
if self._get_settings is not None:
490
settings = self._get_settings()
492
vf = GroupCompressVersionedFiles
493
settings = vf._DEFAULT_COMPRESSOR_SETTINGS
494
self._compressor_settings = settings
495
return self._compressor_settings
469
497
def add_factory(self, key, parents, start, end):
470
498
if not self._factories:
503
531
new_block.set_content(self._block._content[:last_byte])
504
532
self._block = new_block
534
def _make_group_compressor(self):
535
return GroupCompressor(self._get_compressor_settings())
506
537
def _rebuild_block(self):
507
538
"""Create a new GroupCompressBlock with only the referenced texts."""
508
compressor = GroupCompressor()
539
compressor = self._make_group_compressor()
509
540
tstart = time.time()
510
541
old_length = self._block._content_length
523
554
# block? It seems hard to come up with a method that it would
524
555
# expand, since we do full compression again. Perhaps based on a
525
556
# request that ends up poorly ordered?
557
# TODO: If the content would have expanded, then we would want to
558
# handle a case where we need to split the block.
559
# Now that we have a user-tweakable option
560
# (max_bytes_to_index), it is possible that one person set it
561
# to a very low value, causing poor compression.
526
562
delta = time.time() - tstart
527
563
self._block = new_block
528
564
trace.mutter('creating new compressed block on-the-fly in %.3fs'
679
715
z_header_bytes = zlib.compress(header_bytes)
681
717
z_header_bytes_len = len(z_header_bytes)
682
block_bytes = self._block.to_bytes()
718
block_bytes_len, block_chunks = self._block.to_chunks()
683
719
lines.append('%d\n%d\n%d\n' % (z_header_bytes_len, header_bytes_len,
685
721
lines.append(z_header_bytes)
686
lines.append(block_bytes)
687
del z_header_bytes, block_bytes
722
lines.extend(block_chunks)
723
del z_header_bytes, block_chunks
724
# TODO: This is a point where we will double the memory consumption. To
725
# avoid this, we probably have to switch to a 'chunked' api
688
726
return ''.join(lines)
691
729
def from_bytes(cls, bytes):
692
730
# TODO: This does extra string copying, probably better to do it a
731
# different way. At a minimum this creates 2 copies of the
694
733
(storage_kind, z_header_len, header_len,
695
734
block_len, rest) = bytes.split('\n', 4)
855
898
After calling this, the compressor should no longer be used
857
# TODO: this causes us to 'bloat' to 2x the size of content in the
858
# group. This has an impact for 'commit' of large objects.
859
# One possibility is to use self._content_chunks, and be lazy and
860
# only fill out self._content as a full string when we actually
861
# need it. That would at least drop the peak memory consumption
862
# for 'commit' down to ~1x the size of the largest file, at a
863
# cost of increased complexity within this code. 2x is still <<
864
# 3x the size of the largest file, so we are doing ok.
865
900
self._block.set_chunked_content(self.chunks, self.endpoint)
866
901
self.chunks = None
867
902
self._delta_index = None
886
921
class PythonGroupCompressor(_CommonGroupCompressor):
923
def __init__(self, settings=None):
889
924
"""Create a GroupCompressor.
891
926
Used only if the pyrex version is not available.
893
super(PythonGroupCompressor, self).__init__()
928
super(PythonGroupCompressor, self).__init__(settings)
894
929
self._delta_index = LinesDeltaIndex([])
895
930
# The actual content is managed by LinesDeltaIndex
896
931
self.chunks = self._delta_index.lines
934
969
It contains code very similar to SequenceMatcher because of having a similar
935
970
task. However some key differences apply:
936
- there is no junk, we want a minimal edit not a human readable diff.
937
- we don't filter very common lines (because we don't know where a good
938
range will start, and after the first text we want to be emitting minmal
940
- we chain the left side, not the right side
941
- we incrementally update the adjacency matrix as new lines are provided.
942
- we look for matches in all of the left side, so the routine which does
943
the analagous task of find_longest_match does not need to filter on the
972
* there is no junk, we want a minimal edit not a human readable diff.
973
* we don't filter very common lines (because we don't know where a good
974
range will start, and after the first text we want to be emitting minmal
976
* we chain the left side, not the right side
977
* we incrementally update the adjacency matrix as new lines are provided.
978
* we look for matches in all of the left side, so the routine which does
979
the analagous task of find_longest_match does not need to filter on the
948
super(PyrexGroupCompressor, self).__init__()
949
self._delta_index = DeltaIndex()
983
def __init__(self, settings=None):
984
super(PyrexGroupCompressor, self).__init__(settings)
985
max_bytes_to_index = self._settings.get('max_bytes_to_index', 0)
986
self._delta_index = DeltaIndex(max_bytes_to_index=max_bytes_to_index)
951
988
def _compress(self, key, bytes, max_delta_size, soft=False):
952
989
"""see _CommonGroupCompressor._compress"""
1027
1064
index = _GCGraphIndex(graph_index, lambda:True, parents=parents,
1028
1065
add_callback=graph_index.add_nodes,
1029
1066
inconsistency_fatal=inconsistency_fatal)
1030
access = knit._DirectPackAccess({})
1067
access = pack_repo._DirectPackAccess({})
1031
1068
access.set_writer(writer, graph_index, (transport, 'newpack'))
1032
1069
result = GroupCompressVersionedFiles(index, access, delta)
1033
1070
result.stream = stream
1044
1081
class _BatchingBlockFetcher(object):
1045
1082
"""Fetch group compress blocks in batches.
1047
1084
:ivar total_bytes: int of expected number of bytes needed to fetch the
1048
1085
currently pending batch.
1051
def __init__(self, gcvf, locations):
1088
def __init__(self, gcvf, locations, get_compressor_settings=None):
1052
1089
self.gcvf = gcvf
1053
1090
self.locations = locations
1102
1140
def yield_factories(self, full_flush=False):
1103
1141
"""Yield factories for keys added since the last yield. They will be
1104
1142
returned in the order they were added via add_key.
1106
1144
:param full_flush: by default, some results may not be returned in case
1107
1145
they can be part of the next batch. If full_flush is True, then
1108
1146
all results are returned.
1136
1174
memos_to_get_stack.pop()
1138
1176
block = self.batch_memos[read_memo]
1139
self.manager = _LazyGroupContentManager(block)
1177
self.manager = _LazyGroupContentManager(block,
1178
get_compressor_settings=self._get_compressor_settings)
1140
1179
self.last_read_memo = read_memo
1141
1180
start, end = index_memo[3:5]
1142
1181
self.manager.add_factory(key, parents, start, end)
1149
1188
self.total_bytes = 0
1152
class GroupCompressVersionedFiles(VersionedFiles):
1191
class GroupCompressVersionedFiles(VersionedFilesWithFallbacks):
1153
1192
"""A group-compress based VersionedFiles implementation."""
1155
def __init__(self, index, access, delta=True, _unadded_refs=None):
1194
# This controls how the GroupCompress DeltaIndex works. Basically, we
1195
# compute hash pointers into the source blocks (so hash(text) => text).
1196
# However each of these references costs some memory in trade against a
1197
# more accurate match result. For very large files, they either are
1198
# pre-compressed and change in bulk whenever they change, or change in just
1199
# local blocks. Either way, 'improved resolution' is not very helpful,
1200
# versus running out of memory trying to track everything. The default max
1201
# gives 100% sampling of a 1MB file.
1202
_DEFAULT_MAX_BYTES_TO_INDEX = 1024 * 1024
1203
_DEFAULT_COMPRESSOR_SETTINGS = {'max_bytes_to_index':
1204
_DEFAULT_MAX_BYTES_TO_INDEX}
1206
def __init__(self, index, access, delta=True, _unadded_refs=None,
1156
1208
"""Create a GroupCompressVersionedFiles object.
1158
1210
:param index: The index object storing access and graph data.
1159
1211
:param access: The access object storing raw data.
1160
1212
:param delta: Whether to delta compress or just entropy compress.
1161
1213
:param _unadded_refs: private parameter, don't use.
1214
:param _group_cache: private parameter, don't use.
1163
1216
self._index = index
1164
1217
self._access = access
1166
1219
if _unadded_refs is None:
1167
1220
_unadded_refs = {}
1168
1221
self._unadded_refs = _unadded_refs
1169
self._group_cache = LRUSizeCache(max_size=50*1024*1024)
1170
self._fallback_vfs = []
1222
if _group_cache is None:
1223
_group_cache = LRUSizeCache(max_size=50*1024*1024)
1224
self._group_cache = _group_cache
1225
self._immediate_fallback_vfs = []
1226
self._max_bytes_to_index = None
1172
1228
def without_fallbacks(self):
1173
1229
"""Return a clone of this object without any fallbacks configured."""
1174
1230
return GroupCompressVersionedFiles(self._index, self._access,
1175
self._delta, _unadded_refs=dict(self._unadded_refs))
1231
self._delta, _unadded_refs=dict(self._unadded_refs),
1232
_group_cache=self._group_cache)
1177
1234
def add_lines(self, key, parents, lines, parent_texts=None,
1178
1235
left_matching_blocks=None, nostore_sha=None, random_id=False,
1182
1239
:param key: The key tuple of the text to add.
1183
1240
:param parents: The parents key tuples of the text to add.
1184
1241
:param lines: A list of lines. Each line must be a bytestring. And all
1185
of them except the last must be terminated with \n and contain no
1186
other \n's. The last line may either contain no \n's or a single
1187
terminating \n. If the lines list does meet this constraint the add
1188
routine may error or may succeed - but you will be unable to read
1189
the data back accurately. (Checking the lines have been split
1242
of them except the last must be terminated with \\n and contain no
1243
other \\n's. The last line may either contain no \\n's or a single
1244
terminating \\n. If the lines list does meet this constraint the
1245
add routine may error or may succeed - but you will be unable to
1246
read the data back accurately. (Checking the lines have been split
1190
1247
correctly is expensive and extremely unlikely to catch bugs so it
1191
1248
is not done at runtime unless check_content is True.)
1192
1249
:param parent_texts: An optional dictionary containing the opaque
1287
1344
self._check_lines_not_unicode(lines)
1288
1345
self._check_lines_are_lines(lines)
1290
def get_known_graph_ancestry(self, keys):
1291
"""Get a KnownGraph instance with the ancestry of keys."""
1292
# Note that this is identical to
1293
# KnitVersionedFiles.get_known_graph_ancestry, but they don't share
1295
parent_map, missing_keys = self._index.find_ancestry(keys)
1296
for fallback in self._fallback_vfs:
1297
if not missing_keys:
1299
(f_parent_map, f_missing_keys) = fallback._index.find_ancestry(
1301
parent_map.update(f_parent_map)
1302
missing_keys = f_missing_keys
1303
kg = _mod_graph.KnownGraph(parent_map)
1306
1347
def get_parent_map(self, keys):
1307
1348
"""Get a map of the graph parents of keys.
1447
1488
The returned objects should be in the order defined by 'ordering',
1448
1489
which can weave between different sources.
1449
1491
:param ordering: Must be one of 'topological' or 'groupcompress'
1450
1492
:return: List of [(source, [keys])] tuples, such that all keys are in
1451
1493
the defined order, regardless of source.
1453
1495
if ordering == 'topological':
1454
present_keys = topo_sort(parent_map)
1496
present_keys = tsort.topo_sort(parent_map)
1456
1498
# ordering == 'groupcompress'
1457
1499
# XXX: This only optimizes for the target ordering. We may need
1546
1588
# - we encounter an unadded ref, or
1547
1589
# - we run out of keys, or
1548
1590
# - the total bytes to retrieve for this batch > BATCH_SIZE
1549
batcher = _BatchingBlockFetcher(self, locations)
1591
batcher = _BatchingBlockFetcher(self, locations,
1592
get_compressor_settings=self._get_compressor_settings)
1550
1593
for source, keys in source_keys:
1551
1594
if source is self:
1552
1595
for key in keys:
1598
1641
for _ in self._insert_record_stream(stream, random_id=False):
1644
def _get_compressor_settings(self):
1645
if self._max_bytes_to_index is None:
1646
# TODO: VersionedFiles don't know about their containing
1647
# repository, so they don't have much of an idea about their
1648
# location. So for now, this is only a global option.
1649
c = config.GlobalConfig()
1650
val = c.get_user_option('bzr.groupcompress.max_bytes_to_index')
1654
except ValueError, e:
1655
trace.warning('Value for '
1656
'"bzr.groupcompress.max_bytes_to_index"'
1657
' %r is not an integer'
1661
val = self._DEFAULT_MAX_BYTES_TO_INDEX
1662
self._max_bytes_to_index = val
1663
return {'max_bytes_to_index': self._max_bytes_to_index}
1665
def _make_group_compressor(self):
1666
return GroupCompressor(self._get_compressor_settings())
1601
1668
def _insert_record_stream(self, stream, random_id=False, nostore_sha=None,
1602
1669
reuse_blocks=True):
1603
1670
"""Internal core to insert a record stream into this container.
1627
1694
# This will go up to fulltexts for gc to gc fetching, which isn't
1629
self._compressor = GroupCompressor()
1696
self._compressor = self._make_group_compressor()
1630
1697
self._unadded_refs = {}
1631
1698
keys_to_add = []
1633
bytes = self._compressor.flush().to_bytes()
1700
bytes_len, chunks = self._compressor.flush().to_chunks()
1701
self._compressor = self._make_group_compressor()
1702
# Note: At this point we still have 1 copy of the fulltext (in
1703
# record and the var 'bytes'), and this generates 2 copies of
1704
# the compressed text (one for bytes, one in chunks)
1705
# TODO: Push 'chunks' down into the _access api, so that we don't
1706
# have to double compressed memory here
1707
# TODO: Figure out how to indicate that we would be happy to free
1708
# the fulltext content at this point. Note that sometimes we
1709
# will want it later (streaming CHK pages), but most of the
1710
# time we won't (everything else)
1711
bytes = ''.join(chunks)
1634
1713
index, start, length = self._access.add_raw_records(
1635
1714
[(None, len(bytes))], bytes)[0]
1656
1734
raise errors.RevisionNotPresent(record.key, self)
1658
1736
if record.key in inserted_keys:
1659
trace.note('Insert claimed random_id=True,'
1660
' but then inserted %r two times', record.key)
1737
trace.note(gettext('Insert claimed random_id=True,'
1738
' but then inserted %r two times'), record.key)
1662
1740
inserted_keys.add(record.key)
1663
1741
if reuse_blocks:
1802
1880
"""See VersionedFiles.keys."""
1803
1881
if 'evil' in debug.debug_flags:
1804
1882
trace.mutter_callsite(2, "keys scales with size of history")
1805
sources = [self._index] + self._fallback_vfs
1883
sources = [self._index] + self._immediate_fallback_vfs
1807
1885
for source in sources:
1808
1886
result.update(source.keys())
1890
class _GCBuildDetails(object):
1891
"""A blob of data about the build details.
1893
This stores the minimal data, which then allows compatibility with the old
1894
api, without taking as much memory.
1897
__slots__ = ('_index', '_group_start', '_group_end', '_basis_end',
1898
'_delta_end', '_parents')
1901
compression_parent = None
1903
def __init__(self, parents, position_info):
1904
self._parents = parents
1905
(self._index, self._group_start, self._group_end, self._basis_end,
1906
self._delta_end) = position_info
1909
return '%s(%s, %s)' % (self.__class__.__name__,
1910
self.index_memo, self._parents)
1913
def index_memo(self):
1914
return (self._index, self._group_start, self._group_end,
1915
self._basis_end, self._delta_end)
1918
def record_details(self):
1919
return static_tuple.StaticTuple(self.method, None)
1921
def __getitem__(self, offset):
1922
"""Compatibility thunk to act like a tuple."""
1924
return self.index_memo
1926
return self.compression_parent # Always None
1928
return self._parents
1930
return self.record_details
1932
raise IndexError('offset out of range')
1812
1938
class _GCGraphIndex(object):
1813
1939
"""Mapper from GroupCompressVersionedFiles needs into GraphIndex storage."""
1989
2115
:param keys: An iterable of keys.
1990
2116
:return: A dict of key:
1991
2117
(index_memo, compression_parent, parents, record_details).
1993
opaque structure to pass to read_records to extract the raw
1996
Content that this record is built upon, may be None
1998
Logical parents of this node
2000
extra information about the content which needs to be passed to
2001
Factory.parse_record
2119
* index_memo: opaque structure to pass to read_records to extract
2121
* compression_parent: Content that this record is built upon, may
2123
* parents: Logical parents of this node
2124
* record_details: extra information about the content which needs
2125
to be passed to Factory.parse_record
2003
2127
self._check_read()
2033
2156
# each, or about 7MB. Note that it might be even more when you consider
2034
2157
# how PyInt is allocated in separate slabs. And you can't return a slab
2035
2158
# to the OS if even 1 int on it is in use. Note though that Python uses
2036
# a LIFO when re-using PyInt slots, which probably causes more
2159
# a LIFO when re-using PyInt slots, which might cause more
2037
2160
# fragmentation.
2038
2161
start = int(bits[0])
2039
2162
start = self._int_cache.setdefault(start, start)
added
removed
bzrlib/groupcompress.py
