1
# Copyright (C) 2005, 2006, 2007, 2008 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
20
from bzrlib.lazy_import import lazy_import
21
lazy_import(globals(), """
22
from itertools import izip
37
from bzrlib.index import (
39
GraphIndexPrefixAdapter,
41
from bzrlib.knit import (
47
from bzrlib import tsort
54
revision as _mod_revision,
57
from bzrlib.decorators import needs_write_lock
58
from bzrlib.btree_index import (
62
from bzrlib.index import (
66
from bzrlib.repofmt.knitrepo import KnitRepository
67
from bzrlib.repository import (
69
MetaDirRepositoryFormat,
74
from bzrlib.trace import (
80
class PackCommitBuilder(CommitBuilder):
81
"""A subclass of CommitBuilder to add texts with pack semantics.
83
Specifically this uses one knit object rather than one knit object per
84
added text, reducing memory and object pressure.
87
def __init__(self, repository, parents, config, timestamp=None,
88
timezone=None, committer=None, revprops=None,
90
CommitBuilder.__init__(self, repository, parents, config,
91
timestamp=timestamp, timezone=timezone, committer=committer,
92
revprops=revprops, revision_id=revision_id)
93
self._file_graph = graph.Graph(
94
repository._pack_collection.text_index.combined_index)
96
def _heads(self, file_id, revision_ids):
97
keys = [(file_id, revision_id) for revision_id in revision_ids]
98
return set([key[1] for key in self._file_graph.heads(keys)])
101
class PackRootCommitBuilder(RootCommitBuilder):
102
"""A subclass of RootCommitBuilder to add texts with pack semantics.
104
Specifically this uses one knit object rather than one knit object per
105
added text, reducing memory and object pressure.
108
def __init__(self, repository, parents, config, timestamp=None,
109
timezone=None, committer=None, revprops=None,
111
CommitBuilder.__init__(self, repository, parents, config,
112
timestamp=timestamp, timezone=timezone, committer=committer,
113
revprops=revprops, revision_id=revision_id)
114
self._file_graph = graph.Graph(
115
repository._pack_collection.text_index.combined_index)
117
def _heads(self, file_id, revision_ids):
118
keys = [(file_id, revision_id) for revision_id in revision_ids]
119
return set([key[1] for key in self._file_graph.heads(keys)])
123
"""An in memory proxy for a pack and its indices.
125
This is a base class that is not directly used, instead the classes
126
ExistingPack and NewPack are used.
129
# A map of index 'type' to the file extension and position in the
131
index_definitions = {
133
'revision': ('.rix', 0),
134
'inventory': ('.iix', 1),
136
'signature': ('.six', 3),
139
def __init__(self, revision_index, inventory_index, text_index,
140
signature_index, chk_index=None):
141
"""Create a pack instance.
143
:param revision_index: A GraphIndex for determining what revisions are
144
present in the Pack and accessing the locations of their texts.
145
:param inventory_index: A GraphIndex for determining what inventories are
146
present in the Pack and accessing the locations of their
148
:param text_index: A GraphIndex for determining what file texts
149
are present in the pack and accessing the locations of their
150
texts/deltas (via (fileid, revisionid) tuples).
151
:param signature_index: A GraphIndex for determining what signatures are
152
present in the Pack and accessing the locations of their texts.
153
:param chk_index: A GraphIndex for accessing content by CHK, if the
156
self.revision_index = revision_index
157
self.inventory_index = inventory_index
158
self.text_index = text_index
159
self.signature_index = signature_index
160
self.chk_index = chk_index
162
def access_tuple(self):
163
"""Return a tuple (transport, name) for the pack content."""
164
return self.pack_transport, self.file_name()
166
def _check_references(self):
167
"""Make sure our external references are present.
169
Packs are allowed to have deltas whose base is not in the pack, but it
170
must be present somewhere in this collection. It is not allowed to
171
have deltas based on a fallback repository.
172
(See <https://bugs.launchpad.net/bzr/+bug/288751>)
175
for (index_name, external_refs, index) in [
177
self._get_external_refs(self.text_index),
178
self._pack_collection.text_index.combined_index),
180
self._get_external_refs(self.inventory_index),
181
self._pack_collection.inventory_index.combined_index),
183
missing = external_refs.difference(
184
k for (idx, k, v, r) in
185
index.iter_entries(external_refs))
187
missing_items[index_name] = sorted(list(missing))
189
from pprint import pformat
190
raise errors.BzrCheckError(
191
"Newly created pack file %r has delta references to "
192
"items not in its repository:\n%s"
193
% (self, pformat(missing_items)))
196
"""Get the file name for the pack on disk."""
197
return self.name + '.pack'
199
def get_revision_count(self):
200
return self.revision_index.key_count()
202
def index_name(self, index_type, name):
203
"""Get the disk name of an index type for pack name 'name'."""
204
return name + Pack.index_definitions[index_type][0]
206
def index_offset(self, index_type):
207
"""Get the position in a index_size array for a given index type."""
208
return Pack.index_definitions[index_type][1]
210
def inventory_index_name(self, name):
211
"""The inv index is the name + .iix."""
212
return self.index_name('inventory', name)
214
def revision_index_name(self, name):
215
"""The revision index is the name + .rix."""
216
return self.index_name('revision', name)
218
def signature_index_name(self, name):
219
"""The signature index is the name + .six."""
220
return self.index_name('signature', name)
222
def text_index_name(self, name):
223
"""The text index is the name + .tix."""
224
return self.index_name('text', name)
226
def _replace_index_with_readonly(self, index_type):
227
setattr(self, index_type + '_index',
228
self.index_class(self.index_transport,
229
self.index_name(index_type, self.name),
230
self.index_sizes[self.index_offset(index_type)]))
233
class ExistingPack(Pack):
234
"""An in memory proxy for an existing .pack and its disk indices."""
236
def __init__(self, pack_transport, name, revision_index, inventory_index,
237
text_index, signature_index, chk_index=None):
238
"""Create an ExistingPack object.
240
:param pack_transport: The transport where the pack file resides.
241
:param name: The name of the pack on disk in the pack_transport.
243
Pack.__init__(self, revision_index, inventory_index, text_index,
244
signature_index, chk_index)
246
self.pack_transport = pack_transport
247
if None in (revision_index, inventory_index, text_index,
248
signature_index, name, pack_transport):
249
raise AssertionError()
251
def __eq__(self, other):
252
return self.__dict__ == other.__dict__
254
def __ne__(self, other):
255
return not self.__eq__(other)
258
return "<%s.%s object at 0x%x, %s, %s" % (
259
self.__class__.__module__, self.__class__.__name__, id(self),
260
self.pack_transport, self.name)
263
class ResumedPack(ExistingPack):
265
def __init__(self, name, revision_index, inventory_index, text_index,
266
signature_index, upload_transport, pack_transport, index_transport,
267
pack_collection, chk_index=None):
268
"""Create a ResumedPack object."""
269
ExistingPack.__init__(self, pack_transport, name, revision_index,
270
inventory_index, text_index, signature_index,
272
self.upload_transport = upload_transport
273
self.index_transport = index_transport
274
self.index_sizes = [None, None, None, None]
276
('revision', revision_index),
277
('inventory', inventory_index),
278
('text', text_index),
279
('signature', signature_index),
281
if chk_index is not None:
282
indices.append(('chk', chk_index))
283
self.index_sizes.append(None)
284
for index_type, index in indices:
285
offset = self.index_offset(index_type)
286
self.index_sizes[offset] = index._size
287
self.index_class = pack_collection._index_class
288
self._pack_collection = pack_collection
289
self._state = 'resumed'
290
# XXX: perhaps check that the .pack file exists?
292
def access_tuple(self):
293
if self._state == 'finished':
294
return Pack.access_tuple(self)
295
elif self._state == 'resumed':
296
return self.upload_transport, self.file_name()
298
raise AssertionError(self._state)
301
self.upload_transport.delete(self.file_name())
302
indices = [self.revision_index, self.inventory_index, self.text_index,
303
self.signature_index]
304
if self.chk_index is not None:
305
indices.append(self.chk_index)
306
for index in indices:
307
index._transport.delete(index._name)
310
self._check_references()
311
index_types = ['revision', 'inventory', 'text', 'signature']
312
if self.chk_index is not None:
313
index_types.append('chk')
314
for index_type in index_types:
315
old_name = self.index_name(index_type, self.name)
316
new_name = '../indices/' + old_name
317
self.upload_transport.rename(old_name, new_name)
318
self._replace_index_with_readonly(index_type)
319
new_name = '../packs/' + self.file_name()
320
self.upload_transport.rename(self.file_name(), new_name)
321
self._state = 'finished'
323
def _get_external_refs(self, index):
324
"""Return compression parents for this index that are not present.
326
This returns any compression parents that are referenced by this index,
327
which are not contained *in* this index. They may be present elsewhere.
329
return index.external_references(1)
333
"""An in memory proxy for a pack which is being created."""
335
def __init__(self, pack_collection, upload_suffix='', file_mode=None):
336
"""Create a NewPack instance.
338
:param pack_collection: A PackCollection into which this is being inserted.
339
:param upload_suffix: An optional suffix to be given to any temporary
340
files created during the pack creation. e.g '.autopack'
341
:param file_mode: Unix permissions for newly created file.
343
# The relative locations of the packs are constrained, but all are
344
# passed in because the caller has them, so as to avoid object churn.
345
index_builder_class = pack_collection._index_builder_class
346
if pack_collection.chk_index is not None:
347
chk_index = index_builder_class(reference_lists=0)
351
# Revisions: parents list, no text compression.
352
index_builder_class(reference_lists=1),
353
# Inventory: We want to map compression only, but currently the
354
# knit code hasn't been updated enough to understand that, so we
355
# have a regular 2-list index giving parents and compression
357
index_builder_class(reference_lists=2),
358
# Texts: compression and per file graph, for all fileids - so two
359
# reference lists and two elements in the key tuple.
360
index_builder_class(reference_lists=2, key_elements=2),
361
# Signatures: Just blobs to store, no compression, no parents
363
index_builder_class(reference_lists=0),
364
# CHK based storage - just blobs, no compression or parents.
367
self._pack_collection = pack_collection
368
# When we make readonly indices, we need this.
369
self.index_class = pack_collection._index_class
370
# where should the new pack be opened
371
self.upload_transport = pack_collection._upload_transport
372
# where are indices written out to
373
self.index_transport = pack_collection._index_transport
374
# where is the pack renamed to when it is finished?
375
self.pack_transport = pack_collection._pack_transport
376
# What file mode to upload the pack and indices with.
377
self._file_mode = file_mode
378
# tracks the content written to the .pack file.
379
self._hash = osutils.md5()
380
# a tuple with the length in bytes of the indices, once the pack
381
# is finalised. (rev, inv, text, sigs, chk_if_in_use)
382
self.index_sizes = None
383
# How much data to cache when writing packs. Note that this is not
384
# synchronised with reads, because it's not in the transport layer, so
385
# is not safe unless the client knows it won't be reading from the pack
387
self._cache_limit = 0
388
# the temporary pack file name.
389
self.random_name = osutils.rand_chars(20) + upload_suffix
390
# when was this pack started ?
391
self.start_time = time.time()
392
# open an output stream for the data added to the pack.
393
self.write_stream = self.upload_transport.open_write_stream(
394
self.random_name, mode=self._file_mode)
395
if 'pack' in debug.debug_flags:
396
mutter('%s: create_pack: pack stream open: %s%s t+%6.3fs',
397
time.ctime(), self.upload_transport.base, self.random_name,
398
time.time() - self.start_time)
399
# A list of byte sequences to be written to the new pack, and the
400
# aggregate size of them. Stored as a list rather than separate
401
# variables so that the _write_data closure below can update them.
402
self._buffer = [[], 0]
403
# create a callable for adding data
405
# robertc says- this is a closure rather than a method on the object
406
# so that the variables are locals, and faster than accessing object
408
def _write_data(bytes, flush=False, _buffer=self._buffer,
409
_write=self.write_stream.write, _update=self._hash.update):
410
_buffer[0].append(bytes)
411
_buffer[1] += len(bytes)
413
if _buffer[1] > self._cache_limit or flush:
414
bytes = ''.join(_buffer[0])
418
# expose this on self, for the occasion when clients want to add data.
419
self._write_data = _write_data
420
# a pack writer object to serialise pack records.
421
self._writer = pack.ContainerWriter(self._write_data)
423
# what state is the pack in? (open, finished, aborted)
427
"""Cancel creating this pack."""
428
self._state = 'aborted'
429
self.write_stream.close()
430
# Remove the temporary pack file.
431
self.upload_transport.delete(self.random_name)
432
# The indices have no state on disk.
434
def access_tuple(self):
435
"""Return a tuple (transport, name) for the pack content."""
436
if self._state == 'finished':
437
return Pack.access_tuple(self)
438
elif self._state == 'open':
439
return self.upload_transport, self.random_name
441
raise AssertionError(self._state)
443
def data_inserted(self):
444
"""True if data has been added to this pack."""
445
return bool(self.get_revision_count() or
446
self.inventory_index.key_count() or
447
self.text_index.key_count() or
448
self.signature_index.key_count() or
449
(self.chk_index is not None and self.chk_index.key_count()))
451
def finish(self, suspend=False):
452
"""Finish the new pack.
455
- finalises the content
456
- assigns a name (the md5 of the content, currently)
457
- writes out the associated indices
458
- renames the pack into place.
459
- stores the index size tuple for the pack in the index_sizes
464
self._write_data('', flush=True)
465
self.name = self._hash.hexdigest()
467
self._check_references()
469
# XXX: It'd be better to write them all to temporary names, then
470
# rename them all into place, so that the window when only some are
471
# visible is smaller. On the other hand none will be seen until
472
# they're in the names list.
473
self.index_sizes = [None, None, None, None]
474
self._write_index('revision', self.revision_index, 'revision', suspend)
475
self._write_index('inventory', self.inventory_index, 'inventory',
477
self._write_index('text', self.text_index, 'file texts', suspend)
478
self._write_index('signature', self.signature_index,
479
'revision signatures', suspend)
480
if self.chk_index is not None:
481
self.index_sizes.append(None)
482
self._write_index('chk', self.chk_index,
483
'content hash bytes', suspend)
484
self.write_stream.close()
485
# Note that this will clobber an existing pack with the same name,
486
# without checking for hash collisions. While this is undesirable this
487
# is something that can be rectified in a subsequent release. One way
488
# to rectify it may be to leave the pack at the original name, writing
489
# its pack-names entry as something like 'HASH: index-sizes
490
# temporary-name'. Allocate that and check for collisions, if it is
491
# collision free then rename it into place. If clients know this scheme
492
# they can handle missing-file errors by:
493
# - try for HASH.pack
494
# - try for temporary-name
495
# - refresh the pack-list to see if the pack is now absent
496
new_name = self.name + '.pack'
498
new_name = '../packs/' + new_name
499
self.upload_transport.rename(self.random_name, new_name)
500
self._state = 'finished'
501
if 'pack' in debug.debug_flags:
502
# XXX: size might be interesting?
503
mutter('%s: create_pack: pack finished: %s%s->%s t+%6.3fs',
504
time.ctime(), self.upload_transport.base, self.random_name,
505
new_name, time.time() - self.start_time)
508
"""Flush any current data."""
510
bytes = ''.join(self._buffer[0])
511
self.write_stream.write(bytes)
512
self._hash.update(bytes)
513
self._buffer[:] = [[], 0]
515
def _get_external_refs(self, index):
516
return index._external_references()
518
def set_write_cache_size(self, size):
519
self._cache_limit = size
521
def _write_index(self, index_type, index, label, suspend=False):
522
"""Write out an index.
524
:param index_type: The type of index to write - e.g. 'revision'.
525
:param index: The index object to serialise.
526
:param label: What label to give the index e.g. 'revision'.
528
index_name = self.index_name(index_type, self.name)
530
transport = self.upload_transport
532
transport = self.index_transport
533
self.index_sizes[self.index_offset(index_type)] = transport.put_file(
534
index_name, index.finish(), mode=self._file_mode)
535
if 'pack' in debug.debug_flags:
536
# XXX: size might be interesting?
537
mutter('%s: create_pack: wrote %s index: %s%s t+%6.3fs',
538
time.ctime(), label, self.upload_transport.base,
539
self.random_name, time.time() - self.start_time)
540
# Replace the writable index on this object with a readonly,
541
# presently unloaded index. We should alter
542
# the index layer to make its finish() error if add_node is
543
# subsequently used. RBC
544
self._replace_index_with_readonly(index_type)
547
class AggregateIndex(object):
548
"""An aggregated index for the RepositoryPackCollection.
550
AggregateIndex is reponsible for managing the PackAccess object,
551
Index-To-Pack mapping, and all indices list for a specific type of index
552
such as 'revision index'.
554
A CombinedIndex provides an index on a single key space built up
555
from several on-disk indices. The AggregateIndex builds on this
556
to provide a knit access layer, and allows having up to one writable
557
index within the collection.
559
# XXX: Probably 'can be written to' could/should be separated from 'acts
560
# like a knit index' -- mbp 20071024
562
def __init__(self, reload_func=None, flush_func=None):
563
"""Create an AggregateIndex.
565
:param reload_func: A function to call if we find we are missing an
566
index. Should have the form reload_func() => True if the list of
567
active pack files has changed.
569
self._reload_func = reload_func
570
self.index_to_pack = {}
571
self.combined_index = CombinedGraphIndex([], reload_func=reload_func)
572
self.data_access = _DirectPackAccess(self.index_to_pack,
573
reload_func=reload_func,
574
flush_func=flush_func)
575
self.add_callback = None
577
def replace_indices(self, index_to_pack, indices):
578
"""Replace the current mappings with fresh ones.
580
This should probably not be used eventually, rather incremental add and
581
removal of indices. It has been added during refactoring of existing
584
:param index_to_pack: A mapping from index objects to
585
(transport, name) tuples for the pack file data.
586
:param indices: A list of indices.
588
# refresh the revision pack map dict without replacing the instance.
589
self.index_to_pack.clear()
590
self.index_to_pack.update(index_to_pack)
591
# XXX: API break - clearly a 'replace' method would be good?
592
self.combined_index._indices[:] = indices
593
# the current add nodes callback for the current writable index if
595
self.add_callback = None
597
def add_index(self, index, pack):
598
"""Add index to the aggregate, which is an index for Pack pack.
600
Future searches on the aggregate index will seach this new index
601
before all previously inserted indices.
603
:param index: An Index for the pack.
604
:param pack: A Pack instance.
606
# expose it to the index map
607
self.index_to_pack[index] = pack.access_tuple()
608
# put it at the front of the linear index list
609
self.combined_index.insert_index(0, index)
611
def add_writable_index(self, index, pack):
612
"""Add an index which is able to have data added to it.
614
There can be at most one writable index at any time. Any
615
modifications made to the knit are put into this index.
617
:param index: An index from the pack parameter.
618
:param pack: A Pack instance.
620
if self.add_callback is not None:
621
raise AssertionError(
622
"%s already has a writable index through %s" % \
623
(self, self.add_callback))
624
# allow writing: queue writes to a new index
625
self.add_index(index, pack)
626
# Updates the index to packs mapping as a side effect,
627
self.data_access.set_writer(pack._writer, index, pack.access_tuple())
628
self.add_callback = index.add_nodes
631
"""Reset all the aggregate data to nothing."""
632
self.data_access.set_writer(None, None, (None, None))
633
self.index_to_pack.clear()
634
del self.combined_index._indices[:]
635
self.add_callback = None
637
def remove_index(self, index, pack):
638
"""Remove index from the indices used to answer queries.
640
:param index: An index from the pack parameter.
641
:param pack: A Pack instance.
643
del self.index_to_pack[index]
644
self.combined_index._indices.remove(index)
645
if (self.add_callback is not None and
646
getattr(index, 'add_nodes', None) == self.add_callback):
647
self.add_callback = None
648
self.data_access.set_writer(None, None, (None, None))
651
class Packer(object):
652
"""Create a pack from packs."""
654
def __init__(self, pack_collection, packs, suffix, revision_ids=None,
658
:param pack_collection: A RepositoryPackCollection object where the
659
new pack is being written to.
660
:param packs: The packs to combine.
661
:param suffix: The suffix to use on the temporary files for the pack.
662
:param revision_ids: Revision ids to limit the pack to.
663
:param reload_func: A function to call if a pack file/index goes
664
missing. The side effect of calling this function should be to
665
update self.packs. See also AggregateIndex
669
self.revision_ids = revision_ids
670
# The pack object we are creating.
672
self._pack_collection = pack_collection
673
self._reload_func = reload_func
674
# The index layer keys for the revisions being copied. None for 'all
676
self._revision_keys = None
677
# What text keys to copy. None for 'all texts'. This is set by
678
# _copy_inventory_texts
679
self._text_filter = None
682
def _extra_init(self):
683
"""A template hook to allow extending the constructor trivially."""
685
def _pack_map_and_index_list(self, index_attribute):
686
"""Convert a list of packs to an index pack map and index list.
688
:param index_attribute: The attribute that the desired index is found
690
:return: A tuple (map, list) where map contains the dict from
691
index:pack_tuple, and list contains the indices in the preferred
696
for pack_obj in self.packs:
697
index = getattr(pack_obj, index_attribute)
698
indices.append(index)
699
pack_map[index] = pack_obj
700
return pack_map, indices
702
def _index_contents(self, indices, key_filter=None):
703
"""Get an iterable of the index contents from a pack_map.
705
:param indices: The list of indices to query
706
:param key_filter: An optional filter to limit the keys returned.
708
all_index = CombinedGraphIndex(indices)
709
if key_filter is None:
710
return all_index.iter_all_entries()
712
return all_index.iter_entries(key_filter)
714
def pack(self, pb=None):
715
"""Create a new pack by reading data from other packs.
717
This does little more than a bulk copy of data. One key difference
718
is that data with the same item key across multiple packs is elided
719
from the output. The new pack is written into the current pack store
720
along with its indices, and the name added to the pack names. The
721
source packs are not altered and are not required to be in the current
724
:param pb: An optional progress bar to use. A nested bar is created if
726
:return: A Pack object, or None if nothing was copied.
728
# open a pack - using the same name as the last temporary file
729
# - which has already been flushed, so its safe.
730
# XXX: - duplicate code warning with start_write_group; fix before
731
# considering 'done'.
732
if self._pack_collection._new_pack is not None:
733
raise errors.BzrError('call to %s.pack() while another pack is'
735
% (self.__class__.__name__,))
736
if self.revision_ids is not None:
737
if len(self.revision_ids) == 0:
738
# silly fetch request.
741
self.revision_ids = frozenset(self.revision_ids)
742
self.revision_keys = frozenset((revid,) for revid in
745
self.pb = ui.ui_factory.nested_progress_bar()
749
return self._create_pack_from_packs()
755
"""Open a pack for the pack we are creating."""
756
new_pack = self._pack_collection.pack_factory(self._pack_collection,
757
upload_suffix=self.suffix,
758
file_mode=self._pack_collection.repo.bzrdir._get_file_mode())
759
# We know that we will process all nodes in order, and don't need to
760
# query, so don't combine any indices spilled to disk until we are done
761
new_pack.revision_index.set_optimize(combine_backing_indices=False)
762
new_pack.inventory_index.set_optimize(combine_backing_indices=False)
763
new_pack.text_index.set_optimize(combine_backing_indices=False)
764
new_pack.signature_index.set_optimize(combine_backing_indices=False)
767
def _update_pack_order(self, entries, index_to_pack_map):
768
"""Determine how we want our packs to be ordered.
770
This changes the sort order of the self.packs list so that packs unused
771
by 'entries' will be at the end of the list, so that future requests
772
can avoid probing them. Used packs will be at the front of the
773
self.packs list, in the order of their first use in 'entries'.
775
:param entries: A list of (index, ...) tuples
776
:param index_to_pack_map: A mapping from index objects to pack objects.
780
for entry in entries:
782
if index not in seen_indexes:
783
packs.append(index_to_pack_map[index])
784
seen_indexes.add(index)
785
if len(packs) == len(self.packs):
786
if 'pack' in debug.debug_flags:
787
mutter('Not changing pack list, all packs used.')
789
seen_packs = set(packs)
790
for pack in self.packs:
791
if pack not in seen_packs:
794
if 'pack' in debug.debug_flags:
795
old_names = [p.access_tuple()[1] for p in self.packs]
796
new_names = [p.access_tuple()[1] for p in packs]
797
mutter('Reordering packs\nfrom: %s\n to: %s',
798
old_names, new_names)
801
def _copy_revision_texts(self):
802
"""Copy revision data to the new pack."""
804
if self.revision_ids:
805
revision_keys = [(revision_id,) for revision_id in self.revision_ids]
808
# select revision keys
809
revision_index_map, revision_indices = self._pack_map_and_index_list(
811
revision_nodes = self._index_contents(revision_indices, revision_keys)
812
revision_nodes = list(revision_nodes)
813
self._update_pack_order(revision_nodes, revision_index_map)
814
# copy revision keys and adjust values
815
self.pb.update("Copying revision texts", 1)
816
total_items, readv_group_iter = self._revision_node_readv(revision_nodes)
817
list(self._copy_nodes_graph(revision_index_map, self.new_pack._writer,
818
self.new_pack.revision_index, readv_group_iter, total_items))
819
if 'pack' in debug.debug_flags:
820
mutter('%s: create_pack: revisions copied: %s%s %d items t+%6.3fs',
821
time.ctime(), self._pack_collection._upload_transport.base,
822
self.new_pack.random_name,
823
self.new_pack.revision_index.key_count(),
824
time.time() - self.new_pack.start_time)
825
self._revision_keys = revision_keys
827
def _copy_inventory_texts(self):
828
"""Copy the inventory texts to the new pack.
830
self._revision_keys is used to determine what inventories to copy.
832
Sets self._text_filter appropriately.
834
# select inventory keys
835
inv_keys = self._revision_keys # currently the same keyspace, and note that
836
# querying for keys here could introduce a bug where an inventory item
837
# is missed, so do not change it to query separately without cross
838
# checking like the text key check below.
839
inventory_index_map, inventory_indices = self._pack_map_and_index_list(
841
inv_nodes = self._index_contents(inventory_indices, inv_keys)
842
# copy inventory keys and adjust values
843
# XXX: Should be a helper function to allow different inv representation
845
self.pb.update("Copying inventory texts", 2)
846
total_items, readv_group_iter = self._least_readv_node_readv(inv_nodes)
847
# Only grab the output lines if we will be processing them
848
output_lines = bool(self.revision_ids)
849
inv_lines = self._copy_nodes_graph(inventory_index_map,
850
self.new_pack._writer, self.new_pack.inventory_index,
851
readv_group_iter, total_items, output_lines=output_lines)
852
if self.revision_ids:
853
self._process_inventory_lines(inv_lines)
855
# eat the iterator to cause it to execute.
857
self._text_filter = None
858
if 'pack' in debug.debug_flags:
859
mutter('%s: create_pack: inventories copied: %s%s %d items t+%6.3fs',
860
time.ctime(), self._pack_collection._upload_transport.base,
861
self.new_pack.random_name,
862
self.new_pack.inventory_index.key_count(),
863
time.time() - self.new_pack.start_time)
865
def _copy_text_texts(self):
867
text_index_map, text_nodes = self._get_text_nodes()
868
if self._text_filter is not None:
869
# We could return the keys copied as part of the return value from
870
# _copy_nodes_graph but this doesn't work all that well with the
871
# need to get line output too, so we check separately, and as we're
872
# going to buffer everything anyway, we check beforehand, which
873
# saves reading knit data over the wire when we know there are
875
text_nodes = set(text_nodes)
876
present_text_keys = set(_node[1] for _node in text_nodes)
877
missing_text_keys = set(self._text_filter) - present_text_keys
878
if missing_text_keys:
879
# TODO: raise a specific error that can handle many missing
881
mutter("missing keys during fetch: %r", missing_text_keys)
882
a_missing_key = missing_text_keys.pop()
883
raise errors.RevisionNotPresent(a_missing_key[1],
885
# copy text keys and adjust values
886
self.pb.update("Copying content texts", 3)
887
total_items, readv_group_iter = self._least_readv_node_readv(text_nodes)
888
list(self._copy_nodes_graph(text_index_map, self.new_pack._writer,
889
self.new_pack.text_index, readv_group_iter, total_items))
890
self._log_copied_texts()
892
def _create_pack_from_packs(self):
893
self.pb.update("Opening pack", 0, 5)
894
self.new_pack = self.open_pack()
895
new_pack = self.new_pack
896
# buffer data - we won't be reading-back during the pack creation and
897
# this makes a significant difference on sftp pushes.
898
new_pack.set_write_cache_size(1024*1024)
899
if 'pack' in debug.debug_flags:
900
plain_pack_list = ['%s%s' % (a_pack.pack_transport.base, a_pack.name)
901
for a_pack in self.packs]
902
if self.revision_ids is not None:
903
rev_count = len(self.revision_ids)
906
mutter('%s: create_pack: creating pack from source packs: '
907
'%s%s %s revisions wanted %s t=0',
908
time.ctime(), self._pack_collection._upload_transport.base, new_pack.random_name,
909
plain_pack_list, rev_count)
910
self._copy_revision_texts()
911
self._copy_inventory_texts()
912
self._copy_text_texts()
913
# select signature keys
914
signature_filter = self._revision_keys # same keyspace
915
signature_index_map, signature_indices = self._pack_map_and_index_list(
917
signature_nodes = self._index_contents(signature_indices,
919
# copy signature keys and adjust values
920
self.pb.update("Copying signature texts", 4)
921
self._copy_nodes(signature_nodes, signature_index_map, new_pack._writer,
922
new_pack.signature_index)
923
if 'pack' in debug.debug_flags:
924
mutter('%s: create_pack: revision signatures copied: %s%s %d items t+%6.3fs',
925
time.ctime(), self._pack_collection._upload_transport.base, new_pack.random_name,
926
new_pack.signature_index.key_count(),
927
time.time() - new_pack.start_time)
929
# NB XXX: how to check CHK references are present? perhaps by yielding
930
# the items? How should that interact with stacked repos?
931
if new_pack.chk_index is not None:
933
if 'pack' in debug.debug_flags:
934
mutter('%s: create_pack: chk content copied: %s%s %d items t+%6.3fs',
935
time.ctime(), self._pack_collection._upload_transport.base,
936
new_pack.random_name,
937
new_pack.chk_index.key_count(),
938
time.time() - new_pack.start_time)
939
new_pack._check_references()
940
if not self._use_pack(new_pack):
943
self.pb.update("Finishing pack", 5)
945
self._pack_collection.allocate(new_pack)
948
def _copy_chks(self, refs=None):
949
# XXX: Todo, recursive follow-pointers facility when fetching some
951
chk_index_map, chk_indices = self._pack_map_and_index_list(
953
chk_nodes = self._index_contents(chk_indices, refs)
955
# TODO: This isn't strictly tasteful as we are accessing some private
956
# variables (_serializer). Perhaps a better way would be to have
957
# Repository._deserialise_chk_node()
958
search_key_func = chk_map.search_key_registry.get(
959
self._pack_collection.repo._serializer.search_key_name)
960
def accumlate_refs(lines):
961
# XXX: move to a generic location
963
bytes = ''.join(lines)
964
node = chk_map._deserialise(bytes, ("unknown",), search_key_func)
965
new_refs.update(node.refs())
966
self._copy_nodes(chk_nodes, chk_index_map, self.new_pack._writer,
967
self.new_pack.chk_index, output_lines=accumlate_refs)
970
def _copy_nodes(self, nodes, index_map, writer, write_index,
972
"""Copy knit nodes between packs with no graph references.
974
:param output_lines: Output full texts of copied items.
976
pb = ui.ui_factory.nested_progress_bar()
978
return self._do_copy_nodes(nodes, index_map, writer,
979
write_index, pb, output_lines=output_lines)
983
def _do_copy_nodes(self, nodes, index_map, writer, write_index, pb,
985
# for record verification
986
knit = KnitVersionedFiles(None, None)
987
# plan a readv on each source pack:
989
nodes = sorted(nodes)
990
# how to map this into knit.py - or knit.py into this?
991
# we don't want the typical knit logic, we want grouping by pack
992
# at this point - perhaps a helper library for the following code
993
# duplication points?
995
for index, key, value in nodes:
996
if index not in request_groups:
997
request_groups[index] = []
998
request_groups[index].append((key, value))
1000
pb.update("Copied record", record_index, len(nodes))
1001
for index, items in request_groups.iteritems():
1002
pack_readv_requests = []
1003
for key, value in items:
1004
# ---- KnitGraphIndex.get_position
1005
bits = value[1:].split(' ')
1006
offset, length = int(bits[0]), int(bits[1])
1007
pack_readv_requests.append((offset, length, (key, value[0])))
1008
# linear scan up the pack
1009
pack_readv_requests.sort()
1011
pack_obj = index_map[index]
1012
transport, path = pack_obj.access_tuple()
1014
reader = pack.make_readv_reader(transport, path,
1015
[offset[0:2] for offset in pack_readv_requests])
1016
except errors.NoSuchFile:
1017
if self._reload_func is not None:
1020
for (names, read_func), (_1, _2, (key, eol_flag)) in \
1021
izip(reader.iter_records(), pack_readv_requests):
1022
raw_data = read_func(None)
1023
# check the header only
1024
if output_lines is not None:
1025
output_lines(knit._parse_record(key[-1], raw_data)[0])
1027
df, _ = knit._parse_record_header(key, raw_data)
1029
pos, size = writer.add_bytes_record(raw_data, names)
1030
write_index.add_node(key, eol_flag + "%d %d" % (pos, size))
1031
pb.update("Copied record", record_index)
1034
def _copy_nodes_graph(self, index_map, writer, write_index,
1035
readv_group_iter, total_items, output_lines=False):
1036
"""Copy knit nodes between packs.
1038
:param output_lines: Return lines present in the copied data as
1039
an iterator of line,version_id.
1041
pb = ui.ui_factory.nested_progress_bar()
1043
for result in self._do_copy_nodes_graph(index_map, writer,
1044
write_index, output_lines, pb, readv_group_iter, total_items):
1047
# Python 2.4 does not permit try:finally: in a generator.
1053
def _do_copy_nodes_graph(self, index_map, writer, write_index,
1054
output_lines, pb, readv_group_iter, total_items):
1055
# for record verification
1056
knit = KnitVersionedFiles(None, None)
1057
# for line extraction when requested (inventories only)
1059
factory = KnitPlainFactory()
1061
pb.update("Copied record", record_index, total_items)
1062
for index, readv_vector, node_vector in readv_group_iter:
1064
pack_obj = index_map[index]
1065
transport, path = pack_obj.access_tuple()
1067
reader = pack.make_readv_reader(transport, path, readv_vector)
1068
except errors.NoSuchFile:
1069
if self._reload_func is not None:
1072
for (names, read_func), (key, eol_flag, references) in \
1073
izip(reader.iter_records(), node_vector):
1074
raw_data = read_func(None)
1076
# read the entire thing
1077
content, _ = knit._parse_record(key[-1], raw_data)
1078
if len(references[-1]) == 0:
1079
line_iterator = factory.get_fulltext_content(content)
1081
line_iterator = factory.get_linedelta_content(content)
1082
for line in line_iterator:
1085
# check the header only
1086
df, _ = knit._parse_record_header(key, raw_data)
1088
pos, size = writer.add_bytes_record(raw_data, names)
1089
write_index.add_node(key, eol_flag + "%d %d" % (pos, size), references)
1090
pb.update("Copied record", record_index)
1093
def _get_text_nodes(self):
1094
text_index_map, text_indices = self._pack_map_and_index_list(
1096
return text_index_map, self._index_contents(text_indices,
1099
def _least_readv_node_readv(self, nodes):
1100
"""Generate request groups for nodes using the least readv's.
1102
:param nodes: An iterable of graph index nodes.
1103
:return: Total node count and an iterator of the data needed to perform
1104
readvs to obtain the data for nodes. Each item yielded by the
1105
iterator is a tuple with:
1106
index, readv_vector, node_vector. readv_vector is a list ready to
1107
hand to the transport readv method, and node_vector is a list of
1108
(key, eol_flag, references) for the the node retrieved by the
1109
matching readv_vector.
1111
# group by pack so we do one readv per pack
1112
nodes = sorted(nodes)
1115
for index, key, value, references in nodes:
1116
if index not in request_groups:
1117
request_groups[index] = []
1118
request_groups[index].append((key, value, references))
1120
for index, items in request_groups.iteritems():
1121
pack_readv_requests = []
1122
for key, value, references in items:
1123
# ---- KnitGraphIndex.get_position
1124
bits = value[1:].split(' ')
1125
offset, length = int(bits[0]), int(bits[1])
1126
pack_readv_requests.append(
1127
((offset, length), (key, value[0], references)))
1128
# linear scan up the pack to maximum range combining.
1129
pack_readv_requests.sort()
1130
# split out the readv and the node data.
1131
pack_readv = [readv for readv, node in pack_readv_requests]
1132
node_vector = [node for readv, node in pack_readv_requests]
1133
result.append((index, pack_readv, node_vector))
1134
return total, result
1136
def _log_copied_texts(self):
1137
if 'pack' in debug.debug_flags:
1138
mutter('%s: create_pack: file texts copied: %s%s %d items t+%6.3fs',
1139
time.ctime(), self._pack_collection._upload_transport.base,
1140
self.new_pack.random_name,
1141
self.new_pack.text_index.key_count(),
1142
time.time() - self.new_pack.start_time)
1144
def _process_inventory_lines(self, inv_lines):
1145
"""Use up the inv_lines generator and setup a text key filter."""
1146
repo = self._pack_collection.repo
1147
fileid_revisions = repo._find_file_ids_from_xml_inventory_lines(
1148
inv_lines, self.revision_keys)
1150
for fileid, file_revids in fileid_revisions.iteritems():
1151
text_filter.extend([(fileid, file_revid) for file_revid in file_revids])
1152
self._text_filter = text_filter
1154
def _revision_node_readv(self, revision_nodes):
1155
"""Return the total revisions and the readv's to issue.
1157
:param revision_nodes: The revision index contents for the packs being
1158
incorporated into the new pack.
1159
:return: As per _least_readv_node_readv.
1161
return self._least_readv_node_readv(revision_nodes)
1163
def _use_pack(self, new_pack):
1164
"""Return True if new_pack should be used.
1166
:param new_pack: The pack that has just been created.
1167
:return: True if the pack should be used.
1169
return new_pack.data_inserted()
1172
class OptimisingPacker(Packer):
1173
"""A packer which spends more time to create better disk layouts."""
1175
def _revision_node_readv(self, revision_nodes):
1176
"""Return the total revisions and the readv's to issue.
1178
This sort places revisions in topological order with the ancestors
1181
:param revision_nodes: The revision index contents for the packs being
1182
incorporated into the new pack.
1183
:return: As per _least_readv_node_readv.
1185
# build an ancestors dict
1188
for index, key, value, references in revision_nodes:
1189
ancestors[key] = references[0]
1190
by_key[key] = (index, value, references)
1191
order = tsort.topo_sort(ancestors)
1193
# Single IO is pathological, but it will work as a starting point.
1195
for key in reversed(order):
1196
index, value, references = by_key[key]
1197
# ---- KnitGraphIndex.get_position
1198
bits = value[1:].split(' ')
1199
offset, length = int(bits[0]), int(bits[1])
1201
(index, [(offset, length)], [(key, value[0], references)]))
1202
# TODO: combine requests in the same index that are in ascending order.
1203
return total, requests
1205
def open_pack(self):
1206
"""Open a pack for the pack we are creating."""
1207
new_pack = super(OptimisingPacker, self).open_pack()
1208
# Turn on the optimization flags for all the index builders.
1209
new_pack.revision_index.set_optimize(for_size=True)
1210
new_pack.inventory_index.set_optimize(for_size=True)
1211
new_pack.text_index.set_optimize(for_size=True)
1212
new_pack.signature_index.set_optimize(for_size=True)
1216
class ReconcilePacker(Packer):
1217
"""A packer which regenerates indices etc as it copies.
1219
This is used by ``bzr reconcile`` to cause parent text pointers to be
1223
def _extra_init(self):
1224
self._data_changed = False
1226
def _process_inventory_lines(self, inv_lines):
1227
"""Generate a text key reference map rather for reconciling with."""
1228
repo = self._pack_collection.repo
1229
refs = repo._find_text_key_references_from_xml_inventory_lines(
1231
self._text_refs = refs
1232
# during reconcile we:
1233
# - convert unreferenced texts to full texts
1234
# - correct texts which reference a text not copied to be full texts
1235
# - copy all others as-is but with corrected parents.
1236
# - so at this point we don't know enough to decide what becomes a full
1238
self._text_filter = None
1240
def _copy_text_texts(self):
1241
"""generate what texts we should have and then copy."""
1242
self.pb.update("Copying content texts", 3)
1243
# we have three major tasks here:
1244
# 1) generate the ideal index
1245
repo = self._pack_collection.repo
1246
ancestors = dict([(key[0], tuple(ref[0] for ref in refs[0])) for
1247
_1, key, _2, refs in
1248
self.new_pack.revision_index.iter_all_entries()])
1249
ideal_index = repo._generate_text_key_index(self._text_refs, ancestors)
1250
# 2) generate a text_nodes list that contains all the deltas that can
1251
# be used as-is, with corrected parents.
1254
discarded_nodes = []
1255
NULL_REVISION = _mod_revision.NULL_REVISION
1256
text_index_map, text_nodes = self._get_text_nodes()
1257
for node in text_nodes:
1263
ideal_parents = tuple(ideal_index[node[1]])
1265
discarded_nodes.append(node)
1266
self._data_changed = True
1268
if ideal_parents == (NULL_REVISION,):
1270
if ideal_parents == node[3][0]:
1272
ok_nodes.append(node)
1273
elif ideal_parents[0:1] == node[3][0][0:1]:
1274
# the left most parent is the same, or there are no parents
1275
# today. Either way, we can preserve the representation as
1276
# long as we change the refs to be inserted.
1277
self._data_changed = True
1278
ok_nodes.append((node[0], node[1], node[2],
1279
(ideal_parents, node[3][1])))
1280
self._data_changed = True
1282
# Reinsert this text completely
1283
bad_texts.append((node[1], ideal_parents))
1284
self._data_changed = True
1285
# we're finished with some data.
1288
# 3) bulk copy the ok data
1289
total_items, readv_group_iter = self._least_readv_node_readv(ok_nodes)
1290
list(self._copy_nodes_graph(text_index_map, self.new_pack._writer,
1291
self.new_pack.text_index, readv_group_iter, total_items))
1292
# 4) adhoc copy all the other texts.
1293
# We have to topologically insert all texts otherwise we can fail to
1294
# reconcile when parts of a single delta chain are preserved intact,
1295
# and other parts are not. E.g. Discarded->d1->d2->d3. d1 will be
1296
# reinserted, and if d3 has incorrect parents it will also be
1297
# reinserted. If we insert d3 first, d2 is present (as it was bulk
1298
# copied), so we will try to delta, but d2 is not currently able to be
1299
# extracted because it's basis d1 is not present. Topologically sorting
1300
# addresses this. The following generates a sort for all the texts that
1301
# are being inserted without having to reference the entire text key
1302
# space (we only topo sort the revisions, which is smaller).
1303
topo_order = tsort.topo_sort(ancestors)
1304
rev_order = dict(zip(topo_order, range(len(topo_order))))
1305
bad_texts.sort(key=lambda key:rev_order.get(key[0][1], 0))
1306
transaction = repo.get_transaction()
1307
file_id_index = GraphIndexPrefixAdapter(
1308
self.new_pack.text_index,
1310
add_nodes_callback=self.new_pack.text_index.add_nodes)
1311
data_access = _DirectPackAccess(
1312
{self.new_pack.text_index:self.new_pack.access_tuple()})
1313
data_access.set_writer(self.new_pack._writer, self.new_pack.text_index,
1314
self.new_pack.access_tuple())
1315
output_texts = KnitVersionedFiles(
1316
_KnitGraphIndex(self.new_pack.text_index,
1317
add_callback=self.new_pack.text_index.add_nodes,
1318
deltas=True, parents=True, is_locked=repo.is_locked),
1319
data_access=data_access, max_delta_chain=200)
1320
for key, parent_keys in bad_texts:
1321
# We refer to the new pack to delta data being output.
1322
# A possible improvement would be to catch errors on short reads
1323
# and only flush then.
1324
self.new_pack.flush()
1326
for parent_key in parent_keys:
1327
if parent_key[0] != key[0]:
1328
# Graph parents must match the fileid
1329
raise errors.BzrError('Mismatched key parent %r:%r' %
1331
parents.append(parent_key[1])
1332
text_lines = osutils.split_lines(repo.texts.get_record_stream(
1333
[key], 'unordered', True).next().get_bytes_as('fulltext'))
1334
output_texts.add_lines(key, parent_keys, text_lines,
1335
random_id=True, check_content=False)
1336
# 5) check that nothing inserted has a reference outside the keyspace.
1337
missing_text_keys = self.new_pack.text_index._external_references()
1338
if missing_text_keys:
1339
raise errors.BzrCheckError('Reference to missing compression parents %r'
1340
% (missing_text_keys,))
1341
self._log_copied_texts()
1343
def _use_pack(self, new_pack):
1344
"""Override _use_pack to check for reconcile having changed content."""
1345
# XXX: we might be better checking this at the copy time.
1346
original_inventory_keys = set()
1347
inv_index = self._pack_collection.inventory_index.combined_index
1348
for entry in inv_index.iter_all_entries():
1349
original_inventory_keys.add(entry[1])
1350
new_inventory_keys = set()
1351
for entry in new_pack.inventory_index.iter_all_entries():
1352
new_inventory_keys.add(entry[1])
1353
if new_inventory_keys != original_inventory_keys:
1354
self._data_changed = True
1355
return new_pack.data_inserted() and self._data_changed
1358
class RepositoryPackCollection(object):
1359
"""Management of packs within a repository.
1361
:ivar _names: map of {pack_name: (index_size,)}
1364
pack_factory = NewPack
1365
resumed_pack_factory = ResumedPack
1367
def __init__(self, repo, transport, index_transport, upload_transport,
1368
pack_transport, index_builder_class, index_class,
1370
"""Create a new RepositoryPackCollection.
1372
:param transport: Addresses the repository base directory
1373
(typically .bzr/repository/).
1374
:param index_transport: Addresses the directory containing indices.
1375
:param upload_transport: Addresses the directory into which packs are written
1376
while they're being created.
1377
:param pack_transport: Addresses the directory of existing complete packs.
1378
:param index_builder_class: The index builder class to use.
1379
:param index_class: The index class to use.
1380
:param use_chk_index: Whether to setup and manage a CHK index.
1382
# XXX: This should call self.reset()
1384
self.transport = transport
1385
self._index_transport = index_transport
1386
self._upload_transport = upload_transport
1387
self._pack_transport = pack_transport
1388
self._index_builder_class = index_builder_class
1389
self._index_class = index_class
1390
self._suffix_offsets = {'.rix': 0, '.iix': 1, '.tix': 2, '.six': 3,
1395
self._packs_by_name = {}
1396
# the previous pack-names content
1397
self._packs_at_load = None
1398
# when a pack is being created by this object, the state of that pack.
1399
self._new_pack = None
1400
# aggregated revision index data
1401
flush = self._flush_new_pack
1402
self.revision_index = AggregateIndex(self.reload_pack_names, flush)
1403
self.inventory_index = AggregateIndex(self.reload_pack_names, flush)
1404
self.text_index = AggregateIndex(self.reload_pack_names, flush)
1405
self.signature_index = AggregateIndex(self.reload_pack_names, flush)
1407
self.chk_index = AggregateIndex(self.reload_pack_names, flush)
1409
# used to determine if we're using a chk_index elsewhere.
1410
self.chk_index = None
1412
self._resumed_packs = []
1414
def add_pack_to_memory(self, pack):
1415
"""Make a Pack object available to the repository to satisfy queries.
1417
:param pack: A Pack object.
1419
if pack.name in self._packs_by_name:
1420
raise AssertionError(
1421
'pack %s already in _packs_by_name' % (pack.name,))
1422
self.packs.append(pack)
1423
self._packs_by_name[pack.name] = pack
1424
self.revision_index.add_index(pack.revision_index, pack)
1425
self.inventory_index.add_index(pack.inventory_index, pack)
1426
self.text_index.add_index(pack.text_index, pack)
1427
self.signature_index.add_index(pack.signature_index, pack)
1428
if self.chk_index is not None:
1429
self.chk_index.add_index(pack.chk_index, pack)
1431
def all_packs(self):
1432
"""Return a list of all the Pack objects this repository has.
1434
Note that an in-progress pack being created is not returned.
1436
:return: A list of Pack objects for all the packs in the repository.
1439
for name in self.names():
1440
result.append(self.get_pack_by_name(name))
1444
"""Pack the pack collection incrementally.
1446
This will not attempt global reorganisation or recompression,
1447
rather it will just ensure that the total number of packs does
1448
not grow without bound. It uses the _max_pack_count method to
1449
determine if autopacking is needed, and the pack_distribution
1450
method to determine the number of revisions in each pack.
1452
If autopacking takes place then the packs name collection will have
1453
been flushed to disk - packing requires updating the name collection
1454
in synchronisation with certain steps. Otherwise the names collection
1457
:return: Something evaluating true if packing took place.
1461
return self._do_autopack()
1462
except errors.RetryAutopack:
1463
# If we get a RetryAutopack exception, we should abort the
1464
# current action, and retry.
1467
def _do_autopack(self):
1468
# XXX: Should not be needed when the management of indices is sane.
1469
total_revisions = self.revision_index.combined_index.key_count()
1470
total_packs = len(self._names)
1471
if self._max_pack_count(total_revisions) >= total_packs:
1473
# determine which packs need changing
1474
pack_distribution = self.pack_distribution(total_revisions)
1476
for pack in self.all_packs():
1477
revision_count = pack.get_revision_count()
1478
if revision_count == 0:
1479
# revision less packs are not generated by normal operation,
1480
# only by operations like sign-my-commits, and thus will not
1481
# tend to grow rapdily or without bound like commit containing
1482
# packs do - leave them alone as packing them really should
1483
# group their data with the relevant commit, and that may
1484
# involve rewriting ancient history - which autopack tries to
1485
# avoid. Alternatively we could not group the data but treat
1486
# each of these as having a single revision, and thus add
1487
# one revision for each to the total revision count, to get
1488
# a matching distribution.
1490
existing_packs.append((revision_count, pack))
1491
pack_operations = self.plan_autopack_combinations(
1492
existing_packs, pack_distribution)
1493
num_new_packs = len(pack_operations)
1494
num_old_packs = sum([len(po[1]) for po in pack_operations])
1495
num_revs_affected = sum([po[0] for po in pack_operations])
1496
mutter('Auto-packing repository %s, which has %d pack files, '
1497
'containing %d revisions. Packing %d files into %d affecting %d'
1498
' revisions', self, total_packs, total_revisions, num_old_packs,
1499
num_new_packs, num_revs_affected)
1500
result = self._execute_pack_operations(pack_operations,
1501
reload_func=self._restart_autopack)
1502
mutter('Auto-packing repository %s completed', self)
1505
def _execute_pack_operations(self, pack_operations, _packer_class=Packer,
1507
"""Execute a series of pack operations.
1509
:param pack_operations: A list of [revision_count, packs_to_combine].
1510
:param _packer_class: The class of packer to use (default: Packer).
1511
:return: The new pack names.
1513
for revision_count, packs in pack_operations:
1514
# we may have no-ops from the setup logic
1517
packer = _packer_class(self, packs, '.autopack',
1518
reload_func=reload_func)
1521
except errors.RetryWithNewPacks:
1522
# An exception is propagating out of this context, make sure
1523
# this packer has cleaned up. Packer() doesn't set its new_pack
1524
# state into the RepositoryPackCollection object, so we only
1525
# have access to it directly here.
1526
if packer.new_pack is not None:
1527
packer.new_pack.abort()
1530
self._remove_pack_from_memory(pack)
1531
# record the newly available packs and stop advertising the old
1533
result = self._save_pack_names(clear_obsolete_packs=True)
1534
# Move the old packs out of the way now they are no longer referenced.
1535
for revision_count, packs in pack_operations:
1536
self._obsolete_packs(packs)
1539
def _flush_new_pack(self):
1540
if self._new_pack is not None:
1541
self._new_pack.flush()
1543
def lock_names(self):
1544
"""Acquire the mutex around the pack-names index.
1546
This cannot be used in the middle of a read-only transaction on the
1549
self.repo.control_files.lock_write()
1551
def _already_packed(self):
1552
"""Is the collection already packed?"""
1553
return not (self.repo._format.pack_compresses or (len(self._names) > 1))
1555
def pack(self, hint=None):
1556
"""Pack the pack collection totally."""
1557
self.ensure_loaded()
1558
total_packs = len(self._names)
1559
if self._already_packed():
1561
total_revisions = self.revision_index.combined_index.key_count()
1562
# XXX: the following may want to be a class, to pack with a given
1564
mutter('Packing repository %s, which has %d pack files, '
1565
'containing %d revisions with hint %r.', self, total_packs,
1566
total_revisions, hint)
1567
# determine which packs need changing
1568
pack_operations = [[0, []]]
1569
for pack in self.all_packs():
1570
if not hint or pack.name in hint:
1571
pack_operations[-1][0] += pack.get_revision_count()
1572
pack_operations[-1][1].append(pack)
1573
self._execute_pack_operations(pack_operations, OptimisingPacker)
1575
def plan_autopack_combinations(self, existing_packs, pack_distribution):
1576
"""Plan a pack operation.
1578
:param existing_packs: The packs to pack. (A list of (revcount, Pack)
1580
:param pack_distribution: A list with the number of revisions desired
1583
if len(existing_packs) <= len(pack_distribution):
1585
existing_packs.sort(reverse=True)
1586
pack_operations = [[0, []]]
1587
# plan out what packs to keep, and what to reorganise
1588
while len(existing_packs):
1589
# take the largest pack, and if its less than the head of the
1590
# distribution chart we will include its contents in the new pack
1591
# for that position. If its larger, we remove its size from the
1592
# distribution chart
1593
next_pack_rev_count, next_pack = existing_packs.pop(0)
1594
if next_pack_rev_count >= pack_distribution[0]:
1595
# this is already packed 'better' than this, so we can
1596
# not waste time packing it.
1597
while next_pack_rev_count > 0:
1598
next_pack_rev_count -= pack_distribution[0]
1599
if next_pack_rev_count >= 0:
1601
del pack_distribution[0]
1603
# didn't use that entire bucket up
1604
pack_distribution[0] = -next_pack_rev_count
1606
# add the revisions we're going to add to the next output pack
1607
pack_operations[-1][0] += next_pack_rev_count
1608
# allocate this pack to the next pack sub operation
1609
pack_operations[-1][1].append(next_pack)
1610
if pack_operations[-1][0] >= pack_distribution[0]:
1611
# this pack is used up, shift left.
1612
del pack_distribution[0]
1613
pack_operations.append([0, []])
1614
# Now that we know which pack files we want to move, shove them all
1615
# into a single pack file.
1617
final_pack_list = []
1618
for num_revs, pack_files in pack_operations:
1619
final_rev_count += num_revs
1620
final_pack_list.extend(pack_files)
1621
if len(final_pack_list) == 1:
1622
raise AssertionError('We somehow generated an autopack with a'
1623
' single pack file being moved.')
1625
return [[final_rev_count, final_pack_list]]
1627
def ensure_loaded(self):
1628
"""Ensure we have read names from disk.
1630
:return: True if the disk names had not been previously read.
1632
# NB: if you see an assertion error here, its probably access against
1633
# an unlocked repo. Naughty.
1634
if not self.repo.is_locked():
1635
raise errors.ObjectNotLocked(self.repo)
1636
if self._names is None:
1638
self._packs_at_load = set()
1639
for index, key, value in self._iter_disk_pack_index():
1641
self._names[name] = self._parse_index_sizes(value)
1642
self._packs_at_load.add((key, value))
1646
# populate all the metadata.
1650
def _parse_index_sizes(self, value):
1651
"""Parse a string of index sizes."""
1652
return tuple([int(digits) for digits in value.split(' ')])
1654
def get_pack_by_name(self, name):
1655
"""Get a Pack object by name.
1657
:param name: The name of the pack - e.g. '123456'
1658
:return: A Pack object.
1661
return self._packs_by_name[name]
1663
rev_index = self._make_index(name, '.rix')
1664
inv_index = self._make_index(name, '.iix')
1665
txt_index = self._make_index(name, '.tix')
1666
sig_index = self._make_index(name, '.six')
1667
if self.chk_index is not None:
1668
chk_index = self._make_index(name, '.cix')
1671
result = ExistingPack(self._pack_transport, name, rev_index,
1672
inv_index, txt_index, sig_index, chk_index)
1673
self.add_pack_to_memory(result)
1676
def _resume_pack(self, name):
1677
"""Get a suspended Pack object by name.
1679
:param name: The name of the pack - e.g. '123456'
1680
:return: A Pack object.
1682
if not re.match('[a-f0-9]{32}', name):
1683
# Tokens should be md5sums of the suspended pack file, i.e. 32 hex
1685
raise errors.UnresumableWriteGroup(
1686
self.repo, [name], 'Malformed write group token')
1688
rev_index = self._make_index(name, '.rix', resume=True)
1689
inv_index = self._make_index(name, '.iix', resume=True)
1690
txt_index = self._make_index(name, '.tix', resume=True)
1691
sig_index = self._make_index(name, '.six', resume=True)
1692
if self.chk_index is not None:
1693
chk_index = self._make_index(name, '.cix', resume=True)
1696
result = self.resumed_pack_factory(name, rev_index, inv_index,
1697
txt_index, sig_index, self._upload_transport,
1698
self._pack_transport, self._index_transport, self,
1699
chk_index=chk_index)
1700
except errors.NoSuchFile, e:
1701
raise errors.UnresumableWriteGroup(self.repo, [name], str(e))
1702
self.add_pack_to_memory(result)
1703
self._resumed_packs.append(result)
1706
def allocate(self, a_new_pack):
1707
"""Allocate name in the list of packs.
1709
:param a_new_pack: A NewPack instance to be added to the collection of
1710
packs for this repository.
1712
self.ensure_loaded()
1713
if a_new_pack.name in self._names:
1714
raise errors.BzrError(
1715
'Pack %r already exists in %s' % (a_new_pack.name, self))
1716
self._names[a_new_pack.name] = tuple(a_new_pack.index_sizes)
1717
self.add_pack_to_memory(a_new_pack)
1719
def _iter_disk_pack_index(self):
1720
"""Iterate over the contents of the pack-names index.
1722
This is used when loading the list from disk, and before writing to
1723
detect updates from others during our write operation.
1724
:return: An iterator of the index contents.
1726
return self._index_class(self.transport, 'pack-names', None
1727
).iter_all_entries()
1729
def _make_index(self, name, suffix, resume=False):
1730
size_offset = self._suffix_offsets[suffix]
1731
index_name = name + suffix
1733
transport = self._upload_transport
1734
index_size = transport.stat(index_name).st_size
1736
transport = self._index_transport
1737
index_size = self._names[name][size_offset]
1738
return self._index_class(transport, index_name, index_size)
1740
def _max_pack_count(self, total_revisions):
1741
"""Return the maximum number of packs to use for total revisions.
1743
:param total_revisions: The total number of revisions in the
1746
if not total_revisions:
1748
digits = str(total_revisions)
1750
for digit in digits:
1751
result += int(digit)
1755
"""Provide an order to the underlying names."""
1756
return sorted(self._names.keys())
1758
def _obsolete_packs(self, packs):
1759
"""Move a number of packs which have been obsoleted out of the way.
1761
Each pack and its associated indices are moved out of the way.
1763
Note: for correctness this function should only be called after a new
1764
pack names index has been written without these pack names, and with
1765
the names of packs that contain the data previously available via these
1768
:param packs: The packs to obsolete.
1769
:param return: None.
1772
pack.pack_transport.rename(pack.file_name(),
1773
'../obsolete_packs/' + pack.file_name())
1774
# TODO: Probably needs to know all possible indices for this pack
1775
# - or maybe list the directory and move all indices matching this
1776
# name whether we recognize it or not?
1777
suffixes = ['.iix', '.six', '.tix', '.rix']
1778
if self.chk_index is not None:
1779
suffixes.append('.cix')
1780
for suffix in suffixes:
1781
self._index_transport.rename(pack.name + suffix,
1782
'../obsolete_packs/' + pack.name + suffix)
1784
def pack_distribution(self, total_revisions):
1785
"""Generate a list of the number of revisions to put in each pack.
1787
:param total_revisions: The total number of revisions in the
1790
if total_revisions == 0:
1792
digits = reversed(str(total_revisions))
1794
for exponent, count in enumerate(digits):
1795
size = 10 ** exponent
1796
for pos in range(int(count)):
1798
return list(reversed(result))
1800
def _pack_tuple(self, name):
1801
"""Return a tuple with the transport and file name for a pack name."""
1802
return self._pack_transport, name + '.pack'
1804
def _remove_pack_from_memory(self, pack):
1805
"""Remove pack from the packs accessed by this repository.
1807
Only affects memory state, until self._save_pack_names() is invoked.
1809
self._names.pop(pack.name)
1810
self._packs_by_name.pop(pack.name)
1811
self._remove_pack_indices(pack)
1812
self.packs.remove(pack)
1814
def _remove_pack_indices(self, pack):
1815
"""Remove the indices for pack from the aggregated indices."""
1816
self.revision_index.remove_index(pack.revision_index, pack)
1817
self.inventory_index.remove_index(pack.inventory_index, pack)
1818
self.text_index.remove_index(pack.text_index, pack)
1819
self.signature_index.remove_index(pack.signature_index, pack)
1820
if self.chk_index is not None:
1821
self.chk_index.remove_index(pack.chk_index, pack)
1824
"""Clear all cached data."""
1825
# cached revision data
1826
self.revision_index.clear()
1827
# cached signature data
1828
self.signature_index.clear()
1829
# cached file text data
1830
self.text_index.clear()
1831
# cached inventory data
1832
self.inventory_index.clear()
1834
if self.chk_index is not None:
1835
self.chk_index.clear()
1836
# remove the open pack
1837
self._new_pack = None
1838
# information about packs.
1841
self._packs_by_name = {}
1842
self._packs_at_load = None
1844
def _unlock_names(self):
1845
"""Release the mutex around the pack-names index."""
1846
self.repo.control_files.unlock()
1848
def _diff_pack_names(self):
1849
"""Read the pack names from disk, and compare it to the one in memory.
1851
:return: (disk_nodes, deleted_nodes, new_nodes)
1852
disk_nodes The final set of nodes that should be referenced
1853
deleted_nodes Nodes which have been removed from when we started
1854
new_nodes Nodes that are newly introduced
1856
# load the disk nodes across
1858
for index, key, value in self._iter_disk_pack_index():
1859
disk_nodes.add((key, value))
1861
# do a two-way diff against our original content
1862
current_nodes = set()
1863
for name, sizes in self._names.iteritems():
1865
((name, ), ' '.join(str(size) for size in sizes)))
1867
# Packs no longer present in the repository, which were present when we
1868
# locked the repository
1869
deleted_nodes = self._packs_at_load - current_nodes
1870
# Packs which this process is adding
1871
new_nodes = current_nodes - self._packs_at_load
1873
# Update the disk_nodes set to include the ones we are adding, and
1874
# remove the ones which were removed by someone else
1875
disk_nodes.difference_update(deleted_nodes)
1876
disk_nodes.update(new_nodes)
1878
return disk_nodes, deleted_nodes, new_nodes
1880
def _syncronize_pack_names_from_disk_nodes(self, disk_nodes):
1881
"""Given the correct set of pack files, update our saved info.
1883
:return: (removed, added, modified)
1884
removed pack names removed from self._names
1885
added pack names added to self._names
1886
modified pack names that had changed value
1891
## self._packs_at_load = disk_nodes
1892
new_names = dict(disk_nodes)
1893
# drop no longer present nodes
1894
for pack in self.all_packs():
1895
if (pack.name,) not in new_names:
1896
removed.append(pack.name)
1897
self._remove_pack_from_memory(pack)
1898
# add new nodes/refresh existing ones
1899
for key, value in disk_nodes:
1901
sizes = self._parse_index_sizes(value)
1902
if name in self._names:
1904
if sizes != self._names[name]:
1905
# the pack for name has had its indices replaced - rare but
1906
# important to handle. XXX: probably can never happen today
1907
# because the three-way merge code above does not handle it
1908
# - you may end up adding the same key twice to the new
1909
# disk index because the set values are the same, unless
1910
# the only index shows up as deleted by the set difference
1911
# - which it may. Until there is a specific test for this,
1912
# assume its broken. RBC 20071017.
1913
self._remove_pack_from_memory(self.get_pack_by_name(name))
1914
self._names[name] = sizes
1915
self.get_pack_by_name(name)
1916
modified.append(name)
1919
self._names[name] = sizes
1920
self.get_pack_by_name(name)
1922
return removed, added, modified
1924
def _save_pack_names(self, clear_obsolete_packs=False):
1925
"""Save the list of packs.
1927
This will take out the mutex around the pack names list for the
1928
duration of the method call. If concurrent updates have been made, a
1929
three-way merge between the current list and the current in memory list
1932
:param clear_obsolete_packs: If True, clear out the contents of the
1933
obsolete_packs directory.
1934
:return: A list of the names saved that were not previously on disk.
1938
builder = self._index_builder_class()
1939
disk_nodes, deleted_nodes, new_nodes = self._diff_pack_names()
1940
# TODO: handle same-name, index-size-changes here -
1941
# e.g. use the value from disk, not ours, *unless* we're the one
1943
for key, value in disk_nodes:
1944
builder.add_node(key, value)
1945
self.transport.put_file('pack-names', builder.finish(),
1946
mode=self.repo.bzrdir._get_file_mode())
1947
# move the baseline forward
1948
self._packs_at_load = disk_nodes
1949
if clear_obsolete_packs:
1950
self._clear_obsolete_packs()
1952
self._unlock_names()
1953
# synchronise the memory packs list with what we just wrote:
1954
self._syncronize_pack_names_from_disk_nodes(disk_nodes)
1955
return [new_node[0][0] for new_node in new_nodes]
1957
def reload_pack_names(self):
1958
"""Sync our pack listing with what is present in the repository.
1960
This should be called when we find out that something we thought was
1961
present is now missing. This happens when another process re-packs the
1964
:return: True if the in-memory list of packs has been altered at all.
1966
# The ensure_loaded call is to handle the case where the first call
1967
# made involving the collection was to reload_pack_names, where we
1968
# don't have a view of disk contents. Its a bit of a bandaid, and
1969
# causes two reads of pack-names, but its a rare corner case not struck
1970
# with regular push/pull etc.
1971
first_read = self.ensure_loaded()
1974
# out the new value.
1975
disk_nodes, _, _ = self._diff_pack_names()
1976
self._packs_at_load = disk_nodes
1978
modified) = self._syncronize_pack_names_from_disk_nodes(disk_nodes)
1979
if removed or added or modified:
1983
def _restart_autopack(self):
1984
"""Reload the pack names list, and restart the autopack code."""
1985
if not self.reload_pack_names():
1986
# Re-raise the original exception, because something went missing
1987
# and a restart didn't find it
1989
raise errors.RetryAutopack(self.repo, False, sys.exc_info())
1991
def _clear_obsolete_packs(self):
1992
"""Delete everything from the obsolete-packs directory.
1994
obsolete_pack_transport = self.transport.clone('obsolete_packs')
1995
for filename in obsolete_pack_transport.list_dir('.'):
1997
obsolete_pack_transport.delete(filename)
1998
except (errors.PathError, errors.TransportError), e:
1999
warning("couldn't delete obsolete pack, skipping it:\n%s" % (e,))
2001
def _start_write_group(self):
2002
# Do not permit preparation for writing if we're not in a 'write lock'.
2003
if not self.repo.is_write_locked():
2004
raise errors.NotWriteLocked(self)
2005
self._new_pack = self.pack_factory(self, upload_suffix='.pack',
2006
file_mode=self.repo.bzrdir._get_file_mode())
2007
# allow writing: queue writes to a new index
2008
self.revision_index.add_writable_index(self._new_pack.revision_index,
2010
self.inventory_index.add_writable_index(self._new_pack.inventory_index,
2012
self.text_index.add_writable_index(self._new_pack.text_index,
2014
self._new_pack.text_index.set_optimize(combine_backing_indices=False)
2015
self.signature_index.add_writable_index(self._new_pack.signature_index,
2017
if self.chk_index is not None:
2018
self.chk_index.add_writable_index(self._new_pack.chk_index,
2020
self.repo.chk_bytes._index._add_callback = self.chk_index.add_callback
2021
self._new_pack.chk_index.set_optimize(combine_backing_indices=False)
2023
self.repo.inventories._index._add_callback = self.inventory_index.add_callback
2024
self.repo.revisions._index._add_callback = self.revision_index.add_callback
2025
self.repo.signatures._index._add_callback = self.signature_index.add_callback
2026
self.repo.texts._index._add_callback = self.text_index.add_callback
2028
def _abort_write_group(self):
2029
# FIXME: just drop the transient index.
2030
# forget what names there are
2031
if self._new_pack is not None:
2033
self._new_pack.abort()
2035
# XXX: If we aborted while in the middle of finishing the write
2036
# group, _remove_pack_indices can fail because the indexes are
2037
# already gone. If they're not there we shouldn't fail in this
2038
# case. -- mbp 20081113
2039
self._remove_pack_indices(self._new_pack)
2040
self._new_pack = None
2041
for resumed_pack in self._resumed_packs:
2043
resumed_pack.abort()
2045
# See comment in previous finally block.
2047
self._remove_pack_indices(resumed_pack)
2050
del self._resumed_packs[:]
2052
def _remove_resumed_pack_indices(self):
2053
for resumed_pack in self._resumed_packs:
2054
self._remove_pack_indices(resumed_pack)
2055
del self._resumed_packs[:]
2057
def _commit_write_group(self):
2059
for prefix, versioned_file in (
2060
('revisions', self.repo.revisions),
2061
('inventories', self.repo.inventories),
2062
('texts', self.repo.texts),
2063
('signatures', self.repo.signatures),
2065
missing = versioned_file.get_missing_compression_parent_keys()
2066
all_missing.update([(prefix,) + key for key in missing])
2068
raise errors.BzrCheckError(
2069
"Repository %s has missing compression parent(s) %r "
2070
% (self.repo, sorted(all_missing)))
2071
self._remove_pack_indices(self._new_pack)
2072
should_autopack = False
2073
if self._new_pack.data_inserted():
2074
# get all the data to disk and read to use
2075
self._new_pack.finish()
2076
self.allocate(self._new_pack)
2077
self._new_pack = None
2078
should_autopack = True
2080
self._new_pack.abort()
2081
self._new_pack = None
2082
for resumed_pack in self._resumed_packs:
2083
# XXX: this is a pretty ugly way to turn the resumed pack into a
2084
# properly committed pack.
2085
self._names[resumed_pack.name] = None
2086
self._remove_pack_from_memory(resumed_pack)
2087
resumed_pack.finish()
2088
self.allocate(resumed_pack)
2089
should_autopack = True
2090
del self._resumed_packs[:]
2092
if not self.autopack():
2093
# when autopack takes no steps, the names list is still
2095
return self._save_pack_names()
2097
def _suspend_write_group(self):
2098
tokens = [pack.name for pack in self._resumed_packs]
2099
self._remove_pack_indices(self._new_pack)
2100
if self._new_pack.data_inserted():
2101
# get all the data to disk and read to use
2102
self._new_pack.finish(suspend=True)
2103
tokens.append(self._new_pack.name)
2104
self._new_pack = None
2106
self._new_pack.abort()
2107
self._new_pack = None
2108
self._remove_resumed_pack_indices()
2111
def _resume_write_group(self, tokens):
2112
for token in tokens:
2113
self._resume_pack(token)
2116
class KnitPackRepository(KnitRepository):
2117
"""Repository with knit objects stored inside pack containers.
2119
The layering for a KnitPackRepository is:
2121
Graph | HPSS | Repository public layer |
2122
===================================================
2123
Tuple based apis below, string based, and key based apis above
2124
---------------------------------------------------
2126
Provides .texts, .revisions etc
2127
This adapts the N-tuple keys to physical knit records which only have a
2128
single string identifier (for historical reasons), which in older formats
2129
was always the revision_id, and in the mapped code for packs is always
2130
the last element of key tuples.
2131
---------------------------------------------------
2133
A separate GraphIndex is used for each of the
2134
texts/inventories/revisions/signatures contained within each individual
2135
pack file. The GraphIndex layer works in N-tuples and is unaware of any
2137
===================================================
2141
def __init__(self, _format, a_bzrdir, control_files, _commit_builder_class,
2143
KnitRepository.__init__(self, _format, a_bzrdir, control_files,
2144
_commit_builder_class, _serializer)
2145
index_transport = self._transport.clone('indices')
2146
self._pack_collection = RepositoryPackCollection(self, self._transport,
2148
self._transport.clone('upload'),
2149
self._transport.clone('packs'),
2150
_format.index_builder_class,
2151
_format.index_class,
2152
use_chk_index=self._format.supports_chks,
2154
self.inventories = KnitVersionedFiles(
2155
_KnitGraphIndex(self._pack_collection.inventory_index.combined_index,
2156
add_callback=self._pack_collection.inventory_index.add_callback,
2157
deltas=True, parents=True, is_locked=self.is_locked),
2158
data_access=self._pack_collection.inventory_index.data_access,
2159
max_delta_chain=200)
2160
self.revisions = KnitVersionedFiles(
2161
_KnitGraphIndex(self._pack_collection.revision_index.combined_index,
2162
add_callback=self._pack_collection.revision_index.add_callback,
2163
deltas=False, parents=True, is_locked=self.is_locked,
2164
track_external_parent_refs=True),
2165
data_access=self._pack_collection.revision_index.data_access,
2167
self.signatures = KnitVersionedFiles(
2168
_KnitGraphIndex(self._pack_collection.signature_index.combined_index,
2169
add_callback=self._pack_collection.signature_index.add_callback,
2170
deltas=False, parents=False, is_locked=self.is_locked),
2171
data_access=self._pack_collection.signature_index.data_access,
2173
self.texts = KnitVersionedFiles(
2174
_KnitGraphIndex(self._pack_collection.text_index.combined_index,
2175
add_callback=self._pack_collection.text_index.add_callback,
2176
deltas=True, parents=True, is_locked=self.is_locked),
2177
data_access=self._pack_collection.text_index.data_access,
2178
max_delta_chain=200)
2179
if _format.supports_chks:
2180
# No graph, no compression:- references from chks are between
2181
# different objects not temporal versions of the same; and without
2182
# some sort of temporal structure knit compression will just fail.
2183
self.chk_bytes = KnitVersionedFiles(
2184
_KnitGraphIndex(self._pack_collection.chk_index.combined_index,
2185
add_callback=self._pack_collection.chk_index.add_callback,
2186
deltas=False, parents=False, is_locked=self.is_locked),
2187
data_access=self._pack_collection.chk_index.data_access,
2190
self.chk_bytes = None
2191
# True when the repository object is 'write locked' (as opposed to the
2192
# physical lock only taken out around changes to the pack-names list.)
2193
# Another way to represent this would be a decorator around the control
2194
# files object that presents logical locks as physical ones - if this
2195
# gets ugly consider that alternative design. RBC 20071011
2196
self._write_lock_count = 0
2197
self._transaction = None
2199
self._reconcile_does_inventory_gc = True
2200
self._reconcile_fixes_text_parents = True
2201
self._reconcile_backsup_inventory = False
2203
def _warn_if_deprecated(self):
2204
# This class isn't deprecated, but one sub-format is
2205
if isinstance(self._format, RepositoryFormatKnitPack5RichRootBroken):
2206
from bzrlib import repository
2207
if repository._deprecation_warning_done:
2209
repository._deprecation_warning_done = True
2210
warning("Format %s for %s is deprecated - please use"
2211
" 'bzr upgrade --1.6.1-rich-root'"
2212
% (self._format, self.bzrdir.transport.base))
2214
def _abort_write_group(self):
2215
self.revisions._index._key_dependencies.refs.clear()
2216
self._pack_collection._abort_write_group()
2218
def _get_source(self, to_format):
2219
if to_format.network_name() == self._format.network_name():
2220
return KnitPackStreamSource(self, to_format)
2221
return super(KnitPackRepository, self)._get_source(to_format)
2223
def _make_parents_provider(self):
2224
return graph.CachingParentsProvider(self)
2226
def _refresh_data(self):
2227
if not self.is_locked():
2229
self._pack_collection.reload_pack_names()
2231
def _start_write_group(self):
2232
self._pack_collection._start_write_group()
2234
def _commit_write_group(self):
2235
self.revisions._index._key_dependencies.refs.clear()
2236
return self._pack_collection._commit_write_group()
2238
def suspend_write_group(self):
2239
# XXX check self._write_group is self.get_transaction()?
2240
tokens = self._pack_collection._suspend_write_group()
2241
self.revisions._index._key_dependencies.refs.clear()
2242
self._write_group = None
2245
def _resume_write_group(self, tokens):
2246
self._start_write_group()
2248
self._pack_collection._resume_write_group(tokens)
2249
except errors.UnresumableWriteGroup:
2250
self._abort_write_group()
2252
for pack in self._pack_collection._resumed_packs:
2253
self.revisions._index.scan_unvalidated_index(pack.revision_index)
2255
def get_transaction(self):
2256
if self._write_lock_count:
2257
return self._transaction
2259
return self.control_files.get_transaction()
2261
def is_locked(self):
2262
return self._write_lock_count or self.control_files.is_locked()
2264
def is_write_locked(self):
2265
return self._write_lock_count
2267
def lock_write(self, token=None):
2268
locked = self.is_locked()
2269
if not self._write_lock_count and locked:
2270
raise errors.ReadOnlyError(self)
2271
self._write_lock_count += 1
2272
if self._write_lock_count == 1:
2273
self._transaction = transactions.WriteTransaction()
2275
for repo in self._fallback_repositories:
2276
# Writes don't affect fallback repos
2278
self._refresh_data()
2280
def lock_read(self):
2281
locked = self.is_locked()
2282
if self._write_lock_count:
2283
self._write_lock_count += 1
2285
self.control_files.lock_read()
2287
for repo in self._fallback_repositories:
2289
self._refresh_data()
2291
def leave_lock_in_place(self):
2292
# not supported - raise an error
2293
raise NotImplementedError(self.leave_lock_in_place)
2295
def dont_leave_lock_in_place(self):
2296
# not supported - raise an error
2297
raise NotImplementedError(self.dont_leave_lock_in_place)
2300
def pack(self, hint=None):
2301
"""Compress the data within the repository.
2303
This will pack all the data to a single pack. In future it may
2304
recompress deltas or do other such expensive operations.
2306
self._pack_collection.pack(hint=hint)
2309
def reconcile(self, other=None, thorough=False):
2310
"""Reconcile this repository."""
2311
from bzrlib.reconcile import PackReconciler
2312
reconciler = PackReconciler(self, thorough=thorough)
2313
reconciler.reconcile()
2316
def _reconcile_pack(self, collection, packs, extension, revs, pb):
2317
packer = ReconcilePacker(collection, packs, extension, revs)
2318
return packer.pack(pb)
2321
if self._write_lock_count == 1 and self._write_group is not None:
2322
self.abort_write_group()
2323
self._transaction = None
2324
self._write_lock_count = 0
2325
raise errors.BzrError(
2326
'Must end write group before releasing write lock on %s'
2328
if self._write_lock_count:
2329
self._write_lock_count -= 1
2330
if not self._write_lock_count:
2331
transaction = self._transaction
2332
self._transaction = None
2333
transaction.finish()
2335
self.control_files.unlock()
2337
if not self.is_locked():
2338
for repo in self._fallback_repositories:
2342
class KnitPackStreamSource(StreamSource):
2343
"""A StreamSource used to transfer data between same-format KnitPack repos.
2345
This source assumes:
2346
1) Same serialization format for all objects
2347
2) Same root information
2348
3) XML format inventories
2349
4) Atomic inserts (so we can stream inventory texts before text
2354
def __init__(self, from_repository, to_format):
2355
super(KnitPackStreamSource, self).__init__(from_repository, to_format)
2356
self._text_keys = None
2357
self._text_fetch_order = 'unordered'
2359
def _get_filtered_inv_stream(self, revision_ids):
2360
from_repo = self.from_repository
2361
parent_ids = from_repo._find_parent_ids_of_revisions(revision_ids)
2362
parent_keys = [(p,) for p in parent_ids]
2363
find_text_keys = from_repo._find_text_key_references_from_xml_inventory_lines
2364
parent_text_keys = set(find_text_keys(
2365
from_repo._inventory_xml_lines_for_keys(parent_keys)))
2366
content_text_keys = set()
2367
knit = KnitVersionedFiles(None, None)
2368
factory = KnitPlainFactory()
2369
def find_text_keys_from_content(record):
2370
if record.storage_kind not in ('knit-delta-gz', 'knit-ft-gz'):
2371
raise ValueError("Unknown content storage kind for"
2372
" inventory text: %s" % (record.storage_kind,))
2373
# It's a knit record, it has a _raw_record field (even if it was
2374
# reconstituted from a network stream).
2375
raw_data = record._raw_record
2376
# read the entire thing
2377
revision_id = record.key[-1]
2378
content, _ = knit._parse_record(revision_id, raw_data)
2379
if record.storage_kind == 'knit-delta-gz':
2380
line_iterator = factory.get_linedelta_content(content)
2381
elif record.storage_kind == 'knit-ft-gz':
2382
line_iterator = factory.get_fulltext_content(content)
2383
content_text_keys.update(find_text_keys(
2384
[(line, revision_id) for line in line_iterator]))
2385
revision_keys = [(r,) for r in revision_ids]
2386
def _filtered_inv_stream():
2387
source_vf = from_repo.inventories
2388
stream = source_vf.get_record_stream(revision_keys,
2390
for record in stream:
2391
if record.storage_kind == 'absent':
2392
raise errors.NoSuchRevision(from_repo, record.key)
2393
find_text_keys_from_content(record)
2395
self._text_keys = content_text_keys - parent_text_keys
2396
return ('inventories', _filtered_inv_stream())
2398
def _get_text_stream(self):
2399
# Note: We know we don't have to handle adding root keys, because both
2400
# the source and target are the identical network name.
2401
text_stream = self.from_repository.texts.get_record_stream(
2402
self._text_keys, self._text_fetch_order, False)
2403
return ('texts', text_stream)
2405
def get_stream(self, search):
2406
revision_ids = search.get_keys()
2407
for stream_info in self._fetch_revision_texts(revision_ids):
2409
self._revision_keys = [(rev_id,) for rev_id in revision_ids]
2410
yield self._get_filtered_inv_stream(revision_ids)
2411
yield self._get_text_stream()
2415
class RepositoryFormatPack(MetaDirRepositoryFormat):
2416
"""Format logic for pack structured repositories.
2418
This repository format has:
2419
- a list of packs in pack-names
2420
- packs in packs/NAME.pack
2421
- indices in indices/NAME.{iix,six,tix,rix}
2422
- knit deltas in the packs, knit indices mapped to the indices.
2423
- thunk objects to support the knits programming API.
2424
- a format marker of its own
2425
- an optional 'shared-storage' flag
2426
- an optional 'no-working-trees' flag
2430
# Set this attribute in derived classes to control the repository class
2431
# created by open and initialize.
2432
repository_class = None
2433
# Set this attribute in derived classes to control the
2434
# _commit_builder_class that the repository objects will have passed to
2435
# their constructor.
2436
_commit_builder_class = None
2437
# Set this attribute in derived clases to control the _serializer that the
2438
# repository objects will have passed to their constructor.
2440
# Packs are not confused by ghosts.
2441
supports_ghosts = True
2442
# External references are not supported in pack repositories yet.
2443
supports_external_lookups = False
2444
# Most pack formats do not use chk lookups.
2445
supports_chks = False
2446
# What index classes to use
2447
index_builder_class = None
2449
_fetch_uses_deltas = True
2452
def initialize(self, a_bzrdir, shared=False):
2453
"""Create a pack based repository.
2455
:param a_bzrdir: bzrdir to contain the new repository; must already
2457
:param shared: If true the repository will be initialized as a shared
2460
mutter('creating repository in %s.', a_bzrdir.transport.base)
2461
dirs = ['indices', 'obsolete_packs', 'packs', 'upload']
2462
builder = self.index_builder_class()
2463
files = [('pack-names', builder.finish())]
2464
utf8_files = [('format', self.get_format_string())]
2466
self._upload_blank_content(a_bzrdir, dirs, files, utf8_files, shared)
2467
return self.open(a_bzrdir=a_bzrdir, _found=True)
2469
def open(self, a_bzrdir, _found=False, _override_transport=None):
2470
"""See RepositoryFormat.open().
2472
:param _override_transport: INTERNAL USE ONLY. Allows opening the
2473
repository at a slightly different url
2474
than normal. I.e. during 'upgrade'.
2477
format = RepositoryFormat.find_format(a_bzrdir)
2478
if _override_transport is not None:
2479
repo_transport = _override_transport
2481
repo_transport = a_bzrdir.get_repository_transport(None)
2482
control_files = lockable_files.LockableFiles(repo_transport,
2483
'lock', lockdir.LockDir)
2484
return self.repository_class(_format=self,
2486
control_files=control_files,
2487
_commit_builder_class=self._commit_builder_class,
2488
_serializer=self._serializer)
2491
class RepositoryFormatKnitPack1(RepositoryFormatPack):
2492
"""A no-subtrees parameterized Pack repository.
2494
This format was introduced in 0.92.
2497
repository_class = KnitPackRepository
2498
_commit_builder_class = PackCommitBuilder
2500
def _serializer(self):
2501
return xml5.serializer_v5
2502
# What index classes to use
2503
index_builder_class = InMemoryGraphIndex
2504
index_class = GraphIndex
2506
def _get_matching_bzrdir(self):
2507
return bzrdir.format_registry.make_bzrdir('pack-0.92')
2509
def _ignore_setting_bzrdir(self, format):
2512
_matchingbzrdir = property(_get_matching_bzrdir, _ignore_setting_bzrdir)
2514
def get_format_string(self):
2515
"""See RepositoryFormat.get_format_string()."""
2516
return "Bazaar pack repository format 1 (needs bzr 0.92)\n"
2518
def get_format_description(self):
2519
"""See RepositoryFormat.get_format_description()."""
2520
return "Packs containing knits without subtree support"
2522
def check_conversion_target(self, target_format):
2526
class RepositoryFormatKnitPack3(RepositoryFormatPack):
2527
"""A subtrees parameterized Pack repository.
2529
This repository format uses the xml7 serializer to get:
2530
- support for recording full info about the tree root
2531
- support for recording tree-references
2533
This format was introduced in 0.92.
2536
repository_class = KnitPackRepository
2537
_commit_builder_class = PackRootCommitBuilder
2538
rich_root_data = True
2539
supports_tree_reference = True
2541
def _serializer(self):
2542
return xml7.serializer_v7
2543
# What index classes to use
2544
index_builder_class = InMemoryGraphIndex
2545
index_class = GraphIndex
2547
def _get_matching_bzrdir(self):
2548
return bzrdir.format_registry.make_bzrdir(
2549
'pack-0.92-subtree')
2551
def _ignore_setting_bzrdir(self, format):
2554
_matchingbzrdir = property(_get_matching_bzrdir, _ignore_setting_bzrdir)
2556
def check_conversion_target(self, target_format):
2557
if not target_format.rich_root_data:
2558
raise errors.BadConversionTarget(
2559
'Does not support rich root data.', target_format)
2560
if not getattr(target_format, 'supports_tree_reference', False):
2561
raise errors.BadConversionTarget(
2562
'Does not support nested trees', target_format)
2564
def get_format_string(self):
2565
"""See RepositoryFormat.get_format_string()."""
2566
return "Bazaar pack repository format 1 with subtree support (needs bzr 0.92)\n"
2568
def get_format_description(self):
2569
"""See RepositoryFormat.get_format_description()."""
2570
return "Packs containing knits with subtree support\n"
2573
class RepositoryFormatKnitPack4(RepositoryFormatPack):
2574
"""A rich-root, no subtrees parameterized Pack repository.
2576
This repository format uses the xml6 serializer to get:
2577
- support for recording full info about the tree root
2579
This format was introduced in 1.0.
2582
repository_class = KnitPackRepository
2583
_commit_builder_class = PackRootCommitBuilder
2584
rich_root_data = True
2585
supports_tree_reference = False
2587
def _serializer(self):
2588
return xml6.serializer_v6
2589
# What index classes to use
2590
index_builder_class = InMemoryGraphIndex
2591
index_class = GraphIndex
2593
def _get_matching_bzrdir(self):
2594
return bzrdir.format_registry.make_bzrdir(
2597
def _ignore_setting_bzrdir(self, format):
2600
_matchingbzrdir = property(_get_matching_bzrdir, _ignore_setting_bzrdir)
2602
def check_conversion_target(self, target_format):
2603
if not target_format.rich_root_data:
2604
raise errors.BadConversionTarget(
2605
'Does not support rich root data.', target_format)
2607
def get_format_string(self):
2608
"""See RepositoryFormat.get_format_string()."""
2609
return ("Bazaar pack repository format 1 with rich root"
2610
" (needs bzr 1.0)\n")
2612
def get_format_description(self):
2613
"""See RepositoryFormat.get_format_description()."""
2614
return "Packs containing knits with rich root support\n"
2617
class RepositoryFormatKnitPack5(RepositoryFormatPack):
2618
"""Repository that supports external references to allow stacking.
2622
Supports external lookups, which results in non-truncated ghosts after
2623
reconcile compared to pack-0.92 formats.
2626
repository_class = KnitPackRepository
2627
_commit_builder_class = PackCommitBuilder
2628
supports_external_lookups = True
2629
# What index classes to use
2630
index_builder_class = InMemoryGraphIndex
2631
index_class = GraphIndex
2634
def _serializer(self):
2635
return xml5.serializer_v5
2637
def _get_matching_bzrdir(self):
2638
return bzrdir.format_registry.make_bzrdir('1.6')
2640
def _ignore_setting_bzrdir(self, format):
2643
_matchingbzrdir = property(_get_matching_bzrdir, _ignore_setting_bzrdir)
2645
def get_format_string(self):
2646
"""See RepositoryFormat.get_format_string()."""
2647
return "Bazaar RepositoryFormatKnitPack5 (bzr 1.6)\n"
2649
def get_format_description(self):
2650
"""See RepositoryFormat.get_format_description()."""
2651
return "Packs 5 (adds stacking support, requires bzr 1.6)"
2653
def check_conversion_target(self, target_format):
2657
class RepositoryFormatKnitPack5RichRoot(RepositoryFormatPack):
2658
"""A repository with rich roots and stacking.
2660
New in release 1.6.1.
2662
Supports stacking on other repositories, allowing data to be accessed
2663
without being stored locally.
2666
repository_class = KnitPackRepository
2667
_commit_builder_class = PackRootCommitBuilder
2668
rich_root_data = True
2669
supports_tree_reference = False # no subtrees
2670
supports_external_lookups = True
2671
# What index classes to use
2672
index_builder_class = InMemoryGraphIndex
2673
index_class = GraphIndex
2676
def _serializer(self):
2677
return xml6.serializer_v6
2679
def _get_matching_bzrdir(self):
2680
return bzrdir.format_registry.make_bzrdir(
2683
def _ignore_setting_bzrdir(self, format):
2686
_matchingbzrdir = property(_get_matching_bzrdir, _ignore_setting_bzrdir)
2688
def check_conversion_target(self, target_format):
2689
if not target_format.rich_root_data:
2690
raise errors.BadConversionTarget(
2691
'Does not support rich root data.', target_format)
2693
def get_format_string(self):
2694
"""See RepositoryFormat.get_format_string()."""
2695
return "Bazaar RepositoryFormatKnitPack5RichRoot (bzr 1.6.1)\n"
2697
def get_format_description(self):
2698
return "Packs 5 rich-root (adds stacking support, requires bzr 1.6.1)"
2701
class RepositoryFormatKnitPack5RichRootBroken(RepositoryFormatPack):
2702
"""A repository with rich roots and external references.
2706
Supports external lookups, which results in non-truncated ghosts after
2707
reconcile compared to pack-0.92 formats.
2709
This format was deprecated because the serializer it uses accidentally
2710
supported subtrees, when the format was not intended to. This meant that
2711
someone could accidentally fetch from an incorrect repository.
2714
repository_class = KnitPackRepository
2715
_commit_builder_class = PackRootCommitBuilder
2716
rich_root_data = True
2717
supports_tree_reference = False # no subtrees
2719
supports_external_lookups = True
2720
# What index classes to use
2721
index_builder_class = InMemoryGraphIndex
2722
index_class = GraphIndex
2725
def _serializer(self):
2726
return xml7.serializer_v7
2728
def _get_matching_bzrdir(self):
2729
matching = bzrdir.format_registry.make_bzrdir(
2731
matching.repository_format = self
2734
def _ignore_setting_bzrdir(self, format):
2737
_matchingbzrdir = property(_get_matching_bzrdir, _ignore_setting_bzrdir)
2739
def check_conversion_target(self, target_format):
2740
if not target_format.rich_root_data:
2741
raise errors.BadConversionTarget(
2742
'Does not support rich root data.', target_format)
2744
def get_format_string(self):
2745
"""See RepositoryFormat.get_format_string()."""
2746
return "Bazaar RepositoryFormatKnitPack5RichRoot (bzr 1.6)\n"
2748
def get_format_description(self):
2749
return ("Packs 5 rich-root (adds stacking support, requires bzr 1.6)"
2753
class RepositoryFormatKnitPack6(RepositoryFormatPack):
2754
"""A repository with stacking and btree indexes,
2755
without rich roots or subtrees.
2757
This is equivalent to pack-1.6 with B+Tree indices.
2760
repository_class = KnitPackRepository
2761
_commit_builder_class = PackCommitBuilder
2762
supports_external_lookups = True
2763
# What index classes to use
2764
index_builder_class = BTreeBuilder
2765
index_class = BTreeGraphIndex
2768
def _serializer(self):
2769
return xml5.serializer_v5
2771
def _get_matching_bzrdir(self):
2772
return bzrdir.format_registry.make_bzrdir('1.9')
2774
def _ignore_setting_bzrdir(self, format):
2777
_matchingbzrdir = property(_get_matching_bzrdir, _ignore_setting_bzrdir)
2779
def get_format_string(self):
2780
"""See RepositoryFormat.get_format_string()."""
2781
return "Bazaar RepositoryFormatKnitPack6 (bzr 1.9)\n"
2783
def get_format_description(self):
2784
"""See RepositoryFormat.get_format_description()."""
2785
return "Packs 6 (uses btree indexes, requires bzr 1.9)"
2787
def check_conversion_target(self, target_format):
2791
class RepositoryFormatKnitPack6RichRoot(RepositoryFormatPack):
2792
"""A repository with rich roots, no subtrees, stacking and btree indexes.
2794
1.6-rich-root with B+Tree indices.
2797
repository_class = KnitPackRepository
2798
_commit_builder_class = PackRootCommitBuilder
2799
rich_root_data = True
2800
supports_tree_reference = False # no subtrees
2801
supports_external_lookups = True
2802
# What index classes to use
2803
index_builder_class = BTreeBuilder
2804
index_class = BTreeGraphIndex
2807
def _serializer(self):
2808
return xml6.serializer_v6
2810
def _get_matching_bzrdir(self):
2811
return bzrdir.format_registry.make_bzrdir(
2814
def _ignore_setting_bzrdir(self, format):
2817
_matchingbzrdir = property(_get_matching_bzrdir, _ignore_setting_bzrdir)
2819
def check_conversion_target(self, target_format):
2820
if not target_format.rich_root_data:
2821
raise errors.BadConversionTarget(
2822
'Does not support rich root data.', target_format)
2824
def get_format_string(self):
2825
"""See RepositoryFormat.get_format_string()."""
2826
return "Bazaar RepositoryFormatKnitPack6RichRoot (bzr 1.9)\n"
2828
def get_format_description(self):
2829
return "Packs 6 rich-root (uses btree indexes, requires bzr 1.9)"
2832
class RepositoryFormatPackDevelopment2Subtree(RepositoryFormatPack):
2833
"""A subtrees development repository.
2835
This format should be retained until the second release after bzr 1.7.
2837
1.6.1-subtree[as it might have been] with B+Tree indices.
2839
This is [now] retained until we have a CHK based subtree format in
2843
repository_class = KnitPackRepository
2844
_commit_builder_class = PackRootCommitBuilder
2845
rich_root_data = True
2846
supports_tree_reference = True
2847
supports_external_lookups = True
2848
# What index classes to use
2849
index_builder_class = BTreeBuilder
2850
index_class = BTreeGraphIndex
2853
def _serializer(self):
2854
return xml7.serializer_v7
2856
def _get_matching_bzrdir(self):
2857
return bzrdir.format_registry.make_bzrdir(
2858
'development-subtree')
2860
def _ignore_setting_bzrdir(self, format):
2863
_matchingbzrdir = property(_get_matching_bzrdir, _ignore_setting_bzrdir)
2865
def check_conversion_target(self, target_format):
2866
if not target_format.rich_root_data:
2867
raise errors.BadConversionTarget(
2868
'Does not support rich root data.', target_format)
2869
if not getattr(target_format, 'supports_tree_reference', False):
2870
raise errors.BadConversionTarget(
2871
'Does not support nested trees', target_format)
2873
def get_format_string(self):
2874
"""See RepositoryFormat.get_format_string()."""
2875
return ("Bazaar development format 2 with subtree support "
2876
"(needs bzr.dev from before 1.8)\n")
2878
def get_format_description(self):
2879
"""See RepositoryFormat.get_format_description()."""
2880
return ("Development repository format, currently the same as "
2881
"1.6.1-subtree with B+Tree indices.\n")
added
removed
bzrlib/repofmt/pack_repo.py
