~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/knit.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2011-05-04 22:17:22 UTC
  • mfrom: (5815.3.15 use-tree-annotate)
  • Revision ID: pqm@pqm.ubuntu.com-20110504221722-fz5hr1xagchptyje
(jelmer) Avoid directly accessing VersionedFiles.annotate();
 rather, access it through RevisionTree.annotate_iter(). (Jelmer Vernooij)

Show diffs side-by-side

added added

removed removed

Lines of Context:
 
1
# Copyright (C) 2006-2011 Canonical Ltd
 
2
#
 
3
# This program is free software; you can redistribute it and/or modify
 
4
# it under the terms of the GNU General Public License as published by
 
5
# the Free Software Foundation; either version 2 of the License, or
 
6
# (at your option) any later version.
 
7
#
 
8
# This program is distributed in the hope that it will be useful,
 
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
 
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
11
# GNU General Public License for more details.
 
12
#
 
13
# You should have received a copy of the GNU General Public License
 
14
# along with this program; if not, write to the Free Software
 
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 
16
 
 
17
"""Knit versionedfile implementation.
 
18
 
 
19
A knit is a versioned file implementation that supports efficient append only
 
20
updates.
 
21
 
 
22
Knit file layout:
 
23
lifeless: the data file is made up of "delta records".  each delta record has a delta header
 
24
that contains; (1) a version id, (2) the size of the delta (in lines), and (3)  the digest of
 
25
the -expanded data- (ie, the delta applied to the parent).  the delta also ends with a
 
26
end-marker; simply "end VERSION"
 
27
 
 
28
delta can be line or full contents.a
 
29
... the 8's there are the index number of the annotation.
 
30
version robertc@robertcollins.net-20051003014215-ee2990904cc4c7ad 7 c7d23b2a5bd6ca00e8e266cec0ec228158ee9f9e
 
31
59,59,3
 
32
8
 
33
8         if ie.executable:
 
34
8             e.set('executable', 'yes')
 
35
130,130,2
 
36
8         if elt.get('executable') == 'yes':
 
37
8             ie.executable = True
 
38
end robertc@robertcollins.net-20051003014215-ee2990904cc4c7ad
 
39
 
 
40
 
 
41
whats in an index:
 
42
09:33 < jrydberg> lifeless: each index is made up of a tuple of; version id, options, position, size, parents
 
43
09:33 < jrydberg> lifeless: the parents are currently dictionary compressed
 
44
09:33 < jrydberg> lifeless: (meaning it currently does not support ghosts)
 
45
09:33 < lifeless> right
 
46
09:33 < jrydberg> lifeless: the position and size is the range in the data file
 
47
 
 
48
 
 
49
so the index sequence is the dictionary compressed sequence number used
 
50
in the deltas to provide line annotation
 
51
 
 
52
"""
 
53
 
 
54
 
 
55
from cStringIO import StringIO
 
56
from itertools import izip
 
57
import operator
 
58
import os
 
59
 
 
60
from bzrlib.lazy_import import lazy_import
 
61
lazy_import(globals(), """
 
62
import gzip
 
63
 
 
64
from bzrlib import (
 
65
    debug,
 
66
    diff,
 
67
    graph as _mod_graph,
 
68
    index as _mod_index,
 
69
    pack,
 
70
    patiencediff,
 
71
    static_tuple,
 
72
    trace,
 
73
    tsort,
 
74
    tuned_gzip,
 
75
    ui,
 
76
    )
 
77
 
 
78
from bzrlib.repofmt import pack_repo
 
79
""")
 
80
from bzrlib import (
 
81
    annotate,
 
82
    errors,
 
83
    osutils,
 
84
    )
 
85
from bzrlib.errors import (
 
86
    NoSuchFile,
 
87
    InvalidRevisionId,
 
88
    KnitCorrupt,
 
89
    KnitHeaderError,
 
90
    RevisionNotPresent,
 
91
    SHA1KnitCorrupt,
 
92
    )
 
93
from bzrlib.osutils import (
 
94
    contains_whitespace,
 
95
    sha_string,
 
96
    sha_strings,
 
97
    split_lines,
 
98
    )
 
99
from bzrlib.versionedfile import (
 
100
    _KeyRefs,
 
101
    AbsentContentFactory,
 
102
    adapter_registry,
 
103
    ConstantMapper,
 
104
    ContentFactory,
 
105
    sort_groupcompress,
 
106
    VersionedFiles,
 
107
    )
 
108
 
 
109
 
 
110
# TODO: Split out code specific to this format into an associated object.
 
111
 
 
112
# TODO: Can we put in some kind of value to check that the index and data
 
113
# files belong together?
 
114
 
 
115
# TODO: accommodate binaries, perhaps by storing a byte count
 
116
 
 
117
# TODO: function to check whole file
 
118
 
 
119
# TODO: atomically append data, then measure backwards from the cursor
 
120
# position after writing to work out where it was located.  we may need to
 
121
# bypass python file buffering.
 
122
 
 
123
DATA_SUFFIX = '.knit'
 
124
INDEX_SUFFIX = '.kndx'
 
125
_STREAM_MIN_BUFFER_SIZE = 5*1024*1024
 
126
 
 
127
 
 
128
class KnitAdapter(object):
 
129
    """Base class for knit record adaption."""
 
130
 
 
131
    def __init__(self, basis_vf):
 
132
        """Create an adapter which accesses full texts from basis_vf.
 
133
 
 
134
        :param basis_vf: A versioned file to access basis texts of deltas from.
 
135
            May be None for adapters that do not need to access basis texts.
 
136
        """
 
137
        self._data = KnitVersionedFiles(None, None)
 
138
        self._annotate_factory = KnitAnnotateFactory()
 
139
        self._plain_factory = KnitPlainFactory()
 
140
        self._basis_vf = basis_vf
 
141
 
 
142
 
 
143
class FTAnnotatedToUnannotated(KnitAdapter):
 
144
    """An adapter from FT annotated knits to unannotated ones."""
 
145
 
 
146
    def get_bytes(self, factory):
 
147
        annotated_compressed_bytes = factory._raw_record
 
148
        rec, contents = \
 
149
            self._data._parse_record_unchecked(annotated_compressed_bytes)
 
150
        content = self._annotate_factory.parse_fulltext(contents, rec[1])
 
151
        size, bytes = self._data._record_to_data((rec[1],), rec[3], content.text())
 
152
        return bytes
 
153
 
 
154
 
 
155
class DeltaAnnotatedToUnannotated(KnitAdapter):
 
156
    """An adapter for deltas from annotated to unannotated."""
 
157
 
 
158
    def get_bytes(self, factory):
 
159
        annotated_compressed_bytes = factory._raw_record
 
160
        rec, contents = \
 
161
            self._data._parse_record_unchecked(annotated_compressed_bytes)
 
162
        delta = self._annotate_factory.parse_line_delta(contents, rec[1],
 
163
            plain=True)
 
164
        contents = self._plain_factory.lower_line_delta(delta)
 
165
        size, bytes = self._data._record_to_data((rec[1],), rec[3], contents)
 
166
        return bytes
 
167
 
 
168
 
 
169
class FTAnnotatedToFullText(KnitAdapter):
 
170
    """An adapter from FT annotated knits to unannotated ones."""
 
171
 
 
172
    def get_bytes(self, factory):
 
173
        annotated_compressed_bytes = factory._raw_record
 
174
        rec, contents = \
 
175
            self._data._parse_record_unchecked(annotated_compressed_bytes)
 
176
        content, delta = self._annotate_factory.parse_record(factory.key[-1],
 
177
            contents, factory._build_details, None)
 
178
        return ''.join(content.text())
 
179
 
 
180
 
 
181
class DeltaAnnotatedToFullText(KnitAdapter):
 
182
    """An adapter for deltas from annotated to unannotated."""
 
183
 
 
184
    def get_bytes(self, factory):
 
185
        annotated_compressed_bytes = factory._raw_record
 
186
        rec, contents = \
 
187
            self._data._parse_record_unchecked(annotated_compressed_bytes)
 
188
        delta = self._annotate_factory.parse_line_delta(contents, rec[1],
 
189
            plain=True)
 
190
        compression_parent = factory.parents[0]
 
191
        basis_entry = self._basis_vf.get_record_stream(
 
192
            [compression_parent], 'unordered', True).next()
 
193
        if basis_entry.storage_kind == 'absent':
 
194
            raise errors.RevisionNotPresent(compression_parent, self._basis_vf)
 
195
        basis_chunks = basis_entry.get_bytes_as('chunked')
 
196
        basis_lines = osutils.chunks_to_lines(basis_chunks)
 
197
        # Manually apply the delta because we have one annotated content and
 
198
        # one plain.
 
199
        basis_content = PlainKnitContent(basis_lines, compression_parent)
 
200
        basis_content.apply_delta(delta, rec[1])
 
201
        basis_content._should_strip_eol = factory._build_details[1]
 
202
        return ''.join(basis_content.text())
 
203
 
 
204
 
 
205
class FTPlainToFullText(KnitAdapter):
 
206
    """An adapter from FT plain knits to unannotated ones."""
 
207
 
 
208
    def get_bytes(self, factory):
 
209
        compressed_bytes = factory._raw_record
 
210
        rec, contents = \
 
211
            self._data._parse_record_unchecked(compressed_bytes)
 
212
        content, delta = self._plain_factory.parse_record(factory.key[-1],
 
213
            contents, factory._build_details, None)
 
214
        return ''.join(content.text())
 
215
 
 
216
 
 
217
class DeltaPlainToFullText(KnitAdapter):
 
218
    """An adapter for deltas from annotated to unannotated."""
 
219
 
 
220
    def get_bytes(self, factory):
 
221
        compressed_bytes = factory._raw_record
 
222
        rec, contents = \
 
223
            self._data._parse_record_unchecked(compressed_bytes)
 
224
        delta = self._plain_factory.parse_line_delta(contents, rec[1])
 
225
        compression_parent = factory.parents[0]
 
226
        # XXX: string splitting overhead.
 
227
        basis_entry = self._basis_vf.get_record_stream(
 
228
            [compression_parent], 'unordered', True).next()
 
229
        if basis_entry.storage_kind == 'absent':
 
230
            raise errors.RevisionNotPresent(compression_parent, self._basis_vf)
 
231
        basis_chunks = basis_entry.get_bytes_as('chunked')
 
232
        basis_lines = osutils.chunks_to_lines(basis_chunks)
 
233
        basis_content = PlainKnitContent(basis_lines, compression_parent)
 
234
        # Manually apply the delta because we have one annotated content and
 
235
        # one plain.
 
236
        content, _ = self._plain_factory.parse_record(rec[1], contents,
 
237
            factory._build_details, basis_content)
 
238
        return ''.join(content.text())
 
239
 
 
240
 
 
241
class KnitContentFactory(ContentFactory):
 
242
    """Content factory for streaming from knits.
 
243
 
 
244
    :seealso ContentFactory:
 
245
    """
 
246
 
 
247
    def __init__(self, key, parents, build_details, sha1, raw_record,
 
248
        annotated, knit=None, network_bytes=None):
 
249
        """Create a KnitContentFactory for key.
 
250
 
 
251
        :param key: The key.
 
252
        :param parents: The parents.
 
253
        :param build_details: The build details as returned from
 
254
            get_build_details.
 
255
        :param sha1: The sha1 expected from the full text of this object.
 
256
        :param raw_record: The bytes of the knit data from disk.
 
257
        :param annotated: True if the raw data is annotated.
 
258
        :param network_bytes: None to calculate the network bytes on demand,
 
259
            not-none if they are already known.
 
260
        """
 
261
        ContentFactory.__init__(self)
 
262
        self.sha1 = sha1
 
263
        self.key = key
 
264
        self.parents = parents
 
265
        if build_details[0] == 'line-delta':
 
266
            kind = 'delta'
 
267
        else:
 
268
            kind = 'ft'
 
269
        if annotated:
 
270
            annotated_kind = 'annotated-'
 
271
        else:
 
272
            annotated_kind = ''
 
273
        self.storage_kind = 'knit-%s%s-gz' % (annotated_kind, kind)
 
274
        self._raw_record = raw_record
 
275
        self._network_bytes = network_bytes
 
276
        self._build_details = build_details
 
277
        self._knit = knit
 
278
 
 
279
    def _create_network_bytes(self):
 
280
        """Create a fully serialised network version for transmission."""
 
281
        # storage_kind, key, parents, Noeol, raw_record
 
282
        key_bytes = '\x00'.join(self.key)
 
283
        if self.parents is None:
 
284
            parent_bytes = 'None:'
 
285
        else:
 
286
            parent_bytes = '\t'.join('\x00'.join(key) for key in self.parents)
 
287
        if self._build_details[1]:
 
288
            noeol = 'N'
 
289
        else:
 
290
            noeol = ' '
 
291
        network_bytes = "%s\n%s\n%s\n%s%s" % (self.storage_kind, key_bytes,
 
292
            parent_bytes, noeol, self._raw_record)
 
293
        self._network_bytes = network_bytes
 
294
 
 
295
    def get_bytes_as(self, storage_kind):
 
296
        if storage_kind == self.storage_kind:
 
297
            if self._network_bytes is None:
 
298
                self._create_network_bytes()
 
299
            return self._network_bytes
 
300
        if ('-ft-' in self.storage_kind and
 
301
            storage_kind in ('chunked', 'fulltext')):
 
302
            adapter_key = (self.storage_kind, 'fulltext')
 
303
            adapter_factory = adapter_registry.get(adapter_key)
 
304
            adapter = adapter_factory(None)
 
305
            bytes = adapter.get_bytes(self)
 
306
            if storage_kind == 'chunked':
 
307
                return [bytes]
 
308
            else:
 
309
                return bytes
 
310
        if self._knit is not None:
 
311
            # Not redundant with direct conversion above - that only handles
 
312
            # fulltext cases.
 
313
            if storage_kind == 'chunked':
 
314
                return self._knit.get_lines(self.key[0])
 
315
            elif storage_kind == 'fulltext':
 
316
                return self._knit.get_text(self.key[0])
 
317
        raise errors.UnavailableRepresentation(self.key, storage_kind,
 
318
            self.storage_kind)
 
319
 
 
320
 
 
321
class LazyKnitContentFactory(ContentFactory):
 
322
    """A ContentFactory which can either generate full text or a wire form.
 
323
 
 
324
    :seealso ContentFactory:
 
325
    """
 
326
 
 
327
    def __init__(self, key, parents, generator, first):
 
328
        """Create a LazyKnitContentFactory.
 
329
 
 
330
        :param key: The key of the record.
 
331
        :param parents: The parents of the record.
 
332
        :param generator: A _ContentMapGenerator containing the record for this
 
333
            key.
 
334
        :param first: Is this the first content object returned from generator?
 
335
            if it is, its storage kind is knit-delta-closure, otherwise it is
 
336
            knit-delta-closure-ref
 
337
        """
 
338
        self.key = key
 
339
        self.parents = parents
 
340
        self.sha1 = None
 
341
        self._generator = generator
 
342
        self.storage_kind = "knit-delta-closure"
 
343
        if not first:
 
344
            self.storage_kind = self.storage_kind + "-ref"
 
345
        self._first = first
 
346
 
 
347
    def get_bytes_as(self, storage_kind):
 
348
        if storage_kind == self.storage_kind:
 
349
            if self._first:
 
350
                return self._generator._wire_bytes()
 
351
            else:
 
352
                # all the keys etc are contained in the bytes returned in the
 
353
                # first record.
 
354
                return ''
 
355
        if storage_kind in ('chunked', 'fulltext'):
 
356
            chunks = self._generator._get_one_work(self.key).text()
 
357
            if storage_kind == 'chunked':
 
358
                return chunks
 
359
            else:
 
360
                return ''.join(chunks)
 
361
        raise errors.UnavailableRepresentation(self.key, storage_kind,
 
362
            self.storage_kind)
 
363
 
 
364
 
 
365
def knit_delta_closure_to_records(storage_kind, bytes, line_end):
 
366
    """Convert a network record to a iterator over stream records.
 
367
 
 
368
    :param storage_kind: The storage kind of the record.
 
369
        Must be 'knit-delta-closure'.
 
370
    :param bytes: The bytes of the record on the network.
 
371
    """
 
372
    generator = _NetworkContentMapGenerator(bytes, line_end)
 
373
    return generator.get_record_stream()
 
374
 
 
375
 
 
376
def knit_network_to_record(storage_kind, bytes, line_end):
 
377
    """Convert a network record to a record object.
 
378
 
 
379
    :param storage_kind: The storage kind of the record.
 
380
    :param bytes: The bytes of the record on the network.
 
381
    """
 
382
    start = line_end
 
383
    line_end = bytes.find('\n', start)
 
384
    key = tuple(bytes[start:line_end].split('\x00'))
 
385
    start = line_end + 1
 
386
    line_end = bytes.find('\n', start)
 
387
    parent_line = bytes[start:line_end]
 
388
    if parent_line == 'None:':
 
389
        parents = None
 
390
    else:
 
391
        parents = tuple(
 
392
            [tuple(segment.split('\x00')) for segment in parent_line.split('\t')
 
393
             if segment])
 
394
    start = line_end + 1
 
395
    noeol = bytes[start] == 'N'
 
396
    if 'ft' in storage_kind:
 
397
        method = 'fulltext'
 
398
    else:
 
399
        method = 'line-delta'
 
400
    build_details = (method, noeol)
 
401
    start = start + 1
 
402
    raw_record = bytes[start:]
 
403
    annotated = 'annotated' in storage_kind
 
404
    return [KnitContentFactory(key, parents, build_details, None, raw_record,
 
405
        annotated, network_bytes=bytes)]
 
406
 
 
407
 
 
408
class KnitContent(object):
 
409
    """Content of a knit version to which deltas can be applied.
 
410
 
 
411
    This is always stored in memory as a list of lines with \n at the end,
 
412
    plus a flag saying if the final ending is really there or not, because that
 
413
    corresponds to the on-disk knit representation.
 
414
    """
 
415
 
 
416
    def __init__(self):
 
417
        self._should_strip_eol = False
 
418
 
 
419
    def apply_delta(self, delta, new_version_id):
 
420
        """Apply delta to this object to become new_version_id."""
 
421
        raise NotImplementedError(self.apply_delta)
 
422
 
 
423
    def line_delta_iter(self, new_lines):
 
424
        """Generate line-based delta from this content to new_lines."""
 
425
        new_texts = new_lines.text()
 
426
        old_texts = self.text()
 
427
        s = patiencediff.PatienceSequenceMatcher(None, old_texts, new_texts)
 
428
        for tag, i1, i2, j1, j2 in s.get_opcodes():
 
429
            if tag == 'equal':
 
430
                continue
 
431
            # ofrom, oto, length, data
 
432
            yield i1, i2, j2 - j1, new_lines._lines[j1:j2]
 
433
 
 
434
    def line_delta(self, new_lines):
 
435
        return list(self.line_delta_iter(new_lines))
 
436
 
 
437
    @staticmethod
 
438
    def get_line_delta_blocks(knit_delta, source, target):
 
439
        """Extract SequenceMatcher.get_matching_blocks() from a knit delta"""
 
440
        target_len = len(target)
 
441
        s_pos = 0
 
442
        t_pos = 0
 
443
        for s_begin, s_end, t_len, new_text in knit_delta:
 
444
            true_n = s_begin - s_pos
 
445
            n = true_n
 
446
            if n > 0:
 
447
                # knit deltas do not provide reliable info about whether the
 
448
                # last line of a file matches, due to eol handling.
 
449
                if source[s_pos + n -1] != target[t_pos + n -1]:
 
450
                    n-=1
 
451
                if n > 0:
 
452
                    yield s_pos, t_pos, n
 
453
            t_pos += t_len + true_n
 
454
            s_pos = s_end
 
455
        n = target_len - t_pos
 
456
        if n > 0:
 
457
            if source[s_pos + n -1] != target[t_pos + n -1]:
 
458
                n-=1
 
459
            if n > 0:
 
460
                yield s_pos, t_pos, n
 
461
        yield s_pos + (target_len - t_pos), target_len, 0
 
462
 
 
463
 
 
464
class AnnotatedKnitContent(KnitContent):
 
465
    """Annotated content."""
 
466
 
 
467
    def __init__(self, lines):
 
468
        KnitContent.__init__(self)
 
469
        self._lines = lines
 
470
 
 
471
    def annotate(self):
 
472
        """Return a list of (origin, text) for each content line."""
 
