/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to breezy/bzr/groupcompress.py

  • Committer: Jelmer Vernooij
  • Date: 2017-06-10 18:34:12 UTC
  • mto: This revision was merged to the branch mainline in revision 6690.
  • Revision ID: jelmer@jelmer.uk-20170610183412-s9fro6la0e1848x6
More moves.

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2008, 2009, 2010 Canonical Ltd
 
1
# Copyright (C) 2008-2011 Canonical Ltd
2
2
#
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
16
16
 
17
17
"""Core compression logic for compressing streams of related files."""
18
18
 
 
19
from __future__ import absolute_import
 
20
 
19
21
import time
20
22
import zlib
21
 
try:
22
 
    import pylzma
23
 
except ImportError:
24
 
    pylzma = None
25
23
 
26
 
from bzrlib import (
 
24
from ..lazy_import import lazy_import
 
25
lazy_import(globals(), """
 
26
from breezy import (
27
27
    annotate,
 
28
    config,
28
29
    debug,
29
30
    errors,
30
31
    graph as _mod_graph,
31
 
    knit,
32
32
    osutils,
33
 
    pack,
34
33
    static_tuple,
35
34
    trace,
36
 
    )
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
    tsort,
 
36
    )
 
37
from breezy.bzr import (
 
38
    pack,
 
39
    )
 
40
 
 
41
from breezy.bzr import pack_repo
 
42
from breezy.i18n import gettext
 
43
""")
 
44
 
 
45
from .btree_index import BTreeBuilder
 
46
from ..lru_cache import LRUSizeCache
 
47
from ..sixish import (
 
48
    map,
 
49
    range,
 
50
    viewitems,
 
51
    )
 
52
from .versionedfile import (
 
53
    _KeyRefs,
41
54
    adapter_registry,
42
55
    AbsentContentFactory,
43
56
    ChunkedContentFactory,
44
57
    FulltextContentFactory,
45
 
    VersionedFiles,
 
58
    VersionedFilesWithFallbacks,
46
59
    )
47
60
 
48
61
# Minimum number of uncompressed bytes to try fetch at once when retrieving
49
62
# groupcompress blocks.
50
63
BATCH_SIZE = 2**16
51
64
 
52
 
_USE_LZMA = False and (pylzma is not None)
53
 
 
54
65
# osutils.sha_string('')
55
66
_null_sha1 = 'da39a3ee5e6b4b0d3255bfef95601890afd80709'
56
67
 
65
76
    # groupcompress ordering is approximately reverse topological,
66
77
    # properly grouped by file-id.
67
78
    per_prefix_map = {}
68
 
    for key, value in parent_map.iteritems():
 
79
    for key, value in viewitems(parent_map):
69
80
        if isinstance(key, str) or len(key) == 1:
70
81
            prefix = ''
71
82
        else:
77
88
 
78
89
    present_keys = []
79
90
    for prefix in sorted(per_prefix_map):
80
 
        present_keys.extend(reversed(topo_sort(per_prefix_map[prefix])))
 
91
        present_keys.extend(reversed(tsort.topo_sort(per_prefix_map[prefix])))
81
92
    return present_keys
82
93
 
83
94
 
101
112
    def __init__(self):
102
113
        # map by key? or just order in file?
103
114
        self._compressor_name = None
104
 
        self._z_content = None
 
115
        self._z_content_chunks = None
105
116
        self._z_content_decompressor = None
106
117
        self._z_content_length = None
107
118
        self._content_length = None
135
146
                self._content = ''.join(self._content_chunks)
136
147
                self._content_chunks = None
137
148
        if self._content is None:
138
 
            if self._z_content is None:
 
149
            # We join self._z_content_chunks here, because if we are
 
150
            # decompressing, then it is *very* likely that we have a single
 
151
            # chunk
 
152
            if self._z_content_chunks is None:
139
153
                raise AssertionError('No content to decompress')
140
 
            if self._z_content == '':
 
154
            z_content = ''.join(self._z_content_chunks)
 
155
            if z_content == '':
141
156
                self._content = ''
142
157
            elif self._compressor_name == 'lzma':
143
158
                # We don't do partial lzma decomp yet
144
 
                self._content = pylzma.decompress(self._z_content)
 
159
                import pylzma
 
160
                self._content = pylzma.decompress(z_content)
145
161
            elif self._compressor_name == 'zlib':
146
162
                # Start a zlib decompressor
147
163
                if num_bytes * 4 > self._content_length * 3:
148
164
                    # If we are requesting more that 3/4ths of the content,
149
165
                    # just extract the whole thing in a single pass
150
166
                    num_bytes = self._content_length
151
 
                    self._content = zlib.decompress(self._z_content)
 
167
                    self._content = zlib.decompress(z_content)
152
168
                else:
153
169
                    self._z_content_decompressor = zlib.decompressobj()
154
170
                    # Seed the decompressor with the uncompressed bytes, so
155
171
                    # that the rest of the code is simplified
156
172
                    self._content = self._z_content_decompressor.decompress(
157
 
                        self._z_content, num_bytes + _ZLIB_DECOMP_WINDOW)
 
173
                        z_content, num_bytes + _ZLIB_DECOMP_WINDOW)
