/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/repository.py

  • Committer: Robert Collins
  • Date: 2010-05-06 11:08:10 UTC
  • mto: This revision was merged to the branch mainline in revision 5223.
  • Revision ID: robertc@robertcollins.net-20100506110810-h3j07fh5gmw54s25
Cleaner matcher matching revised unlocking protocol.

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2005, 2006, 2007, 2008, 2009 Canonical Ltd
 
1
# Copyright (C) 2005-2010 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
24
24
    bzrdir,
25
25
    check,
26
26
    chk_map,
 
27
    config,
27
28
    debug,
28
29
    errors,
 
30
    fetch as _mod_fetch,
29
31
    fifo_cache,
30
32
    generate_ids,
31
33
    gpg,
32
34
    graph,
33
35
    inventory,
 
36
    inventory_delta,
34
37
    lazy_regex,
35
38
    lockable_files,
36
39
    lockdir,
37
40
    lru_cache,
38
41
    osutils,
39
42
    revision as _mod_revision,
 
43
    static_tuple,
40
44
    symbol_versioning,
 
45
    trace,
41
46
    tsort,
42
47
    ui,
43
48
    versionedfile,
48
53
from bzrlib.testament import Testament
49
54
""")
50
55
 
51
 
from bzrlib.decorators import needs_read_lock, needs_write_lock
 
56
from bzrlib.decorators import needs_read_lock, needs_write_lock, only_raises
52
57
from bzrlib.inter import InterObject
53
58
from bzrlib.inventory import (
54
59
    Inventory,
56
61
    ROOT_ID,
57
62
    entry_factory,
58
63
    )
 
64
from bzrlib.lock import _RelockDebugMixin
59
65
from bzrlib import registry
60
66
from bzrlib.trace import (
61
67
    log_exception_quietly, note, mutter, mutter_callsite, warning)
204
210
            # an inventory delta was accumulated without creating a new
205
211
            # inventory.
206
212
            basis_id = self.basis_delta_revision
207
 
            self.inv_sha1 = self.repository.add_inventory_by_delta(
 
213
            # We ignore the 'inventory' returned by add_inventory_by_delta
 
214
            # because self.new_inventory is used to hint to the rest of the
 
215
            # system what code path was taken
 
216
            self.inv_sha1, _ = self.repository.add_inventory_by_delta(
208
217
                basis_id, self._basis_delta, self._new_revision_id,
209
218
                self.parents)
210
219
        else:
464
473
            if content_summary[2] is None:
465
474
                raise ValueError("Files must not have executable = None")
466
475
            if not store:
467
 
                if (# if the file length changed we have to store:
468
 
                    parent_entry.text_size != content_summary[1] or
469
 
                    # if the exec bit has changed we have to store:
 
476
                # We can't trust a check of the file length because of content
 
477
                # filtering...
 
478
                if (# if the exec bit has changed we have to store:
470
479
                    parent_entry.executable != content_summary[2]):
471
480
                    store = True
472
481
                elif parent_entry.text_sha1 == content_summary[3]:
539
548
                ie.revision = parent_entry.revision
540
549
                return self._get_delta(ie, basis_inv, path), False, None
541
550
            ie.reference_revision = content_summary[3]
 
551
            if ie.reference_revision is None:
 
552
                raise AssertionError("invalid content_summary for nested tree: %r"
 
553
                    % (content_summary,))
542
554
            self._add_text_to_weave(ie.file_id, '', heads, None)
543
555
        else:
544
556
            raise NotImplementedError('unknown kind')
806
818
                seen_root = True
807
819
        self.new_inventory = None
808
820
        if len(inv_delta):
 
821
            # This should perhaps be guarded by a check that the basis we
 
822
            # commit against is the basis for the commit and if not do a delta
 
823
            # against the basis.
809
824
            self._any_changes = True
810
825
        if not seen_root:
811
826
            # housekeeping root entry changes do not affect no-change commits.
849
864
# Repositories
850
865
 
851
866
 
852
 
class Repository(object):
 
867
class Repository(_RelockDebugMixin, bzrdir.ControlComponent):
853
868
    """Repository holding history for one or more branches.
854
869
 
855
870
    The repository holds and retrieves historical information including
924
939
        """
925
940
        if self._write_group is not self.get_transaction():
926
941
            # has an unlock or relock occured ?
 
942
            if suppress_errors:
 
943
                mutter(
 
944
                '(suppressed) mismatched lock context and write group. %r, %r',
 
945
                self._write_group, self.get_transaction())
 
946
                return
927
947
            raise errors.BzrError(
928
948
                'mismatched lock context and write group. %r, %r' %
929
949
                (self._write_group, self.get_transaction()))
1008
1028
 
1009
1029
        :seealso: add_inventory, for the contract.
1010
1030
        """
1011
 
        inv_lines = self._serialise_inventory_to_lines(inv)
 
1031
        inv_lines = self._serializer.write_inventory_to_lines(inv)
1012
1032
        return self._inventory_add_lines(revision_id, parents,
1013
1033
            inv_lines, check_content=False)
1014
1034
 
1063
1083
        check_content=True):
1064
1084
        """Store lines in inv_vf and return the sha1 of the inventory."""
1065
1085
        parents = [(parent,) for parent in parents]
