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

  • Committer: Robert Collins
  • Date: 2010-05-06 11:08:10 UTC
  • mto: This revision was merged to the branch mainline in revision 5223.
  • Revision ID: robertc@robertcollins.net-20100506110810-h3j07fh5gmw54s25
Cleaner matcher matching revised unlocking protocol.

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2008-2011 Canonical Ltd
 
1
# Copyright (C) 2008, 2009, 2010 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
 
 
21
19
import time
22
20
import zlib
 
21
try:
 
22
    import pylzma
 
23
except ImportError:
 
24
    pylzma = None
23
25
 
24
 
from ..lazy_import import lazy_import
25
 
lazy_import(globals(), """
26
 
from breezy import (
 
26
from bzrlib import (
27
27
    annotate,
28
 
    config,
29
28
    debug,
 
29
    errors,
30
30
    graph as _mod_graph,
 
31
    knit,
31
32
    osutils,
 
33
    pack,
32
34
    static_tuple,
33
35
    trace,
34
 
    tsort,
35
 
    )
36
 
from breezy.bzr import (
37
 
    knit,
38
 
    pack,
39
 
    pack_repo,
40
 
    )
41
 
 
42
 
from breezy.i18n import gettext
43
 
""")
44
 
 
45
 
from .. import (
46
 
    errors,
47
 
    )
48
 
from .btree_index import BTreeBuilder
49
 
from ..lru_cache import LRUSizeCache
50
 
from ..sixish import (
51
 
    indexbytes,
52
 
    map,
53
 
    range,
54
 
    viewitems,
55
 
    )
56
 
from .versionedfile import (
57
 
    _KeyRefs,
 
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 (
58
41
    adapter_registry,
59
42
    AbsentContentFactory,
60
43
    ChunkedContentFactory,
61
44
    FulltextContentFactory,
62
 
    VersionedFilesWithFallbacks,
 
45
    VersionedFiles,
63
46
    )
64
47
 
65
48
# Minimum number of uncompressed bytes to try fetch at once when retrieving
66
49
# groupcompress blocks.
67
50
BATCH_SIZE = 2**16
68
51
 
 
52
_USE_LZMA = False and (pylzma is not None)
 
53
 
69
54
# osutils.sha_string('')
70
 
_null_sha1 = b'da39a3ee5e6b4b0d3255bfef95601890afd80709'
 
55
_null_sha1 = 'da39a3ee5e6b4b0d3255bfef95601890afd80709'
71
56
 
72
57
def sort_gc_optimal(parent_map):
73
58
    """Sort and group the keys in parent_map into groupcompress order.
80
65
    # groupcompress ordering is approximately reverse topological,
81
66
    # properly grouped by file-id.
82
67
    per_prefix_map = {}
83
 
    for key, value in viewitems(parent_map):
84
 
        if isinstance(key, bytes) or len(key) == 1:
85
 
            prefix = b''
 
68
    for key, value in parent_map.iteritems():
 
69
        if isinstance(key, str) or len(key) == 1:
 
70
            prefix = ''
86
71
        else:
87
72
            prefix = key[0]
88
73
        try:
92
77
 
93
78
    present_keys = []
94
79
    for prefix in sorted(per_prefix_map):
95
 
        present_keys.extend(reversed(tsort.topo_sort(per_prefix_map[prefix])))
 
80
        present_keys.extend(reversed(topo_sort(per_prefix_map[prefix])))
96
81
    return present_keys
97
82
 
98
83
 
99
 
class DecompressCorruption(errors.BzrError):
100
 
 
101
 
    _fmt = "Corruption while decompressing repository file%(orig_error)s"
102
 
 
103
 
    def __init__(self, orig_error=None):
104
 
        if orig_error is not None:
105
 
            self.orig_error = ", %s" % (orig_error,)
106
 
        else:
107
 
            self.orig_error = ""
108
 
        errors.BzrError.__init__(self)
109
 
 
110
 
 
111
84
# The max zlib window size is 32kB, so if we set 'max_size' output of the
112
85
# decompressor to the requested bytes + 32kB, then we should guarantee
113
86
# num_bytes coming out.
120
93
    """
121
94
 
122
95
    # Group Compress Block v1 Zlib
123
 
    GCB_HEADER = b'gcb1z\n'
 
96
    GCB_HEADER = 'gcb1z\n'
124
97
    # Group Compress Block v1 Lzma
125
 
    GCB_LZ_HEADER = b'gcb1l\n'
 
98
    GCB_LZ_HEADER = 'gcb1l\n'
126
99
    GCB_KNOWN_HEADERS = (GCB_HEADER, GCB_LZ_HEADER)
127
100
 
128
101
    def __init__(self):
129
102
        # map by key? or just order in file?
130
103
        self._compressor_name = None
131
 
        self._z_content_chunks = None
 
104
        self._z_content = None
132
105
        self._z_content_decompressor = None
133
106
        self._z_content_length = None
134
107
        self._content_length = None
159
132
        # Expand the content if required
160
133
        if self._content is None:
161
134
            if self._content_chunks is not None:
162
 
                self._content = b''.join(self._content_chunks)
 
135
                self._content = ''.join(self._content_chunks)
163
136
                self._content_chunks = None
164
137
        if self._content is None:
165
 
            # We join self._z_content_chunks here, because if we are
166
 
            # decompressing, then it is *very* likely that we have a single
167
 
            # chunk
168
 
            if self._z_content_chunks is None:
 
138
            if self._z_content is None:
169
139
                raise AssertionError('No content to decompress')
170
 
            z_content = b''.join(self._z_content_chunks)
171
 
            if z_content == b'':
172
 
                self._content = b''
 
140
            if self._z_content == '':
 
141
                self._content = ''
173
142
            elif self._compressor_name == 'lzma':
174
143
                # We don't do partial lzma decomp yet
175
 
                import pylzma
176
 
                self._content = pylzma.decompress(z_content)
 
144
                self._content = pylzma.decompress(self._z_content)
177
145
            elif self._compressor_name == 'zlib':
178
146
                # Start a zlib decompressor
179
147
                if num_bytes * 4 > self._content_length * 3:
180
148
                    # If we are requesting more that 3/4ths of the content,
181
149
                    # just extract the whole thing in a single pass
182
150
                    num_bytes = self._content_length
183
 
                    self._content = zlib.decompress(z_content)
 
151
                    self._content = zlib.decompress(self._z_content)
184
152
                else:
185
153
                    self._z_content_decompressor = zlib.decompressobj()
186
154
                    # Seed the decompressor with the uncompressed bytes, so
187
155
                    # that the rest of the code is simplified
188
156
                    self._content = self._z_content_decompressor.decompress(
189
 
                        z_content, num_bytes + _ZLIB_DECOMP_WINDOW)
 
157
                        self._z_content, num_bytes + _ZLIB_DECOMP_WINDOW)
190
158
                    if not self._z_content_decompressor.unconsumed_tail:
191
159
                        self._z_content_decompressor = None
192
160
            else:
219
187
            # The stream is finished