158
174
                    if not self._z_content_decompressor.unconsumed_tail:
159
175
                        self._z_content_decompressor = None
160
176
            else:
207
223
            # XXX: Define some GCCorrupt error ?
208
224
            raise AssertionError('Invalid bytes: (%d) != %d + %d' %
209
225
                                 (len(bytes), pos, self._z_content_length))
210
 
        self._z_content = bytes[pos:]
 
226
        self._z_content_chunks = (bytes[pos:],)
 
227
 
 
228
    @property
 
229
    def _z_content(self):
 
230
        """Return z_content_chunks as a simple string.
 
231
 
 
232
        Meant only to be used by the test suite.
 
233
        """
 
234
        if self._z_content_chunks is not None:
 
235
            return ''.join(self._z_content_chunks)
 
236
        return None
211
237
 
212
238
    @classmethod
213
239
    def from_bytes(cls, bytes):
269
295
        self._content_length = length
270
296
        self._content_chunks = content_chunks
271
297
        self._content = None
272
 
        self._z_content = None
 
298
        self._z_content_chunks = None
273
299
 
274
300
    def set_content(self, content):
275
301
        """Set the content of this block."""
276
302
        self._content_length = len(content)
277
303
        self._content = content
278
 
        self._z_content = None
279
 
 
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)
288
 
 
289
 
    def _create_z_content_from_chunks(self):
 
304
        self._z_content_chunks = None
 
305
 
 
306
    def _create_z_content_from_chunks(self, chunks):
290
307
        compressor = zlib.compressobj(zlib.Z_DEFAULT_COMPRESSION)
291
 
        compressed_chunks = map(compressor.compress, self._content_chunks)
 
308
        # Peak in this point is 1 fulltext, 1 compressed text, + zlib overhead
 
309
        # (measured peak is maybe 30MB over the above...)
 
310
        compressed_chunks = list(map(compressor.compress, chunks))
292
311
        compressed_chunks.append(compressor.flush())
293
 
        self._z_content = ''.join(compressed_chunks)
294
 
        self._z_content_length = len(self._z_content)
 
312
        # Ignore empty chunks
 
313
        self._z_content_chunks = [c for c in compressed_chunks if c]
 
314
        self._z_content_length = sum(map(len, self._z_content_chunks))
295
315
 
296
316
    def _create_z_content(self):
297
 
        if self._z_content is not None:
298
 
            return
299
 
        if _USE_LZMA:
300
 
            self._create_z_content_using_lzma()
 
317
        if self._z_content_chunks is not None:
301
318
            return
302
319
        if self._content_chunks is not None:
303
 
            self._create_z_content_from_chunks()
304
 
            return
305
 
        self._z_content = zlib.compress(self._content)
306
 
        self._z_content_length = len(self._z_content)
 
320
            chunks = self._content_chunks
 
321
        else:
 
322
            chunks = (self._content,)
 
323
        self._create_z_content_from_chunks(chunks)
 
324
 
 
325
    def to_chunks(self):
 
326
        """Create the byte stream as a series of 'chunks'"""
 
327
        self._create_z_content()
 
328
        header = self.GCB_HEADER
 
329
        chunks = ['%s%d\n%d\n'
 
330
                  % (header, self._z_content_length, self._content_length),
 
331
                 ]
 
332
        chunks.extend(self._z_content_chunks)
 
333
        total_len = sum(map(len, chunks))
 
334
        return total_len, chunks
307
335
 
308
336
    def to_bytes(self):
309
337
        """Encode the information into a byte stream."""
310
 
        self._create_z_content()
311
 
        if _USE_LZMA:
312
 
            header = self.GCB_LZ_HEADER
313
 
        else:
314
 
            header = self.GCB_HEADER
315
 
        chunks = [header,
316
 
                  '%d\n%d\n' % (self._z_content_length, self._content_length),
317
 
                  self._z_content,
318
 
                 ]
 
338
        total_len, chunks = self.to_chunks()
