~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/knit.py

  • Committer: Patch Queue Manager
  • Date: 2012-10-25 11:13:27 UTC
  • mfrom: (6570.1.6 rubberstamp)
  • Revision ID: pqm@pqm.ubuntu.com-20121025111327-p0ylql0nh9fla0rs
(gz) Set approved revision and vote "Approve" when using lp-propose
 --approve (Jonathan Lange)

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2006-2010 Canonical Ltd
 
1
# Copyright (C) 2006-2011 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
51
51
 
52
52
"""
53
53
 
 
54
from __future__ import absolute_import
 
55
 
54
56
 
55
57
from cStringIO import StringIO
56
58
from itertools import izip
57
59
import operator
58
60
import os
59
 
import sys
60
61
 
61
62
from bzrlib.lazy_import import lazy_import
62
63
lazy_import(globals(), """
 
64
import gzip
 
65
 
63
66
from bzrlib import (
64
 
    annotate,
65
67
    debug,
66
68
    diff,
67
69
    graph as _mod_graph,
68
70
    index as _mod_index,
69
 
    lru_cache,
70
71
    pack,
71
72
    patiencediff,
72
 
    progress,
73
73
    static_tuple,
74
74
    trace,
75
75
    tsort,
76
76
    tuned_gzip,
77
77
    ui,
78
78
    )
 
79
 
 
80
from bzrlib.repofmt import pack_repo
 
81
from bzrlib.i18n import gettext
79
82
""")
80
83
from bzrlib import (
 
84
    annotate,
81
85
    errors,
82
86
    osutils,
83
87
    )
84
88
from bzrlib.errors import (
85
 
    FileExists,
86
89
    NoSuchFile,
87
 
    KnitError,
88
90
    InvalidRevisionId,
89
91
    KnitCorrupt,
90
92
    KnitHeaderError,
91
93
    RevisionNotPresent,
92
 
    RevisionAlreadyPresent,
93
94
    SHA1KnitCorrupt,
94
95
    )
95
96
from bzrlib.osutils import (
96
97
    contains_whitespace,
97
 
    contains_linebreaks,
98
98
    sha_string,
99
99
    sha_strings,
100
100
    split_lines,
101
101
    )
102
102
from bzrlib.versionedfile import (
 
103
    _KeyRefs,
103
104
    AbsentContentFactory,
104
105
    adapter_registry,
105
106
    ConstantMapper,
106
107
    ContentFactory,
107
 
    ChunkedContentFactory,
108
108
    sort_groupcompress,
109
 
    VersionedFile,
110
 
    VersionedFiles,
 
109
    VersionedFilesWithFallbacks,
111
110
    )
112
111
 
113
112
 
412
411
class KnitContent(object):
413
412
    """Content of a knit version to which deltas can be applied.
414
413
 
415
 
    This is always stored in memory as a list of lines with \n at the end,
 
414
    This is always stored in memory as a list of lines with \\n at the end,
416
415
    plus a flag saying if the final ending is really there or not, because that
417
416
    corresponds to the on-disk knit representation.
418
417
    """
805
804
        writer.begin()
806
805
        index = _KnitGraphIndex(graph_index, lambda:True, parents=parents,
807
806
            deltas=delta, add_callback=graph_index.add_nodes)
808
 
        access = _DirectPackAccess({})
 
807
        access = pack_repo._DirectPackAccess({})
809
808
        access.set_writer(writer, graph_index, (transport, 'newpack'))
810
809
        result = KnitVersionedFiles(index, access,
811
810
            max_delta_chain=max_delta_chain)
849
848
                in all_build_index_memos.itervalues()])
850
849
 
851
850
 
852
 
class KnitVersionedFiles(VersionedFiles):
 
851
class KnitVersionedFiles(VersionedFilesWithFallbacks):
853
852
    """Storage for many versioned files using knit compression.
854
853
 
855
854
    Backend storage is managed by indices and data objects.
882
881
            self._factory = KnitAnnotateFactory()
883
882
        else:
884
883
            self._factory = KnitPlainFactory()
885
 
        self._fallback_vfs = []
 
884
        self._immediate_fallback_vfs = []
886
885
        self._reload_func = reload_func
887
886
 
