~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/repofmt/groupcompress_repo.py

  • Committer: Ian Clatworthy
  • Date: 2007-08-13 14:33:10 UTC
  • mto: (2733.1.1 ianc-integration)
  • mto: This revision was merged to the branch mainline in revision 2734.
  • Revision ID: ian.clatworthy@internode.on.net-20070813143310-twhj4la0qnupvze8
Added Quick Start Summary

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2008-2011 Canonical Ltd
2
 
#
3
 
# This program is free software; you can redistribute it and/or modify
4
 
# it under the terms of the GNU General Public License as published by
5
 
# the Free Software Foundation; either version 2 of the License, or
6
 
# (at your option) any later version.
7
 
#
8
 
# This program is distributed in the hope that it will be useful,
9
 
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
 
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
11
 
# GNU General Public License for more details.
12
 
#
13
 
# You should have received a copy of the GNU General Public License
14
 
# along with this program; if not, write to the Free Software
15
 
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
16
 
 
17
 
"""Repository formats using CHK inventories and groupcompress compression."""
18
 
 
19
 
import time
20
 
 
21
 
from bzrlib import (
22
 
    bzrdir,
23
 
    chk_map,
24
 
    chk_serializer,
25
 
    debug,
26
 
    errors,
27
 
    index as _mod_index,
28
 
    inventory,
29
 
    osutils,
30
 
    pack,
31
 
    revision as _mod_revision,
32
 
    trace,
33
 
    ui,
34
 
    versionedfile,
35
 
    )
36
 
from bzrlib.btree_index import (
37
 
    BTreeGraphIndex,
38
 
    BTreeBuilder,
39
 
    )
40
 
from bzrlib.decorators import needs_write_lock
41
 
from bzrlib.groupcompress import (
42
 
    _GCGraphIndex,
43
 
    GroupCompressVersionedFiles,
44
 
    )
45
 
from bzrlib.repofmt.pack_repo import (
46
 
    _DirectPackAccess,
47
 
    Pack,
48
 
    NewPack,
49
 
    PackRepository,
50
 
    PackRootCommitBuilder,
51
 
    RepositoryPackCollection,
52
 
    RepositoryFormatPack,
53
 
    ResumedPack,
54
 
    Packer,
55
 
    )
56
 
from bzrlib.vf_repository import (
57
 
    StreamSource,
58
 
    )
59
 
from bzrlib.static_tuple import StaticTuple
60
 
 
61
 
 
62
 
class GCPack(NewPack):
63
 
 
64
 
    def __init__(self, pack_collection, upload_suffix='', file_mode=None):
65
 
        """Create a NewPack instance.
66
 
 
67
 
        :param pack_collection: A PackCollection into which this is being
68
 
            inserted.
69
 
        :param upload_suffix: An optional suffix to be given to any temporary
70
 
            files created during the pack creation. e.g '.autopack'
71
 
        :param file_mode: An optional file mode to create the new files with.
72
 
        """
73
 
        # replaced from NewPack to:
74
 
        # - change inventory reference list length to 1
75
 
        # - change texts reference lists to 1
76
 
        # TODO: patch this to be parameterised
77
 
 
78
 
        # The relative locations of the packs are constrained, but all are
79
 
        # passed in because the caller has them, so as to avoid object churn.
80
 
        index_builder_class = pack_collection._index_builder_class
81
 
        # from brisbane-core
82
 
        if pack_collection.chk_index is not None:
83
 
            chk_index = index_builder_class(reference_lists=0)
84
 
        else:
85
 
            chk_index = None
86
 
        Pack.__init__(self,
87
 
            # Revisions: parents list, no text compression.
88
 
            index_builder_class(reference_lists=1),
89
 
            # Inventory: We want to map compression only, but currently the
90
 
            # knit code hasn't been updated enough to understand that, so we
91
 
            # have a regular 2-list index giving parents and compression
92
 
            # source.
93
 
            index_builder_class(reference_lists=1),
94
 
            # Texts: per file graph, for all fileids - so one reference list
95
 
            # and two elements in the key tuple.
96
 
            index_builder_class(reference_lists=1, key_elements=2),
97
 
            # Signatures: Just blobs to store, no compression, no parents
98
 
            # listing.
99
 
            index_builder_class(reference_lists=0),
100
 
            # CHK based storage - just blobs, no compression or parents.
101
 
            chk_index=chk_index
102
 
            )
103
 
        self._pack_collection = pack_collection
104
 
        # When we make readonly indices, we need this.
105
 
        self.index_class = pack_collection._index_class
106
 
        # where should the new pack be opened
107
 
        self.upload_transport = pack_collection._upload_transport
108
 
        # where are indices written out to
109
 
        self.index_transport = pack_collection._index_transport
110
 
        # where is the pack renamed to when it is finished?
111
 
        self.pack_transport = pack_collection._pack_transport
112
 
        # What file mode to upload the pack and indices with.
113
 
        self._file_mode = file_mode
114
 
        # tracks the content written to the .pack file.
115
 
        self._hash = osutils.md5()
116
 
        # a four-tuple with the length in bytes of the indices, once the pack
117
 
        # is finalised. (rev, inv, text, sigs)
118
 
        self.index_sizes = None
119
 
        # How much data to cache when writing packs. Note that this is not
120
 
        # synchronised with reads, because it's not in the transport layer, so
121
 
        # is not safe unless the client knows it won't be reading from the pack
122
 
        # under creation.
123
 
        self._cache_limit = 0
124
 
        # the temporary pack file name.
125
 
        self.random_name = osutils.rand_chars(20) + upload_suffix
126
 
        # when was this pack started ?
127
 
        self.start_time = time.time()
128
 
        # open an output stream for the data added to the pack.
129
 
        self.write_stream = self.upload_transport.open_write_stream(
130
 
            self.random_name, mode=self._file_mode)
131
 
        if 'pack' in debug.debug_flags:
132
 
            trace.mutter('%s: create_pack: pack stream open: %s%s t+%6.3fs',
133
 
                time.ctime(), self.upload_transport.base, self.random_name,
134
 
                time.time() - self.start_time)
135
 
        # A list of byte sequences to be written to the new pack, and the
136
 
        # aggregate size of them.  Stored as a list rather than separate
137
 
        # variables so that the _write_data closure below can update them.
138
 
        self._buffer = [[], 0]
139
 
        # create a callable for adding data
140
 
        #
141
 
        # robertc says- this is a closure rather than a method on the object
142
 
        # so that the variables are locals, and faster than accessing object
143
 
        # members.
144
 
        def _write_data(bytes, flush=False, _buffer=self._buffer,
145
 
            _write=self.write_stream.write, _update=self._hash.update):
146
 
            _buffer[0].append(bytes)
147
 
            _buffer[1] += len(bytes)
148
 
            # buffer cap
149
 
            if _buffer[1] > self._cache_limit or flush:
150
 
                bytes = ''.join(_buffer[0])
151
 
                _write(bytes)
152
 
                _update(bytes)
153
 
                _buffer[:] = [[], 0]
154
 
        # expose this on self, for the occasion when clients want to add data.
155
 
        self._write_data = _write_data
156
 
        # a pack writer object to serialise pack records.
157
 
        self._writer = pack.ContainerWriter(self._write_data)
158
 
        self._writer.begin()
159
 
        # what state is the pack in? (open, finished, aborted)
160
 
        self._state = 'open'
161
 
        # no name until we finish writing the content
162
 
        self.name = None
163
 
 
164
 
    def _check_references(self):
165
 
        """Make sure our external references are present.
166
 
 
167
 
        Packs are allowed to have deltas whose base is not in the pack, but it
168
 
        must be present somewhere in this collection.  It is not allowed to
169
 
        have deltas based on a fallback repository.
170
 
        (See <https://bugs.launchpad.net/bzr/+bug/288751>)
171
 
        """
172
 
        # Groupcompress packs don't have any external references, arguably CHK
173
 
        # pages have external references, but we cannot 'cheaply' determine
174
 
        # them without actually walking all of the chk pages.
175
 
 
176
 
 
177
 
class ResumedGCPack(ResumedPack):
178
 
 
179
 
    def _check_references(self):
180
 
        """Make sure our external compression parents are present."""
181
 
        # See GCPack._check_references for why this is empty
182
 
 
183
 
    def _get_external_refs(self, index):
184
 
        # GC repositories don't have compression parents external to a given
185
 
        # pack file
186
 
        return set()
187
 
 
188
 
 
189
 
class GCCHKPacker(Packer):
190
 
    """This class understand what it takes to collect a GCCHK repo."""
191
 
 
192
 
    def __init__(self, pack_collection, packs, suffix, revision_ids=None,
193
 
                 reload_func=None):
194
 
        super(GCCHKPacker, self).__init__(pack_collection, packs, suffix,
195
 
                                          revision_ids=revision_ids,
196
 
                                          reload_func=reload_func)
197
 
        self._pack_collection = pack_collection
198
 
        # ATM, We only support this for GCCHK repositories
199
 
        if pack_collection.chk_index is None:
