/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/groupcompress.py

  • Committer: Jelmer Vernooij
  • Date: 2017-06-10 01:35:53 UTC
  • mto: (6670.4.8 move-bzr)
  • mto: This revision was merged to the branch mainline in revision 6681.
  • Revision ID: jelmer@jelmer.uk-20170610013553-560y7mn3su4pp763
Fix remaining tests.

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