888
887
    def __repr__(self):
891
890
            self._index,
892
891
            self._access)
893
892
 
 
893
    def without_fallbacks(self):
 
894
        """Return a clone of this object without any fallbacks configured."""
 
895
        return KnitVersionedFiles(self._index, self._access,
 
896
            self._max_delta_chain, self._factory.annotated,
 
897
            self._reload_func)
 
898
 
894
899
    def add_fallback_versioned_files(self, a_versioned_files):
895
900
        """Add a source of texts for texts not present in this knit.
896
901
 
897
902
        :param a_versioned_files: A VersionedFiles object.
898
903
        """
899
 
        self._fallback_vfs.append(a_versioned_files)
 
904
        self._immediate_fallback_vfs.append(a_versioned_files)
900
905
 
901
906
    def add_lines(self, key, parents, lines, parent_texts=None,
902
907
        left_matching_blocks=None, nostore_sha=None, random_id=False,
1069
1074
                    raise errors.KnitCorrupt(self,
1070
1075
                        "Missing basis parent %s for %s" % (
1071
1076
                        compression_parent, key))
1072
 
        for fallback_vfs in self._fallback_vfs:
 
1077
        for fallback_vfs in self._immediate_fallback_vfs:
1073
1078
            fallback_vfs.check()
1074
1079
 
1075
1080
    def _check_add(self, key, lines, random_id, check_content):
1153
1158
 
1154
1159
        A dict of key to (record_details, index_memo, next, parents) is
1155
1160
        returned.
1156
 
        method is the way referenced data should be applied.
1157
 
        index_memo is the handle to pass to the data access to actually get the
1158
 
            data
1159
 
        next is the build-parent of the version, or None for fulltexts.
1160
 
        parents is the version_ids of the parents of this version
1161
 
 
1162
 
        :param allow_missing: If True do not raise an error on a missing component,
1163
 
            just ignore it.
 
1161
 
 
1162
        * method is the way referenced data should be applied.
 
1163
        * index_memo is the handle to pass to the data access to actually get
 
1164
          the data
 
1165
        * next is the build-parent of the version, or None for fulltexts.
 
1166
        * parents is the version_ids of the parents of this version
 
1167
 
 
1168
        :param allow_missing: If True do not raise an error on a missing
 
1169
            component, just ignore it.
1164
1170
        """
1165
1171
        component_data = {}
1166
1172
        pending_components = keys
1192
1198
        generator = _VFContentMapGenerator(self, [key])
1193
1199
        return generator._get_content(key)
1194
1200
 
1195
 
    def get_known_graph_ancestry(self, keys):
1196
 
        """Get a KnownGraph instance with the ancestry of keys."""
1197
 
        parent_map, missing_keys = self._index.find_ancestry(keys)
1198
 
        for fallback in self._fallback_vfs:
1199
 
            if not missing_keys:
1200
 
                break
1201
 
            (f_parent_map, f_missing_keys) = fallback._index.find_ancestry(
1202
 
                                                missing_keys)
1203
 
            parent_map.update(f_parent_map)
1204
 
            missing_keys = f_missing_keys
1205
 
        kg = _mod_graph.KnownGraph(parent_map)
1206
 
        return kg
1207
 
 
1208
1201
    def get_parent_map(self, keys):
1209
1202
        """Get a map of the graph parents of keys.
1210
1203
 
1225
1218
            and so on.
1226
1219
        """
1227
1220
        result = {}
1228
 
        sources = [self._index] + self._fallback_vfs
 
1221
        sources = [self._index] + self._immediate_fallback_vfs
1229
1222
        source_results = []
1230
1223
        missing = set(keys)
1231
1224
        for source in sources:
1241
1234
        """Produce a dictionary of knit records.
1242
1235
 
1243
1236
        :return: {key:(record, record_details, digest, next)}
1244
 
            record
1245
 
                data returned from read_records (a KnitContentobject)
1246
 
            record_details
1247
 
                opaque information to pass to parse_record
1248
 
            digest
1249
 
                SHA1 digest of the full text after all steps are done
1250
 
            next
1251
 
                build-parent of the version, i.e. the leftmost ancestor.
 
1237
 
 
1238
            * record: data returned from read_records (a KnitContentobject)
 
1239
            * record_details: opaque information to pass to parse_record
 
1240
            * digest: SHA1 digest of the full text after all steps are done
 
1241
            * next: build-parent of the version, i.e. the leftmost ancestor.
1252
1242
                Will be None if the record is not a delta.
 
1243
 
1253
1244
        :param keys: The keys to build a map for
1254
1245
        :param allow_missing: If some records are missing, rather than
1255
1246
            error, just return the data that could be generated.
1525
1516
                        yield KnitContentFactory(key, global_map[key],
1526
1517
                            record_details, None, raw_data, self._factory.annotated, None)
1527
1518
                else:
1528
 
                    vf = self._fallback_vfs[parent_maps.index(source) - 1]
 
1519
                    vf = self._immediate_fallback_vfs[parent_maps.index(source) - 1]
1529
1520
                    for record in vf.get_record_stream(keys, ordering,
1530
1521
                        include_delta_closure):
1531
1522
                        yield record
1541
1532
            # record entry 2 is the 'digest'.
1542
1533
            result[key] = details[2]
1543
1534
        missing.difference_update(set(result))
1544
 
        for source in self._fallback_vfs:
 
1535
        for source in self._immediate_fallback_vfs:
1545
1536
            if not missing:
1546
1537
                break
1547
1538
            new_result = source.get_sha1s(missing)
1618
1609
                raise RevisionNotPresent([record.key], self)
1619
1610
            elif ((record.storage_kind in knit_types)
1620
1611
                  and (compression_parent is None
1621
 
                       or not self._fallback_vfs
 
1612
                       or not self._immediate_fallback_vfs
1622
1613
                       or self._index.has_key(compression_parent)
1623
1614
                       or not self.has_key(compression_parent))):
1624
1615
                # we can insert the knit record literally if either it has no
1772
1763
                        key_records.append((key, details[0]))
1773
1764
                records_iter = enumerate(self._read_records_iter(key_records))
1774
1765
                for (key_idx, (key, data, sha_value)) in records_iter:
1775
 
                    pb.update('Walking content', key_idx, total)
 
1766
                    pb.update(gettext('Walking content'), key_idx, total)
1776
1767
                    compression_parent = build_details[key][1]
1777
1768
                    if compression_parent is None:
1778
1769
                        # fulltext
1796
1787
        # vfs, and hope to find them there.  Note that if the keys are found
1797
1788
        # but had no changes or no content, the fallback may not return
1798
1789
        # anything.
1799
 
        if keys and not self._fallback_vfs:
 
1790
        if keys and not self._immediate_fallback_vfs:
1800
1791
            # XXX: strictly the second parameter is meant to be the file id
1801
1792
            # but it's not easily accessible here.
1802
1793
            raise RevisionNotPresent(keys, repr(self))
1803
 
        for source in self._fallback_vfs:
 
1794
        for source in self._immediate_fallback_vfs:
1804
1795
            if not keys:
1805
1796
                break
1806
1797
            source_keys = set()
1808
1799
                source_keys.add(key)
1809
1800
                yield line, key
1810
1801
            keys.difference_update(source_keys)
1811
 
        pb.update('Walking content', total, total)
 
1802
        pb.update(gettext('Walking content'), total, total)
1812
1803
 
1813
1804
    def _make_line_delta(self, delta_seq, new_content):
1814
1805
        """Generate a line delta from delta_seq and new_content."""
1879
1870
        :return: the header and the decompressor stream.
1880
1871
                 as (stream, header_record)
1881
1872
        """
1882
 
        df = tuned_gzip.GzipFile(mode='rb', fileobj=StringIO(raw_data))
 
1873
        df = gzip.GzipFile(mode='rb', fileobj=StringIO(raw_data))
1883
1874
        try:
1884
1875
            # Current serialise
1885
1876
            rec = self._check_header(key, df.readline())
1894
1885
        # 4168 calls in 2880 217 internal
1895
1886
        # 4168 calls to _parse_record_header in 2121
1896
1887
        # 4168 calls to readlines in 330
1897
 
        df = tuned_gzip.GzipFile(mode='rb', fileobj=StringIO(data))
 
1888
        df = gzip.GzipFile(mode='rb', fileobj=StringIO(data))
1898
1889
        try:
1899
1890
            record_contents = df.readlines()
1900
1891
        except Exception, e:
1922
1913
        The result will be returned in whatever is the fastest to read.
1923
1914
        Not by the order requested. Also, multiple requests for the same
1924
1915
        record will only yield 1 response.
 
1916
 
1925
1917
        :param records: A list of (key, access_memo) entries
1926
1918
        :return: Yields (key, contents, digest) in the order
1927
1919
                 read, not the order requested
1985
1977
        :param key: The key of the record. Currently keys are always serialised
1986
1978
            using just the trailing component.
1987
1979
        :param dense_lines: The bytes of lines but in a denser form. For
1988
 
            instance, if lines is a list of 1000 bytestrings each ending in \n,
1989
 
            dense_lines may be a list with one line in it, containing all the
1990
 
            1000's lines and their \n's. Using dense_lines if it is already
1991
 
            known is a win because the string join to create bytes in this
1992
 
            function spends less time resizing the final string.
 
1980
            instance, if lines is a list of 1000 bytestrings each ending in
 
1981
            \\n, dense_lines may be a list with one line in it, containing all
 
1982
            the 1000's lines and their \\n's. Using dense_lines if it is
 
1983
            already known is a win because the string join to create bytes in
 
1984
            this function spends less time resizing the final string.
1993
1985
        :return: (len, a StringIO instance with the raw data ready to read.)
1994
1986
        """
1995
1987
        chunks = ["version %s %d %s\n" % (key[-1], len(lines), digest)]
2015
2007
        """See VersionedFiles.keys."""
2016
2008
        if 'evil' in debug.debug_flags:
2017
2009
            trace.mutter_callsite(2, "keys scales with size of history")
2018
 
        sources = [self._index] + self._fallback_vfs
 
2010
        sources = [self._index] + self._immediate_fallback_vfs
2019
2011
        result = set()
2020
2012
        for source in sources:
2021
2013
            result.update(source.keys())
2061
2053
 
2062
2054
        missing_keys = set(nonlocal_keys)
2063
2055
        # Read from remote versioned file instances and provide to our caller.
2064
 
        for source in self.vf._fallback_vfs:
 
2056
        for source in self.vf._immediate_fallback_vfs:
2065
2057
            if not missing_keys:
2066
2058
                break
2067
2059
            # Loop over fallback repositories asking them for texts - ignore
2786
2778
        return key[:-1], key[-1]
2787
2779
 
2788
2780
 
2789
 
class _KeyRefs(object):
2790
 
 
2791
 
    def __init__(self, track_new_keys=False):
2792
 
        # dict mapping 'key' to 'set of keys referring to that key'
2793
 
        self.refs = {}
2794
 
        if track_new_keys:
2795
 
            # set remembering all new keys
2796
 
            self.new_keys = set()
2797
 
        else:
2798
 
            self.new_keys = None
2799
 
 
2800
 
    def clear(self):
2801
 
        if self.refs:
2802
 
            self.refs.clear()
2803
 
        if self.new_keys:
2804
 
            self.new_keys.clear()
2805
 
 
2806
 
    def add_references(self, key, refs):
2807
 
        # Record the new references
2808
 
        for referenced in refs:
2809
 
            try:
2810
 
                needed_by = self.refs[referenced]
2811
 
            except KeyError:
2812
 
                needed_by = self.refs[referenced] = set()
2813
 
            needed_by.add(key)
2814
 
        # Discard references satisfied by the new key
2815
 
        self.add_key(key)
2816
 
 
2817
 
    def get_new_keys(self):
2818
 
        return self.new_keys
2819
 
    
2820
 
    def get_unsatisfied_refs(self):
2821
 
        return self.refs.iterkeys()
2822
 
 
2823
 
    def _satisfy_refs_for_key(self, key):
2824
 
        try:
2825
 
            del self.refs[key]
2826
 
        except KeyError:
2827
 
            # No keys depended on this key.  That's ok.
2828
 
            pass
2829
 
 
2830
 
    def add_key(self, key):
2831
 
        # satisfy refs for key, and remember that we've seen this key.
2832
 
        self._satisfy_refs_for_key(key)
2833
 
        if self.new_keys is not None:
2834
 
            self.new_keys.add(key)
2835
 
 
2836
 
    def satisfy_refs_for_keys(self, keys):
2837
 
        for key in keys:
2838
 
            self._satisfy_refs_for_key(key)
2839
 
 
2840
 
    def get_referrers(self):
2841
 
        result = set()
2842
 
        for referrers in self.refs.itervalues():
2843
 
            result.update(referrers)
2844
 
        return result
2845
 
 
2846
 
 
2847
2781
class _KnitGraphIndex(object):
2848
2782
    """A KnitVersionedFiles index layered on GraphIndex."""
2849
2783
 
3278
3212
                yield data
3279
3213
 
3280
3214
 
3281
 
class _DirectPackAccess(object):
3282
 
    """Access to data in one or more packs with less translation."""
3283
 
 
3284
 
    def __init__(self, index_to_packs, reload_func=None, flush_func=None):
3285
 
        """Create a _DirectPackAccess object.
3286
 
 
3287
 
        :param index_to_packs: A dict mapping index objects to the transport
3288
 
            and file names for obtaining data.
3289
 
        :param reload_func: A function to call if we determine that the pack
3290
 
            files have moved and we need to reload our caches. See
3291
 
            bzrlib.repo_fmt.pack_repo.AggregateIndex for more details.
3292
 
        """
3293
 
        self._container_writer = None
3294
 
        self._write_index = None
3295
 
        self._indices = index_to_packs
3296
 
        self._reload_func = reload_func
3297
 
        self._flush_func = flush_func
3298
 
 
3299
 
    def add_raw_records(self, key_sizes, raw_data):
3300
 
        """Add raw knit bytes to a storage area.
3301
 
 
3302
 
        The data is spooled to the container writer in one bytes-record per
3303
 
        raw data item.
3304
 
 
3305
 
        :param sizes: An iterable of tuples containing the key and size of each
3306
 
            raw data segment.
3307
 
        :param raw_data: A bytestring containing the data.
3308
 
        :return: A list of memos to retrieve the record later. Each memo is an
3309
 
            opaque index memo. For _DirectPackAccess the memo is (index, pos,
3310
 
            length), where the index field is the write_index object supplied
3311
 
            to the PackAccess object.
3312
 
        """
3313
 
        if type(raw_data) is not str:
3314
 
            raise AssertionError(
3315
 
                'data must be plain bytes was %s' % type(raw_data))
3316
 
        result = []
3317
 
        offset = 0
3318
 
        for key, size in key_sizes:
3319
 
            p_offset, p_length = self._container_writer.add_bytes_record(
3320
 
                raw_data[offset:offset+size], [])
3321
 
            offset += size
3322
 
            result.append((self._write_index, p_offset, p_length))
3323
 
        return result
3324
 
 
3325
 
    def flush(self):
3326
 
        """Flush pending writes on this access object.
3327
 
 
3328
 
        This will flush any buffered writes to a NewPack.
3329
 
        """
3330
 
        if self._flush_func is not None:
3331
 
            self._flush_func()
3332
 
            
3333
 
    def get_raw_records(self, memos_for_retrieval):
3334
 
        """Get the raw bytes for a records.
3335
 
 
3336
 
        :param memos_for_retrieval: An iterable containing the (index, pos,
3337
 
            length) memo for retrieving the bytes. The Pack access method
3338
 
            looks up the pack to use for a given record in its index_to_pack
3339
 
            map.
3340
 
        :return: An iterator over the bytes of the records.
3341
 
        """
3342
 
        # first pass, group into same-index requests
3343
 
        request_lists = []
3344
 
        current_index = None
3345
 
        for (index, offset, length) in memos_for_retrieval:
3346
 
            if current_index == index:
3347
 
                current_list.append((offset, length))
3348
 
            else:
3349
 
                if current_index is not None:
3350
 
                    request_lists.append((current_index, current_list))
3351
 
                current_index = index
3352
 
                current_list = [(offset, length)]
3353
 
        # handle the last entry
3354
 
        if current_index is not None:
3355
 
            request_lists.append((current_index, current_list))
3356
 
        for index, offsets in request_lists:
3357
 
            try:
3358
 
                transport, path = self._indices[index]
3359
 
            except KeyError:
3360
 
                # A KeyError here indicates that someone has triggered an index
3361
 
                # reload, and this index has gone missing, we need to start
3362
 
                # over.
3363
 
                if self._reload_func is None:
3364
 
                    # If we don't have a _reload_func there is nothing that can
3365
 
                    # be done
3366
 
                    raise
3367
 
                raise errors.RetryWithNewPacks(index,
3368
 
                                               reload_occurred=True,
3369
 
                                               exc_info=sys.exc_info())
3370
 
            try:
3371
 
                reader = pack.make_readv_reader(transport, path, offsets)
3372
 
                for names, read_func in reader.iter_records():
3373
 
                    yield read_func(None)
3374
 
            except errors.NoSuchFile:
3375
 
                # A NoSuchFile error indicates that a pack file has gone
3376
 
                # missing on disk, we need to trigger a reload, and start over.
3377
 
                if self._reload_func is None:
3378
 
                    raise
3379
 
                raise errors.RetryWithNewPacks(transport.abspath(path),
3380
 
                                               reload_occurred=False,
3381
 
                                               exc_info=sys.exc_info())
3382
 
 
3383
 
    def set_writer(self, writer, index, transport_packname):
3384
 
        """Set a writer to use for adding data."""
3385
 
        if index is not None:
3386
 
            self._indices[index] = transport_packname
3387
 
        self._container_writer = writer
3388
 
        self._write_index = index
3389
 
 
3390
 
    def reload_or_raise(self, retry_exc):
3391
 
        """Try calling the reload function, or re-raise the original exception.
3392
 
 
3393
 
        This should be called after _DirectPackAccess raises a
3394
 
        RetryWithNewPacks exception. This function will handle the common logic
3395
 
        of determining when the error is fatal versus being temporary.
3396
 
        It will also make sure that the original exception is raised, rather
3397
 
        than the RetryWithNewPacks exception.
3398
 
 
3399
 
        If this function returns, then the calling function should retry
3400
 
        whatever operation was being performed. Otherwise an exception will
3401
 
        be raised.
3402
 
 
3403
 
        :param retry_exc: A RetryWithNewPacks exception.
3404
 
        """
3405
 
        is_error = False
3406
 
        if self._reload_func is None:
3407
 
            is_error = True
3408
 
        elif not self._reload_func():
3409
 
            # The reload claimed that nothing changed
3410
 
            if not retry_exc.reload_occurred:
3411
 
                # If there wasn't an earlier reload, then we really were
3412
 
                # expecting to find changes. We didn't find them, so this is a
3413
 
                # hard error
3414
 
                is_error = True
3415
 
        if is_error:
3416
 
            exc_class, exc_value, exc_traceback = retry_exc.exc_info
3417
 
            raise exc_class, exc_value, exc_traceback
3418
 
 
3419
 
 
3420
3215
def annotate_knit(knit, revision_id):
3421
3216
    """Annotate a knit with no cached annotations.
3422
3217
 
3520
3315
        return records, ann_keys
3521
3316
 
3522
3317
    def _get_needed_texts(self, key, pb=None):
3523
 
        # if True or len(self._vf._fallback_vfs) > 0:
3524
 
        if len(self._vf._fallback_vfs) > 0:
 
3318
        # if True or len(self._vf._immediate_fallback_vfs) > 0:
 
3319
        if len(self._vf._immediate_fallback_vfs) > 0:
3525
3320
            # If we have fallbacks, go to the generic path
3526
3321
            for v in annotate.Annotator._get_needed_texts(self, key, pb=pb):
3527
3322
                yield v
3532
3327
                for idx, (sub_key, text, num_lines) in enumerate(
3533
3328
                                                self._extract_texts(records)):
3534
3329
                    if pb is not None:
3535
 
                        pb.update('annotating', idx, len(records))
 
3330
                        pb.update(gettext('annotating'), idx, len(records))
3536
3331
                    yield sub_key, text, num_lines
3537
3332
                for sub_key in ann_keys:
3538
3333
                    text = self._text_cache[sub_key]