~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/repofmt/groupcompress_repo.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2009-04-08 12:39:40 UTC
  • mfrom: (4266.2.1 tests.output)
  • Revision ID: pqm@pqm.ubuntu.com-20090408123940-kaho6cwr21163fjn
(robertc) Remove clutter from bzr selftest --list. (Robert Collins)

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2008, 2009, 2010 Canonical Ltd
 
1
# Copyright (C) 2008, 2009 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
14
14
# along with this program; if not, write to the Free Software
15
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
16
16
 
17
 
"""Repository formats using CHK inventories and groupcompress compression."""
 
17
"""Repostory formats using CHK inventories and groupcompress compression."""
18
18
 
19
19
import time
20
20
 
29
29
    knit,
30
30
    osutils,
31
31
    pack,
 
32
    remote,
 
33
    repository,
32
34
    revision as _mod_revision,
33
35
    trace,
34
36
    ui,
37
39
    BTreeGraphIndex,
38
40
    BTreeBuilder,
39
41
    )
 
42
from bzrlib.index import GraphIndex, GraphIndexBuilder
40
43
from bzrlib.groupcompress import (
41
44
    _GCGraphIndex,
42
45
    GroupCompressVersionedFiles,
45
48
    Pack,
46
49
    NewPack,
47
50
    KnitPackRepository,
48
 
    KnitPackStreamSource,
49
51
    PackRootCommitBuilder,
50
52
    RepositoryPackCollection,
51
53
    RepositoryFormatPack,
52
 
    ResumedPack,
53
54
    Packer,
54
55
    )
55
 
from bzrlib.static_tuple import StaticTuple
56
56
 
57
57
 
58
58
class GCPack(NewPack):
87
87
            # have a regular 2-list index giving parents and compression
88
88
            # source.
89
89
            index_builder_class(reference_lists=1),
90
 
            # Texts: per file graph, for all fileids - so one reference list
91
 
            # and two elements in the key tuple.
 
90
            # Texts: compression and per file graph, for all fileids - so two
 
91
            # reference lists and two elements in the key tuple.
92
92
            index_builder_class(reference_lists=1, key_elements=2),
93
93
            # Signatures: Just blobs to store, no compression, no parents
94
94
            # listing.
154
154
        self._writer.begin()
155
155
        # what state is the pack in? (open, finished, aborted)
156
156
        self._state = 'open'
157
 
        # no name until we finish writing the content
158
 
        self.name = None
159
157
 
160
158
    def _check_references(self):
161
159
        """Make sure our external references are present.
165
163
        have deltas based on a fallback repository.
166
164
        (See <https://bugs.launchpad.net/bzr/+bug/288751>)
167
165
        """
168
 
        # Groupcompress packs don't have any external references, arguably CHK
169
 
        # pages have external references, but we cannot 'cheaply' determine
170
 
        # them without actually walking all of the chk pages.
171
 
 
172
 
 
173
 
class ResumedGCPack(ResumedPack):
174
 
 
175
 
    def _check_references(self):
176
 
        """Make sure our external compression parents are present."""
177
 
        # See GCPack._check_references for why this is empty
178
 
 
179
 
    def _get_external_refs(self, index):
180
 
        # GC repositories don't have compression parents external to a given
181
 
        # pack file
182
 
        return set()
 
166
        # Groupcompress packs don't have any external references
183
167
 
184
168
 
185
169
class GCCHKPacker(Packer):
218
202
            p_id_roots_set = set()
219
203
            stream = source_vf.get_record_stream(keys, 'groupcompress', True)
220
204
            for idx, record in enumerate(stream):
221
 
                # Inventories should always be with revisions; assume success.
222
205
                bytes = record.get_bytes_as('fulltext')
223
206
                chk_inv = inventory.CHKInventory.deserialise(None, bytes,
224
207
                                                             record.key)
262
245
        remaining_keys = set(keys)
263
246
        counter = [0]
264
247
        if self._gather_text_refs:
 
248
            bytes_to_info = inventory.CHKInventory._bytes_to_utf8name_key
265
249
            self._text_refs = set()
266
250
        def _get_referenced_stream(root_keys, parse_leaf_nodes=False):
267
251
            cur_keys = root_keys
288
272
                    # Store is None, because we know we have a LeafNode, and we
289
273
                    # just want its entries
290
274
                    for file_id, bytes in node.iteritems(None):
291
 
                        self._text_refs.add(chk_map._bytes_to_text_key(bytes))
 
275
                        name_utf8, file_id, revision_id = bytes_to_info(bytes)
 
276
                        self._text_refs.add((file_id, revision_id))
292
277
                def next_stream():
293
278
                    stream = source_vf.get_record_stream(cur_keys,
294
279
                                                         'as-requested', True)
295
280
                    for record in stream:
296
 
                        if record.storage_kind == 'absent':
297
 
                            # An absent CHK record: we assume that the missing
298
 
                            # record is in a different pack - e.g. a page not
299
 
                            # altered by the commit we're packing.
300
 
                            continue
301
281
                        bytes = record.get_bytes_as('fulltext')
302
282
                        # We don't care about search_key_func for this code,
303
283
                        # because we only care about external references.
350
330
        """Build a VersionedFiles instance on top of this group of packs."""
351
331
        index_name = index_name + '_index'
352
332
        index_to_pack = {}
353
 
        access = knit._DirectPackAccess(index_to_pack,
354
 
                                        reload_func=self._reload_func)
 
333
        access = knit._DirectPackAccess(index_to_pack)
355
334
        if for_write:
356
335
            # Use new_pack
357
336
            if self.new_pack is None:
411
390
 
412
391
    def _copy_inventory_texts(self):
413
392
        source_vf, target_vf = self._build_vfs('inventory', True, True)