1066
 
        return self.inventories.add_lines((revision_id,), parents, lines,
 
1086
        result = self.inventories.add_lines((revision_id,), parents, lines,
1067
1087
            check_content=check_content)[0]
 
1088
        self.inventories._access.flush()
 
1089
        return result
1068
1090
 
1069
1091
    def add_revision(self, revision_id, rev, inv=None, config=None):
1070
1092
        """Add rev to the revision store as revision_id.
1146
1168
        # The old API returned a list, should this actually be a set?
1147
1169
        return parent_map.keys()
1148
1170
 
 
1171
    def _check_inventories(self, checker):
 
1172
        """Check the inventories found from the revision scan.
 
1173
        
 
1174
        This is responsible for verifying the sha1 of inventories and
 
1175
        creating a pending_keys set that covers data referenced by inventories.
 
1176
        """
 
1177
        bar = ui.ui_factory.nested_progress_bar()
 
1178
        try:
 
1179
            self._do_check_inventories(checker, bar)
 
1180
        finally:
 
1181
            bar.finished()
 
1182
 
 
1183
    def _do_check_inventories(self, checker, bar):
 
1184
        """Helper for _check_inventories."""
 
1185
        revno = 0
 
1186
        keys = {'chk_bytes':set(), 'inventories':set(), 'texts':set()}
 
1187
        kinds = ['chk_bytes', 'texts']
 
1188
        count = len(checker.pending_keys)
 
1189
        bar.update("inventories", 0, 2)
 
1190
        current_keys = checker.pending_keys
 
1191
        checker.pending_keys = {}
 
1192
        # Accumulate current checks.
 
1193
        for key in current_keys:
 
1194
            if key[0] != 'inventories' and key[0] not in kinds:
 
1195
                checker._report_items.append('unknown key type %r' % (key,))
 
1196
            keys[key[0]].add(key[1:])
 
1197
        if keys['inventories']:
 
1198
            # NB: output order *should* be roughly sorted - topo or
 
1199
            # inverse topo depending on repository - either way decent
 
1200
            # to just delta against. However, pre-CHK formats didn't
 
1201
            # try to optimise inventory layout on disk. As such the
 
1202
            # pre-CHK code path does not use inventory deltas.
 
1203
            last_object = None
 
1204
            for record in self.inventories.check(keys=keys['inventories']):
 
1205
                if record.storage_kind == 'absent':
 
1206
                    checker._report_items.append(
 
1207
                        'Missing inventory {%s}' % (record.key,))
 
1208
                else:
 
1209
                    last_object = self._check_record('inventories', record,
 
1210
                        checker, last_object,
 
1211
                        current_keys[('inventories',) + record.key])
 
1212
            del keys['inventories']
 
1213
        else:
 
1214
            return
 
1215
        bar.update("texts", 1)
 
1216
        while (checker.pending_keys or keys['chk_bytes']
 
1217
            or keys['texts']):
 
1218
            # Something to check.
 
1219
            current_keys = checker.pending_keys
 
1220
            checker.pending_keys = {}
 
1221
            # Accumulate current checks.
 
1222
            for key in current_keys:
 
1223
                if key[0] not in kinds:
 
1224
                    checker._report_items.append('unknown key type %r' % (key,))
 
1225
                keys[key[0]].add(key[1:])
 
1226
            # Check the outermost kind only - inventories || chk_bytes || texts
 
1227
            for kind in kinds:
 
1228
                if keys[kind]:
 
1229
                    last_object = None
 
1230
                    for record in getattr(self, kind).check(keys=keys[kind]):
 
1231
                        if record.storage_kind == 'absent':
 
1232
                            checker._report_items.append(
 
1233
                                'Missing %s {%s}' % (kind, record.key,))
 
1234
                        else:
 
1235
                            last_object = self._check_record(kind, record,
 
1236
                                checker, last_object, current_keys[(kind,) + record.key])
 
1237
                    keys[kind] = set()
 
1238
                    break
 
1239
 
 
1240
    def _check_record(self, kind, record, checker, last_object, item_data):
 
1241
        """Check a single text from this repository."""
 
1242
        if kind == 'inventories':
 
1243
            rev_id = record.key[0]
 
1244
            inv = self._deserialise_inventory(rev_id,
 
1245
                record.get_bytes_as('fulltext'))
 
1246
            if last_object is not None:
 
1247
                delta = inv._make_delta(last_object)
 
1248
                for old_path, path, file_id, ie in delta:
 
1249
                    if ie is None:
 
1250
                        continue
 
1251
                    ie.check(checker, rev_id, inv)
 
1252
            else:
 
1253
                for path, ie in inv.iter_entries():
 
1254
                    ie.check(checker, rev_id, inv)
 
1255
            if self._format.fast_deltas:
 
1256
                return inv
 
1257
        elif kind == 'chk_bytes':
 
1258
            # No code written to check chk_bytes for this repo format.
 
1259
            checker._report_items.append(
 
1260
                'unsupported key type chk_bytes for %s' % (record.key,))
 
1261
        elif kind == 'texts':
 
1262
            self._check_text(record, checker, item_data)
 
1263
        else:
 
1264
            checker._report_items.append(
 
1265
                'unknown key type %s for %s' % (kind, record.key))
 
1266
 
 
1267
    def _check_text(self, record, checker, item_data):
 
1268
        """Check a single text."""
 
1269
        # Check it is extractable.
 
1270
        # TODO: check length.
 
1271
        if record.storage_kind == 'chunked':
 
1272
            chunks = record.get_bytes_as(record.storage_kind)
 
1273
            sha1 = osutils.sha_strings(chunks)
 
1274
            length = sum(map(len, chunks))
 
1275
        else:
 
1276
            content = record.get_bytes_as('fulltext')
 
1277
            sha1 = osutils.sha_string(content)
 
1278
            length = len(content)
 
1279
        if item_data and sha1 != item_data[1]:
 
1280
            checker._report_items.append(
 
1281
                'sha1 mismatch: %s has sha1 %s expected %s referenced by %s' %
 
1282
                (record.key, sha1, item_data[1], item_data[2]))
 
1283
 
1149
1284
    @staticmethod
1150
1285
    def create(a_bzrdir):
1151
1286
        """Construct the current default format repository in a_bzrdir."""
1156
1291
 
1157
1292
        :param _format: The format of the repository on disk.
1158
1293
        :param a_bzrdir: The BzrDir of the repository.
1159
 
 
1160
 
        In the future we will have a single api for all stores for
1161
 
        getting file texts, inventories and revisions, then
1162
 
        this construct will accept instances of those things.
1163
1294
        """
 
1295
        # In the future we will have a single api for all stores for
 
1296
        # getting file texts, inventories and revisions, then
 
1297
        # this construct will accept instances of those things.
1164
1298
        super(Repository, self).__init__()
1165
1299
        self._format = _format
1166
1300
        # the following are part of the public API for Repository:
1172
1306
        self._reconcile_does_inventory_gc = True
1173
1307
        self._reconcile_fixes_text_parents = False
1174
1308
        self._reconcile_backsup_inventory = True
1175
 
        # not right yet - should be more semantically clear ?
1176
 
        #
1177
 
        # TODO: make sure to construct the right store classes, etc, depending
1178
 
        # on whether escaping is required.
1179
 
        self._warn_if_deprecated()
1180
1309
        self._write_group = None
1181
1310
        # Additional places to query for data.
1182
1311
        self._fallback_repositories = []
1183
1312
        # An InventoryEntry cache, used during deserialization
1184
1313
        self._inventory_entry_cache = fifo_cache.FIFOCache(10*1024)
 
1314
        # Is it safe to return inventory entries directly from the entry cache,
 
1315
        # rather copying them?
 
1316
        self._safe_to_return_from_cache = False
 
1317
 
 
1318
    @property
 
1319
    def user_transport(self):
 
1320
        return self.bzrdir.user_transport
 
1321
 
 
1322
    @property
 
1323
    def control_transport(self):
 
1324
        return self._transport
1185
1325
 
1186
1326
    def __repr__(self):
1187
1327
        if self._fallback_repositories:
1254
1394
        locked = self.is_locked()
1255
1395
        result = self.control_files.lock_write(token=token)
1256
1396
        if not locked:
 
1397
            self._warn_if_deprecated()
 
1398
            self._note_lock('w')
1257
1399
            for repo in self._fallback_repositories:
1258
1400
                # Writes don't affect fallback repos
1259
1401
                repo.lock_read()
1264
1406
        locked = self.is_locked()
1265
1407
        self.control_files.lock_read()
1266
1408
        if not locked:
 
1409
            self._warn_if_deprecated()
 
1410
            self._note_lock('r')
1267
1411
            for repo in self._fallback_repositories:
1268
1412
                repo.lock_read()
1269
1413
            self._refresh_data()
1332
1476
 
1333
1477
        # now gather global repository information
1334
1478
        # XXX: This is available for many repos regardless of listability.
1335
 
        if self.bzrdir.root_transport.listable():
 
1479
        if self.user_transport.listable():
1336
1480
            # XXX: do we want to __define len__() ?
1337
1481
            # Maybe the versionedfiles object should provide a different
1338
1482
            # method to get the number of keys.
1348
1492
        :param using: If True, list only branches using this repository.
1349
1493
        """
1350
1494
        if using and not self.is_shared():
1351
 
            try:
1352
 
                return [self.bzrdir.open_branch()]
1353
 
            except errors.NotBranchError:
1354
 
                return []
 
1495
            return self.bzrdir.list_branches()
1355
1496
        class Evaluator(object):
1356
1497
 
1357
1498
            def __init__(self):
1366
1507
                    except errors.NoRepositoryPresent:
1367
1508
                        pass
1368
1509
                    else:
1369
 
                        return False, (None, repository)
 
1510
                        return False, ([], repository)
1370
1511
                self.first_call = False
1371
 
                try:
1372
 
                    value = (bzrdir.open_branch(), None)
1373
 
                except errors.NotBranchError:
1374
 
                    value = (None, None)
 
1512
                value = (bzrdir.list_branches(), None)
1375
1513
                return True, value
1376
1514
 
1377
 
        branches = []
1378
 
        for branch, repository in bzrdir.BzrDir.find_bzrdirs(
1379
 
                self.bzrdir.root_transport, evaluate=Evaluator()):
1380
 
            if branch is not None:
1381
 
                branches.append(branch)
 
1515
        ret = []
 
1516
        for branches, repository in bzrdir.BzrDir.find_bzrdirs(
 
1517
                self.user_transport, evaluate=Evaluator()):
 
1518
            if branches is not None:
 
1519
                ret.extend(branches)
1382
1520
            if not using and repository is not None:
1383
 
                branches.extend(repository.find_branches())
1384
 
        return branches
 
1521
                ret.extend(repository.find_branches())
 
1522
        return ret
1385
1523
 
1386
1524
    @needs_read_lock
1387
1525
    def search_missing_revision_ids(self, other, revision_id=None, find_ghosts=True):
1416
1554
        """Commit the contents accrued within the current write group.
1417
1555
 
1418
1556
        :seealso: start_write_group.
 
1557
        
 
1558
        :return: it may return an opaque hint that can be passed to 'pack'.
1419
1559
        """
1420
1560
        if self._write_group is not self.get_transaction():
1421
1561
            # has an unlock or relock occured ?
1475
1615
        # but at the moment we're only checking for texts referenced by
1476
1616
        # inventories at the graph's edge.
1477
1617
        key_deps = self.revisions._index._key_dependencies
1478
 
        key_deps.add_keys(present_inventories)
 
1618
        key_deps.satisfy_refs_for_keys(present_inventories)
1479
1619
        referrers = frozenset(r[0] for r in key_deps.get_referrers())
1480
1620
        file_ids = self.fileids_altered_by_revision_ids(referrers)
1481
1621
        missing_texts = set()
1582
1722
        :param revprops: Optional dictionary of revision properties.
1583
1723
        :param revision_id: Optional revision id.
1584
1724
        """
 
1725
        if self._fallback_repositories:
 
1726
            raise errors.BzrError("Cannot commit from a lightweight checkout "
 
1727
                "to a stacked branch. See "
 
1728
                "https://bugs.launchpad.net/bzr/+bug/375013 for details.")
1585
1729
        result = self._commit_builder_class(self, parents, config,
1586
1730
            timestamp, timezone, committer, revprops, revision_id)
1587
1731
        self.start_write_group()
1588
1732
        return result
1589
1733
 
 
1734
    @only_raises(errors.LockNotHeld, errors.LockBroken)
1590
1735
    def unlock(self):
1591
1736
        if (self.control_files._lock_count == 1 and
1592
1737
            self.control_files._lock_mode == 'w'):
1714
1859
 
1715
1860
    @needs_read_lock
1716
1861
    def get_revisions(self, revision_ids):
1717
 
        """Get many revisions at once."""
 
1862
        """Get many revisions at once.
 
1863
        
 
1864
        Repositories that need to check data on every revision read should 
 
1865
        subclass this method.
 
1866
        """
1718
1867
        return self._get_revisions(revision_ids)
1719
1868
 
1720
1869
    @needs_read_lock
1721
1870
    def _get_revisions(self, revision_ids):
1722
1871
        """Core work logic to get many revisions without sanity checks."""
1723
 
        for rev_id in revision_ids:
1724
 
            if not rev_id or not isinstance(rev_id, basestring):
1725
 
                raise errors.InvalidRevisionId(revision_id=rev_id, branch=self)
 
1872
        revs = {}
 
1873
        for revid, rev in self._iter_revisions(revision_ids):
 
1874
            if rev is None:
 
1875
                raise errors.NoSuchRevision(self, revid)
 
1876
            revs[revid] = rev
 
1877
        return [revs[revid] for revid in revision_ids]
 
1878
 
 
1879
    def _iter_revisions(self, revision_ids):
 
1880
        """Iterate over revision objects.
 
1881
 
 
1882
        :param revision_ids: An iterable of revisions to examine. None may be
 
1883
            passed to request all revisions known to the repository. Note that
 
1884
            not all repositories can find unreferenced revisions; for those
 
1885
            repositories only referenced ones will be returned.
 
1886
        :return: An iterator of (revid, revision) tuples. Absent revisions (
 
1887
            those asked for but not available) are returned as (revid, None).
 
1888
        """
 
1889
        if revision_ids is None:
 
1890
            revision_ids = self.all_revision_ids()
 
1891
        else:
 
1892
            for rev_id in revision_ids:
 
1893
                if not rev_id or not isinstance(rev_id, basestring):
 
1894
                    raise errors.InvalidRevisionId(revision_id=rev_id, branch=self)
1726
1895
        keys = [(key,) for key in revision_ids]
1727
1896
        stream = self.revisions.get_record_stream(keys, 'unordered', True)
1728
 
        revs = {}
1729
1897
        for record in stream:
 
1898
            revid = record.key[0]
1730
1899
            if record.storage_kind == 'absent':
1731
 
                raise errors.NoSuchRevision(self, record.key[0])
1732
 
            text = record.get_bytes_as('fulltext')
1733
 
            rev = self._serializer.read_revision_from_string(text)
1734
 
            revs[record.key[0]] = rev
1735
 
        return [revs[revid] for revid in revision_ids]
1736
 
 
1737
 
    @needs_read_lock
1738
 
    def get_revision_xml(self, revision_id):
1739
 
        # TODO: jam 20070210 This shouldn't be necessary since get_revision
1740
 
        #       would have already do it.
1741
 
        # TODO: jam 20070210 Just use _serializer.write_revision_to_string()
1742
 
        # TODO: this can't just be replaced by:
1743
 
        # return self._serializer.write_revision_to_string(
1744
 
        #     self.get_revision(revision_id))
1745
 
        # as cStringIO preservers the encoding unlike write_revision_to_string
1746
 
        # or some other call down the path.
1747
 
        rev = self.get_revision(revision_id)
1748
 
        rev_tmp = cStringIO.StringIO()
1749
 
        # the current serializer..
1750
 
        self._serializer.write_revision(rev, rev_tmp)
1751
 
        rev_tmp.seek(0)
1752
 
        return rev_tmp.getvalue()
 
1900
                yield (revid, None)
 
1901
            else:
 
1902
                text = record.get_bytes_as('fulltext')
 
1903
                rev = self._serializer.read_revision_from_string(text)
 
1904
                yield (revid, rev)
1753
1905
 
1754
1906
    def get_deltas_for_revisions(self, revisions, specific_fileids=None):
1755
1907
        """Produce a generator of revision deltas.
1998
2150
        """
1999
2151
        selected_keys = set((revid,) for revid in revision_ids)
2000
2152
        w = _inv_weave or self.inventories
2001
 
        pb = ui.ui_factory.nested_progress_bar()
2002
 
        try:
2003
 
            return self._find_file_ids_from_xml_inventory_lines(
2004
 
                w.iter_lines_added_or_present_in_keys(
2005
 
                    selected_keys, pb=pb),
2006
 
                selected_keys)
2007
 
        finally:
2008
 
            pb.finished()
 
2153
        return self._find_file_ids_from_xml_inventory_lines(
 
2154
            w.iter_lines_added_or_present_in_keys(
 
2155
                selected_keys, pb=None),
 
2156
            selected_keys)
2009
2157
 
2010
2158
    def iter_files_bytes(self, desired_files):
2011
2159
        """Iterate through file versions.
2093
2241
                batch_size]
2094
2242
            if not to_query:
2095
2243
                break
2096
 
            for rev_tree in self.revision_trees(to_query):
2097
 
                revision_id = rev_tree.get_revision_id()
 
2244
            for revision_id in to_query:
2098
2245
                parent_ids = ancestors[revision_id]
2099
2246
                for text_key in revision_keys[revision_id]:
2100
2247
                    pb.update("Calculating text parents", processed_texts)
2173
2320
        num_file_ids = len(file_ids)
2174
2321
        for file_id, altered_versions in file_ids.iteritems():
2175
2322
            if pb is not None:
2176
 
                pb.update("fetch texts", count, num_file_ids)
 
2323
                pb.update("Fetch texts", count, num_file_ids)
2177
2324
            count += 1
2178
2325
            yield ("file", file_id, altered_versions)
2179
2326
 
2200
2347
        """Get Inventory object by revision id."""
2201
2348
        return self.iter_inventories([revision_id]).next()
2202
2349
 
2203
 
    def iter_inventories(self, revision_ids):
 
2350
    def iter_inventories(self, revision_ids, ordering=None):
2204
2351
        """Get many inventories by revision_ids.
2205
2352
 
2206
2353
        This will buffer some or all of the texts used in constructing the
2208
2355
        time.
2209
2356
 
2210
2357
        :param revision_ids: The expected revision ids of the inventories.
 
2358
        :param ordering: optional ordering, e.g. 'topological'.  If not
 
2359
            specified, the order of revision_ids will be preserved (by
 
2360
            buffering if necessary).
2211
2361
        :return: An iterator of inventories.
2212
2362
        """
2213
2363
        if ((None in revision_ids)
2214
2364
            or (_mod_revision.NULL_REVISION in revision_ids)):
2215
2365
            raise ValueError('cannot get null revision inventory')
2216
 
        return self._iter_inventories(revision_ids)
 
2366
        return self._iter_inventories(revision_ids, ordering)
2217
2367
 
2218
 
    def _iter_inventories(self, revision_ids):
 
2368
    def _iter_inventories(self, revision_ids, ordering):
2219
2369
        """single-document based inventory iteration."""
2220
 
        for text, revision_id in self._iter_inventory_xmls(revision_ids):
2221
 
            yield self.deserialise_inventory(revision_id, text)
 
2370
        inv_xmls = self._iter_inventory_xmls(revision_ids, ordering)
 
2371
        for text, revision_id in inv_xmls:
 
2372
            yield self._deserialise_inventory(revision_id, text)
2222
2373
 
2223
 
    def _iter_inventory_xmls(self, revision_ids):
 
2374
    def _iter_inventory_xmls(self, revision_ids, ordering):
 
2375
        if ordering is None:
 
2376
            order_as_requested = True
 
2377
            ordering = 'unordered'
 
2378
        else:
 
2379
            order_as_requested = False
2224
2380
        keys = [(revision_id,) for revision_id in revision_ids]
2225
 
        stream = self.inventories.get_record_stream(keys, 'unordered', True)
 
2381
        if not keys:
 
2382
            return
 
2383
        if order_as_requested:
 
2384
            key_iter = iter(keys)
 
2385
            next_key = key_iter.next()
 
2386
        stream = self.inventories.get_record_stream(keys, ordering, True)
2226
2387
        text_chunks = {}
2227
2388
        for record in stream:
2228
2389
            if record.storage_kind != 'absent':
2229
 
                text_chunks[record.key] = record.get_bytes_as('chunked')
 
2390
                chunks = record.get_bytes_as('chunked')
 
2391
                if order_as_requested:
 
2392
                    text_chunks[record.key] = chunks
 
2393
                else:
 
2394
                    yield ''.join(chunks), record.key[-1]
2230
2395
            else:
2231
2396
                raise errors.NoSuchRevision(self, record.key)
2232
 
        for key in keys:
2233
 
            chunks = text_chunks.pop(key)
2234
 
            yield ''.join(chunks), key[-1]
 
2397
            if order_as_requested:
 
2398
                # Yield as many results as we can while preserving order.
 
2399
                while next_key in text_chunks:
 
2400
                    chunks = text_chunks.pop(next_key)
 
2401
                    yield ''.join(chunks), next_key[-1]
 
2402
                    try:
 
2403
                        next_key = key_iter.next()
 
2404
                    except StopIteration:
 
2405
                        # We still want to fully consume the get_record_stream,
 
2406
                        # just in case it is not actually finished at this point
 
2407
                        next_key = None
 
2408
                        break
2235
2409
 
2236
 
    def deserialise_inventory(self, revision_id, xml):
 
2410
    def _deserialise_inventory(self, revision_id, xml):
2237
2411
        """Transform the xml into an inventory object.
2238
2412
 
2239
2413
        :param revision_id: The expected revision id of the inventory.
2240
2414
        :param xml: A serialised inventory.
2241
2415
        """
2242
2416
        result = self._serializer.read_inventory_from_string(xml, revision_id,
2243
 
                    entry_cache=self._inventory_entry_cache)
 
2417
                    entry_cache=self._inventory_entry_cache,
 
2418
                    return_from_cache=self._safe_to_return_from_cache)
2244
2419
        if result.revision_id != revision_id:
2245
2420
            raise AssertionError('revision id mismatch %s != %s' % (
2246
2421
                result.revision_id, revision_id))
2247
2422
        return result
2248
2423
 
2249
 
    def serialise_inventory(self, inv):
2250
 
        return self._serializer.write_inventory_to_string(inv)
2251
 
 
2252
 
    def _serialise_inventory_to_lines(self, inv):
2253
 
        return self._serializer.write_inventory_to_lines(inv)
2254
 
 
2255
2424
    def get_serializer_format(self):
2256
2425
        return self._serializer.format_num
2257
2426
 
2258
2427
    @needs_read_lock
2259
 
    def get_inventory_xml(self, revision_id):
2260
 
        """Get inventory XML as a file object."""
2261
 
        texts = self._iter_inventory_xmls([revision_id])
 
2428
    def _get_inventory_xml(self, revision_id):
 
2429
        """Get serialized inventory as a string."""
 
2430
        texts = self._iter_inventory_xmls([revision_id], 'unordered')
2262
2431
        try:
2263
2432
            text, revision_id = texts.next()
2264
2433
        except StopIteration:
2265
2434
            raise errors.HistoryMissing(self, 'inventory', revision_id)
2266
2435
        return text
2267
2436
 
2268
 
    @needs_read_lock
2269
 
    def get_inventory_sha1(self, revision_id):
2270
 
        """Return the sha1 hash of the inventory entry
2271
 
        """
2272
 
        return self.get_revision(revision_id).inventory_sha1
2273
 
 
2274
2437
    def get_rev_id_for_revno(self, revno, known_pair):
2275
2438
        """Return the revision id of a revno, given a later (revno, revid)
2276
2439
        pair in the same history.
2327
2490
            else:
2328
2491
                next_id = parents[0]
2329
2492
 
2330
 
    @needs_read_lock
2331
 
    def get_revision_inventory(self, revision_id):
2332
 
        """Return inventory of a past revision."""
2333
 
        # TODO: Unify this with get_inventory()
2334
 
        # bzr 0.0.6 and later imposes the constraint that the inventory_id
2335
 
        # must be the same as its revision, so this is trivial.
2336
 
        if revision_id is None:
2337
 
            # This does not make sense: if there is no revision,
2338
 
            # then it is the current tree inventory surely ?!
2339
 
            # and thus get_root_id() is something that looks at the last
2340
 
            # commit on the branch, and the get_root_id is an inventory check.
2341
 
            raise NotImplementedError
2342
 
            # return Inventory(self.get_root_id())
2343
 
        else:
2344
 
            return self.get_inventory(revision_id)
2345
 
 
2346
2493
    def is_shared(self):
2347
2494
        """Return True if this repository is flagged as a shared repository."""
2348
2495
        raise NotImplementedError(self.is_shared)
2382
2529
            return RevisionTree(self, Inventory(root_id=None),
2383
2530
                                _mod_revision.NULL_REVISION)
2384
2531
        else:
2385
 
            inv = self.get_revision_inventory(revision_id)
 
2532
            inv = self.get_inventory(revision_id)
2386
2533
            return RevisionTree(self, inv, revision_id)
2387
2534
 
2388
2535
    def revision_trees(self, revision_ids):
2441
2588
            keys = tsort.topo_sort(parent_map)
2442
2589
        return [None] + list(keys)
2443
2590
 
2444
 
    def pack(self, hint=None):
 
2591
    def pack(self, hint=None, clean_obsolete_packs=False):
2445
2592
        """Compress the data within the repository.
2446
2593
 
2447
2594
        This operation only makes sense for some repository types. For other
2457
2604
            obtained from the result of commit_write_group(). Out of
2458
2605
            date hints are simply ignored, because concurrent operations
2459
2606
            can obsolete them rapidly.
 
2607
 
 
2608
        :param clean_obsolete_packs: Clean obsolete packs immediately after
 
2609
            the pack operation.
2460
2610
        """
2461
2611
 
2462
2612
    def get_transaction(self):
2478
2628
        for ((revision_id,), parent_keys) in \
2479
2629
                self.revisions.get_parent_map(query_keys).iteritems():
2480
2630
            if parent_keys:
2481
 
                result[revision_id] = tuple(parent_revid
2482
 
                    for (parent_revid,) in parent_keys)
 
2631
                result[revision_id] = tuple([parent_revid
 
2632
                    for (parent_revid,) in parent_keys])
2483
2633
            else:
2484
2634
                result[revision_id] = (_mod_revision.NULL_REVISION,)
2485
2635
        return result
2487
2637
    def _make_parents_provider(self):
2488
2638
        return self
2489
2639
 
 
2640
    @needs_read_lock
 
2641
    def get_known_graph_ancestry(self, revision_ids):
 
2642
        """Return the known graph for a set of revision ids and their ancestors.
 
2643
        """
 
2644
        st = static_tuple.StaticTuple
 
2645
        revision_keys = [st(r_id).intern() for r_id in revision_ids]
 
2646
        known_graph = self.revisions.get_known_graph_ancestry(revision_keys)
 
2647
        return graph.GraphThunkIdsToKeys(known_graph)
 
2648
 
2490
2649
    def get_graph(self, other_repository=None):
2491
2650
        """Return the graph walker for this repository format"""
2492
2651
        parents_provider = self._make_parents_provider()
2496
2655
                [parents_provider, other_repository._make_parents_provider()])