200
 
            raise AssertionError('pack_collection.chk_index should not be None')
201
 
        self._gather_text_refs = False
202
 
        self._chk_id_roots = []
203
 
        self._chk_p_id_roots = []
204
 
        self._text_refs = None
205
 
        # set by .pack() if self.revision_ids is not None
206
 
        self.revision_keys = None
207
 
 
208
 
    def _get_progress_stream(self, source_vf, keys, message, pb):
209
 
        def pb_stream():
210
 
            substream = source_vf.get_record_stream(keys, 'groupcompress', True)
211
 
            for idx, record in enumerate(substream):
212
 
                if pb is not None:
213
 
                    pb.update(message, idx + 1, len(keys))
214
 
                yield record
215
 
        return pb_stream()
216
 
 
217
 
    def _get_filtered_inv_stream(self, source_vf, keys, message, pb=None):
218
 
        """Filter the texts of inventories, to find the chk pages."""
219
 
        total_keys = len(keys)
220
 
        def _filtered_inv_stream():
221
 
            id_roots_set = set()
222
 
            p_id_roots_set = set()
223
 
            stream = source_vf.get_record_stream(keys, 'groupcompress', True)
224
 
            for idx, record in enumerate(stream):
225
 
                # Inventories should always be with revisions; assume success.
226
 
                bytes = record.get_bytes_as('fulltext')
227
 
                chk_inv = inventory.CHKInventory.deserialise(None, bytes,
228
 
                                                             record.key)
229
 
                if pb is not None:
230
 
                    pb.update('inv', idx, total_keys)
231
 
                key = chk_inv.id_to_entry.key()
232
 
                if key not in id_roots_set:
233
 
                    self._chk_id_roots.append(key)
234
 
                    id_roots_set.add(key)
235
 
                p_id_map = chk_inv.parent_id_basename_to_file_id
236
 
                if p_id_map is None:
237
 
                    raise AssertionError('Parent id -> file_id map not set')
238
 
                key = p_id_map.key()
239
 
                if key not in p_id_roots_set:
240
 
                    p_id_roots_set.add(key)
241
 
                    self._chk_p_id_roots.append(key)
242
 
                yield record
243
 
            # We have finished processing all of the inventory records, we
244
 
            # don't need these sets anymore
245
 
            id_roots_set.clear()
246
 
            p_id_roots_set.clear()
247
 
        return _filtered_inv_stream()
248
 
 
249
 
    def _get_chk_streams(self, source_vf, keys, pb=None):
250
 
        # We want to stream the keys from 'id_roots', and things they
251
 
        # reference, and then stream things from p_id_roots and things they
252
 
        # reference, and then any remaining keys that we didn't get to.
253
 
 
254
 
        # We also group referenced texts together, so if one root references a
255
 
        # text with prefix 'a', and another root references a node with prefix
256
 
        # 'a', we want to yield those nodes before we yield the nodes for 'b'
257
 
        # This keeps 'similar' nodes together.
258
 
 
259
 
        # Note: We probably actually want multiple streams here, to help the
260
 
        #       client understand that the different levels won't compress well
261
 
        #       against each other.
262
 
        #       Test the difference between using one Group per level, and
263
 
        #       using 1 Group per prefix. (so '' (root) would get a group, then
264
 
        #       all the references to search-key 'a' would get a group, etc.)
265
 
        total_keys = len(keys)
266
 
        remaining_keys = set(keys)
267
 
        counter = [0]
268
 
        if self._gather_text_refs:
269
 
            self._text_refs = set()
270
 
        def _get_referenced_stream(root_keys, parse_leaf_nodes=False):
271
 
            cur_keys = root_keys
272
 
            while cur_keys:
273
 
                keys_by_search_prefix = {}
274
 
                remaining_keys.difference_update(cur_keys)
275
 
                next_keys = set()
276
 
                def handle_internal_node(node):
277
 
                    for prefix, value in node._items.iteritems():
278
 
                        # We don't want to request the same key twice, and we
279
 
                        # want to order it by the first time it is seen.
280
 
                        # Even further, we don't want to request a key which is
281
 
                        # not in this group of pack files (it should be in the
282
 
                        # repo, but it doesn't have to be in the group being
283
 
                        # packed.)
284
 
                        # TODO: consider how to treat externally referenced chk
285
 
                        #       pages as 'external_references' so that we
286
 
                        #       always fill them in for stacked branches
287
 
                        if value not in next_keys and value in remaining_keys:
288
 
                            keys_by_search_prefix.setdefault(prefix,
289
 
                                []).append(value)
290
 
                            next_keys.add(value)
291
 
                def handle_leaf_node(node):
292
 
                    # Store is None, because we know we have a LeafNode, and we
293
 
                    # just want its entries
294
 
                    for file_id, bytes in node.iteritems(None):
295
 
                        self._text_refs.add(chk_map._bytes_to_text_key(bytes))
296
 
                def next_stream():
297
 
                    stream = source_vf.get_record_stream(cur_keys,
298
 
                                                         'as-requested', True)
299
 
                    for record in stream:
300
 
                        if record.storage_kind == 'absent':
301
 
                            # An absent CHK record: we assume that the missing
302
 
                            # record is in a different pack - e.g. a page not
303
 
                            # altered by the commit we're packing.
304
 
                            continue
305
 
                        bytes = record.get_bytes_as('fulltext')
306
 
                        # We don't care about search_key_func for this code,
307
 
                        # because we only care about external references.
308
 
                        node = chk_map._deserialise(bytes, record.key,
309
 
                                                    search_key_func=None)
310
 
                        common_base = node._search_prefix
311
 
                        if isinstance(node, chk_map.InternalNode):
312
 
                            handle_internal_node(node)
313
 
                        elif parse_leaf_nodes:
314
 
                            handle_leaf_node(node)
315
 
                        counter[0] += 1
316
 
                        if pb is not None:
317
 
                            pb.update('chk node', counter[0], total_keys)
318
 
                        yield record
319
 
                yield next_stream()
320
 
                # Double check that we won't be emitting any keys twice
321
 
                # If we get rid of the pre-calculation of all keys, we could
322
 
                # turn this around and do
323
 
                # next_keys.difference_update(seen_keys)
324
 
                # However, we also may have references to chk pages in another
325
 
                # pack file during autopack. We filter earlier, so we should no
326
 
                # longer need to do this
327
 
                # next_keys = next_keys.intersection(remaining_keys)
328
 
                cur_keys = []
329
 
                for prefix in sorted(keys_by_search_prefix):
330
 
                    cur_keys.extend(keys_by_search_prefix.pop(prefix))
331
 
        for stream in _get_referenced_stream(self._chk_id_roots,
332
 
                                             self._gather_text_refs):
333
 
            yield stream
334
 
        del self._chk_id_roots
335
 
        # while it isn't really possible for chk_id_roots to not be in the
336
 
        # local group of packs, it is possible that the tree shape has not
337
 
        # changed recently, so we need to filter _chk_p_id_roots by the
338
 
        # available keys
339
 
        chk_p_id_roots = [key for key in self._chk_p_id_roots
340
 
                          if key in remaining_keys]
341
 
        del self._chk_p_id_roots
342
 
        for stream in _get_referenced_stream(chk_p_id_roots, False):
343
 
            yield stream
344
 
        if remaining_keys:
345
 
            trace.mutter('There were %d keys in the chk index, %d of which'
346
 
                         ' were not referenced', total_keys,
347
 
                         len(remaining_keys))
348
 
            if self.revision_ids is None:
349
 
                stream = source_vf.get_record_stream(remaining_keys,
350
 
                                                     'unordered', True)
351
 
                yield stream
352
 
 
353
 
    def _build_vf(self, index_name, parents, delta, for_write=False):
354
 
        """Build a VersionedFiles instance on top of this group of packs."""
355
 
        index_name = index_name + '_index'
356
 
        index_to_pack = {}
357
 
        access = _DirectPackAccess(index_to_pack,
358
 
                                   reload_func=self._reload_func)
359
 
        if for_write:
360
 
            # Use new_pack
361
 
            if self.new_pack is None:
362
 
                raise AssertionError('No new pack has been set')
363
 
            index = getattr(self.new_pack, index_name)
364
 
            index_to_pack[index] = self.new_pack.access_tuple()
365
 
            index.set_optimize(for_size=True)
366
 
            access.set_writer(self.new_pack._writer, index,
367
 
                              self.new_pack.access_tuple())
368
 
            add_callback = index.add_nodes
369
 
        else:
370
 
            indices = []
371
 
            for pack in self.packs:
372
 
                sub_index = getattr(pack, index_name)
373
 
                index_to_pack[sub_index] = pack.access_tuple()
374
 
                indices.append(sub_index)
375
 
            index = _mod_index.CombinedGraphIndex(indices)
376
 
            add_callback = None
377
 
        vf = GroupCompressVersionedFiles(
378
 
            _GCGraphIndex(index,
379
 
                          add_callback=add_callback,
380
 
                          parents=parents,
381
 
                          is_locked=self._pack_collection.repo.is_locked),
382
 
            access=access,
383
 
            delta=delta)