220
188
            self._z_content_decompressor = None
221
189
 
222
 
    def _parse_bytes(self, data, pos):
 
190
    def _parse_bytes(self, bytes, pos):
223
191
        """Read the various lengths from the header.
224
192
 
225
193
        This also populates the various 'compressed' buffers.
229
197
        # At present, we have 2 integers for the compressed and uncompressed
230
198
        # content. In base10 (ascii) 14 bytes can represent > 1TB, so to avoid
231
199
        # checking too far, cap the search to 14 bytes.
232
 
        pos2 = data.index(b'\n', pos, pos + 14)
233
 
        self._z_content_length = int(data[pos:pos2])
234
 
        pos = pos2 + 1
235
 
        pos2 = data.index(b'\n', pos, pos + 14)
236
 
        self._content_length = int(data[pos:pos2])
237
 
        pos = pos2 + 1
238
 
        if len(data) != (pos + self._z_content_length):
 
200
        pos2 = bytes.index('\n', pos, pos + 14)
 
201
        self._z_content_length = int(bytes[pos:pos2])
 
202
        pos = pos2 + 1
 
203
        pos2 = bytes.index('\n', pos, pos + 14)
 
204
        self._content_length = int(bytes[pos:pos2])
 
205
        pos = pos2 + 1
 
206
        if len(bytes) != (pos + self._z_content_length):
239
207
            # XXX: Define some GCCorrupt error ?
240
208
            raise AssertionError('Invalid bytes: (%d) != %d + %d' %
241
 
                                 (len(data), pos, self._z_content_length))
242
 
        self._z_content_chunks = (data[pos:],)
243
 
 
244
 
    @property
245
 
    def _z_content(self):
246
 
        """Return z_content_chunks as a simple string.
247
 
 
248
 
        Meant only to be used by the test suite.
249
 
        """
250
 
        if self._z_content_chunks is not None:
251
 
            return b''.join(self._z_content_chunks)
252
 
        return None
 
209
                                 (len(bytes), pos, self._z_content_length))
 
210
        self._z_content = bytes[pos:]
253
211
 
254
212
    @classmethod
255
213
    def from_bytes(cls, bytes):
256
214
        out = cls()
257
 
        header = bytes[:6]
258
 
        if header not in cls.GCB_KNOWN_HEADERS:
 
215
        if bytes[:6] not in cls.GCB_KNOWN_HEADERS:
259
216
            raise ValueError('bytes did not start with any of %r'
260
217
                             % (cls.GCB_KNOWN_HEADERS,))
261
 
        if header == cls.GCB_HEADER:
 
218
        # XXX: why not testing the whole header ?
 
219
        if bytes[4] == 'z':
262
220
            out._compressor_name = 'zlib'
263
 
        elif header == cls.GCB_LZ_HEADER:
 
221
        elif bytes[4] == 'l':
264
222
            out._compressor_name = 'lzma'
265
223
        else:
266
 
            raise ValueError('unknown compressor: %r' % (header,))
 
224
            raise ValueError('unknown compressor: %r' % (bytes,))
267
225
        out._parse_bytes(bytes, 6)
268
226
        return out
269
227
 
275
233
        :return: The bytes for the content
276
234
        """
277
235
        if start == end == 0:
278
 
            return b''
 
236
            return ''
279
237
        self._ensure_content(end)
280
238
        # The bytes are 'f' or 'd' for the type, then a variable-length
281
239
        # base128 integer for the content size, then the actual content
282
240
        # We know that the variable-length integer won't be longer than 5
283
241
        # bytes (it takes 5 bytes to encode 2^32)
284
 
        c = self._content[start:start + 1]
285
 
        if c == b'f':
 
242
        c = self._content[start]
 
243
        if c == 'f':
286
244
            type = 'fulltext'
287
245
        else:
288
 
            if c != b'd':
 
246
            if c != 'd':
289
247
                raise ValueError('Unknown content control code: %s'
290
248
                                 % (c,))
291
249
            type = 'delta'
295
253
        if end != content_start + content_len:
296
254
            raise ValueError('end != len according to field header'
297
255
                ' %s != %s' % (end, content_start + content_len))
298
 
        if c == b'f':
299
 
            return self._content[content_start:end]
300
 
        # Must be type delta as checked above
301
 
        return apply_delta_to_source(self._content, content_start, end)
 
256
        if c == 'f':
 
257
            bytes = self._content[content_start:end]
 
258
        elif c == 'd':
 
259
            bytes = apply_delta_to_source(self._content, content_start, end)
 
260
        return bytes
302
261
 
303
262
    def set_chunked_content(self, content_chunks, length):
304
263
        """Set the content of this block to the given chunks."""
310
269
        self._content_length = length
311
270
        self._content_chunks = content_chunks
312
271
        self._content = None
313
 
        self._z_content_chunks = None
 
272
        self._z_content = None
314
273
 
315
274
    def set_content(self, content):
316
275
        """Set the content of this block."""
317
276
        self._content_length = len(content)
318
277
        self._content = content
319
 
        self._z_content_chunks = None
320
 
 
321
 
    def _create_z_content_from_chunks(self, chunks):
 
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):
322
290
        compressor = zlib.compressobj(zlib.Z_DEFAULT_COMPRESSION)
323
 
        # Peak in this point is 1 fulltext, 1 compressed text, + zlib overhead
324
 
        # (measured peak is maybe 30MB over the above...)
325
 
        compressed_chunks = list(map(compressor.compress, chunks))
 
291
        compressed_chunks = map(compressor.compress, self._content_chunks)
326
292
        compressed_chunks.append(compressor.flush())
327
 
        # Ignore empty chunks
328
 
        self._z_content_chunks = [c for c in compressed_chunks if c]
329
 
        self._z_content_length = sum(map(len, self._z_content_chunks))
 
293
        self._z_content = ''.join(compressed_chunks)
 
294
        self._z_content_length = len(self._z_content)
330
295
 
331
296
    def _create_z_content(self):
332
 
        if self._z_content_chunks is not None:
 
297
        if self._z_content is not None:
 
298
            return
 
299
        if _USE_LZMA:
 
300
            self._create_z_content_using_lzma()
333
301
            return
334
302
        if self._content_chunks is not None:
335
 
            chunks = self._content_chunks
336
 
        else:
337
 
            chunks = (self._content,)
338
 
        self._create_z_content_from_chunks(chunks)
 
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)
339
307
 
340
 
    def to_chunks(self):
341
 
        """Create the byte stream as a series of 'chunks'"""
 
308
    def to_bytes(self):
 
309
        """Encode the information into a byte stream."""
342
310
        self._create_z_content()
343
 
        header = self.GCB_HEADER
344
 
        chunks = [b'%s%d\n%d\n'
345
 
                  % (header, self._z_content_length, self._content_length),
 
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,
346
318
                 ]
347
 
        chunks.extend(self._z_content_chunks)
348
 
        total_len = sum(map(len, chunks))
349
 
        return total_len, chunks
350
 
 
351
 
    def to_bytes(self):