319
339
        return ''.join(chunks)
320
340
 
321
341
    def _dump(self, include_text=False):
435
455
                # Grab and cache the raw bytes for this entry
436
456
                # and break the ref-cycle with _manager since we don't need it
437
457
                # anymore
438
 
                self._manager._prepare_for_extract()
 
458
                try:
 
459
                    self._manager._prepare_for_extract()
 
460
                except zlib.error as value:
 
461
                    raise errors.DecompressCorruption("zlib: " + str(value))
439
462
                block = self._manager._block
440
463
                self._bytes = block.extract(self.key, self._start, self._end)
441
464
                # There are code paths that first extract as fulltext, and then
460
483
    _full_enough_block_size = 3*1024*1024 # size at which we won't repack
461
484
    _full_enough_mixed_block_size = 2*768*1024 # 1.5MB
462
485
 
463
 
    def __init__(self, block):
 
486
    def __init__(self, block, get_compressor_settings=None):
464
487
        self._block = block
465
488
        # We need to preserve the ordering
466
489
        self._factories = []
467
490
        self._last_byte = 0
 
491
        self._get_settings = get_compressor_settings
 
492
        self._compressor_settings = None
 
493
 
 
494
    def _get_compressor_settings(self):
 
495
        if self._compressor_settings is not None:
 
496
            return self._compressor_settings
 
497
        settings = None
 
498
        if self._get_settings is not None:
 
499
            settings = self._get_settings()
 
500
        if settings is None:
 
501
            vf = GroupCompressVersionedFiles
 
502
            settings = vf._DEFAULT_COMPRESSOR_SETTINGS
 
503
        self._compressor_settings = settings
 
504
        return self._compressor_settings
468
505
 
469
506
    def add_factory(self, key, parents, start, end):
470
507
        if not self._factories:
503
540
        new_block.set_content(self._block._content[:last_byte])
504
541
        self._block = new_block
505
542
 
 
543
    def _make_group_compressor(self):
 
544
        return GroupCompressor(self._get_compressor_settings())
 
545
 
506
546
    def _rebuild_block(self):
507
547
        """Create a new GroupCompressBlock with only the referenced texts."""
508
 
        compressor = GroupCompressor()
 
548
        compressor = self._make_group_compressor()
509
549
        tstart = time.time()
510
550
        old_length = self._block._content_length
511
551
        end_point = 0
523
563
        #       block? It seems hard to come up with a method that it would
524
564
        #       expand, since we do full compression again. Perhaps based on a
525
565
        #       request that ends up poorly ordered?
 
566
        # TODO: If the content would have expanded, then we would want to
 
567
        #       handle a case where we need to split the block.
 
568
        #       Now that we have a user-tweakable option
 
569
        #       (max_bytes_to_index), it is possible that one person set it
 
570
        #       to a very low value, causing poor compression.
526
571
        delta = time.time() - tstart
527
572
        self._block = new_block
528
573
        trace.mutter('creating new compressed block on-the-fly in %.3fs'
679
724
        z_header_bytes = zlib.compress(header_bytes)
680
725
        del header_bytes
681
726
        z_header_bytes_len = len(z_header_bytes)
682
 
        block_bytes = self._block.to_bytes()
 
727
        block_bytes_len, block_chunks = self._block.to_chunks()
683
728
        lines.append('%d\n%d\n%d\n' % (z_header_bytes_len, header_bytes_len,
684
 
                                       len(block_bytes)))
 
729
                                       block_bytes_len))
685
730
        lines.append(z_header_bytes)
686
 
        lines.append(block_bytes)
687
 
        del z_header_bytes, block_bytes
 
731
        lines.extend(block_chunks)
 
732
        del z_header_bytes, block_chunks
 
733
        # TODO: This is a point where we will double the memory consumption. To
 
734
        #       avoid this, we probably have to switch to a 'chunked' api
688
735
        return ''.join(lines)
689
736
 
690
737
    @classmethod
691
738
    def from_bytes(cls, bytes):
692
739
        # TODO: This does extra string copying, probably better to do it a
693
 
        #       different way
 
740
        #       different way. At a minimum this creates 2 copies of the
 
741
        #       compressed content
694
742
        (storage_kind, z_header_len, header_len,
695
743
         block_len, rest) = bytes.split('\n', 4)
696
744
        del bytes
723
771
        block = GroupCompressBlock.from_bytes(block_bytes)
724
772
        del block_bytes
725
773
        result = cls(block)
726
 
        for start in xrange(0, len(header_lines), 4):
 
774
        for start in range(0, len(header_lines), 4):