473
        lines = self._lines[:]
 
474
        if self._should_strip_eol:
 
475
            origin, last_line = lines[-1]
 
476
            lines[-1] = (origin, last_line.rstrip('\n'))
 
477
        return lines
 
478
 
 
479
    def apply_delta(self, delta, new_version_id):
 
480
        """Apply delta to this object to become new_version_id."""
 
481
        offset = 0
 
482
        lines = self._lines
 
483
        for start, end, count, delta_lines in delta:
 
484
            lines[offset+start:offset+end] = delta_lines
 
485
            offset = offset + (start - end) + count
 
486
 
 
487
    def text(self):
 
488
        try:
 
489
            lines = [text for origin, text in self._lines]
 
490
        except ValueError, e:
 
491
            # most commonly (only?) caused by the internal form of the knit
 
492
            # missing annotation information because of a bug - see thread
 
493
            # around 20071015
 
494
            raise KnitCorrupt(self,
 
495
                "line in annotated knit missing annotation information: %s"
 
496
                % (e,))
 
497
        if self._should_strip_eol:
 
498
            lines[-1] = lines[-1].rstrip('\n')
 
499
        return lines
 
500
 
 
501
    def copy(self):
 
502
        return AnnotatedKnitContent(self._lines[:])
 
503
 
 
504
 
 
505
class PlainKnitContent(KnitContent):
 
506
    """Unannotated content.
 
507
 
 
508
    When annotate[_iter] is called on this content, the same version is reported
 
509
    for all lines. Generally, annotate[_iter] is not useful on PlainKnitContent
 
510
    objects.
 
511
    """
 
512
 
 
513
    def __init__(self, lines, version_id):
 
514
        KnitContent.__init__(self)
 
515
        self._lines = lines
 
516
        self._version_id = version_id
 
517
 
 
518
    def annotate(self):
 
519
        """Return a list of (origin, text) for each content line."""
 
520
        return [(self._version_id, line) for line in self._lines]
 
521
 
 
522
    def apply_delta(self, delta, new_version_id):
 
523
        """Apply delta to this object to become new_version_id."""
 
524
        offset = 0
 
525
        lines = self._lines
 
526
        for start, end, count, delta_lines in delta:
 
527
            lines[offset+start:offset+end] = delta_lines
 
528
            offset = offset + (start - end) + count
 
529
        self._version_id = new_version_id
 
530
 
 
531
    def copy(self):
 
532
        return PlainKnitContent(self._lines[:], self._version_id)
 
533
 
 
534
    def text(self):
 
535
        lines = self._lines
 
536
        if self._should_strip_eol:
 
537
            lines = lines[:]
 
538
            lines[-1] = lines[-1].rstrip('\n')
 
539
        return lines
 
540
 
 
541
 
 
542
class _KnitFactory(object):
 
543
    """Base class for common Factory functions."""
 
544
 
 
545
    def parse_record(self, version_id, record, record_details,
 
546
                     base_content, copy_base_content=True):
 
547
        """Parse a record into a full content object.
 
548
 
 
549
        :param version_id: The official version id for this content
 
550
        :param record: The data returned by read_records_iter()
 
551
        :param record_details: Details about the record returned by
 
552
            get_build_details
 
553
        :param base_content: If get_build_details returns a compression_parent,
 
554
            you must return a base_content here, else use None
 
555
        :param copy_base_content: When building from the base_content, decide
 
556
            you can either copy it and return a new object, or modify it in
 
557
            place.
 
558
        :return: (content, delta) A Content object and possibly a line-delta,
 
559
            delta may be None
 
560
        """
 
561
        method, noeol = record_details
 
562
        if method == 'line-delta':
 
563
            if copy_base_content:
 
564
                content = base_content.copy()
 
565
            else:
 
566
                content = base_content
 
567
            delta = self.parse_line_delta(record, version_id)
 
568
            content.apply_delta(delta, version_id)
 
569
        else:
 
570
            content = self.parse_fulltext(record, version_id)
 
571
            delta = None
 
572
        content._should_strip_eol = noeol
 
573
        return (content, delta)
 
574
 
 
575
 
 
576
class KnitAnnotateFactory(_KnitFactory):
 
577
    """Factory for creating annotated Content objects."""
 
578
 
 
579
    annotated = True
 
580
 
 
581
    def make(self, lines, version_id):
 
582
        num_lines = len(lines)
 
583
        return AnnotatedKnitContent(zip([version_id] * num_lines, lines))
 
584
 
 
585
    def parse_fulltext(self, content, version_id):
 
586
        """Convert fulltext to internal representation
 
587
 
 
588
        fulltext content is of the format
 
589
        revid(utf8) plaintext\n
 
590
        internal representation is of the format:
 
591
        (revid, plaintext)
 
592
        """
 
593
        # TODO: jam 20070209 The tests expect this to be returned as tuples,
 
594
        #       but the code itself doesn't really depend on that.
 
595
        #       Figure out a way to not require the overhead of turning the
 
596
        #       list back into tuples.
 
597
        lines = [tuple(line.split(' ', 1)) for line in content]
 
598
        return AnnotatedKnitContent(lines)
 
599
 
 
600
    def parse_line_delta_iter(self, lines):
 
601
        return iter(self.parse_line_delta(lines))
 
602
 
 
603
    def parse_line_delta(self, lines, version_id, plain=False):
 
604
        """Convert a line based delta into internal representation.
 
605
 
 
606
        line delta is in the form of:
 
607
        intstart intend intcount
 
608
        1..count lines:
 
609
        revid(utf8) newline\n
 
610
        internal representation is
 
611
        (start, end, count, [1..count tuples (revid, newline)])
 
612
 
 
613
        :param plain: If True, the lines are returned as a plain
 
614
            list without annotations, not as a list of (origin, content) tuples, i.e.
 
615
            (start, end, count, [1..count newline])
 
616
        """
 
617
        result = []
 
618
        lines = iter(lines)
 
619
        next = lines.next
 
620
 
 
621
        cache = {}
 
622
        def cache_and_return(line):
 
623
            origin, text = line.split(' ', 1)
 
624
            return cache.setdefault(origin, origin), text
 
625
 
 
626
        # walk through the lines parsing.
 
627
        # Note that the plain test is explicitly pulled out of the
 
628
        # loop to minimise any performance impact
 
629
        if plain:
 
630
            for header in lines:
 
631
                start, end, count = [int(n) for n in header.split(',')]
 
632
                contents = [next().split(' ', 1)[1] for i in xrange(count)]
 
633
                result.append((start, end, count, contents))
 
634
        else:
 
635
            for header in lines:
 
636
                start, end, count = [int(n) for n in header.split(',')]
 
637
                contents = [tuple(next().split(' ', 1)) for i in xrange(count)]
 
638
                result.append((start, end, count, contents))
 
639
        return result
 
640
 
 
641
    def get_fulltext_content(self, lines):
 
642
        """Extract just the content lines from a fulltext."""
 
643
        return (line.split(' ', 1)[1] for line in lines)
 
644
 
 
645
    def get_linedelta_content(self, lines):
 
646
        """Extract just the content from a line delta.
 
647
 
 
648
        This doesn't return all of the extra information stored in a delta.
 
649
        Only the actual content lines.
 
650
        """
 
651
        lines = iter(lines)
 
652
        next = lines.next
 
653
        for header in lines:
 
654
            header = header.split(',')
 
655
            count = int(header[2])
 
656
            for i in xrange(count):
 
657
                origin, text = next().split(' ', 1)
 
658
                yield text
 
659
 
 
660
    def lower_fulltext(self, content):
 
661
        """convert a fulltext content record into a serializable form.
 
662
 
 
663
        see parse_fulltext which this inverts.
 
664
        """
 
665
        return ['%s %s' % (o, t) for o, t in content._lines]
 
666
 
 
667
    def lower_line_delta(self, delta):
 
668
        """convert a delta into a serializable form.
 
669
 
 
670
        See parse_line_delta which this inverts.
 
671
        """
 
672
        # TODO: jam 20070209 We only do the caching thing to make sure that
 
673
        #       the origin is a valid utf-8 line, eventually we could remove it
 
674
        out = []
 
675
        for start, end, c, lines in delta:
 
676
            out.append('%d,%d,%d\n' % (start, end, c))
 
677
            out.extend(origin + ' ' + text
 
678
                       for origin, text in lines)
 
679
        return out
 
680
 
 
681
    def annotate(self, knit, key):
 
682
        content = knit._get_content(key)
 
683
        # adjust for the fact that serialised annotations are only key suffixes
 
684
        # for this factory.
 
685
        if type(key) is tuple:
 
686
            prefix = key[:-1]
 
687
            origins = content.annotate()
 
688
            result = []
 
689
            for origin, line in origins:
 
690
                result.append((prefix + (origin,), line))
 
691
            return result
 
692
        else:
 
693
            # XXX: This smells a bit.  Why would key ever be a non-tuple here?
 
694
            # Aren't keys defined to be tuples?  -- spiv 20080618
 
695
            return content.annotate()
 
696
 
 
697
 
 
698
class KnitPlainFactory(_KnitFactory):
 
699
    """Factory for creating plain Content objects."""
 
700
 
 
701
    annotated = False
 
702
 
 
703
    def make(self, lines, version_id):
 
704
        return PlainKnitContent(lines, version_id)
 
705
 
 
706
    def parse_fulltext(self, content, version_id):
 
707
        """This parses an unannotated fulltext.
 
708
 
 
709
        Note that this is not a noop - the internal representation
 
710
        has (versionid, line) - its just a constant versionid.
 
711
        """
 
712
        return self.make(content, version_id)
 
713
 
 
714
    def parse_line_delta_iter(self, lines, version_id):
 
715
        cur = 0
 
716
        num_lines = len(lines)
 
717
        while cur < num_lines:
 
718
            header = lines[cur]
 
719
            cur += 1
 
720
            start, end, c = [int(n) for n in header.split(',')]
 
721
            yield start, end, c, lines[cur:cur+c]
 
722
            cur += c
 
723
 
 
724
    def parse_line_delta(self, lines, version_id):
 
725
        return list(self.parse_line_delta_iter(lines, version_id))
 
726
 
 
727
    def get_fulltext_content(self, lines):
 
728
        """Extract just the content lines from a fulltext."""
 
729
        return iter(lines)
 
730
 
 
731
    def get_linedelta_content(self, lines):
 
732
        """Extract just the content from a line delta.
 
733
 
 
734
        This doesn't return all of the extra information stored in a delta.
 
735
        Only the actual content lines.
 
736
        """
 
737
        lines = iter(lines)
 
738
        next = lines.next
 
739
        for header in lines:
 
740
            header = header.split(',')
 
741
            count = int(header[2])
 
742
            for i in xrange(count):
 
743
                yield next()
 
744
 
 
745
    def lower_fulltext(self, content):
 
746
        return content.text()
 
747
 
 
748
    def lower_line_delta(self, delta):
 
749
        out = []
 
750
        for start, end, c, lines in delta:
 
751
            out.append('%d,%d,%d\n' % (start, end, c))
 
752
            out.extend(lines)
 
753
        return out
 
754
 
 
755
    def annotate(self, knit, key):
 
756
        annotator = _KnitAnnotator(knit)
 
757
        return annotator.annotate_flat(key)
 
758
 
 
759
 
 
760
 
 
761
def make_file_factory(annotated, mapper):
 
762
    """Create a factory for creating a file based KnitVersionedFiles.
 
763
 
 
764
    This is only functional enough to run interface tests, it doesn't try to
 
765
    provide a full pack environment.
 
766
 
 
767
    :param annotated: knit annotations are wanted.
 
768
    :param mapper: The mapper from keys to paths.
 
769
    """
 
770
    def factory(transport):
 
771
        index = _KndxIndex(transport, mapper, lambda:None, lambda:True, lambda:True)
 
772
        access = _KnitKeyAccess(transport, mapper)
 
773
        return KnitVersionedFiles(index, access, annotated=annotated)
 
774
    return factory
 
775
 
 
776
 
 
777
def make_pack_factory(graph, delta, keylength):
 
778
    """Create a factory for creating a pack based VersionedFiles.
 
779
 
 
780
    This is only functional enough to run interface tests, it doesn't try to
 
781
    provide a full pack environment.
 
782
 
 
783
    :param graph: Store a graph.
 
784
    :param delta: Delta compress contents.
 
785
    :param keylength: How long should keys be.
 
786
    """
 
787
    def factory(transport):
 
788
        parents = graph or delta
 
789
        ref_length = 0
 
790
        if graph:
 
791
            ref_length += 1
 
792
        if delta:
 
793
            ref_length += 1
 
794
            max_delta_chain = 200
 
795
        else:
 
796
            max_delta_chain = 0
 
797
        graph_index = _mod_index.InMemoryGraphIndex(reference_lists=ref_length,
 
798
            key_elements=keylength)
 
799
        stream = transport.open_write_stream('newpack')
 
800
        writer = pack.ContainerWriter(stream.write)
 
801
        writer.begin()
 
802
        index = _KnitGraphIndex(graph_index, lambda:True, parents=parents,
 
803
            deltas=delta, add_callback=graph_index.add_nodes)
 
804
        access = pack_repo._DirectPackAccess({})
 
805
        access.set_writer(writer, graph_index, (transport, 'newpack'))
 
806
        result = KnitVersionedFiles(index, access,
 
807
            max_delta_chain=max_delta_chain)
 
808
        result.stream = stream
 
809
        result.writer = writer
 
810
        return result
 
811
    return factory
 
812
 
 
813
 
 
814
def cleanup_pack_knit(versioned_files):
 
815
    versioned_files.stream.close()
 
816
    versioned_files.writer.end()
 
817
 
 
818
 
 
819
def _get_total_build_size(self, keys, positions):
 
820
    """Determine the total bytes to build these keys.
 
821
 
 
822
    (helper function because _KnitGraphIndex and _KndxIndex work the same, but
 
823
    don't inherit from a common base.)
 
824
 
 
825
    :param keys: Keys that we want to build
 
826
    :param positions: dict of {key, (info, index_memo, comp_parent)} (such
 
827
        as returned by _get_components_positions)
 
828
    :return: Number of bytes to build those keys
 
829
    """
 
830
    all_build_index_memos = {}
 
831
    build_keys = keys
 
832
    while build_keys:
 
833
        next_keys = set()
 
834
        for key in build_keys:
 
835
            # This is mostly for the 'stacked' case
 
836
            # Where we will be getting the data from a fallback
 
837
            if key not in positions:
 
838
                continue
 
839
            _, index_memo, compression_parent = positions[key]
 
840
            all_build_index_memos[key] = index_memo
 
841
            if compression_parent not in all_build_index_memos:
 
842
                next_keys.add(compression_parent)
 
843
        build_keys = next_keys
 
844
    return sum([index_memo[2] for index_memo
 
845
                in all_build_index_memos.itervalues()])
 
846
 
 
847
 
 
848
class KnitVersionedFiles(VersionedFiles):
 
849
    """Storage for many versioned files using knit compression.
 
850
 
 
851
    Backend storage is managed by indices and data objects.
 
852
 
 
853
    :ivar _index: A _KnitGraphIndex or similar that can describe the
 
854
        parents, graph, compression and data location of entries in this
 
855
        KnitVersionedFiles.  Note that this is only the index for
 
856
        *this* vfs; if there are fallbacks they must be queried separately.
 
857
    """
 
858
 
 
859
    def __init__(self, index, data_access, max_delta_chain=200,
 
860
                 annotated=False, reload_func=None):
 
861
        """Create a KnitVersionedFiles with index and data_access.
 
862
 
 
863
        :param index: The index for the knit data.
 
864
        :param data_access: The access object to store and retrieve knit
 
865
            records.
 
866
        :param max_delta_chain: The maximum number of deltas to permit during
 
867
            insertion. Set to 0 to prohibit the use of deltas.
 
868
        :param annotated: Set to True to cause annotations to be calculated and
 
869
            stored during insertion.
 
870
        :param reload_func: An function that can be called if we think we need
 
871
            to reload the pack listing and try again. See
 
872
            'bzrlib.repofmt.pack_repo.AggregateIndex' for the signature.
 
873
        """
 
874
        self._index = index
 
875
        self._access = data_access
 
876
        self._max_delta_chain = max_delta_chain
 
877
        if annotated:
 
878
            self._factory = KnitAnnotateFactory()
 
879
        else:
 
880
            self._factory = KnitPlainFactory()
 
881
        self._immediate_fallback_vfs = []
 
882
        self._reload_func = reload_func
 
883
 
 
884
    def __repr__(self):
 
885
        return "%s(%r, %r)" % (
 
886
            self.__class__.__name__,
 
887
            self._index,
 
888
            self._access)
 
889
 
 
890
    def add_fallback_versioned_files(self, a_versioned_files):
 
891
        """Add a source of texts for texts not present in this knit.
 
892
 
 
893
        :param a_versioned_files: A VersionedFiles object.
 
894
        """
 
895
        self._immediate_fallback_vfs.append(a_versioned_files)
 
896
 
 
897
    def add_lines(self, key, parents, lines, parent_texts=None,
 
898
        left_matching_blocks=None, nostore_sha=None, random_id=False,
 
899
        check_content=True):
 
900
        """See VersionedFiles.add_lines()."""
 
901
        self._index._check_write_ok()
 
902
        self._check_add(key, lines, random_id, check_content)
 
903
        if parents is None:
 
904
            # The caller might pass None if there is no graph data, but kndx
 
905
            # indexes can't directly store that, so we give them
 
906
            # an empty tuple instead.
 
907
            parents = ()
 
908
        line_bytes = ''.join(lines)
 
909
        return self._add(key, lines, parents,
 
910
            parent_texts, left_matching_blocks, nostore_sha, random_id,
 
911
            line_bytes=line_bytes)
 
912
 
 
913
    def _add_text(self, key, parents, text, nostore_sha=None, random_id=False):
 
914
        """See VersionedFiles._add_text()."""
 
915
        self._index._check_write_ok()
 
916
        self._check_add(key, None, random_id, check_content=False)
 
917
        if text.__class__ is not str:
 
918
            raise errors.BzrBadParameterUnicode("text")
 
919
        if parents is None:
 
920
            # The caller might pass None if there is no graph data, but kndx
 
921
            # indexes can't directly store that, so we give them
 
922
            # an empty tuple instead.
 
923
            parents = ()
 
924
        return self._add(key, None, parents,
 
925
            None, None, nostore_sha, random_id,
 
926
            line_bytes=text)
 
927
 
 
928
    def _add(self, key, lines, parents, parent_texts,
 
929
        left_matching_blocks, nostore_sha, random_id,
 
930
        line_bytes):
 
931
        """Add a set of lines on top of version specified by parents.
 
932
 
 
933
        Any versions not present will be converted into ghosts.
 
934
 
 
935
        :param lines: A list of strings where each one is a single line (has a
 
936
            single newline at the end of the string) This is now optional
 
937
            (callers can pass None). It is left in its location for backwards
 
938
            compatibility. It should ''.join(lines) must == line_bytes
 
939
        :param line_bytes: A single string containing the content
 
940
 
 
941
        We pass both lines and line_bytes because different routes bring the
 
942
        values to this function. And for memory efficiency, we don't want to
 
943
        have to split/join on-demand.
 
944
        """
 
945
        # first thing, if the content is something we don't need to store, find
 
946
        # that out.
 
947
        digest = sha_string(line_bytes)
 
948
        if nostore_sha == digest:
 
949
            raise errors.ExistingContent
 
950
 
 
951
        present_parents = []
 
952
        if parent_texts is None:
 
953
            parent_texts = {}
 
954
        # Do a single query to ascertain parent presence; we only compress
 
955
        # against parents in the same kvf.
 
956
        present_parent_map = self._index.get_parent_map(parents)
 
957
        for parent in parents:
 
958
            if parent in present_parent_map:
 
959
                present_parents.append(parent)
 
960
 
 
961
        # Currently we can only compress against the left most present parent.
 
962
        if (len(present_parents) == 0 or
 
963
            present_parents[0] != parents[0]):
 
964
            delta = False
 
965
        else:
 
966
            # To speed the extract of texts the delta chain is limited
 
967
            # to a fixed number of deltas.  This should minimize both
 
968
            # I/O and the time spend applying deltas.
 
969
            delta = self._check_should_delta(present_parents[0])
 
970
 
 
971
        text_length = len(line_bytes)
 
972
        options = []
 
973
        no_eol = False
 
974
        # Note: line_bytes is not modified to add a newline, that is tracked
 
