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."""
33
revision as _mod_revision,
37
from bzrlib.btree_index import (
41
from bzrlib.groupcompress import (
43
GroupCompressVersionedFiles,
45
from bzrlib.repofmt.pack_repo import (
50
PackRootCommitBuilder,
51
RepositoryPackCollection,
58
class GCPack(NewPack):
60
def __init__(self, pack_collection, upload_suffix='', file_mode=None):
61
"""Create a NewPack instance.
63
:param pack_collection: A PackCollection into which this is being
65
:param upload_suffix: An optional suffix to be given to any temporary
66
files created during the pack creation. e.g '.autopack'
67
:param file_mode: An optional file mode to create the new files with.
69
# replaced from NewPack to:
70
# - change inventory reference list length to 1
71
# - change texts reference lists to 1
72
# TODO: patch this to be parameterised
74
# The relative locations of the packs are constrained, but all are
75
# passed in because the caller has them, so as to avoid object churn.
76
index_builder_class = pack_collection._index_builder_class
78
if pack_collection.chk_index is not None:
79
chk_index = index_builder_class(reference_lists=0)
83
# Revisions: parents list, no text compression.
84
index_builder_class(reference_lists=1),
85
# Inventory: We want to map compression only, but currently the
86
# knit code hasn't been updated enough to understand that, so we
87
# have a regular 2-list index giving parents and compression
89
index_builder_class(reference_lists=1),
90
# Texts: per file graph, for all fileids - so one reference list
91
# and two elements in the key tuple.
92
index_builder_class(reference_lists=1, key_elements=2),
93
# Signatures: Just blobs to store, no compression, no parents
95
index_builder_class(reference_lists=0),
96
# CHK based storage - just blobs, no compression or parents.
99
self._pack_collection = pack_collection
100
# When we make readonly indices, we need this.
101
self.index_class = pack_collection._index_class
102
# where should the new pack be opened
103
self.upload_transport = pack_collection._upload_transport
104
# where are indices written out to
105
self.index_transport = pack_collection._index_transport
106
# where is the pack renamed to when it is finished?
107
self.pack_transport = pack_collection._pack_transport
108
# What file mode to upload the pack and indices with.
109
self._file_mode = file_mode
110
# tracks the content written to the .pack file.
111
self._hash = osutils.md5()
112
# a four-tuple with the length in bytes of the indices, once the pack
113
# is finalised. (rev, inv, text, sigs)
114
self.index_sizes = None
115
# How much data to cache when writing packs. Note that this is not
116
# synchronised with reads, because it's not in the transport layer, so
117
# is not safe unless the client knows it won't be reading from the pack
119
self._cache_limit = 0
120
# the temporary pack file name.
121
self.random_name = osutils.rand_chars(20) + upload_suffix
122
# when was this pack started ?
123
self.start_time = time.time()
124
# open an output stream for the data added to the pack.
125
self.write_stream = self.upload_transport.open_write_stream(
126
self.random_name, mode=self._file_mode)
127
if 'pack' in debug.debug_flags:
128
trace.mutter('%s: create_pack: pack stream open: %s%s t+%6.3fs',
129
time.ctime(), self.upload_transport.base, self.random_name,
130
time.time() - self.start_time)
131
# A list of byte sequences to be written to the new pack, and the
132
# aggregate size of them. Stored as a list rather than separate
133
# variables so that the _write_data closure below can update them.
134
self._buffer = [[], 0]
135
# create a callable for adding data
137
# robertc says- this is a closure rather than a method on the object
138
# so that the variables are locals, and faster than accessing object
140
def _write_data(bytes, flush=False, _buffer=self._buffer,
141
_write=self.write_stream.write, _update=self._hash.update):
142
_buffer[0].append(bytes)
143
_buffer[1] += len(bytes)
145
if _buffer[1] > self._cache_limit or flush:
146
bytes = ''.join(_buffer[0])
150
# expose this on self, for the occasion when clients want to add data.
151
self._write_data = _write_data
152
# a pack writer object to serialise pack records.
153
self._writer = pack.ContainerWriter(self._write_data)
155
# what state is the pack in? (open, finished, aborted)
157
# no name until we finish writing the content
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,
356
reload_func=self._reload_func)
359
if self.new_pack is None:
360
raise AssertionError('No new pack has been set')
361
index = getattr(self.new_pack, index_name)
362
index_to_pack[index] = self.new_pack.access_tuple()
363
index.set_optimize(for_size=True)
364
access.set_writer(self.new_pack._writer, index,
365
self.new_pack.access_tuple())
366
add_callback = index.add_nodes
369
for pack in self.packs:
370
sub_index = getattr(pack, index_name)
371
index_to_pack[sub_index] = pack.access_tuple()
372
indices.append(sub_index)
373
index = _mod_index.CombinedGraphIndex(indices)
375
vf = GroupCompressVersionedFiles(
377
add_callback=add_callback,
379
is_locked=self._pack_collection.repo.is_locked),
384
def _build_vfs(self, index_name, parents, delta):
385
"""Build the source and target VersionedFiles."""
386
source_vf = self._build_vf(index_name, parents,
387
delta, for_write=False)
388
target_vf = self._build_vf(index_name, parents,
389
delta, for_write=True)
390
return source_vf, target_vf
392
def _copy_stream(self, source_vf, target_vf, keys, message, vf_to_stream,
394
trace.mutter('repacking %d %s', len(keys), message)
395
self.pb.update('repacking %s' % (message,), pb_offset)
396
child_pb = ui.ui_factory.nested_progress_bar()
398
stream = vf_to_stream(source_vf, keys, message, child_pb)
399
for _ in target_vf._insert_record_stream(stream,
406
def _copy_revision_texts(self):
407
source_vf, target_vf = self._build_vfs('revision', True, False)
408
if not self.revision_keys:
409
# We are doing a full fetch, aka 'pack'
410
self.revision_keys = source_vf.keys()
411
self._copy_stream(source_vf, target_vf, self.revision_keys,
412
'revisions', self._get_progress_stream, 1)
414
def _copy_inventory_texts(self):
415
source_vf, target_vf = self._build_vfs('inventory', True, True)
416
# It is not sufficient to just use self.revision_keys, as stacked
417
# repositories can have more inventories than they have revisions.
418
# One alternative would be to do something with
419
# get_parent_map(self.revision_keys), but that shouldn't be any faster
421
inventory_keys = source_vf.keys()
422
missing_inventories = set(self.revision_keys).difference(inventory_keys)
423
if missing_inventories:
424
missing_inventories = sorted(missing_inventories)
425
raise ValueError('We are missing inventories for revisions: %s'
426
% (missing_inventories,))
427
self._copy_stream(source_vf, target_vf, inventory_keys,
428
'inventories', self._get_filtered_inv_stream, 2)
430
def _copy_chk_texts(self):
431
source_vf, target_vf = self._build_vfs('chk', False, False)
432
# TODO: This is technically spurious... if it is a performance issue,
434
total_keys = source_vf.keys()
435
trace.mutter('repacking chk: %d id_to_entry roots,'
436
' %d p_id_map roots, %d total keys',
437
len(self._chk_id_roots), len(self._chk_p_id_roots),
439
self.pb.update('repacking chk', 3)
440
child_pb = ui.ui_factory.nested_progress_bar()
442
for stream in self._get_chk_streams(source_vf, total_keys,
444
for _ in target_vf._insert_record_stream(stream,
451
def _copy_text_texts(self):
452
source_vf, target_vf = self._build_vfs('text', True, True)
453
# XXX: We don't walk the chk map to determine referenced (file_id,
454
# revision_id) keys. We don't do it yet because you really need
455
# to filter out the ones that are present in the parents of the
456
# rev just before the ones you are copying, otherwise the filter
457
# is grabbing too many keys...
458
text_keys = source_vf.keys()
459
self._copy_stream(source_vf, target_vf, text_keys,
460
'texts', self._get_progress_stream, 4)
462
def _copy_signature_texts(self):
463
source_vf, target_vf = self._build_vfs('signature', False, False)
464
signature_keys = source_vf.keys()
465
signature_keys.intersection(self.revision_keys)
466
self._copy_stream(source_vf, target_vf, signature_keys,
467
'signatures', self._get_progress_stream, 5)
469
def _create_pack_from_packs(self):
470
self.pb.update('repacking', 0, 7)
471
self.new_pack = self.open_pack()
472
# Is this necessary for GC ?
473
self.new_pack.set_write_cache_size(1024*1024)
474
self._copy_revision_texts()
475
self._copy_inventory_texts()
476
self._copy_chk_texts()
477
self._copy_text_texts()
478
self._copy_signature_texts()
479
self.new_pack._check_references()
480
if not self._use_pack(self.new_pack):
481
self.new_pack.abort()
483
self.new_pack.finish_content()
484
if len(self.packs) == 1:
485
old_pack = self.packs[0]
486
if old_pack.name == self.new_pack._hash.hexdigest():
487
# The single old pack was already optimally packed.
488
trace.mutter('single pack %s was already optimally packed',
490
self.new_pack.abort()
492
self.pb.update('finishing repack', 6, 7)
493
self.new_pack.finish()
494
self._pack_collection.allocate(self.new_pack)
498
class GCCHKReconcilePacker(GCCHKPacker):
499
"""A packer which regenerates indices etc as it copies.
501
This is used by ``bzr reconcile`` to cause parent text pointers to be
505
def __init__(self, *args, **kwargs):
506
super(GCCHKReconcilePacker, self).__init__(*args, **kwargs)
507
self._data_changed = False
508
self._gather_text_refs = True
510
def _copy_inventory_texts(self):
511
source_vf, target_vf = self._build_vfs('inventory', True, True)
512
self._copy_stream(source_vf, target_vf, self.revision_keys,
513
'inventories', self._get_filtered_inv_stream, 2)
514
if source_vf.keys() != self.revision_keys:
515
self._data_changed = True
517
def _copy_text_texts(self):
518
"""generate what texts we should have and then copy."""
519
source_vf, target_vf = self._build_vfs('text', True, True)
520
trace.mutter('repacking %d texts', len(self._text_refs))
521
self.pb.update("repacking texts", 4)
522
# we have three major tasks here:
523
# 1) generate the ideal index
524
repo = self._pack_collection.repo
525
# We want the one we just wrote, so base it on self.new_pack
526
revision_vf = self._build_vf('revision', True, False, for_write=True)
527
ancestor_keys = revision_vf.get_parent_map(revision_vf.keys())
528
# Strip keys back into revision_ids.
529
ancestors = dict((k[0], tuple([p[0] for p in parents]))
530
for k, parents in ancestor_keys.iteritems())
532
# TODO: _generate_text_key_index should be much cheaper to generate from
533
# a chk repository, rather than the current implementation
534
ideal_index = repo._generate_text_key_index(None, ancestors)
535
file_id_parent_map = source_vf.get_parent_map(self._text_refs)
536
# 2) generate a keys list that contains all the entries that can
537
# be used as-is, with corrected parents.
539
new_parent_keys = {} # (key, parent_keys)
541
NULL_REVISION = _mod_revision.NULL_REVISION
542
for key in self._text_refs:
548
ideal_parents = tuple(ideal_index[key])
550
discarded_keys.append(key)
551
self._data_changed = True
553
if ideal_parents == (NULL_REVISION,):
555
source_parents = file_id_parent_map[key]
556
if ideal_parents == source_parents:
560
# We need to change the parent graph, but we don't need to
561
# re-insert the text (since we don't pun the compression
562
# parent with the parents list)
563
self._data_changed = True
564
new_parent_keys[key] = ideal_parents
565
# we're finished with some data.
567
del file_id_parent_map
568
# 3) bulk copy the data, updating records than need it
569
def _update_parents_for_texts():
570
stream = source_vf.get_record_stream(self._text_refs,
571
'groupcompress', False)
572
for record in stream:
573
if record.key in new_parent_keys:
574
record.parents = new_parent_keys[record.key]
576
target_vf.insert_record_stream(_update_parents_for_texts())
578
def _use_pack(self, new_pack):
579
"""Override _use_pack to check for reconcile having changed content."""
580
return new_pack.data_inserted() and self._data_changed
583
class GCRepositoryPackCollection(RepositoryPackCollection):
585
pack_factory = GCPack
586
resumed_pack_factory = ResumedGCPack
588
def _check_new_inventories(self):
589
"""Detect missing inventories or chk root entries for the new revisions
592
:returns: list of strs, summarising any problems found. If the list is
593
empty no problems were found.
595
# Ensure that all revisions added in this write group have:
596
# - corresponding inventories,
597
# - chk root entries for those inventories,
598
# - and any present parent inventories have their chk root
600
# And all this should be independent of any fallback repository.
602
key_deps = self.repo.revisions._index._key_dependencies
603
new_revisions_keys = key_deps.get_new_keys()
604
no_fallback_inv_index = self.repo.inventories._index
605
no_fallback_chk_bytes_index = self.repo.chk_bytes._index
606
no_fallback_texts_index = self.repo.texts._index
607
inv_parent_map = no_fallback_inv_index.get_parent_map(
609
# Are any inventories for corresponding to the new revisions missing?
610
corresponding_invs = set(inv_parent_map)
611
missing_corresponding = set(new_revisions_keys)
612
missing_corresponding.difference_update(corresponding_invs)
613
if missing_corresponding:
614
problems.append("inventories missing for revisions %s" %
615
(sorted(missing_corresponding),))
617
# Are any chk root entries missing for any inventories? This includes
618
# any present parent inventories, which may be used when calculating
619
# deltas for streaming.
620
all_inv_keys = set(corresponding_invs)
621
for parent_inv_keys in inv_parent_map.itervalues():
622
all_inv_keys.update(parent_inv_keys)
623
# Filter out ghost parents.
624
all_inv_keys.intersection_update(
625
no_fallback_inv_index.get_parent_map(all_inv_keys))
626
parent_invs_only_keys = all_inv_keys.symmetric_difference(
629
inv_ids = [key[-1] for key in all_inv_keys]
630
parent_invs_only_ids = [key[-1] for key in parent_invs_only_keys]
631
root_key_info = _build_interesting_key_sets(
632
self.repo, inv_ids, parent_invs_only_ids)
633
expected_chk_roots = root_key_info.all_keys()
634
present_chk_roots = no_fallback_chk_bytes_index.get_parent_map(
636
missing_chk_roots = expected_chk_roots.difference(present_chk_roots)
637
if missing_chk_roots:
638
problems.append("missing referenced chk root keys: %s"
639
% (sorted(missing_chk_roots),))
640
# Don't bother checking any further.
642
# Find all interesting chk_bytes records, and make sure they are
643
# present, as well as the text keys they reference.
644
chk_bytes_no_fallbacks = self.repo.chk_bytes.without_fallbacks()
645
chk_bytes_no_fallbacks._search_key_func = \
646
self.repo.chk_bytes._search_key_func
647
chk_diff = chk_map.iter_interesting_nodes(
648
chk_bytes_no_fallbacks, root_key_info.interesting_root_keys,
649
root_key_info.uninteresting_root_keys)
650
bytes_to_info = inventory.CHKInventory._bytes_to_utf8name_key
653
for record in _filter_text_keys(chk_diff, text_keys, bytes_to_info):
655
except errors.NoSuchRevision, e:
656
# XXX: It would be nice if we could give a more precise error here.
657
problems.append("missing chk node(s) for id_to_entry maps")
658
chk_diff = chk_map.iter_interesting_nodes(
659
chk_bytes_no_fallbacks, root_key_info.interesting_pid_root_keys,
660
root_key_info.uninteresting_pid_root_keys)
662
for interesting_rec, interesting_map in chk_diff:
664
except errors.NoSuchRevision, e:
666
"missing chk node(s) for parent_id_basename_to_file_id maps")
667
present_text_keys = no_fallback_texts_index.get_parent_map(text_keys)
668
missing_text_keys = text_keys.difference(present_text_keys)
669
if missing_text_keys:
670
problems.append("missing text keys: %r"
671
% (sorted(missing_text_keys),))
674
def _execute_pack_operations(self, pack_operations,
675
_packer_class=GCCHKPacker,
677
"""Execute a series of pack operations.
679
:param pack_operations: A list of [revision_count, packs_to_combine].
680
:param _packer_class: The class of packer to use (default: Packer).
683
# XXX: Copied across from RepositoryPackCollection simply because we
684
# want to override the _packer_class ... :(
685
for revision_count, packs in pack_operations:
686
# we may have no-ops from the setup logic
689
packer = GCCHKPacker(self, packs, '.autopack',
690
reload_func=reload_func)
692
result = packer.pack()
693
except errors.RetryWithNewPacks:
694
# An exception is propagating out of this context, make sure
695
# this packer has cleaned up. Packer() doesn't set its new_pack
696
# state into the RepositoryPackCollection object, so we only
697
# have access to it directly here.
698
if packer.new_pack is not None:
699
packer.new_pack.abort()
704
self._remove_pack_from_memory(pack)
705
# record the newly available packs and stop advertising the old
707
result = self._save_pack_names(clear_obsolete_packs=True)
708
# Move the old packs out of the way now they are no longer referenced.
709
for revision_count, packs in pack_operations:
710
self._obsolete_packs(packs)
714
class CHKInventoryRepository(KnitPackRepository):
715
"""subclass of KnitPackRepository that uses CHK based inventories."""
717
def __init__(self, _format, a_bzrdir, control_files, _commit_builder_class,
719
"""Overridden to change pack collection class."""
720
KnitPackRepository.__init__(self, _format, a_bzrdir, control_files,
721
_commit_builder_class, _serializer)
722
# and now replace everything it did :)
723
index_transport = self._transport.clone('indices')
724
self._pack_collection = GCRepositoryPackCollection(self,
725
self._transport, index_transport,
726
self._transport.clone('upload'),
727
self._transport.clone('packs'),
728
_format.index_builder_class,
730
use_chk_index=self._format.supports_chks,
732
self.inventories = GroupCompressVersionedFiles(
733
_GCGraphIndex(self._pack_collection.inventory_index.combined_index,
734
add_callback=self._pack_collection.inventory_index.add_callback,
735
parents=True, is_locked=self.is_locked,
736
inconsistency_fatal=False),
737
access=self._pack_collection.inventory_index.data_access)
738
self.revisions = GroupCompressVersionedFiles(
739
_GCGraphIndex(self._pack_collection.revision_index.combined_index,
740
add_callback=self._pack_collection.revision_index.add_callback,
741
parents=True, is_locked=self.is_locked,
742
track_external_parent_refs=True, track_new_keys=True),
743
access=self._pack_collection.revision_index.data_access,
745
self.signatures = GroupCompressVersionedFiles(
746
_GCGraphIndex(self._pack_collection.signature_index.combined_index,
747
add_callback=self._pack_collection.signature_index.add_callback,
748
parents=False, is_locked=self.is_locked,
749
inconsistency_fatal=False),
750
access=self._pack_collection.signature_index.data_access,
752
self.texts = GroupCompressVersionedFiles(
753
_GCGraphIndex(self._pack_collection.text_index.combined_index,
754
add_callback=self._pack_collection.text_index.add_callback,
755
parents=True, is_locked=self.is_locked,
756
inconsistency_fatal=False),
757
access=self._pack_collection.text_index.data_access)
758
# No parents, individual CHK pages don't have specific ancestry
759
self.chk_bytes = GroupCompressVersionedFiles(
760
_GCGraphIndex(self._pack_collection.chk_index.combined_index,
761
add_callback=self._pack_collection.chk_index.add_callback,
762
parents=False, is_locked=self.is_locked,
763
inconsistency_fatal=False),
764
access=self._pack_collection.chk_index.data_access)
765
search_key_name = self._format._serializer.search_key_name
766
search_key_func = chk_map.search_key_registry.get(search_key_name)
767
self.chk_bytes._search_key_func = search_key_func
768
# True when the repository object is 'write locked' (as opposed to the
769
# physical lock only taken out around changes to the pack-names list.)
770
# Another way to represent this would be a decorator around the control
771
# files object that presents logical locks as physical ones - if this
772
# gets ugly consider that alternative design. RBC 20071011
773
self._write_lock_count = 0
774
self._transaction = None
776
self._reconcile_does_inventory_gc = True
777
self._reconcile_fixes_text_parents = True
778
self._reconcile_backsup_inventory = False
780
def _add_inventory_checked(self, revision_id, inv, parents):
781
"""Add inv to the repository after checking the inputs.
783
This function can be overridden to allow different inventory styles.
785
:seealso: add_inventory, for the contract.
788
serializer = self._format._serializer
789
result = inventory.CHKInventory.from_inventory(self.chk_bytes, inv,
790
maximum_size=serializer.maximum_size,
791
search_key_name=serializer.search_key_name)
792
inv_lines = result.to_lines()
793
return self._inventory_add_lines(revision_id, parents,
794
inv_lines, check_content=False)
796
def _create_inv_from_null(self, delta, revision_id):
797
"""This will mutate new_inv directly.
799
This is a simplified form of create_by_apply_delta which knows that all
800
the old values must be None, so everything is a create.
802
serializer = self._format._serializer
803
new_inv = inventory.CHKInventory(serializer.search_key_name)
804
new_inv.revision_id = revision_id
805
entry_to_bytes = new_inv._entry_to_bytes
806
id_to_entry_dict = {}
807
parent_id_basename_dict = {}
808
for old_path, new_path, file_id, entry in delta:
809
if old_path is not None:
810
raise ValueError('Invalid delta, somebody tried to delete %r'
811
' from the NULL_REVISION'
812
% ((old_path, file_id),))
814
raise ValueError('Invalid delta, delta from NULL_REVISION has'
815
' no new_path %r' % (file_id,))
817
new_inv.root_id = file_id
818
parent_id_basename_key = ('', '')
820
utf8_entry_name = entry.name.encode('utf-8')
821
parent_id_basename_key = (entry.parent_id, utf8_entry_name)
822
new_value = entry_to_bytes(entry)
824
# new_inv._path_to_fileid_cache[new_path] = file_id
825
id_to_entry_dict[(file_id,)] = new_value
826
parent_id_basename_dict[parent_id_basename_key] = file_id
828
new_inv._populate_from_dicts(self.chk_bytes, id_to_entry_dict,
829
parent_id_basename_dict, maximum_size=serializer.maximum_size)
832
def add_inventory_by_delta(self, basis_revision_id, delta, new_revision_id,
833
parents, basis_inv=None, propagate_caches=False):
834
"""Add a new inventory expressed as a delta against another revision.
836
:param basis_revision_id: The inventory id the delta was created
838
:param delta: The inventory delta (see Inventory.apply_delta for
840
:param new_revision_id: The revision id that the inventory is being
842
:param parents: The revision ids of the parents that revision_id is
843
known to have and are in the repository already. These are supplied
844
for repositories that depend on the inventory graph for revision
845
graph access, as well as for those that pun ancestry with delta
847
:param basis_inv: The basis inventory if it is already known,
849
:param propagate_caches: If True, the caches for this inventory are
850
copied to and updated for the result if possible.
852
:returns: (validator, new_inv)
853
The validator(which is a sha1 digest, though what is sha'd is
854
repository format specific) of the serialized inventory, and the
857
if not self.is_in_write_group():
858
raise AssertionError("%r not in write group" % (self,))
859
_mod_revision.check_not_reserved_id(new_revision_id)
861
if basis_inv is None:
862
if basis_revision_id == _mod_revision.NULL_REVISION:
863
new_inv = self._create_inv_from_null(delta, new_revision_id)
864
inv_lines = new_inv.to_lines()
865
return self._inventory_add_lines(new_revision_id, parents,
866
inv_lines, check_content=False), new_inv
868
basis_tree = self.revision_tree(basis_revision_id)
869
basis_tree.lock_read()
870
basis_inv = basis_tree.inventory
872
result = basis_inv.create_by_apply_delta(delta, new_revision_id,
873
propagate_caches=propagate_caches)
874
inv_lines = result.to_lines()
875
return self._inventory_add_lines(new_revision_id, parents,
876
inv_lines, check_content=False), result
878
if basis_tree is not None:
881
def deserialise_inventory(self, revision_id, bytes):
882
return inventory.CHKInventory.deserialise(self.chk_bytes, bytes,
885
def _iter_inventories(self, revision_ids, ordering):
886
"""Iterate over many inventory objects."""
888
ordering = 'unordered'
889
keys = [(revision_id,) for revision_id in revision_ids]
890
stream = self.inventories.get_record_stream(keys, ordering, True)
892
for record in stream:
893
if record.storage_kind != 'absent':
894
texts[record.key] = record.get_bytes_as('fulltext')
896
raise errors.NoSuchRevision(self, record.key)
898
yield inventory.CHKInventory.deserialise(self.chk_bytes, texts[key], key)
900
def _iter_inventory_xmls(self, revision_ids, ordering):
901
# Without a native 'xml' inventory, this method doesn't make sense.
902
# However older working trees, and older bundles want it - so we supply
903
# it allowing get_inventory_xml to work. Bundles currently use the
904
# serializer directly; this also isn't ideal, but there isn't an xml
905
# iteration interface offered at all for repositories. We could make
906
# _iter_inventory_xmls be part of the contract, even if kept private.
907
inv_to_str = self._serializer.write_inventory_to_string
908
for inv in self.iter_inventories(revision_ids, ordering=ordering):
909
yield inv_to_str(inv), inv.revision_id
911
def _find_present_inventory_keys(self, revision_keys):
912
parent_map = self.inventories.get_parent_map(revision_keys)
913
present_inventory_keys = set(k for k in parent_map)
914
return present_inventory_keys
916
def fileids_altered_by_revision_ids(self, revision_ids, _inv_weave=None):
917
"""Find the file ids and versions affected by revisions.
919
:param revisions: an iterable containing revision ids.
920
:param _inv_weave: The inventory weave from this repository or None.
921
If None, the inventory weave will be opened automatically.
922
:return: a dictionary mapping altered file-ids to an iterable of
923
revision_ids. Each altered file-ids has the exact revision_ids that
924
altered it listed explicitly.
926
rich_root = self.supports_rich_root()
927
bytes_to_info = inventory.CHKInventory._bytes_to_utf8name_key
928
file_id_revisions = {}
929
pb = ui.ui_factory.nested_progress_bar()
931
revision_keys = [(r,) for r in revision_ids]
932
parent_keys = self._find_parent_keys_of_revisions(revision_keys)
933
# TODO: instead of using _find_present_inventory_keys, change the
934
# code paths to allow missing inventories to be tolerated.
935
# However, we only want to tolerate missing parent
936
# inventories, not missing inventories for revision_ids
937
present_parent_inv_keys = self._find_present_inventory_keys(
939
present_parent_inv_ids = set(
940
[k[-1] for k in present_parent_inv_keys])
941
inventories_to_read = set(revision_ids)
942
inventories_to_read.update(present_parent_inv_ids)
943
root_key_info = _build_interesting_key_sets(
944
self, inventories_to_read, present_parent_inv_ids)
945
interesting_root_keys = root_key_info.interesting_root_keys
946
uninteresting_root_keys = root_key_info.uninteresting_root_keys
947
chk_bytes = self.chk_bytes
948
for record, items in chk_map.iter_interesting_nodes(chk_bytes,
949
interesting_root_keys, uninteresting_root_keys,
951
for name, bytes in items:
952
(name_utf8, file_id, revision_id) = bytes_to_info(bytes)
953
if not rich_root and name_utf8 == '':
956
file_id_revisions[file_id].add(revision_id)
958
file_id_revisions[file_id] = set([revision_id])
961
return file_id_revisions
963
def find_text_key_references(self):
964
"""Find the text key references within the repository.
966
:return: A dictionary mapping text keys ((fileid, revision_id) tuples)
967
to whether they were referred to by the inventory of the
968
revision_id that they contain. The inventory texts from all present
969
revision ids are assessed to generate this report.
971
# XXX: Slow version but correct: rewrite as a series of delta
972
# examinations/direct tree traversal. Note that that will require care
973
# as a common node is reachable both from the inventory that added it,
974
# and others afterwards.
975
revision_keys = self.revisions.keys()
977
rich_roots = self.supports_rich_root()
978
pb = ui.ui_factory.nested_progress_bar()
980
all_revs = self.all_revision_ids()
981
total = len(all_revs)
982
for pos, inv in enumerate(self.iter_inventories(all_revs)):
983
pb.update("Finding text references", pos, total)
984
for _, entry in inv.iter_entries():
985
if not rich_roots and entry.file_id == inv.root_id:
987
key = (entry.file_id, entry.revision)
988
result.setdefault(key, False)
989
if entry.revision == inv.revision_id:
995
def _reconcile_pack(self, collection, packs, extension, revs, pb):
996
packer = GCCHKReconcilePacker(collection, packs, extension)
997
return packer.pack(pb)
999
def _get_source(self, to_format):
1000
"""Return a source for streaming from this repository."""
1001
if self._format._serializer == to_format._serializer:
1002
# We must be exactly the same format, otherwise stuff like the chk
1003
# page layout might be different.
1004
# Actually, this test is just slightly looser than exact so that
1005
# CHK2 <-> 2a transfers will work.
1006
return GroupCHKStreamSource(self, to_format)
1007
return super(CHKInventoryRepository, self)._get_source(to_format)
1010
class GroupCHKStreamSource(KnitPackStreamSource):
1011
"""Used when both the source and target repo are GroupCHK repos."""
1013
def __init__(self, from_repository, to_format):
1014
"""Create a StreamSource streaming from from_repository."""
1015
super(GroupCHKStreamSource, self).__init__(from_repository, to_format)
1016
self._revision_keys = None
1017
self._text_keys = None
1018
self._text_fetch_order = 'groupcompress'
1019
self._chk_id_roots = None
1020
self._chk_p_id_roots = None
1022
def _get_inventory_stream(self, inventory_keys, allow_absent=False):
1023
"""Get a stream of inventory texts.
1025
When this function returns, self._chk_id_roots and self._chk_p_id_roots
1026
should be populated.
1028
self._chk_id_roots = []
1029
self._chk_p_id_roots = []
1030
def _filtered_inv_stream():
1031
id_roots_set = set()
1032
p_id_roots_set = set()
1033
source_vf = self.from_repository.inventories
1034
stream = source_vf.get_record_stream(inventory_keys,
1035
'groupcompress', True)
1036
for record in stream:
1037
if record.storage_kind == 'absent':
1041
raise errors.NoSuchRevision(self, record.key)
1042
bytes = record.get_bytes_as('fulltext')
1043
chk_inv = inventory.CHKInventory.deserialise(None, bytes,
1045
key = chk_inv.id_to_entry.key()
1046
if key not in id_roots_set:
1047
self._chk_id_roots.append(key)
1048
id_roots_set.add(key)
1049
p_id_map = chk_inv.parent_id_basename_to_file_id
1050
if p_id_map is None:
1051
raise AssertionError('Parent id -> file_id map not set')
1052
key = p_id_map.key()
1053
if key not in p_id_roots_set:
1054
p_id_roots_set.add(key)
1055
self._chk_p_id_roots.append(key)
1057
# We have finished processing all of the inventory records, we
1058
# don't need these sets anymore
1059
id_roots_set.clear()
1060
p_id_roots_set.clear()
1061
return ('inventories', _filtered_inv_stream())
1063
def _get_filtered_chk_streams(self, excluded_revision_keys):
1064
self._text_keys = set()
1065
excluded_revision_keys.discard(_mod_revision.NULL_REVISION)
1066
if not excluded_revision_keys:
1067
uninteresting_root_keys = set()
1068
uninteresting_pid_root_keys = set()
1070
# filter out any excluded revisions whose inventories are not
1072
# TODO: Update Repository.iter_inventories() to add
1073
# ignore_missing=True
1074
present_keys = self.from_repository._find_present_inventory_keys(
1075
excluded_revision_keys)
1076
present_ids = [k[-1] for k in present_keys]
1077
uninteresting_root_keys = set()
1078
uninteresting_pid_root_keys = set()
1079
for inv in self.from_repository.iter_inventories(present_ids):
1080
uninteresting_root_keys.add(inv.id_to_entry.key())
1081
uninteresting_pid_root_keys.add(
1082
inv.parent_id_basename_to_file_id.key())
1083
bytes_to_info = inventory.CHKInventory._bytes_to_utf8name_key
1084
chk_bytes = self.from_repository.chk_bytes
1085
def _filter_id_to_entry():
1086
interesting_nodes = chk_map.iter_interesting_nodes(chk_bytes,
1087
self._chk_id_roots, uninteresting_root_keys)
1088
for record in _filter_text_keys(interesting_nodes, self._text_keys,
1090
if record is not None:
1093
self._chk_id_roots = None
1094
yield 'chk_bytes', _filter_id_to_entry()
1095
def _get_parent_id_basename_to_file_id_pages():
1096
for record, items in chk_map.iter_interesting_nodes(chk_bytes,
1097
self._chk_p_id_roots, uninteresting_pid_root_keys):
1098
if record is not None:
1101
self._chk_p_id_roots = None
1102
yield 'chk_bytes', _get_parent_id_basename_to_file_id_pages()
1104
def get_stream(self, search):
1105
revision_ids = search.get_keys()
1106
for stream_info in self._fetch_revision_texts(revision_ids):
1108
self._revision_keys = [(rev_id,) for rev_id in revision_ids]
1109
yield self._get_inventory_stream(self._revision_keys)
1110
# TODO: The keys to exclude might be part of the search recipe
1111
# For now, exclude all parents that are at the edge of ancestry, for
1112
# which we have inventories
1113
from_repo = self.from_repository
1114
parent_keys = from_repo._find_parent_keys_of_revisions(
1115
self._revision_keys)
1116
for stream_info in self._get_filtered_chk_streams(parent_keys):
1118
yield self._get_text_stream()
1120
def get_stream_for_missing_keys(self, missing_keys):
1121
# missing keys can only occur when we are byte copying and not
1122
# translating (because translation means we don't send
1123
# unreconstructable deltas ever).
1124
missing_inventory_keys = set()
1125
for key in missing_keys:
1126
if key[0] != 'inventories':
1127
raise AssertionError('The only missing keys we should'
1128
' be filling in are inventory keys, not %s'
1130
missing_inventory_keys.add(key[1:])
1131
if self._chk_id_roots or self._chk_p_id_roots:
1132
raise AssertionError('Cannot call get_stream_for_missing_keys'
1133
' until all of get_stream() has been consumed.')
1134
# Yield the inventory stream, so we can find the chk stream
1135
# Some of the missing_keys will be missing because they are ghosts.
1136
# As such, we can ignore them. The Sink is required to verify there are
1137
# no unavailable texts when the ghost inventories are not filled in.
1138
yield self._get_inventory_stream(missing_inventory_keys,
1140
# We use the empty set for excluded_revision_keys, to make it clear
1141
# that we want to transmit all referenced chk pages.
1142
for stream_info in self._get_filtered_chk_streams(set()):
1146
class _InterestingKeyInfo(object):
1148
self.interesting_root_keys = set()
1149
self.interesting_pid_root_keys = set()
1150
self.uninteresting_root_keys = set()
1151
self.uninteresting_pid_root_keys = set()
1153
def all_interesting(self):
1154
return self.interesting_root_keys.union(self.interesting_pid_root_keys)
1156
def all_uninteresting(self):
1157
return self.uninteresting_root_keys.union(
1158
self.uninteresting_pid_root_keys)
1161
return self.all_interesting().union(self.all_uninteresting())
1164
def _build_interesting_key_sets(repo, inventory_ids, parent_only_inv_ids):
1165
result = _InterestingKeyInfo()
1166
for inv in repo.iter_inventories(inventory_ids, 'unordered'):
1167
root_key = inv.id_to_entry.key()
1168
pid_root_key = inv.parent_id_basename_to_file_id.key()
1169
if inv.revision_id in parent_only_inv_ids:
1170
result.uninteresting_root_keys.add(root_key)
1171
result.uninteresting_pid_root_keys.add(pid_root_key)
1173
result.interesting_root_keys.add(root_key)
1174
result.interesting_pid_root_keys.add(pid_root_key)
1178
def _filter_text_keys(interesting_nodes_iterable, text_keys, bytes_to_info):
1179
"""Iterate the result of iter_interesting_nodes, yielding the records
1180
and adding to text_keys.
1182
for record, items in interesting_nodes_iterable:
1183
for name, bytes in items:
1184
# Note: we don't care about name_utf8, because groupcompress repos
1185
# are always rich-root, so there are no synthesised root records to
1187
_, file_id, revision_id = bytes_to_info(bytes)
1188
text_keys.add((file_id, revision_id))
1194
class RepositoryFormatCHK1(RepositoryFormatPack):
1195
"""A hashed CHK+group compress pack repository."""
1197
repository_class = CHKInventoryRepository
1198
supports_external_lookups = True
1199
supports_chks = True
1200
# For right now, setting this to True gives us InterModel1And2 rather
1201
# than InterDifferingSerializer
1202
_commit_builder_class = PackRootCommitBuilder
1203
rich_root_data = True
1204
_serializer = chk_serializer.chk_serializer_255_bigpage
1205
_commit_inv_deltas = True
1206
# What index classes to use
1207
index_builder_class = BTreeBuilder
1208
index_class = BTreeGraphIndex
1209
# Note: We cannot unpack a delta that references a text we haven't
1210
# seen yet. There are 2 options, work in fulltexts, or require
1211
# topological sorting. Using fulltexts is more optimal for local
1212
# operations, because the source can be smart about extracting
1213
# multiple in-a-row (and sharing strings). Topological is better
1214
# for remote, because we access less data.
1215
_fetch_order = 'unordered'
1216
_fetch_uses_deltas = False # essentially ignored by the groupcompress code.
1218
pack_compresses = True
1220
def _get_matching_bzrdir(self):
1221
return bzrdir.format_registry.make_bzrdir('development6-rich-root')
1223
def _ignore_setting_bzrdir(self, format):
1226
_matchingbzrdir = property(_get_matching_bzrdir, _ignore_setting_bzrdir)
1228
def get_format_string(self):
1229
"""See RepositoryFormat.get_format_string()."""
1230
return ('Bazaar development format - group compression and chk inventory'
1231
' (needs bzr.dev from 1.14)\n')
1233
def get_format_description(self):
1234
"""See RepositoryFormat.get_format_description()."""
1235
return ("Development repository format - rich roots, group compression"
1236
" and chk inventories")
1239
class RepositoryFormatCHK2(RepositoryFormatCHK1):
1240
"""A CHK repository that uses the bencode revision serializer."""
1242
_serializer = chk_serializer.chk_bencode_serializer
1244
def _get_matching_bzrdir(self):
1245
return bzrdir.format_registry.make_bzrdir('development7-rich-root')
1247
def _ignore_setting_bzrdir(self, format):
1250
_matchingbzrdir = property(_get_matching_bzrdir, _ignore_setting_bzrdir)
1252
def get_format_string(self):
1253
"""See RepositoryFormat.get_format_string()."""
1254
return ('Bazaar development format - chk repository with bencode '
1255
'revision serialization (needs bzr.dev from 1.16)\n')
1258
class RepositoryFormat2a(RepositoryFormatCHK2):
1259
"""A CHK repository that uses the bencode revision serializer.
1261
This is the same as RepositoryFormatCHK2 but with a public name.
1264
_serializer = chk_serializer.chk_bencode_serializer
1266
def _get_matching_bzrdir(self):
1267
return bzrdir.format_registry.make_bzrdir('2a')
1269
def _ignore_setting_bzrdir(self, format):
1272
_matchingbzrdir = property(_get_matching_bzrdir, _ignore_setting_bzrdir)
1274
def get_format_string(self):
1275
return ('Bazaar repository format 2a (needs bzr 1.16 or later)\n')
1277
def get_format_description(self):
1278
"""See RepositoryFormat.get_format_description()."""
1279
return ("Repository format 2a - rich roots, group compression"
1280
" and chk inventories")
added
removed
bzrlib/repofmt/groupcompress_repo.py