384
 
        return vf
385
 
 
386
 
    def _build_vfs(self, index_name, parents, delta):
387
 
        """Build the source and target VersionedFiles."""
388
 
        source_vf = self._build_vf(index_name, parents,
389
 
                                   delta, for_write=False)
390
 
        target_vf = self._build_vf(index_name, parents,
391
 
                                   delta, for_write=True)
392
 
        return source_vf, target_vf
393
 
 
394
 
    def _copy_stream(self, source_vf, target_vf, keys, message, vf_to_stream,
395
 
                     pb_offset):
396
 
        trace.mutter('repacking %d %s', len(keys), message)
397
 
        self.pb.update('repacking %s' % (message,), pb_offset)
398
 
        child_pb = ui.ui_factory.nested_progress_bar()
399
 
        try:
400
 
            stream = vf_to_stream(source_vf, keys, message, child_pb)
401
 
            for _ in target_vf._insert_record_stream(stream,
402
 
                                                     random_id=True,
403
 
                                                     reuse_blocks=False):
404
 
                pass
405
 
        finally:
406
 
            child_pb.finished()
407
 
 
408
 
    def _copy_revision_texts(self):
409
 
        source_vf, target_vf = self._build_vfs('revision', True, False)
410
 
        if not self.revision_keys:
411
 
            # We are doing a full fetch, aka 'pack'
412
 
            self.revision_keys = source_vf.keys()
413
 
        self._copy_stream(source_vf, target_vf, self.revision_keys,
414
 
                          'revisions', self._get_progress_stream, 1)
415
 
 
416
 
    def _copy_inventory_texts(self):
417
 
        source_vf, target_vf = self._build_vfs('inventory', True, True)
418
 
        # It is not sufficient to just use self.revision_keys, as stacked
419
 
        # repositories can have more inventories than they have revisions.
420
 
        # One alternative would be to do something with
421
 
        # get_parent_map(self.revision_keys), but that shouldn't be any faster
422
 
        # than this.
423
 
        inventory_keys = source_vf.keys()
424
 
        missing_inventories = set(self.revision_keys).difference(inventory_keys)
425
 
        if missing_inventories:
426
 
            # Go back to the original repo, to see if these are really missing
427
 
            # https://bugs.launchpad.net/bzr/+bug/437003
428
 
            # If we are packing a subset of the repo, it is fine to just have
429
 
            # the data in another Pack file, which is not included in this pack
430
 
            # operation.
431
 
            inv_index = self._pack_collection.repo.inventories._index
432
 
            pmap = inv_index.get_parent_map(missing_inventories)
433
 
            really_missing = missing_inventories.difference(pmap)
434
 
            if really_missing:
435
 
                missing_inventories = sorted(really_missing)
436
 
                raise ValueError('We are missing inventories for revisions: %s'
437
 
                    % (missing_inventories,))
438
 
        self._copy_stream(source_vf, target_vf, inventory_keys,
439
 
                          'inventories', self._get_filtered_inv_stream, 2)
440
 
 
441
 
    def _get_chk_vfs_for_copy(self):
442
 
        return self._build_vfs('chk', False, False)
443
 
 
444
 
    def _copy_chk_texts(self):
445
 
        source_vf, target_vf = self._get_chk_vfs_for_copy()
446
 
        # TODO: This is technically spurious... if it is a performance issue,
447
 
        #       remove it
448
 
        total_keys = source_vf.keys()
449
 
        trace.mutter('repacking chk: %d id_to_entry roots,'
450
 
                     ' %d p_id_map roots, %d total keys',
451
 
                     len(self._chk_id_roots), len(self._chk_p_id_roots),
452
 
                     len(total_keys))
453
 
        self.pb.update('repacking chk', 3)
454
 
        child_pb = ui.ui_factory.nested_progress_bar()
455
 
        try:
456
 
            for stream in self._get_chk_streams(source_vf, total_keys,
457
 
                                                pb=child_pb):
458
 
                for _ in target_vf._insert_record_stream(stream,
459
 
                                                         random_id=True,
460
 
                                                         reuse_blocks=False):
461
 
                    pass
462
 
        finally:
463
 
            child_pb.finished()
464
 
 
465
 
    def _copy_text_texts(self):
466
 
        source_vf, target_vf = self._build_vfs('text', True, True)
467
 
        # XXX: We don't walk the chk map to determine referenced (file_id,
468
 
        #      revision_id) keys.  We don't do it yet because you really need
469
 
        #      to filter out the ones that are present in the parents of the
470
 
        #      rev just before the ones you are copying, otherwise the filter
471
 
        #      is grabbing too many keys...
472
 
        text_keys = source_vf.keys()
473
 
        self._copy_stream(source_vf, target_vf, text_keys,
474
 
                          'texts', self._get_progress_stream, 4)
475
 
 
476
 
    def _copy_signature_texts(self):
477
 
        source_vf, target_vf = self._build_vfs('signature', False, False)
478
 
        signature_keys = source_vf.keys()
479
 
        signature_keys.intersection(self.revision_keys)
480
 
        self._copy_stream(source_vf, target_vf, signature_keys,
481
 
                          'signatures', self._get_progress_stream, 5)
482
 
 
483
 
    def _create_pack_from_packs(self):
484
 
        self.pb.update('repacking', 0, 7)
485
 
        self.new_pack = self.open_pack()
486
 
        # Is this necessary for GC ?
487
 
        self.new_pack.set_write_cache_size(1024*1024)
488
 
        self._copy_revision_texts()
489
 
        self._copy_inventory_texts()
490
 
        self._copy_chk_texts()
491
 
        self._copy_text_texts()
492
 
        self._copy_signature_texts()
493
 
        self.new_pack._check_references()
494
 
        if not self._use_pack(self.new_pack):
495
 
            self.new_pack.abort()
496
 
            return None
497
 
        self.new_pack.finish_content()
498
 
        if len(self.packs) == 1:
499
 
            old_pack = self.packs[0]
500
 
            if old_pack.name == self.new_pack._hash.hexdigest():
501
 
                # The single old pack was already optimally packed.
502
 
                trace.mutter('single pack %s was already optimally packed',
503
 
                    old_pack.name)
504
 
                self.new_pack.abort()
505
 
                return None
506
 
        self.pb.update('finishing repack', 6, 7)
507
 
        self.new_pack.finish()
508
 
        self._pack_collection.allocate(self.new_pack)
509
 
        return self.new_pack
510
 
 
511
 
 
512
 
class GCCHKReconcilePacker(GCCHKPacker):
513
 
    """A packer which regenerates indices etc as it copies.
514
 
 
515
 
    This is used by ``bzr reconcile`` to cause parent text pointers to be
516
 
    regenerated.
517
 
    """
518
 
 
519
 
    def __init__(self, *args, **kwargs):
520
 
        super(GCCHKReconcilePacker, self).__init__(*args, **kwargs)
521
 
        self._data_changed = False
522
 
        self._gather_text_refs = True
523
 
 
524
 
    def _copy_inventory_texts(self):
525
 
        source_vf, target_vf = self._build_vfs('inventory', True, True)
526
 
        self._copy_stream(source_vf, target_vf, self.revision_keys,
527
 
                          'inventories', self._get_filtered_inv_stream, 2)
528
 
        if source_vf.keys() != self.revision_keys:
529
 
            self._data_changed = True
530
 
 
531
 
    def _copy_text_texts(self):
532
 
        """generate what texts we should have and then copy."""
533
 
        source_vf, target_vf = self._build_vfs('text', True, True)
534
 
        trace.mutter('repacking %d texts', len(self._text_refs))
535
 
        self.pb.update("repacking texts", 4)
536
 
        # we have three major tasks here:
537
 
        # 1) generate the ideal index
538
 
        repo = self._pack_collection.repo
539
 
        # We want the one we just wrote, so base it on self.new_pack
540
 
        revision_vf = self._build_vf('revision', True, False, for_write=True)
541
 
        ancestor_keys = revision_vf.get_parent_map(revision_vf.keys())
542
 
        # Strip keys back into revision_ids.
543
 
        ancestors = dict((k[0], tuple([p[0] for p in parents]))
544
 
                         for k, parents in ancestor_keys.iteritems())
545
 
        del ancestor_keys
546
 
        # TODO: _generate_text_key_index should be much cheaper to generate from
547
 
        #       a chk repository, rather than the current implementation
548
 
        ideal_index = repo._generate_text_key_index(None, ancestors)
549
 
        file_id_parent_map = source_vf.get_parent_map(self._text_refs)
550
 
        # 2) generate a keys list that contains all the entries that can
551
 
        #    be used as-is, with corrected parents.
552
 
        ok_keys = []
553
 
        new_parent_keys = {} # (key, parent_keys)
554
 
        discarded_keys = []
555
 
        NULL_REVISION = _mod_revision.NULL_REVISION
556
 
        for key in self._text_refs:
557
 
            # 0 - index
558
 
            # 1 - key
559
 
            # 2 - value
560
 
            # 3 - refs