352
 
        """Encode the information into a byte stream."""
353
 
        total_len, chunks = self.to_chunks()
354
 
        return b''.join(chunks)
 
319
        return ''.join(chunks)
355
320
 
356
321
    def _dump(self, include_text=False):
357
322
        """Take this block, and spit out a human-readable structure.
367
332
        result = []
368
333
        pos = 0
369
334
        while pos < self._content_length:
370
 
            kind = self._content[pos:pos + 1]
 
335
            kind = self._content[pos]
371
336
            pos += 1
372
 
            if kind not in (b'f', b'd'):
 
337
            if kind not in ('f', 'd'):
373
338
                raise ValueError('invalid kind character: %r' % (kind,))
374
339
            content_len, len_len = decode_base128_int(
375
340
                                self._content[pos:pos + 5])
377
342
            if content_len + pos > self._content_length:
378
343
                raise ValueError('invalid content_len %d for record @ pos %d'
379
344
                                 % (content_len, pos - len_len - 1))
380
 
            if kind == b'f': # Fulltext
 
345
            if kind == 'f': # Fulltext
381
346
                if include_text:
382
347
                    text = self._content[pos:pos+content_len]
383
 
                    result.append((b'f', content_len, text))
 
348
                    result.append(('f', content_len, text))
384
349
                else:
385
 
                    result.append((b'f', content_len))
386
 
            elif kind == b'd': # Delta
 
350
                    result.append(('f', content_len))
 
351
            elif kind == 'd': # Delta
387
352
                delta_content = self._content[pos:pos+content_len]
388
353
                delta_info = []
389
354
                # The first entry in a delta is the decompressed length
390
355
                decomp_len, delta_pos = decode_base128_int(delta_content)
391
 
                result.append((b'd', content_len, decomp_len, delta_info))
 
356
                result.append(('d', content_len, decomp_len, delta_info))
392
357
                measured_len = 0
393
358
                while delta_pos < content_len:
394
 
                    c = indexbytes(delta_content, delta_pos)
 
359
                    c = ord(delta_content[delta_pos])
395
360
                    delta_pos += 1
396
361
                    if c & 0x80: # Copy
397
362
                        (offset, length,
399
364
                                                              delta_pos)
400
365
                        if include_text:
401
366
                            text = self._content[offset:offset+length]
402
 
                            delta_info.append((b'c', offset, length, text))
 
367
                            delta_info.append(('c', offset, length, text))
403
368
                        else:
404
 
                            delta_info.append((b'c', offset, length))
 
369
                            delta_info.append(('c', offset, length))
405
370
                        measured_len += length
406
371
                    else: # Insert
407
372
                        if include_text:
408
373
                            txt = delta_content[delta_pos:delta_pos+c]
409
374
                        else:
410
 
                            txt = b''
411
 
                        delta_info.append((b'i', c, txt))
 
375
                            txt = ''
 
376
                        delta_info.append(('i', c, txt))
412
377
                        measured_len += c
413
378
                        delta_pos += c
414
379
                if delta_pos != content_len:
464
429
                # wire bytes, something...
465
430
                return self._manager._wire_bytes()
466
431
            else:
467
 
                return b''
 
432
                return ''
468
433
        if storage_kind in ('fulltext', 'chunked'):
469
434
            if self._bytes is None:
470
435
                # Grab and cache the raw bytes for this entry
471
436
                # and break the ref-cycle with _manager since we don't need it
472
437
                # anymore
473
 
                try:
474
 
                    self._manager._prepare_for_extract()
475
 
                except zlib.error as value:
476
 
                    raise DecompressCorruption("zlib: " + str(value))
 
438
                self._manager._prepare_for_extract()
477
439
                block = self._manager._block
478
440
                self._bytes = block.extract(self.key, self._start, self._end)
479
441
                # There are code paths that first extract as fulltext, and then
498
460
    _full_enough_block_size = 3*1024*1024 # size at which we won't repack
499
461
    _full_enough_mixed_block_size = 2*768*1024 # 1.5MB
500
462
 
501
 
    def __init__(self, block, get_compressor_settings=None):
 
463
    def __init__(self, block):
502
464
        self._block = block
503
465
        # We need to preserve the ordering
504
466
        self._factories = []
505
467
        self._last_byte = 0
506
 
        self._get_settings = get_compressor_settings
507
 
        self._compressor_settings = None
508
 
 
509
 
    def _get_compressor_settings(self):
510
 
        if self._compressor_settings is not None:
511
 
            return self._compressor_settings
512
 
        settings = None
513
 
        if self._get_settings is not None:
514
 
            settings = self._get_settings()
515
 
        if settings is None:
516
 
            vf = GroupCompressVersionedFiles
517
 
            settings = vf._DEFAULT_COMPRESSOR_SETTINGS
518
 
        self._compressor_settings = settings
519
 
        return self._compressor_settings
520
468
 
521
469
    def add_factory(self, key, parents, start, end):
522
470
        if not self._factories:
555
503
        new_block.set_content(self._block._content[:last_byte])
556
504
        self._block = new_block
557
505
 
558
 
    def _make_group_compressor(self):
559
 
        return GroupCompressor(self._get_compressor_settings())
560
 
 
561
506
    def _rebuild_block(self):
562
507
        """Create a new GroupCompressBlock with only the referenced texts."""
563
 
        compressor = self._make_group_compressor()
 
508
        compressor = GroupCompressor()
564
509
        tstart = time.time()
565
510
        old_length = self._block._content_length
566
511
        end_point = 0
578
523
        #       block? It seems hard to come up with a method that it would
579
524
        #       expand, since we do full compression again. Perhaps based on a
580
525
        #       request that ends up poorly ordered?
581
 
        # TODO: If the content would have expanded, then we would want to
582
 
        #       handle a case where we need to split the block.
583
 
        #       Now that we have a user-tweakable option
584
 
        #       (max_bytes_to_index), it is possible that one person set it
585
 
        #       to a very low value, causing poor compression.
586
526
        delta = time.time() - tstart
587
527
        self._block = new_block
588
528
        trace.mutter('creating new compressed block on-the-fly in %.3fs'
711
651
        #   <length of gc block>\n
712
652
        #   <header bytes>
713
653
        #   <gc-block>
714
 
        lines = [b'groupcompress-block\n']
 
654
        lines = ['groupcompress-block\n']
715
655
        # The minimal info we need is the key, the start offset, and the
716
656
        # parents. The length and type are encoded in the record itself.
717
657
        # However, passing in the other bits makes it easier.  The list of
722
662
        # 1 line for end byte
723
663
        header_lines = []
724
664
        for factory in self._factories:
725
 
            key_bytes = b'\x00'.join(factory.key)
 
665
            key_bytes = '\x00'.join(factory.key)
726
666
            parents = factory.parents
727
667
            if parents is None:
728
 
                parent_bytes = b'None:'
 
668
                parent_bytes = 'None:'
729
669
            else:
730
 
                parent_bytes = b'\t'.join(b'\x00'.join(key) for key in parents)
731
 
            record_header = b'%s\n%s\n%d\n%d\n' % (
 
670
                parent_bytes = '\t'.join('\x00'.join(key) for key in parents)
 
671
            record_header = '%s\n%s\n%d\n%d\n' % (
732
672
                key_bytes, parent_bytes, factory._start, factory._end)
733
673
            header_lines.append(record_header)
734
674
            # TODO: Can we break the refcycle at this point and set
735
675
            #       factory._manager = None?
736
 
        header_bytes = b''.join(header_lines)
 
676
        header_bytes = ''.join(header_lines)
737
677
        del header_lines
738
678
        header_bytes_len = len(header_bytes)
739
679
        z_header_bytes = zlib.compress(header_bytes)
740
680
        del header_bytes
741
681
        z_header_bytes_len = len(z_header_bytes)
742
 
        block_bytes_len, block_chunks = self._block.to_chunks()
743
 
        lines.append(b'%d\n%d\n%d\n' % (
744
 
            z_header_bytes_len, header_bytes_len, block_bytes_len))
 
682
        block_bytes = self._block.to_bytes()
 
683
        lines.append('%d\n%d\n%d\n' % (z_header_bytes_len, header_bytes_len,
 
684
                                       len(block_bytes)))
745
685
        lines.append(z_header_bytes)
746
 
        lines.extend(block_chunks)
747
 
        del z_header_bytes, block_chunks
748
 
        # TODO: This is a point where we will double the memory consumption. To
749
 
        #       avoid this, we probably have to switch to a 'chunked' api
750
 
        return b''.join(lines)
 
686
        lines.append(block_bytes)
 
687
        del z_header_bytes, block_bytes
 
688
        return ''.join(lines)
751
689
 
752
690
    @classmethod
753
691
    def from_bytes(cls, bytes):
754
692
        # TODO: This does extra string copying, probably better to do it a
755
 
        #       different way. At a minimum this creates 2 copies of the
756
 
        #       compressed content
 
693
        #       different way
757
694
        (storage_kind, z_header_len, header_len,
758
 
         block_len, rest) = bytes.split(b'\n', 4)
 
695
         block_len, rest) = bytes.split('\n', 4)
759
696
        del bytes
760
 
        if storage_kind != b'groupcompress-block':
 
697
        if storage_kind != 'groupcompress-block':
761
698
            raise ValueError('Unknown storage kind: %s' % (storage_kind,))
762
699
        z_header_len = int(z_header_len)
763
700
        if len(rest) < z_header_len:
775
712
        del rest
776
713
        # So now we have a valid GCB, we just need to parse the factories that
777
714
        # were sent to us
778
 
        header_lines = header.split(b'\n')
 
715
        header_lines = header.split('\n')
779
716
        del header
780
717
        last = header_lines.pop()
781
 
        if last != b'':
 
718
        if last != '':
782
719
            raise ValueError('header lines did not end with a trailing'
783
720
                             ' newline')
784
721
        if len(header_lines) % 4 != 0:
786
723
        block = GroupCompressBlock.from_bytes(block_bytes)
787
724
        del block_bytes
788
725
        result = cls(block)
789
 
        for start in range(0, len(header_lines), 4):
 
726
        for start in xrange(0, len(header_lines), 4):
790
727
            # intern()?
791
 
            key = tuple(header_lines[start].split(b'\x00'))
 
728
            key = tuple(header_lines[start].split('\x00'))
792
729
            parents_line = header_lines[start+1]
793
 
            if parents_line == b'None:':
 
730
            if parents_line == 'None:':
794
731
                parents = None
795
732
            else:
796
 
                parents = tuple([tuple(segment.split(b'\x00'))
797
 
                                 for segment in parents_line.split(b'\t')
 
733
                parents = tuple([tuple(segment.split('\x00'))
 
734
                                 for segment in parents_line.split('\t')
798
735
                                  if segment])
799
736
            start_offset = int(header_lines[start+2])
800
737
            end_offset = int(header_lines[start+3])
811
748
 
812
749
class _CommonGroupCompressor(object):
813
750
 
814
 
    def __init__(self, settings=None):
 
751
    def __init__(self):
815
752
        """Create a GroupCompressor."""
816
753
        self.chunks = []
817
754
        self._last = None
820
757
        self.labels_deltas = {}
821
758
        self._delta_index = None # Set by the children
822
759
        self._block = GroupCompressBlock()
823
 
        if settings is None:
824
 
            self._settings = {}
825
 
        else:
826
 
            self._settings = settings
827
760
 
828
761
    def compress(self, key, bytes, expected_sha, nostore_sha=None, soft=False):
829
762
        """Compress lines with label key.
