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 (
31
30
graph as _mod_graph,
39
from bzrlib.repofmt import pack_repo
40
from bzrlib.i18n import gettext
43
36
from bzrlib.btree_index import BTreeBuilder
44
37
from bzrlib.lru_cache import LRUSizeCache
38
from bzrlib.tsort import topo_sort
45
39
from bzrlib.versionedfile import (
48
41
AbsentContentFactory,
49
42
ChunkedContentFactory,
50
43
FulltextContentFactory,
51
VersionedFilesWithFallbacks,
54
47
# Minimum number of uncompressed bytes to try fetch at once when retrieving
55
48
# groupcompress blocks.
51
_USE_LZMA = False and (pylzma is not None)
58
53
# osutils.sha_string('')
59
54
_null_sha1 = 'da39a3ee5e6b4b0d3255bfef95601890afd80709'
124
119
:param num_bytes: Ensure that we have extracted at least num_bytes of
125
120
content. If None, consume everything
127
if self._content_length is None:
128
raise AssertionError('self._content_length should never be None')
122
# TODO: If we re-use the same content block at different times during
123
# get_record_stream(), it is possible that the first pass will
124
# get inserted, triggering an extract/_ensure_content() which
125
# will get rid of _z_content. And then the next use of the block
126
# will try to access _z_content (to send it over the wire), and
127
# fail because it is already extracted. Consider never releasing
128
# _z_content because of this.
129
129
if num_bytes is None:
130
130
num_bytes = self._content_length
131
131
elif (self._content_length is not None
139
139
self._content = ''.join(self._content_chunks)
140
140
self._content_chunks = None
141
141
if self._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:
142
if self._z_content is None:
146
143
raise AssertionError('No content to decompress')
147
z_content = ''.join(self._z_content_chunks)
144
if self._z_content == '':
149
145
self._content = ''
150
146
elif self._compressor_name == 'lzma':
151
147
# We don't do partial lzma decomp yet
153
self._content = pylzma.decompress(z_content)
148
self._content = pylzma.decompress(self._z_content)
154
149
elif self._compressor_name == 'zlib':
155
150
# Start a zlib decompressor
156
if num_bytes * 4 > self._content_length * 3:
157
# If we are requesting more that 3/4ths of the content,
158
# just extract the whole thing in a single pass
159
num_bytes = self._content_length
160
self._content = zlib.decompress(z_content)
151
if num_bytes is None:
152
self._content = zlib.decompress(self._z_content)
162
154
self._z_content_decompressor = zlib.decompressobj()
163
155
# Seed the decompressor with the uncompressed bytes, so
164
156
# that the rest of the code is simplified
165
157
self._content = self._z_content_decompressor.decompress(
166
z_content, num_bytes + _ZLIB_DECOMP_WINDOW)
167
if not self._z_content_decompressor.unconsumed_tail:
168
self._z_content_decompressor = None
158
self._z_content, num_bytes + _ZLIB_DECOMP_WINDOW)
170
160
raise AssertionError('Unknown compressor: %r'
171
161
% self._compressor_name)
173
163
# 'unconsumed_tail'
175
165
# Do we have enough bytes already?
176
if len(self._content) >= num_bytes:
166
if num_bytes is not None and len(self._content) >= num_bytes:
168
if num_bytes is None and self._z_content_decompressor is None:
169
# We must have already decompressed everything
178
171
# If we got this far, and don't have a decompressor, something is wrong
179
172
if self._z_content_decompressor is None:
180
173
raise AssertionError(
181
174
'No decompressor to decompress %d bytes' % num_bytes)
182
175
remaining_decomp = self._z_content_decompressor.unconsumed_tail
183
if not remaining_decomp:
184
raise AssertionError('Nothing left to decompress')
185
needed_bytes = num_bytes - len(self._content)
186
# We always set max_size to 32kB over the minimum needed, so that
187
# zlib will give us as much as we really want.
188
# TODO: If this isn't good enough, we could make a loop here,
189
# that keeps expanding the request until we get enough
190
self._content += self._z_content_decompressor.decompress(
191
remaining_decomp, needed_bytes + _ZLIB_DECOMP_WINDOW)
192
if len(self._content) < num_bytes:
193
raise AssertionError('%d bytes wanted, only %d available'
194
% (num_bytes, len(self._content)))
195
if not self._z_content_decompressor.unconsumed_tail:
196
# The stream is finished
197
self._z_content_decompressor = None
176
if num_bytes is None:
178
# We don't know how much is left, but we'll decompress it all
179
self._content += self._z_content_decompressor.decompress(
181
# Note: There's what I consider a bug in zlib.decompressobj
182
# If you pass back in the entire unconsumed_tail, only
183
# this time you don't pass a max-size, it doesn't
184
# change the unconsumed_tail back to None/''.
185
# However, we know we are done with the whole stream
186
self._z_content_decompressor = None
187
# XXX: Why is this the only place in this routine we set this?
188
self._content_length = len(self._content)
190
if not remaining_decomp:
191
raise AssertionError('Nothing left to decompress')
192
needed_bytes = num_bytes - len(self._content)
193
# We always set max_size to 32kB over the minimum needed, so that
194
# zlib will give us as much as we really want.
195
# TODO: If this isn't good enough, we could make a loop here,
196
# that keeps expanding the request until we get enough
197
self._content += self._z_content_decompressor.decompress(
198
remaining_decomp, needed_bytes + _ZLIB_DECOMP_WINDOW)
199
if len(self._content) < num_bytes:
200
raise AssertionError('%d bytes wanted, only %d available'
201
% (num_bytes, len(self._content)))
202
if not self._z_content_decompressor.unconsumed_tail:
203
# The stream is finished
204
self._z_content_decompressor = None
199
206
def _parse_bytes(self, bytes, pos):
200
207
"""Read the various lengths from the header.
216
223
# XXX: Define some GCCorrupt error ?
217
224
raise AssertionError('Invalid bytes: (%d) != %d + %d' %
218
225
(len(bytes), pos, self._z_content_length))
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)
226
self._z_content = bytes[pos:]
232
229
def from_bytes(cls, bytes):
288
285
self._content_length = length
289
286
self._content_chunks = content_chunks
290
287
self._content = None
291
self._z_content_chunks = None
288
self._z_content = None
293
290
def set_content(self, content):
294
291
"""Set the content of this block."""
295
292
self._content_length = len(content)
296
293
self._content = content
297
self._z_content_chunks = None
299
def _create_z_content_from_chunks(self, chunks):
294
self._z_content = None
296
def _create_z_content_using_lzma(self):
297
if self._content_chunks is not None:
298
self._content = ''.join(self._content_chunks)
299
self._content_chunks = None
300
if self._content is None:
301
raise AssertionError('Nothing to compress')
302
self._z_content = pylzma.compress(self._content)
303
self._z_content_length = len(self._z_content)
305
def _create_z_content_from_chunks(self):
300
306
compressor = zlib.compressobj(zlib.Z_DEFAULT_COMPRESSION)
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)
307
compressed_chunks = map(compressor.compress, self._content_chunks)
304
308
compressed_chunks.append(compressor.flush())
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))
309
self._z_content = ''.join(compressed_chunks)
310
self._z_content_length = len(self._z_content)
309
312
def _create_z_content(self):
310
if self._z_content_chunks is not None:
313
if self._z_content is not None:
316
self._create_z_content_using_lzma()
312
318
if self._content_chunks is not None:
313
chunks = self._content_chunks
315
chunks = (self._content,)
316
self._create_z_content_from_chunks(chunks)
319
self._create_z_content_from_chunks()
321
self._z_content = zlib.compress(self._content)
322
self._z_content_length = len(self._z_content)
319
"""Create the byte stream as a series of 'chunks'"""
325
"""Encode the information into a byte stream."""
320
326
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),
328
header = self.GCB_LZ_HEADER
330
header = self.GCB_HEADER
332
'%d\n%d\n' % (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
330
"""Encode the information into a byte stream."""
331
total_len, chunks = self.to_chunks()
332
335
return ''.join(chunks)
334
337
def _dump(self, include_text=False):
448
451
# Grab and cache the raw bytes for this entry
449
452
# 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))
454
self._manager._prepare_for_extract()
455
455
block = self._manager._block
456
456
self._bytes = block.extract(self.key, self._start, self._end)
457
457
# There are code paths that first extract as fulltext, and then
476
476
_full_enough_block_size = 3*1024*1024 # size at which we won't repack
477
477
_full_enough_mixed_block_size = 2*768*1024 # 1.5MB
479
def __init__(self, block, get_compressor_settings=None):
479
def __init__(self, block):
480
480
self._block = block
481
481
# We need to preserve the ordering
482
482
self._factories = []
483
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
499
485
def add_factory(self, key, parents, start, end):
500
486
if not self._factories:
533
519
new_block.set_content(self._block._content[:last_byte])
534
520
self._block = new_block
536
def _make_group_compressor(self):
537
return GroupCompressor(self._get_compressor_settings())
539
522
def _rebuild_block(self):
540
523
"""Create a new GroupCompressBlock with only the referenced texts."""
541
compressor = self._make_group_compressor()
524
compressor = GroupCompressor()
542
525
tstart = time.time()
543
526
old_length = self._block._content_length
556
539
# block? It seems hard to come up with a method that it would
557
540
# expand, since we do full compression again. Perhaps based on a
558
541
# 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
542
delta = time.time() - tstart
565
543
self._block = new_block
566
544
trace.mutter('creating new compressed block on-the-fly in %.3fs'
717
695
z_header_bytes = zlib.compress(header_bytes)
719
697
z_header_bytes_len = len(z_header_bytes)
720
block_bytes_len, block_chunks = self._block.to_chunks()
698
block_bytes = self._block.to_bytes()
721
699
lines.append('%d\n%d\n%d\n' % (z_header_bytes_len, header_bytes_len,
723
701
lines.append(z_header_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
702
lines.append(block_bytes)
703
del z_header_bytes, block_bytes
728
704
return ''.join(lines)
731
707
def from_bytes(cls, bytes):
732
708
# TODO: This does extra string copying, probably better to do it a
733
# different way. At a minimum this creates 2 copies of the
735
710
(storage_kind, z_header_len, header_len,
736
711
block_len, rest) = bytes.split('\n', 4)
900
871
After calling this, the compressor should no longer be used
873
# TODO: this causes us to 'bloat' to 2x the size of content in the
874
# group. This has an impact for 'commit' of large objects.
875
# One possibility is to use self._content_chunks, and be lazy and
876
# only fill out self._content as a full string when we actually
877
# need it. That would at least drop the peak memory consumption
878
# for 'commit' down to ~1x the size of the largest file, at a
879
# cost of increased complexity within this code. 2x is still <<
880
# 3x the size of the largest file, so we are doing ok.
902
881
self._block.set_chunked_content(self.chunks, self.endpoint)
903
882
self.chunks = None
904
883
self._delta_index = None
923
902
class PythonGroupCompressor(_CommonGroupCompressor):
925
def __init__(self, settings=None):
926
905
"""Create a GroupCompressor.
928
907
Used only if the pyrex version is not available.
930
super(PythonGroupCompressor, self).__init__(settings)
909
super(PythonGroupCompressor, self).__init__()
931
910
self._delta_index = LinesDeltaIndex([])
932
911
# The actual content is managed by LinesDeltaIndex
933
912
self.chunks = self._delta_index.lines
971
950
It contains code very similar to SequenceMatcher because of having a similar
972
951
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
952
- there is no junk, we want a minimal edit not a human readable diff.
953
- we don't filter very common lines (because we don't know where a good
954
range will start, and after the first text we want to be emitting minmal
956
- we chain the left side, not the right side
957
- we incrementally update the adjacency matrix as new lines are provided.
958
- we look for matches in all of the left side, so the routine which does
959
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)
964
super(PyrexGroupCompressor, self).__init__()
965
self._delta_index = DeltaIndex()
990
967
def _compress(self, key, bytes, max_delta_size, soft=False):
991
968
"""see _CommonGroupCompressor._compress"""
1066
1043
index = _GCGraphIndex(graph_index, lambda:True, parents=parents,
1067
1044
add_callback=graph_index.add_nodes,
1068
1045
inconsistency_fatal=inconsistency_fatal)
1069
access = pack_repo._DirectPackAccess({})
1046
access = knit._DirectPackAccess({})
1070
1047
access.set_writer(writer, graph_index, (transport, 'newpack'))
1071
1048
result = GroupCompressVersionedFiles(index, access, delta)
1072
1049
result.stream = stream
1083
1060
class _BatchingBlockFetcher(object):
1084
1061
"""Fetch group compress blocks in batches.
1086
1063
:ivar total_bytes: int of expected number of bytes needed to fetch the
1087
1064
currently pending batch.
1090
def __init__(self, gcvf, locations, get_compressor_settings=None):
1067
def __init__(self, gcvf, locations):
1091
1068
self.gcvf = gcvf
1092
1069
self.locations = locations
1176
1152
memos_to_get_stack.pop()
1178
1154
block = self.batch_memos[read_memo]
1179
self.manager = _LazyGroupContentManager(block,
1180
get_compressor_settings=self._get_compressor_settings)
1155
self.manager = _LazyGroupContentManager(block)
1181
1156
self.last_read_memo = read_memo
1182
1157
start, end = index_memo[3:5]
1183
1158
self.manager.add_factory(key, parents, start, end)
1190
1165
self.total_bytes = 0
1193
class GroupCompressVersionedFiles(VersionedFilesWithFallbacks):
1168
class GroupCompressVersionedFiles(VersionedFiles):
1194
1169
"""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
def __init__(self, index, access, delta=True, _unadded_refs=None,
1171
def __init__(self, index, access, delta=True, _unadded_refs=None):
1210
1172
"""Create a GroupCompressVersionedFiles object.
1212
1174
:param index: The index object storing access and graph data.
1213
1175
:param access: The access object storing raw data.
1214
1176
:param delta: Whether to delta compress or just entropy compress.
1215
1177
:param _unadded_refs: private parameter, don't use.
1216
:param _group_cache: private parameter, don't use.
1218
1179
self._index = index
1219
1180
self._access = access
1221
1182
if _unadded_refs is None:
1222
1183
_unadded_refs = {}
1223
1184
self._unadded_refs = _unadded_refs
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
1185
self._group_cache = LRUSizeCache(max_size=50*1024*1024)
1186
self._fallback_vfs = []
1230
1188
def without_fallbacks(self):
1231
1189
"""Return a clone of this object without any fallbacks configured."""
1232
1190
return GroupCompressVersionedFiles(self._index, self._access,
1233
self._delta, _unadded_refs=dict(self._unadded_refs),
1234
_group_cache=self._group_cache)
1191
self._delta, _unadded_refs=dict(self._unadded_refs))
1236
1193
def add_lines(self, key, parents, lines, parent_texts=None,
1237
1194
left_matching_blocks=None, nostore_sha=None, random_id=False,
1241
1198
:param key: The key tuple of the text to add.
1242
1199
:param parents: The parents key tuples of the text to add.
1243
1200
: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
1201
of them except the last must be terminated with \n and contain no
1202
other \n's. The last line may either contain no \n's or a single
1203
terminating \n. If the lines list does meet this constraint the add
1204
routine may error or may succeed - but you will be unable to read
1205
the data back accurately. (Checking the lines have been split
1249
1206
correctly is expensive and extremely unlikely to catch bugs so it
1250
1207
is not done at runtime unless check_content is True.)
1251
1208
:param parent_texts: An optional dictionary containing the opaque
1326
1283
return self.get_record_stream(keys, 'unordered', True)
1328
def clear_cache(self):
1329
"""See VersionedFiles.clear_cache()"""
1330
self._group_cache.clear()
1331
self._index._graph_index.clear_cache()
1332
self._index._int_cache.clear()
1334
1285
def _check_add(self, key, lines, random_id, check_content):
1335
1286
"""check that version_id and lines are safe to add."""
1336
1287
version_id = key[-1]
1346
1297
self._check_lines_not_unicode(lines)
1347
1298
self._check_lines_are_lines(lines)
1300
def get_known_graph_ancestry(self, keys):
1301
"""Get a KnownGraph instance with the ancestry of keys."""
1302
# Note that this is identical to
1303
# KnitVersionedFiles.get_known_graph_ancestry, but they don't share
1305
parent_map, missing_keys = self._index.find_ancestry(keys)
1306
for fallback in self._fallback_vfs:
1307
if not missing_keys:
1309
(f_parent_map, f_missing_keys) = fallback._index.find_ancestry(
1311
parent_map.update(f_parent_map)
1312
missing_keys = f_missing_keys
1313
kg = _mod_graph.KnownGraph(parent_map)
1349
1316
def get_parent_map(self, keys):
1350
1317
"""Get a map of the graph parents of keys.
1490
1457
The returned objects should be in the order defined by 'ordering',
1491
1458
which can weave between different sources.
1493
1459
:param ordering: Must be one of 'topological' or 'groupcompress'
1494
1460
:return: List of [(source, [keys])] tuples, such that all keys are in
1495
1461
the defined order, regardless of source.
1497
1463
if ordering == 'topological':
1498
present_keys = tsort.topo_sort(parent_map)
1464
present_keys = topo_sort(parent_map)
1500
1466
# ordering == 'groupcompress'
1501
1467
# XXX: This only optimizes for the target ordering. We may need
1590
1556
# - we encounter an unadded ref, or
1591
1557
# - we run out of keys, or
1592
1558
# - the total bytes to retrieve for this batch > BATCH_SIZE
1593
batcher = _BatchingBlockFetcher(self, locations,
1594
get_compressor_settings=self._get_compressor_settings)
1559
batcher = _BatchingBlockFetcher(self, locations)
1595
1560
for source, keys in source_keys:
1596
1561
if source is self:
1597
1562
for key in keys:
1643
1608
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
1611
def _insert_record_stream(self, stream, random_id=False, nostore_sha=None,
1671
1612
reuse_blocks=True):
1672
1613
"""Internal core to insert a record stream into this container.
1696
1637
# This will go up to fulltexts for gc to gc fetching, which isn't
1698
self._compressor = self._make_group_compressor()
1639
self._compressor = GroupCompressor()
1699
1640
self._unadded_refs = {}
1700
1641
keys_to_add = []
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)
1643
bytes = self._compressor.flush().to_bytes()
1715
1644
index, start, length = self._access.add_raw_records(
1716
1645
[(None, len(bytes))], bytes)[0]
1736
1666
raise errors.RevisionNotPresent(record.key, self)
1738
1668
if record.key in inserted_keys:
1739
trace.note(gettext('Insert claimed random_id=True,'
1740
' but then inserted %r two times'), record.key)
1669
trace.note('Insert claimed random_id=True,'
1670
' but then inserted %r two times', record.key)
1742
1672
inserted_keys.add(record.key)
1743
1673
if reuse_blocks:
1827
1757
key = record.key
1828
1758
self._unadded_refs[key] = record.parents
1829
1759
yield found_sha1
1830
as_st = static_tuple.StaticTuple.from_sequence
1831
if record.parents is not None:
1832
parents = as_st([as_st(p) for p in record.parents])
1835
refs = static_tuple.StaticTuple(parents)
1836
keys_to_add.append((key, '%d %d' % (start_point, end_point), refs))
1760
keys_to_add.append((key, '%d %d' % (start_point, end_point),
1837
1762
if len(keys_to_add):
1839
1764
self._compressor = None
1882
1807
"""See VersionedFiles.keys."""
1883
1808
if 'evil' in debug.debug_flags:
1884
1809
trace.mutter_callsite(2, "keys scales with size of history")
1885
sources = [self._index] + self._immediate_fallback_vfs
1810
sources = [self._index] + self._fallback_vfs
1887
1812
for source in sources:
1888
1813
result.update(source.keys())
1892
class _GCBuildDetails(object):
1893
"""A blob of data about the build details.
1895
This stores the minimal data, which then allows compatibility with the old
1896
api, without taking as much memory.
1899
__slots__ = ('_index', '_group_start', '_group_end', '_basis_end',
1900
'_delta_end', '_parents')
1903
compression_parent = None
1905
def __init__(self, parents, position_info):
1906
self._parents = parents
1907
(self._index, self._group_start, self._group_end, self._basis_end,
1908
self._delta_end) = position_info
1911
return '%s(%s, %s)' % (self.__class__.__name__,
1912
self.index_memo, self._parents)
1915
def index_memo(self):
1916
return (self._index, self._group_start, self._group_end,
1917
self._basis_end, self._delta_end)
1920
def record_details(self):
1921
return static_tuple.StaticTuple(self.method, None)
1923
def __getitem__(self, offset):
1924
"""Compatibility thunk to act like a tuple."""
1926
return self.index_memo
1928
return self.compression_parent # Always None
1930
return self._parents
1932
return self.record_details
1934
raise IndexError('offset out of range')
1940
1817
class _GCGraphIndex(object):
1941
1818
"""Mapper from GroupCompressVersionedFiles needs into GraphIndex storage."""
1967
1844
self.has_graph = parents
1968
1845
self._is_locked = is_locked
1969
1846
self._inconsistency_fatal = inconsistency_fatal
1970
# GroupCompress records tend to have the same 'group' start + offset
1971
# repeated over and over, this creates a surplus of ints
1972
self._int_cache = {}
1973
1847
if track_external_parent_refs:
1974
self._key_dependencies = _KeyRefs(
1848
self._key_dependencies = knit._KeyRefs(
1975
1849
track_new_keys=track_new_keys)
1977
1851
self._key_dependencies = None
2011
1885
if not random_id:
2012
1886
present_nodes = self._get_entries(keys)
2013
1887
for (index, key, value, node_refs) in present_nodes:
2014
# Sometimes these are passed as a list rather than a tuple
2015
node_refs = static_tuple.as_tuples(node_refs)
2016
passed = static_tuple.as_tuples(keys[key])
2017
if node_refs != passed[1]:
2018
details = '%s %s %s' % (key, (value, node_refs), passed)
1888
if node_refs != keys[key][1]:
1889
details = '%s %s %s' % (key, (value, node_refs), keys[key])
2019
1890
if self._inconsistency_fatal:
2020
1891
raise errors.KnitCorrupt(self, "inconsistent details"
2021
1892
" in add_records: %s" %
2117
1988
:param keys: An iterable of keys.
2118
1989
:return: A dict of key:
2119
1990
(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
1992
opaque structure to pass to read_records to extract the raw
1995
Content that this record is built upon, may be None
1997
Logical parents of this node
1999
extra information about the content which needs to be passed to
2000
Factory.parse_record
2129
2002
self._check_read()
2151
2025
"""Convert an index value to position details."""
2152
2026
bits = node[2].split(' ')
2153
2027
# It would be nice not to read the entire gzip.
2154
# start and stop are put into _int_cache because they are very common.
2155
# They define the 'group' that an entry is in, and many groups can have
2156
# thousands of objects.
2157
# Branching Launchpad, for example, saves ~600k integers, at 12 bytes
2158
# each, or about 7MB. Note that it might be even more when you consider
2159
# how PyInt is allocated in separate slabs. And you can't return a slab
2160
# to the OS if even 1 int on it is in use. Note though that Python uses
2161
# a LIFO when re-using PyInt slots, which might cause more
2163
2028
start = int(bits[0])
2164
start = self._int_cache.setdefault(start, start)
2165
2029
stop = int(bits[1])
2166
stop = self._int_cache.setdefault(stop, stop)
2167
2030
basis_end = int(bits[2])
2168
2031
delta_end = int(bits[3])
2169
# We can't use StaticTuple here, because node[0] is a BTreeGraphIndex
2171
return (node[0], start, stop, basis_end, delta_end)
2032
return node[0], start, stop, basis_end, delta_end
2173
2034
def scan_unvalidated_index(self, graph_index):
2174
2035
"""Inform this _GCGraphIndex that there is an unvalidated index.
added
removed
bzrlib/groupcompress.py
