← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/knit.py

  • Committer: Sidnei da Silva
  • Date: 2009-07-03 15:06:42 UTC
  • mto: (4531.1.1 integration)
  • mto: This revision was merged to the branch mainline in revision 4532.
  • Revision ID: sidnei.da.silva@canonical.com-20090703150642-hjfra5waj5879cae
- Add top-level make target to build all installers using buildout and another to cleanup

Show diffs side-by-side

added added

removed removed

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