975
        #       via the no_eol flag. 'lines' *is* modified, because that is the
 
976
        #       general values needed by the Content code.
 
977
        if line_bytes and line_bytes[-1] != '\n':
 
978
            options.append('no-eol')
 
979
            no_eol = True
 
980
            # Copy the existing list, or create a new one
 
981
            if lines is None:
 
982
                lines = osutils.split_lines(line_bytes)
 
983
            else:
 
984
                lines = lines[:]
 
985
            # Replace the last line with one that ends in a final newline
 
986
            lines[-1] = lines[-1] + '\n'
 
987
        if lines is None:
 
988
            lines = osutils.split_lines(line_bytes)
 
989
 
 
990
        for element in key[:-1]:
 
991
            if type(element) is not str:
 
992
                raise TypeError("key contains non-strings: %r" % (key,))
 
993
        if key[-1] is None:
 
994
            key = key[:-1] + ('sha1:' + digest,)
 
995
        elif type(key[-1]) is not str:
 
996
                raise TypeError("key contains non-strings: %r" % (key,))
 
997
        # Knit hunks are still last-element only
 
998
        version_id = key[-1]
 
999
        content = self._factory.make(lines, version_id)
 
1000
        if no_eol:
 
1001
            # Hint to the content object that its text() call should strip the
 
1002
            # EOL.
 
1003
            content._should_strip_eol = True
 
1004
        if delta or (self._factory.annotated and len(present_parents) > 0):
 
1005
            # Merge annotations from parent texts if needed.
 
1006
            delta_hunks = self._merge_annotations(content, present_parents,
 
1007
                parent_texts, delta, self._factory.annotated,
 
1008
                left_matching_blocks)
 
1009
 
 
1010
        if delta:
 
1011
            options.append('line-delta')
 
1012
            store_lines = self._factory.lower_line_delta(delta_hunks)
 
1013
            size, bytes = self._record_to_data(key, digest,
 
1014
                store_lines)
 
1015
        else:
 
1016
            options.append('fulltext')
 
1017
            # isinstance is slower and we have no hierarchy.
 
1018
            if self._factory.__class__ is KnitPlainFactory:
 
1019
                # Use the already joined bytes saving iteration time in
 
1020
                # _record_to_data.
 
1021
                dense_lines = [line_bytes]
 
1022
                if no_eol:
 
1023
                    dense_lines.append('\n')
 
1024
                size, bytes = self._record_to_data(key, digest,
 
1025
                    lines, dense_lines)
 
1026
            else:
 
1027
                # get mixed annotation + content and feed it into the
 
1028
                # serialiser.
 
1029
                store_lines = self._factory.lower_fulltext(content)
 
1030
                size, bytes = self._record_to_data(key, digest,
 
1031
                    store_lines)
 
1032
 
 
1033
        access_memo = self._access.add_raw_records([(key, size)], bytes)[0]
 
1034
        self._index.add_records(
 
1035
            ((key, options, access_memo, parents),),
 
1036
            random_id=random_id)
 
1037
        return digest, text_length, content
 
1038
 
 
1039
    def annotate(self, key):
 
1040
        """See VersionedFiles.annotate."""
 
1041
        return self._factory.annotate(self, key)
 
1042
 
 
1043
    def get_annotator(self):
 
1044
        return _KnitAnnotator(self)
 
1045
 
 
1046
    def check(self, progress_bar=None, keys=None):
 
1047
        """See VersionedFiles.check()."""
 
1048
        if keys is None:
 
1049
            return self._logical_check()
 
1050
        else:
 
1051
            # At the moment, check does not extra work over get_record_stream
 
1052
            return self.get_record_stream(keys, 'unordered', True)
 
1053
 
 
1054
    def _logical_check(self):
 
1055
        # This doesn't actually test extraction of everything, but that will
 
1056
        # impact 'bzr check' substantially, and needs to be integrated with
 
1057
        # care. However, it does check for the obvious problem of a delta with
 
1058
        # no basis.
 
1059
        keys = self._index.keys()
 
1060
        parent_map = self.get_parent_map(keys)
 
1061
        for key in keys:
 
1062
            if self._index.get_method(key) != 'fulltext':
 
1063
                compression_parent = parent_map[key][0]
 
1064
                if compression_parent not in parent_map:
 
1065
                    raise errors.KnitCorrupt(self,
 
1066
                        "Missing basis parent %s for %s" % (
 
1067
                        compression_parent, key))
 
1068
        for fallback_vfs in self._immediate_fallback_vfs:
 
1069
            fallback_vfs.check()
 
1070
 
 
1071
    def _check_add(self, key, lines, random_id, check_content):
 
1072
        """check that version_id and lines are safe to add."""
 
1073
        version_id = key[-1]
 
1074
        if version_id is not None:
 
1075
            if contains_whitespace(version_id):
 
1076
                raise InvalidRevisionId(version_id, self)
 
1077
            self.check_not_reserved_id(version_id)
 
1078
        # TODO: If random_id==False and the key is already present, we should
 
1079
        # probably check that the existing content is identical to what is
 
1080
        # being inserted, and otherwise raise an exception.  This would make
 
1081
        # the bundle code simpler.
 
1082
        if check_content:
 
1083
            self._check_lines_not_unicode(lines)
 
1084
            self._check_lines_are_lines(lines)
 
1085
 
 
1086
    def _check_header(self, key, line):
 
1087
        rec = self._split_header(line)
 
1088
        self._check_header_version(rec, key[-1])
 
1089
        return rec
 
1090
 
 
1091
    def _check_header_version(self, rec, version_id):
 
1092
        """Checks the header version on original format knit records.
 
1093
 
 
1094
        These have the last component of the key embedded in the record.
 
1095
        """
 
1096
        if rec[1] != version_id:
 
1097
            raise KnitCorrupt(self,
 
1098
                'unexpected version, wanted %r, got %r' % (version_id, rec[1]))
 
1099
 
 
1100
    def _check_should_delta(self, parent):
 
1101
        """Iterate back through the parent listing, looking for a fulltext.
 
1102
 
 
1103
        This is used when we want to decide whether to add a delta or a new
 
1104
        fulltext. It searches for _max_delta_chain parents. When it finds a
 
1105
        fulltext parent, it sees if the total size of the deltas leading up to
 
1106
        it is large enough to indicate that we want a new full text anyway.
 
1107
 
 
1108
        Return True if we should create a new delta, False if we should use a
 
1109
        full text.
 
1110
        """
 
1111
        delta_size = 0
 
1112
        fulltext_size = None
 
1113
        for count in xrange(self._max_delta_chain):
 
1114
            try:
 
1115
                # Note that this only looks in the index of this particular
 
1116
                # KnitVersionedFiles, not in the fallbacks.  This ensures that
 
1117
                # we won't store a delta spanning physical repository
 
1118
                # boundaries.
 
1119
                build_details = self._index.get_build_details([parent])
 
1120
                parent_details = build_details[parent]
 
1121
            except (RevisionNotPresent, KeyError), e:
 
1122
                # Some basis is not locally present: always fulltext
 
1123
                return False
 
1124
            index_memo, compression_parent, _, _ = parent_details
 
1125
            _, _, size = index_memo
 
1126
            if compression_parent is None:
 
1127
                fulltext_size = size
 
1128
                break
 
1129
            delta_size += size
 
1130
            # We don't explicitly check for presence because this is in an
 
1131
            # inner loop, and if it's missing it'll fail anyhow.
 
1132
            parent = compression_parent
 
1133
        else:
 
1134
            # We couldn't find a fulltext, so we must create a new one
 
1135
            return False
 
1136
        # Simple heuristic - if the total I/O wold be greater as a delta than
 
1137
        # the originally installed fulltext, we create a new fulltext.
 
1138
        return fulltext_size > delta_size
 
1139
 
 
1140
    def _build_details_to_components(self, build_details):
 
1141
        """Convert a build_details tuple to a position tuple."""
 
1142
        # record_details, access_memo, compression_parent
 
1143
        return build_details[3], build_details[0], build_details[1]
 
1144
 
 
1145
    def _get_components_positions(self, keys, allow_missing=False):
 
1146
        """Produce a map of position data for the components of keys.
 
1147
 
 
1148
        This data is intended to be used for retrieving the knit records.
 
1149
 
 
1150
        A dict of key to (record_details, index_memo, next, parents) is
 
1151
        returned.
 
1152
        method is the way referenced data should be applied.
 
1153
        index_memo is the handle to pass to the data access to actually get the
 
1154
            data
 
1155
        next is the build-parent of the version, or None for fulltexts.
 
1156
        parents is the version_ids of the parents of this version
 
1157
 
 
1158
        :param allow_missing: If True do not raise an error on a missing component,
 
1159
            just ignore it.
 
1160
        """
 
1161
        component_data = {}
 
1162
        pending_components = keys
 
1163
        while pending_components:
 
1164
            build_details = self._index.get_build_details(pending_components)
 
1165
            current_components = set(pending_components)
 
1166
            pending_components = set()
 
1167
            for key, details in build_details.iteritems():
 
1168
                (index_memo, compression_parent, parents,
 
1169
                 record_details) = details
 
1170
                method = record_details[0]
 
1171
                if compression_parent is not None:
 
1172
                    pending_components.add(compression_parent)
 
1173
                component_data[key] = self._build_details_to_components(details)
 
1174
            missing = current_components.difference(build_details)
 
1175
            if missing and not allow_missing:
 
1176
                raise errors.RevisionNotPresent(missing.pop(), self)
 
1177
        return component_data
 
1178
 
 
1179
    def _get_content(self, key, parent_texts={}):
 
1180
        """Returns a content object that makes up the specified
 
1181
        version."""
 
1182
        cached_version = parent_texts.get(key, None)
 
1183
        if cached_version is not None:
 
1184
            # Ensure the cache dict is valid.
 
1185
            if not self.get_parent_map([key]):
 
1186
                raise RevisionNotPresent(key, self)
 
1187
            return cached_version
 
1188
        generator = _VFContentMapGenerator(self, [key])
 
1189
        return generator._get_content(key)
 
1190
 
 
1191
    def get_known_graph_ancestry(self, keys):
 
1192
        """Get a KnownGraph instance with the ancestry of keys."""
 
1193
        parent_map, missing_keys = self._index.find_ancestry(keys)
 
1194
        for fallback in self._transitive_fallbacks():
 
1195
            if not missing_keys:
 
1196
                break
 
1197
            (f_parent_map, f_missing_keys) = fallback._index.find_ancestry(
 
1198
                                                missing_keys)
 
1199
            parent_map.update(f_parent_map)
 
1200
            missing_keys = f_missing_keys
 
1201
        kg = _mod_graph.KnownGraph(parent_map)
 
1202
        return kg
 
1203
 
 
1204
    def get_parent_map(self, keys):
 
1205
        """Get a map of the graph parents of keys.
 
1206
 
 
1207
        :param keys: The keys to look up parents for.
 
1208
        :return: A mapping from keys to parents. Absent keys are absent from
 
1209
            the mapping.
 
1210
        """
 
1211
        return self._get_parent_map_with_sources(keys)[0]
 
1212
 
 
1213
    def _get_parent_map_with_sources(self, keys):
 
1214
        """Get a map of the parents of keys.
 
1215
 
 
1216
        :param keys: The keys to look up parents for.
 
1217
        :return: A tuple. The first element is a mapping from keys to parents.
 
1218
            Absent keys are absent from the mapping. The second element is a
 
1219
            list with the locations each key was found in. The first element
 
1220
            is the in-this-knit parents, the second the first fallback source,
 
1221
            and so on.
 
1222
        """
 
1223
        result = {}
 
1224
        sources = [self._index] + self._immediate_fallback_vfs
 
1225
        source_results = []
 
1226
        missing = set(keys)
 
1227
        for source in sources:
 
1228
            if not missing:
 
1229
                break
 
1230
            new_result = source.get_parent_map(missing)
 
1231
            source_results.append(new_result)
 
1232
            result.update(new_result)
 
1233
            missing.difference_update(set(new_result))
 
1234
        return result, source_results
 
1235
 
 
1236
    def _get_record_map(self, keys, allow_missing=False):
 
1237
        """Produce a dictionary of knit records.
 
1238
 
 
1239
        :return: {key:(record, record_details, digest, next)}
 
1240
            record
 
1241
                data returned from read_records (a KnitContentobject)
 
1242
            record_details
 
1243
                opaque information to pass to parse_record
 
1244
            digest
 
1245
                SHA1 digest of the full text after all steps are done
 
1246
            next
 
1247
                build-parent of the version, i.e. the leftmost ancestor.
 
1248
                Will be None if the record is not a delta.
 
1249
        :param keys: The keys to build a map for
 
1250
        :param allow_missing: If some records are missing, rather than
 
1251
            error, just return the data that could be generated.
 
1252
        """
 
1253
        raw_map = self._get_record_map_unparsed(keys,
 
1254
            allow_missing=allow_missing)
 
1255
        return self._raw_map_to_record_map(raw_map)
 
1256
 
 
1257
    def _raw_map_to_record_map(self, raw_map):
 
1258
        """Parse the contents of _get_record_map_unparsed.
 
1259
 
 
1260
        :return: see _get_record_map.
 
1261
        """
 
1262
        result = {}
 
1263
        for key in raw_map:
 
1264
            data, record_details, next = raw_map[key]
 
1265
            content, digest = self._parse_record(key[-1], data)
 
1266
            result[key] = content, record_details, digest, next
 
1267
        return result
 
1268
 
 
1269
    def _get_record_map_unparsed(self, keys, allow_missing=False):
 
1270
        """Get the raw data for reconstructing keys without parsing it.
 
1271
 
 
1272
        :return: A dict suitable for parsing via _raw_map_to_record_map.
 
1273
            key-> raw_bytes, (method, noeol), compression_parent
 
1274
        """
 
1275
        # This retries the whole request if anything fails. Potentially we
 
1276
        # could be a bit more selective. We could track the keys whose records
 
1277
        # we have successfully found, and then only request the new records
 
1278
        # from there. However, _get_components_positions grabs the whole build
 
1279
        # chain, which means we'll likely try to grab the same records again
 
1280
        # anyway. Also, can the build chains change as part of a pack
 
1281
        # operation? We wouldn't want to end up with a broken chain.
 
1282
        while True:
 
1283
            try:
 
1284
                position_map = self._get_components_positions(keys,
 
1285
                    allow_missing=allow_missing)
 
1286
                # key = component_id, r = record_details, i_m = index_memo,
 
1287
                # n = next
 
1288
                records = [(key, i_m) for key, (r, i_m, n)
 
1289
                                       in position_map.iteritems()]
 
1290
                # Sort by the index memo, so that we request records from the
 
1291
                # same pack file together, and in forward-sorted order
 
1292
                records.sort(key=operator.itemgetter(1))
 
1293
                raw_record_map = {}
 
1294
                for key, data in self._read_records_iter_unchecked(records):
 
1295
                    (record_details, index_memo, next) = position_map[key]
 
1296
                    raw_record_map[key] = data, record_details, next
 
1297
                return raw_record_map
 
1298
            except errors.RetryWithNewPacks, e:
 
1299
                self._access.reload_or_raise(e)
 
1300
 
 
1301
    @classmethod
 
1302
    def _split_by_prefix(cls, keys):
 
1303
        """For the given keys, split them up based on their prefix.
 
1304
 
 
1305
        To keep memory pressure somewhat under control, split the
 
1306
        requests back into per-file-id requests, otherwise "bzr co"
 
1307
        extracts the full tree into memory before writing it to disk.
 
1308
        This should be revisited if _get_content_maps() can ever cross
 
1309
        file-id boundaries.
 
1310
 
 
1311
        The keys for a given file_id are kept in the same relative order.
 
1312
        Ordering between file_ids is not, though prefix_order will return the
 
1313
        order that the key was first seen.
 
1314
 
 
1315
        :param keys: An iterable of key tuples
 
1316
        :return: (split_map, prefix_order)
 
1317
            split_map       A dictionary mapping prefix => keys
 
1318
            prefix_order    The order that we saw the various prefixes
 
1319
        """
 
1320
        split_by_prefix = {}
 
1321
        prefix_order = []
 
1322
        for key in keys:
 
1323
            if len(key) == 1:
 
1324
                prefix = ''
 
1325
            else:
 
1326
                prefix = key[0]
 
1327
 
 
1328
            if prefix in split_by_prefix:
 
1329
                split_by_prefix[prefix].append(key)
 
1330
            else:
 
1331
                split_by_prefix[prefix] = [key]
 
1332
                prefix_order.append(prefix)
 
1333
        return split_by_prefix, prefix_order
 
1334
 
 
1335
    def _group_keys_for_io(self, keys, non_local_keys, positions,
 
1336
                           _min_buffer_size=_STREAM_MIN_BUFFER_SIZE):
 
1337
        """For the given keys, group them into 'best-sized' requests.
 
1338
 
 
1339
        The idea is to avoid making 1 request per file, but to never try to
 
1340
        unpack an entire 1.5GB source tree in a single pass. Also when
 
1341
        possible, we should try to group requests to the same pack file
 
1342
        together.
 
1343
 
 
1344
        :return: list of (keys, non_local) tuples that indicate what keys
 
1345
            should be fetched next.
 
1346
        """
 
1347
        # TODO: Ideally we would group on 2 factors. We want to extract texts
 
1348
        #       from the same pack file together, and we want to extract all
 
1349
        #       the texts for a given build-chain together. Ultimately it
 
1350
        #       probably needs a better global view.
 
1351
        total_keys = len(keys)
 
1352
        prefix_split_keys, prefix_order = self._split_by_prefix(keys)
 
1353
        prefix_split_non_local_keys, _ = self._split_by_prefix(non_local_keys)
 
1354
        cur_keys = []
 
1355
        cur_non_local = set()
 
1356
        cur_size = 0
 
1357
        result = []
 
1358
        sizes = []
 
1359
        for prefix in prefix_order:
 
1360
            keys = prefix_split_keys[prefix]
 
1361
            non_local = prefix_split_non_local_keys.get(prefix, [])
 
1362
 
 
1363
            this_size = self._index._get_total_build_size(keys, positions)
 
1364
            cur_size += this_size
 
1365
            cur_keys.extend(keys)
 
1366
            cur_non_local.update(non_local)
 
1367
            if cur_size > _min_buffer_size:
 
1368
                result.append((cur_keys, cur_non_local))
 
1369
                sizes.append(cur_size)
 
1370
                cur_keys = []
 
1371
                cur_non_local = set()
 
1372
                cur_size = 0
 
1373
        if cur_keys:
 
1374
            result.append((cur_keys, cur_non_local))
 
1375
            sizes.append(cur_size)
 
1376
        return result
 
1377
 
 
1378
    def get_record_stream(self, keys, ordering, include_delta_closure):
 
1379
        """Get a stream of records for keys.
 
1380
 
 
1381
        :param keys: The keys to include.
 
1382
        :param ordering: Either 'unordered' or 'topological'. A topologically
 
1383
            sorted stream has compression parents strictly before their
 
1384
            children.
 
1385
        :param include_delta_closure: If True then the closure across any
 
1386
            compression parents will be included (in the opaque data).
 
1387
        :return: An iterator of ContentFactory objects, each of which is only
 
1388
            valid until the iterator is advanced.
 
1389
        """
 
1390
        # keys might be a generator
 
1391
        keys = set(keys)
 
1392
        if not keys:
 
1393
            return
 
1394
        if not self._index.has_graph:
 
1395
            # Cannot sort when no graph has been stored.
 
1396
            ordering = 'unordered'
 
1397
 
 
1398
        remaining_keys = keys
 
1399
        while True:
 
1400
            try:
 
1401
                keys = set(remaining_keys)
 
1402
                for content_factory in self._get_remaining_record_stream(keys,
 
1403
                                            ordering, include_delta_closure):
 
1404
                    remaining_keys.discard(content_factory.key)
 
1405
                    yield content_factory
 
1406
                return
 
1407
            except errors.RetryWithNewPacks, e:
 
1408
                self._access.reload_or_raise(e)
 
1409
 
 
1410
    def _get_remaining_record_stream(self, keys, ordering,
 
1411
                                     include_delta_closure):
 
1412
        """This function is the 'retry' portion for get_record_stream."""
 
1413
        if include_delta_closure:
 
1414
            positions = self._get_components_positions(keys, allow_missing=True)
 
1415
        else:
 
1416
            build_details = self._index.get_build_details(keys)
 
1417
            # map from key to
 
1418
            # (record_details, access_memo, compression_parent_key)
 