727
775
            # intern()?
728
776
            key = tuple(header_lines[start].split('\x00'))
729
777
            parents_line = header_lines[start+1]
748
796
 
749
797
class _CommonGroupCompressor(object):
750
798
 
751
 
    def __init__(self):
 
799
    def __init__(self, settings=None):
752
800
        """Create a GroupCompressor."""
753
801
        self.chunks = []
754
802
        self._last = None
757
805
        self.labels_deltas = {}
758
806
        self._delta_index = None # Set by the children
759
807
        self._block = GroupCompressBlock()
 
808
        if settings is None:
 
809
            self._settings = {}
 
810
        else:
 
811
            self._settings = settings
760
812
 
761
813
    def compress(self, key, bytes, expected_sha, nostore_sha=None, soft=False):
762
814
        """Compress lines with label key.
854
906
 
855
907
        After calling this, the compressor should no longer be used
856
908
        """
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
909
        self._block.set_chunked_content(self.chunks, self.endpoint)
866
910
        self.chunks = None
867
911
        self._delta_index = None
885
929
 
886
930
class PythonGroupCompressor(_CommonGroupCompressor):
887
931
 
888
 
    def __init__(self):
 
932
    def __init__(self, settings=None):
889
933
        """Create a GroupCompressor.
890
934
 
891
935
        Used only if the pyrex version is not available.
892
936
        """
893
 
        super(PythonGroupCompressor, self).__init__()
 
937
        super(PythonGroupCompressor, self).__init__(settings)
894
938
        self._delta_index = LinesDeltaIndex([])
895
939
        # The actual content is managed by LinesDeltaIndex
896
940
        self.chunks = self._delta_index.lines
933
977
 
934
978
    It contains code very similar to SequenceMatcher because of having a similar
935
979
    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
939
 
       edits only.
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
944
 
       left side.
 
980
 
 
981
    * there is no junk, we want a minimal edit not a human readable diff.
 
982
    * we don't filter very common lines (because we don't know where a good
 
983
      range will start, and after the first text we want to be emitting minmal
 
984
      edits only.
 
985
    * we chain the left side, not the right side
 
986
    * we incrementally update the adjacency matrix as new lines are provided.
 
987
    * we look for matches in all of the left side, so the routine which does
 
988
      the analagous task of find_longest_match does not need to filter on the
 
989
      left side.
945
990
    """
946
991
 
947
 
    def __init__(self):
948
 
        super(PyrexGroupCompressor, self).__init__()
949
 
        self._delta_index = DeltaIndex()
 
992
    def __init__(self, settings=None):
 
993
        super(PyrexGroupCompressor, self).__init__(settings)
 
994
        max_bytes_to_index = self._settings.get('max_bytes_to_index', 0)
 
995
        self._delta_index = DeltaIndex(max_bytes_to_index=max_bytes_to_index)
950
996
 
951
997
    def _compress(self, key, bytes, max_delta_size, soft=False):
952
998
        """see _CommonGroupCompressor._compress"""
1027
1073
        index = _GCGraphIndex(graph_index, lambda:True, parents=parents,
1028
1074
            add_callback=graph_index.add_nodes,
1029
1075
            inconsistency_fatal=inconsistency_fatal)
1030
 
        access = knit._DirectPackAccess({})
 
1076
        access = pack_repo._DirectPackAccess({})
1031
1077
        access.set_writer(writer, graph_index, (transport, 'newpack'))
1032
1078
        result = GroupCompressVersionedFiles(index, access, delta)
1033
1079
        result.stream = stream
1043
1089
 
1044
1090
class _BatchingBlockFetcher(object):
1045
1091
    """Fetch group compress blocks in batches.
1046
 
    
 
1092
 
1047
1093
    :ivar total_bytes: int of expected number of bytes needed to fetch the
1048
1094
        currently pending batch.
1049
1095
    """
1050
1096
 
1051
 
    def __init__(self, gcvf, locations):
 
1097
    def __init__(self, gcvf, locations, get_compressor_settings=None):
1052
1098
        self.gcvf = gcvf
1053
1099
        self.locations = locations
1054
1100
        self.keys = []
1057
1103
        self.total_bytes = 0
1058
1104
        self.last_read_memo = None
1059
1105
        self.manager = None
 
1106
        self._get_compressor_settings = get_compressor_settings
1060
1107
 
1061
1108
    def add_key(self, key):
1062
1109
        """Add another to key to fetch.
1063
 
        
 
1110
 
1064
1111
        :return: The estimated number of bytes needed to fetch the batch so