414
 
        # It is not sufficient to just use self.revision_keys, as stacked
415
 
        # repositories can have more inventories than they have revisions.
416
 
        # One alternative would be to do something with
417
 
        # get_parent_map(self.revision_keys), but that shouldn't be any faster
418
 
        # than this.
419
 
        inventory_keys = source_vf.keys()
420
 
        missing_inventories = set(self.revision_keys).difference(inventory_keys)
421
 
        if missing_inventories:
422
 
            missing_inventories = sorted(missing_inventories)
423
 
            raise ValueError('We are missing inventories for revisions: %s'
424
 
                % (missing_inventories,))
425
 
        self._copy_stream(source_vf, target_vf, inventory_keys,
 
393
        self._copy_stream(source_vf, target_vf, self.revision_keys,
426
394
                          'inventories', self._get_filtered_inv_stream, 2)
427
395
 
428
396
    def _copy_chk_texts(self):
455
423
        #      is grabbing too many keys...
456
424
        text_keys = source_vf.keys()
457
425
        self._copy_stream(source_vf, target_vf, text_keys,
458
 
                          'texts', self._get_progress_stream, 4)
 
426
                          'text', self._get_progress_stream, 4)
459
427
 
460
428
    def _copy_signature_texts(self):
461
429
        source_vf, target_vf = self._build_vfs('signature', False, False)
478
446
        if not self._use_pack(self.new_pack):
479
447
            self.new_pack.abort()
480
448
            return None
481
 
        self.new_pack.finish_content()
482
 
        if len(self.packs) == 1:
483
 
            old_pack = self.packs[0]
484
 
            if old_pack.name == self.new_pack._hash.hexdigest():
485
 
                # The single old pack was already optimally packed.
486
 
                trace.mutter('single pack %s was already optimally packed',
487
 
                    old_pack.name)
488
 
                self.new_pack.abort()
489
 
                return None
490
449
        self.pb.update('finishing repack', 6, 7)
491
450
        self.new_pack.finish()
492
451
        self._pack_collection.allocate(self.new_pack)
581
540
class GCRepositoryPackCollection(RepositoryPackCollection):
582
541
 
583
542
    pack_factory = GCPack
584
 
    resumed_pack_factory = ResumedGCPack
585
 
 
586
 
    def _check_new_inventories(self):
587
 
        """Detect missing inventories or chk root entries for the new revisions
588
 
        in this write group.
589
 
 
590
 
        :returns: list of strs, summarising any problems found.  If the list is
591
 
            empty no problems were found.
592
 
        """
593
 
        # Ensure that all revisions added in this write group have:
594
 
        #   - corresponding inventories,
595
 
        #   - chk root entries for those inventories,
596
 
        #   - and any present parent inventories have their chk root
597
 
        #     entries too.
598
 
        # And all this should be independent of any fallback repository.
599
 
        problems = []
600
 
        key_deps = self.repo.revisions._index._key_dependencies
601
 
        new_revisions_keys = key_deps.get_new_keys()
602
 
        no_fallback_inv_index = self.repo.inventories._index
603
 
        no_fallback_chk_bytes_index = self.repo.chk_bytes._index
604
 
        no_fallback_texts_index = self.repo.texts._index
605
 
        inv_parent_map = no_fallback_inv_index.get_parent_map(
606
 
            new_revisions_keys)
607
 
        # Are any inventories for corresponding to the new revisions missing?
608
 
        corresponding_invs = set(inv_parent_map)
609
 
        missing_corresponding = set(new_revisions_keys)
610
 
        missing_corresponding.difference_update(corresponding_invs)
611
 
        if missing_corresponding:
612
 
            problems.append("inventories missing for revisions %s" %
613
 
                (sorted(missing_corresponding),))
614
 
            return problems
615
 
        # Are any chk root entries missing for any inventories?  This includes
616
 
        # any present parent inventories, which may be used when calculating
617
 
        # deltas for streaming.
618
 
        all_inv_keys = set(corresponding_invs)
619
 
        for parent_inv_keys in inv_parent_map.itervalues():
620
 
            all_inv_keys.update(parent_inv_keys)
621
 
        # Filter out ghost parents.
622
 
        all_inv_keys.intersection_update(
623
 
            no_fallback_inv_index.get_parent_map(all_inv_keys))
624
 
        parent_invs_only_keys = all_inv_keys.symmetric_difference(
625
 
            corresponding_invs)
626
 
        all_missing = set()
627
 
        inv_ids = [key[-1] for key in all_inv_keys]
628
 
        parent_invs_only_ids = [key[-1] for key in parent_invs_only_keys]
629
 
        root_key_info = _build_interesting_key_sets(
630
 
            self.repo, inv_ids, parent_invs_only_ids)
631
 
        expected_chk_roots = root_key_info.all_keys()
632
 
        present_chk_roots = no_fallback_chk_bytes_index.get_parent_map(
633
 
            expected_chk_roots)
634
 
        missing_chk_roots = expected_chk_roots.difference(present_chk_roots)
635
 
        if missing_chk_roots:
636
 
            problems.append("missing referenced chk root keys: %s"
637
 
                % (sorted(missing_chk_roots),))
638
 
            # Don't bother checking any further.
639
 
            return problems
640
 
        # Find all interesting chk_bytes records, and make sure they are
641
 
        # present, as well as the text keys they reference.
642
 
        chk_bytes_no_fallbacks = self.repo.chk_bytes.without_fallbacks()
643
 
        chk_bytes_no_fallbacks._search_key_func = \
644
 
            self.repo.chk_bytes._search_key_func
