17
17
"""Core compression logic for compressing streams of related files."""
19
from __future__ import absolute_import
24
26
from bzrlib.lazy_import import lazy_import
25
27
lazy_import(globals(), """
26
28
from bzrlib import (
31
32
graph as _mod_graph,
149
151
self._content = ''
150
152
elif self._compressor_name == 'lzma':
151
153
# We don't do partial lzma decomp yet
153
154
self._content = pylzma.decompress(z_content)
154
155
elif self._compressor_name == 'zlib':
155
156
# Start a zlib decompressor
296
297
self._content = content
297
298
self._z_content_chunks = None
300
def _create_z_content_using_lzma(self):
301
if self._content_chunks is not None:
302
self._content = ''.join(self._content_chunks)
303
self._content_chunks = None
304
if self._content is None:
305
raise AssertionError('Nothing to compress')
306
z_content = pylzma.compress(self._content)
307
self._z_content_chunks = (z_content,)
308
self._z_content_length = len(z_content)
299
310
def _create_z_content_from_chunks(self, chunks):
300
311
compressor = zlib.compressobj(zlib.Z_DEFAULT_COMPRESSION)
301
312
# Peak in this point is 1 fulltext, 1 compressed text, + zlib overhead
309
320
def _create_z_content(self):
310
321
if self._z_content_chunks is not None:
324
self._create_z_content_using_lzma()
312
326
if self._content_chunks is not None:
313
327
chunks = self._content_chunks
318
332
def to_chunks(self):
319
333
"""Create the byte stream as a series of 'chunks'"""
320
334
self._create_z_content()
321
header = self.GCB_HEADER
336
header = self.GCB_LZ_HEADER
338
header = self.GCB_HEADER
322
339
chunks = ['%s%d\n%d\n'
323
340
% (header, self._z_content_length, self._content_length),
448
465
# Grab and cache the raw bytes for this entry
449
466
# and break the ref-cycle with _manager since we don't need it
452
self._manager._prepare_for_extract()
453
except zlib.error as value:
454
raise errors.DecompressCorruption("zlib: " + str(value))
468
self._manager._prepare_for_extract()
455
469
block = self._manager._block
456
470
self._bytes = block.extract(self.key, self._start, self._end)
457
471
# There are code paths that first extract as fulltext, and then
476
490
_full_enough_block_size = 3*1024*1024 # size at which we won't repack
477
491
_full_enough_mixed_block_size = 2*768*1024 # 1.5MB
479
def __init__(self, block, get_compressor_settings=None):
493
def __init__(self, block):
480
494
self._block = block
481
495
# We need to preserve the ordering
482
496
self._factories = []
483
497
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
499
499
def add_factory(self, key, parents, start, end):
500
500
if not self._factories:
533
533
new_block.set_content(self._block._content[:last_byte])
534
534
self._block = new_block
536
def _make_group_compressor(self):
537
return GroupCompressor(self._get_compressor_settings())
539
536
def _rebuild_block(self):
540
537
"""Create a new GroupCompressBlock with only the referenced texts."""
541
compressor = self._make_group_compressor()
538
compressor = GroupCompressor()
542
539
tstart = time.time()
543
540
old_length = self._block._content_length
556
553
# block? It seems hard to come up with a method that it would
557
554
# expand, since we do full compression again. Perhaps based on a
558
555
# 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.
564
556
delta = time.time() - tstart
565
557
self._block = new_block
566
558
trace.mutter('creating new compressed block on-the-fly in %.3fs'
798
790
self.labels_deltas = {}
799
791
self._delta_index = None # Set by the children
800
792
self._block = GroupCompressBlock()
804
self._settings = settings
806
794
def compress(self, key, bytes, expected_sha, nostore_sha=None, soft=False):
807
795
"""Compress lines with label key.
923
911
class PythonGroupCompressor(_CommonGroupCompressor):
925
def __init__(self, settings=None):
926
914
"""Create a GroupCompressor.
928
916
Used only if the pyrex version is not available.
930
super(PythonGroupCompressor, self).__init__(settings)
918
super(PythonGroupCompressor, self).__init__()
931
919
self._delta_index = LinesDeltaIndex([])
932
920
# The actual content is managed by LinesDeltaIndex
933
921
self.chunks = self._delta_index.lines
971
959
It contains code very similar to SequenceMatcher because of having a similar
972
960
task. However some key differences apply:
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
961
- there is no junk, we want a minimal edit not a human readable diff.
962
- we don't filter very common lines (because we don't know where a good
963
range will start, and after the first text we want to be emitting minmal
965
- we chain the left side, not the right side
966
- we incrementally update the adjacency matrix as new lines are provided.
967
- we look for matches in all of the left side, so the routine which does
968
the analagous task of find_longest_match does not need to filter on the
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)
973
super(PyrexGroupCompressor, self).__init__()
974
self._delta_index = DeltaIndex()
990
976
def _compress(self, key, bytes, max_delta_size, soft=False):
991
977
"""see _CommonGroupCompressor._compress"""
1083
1069
class _BatchingBlockFetcher(object):
1084
1070
"""Fetch group compress blocks in batches.
1086
1072
:ivar total_bytes: int of expected number of bytes needed to fetch the
1087
1073
currently pending batch.
1090
def __init__(self, gcvf, locations, get_compressor_settings=None):
1076
def __init__(self, gcvf, locations):
1091
1077
self.gcvf = gcvf
1092
1078
self.locations = locations
1142
1127
def yield_factories(self, full_flush=False):
1143
1128
"""Yield factories for keys added since the last yield. They will be
1144
1129
returned in the order they were added via add_key.
1146
1131
:param full_flush: by default, some results may not be returned in case
1147
1132
they can be part of the next batch. If full_flush is True, then
1148
1133
all results are returned.
1176
1161
memos_to_get_stack.pop()
1178
1163
block = self.batch_memos[read_memo]
1179
self.manager = _LazyGroupContentManager(block,
1180
get_compressor_settings=self._get_compressor_settings)
1164
self.manager = _LazyGroupContentManager(block)
1181
1165
self.last_read_memo = read_memo
1182
1166
start, end = index_memo[3:5]
1183
1167
self.manager.add_factory(key, parents, start, end)
1193
1177
class GroupCompressVersionedFiles(VersionedFilesWithFallbacks):
1194
1178
"""A group-compress based VersionedFiles implementation."""
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
1180
def __init__(self, index, access, delta=True, _unadded_refs=None,
1210
1182
"""Create a GroupCompressVersionedFiles object.
1212
1184
:param index: The index object storing access and graph data.
1225
1197
_group_cache = LRUSizeCache(max_size=50*1024*1024)
1226
1198
self._group_cache = _group_cache
1227
1199
self._immediate_fallback_vfs = []
1228
self._max_bytes_to_index = None
1230
1201
def without_fallbacks(self):
1231
1202
"""Return a clone of this object without any fallbacks configured."""
1241
1212
:param key: The key tuple of the text to add.
1242
1213
:param parents: The parents key tuples of the text to add.
1243
1214
:param lines: A list of lines. Each line must be a bytestring. And all
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
1215
of them except the last must be terminated with \n and contain no
1216
other \n's. The last line may either contain no \n's or a single
1217
terminating \n. If the lines list does meet this constraint the add
1218
routine may error or may succeed - but you will be unable to read
1219
the data back accurately. (Checking the lines have been split
1249
1220
correctly is expensive and extremely unlikely to catch bugs so it
1250
1221
is not done at runtime unless check_content is True.)
1251
1222
:param parent_texts: An optional dictionary containing the opaque
1346
1317
self._check_lines_not_unicode(lines)
1347
1318
self._check_lines_are_lines(lines)
1320
def get_known_graph_ancestry(self, keys):
1321
"""Get a KnownGraph instance with the ancestry of keys."""
1322
# Note that this is identical to
1323
# KnitVersionedFiles.get_known_graph_ancestry, but they don't share
1325
parent_map, missing_keys = self._index.find_ancestry(keys)
1326
for fallback in self._transitive_fallbacks():
1327
if not missing_keys:
1329
(f_parent_map, f_missing_keys) = fallback._index.find_ancestry(
1331
parent_map.update(f_parent_map)
1332
missing_keys = f_missing_keys
1333
kg = _mod_graph.KnownGraph(parent_map)
1349
1336
def get_parent_map(self, keys):
1350
1337
"""Get a map of the graph parents of keys.
1490
1477
The returned objects should be in the order defined by 'ordering',
1491
1478
which can weave between different sources.
1493
1479
:param ordering: Must be one of 'topological' or 'groupcompress'
1494
1480
:return: List of [(source, [keys])] tuples, such that all keys are in
1495
1481
the defined order, regardless of source.
1590
1576
# - we encounter an unadded ref, or
1591
1577
# - we run out of keys, or
1592
1578
# - the total bytes to retrieve for this batch > BATCH_SIZE
1593
batcher = _BatchingBlockFetcher(self, locations,
1594
get_compressor_settings=self._get_compressor_settings)
1579
batcher = _BatchingBlockFetcher(self, locations)
1595
1580
for source, keys in source_keys:
1596
1581
if source is self:
1597
1582
for key in keys:
1643
1628
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())
1670
1631
def _insert_record_stream(self, stream, random_id=False, nostore_sha=None,
1671
1632
reuse_blocks=True):
1672
1633
"""Internal core to insert a record stream into this container.
1696
1657
# This will go up to fulltexts for gc to gc fetching, which isn't
1698
self._compressor = self._make_group_compressor()
1659
self._compressor = GroupCompressor()
1699
1660
self._unadded_refs = {}
1700
1661
keys_to_add = []
1702
1663
bytes_len, chunks = self._compressor.flush().to_chunks()
1703
self._compressor = self._make_group_compressor()
1664
self._compressor = GroupCompressor()
1704
1665
# Note: At this point we still have 1 copy of the fulltext (in
1705
1666
# record and the var 'bytes'), and this generates 2 copies of
1706
1667
# the compressed text (one for bytes, one in chunks)
1736
1697
raise errors.RevisionNotPresent(record.key, self)
1738
1699
if record.key in inserted_keys:
1739
trace.note(gettext('Insert claimed random_id=True,'
1740
' but then inserted %r two times'), record.key)
1700
trace.note('Insert claimed random_id=True,'
1701
' but then inserted %r two times', record.key)
1742
1703
inserted_keys.add(record.key)
1743
1704
if reuse_blocks:
2117
2078
:param keys: An iterable of keys.
2118
2079
:return: A dict of key:
2119
2080
(index_memo, compression_parent, parents, record_details).
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
2082
opaque structure to pass to read_records to extract the raw
2085
Content that this record is built upon, may be None
2087
Logical parents of this node
2089
extra information about the content which needs to be passed to
2090
Factory.parse_record
2129
2092
self._check_read()
added
removed
bzrlib/groupcompress.py
