17
17
"""Core compression logic for compressing streams of related files."""
19
from __future__ import absolute_import
24
from ..lazy_import import lazy_import
25
lazy_import(globals(), """
35
from breezy.bzr import (
41
from breezy.i18n import gettext
47
from .btree_index import BTreeBuilder
48
from ..lru_cache import LRUSizeCache
49
from ..sixish import (
55
from .versionedfile import (
37
from bzrlib.btree_index import BTreeBuilder
38
from bzrlib.lru_cache import LRUSizeCache
39
from bzrlib.tsort import topo_sort
40
from bzrlib.versionedfile import (
58
42
AbsentContentFactory,
59
43
ChunkedContentFactory,
60
44
FulltextContentFactory,
61
VersionedFilesWithFallbacks,
64
48
# Minimum number of uncompressed bytes to try fetch at once when retrieving
65
49
# groupcompress blocks.
68
# osutils.sha_string(b'')
69
_null_sha1 = b'da39a3ee5e6b4b0d3255bfef95601890afd80709'
52
_USE_LZMA = False and (pylzma is not None)
54
# osutils.sha_string('')
55
_null_sha1 = 'da39a3ee5e6b4b0d3255bfef95601890afd80709'
72
57
def sort_gc_optimal(parent_map):
73
58
"""Sort and group the keys in parent_map into groupcompress order.
94
79
for prefix in sorted(per_prefix_map):
95
present_keys.extend(reversed(tsort.topo_sort(per_prefix_map[prefix])))
80
present_keys.extend(reversed(topo_sort(per_prefix_map[prefix])))
96
81
return present_keys
99
class DecompressCorruption(errors.BzrError):
101
_fmt = "Corruption while decompressing repository file%(orig_error)s"
103
def __init__(self, orig_error=None):
104
if orig_error is not None:
105
self.orig_error = ", %s" % (orig_error,)
108
errors.BzrError.__init__(self)
111
84
# The max zlib window size is 32kB, so if we set 'max_size' output of the
112
85
# decompressor to the requested bytes + 32kB, then we should guarantee
113
86
# num_bytes coming out.
114
_ZLIB_DECOMP_WINDOW = 32 * 1024
87
_ZLIB_DECOMP_WINDOW = 32*1024
117
89
class GroupCompressBlock(object):
118
90
"""An object which maintains the internal structure of the compressed data.
123
95
# Group Compress Block v1 Zlib
124
GCB_HEADER = b'gcb1z\n'
96
GCB_HEADER = 'gcb1z\n'
125
97
# Group Compress Block v1 Lzma
126
GCB_LZ_HEADER = b'gcb1l\n'
98
GCB_LZ_HEADER = 'gcb1l\n'
127
99
GCB_KNOWN_HEADERS = (GCB_HEADER, GCB_LZ_HEADER)
129
101
def __init__(self):
130
102
# map by key? or just order in file?
131
103
self._compressor_name = None
132
self._z_content_chunks = None
104
self._z_content = None
133
105
self._z_content_decompressor = None
134
106
self._z_content_length = None
135
107
self._content_length = None
160
132
# Expand the content if required
161
133
if self._content is None:
162
134
if self._content_chunks is not None:
163
self._content = b''.join(self._content_chunks)
135
self._content = ''.join(self._content_chunks)
164
136
self._content_chunks = None
165
137
if self._content is None:
166
# We join self._z_content_chunks here, because if we are
167
# decompressing, then it is *very* likely that we have a single
169
if self._z_content_chunks is None:
138
if self._z_content is None:
170
139
raise AssertionError('No content to decompress')
171
z_content = b''.join(self._z_content_chunks)
140
if self._z_content == '':
174
142
elif self._compressor_name == 'lzma':
175
143
# We don't do partial lzma decomp yet
177
self._content = pylzma.decompress(z_content)
144
self._content = pylzma.decompress(self._z_content)
178
145
elif self._compressor_name == 'zlib':
179
146
# Start a zlib decompressor
180
147
if num_bytes * 4 > self._content_length * 3:
181
148
# If we are requesting more that 3/4ths of the content,
182
149
# just extract the whole thing in a single pass
183
150
num_bytes = self._content_length
184
self._content = zlib.decompress(z_content)
151
self._content = zlib.decompress(self._z_content)
186
153
self._z_content_decompressor = zlib.decompressobj()
187
154
# Seed the decompressor with the uncompressed bytes, so
188
155
# that the rest of the code is simplified
189
156
self._content = self._z_content_decompressor.decompress(
190
z_content, num_bytes + _ZLIB_DECOMP_WINDOW)
157
self._z_content, num_bytes + _ZLIB_DECOMP_WINDOW)
191
158
if not self._z_content_decompressor.unconsumed_tail:
192
159
self._z_content_decompressor = None
230
197
# At present, we have 2 integers for the compressed and uncompressed
231
198
# content. In base10 (ascii) 14 bytes can represent > 1TB, so to avoid
232
199
# checking too far, cap the search to 14 bytes.
233
pos2 = data.index(b'\n', pos, pos + 14)
234
self._z_content_length = int(data[pos:pos2])
236
pos2 = data.index(b'\n', pos, pos + 14)
237
self._content_length = int(data[pos:pos2])
239
if len(data) != (pos + self._z_content_length):
200
pos2 = bytes.index('\n', pos, pos + 14)
201
self._z_content_length = int(bytes[pos:pos2])
203
pos2 = bytes.index('\n', pos, pos + 14)
204
self._content_length = int(bytes[pos:pos2])
206
if len(bytes) != (pos + self._z_content_length):
240
207
# XXX: Define some GCCorrupt error ?
241
208
raise AssertionError('Invalid bytes: (%d) != %d + %d' %
242
(len(data), pos, self._z_content_length))
243
self._z_content_chunks = (data[pos:],)
246
def _z_content(self):
247
"""Return z_content_chunks as a simple string.
249
Meant only to be used by the test suite.
251
if self._z_content_chunks is not None:
252
return b''.join(self._z_content_chunks)
209
(len(bytes), pos, self._z_content_length))
210
self._z_content = bytes[pos:]
256
213
def from_bytes(cls, bytes):
259
if header not in cls.GCB_KNOWN_HEADERS:
215
if bytes[:6] not in cls.GCB_KNOWN_HEADERS:
260
216
raise ValueError('bytes did not start with any of %r'
261
217
% (cls.GCB_KNOWN_HEADERS,))
262
if header == cls.GCB_HEADER:
218
# XXX: why not testing the whole header ?
263
220
out._compressor_name = 'zlib'
264
elif header == cls.GCB_LZ_HEADER:
221
elif bytes[4] == 'l':
265
222
out._compressor_name = 'lzma'
267
raise ValueError('unknown compressor: %r' % (header,))
224
raise ValueError('unknown compressor: %r' % (bytes,))
268
225
out._parse_bytes(bytes, 6)
276
233
:return: The bytes for the content
278
235
if start == end == 0:
280
237
self._ensure_content(end)
281
238
# The bytes are 'f' or 'd' for the type, then a variable-length
282
239
# base128 integer for the content size, then the actual content
283
240
# We know that the variable-length integer won't be longer than 5
284
241
# bytes (it takes 5 bytes to encode 2^32)
285
c = self._content[start:start + 1]
242
c = self._content[start]
287
244
type = 'fulltext'
290
247
raise ValueError('Unknown content control code: %s'
293
250
content_len, len_len = decode_base128_int(
294
self._content[start + 1:start + 6])
251
self._content[start + 1:start + 6])
295
252
content_start = start + 1 + len_len
296
253
if end != content_start + content_len:
297
254
raise ValueError('end != len according to field header'
298
' %s != %s' % (end, content_start + content_len))
300
return self._content[content_start:end]
301
# Must be type delta as checked above
302
return apply_delta_to_source(self._content, content_start, end)
255
' %s != %s' % (end, content_start + content_len))
257
bytes = self._content[content_start:end]
259
bytes = apply_delta_to_source(self._content, content_start, end)
304
262
def set_chunked_content(self, content_chunks, length):
305
263
"""Set the content of this block to the given chunks."""
311
269
self._content_length = length
312
270
self._content_chunks = content_chunks
313
271
self._content = None
314
self._z_content_chunks = None
272
self._z_content = None
316
274
def set_content(self, content):
317
275
"""Set the content of this block."""
318
276
self._content_length = len(content)
319
277
self._content = content
320
self._z_content_chunks = None
322
def _create_z_content_from_chunks(self, chunks):
278
self._z_content = None
280
def _create_z_content_using_lzma(self):
281
if self._content_chunks is not None:
282
self._content = ''.join(self._content_chunks)
283
self._content_chunks = None
284
if self._content is None:
285
raise AssertionError('Nothing to compress')
286
self._z_content = pylzma.compress(self._content)
287
self._z_content_length = len(self._z_content)
289
def _create_z_content_from_chunks(self):
323
290
compressor = zlib.compressobj(zlib.Z_DEFAULT_COMPRESSION)
324
# Peak in this point is 1 fulltext, 1 compressed text, + zlib overhead
325
# (measured peak is maybe 30MB over the above...)
326
compressed_chunks = list(map(compressor.compress, chunks))
291
compressed_chunks = map(compressor.compress, self._content_chunks)
327
292
compressed_chunks.append(compressor.flush())
328
# Ignore empty chunks
329
self._z_content_chunks = [c for c in compressed_chunks if c]
330
self._z_content_length = sum(map(len, self._z_content_chunks))
293
self._z_content = ''.join(compressed_chunks)
294
self._z_content_length = len(self._z_content)
332
296
def _create_z_content(self):
333
if self._z_content_chunks is not None:
297
if self._z_content is not None:
300
self._create_z_content_using_lzma()
335
302
if self._content_chunks is not None:
336
chunks = self._content_chunks
338
chunks = (self._content,)
339
self._create_z_content_from_chunks(chunks)
342
"""Create the byte stream as a series of 'chunks'"""
343
self._create_z_content()
344
header = self.GCB_HEADER
345
chunks = [b'%s%d\n%d\n'
346
% (header, self._z_content_length, self._content_length),
348
chunks.extend(self._z_content_chunks)
349
total_len = sum(map(len, chunks))
350
return total_len, chunks
303
self._create_z_content_from_chunks()
305
self._z_content = zlib.compress(self._content)
306
self._z_content_length = len(self._z_content)
352
308
def to_bytes(self):
353
309
"""Encode the information into a byte stream."""
354
total_len, chunks = self.to_chunks()
355
return b''.join(chunks)
310
self._create_z_content()
312
header = self.GCB_LZ_HEADER
314
header = self.GCB_HEADER
316
'%d\n%d\n' % (self._z_content_length, self._content_length),
319
return ''.join(chunks)
357
321
def _dump(self, include_text=False):
358
322
"""Take this block, and spit out a human-readable structure.
370
334
while pos < self._content_length:
371
kind = self._content[pos:pos + 1]
335
kind = self._content[pos]
373
if kind not in (b'f', b'd'):
337
if kind not in ('f', 'd'):
374
338
raise ValueError('invalid kind character: %r' % (kind,))
375
339
content_len, len_len = decode_base128_int(
376
self._content[pos:pos + 5])
340
self._content[pos:pos + 5])
378
342
if content_len + pos > self._content_length:
379
343
raise ValueError('invalid content_len %d for record @ pos %d'
380
344
% (content_len, pos - len_len - 1))
381
if kind == b'f': # Fulltext
345
if kind == 'f': # Fulltext
383
text = self._content[pos:pos + content_len]
384
result.append((b'f', content_len, text))
347
text = self._content[pos:pos+content_len]
348
result.append(('f', content_len, text))
386
result.append((b'f', content_len))
387
elif kind == b'd': # Delta
388
delta_content = self._content[pos:pos + content_len]
350
result.append(('f', content_len))
351
elif kind == 'd': # Delta
352
delta_content = self._content[pos:pos+content_len]
390
354
# The first entry in a delta is the decompressed length
391
355
decomp_len, delta_pos = decode_base128_int(delta_content)
392
result.append((b'd', content_len, decomp_len, delta_info))
356
result.append(('d', content_len, decomp_len, delta_info))
394
358
while delta_pos < content_len:
395
c = indexbytes(delta_content, delta_pos)
359
c = ord(delta_content[delta_pos])
399
363
delta_pos) = decode_copy_instruction(delta_content, c,
402
text = self._content[offset:offset + length]
403
delta_info.append((b'c', offset, length, text))
366
text = self._content[offset:offset+length]
367
delta_info.append(('c', offset, length, text))
405
delta_info.append((b'c', offset, length))
369
delta_info.append(('c', offset, length))
406
370
measured_len += length
409
txt = delta_content[delta_pos:delta_pos + c]
373
txt = delta_content[delta_pos:delta_pos+c]
412
delta_info.append((b'i', c, txt))
376
delta_info.append(('i', c, txt))
413
377
measured_len += c
415
379
if delta_pos != content_len:
465
429
# wire bytes, something...
466
430
return self._manager._wire_bytes()
469
433
if storage_kind in ('fulltext', 'chunked'):
470
434
if self._bytes is None:
471
435
# Grab and cache the raw bytes for this entry
472
436
# and break the ref-cycle with _manager since we don't need it
475
self._manager._prepare_for_extract()
476
except zlib.error as value:
477
raise DecompressCorruption("zlib: " + str(value))
438
self._manager._prepare_for_extract()
478
439
block = self._manager._block
479
440
self._bytes = block.extract(self.key, self._start, self._end)
480
441
# There are code paths that first extract as fulltext, and then
491
452
class _LazyGroupContentManager(object):
492
453
"""This manages a group of _LazyGroupCompressFactory objects."""
494
_max_cut_fraction = 0.75 # We allow a block to be trimmed to 75% of
495
# current size, and still be considered
497
_full_block_size = 4 * 1024 * 1024
498
_full_mixed_block_size = 2 * 1024 * 1024
499
_full_enough_block_size = 3 * 1024 * 1024 # size at which we won't repack
500
_full_enough_mixed_block_size = 2 * 768 * 1024 # 1.5MB
455
_max_cut_fraction = 0.75 # We allow a block to be trimmed to 75% of
456
# current size, and still be considered
458
_full_block_size = 4*1024*1024
459
_full_mixed_block_size = 2*1024*1024
460
_full_enough_block_size = 3*1024*1024 # size at which we won't repack
461
_full_enough_mixed_block_size = 2*768*1024 # 1.5MB
502
def __init__(self, block, get_compressor_settings=None):
463
def __init__(self, block):
503
464
self._block = block
504
465
# We need to preserve the ordering
505
466
self._factories = []
506
467
self._last_byte = 0
507
self._get_settings = get_compressor_settings
508
self._compressor_settings = None
510
def _get_compressor_settings(self):
511
if self._compressor_settings is not None:
512
return self._compressor_settings
514
if self._get_settings is not None:
515
settings = self._get_settings()
517
vf = GroupCompressVersionedFiles
518
settings = vf._DEFAULT_COMPRESSOR_SETTINGS
519
self._compressor_settings = settings
520
return self._compressor_settings
522
469
def add_factory(self, key, parents, start, end):
523
470
if not self._factories:
579
523
# block? It seems hard to come up with a method that it would
580
524
# expand, since we do full compression again. Perhaps based on a
581
525
# request that ends up poorly ordered?
582
# TODO: If the content would have expanded, then we would want to
583
# handle a case where we need to split the block.
584
# Now that we have a user-tweakable option
585
# (max_bytes_to_index), it is possible that one person set it
586
# to a very low value, causing poor compression.
587
526
delta = time.time() - tstart
588
527
self._block = new_block
589
528
trace.mutter('creating new compressed block on-the-fly in %.3fs'
723
662
# 1 line for end byte
724
663
header_lines = []
725
664
for factory in self._factories:
726
key_bytes = b'\x00'.join(factory.key)
665
key_bytes = '\x00'.join(factory.key)
727
666
parents = factory.parents
728
667
if parents is None:
729
parent_bytes = b'None:'
668
parent_bytes = 'None:'
731
parent_bytes = b'\t'.join(b'\x00'.join(key) for key in parents)
732
record_header = b'%s\n%s\n%d\n%d\n' % (
670
parent_bytes = '\t'.join('\x00'.join(key) for key in parents)
671
record_header = '%s\n%s\n%d\n%d\n' % (
733
672
key_bytes, parent_bytes, factory._start, factory._end)
734
673
header_lines.append(record_header)
735
674
# TODO: Can we break the refcycle at this point and set
736
675
# factory._manager = None?
737
header_bytes = b''.join(header_lines)
676
header_bytes = ''.join(header_lines)
739
678
header_bytes_len = len(header_bytes)
740
679
z_header_bytes = zlib.compress(header_bytes)
742
681
z_header_bytes_len = len(z_header_bytes)
743
block_bytes_len, block_chunks = self._block.to_chunks()
744
lines.append(b'%d\n%d\n%d\n' % (
745
z_header_bytes_len, header_bytes_len, block_bytes_len))
682
block_bytes = self._block.to_bytes()
683
lines.append('%d\n%d\n%d\n' % (z_header_bytes_len, header_bytes_len,
746
685
lines.append(z_header_bytes)
747
lines.extend(block_chunks)
748
del z_header_bytes, block_chunks
749
# TODO: This is a point where we will double the memory consumption. To
750
# avoid this, we probably have to switch to a 'chunked' api
751
return b''.join(lines)
686
lines.append(block_bytes)
687
del z_header_bytes, block_bytes
688
return ''.join(lines)
754
691
def from_bytes(cls, bytes):
755
692
# TODO: This does extra string copying, probably better to do it a
756
# different way. At a minimum this creates 2 copies of the
758
694
(storage_kind, z_header_len, header_len,
759
block_len, rest) = bytes.split(b'\n', 4)
695
block_len, rest) = bytes.split('\n', 4)
761
if storage_kind != b'groupcompress-block':
697
if storage_kind != 'groupcompress-block':
762
698
raise ValueError('Unknown storage kind: %s' % (storage_kind,))
763
699
z_header_len = int(z_header_len)
764
700
if len(rest) < z_header_len:
787
723
block = GroupCompressBlock.from_bytes(block_bytes)
789
725
result = cls(block)
790
for start in range(0, len(header_lines), 4):
726
for start in xrange(0, len(header_lines), 4):
792
key = tuple(header_lines[start].split(b'\x00'))
793
parents_line = header_lines[start + 1]
794
if parents_line == b'None:':
728
key = tuple(header_lines[start].split('\x00'))
729
parents_line = header_lines[start+1]
730
if parents_line == 'None:':
797
parents = tuple([tuple(segment.split(b'\x00'))
798
for segment in parents_line.split(b'\t')
800
start_offset = int(header_lines[start + 2])
801
end_offset = int(header_lines[start + 3])
733
parents = tuple([tuple(segment.split('\x00'))
734
for segment in parents_line.split('\t')
736
start_offset = int(header_lines[start+2])
737
end_offset = int(header_lines[start+3])
802
738
result.add_factory(key, parents, start_offset, end_offset)
813
749
class _CommonGroupCompressor(object):
815
def __init__(self, settings=None):
816
752
"""Create a GroupCompressor."""
818
754
self._last = None
819
755
self.endpoint = 0
820
756
self.input_bytes = 0
821
757
self.labels_deltas = {}
822
self._delta_index = None # Set by the children
758
self._delta_index = None # Set by the children
823
759
self._block = GroupCompressBlock()
827
self._settings = settings
829
761
def compress(self, key, bytes, expected_sha, nostore_sha=None, soft=False):
830
762
"""Compress lines with label key.
832
764
:param key: A key tuple. It is stored in the output
833
765
for identification of the text during decompression. If the last
834
element is b'None' it is replaced with the sha1 of the text -
766
element is 'None' it is replaced with the sha1 of the text -
835
767
e.g. sha1:xxxxxxx.
836
768
:param bytes: The bytes to be compressed
837
769
:param expected_sha: If non-None, the sha the lines are believed to
848
780
:seealso VersionedFiles.add_lines:
850
if not bytes: # empty, like a dir entry, etc
782
if not bytes: # empty, like a dir entry, etc
851
783
if nostore_sha == _null_sha1:
852
784
raise errors.ExistingContent()
853
785
return _null_sha1, 0, 0, 'fulltext'
860
792
if sha1 == nostore_sha:
861
793
raise errors.ExistingContent()
862
794
if key[-1] is None:
863
key = key[:-1] + (b'sha1:' + sha1,)
795
key = key[:-1] + ('sha1:' + sha1,)
865
797
start, end, type = self._compress(key, bytes, len(bytes) / 2, soft)
866
798
return sha1, start, end, type
890
822
:param key: The key to extract.
891
823
:return: An iterable over bytes and the sha1.
893
(start_byte, start_chunk, end_byte,
894
end_chunk) = self.labels_deltas[key]
825
(start_byte, start_chunk, end_byte, end_chunk) = self.labels_deltas[key]
895
826
delta_chunks = self.chunks[start_chunk:end_chunk]
896
stored_bytes = b''.join(delta_chunks)
897
kind = stored_bytes[:1]
827
stored_bytes = ''.join(delta_chunks)
828
if stored_bytes[0] == 'f':
899
829
fulltext_len, offset = decode_base128_int(stored_bytes[1:10])
900
830
data_len = fulltext_len + 1 + offset
901
if data_len != len(stored_bytes):
831
if data_len != len(stored_bytes):
902
832
raise ValueError('Index claimed fulltext len, but stored bytes'
903
833
' claim %s != %s'
904
834
% (len(stored_bytes), data_len))
905
data = stored_bytes[offset + 1:]
835
bytes = stored_bytes[offset + 1:]
908
raise ValueError('Unknown content kind, bytes claim %s' % kind)
909
837
# XXX: This is inefficient at best
910
source = b''.join(self.chunks[:start_chunk])
838
source = ''.join(self.chunks[:start_chunk])
839
if stored_bytes[0] != 'd':
840
raise ValueError('Unknown content kind, bytes claim %s'
841
% (stored_bytes[0],))
911
842
delta_len, offset = decode_base128_int(stored_bytes[1:10])
912
843
data_len = delta_len + 1 + offset
913
844
if data_len != len(stored_bytes):
914
845
raise ValueError('Index claimed delta len, but stored bytes'
915
846
' claim %s != %s'
916
847
% (len(stored_bytes), data_len))
917
data = apply_delta(source, stored_bytes[offset + 1:])
918
data_sha1 = osutils.sha_string(data)
919
return data, data_sha1
848
bytes = apply_delta(source, stored_bytes[offset + 1:])
849
bytes_sha1 = osutils.sha_string(bytes)
850
return bytes, bytes_sha1
922
853
"""Finish this group, creating a formatted stream.
924
855
After calling this, the compressor should no longer be used
857
# TODO: this causes us to 'bloat' to 2x the size of content in the
858
# group. This has an impact for 'commit' of large objects.
859
# One possibility is to use self._content_chunks, and be lazy and
860
# only fill out self._content as a full string when we actually
861
# need it. That would at least drop the peak memory consumption
862
# for 'commit' down to ~1x the size of the largest file, at a
863
# cost of increased complexity within this code. 2x is still <<
864
# 3x the size of the largest file, so we are doing ok.
926
865
self._block.set_chunked_content(self.chunks, self.endpoint)
927
866
self.chunks = None
928
867
self._delta_index = None
947
886
class PythonGroupCompressor(_CommonGroupCompressor):
949
def __init__(self, settings=None):
950
889
"""Create a GroupCompressor.
952
891
Used only if the pyrex version is not available.
954
super(PythonGroupCompressor, self).__init__(settings)
893
super(PythonGroupCompressor, self).__init__()
955
894
self._delta_index = LinesDeltaIndex([])
956
895
# The actual content is managed by LinesDeltaIndex
957
896
self.chunks = self._delta_index.lines
966
905
if delta_length > max_delta_size:
967
906
# The delta is longer than the fulltext, insert a fulltext
968
907
type = 'fulltext'
969
out_lines = [b'f', encode_base128_int(input_len)]
908
out_lines = ['f', encode_base128_int(input_len)]
970
909
out_lines.extend(new_lines)
971
910
index_lines = [False, False]
972
911
index_lines.extend([True] * len(new_lines))
974
913
# this is a worthy delta, output it
977
916
# Update the delta_length to include those two encoded integers
978
917
out_lines[1] = encode_base128_int(delta_length)
979
918
# Before insertion
995
934
It contains code very similar to SequenceMatcher because of having a similar
996
935
task. However some key differences apply:
998
* there is no junk, we want a minimal edit not a human readable diff.
999
* we don't filter very common lines (because we don't know where a good
1000
range will start, and after the first text we want to be emitting minmal
1002
* we chain the left side, not the right side
1003
* we incrementally update the adjacency matrix as new lines are provided.
1004
* we look for matches in all of the left side, so the routine which does
1005
the analagous task of find_longest_match does not need to filter on the
936
- there is no junk, we want a minimal edit not a human readable diff.
937
- we don't filter very common lines (because we don't know where a good
938
range will start, and after the first text we want to be emitting minmal
940
- we chain the left side, not the right side
941
- we incrementally update the adjacency matrix as new lines are provided.
942
- we look for matches in all of the left side, so the routine which does
943
the analagous task of find_longest_match does not need to filter on the
1009
def __init__(self, settings=None):
1010
super(PyrexGroupCompressor, self).__init__(settings)
1011
max_bytes_to_index = self._settings.get('max_bytes_to_index', 0)
1012
self._delta_index = DeltaIndex(max_bytes_to_index=max_bytes_to_index)
948
super(PyrexGroupCompressor, self).__init__()
949
self._delta_index = DeltaIndex()
1014
951
def _compress(self, key, bytes, max_delta_size, soft=False):
1015
952
"""see _CommonGroupCompressor._compress"""
1025
962
# new_chunks = ['label:%s\nsha1:%s\n' % (label, sha1)]
1026
963
if self._delta_index._source_offset != self.endpoint:
1027
964
raise AssertionError('_source_offset != endpoint'
1028
' somehow the DeltaIndex got out of sync with'
1029
' the output lines')
965
' somehow the DeltaIndex got out of sync with'
1030
967
delta = self._delta_index.make_delta(bytes, max_delta_size)
1031
968
if (delta is None):
1032
969
type = 'fulltext'
1033
970
enc_length = encode_base128_int(len(bytes))
1034
971
len_mini_header = 1 + len(enc_length)
1035
972
self._delta_index.add_source(bytes, len_mini_header)
1036
new_chunks = [b'f', enc_length, bytes]
973
new_chunks = ['f', enc_length, bytes]
1039
976
enc_length = encode_base128_int(len(delta))
1040
977
len_mini_header = 1 + len(enc_length)
1041
new_chunks = [b'd', enc_length, delta]
978
new_chunks = ['d', enc_length, delta]
1042
979
self._delta_index.add_delta_source(delta, len_mini_header)
1043
980
# Before insertion
1044
981
start = self.endpoint
1051
988
self.endpoint, chunk_end)
1052
989
if not self._delta_index._source_offset == self.endpoint:
1053
990
raise AssertionError('the delta index is out of sync'
1054
'with the output lines %s != %s'
1055
% (self._delta_index._source_offset, self.endpoint))
991
'with the output lines %s != %s'
992
% (self._delta_index._source_offset, self.endpoint))
1056
993
return start, self.endpoint, type
1058
995
def _output_chunks(self, new_chunks):
1085
1022
graph_index = BTreeBuilder(reference_lists=ref_length,
1086
key_elements=keylength)
1023
key_elements=keylength)
1087
1024
stream = transport.open_write_stream('newpack')
1088
1025
writer = pack.ContainerWriter(stream.write)
1090
index = _GCGraphIndex(graph_index, lambda: True, parents=parents,
1091
add_callback=graph_index.add_nodes,
1092
inconsistency_fatal=inconsistency_fatal)
1093
access = pack_repo._DirectPackAccess({})
1027
index = _GCGraphIndex(graph_index, lambda:True, parents=parents,
1028
add_callback=graph_index.add_nodes,
1029
inconsistency_fatal=inconsistency_fatal)
1030
access = knit._DirectPackAccess({})
1094
1031
access.set_writer(writer, graph_index, (transport, 'newpack'))
1095
1032
result = GroupCompressVersionedFiles(index, access, delta)
1096
1033
result.stream = stream
1107
1044
class _BatchingBlockFetcher(object):
1108
1045
"""Fetch group compress blocks in batches.
1110
1047
:ivar total_bytes: int of expected number of bytes needed to fetch the
1111
1048
currently pending batch.
1114
def __init__(self, gcvf, locations, get_compressor_settings=None):
1051
def __init__(self, gcvf, locations):
1115
1052
self.gcvf = gcvf
1116
1053
self.locations = locations
1200
1136
memos_to_get_stack.pop()
1202
1138
block = self.batch_memos[read_memo]
1203
self.manager = _LazyGroupContentManager(block,
1204
get_compressor_settings=self._get_compressor_settings)
1139
self.manager = _LazyGroupContentManager(block)
1205
1140
self.last_read_memo = read_memo
1206
1141
start, end = index_memo[3:5]
1207
1142
self.manager.add_factory(key, parents, start, end)
1214
1149
self.total_bytes = 0
1217
class GroupCompressVersionedFiles(VersionedFilesWithFallbacks):
1152
class GroupCompressVersionedFiles(VersionedFiles):
1218
1153
"""A group-compress based VersionedFiles implementation."""
1220
# This controls how the GroupCompress DeltaIndex works. Basically, we
1221
# compute hash pointers into the source blocks (so hash(text) => text).
1222
# However each of these references costs some memory in trade against a
1223
# more accurate match result. For very large files, they either are
1224
# pre-compressed and change in bulk whenever they change, or change in just
1225
# local blocks. Either way, 'improved resolution' is not very helpful,
1226
# versus running out of memory trying to track everything. The default max
1227
# gives 100% sampling of a 1MB file.
1228
_DEFAULT_MAX_BYTES_TO_INDEX = 1024 * 1024
1229
_DEFAULT_COMPRESSOR_SETTINGS = {'max_bytes_to_index':
1230
_DEFAULT_MAX_BYTES_TO_INDEX}
1232
def __init__(self, index, access, delta=True, _unadded_refs=None,
1155
def __init__(self, index, access, delta=True, _unadded_refs=None):
1234
1156
"""Create a GroupCompressVersionedFiles object.
1236
1158
:param index: The index object storing access and graph data.
1237
1159
:param access: The access object storing raw data.
1238
1160
:param delta: Whether to delta compress or just entropy compress.
1239
1161
:param _unadded_refs: private parameter, don't use.
1240
:param _group_cache: private parameter, don't use.
1242
1163
self._index = index
1243
1164
self._access = access
1245
1166
if _unadded_refs is None:
1246
1167
_unadded_refs = {}
1247
1168
self._unadded_refs = _unadded_refs
1248
if _group_cache is None:
1249
_group_cache = LRUSizeCache(max_size=50 * 1024 * 1024)
1250
self._group_cache = _group_cache
1251
self._immediate_fallback_vfs = []
1252
self._max_bytes_to_index = None
1169
self._group_cache = LRUSizeCache(max_size=50*1024*1024)
1170
self._fallback_vfs = []
1254
1172
def without_fallbacks(self):
1255
1173
"""Return a clone of this object without any fallbacks configured."""
1256
1174
return GroupCompressVersionedFiles(self._index, self._access,
1257
self._delta, _unadded_refs=dict(
1258
self._unadded_refs),
1259
_group_cache=self._group_cache)
1175
self._delta, _unadded_refs=dict(self._unadded_refs))
1261
1177
def add_lines(self, key, parents, lines, parent_texts=None,
1262
left_matching_blocks=None, nostore_sha=None, random_id=False,
1263
check_content=True):
1178
left_matching_blocks=None, nostore_sha=None, random_id=False,
1179
check_content=True):
1264
1180
"""Add a text to the store.
1266
1182
:param key: The key tuple of the text to add.
1267
1183
:param parents: The parents key tuples of the text to add.
1268
1184
:param lines: A list of lines. Each line must be a bytestring. And all
1269
of them except the last must be terminated with \\n and contain no
1270
other \\n's. The last line may either contain no \\n's or a single
1271
terminating \\n. If the lines list does meet this constraint the
1272
add routine may error or may succeed - but you will be unable to
1273
read the data back accurately. (Checking the lines have been split
1185
of them except the last must be terminated with \n and contain no
1186
other \n's. The last line may either contain no \n's or a single
1187
terminating \n. If the lines list does meet this constraint the add
1188
routine may error or may succeed - but you will be unable to read
1189
the data back accurately. (Checking the lines have been split
1274
1190
correctly is expensive and extremely unlikely to catch bugs so it
1275
1191
is not done at runtime unless check_content is True.)
1276
1192
:param parent_texts: An optional dictionary containing the opaque
1308
1224
nostore_sha=nostore_sha))[0]
1309
1225
return sha1, length, None
1227
def _add_text(self, key, parents, text, nostore_sha=None, random_id=False):
1228
"""See VersionedFiles._add_text()."""
1229
self._index._check_write_ok()
1230
self._check_add(key, None, random_id, check_content=False)
1231
if text.__class__ is not str:
1232
raise errors.BzrBadParameterUnicode("text")
1234
# The caller might pass None if there is no graph data, but kndx
1235
# indexes can't directly store that, so we give them
1236
# an empty tuple instead.
1238
# double handling for now. Make it work until then.
1240
record = FulltextContentFactory(key, parents, None, text)
1241
sha1 = list(self._insert_record_stream([record], random_id=random_id,
1242
nostore_sha=nostore_sha))[0]
1243
return sha1, length, None
1311
1245
def add_fallback_versioned_files(self, a_versioned_files):
1312
1246
"""Add a source of texts for texts not present in this knit.
1314
1248
:param a_versioned_files: A VersionedFiles object.
1316
self._immediate_fallback_vfs.append(a_versioned_files)
1250
self._fallback_vfs.append(a_versioned_files)
1318
1252
def annotate(self, key):
1319
1253
"""See VersionedFiles.annotate."""
1353
1287
self._check_lines_not_unicode(lines)
1354
1288
self._check_lines_are_lines(lines)
1290
def get_known_graph_ancestry(self, keys):
1291
"""Get a KnownGraph instance with the ancestry of keys."""
1292
# Note that this is identical to
1293
# KnitVersionedFiles.get_known_graph_ancestry, but they don't share
1295
parent_map, missing_keys = self._index.find_ancestry(keys)
1296
for fallback in self._fallback_vfs:
1297
if not missing_keys:
1299
(f_parent_map, f_missing_keys) = fallback._index.find_ancestry(
1301
parent_map.update(f_parent_map)
1302
missing_keys = f_missing_keys
1303
kg = _mod_graph.KnownGraph(parent_map)
1356
1306
def get_parent_map(self, keys):
1357
1307
"""Get a map of the graph parents of keys.
1461
1411
keys = set(remaining_keys)
1462
1412
for content_factory in self._get_remaining_record_stream(keys,
1463
orig_keys, ordering, include_delta_closure):
1413
orig_keys, ordering, include_delta_closure):
1464
1414
remaining_keys.discard(content_factory.key)
1465
1415
yield content_factory
1467
except errors.RetryWithNewPacks as e:
1417
except errors.RetryWithNewPacks, e:
1468
1418
self._access.reload_or_raise(e)
1470
1420
def _find_from_fallback(self, missing):
1497
1447
The returned objects should be in the order defined by 'ordering',
1498
1448
which can weave between different sources.
1500
1449
:param ordering: Must be one of 'topological' or 'groupcompress'
1501
1450
:return: List of [(source, [keys])] tuples, such that all keys are in
1502
1451
the defined order, regardless of source.
1504
1453
if ordering == 'topological':
1505
present_keys = tsort.topo_sort(parent_map)
1454
present_keys = topo_sort(parent_map)
1507
1456
# ordering == 'groupcompress'
1508
1457
# XXX: This only optimizes for the target ordering. We may need
1544
1493
# This is the group the bytes are stored in, followed by the
1545
1494
# location in the group
1546
1495
return locations[key][0]
1496
present_keys = sorted(locations.iterkeys(), key=get_group)
1547
1497
# We don't have an ordering for keys in the in-memory object, but
1548
1498
# lets process the in-memory ones first.
1549
present_keys = list(unadded_keys)
1550
present_keys.extend(sorted(locations, key=get_group))
1499
present_keys = list(unadded_keys) + present_keys
1551
1500
# Now grab all of the ones from other sources
1552
1501
source_keys = [(self, present_keys)]
1553
1502
source_keys.extend(source_result)
1585
1534
key_to_source_map)
1586
1535
elif ordering == 'as-requested':
1587
1536
source_keys = self._get_as_requested_source_keys(orig_keys,
1588
locations, unadded_keys, key_to_source_map)
1537
locations, unadded_keys, key_to_source_map)
1590
1539
# We want to yield the keys in a semi-optimal (read-wise) ordering.
1591
1540
# Otherwise we thrash the _group_cache and destroy performance
1592
1541
source_keys = self._get_io_ordered_source_keys(locations,
1593
unadded_keys, source_result)
1542
unadded_keys, source_result)
1594
1543
for key in missing:
1595
1544
yield AbsentContentFactory(key)
1596
1545
# Batch up as many keys as we can until either:
1597
1546
# - we encounter an unadded ref, or
1598
1547
# - we run out of keys, or
1599
1548
# - the total bytes to retrieve for this batch > BATCH_SIZE
1600
batcher = _BatchingBlockFetcher(self, locations,
1601
get_compressor_settings=self._get_compressor_settings)
1549
batcher = _BatchingBlockFetcher(self, locations)
1602
1550
for source, keys in source_keys:
1603
1551
if source is self:
1604
1552
for key in keys:
1650
1598
for _ in self._insert_record_stream(stream, random_id=False):
1653
def _get_compressor_settings(self):
1654
if self._max_bytes_to_index is None:
1655
# TODO: VersionedFiles don't know about their containing
1656
# repository, so they don't have much of an idea about their
1657
# location. So for now, this is only a global option.
1658
c = config.GlobalConfig()
1659
val = c.get_user_option('bzr.groupcompress.max_bytes_to_index')
1663
except ValueError as e:
1664
trace.warning('Value for '
1665
'"bzr.groupcompress.max_bytes_to_index"'
1666
' %r is not an integer'
1670
val = self._DEFAULT_MAX_BYTES_TO_INDEX
1671
self._max_bytes_to_index = val
1672
return {'max_bytes_to_index': self._max_bytes_to_index}
1674
def _make_group_compressor(self):
1675
return GroupCompressor(self._get_compressor_settings())
1677
1601
def _insert_record_stream(self, stream, random_id=False, nostore_sha=None,
1678
1602
reuse_blocks=True):
1679
1603
"""Internal core to insert a record stream into this container.
1704
1627
# This will go up to fulltexts for gc to gc fetching, which isn't
1706
self._compressor = self._make_group_compressor()
1629
self._compressor = GroupCompressor()
1707
1630
self._unadded_refs = {}
1708
1631
keys_to_add = []
1711
bytes_len, chunks = self._compressor.flush().to_chunks()
1712
self._compressor = self._make_group_compressor()
1713
# Note: At this point we still have 1 copy of the fulltext (in
1714
# record and the var 'bytes'), and this generates 2 copies of
1715
# the compressed text (one for bytes, one in chunks)
1716
# TODO: Push 'chunks' down into the _access api, so that we don't
1717
# have to double compressed memory here
1718
# TODO: Figure out how to indicate that we would be happy to free
1719
# the fulltext content at this point. Note that sometimes we
1720
# will want it later (streaming CHK pages), but most of the
1721
# time we won't (everything else)
1722
data = b''.join(chunks)
1633
bytes = self._compressor.flush().to_bytes()
1634
self._compressor = GroupCompressor()
1724
1635
index, start, length = self._access.add_raw_records(
1725
[(None, len(data))], data)[0]
1636
[(None, len(bytes))], bytes)[0]
1727
1638
for key, reads, refs in keys_to_add:
1728
nodes.append((key, b"%d %d %s" % (start, length, reads), refs))
1639
nodes.append((key, "%d %d %s" % (start, length, reads), refs))
1729
1640
self._index.add_records(nodes, random_id=random_id)
1730
1641
self._unadded_refs = {}
1731
1642
del keys_to_add[:]
1745
1656
raise errors.RevisionNotPresent(record.key, self)
1747
1658
if record.key in inserted_keys:
1748
trace.note(gettext('Insert claimed random_id=True,'
1749
' but then inserted %r two times'), record.key)
1659
trace.note('Insert claimed random_id=True,'
1660
' but then inserted %r two times', record.key)
1751
1662
inserted_keys.add(record.key)
1752
1663
if reuse_blocks:
1778
1689
raise AssertionError('No insert_manager set')
1779
1690
if insert_manager is not record._manager:
1780
1691
raise AssertionError('insert_manager does not match'
1781
' the current record, we cannot be positive'
1782
' that the appropriate content was inserted.'
1784
value = b"%d %d %d %d" % (block_start, block_length,
1785
record._start, record._end)
1692
' the current record, we cannot be positive'
1693
' that the appropriate content was inserted.'
1695
value = "%d %d %d %d" % (block_start, block_length,
1696
record._start, record._end)
1786
1697
nodes = [(record.key, value, (record.parents,))]
1787
1698
# TODO: Consider buffering up many nodes to be added, not
1788
1699
# sure how much overhead this has, but we're seeing
1811
1722
# delta_ratio = float(len(bytes)) / (end_point - start_point)
1812
1723
# Check if we want to continue to include that text
1813
1724
if (prefix == max_fulltext_prefix
1814
and end_point < 2 * max_fulltext_len):
1725
and end_point < 2 * max_fulltext_len):
1815
1726
# As long as we are on the same file_id, we will fill at least
1816
1727
# 2 * max_fulltext_len
1817
1728
start_new_block = False
1818
elif end_point > 4 * 1024 * 1024:
1729
elif end_point > 4*1024*1024:
1819
1730
start_new_block = True
1820
1731
elif (prefix is not None and prefix != last_prefix
1821
and end_point > 2 * 1024 * 1024):
1732
and end_point > 2*1024*1024):
1822
1733
start_new_block = True
1824
1735
start_new_block = False
1892
1802
"""See VersionedFiles.keys."""
1893
1803
if 'evil' in debug.debug_flags:
1894
1804
trace.mutter_callsite(2, "keys scales with size of history")
1895
sources = [self._index] + self._immediate_fallback_vfs
1805
sources = [self._index] + self._fallback_vfs
1897
1807
for source in sources:
1898
1808
result.update(source.keys())
1902
class _GCBuildDetails(object):
1903
"""A blob of data about the build details.
1905
This stores the minimal data, which then allows compatibility with the old
1906
api, without taking as much memory.
1909
__slots__ = ('_index', '_group_start', '_group_end', '_basis_end',
1910
'_delta_end', '_parents')
1913
compression_parent = None
1915
def __init__(self, parents, position_info):
1916
self._parents = parents
1917
(self._index, self._group_start, self._group_end, self._basis_end,
1918
self._delta_end) = position_info
1921
return '%s(%s, %s)' % (self.__class__.__name__,
1922
self.index_memo, self._parents)
1925
def index_memo(self):
1926
return (self._index, self._group_start, self._group_end,
1927
self._basis_end, self._delta_end)
1930
def record_details(self):
1931
return static_tuple.StaticTuple(self.method, None)
1933
def __getitem__(self, offset):
1934
"""Compatibility thunk to act like a tuple."""
1936
return self.index_memo
1938
return self.compression_parent # Always None
1940
return self._parents
1942
return self.record_details
1944
raise IndexError('offset out of range')
1950
1812
class _GCGraphIndex(object):
1951
1813
"""Mapper from GroupCompressVersionedFiles needs into GraphIndex storage."""
1953
1815
def __init__(self, graph_index, is_locked, parents=True,
1954
add_callback=None, track_external_parent_refs=False,
1955
inconsistency_fatal=True, track_new_keys=False):
1816
add_callback=None, track_external_parent_refs=False,
1817
inconsistency_fatal=True, track_new_keys=False):
1956
1818
"""Construct a _GCGraphIndex on a graph_index.
1958
:param graph_index: An implementation of breezy.index.GraphIndex.
1820
:param graph_index: An implementation of bzrlib.index.GraphIndex.
1959
1821
:param is_locked: A callback, returns True if the index is locked and
1961
1823
:param parents: If True, record knits parents, if not do not record
2027
1889
if node_refs != passed[1]:
2028
1890
details = '%s %s %s' % (key, (value, node_refs), passed)
2029
1891
if self._inconsistency_fatal:
2030
raise knit.KnitCorrupt(self, "inconsistent details"
2031
" in add_records: %s" %
1892
raise errors.KnitCorrupt(self, "inconsistent details"
1893
" in add_records: %s" %
2034
1896
trace.warning("inconsistent details in skipped"
2035
1897
" record: %s", details)
2040
1902
if self._parents:
2041
for key, (value, node_refs) in viewitems(keys):
1903
for key, (value, node_refs) in keys.iteritems():
2042
1904
result.append((key, value, node_refs))
2044
for key, (value, node_refs) in viewitems(keys):
1906
for key, (value, node_refs) in keys.iteritems():
2045
1907
result.append((key, value))
2046
1908
records = result
2047
1909
key_dependencies = self._key_dependencies
2127
1989
:param keys: An iterable of keys.
2128
1990
:return: A dict of key:
2129
1991
(index_memo, compression_parent, parents, record_details).
2131
* index_memo: opaque structure to pass to read_records to extract
2133
* compression_parent: Content that this record is built upon, may
2135
* parents: Logical parents of this node
2136
* record_details: extra information about the content which needs
2137
to be passed to Factory.parse_record
1993
opaque structure to pass to read_records to extract the raw
1996
Content that this record is built upon, may be None
1998
Logical parents of this node
2000
extra information about the content which needs to be passed to
2001
Factory.parse_record
2139
2003
self._check_read()
2168
2033
# each, or about 7MB. Note that it might be even more when you consider
2169
2034
# how PyInt is allocated in separate slabs. And you can't return a slab
2170
2035
# to the OS if even 1 int on it is in use. Note though that Python uses
2171
# a LIFO when re-using PyInt slots, which might cause more
2036
# a LIFO when re-using PyInt slots, which probably causes more
2172
2037
# fragmentation.
2173
2038
start = int(bits[0])
2174
2039
start = self._int_cache.setdefault(start, start)
added
removed
bzrlib/groupcompress.py