645
 
        chk_diff = chk_map.iter_interesting_nodes(
646
 
            chk_bytes_no_fallbacks, root_key_info.interesting_root_keys,
647
 
            root_key_info.uninteresting_root_keys)
648
 
        text_keys = set()
649
 
        try:
650
 
            for record in _filter_text_keys(chk_diff, text_keys,
651
 
                                            chk_map._bytes_to_text_key):
652
 
                pass
653
 
        except errors.NoSuchRevision, e:
654
 
            # XXX: It would be nice if we could give a more precise error here.
655
 
            problems.append("missing chk node(s) for id_to_entry maps")
656
 
        chk_diff = chk_map.iter_interesting_nodes(
657
 
            chk_bytes_no_fallbacks, root_key_info.interesting_pid_root_keys,
658
 
            root_key_info.uninteresting_pid_root_keys)
659
 
        try:
660
 
            for interesting_rec, interesting_map in chk_diff:
661
 
                pass
662
 
        except errors.NoSuchRevision, e:
663
 
            problems.append(
664
 
                "missing chk node(s) for parent_id_basename_to_file_id maps")
665
 
        present_text_keys = no_fallback_texts_index.get_parent_map(text_keys)
666
 
        missing_text_keys = text_keys.difference(present_text_keys)
667
 
        if missing_text_keys:
668
 
            problems.append("missing text keys: %r"
669
 
                % (sorted(missing_text_keys),))
670
 
        return problems
 
543
 
 
544
    def _already_packed(self):
 
545
        """Is the collection already packed?"""
 
546
        # Always repack GC repositories for now
 
547
        return False
671
548
 
672
549
    def _execute_pack_operations(self, pack_operations,
673
550
                                 _packer_class=GCCHKPacker,
687
564
            packer = GCCHKPacker(self, packs, '.autopack',
688
565
                                 reload_func=reload_func)
689
566
            try:
690
 
                result = packer.pack()
 
567
                packer.pack()
691
568
            except errors.RetryWithNewPacks:
692
569
                # An exception is propagating out of this context, make sure
693
570
                # this packer has cleaned up. Packer() doesn't set its new_pack
696
573
                if packer.new_pack is not None:
697
574
                    packer.new_pack.abort()
698
575
                raise
699
 
            if result is None:
700
 
                return
701
576
            for pack in packs:
702
577
                self._remove_pack_from_memory(pack)
703
578
        # record the newly available packs and stop advertising the old
704
579
        # packs
705
 
        to_be_obsoleted = []
706
 
        for _, packs in pack_operations:
707
 
            to_be_obsoleted.extend(packs)
708
 
        result = self._save_pack_names(clear_obsolete_packs=True,
709
 
                                       obsolete_packs=to_be_obsoleted)
710
 
        return result
 
580
        self._save_pack_names(clear_obsolete_packs=True)
 
581
        # Move the old packs out of the way now they are no longer referenced.
 
582
        for revision_count, packs in pack_operations:
 
583
            self._obsolete_packs(packs)
711
584
 
712
585
 
713
586
class CHKInventoryRepository(KnitPackRepository):
731
604
        self.inventories = GroupCompressVersionedFiles(
732
605
            _GCGraphIndex(self._pack_collection.inventory_index.combined_index,
733
606
                add_callback=self._pack_collection.inventory_index.add_callback,
734
 
                parents=True, is_locked=self.is_locked,
735
 
                inconsistency_fatal=False),
 
607
                parents=True, is_locked=self.is_locked),
736
608
            access=self._pack_collection.inventory_index.data_access)
737
609
        self.revisions = GroupCompressVersionedFiles(
738
610
            _GCGraphIndex(self._pack_collection.revision_index.combined_index,
739
611
                add_callback=self._pack_collection.revision_index.add_callback,
740
 
                parents=True, is_locked=self.is_locked,
741
 
                track_external_parent_refs=True, track_new_keys=True),
 
612
                parents=True, is_locked=self.is_locked),
742
613
            access=self._pack_collection.revision_index.data_access,
743
614
            delta=False)
744
615
        self.signatures = GroupCompressVersionedFiles(
745
616
            _GCGraphIndex(self._pack_collection.signature_index.combined_index,
746
617
                add_callback=self._pack_collection.signature_index.add_callback,
747
 
                parents=False, is_locked=self.is_locked,
748
 
                inconsistency_fatal=False),
 
618
                parents=False, is_locked=self.is_locked),
749
619
            access=self._pack_collection.signature_index.data_access,
750
620
            delta=False)
751
621
        self.texts = GroupCompressVersionedFiles(
752
622
            _GCGraphIndex(self._pack_collection.text_index.combined_index,
753
623
                add_callback=self._pack_collection.text_index.add_callback,
754
 
                parents=True, is_locked=self.is_locked,
755
 
                inconsistency_fatal=False),
 
624
                parents=True, is_locked=self.is_locked),
756
625
            access=self._pack_collection.text_index.data_access)
757
626
        # No parents, individual CHK pages don't have specific ancestry
758
627
        self.chk_bytes = GroupCompressVersionedFiles(
759
628
            _GCGraphIndex(self._pack_collection.chk_index.combined_index,
760
629
                add_callback=self._pack_collection.chk_index.add_callback,
761
 
                parents=False, is_locked=self.is_locked,
762
 
                inconsistency_fatal=False),
 
630
                parents=False, is_locked=self.is_locked),
763
631
            access=self._pack_collection.chk_index.data_access)
764
 
        search_key_name = self._format._serializer.search_key_name
765
 
        search_key_func = chk_map.search_key_registry.get(search_key_name)
766
 
        self.chk_bytes._search_key_func = search_key_func
767
632
        # True when the repository object is 'write locked' (as opposed to the
768
633
        # physical lock only taken out around changes to the pack-names list.)