1065
1112
            far.
1066
1113
        """
1091
1138
            # and then.
1092
1139
            self.batch_memos[read_memo] = cached_block
1093
1140
        return self.total_bytes
1094
 
        
 
1141
 
1095
1142
    def _flush_manager(self):
1096
1143
        if self.manager is not None:
1097
1144
            for factory in self.manager.get_record_stream():
1102
1149
    def yield_factories(self, full_flush=False):
1103
1150
        """Yield factories for keys added since the last yield.  They will be
1104
1151
        returned in the order they were added via add_key.
1105
 
        
 
1152
 
1106
1153
        :param full_flush: by default, some results may not be returned in case
1107
1154
            they can be part of the next batch.  If full_flush is True, then
1108
1155
            all results are returned.
1127
1174
                if memos_to_get_stack and memos_to_get_stack[-1] == read_memo:
1128
1175
                    # The next block from _get_blocks will be the block we
1129
1176
                    # need.
1130
 
                    block_read_memo, block = blocks.next()
 
1177
                    block_read_memo, block = next(blocks)
1131
1178
                    if block_read_memo != read_memo:
1132
1179
                        raise AssertionError(
1133
1180
                            "block_read_memo out of sync with read_memo"
1136
1183
                    memos_to_get_stack.pop()
1137
1184
                else:
1138
1185
                    block = self.batch_memos[read_memo]
1139
 
                self.manager = _LazyGroupContentManager(block)
 
1186
                self.manager = _LazyGroupContentManager(block,
 
1187
                    get_compressor_settings=self._get_compressor_settings)
1140
1188
                self.last_read_memo = read_memo
1141
1189
            start, end = index_memo[3:5]
1142
1190
            self.manager.add_factory(key, parents, start, end)
1149
1197
        self.total_bytes = 0
1150
1198
 
1151
1199
 
1152
 
class GroupCompressVersionedFiles(VersionedFiles):
 
1200
class GroupCompressVersionedFiles(VersionedFilesWithFallbacks):
1153
1201
    """A group-compress based VersionedFiles implementation."""
1154
1202
 
1155
 
    def __init__(self, index, access, delta=True, _unadded_refs=None):
 
1203
    # This controls how the GroupCompress DeltaIndex works. Basically, we
 
1204
    # compute hash pointers into the source blocks (so hash(text) => text).
 
1205
    # However each of these references costs some memory in trade against a
 
1206
    # more accurate match result. For very large files, they either are
 
1207
    # pre-compressed and change in bulk whenever they change, or change in just
 
1208
    # local blocks. Either way, 'improved resolution' is not very helpful,
 
1209
    # versus running out of memory trying to track everything. The default max
 
1210
    # gives 100% sampling of a 1MB file.
 
1211
    _DEFAULT_MAX_BYTES_TO_INDEX = 1024 * 1024
 
1212
    _DEFAULT_COMPRESSOR_SETTINGS = {'max_bytes_to_index':
 
1213
                                     _DEFAULT_MAX_BYTES_TO_INDEX}
 
1214
 
 
1215
    def __init__(self, index, access, delta=True, _unadded_refs=None,
 
1216
                 _group_cache=None):
1156
1217
        """Create a GroupCompressVersionedFiles object.
1157
1218
 
1158
1219
        :param index: The index object storing access and graph data.
1159
1220
        :param access: The access object storing raw data.
1160
1221
        :param delta: Whether to delta compress or just entropy compress.
1161
1222
        :param _unadded_refs: private parameter, don't use.
 
1223
        :param _group_cache: private parameter, don't use.
1162
1224
        """
1163
1225
        self._index = index
1164
1226
        self._access = access
1166
1228
        if _unadded_refs is None:
1167
1229
            _unadded_refs = {}
1168
1230
        self._unadded_refs = _unadded_refs
1169
 
        self._group_cache = LRUSizeCache(max_size=50*1024*1024)
1170
 
        self._fallback_vfs = []
 
1231
        if _group_cache is None:
 
1232
            _group_cache = LRUSizeCache(max_size=50*1024*1024)
 
1233
        self._group_cache = _group_cache
 
1234
        self._immediate_fallback_vfs = []
 
1235
        self._max_bytes_to_index = None
1171
1236
 
1172
1237
    def without_fallbacks(self):
1173
1238
        """Return a clone of this object without any fallbacks configured."""
1174
1239
        return GroupCompressVersionedFiles(self._index, self._access,
1175
 
            self._delta, _unadded_refs=dict(self._unadded_refs))
 
1240
            self._delta, _unadded_refs=dict(self._unadded_refs),
 
1241
            _group_cache=self._group_cache)
