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
bytes = record.get_bytes_as('fulltext')
222
chk_inv = inventory.CHKInventory.deserialise(None, bytes,
225
pb.update('inv', idx, total_keys)
226
key = chk_inv.id_to_entry.key()
227
if key not in id_roots_set:
228
self._chk_id_roots.append(key)
229
id_roots_set.add(key)
230
p_id_map = chk_inv.parent_id_basename_to_file_id
232
raise AssertionError('Parent id -> file_id map not set')
234
if key not in p_id_roots_set:
235
p_id_roots_set.add(key)
236
self._chk_p_id_roots.append(key)
238
# We have finished processing all of the inventory records, we
239
# don't need these sets anymore
241
p_id_roots_set.clear()
242
return _filtered_inv_stream()
244
def _get_chk_streams(self, source_vf, keys, pb=None):
245
# We want to stream the keys from 'id_roots', and things they
246
# reference, and then stream things from p_id_roots and things they
247
# reference, and then any remaining keys that we didn't get to.
249
# We also group referenced texts together, so if one root references a
250
# text with prefix 'a', and another root references a node with prefix
251
# 'a', we want to yield those nodes before we yield the nodes for 'b'
252
# This keeps 'similar' nodes together.
254
# Note: We probably actually want multiple streams here, to help the
255
# client understand that the different levels won't compress well
256
# against each other.
257
# Test the difference between using one Group per level, and
258
# using 1 Group per prefix. (so '' (root) would get a group, then
259
# all the references to search-key 'a' would get a group, etc.)
260
total_keys = len(keys)
261
remaining_keys = set(keys)
263
if self._gather_text_refs:
264
bytes_to_info = inventory.CHKInventory._bytes_to_utf8name_key
265
self._text_refs = set()
266
def _get_referenced_stream(root_keys, parse_leaf_nodes=False):
269
keys_by_search_prefix = {}
270
remaining_keys.difference_update(cur_keys)
272
def handle_internal_node(node):
273
for prefix, value in node._items.iteritems():
274
# We don't want to request the same key twice, and we
275
# want to order it by the first time it is seen.
276
# Even further, we don't want to request a key which is
277
# not in this group of pack files (it should be in the
278
# repo, but it doesn't have to be in the group being
280
# TODO: consider how to treat externally referenced chk
281
# pages as 'external_references' so that we
282
# always fill them in for stacked branches
283
if value not in next_keys and value in remaining_keys:
284
keys_by_search_prefix.setdefault(prefix,
287
def handle_leaf_node(node):
288
# Store is None, because we know we have a LeafNode, and we
289
# just want its entries
290
for file_id, bytes in node.iteritems(None):
291
name_utf8, file_id, revision_id = bytes_to_info(bytes)
292
self._text_refs.add((file_id, revision_id))
294
stream = source_vf.get_record_stream(cur_keys,
295
'as-requested', True)
296
for record in stream:
297
bytes = record.get_bytes_as('fulltext')
298
# We don't care about search_key_func for this code,
299
# because we only care about external references.
300
node = chk_map._deserialise(bytes, record.key,
301
search_key_func=None)
302
common_base = node._search_prefix
303
if isinstance(node, chk_map.InternalNode):
304
handle_internal_node(node)
305
elif parse_leaf_nodes:
306
handle_leaf_node(node)
309
pb.update('chk node', counter[0], total_keys)
312
# Double check that we won't be emitting any keys twice
313
# If we get rid of the pre-calculation of all keys, we could
314
# turn this around and do
315
# next_keys.difference_update(seen_keys)
316
# However, we also may have references to chk pages in another
317
# pack file during autopack. We filter earlier, so we should no
318
# longer need to do this
319
# next_keys = next_keys.intersection(remaining_keys)
321
for prefix in sorted(keys_by_search_prefix):
322
cur_keys.extend(keys_by_search_prefix.pop(prefix))
323
for stream in _get_referenced_stream(self._chk_id_roots,
324
self._gather_text_refs):
326
del self._chk_id_roots
327
# while it isn't really possible for chk_id_roots to not be in the
328
# local group of packs, it is possible that the tree shape has not
329
# changed recently, so we need to filter _chk_p_id_roots by the
331
chk_p_id_roots = [key for key in self._chk_p_id_roots
332
if key in remaining_keys]
333
del self._chk_p_id_roots
334
for stream in _get_referenced_stream(chk_p_id_roots, False):
337
trace.mutter('There were %d keys in the chk index, %d of which'
338
' were not referenced', total_keys,
340
if self.revision_ids is None:
341
stream = source_vf.get_record_stream(remaining_keys,
345
def _build_vf(self, index_name, parents, delta, for_write=False):
346
"""Build a VersionedFiles instance on top of this group of packs."""
347
index_name = index_name + '_index'
349
access = knit._DirectPackAccess(index_to_pack)
352
if self.new_pack is None:
353
raise AssertionError('No new pack has been set')
354
index = getattr(self.new_pack, index_name)
355
index_to_pack[index] = self.new_pack.access_tuple()
356
index.set_optimize(for_size=True)
357
access.set_writer(self.new_pack._writer, index,
358
self.new_pack.access_tuple())
359
add_callback = index.add_nodes
362
for pack in self.packs:
363
sub_index = getattr(pack, index_name)
364
index_to_pack[sub_index] = pack.access_tuple()
365
indices.append(sub_index)
366
index = _mod_index.CombinedGraphIndex(indices)
368
vf = GroupCompressVersionedFiles(
370
add_callback=add_callback,
372
is_locked=self._pack_collection.repo.is_locked),
377
def _build_vfs(self, index_name, parents, delta):
378
"""Build the source and target VersionedFiles."""
379
source_vf = self._build_vf(index_name, parents,
380
delta, for_write=False)
381
target_vf = self._build_vf(index_name, parents,
382
delta, for_write=True)
383
return source_vf, target_vf
385
def _copy_stream(self, source_vf, target_vf, keys, message, vf_to_stream,
387
trace.mutter('repacking %d %s', len(keys), message)
388
self.pb.update('repacking %s' % (message,), pb_offset)
389
child_pb = ui.ui_factory.nested_progress_bar()
391
stream = vf_to_stream(source_vf, keys, message, child_pb)
392
for _ in target_vf._insert_record_stream(stream,
399
def _copy_revision_texts(self):
400
source_vf, target_vf = self._build_vfs('revision', True, False)
401
if not self.revision_keys:
402
# We are doing a full fetch, aka 'pack'
403
self.revision_keys = source_vf.keys()
404
self._copy_stream(source_vf, target_vf, self.revision_keys,
405
'revisions', self._get_progress_stream, 1)
407
def _copy_inventory_texts(self):
408
source_vf, target_vf = self._build_vfs('inventory', True, True)
409
self._copy_stream(source_vf, target_vf, self.revision_keys,
410
'inventories', self._get_filtered_inv_stream, 2)
412
def _copy_chk_texts(self):
413
source_vf, target_vf = self._build_vfs('chk', False, False)
414
# TODO: This is technically spurious... if it is a performance issue,
416
total_keys = source_vf.keys()
417
trace.mutter('repacking chk: %d id_to_entry roots,'
418
' %d p_id_map roots, %d total keys',
419
len(self._chk_id_roots), len(self._chk_p_id_roots),
421
self.pb.update('repacking chk', 3)
422
child_pb = ui.ui_factory.nested_progress_bar()
424
for stream in self._get_chk_streams(source_vf, total_keys,
426
for _ in target_vf._insert_record_stream(stream,
433
def _copy_text_texts(self):
434
source_vf, target_vf = self._build_vfs('text', True, True)
435
# XXX: We don't walk the chk map to determine referenced (file_id,
436
# revision_id) keys. We don't do it yet because you really need
437
# to filter out the ones that are present in the parents of the
438
# rev just before the ones you are copying, otherwise the filter
439
# is grabbing too many keys...
440
text_keys = source_vf.keys()
441
self._copy_stream(source_vf, target_vf, text_keys,
442
'text', self._get_progress_stream, 4)
444
def _copy_signature_texts(self):
445
source_vf, target_vf = self._build_vfs('signature', False, False)
446
signature_keys = source_vf.keys()
447
signature_keys.intersection(self.revision_keys)
448
self._copy_stream(source_vf, target_vf, signature_keys,
449
'signatures', self._get_progress_stream, 5)
451
def _create_pack_from_packs(self):
452
self.pb.update('repacking', 0, 7)
453
self.new_pack = self.open_pack()
454
# Is this necessary for GC ?
455
self.new_pack.set_write_cache_size(1024*1024)
456
self._copy_revision_texts()
457
self._copy_inventory_texts()
458
self._copy_chk_texts()
459
self._copy_text_texts()
460
self._copy_signature_texts()
461
self.new_pack._check_references()
462
if not self._use_pack(self.new_pack):
463
self.new_pack.abort()
465
self.pb.update('finishing repack', 6, 7)
466
self.new_pack.finish()
467
self._pack_collection.allocate(self.new_pack)
471
class GCCHKReconcilePacker(GCCHKPacker):
472
"""A packer which regenerates indices etc as it copies.
474
This is used by ``bzr reconcile`` to cause parent text pointers to be
478
def __init__(self, *args, **kwargs):
479
super(GCCHKReconcilePacker, self).__init__(*args, **kwargs)
480
self._data_changed = False
481
self._gather_text_refs = True
483
def _copy_inventory_texts(self):
484
source_vf, target_vf = self._build_vfs('inventory', True, True)
485
self._copy_stream(source_vf, target_vf, self.revision_keys,
486
'inventories', self._get_filtered_inv_stream, 2)
487
if source_vf.keys() != self.revision_keys:
488
self._data_changed = True
490
def _copy_text_texts(self):
491
"""generate what texts we should have and then copy."""
492
source_vf, target_vf = self._build_vfs('text', True, True)
493
trace.mutter('repacking %d texts', len(self._text_refs))
494
self.pb.update("repacking texts", 4)
495
# we have three major tasks here:
496
# 1) generate the ideal index
497
repo = self._pack_collection.repo
498
# We want the one we just wrote, so base it on self.new_pack
499
revision_vf = self._build_vf('revision', True, False, for_write=True)
500
ancestor_keys = revision_vf.get_parent_map(revision_vf.keys())
501
# Strip keys back into revision_ids.
502
ancestors = dict((k[0], tuple([p[0] for p in parents]))
503
for k, parents in ancestor_keys.iteritems())
505
# TODO: _generate_text_key_index should be much cheaper to generate from
506
# a chk repository, rather than the current implementation
507
ideal_index = repo._generate_text_key_index(None, ancestors)
508
file_id_parent_map = source_vf.get_parent_map(self._text_refs)
509
# 2) generate a keys list that contains all the entries that can
510
# be used as-is, with corrected parents.
512
new_parent_keys = {} # (key, parent_keys)
514
NULL_REVISION = _mod_revision.NULL_REVISION
515
for key in self._text_refs:
521
ideal_parents = tuple(ideal_index[key])
523
discarded_keys.append(key)
524
self._data_changed = True
526
if ideal_parents == (NULL_REVISION,):
528
source_parents = file_id_parent_map[key]
529
if ideal_parents == source_parents:
533
# We need to change the parent graph, but we don't need to
534
# re-insert the text (since we don't pun the compression
535
# parent with the parents list)
536
self._data_changed = True
537
new_parent_keys[key] = ideal_parents
538
# we're finished with some data.
540
del file_id_parent_map
541
# 3) bulk copy the data, updating records than need it
542
def _update_parents_for_texts():
543
stream = source_vf.get_record_stream(self._text_refs,
544
'groupcompress', False)
545
for record in stream:
546
if record.key in new_parent_keys:
547
record.parents = new_parent_keys[record.key]
549
target_vf.insert_record_stream(_update_parents_for_texts())
551
def _use_pack(self, new_pack):
552
"""Override _use_pack to check for reconcile having changed content."""
553
return new_pack.data_inserted() and self._data_changed
556
class GCRepositoryPackCollection(RepositoryPackCollection):
558
pack_factory = GCPack
559
resumed_pack_factory = ResumedGCPack
561
def _already_packed(self):
562
"""Is the collection already packed?"""
563
# Always repack GC repositories for now
566
def _execute_pack_operations(self, pack_operations,
567
_packer_class=GCCHKPacker,
569
"""Execute a series of pack operations.
571
:param pack_operations: A list of [revision_count, packs_to_combine].
572
:param _packer_class: The class of packer to use (default: Packer).
575
# XXX: Copied across from RepositoryPackCollection simply because we
576
# want to override the _packer_class ... :(
577
for revision_count, packs in pack_operations:
578
# we may have no-ops from the setup logic
581
packer = GCCHKPacker(self, packs, '.autopack',
582
reload_func=reload_func)
585
except errors.RetryWithNewPacks:
586
# An exception is propagating out of this context, make sure
587
# this packer has cleaned up. Packer() doesn't set its new_pack
588
# state into the RepositoryPackCollection object, so we only
589
# have access to it directly here.
590
if packer.new_pack is not None:
591
packer.new_pack.abort()
594
self._remove_pack_from_memory(pack)
595
# record the newly available packs and stop advertising the old
597
self._save_pack_names(clear_obsolete_packs=True)
598
# Move the old packs out of the way now they are no longer referenced.
599
for revision_count, packs in pack_operations:
600
self._obsolete_packs(packs)
603
class CHKInventoryRepository(KnitPackRepository):
604
"""subclass of KnitPackRepository that uses CHK based inventories."""
606
def __init__(self, _format, a_bzrdir, control_files, _commit_builder_class,
608
"""Overridden to change pack collection class."""
609
KnitPackRepository.__init__(self, _format, a_bzrdir, control_files,
610
_commit_builder_class, _serializer)
611
# and now replace everything it did :)
612
index_transport = self._transport.clone('indices')
613
self._pack_collection = GCRepositoryPackCollection(self,
614
self._transport, index_transport,
615
self._transport.clone('upload'),
616
self._transport.clone('packs'),
617
_format.index_builder_class,
619
use_chk_index=self._format.supports_chks,
621
self.inventories = GroupCompressVersionedFiles(
622
_GCGraphIndex(self._pack_collection.inventory_index.combined_index,
623
add_callback=self._pack_collection.inventory_index.add_callback,
624
parents=True, is_locked=self.is_locked,
625
inconsistency_fatal=False),
626
access=self._pack_collection.inventory_index.data_access)
627
self.revisions = GroupCompressVersionedFiles(
628
_GCGraphIndex(self._pack_collection.revision_index.combined_index,
629
add_callback=self._pack_collection.revision_index.add_callback,
630
parents=True, is_locked=self.is_locked,
631
track_external_parent_refs=True),
632
access=self._pack_collection.revision_index.data_access,
634
self.signatures = GroupCompressVersionedFiles(
635
_GCGraphIndex(self._pack_collection.signature_index.combined_index,
636
add_callback=self._pack_collection.signature_index.add_callback,
637
parents=False, is_locked=self.is_locked,
638
inconsistency_fatal=False),
639
access=self._pack_collection.signature_index.data_access,
641
self.texts = GroupCompressVersionedFiles(
642
_GCGraphIndex(self._pack_collection.text_index.combined_index,
643
add_callback=self._pack_collection.text_index.add_callback,
644
parents=True, is_locked=self.is_locked,
645
inconsistency_fatal=False),
646
access=self._pack_collection.text_index.data_access)
647
# No parents, individual CHK pages don't have specific ancestry
648
self.chk_bytes = GroupCompressVersionedFiles(
649
_GCGraphIndex(self._pack_collection.chk_index.combined_index,
650
add_callback=self._pack_collection.chk_index.add_callback,
651
parents=False, is_locked=self.is_locked,
652
inconsistency_fatal=False),
653
access=self._pack_collection.chk_index.data_access)
654
# True when the repository object is 'write locked' (as opposed to the
655
# physical lock only taken out around changes to the pack-names list.)
656
# Another way to represent this would be a decorator around the control
657
# files object that presents logical locks as physical ones - if this
658
# gets ugly consider that alternative design. RBC 20071011
659
self._write_lock_count = 0
660
self._transaction = None
662
self._reconcile_does_inventory_gc = True
663
self._reconcile_fixes_text_parents = True
664
self._reconcile_backsup_inventory = False
666
def _add_inventory_checked(self, revision_id, inv, parents):
667
"""Add inv to the repository after checking the inputs.
669
This function can be overridden to allow different inventory styles.
671
:seealso: add_inventory, for the contract.
674
serializer = self._format._serializer
675
result = inventory.CHKInventory.from_inventory(self.chk_bytes, inv,
676
maximum_size=serializer.maximum_size,
677
search_key_name=serializer.search_key_name)
678
inv_lines = result.to_lines()
679
return self._inventory_add_lines(revision_id, parents,
680
inv_lines, check_content=False)
682
def _create_inv_from_null(self, delta, revision_id):
683
"""This will mutate new_inv directly.
685
This is a simplified form of create_by_apply_delta which knows that all
686
the old values must be None, so everything is a create.
688
serializer = self._format._serializer
689
new_inv = inventory.CHKInventory(serializer.search_key_name)
690
new_inv.revision_id = revision_id
691
entry_to_bytes = new_inv._entry_to_bytes
692
id_to_entry_dict = {}
693
parent_id_basename_dict = {}
694
for old_path, new_path, file_id, entry in delta:
695
if old_path is not None:
696
raise ValueError('Invalid delta, somebody tried to delete %r'
697
' from the NULL_REVISION'
698
% ((old_path, file_id),))
700
raise ValueError('Invalid delta, delta from NULL_REVISION has'
701
' no new_path %r' % (file_id,))
703
new_inv.root_id = file_id
704
parent_id_basename_key = ('', '')
706
utf8_entry_name = entry.name.encode('utf-8')
707
parent_id_basename_key = (entry.parent_id, utf8_entry_name)
708
new_value = entry_to_bytes(entry)
710
# new_inv._path_to_fileid_cache[new_path] = file_id
711
id_to_entry_dict[(file_id,)] = new_value
712
parent_id_basename_dict[parent_id_basename_key] = file_id
714
new_inv._populate_from_dicts(self.chk_bytes, id_to_entry_dict,
715
parent_id_basename_dict, maximum_size=serializer.maximum_size)
718
def add_inventory_by_delta(self, basis_revision_id, delta, new_revision_id,
719
parents, basis_inv=None, propagate_caches=False):
720
"""Add a new inventory expressed as a delta against another revision.
722
:param basis_revision_id: The inventory id the delta was created
724
:param delta: The inventory delta (see Inventory.apply_delta for
726
:param new_revision_id: The revision id that the inventory is being
728
:param parents: The revision ids of the parents that revision_id is
729
known to have and are in the repository already. These are supplied
730
for repositories that depend on the inventory graph for revision
731
graph access, as well as for those that pun ancestry with delta
733
:param basis_inv: The basis inventory if it is already known,
735
:param propagate_caches: If True, the caches for this inventory are
736
copied to and updated for the result if possible.
738
:returns: (validator, new_inv)
739
The validator(which is a sha1 digest, though what is sha'd is
740
repository format specific) of the serialized inventory, and the
743
if not self.is_in_write_group():
744
raise AssertionError("%r not in write group" % (self,))
745
_mod_revision.check_not_reserved_id(new_revision_id)
747
if basis_inv is None:
748
if basis_revision_id == _mod_revision.NULL_REVISION:
749
new_inv = self._create_inv_from_null(delta, new_revision_id)
750
inv_lines = new_inv.to_lines()
751
return self._inventory_add_lines(new_revision_id, parents,
752
inv_lines, check_content=False), new_inv
754
basis_tree = self.revision_tree(basis_revision_id)
755
basis_tree.lock_read()
756
basis_inv = basis_tree.inventory
758
result = basis_inv.create_by_apply_delta(delta, new_revision_id,
759
propagate_caches=propagate_caches)
760
inv_lines = result.to_lines()
761
return self._inventory_add_lines(new_revision_id, parents,
762
inv_lines, check_content=False), result
764
if basis_tree is not None:
767
def _iter_inventories(self, revision_ids):
768
"""Iterate over many inventory objects."""
769
keys = [(revision_id,) for revision_id in revision_ids]
770
stream = self.inventories.get_record_stream(keys, 'unordered', True)
772
for record in stream:
773
if record.storage_kind != 'absent':
774
texts[record.key] = record.get_bytes_as('fulltext')
776
raise errors.NoSuchRevision(self, record.key)
778
yield inventory.CHKInventory.deserialise(self.chk_bytes, texts[key], key)
780
def _iter_inventory_xmls(self, revision_ids):
781
# Without a native 'xml' inventory, this method doesn't make sense, so
782
# make it raise to trap naughty direct users.
783
raise NotImplementedError(self._iter_inventory_xmls)
785
def _find_present_inventory_keys(self, revision_keys):
786
parent_map = self.inventories.get_parent_map(revision_keys)
787
present_inventory_keys = set(k for k in parent_map)
788
return present_inventory_keys
790
def fileids_altered_by_revision_ids(self, revision_ids, _inv_weave=None):
791
"""Find the file ids and versions affected by revisions.
793
:param revisions: an iterable containing revision ids.
794
:param _inv_weave: The inventory weave from this repository or None.
795
If None, the inventory weave will be opened automatically.
796
:return: a dictionary mapping altered file-ids to an iterable of
797
revision_ids. Each altered file-ids has the exact revision_ids that
798
altered it listed explicitly.
800
rich_root = self.supports_rich_root()
801
bytes_to_info = inventory.CHKInventory._bytes_to_utf8name_key
802
file_id_revisions = {}
803
pb = ui.ui_factory.nested_progress_bar()
805
revision_keys = [(r,) for r in revision_ids]
806
parent_keys = self._find_parent_keys_of_revisions(revision_keys)
807
# TODO: instead of using _find_present_inventory_keys, change the
808
# code paths to allow missing inventories to be tolerated.
809
# However, we only want to tolerate missing parent
810
# inventories, not missing inventories for revision_ids
811
present_parent_inv_keys = self._find_present_inventory_keys(
813
present_parent_inv_ids = set(
814
[k[-1] for k in present_parent_inv_keys])
815
uninteresting_root_keys = set()
816
interesting_root_keys = set()
817
inventories_to_read = set(revision_ids)
818
inventories_to_read.update(present_parent_inv_ids)
819
for inv in self.iter_inventories(inventories_to_read):
820
entry_chk_root_key = inv.id_to_entry.key()
821
if inv.revision_id in present_parent_inv_ids:
822
uninteresting_root_keys.add(entry_chk_root_key)
824
interesting_root_keys.add(entry_chk_root_key)
826
chk_bytes = self.chk_bytes
827
for record, items in chk_map.iter_interesting_nodes(chk_bytes,
828
interesting_root_keys, uninteresting_root_keys,
830
for name, bytes in items:
831
(name_utf8, file_id, revision_id) = bytes_to_info(bytes)
832
if not rich_root and name_utf8 == '':
835
file_id_revisions[file_id].add(revision_id)
837
file_id_revisions[file_id] = set([revision_id])
840
return file_id_revisions
842
def find_text_key_references(self):
843
"""Find the text key references within the repository.
845
:return: A dictionary mapping text keys ((fileid, revision_id) tuples)
846
to whether they were referred to by the inventory of the
847
revision_id that they contain. The inventory texts from all present
848
revision ids are assessed to generate this report.
850
# XXX: Slow version but correct: rewrite as a series of delta
851
# examinations/direct tree traversal. Note that that will require care
852
# as a common node is reachable both from the inventory that added it,
853
# and others afterwards.
854
revision_keys = self.revisions.keys()
856
rich_roots = self.supports_rich_root()
857
pb = ui.ui_factory.nested_progress_bar()
859
all_revs = self.all_revision_ids()
860
total = len(all_revs)
861
for pos, inv in enumerate(self.iter_inventories(all_revs)):
862
pb.update("Finding text references", pos, total)
863
for _, entry in inv.iter_entries():
864
if not rich_roots and entry.file_id == inv.root_id:
866
key = (entry.file_id, entry.revision)
867
result.setdefault(key, False)
868
if entry.revision == inv.revision_id:
874
def _reconcile_pack(self, collection, packs, extension, revs, pb):
875
packer = GCCHKReconcilePacker(collection, packs, extension)
876
return packer.pack(pb)
878
def _get_source(self, to_format):
879
"""Return a source for streaming from this repository."""
880
if isinstance(to_format, remote.RemoteRepositoryFormat):
881
# Can't just check attributes on to_format with the current code,
883
to_format._ensure_real()
884
to_format = to_format._custom_format
885
if to_format.__class__ is self._format.__class__:
886
# We must be exactly the same format, otherwise stuff like the chk
887
# page layout might be different
888
return GroupCHKStreamSource(self, to_format)
889
return super(CHKInventoryRepository, self)._get_source(to_format)
892
class GroupCHKStreamSource(KnitPackStreamSource):
893
"""Used when both the source and target repo are GroupCHK repos."""
895
def __init__(self, from_repository, to_format):
896
"""Create a StreamSource streaming from from_repository."""
897
super(GroupCHKStreamSource, self).__init__(from_repository, to_format)
898
self._revision_keys = None
899
self._text_keys = None
900
self._text_fetch_order = 'groupcompress'
901
self._chk_id_roots = None
902
self._chk_p_id_roots = None
904
def _get_inventory_stream(self, inventory_keys, allow_absent=False):
905
"""Get a stream of inventory texts.
907
When this function returns, self._chk_id_roots and self._chk_p_id_roots
910
self._chk_id_roots = []
911
self._chk_p_id_roots = []
912
def _filtered_inv_stream():
914
p_id_roots_set = set()
915
source_vf = self.from_repository.inventories
916
stream = source_vf.get_record_stream(inventory_keys,
917
'groupcompress', True)
918
for record in stream:
919
if record.storage_kind == 'absent':
923
raise errors.NoSuchRevision(self, record.key)
924
bytes = record.get_bytes_as('fulltext')
925
chk_inv = inventory.CHKInventory.deserialise(None, bytes,
927
key = chk_inv.id_to_entry.key()
928
if key not in id_roots_set:
929
self._chk_id_roots.append(key)
930
id_roots_set.add(key)
931
p_id_map = chk_inv.parent_id_basename_to_file_id
933
raise AssertionError('Parent id -> file_id map not set')
935
if key not in p_id_roots_set:
936
p_id_roots_set.add(key)
937
self._chk_p_id_roots.append(key)
939
# We have finished processing all of the inventory records, we
940
# don't need these sets anymore
942
p_id_roots_set.clear()
943
return ('inventories', _filtered_inv_stream())
945
def _get_filtered_chk_streams(self, excluded_revision_keys):
946
self._text_keys = set()
947
excluded_revision_keys.discard(_mod_revision.NULL_REVISION)
948
if not excluded_revision_keys:
949
uninteresting_root_keys = set()
950
uninteresting_pid_root_keys = set()
952
# filter out any excluded revisions whose inventories are not
954
# TODO: Update Repository.iter_inventories() to add
955
# ignore_missing=True
956
present_keys = self.from_repository._find_present_inventory_keys(
957
excluded_revision_keys)
958
present_ids = [k[-1] for k in present_keys]
959
uninteresting_root_keys = set()
960
uninteresting_pid_root_keys = set()
961
for inv in self.from_repository.iter_inventories(present_ids):
962
uninteresting_root_keys.add(inv.id_to_entry.key())
963
uninteresting_pid_root_keys.add(
964
inv.parent_id_basename_to_file_id.key())
965
bytes_to_info = inventory.CHKInventory._bytes_to_utf8name_key
966
chk_bytes = self.from_repository.chk_bytes
967
def _filter_id_to_entry():
968
for record, items in chk_map.iter_interesting_nodes(chk_bytes,
969
self._chk_id_roots, uninteresting_root_keys):
970
for name, bytes in items:
971
# Note: we don't care about name_utf8, because we are always
973
_, file_id, revision_id = bytes_to_info(bytes)
974
self._text_keys.add((file_id, revision_id))
975
if record is not None:
978
self._chk_id_roots = None
979
yield 'chk_bytes', _filter_id_to_entry()
980
def _get_parent_id_basename_to_file_id_pages():
981
for record, items in chk_map.iter_interesting_nodes(chk_bytes,
982
self._chk_p_id_roots, uninteresting_pid_root_keys):
983
if record is not None:
986
self._chk_p_id_roots = None
987
yield 'chk_bytes', _get_parent_id_basename_to_file_id_pages()
989
def get_stream(self, search):
990
revision_ids = search.get_keys()
991
for stream_info in self._fetch_revision_texts(revision_ids):
993
self._revision_keys = [(rev_id,) for rev_id in revision_ids]
994
yield self._get_inventory_stream(self._revision_keys)
995
# TODO: The keys to exclude might be part of the search recipe
996
# For now, exclude all parents that are at the edge of ancestry, for
997
# which we have inventories
998
from_repo = self.from_repository
999
parent_keys = from_repo._find_parent_keys_of_revisions(
1000
self._revision_keys)
1001
for stream_info in self._get_filtered_chk_streams(parent_keys):
1003
yield self._get_text_stream()
1005
def get_stream_for_missing_keys(self, missing_keys):
1006
# missing keys can only occur when we are byte copying and not
1007
# translating (because translation means we don't send
1008
# unreconstructable deltas ever).
1009
missing_inventory_keys = set()
1010
for key in missing_keys:
1011
if key[0] != 'inventories':
1012
raise AssertionError('The only missing keys we should'
1013
' be filling in are inventory keys, not %s'
1015
missing_inventory_keys.add(key[1:])
1016
if self._chk_id_roots or self._chk_p_id_roots:
1017
raise AssertionError('Cannot call get_stream_for_missing_keys'
1018
' untill all of get_stream() has been consumed.')
1019
# Yield the inventory stream, so we can find the chk stream
1020
# Some of the missing_keys will be missing because they are ghosts.
1021
# As such, we can ignore them. The Sink is required to verify there are
1022
# no unavailable texts when the ghost inventories are not filled in.
1023
yield self._get_inventory_stream(missing_inventory_keys,
1025
# We use the empty set for excluded_revision_keys, to make it clear
1026
# that we want to transmit all referenced chk pages.
1027
for stream_info in self._get_filtered_chk_streams(set()):
1031
class RepositoryFormatCHK1(RepositoryFormatPack):
1032
"""A hashed CHK+group compress pack repository."""
1034
repository_class = CHKInventoryRepository
1035
supports_external_lookups = True
1036
supports_chks = True
1037
# For right now, setting this to True gives us InterModel1And2 rather
1038
# than InterDifferingSerializer
1039
_commit_builder_class = PackRootCommitBuilder
1040
rich_root_data = True
1041
_serializer = chk_serializer.chk_serializer_255_bigpage
1042
_commit_inv_deltas = True
1043
# What index classes to use
1044
index_builder_class = BTreeBuilder
1045
index_class = BTreeGraphIndex
1046
# Note: We cannot unpack a delta that references a text we haven't
1047
# seen yet. There are 2 options, work in fulltexts, or require
1048
# topological sorting. Using fulltexts is more optimal for local
1049
# operations, because the source can be smart about extracting
1050
# multiple in-a-row (and sharing strings). Topological is better
1051
# for remote, because we access less data.
1052
_fetch_order = 'unordered'
1053
_fetch_uses_deltas = False # essentially ignored by the groupcompress code.
1056
def _get_matching_bzrdir(self):
1057
return bzrdir.format_registry.make_bzrdir('development6-rich-root')
1059
def _ignore_setting_bzrdir(self, format):
1062
_matchingbzrdir = property(_get_matching_bzrdir, _ignore_setting_bzrdir)
1064
def get_format_string(self):
1065
"""See RepositoryFormat.get_format_string()."""
1066
return ('Bazaar development format - group compression and chk inventory'
1067
' (needs bzr.dev from 1.14)\n')
1069
def get_format_description(self):
1070
"""See RepositoryFormat.get_format_description()."""
1071
return ("Development repository format - rich roots, group compression"
1072
" and chk inventories")
1074
def check_conversion_target(self, target_format):
1075
if not target_format.rich_root_data:
1076
raise errors.BadConversionTarget(
1077
'Does not support rich root data.', target_format)
1078
if not getattr(target_format, 'supports_tree_reference', False):
1079
raise errors.BadConversionTarget(
1080
'Does not support nested trees', target_format)
1084
class RepositoryFormatCHK2(RepositoryFormatCHK1):
1085
"""A CHK repository that uses the bencode revision serializer."""
1087
_serializer = chk_serializer.chk_bencode_serializer
1089
def _get_matching_bzrdir(self):
1090
return bzrdir.format_registry.make_bzrdir('development7-rich-root')
1092
def _ignore_setting_bzrdir(self, format):
1095
_matchingbzrdir = property(_get_matching_bzrdir, _ignore_setting_bzrdir)
1097
def get_format_string(self):
1098
"""See RepositoryFormat.get_format_string()."""
1099
return ('Bazaar development format - chk repository with bencode '
1100
'revision serialization (needs bzr.dev from 1.16)\n')
1103
class RepositoryFormat2a(RepositoryFormatCHK2):
1104
"""A CHK repository that uses the bencode revision serializer.
1106
This is the same as RepositoryFormatCHK2 but with a public name.
1109
_serializer = chk_serializer.chk_bencode_serializer
1111
def _get_matching_bzrdir(self):
1112
return bzrdir.format_registry.make_bzrdir('2a')
1114
def _ignore_setting_bzrdir(self, format):
1117
_matchingbzrdir = property(_get_matching_bzrdir, _ignore_setting_bzrdir)
1119
def get_format_string(self):
1120
return ('Bazaar repository format 2a (needs bzr 1.16 or later)\n')
added
removed
bzrlib/repofmt/groupcompress_repo.py