830
763
 
831
764
        :param key: A key tuple. It is stored in the output
832
765
            for identification of the text during decompression. If the last
833
 
            element is b'None' it is replaced with the sha1 of the text -
 
766
            element is 'None' it is replaced with the sha1 of the text -
834
767
            e.g. sha1:xxxxxxx.
835
768
        :param bytes: The bytes to be compressed
836
769
        :param expected_sha: If non-None, the sha the lines are believed to
859
792
            if sha1 == nostore_sha:
860
793
                raise errors.ExistingContent()
861
794
        if key[-1] is None:
862
 
            key = key[:-1] + (b'sha1:' + sha1,)
 
795
            key = key[:-1] + ('sha1:' + sha1,)
863
796
 
864
797
        start, end, type = self._compress(key, bytes, len(bytes) / 2, soft)
865
798
        return sha1, start, end, type
891
824
        """
892
825
        (start_byte, start_chunk, end_byte, end_chunk) = self.labels_deltas[key]
893
826
        delta_chunks = self.chunks[start_chunk:end_chunk]
894
 
        stored_bytes = b''.join(delta_chunks)
895
 
        kind = stored_bytes[:1]
896
 
        if kind == b'f':
 
827
        stored_bytes = ''.join(delta_chunks)
 
828
        if stored_bytes[0] == 'f':
897
829
            fulltext_len, offset = decode_base128_int(stored_bytes[1:10])
898
830
            data_len = fulltext_len + 1 + offset
899
831
            if  data_len != len(stored_bytes):
900
832
                raise ValueError('Index claimed fulltext len, but stored bytes'
901
833
                                 ' claim %s != %s'
902
834
                                 % (len(stored_bytes), data_len))
903
 
            data = stored_bytes[offset + 1:]
 
835
            bytes = stored_bytes[offset + 1:]
904
836
        else:
905
 
            if kind != b'd':
906
 
                raise ValueError('Unknown content kind, bytes claim %s' % kind)
907
837
            # XXX: This is inefficient at best
908
 
            source = b''.join(self.chunks[:start_chunk])
 
838
            source = ''.join(self.chunks[:start_chunk])
 
839
            if stored_bytes[0] != 'd':
 
840
                raise ValueError('Unknown content kind, bytes claim %s'
 
841
                                 % (stored_bytes[0],))
909
842
            delta_len, offset = decode_base128_int(stored_bytes[1:10])
910
843
            data_len = delta_len + 1 + offset
911
844
            if data_len != len(stored_bytes):
912
845
                raise ValueError('Index claimed delta len, but stored bytes'
913
846
                                 ' claim %s != %s'
914
847
                                 % (len(stored_bytes), data_len))
915
 
            data = apply_delta(source, stored_bytes[offset + 1:])
916
 
        data_sha1 = osutils.sha_string(data)
917
 
        return data, data_sha1
 
848
            bytes = apply_delta(source, stored_bytes[offset + 1:])
 
849
        bytes_sha1 = osutils.sha_string(bytes)
 
850
        return bytes, bytes_sha1
918
851
 
919
852
    def flush(self):
920
853
        """Finish this group, creating a formatted stream.