1176
1242
 
1177
1243
    def add_lines(self, key, parents, lines, parent_texts=None,
1178
1244
        left_matching_blocks=None, nostore_sha=None, random_id=False,
1182
1248
        :param key: The key tuple of the text to add.
1183
1249
        :param parents: The parents key tuples of the text to add.
1184
1250
        :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
 
1251
            of them except the last must be terminated with \\n and contain no
 
1252
            other \\n's. The last line may either contain no \\n's or a single
 
1253
            terminating \\n. If the lines list does meet this constraint the
 
1254
            add routine may error or may succeed - but you will be unable to
 
1255
            read the data back accurately. (Checking the lines have been split
1190
1256
            correctly is expensive and extremely unlikely to catch bugs so it
1191
1257
            is not done at runtime unless check_content is True.)
1192
1258
        :param parent_texts: An optional dictionary containing the opaque
1247
1313
 
1248
1314
        :param a_versioned_files: A VersionedFiles object.
1249
1315
        """
1250
 
        self._fallback_vfs.append(a_versioned_files)
 
1316
        self._immediate_fallback_vfs.append(a_versioned_files)
1251
1317
 
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)
1289
1355
 
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
1294
 
        # ancestry.
1295
 
        parent_map, missing_keys = self._index.find_ancestry(keys)
1296
 
        for fallback in self._fallback_vfs:
1297
 
            if not missing_keys:
1298
 
                break
1299
 
            (f_parent_map, f_missing_keys) = fallback._index.find_ancestry(
1300
 
                                                missing_keys)
1301
 
            parent_map.update(f_parent_map)
1302
 
            missing_keys = f_missing_keys
1303
 
        kg = _mod_graph.KnownGraph(parent_map)
1304
 
        return kg
1305
 
 
1306
1356
    def get_parent_map(self, keys):
1307
1357
        """Get a map of the graph parents of keys.
1308
1358
 
1323
1373
            and so on.
1324
1374
        """
1325
1375
        result = {}
1326
 
        sources = [self._index] + self._fallback_vfs
 
1376
        sources = [self._index] + self._immediate_fallback_vfs
1327
1377
        source_results = []
1328
1378
        missing = set(keys)
1329
1379
        for source in sources:
1366
1416
                yield read_memo, cached[read_memo]
1367
1417
            except KeyError:
1368
1418
                # Read the block, and cache it.
1369
 
                zdata = raw_records.next()
 
1419
                zdata = next(raw_records)
1370
1420
                block = GroupCompressBlock.from_bytes(zdata)
1371
1421
                self._group_cache[read_memo] = block
1372
1422
                cached[read_memo] = block
1414
1464
                    remaining_keys.discard(content_factory.key)
1415
1465
                    yield content_factory
1416
1466
                return
1417
 
            except errors.RetryWithNewPacks, e:
 
1467
            except errors.RetryWithNewPacks as e:
1418
1468
                self._access.reload_or_raise(e)
1419
1469
 
1420
1470
    def _find_from_fallback(self, missing):
1430
1480
        parent_map = {}
1431
1481
        key_to_source_map = {}
1432
1482
        source_results = []
1433
 
        for source in self._fallback_vfs:
 
1483
        for source in self._immediate_fallback_vfs:
1434
1484
            if not missing:
1435
1485
                break
1436
1486
            source_parents = source.get_parent_map(missing)
1446
1496
 
1447
1497
        The returned objects should be in the order defined by 'ordering',
1448
1498
        which can weave between different sources.
 
1499
 
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.
1452
1503
        """
1453
1504
        if ordering == 'topological':
1454
 
            present_keys = topo_sort(parent_map)
 
1505
            present_keys = tsort.topo_sort(parent_map)
1455
1506
        else:
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)
1526
1577
            # start with one key, recurse to its oldest parent, then grab
1527
1578
            # everything in the same group, etc.
1528
1579
            parent_map = dict((key, details[2]) for key, details in
1529
 
                locations.iteritems())
 
1580
                viewitems(locations))
1530
1581
            for key in unadded_keys:
1531
1582
                parent_map[key] = self._unadded_refs[key]
1532
1583
            parent_map.update(fallback_parent_map)
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):
1599
1651
            pass
1600
1652
 
 
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')
 
1660
            if val is not None:
 
1661
                try:
 
1662
                    val = int(val)
 
1663
                except ValueError as e:
 
1664
                    trace.warning('Value for '
 
1665
                                  '"bzr.groupcompress.max_bytes_to_index"'
 
1666
                                  ' %r is not an integer'
 
1667
                                  % (val,))
 