1419
            positions = dict((key, self._build_details_to_components(details))
 
1420
                for key, details in build_details.iteritems())
 
1421
        absent_keys = keys.difference(set(positions))
 
1422
        # There may be more absent keys : if we're missing the basis component
 
1423
        # and are trying to include the delta closure.
 
1424
        # XXX: We should not ever need to examine remote sources because we do
 
1425
        # not permit deltas across versioned files boundaries.
 
1426
        if include_delta_closure:
 
1427
            needed_from_fallback = set()
 
1428
            # Build up reconstructable_keys dict.  key:True in this dict means
 
1429
            # the key can be reconstructed.
 
1430
            reconstructable_keys = {}
 
1431
            for key in keys:
 
1432
                # the delta chain
 
1433
                try:
 
1434
                    chain = [key, positions[key][2]]
 
1435
                except KeyError:
 
1436
                    needed_from_fallback.add(key)
 
1437
                    continue
 
1438
                result = True
 
1439
                while chain[-1] is not None:
 
1440
                    if chain[-1] in reconstructable_keys:
 
1441
                        result = reconstructable_keys[chain[-1]]
 
1442
                        break
 
1443
                    else:
 
1444
                        try:
 
1445
                            chain.append(positions[chain[-1]][2])
 
1446
                        except KeyError:
 
1447
                            # missing basis component
 
1448
                            needed_from_fallback.add(chain[-1])
 
1449
                            result = True
 
1450
                            break
 
1451
                for chain_key in chain[:-1]:
 
1452
                    reconstructable_keys[chain_key] = result
 
1453
                if not result:
 
1454
                    needed_from_fallback.add(key)
 
1455
        # Double index lookups here : need a unified api ?
 
1456
        global_map, parent_maps = self._get_parent_map_with_sources(keys)
 
1457
        if ordering in ('topological', 'groupcompress'):
 
1458
            if ordering == 'topological':
 
1459
                # Global topological sort
 
1460
                present_keys = tsort.topo_sort(global_map)
 
1461
            else:
 
1462
                present_keys = sort_groupcompress(global_map)
 
1463
            # Now group by source:
 
1464
            source_keys = []
 
1465
            current_source = None
 
1466
            for key in present_keys:
 
1467
                for parent_map in parent_maps:
 
1468
                    if key in parent_map:
 
1469
                        key_source = parent_map
 
1470
                        break
 
1471
                if current_source is not key_source:
 
1472
                    source_keys.append((key_source, []))
 
1473
                    current_source = key_source
 
1474
                source_keys[-1][1].append(key)
 
1475
        else:
 
1476
            if ordering != 'unordered':
 
1477
                raise AssertionError('valid values for ordering are:'
 
1478
                    ' "unordered", "groupcompress" or "topological" not: %r'
 
1479
                    % (ordering,))
 
1480
            # Just group by source; remote sources first.
 
1481
            present_keys = []
 
1482
            source_keys = []
 
1483
            for parent_map in reversed(parent_maps):
 
1484
                source_keys.append((parent_map, []))
 
1485
                for key in parent_map:
 
1486
                    present_keys.append(key)
 
1487
                    source_keys[-1][1].append(key)
 
1488
            # We have been requested to return these records in an order that
 
1489
            # suits us. So we ask the index to give us an optimally sorted
 
1490
            # order.
 
1491
            for source, sub_keys in source_keys:
 
1492
                if source is parent_maps[0]:
 
1493
                    # Only sort the keys for this VF
 
1494
                    self._index._sort_keys_by_io(sub_keys, positions)
 
1495
        absent_keys = keys - set(global_map)
 
1496
        for key in absent_keys:
 
1497
            yield AbsentContentFactory(key)
 
1498
        # restrict our view to the keys we can answer.
 
1499
        # XXX: Memory: TODO: batch data here to cap buffered data at (say) 1MB.
 
1500
        # XXX: At that point we need to consider the impact of double reads by
 
1501
        # utilising components multiple times.
 
1502
        if include_delta_closure:
 
1503
            # XXX: get_content_maps performs its own index queries; allow state
 
1504
            # to be passed in.
 
1505
            non_local_keys = needed_from_fallback - absent_keys
 
1506
            for keys, non_local_keys in self._group_keys_for_io(present_keys,
 
1507
                                                                non_local_keys,
 
1508
                                                                positions):
 
1509
                generator = _VFContentMapGenerator(self, keys, non_local_keys,
 
1510
                                                   global_map,
 
1511
                                                   ordering=ordering)
 
1512
                for record in generator.get_record_stream():
 
1513
                    yield record
 
1514
        else:
 
1515
            for source, keys in source_keys:
 
1516
                if source is parent_maps[0]:
 
1517
                    # this KnitVersionedFiles
 
1518
                    records = [(key, positions[key][1]) for key in keys]
 
1519
                    for key, raw_data in self._read_records_iter_unchecked(records):
 
1520
                        (record_details, index_memo, _) = positions[key]
 
1521
                        yield KnitContentFactory(key, global_map[key],
 
1522
                            record_details, None, raw_data, self._factory.annotated, None)
 
1523
                else:
 
1524
                    vf = self._immediate_fallback_vfs[parent_maps.index(source) - 1]
 
1525
                    for record in vf.get_record_stream(keys, ordering,
 
1526
                        include_delta_closure):
 
1527
                        yield record
 
1528
 
 
1529
    def get_sha1s(self, keys):
 
1530
        """See VersionedFiles.get_sha1s()."""
 
1531
        missing = set(keys)
 
1532
        record_map = self._get_record_map(missing, allow_missing=True)
 
1533
        result = {}
 
1534
        for key, details in record_map.iteritems():
 
1535
            if key not in missing:
 
1536
                continue
 
1537
            # record entry 2 is the 'digest'.
 
1538
            result[key] = details[2]
 
1539
        missing.difference_update(set(result))
 
1540
        for source in self._immediate_fallback_vfs:
 
1541
            if not missing:
 
1542
                break
 
1543
            new_result = source.get_sha1s(missing)
 
1544
            result.update(new_result)
 
1545
            missing.difference_update(set(new_result))
 
1546
        return result
 
1547
 
 
1548
    def insert_record_stream(self, stream):
 
1549
        """Insert a record stream into this container.
 
1550
 
 
1551
        :param stream: A stream of records to insert.
 
1552
        :return: None
 
1553
        :seealso VersionedFiles.get_record_stream:
 
1554
        """
 
1555
        def get_adapter(adapter_key):
 
1556
            try:
 
1557
                return adapters[adapter_key]
 
1558
            except KeyError:
 
1559
                adapter_factory = adapter_registry.get(adapter_key)
 
1560
                adapter = adapter_factory(self)
 
1561
                adapters[adapter_key] = adapter
 
1562
                return adapter
 
1563
        delta_types = set()
 
1564
        if self._factory.annotated:
 
1565
            # self is annotated, we need annotated knits to use directly.
 
1566
            annotated = "annotated-"
 
1567
            convertibles = []
 
1568
        else:
 
1569
            # self is not annotated, but we can strip annotations cheaply.
 
1570
            annotated = ""
 
1571
            convertibles = set(["knit-annotated-ft-gz"])
 
1572
            if self._max_delta_chain:
 
1573
                delta_types.add("knit-annotated-delta-gz")
 
1574
                convertibles.add("knit-annotated-delta-gz")
 
1575
        # The set of types we can cheaply adapt without needing basis texts.
 
1576
        native_types = set()
 
1577
        if self._max_delta_chain:
 
1578
            native_types.add("knit-%sdelta-gz" % annotated)
 
1579
            delta_types.add("knit-%sdelta-gz" % annotated)
 
1580
        native_types.add("knit-%sft-gz" % annotated)
 
1581
        knit_types = native_types.union(convertibles)
 
1582
        adapters = {}
 
1583
        # Buffer all index entries that we can't add immediately because their
 
1584
        # basis parent is missing. We don't buffer all because generating
 
1585
        # annotations may require access to some of the new records. However we
 
1586
        # can't generate annotations from new deltas until their basis parent
 
1587
        # is present anyway, so we get away with not needing an index that
 
1588
        # includes the new keys.
 
1589
        #
 
1590
        # See <http://launchpad.net/bugs/300177> about ordering of compression
 
1591
        # parents in the records - to be conservative, we insist that all
 
1592
        # parents must be present to avoid expanding to a fulltext.
 
1593
        #
 
1594
        # key = basis_parent, value = index entry to add
 
1595
        buffered_index_entries = {}
 
1596
        for record in stream:
 
1597
            kind = record.storage_kind
 
1598
            if kind.startswith('knit-') and kind.endswith('-gz'):
 
1599
                # Check that the ID in the header of the raw knit bytes matches
 
1600
                # the record metadata.
 
1601
                raw_data = record._raw_record
 
1602
                df, rec = self._parse_record_header(record.key, raw_data)
 
1603
                df.close()
 
1604
            buffered = False
 
1605
            parents = record.parents
 
1606
            if record.storage_kind in delta_types:
 
1607
                # TODO: eventually the record itself should track
 
1608
                #       compression_parent
 
1609
                compression_parent = parents[0]
 
1610
            else:
 
1611
                compression_parent = None
 
1612
            # Raise an error when a record is missing.
 
1613
            if record.storage_kind == 'absent':
 
1614
                raise RevisionNotPresent([record.key], self)
 
1615
            elif ((record.storage_kind in knit_types)
 
1616
                  and (compression_parent is None
 
1617
                       or not self._immediate_fallback_vfs
 
1618
                       or self._index.has_key(compression_parent)
 
1619
                       or not self.has_key(compression_parent))):
 
1620
                # we can insert the knit record literally if either it has no
 
1621
                # compression parent OR we already have its basis in this kvf
 
1622
                # OR the basis is not present even in the fallbacks.  In the
 
1623
                # last case it will either turn up later in the stream and all
 
1624
                # will be well, or it won't turn up at all and we'll raise an
 
1625
                # error at the end.
 
1626
                #
 
1627
                # TODO: self.has_key is somewhat redundant with
 
1628
                # self._index.has_key; we really want something that directly
 
1629
                # asks if it's only present in the fallbacks. -- mbp 20081119
 
1630
                if record.storage_kind not in native_types:
 
1631
                    try:
 
1632
                        adapter_key = (record.storage_kind, "knit-delta-gz")
 
1633
                        adapter = get_adapter(adapter_key)
 
1634
                    except KeyError:
 
1635
                        adapter_key = (record.storage_kind, "knit-ft-gz")
 
1636
                        adapter = get_adapter(adapter_key)
 
1637
                    bytes = adapter.get_bytes(record)
 
1638
                else:
 
1639
                    # It's a knit record, it has a _raw_record field (even if
 
1640
                    # it was reconstituted from a network stream).
 
1641
                    bytes = record._raw_record
 
1642
                options = [record._build_details[0]]
 
1643
                if record._build_details[1]:
 
1644
                    options.append('no-eol')
 
1645
                # Just blat it across.
 
1646
                # Note: This does end up adding data on duplicate keys. As
 
1647
                # modern repositories use atomic insertions this should not
 
1648
                # lead to excessive growth in the event of interrupted fetches.
 
1649
                # 'knit' repositories may suffer excessive growth, but as a
 
1650
                # deprecated format this is tolerable. It can be fixed if
 
1651
                # needed by in the kndx index support raising on a duplicate
 
1652
                # add with identical parents and options.
 
1653
                access_memo = self._access.add_raw_records(
 
1654
                    [(record.key, len(bytes))], bytes)[0]
 
1655
                index_entry = (record.key, options, access_memo, parents)
 
1656
                if 'fulltext' not in options:
 
1657
                    # Not a fulltext, so we need to make sure the compression
 
1658
                    # parent will also be present.
 
1659
                    # Note that pack backed knits don't need to buffer here
 
1660
                    # because they buffer all writes to the transaction level,
 
1661
                    # but we don't expose that difference at the index level. If
 
1662
                    # the query here has sufficient cost to show up in
 
1663
                    # profiling we should do that.
 
1664
                    #
 
1665
                    # They're required to be physically in this
 
1666
                    # KnitVersionedFiles, not in a fallback.
 
1667
                    if not self._index.has_key(compression_parent):
 
1668
                        pending = buffered_index_entries.setdefault(
 
1669
                            compression_parent, [])
 
1670
                        pending.append(index_entry)
 
1671
                        buffered = True
 
1672
                if not buffered:
 
1673
                    self._index.add_records([index_entry])
 
1674
            elif record.storage_kind == 'chunked':
 
1675
                self.add_lines(record.key, parents,
 
1676
                    osutils.chunks_to_lines(record.get_bytes_as('chunked')))
 
1677
            else:
 
1678
                # Not suitable for direct insertion as a
 
1679
                # delta, either because it's not the right format, or this
 
1680
                # KnitVersionedFiles doesn't permit deltas (_max_delta_chain ==
 
1681
                # 0) or because it depends on a base only present in the
 
1682
                # fallback kvfs.
 
1683
                self._access.flush()
 
1684
                try:
 
1685
                    # Try getting a fulltext directly from the record.
 
1686
                    bytes = record.get_bytes_as('fulltext')
 
1687
                except errors.UnavailableRepresentation:
 
1688
                    adapter_key = record.storage_kind, 'fulltext'
 
1689
                    adapter = get_adapter(adapter_key)
 
1690
                    bytes = adapter.get_bytes(record)
 
1691
                lines = split_lines(bytes)
 
1692
                try:
 
1693
                    self.add_lines(record.key, parents, lines)
 
1694
                except errors.RevisionAlreadyPresent:
 
1695
                    pass
 
1696
            # Add any records whose basis parent is now available.
 
1697
            if not buffered:
 
1698
                added_keys = [record.key]
 
1699
                while added_keys:
 
1700
                    key = added_keys.pop(0)
 
1701
                    if key in buffered_index_entries:
 
1702
                        index_entries = buffered_index_entries[key]
 
1703
                        self._index.add_records(index_entries)
 
1704
                        added_keys.extend(
 
1705
                            [index_entry[0] for index_entry in index_entries])
 
1706
                        del buffered_index_entries[key]
 
1707
        if buffered_index_entries:
 
1708
            # There were index entries buffered at the end of the stream,
 
1709
            # So these need to be added (if the index supports holding such
 
1710
            # entries for later insertion)
 
1711
            all_entries = []
 
1712
            for key in buffered_index_entries:
 
1713
                index_entries = buffered_index_entries[key]
 
1714
                all_entries.extend(index_entries)
 
1715
            self._index.add_records(
 
1716
                all_entries, missing_compression_parents=True)
 
1717
 
 
1718
    def get_missing_compression_parent_keys(self):
 
1719
        """Return an iterable of keys of missing compression parents.
 
1720
 
 
1721
        Check this after calling insert_record_stream to find out if there are
 
1722
        any missing compression parents.  If there are, the records that
 
1723
        depend on them are not able to be inserted safely. For atomic
 
1724
        KnitVersionedFiles built on packs, the transaction should be aborted or
 
1725
        suspended - commit will fail at this point. Nonatomic knits will error
 
1726
        earlier because they have no staging area to put pending entries into.
 
1727
        """
 
1728
        return self._index.get_missing_compression_parents()
 
1729
 
 
1730
    def iter_lines_added_or_present_in_keys(self, keys, pb=None):
 
1731
        """Iterate over the lines in the versioned files from keys.
 
1732
 
 
1733
        This may return lines from other keys. Each item the returned
 
1734
        iterator yields is a tuple of a line and a text version that that line
 
1735
        is present in (not introduced in).
 
1736
 
 
1737
        Ordering of results is in whatever order is most suitable for the
 
1738
        underlying storage format.
 
1739
 
 
1740
        If a progress bar is supplied, it may be used to indicate progress.
 
1741
        The caller is responsible for cleaning up progress bars (because this
 
1742
        is an iterator).
 
1743
 
 
1744
        NOTES:
 
1745
         * Lines are normalised by the underlying store: they will all have \\n
 
1746
           terminators.
 
1747
         * Lines are returned in arbitrary order.
 
1748
         * If a requested key did not change any lines (or didn't have any
 
1749
           lines), it may not be mentioned at all in the result.
 
1750
 
 
1751
        :param pb: Progress bar supplied by caller.
 
1752
        :return: An iterator over (line, key).
 
1753
        """
 
1754
        if pb is None:
 
1755
            pb = ui.ui_factory.nested_progress_bar()
 
1756
        keys = set(keys)
 
1757
        total = len(keys)
 
1758
        done = False
 
1759
        while not done:
 
1760
            try:
 
1761
                # we don't care about inclusions, the caller cares.
 
1762
                # but we need to setup a list of records to visit.
 
1763
                # we need key, position, length
 
1764
                key_records = []
 
1765
                build_details = self._index.get_build_details(keys)
 
1766
                for key, details in build_details.iteritems():
 
1767
                    if key in keys:
 
1768
                        key_records.append((key, details[0]))
 
1769
                records_iter = enumerate(self._read_records_iter(key_records))
 
1770
                for (key_idx, (key, data, sha_value)) in records_iter:
 
1771
                    pb.update('Walking content', key_idx, total)
 
1772
                    compression_parent = build_details[key][1]
 
1773
                    if compression_parent is None:
 
1774
                        # fulltext
 
1775
                        line_iterator = self._factory.get_fulltext_content(data)
 
1776
                    else:
 
1777
                        # Delta
 
1778
                        line_iterator = self._factory.get_linedelta_content(data)
 
1779
                    # Now that we are yielding the data for this key, remove it
 
1780
                    # from the list
 
1781
                    keys.remove(key)
 
1782
                    # XXX: It might be more efficient to yield (key,
 
1783
                    # line_iterator) in the future. However for now, this is a
 
1784
                    # simpler change to integrate into the rest of the
 
1785
                    # codebase. RBC 20071110
 
1786
                    for line in line_iterator:
 
1787
                        yield line, key
 
1788
                done = True
 
1789
            except errors.RetryWithNewPacks, e:
 
1790
                self._access.reload_or_raise(e)
 
1791
        # If there are still keys we've not yet found, we look in the fallback
 
1792
        # vfs, and hope to find them there.  Note that if the keys are found
 
1793
        # but had no changes or no content, the fallback may not return
 
1794
        # anything.
 
1795
        if keys and not self._immediate_fallback_vfs:
 
1796
            # XXX: strictly the second parameter is meant to be the file id
 
1797
            # but it's not easily accessible here.
 
1798
            raise RevisionNotPresent(keys, repr(self))
 
1799
        for source in self._immediate_fallback_vfs:
 
1800
            if not keys:
 
1801
                break
 
1802
            source_keys = set()
 
1803
            for line, key in source.iter_lines_added_or_present_in_keys(keys):
 
1804
                source_keys.add(key)
 
1805
                yield line, key
 
1806
            keys.difference_update(source_keys)
 
1807
        pb.update('Walking content', total, total)
 
1808
 
 
1809
    def _make_line_delta(self, delta_seq, new_content):
 
1810
        """Generate a line delta from delta_seq and new_content."""
 
1811
        diff_hunks = []
 
1812
        for op in delta_seq.get_opcodes():
 
1813
            if op[0] == 'equal':
 
1814
                continue
 
1815
            diff_hunks.append((op[1], op[2], op[4]-op[3], new_content._lines[op[3]:op[4]]))
 
1816
        return diff_hunks
 
1817
 
 
1818
    def _merge_annotations(self, content, parents, parent_texts={},
 
1819
                           delta=None, annotated=None,
 
1820
                           left_matching_blocks=None):
 
1821
        """Merge annotations for content and generate deltas.
 
1822
 
 
1823
        This is done by comparing the annotations based on changes to the text
 
1824
        and generating a delta on the resulting full texts. If annotations are
 
1825
        not being created then a simple delta is created.
 
1826
        """
 
1827
        if left_matching_blocks is not None:
 
1828
            delta_seq = diff._PrematchedMatcher(left_matching_blocks)
 
1829
        else:
 
1830
            delta_seq = None
 
1831
        if annotated:
 
1832
            for parent_key in parents:
 
1833
                merge_content = self._get_content(parent_key, parent_texts)
 
1834
                if (parent_key == parents[0] and delta_seq is not None):
 
1835
                    seq = delta_seq
 
1836
                else:
 
1837
                    seq = patiencediff.PatienceSequenceMatcher(
 
1838
                        None, merge_content.text(), content.text())
 
1839
                for i, j, n in seq.get_matching_blocks():
 
1840
                    if n == 0:
 
1841
                        continue
 