2497
2656
        return graph.Graph(parents_provider)
2498
2657
 
2499
 
    def _get_versioned_file_checker(self, text_key_references=None):
 
2658
    def _get_versioned_file_checker(self, text_key_references=None,
 
2659
        ancestors=None):
2500
2660
        """Return an object suitable for checking versioned files.
2501
2661
        
2502
2662
        :param text_key_references: if non-None, an already built
2504
2664
            to whether they were referred to by the inventory of the
2505
2665
            revision_id that they contain. If None, this will be
2506
2666
            calculated.
 
2667
        :param ancestors: Optional result from
 
2668
            self.get_graph().get_parent_map(self.all_revision_ids()) if already
 
2669
            available.
2507
2670
        """
2508
2671
        return _VersionedFileChecker(self,
2509
 
            text_key_references=text_key_references)
 
2672
            text_key_references=text_key_references, ancestors=ancestors)
2510
2673
 
2511
2674
    def revision_ids_to_search_result(self, result_set):
2512
2675
        """Convert a set of revision ids to a graph SearchResult."""
2562
2725
        return record.get_bytes_as('fulltext')
2563
2726
 
2564
2727
    @needs_read_lock
2565
 
    def check(self, revision_ids=None):
 
2728
    def check(self, revision_ids=None, callback_refs=None, check_repo=True):
