← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/knit.py

  • Committer: John Arbash Meinel
  • Date: 2009-08-25 18:45:40 UTC
  • mto: (4634.6.15 2.0)
  • mto: This revision was merged to the branch mainline in revision 4667.
  • Revision ID: john@arbash-meinel.com-20090825184540-6dn3xjq62xhgj2gq
Add support for skipping ghost nodes.

Show diffs side-by-side

added added

removed removed

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