1842
                    # this copies (origin, text) pairs across to the new
 
1843
                    # content for any line that matches the last-checked
 
1844
                    # parent.
 
1845
                    content._lines[j:j+n] = merge_content._lines[i:i+n]
 
1846
            # XXX: Robert says the following block is a workaround for a
 
1847
            # now-fixed bug and it can probably be deleted. -- mbp 20080618
 
1848
            if content._lines and content._lines[-1][1][-1] != '\n':
 
1849
                # The copied annotation was from a line without a trailing EOL,
 
1850
                # reinstate one for the content object, to ensure correct
 
1851
                # serialization.
 
1852
                line = content._lines[-1][1] + '\n'
 
1853
                content._lines[-1] = (content._lines[-1][0], line)
 
1854
        if delta:
 
1855
            if delta_seq is None:
 
1856
                reference_content = self._get_content(parents[0], parent_texts)
 
1857
                new_texts = content.text()
 
1858
                old_texts = reference_content.text()
 
1859
                delta_seq = patiencediff.PatienceSequenceMatcher(
 
1860
                                                 None, old_texts, new_texts)
 
1861
            return self._make_line_delta(delta_seq, content)
 
1862
 
 
1863
    def _parse_record(self, version_id, data):
 
1864
        """Parse an original format knit record.
 
1865
 
 
1866
        These have the last element of the key only present in the stored data.
 
1867
        """
 
1868
        rec, record_contents = self._parse_record_unchecked(data)
 
1869
        self._check_header_version(rec, version_id)
 
1870
        return record_contents, rec[3]
 
1871
 
 
1872
    def _parse_record_header(self, key, raw_data):
 
1873
        """Parse a record header for consistency.
 
1874
 
 
1875
        :return: the header and the decompressor stream.
 
1876
                 as (stream, header_record)
 
1877
        """
 
1878
        df = gzip.GzipFile(mode='rb', fileobj=StringIO(raw_data))
 
1879
        try:
 
1880
            # Current serialise
 
1881
            rec = self._check_header(key, df.readline())
 
1882
        except Exception, e:
 
1883
            raise KnitCorrupt(self,
 
1884
                              "While reading {%s} got %s(%s)"
 
1885
                              % (key, e.__class__.__name__, str(e)))
 
1886
        return df, rec
 
1887
 
 
1888
    def _parse_record_unchecked(self, data):
 
1889
        # profiling notes:
 
1890
        # 4168 calls in 2880 217 internal
 
1891
        # 4168 calls to _parse_record_header in 2121
 
1892
        # 4168 calls to readlines in 330
 
1893
        df = gzip.GzipFile(mode='rb', fileobj=StringIO(data))
 
1894
        try:
 
1895
            record_contents = df.readlines()
 
1896
        except Exception, e:
 
1897
            raise KnitCorrupt(self, "Corrupt compressed record %r, got %s(%s)" %
 
1898
                (data, e.__class__.__name__, str(e)))
 
1899
        header = record_contents.pop(0)
 
1900
        rec = self._split_header(header)
 
1901
        last_line = record_contents.pop()
 
1902
        if len(record_contents) != int(rec[2]):
 
1903
            raise KnitCorrupt(self,
 
1904
                              'incorrect number of lines %s != %s'
 
1905
                              ' for version {%s} %s'
 
1906
                              % (len(record_contents), int(rec[2]),
 
1907
                                 rec[1], record_contents))
 
1908
        if last_line != 'end %s\n' % rec[1]:
 
1909
            raise KnitCorrupt(self,
 
1910
                              'unexpected version end line %r, wanted %r'
 
1911
                              % (last_line, rec[1]))
 
1912
        df.close()
 
1913
        return rec, record_contents
 
1914
 
 
1915
    def _read_records_iter(self, records):
 
1916
        """Read text records from data file and yield result.
 
1917
 
 
1918
        The result will be returned in whatever is the fastest to read.
 
1919
        Not by the order requested. Also, multiple requests for the same
 
1920
        record will only yield 1 response.
 
1921
        :param records: A list of (key, access_memo) entries
 
1922
        :return: Yields (key, contents, digest) in the order
 
1923
                 read, not the order requested
 
1924
        """
 
1925
        if not records:
 
1926
            return
 
1927
 
 
1928
        # XXX: This smells wrong, IO may not be getting ordered right.
 
1929
        needed_records = sorted(set(records), key=operator.itemgetter(1))
 
1930
        if not needed_records:
 
1931
            return
 
1932
 
 
1933
        # The transport optimizes the fetching as well
 
1934
        # (ie, reads continuous ranges.)
 
1935
        raw_data = self._access.get_raw_records(
 
1936
            [index_memo for key, index_memo in needed_records])
 
1937
 
 
1938
        for (key, index_memo), data in \
 
1939
                izip(iter(needed_records), raw_data):
 
1940
            content, digest = self._parse_record(key[-1], data)
 
1941
            yield key, content, digest
 
1942
 
 
1943
    def _read_records_iter_raw(self, records):
 
1944
        """Read text records from data file and yield raw data.
 
1945
 
 
1946
        This unpacks enough of the text record to validate the id is
 
1947
        as expected but thats all.
 
1948
 
 
1949
        Each item the iterator yields is (key, bytes,
 
1950
            expected_sha1_of_full_text).
 
1951
        """
 
1952
        for key, data in self._read_records_iter_unchecked(records):
 
1953
            # validate the header (note that we can only use the suffix in
 
1954
            # current knit records).
 
1955
            df, rec = self._parse_record_header(key, data)
 
1956
            df.close()
 
1957
            yield key, data, rec[3]
 
1958
 
 
1959
    def _read_records_iter_unchecked(self, records):
 
1960
        """Read text records from data file and yield raw data.
 
1961
 
 
1962
        No validation is done.
 
1963
 
 
1964
        Yields tuples of (key, data).
 
1965
        """
 
1966
        # setup an iterator of the external records:
 
1967
        # uses readv so nice and fast we hope.
 
1968
        if len(records):
 
1969
            # grab the disk data needed.
 
1970
            needed_offsets = [index_memo for key, index_memo
 
1971
                                           in records]
 
1972
            raw_records = self._access.get_raw_records(needed_offsets)
 
1973
 
 
1974
        for key, index_memo in records:
 
1975
            data = raw_records.next()
 
1976
            yield key, data
 
1977
 
 
1978
    def _record_to_data(self, key, digest, lines, dense_lines=None):
 
1979
        """Convert key, digest, lines into a raw data block.
 
1980
 
 
1981
        :param key: The key of the record. Currently keys are always serialised
 
1982
            using just the trailing component.
 
1983
        :param dense_lines: The bytes of lines but in a denser form. For
 
1984
            instance, if lines is a list of 1000 bytestrings each ending in \n,
 
1985
            dense_lines may be a list with one line in it, containing all the
 
1986
            1000's lines and their \n's. Using dense_lines if it is already
 
1987
            known is a win because the string join to create bytes in this
 
1988
            function spends less time resizing the final string.
 
1989
        :return: (len, a StringIO instance with the raw data ready to read.)
 
1990
        """
 
1991
        chunks = ["version %s %d %s\n" % (key[-1], len(lines), digest)]
 
1992
        chunks.extend(dense_lines or lines)
 
1993
        chunks.append("end %s\n" % key[-1])
 
1994
        for chunk in chunks:
 
1995
            if type(chunk) is not str:
 
1996
                raise AssertionError(
 
1997
                    'data must be plain bytes was %s' % type(chunk))
 
1998
        if lines and lines[-1][-1] != '\n':
 
1999
            raise ValueError('corrupt lines value %r' % lines)
 
2000
        compressed_bytes = tuned_gzip.chunks_to_gzip(chunks)
 
2001
        return len(compressed_bytes), compressed_bytes
 
2002
 
 
2003
    def _split_header(self, line):
 
2004
        rec = line.split()
 
2005
        if len(rec) != 4:
 
2006
            raise KnitCorrupt(self,
 
2007
                              'unexpected number of elements in record header')
 
2008
        return rec
 
2009
 
 
2010
    def keys(self):
 
2011
        """See VersionedFiles.keys."""
 
2012
        if 'evil' in debug.debug_flags:
 
2013
            trace.mutter_callsite(2, "keys scales with size of history")
 
2014
        sources = [self._index] + self._immediate_fallback_vfs
 
2015
        result = set()
 
2016
        for source in sources:
 
2017
            result.update(source.keys())
 
2018
        return result
 
2019
 
 
2020
 
 
2021
class _ContentMapGenerator(object):
 
2022
    """Generate texts or expose raw deltas for a set of texts."""
 
2023
 
 
2024
    def __init__(self, ordering='unordered'):
 
2025
        self._ordering = ordering
 
2026
 
 
2027
    def _get_content(self, key):
 
2028
        """Get the content object for key."""
 
2029
        # Note that _get_content is only called when the _ContentMapGenerator
 
2030
        # has been constructed with just one key requested for reconstruction.
 
2031
        if key in self.nonlocal_keys:
 
2032
            record = self.get_record_stream().next()
 
2033
            # Create a content object on the fly
 
2034
            lines = osutils.chunks_to_lines(record.get_bytes_as('chunked'))
 
2035
            return PlainKnitContent(lines, record.key)
 
2036
        else:
 
2037
            # local keys we can ask for directly
 
2038
            return self._get_one_work(key)
 
2039
 
 
2040
    def get_record_stream(self):
 
2041
        """Get a record stream for the keys requested during __init__."""
 
2042
        for record in self._work():
 
2043
            yield record
 
2044
 
 
2045
    def _work(self):
 
2046
        """Produce maps of text and KnitContents as dicts.
 
2047
 
 
2048
        :return: (text_map, content_map) where text_map contains the texts for
 
2049
            the requested versions and content_map contains the KnitContents.
 
2050
        """
 
2051
        # NB: By definition we never need to read remote sources unless texts
 
2052
        # are requested from them: we don't delta across stores - and we
 
2053
        # explicitly do not want to to prevent data loss situations.
 
2054
        if self.global_map is None:
 
2055
            self.global_map = self.vf.get_parent_map(self.keys)
 
2056
        nonlocal_keys = self.nonlocal_keys
 
2057
 
 
2058
        missing_keys = set(nonlocal_keys)
 
2059
        # Read from remote versioned file instances and provide to our caller.
 
2060
        for source in self.vf._immediate_fallback_vfs:
 
2061
            if not missing_keys:
 
2062
                break
 
2063
            # Loop over fallback repositories asking them for texts - ignore
 
2064
            # any missing from a particular fallback.
 
2065
            for record in source.get_record_stream(missing_keys,
 
2066
                self._ordering, True):
 
2067
                if record.storage_kind == 'absent':
 
2068
                    # Not in thie particular stream, may be in one of the
 
2069
                    # other fallback vfs objects.
 
2070
                    continue
 
2071
                missing_keys.remove(record.key)
 
2072
                yield record
 
2073
 
 
2074
        if self._raw_record_map is None:
 
2075
            raise AssertionError('_raw_record_map should have been filled')
 
2076
        first = True
 
2077
        for key in self.keys:
 
2078
            if key in self.nonlocal_keys:
 
2079
                continue
 
2080
            yield LazyKnitContentFactory(key, self.global_map[key], self, first)
 
2081
            first = False
 
2082
 
 
2083
    def _get_one_work(self, requested_key):
 
2084
        # Now, if we have calculated everything already, just return the
 
2085
        # desired text.
 
2086
        if requested_key in self._contents_map:
 
2087
            return self._contents_map[requested_key]
 
2088
        # To simplify things, parse everything at once - code that wants one text
 
2089
        # probably wants them all.
 
2090
        # FUTURE: This function could be improved for the 'extract many' case
 
2091
        # by tracking each component and only doing the copy when the number of
 
2092
        # children than need to apply delta's to it is > 1 or it is part of the
 
2093
        # final output.
 
2094
        multiple_versions = len(self.keys) != 1
 
2095
        if self._record_map is None:
 
2096
            self._record_map = self.vf._raw_map_to_record_map(
 
2097
                self._raw_record_map)
 
2098
        record_map = self._record_map
 
2099
        # raw_record_map is key:
 
2100
        # Have read and parsed records at this point.
 
2101
        for key in self.keys:
 
2102
            if key in self.nonlocal_keys:
 
2103
                # already handled
 
2104
                continue
 
2105
            components = []
 
2106
            cursor = key
 
2107
            while cursor is not None:
 
2108
                try:
 
2109
                    record, record_details, digest, next = record_map[cursor]
 
2110
                except KeyError:
 
2111
                    raise RevisionNotPresent(cursor, self)
 
2112
                components.append((cursor, record, record_details, digest))
 
2113
                cursor = next
 
2114
                if cursor in self._contents_map:
 
2115
                    # no need to plan further back
 
2116
                    components.append((cursor, None, None, None))
 
2117
                    break
 
2118
 
 
2119
            content = None
 
2120
            for (component_id, record, record_details,
 
2121
                 digest) in reversed(components):
 
2122
                if component_id in self._contents_map:
 
2123
                    content = self._contents_map[component_id]
 
2124
                else:
 
2125
                    content, delta = self._factory.parse_record(key[-1],
 
2126
                        record, record_details, content,
 
2127
                        copy_base_content=multiple_versions)
 
2128
                    if multiple_versions:
 
2129
                        self._contents_map[component_id] = content
 
2130
 
 
2131
            # digest here is the digest from the last applied component.
 
2132
            text = content.text()
 
2133
            actual_sha = sha_strings(text)
 
2134
            if actual_sha != digest:
 
2135
                raise SHA1KnitCorrupt(self, actual_sha, digest, key, text)
 
2136
        if multiple_versions:
 
2137
            return self._contents_map[requested_key]
 
2138
        else:
 
2139
            return content
 
2140
 
 
2141
    def _wire_bytes(self):
 
2142
        """Get the bytes to put on the wire for 'key'.
 
2143
 
 
2144
        The first collection of bytes asked for returns the serialised
 
2145
        raw_record_map and the additional details (key, parent) for key.
 
2146
        Subsequent calls return just the additional details (key, parent).
 
2147
        The wire storage_kind given for the first key is 'knit-delta-closure',
 
2148
        For subsequent keys it is 'knit-delta-closure-ref'.
 
2149
 
 
2150
        :param key: A key from the content generator.
 
2151
        :return: Bytes to put on the wire.
 
2152
        """
 
2153
        lines = []
 
2154
        # kind marker for dispatch on the far side,
 
2155
        lines.append('knit-delta-closure')
 
2156
        # Annotated or not
 
2157
        if self.vf._factory.annotated:
 
2158
            lines.append('annotated')
 
2159
        else:
 
2160
            lines.append('')
 
2161
        # then the list of keys
 
2162
        lines.append('\t'.join(['\x00'.join(key) for key in self.keys
 
2163
            if key not in self.nonlocal_keys]))
 
2164
        # then the _raw_record_map in serialised form:
 
2165
        map_byte_list = []
 
2166
        # for each item in the map:
 
2167
        # 1 line with key
 
2168
        # 1 line with parents if the key is to be yielded (None: for None, '' for ())
 
2169
        # one line with method
 
2170
        # one line with noeol
 
2171
        # one line with next ('' for None)
 
2172
        # one line with byte count of the record bytes
 
2173
        # the record bytes
 
2174
        for key, (record_bytes, (method, noeol), next) in \
 
2175
            self._raw_record_map.iteritems():
 
2176
            key_bytes = '\x00'.join(key)
 
2177
            parents = self.global_map.get(key, None)
 
2178
            if parents is None:
 
2179
                parent_bytes = 'None:'
 
2180
            else:
 
2181
                parent_bytes = '\t'.join('\x00'.join(key) for key in parents)
 
2182
            method_bytes = method
 
2183
            if noeol:
 
2184
                noeol_bytes = "T"
 
2185
            else:
 
2186
                noeol_bytes = "F"
 
2187
            if next:
 
2188
                next_bytes = '\x00'.join(next)
 
2189
            else:
 
2190
                next_bytes = ''
 
2191
            map_byte_list.append('%s\n%s\n%s\n%s\n%s\n%d\n%s' % (
 
2192
                key_bytes, parent_bytes, method_bytes, noeol_bytes, next_bytes,
 
2193
                len(record_bytes), record_bytes))
 
2194
        map_bytes = ''.join(map_byte_list)
 
2195
        lines.append(map_bytes)
 
2196
        bytes = '\n'.join(lines)
 
2197
        return bytes
 
2198
 
 
2199
 
 
2200
class _VFContentMapGenerator(_ContentMapGenerator):
 
2201
    """Content map generator reading from a VersionedFiles object."""
 
2202
 
 
2203
    def __init__(self, versioned_files, keys, nonlocal_keys=None,
 
2204
        global_map=None, raw_record_map=None, ordering='unordered'):
 
2205
        """Create a _ContentMapGenerator.
 
2206
 
 
2207
        :param versioned_files: The versioned files that the texts are being
 
2208
            extracted from.
 
2209
        :param keys: The keys to produce content maps for.
 
2210
        :param nonlocal_keys: An iterable of keys(possibly intersecting keys)
 
2211
            which are known to not be in this knit, but rather in one of the
 
2212
            fallback knits.
 
2213
        :param global_map: The result of get_parent_map(keys) (or a supermap).
 
2214
            This is required if get_record_stream() is to be used.
 
2215
        :param raw_record_map: A unparsed raw record map to use for answering
 
2216
            contents.
 
2217
        """
 
2218
        _ContentMapGenerator.__init__(self, ordering=ordering)
 
2219
        # The vf to source data from
 
2220
        self.vf = versioned_files
 
2221
        # The keys desired
 
2222
        self.keys = list(keys)
 
2223
        # Keys known to be in fallback vfs objects
 
2224
        if nonlocal_keys is None:
 
2225
            self.nonlocal_keys = set()
 
2226
        else:
 
2227
            self.nonlocal_keys = frozenset(nonlocal_keys)
 
2228
        # Parents data for keys to be returned in get_record_stream
 
2229
        self.global_map = global_map
 
2230
        # The chunked lists for self.keys in text form
 
2231
        self._text_map = {}
 
2232
        # A cache of KnitContent objects used in extracting texts.
 
2233
        self._contents_map = {}
 
2234
        # All the knit records needed to assemble the requested keys as full
 
2235
        # texts.
 
2236
        self._record_map = None
 
2237
        if raw_record_map is None:
 
2238
            self._raw_record_map = self.vf._get_record_map_unparsed(keys,
 
2239
                allow_missing=True)
 
2240
        else:
 
2241
            self._raw_record_map = raw_record_map
 
2242
        # the factory for parsing records
 
2243
        self._factory = self.vf._factory
 
2244
 
 
2245
 
 
2246
class _NetworkContentMapGenerator(_ContentMapGenerator):
 
2247
    """Content map generator sourced from a network stream."""
 
2248
 
 
2249
    def __init__(self, bytes, line_end):
 
2250
        """Construct a _NetworkContentMapGenerator from a bytes block."""
 
2251
        self._bytes = bytes
 
2252
        self.global_map = {}
 
2253
        self._raw_record_map = {}
 
2254
        self._contents_map = {}
 
2255
        self._record_map = None
 
2256
        self.nonlocal_keys = []
 
2257
        # Get access to record parsing facilities
 
2258
        self.vf = KnitVersionedFiles(None, None)
 
2259
        start = line_end
 
2260
        # Annotated or not
 
2261
        line_end = bytes.find('\n', start)
 
2262
        line = bytes[start:line_end]
 
2263
        start = line_end + 1
 
2264
        if line == 'annotated':
 
2265
            self._factory = KnitAnnotateFactory()
 
2266
        else:
 
2267
            self._factory = KnitPlainFactory()
 
2268
        # list of keys to emit in get_record_stream
 
2269
        line_end = bytes.find('\n', start)
 
2270
        line = bytes[start:line_end]
 
2271
        start = line_end + 1
 
2272
        self.keys = [
 
2273
            tuple(segment.split('\x00')) for segment in line.split('\t')
 
2274
            if segment]
 
2275
        # now a loop until the end. XXX: It would be nice if this was just a
 
2276
        # bunch of the same records as get_record_stream(..., False) gives, but
 
2277
        # there is a decent sized gap stopping that at the moment.
 
2278
        end = len(bytes)
 
2279
        while start < end:
 
2280
            # 1 line with key
 
2281
            line_end = bytes.find('\n', start)
 
2282
            key = tuple(bytes[start:line_end].split('\x00'))
 
2283
            start = line_end + 1
 
