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(), """
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 (
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 (
42
58
AbsentContentFactory,
43
59
ChunkedContentFactory,
44
60
FulltextContentFactory,
61
VersionedFilesWithFallbacks,
48
64
# Minimum number of uncompressed bytes to try fetch at once when retrieving
49
65
# groupcompress blocks.
52
_USE_LZMA = False and (pylzma is not None)
68
# osutils.sha_string(b'')
69
_null_sha1 = b'da39a3ee5e6b4b0d3255bfef95601890afd80709'
54
# osutils.sha_string('')
55
_null_sha1 = 'da39a3ee5e6b4b0d3255bfef95601890afd80709'
57
72
def sort_gc_optimal(parent_map):
58
73
"""Sort and group the keys in parent_map into groupcompress order.
79
94
for prefix in sorted(per_prefix_map):
80
present_keys.extend(reversed(topo_sort(per_prefix_map[prefix])))
95
present_keys.extend(reversed(tsort.topo_sort(per_prefix_map[prefix])))
81
96
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)
84
111
# The max zlib window size is 32kB, so if we set 'max_size' output of the
85
112
# decompressor to the requested bytes + 32kB, then we should guarantee
86
113
# num_bytes coming out.
87
_ZLIB_DECOMP_WINDOW = 32*1024
114
_ZLIB_DECOMP_WINDOW = 32 * 1024
89
117
class GroupCompressBlock(object):
90
118
"""An object which maintains the internal structure of the compressed data.
95
123
# Group Compress Block v1 Zlib
96
GCB_HEADER = 'gcb1z\n'
124
GCB_HEADER = b'gcb1z\n'
97
125
# Group Compress Block v1 Lzma
98
GCB_LZ_HEADER = 'gcb1l\n'
126
GCB_LZ_HEADER = b'gcb1l\n'
99
127
GCB_KNOWN_HEADERS = (GCB_HEADER, GCB_LZ_HEADER)
101
129
def __init__(self):
102
130
# map by key? or just order in file?
103
131
self._compressor_name = None
104
self._z_content = None
132
self._z_content_chunks = None
105
133
self._z_content_decompressor = None
106
134
self._z_content_length = None
107
135
self._content_length = None
132
160
# Expand the content if required
133
161
if self._content is None:
134
162
if self._content_chunks is not None:
135
self._content = ''.join(self._content_chunks)
163
self._content = b''.join(self._content_chunks)
136
164
self._content_chunks = None
137
165
if self._content is None:
138
if self._z_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:
139
170
raise AssertionError('No content to decompress')
140
if self._z_content == '':
171
z_content = b''.join(self._z_content_chunks)
142
174
elif self._compressor_name == 'lzma':
143
175
# We don't do partial lzma decomp yet
144
self._content = pylzma.decompress(self._z_content)
177
self._content = pylzma.decompress(z_content)
145
178
elif self._compressor_name == 'zlib':
146
179
# Start a zlib decompressor
147
180
if num_bytes * 4 > self._content_length * 3:
148
181
# If we are requesting more that 3/4ths of the content,
149
182
# just extract the whole thing in a single pass
150
183
num_bytes = self._content_length
151
self._content = zlib.decompress(self._z_content)
184
self._content = zlib.decompress(z_content)
153
186
self._z_content_decompressor = zlib.decompressobj()
154
187
# Seed the decompressor with the uncompressed bytes, so
155
188
# that the rest of the code is simplified
156
189
self._content = self._z_content_decompressor.decompress(
157
self._z_content, num_bytes + _ZLIB_DECOMP_WINDOW)
190
z_content, num_bytes + _ZLIB_DECOMP_WINDOW)
158
191
if not self._z_content_decompressor.unconsumed_tail:
159
192
self._z_content_decompressor = None
197
230
# At present, we have 2 integers for the compressed and uncompressed
198
231
# content. In base10 (ascii) 14 bytes can represent > 1TB, so to avoid
199
232
# checking too far, cap the search to 14 bytes.
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):
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):
207
240
# XXX: Define some GCCorrupt error ?
208
241
raise AssertionError('Invalid bytes: (%d) != %d + %d' %
209
(len(bytes), pos, self._z_content_length))
210
self._z_content = bytes[pos:]
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)
213
256
def from_bytes(cls, bytes):
215
if bytes[:6] not in cls.GCB_KNOWN_HEADERS:
259
if header not in cls.GCB_KNOWN_HEADERS:
216
260
raise ValueError('bytes did not start with any of %r'
217
261
% (cls.GCB_KNOWN_HEADERS,))
218
# XXX: why not testing the whole header ?
262
if header == cls.GCB_HEADER:
220
263
out._compressor_name = 'zlib'
221
elif bytes[4] == 'l':
264
elif header == cls.GCB_LZ_HEADER:
222
265
out._compressor_name = 'lzma'
224
raise ValueError('unknown compressor: %r' % (bytes,))
267
raise ValueError('unknown compressor: %r' % (header,))
225
268
out._parse_bytes(bytes, 6)
233
276
:return: The bytes for the content
235
278
if start == end == 0:
237
280
self._ensure_content(end)
238
281
# The bytes are 'f' or 'd' for the type, then a variable-length
239
282
# base128 integer for the content size, then the actual content
240
283
# We know that the variable-length integer won't be longer than 5
241
284
# bytes (it takes 5 bytes to encode 2^32)
242
c = self._content[start]
285
c = self._content[start:start + 1]
244
287
type = 'fulltext'
247
290
raise ValueError('Unknown content control code: %s'
250
293
content_len, len_len = decode_base128_int(
251
self._content[start + 1:start + 6])
294
self._content[start + 1:start + 6])
252
295
content_start = start + 1 + len_len
253
296
if end != content_start + content_len:
254
297
raise ValueError('end != len according to field header'
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)
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)
262
304
def set_chunked_content(self, content_chunks, length):
263
305
"""Set the content of this block to the given chunks."""
269
311
self._content_length = length
270
312
self._content_chunks = content_chunks
271
313
self._content = None
272
self._z_content = None
314
self._z_content_chunks = None
274
316
def set_content(self, content):
275
317
"""Set the content of this block."""
276
318
self._content_length = len(content)
277
319
self._content = content
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):
320
self._z_content_chunks = None
322
def _create_z_content_from_chunks(self, chunks):
290
323
compressor = zlib.compressobj(zlib.Z_DEFAULT_COMPRESSION)
291
compressed_chunks = map(compressor.compress, self._content_chunks)
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))
292
327
compressed_chunks.append(compressor.flush())
293
self._z_content = ''.join(compressed_chunks)
294
self._z_content_length = len(self._z_content)
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))
296
332
def _create_z_content(self):
297
if self._z_content is not None:
300
self._create_z_content_using_lzma()
333
if self._z_content_chunks is not None:
302
335
if self._content_chunks is not None:
303
self._create_z_content_from_chunks()
305
self._z_content = zlib.compress(self._content)
306
self._z_content_length = len(self._z_content)
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
308
352
def to_bytes(self):
309
353
"""Encode the information into a byte stream."""
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)
354
total_len, chunks = self.to_chunks()
355
return b''.join(chunks)
321
357
def _dump(self, include_text=False):
322
358
"""Take this block, and spit out a human-readable structure.
334
370
while pos < self._content_length:
335
kind = self._content[pos]
371
kind = self._content[pos:pos + 1]
337
if kind not in ('f', 'd'):
373
if kind not in (b'f', b'd'):
338
374
raise ValueError('invalid kind character: %r' % (kind,))
339
375
content_len, len_len = decode_base128_int(
340
self._content[pos:pos + 5])
376
self._content[pos:pos + 5])
342
378
if content_len + pos > self._content_length:
343
379
raise ValueError('invalid content_len %d for record @ pos %d'
344
380
% (content_len, pos - len_len - 1))
345
if kind == 'f': # Fulltext
381
if kind == b'f': # Fulltext
347
text = self._content[pos:pos+content_len]
348
result.append(('f', content_len, text))
383
text = self._content[pos:pos + content_len]
384
result.append((b'f', content_len, text))
350
result.append(('f', content_len))
351
elif kind == 'd': # Delta
352
delta_content = self._content[pos:pos+content_len]
386
result.append((b'f', content_len))
387
elif kind == b'd': # Delta
388
delta_content = self._content[pos:pos + content_len]
354
390
# The first entry in a delta is the decompressed length
355
391
decomp_len, delta_pos = decode_base128_int(delta_content)
356
result.append(('d', content_len, decomp_len, delta_info))
392
result.append((b'd', content_len, decomp_len, delta_info))
358
394
while delta_pos < content_len:
359
c = ord(delta_content[delta_pos])
395
c = indexbytes(delta_content, delta_pos)
363
399
delta_pos) = decode_copy_instruction(delta_content, c,
366
text = self._content[offset:offset+length]
367
delta_info.append(('c', offset, length, text))
402
text = self._content[offset:offset + length]
403
delta_info.append((b'c', offset, length, text))
369
delta_info.append(('c', offset, length))
405
delta_info.append((b'c', offset, length))
370
406
measured_len += length
373
txt = delta_content[delta_pos:delta_pos+c]
409
txt = delta_content[delta_pos:delta_pos + c]
376
delta_info.append(('i', c, txt))
412
delta_info.append((b'i', c, txt))
377
413
measured_len += c
379
415
if delta_pos != content_len:
429
465
# wire bytes, something...
430
466
return self._manager._wire_bytes()
433
469
if storage_kind in ('fulltext', 'chunked'):
434
470
if self._bytes is None:
435
471
# Grab and cache the raw bytes for this entry
436
472
# and break the ref-cycle with _manager since we don't need it
438
self._manager._prepare_for_extract()
475
self._manager._prepare_for_extract()
476
except zlib.error as value:
477
raise DecompressCorruption("zlib: " + str(value))
439
478
block = self._manager._block
440
479
self._bytes = block.extract(self.key, self._start, self._end)
441
480
# There are code paths that first extract as fulltext, and then
452
491
class _LazyGroupContentManager(object):
453
492
"""This manages a group of _LazyGroupCompressFactory objects."""
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
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
463
def __init__(self, block):
502
def __init__(self, block, get_compressor_settings=None):
464
503
self._block = block
465
504
# We need to preserve the ordering
466
505
self._factories = []
467
506
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
469
522
def add_factory(self, key, parents, start, end):
470
523
if not self._factories:
523
579
# block? It seems hard to come up with a method that it would
524
580
# expand, since we do full compression again. Perhaps based on a
525
581
# 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.
526
587
delta = time.time() - tstart
527
588
self._block = new_block
528
589
trace.mutter('creating new compressed block on-the-fly in %.3fs'
662
723
# 1 line for end byte
663
724
header_lines = []
664
725
for factory in self._factories:
665
key_bytes = '\x00'.join(factory.key)
726
key_bytes = b'\x00'.join(factory.key)
666
727
parents = factory.parents
667
728
if parents is None:
668
parent_bytes = 'None:'
729
parent_bytes = b'None:'
670
parent_bytes = '\t'.join('\x00'.join(key) for key in parents)
671
record_header = '%s\n%s\n%d\n%d\n' % (
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' % (
672
733
key_bytes, parent_bytes, factory._start, factory._end)
673
734
header_lines.append(record_header)
674
735
# TODO: Can we break the refcycle at this point and set
675
736
# factory._manager = None?
676
header_bytes = ''.join(header_lines)
737
header_bytes = b''.join(header_lines)
678
739
header_bytes_len = len(header_bytes)
679
740
z_header_bytes = zlib.compress(header_bytes)
681
742
z_header_bytes_len = len(z_header_bytes)
682
block_bytes = self._block.to_bytes()
683
lines.append('%d\n%d\n%d\n' % (z_header_bytes_len, header_bytes_len,
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))
685
746
lines.append(z_header_bytes)
686
lines.append(block_bytes)
687
del z_header_bytes, block_bytes
688
return ''.join(lines)
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)
691
754
def from_bytes(cls, bytes):
692
755
# TODO: This does extra string copying, probably better to do it a
756
# different way. At a minimum this creates 2 copies of the
694
758
(storage_kind, z_header_len, header_len,
695
block_len, rest) = bytes.split('\n', 4)
759
block_len, rest) = bytes.split(b'\n', 4)
697
if storage_kind != 'groupcompress-block':
761
if storage_kind != b'groupcompress-block':
698
762
raise ValueError('Unknown storage kind: %s' % (storage_kind,))
699
763
z_header_len = int(z_header_len)
700
764
if len(rest) < z_header_len:
723
787
block = GroupCompressBlock.from_bytes(block_bytes)
725
789
result = cls(block)
726
for start in xrange(0, len(header_lines), 4):
790
for start in range(0, len(header_lines), 4):
728
key = tuple(header_lines[start].split('\x00'))
729
parents_line = header_lines[start+1]
730
if parents_line == 'None:':
792
key = tuple(header_lines[start].split(b'\x00'))
793
parents_line = header_lines[start + 1]
794
if parents_line == b'None:':
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])
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])
738
802
result.add_factory(key, parents, start_offset, end_offset)
749
813
class _CommonGroupCompressor(object):
815
def __init__(self, settings=None):
752
816
"""Create a GroupCompressor."""
754
818
self._last = None
755
819
self.endpoint = 0
756
820
self.input_bytes = 0
757
821
self.labels_deltas = {}
758
self._delta_index = None # Set by the children
822
self._delta_index = None # Set by the children
759
823
self._block = GroupCompressBlock()
827
self._settings = settings
761
829
def compress(self, key, bytes, expected_sha, nostore_sha=None, soft=False):
762
830
"""Compress lines with label key.
764
832
:param key: A key tuple. It is stored in the output
765
833
for identification of the text during decompression. If the last
766
element is 'None' it is replaced with the sha1 of the text -
834
element is b'None' it is replaced with the sha1 of the text -
767
835
e.g. sha1:xxxxxxx.
768
836
:param bytes: The bytes to be compressed
769
837
:param expected_sha: If non-None, the sha the lines are believed to
780
848
:seealso VersionedFiles.add_lines:
782
if not bytes: # empty, like a dir entry, etc
850
if not bytes: # empty, like a dir entry, etc
783
851
if nostore_sha == _null_sha1:
784
852
raise errors.ExistingContent()
785
853
return _null_sha1, 0, 0, 'fulltext'
792
860
if sha1 == nostore_sha:
793
861
raise errors.ExistingContent()
794
862
if key[-1] is None:
795
key = key[:-1] + ('sha1:' + sha1,)
863
key = key[:-1] + (b'sha1:' + sha1,)
797
865
start, end, type = self._compress(key, bytes, len(bytes) / 2, soft)
798
866
return sha1, start, end, type
822
890
:param key: The key to extract.
823
891
:return: An iterable over bytes and the sha1.
825
(start_byte, start_chunk, end_byte, end_chunk) = self.labels_deltas[key]
893
(start_byte, start_chunk, end_byte,
894
end_chunk) = self.labels_deltas[key]
826
895
delta_chunks = self.chunks[start_chunk:end_chunk]
827
stored_bytes = ''.join(delta_chunks)
828
if stored_bytes[0] == 'f':
896
stored_bytes = b''.join(delta_chunks)
897
kind = stored_bytes[:1]
829
899
fulltext_len, offset = decode_base128_int(stored_bytes[1:10])
830
900
data_len = fulltext_len + 1 + offset
831
if data_len != len(stored_bytes):
901
if data_len != len(stored_bytes):
832
902
raise ValueError('Index claimed fulltext len, but stored bytes'
833
903
' claim %s != %s'
834
904
% (len(stored_bytes), data_len))
835
bytes = stored_bytes[offset + 1:]
905
data = stored_bytes[offset + 1:]
908
raise ValueError('Unknown content kind, bytes claim %s' % kind)
837
909
# XXX: This is inefficient at best
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],))
910
source = b''.join(self.chunks[:start_chunk])
842
911
delta_len, offset = decode_base128_int(stored_bytes[1:10])
843
912
data_len = delta_len + 1 + offset
844
913
if data_len != len(stored_bytes):
845
914
raise ValueError('Index claimed delta len, but stored bytes'
846
915
' claim %s != %s'
847
916
% (len(stored_bytes), data_len))
848
bytes = apply_delta(source, stored_bytes[offset + 1:])
849
bytes_sha1 = osutils.sha_string(bytes)
850
return bytes, bytes_sha1
917
data = apply_delta(source, stored_bytes[offset + 1:])
918
data_sha1 = osutils.sha_string(data)
919
return data, data_sha1
853
922
"""Finish this group, creating a formatted stream.
855
924
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.
865
926
self._block.set_chunked_content(self.chunks, self.endpoint)
866
927
self.chunks = None
867
928
self._delta_index = None
886
947
class PythonGroupCompressor(_CommonGroupCompressor):
949
def __init__(self, settings=None):
889
950
"""Create a GroupCompressor.
891
952
Used only if the pyrex version is not available.
893
super(PythonGroupCompressor, self).__init__()
954
super(PythonGroupCompressor, self).__init__(settings)
894
955
self._delta_index = LinesDeltaIndex([])
895
956
# The actual content is managed by LinesDeltaIndex
896
957
self.chunks = self._delta_index.lines
905
966
if delta_length > max_delta_size:
906
967
# The delta is longer than the fulltext, insert a fulltext
907
968
type = 'fulltext'
908
out_lines = ['f', encode_base128_int(input_len)]
969
out_lines = [b'f', encode_base128_int(input_len)]
909
970
out_lines.extend(new_lines)
910
971
index_lines = [False, False]
911
972
index_lines.extend([True] * len(new_lines))
913
974
# this is a worthy delta, output it
916
977
# Update the delta_length to include those two encoded integers
917
978
out_lines[1] = encode_base128_int(delta_length)
918
979
# Before insertion
934
995
It contains code very similar to SequenceMatcher because of having a similar
935
996
task. However some key differences apply:
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
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
948
super(PyrexGroupCompressor, self).__init__()
949
self._delta_index = DeltaIndex()
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)
951
1014
def _compress(self, key, bytes, max_delta_size, soft=False):
952
1015
"""see _CommonGroupCompressor._compress"""
962
1025
# new_chunks = ['label:%s\nsha1:%s\n' % (label, sha1)]
963
1026
if self._delta_index._source_offset != self.endpoint:
964
1027
raise AssertionError('_source_offset != endpoint'
965
' somehow the DeltaIndex got out of sync with'
1028
' somehow the DeltaIndex got out of sync with'
1029
' the output lines')
967
1030
delta = self._delta_index.make_delta(bytes, max_delta_size)
968
1031
if (delta is None):
969
1032
type = 'fulltext'
970
1033
enc_length = encode_base128_int(len(bytes))
971
1034
len_mini_header = 1 + len(enc_length)
972
1035
self._delta_index.add_source(bytes, len_mini_header)
973
new_chunks = ['f', enc_length, bytes]
1036
new_chunks = [b'f', enc_length, bytes]
976
1039
enc_length = encode_base128_int(len(delta))
977
1040
len_mini_header = 1 + len(enc_length)
978
new_chunks = ['d', enc_length, delta]
1041
new_chunks = [b'd', enc_length, delta]
979
1042
self._delta_index.add_delta_source(delta, len_mini_header)
980
1043
# Before insertion
981
1044
start = self.endpoint
988
1051
self.endpoint, chunk_end)
989
1052
if not self._delta_index._source_offset == self.endpoint:
990
1053
raise AssertionError('the delta index is out of sync'
991
'with the output lines %s != %s'
992
% (self._delta_index._source_offset, self.endpoint))
1054
'with the output lines %s != %s'
1055
% (self._delta_index._source_offset, self.endpoint))
993
1056
return start, self.endpoint, type
995
1058
def _output_chunks(self, new_chunks):
1022
1085
graph_index = BTreeBuilder(reference_lists=ref_length,
1023
key_elements=keylength)
1086
key_elements=keylength)
1024
1087
stream = transport.open_write_stream('newpack')
1025
1088
writer = pack.ContainerWriter(stream.write)
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({})
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({})
1031
1094
access.set_writer(writer, graph_index, (transport, 'newpack'))
1032
1095
result = GroupCompressVersionedFiles(index, access, delta)
1033
1096
result.stream = stream
1044
1107
class _BatchingBlockFetcher(object):
1045
1108
"""Fetch group compress blocks in batches.
1047
1110
:ivar total_bytes: int of expected number of bytes needed to fetch the
1048
1111
currently pending batch.
1051
def __init__(self, gcvf, locations):
1114
def __init__(self, gcvf, locations, get_compressor_settings=None):
1052
1115
self.gcvf = gcvf
1053
1116
self.locations = locations
1136
1200
memos_to_get_stack.pop()
1138
1202
block = self.batch_memos[read_memo]
1139
self.manager = _LazyGroupContentManager(block)
1203
self.manager = _LazyGroupContentManager(block,
1204
get_compressor_settings=self._get_compressor_settings)
1140
1205
self.last_read_memo = read_memo
1141
1206
start, end = index_memo[3:5]
1142
1207
self.manager.add_factory(key, parents, start, end)
1149
1214
self.total_bytes = 0
1152
class GroupCompressVersionedFiles(VersionedFiles):
1217
class GroupCompressVersionedFiles(VersionedFilesWithFallbacks):
1153
1218
"""A group-compress based VersionedFiles implementation."""
1155
def __init__(self, index, access, delta=True, _unadded_refs=None):
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,
1156
1234
"""Create a GroupCompressVersionedFiles object.
1158
1236
:param index: The index object storing access and graph data.
1159
1237
:param access: The access object storing raw data.
1160
1238
:param delta: Whether to delta compress or just entropy compress.
1161
1239
:param _unadded_refs: private parameter, don't use.
1240
:param _group_cache: private parameter, don't use.
1163
1242
self._index = index
1164
1243
self._access = access
1166
1245
if _unadded_refs is None:
1167
1246
_unadded_refs = {}
1168
1247
self._unadded_refs = _unadded_refs
1169
self._group_cache = LRUSizeCache(max_size=50*1024*1024)
1170
self._fallback_vfs = []
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
1172
1254
def without_fallbacks(self):
1173
1255
"""Return a clone of this object without any fallbacks configured."""
1174
1256
return GroupCompressVersionedFiles(self._index, self._access,
1175
self._delta, _unadded_refs=dict(self._unadded_refs))
1257
self._delta, _unadded_refs=dict(
1258
self._unadded_refs),
1259
_group_cache=self._group_cache)
1177
1261
def add_lines(self, key, parents, lines, parent_texts=None,
1178
left_matching_blocks=None, nostore_sha=None, random_id=False,
1179
check_content=True):
1262
left_matching_blocks=None, nostore_sha=None, random_id=False,
1263
check_content=True):
1180
1264
"""Add a text to the store.
1182
1266
:param key: The key tuple of the text to add.
1183
1267
:param parents: The parents key tuples of the text to add.
1184
1268
:param lines: A list of lines. Each line must be a bytestring. And all
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
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
1190
1274
correctly is expensive and extremely unlikely to catch bugs so it
1191
1275
is not done at runtime unless check_content is True.)
1192
1276
:param parent_texts: An optional dictionary containing the opaque
1224
1308
nostore_sha=nostore_sha))[0]
1225
1309
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
1245
1311
def add_fallback_versioned_files(self, a_versioned_files):
1246
1312
"""Add a source of texts for texts not present in this knit.
1248
1314
:param a_versioned_files: A VersionedFiles object.
1250
self._fallback_vfs.append(a_versioned_files)
1316
self._immediate_fallback_vfs.append(a_versioned_files)
1252
1318
def annotate(self, key):
1253
1319
"""See VersionedFiles.annotate."""
1287
1353
self._check_lines_not_unicode(lines)
1288
1354
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)
1306
1356
def get_parent_map(self, keys):
1307
1357
"""Get a map of the graph parents of keys.
1411
1461
keys = set(remaining_keys)
1412
1462
for content_factory in self._get_remaining_record_stream(keys,
1413
orig_keys, ordering, include_delta_closure):
1463
orig_keys, ordering, include_delta_closure):
1414
1464
remaining_keys.discard(content_factory.key)
1415
1465
yield content_factory
1417
except errors.RetryWithNewPacks, e:
1467
except errors.RetryWithNewPacks as e:
1418
1468
self._access.reload_or_raise(e)
1420
1470
def _find_from_fallback(self, missing):
1447
1497
The returned objects should be in the order defined by 'ordering',
1448
1498
which can weave between different sources.
1449
1500
:param ordering: Must be one of 'topological' or 'groupcompress'
1450
1501
:return: List of [(source, [keys])] tuples, such that all keys are in
1451
1502
the defined order, regardless of source.
1453
1504
if ordering == 'topological':
1454
present_keys = topo_sort(parent_map)
1505
present_keys = tsort.topo_sort(parent_map)
1456
1507
# ordering == 'groupcompress'
1457
1508
# XXX: This only optimizes for the target ordering. We may need
1493
1544
# This is the group the bytes are stored in, followed by the
1494
1545
# location in the group
1495
1546
return locations[key][0]
1496
present_keys = sorted(locations.iterkeys(), key=get_group)
1497
1547
# We don't have an ordering for keys in the in-memory object, but
1498
1548
# lets process the in-memory ones first.
1499
present_keys = list(unadded_keys) + present_keys
1549
present_keys = list(unadded_keys)
1550
present_keys.extend(sorted(locations, key=get_group))
1500
1551
# Now grab all of the ones from other sources
1501
1552
source_keys = [(self, present_keys)]
1502
1553
source_keys.extend(source_result)
1534
1585
key_to_source_map)
1535
1586
elif ordering == 'as-requested':
1536
1587
source_keys = self._get_as_requested_source_keys(orig_keys,
1537
locations, unadded_keys, key_to_source_map)
1588
locations, unadded_keys, key_to_source_map)
1539
1590
# We want to yield the keys in a semi-optimal (read-wise) ordering.
1540
1591
# Otherwise we thrash the _group_cache and destroy performance
1541
1592
source_keys = self._get_io_ordered_source_keys(locations,
1542
unadded_keys, source_result)
1593
unadded_keys, source_result)
1543
1594
for key in missing:
1544
1595
yield AbsentContentFactory(key)
1545
1596
# Batch up as many keys as we can until either:
1546
1597
# - we encounter an unadded ref, or
1547
1598
# - we run out of keys, or
1548
1599
# - the total bytes to retrieve for this batch > BATCH_SIZE
1549
batcher = _BatchingBlockFetcher(self, locations)
1600
batcher = _BatchingBlockFetcher(self, locations,
1601
get_compressor_settings=self._get_compressor_settings)
1550
1602
for source, keys in source_keys:
1551
1603
if source is self:
1552
1604
for key in keys:
1598
1650
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())
1601
1677
def _insert_record_stream(self, stream, random_id=False, nostore_sha=None,
1602
1678
reuse_blocks=True):
1603
1679
"""Internal core to insert a record stream into this container.
1627
1704
# This will go up to fulltexts for gc to gc fetching, which isn't
1629
self._compressor = GroupCompressor()
1706
self._compressor = self._make_group_compressor()
1630
1707
self._unadded_refs = {}
1631
1708
keys_to_add = []
1633
bytes = self._compressor.flush().to_bytes()
1634
self._compressor = GroupCompressor()
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)
1635
1724
index, start, length = self._access.add_raw_records(
1636
[(None, len(bytes))], bytes)[0]
1725
[(None, len(data))], data)[0]
1638
1727
for key, reads, refs in keys_to_add:
1639
nodes.append((key, "%d %d %s" % (start, length, reads), refs))
1728
nodes.append((key, b"%d %d %s" % (start, length, reads), refs))
1640
1729
self._index.add_records(nodes, random_id=random_id)
1641
1730
self._unadded_refs = {}
1642
1731
del keys_to_add[:]
1656
1745
raise errors.RevisionNotPresent(record.key, self)
1658
1747
if record.key in inserted_keys:
1659
trace.note('Insert claimed random_id=True,'
1660
' but then inserted %r two times', record.key)
1748
trace.note(gettext('Insert claimed random_id=True,'
1749
' but then inserted %r two times'), record.key)
1662
1751
inserted_keys.add(record.key)
1663
1752
if reuse_blocks:
1689
1778
raise AssertionError('No insert_manager set')
1690
1779
if insert_manager is not record._manager:
1691
1780
raise AssertionError('insert_manager does not match'
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)
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)
1697
1786
nodes = [(record.key, value, (record.parents,))]
1698
1787
# TODO: Consider buffering up many nodes to be added, not
1699
1788
# sure how much overhead this has, but we're seeing
1722
1811
# delta_ratio = float(len(bytes)) / (end_point - start_point)
1723
1812
# Check if we want to continue to include that text
1724
1813
if (prefix == max_fulltext_prefix
1725
and end_point < 2 * max_fulltext_len):
1814
and end_point < 2 * max_fulltext_len):
1726
1815
# As long as we are on the same file_id, we will fill at least
1727
1816
# 2 * max_fulltext_len
1728
1817
start_new_block = False
1729
elif end_point > 4*1024*1024:
1818
elif end_point > 4 * 1024 * 1024:
1730
1819
start_new_block = True
1731
1820
elif (prefix is not None and prefix != last_prefix
1732
and end_point > 2*1024*1024):
1821
and end_point > 2 * 1024 * 1024):
1733
1822
start_new_block = True
1735
1824
start_new_block = False
1802
1892
"""See VersionedFiles.keys."""
1803
1893
if 'evil' in debug.debug_flags:
1804
1894
trace.mutter_callsite(2, "keys scales with size of history")
1805
sources = [self._index] + self._fallback_vfs
1895
sources = [self._index] + self._immediate_fallback_vfs
1807
1897
for source in sources:
1808
1898
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')
1812
1950
class _GCGraphIndex(object):
1813
1951
"""Mapper from GroupCompressVersionedFiles needs into GraphIndex storage."""
1815
1953
def __init__(self, graph_index, is_locked, parents=True,
1816
add_callback=None, track_external_parent_refs=False,
1817
inconsistency_fatal=True, track_new_keys=False):
1954
add_callback=None, track_external_parent_refs=False,
1955
inconsistency_fatal=True, track_new_keys=False):
1818
1956
"""Construct a _GCGraphIndex on a graph_index.
1820
:param graph_index: An implementation of bzrlib.index.GraphIndex.
1958
:param graph_index: An implementation of breezy.index.GraphIndex.
1821
1959
:param is_locked: A callback, returns True if the index is locked and
1823
1961
:param parents: If True, record knits parents, if not do not record
1889
2027
if node_refs != passed[1]:
1890
2028
details = '%s %s %s' % (key, (value, node_refs), passed)
1891
2029
if self._inconsistency_fatal:
1892
raise errors.KnitCorrupt(self, "inconsistent details"
1893
" in add_records: %s" %
2030
raise knit.KnitCorrupt(self, "inconsistent details"
2031
" in add_records: %s" %
1896
2034
trace.warning("inconsistent details in skipped"
1897
2035
" record: %s", details)
1902
2040
if self._parents:
1903
for key, (value, node_refs) in keys.iteritems():
2041
for key, (value, node_refs) in viewitems(keys):
1904
2042
result.append((key, value, node_refs))
1906
for key, (value, node_refs) in keys.iteritems():
2044
for key, (value, node_refs) in viewitems(keys):
1907
2045
result.append((key, value))
1908
2046
records = result
1909
2047
key_dependencies = self._key_dependencies
1989
2127
:param keys: An iterable of keys.
1990
2128
:return: A dict of key:
1991
2129
(index_memo, compression_parent, parents, record_details).
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
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
2003
2139
self._check_read()
2033
2168
# each, or about 7MB. Note that it might be even more when you consider
2034
2169
# how PyInt is allocated in separate slabs. And you can't return a slab
2035
2170
# to the OS if even 1 int on it is in use. Note though that Python uses
2036
# a LIFO when re-using PyInt slots, which probably causes more
2171
# a LIFO when re-using PyInt slots, which might cause more
2037
2172
# fragmentation.
2038
2173
start = int(bits[0])
2039
2174
start = self._int_cache.setdefault(start, start)
added
removed
breezy/bzr/groupcompress.py
