1
# Copyright (C) 2008-2011 Canonical Ltd
3
# This program is free software; you can redistribute it and/or modify
4
# it under the terms of the GNU General Public License as published by
5
# the Free Software Foundation; either version 2 of the License, or
6
# (at your option) any later version.
8
# This program is distributed in the hope that it will be useful,
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11
# GNU General Public License for more details.
13
# You should have received a copy of the GNU General Public License
14
# along with this program; if not, write to the Free Software
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
17
"""Core compression logic for compressing streams of related files."""
22
from ..lazy_import import lazy_import
23
lazy_import(globals(), """
33
from breezy.bzr import (
39
from breezy.i18n import gettext
45
from .btree_index import BTreeBuilder
46
from ..lru_cache import LRUSizeCache
47
from .versionedfile import (
51
ChunkedContentFactory,
52
FulltextContentFactory,
53
VersionedFilesWithFallbacks,
56
# Minimum number of uncompressed bytes to try fetch at once when retrieving
57
# groupcompress blocks.
60
# osutils.sha_string(b'')
61
_null_sha1 = b'da39a3ee5e6b4b0d3255bfef95601890afd80709'
64
def sort_gc_optimal(parent_map):
65
"""Sort and group the keys in parent_map into groupcompress order.
67
groupcompress is defined (currently) as reverse-topological order, grouped
70
:return: A sorted-list of keys
72
# groupcompress ordering is approximately reverse topological,
73
# properly grouped by file-id.
75
for key, value in parent_map.items():
76
if isinstance(key, bytes) or len(key) == 1:
81
per_prefix_map[prefix][key] = value
83
per_prefix_map[prefix] = {key: value}
86
for prefix in sorted(per_prefix_map):
87
present_keys.extend(reversed(tsort.topo_sort(per_prefix_map[prefix])))
91
class DecompressCorruption(errors.BzrError):
93
_fmt = "Corruption while decompressing repository file%(orig_error)s"
95
def __init__(self, orig_error=None):
96
if orig_error is not None:
97
self.orig_error = ", %s" % (orig_error,)
100
errors.BzrError.__init__(self)
103
# The max zlib window size is 32kB, so if we set 'max_size' output of the
104
# decompressor to the requested bytes + 32kB, then we should guarantee
105
# num_bytes coming out.
106
_ZLIB_DECOMP_WINDOW = 32 * 1024
109
class GroupCompressBlock(object):
110
"""An object which maintains the internal structure of the compressed data.
112
This tracks the meta info (start of text, length, type, etc.)
115
# Group Compress Block v1 Zlib
116
GCB_HEADER = b'gcb1z\n'
117
# Group Compress Block v1 Lzma
118
GCB_LZ_HEADER = b'gcb1l\n'
119
GCB_KNOWN_HEADERS = (GCB_HEADER, GCB_LZ_HEADER)
122
# map by key? or just order in file?
123
self._compressor_name = None
124
self._z_content_chunks = None
125
self._z_content_decompressor = None
126
self._z_content_length = None
127
self._content_length = None
129
self._content_chunks = None
132
# This is the maximum number of bytes this object will reference if
133
# everything is decompressed. However, if we decompress less than
134
# everything... (this would cause some problems for LRUSizeCache)
135
return self._content_length + self._z_content_length
137
def _ensure_content(self, num_bytes=None):
138
"""Make sure that content has been expanded enough.
140
:param num_bytes: Ensure that we have extracted at least num_bytes of
141
content. If None, consume everything
143
if self._content_length is None:
144
raise AssertionError('self._content_length should never be None')
145
if num_bytes is None:
146
num_bytes = self._content_length
147
elif (self._content_length is not None
148
and num_bytes > self._content_length):
149
raise AssertionError(
150
'requested num_bytes (%d) > content length (%d)'
151
% (num_bytes, self._content_length))
152
# Expand the content if required
153
if self._content is None:
154
if self._content_chunks is not None:
155
self._content = b''.join(self._content_chunks)
156
self._content_chunks = None
157
if self._content is None:
158
# We join self._z_content_chunks here, because if we are
159
# decompressing, then it is *very* likely that we have a single
161
if self._z_content_chunks is None:
162
raise AssertionError('No content to decompress')
163
z_content = b''.join(self._z_content_chunks)
166
elif self._compressor_name == 'lzma':
167
# We don't do partial lzma decomp yet
169
self._content = pylzma.decompress(z_content)
170
elif self._compressor_name == 'zlib':
171
# Start a zlib decompressor
172
if num_bytes * 4 > self._content_length * 3:
173
# If we are requesting more that 3/4ths of the content,
174
# just extract the whole thing in a single pass
175
num_bytes = self._content_length
176
self._content = zlib.decompress(z_content)
178
self._z_content_decompressor = zlib.decompressobj()
179
# Seed the decompressor with the uncompressed bytes, so
180
# that the rest of the code is simplified
181
self._content = self._z_content_decompressor.decompress(
182
z_content, num_bytes + _ZLIB_DECOMP_WINDOW)
183
if not self._z_content_decompressor.unconsumed_tail:
184
self._z_content_decompressor = None
186
raise AssertionError('Unknown compressor: %r'
187
% self._compressor_name)
188
# Any bytes remaining to be decompressed will be in the decompressors
191
# Do we have enough bytes already?
192
if len(self._content) >= num_bytes:
194
# If we got this far, and don't have a decompressor, something is wrong
195
if self._z_content_decompressor is None:
196
raise AssertionError(
197
'No decompressor to decompress %d bytes' % num_bytes)
198
remaining_decomp = self._z_content_decompressor.unconsumed_tail
199
if not remaining_decomp:
200
raise AssertionError('Nothing left to decompress')
201
needed_bytes = num_bytes - len(self._content)
202
# We always set max_size to 32kB over the minimum needed, so that
203
# zlib will give us as much as we really want.
204
# TODO: If this isn't good enough, we could make a loop here,
205
# that keeps expanding the request until we get enough
206
self._content += self._z_content_decompressor.decompress(
207
remaining_decomp, needed_bytes + _ZLIB_DECOMP_WINDOW)
208
if len(self._content) < num_bytes:
209
raise AssertionError('%d bytes wanted, only %d available'
210
% (num_bytes, len(self._content)))
211
if not self._z_content_decompressor.unconsumed_tail:
212
# The stream is finished
213
self._z_content_decompressor = None
215
def _parse_bytes(self, data, pos):
216
"""Read the various lengths from the header.
218
This also populates the various 'compressed' buffers.
220
:return: The position in bytes just after the last newline
222
# At present, we have 2 integers for the compressed and uncompressed
223
# content. In base10 (ascii) 14 bytes can represent > 1TB, so to avoid
224
# checking too far, cap the search to 14 bytes.
225
pos2 = data.index(b'\n', pos, pos + 14)
226
self._z_content_length = int(data[pos:pos2])
228
pos2 = data.index(b'\n', pos, pos + 14)
229
self._content_length = int(data[pos:pos2])
231
if len(data) != (pos + self._z_content_length):
232
# XXX: Define some GCCorrupt error ?
233
raise AssertionError('Invalid bytes: (%d) != %d + %d' %
234
(len(data), pos, self._z_content_length))
235
self._z_content_chunks = (data[pos:],)
238
def _z_content(self):
239
"""Return z_content_chunks as a simple string.
241
Meant only to be used by the test suite.
243
if self._z_content_chunks is not None:
244
return b''.join(self._z_content_chunks)
248
def from_bytes(cls, bytes):
251
if header not in cls.GCB_KNOWN_HEADERS:
252
raise ValueError('bytes did not start with any of %r'
253
% (cls.GCB_KNOWN_HEADERS,))
254
if header == cls.GCB_HEADER:
255
out._compressor_name = 'zlib'
256
elif header == cls.GCB_LZ_HEADER:
257
out._compressor_name = 'lzma'
259
raise ValueError('unknown compressor: %r' % (header,))
260
out._parse_bytes(bytes, 6)
263
def extract(self, key, start, end, sha1=None):
264
"""Extract the text for a specific key.
266
:param key: The label used for this content
267
:param sha1: TODO (should we validate only when sha1 is supplied?)
268
:return: The bytes for the content
270
if start == end == 0:
272
self._ensure_content(end)
273
# The bytes are 'f' or 'd' for the type, then a variable-length
274
# base128 integer for the content size, then the actual content
275
# We know that the variable-length integer won't be longer than 5
276
# bytes (it takes 5 bytes to encode 2^32)
277
c = self._content[start:start + 1]
282
raise ValueError('Unknown content control code: %s'
285
content_len, len_len = decode_base128_int(
286
self._content[start + 1:start + 6])
287
content_start = start + 1 + len_len
288
if end != content_start + content_len:
289
raise ValueError('end != len according to field header'
290
' %s != %s' % (end, content_start + content_len))
292
return [self._content[content_start:end]]
293
# Must be type delta as checked above
294
return [apply_delta_to_source(self._content, content_start, end)]
296
def set_chunked_content(self, content_chunks, length):
297
"""Set the content of this block to the given chunks."""
298
# If we have lots of short lines, it is may be more efficient to join
299
# the content ahead of time. If the content is <10MiB, we don't really
300
# care about the extra memory consumption, so we can just pack it and
301
# be done. However, timing showed 18s => 17.9s for repacking 1k revs of
302
# mysql, which is below the noise margin
303
self._content_length = length
304
self._content_chunks = content_chunks
306
self._z_content_chunks = None
308
def set_content(self, content):
309
"""Set the content of this block."""
310
self._content_length = len(content)
311
self._content = content
312
self._z_content_chunks = None
314
def _create_z_content_from_chunks(self, chunks):
315
compressor = zlib.compressobj(zlib.Z_DEFAULT_COMPRESSION)
316
# Peak in this point is 1 fulltext, 1 compressed text, + zlib overhead
317
# (measured peak is maybe 30MB over the above...)
318
compressed_chunks = list(map(compressor.compress, chunks))
319
compressed_chunks.append(compressor.flush())
320
# Ignore empty chunks
321
self._z_content_chunks = [c for c in compressed_chunks if c]
322
self._z_content_length = sum(map(len, self._z_content_chunks))
324
def _create_z_content(self):
325
if self._z_content_chunks is not None:
327
if self._content_chunks is not None:
328
chunks = self._content_chunks
330
chunks = (self._content,)
331
self._create_z_content_from_chunks(chunks)
334
"""Create the byte stream as a series of 'chunks'"""
335
self._create_z_content()
336
header = self.GCB_HEADER
337
chunks = [b'%s%d\n%d\n'
338
% (header, self._z_content_length, self._content_length),
340
chunks.extend(self._z_content_chunks)
341
total_len = sum(map(len, chunks))
342
return total_len, chunks
345
"""Encode the information into a byte stream."""
346
total_len, chunks = self.to_chunks()
347
return b''.join(chunks)
349
def _dump(self, include_text=False):
350
"""Take this block, and spit out a human-readable structure.
352
:param include_text: Inserts also include text bits, chose whether you
353
want this displayed in the dump or not.
354
:return: A dump of the given block. The layout is something like:
355
[('f', length), ('d', delta_length, text_length, [delta_info])]
356
delta_info := [('i', num_bytes, text), ('c', offset, num_bytes),
359
self._ensure_content()
362
while pos < self._content_length:
363
kind = self._content[pos:pos + 1]
365
if kind not in (b'f', b'd'):
366
raise ValueError('invalid kind character: %r' % (kind,))
367
content_len, len_len = decode_base128_int(
368
self._content[pos:pos + 5])
370
if content_len + pos > self._content_length:
371
raise ValueError('invalid content_len %d for record @ pos %d'
372
% (content_len, pos - len_len - 1))
373
if kind == b'f': # Fulltext
375
text = self._content[pos:pos + content_len]
376
result.append((b'f', content_len, text))
378
result.append((b'f', content_len))
379
elif kind == b'd': # Delta
380
delta_content = self._content[pos:pos + content_len]
382
# The first entry in a delta is the decompressed length
383
decomp_len, delta_pos = decode_base128_int(delta_content)
384
result.append((b'd', content_len, decomp_len, delta_info))
386
while delta_pos < content_len:
387
c = delta_content[delta_pos]
391
delta_pos) = decode_copy_instruction(delta_content, c,
394
text = self._content[offset:offset + length]
395
delta_info.append((b'c', offset, length, text))
397
delta_info.append((b'c', offset, length))
398
measured_len += length
401
txt = delta_content[delta_pos:delta_pos + c]
404
delta_info.append((b'i', c, txt))
407
if delta_pos != content_len:
408
raise ValueError('Delta consumed a bad number of bytes:'
409
' %d != %d' % (delta_pos, content_len))
410
if measured_len != decomp_len:
411
raise ValueError('Delta claimed fulltext was %d bytes, but'
412
' extraction resulted in %d bytes'
413
% (decomp_len, measured_len))
418
class _LazyGroupCompressFactory(object):
419
"""Yield content from a GroupCompressBlock on demand."""
421
def __init__(self, key, parents, manager, start, end, first):
422
"""Create a _LazyGroupCompressFactory
424
:param key: The key of just this record
425
:param parents: The parents of this key (possibly None)
426
:param gc_block: A GroupCompressBlock object
427
:param start: Offset of the first byte for this record in the
429
:param end: Offset of the byte just after the end of this record
430
(ie, bytes = content[start:end])
431
:param first: Is this the first Factory for the given block?
434
self.parents = parents
437
# Note: This attribute coupled with Manager._factories creates a
438
# reference cycle. Perhaps we would rather use a weakref(), or
439
# find an appropriate time to release the ref. After the first
440
# get_bytes_as call? After Manager.get_record_stream() returns
442
self._manager = manager
444
self.storage_kind = 'groupcompress-block'
446
self.storage_kind = 'groupcompress-block-ref'
452
return '%s(%s, first=%s)' % (self.__class__.__name__,
453
self.key, self._first)
455
def _extract_bytes(self):
456
# Grab and cache the raw bytes for this entry
457
# and break the ref-cycle with _manager since we don't need it
460
self._manager._prepare_for_extract()
461
except zlib.error as value:
462
raise DecompressCorruption("zlib: " + str(value))
463
block = self._manager._block
464
self._chunks = block.extract(self.key, self._start, self._end)
465
# There are code paths that first extract as fulltext, and then
466
# extract as storage_kind (smart fetch). So we don't break the
467
# refcycle here, but instead in manager.get_record_stream()
469
def get_bytes_as(self, storage_kind):
470
if storage_kind == self.storage_kind:
472
# wire bytes, something...
473
return self._manager._wire_bytes()
476
if storage_kind in ('fulltext', 'chunked', 'lines'):
477
if self._chunks is None:
478
self._extract_bytes()
479
if storage_kind == 'fulltext':
480
return b''.join(self._chunks)
481
elif storage_kind == 'chunked':
484
return osutils.chunks_to_lines(self._chunks)
485
raise errors.UnavailableRepresentation(self.key, storage_kind,
488
def iter_bytes_as(self, storage_kind):
489
if self._chunks is None:
490
self._extract_bytes()
491
if storage_kind == 'chunked':
492
return iter(self._chunks)
493
elif storage_kind == 'lines':
494
return iter(osutils.chunks_to_lines(self._chunks))
495
raise errors.UnavailableRepresentation(self.key, storage_kind,
499
class _LazyGroupContentManager(object):
500
"""This manages a group of _LazyGroupCompressFactory objects."""
502
_max_cut_fraction = 0.75 # We allow a block to be trimmed to 75% of
503
# current size, and still be considered
505
_full_block_size = 4 * 1024 * 1024
506
_full_mixed_block_size = 2 * 1024 * 1024
507
_full_enough_block_size = 3 * 1024 * 1024 # size at which we won't repack
508
_full_enough_mixed_block_size = 2 * 768 * 1024 # 1.5MB
510
def __init__(self, block, get_compressor_settings=None):
512
# We need to preserve the ordering
515
self._get_settings = get_compressor_settings
516
self._compressor_settings = None
518
def _get_compressor_settings(self):
519
if self._compressor_settings is not None:
520
return self._compressor_settings
522
if self._get_settings is not None:
523
settings = self._get_settings()
525
vf = GroupCompressVersionedFiles
526
settings = vf._DEFAULT_COMPRESSOR_SETTINGS
527
self._compressor_settings = settings
528
return self._compressor_settings
530
def add_factory(self, key, parents, start, end):
531
if not self._factories:
535
# Note that this creates a reference cycle....
536
factory = _LazyGroupCompressFactory(key, parents, self,
537
start, end, first=first)
538
# max() works here, but as a function call, doing a compare seems to be
539
# significantly faster, timeit says 250ms for max() and 100ms for the
541
if end > self._last_byte:
542
self._last_byte = end
543
self._factories.append(factory)
545
def get_record_stream(self):
546
"""Get a record for all keys added so far."""
547
for factory in self._factories:
549
# Break the ref-cycle
550
factory._bytes = None
551
factory._manager = None
552
# TODO: Consider setting self._factories = None after the above loop,
553
# as it will break the reference cycle
555
def _trim_block(self, last_byte):
556
"""Create a new GroupCompressBlock, with just some of the content."""
557
# None of the factories need to be adjusted, because the content is
558
# located in an identical place. Just that some of the unreferenced
559
# trailing bytes are stripped
560
trace.mutter('stripping trailing bytes from groupcompress block'
561
' %d => %d', self._block._content_length, last_byte)
562
new_block = GroupCompressBlock()
563
self._block._ensure_content(last_byte)
564
new_block.set_content(self._block._content[:last_byte])
565
self._block = new_block
567
def _make_group_compressor(self):
568
return GroupCompressor(self._get_compressor_settings())
570
def _rebuild_block(self):
571
"""Create a new GroupCompressBlock with only the referenced texts."""
572
compressor = self._make_group_compressor()
574
old_length = self._block._content_length
576
for factory in self._factories:
577
chunks = factory.get_bytes_as('chunked')
578
chunks_len = factory.size
579
if chunks_len is None:
580
chunks_len = sum(map(len, chunks))
581
(found_sha1, start_point, end_point,
582
type) = compressor.compress(
583
factory.key, chunks, chunks_len, factory.sha1)
584
# Now update this factory with the new offsets, etc
585
factory.sha1 = found_sha1
586
factory._start = start_point
587
factory._end = end_point
588
self._last_byte = end_point
589
new_block = compressor.flush()
590
# TODO: Should we check that new_block really *is* smaller than the old
591
# block? It seems hard to come up with a method that it would
592
# expand, since we do full compression again. Perhaps based on a
593
# request that ends up poorly ordered?
594
# TODO: If the content would have expanded, then we would want to
595
# handle a case where we need to split the block.
596
# Now that we have a user-tweakable option
597
# (max_bytes_to_index), it is possible that one person set it
598
# to a very low value, causing poor compression.
599
delta = time.time() - tstart
600
self._block = new_block
601
trace.mutter('creating new compressed block on-the-fly in %.3fs'
602
' %d bytes => %d bytes', delta, old_length,
603
self._block._content_length)
605
def _prepare_for_extract(self):
606
"""A _LazyGroupCompressFactory is about to extract to fulltext."""
607
# We expect that if one child is going to fulltext, all will be. This
608
# helps prevent all of them from extracting a small amount at a time.
609
# Which in itself isn't terribly expensive, but resizing 2MB 32kB at a
610
# time (self._block._content) is a little expensive.
611
self._block._ensure_content(self._last_byte)
613
def _check_rebuild_action(self):
614
"""Check to see if our block should be repacked."""
617
for factory in self._factories:
618
total_bytes_used += factory._end - factory._start
619
if last_byte_used < factory._end:
620
last_byte_used = factory._end
621
# If we are using more than half of the bytes from the block, we have
622
# nothing else to check
623
if total_bytes_used * 2 >= self._block._content_length:
624
return None, last_byte_used, total_bytes_used
625
# We are using less than 50% of the content. Is the content we are
626
# using at the beginning of the block? If so, we can just trim the
627
# tail, rather than rebuilding from scratch.
628
if total_bytes_used * 2 > last_byte_used:
629
return 'trim', last_byte_used, total_bytes_used
631
# We are using a small amount of the data, and it isn't just packed
632
# nicely at the front, so rebuild the content.
633
# Note: This would be *nicer* as a strip-data-from-group, rather than
634
# building it up again from scratch
635
# It might be reasonable to consider the fulltext sizes for
636
# different bits when deciding this, too. As you may have a small
637
# fulltext, and a trivial delta, and you are just trading around
638
# for another fulltext. If we do a simple 'prune' you may end up
639
# expanding many deltas into fulltexts, as well.
640
# If we build a cheap enough 'strip', then we could try a strip,
641
# if that expands the content, we then rebuild.
642
return 'rebuild', last_byte_used, total_bytes_used
644
def check_is_well_utilized(self):
645
"""Is the current block considered 'well utilized'?
647
This heuristic asks if the current block considers itself to be a fully
648
developed group, rather than just a loose collection of data.
650
if len(self._factories) == 1:
651
# A block of length 1 could be improved by combining with other
652
# groups - don't look deeper. Even larger than max size groups
653
# could compress well with adjacent versions of the same thing.
655
action, last_byte_used, total_bytes_used = self._check_rebuild_action()
656
block_size = self._block._content_length
657
if total_bytes_used < block_size * self._max_cut_fraction:
658
# This block wants to trim itself small enough that we want to
659
# consider it under-utilized.
661
# TODO: This code is meant to be the twin of _insert_record_stream's
662
# 'start_new_block' logic. It would probably be better to factor
663
# out that logic into a shared location, so that it stays
665
# We currently assume a block is properly utilized whenever it is >75%
666
# of the size of a 'full' block. In normal operation, a block is
667
# considered full when it hits 4MB of same-file content. So any block
668
# >3MB is 'full enough'.
669
# The only time this isn't true is when a given block has large-object
670
# content. (a single file >4MB, etc.)
671
# Under these circumstances, we allow a block to grow to
672
# 2 x largest_content. Which means that if a given block had a large
673
# object, it may actually be under-utilized. However, given that this
674
# is 'pack-on-the-fly' it is probably reasonable to not repack large
675
# content blobs on-the-fly. Note that because we return False for all
676
# 1-item blobs, we will repack them; we may wish to reevaluate our
677
# treatment of large object blobs in the future.
678
if block_size >= self._full_enough_block_size:
680
# If a block is <3MB, it still may be considered 'full' if it contains
681
# mixed content. The current rule is 2MB of mixed content is considered
682
# full. So check to see if this block contains mixed content, and
683
# set the threshold appropriately.
685
for factory in self._factories:
686
prefix = factory.key[:-1]
687
if common_prefix is None:
688
common_prefix = prefix
689
elif prefix != common_prefix:
690
# Mixed content, check the size appropriately
691
if block_size >= self._full_enough_mixed_block_size:
694
# The content failed both the mixed check and the single-content check
695
# so obviously it is not fully utilized
696
# TODO: there is one other constraint that isn't being checked
697
# namely, that the entries in the block are in the appropriate
698
# order. For example, you could insert the entries in exactly
699
# reverse groupcompress order, and we would think that is ok.
700
# (all the right objects are in one group, and it is fully
701
# utilized, etc.) For now, we assume that case is rare,
702
# especially since we should always fetch in 'groupcompress'
706
def _check_rebuild_block(self):
707
action, last_byte_used, total_bytes_used = self._check_rebuild_action()
711
self._trim_block(last_byte_used)
712
elif action == 'rebuild':
713
self._rebuild_block()
715
raise ValueError('unknown rebuild action: %r' % (action,))
717
def _wire_bytes(self):
718
"""Return a byte stream suitable for transmitting over the wire."""
719
self._check_rebuild_block()
720
# The outer block starts with:
721
# 'groupcompress-block\n'
722
# <length of compressed key info>\n
723
# <length of uncompressed info>\n
724
# <length of gc block>\n
727
lines = [b'groupcompress-block\n']
728
# The minimal info we need is the key, the start offset, and the
729
# parents. The length and type are encoded in the record itself.
730
# However, passing in the other bits makes it easier. The list of
731
# keys, and the start offset, the length
733
# 1 line with parents, '' for ()
734
# 1 line for start offset
735
# 1 line for end byte
737
for factory in self._factories:
738
key_bytes = b'\x00'.join(factory.key)
739
parents = factory.parents
741
parent_bytes = b'None:'
743
parent_bytes = b'\t'.join(b'\x00'.join(key) for key in parents)
744
record_header = b'%s\n%s\n%d\n%d\n' % (
745
key_bytes, parent_bytes, factory._start, factory._end)
746
header_lines.append(record_header)
747
# TODO: Can we break the refcycle at this point and set
748
# factory._manager = None?
749
header_bytes = b''.join(header_lines)
751
header_bytes_len = len(header_bytes)
752
z_header_bytes = zlib.compress(header_bytes)
754
z_header_bytes_len = len(z_header_bytes)
755
block_bytes_len, block_chunks = self._block.to_chunks()
756
lines.append(b'%d\n%d\n%d\n' % (
757
z_header_bytes_len, header_bytes_len, block_bytes_len))
758
lines.append(z_header_bytes)
759
lines.extend(block_chunks)
760
del z_header_bytes, block_chunks
761
# TODO: This is a point where we will double the memory consumption. To
762
# avoid this, we probably have to switch to a 'chunked' api
763
return b''.join(lines)
766
def from_bytes(cls, bytes):
767
# TODO: This does extra string copying, probably better to do it a
768
# different way. At a minimum this creates 2 copies of the
770
(storage_kind, z_header_len, header_len,
771
block_len, rest) = bytes.split(b'\n', 4)
773
if storage_kind != b'groupcompress-block':
774
raise ValueError('Unknown storage kind: %s' % (storage_kind,))
775
z_header_len = int(z_header_len)
776
if len(rest) < z_header_len:
777
raise ValueError('Compressed header len shorter than all bytes')
778
z_header = rest[:z_header_len]
779
header_len = int(header_len)
780
header = zlib.decompress(z_header)
781
if len(header) != header_len:
782
raise ValueError('invalid length for decompressed bytes')
784
block_len = int(block_len)
785
if len(rest) != z_header_len + block_len:
786
raise ValueError('Invalid length for block')
787
block_bytes = rest[z_header_len:]
789
# So now we have a valid GCB, we just need to parse the factories that
791
header_lines = header.split(b'\n')
793
last = header_lines.pop()
795
raise ValueError('header lines did not end with a trailing'
797
if len(header_lines) % 4 != 0:
798
raise ValueError('The header was not an even multiple of 4 lines')
799
block = GroupCompressBlock.from_bytes(block_bytes)
802
for start in range(0, len(header_lines), 4):
804
key = tuple(header_lines[start].split(b'\x00'))
805
parents_line = header_lines[start + 1]
806
if parents_line == b'None:':
809
parents = tuple([tuple(segment.split(b'\x00'))
810
for segment in parents_line.split(b'\t')
812
start_offset = int(header_lines[start + 2])
813
end_offset = int(header_lines[start + 3])
814
result.add_factory(key, parents, start_offset, end_offset)
818
def network_block_to_records(storage_kind, bytes, line_end):
819
if storage_kind != 'groupcompress-block':
820
raise ValueError('Unknown storage kind: %s' % (storage_kind,))
821
manager = _LazyGroupContentManager.from_bytes(bytes)
822
return manager.get_record_stream()
825
class _CommonGroupCompressor(object):
827
def __init__(self, settings=None):
828
"""Create a GroupCompressor."""
833
self.labels_deltas = {}
834
self._delta_index = None # Set by the children
835
self._block = GroupCompressBlock()
839
self._settings = settings
841
def compress(self, key, chunks, length, expected_sha, nostore_sha=None,
843
"""Compress lines with label key.
845
:param key: A key tuple. It is stored in the output
846
for identification of the text during decompression. If the last
847
element is b'None' it is replaced with the sha1 of the text -
849
:param chunks: Chunks of bytes to be compressed
850
:param length: Length of chunks
851
:param expected_sha: If non-None, the sha the lines are believed to
852
have. During compression the sha is calculated; a mismatch will
854
:param nostore_sha: If the computed sha1 sum matches, we will raise
855
ExistingContent rather than adding the text.
856
:param soft: Do a 'soft' compression. This means that we require larger
857
ranges to match to be considered for a copy command.
859
:return: The sha1 of lines, the start and end offsets in the delta, and
860
the type ('fulltext' or 'delta').
862
:seealso VersionedFiles.add_lines:
864
if length == 0: # empty, like a dir entry, etc
865
if nostore_sha == _null_sha1:
866
raise errors.ExistingContent()
867
return _null_sha1, 0, 0, 'fulltext'
868
# we assume someone knew what they were doing when they passed it in
869
if expected_sha is not None:
872
sha1 = osutils.sha_strings(chunks)
873
if nostore_sha is not None:
874
if sha1 == nostore_sha:
875
raise errors.ExistingContent()
877
key = key[:-1] + (b'sha1:' + sha1,)
879
start, end, type = self._compress(key, chunks, length, length / 2, soft)
880
return sha1, start, end, type
882
def _compress(self, key, chunks, input_len, max_delta_size, soft=False):
883
"""Compress lines with label key.
885
:param key: A key tuple. It is stored in the output for identification
886
of the text during decompression.
888
:param chunks: The chunks of bytes to be compressed
890
:param input_len: The length of the chunks
892
:param max_delta_size: The size above which we issue a fulltext instead
895
:param soft: Do a 'soft' compression. This means that we require larger
896
ranges to match to be considered for a copy command.
898
:return: The sha1 of lines, the start and end offsets in the delta, and
899
the type ('fulltext' or 'delta').
901
raise NotImplementedError(self._compress)
903
def extract(self, key):
904
"""Extract a key previously added to the compressor.
906
:param key: The key to extract.
907
:return: An iterable over chunks and the sha1.
909
(start_byte, start_chunk, end_byte,
910
end_chunk) = self.labels_deltas[key]
911
delta_chunks = self.chunks[start_chunk:end_chunk]
912
stored_bytes = b''.join(delta_chunks)
913
kind = stored_bytes[:1]
915
fulltext_len, offset = decode_base128_int(stored_bytes[1:10])
916
data_len = fulltext_len + 1 + offset
917
if data_len != len(stored_bytes):
918
raise ValueError('Index claimed fulltext len, but stored bytes'
920
% (len(stored_bytes), data_len))
921
data = [stored_bytes[offset + 1:]]
924
raise ValueError('Unknown content kind, bytes claim %s' % kind)
925
# XXX: This is inefficient at best
926
source = b''.join(self.chunks[:start_chunk])
927
delta_len, offset = decode_base128_int(stored_bytes[1:10])
928
data_len = delta_len + 1 + offset
929
if data_len != len(stored_bytes):
930
raise ValueError('Index claimed delta len, but stored bytes'
932
% (len(stored_bytes), data_len))
933
data = [apply_delta(source, stored_bytes[offset + 1:])]
934
data_sha1 = osutils.sha_strings(data)
935
return data, data_sha1
938
"""Finish this group, creating a formatted stream.
940
After calling this, the compressor should no longer be used
942
self._block.set_chunked_content(self.chunks, self.endpoint)
944
self._delta_index = None
948
"""Call this if you want to 'revoke' the last compression.
950
After this, the data structures will be rolled back, but you cannot do
953
self._delta_index = None
954
del self.chunks[self._last[0]:]
955
self.endpoint = self._last[1]
959
"""Return the overall compression ratio."""
960
return float(self.input_bytes) / float(self.endpoint)
963
class PythonGroupCompressor(_CommonGroupCompressor):
965
def __init__(self, settings=None):
966
"""Create a GroupCompressor.
968
Used only if the pyrex version is not available.
970
super(PythonGroupCompressor, self).__init__(settings)
971
self._delta_index = LinesDeltaIndex([])
972
# The actual content is managed by LinesDeltaIndex
973
self.chunks = self._delta_index.lines
975
def _compress(self, key, chunks, input_len, max_delta_size, soft=False):
976
"""see _CommonGroupCompressor._compress"""
977
new_lines = osutils.chunks_to_lines(chunks)
978
out_lines, index_lines = self._delta_index.make_delta(
979
new_lines, bytes_length=input_len, soft=soft)
980
delta_length = sum(map(len, out_lines))
981
if delta_length > max_delta_size:
982
# The delta is longer than the fulltext, insert a fulltext
984
out_lines = [b'f', encode_base128_int(input_len)]
985
out_lines.extend(new_lines)
986
index_lines = [False, False]
987
index_lines.extend([True] * len(new_lines))
989
# this is a worthy delta, output it
992
# Update the delta_length to include those two encoded integers
993
out_lines[1] = encode_base128_int(delta_length)
995
start = self.endpoint
996
chunk_start = len(self.chunks)
997
self._last = (chunk_start, self.endpoint)
998
self._delta_index.extend_lines(out_lines, index_lines)
999
self.endpoint = self._delta_index.endpoint
1000
self.input_bytes += input_len
1001
chunk_end = len(self.chunks)
1002
self.labels_deltas[key] = (start, chunk_start,
1003
self.endpoint, chunk_end)
1004
return start, self.endpoint, type
1007
class PyrexGroupCompressor(_CommonGroupCompressor):
1008
"""Produce a serialised group of compressed texts.
1010
It contains code very similar to SequenceMatcher because of having a similar
1011
task. However some key differences apply:
1013
* there is no junk, we want a minimal edit not a human readable diff.
1014
* we don't filter very common lines (because we don't know where a good
1015
range will start, and after the first text we want to be emitting minmal
1017
* we chain the left side, not the right side
1018
* we incrementally update the adjacency matrix as new lines are provided.
1019
* we look for matches in all of the left side, so the routine which does
1020
the analagous task of find_longest_match does not need to filter on the
1024
def __init__(self, settings=None):
1025
super(PyrexGroupCompressor, self).__init__(settings)
1026
max_bytes_to_index = self._settings.get('max_bytes_to_index', 0)
1027
self._delta_index = DeltaIndex(max_bytes_to_index=max_bytes_to_index)
1029
def _compress(self, key, chunks, input_len, max_delta_size, soft=False):
1030
"""see _CommonGroupCompressor._compress"""
1031
# By having action/label/sha1/len, we can parse the group if the index
1032
# was ever destroyed, we have the key in 'label', we know the final
1033
# bytes are valid from sha1, and we know where to find the end of this
1034
# record because of 'len'. (the delta record itself will store the
1035
# total length for the expanded record)
1036
# 'len: %d\n' costs approximately 1% increase in total data
1037
# Having the labels at all costs us 9-10% increase, 38% increase for
1038
# inventory pages, and 5.8% increase for text pages
1039
# new_chunks = ['label:%s\nsha1:%s\n' % (label, sha1)]
1040
if self._delta_index._source_offset != self.endpoint:
1041
raise AssertionError('_source_offset != endpoint'
1042
' somehow the DeltaIndex got out of sync with'
1043
' the output lines')
1044
bytes = b''.join(chunks)
1045
delta = self._delta_index.make_delta(bytes, max_delta_size)
1048
enc_length = encode_base128_int(input_len)
1049
len_mini_header = 1 + len(enc_length)
1050
self._delta_index.add_source(bytes, len_mini_header)
1051
new_chunks = [b'f', enc_length] + chunks
1054
enc_length = encode_base128_int(len(delta))
1055
len_mini_header = 1 + len(enc_length)
1056
new_chunks = [b'd', enc_length, delta]
1057
self._delta_index.add_delta_source(delta, len_mini_header)
1059
start = self.endpoint
1060
chunk_start = len(self.chunks)
1061
# Now output these bytes
1062
self._output_chunks(new_chunks)
1063
self.input_bytes += input_len
1064
chunk_end = len(self.chunks)
1065
self.labels_deltas[key] = (start, chunk_start,
1066
self.endpoint, chunk_end)
1067
if not self._delta_index._source_offset == self.endpoint:
1068
raise AssertionError('the delta index is out of sync'
1069
'with the output lines %s != %s'
1070
% (self._delta_index._source_offset, self.endpoint))
1071
return start, self.endpoint, type
1073
def _output_chunks(self, new_chunks):
1074
"""Output some chunks.
1076
:param new_chunks: The chunks to output.
1078
self._last = (len(self.chunks), self.endpoint)
1079
endpoint = self.endpoint
1080
self.chunks.extend(new_chunks)
1081
endpoint += sum(map(len, new_chunks))
1082
self.endpoint = endpoint
1085
def make_pack_factory(graph, delta, keylength, inconsistency_fatal=True):
1086
"""Create a factory for creating a pack based groupcompress.
1088
This is only functional enough to run interface tests, it doesn't try to
1089
provide a full pack environment.
1091
:param graph: Store a graph.
1092
:param delta: Delta compress contents.
1093
:param keylength: How long should keys be.
1095
def factory(transport):
1100
graph_index = BTreeBuilder(reference_lists=ref_length,
1101
key_elements=keylength)
1102
stream = transport.open_write_stream('newpack')
1103
writer = pack.ContainerWriter(stream.write)
1105
index = _GCGraphIndex(graph_index, lambda: True, parents=parents,
1106
add_callback=graph_index.add_nodes,
1107
inconsistency_fatal=inconsistency_fatal)
1108
access = pack_repo._DirectPackAccess({})
1109
access.set_writer(writer, graph_index, (transport, 'newpack'))
1110
result = GroupCompressVersionedFiles(index, access, delta)
1111
result.stream = stream
1112
result.writer = writer
1117
def cleanup_pack_group(versioned_files):
1118
versioned_files.writer.end()
1119
versioned_files.stream.close()
1122
class _BatchingBlockFetcher(object):
1123
"""Fetch group compress blocks in batches.
1125
:ivar total_bytes: int of expected number of bytes needed to fetch the
1126
currently pending batch.
1129
def __init__(self, gcvf, locations, get_compressor_settings=None):
1131
self.locations = locations
1133
self.batch_memos = {}
1134
self.memos_to_get = []
1135
self.total_bytes = 0
1136
self.last_read_memo = None
1138
self._get_compressor_settings = get_compressor_settings
1140
def add_key(self, key):
1141
"""Add another to key to fetch.
1143
:return: The estimated number of bytes needed to fetch the batch so
1146
self.keys.append(key)
1147
index_memo, _, _, _ = self.locations[key]
1148
read_memo = index_memo[0:3]
1149
# Three possibilities for this read_memo:
1150
# - it's already part of this batch; or
1151
# - it's not yet part of this batch, but is already cached; or
1152
# - it's not yet part of this batch and will need to be fetched.
1153
if read_memo in self.batch_memos:
1154
# This read memo is already in this batch.
1155
return self.total_bytes
1157
cached_block = self.gcvf._group_cache[read_memo]
1159
# This read memo is new to this batch, and the data isn't cached
1161
self.batch_memos[read_memo] = None
1162
self.memos_to_get.append(read_memo)
1163
byte_length = read_memo[2]
1164
self.total_bytes += byte_length
1166
# This read memo is new to this batch, but cached.
1167
# Keep a reference to the cached block in batch_memos because it's
1168
# certain that we'll use it when this batch is processed, but
1169
# there's a risk that it would fall out of _group_cache between now
1171
self.batch_memos[read_memo] = cached_block
1172
return self.total_bytes
1174
def _flush_manager(self):
1175
if self.manager is not None:
1176
for factory in self.manager.get_record_stream():
1179
self.last_read_memo = None
1181
def yield_factories(self, full_flush=False):
1182
"""Yield factories for keys added since the last yield. They will be
1183
returned in the order they were added via add_key.
1185
:param full_flush: by default, some results may not be returned in case
1186
they can be part of the next batch. If full_flush is True, then
1187
all results are returned.
1189
if self.manager is None and not self.keys:
1191
# Fetch all memos in this batch.
1192
blocks = self.gcvf._get_blocks(self.memos_to_get)
1193
# Turn blocks into factories and yield them.
1194
memos_to_get_stack = list(self.memos_to_get)
1195
memos_to_get_stack.reverse()
1196
for key in self.keys:
1197
index_memo, _, parents, _ = self.locations[key]
1198
read_memo = index_memo[:3]
1199
if self.last_read_memo != read_memo:
1200
# We are starting a new block. If we have a
1201
# manager, we have found everything that fits for
1202
# now, so yield records
1203
for factory in self._flush_manager():
1205
# Now start a new manager.
1206
if memos_to_get_stack and memos_to_get_stack[-1] == read_memo:
1207
# The next block from _get_blocks will be the block we
1209
block_read_memo, block = next(blocks)
1210
if block_read_memo != read_memo:
1211
raise AssertionError(
1212
"block_read_memo out of sync with read_memo"
1213
"(%r != %r)" % (block_read_memo, read_memo))
1214
self.batch_memos[read_memo] = block
1215
memos_to_get_stack.pop()
1217
block = self.batch_memos[read_memo]
1218
self.manager = _LazyGroupContentManager(block,
1219
get_compressor_settings=self._get_compressor_settings)
1220
self.last_read_memo = read_memo
1221
start, end = index_memo[3:5]
1222
self.manager.add_factory(key, parents, start, end)
1224
for factory in self._flush_manager():
1227
self.batch_memos.clear()
1228
del self.memos_to_get[:]
1229
self.total_bytes = 0
1232
class GroupCompressVersionedFiles(VersionedFilesWithFallbacks):
1233
"""A group-compress based VersionedFiles implementation."""
1235
# This controls how the GroupCompress DeltaIndex works. Basically, we
1236
# compute hash pointers into the source blocks (so hash(text) => text).
1237
# However each of these references costs some memory in trade against a
1238
# more accurate match result. For very large files, they either are
1239
# pre-compressed and change in bulk whenever they change, or change in just
1240
# local blocks. Either way, 'improved resolution' is not very helpful,
1241
# versus running out of memory trying to track everything. The default max
1242
# gives 100% sampling of a 1MB file.
1243
_DEFAULT_MAX_BYTES_TO_INDEX = 1024 * 1024
1244
_DEFAULT_COMPRESSOR_SETTINGS = {'max_bytes_to_index':
1245
_DEFAULT_MAX_BYTES_TO_INDEX}
1247
def __init__(self, index, access, delta=True, _unadded_refs=None,
1249
"""Create a GroupCompressVersionedFiles object.
1251
:param index: The index object storing access and graph data.
1252
:param access: The access object storing raw data.
1253
:param delta: Whether to delta compress or just entropy compress.
1254
:param _unadded_refs: private parameter, don't use.
1255
:param _group_cache: private parameter, don't use.
1258
self._access = access
1260
if _unadded_refs is None:
1262
self._unadded_refs = _unadded_refs
1263
if _group_cache is None:
1264
_group_cache = LRUSizeCache(max_size=50 * 1024 * 1024)
1265
self._group_cache = _group_cache
1266
self._immediate_fallback_vfs = []
1267
self._max_bytes_to_index = None
1269
def without_fallbacks(self):
1270
"""Return a clone of this object without any fallbacks configured."""
1271
return GroupCompressVersionedFiles(self._index, self._access,
1272
self._delta, _unadded_refs=dict(
1273
self._unadded_refs),
1274
_group_cache=self._group_cache)
1276
def add_lines(self, key, parents, lines, parent_texts=None,
1277
left_matching_blocks=None, nostore_sha=None, random_id=False,
1278
check_content=True):
1279
"""Add a text to the store.
1281
:param key: The key tuple of the text to add.
1282
:param parents: The parents key tuples of the text to add.
1283
:param lines: A list of lines. Each line must be a bytestring. And all
1284
of them except the last must be terminated with \\n and contain no
1285
other \\n's. The last line may either contain no \\n's or a single
1286
terminating \\n. If the lines list does meet this constraint the
1287
add routine may error or may succeed - but you will be unable to
1288
read the data back accurately. (Checking the lines have been split
1289
correctly is expensive and extremely unlikely to catch bugs so it
1290
is not done at runtime unless check_content is True.)
1291
:param parent_texts: An optional dictionary containing the opaque
1292
representations of some or all of the parents of version_id to
1293
allow delta optimisations. VERY IMPORTANT: the texts must be those
1294
returned by add_lines or data corruption can be caused.
1295
:param left_matching_blocks: a hint about which areas are common
1296
between the text and its left-hand-parent. The format is
1297
the SequenceMatcher.get_matching_blocks format.
1298
:param nostore_sha: Raise ExistingContent and do not add the lines to
1299
the versioned file if the digest of the lines matches this.
1300
:param random_id: If True a random id has been selected rather than
1301
an id determined by some deterministic process such as a converter
1302
from a foreign VCS. When True the backend may choose not to check
1303
for uniqueness of the resulting key within the versioned file, so
1304
this should only be done when the result is expected to be unique
1306
:param check_content: If True, the lines supplied are verified to be
1307
bytestrings that are correctly formed lines.
1308
:return: The text sha1, the number of bytes in the text, and an opaque
1309
representation of the inserted version which can be provided
1310
back to future add_lines calls in the parent_texts dictionary.
1312
self._index._check_write_ok()
1314
self._check_lines_not_unicode(lines)
1315
self._check_lines_are_lines(lines)
1316
return self.add_content(
1317
ChunkedContentFactory(
1318
key, parents, osutils.sha_strings(lines), lines, chunks_are_lines=True),
1319
parent_texts, left_matching_blocks, nostore_sha, random_id)
1321
def add_content(self, factory, parent_texts=None,
1322
left_matching_blocks=None, nostore_sha=None,
1324
"""Add a text to the store.
1326
:param factory: A ContentFactory that can be used to retrieve the key,
1327
parents and contents.
1328
:param parent_texts: An optional dictionary containing the opaque
1329
representations of some or all of the parents of version_id to
1330
allow delta optimisations. VERY IMPORTANT: the texts must be those
1331
returned by add_lines or data corruption can be caused.
1332
:param left_matching_blocks: a hint about which areas are common
1333
between the text and its left-hand-parent. The format is
1334
the SequenceMatcher.get_matching_blocks format.
1335
:param nostore_sha: Raise ExistingContent and do not add the lines to
1336
the versioned file if the digest of the lines matches this.
1337
:param random_id: If True a random id has been selected rather than
1338
an id determined by some deterministic process such as a converter
1339
from a foreign VCS. When True the backend may choose not to check
1340
for uniqueness of the resulting key within the versioned file, so
1341
this should only be done when the result is expected to be unique
1343
:return: The text sha1, the number of bytes in the text, and an opaque
1344
representation of the inserted version which can be provided
1345
back to future add_lines calls in the parent_texts dictionary.
1347
self._index._check_write_ok()
1348
parents = factory.parents
1349
self._check_add(factory.key, random_id)
1351
# The caller might pass None if there is no graph data, but kndx
1352
# indexes can't directly store that, so we give them
1353
# an empty tuple instead.
1355
# double handling for now. Make it work until then.
1356
sha1, length = list(self._insert_record_stream(
1357
[factory], random_id=random_id, nostore_sha=nostore_sha))[0]
1358
return sha1, length, None
1360
def add_fallback_versioned_files(self, a_versioned_files):
1361
"""Add a source of texts for texts not present in this knit.
1363
:param a_versioned_files: A VersionedFiles object.
1365
self._immediate_fallback_vfs.append(a_versioned_files)
1367
def annotate(self, key):
1368
"""See VersionedFiles.annotate."""
1369
ann = annotate.Annotator(self)
1370
return ann.annotate_flat(key)
1372
def get_annotator(self):
1373
return annotate.Annotator(self)
1375
def check(self, progress_bar=None, keys=None):
1376
"""See VersionedFiles.check()."""
1379
for record in self.get_record_stream(keys, 'unordered', True):
1380
for chunk in record.iter_bytes_as('chunked'):
1383
return self.get_record_stream(keys, 'unordered', True)
1385
def clear_cache(self):
1386
"""See VersionedFiles.clear_cache()"""
1387
self._group_cache.clear()
1388
self._index._graph_index.clear_cache()
1389
self._index._int_cache.clear()
1391
def _check_add(self, key, random_id):
1392
"""check that version_id and lines are safe to add."""
1393
version_id = key[-1]
1394
if version_id is not None:
1395
if osutils.contains_whitespace(version_id):
1396
raise errors.InvalidRevisionId(version_id, self)
1397
self.check_not_reserved_id(version_id)
1398
# TODO: If random_id==False and the key is already present, we should
1399
# probably check that the existing content is identical to what is
1400
# being inserted, and otherwise raise an exception. This would make
1401
# the bundle code simpler.
1403
def get_parent_map(self, keys):
1404
"""Get a map of the graph parents of keys.
1406
:param keys: The keys to look up parents for.
1407
:return: A mapping from keys to parents. Absent keys are absent from
1410
return self._get_parent_map_with_sources(keys)[0]
1412
def _get_parent_map_with_sources(self, keys):
1413
"""Get a map of the parents of keys.
1415
:param keys: The keys to look up parents for.
1416
:return: A tuple. The first element is a mapping from keys to parents.
1417
Absent keys are absent from the mapping. The second element is a
1418
list with the locations each key was found in. The first element
1419
is the in-this-knit parents, the second the first fallback source,
1423
sources = [self._index] + self._immediate_fallback_vfs
1426
for source in sources:
1429
new_result = source.get_parent_map(missing)
1430
source_results.append(new_result)
1431
result.update(new_result)
1432
missing.difference_update(set(new_result))
1433
return result, source_results
1435
def _get_blocks(self, read_memos):
1436
"""Get GroupCompressBlocks for the given read_memos.
1438
:returns: a series of (read_memo, block) pairs, in the order they were
1442
for read_memo in read_memos:
1444
block = self._group_cache[read_memo]
1448
cached[read_memo] = block
1450
not_cached_seen = set()
1451
for read_memo in read_memos:
1452
if read_memo in cached:
1453
# Don't fetch what we already have
1455
if read_memo in not_cached_seen:
1456
# Don't try to fetch the same data twice
1458
not_cached.append(read_memo)
1459
not_cached_seen.add(read_memo)
1460
raw_records = self._access.get_raw_records(not_cached)
1461
for read_memo in read_memos:
1463
yield read_memo, cached[read_memo]
1465
# Read the block, and cache it.
1466
zdata = next(raw_records)
1467
block = GroupCompressBlock.from_bytes(zdata)
1468
self._group_cache[read_memo] = block
1469
cached[read_memo] = block
1470
yield read_memo, block
1472
def get_missing_compression_parent_keys(self):
1473
"""Return the keys of missing compression parents.
1475
Missing compression parents occur when a record stream was missing
1476
basis texts, or a index was scanned that had missing basis texts.
1478
# GroupCompress cannot currently reference texts that are not in the
1479
# group, so this is valid for now
1482
def get_record_stream(self, keys, ordering, include_delta_closure):
1483
"""Get a stream of records for keys.
1485
:param keys: The keys to include.
1486
:param ordering: Either 'unordered' or 'topological'. A topologically
1487
sorted stream has compression parents strictly before their
1489
:param include_delta_closure: If True then the closure across any
1490
compression parents will be included (in the opaque data).
1491
:return: An iterator of ContentFactory objects, each of which is only
1492
valid until the iterator is advanced.
1494
# keys might be a generator
1495
orig_keys = list(keys)
1499
if (not self._index.has_graph
1500
and ordering in ('topological', 'groupcompress')):
1501
# Cannot topological order when no graph has been stored.
1502
# but we allow 'as-requested' or 'unordered'
1503
ordering = 'unordered'
1505
remaining_keys = keys
1508
keys = set(remaining_keys)
1509
for content_factory in self._get_remaining_record_stream(keys,
1510
orig_keys, ordering, include_delta_closure):
1511
remaining_keys.discard(content_factory.key)
1512
yield content_factory
1514
except errors.RetryWithNewPacks as e:
1515
self._access.reload_or_raise(e)
1517
def _find_from_fallback(self, missing):
1518
"""Find whatever keys you can from the fallbacks.
1520
:param missing: A set of missing keys. This set will be mutated as keys
1521
are found from a fallback_vfs
1522
:return: (parent_map, key_to_source_map, source_results)
1523
parent_map the overall key => parent_keys
1524
key_to_source_map a dict from {key: source}
1525
source_results a list of (source: keys)
1528
key_to_source_map = {}
1530
for source in self._immediate_fallback_vfs:
1533
source_parents = source.get_parent_map(missing)
1534
parent_map.update(source_parents)
1535
source_parents = list(source_parents)
1536
source_results.append((source, source_parents))
1537
key_to_source_map.update((key, source) for key in source_parents)
1538
missing.difference_update(source_parents)
1539
return parent_map, key_to_source_map, source_results
1541
def _get_ordered_source_keys(self, ordering, parent_map, key_to_source_map):
1542
"""Get the (source, [keys]) list.
1544
The returned objects should be in the order defined by 'ordering',
1545
which can weave between different sources.
1547
:param ordering: Must be one of 'topological' or 'groupcompress'
1548
:return: List of [(source, [keys])] tuples, such that all keys are in
1549
the defined order, regardless of source.
1551
if ordering == 'topological':
1552
present_keys = tsort.topo_sort(parent_map)
1554
# ordering == 'groupcompress'
1555
# XXX: This only optimizes for the target ordering. We may need
1556
# to balance that with the time it takes to extract
1557
# ordering, by somehow grouping based on
1558
# locations[key][0:3]
1559
present_keys = sort_gc_optimal(parent_map)
1560
# Now group by source:
1562
current_source = None
1563
for key in present_keys:
1564
source = key_to_source_map.get(key, self)
1565
if source is not current_source:
1566
source_keys.append((source, []))
1567
current_source = source
1568
source_keys[-1][1].append(key)
1571
def _get_as_requested_source_keys(self, orig_keys, locations, unadded_keys,
1574
current_source = None
1575
for key in orig_keys:
1576
if key in locations or key in unadded_keys:
1578
elif key in key_to_source_map:
1579
source = key_to_source_map[key]
1582
if source is not current_source:
1583
source_keys.append((source, []))
1584
current_source = source
1585
source_keys[-1][1].append(key)
1588
def _get_io_ordered_source_keys(self, locations, unadded_keys,
1591
# This is the group the bytes are stored in, followed by the
1592
# location in the group
1593
return locations[key][0]
1594
# We don't have an ordering for keys in the in-memory object, but
1595
# lets process the in-memory ones first.
1596
present_keys = list(unadded_keys)
1597
present_keys.extend(sorted(locations, key=get_group))
1598
# Now grab all of the ones from other sources
1599
source_keys = [(self, present_keys)]
1600
source_keys.extend(source_result)
1603
def _get_remaining_record_stream(self, keys, orig_keys, ordering,
1604
include_delta_closure):
1605
"""Get a stream of records for keys.
1607
:param keys: The keys to include.
1608
:param ordering: one of 'unordered', 'topological', 'groupcompress' or
1610
:param include_delta_closure: If True then the closure across any
1611
compression parents will be included (in the opaque data).
1612
:return: An iterator of ContentFactory objects, each of which is only
1613
valid until the iterator is advanced.
1616
locations = self._index.get_build_details(keys)
1617
unadded_keys = set(self._unadded_refs).intersection(keys)
1618
missing = keys.difference(locations)
1619
missing.difference_update(unadded_keys)
1620
(fallback_parent_map, key_to_source_map,
1621
source_result) = self._find_from_fallback(missing)
1622
if ordering in ('topological', 'groupcompress'):
1623
# would be better to not globally sort initially but instead
1624
# start with one key, recurse to its oldest parent, then grab
1625
# everything in the same group, etc.
1626
parent_map = dict((key, details[2]) for key, details in
1628
for key in unadded_keys:
1629
parent_map[key] = self._unadded_refs[key]
1630
parent_map.update(fallback_parent_map)
1631
source_keys = self._get_ordered_source_keys(ordering, parent_map,
1633
elif ordering == 'as-requested':
1634
source_keys = self._get_as_requested_source_keys(orig_keys,
1635
locations, unadded_keys, key_to_source_map)
1637
# We want to yield the keys in a semi-optimal (read-wise) ordering.
1638
# Otherwise we thrash the _group_cache and destroy performance
1639
source_keys = self._get_io_ordered_source_keys(locations,
1640
unadded_keys, source_result)
1642
yield AbsentContentFactory(key)
1643
# Batch up as many keys as we can until either:
1644
# - we encounter an unadded ref, or
1645
# - we run out of keys, or
1646
# - the total bytes to retrieve for this batch > BATCH_SIZE
1647
batcher = _BatchingBlockFetcher(self, locations,
1648
get_compressor_settings=self._get_compressor_settings)
1649
for source, keys in source_keys:
1652
if key in self._unadded_refs:
1653
# Flush batch, then yield unadded ref from
1655
for factory in batcher.yield_factories(full_flush=True):
1657
chunks, sha1 = self._compressor.extract(key)
1658
parents = self._unadded_refs[key]
1659
yield ChunkedContentFactory(key, parents, sha1, chunks)
1661
if batcher.add_key(key) > BATCH_SIZE:
1662
# Ok, this batch is big enough. Yield some results.
1663
for factory in batcher.yield_factories():
1666
for factory in batcher.yield_factories(full_flush=True):
1668
for record in source.get_record_stream(keys, ordering,
1669
include_delta_closure):
1671
for factory in batcher.yield_factories(full_flush=True):
1674
def get_sha1s(self, keys):
1675
"""See VersionedFiles.get_sha1s()."""
1677
for record in self.get_record_stream(keys, 'unordered', True):
1678
if record.sha1 is not None:
1679
result[record.key] = record.sha1
1681
if record.storage_kind != 'absent':
1682
result[record.key] = osutils.sha_strings(
1683
record.iter_bytes_as('chunked'))
1686
def insert_record_stream(self, stream):
1687
"""Insert a record stream into this container.
1689
:param stream: A stream of records to insert.
1691
:seealso VersionedFiles.get_record_stream:
1693
# XXX: Setting random_id=True makes
1694
# test_insert_record_stream_existing_keys fail for groupcompress and
1695
# groupcompress-nograph, this needs to be revisited while addressing
1696
# 'bzr branch' performance issues.
1697
for _, _ in self._insert_record_stream(stream, random_id=False):
1700
def _get_compressor_settings(self):
1701
if self._max_bytes_to_index is None:
1702
# TODO: VersionedFiles don't know about their containing
1703
# repository, so they don't have much of an idea about their
1704
# location. So for now, this is only a global option.
1705
c = config.GlobalConfig()
1706
val = c.get_user_option('bzr.groupcompress.max_bytes_to_index')
1710
except ValueError as e:
1711
trace.warning('Value for '
1712
'"bzr.groupcompress.max_bytes_to_index"'
1713
' %r is not an integer'
1717
val = self._DEFAULT_MAX_BYTES_TO_INDEX
1718
self._max_bytes_to_index = val
1719
return {'max_bytes_to_index': self._max_bytes_to_index}
1721
def _make_group_compressor(self):
1722
return GroupCompressor(self._get_compressor_settings())
1724
def _insert_record_stream(self, stream, random_id=False, nostore_sha=None,
1726
"""Internal core to insert a record stream into this container.
1728
This helper function has a different interface than insert_record_stream
1729
to allow add_lines to be minimal, but still return the needed data.
1731
:param stream: A stream of records to insert.
1732
:param nostore_sha: If the sha1 of a given text matches nostore_sha,
1733
raise ExistingContent, rather than committing the new text.
1734
:param reuse_blocks: If the source is streaming from
1735
groupcompress-blocks, just insert the blocks as-is, rather than
1736
expanding the texts and inserting again.
1737
:return: An iterator over (sha1, length) of the inserted records.
1738
:seealso insert_record_stream:
1743
def get_adapter(adapter_key):
1745
return adapters[adapter_key]
1747
adapter_factory = adapter_registry.get(adapter_key)
1748
adapter = adapter_factory(self)
1749
adapters[adapter_key] = adapter
1751
# This will go up to fulltexts for gc to gc fetching, which isn't
1753
self._compressor = self._make_group_compressor()
1754
self._unadded_refs = {}
1758
bytes_len, chunks = self._compressor.flush().to_chunks()
1759
self._compressor = self._make_group_compressor()
1760
# Note: At this point we still have 1 copy of the fulltext (in
1761
# record and the var 'bytes'), and this generates 2 copies of
1762
# the compressed text (one for bytes, one in chunks)
1763
# TODO: Figure out how to indicate that we would be happy to free
1764
# the fulltext content at this point. Note that sometimes we
1765
# will want it later (streaming CHK pages), but most of the
1766
# time we won't (everything else)
1767
index, start, length = self._access.add_raw_record(
1768
None, bytes_len, chunks)
1770
for key, reads, refs in keys_to_add:
1771
nodes.append((key, b"%d %d %s" % (start, length, reads), refs))
1772
self._index.add_records(nodes, random_id=random_id)
1773
self._unadded_refs = {}
1777
max_fulltext_len = 0
1778
max_fulltext_prefix = None
1779
insert_manager = None
1782
# XXX: TODO: remove this, it is just for safety checking for now
1783
inserted_keys = set()
1784
reuse_this_block = reuse_blocks
1785
for record in stream:
1786
# Raise an error when a record is missing.
1787
if record.storage_kind == 'absent':
1788
raise errors.RevisionNotPresent(record.key, self)
1790
if record.key in inserted_keys:
1791
trace.note(gettext('Insert claimed random_id=True,'
1792
' but then inserted %r two times'), record.key)
1794
inserted_keys.add(record.key)
1796
# If the reuse_blocks flag is set, check to see if we can just
1797
# copy a groupcompress block as-is.
1798
# We only check on the first record (groupcompress-block) not
1799
# on all of the (groupcompress-block-ref) entries.
1800
# The reuse_this_block flag is then kept for as long as
1801
if record.storage_kind == 'groupcompress-block':
1802
# Check to see if we really want to re-use this block
1803
insert_manager = record._manager
1804
reuse_this_block = insert_manager.check_is_well_utilized()
1806
reuse_this_block = False
1807
if reuse_this_block:
1808
# We still want to reuse this block
1809
if record.storage_kind == 'groupcompress-block':
1810
# Insert the raw block into the target repo
1811
insert_manager = record._manager
1812
bytes_len, chunks = record._manager._block.to_chunks()
1813
_, start, length = self._access.add_raw_record(
1814
None, bytes_len, chunks)
1816
block_length = length
1817
if record.storage_kind in ('groupcompress-block',
1818
'groupcompress-block-ref'):
1819
if insert_manager is None:
1820
raise AssertionError('No insert_manager set')
1821
if insert_manager is not record._manager:
1822
raise AssertionError('insert_manager does not match'
1823
' the current record, we cannot be positive'
1824
' that the appropriate content was inserted.'
1826
value = b"%d %d %d %d" % (block_start, block_length,
1827
record._start, record._end)
1828
nodes = [(record.key, value, (record.parents,))]
1829
# TODO: Consider buffering up many nodes to be added, not
1830
# sure how much overhead this has, but we're seeing
1831
# ~23s / 120s in add_records calls
1832
self._index.add_records(nodes, random_id=random_id)
1835
chunks = record.get_bytes_as('chunked')
1836
except errors.UnavailableRepresentation:
1837
adapter_key = record.storage_kind, 'chunked'
1838
adapter = get_adapter(adapter_key)
1839
chunks = adapter.get_bytes(record, 'chunked')
1840
chunks_len = record.size
1841
if chunks_len is None:
1842
chunks_len = sum(map(len, chunks))
1843
if len(record.key) > 1:
1844
prefix = record.key[0]
1845
soft = (prefix == last_prefix)
1849
if max_fulltext_len < chunks_len:
1850
max_fulltext_len = chunks_len
1851
max_fulltext_prefix = prefix
1852
(found_sha1, start_point, end_point,
1853
type) = self._compressor.compress(
1854
record.key, chunks, chunks_len, record.sha1, soft=soft,
1855
nostore_sha=nostore_sha)
1856
# delta_ratio = float(chunks_len) / (end_point - start_point)
1857
# Check if we want to continue to include that text
1858
if (prefix == max_fulltext_prefix
1859
and end_point < 2 * max_fulltext_len):
1860
# As long as we are on the same file_id, we will fill at least
1861
# 2 * max_fulltext_len
1862
start_new_block = False
1863
elif end_point > 4 * 1024 * 1024:
1864
start_new_block = True
1865
elif (prefix is not None and prefix != last_prefix
1866
and end_point > 2 * 1024 * 1024):
1867
start_new_block = True
1869
start_new_block = False
1870
last_prefix = prefix
1872
self._compressor.pop_last()
1874
max_fulltext_len = chunks_len
1875
(found_sha1, start_point, end_point,
1876
type) = self._compressor.compress(
1877
record.key, chunks, chunks_len, record.sha1)
1878
if record.key[-1] is None:
1879
key = record.key[:-1] + (b'sha1:' + found_sha1,)
1882
self._unadded_refs[key] = record.parents
1883
yield found_sha1, chunks_len
1884
as_st = static_tuple.StaticTuple.from_sequence
1885
if record.parents is not None:
1886
parents = as_st([as_st(p) for p in record.parents])
1889
refs = static_tuple.StaticTuple(parents)
1891
(key, b'%d %d' % (start_point, end_point), refs))
1892
if len(keys_to_add):
1894
self._compressor = None
1896
def iter_lines_added_or_present_in_keys(self, keys, pb=None):
1897
"""Iterate over the lines in the versioned files from keys.
1899
This may return lines from other keys. Each item the returned
1900
iterator yields is a tuple of a line and a text version that that line
1901
is present in (not introduced in).
1903
Ordering of results is in whatever order is most suitable for the
1904
underlying storage format.
1906
If a progress bar is supplied, it may be used to indicate progress.
1907
The caller is responsible for cleaning up progress bars (because this
1911
* Lines are normalised by the underlying store: they will all have \n
1913
* Lines are returned in arbitrary order.
1915
:return: An iterator over (line, key).
1919
# we don't care about inclusions, the caller cares.
1920
# but we need to setup a list of records to visit.
1921
# we need key, position, length
1922
for key_idx, record in enumerate(self.get_record_stream(keys,
1923
'unordered', True)):
1924
# XXX: todo - optimise to use less than full texts.
1927
pb.update('Walking content', key_idx, total)
1928
if record.storage_kind == 'absent':
1929
raise errors.RevisionNotPresent(key, self)
1930
for line in record.iter_bytes_as('lines'):
1933
pb.update('Walking content', total, total)
1936
"""See VersionedFiles.keys."""
1937
if 'evil' in debug.debug_flags:
1938
trace.mutter_callsite(2, "keys scales with size of history")
1939
sources = [self._index] + self._immediate_fallback_vfs
1941
for source in sources:
1942
result.update(source.keys())
1946
class _GCBuildDetails(object):
1947
"""A blob of data about the build details.
1949
This stores the minimal data, which then allows compatibility with the old
1950
api, without taking as much memory.
1953
__slots__ = ('_index', '_group_start', '_group_end', '_basis_end',
1954
'_delta_end', '_parents')
1957
compression_parent = None
1959
def __init__(self, parents, position_info):
1960
self._parents = parents
1961
(self._index, self._group_start, self._group_end, self._basis_end,
1962
self._delta_end) = position_info
1965
return '%s(%s, %s)' % (self.__class__.__name__,
1966
self.index_memo, self._parents)
1969
def index_memo(self):
1970
return (self._index, self._group_start, self._group_end,
1971
self._basis_end, self._delta_end)
1974
def record_details(self):
1975
return static_tuple.StaticTuple(self.method, None)
1977
def __getitem__(self, offset):
1978
"""Compatibility thunk to act like a tuple."""
1980
return self.index_memo
1982
return self.compression_parent # Always None
1984
return self._parents
1986
return self.record_details
1988
raise IndexError('offset out of range')
1994
class _GCGraphIndex(object):
1995
"""Mapper from GroupCompressVersionedFiles needs into GraphIndex storage."""
1997
def __init__(self, graph_index, is_locked, parents=True,
1998
add_callback=None, track_external_parent_refs=False,
1999
inconsistency_fatal=True, track_new_keys=False):
2000
"""Construct a _GCGraphIndex on a graph_index.
2002
:param graph_index: An implementation of breezy.index.GraphIndex.
2003
:param is_locked: A callback, returns True if the index is locked and
2005
:param parents: If True, record knits parents, if not do not record
2007
:param add_callback: If not None, allow additions to the index and call
2008
this callback with a list of added GraphIndex nodes:
2009
[(node, value, node_refs), ...]
2010
:param track_external_parent_refs: As keys are added, keep track of the
2011
keys they reference, so that we can query get_missing_parents(),
2013
:param inconsistency_fatal: When asked to add records that are already
2014
present, and the details are inconsistent with the existing
2015
record, raise an exception instead of warning (and skipping the
2018
self._add_callback = add_callback
2019
self._graph_index = graph_index
2020
self._parents = parents
2021
self.has_graph = parents
2022
self._is_locked = is_locked
2023
self._inconsistency_fatal = inconsistency_fatal
2024
# GroupCompress records tend to have the same 'group' start + offset
2025
# repeated over and over, this creates a surplus of ints
2026
self._int_cache = {}
2027
if track_external_parent_refs:
2028
self._key_dependencies = _KeyRefs(
2029
track_new_keys=track_new_keys)
2031
self._key_dependencies = None
2033
def add_records(self, records, random_id=False):
2034
"""Add multiple records to the index.
2036
This function does not insert data into the Immutable GraphIndex
2037
backing the KnitGraphIndex, instead it prepares data for insertion by
2038
the caller and checks that it is safe to insert then calls
2039
self._add_callback with the prepared GraphIndex nodes.
2041
:param records: a list of tuples:
2042
(key, options, access_memo, parents).
2043
:param random_id: If True the ids being added were randomly generated
2044
and no check for existence will be performed.
2046
if not self._add_callback:
2047
raise errors.ReadOnlyError(self)
2048
# we hope there are no repositories with inconsistent parentage
2053
for (key, value, refs) in records:
2054
if not self._parents:
2058
raise knit.KnitCorrupt(self,
2059
"attempt to add node with parents "
2060
"in parentless index.")
2063
keys[key] = (value, refs)
2066
present_nodes = self._get_entries(keys)
2067
for (index, key, value, node_refs) in present_nodes:
2068
# Sometimes these are passed as a list rather than a tuple
2069
node_refs = static_tuple.as_tuples(node_refs)
2070
passed = static_tuple.as_tuples(keys[key])
2071
if node_refs != passed[1]:
2072
details = '%s %s %s' % (key, (value, node_refs), passed)
2073
if self._inconsistency_fatal:
2074
raise knit.KnitCorrupt(self, "inconsistent details"
2075
" in add_records: %s" %
2078
trace.warning("inconsistent details in skipped"
2079
" record: %s", details)
2085
for key, (value, node_refs) in keys.items():
2086
result.append((key, value, node_refs))
2088
for key, (value, node_refs) in keys.items():
2089
result.append((key, value))
2091
key_dependencies = self._key_dependencies
2092
if key_dependencies is not None:
2094
for key, value, refs in records:
2096
key_dependencies.add_references(key, parents)
2098
for key, value, refs in records:
2099
new_keys.add_key(key)
2100
self._add_callback(records)
2102
def _check_read(self):
2103
"""Raise an exception if reads are not permitted."""
2104
if not self._is_locked():
2105
raise errors.ObjectNotLocked(self)
2107
def _check_write_ok(self):
2108
"""Raise an exception if writes are not permitted."""
2109
if not self._is_locked():
2110
raise errors.ObjectNotLocked(self)
2112
def _get_entries(self, keys, check_present=False):
2113
"""Get the entries for keys.
2115
Note: Callers are responsible for checking that the index is locked
2116
before calling this method.
2118
:param keys: An iterable of index key tuples.
2123
for node in self._graph_index.iter_entries(keys):
2125
found_keys.add(node[1])
2127
# adapt parentless index to the rest of the code.
2128
for node in self._graph_index.iter_entries(keys):
2129
yield node[0], node[1], node[2], ()
2130
found_keys.add(node[1])
2132
missing_keys = keys.difference(found_keys)
2134
raise errors.RevisionNotPresent(missing_keys.pop(), self)
2136
def find_ancestry(self, keys):
2137
"""See CombinedGraphIndex.find_ancestry"""
2138
return self._graph_index.find_ancestry(keys, 0)
2140
def get_parent_map(self, keys):
2141
"""Get a map of the parents of keys.
2143
:param keys: The keys to look up parents for.
2144
:return: A mapping from keys to parents. Absent keys are absent from
2148
nodes = self._get_entries(keys)
2152
result[node[1]] = node[3][0]
2155
result[node[1]] = None
2158
def get_missing_parents(self):
2159
"""Return the keys of missing parents."""
2160
# Copied from _KnitGraphIndex.get_missing_parents
2161
# We may have false positives, so filter those out.
2162
self._key_dependencies.satisfy_refs_for_keys(
2163
self.get_parent_map(self._key_dependencies.get_unsatisfied_refs()))
2164
return frozenset(self._key_dependencies.get_unsatisfied_refs())
2166
def get_build_details(self, keys):
2167
"""Get the various build details for keys.
2169
Ghosts are omitted from the result.
2171
:param keys: An iterable of keys.
2172
:return: A dict of key:
2173
(index_memo, compression_parent, parents, record_details).
2175
* index_memo: opaque structure to pass to read_records to extract
2177
* compression_parent: Content that this record is built upon, may
2179
* parents: Logical parents of this node
2180
* record_details: extra information about the content which needs
2181
to be passed to Factory.parse_record
2185
entries = self._get_entries(keys)
2186
for entry in entries:
2188
if not self._parents:
2191
parents = entry[3][0]
2192
details = _GCBuildDetails(parents, self._node_to_position(entry))
2193
result[key] = details
2197
"""Get all the keys in the collection.
2199
The keys are not ordered.
2202
return [node[1] for node in self._graph_index.iter_all_entries()]
2204
def _node_to_position(self, node):
2205
"""Convert an index value to position details."""
2206
bits = node[2].split(b' ')
2207
# It would be nice not to read the entire gzip.
2208
# start and stop are put into _int_cache because they are very common.
2209
# They define the 'group' that an entry is in, and many groups can have
2210
# thousands of objects.
2211
# Branching Launchpad, for example, saves ~600k integers, at 12 bytes
2212
# each, or about 7MB. Note that it might be even more when you consider
2213
# how PyInt is allocated in separate slabs. And you can't return a slab
2214
# to the OS if even 1 int on it is in use. Note though that Python uses
2215
# a LIFO when re-using PyInt slots, which might cause more
2217
start = int(bits[0])
2218
start = self._int_cache.setdefault(start, start)
2220
stop = self._int_cache.setdefault(stop, stop)
2221
basis_end = int(bits[2])
2222
delta_end = int(bits[3])
2223
# We can't use StaticTuple here, because node[0] is a BTreeGraphIndex
2225
return (node[0], start, stop, basis_end, delta_end)
2227
def scan_unvalidated_index(self, graph_index):
2228
"""Inform this _GCGraphIndex that there is an unvalidated index.
2230
This allows this _GCGraphIndex to keep track of any missing
2231
compression parents we may want to have filled in to make those
2232
indices valid. It also allows _GCGraphIndex to track any new keys.
2234
:param graph_index: A GraphIndex
2236
key_dependencies = self._key_dependencies
2237
if key_dependencies is None:
2239
for node in graph_index.iter_all_entries():
2240
# Add parent refs from graph_index (and discard parent refs
2241
# that the graph_index has).
2242
key_dependencies.add_references(node[1], node[3][0])
2245
from ._groupcompress_py import (
2247
apply_delta_to_source,
2250
decode_copy_instruction,
2254
from ._groupcompress_pyx import (
2256
apply_delta_to_source,
2261
GroupCompressor = PyrexGroupCompressor
2262
except ImportError as e:
2263
osutils.failed_to_load_extension(e)
2264
GroupCompressor = PythonGroupCompressor
added
removed
breezy/bzr/groupcompress.py