561
 
            try:
562
 
                ideal_parents = tuple(ideal_index[key])
563
 
            except KeyError:
564
 
                discarded_keys.append(key)
565
 
                self._data_changed = True
566
 
            else:
567
 
                if ideal_parents == (NULL_REVISION,):
568
 
                    ideal_parents = ()
569
 
                source_parents = file_id_parent_map[key]
570
 
                if ideal_parents == source_parents:
571
 
                    # no change needed.
572
 
                    ok_keys.append(key)
573
 
                else:
574
 
                    # We need to change the parent graph, but we don't need to
575
 
                    # re-insert the text (since we don't pun the compression
576
 
                    # parent with the parents list)
577
 
                    self._data_changed = True
578
 
                    new_parent_keys[key] = ideal_parents
579
 
        # we're finished with some data.
580
 
        del ideal_index
581
 
        del file_id_parent_map
582
 
        # 3) bulk copy the data, updating records than need it
583
 
        def _update_parents_for_texts():
584
 
            stream = source_vf.get_record_stream(self._text_refs,
585
 
                'groupcompress', False)
586
 
            for record in stream:
587
 
                if record.key in new_parent_keys:
588
 
                    record.parents = new_parent_keys[record.key]
589
 
                yield record
590
 
        target_vf.insert_record_stream(_update_parents_for_texts())
591
 
 
592
 
    def _use_pack(self, new_pack):
593
 
        """Override _use_pack to check for reconcile having changed content."""
594
 
        return new_pack.data_inserted() and self._data_changed
595
 
 
596
 
 
597
 
class GCCHKCanonicalizingPacker(GCCHKPacker):
598
 
    """A packer that ensures inventories have canonical-form CHK maps.
599
 
    
600
 
    Ideally this would be part of reconcile, but it's very slow and rarely
601
 
    needed.  (It repairs repositories affected by
602
 
    https://bugs.launchpad.net/bzr/+bug/522637).
603
 
    """
604
 
 
605
 
    def __init__(self, *args, **kwargs):
606
 
        super(GCCHKCanonicalizingPacker, self).__init__(*args, **kwargs)
607
 
        self._data_changed = False
608
 
 
609
 
    def _exhaust_stream(self, source_vf, keys, message, vf_to_stream, pb_offset):
610
 
        """Create and exhaust a stream, but don't insert it.
611
 
 
612
 
        This is useful to get the side-effects of generating a stream.
613
 
        """
614
 
        self.pb.update('scanning %s' % (message,), pb_offset)
615
 
        child_pb = ui.ui_factory.nested_progress_bar()
616
 
        try:
617
 
            list(vf_to_stream(source_vf, keys, message, child_pb))
618
 
        finally:
619
 
            child_pb.finished()
620
 
 
621
 
    def _copy_inventory_texts(self):
622
 
        source_vf, target_vf = self._build_vfs('inventory', True, True)
623
 
        source_chk_vf, target_chk_vf = self._get_chk_vfs_for_copy()
624
 
        inventory_keys = source_vf.keys()
625
 
        # First, copy the existing CHKs on the assumption that most of them
626
 
        # will be correct.  This will save us from having to reinsert (and
627
 
        # recompress) these records later at the cost of perhaps preserving a
628
 
        # few unused CHKs. 
629
 
        # (Iterate but don't insert _get_filtered_inv_stream to populate the
630
 
        # variables needed by GCCHKPacker._copy_chk_texts.)
631
 
        self._exhaust_stream(source_vf, inventory_keys, 'inventories',
632
 
                self._get_filtered_inv_stream, 2)
633
 
        GCCHKPacker._copy_chk_texts(self)
634
 
        # Now copy and fix the inventories, and any regenerated CHKs.
635
 
        def chk_canonicalizing_inv_stream(source_vf, keys, message, pb=None):
636
 
            return self._get_filtered_canonicalizing_inv_stream(
637
 
                source_vf, keys, message, pb, source_chk_vf, target_chk_vf)
638
 
        self._copy_stream(source_vf, target_vf, inventory_keys,
639
 
                          'inventories', chk_canonicalizing_inv_stream, 4)
640
 
 
641
 
    def _copy_chk_texts(self):
642
 
        # No-op; in this class this happens during _copy_inventory_texts.
643
 
        pass
644
 
 
645
 
    def _get_filtered_canonicalizing_inv_stream(self, source_vf, keys, message,
646
 
            pb=None, source_chk_vf=None, target_chk_vf=None):
647
 
        """Filter the texts of inventories, regenerating CHKs to make sure they
648
 
        are canonical.
649
 
        """
650
 
        total_keys = len(keys)
651
 
        target_chk_vf = versionedfile.NoDupeAddLinesDecorator(target_chk_vf)
652
 
        def _filtered_inv_stream():
653
 
            stream = source_vf.get_record_stream(keys, 'groupcompress', True)
654
 
            search_key_name = None
655
 
            for idx, record in enumerate(stream):
656
 
                # Inventories should always be with revisions; assume success.
657
 
                bytes = record.get_bytes_as('fulltext')
658
 
                chk_inv = inventory.CHKInventory.deserialise(
659
 
                    source_chk_vf, bytes, record.key)
660
 
                if pb is not None:
661
 
                    pb.update('inv', idx, total_keys)
662
 
                chk_inv.id_to_entry._ensure_root()
663
 
                if search_key_name is None:
664
 
                    # Find the name corresponding to the search_key_func
665
 
                    search_key_reg = chk_map.search_key_registry
666
 
                    for search_key_name, func in search_key_reg.iteritems():
667
 
                        if func == chk_inv.id_to_entry._search_key_func:
668
 
                            break
669
 
                canonical_inv = inventory.CHKInventory.from_inventory(
670
 
                    target_chk_vf, chk_inv,
671
 
                    maximum_size=chk_inv.id_to_entry._root_node._maximum_size,
672
 
                    search_key_name=search_key_name)
673
 
                if chk_inv.id_to_entry.key() != canonical_inv.id_to_entry.key():
674
 
                    trace.mutter(
675
 
                        'Non-canonical CHK map for id_to_entry of inv: %s '
676
 
                        '(root is %s, should be %s)' % (chk_inv.revision_id,
677
 
                        chk_inv.id_to_entry.key()[0],
678
 
                        canonical_inv.id_to_entry.key()[0]))
679
 
                    self._data_changed = True
680
 
                p_id_map = chk_inv.parent_id_basename_to_file_id
681
 
                p_id_map._ensure_root()
682
 
                canon_p_id_map = canonical_inv.parent_id_basename_to_file_id
683
 
                if p_id_map.key() != canon_p_id_map.key():
684
 
                    trace.mutter(
685
 
                        'Non-canonical CHK map for parent_id_to_basename of '
686
 
                        'inv: %s (root is %s, should be %s)'
687
 
                        % (chk_inv.revision_id, p_id_map.key()[0],
688
 
                           canon_p_id_map.key()[0]))
689
 
                    self._data_changed = True
690
 
                yield versionedfile.ChunkedContentFactory(record.key,
691
 
                        record.parents, record.sha1,
692
 
                        canonical_inv.to_lines())
693
 
            # We have finished processing all of the inventory records, we
694
 
            # don't need these sets anymore
695
 
        return _filtered_inv_stream()
696
 
 
697
 
    def _use_pack(self, new_pack):
698
 
        """Override _use_pack to check for reconcile having changed content."""
699
 
        return new_pack.data_inserted() and self._data_changed
700
 
 
701
 
 
702
 
class GCRepositoryPackCollection(RepositoryPackCollection):
703
 
 
704
 
    pack_factory = GCPack
705
 
    resumed_pack_factory = ResumedGCPack
706
 
    normal_packer_class = GCCHKPacker
707
 
    optimising_packer_class = GCCHKPacker
708
 
 
709
 
    def _check_new_inventories(self):
710
 
        """Detect missing inventories or chk root entries for the new revisions
711
 
        in this write group.
712
 
 
713
 
        :returns: list of strs, summarising any problems found.  If the list is
714
 
            empty no problems were found.
715
 
        """
716
 
        # Ensure that all revisions added in this write group have:
717
 
        #   - corresponding inventories,
718
 
        #   - chk root entries for those inventories,
719
 
        #   - and any present parent inventories have their chk root
720
 
        #     entries too.
721
 
        # And all this should be independent of any fallback repository.
722
 
        problems = []
723
 
        key_deps = self.repo.revisions._index._key_dependencies
724
 
        new_revisions_keys = key_deps.get_new_keys()
725
 
        no_fallback_inv_index = self.repo.inventories._index
726
 
        no_fallback_chk_bytes_index = self.repo.chk_bytes._index
727
 
        no_fallback_texts_index = self.repo.texts._index
728
 
        inv_parent_map = no_fallback_inv_index.get_parent_map(
729
 
            new_revisions_keys)
730
 
        # Are any inventories for corresponding to the new revisions missing?
731
 
        corresponding_invs = set(inv_parent_map)
732
 
        missing_corresponding = set(new_revisions_keys)
733
 
        missing_corresponding.difference_update(corresponding_invs)