2566
2729
        """Check consistency of all history of given revision_ids.
2567
2730
 
2568
2731
        Different repository implementations should override _check().
2569
2732
 
2570
2733
        :param revision_ids: A non-empty list of revision_ids whose ancestry
2571
2734
             will be checked.  Typically the last revision_id of a branch.
 
2735
        :param callback_refs: A dict of check-refs to resolve and callback
 
2736
            the check/_check method on the items listed as wanting the ref.
 
2737
            see bzrlib.check.
 
2738
        :param check_repo: If False do not check the repository contents, just 
 
2739
            calculate the data callback_refs requires and call them back.
2572
2740
        """
2573
 
        return self._check(revision_ids)
 
2741
        return self._check(revision_ids, callback_refs=callback_refs,
 
2742
            check_repo=check_repo)
2574
2743
 
2575
 
    def _check(self, revision_ids):
2576
 
        result = check.Check(self)
2577
 
        result.check()
 
2744
    def _check(self, revision_ids, callback_refs, check_repo):
 
2745
        result = check.Check(self, check_repo=check_repo)
 
2746
        result.check(callback_refs)
2578
2747
        return result
2579
2748
 
2580
 
    def _warn_if_deprecated(self):
 
2749
    def _warn_if_deprecated(self, branch=None):
2581
2750
        global _deprecation_warning_done
2582
2751
        if _deprecation_warning_done:
2583
2752
            return
2584
 
        _deprecation_warning_done = True
2585
 
        warning("Format %s for %s is deprecated - please use 'bzr upgrade' to get better performance"
2586
 
                % (self._format, self.bzrdir.transport.base))
 
2753
        try:
 
2754
            if branch is None:
 
2755
                conf = config.GlobalConfig()
 
2756
            else:
 
2757
                conf = branch.get_config()
 
2758
            if conf.suppress_warning('format_deprecation'):
 
2759
                return
 
2760
            warning("Format %s for %s is deprecated -"
 
2761
                    " please use 'bzr upgrade' to get better performance"
 
2762
                    % (self._format, self.bzrdir.transport.base))
 
2763
        finally:
 
2764
            _deprecation_warning_done = True
2587
2765
 
2588
2766
    def supports_rich_root(self):
2589
2767
        return self._format.rich_root_data
2870
3048
    # help), and for fetching when data won't have come from the same
2871
3049
    # compressor.
2872
3050
    pack_compresses = False
 
3051
    # Does the repository inventory storage understand references to trees?
 
3052
    supports_tree_reference = None
 
3053
    # Is the format experimental ?
 
3054
    experimental = False
2873
3055
 
2874
 
    def __str__(self):
2875
 
        return "<%s>" % self.__class__.__name__
 
3056
    def __repr__(self):
 
3057
        return "%s()" % self.__class__.__name__
2876
3058
 
2877
3059
    def __eq__(self, other):
2878
3060
        # format objects are generally stateless
2891
3073
        """
2892
3074
        try:
2893
3075
            transport = a_bzrdir.get_repository_transport(None)
2894
 
            format_string = transport.get("format").read()
 
3076
            format_string = transport.get_bytes("format")
2895
3077
            return format_registry.get(format_string)
2896
3078
        except errors.NoSuchFile:
2897
3079
            raise errors.NoRepositoryPresent(a_bzrdir)
2979
3161
        raise NotImplementedError(self.network_name)
2980
3162
 
2981
3163
    def check_conversion_target(self, target_format):
2982
 
        raise NotImplementedError(self.check_conversion_target)
 
3164
        if self.rich_root_data and not target_format.rich_root_data:
 
3165
            raise errors.BadConversionTarget(
 
3166
                'Does not support rich root data.', target_format,
 
3167
                from_format=self)
 
3168
        if (self.supports_tree_reference and 
 
3169
            not getattr(target_format, 'supports_tree_reference', False)):
 
3170
            raise errors.BadConversionTarget(
 
3171
                'Does not support nested trees', target_format,
 
3172
                from_format=self)
2983
3173
 
2984
3174
    def open(self, a_bzrdir, _found=False):
2985
3175
        """Return an instance of this format for the bzrdir a_bzrdir.
2988
3178
        """
2989
3179
        raise NotImplementedError(self.open)
2990
3180
 
 
3181
    def _run_post_repo_init_hooks(self, repository, a_bzrdir, shared):
 
3182
        from bzrlib.bzrdir import BzrDir, RepoInitHookParams
 
3183
        hooks = BzrDir.hooks['post_repo_init']
 
3184
        if not hooks:
 
3185
            return
 
3186
        params = RepoInitHookParams(repository, self, a_bzrdir, shared)
 
3187
        for hook in hooks:
 
3188
            hook(params)
 
3189
 
2991
3190
 
2992
3191
class MetaDirRepositoryFormat(RepositoryFormat):
2993
3192
    """Common base class for the new repositories using the metadir layout."""
3198
3397
 
3199
3398
        :param revision_id: if None all content is copied, if NULL_REVISION no
3200
3399
                            content is copied.
3201
 
        :param pb: optional progress bar to use for progress reports. If not
3202
 
                   provided a default one will be created.
 
3400
        :param pb: ignored.
3203
3401
        :return: None.
3204
3402
        """
 
3403
        ui.ui_factory.warn_experimental_format_fetch(self)
3205
3404
        from bzrlib.fetch import RepoFetcher
 
3405
        # See <https://launchpad.net/bugs/456077> asking for a warning here
 
3406
        if self.source._format.network_name() != self.target._format.network_name():
 
3407
            ui.ui_factory.show_user_warning('cross_format_fetch',
 
3408
                from_format=self.source._format,
 
3409
                to_format=self.target._format)
3206
3410
        f = RepoFetcher(to_repository=self.target,
3207
3411
                               from_repository=self.source,
3208
3412
                               last_revision=revision_id,
3209
3413
                               fetch_spec=fetch_spec,
3210
 
                               pb=pb, find_ghosts=find_ghosts)
 
3414
                               find_ghosts=find_ghosts)
3211
3415
 
3212
3416
    def _walk_to_common_revisions(self, revision_ids):
3213
3417
        """Walk out from revision_ids in source to revisions target has.
3382
3586
                self.target.texts.insert_record_stream(
3383
3587
                    self.source.texts.get_record_stream(
3384
3588
                        self.source.texts.keys(), 'topological', False))
3385
 
                pb.update('copying inventory', 0, 1)
 
3589
                pb.update('Copying inventory', 0, 1)
3386
3590
                self.target.inventories.insert_record_stream(
3387
3591
                    self.source.inventories.get_record_stream(
3388
3592
                        self.source.inventories.keys(), 'topological', False))
3514
3718
        # This is redundant with format.check_conversion_target(), however that
3515
3719
        # raises an exception, and we just want to say "False" as in we won't
3516
3720
        # support converting between these formats.
 
3721
        if 'IDS_never' in debug.debug_flags:
 
3722
            return False
3517
3723
        if source.supports_rich_root() and not target.supports_rich_root():
3518
3724
            return False
3519
3725
        if (source._format.supports_tree_reference
3520
3726
            and not target._format.supports_tree_reference):
3521
3727
            return False
 
3728
        if target._fallback_repositories and target._format.supports_chks:
 
3729
            # IDS doesn't know how to copy CHKs for the parent inventories it
 
3730
            # adds to stacked repos.
 
3731
            return False
 
3732
        if 'IDS_always' in debug.debug_flags:
 
3733
            return True
 
3734
        # Only use this code path for local source and target.  IDS does far
 
3735
        # too much IO (both bandwidth and roundtrips) over a network.
 
3736
        if not source.bzrdir.transport.base.startswith('file:///'):
 
3737
            return False
 
3738
        if not target.bzrdir.transport.base.startswith('file:///'):
 
3739
            return False
3522
3740
        return True
3523
3741
 
3524
 
    def _get_delta_for_revision(self, tree, parent_ids, basis_id, cache):
 
3742
    def _get_trees(self, revision_ids, cache):
 
3743
        possible_trees = []
 
3744
        for rev_id in revision_ids:
 
3745
            if rev_id in cache:
 
3746
                possible_trees.append((rev_id, cache[rev_id]))
 
3747
            else:
 
3748
                # Not cached, but inventory might be present anyway.
 
3749
                try:
 
3750
                    tree = self.source.revision_tree(rev_id)
 
3751
                except errors.NoSuchRevision:
 
3752
                    # Nope, parent is ghost.
 
3753
                    pass
 
3754
                else:
 
3755
                    cache[rev_id] = tree
 
3756
                    possible_trees.append((rev_id, tree))
 
3757
        return possible_trees
 
3758
 
 
3759
    def _get_delta_for_revision(self, tree, parent_ids, possible_trees):
3525
3760
        """Get the best delta and base for this revision.