921
854
 
922
855
        After calling this, the compressor should no longer be used
923
856
        """
 
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.
924
865
        self._block.set_chunked_content(self.chunks, self.endpoint)
925
866
        self.chunks = None
926
867
        self._delta_index = None
944
885
 
945
886
class PythonGroupCompressor(_CommonGroupCompressor):
946
887
 
947
 
    def __init__(self, settings=None):
 
888
    def __init__(self):
948
889
        """Create a GroupCompressor.
949
890
 
950
891
        Used only if the pyrex version is not available.
951
892
        """
952
 
        super(PythonGroupCompressor, self).__init__(settings)
 
893
        super(PythonGroupCompressor, self).__init__()
953
894
        self._delta_index = LinesDeltaIndex([])
954
895
        # The actual content is managed by LinesDeltaIndex
955
896
        self.chunks = self._delta_index.lines
964
905
        if delta_length > max_delta_size:
965
906
            # The delta is longer than the fulltext, insert a fulltext
966
907
            type = 'fulltext'
967
 
            out_lines = [b'f', encode_base128_int(input_len)]
 
908
            out_lines = ['f', encode_base128_int(input_len)]
968
909
            out_lines.extend(new_lines)
969
910
            index_lines = [False, False]
970
911
            index_lines.extend([True] * len(new_lines))
971
912
        else:
972
913
            # this is a worthy delta, output it
973
914
            type = 'delta'
974
 
            out_lines[0] = b'd'
 
915
            out_lines[0] = 'd'
975
916
            # Update the delta_length to include those two encoded integers
976
917
            out_lines[1] = encode_base128_int(delta_length)
977
918
        # Before insertion
992
933
 
993
934
    It contains code very similar to SequenceMatcher because of having a similar
994
935
    task. However some key differences apply:
995
 
 
996
 
    * there is no junk, we want a minimal edit not a human readable diff.
997
 
    * we don't filter very common lines (because we don't know where a good
998
 
      range will start, and after the first text we want to be emitting minmal
999
 
      edits only.
1000
 
    * we chain the left side, not the right side
1001
 
    * we incrementally update the adjacency matrix as new lines are provided.
1002
 
    * we look for matches in all of the left side, so the routine which does
1003
 
      the analagous task of find_longest_match does not need to filter on the
1004
 
      left side.
 
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.
1005
945
    """
1006
946
 
1007
 
    def __init__(self, settings=None):
1008
 
        super(PyrexGroupCompressor, self).__init__(settings)
1009
 
        max_bytes_to_index = self._settings.get('max_bytes_to_index', 0)
1010
 
        self._delta_index = DeltaIndex(max_bytes_to_index=max_bytes_to_index)
 
947
    def __init__(self):
 
948
        super(PyrexGroupCompressor, self).__init__()
 
949
        self._delta_index = DeltaIndex()
1011
950
 
1012
951
    def _compress(self, key, bytes, max_delta_size, soft=False):
1013
952
        """see _CommonGroupCompressor._compress"""
1031
970
            enc_length = encode_base128_int(len(bytes))
1032
971
            len_mini_header = 1 + len(enc_length)
1033
972
            self._delta_index.add_source(bytes, len_mini_header)
1034
 
            new_chunks = [b'f', enc_length, bytes]
 
973
            new_chunks = ['f', enc_length, bytes]
1035
974
        else:
1036
975
            type = 'delta'
1037
976
            enc_length = encode_base128_int(len(delta))
1038
977
            len_mini_header = 1 + len(enc_length)
1039
 
            new_chunks = [b'd', enc_length, delta]
 
978
            new_chunks = ['d', enc_length, delta]
1040
979
            self._delta_index.add_delta_source(delta, len_mini_header)
1041
980
        # Before insertion
1042
981
        start = self.endpoint
1088
1027
        index = _GCGraphIndex(graph_index, lambda:True, parents=parents,
1089
1028
            add_callback=graph_index.add_nodes,
1090
1029
            inconsistency_fatal=inconsistency_fatal)
1091
 
        access = pack_repo._DirectPackAccess({})
 
1030
        access = knit._DirectPackAccess({})
1092
1031
        access.set_writer(writer, graph_index, (transport, 'newpack'))
1093
1032
        result = GroupCompressVersionedFiles(index, access, delta)
1094
1033
        result.stream = stream
1104
1043
 
1105
1044
class _BatchingBlockFetcher(object):
1106
1045
    """Fetch group compress blocks in batches.
1107
 
 
 
1046
    
1108
1047
    :ivar total_bytes: int of expected number of bytes needed to fetch the
1109
1048
        currently pending batch.
1110
1049
    """
1111
1050
 
1112
 
    def __init__(self, gcvf, locations, get_compressor_settings=None):
 
1051
    def __init__(self, gcvf, locations):
1113
1052
        self.gcvf = gcvf
1114
1053
        self.locations = locations
1115
1054
        self.keys = []
1118
1057
        self.total_bytes = 0
1119
1058
        self.last_read_memo = None
1120
1059
        self.manager = None
1121
 
        self._get_compressor_settings = get_compressor_settings
1122
1060
 
1123
1061
    def add_key(self, key):
1124
1062
        """Add another to key to fetch.
1125
 
 
 
1063
        
1126
1064
        :return: The estimated number of bytes needed to fetch the batch so
1127
1065
            far.
1128
1066
        """
1153
1091
            # and then.
1154
1092
            self.batch_memos[read_memo] = cached_block
1155
1093
        return self.total_bytes
1156
 
 
 
1094
        
1157
1095
    def _flush_manager(self):
1158
1096
        if self.manager is not None:
1159
1097
            for factory in self.manager.get_record_stream():
1164
1102
    def yield_factories(self, full_flush=False):
