23
23
except ImportError:
26
from bzrlib.lazy_import import lazy_import
27
lazy_import(globals(), """
26
28
from bzrlib import (
30
33
graph as _mod_graph,
41
from bzrlib.repofmt import pack_repo
37
44
from bzrlib.btree_index import BTreeBuilder
38
45
from bzrlib.lru_cache import LRUSizeCache
39
from bzrlib.tsort import topo_sort
40
46
from bzrlib.versionedfile import (
42
49
AbsentContentFactory,
43
50
ChunkedContentFactory,
44
51
FulltextContentFactory,
52
VersionedFilesWithFallbacks,
48
55
# Minimum number of uncompressed bytes to try fetch at once when retrieving
135
142
self._content = ''.join(self._content_chunks)
136
143
self._content_chunks = None
137
144
if self._content is None:
138
if self._z_content is None:
145
# We join self._z_content_chunks here, because if we are
146
# decompressing, then it is *very* likely that we have a single
148
if self._z_content_chunks is None:
139
149
raise AssertionError('No content to decompress')
140
if self._z_content == '':
150
z_content = ''.join(self._z_content_chunks)
141
152
self._content = ''
142
153
elif self._compressor_name == 'lzma':
143
154
# We don't do partial lzma decomp yet
144
self._content = pylzma.decompress(self._z_content)
155
self._content = pylzma.decompress(z_content)
145
156
elif self._compressor_name == 'zlib':
146
157
# Start a zlib decompressor
147
158
if num_bytes * 4 > self._content_length * 3:
148
159
# If we are requesting more that 3/4ths of the content,
149
160
# just extract the whole thing in a single pass
150
161
num_bytes = self._content_length
151
self._content = zlib.decompress(self._z_content)
162
self._content = zlib.decompress(z_content)
153
164
self._z_content_decompressor = zlib.decompressobj()
154
165
# Seed the decompressor with the uncompressed bytes, so
155
166
# that the rest of the code is simplified
156
167
self._content = self._z_content_decompressor.decompress(
157
self._z_content, num_bytes + _ZLIB_DECOMP_WINDOW)
168
z_content, num_bytes + _ZLIB_DECOMP_WINDOW)
158
169
if not self._z_content_decompressor.unconsumed_tail:
159
170
self._z_content_decompressor = None
207
218
# XXX: Define some GCCorrupt error ?
208
219
raise AssertionError('Invalid bytes: (%d) != %d + %d' %
209
220
(len(bytes), pos, self._z_content_length))
210
self._z_content = bytes[pos:]
221
self._z_content_chunks = (bytes[pos:],)
224
def _z_content(self):
225
"""Return z_content_chunks as a simple string.
227
Meant only to be used by the test suite.
229
if self._z_content_chunks is not None:
230
return ''.join(self._z_content_chunks)
213
234
def from_bytes(cls, bytes):
269
290
self._content_length = length
270
291
self._content_chunks = content_chunks
271
292
self._content = None
272
self._z_content = None
293
self._z_content_chunks = None
274
295
def set_content(self, content):
275
296
"""Set the content of this block."""
276
297
self._content_length = len(content)
277
298
self._content = content
278
self._z_content = None
299
self._z_content_chunks = None
280
301
def _create_z_content_using_lzma(self):
281
302
if self._content_chunks is not None:
283
304
self._content_chunks = None
284
305
if self._content is None:
285
306
raise AssertionError('Nothing to compress')
286
self._z_content = pylzma.compress(self._content)
287
self._z_content_length = len(self._z_content)
307
z_content = pylzma.compress(self._content)
308
self._z_content_chunks = (z_content,)
309
self._z_content_length = len(z_content)
289
def _create_z_content_from_chunks(self):
311
def _create_z_content_from_chunks(self, chunks):
290
312
compressor = zlib.compressobj(zlib.Z_DEFAULT_COMPRESSION)
291
compressed_chunks = map(compressor.compress, self._content_chunks)
313
# Peak in this point is 1 fulltext, 1 compressed text, + zlib overhead
314
# (measured peak is maybe 30MB over the above...)
315
compressed_chunks = map(compressor.compress, chunks)
292
316
compressed_chunks.append(compressor.flush())
293
self._z_content = ''.join(compressed_chunks)
294
self._z_content_length = len(self._z_content)
317
# Ignore empty chunks
318
self._z_content_chunks = [c for c in compressed_chunks if c]
319
self._z_content_length = sum(map(len, self._z_content_chunks))
296
321
def _create_z_content(self):
297
if self._z_content is not None:
322
if self._z_content_chunks is not None:
300
325
self._create_z_content_using_lzma()
302
327
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)
328
chunks = self._content_chunks
330
chunks = (self._content,)
331
self._create_z_content_from_chunks(chunks)
309
"""Encode the information into a byte stream."""
334
"""Create the byte stream as a series of 'chunks'"""
310
335
self._create_z_content()
312
337
header = self.GCB_LZ_HEADER
314
339
header = self.GCB_HEADER
316
'%d\n%d\n' % (self._z_content_length, self._content_length),
340
chunks = ['%s%d\n%d\n'
341
% (header, self._z_content_length, self._content_length),
343
chunks.extend(self._z_content_chunks)
344
total_len = sum(map(len, chunks))
345
return total_len, chunks
348
"""Encode the information into a byte stream."""
349
total_len, chunks = self.to_chunks()
319
350
return ''.join(chunks)
321
352
def _dump(self, include_text=False):
435
466
# Grab and cache the raw bytes for this entry
436
467
# and break the ref-cycle with _manager since we don't need it
438
self._manager._prepare_for_extract()
470
self._manager._prepare_for_extract()
471
except zlib.error as value:
472
raise errors.DecompressCorruption("zlib: " + str(value))
439
473
block = self._manager._block
440
474
self._bytes = block.extract(self.key, self._start, self._end)
441
475
# There are code paths that first extract as fulltext, and then
460
494
_full_enough_block_size = 3*1024*1024 # size at which we won't repack
461
495
_full_enough_mixed_block_size = 2*768*1024 # 1.5MB
463
def __init__(self, block):
497
def __init__(self, block, get_compressor_settings=None):
464
498
self._block = block
465
499
# We need to preserve the ordering
466
500
self._factories = []
467
501
self._last_byte = 0
502
self._get_settings = get_compressor_settings
503
self._compressor_settings = None
505
def _get_compressor_settings(self):
506
if self._compressor_settings is not None:
507
return self._compressor_settings
509
if self._get_settings is not None:
510
settings = self._get_settings()
512
vf = GroupCompressVersionedFiles
513
settings = vf._DEFAULT_COMPRESSOR_SETTINGS
514
self._compressor_settings = settings
515
return self._compressor_settings
469
517
def add_factory(self, key, parents, start, end):
470
518
if not self._factories:
503
551
new_block.set_content(self._block._content[:last_byte])
504
552
self._block = new_block
554
def _make_group_compressor(self):
555
return GroupCompressor(self._get_compressor_settings())
506
557
def _rebuild_block(self):
507
558
"""Create a new GroupCompressBlock with only the referenced texts."""
508
compressor = GroupCompressor()
559
compressor = self._make_group_compressor()
509
560
tstart = time.time()
510
561
old_length = self._block._content_length
523
574
# block? It seems hard to come up with a method that it would
524
575
# expand, since we do full compression again. Perhaps based on a
525
576
# request that ends up poorly ordered?
577
# TODO: If the content would have expanded, then we would want to
578
# handle a case where we need to split the block.
579
# Now that we have a user-tweakable option
580
# (max_bytes_to_index), it is possible that one person set it
581
# to a very low value, causing poor compression.
526
582
delta = time.time() - tstart
527
583
self._block = new_block
528
584
trace.mutter('creating new compressed block on-the-fly in %.3fs'
679
735
z_header_bytes = zlib.compress(header_bytes)
681
737
z_header_bytes_len = len(z_header_bytes)
682
block_bytes = self._block.to_bytes()
738
block_bytes_len, block_chunks = self._block.to_chunks()
683
739
lines.append('%d\n%d\n%d\n' % (z_header_bytes_len, header_bytes_len,
685
741
lines.append(z_header_bytes)
686
lines.append(block_bytes)
687
del z_header_bytes, block_bytes
742
lines.extend(block_chunks)
743
del z_header_bytes, block_chunks
744
# TODO: This is a point where we will double the memory consumption. To
745
# avoid this, we probably have to switch to a 'chunked' api
688
746
return ''.join(lines)
691
749
def from_bytes(cls, bytes):
692
750
# TODO: This does extra string copying, probably better to do it a
751
# different way. At a minimum this creates 2 copies of the
694
753
(storage_kind, z_header_len, header_len,
695
754
block_len, rest) = bytes.split('\n', 4)
757
816
self.labels_deltas = {}
758
817
self._delta_index = None # Set by the children
759
818
self._block = GroupCompressBlock()
822
self._settings = settings
761
824
def compress(self, key, bytes, expected_sha, nostore_sha=None, soft=False):
762
825
"""Compress lines with label key.
855
918
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
920
self._block.set_chunked_content(self.chunks, self.endpoint)
866
921
self.chunks = None
867
922
self._delta_index = None
886
941
class PythonGroupCompressor(_CommonGroupCompressor):
943
def __init__(self, settings=None):
889
944
"""Create a GroupCompressor.
891
946
Used only if the pyrex version is not available.
893
super(PythonGroupCompressor, self).__init__()
948
super(PythonGroupCompressor, self).__init__(settings)
894
949
self._delta_index = LinesDeltaIndex([])
895
950
# The actual content is managed by LinesDeltaIndex
896
951
self.chunks = self._delta_index.lines
934
989
It contains code very similar to SequenceMatcher because of having a similar
935
990
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
992
* there is no junk, we want a minimal edit not a human readable diff.
993
* we don't filter very common lines (because we don't know where a good
994
range will start, and after the first text we want to be emitting minmal
996
* we chain the left side, not the right side
997
* we incrementally update the adjacency matrix as new lines are provided.
998
* we look for matches in all of the left side, so the routine which does
999
the analagous task of find_longest_match does not need to filter on the
948
super(PyrexGroupCompressor, self).__init__()
949
self._delta_index = DeltaIndex()
1003
def __init__(self, settings=None):
1004
super(PyrexGroupCompressor, self).__init__(settings)
1005
max_bytes_to_index = self._settings.get('max_bytes_to_index', 0)
1006
self._delta_index = DeltaIndex(max_bytes_to_index=max_bytes_to_index)
951
1008
def _compress(self, key, bytes, max_delta_size, soft=False):
952
1009
"""see _CommonGroupCompressor._compress"""
1027
1084
index = _GCGraphIndex(graph_index, lambda:True, parents=parents,
1028
1085
add_callback=graph_index.add_nodes,
1029
1086
inconsistency_fatal=inconsistency_fatal)
1030
access = knit._DirectPackAccess({})
1087
access = pack_repo._DirectPackAccess({})
1031
1088
access.set_writer(writer, graph_index, (transport, 'newpack'))
1032
1089
result = GroupCompressVersionedFiles(index, access, delta)
1033
1090
result.stream = stream
1044
1101
class _BatchingBlockFetcher(object):
1045
1102
"""Fetch group compress blocks in batches.
1047
1104
:ivar total_bytes: int of expected number of bytes needed to fetch the
1048
1105
currently pending batch.
1051
def __init__(self, gcvf, locations):
1108
def __init__(self, gcvf, locations, get_compressor_settings=None):
1052
1109
self.gcvf = gcvf
1053
1110
self.locations = locations
1102
1160
def yield_factories(self, full_flush=False):
1103
1161
"""Yield factories for keys added since the last yield. They will be
1104
1162
returned in the order they were added via add_key.
1106
1164
:param full_flush: by default, some results may not be returned in case
1107
1165
they can be part of the next batch. If full_flush is True, then
1108
1166
all results are returned.
1136
1194
memos_to_get_stack.pop()
1138
1196
block = self.batch_memos[read_memo]
1139
self.manager = _LazyGroupContentManager(block)
1197
self.manager = _LazyGroupContentManager(block,
1198
get_compressor_settings=self._get_compressor_settings)
1140
1199
self.last_read_memo = read_memo
1141
1200
start, end = index_memo[3:5]
1142
1201
self.manager.add_factory(key, parents, start, end)
1149
1208
self.total_bytes = 0
1152
class GroupCompressVersionedFiles(VersionedFiles):
1211
class GroupCompressVersionedFiles(VersionedFilesWithFallbacks):
1153
1212
"""A group-compress based VersionedFiles implementation."""
1155
def __init__(self, index, access, delta=True, _unadded_refs=None):
1214
# This controls how the GroupCompress DeltaIndex works. Basically, we
1215
# compute hash pointers into the source blocks (so hash(text) => text).
1216
# However each of these references costs some memory in trade against a
1217
# more accurate match result. For very large files, they either are
1218
# pre-compressed and change in bulk whenever they change, or change in just
1219
# local blocks. Either way, 'improved resolution' is not very helpful,
1220
# versus running out of memory trying to track everything. The default max
1221
# gives 100% sampling of a 1MB file.
1222
_DEFAULT_MAX_BYTES_TO_INDEX = 1024 * 1024
1223
_DEFAULT_COMPRESSOR_SETTINGS = {'max_bytes_to_index':
1224
_DEFAULT_MAX_BYTES_TO_INDEX}
1226
def __init__(self, index, access, delta=True, _unadded_refs=None,
1156
1228
"""Create a GroupCompressVersionedFiles object.
1158
1230
:param index: The index object storing access and graph data.
1159
1231
:param access: The access object storing raw data.
1160
1232
:param delta: Whether to delta compress or just entropy compress.
1161
1233
:param _unadded_refs: private parameter, don't use.
1234
:param _group_cache: private parameter, don't use.
1163
1236
self._index = index
1164
1237
self._access = access
1166
1239
if _unadded_refs is None:
1167
1240
_unadded_refs = {}
1168
1241
self._unadded_refs = _unadded_refs
1169
self._group_cache = LRUSizeCache(max_size=50*1024*1024)
1170
self._fallback_vfs = []
1242
if _group_cache is None:
1243
_group_cache = LRUSizeCache(max_size=50*1024*1024)
1244
self._group_cache = _group_cache
1245
self._immediate_fallback_vfs = []
1246
self._max_bytes_to_index = None
1172
1248
def without_fallbacks(self):
1173
1249
"""Return a clone of this object without any fallbacks configured."""
1174
1250
return GroupCompressVersionedFiles(self._index, self._access,
1175
self._delta, _unadded_refs=dict(self._unadded_refs))
1251
self._delta, _unadded_refs=dict(self._unadded_refs),
1252
_group_cache=self._group_cache)
1177
1254
def add_lines(self, key, parents, lines, parent_texts=None,
1178
1255
left_matching_blocks=None, nostore_sha=None, random_id=False,
1182
1259
:param key: The key tuple of the text to add.
1183
1260
:param parents: The parents key tuples of the text to add.
1184
1261
: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
1262
of them except the last must be terminated with \\n and contain no
1263
other \\n's. The last line may either contain no \\n's or a single
1264
terminating \\n. If the lines list does meet this constraint the
1265
add routine may error or may succeed - but you will be unable to
1266
read the data back accurately. (Checking the lines have been split
1190
1267
correctly is expensive and extremely unlikely to catch bugs so it
1191
1268
is not done at runtime unless check_content is True.)
1192
1269
:param parent_texts: An optional dictionary containing the opaque
1287
1364
self._check_lines_not_unicode(lines)
1288
1365
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
1367
def get_parent_map(self, keys):
1307
1368
"""Get a map of the graph parents of keys.
1447
1508
The returned objects should be in the order defined by 'ordering',
1448
1509
which can weave between different sources.
1449
1511
:param ordering: Must be one of 'topological' or 'groupcompress'
1450
1512
:return: List of [(source, [keys])] tuples, such that all keys are in
1451
1513
the defined order, regardless of source.
1453
1515
if ordering == 'topological':
1454
present_keys = topo_sort(parent_map)
1516
present_keys = tsort.topo_sort(parent_map)
1456
1518
# ordering == 'groupcompress'
1457
1519
# XXX: This only optimizes for the target ordering. We may need
1546
1608
# - we encounter an unadded ref, or
1547
1609
# - we run out of keys, or
1548
1610
# - the total bytes to retrieve for this batch > BATCH_SIZE
1549
batcher = _BatchingBlockFetcher(self, locations)
1611
batcher = _BatchingBlockFetcher(self, locations,
1612
get_compressor_settings=self._get_compressor_settings)
1550
1613
for source, keys in source_keys:
1551
1614
if source is self:
1552
1615
for key in keys:
1598
1661
for _ in self._insert_record_stream(stream, random_id=False):
1664
def _get_compressor_settings(self):
1665
if self._max_bytes_to_index is None:
1666
# TODO: VersionedFiles don't know about their containing
1667
# repository, so they don't have much of an idea about their
1668
# location. So for now, this is only a global option.
1669
c = config.GlobalConfig()
1670
val = c.get_user_option('bzr.groupcompress.max_bytes_to_index')
1674
except ValueError, e:
1675
trace.warning('Value for '
1676
'"bzr.groupcompress.max_bytes_to_index"'
1677
' %r is not an integer'
1681
val = self._DEFAULT_MAX_BYTES_TO_INDEX
1682
self._max_bytes_to_index = val
1683
return {'max_bytes_to_index': self._max_bytes_to_index}
1685
def _make_group_compressor(self):
1686
return GroupCompressor(self._get_compressor_settings())
1601
1688
def _insert_record_stream(self, stream, random_id=False, nostore_sha=None,
1602
1689
reuse_blocks=True):
1603
1690
"""Internal core to insert a record stream into this container.
1627
1714
# This will go up to fulltexts for gc to gc fetching, which isn't
1629
self._compressor = GroupCompressor()
1716
self._compressor = self._make_group_compressor()
1630
1717
self._unadded_refs = {}
1631
1718
keys_to_add = []
1633
bytes = self._compressor.flush().to_bytes()
1634
self._compressor = GroupCompressor()
1720
bytes_len, chunks = self._compressor.flush().to_chunks()
1721
self._compressor = self._make_group_compressor()
1722
# Note: At this point we still have 1 copy of the fulltext (in
1723
# record and the var 'bytes'), and this generates 2 copies of
1724
# the compressed text (one for bytes, one in chunks)
1725
# TODO: Push 'chunks' down into the _access api, so that we don't
1726
# have to double compressed memory here
1727
# TODO: Figure out how to indicate that we would be happy to free
1728
# the fulltext content at this point. Note that sometimes we
1729
# will want it later (streaming CHK pages), but most of the
1730
# time we won't (everything else)
1731
bytes = ''.join(chunks)
1635
1733
index, start, length = self._access.add_raw_records(
1636
1734
[(None, len(bytes))], bytes)[0]
1802
1900
"""See VersionedFiles.keys."""
1803
1901
if 'evil' in debug.debug_flags:
1804
1902
trace.mutter_callsite(2, "keys scales with size of history")
1805
sources = [self._index] + self._fallback_vfs
1903
sources = [self._index] + self._immediate_fallback_vfs
1807
1905
for source in sources:
1808
1906
result.update(source.keys())
1891
1989
# repeated over and over, this creates a surplus of ints
1892
1990
self._int_cache = {}
1893
1991
if track_external_parent_refs:
1894
self._key_dependencies = knit._KeyRefs(
1992
self._key_dependencies = _KeyRefs(
1895
1993
track_new_keys=track_new_keys)
1897
1995
self._key_dependencies = None
2037
2135
:param keys: An iterable of keys.
2038
2136
:return: A dict of key:
2039
2137
(index_memo, compression_parent, parents, record_details).
2041
opaque structure to pass to read_records to extract the raw
2044
Content that this record is built upon, may be None
2046
Logical parents of this node
2048
extra information about the content which needs to be passed to
2049
Factory.parse_record
2139
* index_memo: opaque structure to pass to read_records to extract
2141
* compression_parent: Content that this record is built upon, may
2143
* parents: Logical parents of this node
2144
* record_details: extra information about the content which needs
2145
to be passed to Factory.parse_record
2051
2147
self._check_read()
added
removed
bzrlib/groupcompress.py