769
634
        # Another way to represent this would be a decorator around the control
792
657
        return self._inventory_add_lines(revision_id, parents,
793
658
            inv_lines, check_content=False)
794
659
 
795
 
    def _create_inv_from_null(self, delta, revision_id):
796
 
        """This will mutate new_inv directly.
797
 
 
798
 
        This is a simplified form of create_by_apply_delta which knows that all
799
 
        the old values must be None, so everything is a create.
800
 
        """
801
 
        serializer = self._format._serializer
802
 
        new_inv = inventory.CHKInventory(serializer.search_key_name)
803
 
        new_inv.revision_id = revision_id
804
 
        entry_to_bytes = new_inv._entry_to_bytes
805
 
        id_to_entry_dict = {}
806
 
        parent_id_basename_dict = {}
807
 
        for old_path, new_path, file_id, entry in delta:
808
 
            if old_path is not None:
809
 
                raise ValueError('Invalid delta, somebody tried to delete %r'
810
 
                                 ' from the NULL_REVISION'
811
 
                                 % ((old_path, file_id),))
812
 
            if new_path is None:
813
 
                raise ValueError('Invalid delta, delta from NULL_REVISION has'
814
 
                                 ' no new_path %r' % (file_id,))
815
 
            if new_path == '':
816
 
                new_inv.root_id = file_id
817
 
                parent_id_basename_key = StaticTuple('', '').intern()
818
 
            else:
819
 
                utf8_entry_name = entry.name.encode('utf-8')
820
 
                parent_id_basename_key = StaticTuple(entry.parent_id,
821
 
                                                     utf8_entry_name).intern()
822
 
            new_value = entry_to_bytes(entry)
823
 
            # Populate Caches?
824
 
            # new_inv._path_to_fileid_cache[new_path] = file_id
825
 
            key = StaticTuple(file_id).intern()
826
 
            id_to_entry_dict[key] = new_value
827
 
            parent_id_basename_dict[parent_id_basename_key] = file_id
828
 
 
829
 
        new_inv._populate_from_dicts(self.chk_bytes, id_to_entry_dict,
830
 
            parent_id_basename_dict, maximum_size=serializer.maximum_size)
831
 
        return new_inv
832
 
 
833
660
    def add_inventory_by_delta(self, basis_revision_id, delta, new_revision_id,
834
661
                               parents, basis_inv=None, propagate_caches=False):
835
662
        """Add a new inventory expressed as a delta against another revision.
855
682
            repository format specific) of the serialized inventory, and the
856
683
            resulting inventory.
857
684
        """
 
685
        if basis_revision_id == _mod_revision.NULL_REVISION:
 
686
            return KnitPackRepository.add_inventory_by_delta(self,
 
687
                basis_revision_id, delta, new_revision_id, parents)
858
688
        if not self.is_in_write_group():
859
689
            raise AssertionError("%r not in write group" % (self,))
860
690
        _mod_revision.check_not_reserved_id(new_revision_id)
861
 
        basis_tree = None
862
 
        if basis_inv is None:
863
 
            if basis_revision_id == _mod_revision.NULL_REVISION:
864
 
                new_inv = self._create_inv_from_null(delta, new_revision_id)
865
 
                if new_inv.root_id is None:
866
 
                    raise errors.RootMissing()
867
 
                inv_lines = new_inv.to_lines()
868
 
                return self._inventory_add_lines(new_revision_id, parents,
869
 
                    inv_lines, check_content=False), new_inv
870
 
            else:
871
 
                basis_tree = self.revision_tree(basis_revision_id)
872
 
                basis_tree.lock_read()
 
691
        basis_tree = self.revision_tree(basis_revision_id)
 
692
        basis_tree.lock_read()
 
693
        try:
 
694
            if basis_inv is None:
873
695
                basis_inv = basis_tree.inventory
874
 
        try:
875
696
            result = basis_inv.create_by_apply_delta(delta, new_revision_id,
876
697
                propagate_caches=propagate_caches)
877
698
            inv_lines = result.to_lines()
878
699
            return self._inventory_add_lines(new_revision_id, parents,
879
700
                inv_lines, check_content=False), result
880
701
        finally:
881
 
            if basis_tree is not None:
882
 
                basis_tree.unlock()
883
 
 
884
 
    def _deserialise_inventory(self, revision_id, bytes):
885
 
        return inventory.CHKInventory.deserialise(self.chk_bytes, bytes,
886
 
            (revision_id,))
887
 
 
888
 
    def _iter_inventories(self, revision_ids, ordering):
 
702
            basis_tree.unlock()
 
703
 
 
704
    def _iter_inventories(self, revision_ids):
889
705
        """Iterate over many inventory objects."""
890
 
        if ordering is None:
891
 
            ordering = 'unordered'
892
706
        keys = [(revision_id,) for revision_id in revision_ids]
893
 
        stream = self.inventories.get_record_stream(keys, ordering, True)
 
707
        stream = self.inventories.get_record_stream(keys, 'unordered', True)
894
708
        texts = {}
895
709
        for record in stream:
896
710
            if record.storage_kind != 'absent':
900
714
        for key in keys:
901
715
            yield inventory.CHKInventory.deserialise(self.chk_bytes, texts[key], key)
902
716
 
903
 
    def _iter_inventory_xmls(self, revision_ids, ordering):
904
 
        # Without a native 'xml' inventory, this method doesn't make sense.
905
 
        # However older working trees, and older bundles want it - so we supply
906
 
        # it allowing _get_inventory_xml to work. Bundles currently use the
907
 
        # serializer directly; this also isn't ideal, but there isn't an xml
908
 
        # iteration interface offered at all for repositories. We could make
909
 
        # _iter_inventory_xmls be part of the contract, even if kept private.