1165
1103
        """Yield factories for keys added since the last yield.  They will be
1166
1104
        returned in the order they were added via add_key.
1167
 
 
 
1105
        
1168
1106
        :param full_flush: by default, some results may not be returned in case
1169
1107
            they can be part of the next batch.  If full_flush is True, then
1170
1108
            all results are returned.
1189
1127
                if memos_to_get_stack and memos_to_get_stack[-1] == read_memo:
1190
1128
                    # The next block from _get_blocks will be the block we
1191
1129
                    # need.
1192
 
                    block_read_memo, block = next(blocks)
 
1130
                    block_read_memo, block = blocks.next()
1193
1131
                    if block_read_memo != read_memo:
1194
1132
                        raise AssertionError(
1195
1133
                            "block_read_memo out of sync with read_memo"
1198
1136
                    memos_to_get_stack.pop()
1199
1137
                else:
1200
1138
                    block = self.batch_memos[read_memo]
1201
 
                self.manager = _LazyGroupContentManager(block,
1202
 
                    get_compressor_settings=self._get_compressor_settings)
 
1139
                self.manager = _LazyGroupContentManager(block)
1203
1140
                self.last_read_memo = read_memo
1204
1141
            start, end = index_memo[3:5]
1205
1142
            self.manager.add_factory(key, parents, start, end)
1212
1149
        self.total_bytes = 0
1213
1150
 
1214
1151
 
1215
 
class GroupCompressVersionedFiles(VersionedFilesWithFallbacks):
 
1152
class GroupCompressVersionedFiles(VersionedFiles):
1216
1153
    """A group-compress based VersionedFiles implementation."""
1217
1154
 
1218
 
    # This controls how the GroupCompress DeltaIndex works. Basically, we
1219
 
    # compute hash pointers into the source blocks (so hash(text) => text).
1220
 
    # However each of these references costs some memory in trade against a
1221
 
    # more accurate match result. For very large files, they either are
1222
 
    # pre-compressed and change in bulk whenever they change, or change in just
1223
 
    # local blocks. Either way, 'improved resolution' is not very helpful,
1224
 
    # versus running out of memory trying to track everything. The default max
1225
 
    # gives 100% sampling of a 1MB file.
1226
 
    _DEFAULT_MAX_BYTES_TO_INDEX = 1024 * 1024
1227
 
    _DEFAULT_COMPRESSOR_SETTINGS = {'max_bytes_to_index':
1228
 
                                     _DEFAULT_MAX_BYTES_TO_INDEX}
1229
 
 
1230
 
    def __init__(self, index, access, delta=True, _unadded_refs=None,
1231
 
                 _group_cache=None):
 
1155
    def __init__(self, index, access, delta=True, _unadded_refs=None):
1232
1156
        """Create a GroupCompressVersionedFiles object.
1233
1157
 
1234
1158
        :param index: The index object storing access and graph data.
1235
1159
        :param access: The access object storing raw data.
1236
1160
        :param delta: Whether to delta compress or just entropy compress.
1237
1161
        :param _unadded_refs: private parameter, don't use.
1238
 
        :param _group_cache: private parameter, don't use.
1239
1162
        """
1240
1163
        self._index = index
1241
1164
        self._access = access
1243
1166
        if _unadded_refs is None:
1244
1167
            _unadded_refs = {}
1245
1168
        self._unadded_refs = _unadded_refs
1246
 
        if _group_cache is None:
1247
 
            _group_cache = LRUSizeCache(max_size=50*1024*1024)
1248
 
        self._group_cache = _group_cache
1249
 
        self._immediate_fallback_vfs = []
1250
 
        self._max_bytes_to_index = None
 
1169
        self._group_cache = LRUSizeCache(max_size=50*1024*1024)
 
1170
        self._fallback_vfs = []
1251
1171
 
1252
1172
    def without_fallbacks(self):
1253
1173
        """Return a clone of this object without any fallbacks configured."""
1254
1174
        return GroupCompressVersionedFiles(self._index, self._access,
1255
 
            self._delta, _unadded_refs=dict(self._unadded_refs),
1256
 
            _group_cache=self._group_cache)
 
1175
            self._delta, _unadded_refs=dict(self._unadded_refs))
1257
1176
 
1258
1177
    def add_lines(self, key, parents, lines, parent_texts=None,
1259
1178
        left_matching_blocks=None, nostore_sha=None, random_id=False,
1263
1182
        :param key: The key tuple of the text to add.
1264
1183
        :param parents: The parents key tuples of the text to add.
1265
1184
        :param lines: A list of lines. Each line must be a bytestring. And all
1266
 
            of them except the last must be terminated with \\n and contain no
1267
 
            other \\n's. The last line may either contain no \\n's or a single
1268
 
            terminating \\n. If the lines list does meet this constraint the
1269
 
            add routine may error or may succeed - but you will be unable to
1270
 
            read the data back accurately. (Checking the lines have been split
 
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
1271
1190
            correctly is expensive and extremely unlikely to catch bugs so it
1272
1191
            is not done at runtime unless check_content is True.)
1273
1192
        :param parent_texts: An optional dictionary containing the opaque
1305
1224
                                               nostore_sha=nostore_sha))[0]
1306
1225
        return sha1, length, None
1307
1226
 
 
1227
    def _add_text(self, key, parents, text, nostore_sha=None, random_id=False):
 
1228
        """See VersionedFiles._add_text()."""
 
1229
        self._index._check_write_ok()
 
1230
        self._check_add(key, None, random_id, check_content=False)
 
1231
        if text.__class__ is not str:
 
1232
            raise errors.BzrBadParameterUnicode("text")
 
1233
        if parents is None:
 
1234
            # The caller might pass None if there is no graph data, but kndx
 
1235
            # indexes can't directly store that, so we give them
 
1236
            # an empty tuple instead.
 
1237
            parents = ()
 
1238
        # double handling for now. Make it work until then.
 
1239
        length = len(text)
 
1240
        record = FulltextContentFactory(key, parents, None, text)
 
1241
        sha1 = list(self._insert_record_stream([record], random_id=random_id,
 
1242
                                               nostore_sha=nostore_sha))[0]
 
1243
        return sha1, length, None
 
1244
 
1308
1245
    def add_fallback_versioned_files(self, a_versioned_files):
1309
1246
        """Add a source of texts for texts not present in this knit.
1310
1247
 
1311
1248
        :param a_versioned_files: A VersionedFiles object.
1312
1249
        """
1313
 
        self._immediate_fallback_vfs.append(a_versioned_files)
 
1250
        self._fallback_vfs.append(a_versioned_files)
1314
1251
 
1315
1252
    def annotate(self, key):
1316
1253
        """See VersionedFiles.annotate."""
1350
1287
            self._check_lines_not_unicode(lines)
1351
1288
            self._check_lines_are_lines(lines)
1352
1289
 
 
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
 
1353
1306
    def get_parent_map(self, keys):
1354
1307
        """Get a map of the graph parents of keys.
1355
1308
 
1370
1323
            and so on.