3526
3761
 
3527
3762
        :return: (basis_id, delta)
3528
3763
        """
3529
 
        possible_trees = [(parent_id, cache[parent_id])
3530
 
                          for parent_id in parent_ids
3531
 
                           if parent_id in cache]
3532
 
        if len(possible_trees) == 0:
3533
 
            # There either aren't any parents, or the parents aren't in the
3534
 
            # cache, so just use the last converted tree
3535
 
            possible_trees.append((basis_id, cache[basis_id]))
3536
3764
        deltas = []
 
3765
        # Generate deltas against each tree, to find the shortest.
 
3766
        texts_possibly_new_in_tree = set()
3537
3767
        for basis_id, basis_tree in possible_trees:
3538
3768
            delta = tree.inventory._make_delta(basis_tree.inventory)
 
3769
            for old_path, new_path, file_id, new_entry in delta:
 
3770
                if new_path is None:
 
3771
                    # This file_id isn't present in the new rev, so we don't
 
3772
                    # care about it.
 
3773
                    continue
 
3774
                if not new_path:
 
3775
                    # Rich roots are handled elsewhere...
 
3776
                    continue
 
3777
                kind = new_entry.kind
 
3778
                if kind != 'directory' and kind != 'file':
 
3779
                    # No text record associated with this inventory entry.
 
3780
                    continue
 
3781
                # This is a directory or file that has changed somehow.
 
3782
                texts_possibly_new_in_tree.add((file_id, new_entry.revision))
3539
3783
            deltas.append((len(delta), basis_id, delta))
3540
3784
        deltas.sort()
3541
3785
        return deltas[0][1:]
3542
3786
 
3543
 
    def _get_parent_keys(self, root_key, parent_map):
3544
 
        """Get the parent keys for a given root id."""
3545
 
        root_id, rev_id = root_key
3546
 
        # Include direct parents of the revision, but only if they used
3547
 
        # the same root_id and are heads.
3548
 
        parent_keys = []
3549
 
        for parent_id in parent_map[rev_id]:
3550
 
            if parent_id == _mod_revision.NULL_REVISION:
3551
 
                continue
3552
 
            if parent_id not in self._revision_id_to_root_id:
3553
 
                # We probably didn't read this revision, go spend the
3554
 
                # extra effort to actually check
3555
 
                try:
3556
 
                    tree = self.source.revision_tree(parent_id)
3557
 
                except errors.NoSuchRevision:
3558
 
                    # Ghost, fill out _revision_id_to_root_id in case we
3559
 
                    # encounter this again.
3560
 
                    # But set parent_root_id to None since we don't really know
3561
 
                    parent_root_id = None
3562
 
                else:
3563
 
                    parent_root_id = tree.get_root_id()
3564
 
                self._revision_id_to_root_id[parent_id] = None
3565
 
            else:
3566
 
                parent_root_id = self._revision_id_to_root_id[parent_id]
3567
 
            if root_id == parent_root_id:
3568
 
                # With stacking we _might_ want to refer to a non-local
3569
 
                # revision, but this code path only applies when we have the
3570
 
                # full content available, so ghosts really are ghosts, not just
3571
 
                # the edge of local data.
3572
 
                parent_keys.append((parent_id,))
3573
 
            else:
3574
 
                # root_id may be in the parent anyway.
3575
 
                try:
3576
 
                    tree = self.source.revision_tree(parent_id)
3577
 
                except errors.NoSuchRevision:
3578
 
                    # ghost, can't refer to it.
3579
 
                    pass
3580
 
                else:
3581
 
                    try:
3582
 
                        parent_keys.append((tree.inventory[root_id].revision,))
3583
 
                    except errors.NoSuchId:
3584
 
                        # not in the tree
3585
 
                        pass
3586
 
        g = graph.Graph(self.source.revisions)
3587
 
        heads = g.heads(parent_keys)
3588
 
        selected_keys = []
3589
 
        for key in parent_keys:
3590
 
            if key in heads and key not in selected_keys:
3591
 
                selected_keys.append(key)
3592
 
        return tuple([(root_id,)+ key for key in selected_keys])
3593
 
 
3594
 
    def _new_root_data_stream(self, root_keys_to_create, parent_map):
3595
 
        for root_key in root_keys_to_create:
3596
 
            parent_keys = self._get_parent_keys(root_key, parent_map)
3597
 
            yield versionedfile.FulltextContentFactory(root_key,
3598
 
                parent_keys, None, '')
3599
 
 
3600
 
    def _fetch_batch(self, revision_ids, basis_id, cache):
 
3787
    def _fetch_parent_invs_for_stacking(self, parent_map, cache):
 
3788
        """Find all parent revisions that are absent, but for which the
 
3789
        inventory is present, and copy those inventories.
 
3790
 
 
3791
        This is necessary to preserve correctness when the source is stacked
 
3792
        without fallbacks configured.  (Note that in cases like upgrade the
 
3793
        source may be not have _fallback_repositories even though it is
 
3794
        stacked.)
 
3795
        """
 
3796
        parent_revs = set()
 
3797
        for parents in parent_map.values():
 
3798
            parent_revs.update(parents)
 
3799
        present_parents = self.source.get_parent_map(parent_revs)
 
3800
        absent_parents = set(parent_revs).difference(present_parents)
 
3801
        parent_invs_keys_for_stacking = self.source.inventories.get_parent_map(
 
3802
            (rev_id,) for rev_id in absent_parents)
 
3803
        parent_inv_ids = [key[-1] for key in parent_invs_keys_for_stacking]
 
3804
        for parent_tree in self.source.revision_trees(parent_inv_ids):
 
3805
            current_revision_id = parent_tree.get_revision_id()
 
3806
            parents_parents_keys = parent_invs_keys_for_stacking[
 
3807
                (current_revision_id,)]
 
3808
            parents_parents = [key[-1] for key in parents_parents_keys]
 
3809
            basis_id = _mod_revision.NULL_REVISION
 
3810
            basis_tree = self.source.revision_tree(basis_id)
 
3811
            delta = parent_tree.inventory._make_delta(basis_tree.inventory)
 
3812
            self.target.add_inventory_by_delta(
 
3813
                basis_id, delta, current_revision_id, parents_parents)
 
3814
            cache[current_revision_id] = parent_tree
 
3815
 
 
3816
    def _fetch_batch(self, revision_ids, basis_id, cache, a_graph=None):
3601
3817
        """Fetch across a few revisions.
3602
3818
 
3603
3819
        :param revision_ids: The revisions to copy
3604
3820
        :param basis_id: The revision_id of a tree that must be in cache, used
3605
3821
            as a basis for delta when no other base is available
3606
3822
        :param cache: A cache of RevisionTrees that we can use.
 
3823
        :param a_graph: A Graph object to determine the heads() of the
 
3824
            rich-root data stream.
3607
3825
        :return: The revision_id of the last converted tree. The RevisionTree
3608
3826
            for it will be in cache
