17
17
"""Core compression logic for compressing streams of related files."""
19
from __future__ import absolute_import
24
from bzrlib.lazy_import import lazy_import
25
lazy_import(globals(), """
26
26
from bzrlib import (
30
31
graph as _mod_graph,
39
from bzrlib.repofmt import pack_repo
40
from bzrlib.i18n import gettext
37
43
from bzrlib.btree_index import BTreeBuilder
38
44
from bzrlib.lru_cache import LRUSizeCache
39
from bzrlib.tsort import topo_sort
40
45
from bzrlib.versionedfile import (
42
48
AbsentContentFactory,
43
49
ChunkedContentFactory,
44
50
FulltextContentFactory,
51
VersionedFilesWithFallbacks,
48
54
# Minimum number of uncompressed bytes to try fetch at once when retrieving
49
55
# groupcompress blocks.
52
_USE_LZMA = False and (pylzma is not None)
54
58
# osutils.sha_string('')
55
59
_null_sha1 = 'da39a3ee5e6b4b0d3255bfef95601890afd80709'
135
139
self._content = ''.join(self._content_chunks)
136
140
self._content_chunks = None
137
141
if self._content is None:
138
if self._z_content is None:
142
# We join self._z_content_chunks here, because if we are
143
# decompressing, then it is *very* likely that we have a single
145
if self._z_content_chunks is None:
139
146
raise AssertionError('No content to decompress')
140
if self._z_content == '':
147
z_content = ''.join(self._z_content_chunks)
141
149
self._content = ''
142
150
elif self._compressor_name == 'lzma':
143
151
# We don't do partial lzma decomp yet
144
self._content = pylzma.decompress(self._z_content)
153
self._content = pylzma.decompress(z_content)
145
154
elif self._compressor_name == 'zlib':
146
155
# Start a zlib decompressor
147
156
if num_bytes * 4 > self._content_length * 3:
148
157
# If we are requesting more that 3/4ths of the content,
149
158
# just extract the whole thing in a single pass
150
159
num_bytes = self._content_length
151
self._content = zlib.decompress(self._z_content)
160
self._content = zlib.decompress(z_content)
153
162
self._z_content_decompressor = zlib.decompressobj()
154
163
# Seed the decompressor with the uncompressed bytes, so
155
164
# that the rest of the code is simplified
156
165
self._content = self._z_content_decompressor.decompress(
157
self._z_content, num_bytes + _ZLIB_DECOMP_WINDOW)
166
z_content, num_bytes + _ZLIB_DECOMP_WINDOW)
158
167
if not self._z_content_decompressor.unconsumed_tail:
159
168
self._z_content_decompressor = None
207
216
# XXX: Define some GCCorrupt error ?
208
217
raise AssertionError('Invalid bytes: (%d) != %d + %d' %
209
218
(len(bytes), pos, self._z_content_length))
210
self._z_content = bytes[pos:]
219
self._z_content_chunks = (bytes[pos:],)
222
def _z_content(self):
223
"""Return z_content_chunks as a simple string.
225
Meant only to be used by the test suite.
227
if self._z_content_chunks is not None:
228
return ''.join(self._z_content_chunks)
213
232
def from_bytes(cls, bytes):
269
288
self._content_length = length
270
289
self._content_chunks = content_chunks
271
290
self._content = None
272
self._z_content = None
291
self._z_content_chunks = None
274
293
def set_content(self, content):
275
294
"""Set the content of this block."""
276
295
self._content_length = len(content)
277
296
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):
297
self._z_content_chunks = None
299
def _create_z_content_from_chunks(self, chunks):
290
300
compressor = zlib.compressobj(zlib.Z_DEFAULT_COMPRESSION)
291
compressed_chunks = map(compressor.compress, self._content_chunks)
301
# Peak in this point is 1 fulltext, 1 compressed text, + zlib overhead
302
# (measured peak is maybe 30MB over the above...)
303
compressed_chunks = map(compressor.compress, chunks)
292
304
compressed_chunks.append(compressor.flush())
293
self._z_content = ''.join(compressed_chunks)
294
self._z_content_length = len(self._z_content)
305
# Ignore empty chunks
306
self._z_content_chunks = [c for c in compressed_chunks if c]
307
self._z_content_length = sum(map(len, self._z_content_chunks))
296
309
def _create_z_content(self):
297
if self._z_content is not None:
300
self._create_z_content_using_lzma()
310
if self._z_content_chunks is not None:
302
312
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)
313
chunks = self._content_chunks
315
chunks = (self._content,)
316
self._create_z_content_from_chunks(chunks)
319
"""Create the byte stream as a series of 'chunks'"""
320
self._create_z_content()
321
header = self.GCB_HEADER
322
chunks = ['%s%d\n%d\n'
323
% (header, self._z_content_length, self._content_length),
325
chunks.extend(self._z_content_chunks)
326
total_len = sum(map(len, chunks))
327
return total_len, chunks
308
329
def to_bytes(self):
309
330
"""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),
331
total_len, chunks = self.to_chunks()
319
332
return ''.join(chunks)
321
334
def _dump(self, include_text=False):
435
448
# Grab and cache the raw bytes for this entry
436
449
# and break the ref-cycle with _manager since we don't need it
438
self._manager._prepare_for_extract()
452
self._manager._prepare_for_extract()
453
except zlib.error as value:
454
raise errors.DecompressCorruption("zlib: " + str(value))
439
455
block = self._manager._block
440
456
self._bytes = block.extract(self.key, self._start, self._end)
441
457
# There are code paths that first extract as fulltext, and then
460
476
_full_enough_block_size = 3*1024*1024 # size at which we won't repack
461
477
_full_enough_mixed_block_size = 2*768*1024 # 1.5MB
463
def __init__(self, block):
479
def __init__(self, block, get_compressor_settings=None):
464
480
self._block = block
465
481
# We need to preserve the ordering
466
482
self._factories = []
467
483
self._last_byte = 0
484
self._get_settings = get_compressor_settings
485
self._compressor_settings = None
487
def _get_compressor_settings(self):
488
if self._compressor_settings is not None:
489
return self._compressor_settings
491
if self._get_settings is not None:
492
settings = self._get_settings()
494
vf = GroupCompressVersionedFiles
495
settings = vf._DEFAULT_COMPRESSOR_SETTINGS
496
self._compressor_settings = settings
497
return self._compressor_settings
469
499
def add_factory(self, key, parents, start, end):
470
500
if not self._factories:
503
533
new_block.set_content(self._block._content[:last_byte])
504
534
self._block = new_block
536
def _make_group_compressor(self):
537
return GroupCompressor(self._get_compressor_settings())
506
539
def _rebuild_block(self):
507
540
"""Create a new GroupCompressBlock with only the referenced texts."""
508
compressor = GroupCompressor()
541
compressor = self._make_group_compressor()
509
542
tstart = time.time()
510
543
old_length = self._block._content_length
523
556
# block? It seems hard to come up with a method that it would
524
557
# expand, since we do full compression again. Perhaps based on a
525
558
# request that ends up poorly ordered?
559
# TODO: If the content would have expanded, then we would want to
560
# handle a case where we need to split the block.
561
# Now that we have a user-tweakable option
562
# (max_bytes_to_index), it is possible that one person set it
563
# to a very low value, causing poor compression.
526
564
delta = time.time() - tstart
527
565
self._block = new_block
528
566
trace.mutter('creating new compressed block on-the-fly in %.3fs'
679
717
z_header_bytes = zlib.compress(header_bytes)
681
719
z_header_bytes_len = len(z_header_bytes)
682
block_bytes = self._block.to_bytes()
720
block_bytes_len, block_chunks = self._block.to_chunks()
683
721
lines.append('%d\n%d\n%d\n' % (z_header_bytes_len, header_bytes_len,
685
723
lines.append(z_header_bytes)
686
lines.append(block_bytes)
687
del z_header_bytes, block_bytes
724
lines.extend(block_chunks)
725
del z_header_bytes, block_chunks
726
# TODO: This is a point where we will double the memory consumption. To
727
# avoid this, we probably have to switch to a 'chunked' api
688
728
return ''.join(lines)
691
731
def from_bytes(cls, bytes):
692
732
# TODO: This does extra string copying, probably better to do it a
733
# different way. At a minimum this creates 2 copies of the
694
735
(storage_kind, z_header_len, header_len,
695
736
block_len, rest) = bytes.split('\n', 4)
855
900
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
902
self._block.set_chunked_content(self.chunks, self.endpoint)
866
903
self.chunks = None
867
904
self._delta_index = None
886
923
class PythonGroupCompressor(_CommonGroupCompressor):
925
def __init__(self, settings=None):
889
926
"""Create a GroupCompressor.
891
928
Used only if the pyrex version is not available.
893
super(PythonGroupCompressor, self).__init__()
930
super(PythonGroupCompressor, self).__init__(settings)
894
931
self._delta_index = LinesDeltaIndex([])
895
932
# The actual content is managed by LinesDeltaIndex
896
933
self.chunks = self._delta_index.lines
934
971
It contains code very similar to SequenceMatcher because of having a similar
935
972
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
974
* there is no junk, we want a minimal edit not a human readable diff.
975
* we don't filter very common lines (because we don't know where a good
976
range will start, and after the first text we want to be emitting minmal
978
* we chain the left side, not the right side
979
* we incrementally update the adjacency matrix as new lines are provided.
980
* we look for matches in all of the left side, so the routine which does
981
the analagous task of find_longest_match does not need to filter on the
948
super(PyrexGroupCompressor, self).__init__()
949
self._delta_index = DeltaIndex()
985
def __init__(self, settings=None):
986
super(PyrexGroupCompressor, self).__init__(settings)
987
max_bytes_to_index = self._settings.get('max_bytes_to_index', 0)
988
self._delta_index = DeltaIndex(max_bytes_to_index=max_bytes_to_index)
951
990
def _compress(self, key, bytes, max_delta_size, soft=False):
952
991
"""see _CommonGroupCompressor._compress"""
1027
1066
index = _GCGraphIndex(graph_index, lambda:True, parents=parents,
1028
1067
add_callback=graph_index.add_nodes,
1029
1068
inconsistency_fatal=inconsistency_fatal)
1030
access = knit._DirectPackAccess({})
1069
access = pack_repo._DirectPackAccess({})
1031
1070
access.set_writer(writer, graph_index, (transport, 'newpack'))
1032
1071
result = GroupCompressVersionedFiles(index, access, delta)
1033
1072
result.stream = stream
1044
1083
class _BatchingBlockFetcher(object):
1045
1084
"""Fetch group compress blocks in batches.
1047
1086
:ivar total_bytes: int of expected number of bytes needed to fetch the
1048
1087
currently pending batch.
1051
def __init__(self, gcvf, locations):
1090
def __init__(self, gcvf, locations, get_compressor_settings=None):
1052
1091
self.gcvf = gcvf
1053
1092
self.locations = locations
1102
1142
def yield_factories(self, full_flush=False):
1103
1143
"""Yield factories for keys added since the last yield. They will be
1104
1144
returned in the order they were added via add_key.
1106
1146
:param full_flush: by default, some results may not be returned in case
1107
1147
they can be part of the next batch. If full_flush is True, then
1108
1148
all results are returned.
1136
1176
memos_to_get_stack.pop()
1138
1178
block = self.batch_memos[read_memo]
1139
self.manager = _LazyGroupContentManager(block)
1179
self.manager = _LazyGroupContentManager(block,
1180
get_compressor_settings=self._get_compressor_settings)
1140
1181
self.last_read_memo = read_memo
1141
1182
start, end = index_memo[3:5]
1142
1183
self.manager.add_factory(key, parents, start, end)
1149
1190
self.total_bytes = 0
1152
class GroupCompressVersionedFiles(VersionedFiles):
1193
class GroupCompressVersionedFiles(VersionedFilesWithFallbacks):
1153
1194
"""A group-compress based VersionedFiles implementation."""
1155
def __init__(self, index, access, delta=True, _unadded_refs=None):
1196
# This controls how the GroupCompress DeltaIndex works. Basically, we
1197
# compute hash pointers into the source blocks (so hash(text) => text).
1198
# However each of these references costs some memory in trade against a
1199
# more accurate match result. For very large files, they either are
1200
# pre-compressed and change in bulk whenever they change, or change in just
1201
# local blocks. Either way, 'improved resolution' is not very helpful,
1202
# versus running out of memory trying to track everything. The default max
1203
# gives 100% sampling of a 1MB file.
1204
_DEFAULT_MAX_BYTES_TO_INDEX = 1024 * 1024
1205
_DEFAULT_COMPRESSOR_SETTINGS = {'max_bytes_to_index':
1206
_DEFAULT_MAX_BYTES_TO_INDEX}
1208
def __init__(self, index, access, delta=True, _unadded_refs=None,
1156
1210
"""Create a GroupCompressVersionedFiles object.
1158
1212
:param index: The index object storing access and graph data.
1159
1213
:param access: The access object storing raw data.
1160
1214
:param delta: Whether to delta compress or just entropy compress.
1161
1215
:param _unadded_refs: private parameter, don't use.
1216
:param _group_cache: private parameter, don't use.
1163
1218
self._index = index
1164
1219
self._access = access
1166
1221
if _unadded_refs is None:
1167
1222
_unadded_refs = {}
1168
1223
self._unadded_refs = _unadded_refs
1169
self._group_cache = LRUSizeCache(max_size=50*1024*1024)
1170
self._fallback_vfs = []
1224
if _group_cache is None:
1225
_group_cache = LRUSizeCache(max_size=50*1024*1024)
1226
self._group_cache = _group_cache
1227
self._immediate_fallback_vfs = []
1228
self._max_bytes_to_index = None
1172
1230
def without_fallbacks(self):
1173
1231
"""Return a clone of this object without any fallbacks configured."""
1174
1232
return GroupCompressVersionedFiles(self._index, self._access,
1175
self._delta, _unadded_refs=dict(self._unadded_refs))
1233
self._delta, _unadded_refs=dict(self._unadded_refs),
1234
_group_cache=self._group_cache)
1177
1236
def add_lines(self, key, parents, lines, parent_texts=None,
1178
1237
left_matching_blocks=None, nostore_sha=None, random_id=False,
1182
1241
:param key: The key tuple of the text to add.
1183
1242
:param parents: The parents key tuples of the text to add.
1184
1243
: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
1244
of them except the last must be terminated with \\n and contain no
1245
other \\n's. The last line may either contain no \\n's or a single
1246
terminating \\n. If the lines list does meet this constraint the
1247
add routine may error or may succeed - but you will be unable to
1248
read the data back accurately. (Checking the lines have been split
1190
1249
correctly is expensive and extremely unlikely to catch bugs so it
1191
1250
is not done at runtime unless check_content is True.)
1192
1251
:param parent_texts: An optional dictionary containing the opaque
1287
1346
self._check_lines_not_unicode(lines)
1288
1347
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
1349
def get_parent_map(self, keys):
1307
1350
"""Get a map of the graph parents of keys.
1447
1490
The returned objects should be in the order defined by 'ordering',
1448
1491
which can weave between different sources.
1449
1493
:param ordering: Must be one of 'topological' or 'groupcompress'
1450
1494
:return: List of [(source, [keys])] tuples, such that all keys are in
1451
1495
the defined order, regardless of source.
1453
1497
if ordering == 'topological':
1454
present_keys = topo_sort(parent_map)
1498
present_keys = tsort.topo_sort(parent_map)
1456
1500
# ordering == 'groupcompress'
1457
1501
# XXX: This only optimizes for the target ordering. We may need
1546
1590
# - we encounter an unadded ref, or
1547
1591
# - we run out of keys, or
1548
1592
# - the total bytes to retrieve for this batch > BATCH_SIZE
1549
batcher = _BatchingBlockFetcher(self, locations)
1593
batcher = _BatchingBlockFetcher(self, locations,
1594
get_compressor_settings=self._get_compressor_settings)
1550
1595
for source, keys in source_keys:
1551
1596
if source is self:
1552
1597
for key in keys:
1598
1643
for _ in self._insert_record_stream(stream, random_id=False):
1646
def _get_compressor_settings(self):
1647
if self._max_bytes_to_index is None:
1648
# TODO: VersionedFiles don't know about their containing
1649
# repository, so they don't have much of an idea about their
1650
# location. So for now, this is only a global option.
1651
c = config.GlobalConfig()
1652
val = c.get_user_option('bzr.groupcompress.max_bytes_to_index')
1656
except ValueError, e:
1657
trace.warning('Value for '
1658
'"bzr.groupcompress.max_bytes_to_index"'
1659
' %r is not an integer'
1663
val = self._DEFAULT_MAX_BYTES_TO_INDEX
1664
self._max_bytes_to_index = val
1665
return {'max_bytes_to_index': self._max_bytes_to_index}
1667
def _make_group_compressor(self):
1668
return GroupCompressor(self._get_compressor_settings())
1601
1670
def _insert_record_stream(self, stream, random_id=False, nostore_sha=None,
1602
1671
reuse_blocks=True):
1603
1672
"""Internal core to insert a record stream into this container.
1627
1696
# This will go up to fulltexts for gc to gc fetching, which isn't
1629
self._compressor = GroupCompressor()
1698
self._compressor = self._make_group_compressor()
1630
1699
self._unadded_refs = {}
1631
1700
keys_to_add = []
1633
bytes = self._compressor.flush().to_bytes()
1634
self._compressor = GroupCompressor()
1702
bytes_len, chunks = self._compressor.flush().to_chunks()
1703
self._compressor = self._make_group_compressor()
1704
# Note: At this point we still have 1 copy of the fulltext (in
1705
# record and the var 'bytes'), and this generates 2 copies of
1706
# the compressed text (one for bytes, one in chunks)
1707
# TODO: Push 'chunks' down into the _access api, so that we don't
1708
# have to double compressed memory here
1709
# TODO: Figure out how to indicate that we would be happy to free
1710
# the fulltext content at this point. Note that sometimes we
1711
# will want it later (streaming CHK pages), but most of the
1712
# time we won't (everything else)
1713
bytes = ''.join(chunks)
1635
1715
index, start, length = self._access.add_raw_records(
1636
1716
[(None, len(bytes))], bytes)[0]
1656
1736
raise errors.RevisionNotPresent(record.key, self)
1658
1738
if record.key in inserted_keys:
1659
trace.note('Insert claimed random_id=True,'
1660
' but then inserted %r two times', record.key)
1739
trace.note(gettext('Insert claimed random_id=True,'
1740
' but then inserted %r two times'), record.key)
1662
1742
inserted_keys.add(record.key)
1663
1743
if reuse_blocks:
2037
2117
:param keys: An iterable of keys.
2038
2118
:return: A dict of key:
2039
2119
(index_memo, compression_parent, parents, record_details).
2041
opaque structure to pass to read_records to extract the raw
2044
Content that this record is built upon, may be None
2046
Logical parents of this node
2048
extra information about the content which needs to be passed to
2049
Factory.parse_record
2121
* index_memo: opaque structure to pass to read_records to extract
2123
* compression_parent: Content that this record is built upon, may
2125
* parents: Logical parents of this node
2126
* record_details: extra information about the content which needs
2127
to be passed to Factory.parse_record
2051
2129
self._check_read()
added
removed
bzrlib/groupcompress.py