1668
                    val = None
 
1669
            if val is None:
 
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}
 
1673
 
 
1674
    def _make_group_compressor(self):
 
1675
        return GroupCompressor(self._get_compressor_settings())
 
1676
 
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.
1626
1702
                return adapter
1627
1703
        # This will go up to fulltexts for gc to gc fetching, which isn't
1628
1704
        # ideal.
1629
 
        self._compressor = GroupCompressor()
 
1705
        self._compressor = self._make_group_compressor()
1630
1706
        self._unadded_refs = {}
1631
1707
        keys_to_add = []
1632
1708
        def flush():
1633
 
            bytes = self._compressor.flush().to_bytes()
1634
 
            self._compressor = GroupCompressor()
 
1709
            bytes_len, chunks = self._compressor.flush().to_chunks()
 
1710
            self._compressor = self._make_group_compressor()
 
1711
            # Note: At this point we still have 1 copy of the fulltext (in
 
1712
            #       record and the var 'bytes'), and this generates 2 copies of
 
1713
            #       the compressed text (one for bytes, one in chunks)
 
1714
            # TODO: Push 'chunks' down into the _access api, so that we don't
 
1715
            #       have to double compressed memory here
 
1716
            # TODO: Figure out how to indicate that we would be happy to free
 
1717
            #       the fulltext content at this point. Note that sometimes we
 
1718
            #       will want it later (streaming CHK pages), but most of the
 
1719
            #       time we won't (everything else)
 
1720
            bytes = ''.join(chunks)
 
1721
            del chunks
1635
1722
            index, start, length = self._access.add_raw_records(
1636
1723
                [(None, len(bytes))], bytes)[0]
1637
1724
            nodes = []
1656
1743
                raise errors.RevisionNotPresent(record.key, self)
1657
1744
            if random_id:
1658
1745
                if record.key in inserted_keys:
1659
 
                    trace.note('Insert claimed random_id=True,'
1660
 
                               ' but then inserted %r two times', record.key)
 
1746
                    trace.note(gettext('Insert claimed random_id=True,'
 
1747
                               ' but then inserted %r two times'), record.key)
1661
1748
                    continue
1662
1749
                inserted_keys.add(record.key)
1663
1750
            if reuse_blocks:
1802
1889
        """See VersionedFiles.keys."""
1803
1890
        if 'evil' in debug.debug_flags:
1804
1891
            trace.mutter_callsite(2, "keys scales with size of history")
1805
 
        sources = [self._index] + self._fallback_vfs
 
1892
        sources = [self._index] + self._immediate_fallback_vfs
1806
1893
        result = set()
1807
1894
        for source in sources:
1808
1895
            result.update(source.keys())
1809
1896
        return result
1810
1897
 
1811
1898
 
 
1899
class _GCBuildDetails(object):
 
1900
    """A blob of data about the build details.
 
1901
 
 
1902
    This stores the minimal data, which then allows compatibility with the old
 
1903
    api, without taking as much memory.
 
1904
    """
 
1905
 
 
1906
    __slots__ = ('_index', '_group_start', '_group_end', '_basis_end',
 
1907
                 '_delta_end', '_parents')
 
1908
 
 
1909
    method = 'group'
 
1910
    compression_parent = None
 
1911
 
 
1912
    def __init__(self, parents, position_info):
 
1913
        self._parents = parents
 
1914
        (self._index, self._group_start, self._group_end, self._basis_end,
 
1915
         self._delta_end) = position_info
 
1916
 
 
1917
    def __repr__(self):
 
1918
        return '%s(%s, %s)' % (self.__class__.__name__,
 
1919
            self.index_memo, self._parents)
 
1920
 
 
1921
    @property
 
1922
    def index_memo(self):
 
1923
        return (self._index, self._group_start, self._group_end,
 
1924
                self._basis_end, self._delta_end)
 
1925
 
 
1926
    @property
 
1927
    def record_details(self):
 
1928
        return static_tuple.StaticTuple(self.method, None)
 
1929
 
 
1930
    def __getitem__(self, offset):
 
1931
        """Compatibility thunk to act like a tuple."""
 
1932
        if offset == 0:
 
1933
            return self.index_memo
 
1934
        elif offset == 1:
 
1935
            return self.compression_parent # Always None
 
1936
        elif offset == 2:
 
1937
            return self._parents
 
1938
        elif offset == 3:
 
1939
            return self.record_details
 
1940
        else:
 
1941
            raise IndexError('offset out of range')
 