3609
3827
        """
3615
3833
        pending_deltas = []
3616
3834
        pending_revisions = []
3617
3835
        parent_map = self.source.get_parent_map(revision_ids)
 
3836
        self._fetch_parent_invs_for_stacking(parent_map, cache)
 
3837
        self.source._safe_to_return_from_cache = True
3618
3838
        for tree in self.source.revision_trees(revision_ids):
 
3839
            # Find a inventory delta for this revision.
 
3840
            # Find text entries that need to be copied, too.
3619
3841
            current_revision_id = tree.get_revision_id()
3620
3842
            parent_ids = parent_map.get(current_revision_id, ())
 
3843
            parent_trees = self._get_trees(parent_ids, cache)
 
3844
            possible_trees = list(parent_trees)
 
3845
            if len(possible_trees) == 0:
 
3846
                # There either aren't any parents, or the parents are ghosts,
 
3847
                # so just use the last converted tree.
 
3848
                possible_trees.append((basis_id, cache[basis_id]))
3621
3849
            basis_id, delta = self._get_delta_for_revision(tree, parent_ids,
3622
 
                                                           basis_id, cache)
 
3850
                                                           possible_trees)
 
3851
            revision = self.source.get_revision(current_revision_id)
 
3852
            pending_deltas.append((basis_id, delta,
 
3853
                current_revision_id, revision.parent_ids))
3623
3854
            if self._converting_to_rich_root:
3624
3855
                self._revision_id_to_root_id[current_revision_id] = \
3625
3856
                    tree.get_root_id()
3626
 
            # Find text entries that need to be copied
 
3857
            # Determine which texts are in present in this revision but not in
 
3858
            # any of the available parents.
 
3859
            texts_possibly_new_in_tree = set()
3627
3860
            for old_path, new_path, file_id, entry in delta:
3628
 
                if new_path is not None:
3629
 
                    if not new_path:
3630
 
                        # This is the root
3631
 
                        if not self.target.supports_rich_root():
3632
 
                            # The target doesn't support rich root, so we don't
3633
 
                            # copy
3634
 
                            continue
3635
 
                        if self._converting_to_rich_root:
3636
 
                            # This can't be copied normally, we have to insert
3637
 
                            # it specially
3638
 
                            root_keys_to_create.add((file_id, entry.revision))
3639
 
                            continue
3640
 
                    text_keys.add((file_id, entry.revision))
3641
 
            revision = self.source.get_revision(current_revision_id)
3642
 
            pending_deltas.append((basis_id, delta,
3643
 
                current_revision_id, revision.parent_ids))
 
3861
                if new_path is None:
 
3862
                    # This file_id isn't present in the new rev
 
3863
                    continue
 
3864
                if not new_path:
 
3865
                    # This is the root
 
3866
                    if not self.target.supports_rich_root():
 
3867
                        # The target doesn't support rich root, so we don't
 
3868
                        # copy
 
3869
                        continue
 
3870
                    if self._converting_to_rich_root:
 
3871
                        # This can't be copied normally, we have to insert
 
3872
                        # it specially
 
3873
                        root_keys_to_create.add((file_id, entry.revision))
 
3874
                        continue
 
3875
                kind = entry.kind
 
3876
                texts_possibly_new_in_tree.add((file_id, entry.revision))
 
3877
            for basis_id, basis_tree in possible_trees:
 
3878
                basis_inv = basis_tree.inventory
 
3879
                for file_key in list(texts_possibly_new_in_tree):
 
3880
                    file_id, file_revision = file_key
 
3881
                    try:
 
3882
                        entry = basis_inv[file_id]
 
3883
                    except errors.NoSuchId:
 
3884
                        continue
 
3885
                    if entry.revision == file_revision:
 
3886
                        texts_possibly_new_in_tree.remove(file_key)
 
3887
            text_keys.update(texts_possibly_new_in_tree)
3644
3888
            pending_revisions.append(revision)
3645
3889
            cache[current_revision_id] = tree
3646
3890
            basis_id = current_revision_id
 
3891
        self.source._safe_to_return_from_cache = False
3647
3892
        # Copy file texts
3648
3893
        from_texts = self.source.texts
3649
3894
        to_texts = self.target.texts
3650
3895
        if root_keys_to_create:
3651
 
            root_stream = self._new_root_data_stream(root_keys_to_create,
3652
 
                                                     parent_map)
 
3896
            root_stream = _mod_fetch._new_root_data_stream(
 
3897
                root_keys_to_create, self._revision_id_to_root_id, parent_map,
 
3898
                self.source, graph=a_graph)
3653
3899
            to_texts.insert_record_stream(root_stream)
3654
3900
        to_texts.insert_record_stream(from_texts.get_record_stream(
3655
3901
            text_keys, self.target._format._fetch_order,
3662
3908
            # for the new revisions that we are about to insert.  We do this
3663
3909
            # before adding the revisions so that no revision is added until
3664
3910
            # all the inventories it may depend on are added.
 
3911
            # Note that this is overzealous, as we may have fetched these in an
 
3912
            # earlier batch.
3665
3913
            parent_ids = set()
3666
3914
            revision_ids = set()
3667
3915
            for revision in pending_revisions:
3670
3918
            parent_ids.difference_update(revision_ids)
3671
3919
            parent_ids.discard(_mod_revision.NULL_REVISION)
3672
3920
            parent_map = self.source.get_parent_map(parent_ids)
3673
 
            for parent_tree in self.source.revision_trees(parent_ids):
3674
 
                basis_id, delta = self._get_delta_for_revision(tree, parent_ids, basis_id, cache)
 
3921
            # we iterate over parent_map and not parent_ids because we don't
 
3922
            # want to try copying any revision which is a ghost
 
3923
            for parent_tree in self.source.revision_trees(parent_map):
3675
3924
                current_revision_id = parent_tree.get_revision_id()
3676
3925
                parents_parents = parent_map[current_revision_id]
 
3926
                possible_trees = self._get_trees(parents_parents, cache)
 
3927
                if len(possible_trees) == 0:
 
3928
                    # There either aren't any parents, or the parents are
 
3929
                    # ghosts, so just use the last converted tree.
 
3930
                    possible_trees.append((basis_id, cache[basis_id]))
 
3931
                basis_id, delta = self._get_delta_for_revision(parent_tree,
 
3932
                    parents_parents, possible_trees)
3677
3933
                self.target.add_inventory_by_delta(
3678
3934
                    basis_id, delta, current_revision_id, parents_parents)
3679
3935
        # insert signatures and revisions
3693
3949
 
3694
3950
        :param revision_ids: The list of revisions to fetch. Must be in
3695
3951
            topological order.
3696
 
        :param pb: A ProgressBar
 
3952
        :param pb: A ProgressTask
3697
3953
        :return: None
3698
3954
        """
3699
3955
        basis_id, basis_tree = self._get_basis(revision_ids[0])
3702
3958
        cache[basis_id] = basis_tree
3703
3959
        del basis_tree # We don't want to hang on to it here
3704
3960
        hints = []
 
3961
        if self._converting_to_rich_root and len(revision_ids) > 100:
 
3962
            a_graph = _mod_fetch._get_rich_root_heads_graph(self.source,
 
3963
                                                            revision_ids)
 
3964
        else:
 
3965
            a_graph = None
 
3966
 
3705
3967
        for offset in range(0, len(revision_ids), batch_size):
3706
3968
            self.target.start_write_group()
3707
3969
            try:
3708
3970
                pb.update('Transferring revisions', offset,
3709
3971
                          len(revision_ids))
3710
3972
                batch = revision_ids[offset:offset+batch_size]
3711
 
                basis_id = self._fetch_batch(batch, basis_id, cache)
 
3973
                basis_id = self._fetch_batch(batch, basis_id, cache,
 
3974
                                             a_graph=a_graph)
3712
3975
            except:
 
3976
                self.source._safe_to_return_from_cache = False
3713
3977
                self.target.abort_write_group()
3714
3978
                raise
3715
3979
            else:
3727
3991
        """See InterRepository.fetch()."""
3728
3992
        if fetch_spec is not None:
3729
3993
            raise AssertionError("Not implemented yet...")
 
3994
        ui.ui_factory.warn_experimental_format_fetch(self)
3730
3995
        if (not self.source.supports_rich_root()
3731
3996
            and self.target.supports_rich_root()):
3732
3997
            self._converting_to_rich_root = True
3733
3998
            self._revision_id_to_root_id = {}
3734
3999
        else:
3735
4000
            self._converting_to_rich_root = False
 
4001
        # See <https://launchpad.net/bugs/456077> asking for a warning here
 
4002
        if self.source._format.network_name() != self.target._format.network_name():
 
4003
            ui.ui_factory.show_user_warning('cross_format_fetch',
 
4004
                from_format=self.source._format,
 
4005
                to_format=self.target._format)
3736
4006
        revision_ids = self.target.search_missing_revision_ids(self.source,
3737
4007
            revision_id, find_ghosts=find_ghosts).get_keys()
3738
4008
        if not revision_ids:
3744
4014
        # Walk though all revisions; get inventory deltas, copy referenced
3745
4015
        # texts that delta references, insert the delta, revision and
3746
4016
        # signature.
3747
 
        first_rev = self.source.get_revision(revision_ids[0])
3748
4017
        if pb is None:
3749
4018
            my_pb = ui.ui_factory.nested_progress_bar()
3750
4019
            pb = my_pb
3808
4077
        :param to_convert: The disk object to convert.
3809
4078
        :param pb: a progress bar to use for progress information.
3810
4079
        """
3811
 
        self.pb = pb
 
4080
        pb = ui.ui_factory.nested_progress_bar()
3812
4081
        self.count = 0
3813
4082
        self.total = 4
3814
4083
        # this is only useful with metadir layouts - separated repo content.
3815
4084
        # trigger an assertion if not such
3816
4085
        repo._format.get_format_string()
3817
4086
        self.repo_dir = repo.bzrdir
3818
 
        self.step('Moving repository to repository.backup')
 
4087
        pb.update('Moving repository to repository.backup')
3819
4088
        self.repo_dir.transport.move('repository', 'repository.backup')
3820
4089
        backup_transport =  self.repo_dir.transport.clone('repository.backup')
3821
4090
        repo._format.check_conversion_target(self.target_format)
3822
4091
        self.source_repo = repo._format.open(self.repo_dir,
3823
4092
            _found=True,
3824
4093
            _override_transport=backup_transport)
3825
 
        self.step('Creating new repository')
 
4094
        pb.update('Creating new repository')
3826
4095
        converted = self.target_format.initialize(self.repo_dir,
3827
4096
                                                  self.source_repo.is_shared())
3828
4097
        converted.lock_write()
3829
4098
        try:
3830
 
            self.step('Copying content into repository.')
 
4099
            pb.update('Copying content')
3831
4100
            self.source_repo.copy_content_into(converted)
3832
4101
        finally:
3833
4102
            converted.unlock()
3834
 
        self.step('Deleting old repository content.')
 
4103
        pb.update('Deleting old repository content')
3835
4104
        self.repo_dir.transport.delete_tree('repository.backup')
3836
 
        self.pb.note('repository converted')
3837
 
 
3838
 
    def step(self, message):
3839
 
        """Update the pb by a step."""
3840
 
        self.count +=1
3841
 
        self.pb.update(message, self.count, self.total)
 
4105
        ui.ui_factory.note('repository converted')
 
4106
        pb.finished()
3842
4107
 
3843
4108
 
3844
4109
_unescape_map = {
3873
4138
 
3874
4139
class _VersionedFileChecker(object):
3875
4140
 
3876
 
    def __init__(self, repository, text_key_references=None):
 
4141
    def __init__(self, repository, text_key_references=None, ancestors=None):
3877
4142
        self.repository = repository
3878
4143
        self.text_index = self.repository._generate_text_key_index(
3879
 
            text_key_references=text_key_references)
 
4144
            text_key_references=text_key_references, ancestors=ancestors)
3880
4145
 
3881
4146
    def calculate_file_version_parents(self, text_key):
3882
4147
        """Calculate the correct parents for a file version according to
3900
4165
            revision_id) tuples for versions that are present in this versioned
3901
4166
            file, but not used by the corresponding inventory.
