~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/knit.py

  • Committer: John Arbash Meinel
  • Date: 2013-05-19 14:29:37 UTC
  • mfrom: (6437.63.9 2.5)
  • mto: (6437.63.10 2.5)
  • mto: This revision was merged to the branch mainline in revision 6575.
  • Revision ID: john@arbash-meinel.com-20130519142937-21ykz2n2y2f22za9
Merge in the actual 2.5 branch. It seems I failed before

Show diffs side-by-side

added added

removed removed

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