1371
1324
        """
1372
1325
        result = {}
1373
 
        sources = [self._index] + self._immediate_fallback_vfs
 
1326
        sources = [self._index] + self._fallback_vfs
1374
1327
        source_results = []
1375
1328
        missing = set(keys)
1376
1329
        for source in sources:
1413
1366
                yield read_memo, cached[read_memo]
1414
1367
            except KeyError:
1415
1368
                # Read the block, and cache it.
1416
 
                zdata = next(raw_records)
 
1369
                zdata = raw_records.next()
1417
1370
                block = GroupCompressBlock.from_bytes(zdata)
1418
1371
                self._group_cache[read_memo] = block
1419
1372
                cached[read_memo] = block
1461
1414
                    remaining_keys.discard(content_factory.key)
1462
1415
                    yield content_factory
1463
1416
                return
1464
 
            except errors.RetryWithNewPacks as e:
 
1417
            except errors.RetryWithNewPacks, e:
1465
1418
                self._access.reload_or_raise(e)
1466
1419
 
1467
1420
    def _find_from_fallback(self, missing):
1477
1430
        parent_map = {}
1478
1431
        key_to_source_map = {}
1479
1432
        source_results = []
1480
 
        for source in self._immediate_fallback_vfs:
 
1433
        for source in self._fallback_vfs:
1481
1434
            if not missing:
1482
1435
                break
1483
1436
            source_parents = source.get_parent_map(missing)
1493
1446
 
1494
1447
        The returned objects should be in the order defined by 'ordering',
1495
1448
        which can weave between different sources.
1496
 
 
1497
1449
        :param ordering: Must be one of 'topological' or 'groupcompress'
1498
1450
        :return: List of [(source, [keys])] tuples, such that all keys are in
1499
1451
            the defined order, regardless of source.
1500
1452
        """
1501
1453
        if ordering == 'topological':
1502
 
            present_keys = tsort.topo_sort(parent_map)
 
1454
            present_keys = topo_sort(parent_map)
1503
1455
        else:
1504
1456
            # ordering == 'groupcompress'
1505
1457
            # XXX: This only optimizes for the target ordering. We may need
1541
1493
            # This is the group the bytes are stored in, followed by the
1542
1494
            # location in the group
1543
1495
            return locations[key][0]
 
1496
        present_keys = sorted(locations.iterkeys(), key=get_group)
1544
1497
        # We don't have an ordering for keys in the in-memory object, but
1545
1498
        # lets process the in-memory ones first.
1546
 
        present_keys = list(unadded_keys)
1547
 
        present_keys.extend(sorted(locations, key=get_group))
 
1499
        present_keys = list(unadded_keys) + present_keys
1548
1500
        # Now grab all of the ones from other sources
1549
1501
        source_keys = [(self, present_keys)]
1550
1502
        source_keys.extend(source_result)
1574
1526
            # start with one key, recurse to its oldest parent, then grab
1575
1527
            # everything in the same group, etc.
1576
1528
            parent_map = dict((key, details[2]) for key, details in
1577
 
                viewitems(locations))
 
1529
                locations.iteritems())
1578
1530
            for key in unadded_keys:
1579
1531
                parent_map[key] = self._unadded_refs[key]
1580
1532
            parent_map.update(fallback_parent_map)
1594
1546
        #  - we encounter an unadded ref, or
1595
1547
        #  - we run out of keys, or
1596
1548
        #  - the total bytes to retrieve for this batch > BATCH_SIZE
1597
 
        batcher = _BatchingBlockFetcher(self, locations,
1598
 
            get_compressor_settings=self._get_compressor_settings)
 
1549
        batcher = _BatchingBlockFetcher(self, locations)
1599
1550
        for source, keys in source_keys:
1600
1551
            if source is self:
1601
1552
                for key in keys:
1647
1598
        for _ in self._insert_record_stream(stream, random_id=False):
1648
1599
            pass
1649
1600
 
1650
 
    def _get_compressor_settings(self):
1651
 
        if self._max_bytes_to_index is None:
1652
 
            # TODO: VersionedFiles don't know about their containing
1653
 
            #       repository, so they don't have much of an idea about their
1654
 
            #       location. So for now, this is only a global option.
1655
 
            c = config.GlobalConfig()
1656
 
            val = c.get_user_option('bzr.groupcompress.max_bytes_to_index')
1657
 
            if val is not None:
1658
 
                try:
1659
 
                    val = int(val)
1660
 
                except ValueError as e:
1661
 
                    trace.warning('Value for '
1662
 
                                  '"bzr.groupcompress.max_bytes_to_index"'
1663
 
                                  ' %r is not an integer'
1664
 
                                  % (val,))
1665
 
                    val = None
1666
 
            if val is None:
1667
 
                val = self._DEFAULT_MAX_BYTES_TO_INDEX
1668
 
            self._max_bytes_to_index = val
1669
 
        return {'max_bytes_to_index': self._max_bytes_to_index}
1670
 
 
1671
 
    def _make_group_compressor(self):
1672
 
        return GroupCompressor(self._get_compressor_settings())
1673
 
 
1674
1601
    def _insert_record_stream(self, stream, random_id=False, nostore_sha=None,
1675
1602
                              reuse_blocks=True):
1676
1603
        """Internal core to insert a record stream into this container.
1699
1626
                return adapter
1700
1627
        # This will go up to fulltexts for gc to gc fetching, which isn't
1701
1628
        # ideal.
1702
 
        self._compressor = self._make_group_compressor()
 
1629
        self._compressor = GroupCompressor()
1703
1630
        self._unadded_refs = {}
1704
1631
        keys_to_add = []
1705
1632
        def flush():
1706
 
            bytes_len, chunks = self._compressor.flush().to_chunks()
1707
 
            self._compressor = self._make_group_compressor()
1708
 
            # Note: At this point we still have 1 copy of the fulltext (in
1709
 
            #       record and the var 'bytes'), and this generates 2 copies of
1710
 
            #       the compressed text (one for bytes, one in chunks)
1711
 
            # TODO: Push 'chunks' down into the _access api, so that we don't
1712
 
            #       have to double compressed memory here
1713
 
            # TODO: Figure out how to indicate that we would be happy to free
1714
 
            #       the fulltext content at this point. Note that sometimes we
1715
 
            #       will want it later (streaming CHK pages), but most of the
1716
 
            #       time we won't (everything else)
1717
 
            data = b''.join(chunks)
1718
 
            del chunks
 
1633
            bytes = self._compressor.flush().to_bytes()
 
1634
            self._compressor = GroupCompressor()
1719
1635
            index, start, length = self._access.add_raw_records(
1720
 
                [(None, len(data))], data)[0]
 
1636
                [(None, len(bytes))], bytes)[0]
1721
1637
            nodes = []
1722
1638
            for key, reads, refs in keys_to_add:
1723
 
                nodes.append((key, b"%d %d %s" % (start, length, reads), refs))
 
1639
                nodes.append((key, "%d %d %s" % (start, length, reads), refs))
1724
1640
            self._index.add_records(nodes, random_id=random_id)
1725
1641
            self._unadded_refs = {}
1726
1642
            del keys_to_add[:]
1740
1656
                raise errors.RevisionNotPresent(record.key, self)
1741
1657
            if random_id:
1742
1658
                if record.key in inserted_keys:
1743
 
                    trace.note(gettext('Insert claimed random_id=True,'
1744
 
                               ' but then inserted %r two times'), record.key)
 
1659
                    trace.note('Insert claimed random_id=True,'
 
1660
                               ' but then inserted %r two times', record.key)
1745
1661
                    continue
1746
1662
                inserted_keys.add(record.key)
1747
1663
            if reuse_blocks:
1776
1692
                            ' the current record, we cannot be positive'
1777
1693
                            ' that the appropriate content was inserted.'
1778
1694
                            )
1779
 
                    value = b"%d %d %d %d" % (block_start, block_length,
 
1695
                    value = "%d %d %d %d" % (block_start, block_length,
1780
1696
                                             record._start, record._end)
1781
1697
                    nodes = [(record.key, value, (record.parents,))]
1782
1698
                    # TODO: Consider buffering up many nodes to be added, not
1826
1742
                 type) = self._compressor.compress(record.key, bytes,
1827
1743
                                                   record.sha1)
1828
1744
            if record.key[-1] is None:
1829
 
                key = record.key[:-1] + (b'sha1:' + found_sha1,)
 
1745
                key = record.key[:-1] + ('sha1:' + found_sha1,)
1830
1746
            else:
1831
1747
                key = record.key
1832
1748
            self._unadded_refs[key] = record.parents
1837
1753
            else:
1838
1754
                parents = None
1839
1755
            refs = static_tuple.StaticTuple(parents)
1840
 
            keys_to_add.append(
1841
 
                (key, b'%d %d' % (start_point, end_point), refs))
 
1756
            keys_to_add.append((key, '%d %d' % (start_point, end_point), refs))
1842
1757
        if len(keys_to_add):
1843
1758
            flush()
1844
1759
        self._compressor = None
1887
1802
        """See VersionedFiles.keys."""
1888
1803
        if 'evil' in debug.debug_flags:
1889
1804
            trace.mutter_callsite(2, "keys scales with size of history")
1890
 
        sources = [self._index] + self._immediate_fallback_vfs
 
1805
        sources = [self._index] + self._fallback_vfs
1891
1806
        result = set()
1892
1807
        for source in sources:
1893
1808
            result.update(source.keys())
1894
1809
        return result
1895
1810
 
1896
1811
 
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
 
 
1945
1812
class _GCGraphIndex(object):
1946
1813
    """Mapper from GroupCompressVersionedFiles needs into GraphIndex storage."""
1947
1814
 
1950
1817
        inconsistency_fatal=True, track_new_keys=False):
1951
1818
        """Construct a _GCGraphIndex on a graph_index.
