1
# Copyright (C) 2008, 2009, 2010 Canonical Ltd
1
# Copyright (C) 2008-2011 Canonical Ltd
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
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'
145
149
self._content = ''
146
150
elif self._compressor_name == 'lzma':
147
151
# We don't do partial lzma decomp yet
148
153
self._content = pylzma.decompress(z_content)
149
154
elif self._compressor_name == 'zlib':
150
155
# Start a zlib decompressor
291
296
self._content = content
292
297
self._z_content_chunks = None
294
def _create_z_content_using_lzma(self):
295
if self._content_chunks is not None:
296
self._content = ''.join(self._content_chunks)
297
self._content_chunks = None
298
if self._content is None:
299
raise AssertionError('Nothing to compress')
300
z_content = pylzma.compress(self._content)
301
self._z_content_chunks = (z_content,)
302
self._z_content_length = len(z_content)
304
299
def _create_z_content_from_chunks(self, chunks):
305
300
compressor = zlib.compressobj(zlib.Z_DEFAULT_COMPRESSION)
306
301
# Peak in this point is 1 fulltext, 1 compressed text, + zlib overhead
314
309
def _create_z_content(self):
315
310
if self._z_content_chunks is not None:
318
self._create_z_content_using_lzma()
320
312
if self._content_chunks is not None:
321
313
chunks = self._content_chunks
326
318
def to_chunks(self):
327
319
"""Create the byte stream as a series of 'chunks'"""
328
320
self._create_z_content()
330
header = self.GCB_LZ_HEADER
332
header = self.GCB_HEADER
321
header = self.GCB_HEADER
333
322
chunks = ['%s%d\n%d\n'
334
323
% (header, self._z_content_length, self._content_length),
459
448
# Grab and cache the raw bytes for this entry
460
449
# and break the ref-cycle with _manager since we don't need it
462
self._manager._prepare_for_extract()
452
self._manager._prepare_for_extract()
453
except zlib.error as value:
454
raise errors.DecompressCorruption("zlib: " + str(value))
463
455
block = self._manager._block
464
456
self._bytes = block.extract(self.key, self._start, self._end)
465
457
# There are code paths that first extract as fulltext, and then
484
476
_full_enough_block_size = 3*1024*1024 # size at which we won't repack
485
477
_full_enough_mixed_block_size = 2*768*1024 # 1.5MB
487
def __init__(self, block):
479
def __init__(self, block, get_compressor_settings=None):
488
480
self._block = block
489
481
# We need to preserve the ordering
490
482
self._factories = []
491
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
493
499
def add_factory(self, key, parents, start, end):
494
500
if not self._factories:
527
533
new_block.set_content(self._block._content[:last_byte])
528
534
self._block = new_block
536
def _make_group_compressor(self):
537
return GroupCompressor(self._get_compressor_settings())
530
539
def _rebuild_block(self):
531
540
"""Create a new GroupCompressBlock with only the referenced texts."""
532
compressor = GroupCompressor()
541
compressor = self._make_group_compressor()
533
542
tstart = time.time()
534
543
old_length = self._block._content_length
547
556
# block? It seems hard to come up with a method that it would
548
557
# expand, since we do full compression again. Perhaps based on a
549
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.
550
564
delta = time.time() - tstart
551
565
self._block = new_block
552
566
trace.mutter('creating new compressed block on-the-fly in %.3fs'
784
798
self.labels_deltas = {}
785
799
self._delta_index = None # Set by the children
786
800
self._block = GroupCompressBlock()
804
self._settings = settings
788
806
def compress(self, key, bytes, expected_sha, nostore_sha=None, soft=False):
789
807
"""Compress lines with label key.
905
923
class PythonGroupCompressor(_CommonGroupCompressor):
925
def __init__(self, settings=None):
908
926
"""Create a GroupCompressor.
910
928
Used only if the pyrex version is not available.
912
super(PythonGroupCompressor, self).__init__()
930
super(PythonGroupCompressor, self).__init__(settings)
913
931
self._delta_index = LinesDeltaIndex([])
914
932
# The actual content is managed by LinesDeltaIndex
915
933
self.chunks = self._delta_index.lines
953
971
It contains code very similar to SequenceMatcher because of having a similar
954
972
task. However some key differences apply:
955
- there is no junk, we want a minimal edit not a human readable diff.
956
- we don't filter very common lines (because we don't know where a good
957
range will start, and after the first text we want to be emitting minmal
959
- we chain the left side, not the right side
960
- we incrementally update the adjacency matrix as new lines are provided.
961
- we look for matches in all of the left side, so the routine which does
962
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
967
super(PyrexGroupCompressor, self).__init__()
968
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)
970
990
def _compress(self, key, bytes, max_delta_size, soft=False):
971
991
"""see _CommonGroupCompressor._compress"""
1046
1066
index = _GCGraphIndex(graph_index, lambda:True, parents=parents,
1047
1067
add_callback=graph_index.add_nodes,
1048
1068
inconsistency_fatal=inconsistency_fatal)
1049
access = knit._DirectPackAccess({})
1069
access = pack_repo._DirectPackAccess({})
1050
1070
access.set_writer(writer, graph_index, (transport, 'newpack'))
1051
1071
result = GroupCompressVersionedFiles(index, access, delta)
1052
1072
result.stream = stream
1063
1083
class _BatchingBlockFetcher(object):
1064
1084
"""Fetch group compress blocks in batches.
1066
1086
:ivar total_bytes: int of expected number of bytes needed to fetch the
1067
1087
currently pending batch.
1070
def __init__(self, gcvf, locations):
1090
def __init__(self, gcvf, locations, get_compressor_settings=None):
1071
1091
self.gcvf = gcvf
1072
1092
self.locations = locations
1121
1142
def yield_factories(self, full_flush=False):
1122
1143
"""Yield factories for keys added since the last yield. They will be
1123
1144
returned in the order they were added via add_key.
1125
1146
:param full_flush: by default, some results may not be returned in case
1126
1147
they can be part of the next batch. If full_flush is True, then
1127
1148
all results are returned.
1155
1176
memos_to_get_stack.pop()
1157
1178
block = self.batch_memos[read_memo]
1158
self.manager = _LazyGroupContentManager(block)
1179
self.manager = _LazyGroupContentManager(block,
1180
get_compressor_settings=self._get_compressor_settings)
1159
1181
self.last_read_memo = read_memo
1160
1182
start, end = index_memo[3:5]
1161
1183
self.manager.add_factory(key, parents, start, end)
1168
1190
self.total_bytes = 0
1171
class GroupCompressVersionedFiles(VersionedFiles):
1193
class GroupCompressVersionedFiles(VersionedFilesWithFallbacks):
1172
1194
"""A group-compress based VersionedFiles implementation."""
1174
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,
1175
1210
"""Create a GroupCompressVersionedFiles object.
1177
1212
:param index: The index object storing access and graph data.
1178
1213
:param access: The access object storing raw data.
1179
1214
:param delta: Whether to delta compress or just entropy compress.
1180
1215
:param _unadded_refs: private parameter, don't use.
1216
:param _group_cache: private parameter, don't use.
1182
1218
self._index = index
1183
1219
self._access = access
1185
1221
if _unadded_refs is None:
1186
1222
_unadded_refs = {}
1187
1223
self._unadded_refs = _unadded_refs
1188
self._group_cache = LRUSizeCache(max_size=50*1024*1024)
1189
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
1191
1230
def without_fallbacks(self):
1192
1231
"""Return a clone of this object without any fallbacks configured."""
1193
1232
return GroupCompressVersionedFiles(self._index, self._access,
1194
self._delta, _unadded_refs=dict(self._unadded_refs))
1233
self._delta, _unadded_refs=dict(self._unadded_refs),
1234
_group_cache=self._group_cache)
1196
1236
def add_lines(self, key, parents, lines, parent_texts=None,
1197
1237
left_matching_blocks=None, nostore_sha=None, random_id=False,
1201
1241
:param key: The key tuple of the text to add.
1202
1242
:param parents: The parents key tuples of the text to add.
1203
1243
:param lines: A list of lines. Each line must be a bytestring. And all
1204
of them except the last must be terminated with \n and contain no
1205
other \n's. The last line may either contain no \n's or a single
1206
terminating \n. If the lines list does meet this constraint the add
1207
routine may error or may succeed - but you will be unable to read
1208
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
1209
1249
correctly is expensive and extremely unlikely to catch bugs so it
1210
1250
is not done at runtime unless check_content is True.)
1211
1251
:param parent_texts: An optional dictionary containing the opaque
1306
1346
self._check_lines_not_unicode(lines)
1307
1347
self._check_lines_are_lines(lines)
1309
def get_known_graph_ancestry(self, keys):
1310
"""Get a KnownGraph instance with the ancestry of keys."""
1311
# Note that this is identical to
1312
# KnitVersionedFiles.get_known_graph_ancestry, but they don't share
1314
parent_map, missing_keys = self._index.find_ancestry(keys)
1315
for fallback in self._fallback_vfs:
1316
if not missing_keys:
1318
(f_parent_map, f_missing_keys) = fallback._index.find_ancestry(
1320
parent_map.update(f_parent_map)
1321
missing_keys = f_missing_keys
1322
kg = _mod_graph.KnownGraph(parent_map)
1325
1349
def get_parent_map(self, keys):
1326
1350
"""Get a map of the graph parents of keys.
1466
1490
The returned objects should be in the order defined by 'ordering',
1467
1491
which can weave between different sources.
1468
1493
:param ordering: Must be one of 'topological' or 'groupcompress'
1469
1494
:return: List of [(source, [keys])] tuples, such that all keys are in
1470
1495
the defined order, regardless of source.
1472
1497
if ordering == 'topological':
1473
present_keys = topo_sort(parent_map)
1498
present_keys = tsort.topo_sort(parent_map)
1475
1500
# ordering == 'groupcompress'
1476
1501
# XXX: This only optimizes for the target ordering. We may need
1565
1590
# - we encounter an unadded ref, or
1566
1591
# - we run out of keys, or
1567
1592
# - the total bytes to retrieve for this batch > BATCH_SIZE
1568
batcher = _BatchingBlockFetcher(self, locations)
1593
batcher = _BatchingBlockFetcher(self, locations,
1594
get_compressor_settings=self._get_compressor_settings)
1569
1595
for source, keys in source_keys:
1570
1596
if source is self:
1571
1597
for key in keys:
1617
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())
1620
1670
def _insert_record_stream(self, stream, random_id=False, nostore_sha=None,
1621
1671
reuse_blocks=True):
1622
1672
"""Internal core to insert a record stream into this container.
1646
1696
# This will go up to fulltexts for gc to gc fetching, which isn't
1648
self._compressor = GroupCompressor()
1698
self._compressor = self._make_group_compressor()
1649
1699
self._unadded_refs = {}
1650
1700
keys_to_add = []
1652
1702
bytes_len, chunks = self._compressor.flush().to_chunks()
1653
self._compressor = GroupCompressor()
1703
self._compressor = self._make_group_compressor()
1654
1704
# Note: At this point we still have 1 copy of the fulltext (in
1655
1705
# record and the var 'bytes'), and this generates 2 copies of
1656
1706
# the compressed text (one for bytes, one in chunks)
1686
1736
raise errors.RevisionNotPresent(record.key, self)
1688
1738
if record.key in inserted_keys:
1689
trace.note('Insert claimed random_id=True,'
1690
' 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)
1692
1742
inserted_keys.add(record.key)
1693
1743
if reuse_blocks:
1832
1882
"""See VersionedFiles.keys."""
1833
1883
if 'evil' in debug.debug_flags:
1834
1884
trace.mutter_callsite(2, "keys scales with size of history")
1835
sources = [self._index] + self._fallback_vfs
1885
sources = [self._index] + self._immediate_fallback_vfs
1837
1887
for source in sources:
1838
1888
result.update(source.keys())
1921
1971
# repeated over and over, this creates a surplus of ints
1922
1972
self._int_cache = {}
1923
1973
if track_external_parent_refs:
1924
self._key_dependencies = knit._KeyRefs(
1974
self._key_dependencies = _KeyRefs(
1925
1975
track_new_keys=track_new_keys)
1927
1977
self._key_dependencies = None
2067
2117
:param keys: An iterable of keys.
2068
2118
:return: A dict of key:
2069
2119
(index_memo, compression_parent, parents, record_details).
2071
opaque structure to pass to read_records to extract the raw
2074
Content that this record is built upon, may be None
2076
Logical parents of this node
2078
extra information about the content which needs to be passed to
2079
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
2081
2129
self._check_read()
added
removed
bzrlib/groupcompress.py