3902
4167
        """
 
4168
        local_progress = None
 
4169
        if progress_bar is None:
 
4170
            local_progress = ui.ui_factory.nested_progress_bar()
 
4171
            progress_bar = local_progress
 
4172
        try:
 
4173
            return self._check_file_version_parents(texts, progress_bar)
 
4174
        finally:
 
4175
            if local_progress:
 
4176
                local_progress.finished()
 
4177
 
 
4178
    def _check_file_version_parents(self, texts, progress_bar):
 
4179
        """See check_file_version_parents."""
3903
4180
        wrong_parents = {}
3904
4181
        self.file_ids = set([file_id for file_id, _ in
3905
4182
            self.text_index.iterkeys()])
3906
4183
        # text keys is now grouped by file_id
3907
 
        n_weaves = len(self.file_ids)
3908
 
        files_in_revisions = {}
3909
 
        revisions_of_files = {}
3910
4184
        n_versions = len(self.text_index)
3911
4185
        progress_bar.update('loading text store', 0, n_versions)
3912
4186
        parent_map = self.repository.texts.get_parent_map(self.text_index)
3914
4188
        text_keys = self.repository.texts.keys()
3915
4189
        unused_keys = frozenset(text_keys) - set(self.text_index)
3916
4190
        for num, key in enumerate(self.text_index.iterkeys()):
3917
 
            if progress_bar is not None:
3918
 
                progress_bar.update('checking text graph', num, n_versions)
 
4191
            progress_bar.update('checking text graph', num, n_versions)
3919
4192
            correct_parents = self.calculate_file_version_parents(key)
3920
4193
            try:
3921
4194
                knit_parents = parent_map[key]
4006
4279
            else:
4007
4280
                new_pack.set_write_cache_size(1024*1024)
4008
4281
        for substream_type, substream in stream:
 
4282
            if 'stream' in debug.debug_flags:
 
4283
                mutter('inserting substream: %s', substream_type)
4009
4284
            if substream_type == 'texts':
4010
4285
                self.target_repo.texts.insert_record_stream(substream)
4011
4286
            elif substream_type == 'inventories':
4015
4290
                else:
4016
4291
                    self._extract_and_insert_inventories(
4017
4292
                        substream, src_serializer)
 
4293
            elif substream_type == 'inventory-deltas':
 
4294
                self._extract_and_insert_inventory_deltas(
 
4295
                    substream, src_serializer)
4018
4296
            elif substream_type == 'chk_bytes':
4019
4297
                # XXX: This doesn't support conversions, as it assumes the
4020
4298
                #      conversion was done in the fetch code.
4051
4329
                ):
4052
4330
                if versioned_file is None:
4053
4331
                    continue
 
4332
                # TODO: key is often going to be a StaticTuple object
 
4333
                #       I don't believe we can define a method by which
 
4334
                #       (prefix,) + StaticTuple will work, though we could
 
4335
                #       define a StaticTuple.sq_concat that would allow you to
 
4336
                #       pass in either a tuple or a StaticTuple as the second
 
4337
                #       object, so instead we could have:
 
4338
                #       StaticTuple(prefix) + key here...
4054
4339
                missing_keys.update((prefix,) + key for key in
4055
4340
                    versioned_file.get_missing_compression_parent_keys())
4056
4341
        except NotImplementedError:
4071
4356
            self.target_repo.pack(hint=hint)
4072
4357
        return [], set()
4073
4358
 
4074
 
    def _extract_and_insert_inventories(self, substream, serializer):
 
4359
    def _extract_and_insert_inventory_deltas(self, substream, serializer):
 
4360
        target_rich_root = self.target_repo._format.rich_root_data
 
4361
        target_tree_refs = self.target_repo._format.supports_tree_reference
 
4362
        for record in substream:
 
4363
            # Insert the delta directly
 
4364
            inventory_delta_bytes = record.get_bytes_as('fulltext')
 
4365
            deserialiser = inventory_delta.InventoryDeltaDeserializer()
 
4366
            try:
 
4367
                parse_result = deserialiser.parse_text_bytes(
 
4368
                    inventory_delta_bytes)
 
4369
            except inventory_delta.IncompatibleInventoryDelta, err:
 
4370
                trace.mutter("Incompatible delta: %s", err.msg)
 
4371
                raise errors.IncompatibleRevision(self.target_repo._format)
 
4372
            basis_id, new_id, rich_root, tree_refs, inv_delta = parse_result
 
4373
            revision_id = new_id
 
4374
            parents = [key[0] for key in record.parents]
 
4375
            self.target_repo.add_inventory_by_delta(
 
4376
                basis_id, inv_delta, revision_id, parents)
 
4377
 
 
4378
    def _extract_and_insert_inventories(self, substream, serializer,
 
4379
            parse_delta=None):
4075
4380
        """Generate a new inventory versionedfile in target, converting data.
4076
4381
 
4077
4382
        The inventory is retrieved from the source, (deserializing it), and
4078
4383
        stored in the target (reserializing it in a different format).
4079
4384
        """
 
4385
        target_rich_root = self.target_repo._format.rich_root_data
 
4386
        target_tree_refs = self.target_repo._format.supports_tree_reference
4080
4387
        for record in substream:
 
4388
            # It's not a delta, so it must be a fulltext in the source
 
4389
            # serializer's format.
4081
4390
            bytes = record.get_bytes_as('fulltext')
4082
4391
            revision_id = record.key[0]
4083
4392
            inv = serializer.read_inventory_from_string(bytes, revision_id)
4084
4393
            parents = [key[0] for key in record.parents]
4085
4394
            self.target_repo.add_inventory(revision_id, inv, parents)
 
4395
            # No need to keep holding this full inv in memory when the rest of
 
4396
            # the substream is likely to be all deltas.
 
4397
            del inv
4086
4398
 
4087
4399
    def _extract_and_insert_revisions(self, substream, serializer):
4088
4400
        for record in substream:
4137
4449
        return [('signatures', signatures), ('revisions', revisions)]
4138
4450
 
4139
4451
    def _generate_root_texts(self, revs):
4140
 
        """This will be called by __fetch between fetching weave texts and
 
4452
        """This will be called by get_stream between fetching weave texts and
4141
4453
        fetching the inventory weave.
4142
 
 
4143
 
        Subclasses should override this if they need to generate root texts
4144
 
        after fetching weave texts.
4145
4454
        """
4146
4455
        if self._rich_root_upgrade():
4147
 
            import bzrlib.fetch
4148
 
            return bzrlib.fetch.Inter1and2Helper(
 
4456
            return _mod_fetch.Inter1and2Helper(
4149
4457
                self.from_repository).generate_root_texts(revs)
4150
4458
        else:
4151
4459
            return []
4154
4462
        phase = 'file'
4155
4463
        revs = search.get_keys()
4156
4464
        graph = self.from_repository.get_graph()
4157
 
        revs = list(graph.iter_topo_order(revs))
 
4465
        revs = tsort.topo_sort(graph.get_parent_map(revs))
4158
4466
        data_to_fetch = self.from_repository.item_keys_introduced_by(revs)
4159
4467
        text_keys = []
4160
4468
        for knit_kind, file_id, revisions in data_to_fetch:
4179
4487
                # will be valid.
4180
4488
                for _ in self._generate_root_texts(revs):
4181
4489
                    yield _
4182
 
                # NB: This currently reopens the inventory weave in source;
4183
 
                # using a single stream interface instead would avoid this.
4184
 
                from_weave = self.from_repository.inventories
4185
4490
                # we fetch only the referenced inventories because we do not
4186
4491
                # know for unselected inventories whether all their required
4187
4492
                # texts are present in the other repository - it could be
4226
4531
            if not keys:
4227
4532
                # No need to stream something we don't have
4228
4533
                continue
 
4534
            if substream_kind == 'inventories':
 
4535
                # Some missing keys are genuinely ghosts, filter those out.
 
4536
                present = self.from_repository.inventories.get_parent_map(keys)
 
4537
                revs = [key[0] for key in present]
 
4538
                # Get the inventory stream more-or-less as we do for the
 
4539
                # original stream; there's no reason to assume that records
 
4540
                # direct from the source will be suitable for the sink.  (Think
 
4541
                # e.g. 2a -> 1.9-rich-root).
 
4542
                for info in self._get_inventory_stream(revs, missing=True):
 
4543
                    yield info
 
4544
                continue
 
4545
 
4229
4546
            # Ask for full texts always so that we don't need more round trips
4230
4547
            # after this stream.
4231
4548
            # Some of the missing keys are genuinely ghosts, so filter absent
4246
4563
        return (not self.from_repository._format.rich_root_data and
4247
4564
            self.to_format.rich_root_data)
4248
4565
 
4249
 
    def _get_inventory_stream(self, revision_ids):
 
4566
    def _get_inventory_stream(self, revision_ids, missing=False):
4250
4567
        from_format = self.from_repository._format
4251
 
        if (from_format.supports_chks and self.to_format.supports_chks
4252
 
            and (from_format._serializer == self.to_format._serializer)):
4253
 
            # Both sides support chks, and they use the same serializer, so it
4254
 
            # is safe to transmit the chk pages and inventory pages across
4255
 
            # as-is.
4256
 
            return self._get_chk_inventory_stream(revision_ids)
4257
 
        elif (not from_format.supports_chks):
4258
 
            # Source repository doesn't support chks. So we can transmit the
4259
 
            # inventories 'as-is' and either they are just accepted on the
4260
 
            # target, or the Sink will properly convert it.
4261
 
            return self._get_simple_inventory_stream(revision_ids)
 
4568
        if (from_format.supports_chks and self.to_format.supports_chks and
 
4569
            from_format.network_name() == self.to_format.network_name()):
 
4570
            raise AssertionError(
 
4571
                "this case should be handled by GroupCHKStreamSource")
 
4572
        elif 'forceinvdeltas' in debug.debug_flags:
 
4573
            return self._get_convertable_inventory_stream(revision_ids,
 
4574
                    delta_versus_null=missing)
 
4575
        elif from_format.network_name() == self.to_format.network_name():
 
4576
            # Same format.
 
4577
            return self._get_simple_inventory_stream(revision_ids,
 
4578
                    missing=missing)
 
4579
        elif (not from_format.supports_chks and not self.to_format.supports_chks
 
4580
                and from_format._serializer == self.to_format._serializer):
 
4581
            # Essentially the same format.
 
4582
            return self._get_simple_inventory_stream(revision_ids,
 
4583
                    missing=missing)
4262
4584
        else:
4263
 
            # XXX: Hack to make not-chk->chk fetch: copy the inventories as
4264
 
            #      inventories. Note that this should probably be done somehow
4265
 
            #      as part of bzrlib.repository.StreamSink. Except JAM couldn't
4266
 
            #      figure out how a non-chk repository could possibly handle
4267
 
            #      deserializing an inventory stream from a chk repo, as it
4268
 
            #      doesn't have a way to understand individual pages.
4269
 
            return self._get_convertable_inventory_stream(revision_ids)
 
4585
            # Any time we switch serializations, we want to use an
 
4586
            # inventory-delta based approach.
 
4587
            return self._get_convertable_inventory_stream(revision_ids,
 
4588
                    delta_versus_null=missing)
4270
4589
 
4271
 
    def _get_simple_inventory_stream(self, revision_ids):
 
4590
    def _get_simple_inventory_stream(self, revision_ids, missing=False):
 
4591
        # NB: This currently reopens the inventory weave in source;
 
4592
        # using a single stream interface instead would avoid this.
4272
4593
        from_weave = self.from_repository.inventories
 
4594
        if missing:
 
4595
            delta_closure = True
 
4596
        else:
 
4597
            delta_closure = not self.delta_on_metadata()
4273
4598
        yield ('inventories', from_weave.get_record_stream(
4274
4599
            [(rev_id,) for rev_id in revision_ids],
4275
 
            self.inventory_fetch_order(),
4276
 
            not self.delta_on_metadata()))
4277
 
 
4278
 
    def _get_chk_inventory_stream(self, revision_ids):
4279
 
        """Fetch the inventory texts, along with the associated chk maps."""
4280
 
        # We want an inventory outside of the search set, so that we can filter
4281
 
        # out uninteresting chk pages. For now we use
4282
 
        # _find_revision_outside_set, but if we had a Search with cut_revs, we
4283
 
        # could use that instead.
4284
 
        start_rev_id = self.from_repository._find_revision_outside_set(
4285
 
                            revision_ids)
4286
 
        start_rev_key = (start_rev_id,)
4287
 
        inv_keys_to_fetch = [(rev_id,) for rev_id in revision_ids]
4288
 
        if start_rev_id != _mod_revision.NULL_REVISION:
4289
 
            inv_keys_to_fetch.append((start_rev_id,))
4290
 
        # Any repo that supports chk_bytes must also support out-of-order
4291
 
        # insertion. At least, that is how we expect it to work
4292
 
        # We use get_record_stream instead of iter_inventories because we want
4293
 
        # to be able to insert the stream as well. We could instead fetch
4294
 
        # allowing deltas, and then iter_inventories, but we don't know whether
4295
 
        # source or target is more 'local' anway.
4296
 
        inv_stream = self.from_repository.inventories.get_record_stream(
4297
 
            inv_keys_to_fetch, 'unordered',
4298
 
            True) # We need them as full-texts so we can find their references
4299
 
        uninteresting_chk_roots = set()
4300
 
        interesting_chk_roots = set()
4301
 
        def filter_inv_stream(inv_stream):
4302
 
            for idx, record in enumerate(inv_stream):
4303
 
                ### child_pb.update('fetch inv', idx, len(inv_keys_to_fetch))
4304
 
                bytes = record.get_bytes_as('fulltext')
4305
 
                chk_inv = inventory.CHKInventory.deserialise(
4306
 
                    self.from_repository.chk_bytes, bytes, record.key)
4307
 
                if record.key == start_rev_key:
4308
 
                    uninteresting_chk_roots.add(chk_inv.id_to_entry.key())
4309
 
                    p_id_map = chk_inv.parent_id_basename_to_file_id
4310
 
                    if p_id_map is not None:
4311
 
                        uninteresting_chk_roots.add(p_id_map.key())
4312
 
                else:
4313
 
                    yield record
4314
 
                    interesting_chk_roots.add(chk_inv.id_to_entry.key())
4315
 
                    p_id_map = chk_inv.parent_id_basename_to_file_id
4316
 
                    if p_id_map is not None:
4317
 
                        interesting_chk_roots.add(p_id_map.key())
4318
 
        ### pb.update('fetch inventory', 0, 2)
4319
 
        yield ('inventories', filter_inv_stream(inv_stream))
4320
 
        # Now that we have worked out all of the interesting root nodes, grab
4321
 
        # all of the interesting pages and insert them
4322
 
        ### pb.update('fetch inventory', 1, 2)
4323
 
        interesting = chk_map.iter_interesting_nodes(
4324
 
            self.from_repository.chk_bytes, interesting_chk_roots,
4325
 
            uninteresting_chk_roots)
4326
 
        def to_stream_adapter():
4327
 
            """Adapt the iter_interesting_nodes result to a single stream.
4328
 
 
4329
 
            iter_interesting_nodes returns records as it processes them, along
4330
 
            with keys. However, we only want to return the records themselves.
4331
 
            """
4332
 
            for record, items in interesting:
4333
 
                if record is not None:
4334
 
                    yield record
4335
 
        # XXX: We could instead call get_record_stream(records.keys())
4336
 
        #      ATM, this will always insert the records as fulltexts, and
4337
 
        #      requires that you can hang on to records once you have gone
4338
 
        #      on to the next one. Further, it causes the target to
4339
 
        #      recompress the data. Testing shows it to be faster than
4340
 
        #      requesting the records again, though.
4341
 
        yield ('chk_bytes', to_stream_adapter())
4342
 
        ### pb.update('fetch inventory', 2, 2)
4343
 
 
4344
 
    def _get_convertable_inventory_stream(self, revision_ids):
4345
 
        # XXX: One of source or target is using chks, and they don't have
4346
 
        #      compatible serializations. The StreamSink code expects to be
4347
 
        #      able to convert on the target, so we need to put
4348
 
        #      bytes-on-the-wire that can be converted
4349
 
        yield ('inventories', self._stream_invs_as_fulltexts(revision_ids))
4350
 
 
4351
 
    def _stream_invs_as_fulltexts(self, revision_ids):
 
4600
            self.inventory_fetch_order(), delta_closure))
 
