1
# Copyright (C) 2008, 2009 Canonical Ltd
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.
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.
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
17
"""Repository formats using CHK inventories and groupcompress compression."""
34
revision as _mod_revision,
38
from bzrlib.btree_index import (
42
from bzrlib.index import GraphIndex, GraphIndexBuilder
43
from bzrlib.groupcompress import (
45
GroupCompressVersionedFiles,
47
from bzrlib.repofmt.pack_repo import (
52
PackRootCommitBuilder,
53
RepositoryPackCollection,
60
class GCPack(NewPack):
62
def __init__(self, pack_collection, upload_suffix='', file_mode=None):
63
"""Create a NewPack instance.
65
:param pack_collection: A PackCollection into which this is being
67
:param upload_suffix: An optional suffix to be given to any temporary
68
files created during the pack creation. e.g '.autopack'
69
:param file_mode: An optional file mode to create the new files with.
71
# replaced from NewPack to:
72
# - change inventory reference list length to 1
73
# - change texts reference lists to 1
74
# TODO: patch this to be parameterised
76
# The relative locations of the packs are constrained, but all are
77
# passed in because the caller has them, so as to avoid object churn.
78
index_builder_class = pack_collection._index_builder_class
80
if pack_collection.chk_index is not None:
81
chk_index = index_builder_class(reference_lists=0)
85
# Revisions: parents list, no text compression.
86
index_builder_class(reference_lists=1),
87
# Inventory: We want to map compression only, but currently the
88
# knit code hasn't been updated enough to understand that, so we
89
# have a regular 2-list index giving parents and compression
91
index_builder_class(reference_lists=1),
92
# Texts: per file graph, for all fileids - so one reference list
93
# and two elements in the key tuple.
94
index_builder_class(reference_lists=1, key_elements=2),
95
# Signatures: Just blobs to store, no compression, no parents
97
index_builder_class(reference_lists=0),
98
# CHK based storage - just blobs, no compression or parents.
101
self._pack_collection = pack_collection
102
# When we make readonly indices, we need this.
103
self.index_class = pack_collection._index_class
104
# where should the new pack be opened
105
self.upload_transport = pack_collection._upload_transport
106
# where are indices written out to
107
self.index_transport = pack_collection._index_transport
108
# where is the pack renamed to when it is finished?
109
self.pack_transport = pack_collection._pack_transport
110
# What file mode to upload the pack and indices with.
111
self._file_mode = file_mode
112
# tracks the content written to the .pack file.
113
self._hash = osutils.md5()
114
# a four-tuple with the length in bytes of the indices, once the pack
115
# is finalised. (rev, inv, text, sigs)
116
self.index_sizes = None
117
# How much data to cache when writing packs. Note that this is not
118
# synchronised with reads, because it's not in the transport layer, so
119
# is not safe unless the client knows it won't be reading from the pack
121
self._cache_limit = 0
122
# the temporary pack file name.
123
self.random_name = osutils.rand_chars(20) + upload_suffix
124
# when was this pack started ?
125
self.start_time = time.time()
126
# open an output stream for the data added to the pack.
127
self.write_stream = self.upload_transport.open_write_stream(
128
self.random_name, mode=self._file_mode)
129
if 'pack' in debug.debug_flags:
130
trace.mutter('%s: create_pack: pack stream open: %s%s t+%6.3fs',
131
time.ctime(), self.upload_transport.base, self.random_name,
132
time.time() - self.start_time)
133
# A list of byte sequences to be written to the new pack, and the
134
# aggregate size of them. Stored as a list rather than separate
135
# variables so that the _write_data closure below can update them.
136
self._buffer = [[], 0]
137
# create a callable for adding data
139
# robertc says- this is a closure rather than a method on the object
140
# so that the variables are locals, and faster than accessing object
142
def _write_data(bytes, flush=False, _buffer=self._buffer,
143
_write=self.write_stream.write, _update=self._hash.update):
144
_buffer[0].append(bytes)
145
_buffer[1] += len(bytes)
147
if _buffer[1] > self._cache_limit or flush:
148
bytes = ''.join(_buffer[0])
152
# expose this on self, for the occasion when clients want to add data.
153
self._write_data = _write_data
154
# a pack writer object to serialise pack records.
155
self._writer = pack.ContainerWriter(self._write_data)
157
# what state is the pack in? (open, finished, aborted)
160
def _check_references(self):
161
"""Make sure our external references are present.
163
Packs are allowed to have deltas whose base is not in the pack, but it
164
must be present somewhere in this collection. It is not allowed to
165
have deltas based on a fallback repository.
166
(See <https://bugs.launchpad.net/bzr/+bug/288751>)
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.
173
class ResumedGCPack(ResumedPack):
175
def _check_references(self):
176
"""Make sure our external compression parents are present."""
177
# See GCPack._check_references for why this is empty
179
def _get_external_refs(self, index):
180
# GC repositories don't have compression parents external to a given
185
class GCCHKPacker(Packer):
186
"""This class understand what it takes to collect a GCCHK repo."""
188
def __init__(self, pack_collection, packs, suffix, revision_ids=None,
190
super(GCCHKPacker, self).__init__(pack_collection, packs, suffix,
191
revision_ids=revision_ids,
192
reload_func=reload_func)
193
self._pack_collection = pack_collection
194
# ATM, We only support this for GCCHK repositories
195
if pack_collection.chk_index is None:
196
raise AssertionError('pack_collection.chk_index should not be None')
197
self._gather_text_refs = False
198
self._chk_id_roots = []
199
self._chk_p_id_roots = []
200
self._text_refs = None
201
# set by .pack() if self.revision_ids is not None
202
self.revision_keys = None
204
def _get_progress_stream(self, source_vf, keys, message, pb):
206
substream = source_vf.get_record_stream(keys, 'groupcompress', True)
207
for idx, record in enumerate(substream):
209
pb.update(message, idx + 1, len(keys))
213
def _get_filtered_inv_stream(self, source_vf, keys, message, pb=None):
214
"""Filter the texts of inventories, to find the chk pages."""
215
total_keys = len(keys)
216
def _filtered_inv_stream():
218
p_id_roots_set = set()
219
stream = source_vf.get_record_stream(keys, 'groupcompress', True)
220
for idx, record in enumerate(stream):
221
# Inventories should always be with revisions; assume success.
222
bytes = record.get_bytes_as('fulltext')
223
chk_inv = inventory.CHKInventory.deserialise(None, bytes,
226
pb.update('inv', idx, total_keys)
227
key = chk_inv.id_to_entry.key()
228
if key not in id_roots_set:
229
self._chk_id_roots.append(key)
230
id_roots_set.add(key)
231
p_id_map = chk_inv.parent_id_basename_to_file_id
233
raise AssertionError('Parent id -> file_id map not set')
235
if key not in p_id_roots_set:
236
p_id_roots_set.add(key)
237
self._chk_p_id_roots.append(key)
239
# We have finished processing all of the inventory records, we
240
# don't need these sets anymore
242
p_id_roots_set.clear()
243
return _filtered_inv_stream()
245
def _get_chk_streams(self, source_vf, keys, pb=None):
246
# We want to stream the keys from 'id_roots', and things they
247
# reference, and then stream things from p_id_roots and things they
248
# reference, and then any remaining keys that we didn't get to.
250
# We also group referenced texts together, so if one root references a
251
# text with prefix 'a', and another root references a node with prefix
252
# 'a', we want to yield those nodes before we yield the nodes for 'b'
253
# This keeps 'similar' nodes together.
255
# Note: We probably actually want multiple streams here, to help the
256
# client understand that the different levels won't compress well
257
# against each other.
258
# Test the difference between using one Group per level, and
259
# using 1 Group per prefix. (so '' (root) would get a group, then
260
# all the references to search-key 'a' would get a group, etc.)
261
total_keys = len(keys)
262
remaining_keys = set(keys)
264
if self._gather_text_refs:
265
bytes_to_info = inventory.CHKInventory._bytes_to_utf8name_key
266
self._text_refs = set()
267
def _get_referenced_stream(root_keys, parse_leaf_nodes=False):
270
keys_by_search_prefix = {}
271
remaining_keys.difference_update(cur_keys)
273
def handle_internal_node(node):
274
for prefix, value in node._items.iteritems():
275
# We don't want to request the same key twice, and we
276
# want to order it by the first time it is seen.
277
# Even further, we don't want to request a key which is
278
# not in this group of pack files (it should be in the
279
# repo, but it doesn't have to be in the group being
281
# TODO: consider how to treat externally referenced chk
282
# pages as 'external_references' so that we
283
# always fill them in for stacked branches
284
if value not in next_keys and value in remaining_keys:
285
keys_by_search_prefix.setdefault(prefix,
288
def handle_leaf_node(node):
289
# Store is None, because we know we have a LeafNode, and we
290
# just want its entries
291
for file_id, bytes in node.iteritems(None):
292
name_utf8, file_id, revision_id = bytes_to_info(bytes)
293
self._text_refs.add((file_id, revision_id))
295
stream = source_vf.get_record_stream(cur_keys,
296
'as-requested', True)
297
for record in stream:
298
if record.storage_kind == 'absent':
299
# An absent CHK record: we assume that the missing
300
# record is in a different pack - e.g. a page not
301
# altered by the commit we're packing.
303
bytes = record.get_bytes_as('fulltext')
304
# We don't care about search_key_func for this code,
305
# because we only care about external references.
306
node = chk_map._deserialise(bytes, record.key,
307
search_key_func=None)
308
common_base = node._search_prefix
309
if isinstance(node, chk_map.InternalNode):
310
handle_internal_node(node)
311
elif parse_leaf_nodes:
312
handle_leaf_node(node)
315
pb.update('chk node', counter[0], total_keys)
318
# Double check that we won't be emitting any keys twice
319
# If we get rid of the pre-calculation of all keys, we could
320
# turn this around and do
321
# next_keys.difference_update(seen_keys)
322
# However, we also may have references to chk pages in another
323
# pack file during autopack. We filter earlier, so we should no
324
# longer need to do this
325
# next_keys = next_keys.intersection(remaining_keys)
327
for prefix in sorted(keys_by_search_prefix):
328
cur_keys.extend(keys_by_search_prefix.pop(prefix))
329
for stream in _get_referenced_stream(self._chk_id_roots,
330
self._gather_text_refs):
332
del self._chk_id_roots
333
# while it isn't really possible for chk_id_roots to not be in the
334
# local group of packs, it is possible that the tree shape has not
335
# changed recently, so we need to filter _chk_p_id_roots by the
337
chk_p_id_roots = [key for key in self._chk_p_id_roots
338
if key in remaining_keys]
339
del self._chk_p_id_roots
340
for stream in _get_referenced_stream(chk_p_id_roots, False):
343
trace.mutter('There were %d keys in the chk index, %d of which'
344
' were not referenced', total_keys,
346
if self.revision_ids is None:
347
stream = source_vf.get_record_stream(remaining_keys,
351
def _build_vf(self, index_name, parents, delta, for_write=False):
352
"""Build a VersionedFiles instance on top of this group of packs."""
353
index_name = index_name + '_index'
355
access = knit._DirectPackAccess(index_to_pack)
358
if self.new_pack is None:
359
raise AssertionError('No new pack has been set')
360
index = getattr(self.new_pack, index_name)
361
index_to_pack[index] = self.new_pack.access_tuple()
362
index.set_optimize(for_size=True)
363
access.set_writer(self.new_pack._writer, index,
364
self.new_pack.access_tuple())
365
add_callback = index.add_nodes
368
for pack in self.packs:
369
sub_index = getattr(pack, index_name)
370
index_to_pack[sub_index] = pack.access_tuple()
371
indices.append(sub_index)
372
index = _mod_index.CombinedGraphIndex(indices)
374
vf = GroupCompressVersionedFiles(
376
add_callback=add_callback,
378
is_locked=self._pack_collection.repo.is_locked),
383
def _build_vfs(self, index_name, parents, delta):
384
"""Build the source and target VersionedFiles."""
385
source_vf = self._build_vf(index_name, parents,
386
delta, for_write=False)
387
target_vf = self._build_vf(index_name, parents,
388
delta, for_write=True)
389
return source_vf, target_vf
391
def _copy_stream(self, source_vf, target_vf, keys, message, vf_to_stream,
393
trace.mutter('repacking %d %s', len(keys), message)
394
self.pb.update('repacking %s' % (message,), pb_offset)
395
child_pb = ui.ui_factory.nested_progress_bar()
397
stream = vf_to_stream(source_vf, keys, message, child_pb)
398
for _ in target_vf._insert_record_stream(stream,
405
def _copy_revision_texts(self):
406
source_vf, target_vf = self._build_vfs('revision', True, False)
407
if not self.revision_keys:
408
# We are doing a full fetch, aka 'pack'
409
self.revision_keys = source_vf.keys()
410
self._copy_stream(source_vf, target_vf, self.revision_keys,
411
'revisions', self._get_progress_stream, 1)
413
def _copy_inventory_texts(self):
414
source_vf, target_vf = self._build_vfs('inventory', True, True)
415
self._copy_stream(source_vf, target_vf, self.revision_keys,
416
'inventories', self._get_filtered_inv_stream, 2)
418
def _copy_chk_texts(self):
419
source_vf, target_vf = self._build_vfs('chk', False, False)
420
# TODO: This is technically spurious... if it is a performance issue,
422
total_keys = source_vf.keys()
423
trace.mutter('repacking chk: %d id_to_entry roots,'
424
' %d p_id_map roots, %d total keys',
425
len(self._chk_id_roots), len(self._chk_p_id_roots),
427
self.pb.update('repacking chk', 3)
428
child_pb = ui.ui_factory.nested_progress_bar()
430
for stream in self._get_chk_streams(source_vf, total_keys,
432
for _ in target_vf._insert_record_stream(stream,
439
def _copy_text_texts(self):
440
source_vf, target_vf = self._build_vfs('text', True, True)
441
# XXX: We don't walk the chk map to determine referenced (file_id,
442
# revision_id) keys. We don't do it yet because you really need
443
# to filter out the ones that are present in the parents of the
444
# rev just before the ones you are copying, otherwise the filter
445
# is grabbing too many keys...
446
text_keys = source_vf.keys()
447
self._copy_stream(source_vf, target_vf, text_keys,
448
'text', self._get_progress_stream, 4)
450
def _copy_signature_texts(self):
451
source_vf, target_vf = self._build_vfs('signature', False, False)
452
signature_keys = source_vf.keys()
453
signature_keys.intersection(self.revision_keys)
454
self._copy_stream(source_vf, target_vf, signature_keys,
455
'signatures', self._get_progress_stream, 5)
457
def _create_pack_from_packs(self):
458
self.pb.update('repacking', 0, 7)
459
self.new_pack = self.open_pack()
460
# Is this necessary for GC ?
461
self.new_pack.set_write_cache_size(1024*1024)
462
self._copy_revision_texts()
463
self._copy_inventory_texts()
464
self._copy_chk_texts()
465
self._copy_text_texts()
466
self._copy_signature_texts()
467
self.new_pack._check_references()
468
if not self._use_pack(self.new_pack):
469
self.new_pack.abort()
471
self.pb.update('finishing repack', 6, 7)
472
self.new_pack.finish()
473
self._pack_collection.allocate(self.new_pack)
477
class GCCHKReconcilePacker(GCCHKPacker):
478
"""A packer which regenerates indices etc as it copies.
480
This is used by ``bzr reconcile`` to cause parent text pointers to be
484
def __init__(self, *args, **kwargs):
485
super(GCCHKReconcilePacker, self).__init__(*args, **kwargs)
486
self._data_changed = False
487
self._gather_text_refs = True
489
def _copy_inventory_texts(self):
490
source_vf, target_vf = self._build_vfs('inventory', True, True)
491
self._copy_stream(source_vf, target_vf, self.revision_keys,
492
'inventories', self._get_filtered_inv_stream, 2)
493
if source_vf.keys() != self.revision_keys:
494
self._data_changed = True
496
def _copy_text_texts(self):
497
"""generate what texts we should have and then copy."""
498
source_vf, target_vf = self._build_vfs('text', True, True)
499
trace.mutter('repacking %d texts', len(self._text_refs))
500
self.pb.update("repacking texts", 4)
501
# we have three major tasks here:
502
# 1) generate the ideal index
503
repo = self._pack_collection.repo
504
# We want the one we just wrote, so base it on self.new_pack
505
revision_vf = self._build_vf('revision', True, False, for_write=True)
506
ancestor_keys = revision_vf.get_parent_map(revision_vf.keys())
507
# Strip keys back into revision_ids.
508
ancestors = dict((k[0], tuple([p[0] for p in parents]))
509
for k, parents in ancestor_keys.iteritems())
511
# TODO: _generate_text_key_index should be much cheaper to generate from
512
# a chk repository, rather than the current implementation
513
ideal_index = repo._generate_text_key_index(None, ancestors)
514
file_id_parent_map = source_vf.get_parent_map(self._text_refs)
515
# 2) generate a keys list that contains all the entries that can
516
# be used as-is, with corrected parents.
518
new_parent_keys = {} # (key, parent_keys)
520
NULL_REVISION = _mod_revision.NULL_REVISION
521
for key in self._text_refs:
527
ideal_parents = tuple(ideal_index[key])
529
discarded_keys.append(key)
530
self._data_changed = True
532
if ideal_parents == (NULL_REVISION,):
534
source_parents = file_id_parent_map[key]
535
if ideal_parents == source_parents:
539
# We need to change the parent graph, but we don't need to
540
# re-insert the text (since we don't pun the compression
541
# parent with the parents list)
542
self._data_changed = True
543
new_parent_keys[key] = ideal_parents
544
# we're finished with some data.
546
del file_id_parent_map
547
# 3) bulk copy the data, updating records than need it
548
def _update_parents_for_texts():
549
stream = source_vf.get_record_stream(self._text_refs,
550
'groupcompress', False)
551
for record in stream:
552
if record.key in new_parent_keys:
553
record.parents = new_parent_keys[record.key]
555
target_vf.insert_record_stream(_update_parents_for_texts())
557
def _use_pack(self, new_pack):
558
"""Override _use_pack to check for reconcile having changed content."""
559
return new_pack.data_inserted() and self._data_changed
562
class GCRepositoryPackCollection(RepositoryPackCollection):
564
pack_factory = GCPack
565
resumed_pack_factory = ResumedGCPack
567
def _execute_pack_operations(self, pack_operations,
568
_packer_class=GCCHKPacker,
570
"""Execute a series of pack operations.
572
:param pack_operations: A list of [revision_count, packs_to_combine].
573
:param _packer_class: The class of packer to use (default: Packer).
576
# XXX: Copied across from RepositoryPackCollection simply because we
577
# want to override the _packer_class ... :(
578
for revision_count, packs in pack_operations:
579
# we may have no-ops from the setup logic
582
packer = GCCHKPacker(self, packs, '.autopack',
583
reload_func=reload_func)
586
except errors.RetryWithNewPacks:
587
# An exception is propagating out of this context, make sure
588
# this packer has cleaned up. Packer() doesn't set its new_pack
589
# state into the RepositoryPackCollection object, so we only
590
# have access to it directly here.
591
if packer.new_pack is not None:
592
packer.new_pack.abort()
595
self._remove_pack_from_memory(pack)
596
# record the newly available packs and stop advertising the old
598
self._save_pack_names(clear_obsolete_packs=True)
599
# Move the old packs out of the way now they are no longer referenced.
600
for revision_count, packs in pack_operations:
601
self._obsolete_packs(packs)
604
class CHKInventoryRepository(KnitPackRepository):
605
"""subclass of KnitPackRepository that uses CHK based inventories."""
607
def __init__(self, _format, a_bzrdir, control_files, _commit_builder_class,
609
"""Overridden to change pack collection class."""
610
KnitPackRepository.__init__(self, _format, a_bzrdir, control_files,
611
_commit_builder_class, _serializer)
612
# and now replace everything it did :)
613
index_transport = self._transport.clone('indices')
614
self._pack_collection = GCRepositoryPackCollection(self,
615
self._transport, index_transport,
616
self._transport.clone('upload'),
617
self._transport.clone('packs'),
618
_format.index_builder_class,
620
use_chk_index=self._format.supports_chks,
622
self.inventories = GroupCompressVersionedFiles(
623
_GCGraphIndex(self._pack_collection.inventory_index.combined_index,
624
add_callback=self._pack_collection.inventory_index.add_callback,
625
parents=True, is_locked=self.is_locked,
626
inconsistency_fatal=False),
627
access=self._pack_collection.inventory_index.data_access)
628
self.revisions = GroupCompressVersionedFiles(
629
_GCGraphIndex(self._pack_collection.revision_index.combined_index,
630
add_callback=self._pack_collection.revision_index.add_callback,
631
parents=True, is_locked=self.is_locked,
632
track_external_parent_refs=True),
633
access=self._pack_collection.revision_index.data_access,
635
self.signatures = GroupCompressVersionedFiles(
636
_GCGraphIndex(self._pack_collection.signature_index.combined_index,
637
add_callback=self._pack_collection.signature_index.add_callback,
638
parents=False, is_locked=self.is_locked,
639
inconsistency_fatal=False),
640
access=self._pack_collection.signature_index.data_access,
642
self.texts = GroupCompressVersionedFiles(
643
_GCGraphIndex(self._pack_collection.text_index.combined_index,
644
add_callback=self._pack_collection.text_index.add_callback,
645
parents=True, is_locked=self.is_locked,
646
inconsistency_fatal=False),
647
access=self._pack_collection.text_index.data_access)
648
# No parents, individual CHK pages don't have specific ancestry
649
self.chk_bytes = GroupCompressVersionedFiles(
650
_GCGraphIndex(self._pack_collection.chk_index.combined_index,
651
add_callback=self._pack_collection.chk_index.add_callback,
652
parents=False, is_locked=self.is_locked,
653
inconsistency_fatal=False),
654
access=self._pack_collection.chk_index.data_access)
655
# True when the repository object is 'write locked' (as opposed to the
656
# physical lock only taken out around changes to the pack-names list.)
657
# Another way to represent this would be a decorator around the control
658
# files object that presents logical locks as physical ones - if this
659
# gets ugly consider that alternative design. RBC 20071011
660
self._write_lock_count = 0
661
self._transaction = None
663
self._reconcile_does_inventory_gc = True
664
self._reconcile_fixes_text_parents = True
665
self._reconcile_backsup_inventory = False
667
def _add_inventory_checked(self, revision_id, inv, parents):
668
"""Add inv to the repository after checking the inputs.
670
This function can be overridden to allow different inventory styles.
672
:seealso: add_inventory, for the contract.
675
serializer = self._format._serializer
676
result = inventory.CHKInventory.from_inventory(self.chk_bytes, inv,
677
maximum_size=serializer.maximum_size,
678
search_key_name=serializer.search_key_name)
679
inv_lines = result.to_lines()
680
return self._inventory_add_lines(revision_id, parents,
681
inv_lines, check_content=False)
683
def _create_inv_from_null(self, delta, revision_id):
684
"""This will mutate new_inv directly.
686
This is a simplified form of create_by_apply_delta which knows that all
687
the old values must be None, so everything is a create.
689
serializer = self._format._serializer
690
new_inv = inventory.CHKInventory(serializer.search_key_name)
691
new_inv.revision_id = revision_id
692
entry_to_bytes = new_inv._entry_to_bytes
693
id_to_entry_dict = {}
694
parent_id_basename_dict = {}
695
for old_path, new_path, file_id, entry in delta:
696
if old_path is not None:
697
raise ValueError('Invalid delta, somebody tried to delete %r'
698
' from the NULL_REVISION'
699
% ((old_path, file_id),))
701
raise ValueError('Invalid delta, delta from NULL_REVISION has'
702
' no new_path %r' % (file_id,))
704
new_inv.root_id = file_id
705
parent_id_basename_key = ('', '')
707
utf8_entry_name = entry.name.encode('utf-8')
708
parent_id_basename_key = (entry.parent_id, utf8_entry_name)
709
new_value = entry_to_bytes(entry)
711
# new_inv._path_to_fileid_cache[new_path] = file_id
712
id_to_entry_dict[(file_id,)] = new_value
713
parent_id_basename_dict[parent_id_basename_key] = file_id
715
new_inv._populate_from_dicts(self.chk_bytes, id_to_entry_dict,
716
parent_id_basename_dict, maximum_size=serializer.maximum_size)
719
def add_inventory_by_delta(self, basis_revision_id, delta, new_revision_id,
720
parents, basis_inv=None, propagate_caches=False):
721
"""Add a new inventory expressed as a delta against another revision.
723
:param basis_revision_id: The inventory id the delta was created
725
:param delta: The inventory delta (see Inventory.apply_delta for
727
:param new_revision_id: The revision id that the inventory is being
729
:param parents: The revision ids of the parents that revision_id is
730
known to have and are in the repository already. These are supplied
731
for repositories that depend on the inventory graph for revision
732
graph access, as well as for those that pun ancestry with delta
734
:param basis_inv: The basis inventory if it is already known,
736
:param propagate_caches: If True, the caches for this inventory are
737
copied to and updated for the result if possible.
739
:returns: (validator, new_inv)
740
The validator(which is a sha1 digest, though what is sha'd is
741
repository format specific) of the serialized inventory, and the
744
if not self.is_in_write_group():
745
raise AssertionError("%r not in write group" % (self,))
746
_mod_revision.check_not_reserved_id(new_revision_id)
748
if basis_inv is None:
749
if basis_revision_id == _mod_revision.NULL_REVISION:
750
new_inv = self._create_inv_from_null(delta, new_revision_id)
751
inv_lines = new_inv.to_lines()
752
return self._inventory_add_lines(new_revision_id, parents,
753
inv_lines, check_content=False), new_inv
755
basis_tree = self.revision_tree(basis_revision_id)
756
basis_tree.lock_read()
757
basis_inv = basis_tree.inventory
759
result = basis_inv.create_by_apply_delta(delta, new_revision_id,
760
propagate_caches=propagate_caches)
761
inv_lines = result.to_lines()
762
return self._inventory_add_lines(new_revision_id, parents,
763
inv_lines, check_content=False), result
765
if basis_tree is not None:
768
def _iter_inventories(self, revision_ids):
769
"""Iterate over many inventory objects."""
770
keys = [(revision_id,) for revision_id in revision_ids]
771
stream = self.inventories.get_record_stream(keys, 'unordered', True)
773
for record in stream:
774
if record.storage_kind != 'absent':
775
texts[record.key] = record.get_bytes_as('fulltext')
777
raise errors.NoSuchRevision(self, record.key)
779
yield inventory.CHKInventory.deserialise(self.chk_bytes, texts[key], key)
781
def _iter_inventory_xmls(self, revision_ids):
782
# Without a native 'xml' inventory, this method doesn't make sense, so
783
# make it raise to trap naughty direct users.
784
raise NotImplementedError(self._iter_inventory_xmls)
786
def _find_present_inventory_keys(self, revision_keys):
787
parent_map = self.inventories.get_parent_map(revision_keys)
788
present_inventory_keys = set(k for k in parent_map)
789
return present_inventory_keys
791
def fileids_altered_by_revision_ids(self, revision_ids, _inv_weave=None):
792
"""Find the file ids and versions affected by revisions.
794
:param revisions: an iterable containing revision ids.
795
:param _inv_weave: The inventory weave from this repository or None.
796
If None, the inventory weave will be opened automatically.
797
:return: a dictionary mapping altered file-ids to an iterable of
798
revision_ids. Each altered file-ids has the exact revision_ids that
799
altered it listed explicitly.
801
rich_root = self.supports_rich_root()
802
bytes_to_info = inventory.CHKInventory._bytes_to_utf8name_key
803
file_id_revisions = {}
804
pb = ui.ui_factory.nested_progress_bar()
806
revision_keys = [(r,) for r in revision_ids]
807
parent_keys = self._find_parent_keys_of_revisions(revision_keys)
808
# TODO: instead of using _find_present_inventory_keys, change the
809
# code paths to allow missing inventories to be tolerated.
810
# However, we only want to tolerate missing parent
811
# inventories, not missing inventories for revision_ids
812
present_parent_inv_keys = self._find_present_inventory_keys(
814
present_parent_inv_ids = set(
815
[k[-1] for k in present_parent_inv_keys])
816
uninteresting_root_keys = set()
817
interesting_root_keys = set()
818
inventories_to_read = set(revision_ids)
819
inventories_to_read.update(present_parent_inv_ids)
820
for inv in self.iter_inventories(inventories_to_read):
821
entry_chk_root_key = inv.id_to_entry.key()
822
if inv.revision_id in present_parent_inv_ids:
823
uninteresting_root_keys.add(entry_chk_root_key)
825
interesting_root_keys.add(entry_chk_root_key)
827
chk_bytes = self.chk_bytes
828
for record, items in chk_map.iter_interesting_nodes(chk_bytes,
829
interesting_root_keys, uninteresting_root_keys,
831
for name, bytes in items:
832
(name_utf8, file_id, revision_id) = bytes_to_info(bytes)
833
if not rich_root and name_utf8 == '':
836
file_id_revisions[file_id].add(revision_id)
838
file_id_revisions[file_id] = set([revision_id])
841
return file_id_revisions
843
def find_text_key_references(self):
844
"""Find the text key references within the repository.
846
:return: A dictionary mapping text keys ((fileid, revision_id) tuples)
847
to whether they were referred to by the inventory of the
848
revision_id that they contain. The inventory texts from all present
849
revision ids are assessed to generate this report.
851
# XXX: Slow version but correct: rewrite as a series of delta
852
# examinations/direct tree traversal. Note that that will require care
853
# as a common node is reachable both from the inventory that added it,
854
# and others afterwards.
855
revision_keys = self.revisions.keys()
857
rich_roots = self.supports_rich_root()
858
pb = ui.ui_factory.nested_progress_bar()
860
all_revs = self.all_revision_ids()
861
total = len(all_revs)
862
for pos, inv in enumerate(self.iter_inventories(all_revs)):
863
pb.update("Finding text references", pos, total)
864
for _, entry in inv.iter_entries():
865
if not rich_roots and entry.file_id == inv.root_id:
867
key = (entry.file_id, entry.revision)
868
result.setdefault(key, False)
869
if entry.revision == inv.revision_id:
875
def _reconcile_pack(self, collection, packs, extension, revs, pb):
876
packer = GCCHKReconcilePacker(collection, packs, extension)
877
return packer.pack(pb)
879
def _get_source(self, to_format):
880
"""Return a source for streaming from this repository."""
881
if isinstance(to_format, remote.RemoteRepositoryFormat):
882
# Can't just check attributes on to_format with the current code,
884
to_format._ensure_real()
885
to_format = to_format._custom_format
886
if to_format.__class__ is self._format.__class__:
887
# We must be exactly the same format, otherwise stuff like the chk
888
# page layout might be different
889
return GroupCHKStreamSource(self, to_format)
890
return super(CHKInventoryRepository, self)._get_source(to_format)
893
class GroupCHKStreamSource(KnitPackStreamSource):
894
"""Used when both the source and target repo are GroupCHK repos."""
896
def __init__(self, from_repository, to_format):
897
"""Create a StreamSource streaming from from_repository."""
898
super(GroupCHKStreamSource, self).__init__(from_repository, to_format)
899
self._revision_keys = None
900
self._text_keys = None
901
self._text_fetch_order = 'groupcompress'
902
self._chk_id_roots = None
903
self._chk_p_id_roots = None
905
def _get_inventory_stream(self, inventory_keys, allow_absent=False):
906
"""Get a stream of inventory texts.
908
When this function returns, self._chk_id_roots and self._chk_p_id_roots
911
self._chk_id_roots = []
912
self._chk_p_id_roots = []
913
def _filtered_inv_stream():
915
p_id_roots_set = set()
916
source_vf = self.from_repository.inventories
917
stream = source_vf.get_record_stream(inventory_keys,
918
'groupcompress', True)
919
for record in stream:
920
if record.storage_kind == 'absent':
924
raise errors.NoSuchRevision(self, record.key)
925
bytes = record.get_bytes_as('fulltext')
926
chk_inv = inventory.CHKInventory.deserialise(None, bytes,
928
key = chk_inv.id_to_entry.key()
929
if key not in id_roots_set:
930
self._chk_id_roots.append(key)
931
id_roots_set.add(key)
932
p_id_map = chk_inv.parent_id_basename_to_file_id
934
raise AssertionError('Parent id -> file_id map not set')
936
if key not in p_id_roots_set:
937
p_id_roots_set.add(key)
938
self._chk_p_id_roots.append(key)
940
# We have finished processing all of the inventory records, we
941
# don't need these sets anymore
943
p_id_roots_set.clear()
944
return ('inventories', _filtered_inv_stream())
946
def _get_filtered_chk_streams(self, excluded_revision_keys):
947
self._text_keys = set()
948
excluded_revision_keys.discard(_mod_revision.NULL_REVISION)
949
if not excluded_revision_keys:
950
uninteresting_root_keys = set()
951
uninteresting_pid_root_keys = set()
953
# filter out any excluded revisions whose inventories are not
955
# TODO: Update Repository.iter_inventories() to add
956
# ignore_missing=True
957
present_keys = self.from_repository._find_present_inventory_keys(
958
excluded_revision_keys)
959
present_ids = [k[-1] for k in present_keys]
960
uninteresting_root_keys = set()
961
uninteresting_pid_root_keys = set()
962
for inv in self.from_repository.iter_inventories(present_ids):
963
uninteresting_root_keys.add(inv.id_to_entry.key())
964
uninteresting_pid_root_keys.add(
965
inv.parent_id_basename_to_file_id.key())
966
bytes_to_info = inventory.CHKInventory._bytes_to_utf8name_key
967
chk_bytes = self.from_repository.chk_bytes
968
def _filter_id_to_entry():
969
for record, items in chk_map.iter_interesting_nodes(chk_bytes,
970
self._chk_id_roots, uninteresting_root_keys):
971
for name, bytes in items:
972
# Note: we don't care about name_utf8, because we are always
974
_, file_id, revision_id = bytes_to_info(bytes)
975
self._text_keys.add((file_id, revision_id))
976
if record is not None:
979
self._chk_id_roots = None
980
yield 'chk_bytes', _filter_id_to_entry()
981
def _get_parent_id_basename_to_file_id_pages():
982
for record, items in chk_map.iter_interesting_nodes(chk_bytes,
983
self._chk_p_id_roots, uninteresting_pid_root_keys):
984
if record is not None:
987
self._chk_p_id_roots = None
988
yield 'chk_bytes', _get_parent_id_basename_to_file_id_pages()
990
def get_stream(self, search):
991
revision_ids = search.get_keys()
992
for stream_info in self._fetch_revision_texts(revision_ids):
994
self._revision_keys = [(rev_id,) for rev_id in revision_ids]
995
yield self._get_inventory_stream(self._revision_keys)
996
# TODO: The keys to exclude might be part of the search recipe
997
# For now, exclude all parents that are at the edge of ancestry, for
998
# which we have inventories
999
from_repo = self.from_repository
1000
parent_keys = from_repo._find_parent_keys_of_revisions(
1001
self._revision_keys)
1002
for stream_info in self._get_filtered_chk_streams(parent_keys):
1004
yield self._get_text_stream()
1006
def get_stream_for_missing_keys(self, missing_keys):
1007
# missing keys can only occur when we are byte copying and not
1008
# translating (because translation means we don't send
1009
# unreconstructable deltas ever).
1010
missing_inventory_keys = set()
1011
for key in missing_keys:
1012
if key[0] != 'inventories':
1013
raise AssertionError('The only missing keys we should'
1014
' be filling in are inventory keys, not %s'
1016
missing_inventory_keys.add(key[1:])
1017
if self._chk_id_roots or self._chk_p_id_roots:
1018
raise AssertionError('Cannot call get_stream_for_missing_keys'
1019
' untill all of get_stream() has been consumed.')
1020
# Yield the inventory stream, so we can find the chk stream
1021
# Some of the missing_keys will be missing because they are ghosts.
1022
# As such, we can ignore them. The Sink is required to verify there are
1023
# no unavailable texts when the ghost inventories are not filled in.
1024
yield self._get_inventory_stream(missing_inventory_keys,
1026
# We use the empty set for excluded_revision_keys, to make it clear
1027
# that we want to transmit all referenced chk pages.
1028
for stream_info in self._get_filtered_chk_streams(set()):
1032
class RepositoryFormatCHK1(RepositoryFormatPack):
1033
"""A hashed CHK+group compress pack repository."""
1035
repository_class = CHKInventoryRepository
1036
supports_external_lookups = True
1037
supports_chks = True
1038
# For right now, setting this to True gives us InterModel1And2 rather
1039
# than InterDifferingSerializer
1040
_commit_builder_class = PackRootCommitBuilder
1041
rich_root_data = True
1042
_serializer = chk_serializer.chk_serializer_255_bigpage
1043
_commit_inv_deltas = True
1044
# What index classes to use
1045
index_builder_class = BTreeBuilder
1046
index_class = BTreeGraphIndex
1047
# Note: We cannot unpack a delta that references a text we haven't
1048
# seen yet. There are 2 options, work in fulltexts, or require
1049
# topological sorting. Using fulltexts is more optimal for local
1050
# operations, because the source can be smart about extracting
1051
# multiple in-a-row (and sharing strings). Topological is better
1052
# for remote, because we access less data.
1053
_fetch_order = 'unordered'
1054
_fetch_uses_deltas = False # essentially ignored by the groupcompress code.
1056
pack_compresses = True
1058
def _get_matching_bzrdir(self):
1059
return bzrdir.format_registry.make_bzrdir('development6-rich-root')
1061
def _ignore_setting_bzrdir(self, format):
1064
_matchingbzrdir = property(_get_matching_bzrdir, _ignore_setting_bzrdir)
1066
def get_format_string(self):
1067
"""See RepositoryFormat.get_format_string()."""
1068
return ('Bazaar development format - group compression and chk inventory'
1069
' (needs bzr.dev from 1.14)\n')
1071
def get_format_description(self):
1072
"""See RepositoryFormat.get_format_description()."""
1073
return ("Development repository format - rich roots, group compression"
1074
" and chk inventories")
1076
def check_conversion_target(self, target_format):
1077
if not target_format.rich_root_data:
1078
raise errors.BadConversionTarget(
1079
'Does not support rich root data.', target_format)
1080
if (self.supports_tree_reference and
1081
not getattr(target_format, 'supports_tree_reference', False)):
1082
raise errors.BadConversionTarget(
1083
'Does not support nested trees', target_format)
1087
class RepositoryFormatCHK2(RepositoryFormatCHK1):
1088
"""A CHK repository that uses the bencode revision serializer."""
1090
_serializer = chk_serializer.chk_bencode_serializer
1092
def _get_matching_bzrdir(self):
1093
return bzrdir.format_registry.make_bzrdir('development7-rich-root')
1095
def _ignore_setting_bzrdir(self, format):
1098
_matchingbzrdir = property(_get_matching_bzrdir, _ignore_setting_bzrdir)
1100
def get_format_string(self):
1101
"""See RepositoryFormat.get_format_string()."""
1102
return ('Bazaar development format - chk repository with bencode '
1103
'revision serialization (needs bzr.dev from 1.16)\n')
1106
class RepositoryFormat2a(RepositoryFormatCHK2):
1107
"""A CHK repository that uses the bencode revision serializer.
1109
This is the same as RepositoryFormatCHK2 but with a public name.
1112
_serializer = chk_serializer.chk_bencode_serializer
1114
def _get_matching_bzrdir(self):
1115
return bzrdir.format_registry.make_bzrdir('2a')
1117
def _ignore_setting_bzrdir(self, format):
1120
_matchingbzrdir = property(_get_matching_bzrdir, _ignore_setting_bzrdir)
1122
def get_format_string(self):
1123
return ('Bazaar repository format 2a (needs bzr 1.16 or later)\n')
added
removed
bzrlib/repofmt/groupcompress_repo.py