1952
1819
 
1953
 
        :param graph_index: An implementation of breezy.index.GraphIndex.
 
1820
        :param graph_index: An implementation of bzrlib.index.GraphIndex.
1954
1821
        :param is_locked: A callback, returns True if the index is locked and
1955
1822
            thus usable.
1956
1823
        :param parents: If True, record knits parents, if not do not record
1976
1843
        # repeated over and over, this creates a surplus of ints
1977
1844
        self._int_cache = {}
1978
1845
        if track_external_parent_refs:
1979
 
            self._key_dependencies = _KeyRefs(
 
1846
            self._key_dependencies = knit._KeyRefs(
1980
1847
                track_new_keys=track_new_keys)
1981
1848
        else:
1982
1849
            self._key_dependencies = None
2006
1873
                if refs:
2007
1874
                    for ref in refs:
2008
1875
                        if ref:
2009
 
                            raise knit.KnitCorrupt(self,
 
1876
                            raise errors.KnitCorrupt(self,
2010
1877
                                "attempt to add node with parents "
2011
1878
                                "in parentless index.")
2012
1879
                    refs = ()
2022
1889
                if node_refs != passed[1]:
2023
1890
                    details = '%s %s %s' % (key, (value, node_refs), passed)
2024
1891
                    if self._inconsistency_fatal:
2025
 
                        raise knit.KnitCorrupt(self, "inconsistent details"
 
1892
                        raise errors.KnitCorrupt(self, "inconsistent details"
2026
1893
                                                 " in add_records: %s" %
2027
1894
                                                 details)
2028
1895
                    else:
2033
1900
        if changed:
2034
1901
            result = []
2035
1902
            if self._parents:
2036
 
                for key, (value, node_refs) in viewitems(keys):
 
1903
                for key, (value, node_refs) in keys.iteritems():
2037
1904
                    result.append((key, value, node_refs))
2038
1905
            else:
2039
 
                for key, (value, node_refs) in viewitems(keys):
 
1906
                for key, (value, node_refs) in keys.iteritems():
2040
1907
                    result.append((key, value))
2041
1908
            records = result
2042
1909
        key_dependencies = self._key_dependencies
2122
1989
        :param keys: An iterable of keys.
2123
1990
        :return: A dict of key:
2124
1991
            (index_memo, compression_parent, parents, record_details).
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
 
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
2133
2002
        """
2134
2003
        self._check_read()
2135
2004
        result = {}
2140
2009
                parents = None
2141
2010
            else:
2142
2011
                parents = entry[3][0]
2143
 
            details = _GCBuildDetails(parents, self._node_to_position(entry))
2144
 
            result[key] = details
 
2012
            method = 'group'
 
2013
            result[key] = (self._node_to_position(entry),
 
2014
                                  None, parents, (method, None))
2145
2015
        return result
2146
2016
 
2147
2017
    def keys(self):
2154
2024
 
2155
2025
    def _node_to_position(self, node):
2156
2026
        """Convert an index value to position details."""
2157
 
        bits = node[2].split(b' ')
 
2027
        bits = node[2].split(' ')
2158
2028
        # It would be nice not to read the entire gzip.
2159
2029
        # start and stop are put into _int_cache because they are very common.
2160
2030
        # They define the 'group' that an entry is in, and many groups can have
2163
2033
        # each, or about 7MB. Note that it might be even more when you consider
2164
2034
        # how PyInt is allocated in separate slabs. And you can't return a slab
2165
2035
        # to the OS if even 1 int on it is in use. Note though that Python uses
2166
 
        # a LIFO when re-using PyInt slots, which might cause more
 
2036
        # a LIFO when re-using PyInt slots, which probably causes more
2167
2037
        # fragmentation.
2168
2038
        start = int(bits[0])
2169
2039
        start = self._int_cache.setdefault(start, start)
2193
2063
            key_dependencies.add_references(node[1], node[3][0])
2194
2064
 
2195
2065
 
2196
 
from ._groupcompress_py import (
 
2066
from bzrlib._groupcompress_py import (
2197
2067
    apply_delta,
2198
2068
    apply_delta_to_source,
2199
2069
    encode_base128_int,
2202
2072
    LinesDeltaIndex,
2203
2073
    )
2204
2074
try:
2205
 
    from ._groupcompress_pyx import (
 
2075
    from bzrlib._groupcompress_pyx import (
2206
2076
        apply_delta,
2207
2077
        apply_delta_to_source,
2208
2078
        DeltaIndex,
2210
2080
        decode_base128_int,
2211
2081
        )
2212
2082
    GroupCompressor = PyrexGroupCompressor
2213
 
except ImportError as e:
 
2083
except ImportError, e:
2214
2084
    osutils.failed_to_load_extension(e)
2215
2085
    GroupCompressor = PythonGroupCompressor
2216
2086