910
 
        inv_to_str = self._serializer.write_inventory_to_string
911
 
        for inv in self.iter_inventories(revision_ids, ordering=ordering):
912
 
            yield inv_to_str(inv), inv.revision_id
913
 
 
914
 
    def _find_present_inventory_keys(self, revision_keys):
915
 
        parent_map = self.inventories.get_parent_map(revision_keys)
916
 
        present_inventory_keys = set(k for k in parent_map)
917
 
        return present_inventory_keys
 
717
    def _iter_inventory_xmls(self, revision_ids):
 
718
        # Without a native 'xml' inventory, this method doesn't make sense, so
 
719
        # make it raise to trap naughty direct users.
 
720
        raise NotImplementedError(self._iter_inventory_xmls)
 
721
 
 
722
    def _find_revision_outside_set(self, revision_ids):
 
723
        revision_set = frozenset(revision_ids)
 
724
        for revid in revision_ids:
 
725
            parent_ids = self.get_parent_map([revid]).get(revid, ())
 
726
            for parent in parent_ids:
 
727
                if parent in revision_set:
 
728
                    # Parent is not outside the set
 
729
                    continue
 
730
                if parent not in self.get_parent_map([parent]):
 
731
                    # Parent is a ghost
 
732
                    continue
 
733
                return parent
 
734
        return _mod_revision.NULL_REVISION
 
735
 
 
736
    def _find_file_keys_to_fetch(self, revision_ids, pb):
 
737
        rich_root = self.supports_rich_root()
 
738
        revision_outside_set = self._find_revision_outside_set(revision_ids)
 
739
        if revision_outside_set == _mod_revision.NULL_REVISION:
 
740
            uninteresting_root_keys = set()
 
741
        else:
 
742
            uninteresting_inv = self.get_inventory(revision_outside_set)
 
743
            uninteresting_root_keys = set([uninteresting_inv.id_to_entry.key()])
 
744
        interesting_root_keys = set()
 
745
        for idx, inv in enumerate(self.iter_inventories(revision_ids)):
 
746
            interesting_root_keys.add(inv.id_to_entry.key())
 
747
        revision_ids = frozenset(revision_ids)
 
748
        file_id_revisions = {}
 
749
        bytes_to_info = inventory.CHKInventory._bytes_to_utf8name_key
 
750
        for record, items in chk_map.iter_interesting_nodes(self.chk_bytes,
 
751
                    interesting_root_keys, uninteresting_root_keys,
 
752
                    pb=pb):
 
753
            # This is cheating a bit to use the last grabbed 'inv', but it
 
754
            # works
 
755
            for name, bytes in items:
 
756
                (name_utf8, file_id, revision_id) = bytes_to_info(bytes)
 
757
                if not rich_root and name_utf8 == '':
 
758
                    continue
 
759
                if revision_id in revision_ids:
 
760
                    # Would we rather build this up into file_id => revision
 
761
                    # maps?
 
762
                    try:
 
763
                        file_id_revisions[file_id].add(revision_id)
 
764
                    except KeyError:
 
765
                        file_id_revisions[file_id] = set([revision_id])
 
766
        for file_id, revisions in file_id_revisions.iteritems():
 
767
            yield ('file', file_id, revisions)
918
768
 
919
769
    def fileids_altered_by_revision_ids(self, revision_ids, _inv_weave=None):
920
770
        """Find the file ids and versions affected by revisions.
926
776
            revision_ids. Each altered file-ids has the exact revision_ids that
927
777
            altered it listed explicitly.
928
778
        """
929
 
        rich_root = self.supports_rich_root()
930
 
        bytes_to_info = inventory.CHKInventory._bytes_to_utf8name_key
931
 
        file_id_revisions = {}
 
779
        rich_roots = self.supports_rich_root()
 
780
        result = {}
932
781
        pb = ui.ui_factory.nested_progress_bar()
933
782
        try:
934
 
            revision_keys = [(r,) for r in revision_ids]
935
 
            parent_keys = self._find_parent_keys_of_revisions(revision_keys)
936
 
            # TODO: instead of using _find_present_inventory_keys, change the
937
 
            #       code paths to allow missing inventories to be tolerated.
938
 
            #       However, we only want to tolerate missing parent
939
 
            #       inventories, not missing inventories for revision_ids
940
 
            present_parent_inv_keys = self._find_present_inventory_keys(
941
 
                                        parent_keys)
942
 
            present_parent_inv_ids = set(
943
 
                [k[-1] for k in present_parent_inv_keys])
944
 
            inventories_to_read = set(revision_ids)
945
 
            inventories_to_read.update(present_parent_inv_ids)
946
 
            root_key_info = _build_interesting_key_sets(
947
 
                self, inventories_to_read, present_parent_inv_ids)
948
 
            interesting_root_keys = root_key_info.interesting_root_keys
949
 
            uninteresting_root_keys = root_key_info.uninteresting_root_keys
950
 
            chk_bytes = self.chk_bytes
951
 
            for record, items in chk_map.iter_interesting_nodes(chk_bytes,
952
 
                        interesting_root_keys, uninteresting_root_keys,
953
 
                        pb=pb):
954
 
                for name, bytes in items:
955
 
                    (name_utf8, file_id, revision_id) = bytes_to_info(bytes)
956
 
                    # TODO: consider interning file_id, revision_id here, or
957
 
                    #       pushing that intern() into bytes_to_info()
958
 
                    # TODO: rich_root should always be True here, for all
959
 
                    #       repositories that support chk_bytes
960
 
                    if not rich_root and name_utf8 == '':
961
 
                        continue
962
 
                    try:
963
 
                        file_id_revisions[file_id].add(revision_id)
964
 
                    except KeyError:
