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
42
from bzrlib.i18n import gettext
37
45
from bzrlib.btree_index import BTreeBuilder
38
46
from bzrlib.lru_cache import LRUSizeCache
39
from bzrlib.tsort import topo_sort
40
47
from bzrlib.versionedfile import (
42
50
AbsentContentFactory,
43
51
ChunkedContentFactory,
44
52
FulltextContentFactory,
53
VersionedFilesWithFallbacks,
48
56
# Minimum number of uncompressed bytes to try fetch at once when retrieving
135
143
self._content = ''.join(self._content_chunks)
136
144
self._content_chunks = None
137
145
if self._content is None:
138
if self._z_content is None:
146
# We join self._z_content_chunks here, because if we are
147
# decompressing, then it is *very* likely that we have a single
149
if self._z_content_chunks is None:
139
150
raise AssertionError('No content to decompress')
140
if self._z_content == '':
151
z_content = ''.join(self._z_content_chunks)
141
153
self._content = ''
142
154
elif self._compressor_name == 'lzma':
143
155
# We don't do partial lzma decomp yet
144
self._content = pylzma.decompress(self._z_content)
156
self._content = pylzma.decompress(z_content)
145
157
elif self._compressor_name == 'zlib':
146
158
# Start a zlib decompressor
147
159
if num_bytes * 4 > self._content_length * 3:
148
160
# If we are requesting more that 3/4ths of the content,
149
161
# just extract the whole thing in a single pass
150
162
num_bytes = self._content_length
151
self._content = zlib.decompress(self._z_content)
163
self._content = zlib.decompress(z_content)
153
165
self._z_content_decompressor = zlib.decompressobj()
154
166
# Seed the decompressor with the uncompressed bytes, so
155
167
# that the rest of the code is simplified
156
168
self._content = self._z_content_decompressor.decompress(
157
self._z_content, num_bytes + _ZLIB_DECOMP_WINDOW)
169
z_content, num_bytes + _ZLIB_DECOMP_WINDOW)
158
170
if not self._z_content_decompressor.unconsumed_tail:
159
171
self._z_content_decompressor = None
207
219
# XXX: Define some GCCorrupt error ?
208
220
raise AssertionError('Invalid bytes: (%d) != %d + %d' %
209
221
(len(bytes), pos, self._z_content_length))
210
self._z_content = bytes[pos:]
222
self._z_content_chunks = (bytes[pos:],)
225
def _z_content(self):
226
"""Return z_content_chunks as a simple string.
228
Meant only to be used by the test suite.
230
if self._z_content_chunks is not None:
231
return ''.join(self._z_content_chunks)
213
235
def from_bytes(cls, bytes):
269
291
self._content_length = length
270
292
self._content_chunks = content_chunks
271
293
self._content = None
272
self._z_content = None
294
self._z_content_chunks = None
274
296
def set_content(self, content):
275
297
"""Set the content of this block."""
276
298
self._content_length = len(content)
277
299
self._content = content
278
self._z_content = None
300
self._z_content_chunks = None
280
302
def _create_z_content_using_lzma(self):
281
303
if self._content_chunks is not None:
283
305
self._content_chunks = None
284
306
if self._content is None:
285
307
raise AssertionError('Nothing to compress')
286
self._z_content = pylzma.compress(self._content)
287
self._z_content_length = len(self._z_content)
308
z_content = pylzma.compress(self._content)
309
self._z_content_chunks = (z_content,)
310
self._z_content_length = len(z_content)
289
def _create_z_content_from_chunks(self):
312
def _create_z_content_from_chunks(self, chunks):
290
313
compressor = zlib.compressobj(zlib.Z_DEFAULT_COMPRESSION)
291
compressed_chunks = map(compressor.compress, self._content_chunks)
314
# Peak in this point is 1 fulltext, 1 compressed text, + zlib overhead
315
# (measured peak is maybe 30MB over the above...)
316
compressed_chunks = map(compressor.compress, chunks)
292
317
compressed_chunks.append(compressor.flush())
293
self._z_content = ''.join(compressed_chunks)
294
self._z_content_length = len(self._z_content)
318
# Ignore empty chunks
319
self._z_content_chunks = [c for c in compressed_chunks if c]
320
self._z_content_length = sum(map(len, self._z_content_chunks))
296
322
def _create_z_content(self):
297
if self._z_content is not None:
323
if self._z_content_chunks is not None:
300
326
self._create_z_content_using_lzma()
302
328
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)
329
chunks = self._content_chunks
331
chunks = (self._content,)
332
self._create_z_content_from_chunks(chunks)
309
"""Encode the information into a byte stream."""
335
"""Create the byte stream as a series of 'chunks'"""
310
336
self._create_z_content()
312
338
header = self.GCB_LZ_HEADER
314
340
header = self.GCB_HEADER
316
'%d\n%d\n' % (self._z_content_length, self._content_length),
341
chunks = ['%s%d\n%d\n'
342
% (header, self._z_content_length, self._content_length),
344
chunks.extend(self._z_content_chunks)
345
total_len = sum(map(len, chunks))
346
return total_len, chunks
349
"""Encode the information into a byte stream."""
350
total_len, chunks = self.to_chunks()
319
351
return ''.join(chunks)
321
353
def _dump(self, include_text=False):
435
467
# Grab and cache the raw bytes for this entry
436
468
# and break the ref-cycle with _manager since we don't need it
438
self._manager._prepare_for_extract()
471
self._manager._prepare_for_extract()
472
except zlib.error as value:
473
raise errors.DecompressCorruption("zlib: " + str(value))
439
474
block = self._manager._block
440
475
self._bytes = block.extract(self.key, self._start, self._end)
441
476
# There are code paths that first extract as fulltext, and then
460
495
_full_enough_block_size = 3*1024*1024 # size at which we won't repack
461
496
_full_enough_mixed_block_size = 2*768*1024 # 1.5MB
463
def __init__(self, block):
498
def __init__(self, block, get_compressor_settings=None):
464
499
self._block = block
465
500
# We need to preserve the ordering
466
501
self._factories = []
467
502
self._last_byte = 0
503
self._get_settings = get_compressor_settings
504
self._compressor_settings = None
506
def _get_compressor_settings(self):
507
if self._compressor_settings is not None:
508
return self._compressor_settings
510
if self._get_settings is not None:
511
settings = self._get_settings()
513
vf = GroupCompressVersionedFiles
514
settings = vf._DEFAULT_COMPRESSOR_SETTINGS
515
self._compressor_settings = settings
516
return self._compressor_settings
469
518
def add_factory(self, key, parents, start, end):
470
519
if not self._factories:
503
552
new_block.set_content(self._block._content[:last_byte])
504
553
self._block = new_block
555
def _make_group_compressor(self):
556
return GroupCompressor(self._get_compressor_settings())
506
558
def _rebuild_block(self):
507
559
"""Create a new GroupCompressBlock with only the referenced texts."""
508
compressor = GroupCompressor()
560
compressor = self._make_group_compressor()
509
561
tstart = time.time()
510
562
old_length = self._block._content_length
523
575
# block? It seems hard to come up with a method that it would
524
576
# expand, since we do full compression again. Perhaps based on a
525
577
# request that ends up poorly ordered?
578
# TODO: If the content would have expanded, then we would want to
579
# handle a case where we need to split the block.
580
# Now that we have a user-tweakable option
581
# (max_bytes_to_index), it is possible that one person set it
582
# to a very low value, causing poor compression.
526
583
delta = time.time() - tstart
527
584
self._block = new_block
528
585
trace.mutter('creating new compressed block on-the-fly in %.3fs'
679
736
z_header_bytes = zlib.compress(header_bytes)
681
738
z_header_bytes_len = len(z_header_bytes)
682
block_bytes = self._block.to_bytes()
739
block_bytes_len, block_chunks = self._block.to_chunks()
683
740
lines.append('%d\n%d\n%d\n' % (z_header_bytes_len, header_bytes_len,
685
742
lines.append(z_header_bytes)
686
lines.append(block_bytes)
687
del z_header_bytes, block_bytes
743
lines.extend(block_chunks)
744
del z_header_bytes, block_chunks
745
# TODO: This is a point where we will double the memory consumption. To
746
# avoid this, we probably have to switch to a 'chunked' api
688
747
return ''.join(lines)
691
750
def from_bytes(cls, bytes):
692
751
# TODO: This does extra string copying, probably better to do it a
752
# different way. At a minimum this creates 2 copies of the
694
754
(storage_kind, z_header_len, header_len,
695
755
block_len, rest) = bytes.split('\n', 4)
757
817
self.labels_deltas = {}
758
818
self._delta_index = None # Set by the children
759
819
self._block = GroupCompressBlock()
823
self._settings = settings
761
825
def compress(self, key, bytes, expected_sha, nostore_sha=None, soft=False):
762
826
"""Compress lines with label key.
855
919
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
921
self._block.set_chunked_content(self.chunks, self.endpoint)
866
922
self.chunks = None
867
923
self._delta_index = None
886
942
class PythonGroupCompressor(_CommonGroupCompressor):
944
def __init__(self, settings=None):
889
945
"""Create a GroupCompressor.
891
947
Used only if the pyrex version is not available.
893
super(PythonGroupCompressor, self).__init__()
949
super(PythonGroupCompressor, self).__init__(settings)
894
950
self._delta_index = LinesDeltaIndex([])
895
951
# The actual content is managed by LinesDeltaIndex
896
952
self.chunks = self._delta_index.lines
934
990
It contains code very similar to SequenceMatcher because of having a similar
935
991
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
993
* there is no junk, we want a minimal edit not a human readable diff.
994
* we don't filter very common lines (because we don't know where a good
995
range will start, and after the first text we want to be emitting minmal
997
* we chain the left side, not the right side
998
* we incrementally update the adjacency matrix as new lines are provided.
999
* we look for matches in all of the left side, so the routine which does
1000
the analagous task of find_longest_match does not need to filter on the
948
super(PyrexGroupCompressor, self).__init__()
949
self._delta_index = DeltaIndex()
1004
def __init__(self, settings=None):
1005
super(PyrexGroupCompressor, self).__init__(settings)
1006
max_bytes_to_index = self._settings.get('max_bytes_to_index', 0)
1007
self._delta_index = DeltaIndex(max_bytes_to_index=max_bytes_to_index)
951
1009
def _compress(self, key, bytes, max_delta_size, soft=False):
952
1010
"""see _CommonGroupCompressor._compress"""
1027
1085
index = _GCGraphIndex(graph_index, lambda:True, parents=parents,
1028
1086
add_callback=graph_index.add_nodes,
1029
1087
inconsistency_fatal=inconsistency_fatal)
1030
access = knit._DirectPackAccess({})
1088
access = pack_repo._DirectPackAccess({})
1031
1089
access.set_writer(writer, graph_index, (transport, 'newpack'))
1032
1090
result = GroupCompressVersionedFiles(index, access, delta)
1033
1091
result.stream = stream
1044
1102
class _BatchingBlockFetcher(object):
1045
1103
"""Fetch group compress blocks in batches.
1047
1105
:ivar total_bytes: int of expected number of bytes needed to fetch the
1048
1106
currently pending batch.
1051
def __init__(self, gcvf, locations):
1109
def __init__(self, gcvf, locations, get_compressor_settings=None):
1052
1110
self.gcvf = gcvf
1053
1111
self.locations = locations
1102
1161
def yield_factories(self, full_flush=False):
1103
1162
"""Yield factories for keys added since the last yield. They will be
1104
1163
returned in the order they were added via add_key.
1106
1165
:param full_flush: by default, some results may not be returned in case
1107
1166
they can be part of the next batch. If full_flush is True, then
1108
1167
all results are returned.
1136
1195
memos_to_get_stack.pop()
1138
1197
block = self.batch_memos[read_memo]
1139
self.manager = _LazyGroupContentManager(block)
1198
self.manager = _LazyGroupContentManager(block,
1199
get_compressor_settings=self._get_compressor_settings)
1140
1200
self.last_read_memo = read_memo
1141
1201
start, end = index_memo[3:5]
1142
1202
self.manager.add_factory(key, parents, start, end)
1149
1209
self.total_bytes = 0
1152
class GroupCompressVersionedFiles(VersionedFiles):
1212
class GroupCompressVersionedFiles(VersionedFilesWithFallbacks):
1153
1213
"""A group-compress based VersionedFiles implementation."""
1155
def __init__(self, index, access, delta=True, _unadded_refs=None):
1215
# This controls how the GroupCompress DeltaIndex works. Basically, we
1216
# compute hash pointers into the source blocks (so hash(text) => text).
1217
# However each of these references costs some memory in trade against a
1218
# more accurate match result. For very large files, they either are
1219
# pre-compressed and change in bulk whenever they change, or change in just
1220
# local blocks. Either way, 'improved resolution' is not very helpful,
1221
# versus running out of memory trying to track everything. The default max
1222
# gives 100% sampling of a 1MB file.
1223
_DEFAULT_MAX_BYTES_TO_INDEX = 1024 * 1024
1224
_DEFAULT_COMPRESSOR_SETTINGS = {'max_bytes_to_index':
1225
_DEFAULT_MAX_BYTES_TO_INDEX}
1227
def __init__(self, index, access, delta=True, _unadded_refs=None,
1156
1229
"""Create a GroupCompressVersionedFiles object.
1158
1231
:param index: The index object storing access and graph data.
1159
1232
:param access: The access object storing raw data.
1160
1233
:param delta: Whether to delta compress or just entropy compress.
1161
1234
:param _unadded_refs: private parameter, don't use.
1235
:param _group_cache: private parameter, don't use.
1163
1237
self._index = index
1164
1238
self._access = access
1166
1240
if _unadded_refs is None:
1167
1241
_unadded_refs = {}
1168
1242
self._unadded_refs = _unadded_refs
1169
self._group_cache = LRUSizeCache(max_size=50*1024*1024)
1170
self._fallback_vfs = []
1243
if _group_cache is None:
1244
_group_cache = LRUSizeCache(max_size=50*1024*1024)
1245
self._group_cache = _group_cache
1246
self._immediate_fallback_vfs = []
1247
self._max_bytes_to_index = None
1172
1249
def without_fallbacks(self):
1173
1250
"""Return a clone of this object without any fallbacks configured."""
1174
1251
return GroupCompressVersionedFiles(self._index, self._access,
1175
self._delta, _unadded_refs=dict(self._unadded_refs))
1252
self._delta, _unadded_refs=dict(self._unadded_refs),
1253
_group_cache=self._group_cache)
1177
1255
def add_lines(self, key, parents, lines, parent_texts=None,
1178
1256
left_matching_blocks=None, nostore_sha=None, random_id=False,
1182
1260
:param key: The key tuple of the text to add.
1183
1261
:param parents: The parents key tuples of the text to add.
1184
1262
: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
1263
of them except the last must be terminated with \\n and contain no
1264
other \\n's. The last line may either contain no \\n's or a single
1265
terminating \\n. If the lines list does meet this constraint the
1266
add routine may error or may succeed - but you will be unable to
1267
read the data back accurately. (Checking the lines have been split
1190
1268
correctly is expensive and extremely unlikely to catch bugs so it
1191
1269
is not done at runtime unless check_content is True.)
1192
1270
:param parent_texts: An optional dictionary containing the opaque
1287
1365
self._check_lines_not_unicode(lines)
1288
1366
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
1368
def get_parent_map(self, keys):
1307
1369
"""Get a map of the graph parents of keys.
1447
1509
The returned objects should be in the order defined by 'ordering',
1448
1510
which can weave between different sources.
1449
1512
:param ordering: Must be one of 'topological' or 'groupcompress'
1450
1513
:return: List of [(source, [keys])] tuples, such that all keys are in
1451
1514
the defined order, regardless of source.
1453
1516
if ordering == 'topological':
1454
present_keys = topo_sort(parent_map)
1517
present_keys = tsort.topo_sort(parent_map)
1456
1519
# ordering == 'groupcompress'
1457
1520
# XXX: This only optimizes for the target ordering. We may need
1546
1609
# - we encounter an unadded ref, or
1547
1610
# - we run out of keys, or
1548
1611
# - the total bytes to retrieve for this batch > BATCH_SIZE
1549
batcher = _BatchingBlockFetcher(self, locations)
1612
batcher = _BatchingBlockFetcher(self, locations,
1613
get_compressor_settings=self._get_compressor_settings)
1550
1614
for source, keys in source_keys:
1551
1615
if source is self:
1552
1616
for key in keys:
1598
1662
for _ in self._insert_record_stream(stream, random_id=False):
1665
def _get_compressor_settings(self):
1666
if self._max_bytes_to_index is None:
1667
# TODO: VersionedFiles don't know about their containing
1668
# repository, so they don't have much of an idea about their
1669
# location. So for now, this is only a global option.
1670
c = config.GlobalConfig()
1671
val = c.get_user_option('bzr.groupcompress.max_bytes_to_index')
1675
except ValueError, e:
1676
trace.warning('Value for '
1677
'"bzr.groupcompress.max_bytes_to_index"'
1678
' %r is not an integer'
1682
val = self._DEFAULT_MAX_BYTES_TO_INDEX
1683
self._max_bytes_to_index = val
1684
return {'max_bytes_to_index': self._max_bytes_to_index}
1686
def _make_group_compressor(self):
1687
return GroupCompressor(self._get_compressor_settings())
1601
1689
def _insert_record_stream(self, stream, random_id=False, nostore_sha=None,
1602
1690
reuse_blocks=True):
1603
1691
"""Internal core to insert a record stream into this container.
1627
1715
# This will go up to fulltexts for gc to gc fetching, which isn't
1629
self._compressor = GroupCompressor()
1717
self._compressor = self._make_group_compressor()
1630
1718
self._unadded_refs = {}
1631
1719
keys_to_add = []
1633
bytes = self._compressor.flush().to_bytes()
1634
self._compressor = GroupCompressor()
1721
bytes_len, chunks = self._compressor.flush().to_chunks()
1722
self._compressor = self._make_group_compressor()
1723
# Note: At this point we still have 1 copy of the fulltext (in
1724
# record and the var 'bytes'), and this generates 2 copies of
1725
# the compressed text (one for bytes, one in chunks)
1726
# TODO: Push 'chunks' down into the _access api, so that we don't
1727
# have to double compressed memory here
1728
# TODO: Figure out how to indicate that we would be happy to free
1729
# the fulltext content at this point. Note that sometimes we
1730
# will want it later (streaming CHK pages), but most of the
1731
# time we won't (everything else)
1732
bytes = ''.join(chunks)
1635
1734
index, start, length = self._access.add_raw_records(
1636
1735
[(None, len(bytes))], bytes)[0]
1656
1755
raise errors.RevisionNotPresent(record.key, self)
1658
1757
if record.key in inserted_keys:
1659
trace.note('Insert claimed random_id=True,'
1660
' but then inserted %r two times', record.key)
1758
trace.note(gettext('Insert claimed random_id=True,'
1759
' but then inserted %r two times'), record.key)
1662
1761
inserted_keys.add(record.key)
1663
1762
if reuse_blocks:
1802
1901
"""See VersionedFiles.keys."""
1803
1902
if 'evil' in debug.debug_flags:
1804
1903
trace.mutter_callsite(2, "keys scales with size of history")
1805
sources = [self._index] + self._fallback_vfs
1904
sources = [self._index] + self._immediate_fallback_vfs
1807
1906
for source in sources:
1808
1907
result.update(source.keys())
1911
class _GCBuildDetails(object):
1912
"""A blob of data about the build details.
1914
This stores the minimal data, which then allows compatibility with the old
1915
api, without taking as much memory.
1918
__slots__ = ('_index', '_group_start', '_group_end', '_basis_end',
1919
'_delta_end', '_parents')
1922
compression_parent = None
1924
def __init__(self, parents, position_info):
1925
self._parents = parents
1926
(self._index, self._group_start, self._group_end, self._basis_end,
1927
self._delta_end) = position_info
1930
return '%s(%s, %s)' % (self.__class__.__name__,
1931
self.index_memo, self._parents)
1934
def index_memo(self):
1935
return (self._index, self._group_start, self._group_end,
1936
self._basis_end, self._delta_end)
1939
def record_details(self):
1940
return static_tuple.StaticTuple(self.method, None)
1942
def __getitem__(self, offset):
1943
"""Compatibility thunk to act like a tuple."""
1945
return self.index_memo
1947
return self.compression_parent # Always None
1949
return self._parents
1951
return self.record_details
1953
raise IndexError('offset out of range')
1812
1959
class _GCGraphIndex(object):
1813
1960
"""Mapper from GroupCompressVersionedFiles needs into GraphIndex storage."""
1843
1990
# repeated over and over, this creates a surplus of ints
1844
1991
self._int_cache = {}
1845
1992
if track_external_parent_refs:
1846
self._key_dependencies = knit._KeyRefs(
1993
self._key_dependencies = _KeyRefs(
1847
1994
track_new_keys=track_new_keys)
1849
1996
self._key_dependencies = None
1989
2136
:param keys: An iterable of keys.
1990
2137
:return: A dict of key:
1991
2138
(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
2140
* index_memo: opaque structure to pass to read_records to extract
2142
* compression_parent: Content that this record is built upon, may
2144
* parents: Logical parents of this node
2145
* record_details: extra information about the content which needs
2146
to be passed to Factory.parse_record
2003
2148
self._check_read()
2033
2177
# each, or about 7MB. Note that it might be even more when you consider
2034
2178
# how PyInt is allocated in separate slabs. And you can't return a slab
2035
2179
# 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
2180
# a LIFO when re-using PyInt slots, which might cause more
2037
2181
# fragmentation.
2038
2182
start = int(bits[0])
2039
2183
start = self._int_cache.setdefault(start, start)
added
removed
bzrlib/groupcompress.py