734
 
        if missing_corresponding:
735
 
            problems.append("inventories missing for revisions %s" %
736
 
                (sorted(missing_corresponding),))
737
 
            return problems
738
 
        # Are any chk root entries missing for any inventories?  This includes
739
 
        # any present parent inventories, which may be used when calculating
740
 
        # deltas for streaming.
741
 
        all_inv_keys = set(corresponding_invs)
742
 
        for parent_inv_keys in inv_parent_map.itervalues():
743
 
            all_inv_keys.update(parent_inv_keys)
744
 
        # Filter out ghost parents.
745
 
        all_inv_keys.intersection_update(
746
 
            no_fallback_inv_index.get_parent_map(all_inv_keys))
747
 
        parent_invs_only_keys = all_inv_keys.symmetric_difference(
748
 
            corresponding_invs)
749
 
        all_missing = set()
750
 
        inv_ids = [key[-1] for key in all_inv_keys]
751
 
        parent_invs_only_ids = [key[-1] for key in parent_invs_only_keys]
752
 
        root_key_info = _build_interesting_key_sets(
753
 
            self.repo, inv_ids, parent_invs_only_ids)
754
 
        expected_chk_roots = root_key_info.all_keys()
755
 
        present_chk_roots = no_fallback_chk_bytes_index.get_parent_map(
756
 
            expected_chk_roots)
757
 
        missing_chk_roots = expected_chk_roots.difference(present_chk_roots)
758
 
        if missing_chk_roots:
759
 
            problems.append("missing referenced chk root keys: %s"
760
 
                % (sorted(missing_chk_roots),))
761
 
            # Don't bother checking any further.
762
 
            return problems
763
 
        # Find all interesting chk_bytes records, and make sure they are
764
 
        # present, as well as the text keys they reference.
765
 
        chk_bytes_no_fallbacks = self.repo.chk_bytes.without_fallbacks()
766
 
        chk_bytes_no_fallbacks._search_key_func = \
767
 
            self.repo.chk_bytes._search_key_func
768
 
        chk_diff = chk_map.iter_interesting_nodes(
769
 
            chk_bytes_no_fallbacks, root_key_info.interesting_root_keys,
770
 
            root_key_info.uninteresting_root_keys)
771
 
        text_keys = set()
772
 
        try:
773
 
            for record in _filter_text_keys(chk_diff, text_keys,
774
 
                                            chk_map._bytes_to_text_key):
775
 
                pass
776
 
        except errors.NoSuchRevision, e:
777
 
            # XXX: It would be nice if we could give a more precise error here.
778
 
            problems.append("missing chk node(s) for id_to_entry maps")
779
 
        chk_diff = chk_map.iter_interesting_nodes(
780
 
            chk_bytes_no_fallbacks, root_key_info.interesting_pid_root_keys,
781
 
            root_key_info.uninteresting_pid_root_keys)
782
 
        try:
783
 
            for interesting_rec, interesting_map in chk_diff:
784
 
                pass
785
 
        except errors.NoSuchRevision, e:
786
 
            problems.append(
787
 
                "missing chk node(s) for parent_id_basename_to_file_id maps")
788
 
        present_text_keys = no_fallback_texts_index.get_parent_map(text_keys)
789
 
        missing_text_keys = text_keys.difference(present_text_keys)
790
 
        if missing_text_keys:
791
 
            problems.append("missing text keys: %r"
792
 
                % (sorted(missing_text_keys),))
793
 
        return problems
794
 
 
795
 
 
796
 
class CHKInventoryRepository(PackRepository):
797
 
    """subclass of PackRepository that uses CHK based inventories."""
798
 
 
799
 
    def __init__(self, _format, a_bzrdir, control_files, _commit_builder_class,
800
 
        _serializer):
801
 
        """Overridden to change pack collection class."""
802
 
        super(CHKInventoryRepository, self).__init__(_format, a_bzrdir,
803
 
            control_files, _commit_builder_class, _serializer)
804
 
        index_transport = self._transport.clone('indices')
805
 
        self._pack_collection = GCRepositoryPackCollection(self,
806
 
            self._transport, index_transport,
807
 
            self._transport.clone('upload'),
808
 
            self._transport.clone('packs'),
809
 
            _format.index_builder_class,
810
 
            _format.index_class,
811
 
            use_chk_index=self._format.supports_chks,
812
 
            )
813
 
        self.inventories = GroupCompressVersionedFiles(
814
 
            _GCGraphIndex(self._pack_collection.inventory_index.combined_index,
815
 
                add_callback=self._pack_collection.inventory_index.add_callback,
816
 
                parents=True, is_locked=self.is_locked,
817
 
                inconsistency_fatal=False),
818
 
            access=self._pack_collection.inventory_index.data_access)
819
 
        self.revisions = GroupCompressVersionedFiles(
820
 
            _GCGraphIndex(self._pack_collection.revision_index.combined_index,
821
 
                add_callback=self._pack_collection.revision_index.add_callback,
822
 
                parents=True, is_locked=self.is_locked,
823
 
                track_external_parent_refs=True, track_new_keys=True),
824
 
            access=self._pack_collection.revision_index.data_access,
825
 
            delta=False)
826
 
        self.signatures = GroupCompressVersionedFiles(
827
 
            _GCGraphIndex(self._pack_collection.signature_index.combined_index,
828
 
                add_callback=self._pack_collection.signature_index.add_callback,
829
 
                parents=False, is_locked=self.is_locked,
830
 
                inconsistency_fatal=False),
831
 
            access=self._pack_collection.signature_index.data_access,
832
 
            delta=False)
833
 
        self.texts = GroupCompressVersionedFiles(
834
 
            _GCGraphIndex(self._pack_collection.text_index.combined_index,
835
 
                add_callback=self._pack_collection.text_index.add_callback,
836
 
                parents=True, is_locked=self.is_locked,
837
 
                inconsistency_fatal=False),
838
 
            access=self._pack_collection.text_index.data_access)
839
 
        # No parents, individual CHK pages don't have specific ancestry
840
 
        self.chk_bytes = GroupCompressVersionedFiles(
841
 
            _GCGraphIndex(self._pack_collection.chk_index.combined_index,
842
 
                add_callback=self._pack_collection.chk_index.add_callback,
843
 
                parents=False, is_locked=self.is_locked,
844
 
                inconsistency_fatal=False),
845
 
            access=self._pack_collection.chk_index.data_access)
846
 
        search_key_name = self._format._serializer.search_key_name
847
 
        search_key_func = chk_map.search_key_registry.get(search_key_name)
848
 
        self.chk_bytes._search_key_func = search_key_func
849
 
        # True when the repository object is 'write locked' (as opposed to the
850
 
        # physical lock only taken out around changes to the pack-names list.)
851
 
        # Another way to represent this would be a decorator around the control
852
 
        # files object that presents logical locks as physical ones - if this
853
 
        # gets ugly consider that alternative design. RBC 20071011
854
 
        self._write_lock_count = 0
855
 
        self._transaction = None
856
 
        # for tests
857
 
        self._reconcile_does_inventory_gc = True
858
 
        self._reconcile_fixes_text_parents = True
859
 
        self._reconcile_backsup_inventory = False
860
 
 
861
 
    def _add_inventory_checked(self, revision_id, inv, parents):
862
 
        """Add inv to the repository after checking the inputs.
863
 
 
864
 
        This function can be overridden to allow different inventory styles.
865
 
 
866
 
        :seealso: add_inventory, for the contract.
867
 
        """
868
 
        # make inventory
869
 
        serializer = self._format._serializer
870
 
        result = inventory.CHKInventory.from_inventory(self.chk_bytes, inv,
871
 
            maximum_size=serializer.maximum_size,
872
 
            search_key_name=serializer.search_key_name)
873
 
        inv_lines = result.to_lines()
874
 
        return self._inventory_add_lines(revision_id, parents,
875
 
            inv_lines, check_content=False)
876
 
 
877
 
    def _create_inv_from_null(self, delta, revision_id):
878
 
        """This will mutate new_inv directly.
879
 
 
880
 
        This is a simplified form of create_by_apply_delta which knows that all
881
 
        the old values must be None, so everything is a create.
882
 
        """
883
 
        serializer = self._format._serializer
884
 
        new_inv = inventory.CHKInventory(serializer.search_key_name)
885
 
        new_inv.revision_id = revision_id
886
 
        entry_to_bytes = new_inv._entry_to_bytes
887
 
        id_to_entry_dict = {}
888
 
        parent_id_basename_dict = {}
889
 
        for old_path, new_path, file_id, entry in delta:
890
 
            if old_path is not None:
891
 
                raise ValueError('Invalid delta, somebody tried to delete %r'
892
 
                                 ' from the NULL_REVISION'
893
 
                                 % ((old_path, file_id),))
894
 
            if new_path is None:
895
 
                raise ValueError('Invalid delta, delta from NULL_REVISION has'
896
 
                                 ' no new_path %r' % (file_id,))
897
 
            if new_path == '':
898
 
                new_inv.root_id = file_id