965
 
                        file_id_revisions[file_id] = set([revision_id])
 
783
            total = len(revision_ids)
 
784
            for pos, inv in enumerate(self.iter_inventories(revision_ids)):
 
785
                pb.update("Finding text references", pos, total)
 
786
                for entry in inv.iter_just_entries():
 
787
                    if entry.revision != inv.revision_id:
 
788
                        continue
 
789
                    if not rich_roots and entry.file_id == inv.root_id:
 
790
                        continue
 
791
                    alterations = result.setdefault(entry.file_id, set([]))
 
792
                    alterations.add(entry.revision)
 
793
            return result
966
794
        finally:
967
795
            pb.finished()
968
 
        return file_id_revisions
969
796
 
970
797
    def find_text_key_references(self):
971
798
        """Find the text key references within the repository.
1005
832
 
1006
833
    def _get_source(self, to_format):
1007
834
        """Return a source for streaming from this repository."""
1008
 
        if self._format._serializer == to_format._serializer:
 
835
        if isinstance(to_format, remote.RemoteRepositoryFormat):
 
836
            # Can't just check attributes on to_format with the current code,
 
837
            # work around this:
 
838
            to_format._ensure_real()
 
839
            to_format = to_format._custom_format
 
840
        if to_format.__class__ is self._format.__class__:
1009
841
            # We must be exactly the same format, otherwise stuff like the chk
1010
 
            # page layout might be different.
1011
 
            # Actually, this test is just slightly looser than exact so that
1012
 
            # CHK2 <-> 2a transfers will work.
 
842
            # page layout might be different
1013
843
            return GroupCHKStreamSource(self, to_format)
1014
844
        return super(CHKInventoryRepository, self)._get_source(to_format)
1015
845
 
1016
 
 
1017
 
class GroupCHKStreamSource(KnitPackStreamSource):
 
846
    def suspend_write_group(self):
 
847
        raise errors.UnsuspendableWriteGroup(self)
 
848
 
 
849
    def _resume_write_group(self, tokens):
 
850
        raise errors.UnsuspendableWriteGroup(self)
 
851
 
 
852
 
 
853
class GroupCHKStreamSource(repository.StreamSource):
1018
854
    """Used when both the source and target repo are GroupCHK repos."""
1019
855
 
1020
856
    def __init__(self, from_repository, to_format):
1022
858
        super(GroupCHKStreamSource, self).__init__(from_repository, to_format)
1023
859
        self._revision_keys = None
1024
860
        self._text_keys = None
1025
 
        self._text_fetch_order = 'groupcompress'
1026
861
        self._chk_id_roots = None
1027
862
        self._chk_p_id_roots = None
1028
863
 
1029
 
    def _get_inventory_stream(self, inventory_keys, allow_absent=False):
 
864
    def _get_filtered_inv_stream(self):
1030
865
        """Get a stream of inventory texts.
1031
866
 
1032
867
        When this function returns, self._chk_id_roots and self._chk_p_id_roots
1038
873
            id_roots_set = set()
1039
874
            p_id_roots_set = set()
1040
875
            source_vf = self.from_repository.inventories