2284
            # 1 line with parents (None: for None, '' for ())
 
2285
            line_end = bytes.find('\n', start)
 
2286
            line = bytes[start:line_end]
 
2287
            if line == 'None:':
 
2288
                parents = None
 
2289
            else:
 
2290
                parents = tuple(
 
2291
                    [tuple(segment.split('\x00')) for segment in line.split('\t')
 
2292
                     if segment])
 
2293
            self.global_map[key] = parents
 
2294
            start = line_end + 1
 
2295
            # one line with method
 
2296
            line_end = bytes.find('\n', start)
 
2297
            line = bytes[start:line_end]
 
2298
            method = line
 
2299
            start = line_end + 1
 
2300
            # one line with noeol
 
2301
            line_end = bytes.find('\n', start)
 
2302
            line = bytes[start:line_end]
 
2303
            noeol = line == "T"
 
2304
            start = line_end + 1
 
2305
            # one line with next ('' for None)
 
2306
            line_end = bytes.find('\n', start)
 
2307
            line = bytes[start:line_end]
 
2308
            if not line:
 
2309
                next = None
 
2310
            else:
 
2311
                next = tuple(bytes[start:line_end].split('\x00'))
 
2312
            start = line_end + 1
 
2313
            # one line with byte count of the record bytes
 
2314
            line_end = bytes.find('\n', start)
 
2315
            line = bytes[start:line_end]
 
2316
            count = int(line)
 
2317
            start = line_end + 1
 
2318
            # the record bytes
 
2319
            record_bytes = bytes[start:start+count]
 
2320
            start = start + count
 
2321
            # put it in the map
 
2322
            self._raw_record_map[key] = (record_bytes, (method, noeol), next)
 
2323
 
 
2324
    def get_record_stream(self):
 
2325
        """Get a record stream for for keys requested by the bytestream."""
 
2326
        first = True
 
2327
        for key in self.keys:
 
2328
            yield LazyKnitContentFactory(key, self.global_map[key], self, first)
 
2329
            first = False
 
2330
 
 
2331
    def _wire_bytes(self):
 
2332
        return self._bytes
 
2333
 
 
2334
 
 
2335
class _KndxIndex(object):
 
2336
    """Manages knit index files
 
2337
 
 
2338
    The index is kept in memory and read on startup, to enable
 
2339
    fast lookups of revision information.  The cursor of the index
 
2340
    file is always pointing to the end, making it easy to append
 
2341
    entries.
 
2342
 
 
2343
    _cache is a cache for fast mapping from version id to a Index
 
2344
    object.
 
2345
 
 
2346
    _history is a cache for fast mapping from indexes to version ids.
 
2347
 
 
2348
    The index data format is dictionary compressed when it comes to
 
2349
    parent references; a index entry may only have parents that with a
 
2350
    lover index number.  As a result, the index is topological sorted.
 
2351
 
 
2352
    Duplicate entries may be written to the index for a single version id
 
2353
    if this is done then the latter one completely replaces the former:
 
2354
    this allows updates to correct version and parent information.
 
2355
    Note that the two entries may share the delta, and that successive
 
2356
    annotations and references MUST point to the first entry.
 
2357
 
 
2358
    The index file on disc contains a header, followed by one line per knit
 
2359
    record. The same revision can be present in an index file more than once.
 
2360
    The first occurrence gets assigned a sequence number starting from 0.
 
2361
 
 
2362
    The format of a single line is
 
2363
    REVISION_ID FLAGS BYTE_OFFSET LENGTH( PARENT_ID|PARENT_SEQUENCE_ID)* :\n
 
2364
    REVISION_ID is a utf8-encoded revision id
 
2365
    FLAGS is a comma separated list of flags about the record. Values include
 
2366
        no-eol, line-delta, fulltext.
 
2367
    BYTE_OFFSET is the ascii representation of the byte offset in the data file
 
2368
        that the compressed data starts at.
 
2369
    LENGTH is the ascii representation of the length of the data file.
 
2370
    PARENT_ID a utf-8 revision id prefixed by a '.' that is a parent of
 
2371
        REVISION_ID.
 
2372
    PARENT_SEQUENCE_ID the ascii representation of the sequence number of a
 
2373
        revision id already in the knit that is a parent of REVISION_ID.
 
2374
    The ' :' marker is the end of record marker.
 
2375
 
 
2376
    partial writes:
 
2377
    when a write is interrupted to the index file, it will result in a line
 
2378
    that does not end in ' :'. If the ' :' is not present at the end of a line,
 
2379
    or at the end of the file, then the record that is missing it will be
 
2380
    ignored by the parser.
 
2381
 
 
2382
    When writing new records to the index file, the data is preceded by '\n'
 
2383
    to ensure that records always start on new lines even if the last write was
 
2384
    interrupted. As a result its normal for the last line in the index to be
 
2385
    missing a trailing newline. One can be added with no harmful effects.
 
2386
 
 
2387
    :ivar _kndx_cache: dict from prefix to the old state of KnitIndex objects,
 
2388
        where prefix is e.g. the (fileid,) for .texts instances or () for
 
2389
        constant-mapped things like .revisions, and the old state is
 
2390
        tuple(history_vector, cache_dict).  This is used to prevent having an
 
2391
        ABI change with the C extension that reads .kndx files.
 
2392
    """
 
2393
 
 
2394
    HEADER = "# bzr knit index 8\n"
 
2395
 
 
2396
    def __init__(self, transport, mapper, get_scope, allow_writes, is_locked):
 
2397
        """Create a _KndxIndex on transport using mapper."""
 
2398
        self._transport = transport
 
2399
        self._mapper = mapper
 
2400
        self._get_scope = get_scope
 
2401
        self._allow_writes = allow_writes
 
2402
        self._is_locked = is_locked
 
2403
        self._reset_cache()
 
2404
        self.has_graph = True
 
2405
 
 
2406
    def add_records(self, records, random_id=False, missing_compression_parents=False):
 
2407
        """Add multiple records to the index.
 
2408
 
 
2409
        :param records: a list of tuples:
 
2410
                         (key, options, access_memo, parents).
 
2411
        :param random_id: If True the ids being added were randomly generated
 
2412
            and no check for existence will be performed.
 
2413
        :param missing_compression_parents: If True the records being added are
 
2414
            only compressed against texts already in the index (or inside
 
2415
            records). If False the records all refer to unavailable texts (or
 
2416
            texts inside records) as compression parents.
 
2417
        """
 
2418
        if missing_compression_parents:
 
2419
            # It might be nice to get the edge of the records. But keys isn't
 
2420
            # _wrong_.
 
2421
            keys = sorted(record[0] for record in records)
 
2422
            raise errors.RevisionNotPresent(keys, self)
 
2423
        paths = {}
 
2424
        for record in records:
 
2425
            key = record[0]
 
2426
            prefix = key[:-1]
 
2427
            path = self._mapper.map(key) + '.kndx'
 
2428
            path_keys = paths.setdefault(path, (prefix, []))
 
2429
            path_keys[1].append(record)
 
2430
        for path in sorted(paths):
 
2431
            prefix, path_keys = paths[path]
 
2432
            self._load_prefixes([prefix])
 
2433
            lines = []
 
2434
            orig_history = self._kndx_cache[prefix][1][:]
 
2435
            orig_cache = self._kndx_cache[prefix][0].copy()
 
2436
 
 
2437
            try:
 
2438
                for key, options, (_, pos, size), parents in path_keys:
 
2439
                    if parents is None:
 
2440
                        # kndx indices cannot be parentless.
 
2441
                        parents = ()
 
2442
                    line = "\n%s %s %s %s %s :" % (
 
2443
                        key[-1], ','.join(options), pos, size,
 
2444
                        self._dictionary_compress(parents))
 
2445
                    if type(line) is not str:
 
2446
                        raise AssertionError(
 
2447
                            'data must be utf8 was %s' % type(line))
 
2448
                    lines.append(line)
 
2449
                    self._cache_key(key, options, pos, size, parents)
 
2450
                if len(orig_history):
 
2451
                    self._transport.append_bytes(path, ''.join(lines))
 
2452
                else:
 
2453
                    self._init_index(path, lines)
 
2454
            except:
 
2455
                # If any problems happen, restore the original values and re-raise
 
2456
                self._kndx_cache[prefix] = (orig_cache, orig_history)
 
2457
                raise
 
2458
 
 
2459
    def scan_unvalidated_index(self, graph_index):
 
2460
        """See _KnitGraphIndex.scan_unvalidated_index."""
 
2461
        # Because kndx files do not support atomic insertion via separate index
 
2462
        # files, they do not support this method.
 
2463
        raise NotImplementedError(self.scan_unvalidated_index)
 
2464
 
 
2465
    def get_missing_compression_parents(self):
 
2466
        """See _KnitGraphIndex.get_missing_compression_parents."""
 
2467
        # Because kndx files do not support atomic insertion via separate index
 
2468
        # files, they do not support this method.
 
2469
        raise NotImplementedError(self.get_missing_compression_parents)
 
2470
 
 
2471
    def _cache_key(self, key, options, pos, size, parent_keys):
 
2472
        """Cache a version record in the history array and index cache.
 
2473
 
 
2474
        This is inlined into _load_data for performance. KEEP IN SYNC.
 
2475
        (It saves 60ms, 25% of the __init__ overhead on local 4000 record
 
2476
         indexes).
 
2477
        """
 
2478
        prefix = key[:-1]
 
2479
        version_id = key[-1]
 
2480
        # last-element only for compatibilty with the C load_data.
 
2481
        parents = tuple(parent[-1] for parent in parent_keys)
 
2482
        for parent in parent_keys:
 
2483
            if parent[:-1] != prefix:
 
2484
                raise ValueError("mismatched prefixes for %r, %r" % (
 
2485
                    key, parent_keys))
 
2486
        cache, history = self._kndx_cache[prefix]
 
2487
        # only want the _history index to reference the 1st index entry
 
2488
        # for version_id
 
2489
        if version_id not in cache:
 
2490
            index = len(history)
 
2491
            history.append(version_id)
 
2492
        else:
 
2493
            index = cache[version_id][5]
 
2494
        cache[version_id] = (version_id,
 
2495
                                   options,
 
2496
                                   pos,
 
2497
                                   size,
 
2498
                                   parents,
 
2499
                                   index)
 
2500
 
 
2501
    def check_header(self, fp):
 
2502
        line = fp.readline()
 
2503
        if line == '':
 
2504
            # An empty file can actually be treated as though the file doesn't
 
2505
            # exist yet.
 
2506
            raise errors.NoSuchFile(self)
 
2507
        if line != self.HEADER:
 
2508
            raise KnitHeaderError(badline=line, filename=self)
 
2509
 
 
2510
    def _check_read(self):
 
2511
        if not self._is_locked():
 
2512
            raise errors.ObjectNotLocked(self)
 
2513
        if self._get_scope() != self._scope:
 
2514
            self._reset_cache()
 
2515
 
 
2516
    def _check_write_ok(self):
 
2517
        """Assert if not writes are permitted."""
 
2518
        if not self._is_locked():
 
2519
            raise errors.ObjectNotLocked(self)
 
2520
        if self._get_scope() != self._scope:
 
2521
            self._reset_cache()
 
2522
        if self._mode != 'w':
 
2523
            raise errors.ReadOnlyObjectDirtiedError(self)
 
2524
 
 
2525
    def get_build_details(self, keys):
 
2526
        """Get the method, index_memo and compression parent for keys.
 
2527
 
 
2528
        Ghosts are omitted from the result.
 
2529
 
 
2530
        :param keys: An iterable of keys.
 
2531
        :return: A dict of key:(index_memo, compression_parent, parents,
 
2532
            record_details).
 
2533
            index_memo
 
2534
                opaque structure to pass to read_records to extract the raw
 
2535
                data
 
2536
            compression_parent
 
2537
                Content that this record is built upon, may be None
 
2538
            parents
 
2539
                Logical parents of this node
 
2540
            record_details
 
2541
                extra information about the content which needs to be passed to
 
2542
                Factory.parse_record
 
2543
        """
 
2544
        parent_map = self.get_parent_map(keys)
 
2545
        result = {}
 
2546
        for key in keys:
 
2547
            if key not in parent_map:
 
2548
                continue # Ghost
 
2549
            method = self.get_method(key)
 
2550
            parents = parent_map[key]
 
2551
            if method == 'fulltext':
 
2552
                compression_parent = None
 
2553
            else:
 
2554
                compression_parent = parents[0]
 
2555
            noeol = 'no-eol' in self.get_options(key)
 
2556
            index_memo = self.get_position(key)
 
2557
            result[key] = (index_memo, compression_parent,
 
2558
                                  parents, (method, noeol))
 
2559
        return result
 
2560
 
 
2561
    def get_method(self, key):
 
2562
        """Return compression method of specified key."""
 
2563
        options = self.get_options(key)
 
2564
        if 'fulltext' in options:
 
2565
            return 'fulltext'
 
2566
        elif 'line-delta' in options:
 
2567
            return 'line-delta'
 
2568
        else:
 
2569
            raise errors.KnitIndexUnknownMethod(self, options)
 
2570
 
 
2571
    def get_options(self, key):
 
2572
        """Return a list representing options.
 
2573
 
 
2574
        e.g. ['foo', 'bar']
 
2575
        """
 
2576
        prefix, suffix = self._split_key(key)
 
2577
        self._load_prefixes([prefix])
 
2578
        try:
 
2579
            return self._kndx_cache[prefix][0][suffix][1]
 
2580
        except KeyError:
 
2581
            raise RevisionNotPresent(key, self)
 
2582
 
 
2583
    def find_ancestry(self, keys):
 
2584
        """See CombinedGraphIndex.find_ancestry()"""
 
2585
        prefixes = set(key[:-1] for key in keys)
 
2586
        self._load_prefixes(prefixes)
 
2587
        result = {}
 
2588
        parent_map = {}
 
2589
        missing_keys = set()
 
2590
        pending_keys = list(keys)
 
2591
        # This assumes that keys will not reference parents in a different
 
2592
        # prefix, which is accurate so far.
 
2593
        while pending_keys:
 
2594
            key = pending_keys.pop()
 
2595
            if key in parent_map:
 
2596
                continue
 
2597
            prefix = key[:-1]
 
2598
            try:
 
2599
                suffix_parents = self._kndx_cache[prefix][0][key[-1]][4]
 
2600
            except KeyError:
 
2601
                missing_keys.add(key)
 
2602
            else:
 
2603
                parent_keys = tuple([prefix + (suffix,)
 
2604
                                     for suffix in suffix_parents])
 
2605
                parent_map[key] = parent_keys
 
2606
                pending_keys.extend([p for p in parent_keys
 
2607
                                        if p not in parent_map])
 
2608
        return parent_map, missing_keys
 
2609
 
 
2610
    def get_parent_map(self, keys):
 
2611
        """Get a map of the parents of keys.
 
2612
 
 
2613
        :param keys: The keys to look up parents for.
 
2614
        :return: A mapping from keys to parents. Absent keys are absent from
 
2615
            the mapping.
 
2616
        """
 
2617
        # Parse what we need to up front, this potentially trades off I/O
 
2618
        # locality (.kndx and .knit in the same block group for the same file
 
2619
        # id) for less checking in inner loops.
 
2620
        prefixes = set(key[:-1] for key in keys)
 
2621
        self._load_prefixes(prefixes)
 
2622
        result = {}
 
2623
        for key in keys:
 
2624
            prefix = key[:-1]
 
2625
            try:
 
2626
                suffix_parents = self._kndx_cache[prefix][0][key[-1]][4]
 
2627
            except KeyError:
 
2628
                pass
 
2629
            else:
 
2630
                result[key] = tuple(prefix + (suffix,) for
 
2631
                    suffix in suffix_parents)
 
2632
        return result
 
2633
 
 
2634
    def get_position(self, key):
 
2635
        """Return details needed to access the version.
 
2636
 
 
2637
        :return: a tuple (key, data position, size) to hand to the access
 
2638
            logic to get the record.
 
2639
        """
 
2640
        prefix, suffix = self._split_key(key)
 
2641
        self._load_prefixes([prefix])
 
2642
        entry = self._kndx_cache[prefix][0][suffix]
 
2643
        return key, entry[2], entry[3]
 
2644
 
 
2645
    has_key = _mod_index._has_key_from_parent_map
 
2646
 
 
2647
    def _init_index(self, path, extra_lines=[]):
 
2648
        """Initialize an index."""
 
2649
        sio = StringIO()
 
2650
        sio.write(self.HEADER)
 
2651
        sio.writelines(extra_lines)
 
2652
        sio.seek(0)
 
2653
        self._transport.put_file_non_atomic(path, sio,
 
2654
                            create_parent_dir=True)
 
2655
                           # self._create_parent_dir)
 
2656
                           # mode=self._file_mode,
 
2657
                           # dir_mode=self._dir_mode)
 
2658
 
 
2659
    def keys(self):
 
2660
        """Get all the keys in the collection.
 
2661
 
 
2662
        The keys are not ordered.
 
2663
        """
 
2664
        result = set()
 
2665
        # Identify all key prefixes.
 
2666
        # XXX: A bit hacky, needs polish.
 
2667
        if type(self._mapper) is ConstantMapper:
 
2668
            prefixes = [()]
 
2669
        else:
 
2670
            relpaths = set()
 
2671
            for quoted_relpath in self._transport.iter_files_recursive():
 
2672
                path, ext = os.path.splitext(quoted_relpath)
 
2673
                relpaths.add(path)
 
2674
            prefixes = [self._mapper.unmap(path) for path in relpaths]
 
2675
        self._load_prefixes(prefixes)
 
2676
        for prefix in prefixes:
 
2677
            for suffix in self._kndx_cache[prefix][1]:
 
2678
                result.add(prefix + (suffix,))
 
2679
        return result
 
2680
 
 
2681
    def _load_prefixes(self, prefixes):
 
2682
        """Load the indices for prefixes."""
 
2683
        self._check_read()
 
2684
        for prefix in prefixes:
 
2685
            if prefix not in self._kndx_cache:
 
2686
                # the load_data interface writes to these variables.
 
2687
                self._cache = {}
 
2688
                self._history = []
 
2689
                self._filename = prefix
 
2690
                try:
 
2691
                    path = self._mapper.map(prefix) + '.kndx'
 
2692
                    fp = self._transport.get(path)
 
2693
                    try:
 
2694
                        # _load_data may raise NoSuchFile if the target knit is
 
2695
                        # completely empty.
 
2696
                        _load_data(self, fp)
 
2697
                    finally:
 
2698
                        fp.close()
 
2699
                    self._kndx_cache[prefix] = (self._cache, self._history)
 
2700
                    del self._cache
 
2701
                    del self._filename
 
2702
                    del self._history
 
2703
                except NoSuchFile:
 
2704
                    self._kndx_cache[prefix] = ({}, [])
 
2705
                    if type(self._mapper) is ConstantMapper:
 
2706
                        # preserve behaviour for revisions.kndx etc.
 
2707
                        self._init_index(path)
 
2708
                    del self._cache
 
2709
                    del self._filename
 
2710
                    del self._history
 
2711
 
 
2712
    missing_keys = _mod_index._missing_keys_from_parent_map
 
2713
 
 
2714
    def _partition_keys(self, keys):
 
2715
        """Turn keys into a dict of prefix:suffix_list."""
 
2716
        result = {}
 
2717
        for key in keys:
 
2718
            prefix_keys = result.setdefault(key[:-1], [])
 
2719
            prefix_keys.append(key[-1])
 
2720
        return result
 
2721
 
 
2722
    def _dictionary_compress(self, keys):
 
2723
        """Dictionary compress keys.
 
2724
 
 
2725
        :param keys: The keys to generate references to.
 
2726
        :return: A string representation of keys. keys which are present are
 
2727
            dictionary compressed, and others are emitted as fulltext with a
 
2728
            '.' prefix.
 
2729
        """
 
2730
        if not keys:
 
2731
            return ''
 
2732
        result_list = []
 
2733
        prefix = keys[0][:-1]
 
2734
        cache = self._kndx_cache[prefix][0]
 
2735
        for key in keys:
 
2736
            if key[:-1] != prefix:
 
2737
                # kndx indices cannot refer across partitioned storage.
 
2738
                raise ValueError("mismatched prefixes for %r" % keys)
 
2739
            if key[-1] in cache:
 
2740
                # -- inlined lookup() --
 
2741
                result_list.append(str(cache[key[-1]][5]))
 
2742
                # -- end lookup () --
 