4601
 
 
4602
    def _get_convertable_inventory_stream(self, revision_ids,
 
4603
                                          delta_versus_null=False):
 
4604
        # The two formats are sufficiently different that there is no fast
 
4605
        # path, so we need to send just inventorydeltas, which any
 
4606
        # sufficiently modern client can insert into any repository.
 
4607
        # The StreamSink code expects to be able to
 
4608
        # convert on the target, so we need to put bytes-on-the-wire that can
 
4609
        # be converted.  That means inventory deltas (if the remote is <1.19,
 
4610
        # RemoteStreamSink will fallback to VFS to insert the deltas).
 
4611
        yield ('inventory-deltas',
 
4612
           self._stream_invs_as_deltas(revision_ids,
 
4613
                                       delta_versus_null=delta_versus_null))
 
4614
 
 
4615
    def _stream_invs_as_deltas(self, revision_ids, delta_versus_null=False):
 
4616
        """Return a stream of inventory-deltas for the given rev ids.
 
4617
 
 
4618
        :param revision_ids: The list of inventories to transmit
 
4619
        :param delta_versus_null: Don't try to find a minimal delta for this
 
4620
            entry, instead compute the delta versus the NULL_REVISION. This
 
4621
            effectively streams a complete inventory. Used for stuff like
 
4622
            filling in missing parents, etc.
 
4623
        """
4352
4624
        from_repo = self.from_repository
4353
 
        from_serializer = from_repo._format._serializer
4354
4625
        revision_keys = [(rev_id,) for rev_id in revision_ids]
4355
4626
        parent_map = from_repo.inventories.get_parent_map(revision_keys)
4356
 
        for inv in self.from_repository.iter_inventories(revision_ids):
4357
 
            # XXX: This is a bit hackish, but it works. Basically,
4358
 
            #      CHKSerializer 'accidentally' supports
4359
 
            #      read/write_inventory_to_string, even though that is never
4360
 
            #      the format that is stored on disk. It *does* give us a
4361
 
            #      single string representation for an inventory, so live with
4362
 
            #      it for now.
4363
 
            #      This would be far better if we had a 'serialized inventory
4364
 
            #      delta' form. Then we could use 'inventory._make_delta', and
4365
 
            #      transmit that. This would both be faster to generate, and
4366
 
            #      result in fewer bytes-on-the-wire.
4367
 
            as_bytes = from_serializer.write_inventory_to_string(inv)
 
4627
        # XXX: possibly repos could implement a more efficient iter_inv_deltas
 
4628
        # method...
 
4629
        inventories = self.from_repository.iter_inventories(
 
4630
            revision_ids, 'topological')
 
4631
        format = from_repo._format
 
4632
        invs_sent_so_far = set([_mod_revision.NULL_REVISION])
 
4633
        inventory_cache = lru_cache.LRUCache(50)
 
4634
        null_inventory = from_repo.revision_tree(
 
4635
            _mod_revision.NULL_REVISION).inventory
 
4636
        # XXX: ideally the rich-root/tree-refs flags would be per-revision, not
 
4637
        # per-repo (e.g.  streaming a non-rich-root revision out of a rich-root
 
4638
        # repo back into a non-rich-root repo ought to be allowed)
 
4639
        serializer = inventory_delta.InventoryDeltaSerializer(
 
4640
            versioned_root=format.rich_root_data,
 
4641
            tree_references=format.supports_tree_reference)
 
4642
        for inv in inventories:
4368
4643
            key = (inv.revision_id,)
4369
4644
            parent_keys = parent_map.get(key, ())
 
4645
            delta = None
 
4646
            if not delta_versus_null and parent_keys:
 
4647
                # The caller did not ask for complete inventories and we have
 
4648
                # some parents that we can delta against.  Make a delta against
 
4649
                # each parent so that we can find the smallest.
 
4650
                parent_ids = [parent_key[0] for parent_key in parent_keys]
 
4651
                for parent_id in parent_ids:
 
4652
                    if parent_id not in invs_sent_so_far:
 
4653
                        # We don't know that the remote side has this basis, so
 
4654
                        # we can't use it.
 
4655
                        continue
 
4656
                    if parent_id == _mod_revision.NULL_REVISION:
 
4657
                        parent_inv = null_inventory
 
4658
                    else:
 
4659
                        parent_inv = inventory_cache.get(parent_id, None)
 
4660
                        if parent_inv is None:
 
4661
                            parent_inv = from_repo.get_inventory(parent_id)
 
4662
                    candidate_delta = inv._make_delta(parent_inv)
 
4663
                    if (delta is None or
 
4664
                        len(delta) > len(candidate_delta)):
 
4665
                        delta = candidate_delta
 
4666
                        basis_id = parent_id
 
4667
            if delta is None:
 
4668
                # Either none of the parents ended up being suitable, or we
 
4669
                # were asked to delta against NULL
 
4670
                basis_id = _mod_revision.NULL_REVISION
 
4671
                delta = inv._make_delta(null_inventory)
 
4672
            invs_sent_so_far.add(inv.revision_id)
 
4673
            inventory_cache[inv.revision_id] = inv
 
4674
            delta_serialized = ''.join(
 
4675
                serializer.delta_to_lines(basis_id, key[-1], delta))
4370
4676
            yield versionedfile.FulltextContentFactory(
4371
 
                key, parent_keys, None, as_bytes)
 
4677
                key, parent_keys, None, delta_serialized)
4372
4678
 
4373
4679
 
4374
4680
def _iter_for_revno(repo, partial_history_cache, stop_index=None,