1041
 
            stream = source_vf.get_record_stream(inventory_keys,
 
876
            stream = source_vf.get_record_stream(self._revision_keys,
1042
877
                                                 'groupcompress', True)
1043
878
            for record in stream:
1044
 
                if record.storage_kind == 'absent':
1045
 
                    if allow_absent:
1046
 
                        continue
1047
 
                    else:
1048
 
                        raise errors.NoSuchRevision(self, record.key)
1049
879
                bytes = record.get_bytes_as('fulltext')
1050
880
                chk_inv = inventory.CHKInventory.deserialise(None, bytes,
1051
881
                                                             record.key)
1067
897
            p_id_roots_set.clear()
1068
898
        return ('inventories', _filtered_inv_stream())
1069
899
 
1070
 
    def _get_filtered_chk_streams(self, excluded_revision_keys):
 
900
    def _get_filtered_chk_streams(self, excluded_keys):
1071
901
        self._text_keys = set()
1072
 
        excluded_revision_keys.discard(_mod_revision.NULL_REVISION)
1073
 
        if not excluded_revision_keys:
 
902
        excluded_keys.discard(_mod_revision.NULL_REVISION)
 
903
        if not excluded_keys:
1074
904
            uninteresting_root_keys = set()
1075
905
            uninteresting_pid_root_keys = set()
1076
906
        else:
1077
 
            # filter out any excluded revisions whose inventories are not
1078
 
            # actually present
1079
 
            # TODO: Update Repository.iter_inventories() to add
1080
 
            #       ignore_missing=True
1081
 
            present_keys = self.from_repository._find_present_inventory_keys(
1082
 
                            excluded_revision_keys)
1083
 
            present_ids = [k[-1] for k in present_keys]
1084
907
            uninteresting_root_keys = set()
1085
908
            uninteresting_pid_root_keys = set()
1086
 
            for inv in self.from_repository.iter_inventories(present_ids):
 
909
            for inv in self.from_repository.iter_inventories(excluded_keys):
1087
910
                uninteresting_root_keys.add(inv.id_to_entry.key())
1088
911
                uninteresting_pid_root_keys.add(
1089
912
                    inv.parent_id_basename_to_file_id.key())
 
913
        bytes_to_info = inventory.CHKInventory._bytes_to_utf8name_key
1090
914
        chk_bytes = self.from_repository.chk_bytes
1091
915
        def _filter_id_to_entry():
1092
 
            interesting_nodes = chk_map.iter_interesting_nodes(chk_bytes,
1093
 
                        self._chk_id_roots, uninteresting_root_keys)
1094
 
            for record in _filter_text_keys(interesting_nodes, self._text_keys,
1095
 
                    chk_map._bytes_to_text_key):
 
916
            for record, items in chk_map.iter_interesting_nodes(chk_bytes,
 
917
                        self._chk_id_roots, uninteresting_root_keys):
 
918
                for name, bytes in items:
 
919
                    # Note: we don't care about name_utf8, because we are always
 
920
                    # rich-root = True
 
921
                    _, file_id, revision_id = bytes_to_info(bytes)
 
922
                    self._text_keys.add((file_id, revision_id))
1096
923
                if record is not None:
1097
924
                    yield record
1098
 
            # Consumed
1099
 
            self._chk_id_roots = None
1100
925
        yield 'chk_bytes', _filter_id_to_entry()
1101
926
        def _get_parent_id_basename_to_file_id_pages():
1102
927
            for record, items in chk_map.iter_interesting_nodes(chk_bytes,
1103
928
                        self._chk_p_id_roots, uninteresting_pid_root_keys):
1104
929
                if record is not None:
1105
930
                    yield record
1106
 
            # Consumed
1107
 
            self._chk_p_id_roots = None
1108
931
        yield 'chk_bytes', _get_parent_id_basename_to_file_id_pages()
1109
932
 
 
933
    def _get_text_stream(self):
 
934
        # Note: We know we don't have to handle adding root keys, because both
 
935
        # the source and target are GCCHK, and those always support rich-roots
 
936
        # We may want to request as 'unordered', in case the source has done a
 
937
        # 'split' packing
 
938
        return ('texts', self.from_repository.texts.get_record_stream(
 
939
                            self._text_keys, 'groupcompress', False))
 
940
 
1110
941
    def get_stream(self, search):
1111
 
        def wrap_and_count(pb, rc, stream):
1112
 
            """Yield records from stream while showing progress."""
1113
 
            count = 0
1114
 
            for record in stream:
1115
 
                if count == rc.STEP:
1116
 
                    rc.increment(count)
1117
 
                    pb.update('Estimate', rc.current, rc.max)
1118
 
                    count = 0
1119
 
                count += 1
1120
 
                yield record
1121
 
 
1122
942
        revision_ids = search.get_keys()
1123
 
        pb = ui.ui_factory.nested_progress_bar()
1124
 
        rc = self._record_counter
1125
 
        self._record_counter.setup(len(revision_ids))
1126
943
        for stream_info in self._fetch_revision_texts(revision_ids):
1127
 
            yield (stream_info[0],
1128
 
                wrap_and_count(pb, rc, stream_info[1]))
 
944
            yield stream_info
1129
945
        self._revision_keys = [(rev_id,) for rev_id in revision_ids]
1130
 
        self.from_repository.revisions.clear_cache()
1131
 
        self.from_repository.signatures.clear_cache()
1132
 
        s = self._get_inventory_stream(self._revision_keys)
1133
 
        yield (s[0], wrap_and_count(pb, rc, s[1]))
1134
 
        self.from_repository.inventories.clear_cache()
1135
 
        # TODO: The keys to exclude might be part of the search recipe
1136
 
        # For now, exclude all parents that are at the edge of ancestry, for
1137
 
        # which we have inventories
1138
 
        from_repo = self.from_repository
1139
 
        parent_keys = from_repo._find_parent_keys_of_revisions(
1140
 
                        self._revision_keys)
1141
 
        for stream_info in self._get_filtered_chk_streams(parent_keys):
1142
 
            yield (stream_info[0], wrap_and_count(pb, rc, stream_info[1]))
1143
 
        self.from_repository.chk_bytes.clear_cache()
1144
 
        s = self._get_text_stream()
1145
 
        yield (s[0], wrap_and_count(pb, rc, s[1]))
1146
 
        self.from_repository.texts.clear_cache()
1147
 
        pb.update('Done', rc.max, rc.max)
1148
 
        pb.finished()
1149
 
 
1150
 
    def get_stream_for_missing_keys(self, missing_keys):
1151
 
        # missing keys can only occur when we are byte copying and not
1152
 
        # translating (because translation means we don't send
1153
 
        # unreconstructable deltas ever).
1154
 
        missing_inventory_keys = set()
1155
 
        for key in missing_keys:
1156
 
            if key[0] != 'inventories':
1157
 
                raise AssertionError('The only missing keys we should'
1158
 
                    ' be filling in are inventory keys, not %s'
1159
 
                    % (key[0],))
1160
 
            missing_inventory_keys.add(key[1:])
1161
 
        if self._chk_id_roots or self._chk_p_id_roots:
1162
 
            raise AssertionError('Cannot call get_stream_for_missing_keys'
1163
 
                ' until all of get_stream() has been consumed.')
1164
 
        # Yield the inventory stream, so we can find the chk stream
1165
 
        # Some of the missing_keys will be missing because they are ghosts.
1166
 
        # As such, we can ignore them. The Sink is required to verify there are
1167
 
        # no unavailable texts when the ghost inventories are not filled in.
1168
 
        yield self._get_inventory_stream(missing_inventory_keys,
1169
 
                                         allow_absent=True)
1170
 
        # We use the empty set for excluded_revision_keys, to make it clear
1171
 
        # that we want to transmit all referenced chk pages.
1172
 
        for stream_info in self._get_filtered_chk_streams(set()):
 
946
        yield self._get_filtered_inv_stream()
 
947
        # The keys to exclude are part of the search recipe
 
948
        _, _, exclude_keys, _ = search.get_recipe()
 
949
        for stream_info in self._get_filtered_chk_streams(exclude_keys):
1173
950
            yield stream_info
1174
 
 
1175
 
 
1176
 
class _InterestingKeyInfo(object):
1177
 
    def __init__(self):
1178
 
        self.interesting_root_keys = set()
1179
 
        self.interesting_pid_root_keys = set()
1180
 
        self.uninteresting_root_keys = set()
1181
 
        self.uninteresting_pid_root_keys = set()
1182
 
 
1183
 
    def all_interesting(self):
1184
 
        return self.interesting_root_keys.union(self.interesting_pid_root_keys)
1185
 
 
1186
 
    def all_uninteresting(self):
1187
 
        return self.uninteresting_root_keys.union(
1188
 
            self.uninteresting_pid_root_keys)
1189
 
 
1190
 
    def all_keys(self):
1191
 
        return self.all_interesting().union(self.all_uninteresting())
1192
 
 
1193
 
 
1194
 
def _build_interesting_key_sets(repo, inventory_ids, parent_only_inv_ids):
1195
 
    result = _InterestingKeyInfo()
1196
 
    for inv in repo.iter_inventories(inventory_ids, 'unordered'):
1197
 
        root_key = inv.id_to_entry.key()
1198
 
        pid_root_key = inv.parent_id_basename_to_file_id.key()
1199
 
        if inv.revision_id in parent_only_inv_ids:
1200
 
            result.uninteresting_root_keys.add(root_key)
1201
 
            result.uninteresting_pid_root_keys.add(pid_root_key)
1202
 
        else:
1203
 
            result.interesting_root_keys.add(root_key)
1204
 
            result.interesting_pid_root_keys.add(pid_root_key)
1205
 
    return result
1206
 
 
1207
 
 
1208
 
def _filter_text_keys(interesting_nodes_iterable, text_keys, bytes_to_text_key):
1209
 
    """Iterate the result of iter_interesting_nodes, yielding the records
1210
 
    and adding to text_keys.
1211
 
    """
1212
 
    text_keys_update = text_keys.update
1213
 
    for record, items in interesting_nodes_iterable:
1214
 
        text_keys_update([bytes_to_text_key(b) for n,b in items])
1215
 
        yield record
1216
 
 
1217
 
 
 
951
        yield self._get_text_stream()
1218
952
 
1219
953
 
1220
954
class RepositoryFormatCHK1(RepositoryFormatPack):
1221
955
    """A hashed CHK+group compress pack repository."""
1222
956
 
1223
957
    repository_class = CHKInventoryRepository
1224
 
    supports_external_lookups = True
1225
958
    supports_chks = True
1226
959
    # For right now, setting this to True gives us InterModel1And2 rather
1227
960
    # than InterDifferingSerializer
1239
972
    # multiple in-a-row (and sharing strings). Topological is better
1240
973
    # for remote, because we access less data.
1241
974
    _fetch_order = 'unordered'
1242
 
    _fetch_uses_deltas = False # essentially ignored by the groupcompress code.
 
975
    _fetch_uses_deltas = True # essentially ignored by the groupcompress code.
1243
976
    fast_deltas = True
1244
 
    pack_compresses = True
1245
977
 
1246
978
    def _get_matching_bzrdir(self):
1247
979
        return bzrdir.format_registry.make_bzrdir('development6-rich-root')
1261
993
        return ("Development repository format - rich roots, group compression"
1262
994
            " and chk inventories")
1263
995
 
1264
 
 
1265
 
class RepositoryFormatCHK2(RepositoryFormatCHK1):
1266
 
    """A CHK repository that uses the bencode revision serializer."""
1267
 
 
1268
 
    _serializer = chk_serializer.chk_bencode_serializer
1269
 
 
1270
 
    def _get_matching_bzrdir(self):
1271
 
        return bzrdir.format_registry.make_bzrdir('development7-rich-root')
1272
 
 
1273
 
    def _ignore_setting_bzrdir(self, format):
1274
 
        pass
1275
 
 
1276
 
    _matchingbzrdir = property(_get_matching_bzrdir, _ignore_setting_bzrdir)
1277
 
 
1278
 
    def get_format_string(self):
1279
 
        """See RepositoryFormat.get_format_string()."""
1280
 
        return ('Bazaar development format - chk repository with bencode '
1281
 
                'revision serialization (needs bzr.dev from 1.16)\n')
1282
 
 
1283
 
 
1284
 
class RepositoryFormat2a(RepositoryFormatCHK2):
1285
 
    """A CHK repository that uses the bencode revision serializer.
1286
 
 
1287
 
    This is the same as RepositoryFormatCHK2 but with a public name.
1288
 
    """
1289
 
 
1290
 
    _serializer = chk_serializer.chk_bencode_serializer
1291
 
 
1292
 
    def _get_matching_bzrdir(self):
1293
 
        return bzrdir.format_registry.make_bzrdir('2a')
1294
 
 
1295
 
    def _ignore_setting_bzrdir(self, format):
1296
 
        pass
1297
 
 
1298
 
    _matchingbzrdir = property(_get_matching_bzrdir, _ignore_setting_bzrdir)
1299
 
 
1300
 
    def get_format_string(self):
1301
 
        return ('Bazaar repository format 2a (needs bzr 1.16 or later)\n')
1302
 
 
1303
 
    def get_format_description(self):
1304
 
        """See RepositoryFormat.get_format_description()."""
1305
 
        return ("Repository format 2a - rich roots, group compression"
1306
 
            " and chk inventories")
 
996
    def check_conversion_target(self, target_format):
 
997
        if not target_format.rich_root_data:
 
998
            raise errors.BadConversionTarget(
 
999
                'Does not support rich root data.', target_format)
 
1000
        if not getattr(target_format, 'supports_tree_reference', False):
 
1001
            raise errors.BadConversionTarget(
 
1002
                'Does not support nested trees', target_format)
 
1003
 
 
1004