899
 
                parent_id_basename_key = StaticTuple('', '').intern()
900
 
            else:
901
 
                utf8_entry_name = entry.name.encode('utf-8')
902
 
                parent_id_basename_key = StaticTuple(entry.parent_id,
903
 
                                                     utf8_entry_name).intern()
904
 
            new_value = entry_to_bytes(entry)
905
 
            # Populate Caches?
906
 
            # new_inv._path_to_fileid_cache[new_path] = file_id
907
 
            key = StaticTuple(file_id).intern()
908
 
            id_to_entry_dict[key] = new_value
909
 
            parent_id_basename_dict[parent_id_basename_key] = file_id
910
 
 
911
 
        new_inv._populate_from_dicts(self.chk_bytes, id_to_entry_dict,
912
 
            parent_id_basename_dict, maximum_size=serializer.maximum_size)
913
 
        return new_inv
914
 
 
915
 
    def add_inventory_by_delta(self, basis_revision_id, delta, new_revision_id,
916
 
                               parents, basis_inv=None, propagate_caches=False):
917
 
        """Add a new inventory expressed as a delta against another revision.
918
 
 
919
 
        :param basis_revision_id: The inventory id the delta was created
920
 
            against.
921
 
        :param delta: The inventory delta (see Inventory.apply_delta for
922
 
            details).
923
 
        :param new_revision_id: The revision id that the inventory is being
924
 
            added for.
925
 
        :param parents: The revision ids of the parents that revision_id is
926
 
            known to have and are in the repository already. These are supplied
927
 
            for repositories that depend on the inventory graph for revision
928
 
            graph access, as well as for those that pun ancestry with delta
929
 
            compression.
930
 
        :param basis_inv: The basis inventory if it is already known,
931
 
            otherwise None.
932
 
        :param propagate_caches: If True, the caches for this inventory are
933
 
          copied to and updated for the result if possible.
934
 
 
935
 
        :returns: (validator, new_inv)
936
 
            The validator(which is a sha1 digest, though what is sha'd is
937
 
            repository format specific) of the serialized inventory, and the
938
 
            resulting inventory.
939
 
        """
940
 
        if not self.is_in_write_group():
941
 
            raise AssertionError("%r not in write group" % (self,))
942
 
        _mod_revision.check_not_reserved_id(new_revision_id)
943
 
        basis_tree = None
944
 
        if basis_inv is None:
945
 
            if basis_revision_id == _mod_revision.NULL_REVISION:
946
 
                new_inv = self._create_inv_from_null(delta, new_revision_id)
947
 
                if new_inv.root_id is None:
948
 
                    raise errors.RootMissing()
949
 
                inv_lines = new_inv.to_lines()
950
 
                return self._inventory_add_lines(new_revision_id, parents,
951
 
                    inv_lines, check_content=False), new_inv
952
 
            else:
953
 
                basis_tree = self.revision_tree(basis_revision_id)
954
 
                basis_tree.lock_read()
955
 
                basis_inv = basis_tree.inventory
956
 
        try:
957
 
            result = basis_inv.create_by_apply_delta(delta, new_revision_id,
958
 
                propagate_caches=propagate_caches)
959
 
            inv_lines = result.to_lines()
960
 
            return self._inventory_add_lines(new_revision_id, parents,
961
 
                inv_lines, check_content=False), result
962
 
        finally:
963
 
            if basis_tree is not None:
964
 
                basis_tree.unlock()
965
 
 
966
 
    def _deserialise_inventory(self, revision_id, bytes):
967
 
        return inventory.CHKInventory.deserialise(self.chk_bytes, bytes,
968
 
            (revision_id,))
969
 
 
970
 
    def _iter_inventories(self, revision_ids, ordering):
971
 
        """Iterate over many inventory objects."""
972
 
        if ordering is None:
973
 
            ordering = 'unordered'
974
 
        keys = [(revision_id,) for revision_id in revision_ids]
975
 
        stream = self.inventories.get_record_stream(keys, ordering, True)
976
 
        texts = {}
977
 
        for record in stream:
978
 
            if record.storage_kind != 'absent':
979
 
                texts[record.key] = record.get_bytes_as('fulltext')
980
 
            else:
981
 
                raise errors.NoSuchRevision(self, record.key)
982
 
        for key in keys:
983
 
            yield inventory.CHKInventory.deserialise(self.chk_bytes, texts[key], key)
984
 
 
985
 
    def _iter_inventory_xmls(self, revision_ids, ordering):
986
 
        # Without a native 'xml' inventory, this method doesn't make sense.
987
 
        # However older working trees, and older bundles want it - so we supply
988
 
        # it allowing _get_inventory_xml to work. Bundles currently use the
989
 
        # serializer directly; this also isn't ideal, but there isn't an xml
990
 
        # iteration interface offered at all for repositories. We could make
991
 
        # _iter_inventory_xmls be part of the contract, even if kept private.
992
 
        inv_to_str = self._serializer.write_inventory_to_string
993
 
        for inv in self.iter_inventories(revision_ids, ordering=ordering):
994
 
            yield inv_to_str(inv), inv.revision_id
995
 
 
996
 
    def _find_present_inventory_keys(self, revision_keys):
997
 
        parent_map = self.inventories.get_parent_map(revision_keys)
998
 
        present_inventory_keys = set(k for k in parent_map)
999
 
        return present_inventory_keys
1000
 
 
1001
 
    def fileids_altered_by_revision_ids(self, revision_ids, _inv_weave=None):
1002
 
        """Find the file ids and versions affected by revisions.
1003
 
 
1004
 
        :param revisions: an iterable containing revision ids.
1005
 
        :param _inv_weave: The inventory weave from this repository or None.
1006
 
            If None, the inventory weave will be opened automatically.
1007
 
        :return: a dictionary mapping altered file-ids to an iterable of
1008
 
            revision_ids. Each altered file-ids has the exact revision_ids that
1009
 
            altered it listed explicitly.
1010
 
        """
1011
 
        rich_root = self.supports_rich_root()
1012
 
        bytes_to_info = inventory.CHKInventory._bytes_to_utf8name_key
1013
 
        file_id_revisions = {}
1014
 
        pb = ui.ui_factory.nested_progress_bar()
1015
 
        try:
1016
 
            revision_keys = [(r,) for r in revision_ids]
1017
 
            parent_keys = self._find_parent_keys_of_revisions(revision_keys)
1018
 
            # TODO: instead of using _find_present_inventory_keys, change the
1019
 
            #       code paths to allow missing inventories to be tolerated.
1020
 
            #       However, we only want to tolerate missing parent
1021
 
            #       inventories, not missing inventories for revision_ids
1022
 
            present_parent_inv_keys = self._find_present_inventory_keys(
1023
 
                                        parent_keys)
1024
 
            present_parent_inv_ids = set(
1025
 
                [k[-1] for k in present_parent_inv_keys])
1026
 
            inventories_to_read = set(revision_ids)
1027
 
            inventories_to_read.update(present_parent_inv_ids)
1028
 
            root_key_info = _build_interesting_key_sets(
1029
 
                self, inventories_to_read, present_parent_inv_ids)
1030
 
            interesting_root_keys = root_key_info.interesting_root_keys
1031
 
            uninteresting_root_keys = root_key_info.uninteresting_root_keys
1032
 
            chk_bytes = self.chk_bytes
1033
 
            for record, items in chk_map.iter_interesting_nodes(chk_bytes,
1034
 
                        interesting_root_keys, uninteresting_root_keys,
1035
 
                        pb=pb):
1036
 
                for name, bytes in items:
1037
 
                    (name_utf8, file_id, revision_id) = bytes_to_info(bytes)
1038
 
                    # TODO: consider interning file_id, revision_id here, or
1039
 
                    #       pushing that intern() into bytes_to_info()
1040
 
                    # TODO: rich_root should always be True here, for all
1041
 
                    #       repositories that support chk_bytes
1042
 
                    if not rich_root and name_utf8 == '':
1043
 
                        continue
1044
 
                    try:
1045
 
                        file_id_revisions[file_id].add(revision_id)
1046
 
                    except KeyError:
1047
 
                        file_id_revisions[file_id] = set([revision_id])
1048
 
        finally:
1049
 
            pb.finished()
1050
 
        return file_id_revisions
1051
 
 
1052
 
    def find_text_key_references(self):
1053
 
        """Find the text key references within the repository.
1054
 
 
1055
 
        :return: A dictionary mapping text keys ((fileid, revision_id) tuples)
1056
 
            to whether they were referred to by the inventory of the
1057
 
            revision_id that they contain. The inventory texts from all present
1058
 
            revision ids are assessed to generate this report.
1059
 
        """
1060
 
        # XXX: Slow version but correct: rewrite as a series of delta
1061
 
        # examinations/direct tree traversal. Note that that will require care
1062
 
        # as a common node is reachable both from the inventory that added it,
1063
 
        # and others afterwards.
1064
 
        revision_keys = self.revisions.keys()
1065
 
        result = {}
1066
 
        rich_roots = self.supports_rich_root()