2743
            else:
 
2744
                result_list.append('.' + key[-1])
 
2745
        return ' '.join(result_list)
 
2746
 
 
2747
    def _reset_cache(self):
 
2748
        # Possibly this should be a LRU cache. A dictionary from key_prefix to
 
2749
        # (cache_dict, history_vector) for parsed kndx files.
 
2750
        self._kndx_cache = {}
 
2751
        self._scope = self._get_scope()
 
2752
        allow_writes = self._allow_writes()
 
2753
        if allow_writes:
 
2754
            self._mode = 'w'
 
2755
        else:
 
2756
            self._mode = 'r'
 
2757
 
 
2758
    def _sort_keys_by_io(self, keys, positions):
 
2759
        """Figure out an optimal order to read the records for the given keys.
 
2760
 
 
2761
        Sort keys, grouped by index and sorted by position.
 
2762
 
 
2763
        :param keys: A list of keys whose records we want to read. This will be
 
2764
            sorted 'in-place'.
 
2765
        :param positions: A dict, such as the one returned by
 
2766
            _get_components_positions()
 
2767
        :return: None
 
2768
        """
 
2769
        def get_sort_key(key):
 
2770
            index_memo = positions[key][1]
 
2771
            # Group by prefix and position. index_memo[0] is the key, so it is
 
2772
            # (file_id, revision_id) and we don't want to sort on revision_id,
 
2773
            # index_memo[1] is the position, and index_memo[2] is the size,
 
2774
            # which doesn't matter for the sort
 
2775
            return index_memo[0][:-1], index_memo[1]
 
2776
        return keys.sort(key=get_sort_key)
 
2777
 
 
2778
    _get_total_build_size = _get_total_build_size
 
2779
 
 
2780
    def _split_key(self, key):
 
2781
        """Split key into a prefix and suffix."""
 
2782
        return key[:-1], key[-1]
 
2783
 
 
2784
 
 
2785
class _KnitGraphIndex(object):
 
2786
    """A KnitVersionedFiles index layered on GraphIndex."""
 
2787
 
 
2788
    def __init__(self, graph_index, is_locked, deltas=False, parents=True,
 
2789
        add_callback=None, track_external_parent_refs=False):
 
2790
        """Construct a KnitGraphIndex on a graph_index.
 
2791
 
 
2792
        :param graph_index: An implementation of bzrlib.index.GraphIndex.
 
2793
        :param is_locked: A callback to check whether the object should answer
 
2794
            queries.
 
2795
        :param deltas: Allow delta-compressed records.
 
2796
        :param parents: If True, record knits parents, if not do not record
 
2797
            parents.
 
2798
        :param add_callback: If not None, allow additions to the index and call
 
2799
            this callback with a list of added GraphIndex nodes:
 
2800
            [(node, value, node_refs), ...]
 
2801
        :param is_locked: A callback, returns True if the index is locked and
 
2802
            thus usable.
 
2803
        :param track_external_parent_refs: If True, record all external parent
 
2804
            references parents from added records.  These can be retrieved
 
2805
            later by calling get_missing_parents().
 
2806
        """
 
2807
        self._add_callback = add_callback
 
2808
        self._graph_index = graph_index
 
2809
        self._deltas = deltas
 
2810
        self._parents = parents
 
2811
        if deltas and not parents:
 
2812
            # XXX: TODO: Delta tree and parent graph should be conceptually
 
2813
            # separate.
 
2814
            raise KnitCorrupt(self, "Cannot do delta compression without "
 
2815
                "parent tracking.")
 
2816
        self.has_graph = parents
 
2817
        self._is_locked = is_locked
 
2818
        self._missing_compression_parents = set()
 
2819
        if track_external_parent_refs:
 
2820
            self._key_dependencies = _KeyRefs()
 
2821
        else:
 
2822
            self._key_dependencies = None
 
2823
 
 
2824
    def __repr__(self):
 
2825
        return "%s(%r)" % (self.__class__.__name__, self._graph_index)
 
2826
 
 
2827
    def add_records(self, records, random_id=False,
 
2828
        missing_compression_parents=False):
 
2829
        """Add multiple records to the index.
 
2830
 
 
2831
        This function does not insert data into the Immutable GraphIndex
 
2832
        backing the KnitGraphIndex, instead it prepares data for insertion by
 
2833
        the caller and checks that it is safe to insert then calls
 
2834
        self._add_callback with the prepared GraphIndex nodes.
 
2835
 
 
2836
        :param records: a list of tuples:
 
2837
                         (key, options, access_memo, parents).
 
2838
        :param random_id: If True the ids being added were randomly generated
 
2839
            and no check for existence will be performed.
 
2840
        :param missing_compression_parents: If True the records being added are
 
2841
            only compressed against texts already in the index (or inside
 
2842
            records). If False the records all refer to unavailable texts (or
 
2843
            texts inside records) as compression parents.
 
2844
        """
 
2845
        if not self._add_callback:
 
2846
            raise errors.ReadOnlyError(self)
 
2847
        # we hope there are no repositories with inconsistent parentage
 
2848
        # anymore.
 
2849
 
 
2850
        keys = {}
 
2851
        compression_parents = set()
 
2852
        key_dependencies = self._key_dependencies
 
2853
        for (key, options, access_memo, parents) in records:
 
2854
            if self._parents:
 
2855
                parents = tuple(parents)
 
2856
                if key_dependencies is not None:
 
2857
                    key_dependencies.add_references(key, parents)
 
2858
            index, pos, size = access_memo
 
2859
            if 'no-eol' in options:
 
2860
                value = 'N'
 
2861
            else:
 
2862
                value = ' '
 
2863
            value += "%d %d" % (pos, size)
 
2864
            if not self._deltas:
 
2865
                if 'line-delta' in options:
 
2866
                    raise KnitCorrupt(self, "attempt to add line-delta in non-delta knit")
 
2867
            if self._parents:
 
2868
                if self._deltas:
 
2869
                    if 'line-delta' in options:
 
2870
                        node_refs = (parents, (parents[0],))
 
2871
                        if missing_compression_parents:
 
2872
                            compression_parents.add(parents[0])
 
2873
                    else:
 
2874
                        node_refs = (parents, ())
 
2875
                else:
 
2876
                    node_refs = (parents, )
 
2877
            else:
 
2878
                if parents:
 
2879
                    raise KnitCorrupt(self, "attempt to add node with parents "
 
2880
                        "in parentless index.")
 
2881
                node_refs = ()
 
2882
            keys[key] = (value, node_refs)
 
2883
        # check for dups
 
2884
        if not random_id:
 
2885
            present_nodes = self._get_entries(keys)
 
2886
            for (index, key, value, node_refs) in present_nodes:
 
2887
                parents = node_refs[:1]
 
2888
                # Sometimes these are passed as a list rather than a tuple
 
2889
                passed = static_tuple.as_tuples(keys[key])
 
2890
                passed_parents = passed[1][:1]
 
2891
                if (value[0] != keys[key][0][0] or
 
2892
                    parents != passed_parents):
 
2893
                    node_refs = static_tuple.as_tuples(node_refs)
 
2894
                    raise KnitCorrupt(self, "inconsistent details in add_records"
 
2895
                        ": %s %s" % ((value, node_refs), passed))
 
2896
                del keys[key]
 
2897
        result = []
 
2898
        if self._parents:
 
2899
            for key, (value, node_refs) in keys.iteritems():
 
2900
                result.append((key, value, node_refs))
 
2901
        else:
 
2902
            for key, (value, node_refs) in keys.iteritems():
 
2903
                result.append((key, value))
 
2904
        self._add_callback(result)
 
2905
        if missing_compression_parents:
 
2906
            # This may appear to be incorrect (it does not check for
 
2907
            # compression parents that are in the existing graph index),
 
2908
            # but such records won't have been buffered, so this is
 
2909
            # actually correct: every entry when
 
2910
            # missing_compression_parents==True either has a missing parent, or
 
2911
            # a parent that is one of the keys in records.
 
2912
            compression_parents.difference_update(keys)
 
2913
            self._missing_compression_parents.update(compression_parents)
 
2914
        # Adding records may have satisfied missing compression parents.
 
2915
        self._missing_compression_parents.difference_update(keys)
 
2916
 
 
2917
    def scan_unvalidated_index(self, graph_index):
 
2918
        """Inform this _KnitGraphIndex that there is an unvalidated index.
 
2919
 
 
2920
        This allows this _KnitGraphIndex to keep track of any missing
 
2921
        compression parents we may want to have filled in to make those
 
2922
        indices valid.
 
2923
 
 
2924
        :param graph_index: A GraphIndex
 
2925
        """
 
2926
        if self._deltas:
 
2927
            new_missing = graph_index.external_references(ref_list_num=1)
 
2928
            new_missing.difference_update(self.get_parent_map(new_missing))
 
2929
            self._missing_compression_parents.update(new_missing)
 
2930
        if self._key_dependencies is not None:
 
2931
            # Add parent refs from graph_index (and discard parent refs that
 
2932
            # the graph_index has).
 
2933
            for node in graph_index.iter_all_entries():
 
2934
                self._key_dependencies.add_references(node[1], node[3][0])
 
2935
 
 
2936
    def get_missing_compression_parents(self):
 
2937
        """Return the keys of missing compression parents.
 
2938
 
 
2939
        Missing compression parents occur when a record stream was missing
 
2940
        basis texts, or a index was scanned that had missing basis texts.
 
2941
        """
 
2942
        return frozenset(self._missing_compression_parents)
 
2943
 
 
2944
    def get_missing_parents(self):
 
2945
        """Return the keys of missing parents."""
 
2946
        # If updating this, you should also update
 
2947
        # groupcompress._GCGraphIndex.get_missing_parents
 
2948
        # We may have false positives, so filter those out.
 
2949
        self._key_dependencies.satisfy_refs_for_keys(
 
2950
            self.get_parent_map(self._key_dependencies.get_unsatisfied_refs()))
 
2951
        return frozenset(self._key_dependencies.get_unsatisfied_refs())
 
2952
 
 
2953
    def _check_read(self):
 
2954
        """raise if reads are not permitted."""
 
2955
        if not self._is_locked():
 
2956
            raise errors.ObjectNotLocked(self)
 
2957
 
 
2958
    def _check_write_ok(self):
 
2959
        """Assert if writes are not permitted."""
 
2960
        if not self._is_locked():
 
2961
            raise errors.ObjectNotLocked(self)
 
2962
 
 
2963
    def _compression_parent(self, an_entry):
 
2964
        # return the key that an_entry is compressed against, or None
 
2965
        # Grab the second parent list (as deltas implies parents currently)
 
2966
        compression_parents = an_entry[3][1]
 
2967
        if not compression_parents:
 
2968
            return None
 
2969
        if len(compression_parents) != 1:
 
2970
            raise AssertionError(
 
2971
                "Too many compression parents: %r" % compression_parents)
 
2972
        return compression_parents[0]
 
2973
 
 
2974
    def get_build_details(self, keys):
 
2975
        """Get the method, index_memo and compression parent for version_ids.
 
2976
 
 
2977
        Ghosts are omitted from the result.
 
2978
 
 
2979
        :param keys: An iterable of keys.
 
2980
        :return: A dict of key:
 
2981
            (index_memo, compression_parent, parents, record_details).
 
2982
            index_memo
 
2983
                opaque structure to pass to read_records to extract the raw
 
2984
                data
 
2985
            compression_parent
 
2986
                Content that this record is built upon, may be None
 
2987
            parents
 
2988
                Logical parents of this node
 
2989
            record_details
 
2990
                extra information about the content which needs to be passed to
 
2991
                Factory.parse_record
 
2992
        """
 
2993
        self._check_read()
 
2994
        result = {}
 
2995
        entries = self._get_entries(keys, False)
 
2996
        for entry in entries:
 
2997
            key = entry[1]
 
2998
            if not self._parents:
 
2999
                parents = ()
 
3000
            else:
 
3001
                parents = entry[3][0]
 
3002
            if not self._deltas:
 
3003
                compression_parent_key = None
 
3004
            else:
 
3005
                compression_parent_key = self._compression_parent(entry)
 
3006
            noeol = (entry[2][0] == 'N')
 
3007
            if compression_parent_key:
 
3008
                method = 'line-delta'
 
3009
            else:
 
3010
                method = 'fulltext'
 
3011
            result[key] = (self._node_to_position(entry),
 
3012
                                  compression_parent_key, parents,
 
3013
                                  (method, noeol))
 
3014
        return result
 
3015
 
 
3016
    def _get_entries(self, keys, check_present=False):
 
3017
        """Get the entries for keys.
 
3018
 
 
3019
        :param keys: An iterable of index key tuples.
 
3020
        """
 
3021
        keys = set(keys)
 
3022
        found_keys = set()
 
3023
        if self._parents:
 
3024
            for node in self._graph_index.iter_entries(keys):
 
3025
                yield node
 
3026
                found_keys.add(node[1])
 
3027
        else:
 
3028
            # adapt parentless index to the rest of the code.
 
3029
            for node in self._graph_index.iter_entries(keys):
 
3030
                yield node[0], node[1], node[2], ()
 
3031
                found_keys.add(node[1])
 
3032
        if check_present:
 
3033
            missing_keys = keys.difference(found_keys)
 
3034
            if missing_keys:
 
3035
                raise RevisionNotPresent(missing_keys.pop(), self)
 
3036
 
 
3037
    def get_method(self, key):
 
3038
        """Return compression method of specified key."""
 
3039
        return self._get_method(self._get_node(key))
 
3040
 
 
3041
    def _get_method(self, node):
 
3042
        if not self._deltas:
 
3043
            return 'fulltext'
 
3044
        if self._compression_parent(node):
 
3045
            return 'line-delta'
 
3046
        else:
 
3047
            return 'fulltext'
 
3048
 
 
3049
    def _get_node(self, key):
 
3050
        try:
 
3051
            return list(self._get_entries([key]))[0]
 
3052
        except IndexError:
 
3053
            raise RevisionNotPresent(key, self)
 
3054
 
 
3055
    def get_options(self, key):
 
3056
        """Return a list representing options.
 
3057
 
 
3058
        e.g. ['foo', 'bar']
 
3059
        """
 
3060
        node = self._get_node(key)
 
3061
        options = [self._get_method(node)]
 
3062
        if node[2][0] == 'N':
 
3063
            options.append('no-eol')
 
3064
        return options
 
3065
 
 
3066
    def find_ancestry(self, keys):
 
3067
        """See CombinedGraphIndex.find_ancestry()"""
 
3068
        return self._graph_index.find_ancestry(keys, 0)
 
3069
 
 
3070
    def get_parent_map(self, keys):
 
3071
        """Get a map of the parents of keys.
 
3072
 
 
3073
        :param keys: The keys to look up parents for.
 
3074
        :return: A mapping from keys to parents. Absent keys are absent from
 
3075
            the mapping.
 
3076
        """
 
3077
        self._check_read()
 
3078
        nodes = self._get_entries(keys)
 
3079
        result = {}
 
3080
        if self._parents:
 
3081
            for node in nodes:
 
3082
                result[node[1]] = node[3][0]
 
3083
        else:
 
3084
            for node in nodes:
 
3085
                result[node[1]] = None
 
3086
        return result
 
3087
 
 
3088
    def get_position(self, key):
 
3089
        """Return details needed to access the version.
 
3090
 
 
3091
        :return: a tuple (index, data position, size) to hand to the access
 
3092
            logic to get the record.
 
3093
        """
 
3094
        node = self._get_node(key)
 
3095
        return self._node_to_position(node)
 
3096
 
 
3097
    has_key = _mod_index._has_key_from_parent_map
 
3098
 
 
3099
    def keys(self):
 
3100
        """Get all the keys in the collection.
 
3101
 
 
3102
        The keys are not ordered.
 
3103
        """
 
3104
        self._check_read()
 
3105
        return [node[1] for node in self._graph_index.iter_all_entries()]
 
3106
 
 
3107
    missing_keys = _mod_index._missing_keys_from_parent_map
 
3108
 
 
3109
    def _node_to_position(self, node):
 
3110
        """Convert an index value to position details."""
 
3111
        bits = node[2][1:].split(' ')
 
3112
        return node[0], int(bits[0]), int(bits[1])
 
3113
 
 
3114
    def _sort_keys_by_io(self, keys, positions):
 
3115
        """Figure out an optimal order to read the records for the given keys.
 
3116
 
 
3117
        Sort keys, grouped by index and sorted by position.
 
3118
 
 
3119
        :param keys: A list of keys whose records we want to read. This will be
 
3120
            sorted 'in-place'.
 
3121
        :param positions: A dict, such as the one returned by
 
3122
            _get_components_positions()
 
3123
        :return: None
 
3124
        """
 
3125
        def get_index_memo(key):
 
3126
            # index_memo is at offset [1]. It is made up of (GraphIndex,
 
3127
            # position, size). GI is an object, which will be unique for each
 
3128
            # pack file. This causes us to group by pack file, then sort by
 
3129
            # position. Size doesn't matter, but it isn't worth breaking up the
 
3130
            # tuple.
 
3131
            return positions[key][1]
 
3132
        return keys.sort(key=get_index_memo)
 
3133
 
 
3134
    _get_total_build_size = _get_total_build_size
 
3135
 
 
3136
 
 
3137
class _KnitKeyAccess(object):
 
3138
    """Access to records in .knit files."""
 
3139
 
 
3140
    def __init__(self, transport, mapper):
 
3141
        """Create a _KnitKeyAccess with transport and mapper.
 
3142
 
 
3143
        :param transport: The transport the access object is rooted at.
 
3144
        :param mapper: The mapper used to map keys to .knit files.
 
3145
        """
 
3146
        self._transport = transport
 
3147
        self._mapper = mapper
 
3148
 
 
3149
    def add_raw_records(self, key_sizes, raw_data):
 
3150
        """Add raw knit bytes to a storage area.
 
3151
 
 
3152
        The data is spooled to the container writer in one bytes-record per
 
3153
        raw data item.
 
3154
 
 
3155
        :param sizes: An iterable of tuples containing the key and size of each
 
3156
            raw data segment.
 
3157
        :param raw_data: A bytestring containing the data.
 
3158
        :return: A list of memos to retrieve the record later. Each memo is an
 
3159
            opaque index memo. For _KnitKeyAccess the memo is (key, pos,
 
3160
            length), where the key is the record key.
 
3161
        """
 
3162
        if type(raw_data) is not str:
 
3163
            raise AssertionError(
 
3164
                'data must be plain bytes was %s' % type(raw_data))
 
3165
        result = []
 
3166
        offset = 0
 
3167
        # TODO: This can be tuned for writing to sftp and other servers where
 
3168
        # append() is relatively expensive by grouping the writes to each key
 
3169
        # prefix.
 
3170
        for key, size in key_sizes:
 
3171
            path = self._mapper.map(key)
 
3172
            try:
 
3173
                base = self._transport.append_bytes(path + '.knit',
 
3174
                    raw_data[offset:offset+size])
 
3175
            except errors.NoSuchFile:
 
3176
                self._transport.mkdir(osutils.dirname(path))
 
3177
                base = self._transport.append_bytes(path + '.knit',
 
3178
                    raw_data[offset:offset+size])
 
3179
            # if base == 0:
 
3180
            # chmod.
 
3181
            offset += size
 
3182
            result.append((key, base, size))
 
3183
        return result
 
3184
 
 
3185
    def flush(self):
 
3186
        """Flush pending writes on this access object.
 
3187
        
 
3188
        For .knit files this is a no-op.
 
3189
        """
 
3190
        pass
 
3191
 
 
3192
    def get_raw_records(self, memos_for_retrieval):
 
3193
        """Get the raw bytes for a records.
 
3194
 
 
3195
        :param memos_for_retrieval: An iterable containing the access memo for
 
3196
            retrieving the bytes.
 
3197
        :return: An iterator over the bytes of the records.
 
3198
        """
 
3199
        # first pass, group into same-index request to minimise readv's issued.
 
3200
        request_lists = []
 
3201
        current_prefix = None
 
3202
        for (key, offset, length) in memos_for_retrieval:
 
3203
            if current_prefix == key[:-1]:
 
3204
                current_list.append((offset, length))
 
3205
            else:
 
3206
                if current_prefix is not None:
 
3207
                    request_lists.append((current_prefix, current_list))
 
3208
                current_prefix = key[:-1]
 
3209
                current_list = [(offset, length)]
 
3210
        # handle the last entry
 
3211
        if current_prefix is not None:
 
