← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/knit.py

  • Committer: Vincent Ladeuil
  • Date: 2010-02-09 17:27:46 UTC
  • mto: (5029.1.1 integration)
  • mto: This revision was merged to the branch mainline in revision 5030.
  • Revision ID: v.ladeuil+lp@free.fr-20100209172746-6f4mvlnr2mac807j
Move NoSmartTransportServer to bzrlib.tests.test_server

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