1942
            
 
1943
    def __len__(self):
 
1944
        return 4
 
1945
 
 
1946
 
1812
1947
class _GCGraphIndex(object):
1813
1948
    """Mapper from GroupCompressVersionedFiles needs into GraphIndex storage."""
1814
1949
 
1817
1952
        inconsistency_fatal=True, track_new_keys=False):
1818
1953
        """Construct a _GCGraphIndex on a graph_index.
1819
1954
 
1820
 
        :param graph_index: An implementation of bzrlib.index.GraphIndex.
 
1955
        :param graph_index: An implementation of breezy.index.GraphIndex.
1821
1956
        :param is_locked: A callback, returns True if the index is locked and
1822
1957
            thus usable.
1823
1958
        :param parents: If True, record knits parents, if not do not record
1843
1978
        # repeated over and over, this creates a surplus of ints
1844
1979
        self._int_cache = {}
1845
1980
        if track_external_parent_refs:
1846
 
            self._key_dependencies = knit._KeyRefs(
 
1981
            self._key_dependencies = _KeyRefs(
1847
1982
                track_new_keys=track_new_keys)
1848
1983
        else:
1849
1984
            self._key_dependencies = None
1900
2035
        if changed:
1901
2036
            result = []
1902
2037
            if self._parents:
1903
 
                for key, (value, node_refs) in keys.iteritems():
 
2038
                for key, (value, node_refs) in viewitems(keys):
1904
2039
                    result.append((key, value, node_refs))
1905
2040
            else:
1906
 
                for key, (value, node_refs) in keys.iteritems():
 
2041
                for key, (value, node_refs) in viewitems(keys):
1907
2042
                    result.append((key, value))
1908
2043
            records = result
1909
2044
        key_dependencies = self._key_dependencies
1989
2124
        :param keys: An iterable of keys.
1990
2125
        :return: A dict of key:
1991
2126
            (index_memo, compression_parent, parents, record_details).
1992
 
            index_memo
1993
 
                opaque structure to pass to read_records to extract the raw
1994
 
                data
1995
 
            compression_parent
1996
 
                Content that this record is built upon, may be None
1997
 
            parents
1998
 
                Logical parents of this node
1999
 
            record_details
2000
 
                extra information about the content which needs to be passed to
2001
 
                Factory.parse_record
 
2127
 
 
2128
            * index_memo: opaque structure to pass to read_records to extract
 
2129
              the raw data
 
2130
            * compression_parent: Content that this record is built upon, may
 
2131
              be None
 
2132
            * parents: Logical parents of this node
 
2133
            * record_details: extra information about the content which needs
 
2134
              to be passed to Factory.parse_record
2002
2135
        """
2003
2136
        self._check_read()
2004
2137
        result = {}
2009
2142
                parents = None
2010
2143
            else:
2011
2144
                parents = entry[3][0]
2012
 
            method = 'group'
2013
 
            result[key] = (self._node_to_position(entry),
2014
 
                                  None, parents, (method, None))
 
2145
            details = _GCBuildDetails(parents, self._node_to_position(entry))
 
2146
            result[key] = details
2015
2147
        return result
2016
2148
 
2017
2149
    def keys(self):
2033
2165
        # each, or about 7MB. Note that it might be even more when you consider
2034
2166
        # how PyInt is allocated in separate slabs. And you can't return a slab
2035
2167
        # 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
 
2168
        # a LIFO when re-using PyInt slots, which might cause more
2037
2169
        # fragmentation.
2038
2170
        start = int(bits[0])
2039
2171
        start = self._int_cache.setdefault(start, start)
2063
2195
            key_dependencies.add_references(node[1], node[3][0])
2064
2196
 
2065
2197
 
2066
 
from bzrlib._groupcompress_py import (
 
2198
from ._groupcompress_py import (
2067
2199
    apply_delta,
2068
2200
    apply_delta_to_source,
2069
2201
    encode_base128_int,
2072
2204
    LinesDeltaIndex,
2073
2205
    )
2074
2206
try:
2075
 
    from bzrlib._groupcompress_pyx import (
 
2207
    from ._groupcompress_pyx import (
2076
2208
        apply_delta,
2077
2209
        apply_delta_to_source,
2078
2210
        DeltaIndex,
2080
2212
        decode_base128_int,
2081
2213
        )
2082
2214
    GroupCompressor = PyrexGroupCompressor
2083
 
except ImportError, e:
 
2215
except ImportError as e:
2084
2216
    osutils.failed_to_load_extension(e)
2085
2217
    GroupCompressor = PythonGroupCompressor
2086
2218