1
# Copyright (C) 2008, 2009, 2010 Canonical Ltd
1
# Copyright (C) 2008-2011 Canonical Ltd
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
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
459
467
# Grab and cache the raw bytes for this entry
460
468
# and break the ref-cycle with _manager since we don't need it
462
self._manager._prepare_for_extract()
471
self._manager._prepare_for_extract()
472
except zlib.error as value:
473
raise errors.DecompressCorruption("zlib: " + str(value))
463
474
block = self._manager._block
464
475
self._bytes = block.extract(self.key, self._start, self._end)
465
476
# There are code paths that first extract as fulltext, and then
484
495
_full_enough_block_size = 3*1024*1024 # size at which we won't repack
485
496
_full_enough_mixed_block_size = 2*768*1024 # 1.5MB
487
def __init__(self, block):
498
def __init__(self, block, get_compressor_settings=None):
488
499
self._block = block
489
500
# We need to preserve the ordering
490
501
self._factories = []
491
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
493
518
def add_factory(self, key, parents, start, end):
494
519
if not self._factories:
527
552
new_block.set_content(self._block._content[:last_byte])
528
553
self._block = new_block
555
def _make_group_compressor(self):
556
return GroupCompressor(self._get_compressor_settings())
530
558
def _rebuild_block(self):
531
559
"""Create a new GroupCompressBlock with only the referenced texts."""
532
compressor = GroupCompressor()
560
compressor = self._make_group_compressor()
533
561
tstart = time.time()
534
562
old_length = self._block._content_length
547
575
# block? It seems hard to come up with a method that it would
548
576
# expand, since we do full compression again. Perhaps based on a
549
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.
550
583
delta = time.time() - tstart
551
584
self._block = new_block
552
585
trace.mutter('creating new compressed block on-the-fly in %.3fs'
784
817
self.labels_deltas = {}
785
818
self._delta_index = None # Set by the children
786
819
self._block = GroupCompressBlock()
823
self._settings = settings
788
825
def compress(self, key, bytes, expected_sha, nostore_sha=None, soft=False):
789
826
"""Compress lines with label key.
905
942
class PythonGroupCompressor(_CommonGroupCompressor):
944
def __init__(self, settings=None):
908
945
"""Create a GroupCompressor.
910
947
Used only if the pyrex version is not available.
912
super(PythonGroupCompressor, self).__init__()
949
super(PythonGroupCompressor, self).__init__(settings)
913
950
self._delta_index = LinesDeltaIndex([])
914
951
# The actual content is managed by LinesDeltaIndex
915
952
self.chunks = self._delta_index.lines
953
990
It contains code very similar to SequenceMatcher because of having a similar
954
991
task. However some key differences apply:
955
- there is no junk, we want a minimal edit not a human readable diff.
956
- we don't filter very common lines (because we don't know where a good
957
range will start, and after the first text we want to be emitting minmal
959
- we chain the left side, not the right side
960
- we incrementally update the adjacency matrix as new lines are provided.
961
- we look for matches in all of the left side, so the routine which does
962
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
967
super(PyrexGroupCompressor, self).__init__()
968
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)
970
1009
def _compress(self, key, bytes, max_delta_size, soft=False):
971
1010
"""see _CommonGroupCompressor._compress"""
1046
1085
index = _GCGraphIndex(graph_index, lambda:True, parents=parents,
1047
1086
add_callback=graph_index.add_nodes,
1048
1087
inconsistency_fatal=inconsistency_fatal)
1049
access = knit._DirectPackAccess({})
1088
access = pack_repo._DirectPackAccess({})
1050
1089
access.set_writer(writer, graph_index, (transport, 'newpack'))
1051
1090
result = GroupCompressVersionedFiles(index, access, delta)
1052
1091
result.stream = stream
1063
1102
class _BatchingBlockFetcher(object):
1064
1103
"""Fetch group compress blocks in batches.
1066
1105
:ivar total_bytes: int of expected number of bytes needed to fetch the
1067
1106
currently pending batch.
1070
def __init__(self, gcvf, locations):
1109
def __init__(self, gcvf, locations, get_compressor_settings=None):
1071
1110
self.gcvf = gcvf
1072
1111
self.locations = locations
1121
1161
def yield_factories(self, full_flush=False):
1122
1162
"""Yield factories for keys added since the last yield. They will be
1123
1163
returned in the order they were added via add_key.
1125
1165
:param full_flush: by default, some results may not be returned in case
1126
1166
they can be part of the next batch. If full_flush is True, then
1127
1167
all results are returned.
1155
1195
memos_to_get_stack.pop()
1157
1197
block = self.batch_memos[read_memo]
1158
self.manager = _LazyGroupContentManager(block)
1198
self.manager = _LazyGroupContentManager(block,
1199
get_compressor_settings=self._get_compressor_settings)
1159
1200
self.last_read_memo = read_memo
1160
1201
start, end = index_memo[3:5]
1161
1202
self.manager.add_factory(key, parents, start, end)
1168
1209
self.total_bytes = 0
1171
class GroupCompressVersionedFiles(VersionedFiles):
1212
class GroupCompressVersionedFiles(VersionedFilesWithFallbacks):
1172
1213
"""A group-compress based VersionedFiles implementation."""
1174
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,
1175
1229
"""Create a GroupCompressVersionedFiles object.
1177
1231
:param index: The index object storing access and graph data.
1178
1232
:param access: The access object storing raw data.
1179
1233
:param delta: Whether to delta compress or just entropy compress.
1180
1234
:param _unadded_refs: private parameter, don't use.
1235
:param _group_cache: private parameter, don't use.
1182
1237
self._index = index
1183
1238
self._access = access
1185
1240
if _unadded_refs is None:
1186
1241
_unadded_refs = {}
1187
1242
self._unadded_refs = _unadded_refs
1188
self._group_cache = LRUSizeCache(max_size=50*1024*1024)
1189
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
1191
1249
def without_fallbacks(self):
1192
1250
"""Return a clone of this object without any fallbacks configured."""
1193
1251
return GroupCompressVersionedFiles(self._index, self._access,
1194
self._delta, _unadded_refs=dict(self._unadded_refs))
1252
self._delta, _unadded_refs=dict(self._unadded_refs),
1253
_group_cache=self._group_cache)
1196
1255
def add_lines(self, key, parents, lines, parent_texts=None,
1197
1256
left_matching_blocks=None, nostore_sha=None, random_id=False,
1201
1260
:param key: The key tuple of the text to add.
1202
1261
:param parents: The parents key tuples of the text to add.
1203
1262
:param lines: A list of lines. Each line must be a bytestring. And all
1204
of them except the last must be terminated with \n and contain no
1205
other \n's. The last line may either contain no \n's or a single
1206
terminating \n. If the lines list does meet this constraint the add
1207
routine may error or may succeed - but you will be unable to read
1208
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
1209
1268
correctly is expensive and extremely unlikely to catch bugs so it
1210
1269
is not done at runtime unless check_content is True.)
1211
1270
:param parent_texts: An optional dictionary containing the opaque
1306
1365
self._check_lines_not_unicode(lines)
1307
1366
self._check_lines_are_lines(lines)
1309
def get_known_graph_ancestry(self, keys):
1310
"""Get a KnownGraph instance with the ancestry of keys."""
1311
# Note that this is identical to
1312
# KnitVersionedFiles.get_known_graph_ancestry, but they don't share
1314
parent_map, missing_keys = self._index.find_ancestry(keys)
1315
for fallback in self._fallback_vfs:
1316
if not missing_keys:
1318
(f_parent_map, f_missing_keys) = fallback._index.find_ancestry(
1320
parent_map.update(f_parent_map)
1321
missing_keys = f_missing_keys
1322
kg = _mod_graph.KnownGraph(parent_map)
1325
1368
def get_parent_map(self, keys):
1326
1369
"""Get a map of the graph parents of keys.
1466
1509
The returned objects should be in the order defined by 'ordering',
1467
1510
which can weave between different sources.
1468
1512
:param ordering: Must be one of 'topological' or 'groupcompress'
1469
1513
:return: List of [(source, [keys])] tuples, such that all keys are in
1470
1514
the defined order, regardless of source.
1472
1516
if ordering == 'topological':
1473
present_keys = topo_sort(parent_map)
1517
present_keys = tsort.topo_sort(parent_map)
1475
1519
# ordering == 'groupcompress'
1476
1520
# XXX: This only optimizes for the target ordering. We may need
1565
1609
# - we encounter an unadded ref, or
1566
1610
# - we run out of keys, or
1567
1611
# - the total bytes to retrieve for this batch > BATCH_SIZE
1568
batcher = _BatchingBlockFetcher(self, locations)
1612
batcher = _BatchingBlockFetcher(self, locations,
1613
get_compressor_settings=self._get_compressor_settings)
1569
1614
for source, keys in source_keys:
1570
1615
if source is self:
1571
1616
for key in keys:
1617
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())
1620
1689
def _insert_record_stream(self, stream, random_id=False, nostore_sha=None,
1621
1690
reuse_blocks=True):
1622
1691
"""Internal core to insert a record stream into this container.
1646
1715
# This will go up to fulltexts for gc to gc fetching, which isn't
1648
self._compressor = GroupCompressor()
1717
self._compressor = self._make_group_compressor()
1649
1718
self._unadded_refs = {}
1650
1719
keys_to_add = []
1652
1721
bytes_len, chunks = self._compressor.flush().to_chunks()
1653
self._compressor = GroupCompressor()
1722
self._compressor = self._make_group_compressor()
1654
1723
# Note: At this point we still have 1 copy of the fulltext (in
1655
1724
# record and the var 'bytes'), and this generates 2 copies of
1656
1725
# the compressed text (one for bytes, one in chunks)
1686
1755
raise errors.RevisionNotPresent(record.key, self)
1688
1757
if record.key in inserted_keys:
1689
trace.note('Insert claimed random_id=True,'
1690
' 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)
1692
1761
inserted_keys.add(record.key)
1693
1762
if reuse_blocks:
1832
1901
"""See VersionedFiles.keys."""
1833
1902
if 'evil' in debug.debug_flags:
1834
1903
trace.mutter_callsite(2, "keys scales with size of history")
1835
sources = [self._index] + self._fallback_vfs
1904
sources = [self._index] + self._immediate_fallback_vfs
1837
1906
for source in sources:
1838
1907
result.update(source.keys())
1921
1990
# repeated over and over, this creates a surplus of ints
1922
1991
self._int_cache = {}
1923
1992
if track_external_parent_refs:
1924
self._key_dependencies = knit._KeyRefs(
1993
self._key_dependencies = _KeyRefs(
1925
1994
track_new_keys=track_new_keys)
1927
1996
self._key_dependencies = None
2067
2136
:param keys: An iterable of keys.
2068
2137
:return: A dict of key:
2069
2138
(index_memo, compression_parent, parents, record_details).
2071
opaque structure to pass to read_records to extract the raw
2074
Content that this record is built upon, may be None
2076
Logical parents of this node
2078
extra information about the content which needs to be passed to
2079
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
2081
2148
self._check_read()
added
removed
bzrlib/groupcompress.py