1067
 
        pb = ui.ui_factory.nested_progress_bar()
1068
 
        try:
1069
 
            all_revs = self.all_revision_ids()
1070
 
            total = len(all_revs)
1071
 
            for pos, inv in enumerate(self.iter_inventories(all_revs)):
1072
 
                pb.update("Finding text references", pos, total)
1073
 
                for _, entry in inv.iter_entries():
1074
 
                    if not rich_roots and entry.file_id == inv.root_id:
1075
 
                        continue
1076
 
                    key = (entry.file_id, entry.revision)
1077
 
                    result.setdefault(key, False)
1078
 
                    if entry.revision == inv.revision_id:
1079
 
                        result[key] = True
1080
 
            return result
1081
 
        finally:
1082
 
            pb.finished()
1083
 
 
1084
 
    @needs_write_lock
1085
 
    def reconcile_canonicalize_chks(self):
1086
 
        """Reconcile this repository to make sure all CHKs are in canonical
1087
 
        form.
1088
 
        """
1089
 
        from bzrlib.reconcile import PackReconciler
1090
 
        reconciler = PackReconciler(self, thorough=True, canonicalize_chks=True)
1091
 
        reconciler.reconcile()
1092
 
        return reconciler
1093
 
 
1094
 
    def _reconcile_pack(self, collection, packs, extension, revs, pb):
1095
 
        packer = GCCHKReconcilePacker(collection, packs, extension)
1096
 
        return packer.pack(pb)
1097
 
 
1098
 
    def _canonicalize_chks_pack(self, collection, packs, extension, revs, pb):
1099
 
        packer = GCCHKCanonicalizingPacker(collection, packs, extension, revs)
1100
 
        return packer.pack(pb)
1101
 
 
1102
 
    def _get_source(self, to_format):
1103
 
        """Return a source for streaming from this repository."""
1104
 
        if self._format._serializer == to_format._serializer:
1105
 
            # We must be exactly the same format, otherwise stuff like the chk
1106
 
            # page layout might be different.
1107
 
            # Actually, this test is just slightly looser than exact so that
1108
 
            # CHK2 <-> 2a transfers will work.
1109
 
            return GroupCHKStreamSource(self, to_format)
1110
 
        return super(CHKInventoryRepository, self)._get_source(to_format)
1111
 
 
1112
 
    def _find_inconsistent_revision_parents(self, revisions_iterator=None):
1113
 
        """Find revisions with different parent lists in the revision object
1114
 
        and in the index graph.
1115
 
 
1116
 
        :param revisions_iterator: None, or an iterator of (revid,
1117
 
            Revision-or-None). This iterator controls the revisions checked.
1118
 
        :returns: an iterator yielding tuples of (revison-id, parents-in-index,
1119
 
            parents-in-revision).
1120
 
        """
1121
 
        if not self.is_locked():
1122
 
            raise AssertionError()
1123
 
        vf = self.revisions
1124
 
        if revisions_iterator is None:
1125
 
            revisions_iterator = self._iter_revisions(None)
1126
 
        for revid, revision in revisions_iterator:
1127
 
            if revision is None:
1128
 
                pass
1129
 
            parent_map = vf.get_parent_map([(revid,)])
1130
 
            parents_according_to_index = tuple(parent[-1] for parent in
1131
 
                parent_map[(revid,)])
1132
 
            parents_according_to_revision = tuple(revision.parent_ids)
1133
 
            if parents_according_to_index != parents_according_to_revision:
1134
 
                yield (revid, parents_according_to_index,
1135
 
                    parents_according_to_revision)
1136
 
 
1137
 
    def _check_for_inconsistent_revision_parents(self):
1138
 
        inconsistencies = list(self._find_inconsistent_revision_parents())
1139
 
        if inconsistencies:
1140
 
            raise errors.BzrCheckError(
1141
 
                "Revision index has inconsistent parents.")
1142
 
 
1143
 
 
1144
 
class GroupCHKStreamSource(StreamSource):
1145
 
    """Used when both the source and target repo are GroupCHK repos."""
1146
 
 
1147
 
    def __init__(self, from_repository, to_format):
1148
 
        """Create a StreamSource streaming from from_repository."""
1149
 
        super(GroupCHKStreamSource, self).__init__(from_repository, to_format)
1150
 
        self._revision_keys = None
1151
 
        self._text_keys = None
1152
 
        self._text_fetch_order = 'groupcompress'
1153
 
        self._chk_id_roots = None
1154
 
        self._chk_p_id_roots = None
1155
 
 
1156
 
    def _get_inventory_stream(self, inventory_keys, allow_absent=False):
1157
 
        """Get a stream of inventory texts.
1158
 
 
1159
 
        When this function returns, self._chk_id_roots and self._chk_p_id_roots
1160
 
        should be populated.
1161
 
        """
1162
 
        self._chk_id_roots = []
1163
 
        self._chk_p_id_roots = []
1164
 
        def _filtered_inv_stream():
1165
 
            id_roots_set = set()
1166
 
            p_id_roots_set = set()
1167
 
            source_vf = self.from_repository.inventories
1168
 
            stream = source_vf.get_record_stream(inventory_keys,
1169
 
                                                 'groupcompress', True)
1170
 
            for record in stream:
1171
 
                if record.storage_kind == 'absent':
1172
 
                    if allow_absent:
1173
 
                        continue
1174
 
                    else:
1175
 
                        raise errors.NoSuchRevision(self, record.key)
1176
 
                bytes = record.get_bytes_as('fulltext')
1177
 
                chk_inv = inventory.CHKInventory.deserialise(None, bytes,
1178
 
                                                             record.key)
1179
 
                key = chk_inv.id_to_entry.key()
1180
 
                if key not in id_roots_set:
1181
 
                    self._chk_id_roots.append(key)
1182
 
                    id_roots_set.add(key)
1183
 
                p_id_map = chk_inv.parent_id_basename_to_file_id
1184
 
                if p_id_map is None:
1185
 
                    raise AssertionError('Parent id -> file_id map not set')
1186
 
                key = p_id_map.key()
1187
 
                if key not in p_id_roots_set:
1188
 
                    p_id_roots_set.add(key)
1189
 
                    self._chk_p_id_roots.append(key)
1190
 
                yield record
1191
 
            # We have finished processing all of the inventory records, we
1192
 
            # don't need these sets anymore
1193
 
            id_roots_set.clear()
1194
 
            p_id_roots_set.clear()
1195
 
        return ('inventories', _filtered_inv_stream())
1196
 
 
1197
 
    def _get_filtered_chk_streams(self, excluded_revision_keys):
1198
 
        self._text_keys = set()
1199
 
        excluded_revision_keys.discard(_mod_revision.NULL_REVISION)
1200
 
        if not excluded_revision_keys:
1201
 
            uninteresting_root_keys = set()
1202
 
            uninteresting_pid_root_keys = set()
1203
 
        else:
1204
 
            # filter out any excluded revisions whose inventories are not
1205
 
            # actually present
1206
 
            # TODO: Update Repository.iter_inventories() to add
1207
 
            #       ignore_missing=True
1208
 
            present_keys = self.from_repository._find_present_inventory_keys(
1209
 
                            excluded_revision_keys)
1210
 
            present_ids = [k[-1] for k in present_keys]
1211
 
            uninteresting_root_keys = set()
1212
 
            uninteresting_pid_root_keys = set()
1213
 
            for inv in self.from_repository.iter_inventories(present_ids):
1214
 
                uninteresting_root_keys.add(inv.id_to_entry.key())
1215
 
                uninteresting_pid_root_keys.add(
1216
 
                    inv.parent_id_basename_to_file_id.key())
1217
 
        chk_bytes = self.from_repository.chk_bytes
1218
 
        def _filter_id_to_entry():
1219
 
            interesting_nodes = chk_map.iter_interesting_nodes(chk_bytes,
1220
 
                        self._chk_id_roots, uninteresting_root_keys)
1221
 
            for record in _filter_text_keys(interesting_nodes, self._text_keys,
1222
 
                    chk_map._bytes_to_text_key):
1223
 
                if record is not None:
1224
 
                    yield record
1225
 
            # Consumed
1226
 
            self._chk_id_roots = None
1227
 
        yield 'chk_bytes', _filter_id_to_entry()
1228
 
        def _get_parent_id_basename_to_file_id_pages():
1229
 
            for record, items in chk_map.iter_interesting_nodes(chk_bytes,
1230
 
                        self._chk_p_id_roots, uninteresting_pid_root_keys):
1231
 
                if record is not None:
1232
 
                    yield record
1233
 
            # Consumed
1234
 
            self._chk_p_id_roots = None
1235
 
        yield 'chk_bytes', _get_parent_id_basename_to_file_id_pages()
1236
 
 
1237
 
    def _get_text_stream(self):
1238
 
        # Note: We know we don't have to handle adding root keys, because both
1239
 
        # the source and target are the identical network name.
1240
 
        text_stream = self.from_repository.texts.get_record_stream(
1241
 
                        self._text_keys, self._text_fetch_order, False)
