1
# Copyright (C) 2008, 2009, 2010 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."""
32
revision as _mod_revision,
36
from bzrlib.btree_index import (
40
from bzrlib.groupcompress import (
42
GroupCompressVersionedFiles,
44
from bzrlib.repofmt.pack_repo import (
49
PackRootCommitBuilder,
50
RepositoryPackCollection,
55
from bzrlib.static_tuple import StaticTuple
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
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
self._text_refs.add(chk_map._bytes_to_text_key(bytes))
293
stream = source_vf.get_record_stream(cur_keys,
294
'as-requested', True)
295
for record in stream:
296
if record.storage_kind == 'absent':
297
# An absent CHK record: we assume that the missing
298
# record is in a different pack - e.g. a page not
299
# altered by the commit we're packing.
301
bytes = record.get_bytes_as('fulltext')
302
# We don't care about search_key_func for this code,
303
# because we only care about external references.
304
node = chk_map._deserialise(bytes, record.key,
305
search_key_func=None)
306
common_base = node._search_prefix
307
if isinstance(node, chk_map.InternalNode):
308
handle_internal_node(node)
309
elif parse_leaf_nodes:
310
handle_leaf_node(node)
313
pb.update('chk node', counter[0], total_keys)
316
# Double check that we won't be emitting any keys twice
317
# If we get rid of the pre-calculation of all keys, we could
318
# turn this around and do
319
# next_keys.difference_update(seen_keys)
320
# However, we also may have references to chk pages in another
321
# pack file during autopack. We filter earlier, so we should no
322
# longer need to do this
323
# next_keys = next_keys.intersection(remaining_keys)
325
for prefix in sorted(keys_by_search_prefix):
326
cur_keys.extend(keys_by_search_prefix.pop(prefix))
327
for stream in _get_referenced_stream(self._chk_id_roots,
328
self._gather_text_refs):
330
del self._chk_id_roots
331
# while it isn't really possible for chk_id_roots to not be in the
332
# local group of packs, it is possible that the tree shape has not
333
# changed recently, so we need to filter _chk_p_id_roots by the
335
chk_p_id_roots = [key for key in self._chk_p_id_roots
336
if key in remaining_keys]
337
del self._chk_p_id_roots
338
for stream in _get_referenced_stream(chk_p_id_roots, False):
341
trace.mutter('There were %d keys in the chk index, %d of which'
342
' were not referenced', total_keys,
344
if self.revision_ids is None:
345
stream = source_vf.get_record_stream(remaining_keys,
349
def _build_vf(self, index_name, parents, delta, for_write=False):
350
"""Build a VersionedFiles instance on top of this group of packs."""
351
index_name = index_name + '_index'
353
access = knit._DirectPackAccess(index_to_pack,
354
reload_func=self._reload_func)
357
if self.new_pack is None:
358
raise AssertionError('No new pack has been set')
359
index = getattr(self.new_pack, index_name)
360
index_to_pack[index] = self.new_pack.access_tuple()
361
index.set_optimize(for_size=True)
362
access.set_writer(self.new_pack._writer, index,
363
self.new_pack.access_tuple())
364
add_callback = index.add_nodes
367
for pack in self.packs:
368
sub_index = getattr(pack, index_name)
369
index_to_pack[sub_index] = pack.access_tuple()
370
indices.append(sub_index)
371
index = _mod_index.CombinedGraphIndex(indices)
373
vf = GroupCompressVersionedFiles(
375
add_callback=add_callback,
377
is_locked=self._pack_collection.repo.is_locked),
382
def _build_vfs(self, index_name, parents, delta):
383
"""Build the source and target VersionedFiles."""
384
source_vf = self._build_vf(index_name, parents,
385
delta, for_write=False)
386
target_vf = self._build_vf(index_name, parents,
387
delta, for_write=True)
388
return source_vf, target_vf
390
def _copy_stream(self, source_vf, target_vf, keys, message, vf_to_stream,
392
trace.mutter('repacking %d %s', len(keys), message)
393
self.pb.update('repacking %s' % (message,), pb_offset)
394
child_pb = ui.ui_factory.nested_progress_bar()
396
stream = vf_to_stream(source_vf, keys, message, child_pb)
397
for _ in target_vf._insert_record_stream(stream,
404
def _copy_revision_texts(self):
405
source_vf, target_vf = self._build_vfs('revision', True, False)
406
if not self.revision_keys:
407
# We are doing a full fetch, aka 'pack'
408
self.revision_keys = source_vf.keys()
409
self._copy_stream(source_vf, target_vf, self.revision_keys,
410
'revisions', self._get_progress_stream, 1)
412
def _copy_inventory_texts(self):
413
source_vf, target_vf = self._build_vfs('inventory', True, True)
414
# It is not sufficient to just use self.revision_keys, as stacked
415
# repositories can have more inventories than they have revisions.
416
# One alternative would be to do something with
417
# get_parent_map(self.revision_keys), but that shouldn't be any faster
419
inventory_keys = source_vf.keys()
420
missing_inventories = set(self.revision_keys).difference(inventory_keys)
421
if missing_inventories:
422
missing_inventories = sorted(missing_inventories)
423
raise ValueError('We are missing inventories for revisions: %s'
424
% (missing_inventories,))
425
self._copy_stream(source_vf, target_vf, inventory_keys,
426
'inventories', self._get_filtered_inv_stream, 2)
428
def _copy_chk_texts(self):
429
source_vf, target_vf = self._build_vfs('chk', False, False)
430
# TODO: This is technically spurious... if it is a performance issue,
432
total_keys = source_vf.keys()
433
trace.mutter('repacking chk: %d id_to_entry roots,'
434
' %d p_id_map roots, %d total keys',
435
len(self._chk_id_roots), len(self._chk_p_id_roots),
437
self.pb.update('repacking chk', 3)
438
child_pb = ui.ui_factory.nested_progress_bar()
440
for stream in self._get_chk_streams(source_vf, total_keys,
442
for _ in target_vf._insert_record_stream(stream,
449
def _copy_text_texts(self):
450
source_vf, target_vf = self._build_vfs('text', True, True)
451
# XXX: We don't walk the chk map to determine referenced (file_id,
452
# revision_id) keys. We don't do it yet because you really need
453
# to filter out the ones that are present in the parents of the
454
# rev just before the ones you are copying, otherwise the filter
455
# is grabbing too many keys...
456
text_keys = source_vf.keys()
457
self._copy_stream(source_vf, target_vf, text_keys,
458
'texts', self._get_progress_stream, 4)
460
def _copy_signature_texts(self):
461
source_vf, target_vf = self._build_vfs('signature', False, False)
462
signature_keys = source_vf.keys()
463
signature_keys.intersection(self.revision_keys)
464
self._copy_stream(source_vf, target_vf, signature_keys,
465
'signatures', self._get_progress_stream, 5)
467
def _create_pack_from_packs(self):
468
self.pb.update('repacking', 0, 7)
469
self.new_pack = self.open_pack()
470
# Is this necessary for GC ?
471
self.new_pack.set_write_cache_size(1024*1024)
472
self._copy_revision_texts()
473
self._copy_inventory_texts()
474
self._copy_chk_texts()
475
self._copy_text_texts()
476
self._copy_signature_texts()
477
self.new_pack._check_references()
478
if not self._use_pack(self.new_pack):
479
self.new_pack.abort()
481
self.new_pack.finish_content()
482
if len(self.packs) == 1:
483
old_pack = self.packs[0]
484
if old_pack.name == self.new_pack._hash.hexdigest():
485
# The single old pack was already optimally packed.
486
trace.mutter('single pack %s was already optimally packed',
488
self.new_pack.abort()
490
self.pb.update('finishing repack', 6, 7)
491
self.new_pack.finish()
492
self._pack_collection.allocate(self.new_pack)
496
class GCCHKReconcilePacker(GCCHKPacker):
497
"""A packer which regenerates indices etc as it copies.
499
This is used by ``bzr reconcile`` to cause parent text pointers to be
503
def __init__(self, *args, **kwargs):
504
super(GCCHKReconcilePacker, self).__init__(*args, **kwargs)
505
self._data_changed = False
506
self._gather_text_refs = True
508
def _copy_inventory_texts(self):
509
source_vf, target_vf = self._build_vfs('inventory', True, True)
510
self._copy_stream(source_vf, target_vf, self.revision_keys,
511
'inventories', self._get_filtered_inv_stream, 2)
512
if source_vf.keys() != self.revision_keys:
513
self._data_changed = True
515
def _copy_text_texts(self):
516
"""generate what texts we should have and then copy."""
517
source_vf, target_vf = self._build_vfs('text', True, True)
518
trace.mutter('repacking %d texts', len(self._text_refs))
519
self.pb.update("repacking texts", 4)
520
# we have three major tasks here:
521
# 1) generate the ideal index
522
repo = self._pack_collection.repo
523
# We want the one we just wrote, so base it on self.new_pack
524
revision_vf = self._build_vf('revision', True, False, for_write=True)
525
ancestor_keys = revision_vf.get_parent_map(revision_vf.keys())
526
# Strip keys back into revision_ids.
527
ancestors = dict((k[0], tuple([p[0] for p in parents]))
528
for k, parents in ancestor_keys.iteritems())
530
# TODO: _generate_text_key_index should be much cheaper to generate from
531
# a chk repository, rather than the current implementation
532
ideal_index = repo._generate_text_key_index(None, ancestors)
533
file_id_parent_map = source_vf.get_parent_map(self._text_refs)
534
# 2) generate a keys list that contains all the entries that can
535
# be used as-is, with corrected parents.
537
new_parent_keys = {} # (key, parent_keys)
539
NULL_REVISION = _mod_revision.NULL_REVISION
540
for key in self._text_refs:
546
ideal_parents = tuple(ideal_index[key])
548
discarded_keys.append(key)
549
self._data_changed = True
551
if ideal_parents == (NULL_REVISION,):
553
source_parents = file_id_parent_map[key]
554
if ideal_parents == source_parents:
558
# We need to change the parent graph, but we don't need to
559
# re-insert the text (since we don't pun the compression
560
# parent with the parents list)
561
self._data_changed = True
562
new_parent_keys[key] = ideal_parents
563
# we're finished with some data.
565
del file_id_parent_map
566
# 3) bulk copy the data, updating records than need it
567
def _update_parents_for_texts():
568
stream = source_vf.get_record_stream(self._text_refs,
569
'groupcompress', False)
570
for record in stream:
571
if record.key in new_parent_keys:
572
record.parents = new_parent_keys[record.key]
574
target_vf.insert_record_stream(_update_parents_for_texts())
576
def _use_pack(self, new_pack):
577
"""Override _use_pack to check for reconcile having changed content."""
578
return new_pack.data_inserted() and self._data_changed
581
class GCRepositoryPackCollection(RepositoryPackCollection):
583
pack_factory = GCPack
584
resumed_pack_factory = ResumedGCPack
586
def _check_new_inventories(self):
587
"""Detect missing inventories or chk root entries for the new revisions
590
:returns: list of strs, summarising any problems found. If the list is
591
empty no problems were found.
593
# Ensure that all revisions added in this write group have:
594
# - corresponding inventories,
595
# - chk root entries for those inventories,
596
# - and any present parent inventories have their chk root
598
# And all this should be independent of any fallback repository.
600
key_deps = self.repo.revisions._index._key_dependencies
601
new_revisions_keys = key_deps.get_new_keys()
602
no_fallback_inv_index = self.repo.inventories._index
603
no_fallback_chk_bytes_index = self.repo.chk_bytes._index
604
no_fallback_texts_index = self.repo.texts._index
605
inv_parent_map = no_fallback_inv_index.get_parent_map(
607
# Are any inventories for corresponding to the new revisions missing?
608
corresponding_invs = set(inv_parent_map)
609
missing_corresponding = set(new_revisions_keys)
610
missing_corresponding.difference_update(corresponding_invs)
611
if missing_corresponding:
612
problems.append("inventories missing for revisions %s" %
613
(sorted(missing_corresponding),))
615
# Are any chk root entries missing for any inventories? This includes
616
# any present parent inventories, which may be used when calculating
617
# deltas for streaming.
618
all_inv_keys = set(corresponding_invs)
619
for parent_inv_keys in inv_parent_map.itervalues():
620
all_inv_keys.update(parent_inv_keys)
621
# Filter out ghost parents.
622
all_inv_keys.intersection_update(
623
no_fallback_inv_index.get_parent_map(all_inv_keys))
624
parent_invs_only_keys = all_inv_keys.symmetric_difference(
627
inv_ids = [key[-1] for key in all_inv_keys]
628
parent_invs_only_ids = [key[-1] for key in parent_invs_only_keys]
629
root_key_info = _build_interesting_key_sets(
630
self.repo, inv_ids, parent_invs_only_ids)
631
expected_chk_roots = root_key_info.all_keys()
632
present_chk_roots = no_fallback_chk_bytes_index.get_parent_map(
634
missing_chk_roots = expected_chk_roots.difference(present_chk_roots)
635
if missing_chk_roots:
636
problems.append("missing referenced chk root keys: %s"
637
% (sorted(missing_chk_roots),))
638
# Don't bother checking any further.
640
# Find all interesting chk_bytes records, and make sure they are
641
# present, as well as the text keys they reference.
642
chk_bytes_no_fallbacks = self.repo.chk_bytes.without_fallbacks()
643
chk_bytes_no_fallbacks._search_key_func = \
644
self.repo.chk_bytes._search_key_func
645
chk_diff = chk_map.iter_interesting_nodes(
646
chk_bytes_no_fallbacks, root_key_info.interesting_root_keys,
647
root_key_info.uninteresting_root_keys)
650
for record in _filter_text_keys(chk_diff, text_keys,
651
chk_map._bytes_to_text_key):
653
except errors.NoSuchRevision, e:
654
# XXX: It would be nice if we could give a more precise error here.
655
problems.append("missing chk node(s) for id_to_entry maps")
656
chk_diff = chk_map.iter_interesting_nodes(
657
chk_bytes_no_fallbacks, root_key_info.interesting_pid_root_keys,
658
root_key_info.uninteresting_pid_root_keys)
660
for interesting_rec, interesting_map in chk_diff:
662
except errors.NoSuchRevision, e:
664
"missing chk node(s) for parent_id_basename_to_file_id maps")
665
present_text_keys = no_fallback_texts_index.get_parent_map(text_keys)
666
missing_text_keys = text_keys.difference(present_text_keys)
667
if missing_text_keys:
668
problems.append("missing text keys: %r"
669
% (sorted(missing_text_keys),))
672
def _execute_pack_operations(self, pack_operations,
673
_packer_class=GCCHKPacker,
675
"""Execute a series of pack operations.
677
:param pack_operations: A list of [revision_count, packs_to_combine].
678
:param _packer_class: The class of packer to use (default: Packer).
681
# XXX: Copied across from RepositoryPackCollection simply because we
682
# want to override the _packer_class ... :(
683
for revision_count, packs in pack_operations:
684
# we may have no-ops from the setup logic
687
packer = GCCHKPacker(self, packs, '.autopack',
688
reload_func=reload_func)
690
result = packer.pack()
691
except errors.RetryWithNewPacks:
692
# An exception is propagating out of this context, make sure
693
# this packer has cleaned up. Packer() doesn't set its new_pack
694
# state into the RepositoryPackCollection object, so we only
695
# have access to it directly here.
696
if packer.new_pack is not None:
697
packer.new_pack.abort()
702
self._remove_pack_from_memory(pack)
703
# record the newly available packs and stop advertising the old
706
for _, packs in pack_operations:
707
to_be_obsoleted.extend(packs)
708
result = self._save_pack_names(clear_obsolete_packs=True,
709
obsolete_packs=to_be_obsoleted)
713
class CHKInventoryRepository(KnitPackRepository):
714
"""subclass of KnitPackRepository that uses CHK based inventories."""
716
def __init__(self, _format, a_bzrdir, control_files, _commit_builder_class,
718
"""Overridden to change pack collection class."""
719
KnitPackRepository.__init__(self, _format, a_bzrdir, control_files,
720
_commit_builder_class, _serializer)
721
# and now replace everything it did :)
722
index_transport = self._transport.clone('indices')
723
self._pack_collection = GCRepositoryPackCollection(self,
724
self._transport, index_transport,
725
self._transport.clone('upload'),
726
self._transport.clone('packs'),
727
_format.index_builder_class,
729
use_chk_index=self._format.supports_chks,
731
self.inventories = GroupCompressVersionedFiles(
732
_GCGraphIndex(self._pack_collection.inventory_index.combined_index,
733
add_callback=self._pack_collection.inventory_index.add_callback,
734
parents=True, is_locked=self.is_locked,
735
inconsistency_fatal=False),
736
access=self._pack_collection.inventory_index.data_access)
737
self.revisions = GroupCompressVersionedFiles(
738
_GCGraphIndex(self._pack_collection.revision_index.combined_index,
739
add_callback=self._pack_collection.revision_index.add_callback,
740
parents=True, is_locked=self.is_locked,
741
track_external_parent_refs=True, track_new_keys=True),
742
access=self._pack_collection.revision_index.data_access,
744
self.signatures = GroupCompressVersionedFiles(
745
_GCGraphIndex(self._pack_collection.signature_index.combined_index,
746
add_callback=self._pack_collection.signature_index.add_callback,
747
parents=False, is_locked=self.is_locked,
748
inconsistency_fatal=False),
749
access=self._pack_collection.signature_index.data_access,
751
self.texts = GroupCompressVersionedFiles(
752
_GCGraphIndex(self._pack_collection.text_index.combined_index,
753
add_callback=self._pack_collection.text_index.add_callback,
754
parents=True, is_locked=self.is_locked,
755
inconsistency_fatal=False),
756
access=self._pack_collection.text_index.data_access)
757
# No parents, individual CHK pages don't have specific ancestry
758
self.chk_bytes = GroupCompressVersionedFiles(
759
_GCGraphIndex(self._pack_collection.chk_index.combined_index,
760
add_callback=self._pack_collection.chk_index.add_callback,
761
parents=False, is_locked=self.is_locked,
762
inconsistency_fatal=False),
763
access=self._pack_collection.chk_index.data_access)
764
search_key_name = self._format._serializer.search_key_name
765
search_key_func = chk_map.search_key_registry.get(search_key_name)
766
self.chk_bytes._search_key_func = search_key_func
767
# True when the repository object is 'write locked' (as opposed to the
768
# physical lock only taken out around changes to the pack-names list.)
769
# Another way to represent this would be a decorator around the control
770
# files object that presents logical locks as physical ones - if this
771
# gets ugly consider that alternative design. RBC 20071011
772
self._write_lock_count = 0
773
self._transaction = None
775
self._reconcile_does_inventory_gc = True
776
self._reconcile_fixes_text_parents = True
777
self._reconcile_backsup_inventory = False
779
def _add_inventory_checked(self, revision_id, inv, parents):
780
"""Add inv to the repository after checking the inputs.
782
This function can be overridden to allow different inventory styles.
784
:seealso: add_inventory, for the contract.
787
serializer = self._format._serializer
788
result = inventory.CHKInventory.from_inventory(self.chk_bytes, inv,
789
maximum_size=serializer.maximum_size,
790
search_key_name=serializer.search_key_name)
791
inv_lines = result.to_lines()
792
return self._inventory_add_lines(revision_id, parents,
793
inv_lines, check_content=False)
795
def _create_inv_from_null(self, delta, revision_id):
796
"""This will mutate new_inv directly.
798
This is a simplified form of create_by_apply_delta which knows that all
799
the old values must be None, so everything is a create.
801
serializer = self._format._serializer
802
new_inv = inventory.CHKInventory(serializer.search_key_name)
803
new_inv.revision_id = revision_id
804
entry_to_bytes = new_inv._entry_to_bytes
805
id_to_entry_dict = {}
806
parent_id_basename_dict = {}
807
for old_path, new_path, file_id, entry in delta:
808
if old_path is not None:
809
raise ValueError('Invalid delta, somebody tried to delete %r'
810
' from the NULL_REVISION'
811
% ((old_path, file_id),))
813
raise ValueError('Invalid delta, delta from NULL_REVISION has'
814
' no new_path %r' % (file_id,))
816
new_inv.root_id = file_id
817
parent_id_basename_key = StaticTuple('', '').intern()
819
utf8_entry_name = entry.name.encode('utf-8')
820
parent_id_basename_key = StaticTuple(entry.parent_id,
821
utf8_entry_name).intern()
822
new_value = entry_to_bytes(entry)
824
# new_inv._path_to_fileid_cache[new_path] = file_id
825
key = StaticTuple(file_id).intern()
826
id_to_entry_dict[key] = new_value
827
parent_id_basename_dict[parent_id_basename_key] = file_id
829
new_inv._populate_from_dicts(self.chk_bytes, id_to_entry_dict,
830
parent_id_basename_dict, maximum_size=serializer.maximum_size)
833
def add_inventory_by_delta(self, basis_revision_id, delta, new_revision_id,
834
parents, basis_inv=None, propagate_caches=False):
835
"""Add a new inventory expressed as a delta against another revision.
837
:param basis_revision_id: The inventory id the delta was created
839
:param delta: The inventory delta (see Inventory.apply_delta for
841
:param new_revision_id: The revision id that the inventory is being
843
:param parents: The revision ids of the parents that revision_id is
844
known to have and are in the repository already. These are supplied
845
for repositories that depend on the inventory graph for revision
846
graph access, as well as for those that pun ancestry with delta
848
:param basis_inv: The basis inventory if it is already known,
850
:param propagate_caches: If True, the caches for this inventory are
851
copied to and updated for the result if possible.
853
:returns: (validator, new_inv)
854
The validator(which is a sha1 digest, though what is sha'd is
855
repository format specific) of the serialized inventory, and the
858
if not self.is_in_write_group():
859
raise AssertionError("%r not in write group" % (self,))
860
_mod_revision.check_not_reserved_id(new_revision_id)
862
if basis_inv is None:
863
if basis_revision_id == _mod_revision.NULL_REVISION:
864
new_inv = self._create_inv_from_null(delta, new_revision_id)
865
if new_inv.root_id is None:
866
raise errors.RootMissing()
867
inv_lines = new_inv.to_lines()
868
return self._inventory_add_lines(new_revision_id, parents,
869
inv_lines, check_content=False), new_inv
871
basis_tree = self.revision_tree(basis_revision_id)
872
basis_tree.lock_read()
873
basis_inv = basis_tree.inventory
875
result = basis_inv.create_by_apply_delta(delta, new_revision_id,
876
propagate_caches=propagate_caches)
877
inv_lines = result.to_lines()
878
return self._inventory_add_lines(new_revision_id, parents,
879
inv_lines, check_content=False), result
881
if basis_tree is not None:
884
def _deserialise_inventory(self, revision_id, bytes):
885
return inventory.CHKInventory.deserialise(self.chk_bytes, bytes,
888
def _iter_inventories(self, revision_ids, ordering):
889
"""Iterate over many inventory objects."""
891
ordering = 'unordered'
892
keys = [(revision_id,) for revision_id in revision_ids]
893
stream = self.inventories.get_record_stream(keys, ordering, True)
895
for record in stream:
896
if record.storage_kind != 'absent':
897
texts[record.key] = record.get_bytes_as('fulltext')
899
raise errors.NoSuchRevision(self, record.key)
901
yield inventory.CHKInventory.deserialise(self.chk_bytes, texts[key], key)
903
def _iter_inventory_xmls(self, revision_ids, ordering):
904
# Without a native 'xml' inventory, this method doesn't make sense.
905
# However older working trees, and older bundles want it - so we supply
906
# it allowing _get_inventory_xml to work. Bundles currently use the
907
# serializer directly; this also isn't ideal, but there isn't an xml
908
# iteration interface offered at all for repositories. We could make
909
# _iter_inventory_xmls be part of the contract, even if kept private.
910
inv_to_str = self._serializer.write_inventory_to_string
911
for inv in self.iter_inventories(revision_ids, ordering=ordering):
912
yield inv_to_str(inv), inv.revision_id
914
def _find_present_inventory_keys(self, revision_keys):
915
parent_map = self.inventories.get_parent_map(revision_keys)
916
present_inventory_keys = set(k for k in parent_map)
917
return present_inventory_keys
919
def fileids_altered_by_revision_ids(self, revision_ids, _inv_weave=None):
920
"""Find the file ids and versions affected by revisions.
922
:param revisions: an iterable containing revision ids.
923
:param _inv_weave: The inventory weave from this repository or None.
924
If None, the inventory weave will be opened automatically.
925
:return: a dictionary mapping altered file-ids to an iterable of
926
revision_ids. Each altered file-ids has the exact revision_ids that
927
altered it listed explicitly.
929
rich_root = self.supports_rich_root()
930
bytes_to_info = inventory.CHKInventory._bytes_to_utf8name_key
931
file_id_revisions = {}
932
pb = ui.ui_factory.nested_progress_bar()
934
revision_keys = [(r,) for r in revision_ids]
935
parent_keys = self._find_parent_keys_of_revisions(revision_keys)
936
# TODO: instead of using _find_present_inventory_keys, change the
937
# code paths to allow missing inventories to be tolerated.
938
# However, we only want to tolerate missing parent
939
# inventories, not missing inventories for revision_ids
940
present_parent_inv_keys = self._find_present_inventory_keys(
942
present_parent_inv_ids = set(
943
[k[-1] for k in present_parent_inv_keys])
944
inventories_to_read = set(revision_ids)
945
inventories_to_read.update(present_parent_inv_ids)
946
root_key_info = _build_interesting_key_sets(
947
self, inventories_to_read, present_parent_inv_ids)
948
interesting_root_keys = root_key_info.interesting_root_keys
949
uninteresting_root_keys = root_key_info.uninteresting_root_keys
950
chk_bytes = self.chk_bytes
951
for record, items in chk_map.iter_interesting_nodes(chk_bytes,
952
interesting_root_keys, uninteresting_root_keys,
954
for name, bytes in items:
955
(name_utf8, file_id, revision_id) = bytes_to_info(bytes)
956
# TODO: consider interning file_id, revision_id here, or
957
# pushing that intern() into bytes_to_info()
958
# TODO: rich_root should always be True here, for all
959
# repositories that support chk_bytes
960
if not rich_root and name_utf8 == '':
963
file_id_revisions[file_id].add(revision_id)
965
file_id_revisions[file_id] = set([revision_id])
968
return file_id_revisions
970
def find_text_key_references(self):
971
"""Find the text key references within the repository.
973
:return: A dictionary mapping text keys ((fileid, revision_id) tuples)
974
to whether they were referred to by the inventory of the
975
revision_id that they contain. The inventory texts from all present
976
revision ids are assessed to generate this report.
978
# XXX: Slow version but correct: rewrite as a series of delta
979
# examinations/direct tree traversal. Note that that will require care
980
# as a common node is reachable both from the inventory that added it,
981
# and others afterwards.
982
revision_keys = self.revisions.keys()
984
rich_roots = self.supports_rich_root()
985
pb = ui.ui_factory.nested_progress_bar()
987
all_revs = self.all_revision_ids()
988
total = len(all_revs)
989
for pos, inv in enumerate(self.iter_inventories(all_revs)):
990
pb.update("Finding text references", pos, total)
991
for _, entry in inv.iter_entries():
992
if not rich_roots and entry.file_id == inv.root_id:
994
key = (entry.file_id, entry.revision)
995
result.setdefault(key, False)
996
if entry.revision == inv.revision_id:
1002
def _reconcile_pack(self, collection, packs, extension, revs, pb):
1003
packer = GCCHKReconcilePacker(collection, packs, extension)
1004
return packer.pack(pb)
1006
def _get_source(self, to_format):
1007
"""Return a source for streaming from this repository."""
1008
if self._format._serializer == to_format._serializer:
1009
# We must be exactly the same format, otherwise stuff like the chk
1010
# page layout might be different.
1011
# Actually, this test is just slightly looser than exact so that
1012
# CHK2 <-> 2a transfers will work.
1013
return GroupCHKStreamSource(self, to_format)
1014
return super(CHKInventoryRepository, self)._get_source(to_format)
1017
class GroupCHKStreamSource(KnitPackStreamSource):
1018
"""Used when both the source and target repo are GroupCHK repos."""
1020
def __init__(self, from_repository, to_format):
1021
"""Create a StreamSource streaming from from_repository."""
1022
super(GroupCHKStreamSource, self).__init__(from_repository, to_format)
1023
self._revision_keys = None
1024
self._text_keys = None
1025
self._text_fetch_order = 'groupcompress'
1026
self._chk_id_roots = None
1027
self._chk_p_id_roots = None
1029
def _get_inventory_stream(self, inventory_keys, allow_absent=False):
1030
"""Get a stream of inventory texts.
1032
When this function returns, self._chk_id_roots and self._chk_p_id_roots
1033
should be populated.
1035
self._chk_id_roots = []
1036
self._chk_p_id_roots = []
1037
def _filtered_inv_stream():
1038
id_roots_set = set()
1039
p_id_roots_set = set()
1040
source_vf = self.from_repository.inventories
1041
stream = source_vf.get_record_stream(inventory_keys,
1042
'groupcompress', True)
1043
for record in stream:
1044
if record.storage_kind == 'absent':
1048
raise errors.NoSuchRevision(self, record.key)
1049
bytes = record.get_bytes_as('fulltext')
1050
chk_inv = inventory.CHKInventory.deserialise(None, bytes,
1052
key = chk_inv.id_to_entry.key()
1053
if key not in id_roots_set:
1054
self._chk_id_roots.append(key)
1055
id_roots_set.add(key)
1056
p_id_map = chk_inv.parent_id_basename_to_file_id
1057
if p_id_map is None:
1058
raise AssertionError('Parent id -> file_id map not set')
1059
key = p_id_map.key()
1060
if key not in p_id_roots_set:
1061
p_id_roots_set.add(key)
1062
self._chk_p_id_roots.append(key)
1064
# We have finished processing all of the inventory records, we
1065
# don't need these sets anymore
1066
id_roots_set.clear()
1067
p_id_roots_set.clear()
1068
return ('inventories', _filtered_inv_stream())
1070
def _get_filtered_chk_streams(self, excluded_revision_keys):
1071
self._text_keys = set()
1072
excluded_revision_keys.discard(_mod_revision.NULL_REVISION)
1073
if not excluded_revision_keys:
1074
uninteresting_root_keys = set()
1075
uninteresting_pid_root_keys = set()
1077
# filter out any excluded revisions whose inventories are not
1079
# TODO: Update Repository.iter_inventories() to add
1080
# ignore_missing=True
1081
present_keys = self.from_repository._find_present_inventory_keys(
1082
excluded_revision_keys)
1083
present_ids = [k[-1] for k in present_keys]
1084
uninteresting_root_keys = set()
1085
uninteresting_pid_root_keys = set()
1086
for inv in self.from_repository.iter_inventories(present_ids):
1087
uninteresting_root_keys.add(inv.id_to_entry.key())
1088
uninteresting_pid_root_keys.add(
1089
inv.parent_id_basename_to_file_id.key())
1090
chk_bytes = self.from_repository.chk_bytes
1091
def _filter_id_to_entry():
1092
interesting_nodes = chk_map.iter_interesting_nodes(chk_bytes,
1093
self._chk_id_roots, uninteresting_root_keys)
1094
for record in _filter_text_keys(interesting_nodes, self._text_keys,
1095
chk_map._bytes_to_text_key):
1096
if record is not None:
1099
self._chk_id_roots = None
1100
yield 'chk_bytes', _filter_id_to_entry()
1101
def _get_parent_id_basename_to_file_id_pages():
1102
for record, items in chk_map.iter_interesting_nodes(chk_bytes,
1103
self._chk_p_id_roots, uninteresting_pid_root_keys):
1104
if record is not None:
1107
self._chk_p_id_roots = None
1108
yield 'chk_bytes', _get_parent_id_basename_to_file_id_pages()
1110
def get_stream(self, search):
1111
def wrap_and_count(pb, rc, stream):
1112
"""Yield records from stream while showing progress."""
1114
for record in stream:
1115
if count == rc.STEP:
1117
pb.update('Estimate', rc.current, rc.max)
1122
revision_ids = search.get_keys()
1123
pb = ui.ui_factory.nested_progress_bar()
1124
rc = self._record_counter
1125
self._record_counter.setup(len(revision_ids))
1126
for stream_info in self._fetch_revision_texts(revision_ids):
1127
yield (stream_info[0],
1128
wrap_and_count(pb, rc, stream_info[1]))
1129
self._revision_keys = [(rev_id,) for rev_id in revision_ids]
1130
self.from_repository.revisions.clear_cache()
1131
self.from_repository.signatures.clear_cache()
1132
s = self._get_inventory_stream(self._revision_keys)
1133
yield (s[0], wrap_and_count(pb, rc, s[1]))
1134
self.from_repository.inventories.clear_cache()
1135
# TODO: The keys to exclude might be part of the search recipe
1136
# For now, exclude all parents that are at the edge of ancestry, for
1137
# which we have inventories
1138
from_repo = self.from_repository
1139
parent_keys = from_repo._find_parent_keys_of_revisions(
1140
self._revision_keys)
1141
for stream_info in self._get_filtered_chk_streams(parent_keys):
1142
yield (stream_info[0], wrap_and_count(pb, rc, stream_info[1]))
1143
self.from_repository.chk_bytes.clear_cache()
1144
s = self._get_text_stream()
1145
yield (s[0], wrap_and_count(pb, rc, s[1]))
1146
self.from_repository.texts.clear_cache()
1147
pb.update('Done', rc.max, rc.max)
1150
def get_stream_for_missing_keys(self, missing_keys):
1151
# missing keys can only occur when we are byte copying and not
1152
# translating (because translation means we don't send
1153
# unreconstructable deltas ever).
1154
missing_inventory_keys = set()
1155
for key in missing_keys:
1156
if key[0] != 'inventories':
1157
raise AssertionError('The only missing keys we should'
1158
' be filling in are inventory keys, not %s'
1160
missing_inventory_keys.add(key[1:])
1161
if self._chk_id_roots or self._chk_p_id_roots:
1162
raise AssertionError('Cannot call get_stream_for_missing_keys'
1163
' until all of get_stream() has been consumed.')
1164
# Yield the inventory stream, so we can find the chk stream
1165
# Some of the missing_keys will be missing because they are ghosts.
1166
# As such, we can ignore them. The Sink is required to verify there are
1167
# no unavailable texts when the ghost inventories are not filled in.
1168
yield self._get_inventory_stream(missing_inventory_keys,
1170
# We use the empty set for excluded_revision_keys, to make it clear
1171
# that we want to transmit all referenced chk pages.
1172
for stream_info in self._get_filtered_chk_streams(set()):
1176
class _InterestingKeyInfo(object):
1178
self.interesting_root_keys = set()
1179
self.interesting_pid_root_keys = set()
1180
self.uninteresting_root_keys = set()
1181
self.uninteresting_pid_root_keys = set()
1183
def all_interesting(self):
1184
return self.interesting_root_keys.union(self.interesting_pid_root_keys)
1186
def all_uninteresting(self):
1187
return self.uninteresting_root_keys.union(
1188
self.uninteresting_pid_root_keys)
1191
return self.all_interesting().union(self.all_uninteresting())
1194
def _build_interesting_key_sets(repo, inventory_ids, parent_only_inv_ids):
1195
result = _InterestingKeyInfo()
1196
for inv in repo.iter_inventories(inventory_ids, 'unordered'):
1197
root_key = inv.id_to_entry.key()
1198
pid_root_key = inv.parent_id_basename_to_file_id.key()
1199
if inv.revision_id in parent_only_inv_ids:
1200
result.uninteresting_root_keys.add(root_key)
1201
result.uninteresting_pid_root_keys.add(pid_root_key)
1203
result.interesting_root_keys.add(root_key)
1204
result.interesting_pid_root_keys.add(pid_root_key)
1208
def _filter_text_keys(interesting_nodes_iterable, text_keys, bytes_to_text_key):
1209
"""Iterate the result of iter_interesting_nodes, yielding the records
1210
and adding to text_keys.
1212
text_keys_update = text_keys.update
1213
for record, items in interesting_nodes_iterable:
1214
text_keys_update([bytes_to_text_key(b) for n,b in items])
1220
class RepositoryFormatCHK1(RepositoryFormatPack):
1221
"""A hashed CHK+group compress pack repository."""
1223
repository_class = CHKInventoryRepository
1224
supports_external_lookups = True
1225
supports_chks = True
1226
# For right now, setting this to True gives us InterModel1And2 rather
1227
# than InterDifferingSerializer
1228
_commit_builder_class = PackRootCommitBuilder
1229
rich_root_data = True
1230
_serializer = chk_serializer.chk_serializer_255_bigpage
1231
_commit_inv_deltas = True
1232
# What index classes to use
1233
index_builder_class = BTreeBuilder
1234
index_class = BTreeGraphIndex
1235
# Note: We cannot unpack a delta that references a text we haven't
1236
# seen yet. There are 2 options, work in fulltexts, or require
1237
# topological sorting. Using fulltexts is more optimal for local
1238
# operations, because the source can be smart about extracting
1239
# multiple in-a-row (and sharing strings). Topological is better
1240
# for remote, because we access less data.
1241
_fetch_order = 'unordered'
1242
_fetch_uses_deltas = False # essentially ignored by the groupcompress code.
1244
pack_compresses = True
1246
def _get_matching_bzrdir(self):
1247
return bzrdir.format_registry.make_bzrdir('development6-rich-root')
1249
def _ignore_setting_bzrdir(self, format):
1252
_matchingbzrdir = property(_get_matching_bzrdir, _ignore_setting_bzrdir)
1254
def get_format_string(self):
1255
"""See RepositoryFormat.get_format_string()."""
1256
return ('Bazaar development format - group compression and chk inventory'
1257
' (needs bzr.dev from 1.14)\n')
1259
def get_format_description(self):
1260
"""See RepositoryFormat.get_format_description()."""
1261
return ("Development repository format - rich roots, group compression"
1262
" and chk inventories")
1265
class RepositoryFormatCHK2(RepositoryFormatCHK1):
1266
"""A CHK repository that uses the bencode revision serializer."""
1268
_serializer = chk_serializer.chk_bencode_serializer
1270
def _get_matching_bzrdir(self):
1271
return bzrdir.format_registry.make_bzrdir('development7-rich-root')
1273
def _ignore_setting_bzrdir(self, format):
1276
_matchingbzrdir = property(_get_matching_bzrdir, _ignore_setting_bzrdir)
1278
def get_format_string(self):
1279
"""See RepositoryFormat.get_format_string()."""
1280
return ('Bazaar development format - chk repository with bencode '
1281
'revision serialization (needs bzr.dev from 1.16)\n')
1284
class RepositoryFormat2a(RepositoryFormatCHK2):
1285
"""A CHK repository that uses the bencode revision serializer.
1287
This is the same as RepositoryFormatCHK2 but with a public name.
1290
_serializer = chk_serializer.chk_bencode_serializer
1292
def _get_matching_bzrdir(self):
1293
return bzrdir.format_registry.make_bzrdir('2a')
1295
def _ignore_setting_bzrdir(self, format):
1298
_matchingbzrdir = property(_get_matching_bzrdir, _ignore_setting_bzrdir)
1300
def get_format_string(self):
1301
return ('Bazaar repository format 2a (needs bzr 1.16 or later)\n')
1303
def get_format_description(self):
1304
"""See RepositoryFormat.get_format_description()."""
1305
return ("Repository format 2a - rich roots, group compression"
1306
" and chk inventories")
added
removed
bzrlib/repofmt/groupcompress_repo.py