3212
            request_lists.append((current_prefix, current_list))
 
3213
        for prefix, read_vector in request_lists:
 
3214
            path = self._mapper.map(prefix) + '.knit'
 
3215
            for pos, data in self._transport.readv(path, read_vector):
 
3216
                yield data
 
3217
 
 
3218
 
 
3219
class _DirectPackAccess(object):
 
3220
    """Access to data in one or more packs with less translation."""
 
3221
 
 
3222
    def __init__(self, index_to_packs, reload_func=None, flush_func=None):
 
3223
        """Create a _DirectPackAccess object.
 
3224
 
 
3225
        :param index_to_packs: A dict mapping index objects to the transport
 
3226
            and file names for obtaining data.
 
3227
        :param reload_func: A function to call if we determine that the pack
 
3228
            files have moved and we need to reload our caches. See
 
3229
            bzrlib.repo_fmt.pack_repo.AggregateIndex for more details.
 
3230
        """
 
3231
        self._container_writer = None
 
3232
        self._write_index = None
 
3233
        self._indices = index_to_packs
 
3234
        self._reload_func = reload_func
 
3235
        self._flush_func = flush_func
 
3236
 
 
3237
    def add_raw_records(self, key_sizes, raw_data):
 
3238
        """Add raw knit bytes to a storage area.
 
3239
 
 
3240
        The data is spooled to the container writer in one bytes-record per
 
3241
        raw data item.
 
3242
 
 
3243
        :param sizes: An iterable of tuples containing the key and size of each
 
3244
            raw data segment.
 
3245
        :param raw_data: A bytestring containing the data.
 
3246
        :return: A list of memos to retrieve the record later. Each memo is an
 
3247
            opaque index memo. For _DirectPackAccess the memo is (index, pos,
 
3248
            length), where the index field is the write_index object supplied
 
3249
            to the PackAccess object.
 
3250
        """
 
3251
        if type(raw_data) is not str:
 
3252
            raise AssertionError(
 
3253
                'data must be plain bytes was %s' % type(raw_data))
 
3254
        result = []
 
3255
        offset = 0
 
3256
        for key, size in key_sizes:
 
3257
            p_offset, p_length = self._container_writer.add_bytes_record(
 
3258
                raw_data[offset:offset+size], [])
 
3259
            offset += size
 
3260
            result.append((self._write_index, p_offset, p_length))
 
3261
        return result
 
3262
 
 
3263
    def flush(self):
 
3264
        """Flush pending writes on this access object.
 
3265
 
 
3266
        This will flush any buffered writes to a NewPack.
 
3267
        """
 
3268
        if self._flush_func is not None:
 
3269
            self._flush_func()
 
3270
            
 
3271
    def get_raw_records(self, memos_for_retrieval):
 
3272
        """Get the raw bytes for a records.
 
3273
 
 
3274
        :param memos_for_retrieval: An iterable containing the (index, pos,
 
3275
            length) memo for retrieving the bytes. The Pack access method
 
3276
            looks up the pack to use for a given record in its index_to_pack
 
3277
            map.
 
3278
        :return: An iterator over the bytes of the records.
 
3279
        """
 
3280
        # first pass, group into same-index requests
 
3281
        request_lists = []
 
3282
        current_index = None
 
3283
        for (index, offset, length) in memos_for_retrieval:
 
3284
            if current_index == index:
 
3285
                current_list.append((offset, length))
 
3286
            else:
 
3287
                if current_index is not None:
 
3288
                    request_lists.append((current_index, current_list))
 
3289
                current_index = index
 
3290
                current_list = [(offset, length)]
 
3291
        # handle the last entry
 
3292
        if current_index is not None:
 
3293
            request_lists.append((current_index, current_list))
 
3294
        for index, offsets in request_lists:
 
3295
            try:
 
3296
                transport, path = self._indices[index]
 
3297
            except KeyError:
 
3298
                # A KeyError here indicates that someone has triggered an index
 
3299
                # reload, and this index has gone missing, we need to start
 
3300
                # over.
 
3301
                if self._reload_func is None:
 
3302
                    # If we don't have a _reload_func there is nothing that can
 
3303
                    # be done
 
3304
                    raise
 
3305
                raise errors.RetryWithNewPacks(index,
 
3306
                                               reload_occurred=True,
 
3307
                                               exc_info=sys.exc_info())
 
3308
            try:
 
3309
                reader = pack.make_readv_reader(transport, path, offsets)
 
3310
                for names, read_func in reader.iter_records():
 
3311
                    yield read_func(None)
 
3312
            except errors.NoSuchFile:
 
3313
                # A NoSuchFile error indicates that a pack file has gone
 
3314
                # missing on disk, we need to trigger a reload, and start over.
 
3315
                if self._reload_func is None:
 
3316
                    raise
 
3317
                raise errors.RetryWithNewPacks(transport.abspath(path),
 
3318
                                               reload_occurred=False,
 
3319
                                               exc_info=sys.exc_info())
 
3320
 
 
3321
    def set_writer(self, writer, index, transport_packname):
 
3322
        """Set a writer to use for adding data."""
 
3323
        if index is not None:
 
3324
            self._indices[index] = transport_packname
 
3325
        self._container_writer = writer
 
3326
        self._write_index = index
 
3327
 
 
3328
    def reload_or_raise(self, retry_exc):
 
3329
        """Try calling the reload function, or re-raise the original exception.
 
3330
 
 
3331
        This should be called after _DirectPackAccess raises a
 
3332
        RetryWithNewPacks exception. This function will handle the common logic
 
3333
        of determining when the error is fatal versus being temporary.
 
3334
        It will also make sure that the original exception is raised, rather
 
3335
        than the RetryWithNewPacks exception.
 
3336
 
 
3337
        If this function returns, then the calling function should retry
 
3338
        whatever operation was being performed. Otherwise an exception will
 
3339
        be raised.
 
3340
 
 
3341
        :param retry_exc: A RetryWithNewPacks exception.
 
3342
        """
 
3343
        is_error = False
 
3344
        if self._reload_func is None:
 
3345
            is_error = True
 
3346
        elif not self._reload_func():
 
3347
            # The reload claimed that nothing changed
 
3348
            if not retry_exc.reload_occurred:
 
3349
                # If there wasn't an earlier reload, then we really were
 
3350
                # expecting to find changes. We didn't find them, so this is a
 
3351
                # hard error
 
3352
                is_error = True
 
3353
        if is_error:
 
3354
            exc_class, exc_value, exc_traceback = retry_exc.exc_info
 
3355
            raise exc_class, exc_value, exc_traceback
 
3356
 
 
3357
 
 
3358
def annotate_knit(knit, revision_id):
 
3359
    """Annotate a knit with no cached annotations.
 
3360
 
 
3361
    This implementation is for knits with no cached annotations.
 
3362
    It will work for knits with cached annotations, but this is not
 
3363
    recommended.
 
3364
    """
 
3365
    annotator = _KnitAnnotator(knit)
 
3366
    return iter(annotator.annotate_flat(revision_id))
 
3367
 
 
3368
 
 
3369
class _KnitAnnotator(annotate.Annotator):
 
3370
    """Build up the annotations for a text."""
 
3371
 
 
3372
    def __init__(self, vf):
 
3373
        annotate.Annotator.__init__(self, vf)
 
3374
 
 
3375
        # TODO: handle Nodes which cannot be extracted
 
3376
        # self._ghosts = set()
 
3377
 
 
3378
        # Map from (key, parent_key) => matching_blocks, should be 'use once'
 
3379
        self._matching_blocks = {}
 
3380
 
 
3381
        # KnitContent objects
 
3382
        self._content_objects = {}
 
3383
        # The number of children that depend on this fulltext content object
 
3384
        self._num_compression_children = {}
 
3385
        # Delta records that need their compression parent before they can be
 
3386
        # expanded
 
3387
        self._pending_deltas = {}
 
3388
        # Fulltext records that are waiting for their parents fulltexts before
 
3389
        # they can be yielded for annotation
 
3390
        self._pending_annotation = {}
 
3391
 
 
3392
        self._all_build_details = {}
 
3393
 
 
3394
    def _get_build_graph(self, key):
 
3395
        """Get the graphs for building texts and annotations.
 
3396
 
 
3397
        The data you need for creating a full text may be different than the
 
3398
        data you need to annotate that text. (At a minimum, you need both
 
3399
        parents to create an annotation, but only need 1 parent to generate the
 
3400
        fulltext.)
 
3401
 
 
3402
        :return: A list of (key, index_memo) records, suitable for
 
3403
            passing to read_records_iter to start reading in the raw data from
 
3404
            the pack file.
 
3405
        """
 
3406
        pending = set([key])
 
3407
        records = []
 
3408
        ann_keys = set()
 
3409
        self._num_needed_children[key] = 1
 
3410
        while pending:
 
3411
            # get all pending nodes
 
3412
            this_iteration = pending
 
3413
            build_details = self._vf._index.get_build_details(this_iteration)
 
3414
            self._all_build_details.update(build_details)
 
3415
            # new_nodes = self._vf._index._get_entries(this_iteration)
 
3416
            pending = set()
 
3417
            for key, details in build_details.iteritems():
 
3418
                (index_memo, compression_parent, parent_keys,
 
3419
                 record_details) = details
 
3420
                self._parent_map[key] = parent_keys
 
3421
                self._heads_provider = None
 
3422
                records.append((key, index_memo))
 
3423
                # Do we actually need to check _annotated_lines?
 
3424
                pending.update([p for p in parent_keys
 
3425
                                   if p not in self._all_build_details])
 
3426
                if parent_keys:
 
3427
                    for parent_key in parent_keys:
 
3428
                        if parent_key in self._num_needed_children:
 
3429
                            self._num_needed_children[parent_key] += 1
 
3430
                        else:
 
3431
                            self._num_needed_children[parent_key] = 1
 
3432
                if compression_parent:
 
3433
                    if compression_parent in self._num_compression_children:
 
3434
                        self._num_compression_children[compression_parent] += 1
 
3435
                    else:
 
3436
                        self._num_compression_children[compression_parent] = 1
 
3437
 
 
3438
            missing_versions = this_iteration.difference(build_details.keys())
 
3439
            if missing_versions:
 
3440
                for key in missing_versions:
 
3441
                    if key in self._parent_map and key in self._text_cache:
 
3442
                        # We already have this text ready, we just need to
 
3443
                        # yield it later so we get it annotated
 
3444
                        ann_keys.add(key)
 
3445
                        parent_keys = self._parent_map[key]
 
3446
                        for parent_key in parent_keys:
 
3447
                            if parent_key in self._num_needed_children:
 
3448
                                self._num_needed_children[parent_key] += 1
 
3449
                            else:
 
3450
                                self._num_needed_children[parent_key] = 1
 
3451
                        pending.update([p for p in parent_keys
 
3452
                                           if p not in self._all_build_details])
 
3453
                    else:
 
3454
                        raise errors.RevisionNotPresent(key, self._vf)
 
3455
        # Generally we will want to read the records in reverse order, because
 
3456
        # we find the parent nodes after the children
 
3457
        records.reverse()
 
3458
        return records, ann_keys
 
3459
 
 
3460
    def _get_needed_texts(self, key, pb=None):
 
3461
        # if True or len(self._vf._immediate_fallback_vfs) > 0:
 
3462
        if len(self._vf._immediate_fallback_vfs) > 0:
 
3463
            # If we have fallbacks, go to the generic path
 
3464
            for v in annotate.Annotator._get_needed_texts(self, key, pb=pb):
 
3465
                yield v
 
3466
            return
 
3467
        while True:
 
3468
            try:
 
3469
                records, ann_keys = self._get_build_graph(key)
 
3470
                for idx, (sub_key, text, num_lines) in enumerate(
 
3471
                                                self._extract_texts(records)):
 
3472
                    if pb is not None:
 
3473
                        pb.update('annotating', idx, len(records))
 
3474
                    yield sub_key, text, num_lines
 
3475
                for sub_key in ann_keys:
 
3476
                    text = self._text_cache[sub_key]
 
3477
                    num_lines = len(text) # bad assumption
 
3478
                    yield sub_key, text, num_lines
 
3479
                return
 
3480
            except errors.RetryWithNewPacks, e:
 
3481
                self._vf._access.reload_or_raise(e)
 
3482
                # The cached build_details are no longer valid
 
3483
                self._all_build_details.clear()
 
3484
 
 
3485
    def _cache_delta_blocks(self, key, compression_parent, delta, lines):
 
3486
        parent_lines = self._text_cache[compression_parent]
 
3487
        blocks = list(KnitContent.get_line_delta_blocks(delta, parent_lines, lines))
 
3488
        self._matching_blocks[(key, compression_parent)] = blocks
 
3489
 
 
3490
    def _expand_record(self, key, parent_keys, compression_parent, record,
 
3491
                       record_details):
 
3492
        delta = None
 
3493
        if compression_parent:
 
3494
            if compression_parent not in self._content_objects:
 
3495
                # Waiting for the parent
 
3496
                self._pending_deltas.setdefault(compression_parent, []).append(
 
3497
                    (key, parent_keys, record, record_details))
 
3498
                return None
 
3499
            # We have the basis parent, so expand the delta
 
3500
            num = self._num_compression_children[compression_parent]
 
3501
            num -= 1
 
3502
            if num == 0:
 
3503
                base_content = self._content_objects.pop(compression_parent)
 
3504
                self._num_compression_children.pop(compression_parent)
 
3505
            else:
 
3506
                self._num_compression_children[compression_parent] = num
 
3507
                base_content = self._content_objects[compression_parent]
 
3508
            # It is tempting to want to copy_base_content=False for the last
 
3509
            # child object. However, whenever noeol=False,
 
3510
            # self._text_cache[parent_key] is content._lines. So mutating it
 
3511
            # gives very bad results.
 
3512
            # The alternative is to copy the lines into text cache, but then we
 
3513
            # are copying anyway, so just do it here.
 
3514
            content, delta = self._vf._factory.parse_record(
 
3515
                key, record, record_details, base_content,
 
3516
                copy_base_content=True)
 
3517
        else:
 
3518
            # Fulltext record
 
3519
            content, _ = self._vf._factory.parse_record(
 
3520
                key, record, record_details, None)
 
3521
        if self._num_compression_children.get(key, 0) > 0:
 
3522
            self._content_objects[key] = content
 
3523
        lines = content.text()
 
3524
        self._text_cache[key] = lines
 
3525
        if delta is not None:
 
3526
            self._cache_delta_blocks(key, compression_parent, delta, lines)
 
3527
        return lines
 
3528
 
 
3529
    def _get_parent_annotations_and_matches(self, key, text, parent_key):
 
3530
        """Get the list of annotations for the parent, and the matching lines.
 
3531
 
 
3532
        :param text: The opaque value given by _get_needed_texts
 
3533
        :param parent_key: The key for the parent text
 
3534
        :return: (parent_annotations, matching_blocks)
 
3535
            parent_annotations is a list as long as the number of lines in
 
3536
                parent
 
3537
            matching_blocks is a list of (parent_idx, text_idx, len) tuples
 
3538
                indicating which lines match between the two texts
 
3539
        """
 
3540
        block_key = (key, parent_key)
 
3541
        if block_key in self._matching_blocks:
 
3542
            blocks = self._matching_blocks.pop(block_key)
 
3543
            parent_annotations = self._annotations_cache[parent_key]
 
3544
            return parent_annotations, blocks
 
3545
        return annotate.Annotator._get_parent_annotations_and_matches(self,
 
3546
            key, text, parent_key)
 
3547
 
 
3548
    def _process_pending(self, key):
 
3549
        """The content for 'key' was just processed.
 
3550
 
 
3551
        Determine if there is any more pending work to be processed.
 
3552
        """
 
3553
        to_return = []
 
3554
        if key in self._pending_deltas:
 
3555
            compression_parent = key
 
3556
            children = self._pending_deltas.pop(key)
 
3557
            for child_key, parent_keys, record, record_details in children:
 
3558
                lines = self._expand_record(child_key, parent_keys,
 
3559
                                            compression_parent,
 
3560
                                            record, record_details)
 
3561
                if self._check_ready_for_annotations(child_key, parent_keys):
 
3562
                    to_return.append(child_key)
 
3563
        # Also check any children that are waiting for this parent to be
 
3564
        # annotation ready
 
3565
        if key in self._pending_annotation:
 
3566
            children = self._pending_annotation.pop(key)
 
3567
            to_return.extend([c for c, p_keys in children
 
3568
                              if self._check_ready_for_annotations(c, p_keys)])
 
3569
        return to_return
 
3570
 
 
3571
    def _check_ready_for_annotations(self, key, parent_keys):
 
3572
        """return true if this text is ready to be yielded.
 
3573
 
 
3574
        Otherwise, this will return False, and queue the text into
 
3575
        self._pending_annotation
 
3576
        """
 
3577
        for parent_key in parent_keys:
 
3578
            if parent_key not in self._annotations_cache:
 
3579
                # still waiting on at least one parent text, so queue it up
 
3580
                # Note that if there are multiple parents, we need to wait
 
3581
                # for all of them.
 
3582
                self._pending_annotation.setdefault(parent_key,
 
3583
                    []).append((key, parent_keys))
 
3584
                return False
 
3585
        return True
 
3586
 
 
3587
    def _extract_texts(self, records):
 
3588
        """Extract the various texts needed based on records"""
 
3589
        # We iterate in the order read, rather than a strict order requested
 
3590
        # However, process what we can, and put off to the side things that
 
3591
        # still need parents, cleaning them up when those parents are
 
3592
        # processed.
 
3593
        # Basic data flow:
 
3594
        #   1) As 'records' are read, see if we can expand these records into
 
3595
        #      Content objects (and thus lines)
 
3596
        #   2) If a given line-delta is waiting on its compression parent, it
 
3597
        #      gets queued up into self._pending_deltas, otherwise we expand
 
3598
        #      it, and put it into self._text_cache and self._content_objects
 
3599
        #   3) If we expanded the text, we will then check to see if all
 
3600
        #      parents have also been processed. If so, this text gets yielded,
 
3601
        #      else this record gets set aside into pending_annotation
 
3602
        #   4) Further, if we expanded the text in (2), we will then check to
 
3603
        #      see if there are any children in self._pending_deltas waiting to
 
3604
        #      also be processed. If so, we go back to (2) for those
 
3605
        #   5) Further again, if we yielded the text, we can then check if that
 
3606
        #      'unlocks' any of the texts in pending_annotations, which should
 
3607
        #      then get yielded as well
 
3608
        # Note that both steps 4 and 5 are 'recursive' in that unlocking one
 
3609
        # compression child could unlock yet another, and yielding a fulltext
 
3610
        # will also 'unlock' the children that are waiting on that annotation.
 
3611
        # (Though also, unlocking 1 parent's fulltext, does not unlock a child
 
3612
        # if other parents are also waiting.)
 
3613
        # We want to yield content before expanding child content objects, so
 
3614
        # that we know when we can re-use the content lines, and the annotation
 
3615
        # code can know when it can stop caching fulltexts, as well.
 
3616
 
 
3617
        # Children that are missing their compression parent
 
3618
        pending_deltas = {}
 
3619
        for (key, record, digest) in self._vf._read_records_iter(records):
 
3620
            # ghosts?
 
3621
            details = self._all_build_details[key]
 
3622
            (_, compression_parent, parent_keys, record_details) = details
 
3623
            lines = self._expand_record(key, parent_keys, compression_parent,
 
3624
                                        record, record_details)
 
3625
            if lines is None:
 
3626
                # Pending delta should be queued up
 
3627
                continue
 
3628
            # At this point, we may be able to yield this content, if all
 
3629
            # parents are also finished
 
3630
            yield_this_text = self._check_ready_for_annotations(key,
 
3631
                                                                parent_keys)
 
3632
            if yield_this_text:
 
3633
                # All parents present
 
3634
                yield key, lines, len(lines)
 
3635
            to_process = self._process_pending(key)
 
3636
            while to_process:
 
3637
                this_process = to_process
 
3638
                to_process = []
 
3639
                for key in this_process:
 
3640
                    lines = self._text_cache[key]
 
3641
                    yield key, lines, len(lines)
 
3642
                    to_process.extend(self._process_pending(key))
 
3643
 
 
3644
try:
 
3645
    from bzrlib._knit_load_data_pyx import _load_data_c as _load_data
 
3646
except ImportError, e:
 
3647
    osutils.failed_to_load_extension(e)
 
3648
    from bzrlib._knit_load_data_py import _load_data_py as _load_data