1242
 
        return ('texts', text_stream)
1243
 
 
1244
 
    def get_stream(self, search):
1245
 
        def wrap_and_count(pb, rc, stream):
1246
 
            """Yield records from stream while showing progress."""
1247
 
            count = 0
1248
 
            for record in stream:
1249
 
                if count == rc.STEP:
1250
 
                    rc.increment(count)
1251
 
                    pb.update('Estimate', rc.current, rc.max)
1252
 
                    count = 0
1253
 
                count += 1
1254
 
                yield record
1255
 
 
1256
 
        revision_ids = search.get_keys()
1257
 
        pb = ui.ui_factory.nested_progress_bar()
1258
 
        rc = self._record_counter
1259
 
        self._record_counter.setup(len(revision_ids))
1260
 
        for stream_info in self._fetch_revision_texts(revision_ids):
1261
 
            yield (stream_info[0],
1262
 
                wrap_and_count(pb, rc, stream_info[1]))
1263
 
        self._revision_keys = [(rev_id,) for rev_id in revision_ids]
1264
 
        # TODO: The keys to exclude might be part of the search recipe
1265
 
        # For now, exclude all parents that are at the edge of ancestry, for
1266
 
        # which we have inventories
1267
 
        from_repo = self.from_repository
1268
 
        parent_keys = from_repo._find_parent_keys_of_revisions(
1269
 
                        self._revision_keys)
1270
 
        self.from_repository.revisions.clear_cache()
1271
 
        self.from_repository.signatures.clear_cache()
1272
 
        # Clear the repo's get_parent_map cache too.
1273
 
        self.from_repository._unstacked_provider.disable_cache()
1274
 
        self.from_repository._unstacked_provider.enable_cache()
1275
 
        s = self._get_inventory_stream(self._revision_keys)
1276
 
        yield (s[0], wrap_and_count(pb, rc, s[1]))
1277
 
        self.from_repository.inventories.clear_cache()
1278
 
        for stream_info in self._get_filtered_chk_streams(parent_keys):
1279
 
            yield (stream_info[0], wrap_and_count(pb, rc, stream_info[1]))
1280
 
        self.from_repository.chk_bytes.clear_cache()
1281
 
        s = self._get_text_stream()
1282
 
        yield (s[0], wrap_and_count(pb, rc, s[1]))
1283
 
        self.from_repository.texts.clear_cache()
1284
 
        pb.update('Done', rc.max, rc.max)
1285
 
        pb.finished()
1286
 
 
1287
 
    def get_stream_for_missing_keys(self, missing_keys):
1288
 
        # missing keys can only occur when we are byte copying and not
1289
 
        # translating (because translation means we don't send
1290
 
        # unreconstructable deltas ever).
1291
 
        missing_inventory_keys = set()
1292
 
        for key in missing_keys:
1293
 
            if key[0] != 'inventories':
1294
 
                raise AssertionError('The only missing keys we should'
1295
 
                    ' be filling in are inventory keys, not %s'
1296
 
                    % (key[0],))
1297
 
            missing_inventory_keys.add(key[1:])
1298
 
        if self._chk_id_roots or self._chk_p_id_roots:
1299
 
            raise AssertionError('Cannot call get_stream_for_missing_keys'
1300
 
                ' until all of get_stream() has been consumed.')
1301
 
        # Yield the inventory stream, so we can find the chk stream
1302
 
        # Some of the missing_keys will be missing because they are ghosts.
1303
 
        # As such, we can ignore them. The Sink is required to verify there are
1304
 
        # no unavailable texts when the ghost inventories are not filled in.
1305
 
        yield self._get_inventory_stream(missing_inventory_keys,
1306
 
                                         allow_absent=True)
1307
 
        # We use the empty set for excluded_revision_keys, to make it clear
1308
 
        # that we want to transmit all referenced chk pages.
1309
 
        for stream_info in self._get_filtered_chk_streams(set()):
1310
 
            yield stream_info
1311
 
 
1312
 
 
1313
 
class _InterestingKeyInfo(object):
1314
 
    def __init__(self):
1315
 
        self.interesting_root_keys = set()
1316
 
        self.interesting_pid_root_keys = set()
1317
 
        self.uninteresting_root_keys = set()
1318
 
        self.uninteresting_pid_root_keys = set()
1319
 
 
1320
 
    def all_interesting(self):
1321
 
        return self.interesting_root_keys.union(self.interesting_pid_root_keys)
1322
 
 
1323
 
    def all_uninteresting(self):
1324
 
        return self.uninteresting_root_keys.union(
1325
 
            self.uninteresting_pid_root_keys)
1326
 
 
1327
 
    def all_keys(self):
1328
 
        return self.all_interesting().union(self.all_uninteresting())
1329
 
 
1330
 
 
1331
 
def _build_interesting_key_sets(repo, inventory_ids, parent_only_inv_ids):
1332
 
    result = _InterestingKeyInfo()
1333
 
    for inv in repo.iter_inventories(inventory_ids, 'unordered'):
1334
 
        root_key = inv.id_to_entry.key()
1335
 
        pid_root_key = inv.parent_id_basename_to_file_id.key()
1336
 
        if inv.revision_id in parent_only_inv_ids:
1337
 
            result.uninteresting_root_keys.add(root_key)
1338
 
            result.uninteresting_pid_root_keys.add(pid_root_key)
1339
 
        else:
1340
 
            result.interesting_root_keys.add(root_key)
1341
 
            result.interesting_pid_root_keys.add(pid_root_key)
1342
 
    return result
1343
 
 
1344
 
 
1345
 
def _filter_text_keys(interesting_nodes_iterable, text_keys, bytes_to_text_key):
1346
 
    """Iterate the result of iter_interesting_nodes, yielding the records
1347
 
    and adding to text_keys.
1348
 
    """
1349
 
    text_keys_update = text_keys.update
1350
 
    for record, items in interesting_nodes_iterable:
1351
 
        text_keys_update([bytes_to_text_key(b) for n,b in items])
1352
 
        yield record
1353
 
 
1354
 
 
1355
 
class RepositoryFormat2a(RepositoryFormatPack):
1356
 
    """A CHK repository that uses the bencode revision serializer."""
1357
 
 
1358
 
    repository_class = CHKInventoryRepository
1359
 
    supports_external_lookups = True
1360
 
    supports_chks = True
1361
 
    _commit_builder_class = PackRootCommitBuilder
1362
 
    rich_root_data = True
1363
 
    _serializer = chk_serializer.chk_bencode_serializer
1364
 
    _commit_inv_deltas = True
1365
 
    # What index classes to use
1366
 
    index_builder_class = BTreeBuilder
1367
 
    index_class = BTreeGraphIndex
1368
 
    # Note: We cannot unpack a delta that references a text we haven't
1369
 
    # seen yet. There are 2 options, work in fulltexts, or require
1370
 
    # topological sorting. Using fulltexts is more optimal for local
1371
 
    # operations, because the source can be smart about extracting
1372
 
    # multiple in-a-row (and sharing strings). Topological is better
1373
 
    # for remote, because we access less data.
1374
 
    _fetch_order = 'unordered'
1375
 
    _fetch_uses_deltas = False # essentially ignored by the groupcompress code.
1376
 
    fast_deltas = True
1377
 
    pack_compresses = True
1378
 
 
1379
 
    def _get_matching_bzrdir(self):
1380
 
        return bzrdir.format_registry.make_bzrdir('2a')
1381
 
 
1382
 
    def _ignore_setting_bzrdir(self, format):
1383
 
        pass
1384
 
 
1385
 
    _matchingbzrdir = property(_get_matching_bzrdir, _ignore_setting_bzrdir)
1386
 
 
1387
 
    def get_format_string(self):
1388
 
        return ('Bazaar repository format 2a (needs bzr 1.16 or later)\n')
1389
 
 
1390
 
    def get_format_description(self):
1391
 
        """See RepositoryFormat.get_format_description()."""
1392
 
        return ("Repository format 2a - rich roots, group compression"
1393
 
            " and chk inventories")
1394
 
 
1395
 
 
1396
 
class RepositoryFormat2aSubtree(RepositoryFormat2a):
1397
 
    """A 2a repository format that supports nested trees.
1398
 
 
1399
 
    """
1400
 
 
1401
 
    def _get_matching_bzrdir(self):
1402
 
        return bzrdir.format_registry.make_bzrdir('development-subtree')
1403
 
 
1404
 
    def _ignore_setting_bzrdir(self, format):
1405
 
        pass
1406
 
 
1407
 
    _matchingbzrdir = property(_get_matching_bzrdir, _ignore_setting_bzrdir)
1408
 
 
1409
 
    def get_format_string(self):
1410
 
        return ('Bazaar development format 8\n')
1411
 
 
1412
 
    def get_format_description(self):
1413
 
        """See RepositoryFormat.get_format_description()."""
1414
 
        return ("Development repository format 8 - nested trees, "
1415
 
                "group compression and chk inventories")
1416
 
 
1417
 
    experimental = True
1418
 
    supports_tree_reference = True