~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/repofmt/pack_repo.py

  • Committer: Aaron Bentley
  • Date: 2012-07-19 16:57:16 UTC
  • mto: This revision was merged to the branch mainline in revision 6540.
  • Revision ID: aaron@aaronbentley.com-20120719165716-b4iupzkb17b9l9wx
Avoid branch write lock to preserve VFS call count.

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2005, 2006, 2007, 2008 Canonical Ltd
 
1
# Copyright (C) 2007-2011 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
12
12
#
13
13
# You should have received a copy of the GNU General Public License
14
14
# along with this program; if not, write to the Free Software
15
 
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 
16
 
 
17
from __future__ import absolute_import
 
18
 
 
19
import re
 
20
import sys
16
21
 
17
22
from bzrlib.lazy_import import lazy_import
18
23
lazy_import(globals(), """
19
24
from itertools import izip
20
 
import math
21
 
import md5
22
25
import time
23
26
 
24
27
from bzrlib import (
25
 
        debug,
26
 
        graph,
27
 
        pack,
28
 
        ui,
29
 
        )
 
28
    chk_map,
 
29
    cleanup,
 
30
    config,
 
31
    debug,
 
32
    graph,
 
33
    osutils,
 
34
    pack,
 
35
    transactions,
 
36
    tsort,
 
37
    ui,
 
38
    )
30
39
from bzrlib.index import (
31
 
    GraphIndex,
32
 
    GraphIndexBuilder,
33
 
    InMemoryGraphIndex,
34
40
    CombinedGraphIndex,
35
41
    GraphIndexPrefixAdapter,
36
42
    )
37
 
from bzrlib.knit import KnitGraphIndex, _PackAccess, _KnitData
38
 
from bzrlib.osutils import rand_chars
39
 
from bzrlib.pack import ContainerWriter
40
 
from bzrlib.store import revision
41
 
from bzrlib import tsort
42
43
""")
43
44
from bzrlib import (
44
 
    bzrdir,
 
45
    btree_index,
45
46
    errors,
46
 
    knit,
47
47
    lockable_files,
48
48
    lockdir,
49
 
    osutils,
50
 
    symbol_versioning,
51
 
    transactions,
52
 
    xml5,
53
 
    xml6,
54
 
    xml7,
55
49
    )
56
50
 
57
 
from bzrlib.decorators import needs_read_lock, needs_write_lock
58
 
from bzrlib.repofmt.knitrepo import KnitRepository
 
51
from bzrlib.decorators import (
 
52
    needs_read_lock,
 
53
    needs_write_lock,
 
54
    only_raises,
 
55
    )
 
56
from bzrlib.lock import LogicalLockResult
59
57
from bzrlib.repository import (
60
 
    CommitBuilder,
 
58
    _LazyListJoin,
61
59
    MetaDirRepository,
62
 
    MetaDirRepositoryFormat,
63
 
    RepositoryFormat,
64
 
    RootCommitBuilder,
65
 
    )
66
 
import bzrlib.revision as _mod_revision
67
 
from bzrlib.store.revision.knit import KnitRevisionStore
68
 
from bzrlib.store.versioned import VersionedFileStore
 
60
    RepositoryFormatMetaDir,
 
61
    RepositoryWriteLockResult,
 
62
    )
 
63
from bzrlib.vf_repository import (
 
64
    MetaDirVersionedFileRepository,
 
65
    MetaDirVersionedFileRepositoryFormat,
 
66
    VersionedFileCommitBuilder,
 
67
    VersionedFileRootCommitBuilder,
 
68
    )
69
69
from bzrlib.trace import (
70
70
    mutter,
71
 
    mutter_callsite,
72
71
    note,
73
72
    warning,
74
73
    )
75
74
 
76
75
 
77
 
class PackCommitBuilder(CommitBuilder):
78
 
    """A subclass of CommitBuilder to add texts with pack semantics.
79
 
    
 
76
class PackCommitBuilder(VersionedFileCommitBuilder):
 
77
    """Subclass of VersionedFileCommitBuilder to add texts with pack semantics.
 
78
 
80
79
    Specifically this uses one knit object rather than one knit object per
81
80
    added text, reducing memory and object pressure.
82
81
    """
83
82
 
84
83
    def __init__(self, repository, parents, config, timestamp=None,
85
84
                 timezone=None, committer=None, revprops=None,
86
 
                 revision_id=None):
87
 
        CommitBuilder.__init__(self, repository, parents, config,
 
85
                 revision_id=None, lossy=False):
 
86
        VersionedFileCommitBuilder.__init__(self, repository, parents, config,
88
87
            timestamp=timestamp, timezone=timezone, committer=committer,
89
 
            revprops=revprops, revision_id=revision_id)
 
88
            revprops=revprops, revision_id=revision_id, lossy=lossy)
90
89
        self._file_graph = graph.Graph(
91
90
            repository._pack_collection.text_index.combined_index)
92
91
 
93
 
    def _add_text_to_weave(self, file_id, new_lines, parents, nostore_sha):
94
 
        return self.repository._pack_collection._add_text_to_weave(file_id,
95
 
            self._new_revision_id, new_lines, parents, nostore_sha,
96
 
            self.random_revid)
97
 
 
98
92
    def _heads(self, file_id, revision_ids):
99
93
        keys = [(file_id, revision_id) for revision_id in revision_ids]
100
94
        return set([key[1] for key in self._file_graph.heads(keys)])
101
95
 
102
96
 
103
 
class PackRootCommitBuilder(RootCommitBuilder):
 
97
class PackRootCommitBuilder(VersionedFileRootCommitBuilder):
104
98
    """A subclass of RootCommitBuilder to add texts with pack semantics.
105
 
    
 
99
 
106
100
    Specifically this uses one knit object rather than one knit object per
107
101
    added text, reducing memory and object pressure.
108
102
    """
109
103
 
110
104
    def __init__(self, repository, parents, config, timestamp=None,
111
105
                 timezone=None, committer=None, revprops=None,
112
 
                 revision_id=None):
113
 
        CommitBuilder.__init__(self, repository, parents, config,
114
 
            timestamp=timestamp, timezone=timezone, committer=committer,
115
 
            revprops=revprops, revision_id=revision_id)
 
106
                 revision_id=None, lossy=False):
 
107
        super(PackRootCommitBuilder, self).__init__(repository, parents,
 
108
            config, timestamp=timestamp, timezone=timezone,
 
109
            committer=committer, revprops=revprops, revision_id=revision_id,
 
110
            lossy=lossy)
116
111
        self._file_graph = graph.Graph(
117
112
            repository._pack_collection.text_index.combined_index)
118
113
 
119
 
    def _add_text_to_weave(self, file_id, new_lines, parents, nostore_sha):
120
 
        return self.repository._pack_collection._add_text_to_weave(file_id,
121
 
            self._new_revision_id, new_lines, parents, nostore_sha,
122
 
            self.random_revid)
123
 
 
124
114
    def _heads(self, file_id, revision_ids):
125
115
        keys = [(file_id, revision_id) for revision_id in revision_ids]
126
116
        return set([key[1] for key in self._file_graph.heads(keys)])
133
123
    ExistingPack and NewPack are used.
134
124
    """
135
125
 
 
126
    # A map of index 'type' to the file extension and position in the
 
127
    # index_sizes array.
 
128
    index_definitions = {
 
129
        'chk': ('.cix', 4),
 
130
        'revision': ('.rix', 0),
 
131
        'inventory': ('.iix', 1),
 
132
        'text': ('.tix', 2),
 
133
        'signature': ('.six', 3),
 
134
        }
 
135
 
136
136
    def __init__(self, revision_index, inventory_index, text_index,
137
 
        signature_index):
 
137
        signature_index, chk_index=None):
138
138
        """Create a pack instance.
139
139
 
140
140
        :param revision_index: A GraphIndex for determining what revisions are
145
145
        :param text_index: A GraphIndex for determining what file texts
146
146
            are present in the pack and accessing the locations of their
147
147
            texts/deltas (via (fileid, revisionid) tuples).
148
 
        :param revision_index: A GraphIndex for determining what signatures are
 
148
        :param signature_index: A GraphIndex for determining what signatures are
149
149
            present in the Pack and accessing the locations of their texts.
 
150
        :param chk_index: A GraphIndex for accessing content by CHK, if the
 
151
            pack has one.
150
152
        """
151
153
        self.revision_index = revision_index
152
154
        self.inventory_index = inventory_index
153
155
        self.text_index = text_index
154
156
        self.signature_index = signature_index
 
157
        self.chk_index = chk_index
155
158
 
156
159
    def access_tuple(self):
157
160
        """Return a tuple (transport, name) for the pack content."""
158
161
        return self.pack_transport, self.file_name()
159
162
 
 
163
    def _check_references(self):
 
164
        """Make sure our external references are present.
 
165
 
 
166
        Packs are allowed to have deltas whose base is not in the pack, but it
 
167
        must be present somewhere in this collection.  It is not allowed to
 
168
        have deltas based on a fallback repository.
 
169
        (See <https://bugs.launchpad.net/bzr/+bug/288751>)
 
170
        """
 
171
        missing_items = {}
 
172
        for (index_name, external_refs, index) in [
 
173
            ('texts',
 
174
                self._get_external_refs(self.text_index),
 
175
                self._pack_collection.text_index.combined_index),
 
176
            ('inventories',
 
177
                self._get_external_refs(self.inventory_index),
 
178
                self._pack_collection.inventory_index.combined_index),
 
179
            ]:
 
180
            missing = external_refs.difference(
 
181
                k for (idx, k, v, r) in
 
182
                index.iter_entries(external_refs))
 
183
            if missing:
 
184
                missing_items[index_name] = sorted(list(missing))
 
185
        if missing_items:
 
186
            from pprint import pformat
 
187
            raise errors.BzrCheckError(
 
188
                "Newly created pack file %r has delta references to "
 
189
                "items not in its repository:\n%s"
 
190
                % (self, pformat(missing_items)))
 
191
 
160
192
    def file_name(self):
161
193
        """Get the file name for the pack on disk."""
162
194
        return self.name + '.pack'
164
196
    def get_revision_count(self):
165
197
        return self.revision_index.key_count()
166
198
 
 
199
    def index_name(self, index_type, name):
 
200
        """Get the disk name of an index type for pack name 'name'."""
 
201
        return name + Pack.index_definitions[index_type][0]
 
202
 
 
203
    def index_offset(self, index_type):
 
204
        """Get the position in a index_size array for a given index type."""
 
205
        return Pack.index_definitions[index_type][1]
 
206
 
167
207
    def inventory_index_name(self, name):
168
208
        """The inv index is the name + .iix."""
169
209
        return self.index_name('inventory', name)
180
220
        """The text index is the name + .tix."""
181
221
        return self.index_name('text', name)
182
222
 
183
 
    def _external_compression_parents_of_texts(self):
184
 
        keys = set()
185
 
        refs = set()
186
 
        for node in self.text_index.iter_all_entries():
187
 
            keys.add(node[1])
188
 
            refs.update(node[3][1])
189
 
        return refs - keys
 
223
    def _replace_index_with_readonly(self, index_type):
 
224
        unlimited_cache = False
 
225
        if index_type == 'chk':
 
226
            unlimited_cache = True
 
227
        index = self.index_class(self.index_transport,
 
228
                    self.index_name(index_type, self.name),
 
229
                    self.index_sizes[self.index_offset(index_type)],
 
230
                    unlimited_cache=unlimited_cache)
 
231
        if index_type == 'chk':
 
232
            index._leaf_factory = btree_index._gcchk_factory
 
233
        setattr(self, index_type + '_index', index)
190
234
 
191
235
 
192
236
class ExistingPack(Pack):
193
237
    """An in memory proxy for an existing .pack and its disk indices."""
194
238
 
195
239
    def __init__(self, pack_transport, name, revision_index, inventory_index,
196
 
        text_index, signature_index):
 
240
        text_index, signature_index, chk_index=None):
197
241
        """Create an ExistingPack object.
198
242
 
199
243
        :param pack_transport: The transport where the pack file resides.
200
244
        :param name: The name of the pack on disk in the pack_transport.
201
245
        """
202
246
        Pack.__init__(self, revision_index, inventory_index, text_index,
203
 
            signature_index)
 
247
            signature_index, chk_index)
204
248
        self.name = name
205
249
        self.pack_transport = pack_transport
206
250
        if None in (revision_index, inventory_index, text_index,
214
258
        return not self.__eq__(other)
215
259
 
216
260
    def __repr__(self):
217
 
        return "<bzrlib.repofmt.pack_repo.Pack object at 0x%x, %s, %s" % (
218
 
            id(self), self.transport, self.name)
 
261
        return "<%s.%s object at 0x%x, %s, %s" % (
 
262
            self.__class__.__module__, self.__class__.__name__, id(self),
 
263
            self.pack_transport, self.name)
 
264
 
 
265
 
 
266
class ResumedPack(ExistingPack):
 
267
 
 
268
    def __init__(self, name, revision_index, inventory_index, text_index,
 
269
        signature_index, upload_transport, pack_transport, index_transport,
 
270
        pack_collection, chk_index=None):
 
271
        """Create a ResumedPack object."""
 
272
        ExistingPack.__init__(self, pack_transport, name, revision_index,
 
273
            inventory_index, text_index, signature_index,
 
274
            chk_index=chk_index)
 
275
        self.upload_transport = upload_transport
 
276
        self.index_transport = index_transport
 
277
        self.index_sizes = [None, None, None, None]
 
278
        indices = [
 
279
            ('revision', revision_index),
 
280
            ('inventory', inventory_index),
 
281
            ('text', text_index),
 
282
            ('signature', signature_index),
 
283
            ]
 
284
        if chk_index is not None:
 
285
            indices.append(('chk', chk_index))
 
286
            self.index_sizes.append(None)
 
287
        for index_type, index in indices:
 
288
            offset = self.index_offset(index_type)
 
289
            self.index_sizes[offset] = index._size
 
290
        self.index_class = pack_collection._index_class
 
291
        self._pack_collection = pack_collection
 
292
        self._state = 'resumed'
 
293
        # XXX: perhaps check that the .pack file exists?
 
294
 
 
295
    def access_tuple(self):
 
296
        if self._state == 'finished':
 
297
            return Pack.access_tuple(self)
 
298
        elif self._state == 'resumed':
 
299
            return self.upload_transport, self.file_name()
 
300
        else:
 
301
            raise AssertionError(self._state)
 
302
 
 
303
    def abort(self):
 
304
        self.upload_transport.delete(self.file_name())
 
305
        indices = [self.revision_index, self.inventory_index, self.text_index,
 
306
            self.signature_index]
 
307
        if self.chk_index is not None:
 
308
            indices.append(self.chk_index)
 
309
        for index in indices:
 
310
            index._transport.delete(index._name)
 
311
 
 
312
    def finish(self):
 
313
        self._check_references()
 
314
        index_types = ['revision', 'inventory', 'text', 'signature']
 
315
        if self.chk_index is not None:
 
316
            index_types.append('chk')
 
317
        for index_type in index_types:
 
318
            old_name = self.index_name(index_type, self.name)
 
319
            new_name = '../indices/' + old_name
 
320
            self.upload_transport.move(old_name, new_name)
 
321
            self._replace_index_with_readonly(index_type)
 
322
        new_name = '../packs/' + self.file_name()
 
323
        self.upload_transport.move(self.file_name(), new_name)
 
324
        self._state = 'finished'
 
325
 
 
326
    def _get_external_refs(self, index):
 
327
        """Return compression parents for this index that are not present.
 
328
 
 
329
        This returns any compression parents that are referenced by this index,
 
330
        which are not contained *in* this index. They may be present elsewhere.
 
331
        """
 
332
        return index.external_references(1)
219
333
 
220
334
 
221
335
class NewPack(Pack):
222
336
    """An in memory proxy for a pack which is being created."""
223
337
 
224
 
    # A map of index 'type' to the file extension and position in the
225
 
    # index_sizes array.
226
 
    index_definitions = {
227
 
        'revision': ('.rix', 0),
228
 
        'inventory': ('.iix', 1),
229
 
        'text': ('.tix', 2),
230
 
        'signature': ('.six', 3),
231
 
        }
232
 
 
233
 
    def __init__(self, upload_transport, index_transport, pack_transport,
234
 
        upload_suffix='', file_mode=None):
 
338
    def __init__(self, pack_collection, upload_suffix='', file_mode=None):
235
339
        """Create a NewPack instance.
236
340
 
237
 
        :param upload_transport: A writable transport for the pack to be
238
 
            incrementally uploaded to.
239
 
        :param index_transport: A writable transport for the pack's indices to
240
 
            be written to when the pack is finished.
241
 
        :param pack_transport: A writable transport for the pack to be renamed
242
 
            to when the upload is complete. This *must* be the same as
243
 
            upload_transport.clone('../packs').
 
341
        :param pack_collection: A PackCollection into which this is being inserted.
244
342
        :param upload_suffix: An optional suffix to be given to any temporary
245
343
            files created during the pack creation. e.g '.autopack'
246
 
        :param file_mode: An optional file mode to create the new files with.
 
344
        :param file_mode: Unix permissions for newly created file.
247
345
        """
248
346
        # The relative locations of the packs are constrained, but all are
249
347
        # passed in because the caller has them, so as to avoid object churn.
 
348
        index_builder_class = pack_collection._index_builder_class
 
349
        if pack_collection.chk_index is not None:
 
350
            chk_index = index_builder_class(reference_lists=0)
 
351
        else:
 
352
            chk_index = None
250
353
        Pack.__init__(self,
251
354
            # Revisions: parents list, no text compression.
252
 
            InMemoryGraphIndex(reference_lists=1),
 
355
            index_builder_class(reference_lists=1),
253
356
            # Inventory: We want to map compression only, but currently the
254
357
            # knit code hasn't been updated enough to understand that, so we
255
358
            # have a regular 2-list index giving parents and compression
256
359
            # source.
257
 
            InMemoryGraphIndex(reference_lists=2),
 
360
            index_builder_class(reference_lists=2),
258
361
            # Texts: compression and per file graph, for all fileids - so two
259
362
            # reference lists and two elements in the key tuple.
260
 
            InMemoryGraphIndex(reference_lists=2, key_elements=2),
 
363
            index_builder_class(reference_lists=2, key_elements=2),
261
364
            # Signatures: Just blobs to store, no compression, no parents
262
365
            # listing.
263
 
            InMemoryGraphIndex(reference_lists=0),
 
366
            index_builder_class(reference_lists=0),
 
367
            # CHK based storage - just blobs, no compression or parents.
 
368
            chk_index=chk_index
264
369
            )
 
370
        self._pack_collection = pack_collection
 
371
        # When we make readonly indices, we need this.
 
372
        self.index_class = pack_collection._index_class
265
373
        # where should the new pack be opened
266
 
        self.upload_transport = upload_transport
 
374
        self.upload_transport = pack_collection._upload_transport
267
375
        # where are indices written out to
268
 
        self.index_transport = index_transport
 
376
        self.index_transport = pack_collection._index_transport
269
377
        # where is the pack renamed to when it is finished?
270
 
        self.pack_transport = pack_transport
 
378
        self.pack_transport = pack_collection._pack_transport
271
379
        # What file mode to upload the pack and indices with.
272
380
        self._file_mode = file_mode
273
381
        # tracks the content written to the .pack file.
274
 
        self._hash = md5.new()
275
 
        # a four-tuple with the length in bytes of the indices, once the pack
276
 
        # is finalised. (rev, inv, text, sigs)
 
382
        self._hash = osutils.md5()
 
383
        # a tuple with the length in bytes of the indices, once the pack
 
384
        # is finalised. (rev, inv, text, sigs, chk_if_in_use)
277
385
        self.index_sizes = None
278
386
        # How much data to cache when writing packs. Note that this is not
279
387
        # synchronised with reads, because it's not in the transport layer, so
281
389
        # under creation.
282
390
        self._cache_limit = 0
283
391
        # the temporary pack file name.
284
 
        self.random_name = rand_chars(20) + upload_suffix
 
392
        self.random_name = osutils.rand_chars(20) + upload_suffix
285
393
        # when was this pack started ?
286
394
        self.start_time = time.time()
287
395
        # open an output stream for the data added to the pack.
291
399
            mutter('%s: create_pack: pack stream open: %s%s t+%6.3fs',
292
400
                time.ctime(), self.upload_transport.base, self.random_name,
293
401
                time.time() - self.start_time)
294
 
        # A list of byte sequences to be written to the new pack, and the 
295
 
        # aggregate size of them.  Stored as a list rather than separate 
 
402
        # A list of byte sequences to be written to the new pack, and the
 
403
        # aggregate size of them.  Stored as a list rather than separate
296
404
        # variables so that the _write_data closure below can update them.
297
405
        self._buffer = [[], 0]
298
 
        # create a callable for adding data 
 
406
        # create a callable for adding data
299
407
        #
300
408
        # robertc says- this is a closure rather than a method on the object
301
409
        # so that the variables are locals, and faster than accessing object
317
425
        self._writer.begin()
318
426
        # what state is the pack in? (open, finished, aborted)
319
427
        self._state = 'open'
 
428
        # no name until we finish writing the content
 
429
        self.name = None
320
430
 
321
431
    def abort(self):
322
432
        """Cancel creating this pack."""
340
450
        return bool(self.get_revision_count() or
341
451
            self.inventory_index.key_count() or
342
452
            self.text_index.key_count() or
343
 
            self.signature_index.key_count())
344
 
 
345
 
    def finish(self):
 
453
            self.signature_index.key_count() or
 
454
            (self.chk_index is not None and self.chk_index.key_count()))
 
455
 
 
456
    def finish_content(self):
 
457
        if self.name is not None:
 
458
            return
 
459
        self._writer.end()
 
460
        if self._buffer[1]:
 
461
            self._write_data('', flush=True)
 
462
        self.name = self._hash.hexdigest()
 
463
 
 
464
    def finish(self, suspend=False):
346
465
        """Finish the new pack.
347
466
 
348
467
        This:
353
472
         - stores the index size tuple for the pack in the index_sizes
354
473
           attribute.
355
474
        """
356
 
        self._writer.end()
357
 
        if self._buffer[1]:
358
 
            self._write_data('', flush=True)
359
 
        self.name = self._hash.hexdigest()
 
475
        self.finish_content()
 
476
        if not suspend:
 
477
            self._check_references()
360
478
        # write indices
361
479
        # XXX: It'd be better to write them all to temporary names, then
362
480
        # rename them all into place, so that the window when only some are
363
481
        # visible is smaller.  On the other hand none will be seen until
364
482
        # they're in the names list.
365
483
        self.index_sizes = [None, None, None, None]
366
 
        self._write_index('revision', self.revision_index, 'revision')
367
 
        self._write_index('inventory', self.inventory_index, 'inventory')
368
 
        self._write_index('text', self.text_index, 'file texts')
 
484
        self._write_index('revision', self.revision_index, 'revision',
 
485
            suspend)
 
486
        self._write_index('inventory', self.inventory_index, 'inventory',
 
487
            suspend)
 
488
        self._write_index('text', self.text_index, 'file texts', suspend)
369
489
        self._write_index('signature', self.signature_index,
370
 
            'revision signatures')
371
 
        self.write_stream.close()
 
490
            'revision signatures', suspend)
 
491
        if self.chk_index is not None:
 
492
            self.index_sizes.append(None)
 
493
            self._write_index('chk', self.chk_index,
 
494
                'content hash bytes', suspend)
 
495
        self.write_stream.close(
 
496
            want_fdatasync=self._pack_collection.config_stack.get('repository.fdatasync'))
372
497
        # Note that this will clobber an existing pack with the same name,
373
498
        # without checking for hash collisions. While this is undesirable this
374
499
        # is something that can be rectified in a subsequent release. One way
380
505
        #  - try for HASH.pack
381
506
        #  - try for temporary-name
382
507
        #  - refresh the pack-list to see if the pack is now absent
383
 
        self.upload_transport.rename(self.random_name,
384
 
                '../packs/' + self.name + '.pack')
 
508
        new_name = self.name + '.pack'
 
509
        if not suspend:
 
510
            new_name = '../packs/' + new_name
 
511
        self.upload_transport.move(self.random_name, new_name)
385
512
        self._state = 'finished'
386
513
        if 'pack' in debug.debug_flags:
387
514
            # XXX: size might be interesting?
388
 
            mutter('%s: create_pack: pack renamed into place: %s%s->%s%s t+%6.3fs',
 
515
            mutter('%s: create_pack: pack finished: %s%s->%s t+%6.3fs',
389
516
                time.ctime(), self.upload_transport.base, self.random_name,
390
 
                self.pack_transport, self.name,
391
 
                time.time() - self.start_time)
 
517
                new_name, time.time() - self.start_time)
392
518
 
393
519
    def flush(self):
394
520
        """Flush any current data."""
398
524
            self._hash.update(bytes)
399
525
            self._buffer[:] = [[], 0]
400
526
 
401
 
    def index_name(self, index_type, name):
402
 
        """Get the disk name of an index type for pack name 'name'."""
403
 
        return name + NewPack.index_definitions[index_type][0]
404
 
 
405
 
    def index_offset(self, index_type):
406
 
        """Get the position in a index_size array for a given index type."""
407
 
        return NewPack.index_definitions[index_type][1]
408
 
 
409
 
    def _replace_index_with_readonly(self, index_type):
410
 
        setattr(self, index_type + '_index',
411
 
            GraphIndex(self.index_transport,
412
 
                self.index_name(index_type, self.name),
413
 
                self.index_sizes[self.index_offset(index_type)]))
 
527
    def _get_external_refs(self, index):
 
528
        return index._external_references()
414
529
 
415
530
    def set_write_cache_size(self, size):
416
531
        self._cache_limit = size
417
532
 
418
 
    def _write_index(self, index_type, index, label):
 
533
    def _write_index(self, index_type, index, label, suspend=False):
419
534
        """Write out an index.
420
535
 
421
536
        :param index_type: The type of index to write - e.g. 'revision'.
423
538
        :param label: What label to give the index e.g. 'revision'.
424
539
        """
425
540
        index_name = self.index_name(index_type, self.name)
426
 
        self.index_sizes[self.index_offset(index_type)] = \
427
 
            self.index_transport.put_file(index_name, index.finish(),
 
541
        if suspend:
 
542
            transport = self.upload_transport
 
543
        else:
 
544
            transport = self.index_transport
 
545
        index_tempfile = index.finish()
 
546
        index_bytes = index_tempfile.read()
 
547
        write_stream = transport.open_write_stream(index_name,
428
548
            mode=self._file_mode)
 
549
        write_stream.write(index_bytes)
 
550
        write_stream.close(
 
551
            want_fdatasync=self._pack_collection.config_stack.get('repository.fdatasync'))
 
552
        self.index_sizes[self.index_offset(index_type)] = len(index_bytes)
429
553
        if 'pack' in debug.debug_flags:
430
554
            # XXX: size might be interesting?
431
555
            mutter('%s: create_pack: wrote %s index: %s%s t+%6.3fs',
432
556
                time.ctime(), label, self.upload_transport.base,
433
557
                self.random_name, time.time() - self.start_time)
434
 
        # Replace the writable index on this object with a readonly, 
 
558
        # Replace the writable index on this object with a readonly,
435
559
        # presently unloaded index. We should alter
436
560
        # the index layer to make its finish() error if add_node is
437
561
        # subsequently used. RBC
446
570
    such as 'revision index'.
447
571
 
448
572
    A CombinedIndex provides an index on a single key space built up
449
 
    from several on-disk indices.  The AggregateIndex builds on this 
 
573
    from several on-disk indices.  The AggregateIndex builds on this
450
574
    to provide a knit access layer, and allows having up to one writable
451
575
    index within the collection.
452
576
    """
453
577
    # XXX: Probably 'can be written to' could/should be separated from 'acts
454
578
    # like a knit index' -- mbp 20071024
455
579
 
456
 
    def __init__(self):
457
 
        """Create an AggregateIndex."""
 
580
    def __init__(self, reload_func=None, flush_func=None):
 
581
        """Create an AggregateIndex.
 
582
 
 
583
        :param reload_func: A function to call if we find we are missing an
 
584
            index. Should have the form reload_func() => True if the list of
 
585
            active pack files has changed.
 
586
        """
 
587
        self._reload_func = reload_func
458
588
        self.index_to_pack = {}
459
 
        self.combined_index = CombinedGraphIndex([])
460
 
        self.knit_access = _PackAccess(self.index_to_pack)
461
 
 
462
 
    def replace_indices(self, index_to_pack, indices):
463
 
        """Replace the current mappings with fresh ones.
464
 
 
465
 
        This should probably not be used eventually, rather incremental add and
466
 
        removal of indices. It has been added during refactoring of existing
467
 
        code.
468
 
 
469
 
        :param index_to_pack: A mapping from index objects to
470
 
            (transport, name) tuples for the pack file data.
471
 
        :param indices: A list of indices.
472
 
        """
473
 
        # refresh the revision pack map dict without replacing the instance.
474
 
        self.index_to_pack.clear()
475
 
        self.index_to_pack.update(index_to_pack)
476
 
        # XXX: API break - clearly a 'replace' method would be good?
477
 
        self.combined_index._indices[:] = indices
478
 
        # the current add nodes callback for the current writable index if
479
 
        # there is one.
 
589
        self.combined_index = CombinedGraphIndex([], reload_func=reload_func)
 
590
        self.data_access = _DirectPackAccess(self.index_to_pack,
 
591
                                             reload_func=reload_func,
 
592
                                             flush_func=flush_func)
480
593
        self.add_callback = None
481
594
 
482
595
    def add_index(self, index, pack):
484
597
 
485
598
        Future searches on the aggregate index will seach this new index
486
599
        before all previously inserted indices.
487
 
        
 
600
 
488
601
        :param index: An Index for the pack.
489
602
        :param pack: A Pack instance.
490
603
        """
491
604
        # expose it to the index map
492
605
        self.index_to_pack[index] = pack.access_tuple()
493
606
        # put it at the front of the linear index list
494
 
        self.combined_index.insert_index(0, index)
 
607
        self.combined_index.insert_index(0, index, pack.name)
495
608
 
496
609
    def add_writable_index(self, index, pack):
497
610
        """Add an index which is able to have data added to it.
498
611
 
499
612
        There can be at most one writable index at any time.  Any
500
613
        modifications made to the knit are put into this index.
501
 
        
 
614
 
502
615
        :param index: An index from the pack parameter.
503
616
        :param pack: A Pack instance.
504
617
        """
509
622
        # allow writing: queue writes to a new index
510
623
        self.add_index(index, pack)
511
624
        # Updates the index to packs mapping as a side effect,
512
 
        self.knit_access.set_writer(pack._writer, index, pack.access_tuple())
 
625
        self.data_access.set_writer(pack._writer, index, pack.access_tuple())
513
626
        self.add_callback = index.add_nodes
514
627
 
515
628
    def clear(self):
516
629
        """Reset all the aggregate data to nothing."""
517
 
        self.knit_access.set_writer(None, None, (None, None))
 
630
        self.data_access.set_writer(None, None, (None, None))
518
631
        self.index_to_pack.clear()
519
632
        del self.combined_index._indices[:]
 
633
        del self.combined_index._index_names[:]
520
634
        self.add_callback = None
521
635
 
522
 
    def remove_index(self, index, pack):
 
636
    def remove_index(self, index):
523
637
        """Remove index from the indices used to answer queries.
524
 
        
 
638
 
525
639
        :param index: An index from the pack parameter.
526
 
        :param pack: A Pack instance.
527
640
        """
528
641
        del self.index_to_pack[index]
529
 
        self.combined_index._indices.remove(index)
 
642
        pos = self.combined_index._indices.index(index)
 
643
        del self.combined_index._indices[pos]
 
644
        del self.combined_index._index_names[pos]
530
645
        if (self.add_callback is not None and
531
646
            getattr(index, 'add_nodes', None) == self.add_callback):
532
647
            self.add_callback = None
533
 
            self.knit_access.set_writer(None, None, (None, None))
 
648
            self.data_access.set_writer(None, None, (None, None))
534
649
 
535
650
 
536
651
class Packer(object):
537
652
    """Create a pack from packs."""
538
653
 
539
 
    def __init__(self, pack_collection, packs, suffix, revision_ids=None):
 
654
    def __init__(self, pack_collection, packs, suffix, revision_ids=None,
 
655
                 reload_func=None):
540
656
        """Create a Packer.
541
657
 
542
658
        :param pack_collection: A RepositoryPackCollection object where the
544
660
        :param packs: The packs to combine.
545
661
        :param suffix: The suffix to use on the temporary files for the pack.
546
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
547
666
        """
548
667
        self.packs = packs
549
668
        self.suffix = suffix
551
670
        # The pack object we are creating.
552
671
        self.new_pack = None
553
672
        self._pack_collection = pack_collection
 
673
        self._reload_func = reload_func
554
674
        # The index layer keys for the revisions being copied. None for 'all
555
675
        # objects'.
556
676
        self._revision_keys = None
557
677
        # What text keys to copy. None for 'all texts'. This is set by
558
678
        # _copy_inventory_texts
559
679
        self._text_filter = None
560
 
        self._extra_init()
561
 
 
562
 
    def _extra_init(self):
563
 
        """A template hook to allow extending the constructor trivially."""
564
680
 
565
681
    def pack(self, pb=None):
566
682
        """Create a new pack by reading data from other packs.
568
684
        This does little more than a bulk copy of data. One key difference
569
685
        is that data with the same item key across multiple packs is elided
570
686
        from the output. The new pack is written into the current pack store
571
 
        along with its indices, and the name added to the pack names. The 
 
687
        along with its indices, and the name added to the pack names. The
572
688
        source packs are not altered and are not required to be in the current
573
689
        pack collection.
574
690
 
577
693
        :return: A Pack object, or None if nothing was copied.
578
694
        """
579
695
        # open a pack - using the same name as the last temporary file
580
 
        # - which has already been flushed, so its safe.
 
696
        # - which has already been flushed, so it's safe.
581
697
        # XXX: - duplicate code warning with start_write_group; fix before
582
698
        #      considering 'done'.
583
699
        if self._pack_collection._new_pack is not None:
584
 
            raise errors.BzrError('call to create_pack_from_packs while '
585
 
                'another pack is being written.')
 
700
            raise errors.BzrError('call to %s.pack() while another pack is'
 
701
                                  ' being written.'
 
702
                                  % (self.__class__.__name__,))
586
703
        if self.revision_ids is not None:
587
704
            if len(self.revision_ids) == 0:
588
705
                # silly fetch request.
589
706
                return None
590
707
            else:
591
708
                self.revision_ids = frozenset(self.revision_ids)
 
709
                self.revision_keys = frozenset((revid,) for revid in
 
710
                    self.revision_ids)
592
711
        if pb is None:
593
712
            self.pb = ui.ui_factory.nested_progress_bar()
594
713
        else:
601
720
 
602
721
    def open_pack(self):
603
722
        """Open a pack for the pack we are creating."""
604
 
        return NewPack(self._pack_collection._upload_transport,
605
 
            self._pack_collection._index_transport,
606
 
            self._pack_collection._pack_transport, upload_suffix=self.suffix,
607
 
            file_mode=self._pack_collection.repo.control_files._file_mode)
 
723
        new_pack = self._pack_collection.pack_factory(self._pack_collection,
 
724
                upload_suffix=self.suffix,
 
725
                file_mode=self._pack_collection.repo.bzrdir._get_file_mode())
 
726
        # We know that we will process all nodes in order, and don't need to
 
727
        # query, so don't combine any indices spilled to disk until we are done
 
728
        new_pack.revision_index.set_optimize(combine_backing_indices=False)
 
729
        new_pack.inventory_index.set_optimize(combine_backing_indices=False)
 
730
        new_pack.text_index.set_optimize(combine_backing_indices=False)
 
731
        new_pack.signature_index.set_optimize(combine_backing_indices=False)
 
732
        return new_pack
608
733
 
609
734
    def _copy_revision_texts(self):
610
735
        """Copy revision data to the new pack."""
611
 
        # select revisions
612
 
        if self.revision_ids:
613
 
            revision_keys = [(revision_id,) for revision_id in self.revision_ids]
614
 
        else:
615
 
            revision_keys = None
616
 
        # select revision keys
617
 
        revision_index_map = self._pack_collection._packs_list_to_pack_map_and_index_list(
618
 
            self.packs, 'revision_index')[0]
619
 
        revision_nodes = self._pack_collection._index_contents(revision_index_map, revision_keys)
620
 
        # copy revision keys and adjust values
621
 
        self.pb.update("Copying revision texts", 1)
622
 
        total_items, readv_group_iter = self._revision_node_readv(revision_nodes)
623
 
        list(self._copy_nodes_graph(revision_index_map, self.new_pack._writer,
624
 
            self.new_pack.revision_index, readv_group_iter, total_items))
625
 
        if 'pack' in debug.debug_flags:
626
 
            mutter('%s: create_pack: revisions copied: %s%s %d items t+%6.3fs',
627
 
                time.ctime(), self._pack_collection._upload_transport.base,
628
 
                self.new_pack.random_name,
629
 
                self.new_pack.revision_index.key_count(),
630
 
                time.time() - self.new_pack.start_time)
631
 
        self._revision_keys = revision_keys
 
736
        raise NotImplementedError(self._copy_revision_texts)
632
737
 
633
738
    def _copy_inventory_texts(self):
634
739
        """Copy the inventory texts to the new pack.
637
742
 
638
743
        Sets self._text_filter appropriately.
639
744
        """
640
 
        # select inventory keys
641
 
        inv_keys = self._revision_keys # currently the same keyspace, and note that
642
 
        # querying for keys here could introduce a bug where an inventory item
643
 
        # is missed, so do not change it to query separately without cross
644
 
        # checking like the text key check below.
645
 
        inventory_index_map = self._pack_collection._packs_list_to_pack_map_and_index_list(
646
 
            self.packs, 'inventory_index')[0]
647
 
        inv_nodes = self._pack_collection._index_contents(inventory_index_map, inv_keys)
648
 
        # copy inventory keys and adjust values
649
 
        # XXX: Should be a helper function to allow different inv representation
650
 
        # at this point.
651
 
        self.pb.update("Copying inventory texts", 2)
652
 
        total_items, readv_group_iter = self._least_readv_node_readv(inv_nodes)
653
 
        # Only grab the output lines if we will be processing them
654
 
        output_lines = bool(self.revision_ids)
655
 
        inv_lines = self._copy_nodes_graph(inventory_index_map,
656
 
            self.new_pack._writer, self.new_pack.inventory_index,
657
 
            readv_group_iter, total_items, output_lines=output_lines)
658
 
        if self.revision_ids:
659
 
            self._process_inventory_lines(inv_lines)
660
 
        else:
661
 
            # eat the iterator to cause it to execute.
662
 
            list(inv_lines)
663
 
            self._text_filter = None
664
 
        if 'pack' in debug.debug_flags:
665
 
            mutter('%s: create_pack: inventories copied: %s%s %d items t+%6.3fs',
666
 
                time.ctime(), self._pack_collection._upload_transport.base,
667
 
                self.new_pack.random_name,
668
 
                self.new_pack.inventory_index.key_count(),
669
 
                time.time() - self.new_pack.start_time)
 
745
        raise NotImplementedError(self._copy_inventory_texts)
670
746
 
671
747
    def _copy_text_texts(self):
672
 
        # select text keys
673
 
        text_index_map, text_nodes = self._get_text_nodes()
674
 
        if self._text_filter is not None:
675
 
            # We could return the keys copied as part of the return value from
676
 
            # _copy_nodes_graph but this doesn't work all that well with the
677
 
            # need to get line output too, so we check separately, and as we're
678
 
            # going to buffer everything anyway, we check beforehand, which
679
 
            # saves reading knit data over the wire when we know there are
680
 
            # mising records.
681
 
            text_nodes = set(text_nodes)
682
 
            present_text_keys = set(_node[1] for _node in text_nodes)
683
 
            missing_text_keys = set(self._text_filter) - present_text_keys
684
 
            if missing_text_keys:
685
 
                # TODO: raise a specific error that can handle many missing
686
 
                # keys.
687
 
                a_missing_key = missing_text_keys.pop()
688
 
                raise errors.RevisionNotPresent(a_missing_key[1],
689
 
                    a_missing_key[0])
690
 
        # copy text keys and adjust values
691
 
        self.pb.update("Copying content texts", 3)
692
 
        total_items, readv_group_iter = self._least_readv_node_readv(text_nodes)
693
 
        list(self._copy_nodes_graph(text_index_map, self.new_pack._writer,
694
 
            self.new_pack.text_index, readv_group_iter, total_items))
695
 
        self._log_copied_texts()
696
 
 
697
 
    def _check_references(self):
698
 
        """Make sure our external refereneces are present."""
699
 
        external_refs = self.new_pack._external_compression_parents_of_texts()
700
 
        if external_refs:
701
 
            index = self._pack_collection.text_index.combined_index
702
 
            found_items = list(index.iter_entries(external_refs))
703
 
            if len(found_items) != len(external_refs):
704
 
                found_keys = set(k for idx, k, refs, value in found_items)
705
 
                missing_items = external_refs - found_keys
706
 
                missing_file_id, missing_revision_id = missing_items.pop()
707
 
                raise errors.RevisionNotPresent(missing_revision_id,
708
 
                                                missing_file_id)
 
748
        raise NotImplementedError(self._copy_text_texts)
709
749
 
710
750
    def _create_pack_from_packs(self):
711
 
        self.pb.update("Opening pack", 0, 5)
712
 
        self.new_pack = self.open_pack()
713
 
        new_pack = self.new_pack
714
 
        # buffer data - we won't be reading-back during the pack creation and
715
 
        # this makes a significant difference on sftp pushes.
716
 
        new_pack.set_write_cache_size(1024*1024)
717
 
        if 'pack' in debug.debug_flags:
718
 
            plain_pack_list = ['%s%s' % (a_pack.pack_transport.base, a_pack.name)
719
 
                for a_pack in self.packs]
720
 
            if self.revision_ids is not None:
721
 
                rev_count = len(self.revision_ids)
722
 
            else:
723
 
                rev_count = 'all'
724
 
            mutter('%s: create_pack: creating pack from source packs: '
725
 
                '%s%s %s revisions wanted %s t=0',
726
 
                time.ctime(), self._pack_collection._upload_transport.base, new_pack.random_name,
727
 
                plain_pack_list, rev_count)
728
 
        self._copy_revision_texts()
729
 
        self._copy_inventory_texts()
730
 
        self._copy_text_texts()
731
 
        # select signature keys
732
 
        signature_filter = self._revision_keys # same keyspace
733
 
        signature_index_map = self._pack_collection._packs_list_to_pack_map_and_index_list(
734
 
            self.packs, 'signature_index')[0]
735
 
        signature_nodes = self._pack_collection._index_contents(signature_index_map,
736
 
            signature_filter)
737
 
        # copy signature keys and adjust values
738
 
        self.pb.update("Copying signature texts", 4)
739
 
        self._copy_nodes(signature_nodes, signature_index_map, new_pack._writer,
740
 
            new_pack.signature_index)
741
 
        if 'pack' in debug.debug_flags:
742
 
            mutter('%s: create_pack: revision signatures copied: %s%s %d items t+%6.3fs',
743
 
                time.ctime(), self._pack_collection._upload_transport.base, new_pack.random_name,
744
 
                new_pack.signature_index.key_count(),
745
 
                time.time() - new_pack.start_time)
746
 
        self._check_references()
747
 
        if not self._use_pack(new_pack):
748
 
            new_pack.abort()
749
 
            return None
750
 
        self.pb.update("Finishing pack", 5)
751
 
        new_pack.finish()
752
 
        self._pack_collection.allocate(new_pack)
753
 
        return new_pack
754
 
 
755
 
    def _copy_nodes(self, nodes, index_map, writer, write_index):
756
 
        """Copy knit nodes between packs with no graph references."""
757
 
        pb = ui.ui_factory.nested_progress_bar()
758
 
        try:
759
 
            return self._do_copy_nodes(nodes, index_map, writer,
760
 
                write_index, pb)
761
 
        finally:
762
 
            pb.finished()
763
 
 
764
 
    def _do_copy_nodes(self, nodes, index_map, writer, write_index, pb):
765
 
        # for record verification
766
 
        knit_data = _KnitData(None)
767
 
        # plan a readv on each source pack:
768
 
        # group by pack
769
 
        nodes = sorted(nodes)
770
 
        # how to map this into knit.py - or knit.py into this?
771
 
        # we don't want the typical knit logic, we want grouping by pack
772
 
        # at this point - perhaps a helper library for the following code 
773
 
        # duplication points?
774
 
        request_groups = {}
775
 
        for index, key, value in nodes:
776
 
            if index not in request_groups:
777
 
                request_groups[index] = []
778
 
            request_groups[index].append((key, value))
779
 
        record_index = 0
780
 
        pb.update("Copied record", record_index, len(nodes))
781
 
        for index, items in request_groups.iteritems():
782
 
            pack_readv_requests = []
783
 
            for key, value in items:
784
 
                # ---- KnitGraphIndex.get_position
785
 
                bits = value[1:].split(' ')
786
 
                offset, length = int(bits[0]), int(bits[1])
787
 
                pack_readv_requests.append((offset, length, (key, value[0])))
788
 
            # linear scan up the pack
789
 
            pack_readv_requests.sort()
790
 
            # copy the data
791
 
            transport, path = index_map[index]
792
 
            reader = pack.make_readv_reader(transport, path,
793
 
                [offset[0:2] for offset in pack_readv_requests])
794
 
            for (names, read_func), (_1, _2, (key, eol_flag)) in \
795
 
                izip(reader.iter_records(), pack_readv_requests):
796
 
                raw_data = read_func(None)
797
 
                # check the header only
798
 
                df, _ = knit_data._parse_record_header(key[-1], raw_data)
799
 
                df.close()
800
 
                pos, size = writer.add_bytes_record(raw_data, names)
801
 
                write_index.add_node(key, eol_flag + "%d %d" % (pos, size))
802
 
                pb.update("Copied record", record_index)
803
 
                record_index += 1
804
 
 
805
 
    def _copy_nodes_graph(self, index_map, writer, write_index,
806
 
        readv_group_iter, total_items, output_lines=False):
807
 
        """Copy knit nodes between packs.
808
 
 
809
 
        :param output_lines: Return lines present in the copied data as
810
 
            an iterator of line,version_id.
811
 
        """
812
 
        pb = ui.ui_factory.nested_progress_bar()
813
 
        try:
814
 
            for result in self._do_copy_nodes_graph(index_map, writer,
815
 
                write_index, output_lines, pb, readv_group_iter, total_items):
816
 
                yield result
817
 
        except Exception:
818
 
            # Python 2.4 does not permit try:finally: in a generator.
819
 
            pb.finished()
820
 
            raise
821
 
        else:
822
 
            pb.finished()
823
 
 
824
 
    def _do_copy_nodes_graph(self, index_map, writer, write_index,
825
 
        output_lines, pb, readv_group_iter, total_items):
826
 
        # for record verification
827
 
        knit_data = _KnitData(None)
828
 
        # for line extraction when requested (inventories only)
829
 
        if output_lines:
830
 
            factory = knit.KnitPlainFactory()
831
 
        record_index = 0
832
 
        pb.update("Copied record", record_index, total_items)
833
 
        for index, readv_vector, node_vector in readv_group_iter:
834
 
            # copy the data
835
 
            transport, path = index_map[index]
836
 
            reader = pack.make_readv_reader(transport, path, readv_vector)
837
 
            for (names, read_func), (key, eol_flag, references) in \
838
 
                izip(reader.iter_records(), node_vector):
839
 
                raw_data = read_func(None)
840
 
                version_id = key[-1]
841
 
                if output_lines:
842
 
                    # read the entire thing
843
 
                    content, _ = knit_data._parse_record(version_id, raw_data)
844
 
                    if len(references[-1]) == 0:
845
 
                        line_iterator = factory.get_fulltext_content(content)
846
 
                    else:
847
 
                        line_iterator = factory.get_linedelta_content(content)
848
 
                    for line in line_iterator:
849
 
                        yield line, version_id
850
 
                else:
851
 
                    # check the header only
852
 
                    df, _ = knit_data._parse_record_header(version_id, raw_data)
853
 
                    df.close()
854
 
                pos, size = writer.add_bytes_record(raw_data, names)
855
 
                write_index.add_node(key, eol_flag + "%d %d" % (pos, size), references)
856
 
                pb.update("Copied record", record_index)
857
 
                record_index += 1
858
 
 
859
 
    def _get_text_nodes(self):
860
 
        text_index_map = self._pack_collection._packs_list_to_pack_map_and_index_list(
861
 
            self.packs, 'text_index')[0]
862
 
        return text_index_map, self._pack_collection._index_contents(text_index_map,
863
 
            self._text_filter)
864
 
 
865
 
    def _least_readv_node_readv(self, nodes):
866
 
        """Generate request groups for nodes using the least readv's.
867
 
        
868
 
        :param nodes: An iterable of graph index nodes.
869
 
        :return: Total node count and an iterator of the data needed to perform
870
 
            readvs to obtain the data for nodes. Each item yielded by the
871
 
            iterator is a tuple with:
872
 
            index, readv_vector, node_vector. readv_vector is a list ready to
873
 
            hand to the transport readv method, and node_vector is a list of
874
 
            (key, eol_flag, references) for the the node retrieved by the
875
 
            matching readv_vector.
876
 
        """
877
 
        # group by pack so we do one readv per pack
878
 
        nodes = sorted(nodes)
879
 
        total = len(nodes)
880
 
        request_groups = {}
881
 
        for index, key, value, references in nodes:
882
 
            if index not in request_groups:
883
 
                request_groups[index] = []
884
 
            request_groups[index].append((key, value, references))
885
 
        result = []
886
 
        for index, items in request_groups.iteritems():
887
 
            pack_readv_requests = []
888
 
            for key, value, references in items:
889
 
                # ---- KnitGraphIndex.get_position
890
 
                bits = value[1:].split(' ')
891
 
                offset, length = int(bits[0]), int(bits[1])
892
 
                pack_readv_requests.append(
893
 
                    ((offset, length), (key, value[0], references)))
894
 
            # linear scan up the pack to maximum range combining.
895
 
            pack_readv_requests.sort()
896
 
            # split out the readv and the node data.
897
 
            pack_readv = [readv for readv, node in pack_readv_requests]
898
 
            node_vector = [node for readv, node in pack_readv_requests]
899
 
            result.append((index, pack_readv, node_vector))
900
 
        return total, result
 
751
        raise NotImplementedError(self._create_pack_from_packs)
901
752
 
902
753
    def _log_copied_texts(self):
903
754
        if 'pack' in debug.debug_flags:
907
758
                self.new_pack.text_index.key_count(),
908
759
                time.time() - self.new_pack.start_time)
909
760
 
910
 
    def _process_inventory_lines(self, inv_lines):
911
 
        """Use up the inv_lines generator and setup a text key filter."""
912
 
        repo = self._pack_collection.repo
913
 
        fileid_revisions = repo._find_file_ids_from_xml_inventory_lines(
914
 
            inv_lines, self.revision_ids)
915
 
        text_filter = []
916
 
        for fileid, file_revids in fileid_revisions.iteritems():
917
 
            text_filter.extend([(fileid, file_revid) for file_revid in file_revids])
918
 
        self._text_filter = text_filter
919
 
 
920
 
    def _revision_node_readv(self, revision_nodes):
921
 
        """Return the total revisions and the readv's to issue.
922
 
 
923
 
        :param revision_nodes: The revision index contents for the packs being
924
 
            incorporated into the new pack.
925
 
        :return: As per _least_readv_node_readv.
926
 
        """
927
 
        return self._least_readv_node_readv(revision_nodes)
928
 
 
929
761
    def _use_pack(self, new_pack):
930
762
        """Return True if new_pack should be used.
931
763
 
935
767
        return new_pack.data_inserted()
936
768
 
937
769
 
938
 
class OptimisingPacker(Packer):
939
 
    """A packer which spends more time to create better disk layouts."""
940
 
 
941
 
    def _revision_node_readv(self, revision_nodes):
942
 
        """Return the total revisions and the readv's to issue.
943
 
 
944
 
        This sort places revisions in topological order with the ancestors
945
 
        after the children.
946
 
 
947
 
        :param revision_nodes: The revision index contents for the packs being
948
 
            incorporated into the new pack.
949
 
        :return: As per _least_readv_node_readv.
950
 
        """
951
 
        # build an ancestors dict
952
 
        ancestors = {}
953
 
        by_key = {}
954
 
        for index, key, value, references in revision_nodes:
955
 
            ancestors[key] = references[0]
956
 
            by_key[key] = (index, value, references)
957
 
        order = tsort.topo_sort(ancestors)
958
 
        total = len(order)
959
 
        # Single IO is pathological, but it will work as a starting point.
960
 
        requests = []
961
 
        for key in reversed(order):
962
 
            index, value, references = by_key[key]
963
 
            # ---- KnitGraphIndex.get_position
964
 
            bits = value[1:].split(' ')
965
 
            offset, length = int(bits[0]), int(bits[1])
966
 
            requests.append(
967
 
                (index, [(offset, length)], [(key, value[0], references)]))
968
 
        # TODO: combine requests in the same index that are in ascending order.
969
 
        return total, requests
970
 
 
971
 
 
972
 
class ReconcilePacker(Packer):
973
 
    """A packer which regenerates indices etc as it copies.
974
 
    
975
 
    This is used by ``bzr reconcile`` to cause parent text pointers to be
976
 
    regenerated.
977
 
    """
978
 
 
979
 
    def _extra_init(self):
980
 
        self._data_changed = False
981
 
 
982
 
    def _process_inventory_lines(self, inv_lines):
983
 
        """Generate a text key reference map rather for reconciling with."""
984
 
        repo = self._pack_collection.repo
985
 
        refs = repo._find_text_key_references_from_xml_inventory_lines(
986
 
            inv_lines)
987
 
        self._text_refs = refs
988
 
        # during reconcile we:
989
 
        #  - convert unreferenced texts to full texts
990
 
        #  - correct texts which reference a text not copied to be full texts
991
 
        #  - copy all others as-is but with corrected parents.
992
 
        #  - so at this point we don't know enough to decide what becomes a full
993
 
        #    text.
994
 
        self._text_filter = None
995
 
 
996
 
    def _copy_text_texts(self):
997
 
        """generate what texts we should have and then copy."""
998
 
        self.pb.update("Copying content texts", 3)
999
 
        # we have three major tasks here:
1000
 
        # 1) generate the ideal index
1001
 
        repo = self._pack_collection.repo
1002
 
        ancestors = dict([(key[0], tuple(ref[0] for ref in refs[0])) for
1003
 
            _1, key, _2, refs in 
1004
 
            self.new_pack.revision_index.iter_all_entries()])
1005
 
        ideal_index = repo._generate_text_key_index(self._text_refs, ancestors)
1006
 
        # 2) generate a text_nodes list that contains all the deltas that can
1007
 
        #    be used as-is, with corrected parents.
1008
 
        ok_nodes = []
1009
 
        bad_texts = []
1010
 
        discarded_nodes = []
1011
 
        NULL_REVISION = _mod_revision.NULL_REVISION
1012
 
        text_index_map, text_nodes = self._get_text_nodes()
1013
 
        for node in text_nodes:
1014
 
            # 0 - index
1015
 
            # 1 - key 
1016
 
            # 2 - value
1017
 
            # 3 - refs
1018
 
            try:
1019
 
                ideal_parents = tuple(ideal_index[node[1]])
1020
 
            except KeyError:
1021
 
                discarded_nodes.append(node)
1022
 
                self._data_changed = True
1023
 
            else:
1024
 
                if ideal_parents == (NULL_REVISION,):
1025
 
                    ideal_parents = ()
1026
 
                if ideal_parents == node[3][0]:
1027
 
                    # no change needed.
1028
 
                    ok_nodes.append(node)
1029
 
                elif ideal_parents[0:1] == node[3][0][0:1]:
1030
 
                    # the left most parent is the same, or there are no parents
1031
 
                    # today. Either way, we can preserve the representation as
1032
 
                    # long as we change the refs to be inserted.
1033
 
                    self._data_changed = True
1034
 
                    ok_nodes.append((node[0], node[1], node[2],
1035
 
                        (ideal_parents, node[3][1])))
1036
 
                    self._data_changed = True
1037
 
                else:
1038
 
                    # Reinsert this text completely
1039
 
                    bad_texts.append((node[1], ideal_parents))
1040
 
                    self._data_changed = True
1041
 
        # we're finished with some data.
1042
 
        del ideal_index
1043
 
        del text_nodes
1044
 
        # 3) bulk copy the ok data
1045
 
        total_items, readv_group_iter = self._least_readv_node_readv(ok_nodes)
1046
 
        list(self._copy_nodes_graph(text_index_map, self.new_pack._writer,
1047
 
            self.new_pack.text_index, readv_group_iter, total_items))
1048
 
        # 4) adhoc copy all the other texts.
1049
 
        # We have to topologically insert all texts otherwise we can fail to
1050
 
        # reconcile when parts of a single delta chain are preserved intact,
1051
 
        # and other parts are not. E.g. Discarded->d1->d2->d3. d1 will be
1052
 
        # reinserted, and if d3 has incorrect parents it will also be
1053
 
        # reinserted. If we insert d3 first, d2 is present (as it was bulk
1054
 
        # copied), so we will try to delta, but d2 is not currently able to be
1055
 
        # extracted because it's basis d1 is not present. Topologically sorting
1056
 
        # addresses this. The following generates a sort for all the texts that
1057
 
        # are being inserted without having to reference the entire text key
1058
 
        # space (we only topo sort the revisions, which is smaller).
1059
 
        topo_order = tsort.topo_sort(ancestors)
1060
 
        rev_order = dict(zip(topo_order, range(len(topo_order))))
1061
 
        bad_texts.sort(key=lambda key:rev_order[key[0][1]])
1062
 
        transaction = repo.get_transaction()
1063
 
        file_id_index = GraphIndexPrefixAdapter(
1064
 
            self.new_pack.text_index,
1065
 
            ('blank', ), 1,
1066
 
            add_nodes_callback=self.new_pack.text_index.add_nodes)
1067
 
        knit_index = KnitGraphIndex(file_id_index,
1068
 
            add_callback=file_id_index.add_nodes,
1069
 
            deltas=True, parents=True)
1070
 
        output_knit = knit.KnitVersionedFile('reconcile-texts',
1071
 
            self._pack_collection.transport,
1072
 
            index=knit_index,
1073
 
            access_method=_PackAccess(
1074
 
                {self.new_pack.text_index:self.new_pack.access_tuple()},
1075
 
                (self.new_pack._writer, self.new_pack.text_index)),
1076
 
            factory=knit.KnitPlainFactory())
1077
 
        for key, parent_keys in bad_texts:
1078
 
            # We refer to the new pack to delta data being output.
1079
 
            # A possible improvement would be to catch errors on short reads
1080
 
            # and only flush then.
1081
 
            self.new_pack.flush()
1082
 
            parents = []
1083
 
            for parent_key in parent_keys:
1084
 
                if parent_key[0] != key[0]:
1085
 
                    # Graph parents must match the fileid
1086
 
                    raise errors.BzrError('Mismatched key parent %r:%r' %
1087
 
                        (key, parent_keys))
1088
 
                parents.append(parent_key[1])
1089
 
            source_weave = repo.weave_store.get_weave(key[0], transaction)
1090
 
            text_lines = source_weave.get_lines(key[1])
1091
 
            # adapt the 'knit' to the current file_id.
1092
 
            file_id_index = GraphIndexPrefixAdapter(
1093
 
                self.new_pack.text_index,
1094
 
                (key[0], ), 1,
1095
 
                add_nodes_callback=self.new_pack.text_index.add_nodes)
1096
 
            knit_index._graph_index = file_id_index
1097
 
            knit_index._add_callback = file_id_index.add_nodes
1098
 
            output_knit.add_lines_with_ghosts(
1099
 
                key[1], parents, text_lines, random_id=True, check_content=False)
1100
 
        # 5) check that nothing inserted has a reference outside the keyspace.
1101
 
        missing_text_keys = self.new_pack._external_compression_parents_of_texts()
1102
 
        if missing_text_keys:
1103
 
            raise errors.BzrError('Reference to missing compression parents %r'
1104
 
                % (missing_text_keys,))
1105
 
        self._log_copied_texts()
1106
 
 
1107
 
    def _use_pack(self, new_pack):
1108
 
        """Override _use_pack to check for reconcile having changed content."""
1109
 
        # XXX: we might be better checking this at the copy time.
1110
 
        original_inventory_keys = set()
1111
 
        inv_index = self._pack_collection.inventory_index.combined_index
1112
 
        for entry in inv_index.iter_all_entries():
1113
 
            original_inventory_keys.add(entry[1])
1114
 
        new_inventory_keys = set()
1115
 
        for entry in new_pack.inventory_index.iter_all_entries():
1116
 
            new_inventory_keys.add(entry[1])
1117
 
        if new_inventory_keys != original_inventory_keys:
1118
 
            self._data_changed = True
1119
 
        return new_pack.data_inserted() and self._data_changed
1120
 
 
1121
 
 
1122
770
class RepositoryPackCollection(object):
1123
 
    """Management of packs within a repository."""
 
771
    """Management of packs within a repository.
 
772
 
 
773
    :ivar _names: map of {pack_name: (index_size,)}
 
774
    """
 
775
 
 
776
    pack_factory = None
 
777
    resumed_pack_factory = None
 
778
    normal_packer_class = None
 
779
    optimising_packer_class = None
1124
780
 
1125
781
    def __init__(self, repo, transport, index_transport, upload_transport,
1126
 
                 pack_transport):
 
782
                 pack_transport, index_builder_class, index_class,
 
783
                 use_chk_index):
1127
784
        """Create a new RepositoryPackCollection.
1128
785
 
1129
 
        :param transport: Addresses the repository base directory 
 
786
        :param transport: Addresses the repository base directory
1130
787
            (typically .bzr/repository/).
1131
788
        :param index_transport: Addresses the directory containing indices.
1132
789
        :param upload_transport: Addresses the directory into which packs are written
1133
790
            while they're being created.
1134
791
        :param pack_transport: Addresses the directory of existing complete packs.
 
792
        :param index_builder_class: The index builder class to use.
 
793
        :param index_class: The index class to use.
 
794
        :param use_chk_index: Whether to setup and manage a CHK index.
1135
795
        """
 
796
        # XXX: This should call self.reset()
1136
797
        self.repo = repo
1137
798
        self.transport = transport
1138
799
        self._index_transport = index_transport
1139
800
        self._upload_transport = upload_transport
1140
801
        self._pack_transport = pack_transport
1141
 
        self._suffix_offsets = {'.rix': 0, '.iix': 1, '.tix': 2, '.six': 3}
 
802
        self._index_builder_class = index_builder_class
 
803
        self._index_class = index_class
 
804
        self._suffix_offsets = {'.rix': 0, '.iix': 1, '.tix': 2, '.six': 3,
 
805
            '.cix': 4}
1142
806
        self.packs = []
1143
807
        # name:Pack mapping
 
808
        self._names = None
1144
809
        self._packs_by_name = {}
1145
810
        # the previous pack-names content
1146
811
        self._packs_at_load = None
1147
812
        # when a pack is being created by this object, the state of that pack.
1148
813
        self._new_pack = None
1149
814
        # aggregated revision index data
1150
 
        self.revision_index = AggregateIndex()
1151
 
        self.inventory_index = AggregateIndex()
1152
 
        self.text_index = AggregateIndex()
1153
 
        self.signature_index = AggregateIndex()
 
815
        flush = self._flush_new_pack
 
816
        self.revision_index = AggregateIndex(self.reload_pack_names, flush)
 
817
        self.inventory_index = AggregateIndex(self.reload_pack_names, flush)
 
818
        self.text_index = AggregateIndex(self.reload_pack_names, flush)
 
819
        self.signature_index = AggregateIndex(self.reload_pack_names, flush)
 
820
        all_indices = [self.revision_index, self.inventory_index,
 
821
                self.text_index, self.signature_index]
 
822
        if use_chk_index:
 
823
            self.chk_index = AggregateIndex(self.reload_pack_names, flush)
 
824
            all_indices.append(self.chk_index)
 
825
        else:
 
826
            # used to determine if we're using a chk_index elsewhere.
 
827
            self.chk_index = None
 
828
        # Tell all the CombinedGraphIndex objects about each other, so they can
 
829
        # share hints about which pack names to search first.
 
830
        all_combined = [agg_idx.combined_index for agg_idx in all_indices]
 
831
        for combined_idx in all_combined:
 
832
            combined_idx.set_sibling_indices(
 
833
                set(all_combined).difference([combined_idx]))
 
834
        # resumed packs
 
835
        self._resumed_packs = []
 
836
        self.config_stack = config.LocationStack(self.transport.base)
 
837
 
 
838
    def __repr__(self):
 
839
        return '%s(%r)' % (self.__class__.__name__, self.repo)
1154
840
 
1155
841
    def add_pack_to_memory(self, pack):
1156
842
        """Make a Pack object available to the repository to satisfy queries.
1157
 
        
 
843
 
1158
844
        :param pack: A Pack object.
1159
845
        """
1160
846
        if pack.name in self._packs_by_name:
1161
 
            raise AssertionError()
 
847
            raise AssertionError(
 
848
                'pack %s already in _packs_by_name' % (pack.name,))
1162
849
        self.packs.append(pack)
1163
850
        self._packs_by_name[pack.name] = pack
1164
851
        self.revision_index.add_index(pack.revision_index, pack)
1165
852
        self.inventory_index.add_index(pack.inventory_index, pack)
1166
853
        self.text_index.add_index(pack.text_index, pack)
1167
854
        self.signature_index.add_index(pack.signature_index, pack)
1168
 
        
1169
 
    def _add_text_to_weave(self, file_id, revision_id, new_lines, parents,
1170
 
        nostore_sha, random_revid):
1171
 
        file_id_index = GraphIndexPrefixAdapter(
1172
 
            self.text_index.combined_index,
1173
 
            (file_id, ), 1,
1174
 
            add_nodes_callback=self.text_index.add_callback)
1175
 
        self.repo._text_knit._index._graph_index = file_id_index
1176
 
        self.repo._text_knit._index._add_callback = file_id_index.add_nodes
1177
 
        return self.repo._text_knit.add_lines_with_ghosts(
1178
 
            revision_id, parents, new_lines, nostore_sha=nostore_sha,
1179
 
            random_id=random_revid, check_content=False)[0:2]
 
855
        if self.chk_index is not None:
 
856
            self.chk_index.add_index(pack.chk_index, pack)
1180
857
 
1181
858
    def all_packs(self):
1182
859
        """Return a list of all the Pack objects this repository has.
1192
869
 
1193
870
    def autopack(self):
1194
871
        """Pack the pack collection incrementally.
1195
 
        
 
872
 
1196
873
        This will not attempt global reorganisation or recompression,
1197
874
        rather it will just ensure that the total number of packs does
1198
875
        not grow without bound. It uses the _max_pack_count method to
1204
881
        in synchronisation with certain steps. Otherwise the names collection
1205
882
        is not flushed.
1206
883
 
1207
 
        :return: True if packing took place.
 
884
        :return: Something evaluating true if packing took place.
1208
885
        """
 
886
        while True:
 
887
            try:
 
888
                return self._do_autopack()
 
889
            except errors.RetryAutopack:
 
890
                # If we get a RetryAutopack exception, we should abort the
 
891
                # current action, and retry.
 
892
                pass
 
893
 
 
894
    def _do_autopack(self):
1209
895
        # XXX: Should not be needed when the management of indices is sane.
1210
896
        total_revisions = self.revision_index.combined_index.key_count()
1211
897
        total_packs = len(self._names)
1212
898
        if self._max_pack_count(total_revisions) >= total_packs:
1213
 
            return False
1214
 
        # XXX: the following may want to be a class, to pack with a given
1215
 
        # policy.
1216
 
        mutter('Auto-packing repository %s, which has %d pack files, '
1217
 
            'containing %d revisions into %d packs.', self, total_packs,
1218
 
            total_revisions, self._max_pack_count(total_revisions))
 
899
            return None
1219
900
        # determine which packs need changing
1220
901
        pack_distribution = self.pack_distribution(total_revisions)
1221
902
        existing_packs = []
1229
910
                # group their data with the relevant commit, and that may
1230
911
                # involve rewriting ancient history - which autopack tries to
1231
912
                # avoid. Alternatively we could not group the data but treat
1232
 
                # each of these as having a single revision, and thus add 
 
913
                # each of these as having a single revision, and thus add
1233
914
                # one revision for each to the total revision count, to get
1234
915
                # a matching distribution.
1235
916
                continue
1236
917
            existing_packs.append((revision_count, pack))
1237
918
        pack_operations = self.plan_autopack_combinations(
1238
919
            existing_packs, pack_distribution)
1239
 
        self._execute_pack_operations(pack_operations)
1240
 
        return True
 
920
        num_new_packs = len(pack_operations)
 
921
        num_old_packs = sum([len(po[1]) for po in pack_operations])
 
922
        num_revs_affected = sum([po[0] for po in pack_operations])
 
923
        mutter('Auto-packing repository %s, which has %d pack files, '
 
924
            'containing %d revisions. Packing %d files into %d affecting %d'
 
925
            ' revisions', self, total_packs, total_revisions, num_old_packs,
 
926
            num_new_packs, num_revs_affected)
 
927
        result = self._execute_pack_operations(pack_operations, packer_class=self.normal_packer_class,
 
928
                                      reload_func=self._restart_autopack)
 
929
        mutter('Auto-packing repository %s completed', self)
 
930
        return result
1241
931
 
1242
 
    def _execute_pack_operations(self, pack_operations, _packer_class=Packer):
 
932
    def _execute_pack_operations(self, pack_operations, packer_class,
 
933
            reload_func=None):
1243
934
        """Execute a series of pack operations.
1244
935
 
1245
936
        :param pack_operations: A list of [revision_count, packs_to_combine].
1246
 
        :param _packer_class: The class of packer to use (default: Packer).
1247
 
        :return: None.
 
937
        :param packer_class: The class of packer to use
 
938
        :return: The new pack names.
1248
939
        """
1249
940
        for revision_count, packs in pack_operations:
1250
941
            # we may have no-ops from the setup logic
1251
942
            if len(packs) == 0:
1252
943
                continue
1253
 
            _packer_class(self, packs, '.autopack').pack()
 
944
            packer = packer_class(self, packs, '.autopack',
 
945
                                   reload_func=reload_func)
 
946
            try:
 
947
                result = packer.pack()
 
948
            except errors.RetryWithNewPacks:
 
949
                # An exception is propagating out of this context, make sure
 
950
                # this packer has cleaned up. Packer() doesn't set its new_pack
 
951
                # state into the RepositoryPackCollection object, so we only
 
952
                # have access to it directly here.
 
953
                if packer.new_pack is not None:
 
954
                    packer.new_pack.abort()
 
955
                raise
 
956
            if result is None:
 
957
                return
1254
958
            for pack in packs:
1255
959
                self._remove_pack_from_memory(pack)
1256
960
        # record the newly available packs and stop advertising the old
1257
961
        # packs
1258
 
        self._save_pack_names(clear_obsolete_packs=True)
1259
 
        # Move the old packs out of the way now they are no longer referenced.
1260
 
        for revision_count, packs in pack_operations:
1261
 
            self._obsolete_packs(packs)
 
962
        to_be_obsoleted = []
 
963
        for _, packs in pack_operations:
 
964
            to_be_obsoleted.extend(packs)
 
965
        result = self._save_pack_names(clear_obsolete_packs=True,
 
966
                                       obsolete_packs=to_be_obsoleted)
 
967
        return result
 
968
 
 
969
    def _flush_new_pack(self):
 
970
        if self._new_pack is not None:
 
971
            self._new_pack.flush()
1262
972
 
1263
973
    def lock_names(self):
1264
974
        """Acquire the mutex around the pack-names index.
1265
 
        
 
975
 
1266
976
        This cannot be used in the middle of a read-only transaction on the
1267
977
        repository.
1268
978
        """
1269
979
        self.repo.control_files.lock_write()
1270
980
 
1271
 
    def pack(self):
 
981
    def _already_packed(self):
 
982
        """Is the collection already packed?"""
 
983
        return not (self.repo._format.pack_compresses or (len(self._names) > 1))
 
984
 
 
985
    def pack(self, hint=None, clean_obsolete_packs=False):
1272
986
        """Pack the pack collection totally."""
1273
987
        self.ensure_loaded()
1274
988
        total_packs = len(self._names)
1275
 
        if total_packs < 2:
1276
 
            # This is arguably wrong because we might not be optimal, but for
1277
 
            # now lets leave it in. (e.g. reconcile -> one pack. But not
1278
 
            # optimal.
 
989
        if self._already_packed():
1279
990
            return
1280
991
        total_revisions = self.revision_index.combined_index.key_count()
1281
992
        # XXX: the following may want to be a class, to pack with a given
1282
993
        # policy.
1283
994
        mutter('Packing repository %s, which has %d pack files, '
1284
 
            'containing %d revisions into 1 packs.', self, total_packs,
1285
 
            total_revisions)
 
995
            'containing %d revisions with hint %r.', self, total_packs,
 
996
            total_revisions, hint)
 
997
        while True:
 
998
            try:
 
999
                self._try_pack_operations(hint)
 
1000
            except RetryPackOperations:
 
1001
                continue
 
1002
            break
 
1003
 
 
1004
        if clean_obsolete_packs:
 
1005
            self._clear_obsolete_packs()
 
1006
 
 
1007
    def _try_pack_operations(self, hint):
 
1008
        """Calculate the pack operations based on the hint (if any), and
 
1009
        execute them.
 
1010
        """
1286
1011
        # determine which packs need changing
1287
 
        pack_distribution = [1]
1288
1012
        pack_operations = [[0, []]]
1289
1013
        for pack in self.all_packs():
1290
 
            pack_operations[-1][0] += pack.get_revision_count()
1291
 
            pack_operations[-1][1].append(pack)
1292
 
        self._execute_pack_operations(pack_operations, OptimisingPacker)
 
1014
            if hint is None or pack.name in hint:
 
1015
                # Either no hint was provided (so we are packing everything),
 
1016
                # or this pack was included in the hint.
 
1017
                pack_operations[-1][0] += pack.get_revision_count()
 
1018
                pack_operations[-1][1].append(pack)
 
1019
        self._execute_pack_operations(pack_operations,
 
1020
            packer_class=self.optimising_packer_class,
 
1021
            reload_func=self._restart_pack_operations)
1293
1022
 
1294
1023
    def plan_autopack_combinations(self, existing_packs, pack_distribution):
1295
1024
        """Plan a pack operation.
1305
1034
        pack_operations = [[0, []]]
1306
1035
        # plan out what packs to keep, and what to reorganise
1307
1036
        while len(existing_packs):
1308
 
            # take the largest pack, and if its less than the head of the
1309
 
            # distribution chart we will include its contents in the new pack for
1310
 
            # that position. If its larger, we remove its size from the
 
1037
            # take the largest pack, and if it's less than the head of the
 
1038
            # distribution chart we will include its contents in the new pack
 
1039
            # for that position. If it's larger, we remove its size from the
1311
1040
            # distribution chart
1312
1041
            next_pack_rev_count, next_pack = existing_packs.pop(0)
1313
1042
            if next_pack_rev_count >= pack_distribution[0]:
1330
1059
                    # this pack is used up, shift left.
1331
1060
                    del pack_distribution[0]
1332
1061
                    pack_operations.append([0, []])
1333
 
        
1334
 
        return pack_operations
 
1062
        # Now that we know which pack files we want to move, shove them all
 
1063
        # into a single pack file.
 
1064
        final_rev_count = 0
 
1065
        final_pack_list = []
 
1066
        for num_revs, pack_files in pack_operations:
 
1067
            final_rev_count += num_revs
 
1068
            final_pack_list.extend(pack_files)
 
1069
        if len(final_pack_list) == 1:
 
1070
            raise AssertionError('We somehow generated an autopack with a'
 
1071
                ' single pack file being moved.')
 
1072
            return []
 
1073
        return [[final_rev_count, final_pack_list]]
1335
1074
 
1336
1075
    def ensure_loaded(self):
1337
 
        # NB: if you see an assertion error here, its probably access against
 
1076
        """Ensure we have read names from disk.
 
1077
 
 
1078
        :return: True if the disk names had not been previously read.
 
1079
        """
 
1080
        # NB: if you see an assertion error here, it's probably access against
1338
1081
        # an unlocked repo. Naughty.
1339
1082
        if not self.repo.is_locked():
1340
1083
            raise errors.ObjectNotLocked(self.repo)
1345
1088
                name = key[0]
1346
1089
                self._names[name] = self._parse_index_sizes(value)
1347
1090
                self._packs_at_load.add((key, value))
 
1091
            result = True
 
1092
        else:
 
1093
            result = False
1348
1094
        # populate all the metadata.
1349
1095
        self.all_packs()
 
1096
        return result
1350
1097
 
1351
1098
    def _parse_index_sizes(self, value):
1352
1099
        """Parse a string of index sizes."""
1365
1112
            inv_index = self._make_index(name, '.iix')
1366
1113
            txt_index = self._make_index(name, '.tix')
1367
1114
            sig_index = self._make_index(name, '.six')
 
1115
            if self.chk_index is not None:
 
1116
                chk_index = self._make_index(name, '.cix', is_chk=True)
 
1117
            else:
 
1118
                chk_index = None
1368
1119
            result = ExistingPack(self._pack_transport, name, rev_index,
1369
 
                inv_index, txt_index, sig_index)
 
1120
                inv_index, txt_index, sig_index, chk_index)
1370
1121
            self.add_pack_to_memory(result)
1371
1122
            return result
1372
1123
 
 
1124
    def _resume_pack(self, name):
 
1125
        """Get a suspended Pack object by name.
 
1126
 
 
1127
        :param name: The name of the pack - e.g. '123456'
 
1128
        :return: A Pack object.
 
1129
        """
 
1130
        if not re.match('[a-f0-9]{32}', name):
 
1131
            # Tokens should be md5sums of the suspended pack file, i.e. 32 hex
 
1132
            # digits.
 
1133
            raise errors.UnresumableWriteGroup(
 
1134
                self.repo, [name], 'Malformed write group token')
 
1135
        try:
 
1136
            rev_index = self._make_index(name, '.rix', resume=True)
 
1137
            inv_index = self._make_index(name, '.iix', resume=True)
 
1138
            txt_index = self._make_index(name, '.tix', resume=True)
 
1139
            sig_index = self._make_index(name, '.six', resume=True)
 
1140
            if self.chk_index is not None:
 
1141
                chk_index = self._make_index(name, '.cix', resume=True,
 
1142
                                             is_chk=True)
 
1143
            else:
 
1144
                chk_index = None
 
1145
            result = self.resumed_pack_factory(name, rev_index, inv_index,
 
1146
                txt_index, sig_index, self._upload_transport,
 
1147
                self._pack_transport, self._index_transport, self,
 
1148
                chk_index=chk_index)
 
1149
        except errors.NoSuchFile, e:
 
1150
            raise errors.UnresumableWriteGroup(self.repo, [name], str(e))
 
1151
        self.add_pack_to_memory(result)
 
1152
        self._resumed_packs.append(result)
 
1153
        return result
 
1154
 
1373
1155
    def allocate(self, a_new_pack):
1374
1156
        """Allocate name in the list of packs.
1375
1157
 
1385
1167
 
1386
1168
    def _iter_disk_pack_index(self):
1387
1169
        """Iterate over the contents of the pack-names index.
1388
 
        
 
1170
 
1389
1171
        This is used when loading the list from disk, and before writing to
1390
1172
        detect updates from others during our write operation.
1391
1173
        :return: An iterator of the index contents.
1392
1174
        """
1393
 
        return GraphIndex(self.transport, 'pack-names', None
 
1175
        return self._index_class(self.transport, 'pack-names', None
1394
1176
                ).iter_all_entries()
1395
1177
 
1396
 
    def _make_index(self, name, suffix):
 
1178
    def _make_index(self, name, suffix, resume=False, is_chk=False):
1397
1179
        size_offset = self._suffix_offsets[suffix]
1398
1180
        index_name = name + suffix
1399
 
        index_size = self._names[name][size_offset]
1400
 
        return GraphIndex(
1401
 
            self._index_transport, index_name, index_size)
 
1181
        if resume:
 
1182
            transport = self._upload_transport
 
1183
            index_size = transport.stat(index_name).st_size
 
1184
        else:
 
1185
            transport = self._index_transport
 
1186
            index_size = self._names[name][size_offset]
 
1187
        index = self._index_class(transport, index_name, index_size,
 
1188
                                  unlimited_cache=is_chk)
 
1189
        if is_chk and self._index_class is btree_index.BTreeGraphIndex: 
 
1190
            index._leaf_factory = btree_index._gcchk_factory
 
1191
        return index
1402
1192
 
1403
1193
    def _max_pack_count(self, total_revisions):
1404
1194
        """Return the maximum number of packs to use for total revisions.
1405
 
        
 
1195
 
1406
1196
        :param total_revisions: The total number of revisions in the
1407
1197
            repository.
1408
1198
        """
1432
1222
        :param return: None.
1433
1223
        """
1434
1224
        for pack in packs:
1435
 
            pack.pack_transport.rename(pack.file_name(),
1436
 
                '../obsolete_packs/' + pack.file_name())
 
1225
            try:
 
1226
                try:
 
1227
                    pack.pack_transport.move(pack.file_name(),
 
1228
                        '../obsolete_packs/' + pack.file_name())
 
1229
                except errors.NoSuchFile:
 
1230
                    # perhaps obsolete_packs was removed? Let's create it and
 
1231
                    # try again
 
1232
                    try:
 
1233
                        pack.pack_transport.mkdir('../obsolete_packs/')
 
1234
                    except errors.FileExists:
 
1235
                        pass
 
1236
                    pack.pack_transport.move(pack.file_name(),
 
1237
                        '../obsolete_packs/' + pack.file_name())
 
1238
            except (errors.PathError, errors.TransportError), e:
 
1239
                # TODO: Should these be warnings or mutters?
 
1240
                mutter("couldn't rename obsolete pack, skipping it:\n%s"
 
1241
                       % (e,))
1437
1242
            # TODO: Probably needs to know all possible indices for this pack
1438
1243
            # - or maybe list the directory and move all indices matching this
1439
1244
            # name whether we recognize it or not?
1440
 
            for suffix in ('.iix', '.six', '.tix', '.rix'):
1441
 
                self._index_transport.rename(pack.name + suffix,
1442
 
                    '../obsolete_packs/' + pack.name + suffix)
 
1245
            suffixes = ['.iix', '.six', '.tix', '.rix']
 
1246
            if self.chk_index is not None:
 
1247
                suffixes.append('.cix')
 
1248
            for suffix in suffixes:
 
1249
                try:
 
1250
                    self._index_transport.move(pack.name + suffix,
 
1251
                        '../obsolete_packs/' + pack.name + suffix)
 
1252
                except (errors.PathError, errors.TransportError), e:
 
1253
                    mutter("couldn't rename obsolete index, skipping it:\n%s"
 
1254
                           % (e,))
1443
1255
 
1444
1256
    def pack_distribution(self, total_revisions):
1445
1257
        """Generate a list of the number of revisions to put in each pack.
1463
1275
 
1464
1276
    def _remove_pack_from_memory(self, pack):
1465
1277
        """Remove pack from the packs accessed by this repository.
1466
 
        
 
1278
 
1467
1279
        Only affects memory state, until self._save_pack_names() is invoked.
1468
1280
        """
1469
1281
        self._names.pop(pack.name)
1470
1282
        self._packs_by_name.pop(pack.name)
1471
1283
        self._remove_pack_indices(pack)
 
1284
        self.packs.remove(pack)
1472
1285
 
1473
 
    def _remove_pack_indices(self, pack):
1474
 
        """Remove the indices for pack from the aggregated indices."""
1475
 
        self.revision_index.remove_index(pack.revision_index, pack)
1476
 
        self.inventory_index.remove_index(pack.inventory_index, pack)
1477
 
        self.text_index.remove_index(pack.text_index, pack)
1478
 
        self.signature_index.remove_index(pack.signature_index, pack)
 
1286
    def _remove_pack_indices(self, pack, ignore_missing=False):
 
1287
        """Remove the indices for pack from the aggregated indices.
 
1288
        
 
1289
        :param ignore_missing: Suppress KeyErrors from calling remove_index.
 
1290
        """
 
1291
        for index_type in Pack.index_definitions.keys():
 
1292
            attr_name = index_type + '_index'
 
1293
            aggregate_index = getattr(self, attr_name)
 
1294
            if aggregate_index is not None:
 
1295
                pack_index = getattr(pack, attr_name)
 
1296
                try:
 
1297
                    aggregate_index.remove_index(pack_index)
 
1298
                except KeyError:
 
1299
                    if ignore_missing:
 
1300
                        continue
 
1301
                    raise
1479
1302
 
1480
1303
    def reset(self):
1481
1304
        """Clear all cached data."""
1482
1305
        # cached revision data
1483
 
        self.repo._revision_knit = None
1484
1306
        self.revision_index.clear()
1485
1307
        # cached signature data
1486
 
        self.repo._signature_knit = None
1487
1308
        self.signature_index.clear()
1488
1309
        # cached file text data
1489
1310
        self.text_index.clear()
1490
 
        self.repo._text_knit = None
1491
1311
        # cached inventory data
1492
1312
        self.inventory_index.clear()
 
1313
        # cached chk data
 
1314
        if self.chk_index is not None:
 
1315
            self.chk_index.clear()
1493
1316
        # remove the open pack
1494
1317
        self._new_pack = None
1495
1318
        # information about packs.
1498
1321
        self._packs_by_name = {}
1499
1322
        self._packs_at_load = None
1500
1323
 
1501
 
    def _make_index_map(self, index_suffix):
1502
 
        """Return information on existing indices.
1503
 
 
1504
 
        :param suffix: Index suffix added to pack name.
1505
 
 
1506
 
        :returns: (pack_map, indices) where indices is a list of GraphIndex 
1507
 
        objects, and pack_map is a mapping from those objects to the 
1508
 
        pack tuple they describe.
1509
 
        """
1510
 
        # TODO: stop using this; it creates new indices unnecessarily.
1511
 
        self.ensure_loaded()
1512
 
        suffix_map = {'.rix': 'revision_index',
1513
 
            '.six': 'signature_index',
1514
 
            '.iix': 'inventory_index',
1515
 
            '.tix': 'text_index',
1516
 
        }
1517
 
        return self._packs_list_to_pack_map_and_index_list(self.all_packs(),
1518
 
            suffix_map[index_suffix])
1519
 
 
1520
 
    def _packs_list_to_pack_map_and_index_list(self, packs, index_attribute):
1521
 
        """Convert a list of packs to an index pack map and index list.
1522
 
 
1523
 
        :param packs: The packs list to process.
1524
 
        :param index_attribute: The attribute that the desired index is found
1525
 
            on.
1526
 
        :return: A tuple (map, list) where map contains the dict from
1527
 
            index:pack_tuple, and lsit contains the indices in the same order
1528
 
            as the packs list.
1529
 
        """
1530
 
        indices = []
1531
 
        pack_map = {}
1532
 
        for pack in packs:
1533
 
            index = getattr(pack, index_attribute)
1534
 
            indices.append(index)
1535
 
            pack_map[index] = (pack.pack_transport, pack.file_name())
1536
 
        return pack_map, indices
1537
 
 
1538
 
    def _index_contents(self, pack_map, key_filter=None):
1539
 
        """Get an iterable of the index contents from a pack_map.
1540
 
 
1541
 
        :param pack_map: A map from indices to pack details.
1542
 
        :param key_filter: An optional filter to limit the
1543
 
            keys returned.
1544
 
        """
1545
 
        indices = [index for index in pack_map.iterkeys()]
1546
 
        all_index = CombinedGraphIndex(indices)
1547
 
        if key_filter is None:
1548
 
            return all_index.iter_all_entries()
1549
 
        else:
1550
 
            return all_index.iter_entries(key_filter)
1551
 
 
1552
1324
    def _unlock_names(self):
1553
1325
        """Release the mutex around the pack-names index."""
1554
1326
        self.repo.control_files.unlock()
1555
1327
 
1556
 
    def _save_pack_names(self, clear_obsolete_packs=False):
1557
 
        """Save the list of packs.
1558
 
 
1559
 
        This will take out the mutex around the pack names list for the
1560
 
        duration of the method call. If concurrent updates have been made, a
1561
 
        three-way merge between the current list and the current in memory list
1562
 
        is performed.
1563
 
 
1564
 
        :param clear_obsolete_packs: If True, clear out the contents of the
1565
 
            obsolete_packs directory.
1566
 
        """
1567
 
        self.lock_names()
1568
 
        try:
1569
 
            builder = GraphIndexBuilder()
1570
 
            # load the disk nodes across
1571
 
            disk_nodes = set()
1572
 
            for index, key, value in self._iter_disk_pack_index():
1573
 
                disk_nodes.add((key, value))
1574
 
            # do a two-way diff against our original content
1575
 
            current_nodes = set()
1576
 
            for name, sizes in self._names.iteritems():
1577
 
                current_nodes.add(
1578
 
                    ((name, ), ' '.join(str(size) for size in sizes)))
1579
 
            deleted_nodes = self._packs_at_load - current_nodes
1580
 
            new_nodes = current_nodes - self._packs_at_load
1581
 
            disk_nodes.difference_update(deleted_nodes)
1582
 
            disk_nodes.update(new_nodes)
1583
 
            # TODO: handle same-name, index-size-changes here - 
1584
 
            # e.g. use the value from disk, not ours, *unless* we're the one
1585
 
            # changing it.
1586
 
            for key, value in disk_nodes:
1587
 
                builder.add_node(key, value)
1588
 
            self.transport.put_file('pack-names', builder.finish(),
1589
 
                mode=self.repo.control_files._file_mode)
1590
 
            # move the baseline forward
1591
 
            self._packs_at_load = disk_nodes
1592
 
            # now clear out the obsolete packs directory
1593
 
            if clear_obsolete_packs:
1594
 
                self.transport.clone('obsolete_packs').delete_multi(
1595
 
                    self.transport.list_dir('obsolete_packs'))
1596
 
        finally:
1597
 
            self._unlock_names()
1598
 
        # synchronise the memory packs list with what we just wrote:
 
1328
    def _diff_pack_names(self):
 
1329
        """Read the pack names from disk, and compare it to the one in memory.
 
1330
 
 
1331
        :return: (disk_nodes, deleted_nodes, new_nodes)
 
1332
            disk_nodes    The final set of nodes that should be referenced
 
1333
            deleted_nodes Nodes which have been removed from when we started
 
1334
            new_nodes     Nodes that are newly introduced
 
1335
        """
 
1336
        # load the disk nodes across
 
1337
        disk_nodes = set()
 
1338
        for index, key, value in self._iter_disk_pack_index():
 
1339
            disk_nodes.add((key, value))
 
1340
        orig_disk_nodes = set(disk_nodes)
 
1341
 
 
1342
        # do a two-way diff against our original content
 
1343
        current_nodes = set()
 
1344
        for name, sizes in self._names.iteritems():
 
1345
            current_nodes.add(
 
1346
                ((name, ), ' '.join(str(size) for size in sizes)))
 
1347
 
 
1348
        # Packs no longer present in the repository, which were present when we
 
1349
        # locked the repository
 
1350
        deleted_nodes = self._packs_at_load - current_nodes
 
1351
        # Packs which this process is adding
 
1352
        new_nodes = current_nodes - self._packs_at_load
 
1353
 
 
1354
        # Update the disk_nodes set to include the ones we are adding, and
 
1355
        # remove the ones which were removed by someone else
 
1356
        disk_nodes.difference_update(deleted_nodes)
 
1357
        disk_nodes.update(new_nodes)
 
1358
 
 
1359
        return disk_nodes, deleted_nodes, new_nodes, orig_disk_nodes
 
1360
 
 
1361
    def _syncronize_pack_names_from_disk_nodes(self, disk_nodes):
 
1362
        """Given the correct set of pack files, update our saved info.
 
1363
 
 
1364
        :return: (removed, added, modified)
 
1365
            removed     pack names removed from self._names
 
1366
            added       pack names added to self._names
 
1367
            modified    pack names that had changed value
 
1368
        """
 
1369
        removed = []
 
1370
        added = []
 
1371
        modified = []
 
1372
        ## self._packs_at_load = disk_nodes
1599
1373
        new_names = dict(disk_nodes)
1600
1374
        # drop no longer present nodes
1601
1375
        for pack in self.all_packs():
1602
1376
            if (pack.name,) not in new_names:
 
1377
                removed.append(pack.name)
1603
1378
                self._remove_pack_from_memory(pack)
1604
1379
        # add new nodes/refresh existing ones
1605
1380
        for key, value in disk_nodes:
1615
1390
                    # disk index because the set values are the same, unless
1616
1391
                    # the only index shows up as deleted by the set difference
1617
1392
                    # - which it may. Until there is a specific test for this,
1618
 
                    # assume its broken. RBC 20071017.
 
1393
                    # assume it's broken. RBC 20071017.
1619
1394
                    self._remove_pack_from_memory(self.get_pack_by_name(name))
1620
1395
                    self._names[name] = sizes
1621
1396
                    self.get_pack_by_name(name)
 
1397
                    modified.append(name)
1622
1398
            else:
1623
1399
                # new
1624
1400
                self._names[name] = sizes
1625
1401
                self.get_pack_by_name(name)
 
1402
                added.append(name)
 
1403
        return removed, added, modified
 
1404
 
 
1405
    def _save_pack_names(self, clear_obsolete_packs=False, obsolete_packs=None):
 
1406
        """Save the list of packs.
 
1407
 
 
1408
        This will take out the mutex around the pack names list for the
 
1409
        duration of the method call. If concurrent updates have been made, a
 
1410
        three-way merge between the current list and the current in memory list
 
1411
        is performed.
 
1412
 
 
1413
        :param clear_obsolete_packs: If True, clear out the contents of the
 
1414
            obsolete_packs directory.
 
1415
        :param obsolete_packs: Packs that are obsolete once the new pack-names
 
1416
            file has been written.
 
1417
        :return: A list of the names saved that were not previously on disk.
 
1418
        """
 
1419
        already_obsolete = []
 
1420
        self.lock_names()
 
1421
        try:
 
1422
            builder = self._index_builder_class()
 
1423
            (disk_nodes, deleted_nodes, new_nodes,
 
1424
             orig_disk_nodes) = self._diff_pack_names()
 
1425
            # TODO: handle same-name, index-size-changes here -
 
1426
            # e.g. use the value from disk, not ours, *unless* we're the one
 
1427
            # changing it.
 
1428
            for key, value in disk_nodes:
 
1429
                builder.add_node(key, value)
 
1430
            self.transport.put_file('pack-names', builder.finish(),
 
1431
                mode=self.repo.bzrdir._get_file_mode())
 
1432
            self._packs_at_load = disk_nodes
 
1433
            if clear_obsolete_packs:
 
1434
                to_preserve = None
 
1435
                if obsolete_packs:
 
1436
                    to_preserve = set([o.name for o in obsolete_packs])
 
1437
                already_obsolete = self._clear_obsolete_packs(to_preserve)
 
1438
        finally:
 
1439
            self._unlock_names()
 
1440
        # synchronise the memory packs list with what we just wrote:
 
1441
        self._syncronize_pack_names_from_disk_nodes(disk_nodes)
 
1442
        if obsolete_packs:
 
1443
            # TODO: We could add one more condition here. "if o.name not in
 
1444
            #       orig_disk_nodes and o != the new_pack we haven't written to
 
1445
            #       disk yet. However, the new pack object is not easily
 
1446
            #       accessible here (it would have to be passed through the
 
1447
            #       autopacking code, etc.)
 
1448
            obsolete_packs = [o for o in obsolete_packs
 
1449
                              if o.name not in already_obsolete]
 
1450
            self._obsolete_packs(obsolete_packs)
 
1451
        return [new_node[0][0] for new_node in new_nodes]
 
1452
 
 
1453
    def reload_pack_names(self):
 
1454
        """Sync our pack listing with what is present in the repository.
 
1455
 
 
1456
        This should be called when we find out that something we thought was
 
1457
        present is now missing. This happens when another process re-packs the
 
1458
        repository, etc.
 
1459
 
 
1460
        :return: True if the in-memory list of packs has been altered at all.
 
1461
        """
 
1462
        # The ensure_loaded call is to handle the case where the first call
 
1463
        # made involving the collection was to reload_pack_names, where we 
 
1464
        # don't have a view of disk contents. It's a bit of a bandaid, and
 
1465
        # causes two reads of pack-names, but it's a rare corner case not
 
1466
        # struck with regular push/pull etc.
 
1467
        first_read = self.ensure_loaded()
 
1468
        if first_read:
 
1469
            return True
 
1470
        # out the new value.
 
1471
        (disk_nodes, deleted_nodes, new_nodes,
 
1472
         orig_disk_nodes) = self._diff_pack_names()
 
1473
        # _packs_at_load is meant to be the explicit list of names in
 
1474
        # 'pack-names' at then start. As such, it should not contain any
 
1475
        # pending names that haven't been written out yet.
 
1476
        self._packs_at_load = orig_disk_nodes
 
1477
        (removed, added,
 
1478
         modified) = self._syncronize_pack_names_from_disk_nodes(disk_nodes)
 
1479
        if removed or added or modified:
 
1480
            return True
 
1481
        return False
 
1482
 
 
1483
    def _restart_autopack(self):
 
1484
        """Reload the pack names list, and restart the autopack code."""
 
1485
        if not self.reload_pack_names():
 
1486
            # Re-raise the original exception, because something went missing
 
1487
            # and a restart didn't find it
 
1488
            raise
 
1489
        raise errors.RetryAutopack(self.repo, False, sys.exc_info())
 
1490
 
 
1491
    def _restart_pack_operations(self):
 
1492
        """Reload the pack names list, and restart the autopack code."""
 
1493
        if not self.reload_pack_names():
 
1494
            # Re-raise the original exception, because something went missing
 
1495
            # and a restart didn't find it
 
1496
            raise
 
1497
        raise RetryPackOperations(self.repo, False, sys.exc_info())
 
1498
 
 
1499
    def _clear_obsolete_packs(self, preserve=None):
 
1500
        """Delete everything from the obsolete-packs directory.
 
1501
 
 
1502
        :return: A list of pack identifiers (the filename without '.pack') that
 
1503
            were found in obsolete_packs.
 
1504
        """
 
1505
        found = []
 
1506
        obsolete_pack_transport = self.transport.clone('obsolete_packs')
 
1507
        if preserve is None:
 
1508
            preserve = set()
 
1509
        try:
 
1510
            obsolete_pack_files = obsolete_pack_transport.list_dir('.')
 
1511
        except errors.NoSuchFile:
 
1512
            return found
 
1513
        for filename in obsolete_pack_files:
 
1514
            name, ext = osutils.splitext(filename)
 
1515
            if ext == '.pack':
 
1516
                found.append(name)
 
1517
            if name in preserve:
 
1518
                continue
 
1519
            try:
 
1520
                obsolete_pack_transport.delete(filename)
 
1521
            except (errors.PathError, errors.TransportError), e:
 
1522
                warning("couldn't delete obsolete pack, skipping it:\n%s"
 
1523
                        % (e,))
 
1524
        return found
1626
1525
 
1627
1526
    def _start_write_group(self):
1628
1527
        # Do not permit preparation for writing if we're not in a 'write lock'.
1629
1528
        if not self.repo.is_write_locked():
1630
1529
            raise errors.NotWriteLocked(self)
1631
 
        self._new_pack = NewPack(self._upload_transport, self._index_transport,
1632
 
            self._pack_transport, upload_suffix='.pack',
1633
 
            file_mode=self.repo.control_files._file_mode)
 
1530
        self._new_pack = self.pack_factory(self, upload_suffix='.pack',
 
1531
            file_mode=self.repo.bzrdir._get_file_mode())
1634
1532
        # allow writing: queue writes to a new index
1635
1533
        self.revision_index.add_writable_index(self._new_pack.revision_index,
1636
1534
            self._new_pack)
1638
1536
            self._new_pack)
1639
1537
        self.text_index.add_writable_index(self._new_pack.text_index,
1640
1538
            self._new_pack)
 
1539
        self._new_pack.text_index.set_optimize(combine_backing_indices=False)
1641
1540
        self.signature_index.add_writable_index(self._new_pack.signature_index,
1642
1541
            self._new_pack)
 
1542
        if self.chk_index is not None:
 
1543
            self.chk_index.add_writable_index(self._new_pack.chk_index,
 
1544
                self._new_pack)
 
1545
            self.repo.chk_bytes._index._add_callback = self.chk_index.add_callback
 
1546
            self._new_pack.chk_index.set_optimize(combine_backing_indices=False)
1643
1547
 
1644
 
        # reused revision and signature knits may need updating
1645
 
        #
1646
 
        # "Hysterical raisins. client code in bzrlib grabs those knits outside
1647
 
        # of write groups and then mutates it inside the write group."
1648
 
        if self.repo._revision_knit is not None:
1649
 
            self.repo._revision_knit._index._add_callback = \
1650
 
                self.revision_index.add_callback
1651
 
        if self.repo._signature_knit is not None:
1652
 
            self.repo._signature_knit._index._add_callback = \
1653
 
                self.signature_index.add_callback
1654
 
        # create a reused knit object for text addition in commit.
1655
 
        self.repo._text_knit = self.repo.weave_store.get_weave_or_empty(
1656
 
            'all-texts', None)
 
1548
        self.repo.inventories._index._add_callback = self.inventory_index.add_callback
 
1549
        self.repo.revisions._index._add_callback = self.revision_index.add_callback
 
1550
        self.repo.signatures._index._add_callback = self.signature_index.add_callback
 
1551
        self.repo.texts._index._add_callback = self.text_index.add_callback
1657
1552
 
1658
1553
    def _abort_write_group(self):
1659
1554
        # FIXME: just drop the transient index.
1660
1555
        # forget what names there are
1661
1556
        if self._new_pack is not None:
1662
 
            self._new_pack.abort()
1663
 
            self._remove_pack_indices(self._new_pack)
1664
 
            self._new_pack = None
1665
 
        self.repo._text_knit = None
1666
 
 
 
1557
            operation = cleanup.OperationWithCleanups(self._new_pack.abort)
 
1558
            operation.add_cleanup(setattr, self, '_new_pack', None)
 
1559
            # If we aborted while in the middle of finishing the write
 
1560
            # group, _remove_pack_indices could fail because the indexes are
 
1561
            # already gone.  But they're not there we shouldn't fail in this
 
1562
            # case, so we pass ignore_missing=True.
 
1563
            operation.add_cleanup(self._remove_pack_indices, self._new_pack,
 
1564
                ignore_missing=True)
 
1565
            operation.run_simple()
 
1566
        for resumed_pack in self._resumed_packs:
 
1567
            operation = cleanup.OperationWithCleanups(resumed_pack.abort)
 
1568
            # See comment in previous finally block.
 
1569
            operation.add_cleanup(self._remove_pack_indices, resumed_pack,
 
1570
                ignore_missing=True)
 
1571
            operation.run_simple()
 
1572
        del self._resumed_packs[:]
 
1573
 
 
1574
    def _remove_resumed_pack_indices(self):
 
1575
        for resumed_pack in self._resumed_packs:
 
1576
            self._remove_pack_indices(resumed_pack)
 
1577
        del self._resumed_packs[:]
 
1578
 
 
1579
    def _check_new_inventories(self):
 
1580
        """Detect missing inventories in this write group.
 
1581
 
 
1582
        :returns: list of strs, summarising any problems found.  If the list is
 
1583
            empty no problems were found.
 
1584
        """
 
1585
        # The base implementation does no checks.  GCRepositoryPackCollection
 
1586
        # overrides this.
 
1587
        return []
 
1588
        
1667
1589
    def _commit_write_group(self):
 
1590
        all_missing = set()
 
1591
        for prefix, versioned_file in (
 
1592
                ('revisions', self.repo.revisions),
 
1593
                ('inventories', self.repo.inventories),
 
1594
                ('texts', self.repo.texts),
 
1595
                ('signatures', self.repo.signatures),
 
1596
                ):
 
1597
            missing = versioned_file.get_missing_compression_parent_keys()
 
1598
            all_missing.update([(prefix,) + key for key in missing])
 
1599
        if all_missing:
 
1600
            raise errors.BzrCheckError(
 
1601
                "Repository %s has missing compression parent(s) %r "
 
1602
                 % (self.repo, sorted(all_missing)))
 
1603
        problems = self._check_new_inventories()
 
1604
        if problems:
 
1605
            problems_summary = '\n'.join(problems)
 
1606
            raise errors.BzrCheckError(
 
1607
                "Cannot add revision(s) to repository: " + problems_summary)
1668
1608
        self._remove_pack_indices(self._new_pack)
 
1609
        any_new_content = False
1669
1610
        if self._new_pack.data_inserted():
1670
1611
            # get all the data to disk and read to use
1671
1612
            self._new_pack.finish()
1672
1613
            self.allocate(self._new_pack)
1673
1614
            self._new_pack = None
1674
 
            if not self.autopack():
 
1615
            any_new_content = True
 
1616
        else:
 
1617
            self._new_pack.abort()
 
1618
            self._new_pack = None
 
1619
        for resumed_pack in self._resumed_packs:
 
1620
            # XXX: this is a pretty ugly way to turn the resumed pack into a
 
1621
            # properly committed pack.
 
1622
            self._names[resumed_pack.name] = None
 
1623
            self._remove_pack_from_memory(resumed_pack)
 
1624
            resumed_pack.finish()
 
1625
            self.allocate(resumed_pack)
 
1626
            any_new_content = True
 
1627
        del self._resumed_packs[:]
 
1628
        if any_new_content:
 
1629
            result = self.autopack()
 
1630
            if not result:
1675
1631
                # when autopack takes no steps, the names list is still
1676
1632
                # unsaved.
1677
 
                self._save_pack_names()
 
1633
                return self._save_pack_names()
 
1634
            return result
 
1635
        return []
 
1636
 
 
1637
    def _suspend_write_group(self):
 
1638
        tokens = [pack.name for pack in self._resumed_packs]
 
1639
        self._remove_pack_indices(self._new_pack)
 
1640
        if self._new_pack.data_inserted():
 
1641
            # get all the data to disk and read to use
 
1642
            self._new_pack.finish(suspend=True)
 
1643
            tokens.append(self._new_pack.name)
 
1644
            self._new_pack = None
1678
1645
        else:
1679
1646
            self._new_pack.abort()
1680
1647
            self._new_pack = None
1681
 
        self.repo._text_knit = None
1682
 
 
1683
 
 
1684
 
class KnitPackRevisionStore(KnitRevisionStore):
1685
 
    """An object to adapt access from RevisionStore's to use KnitPacks.
1686
 
 
1687
 
    This class works by replacing the original RevisionStore.
1688
 
    We need to do this because the KnitPackRevisionStore is less
1689
 
    isolated in its layering - it uses services from the repo.
1690
 
    """
1691
 
 
1692
 
    def __init__(self, repo, transport, revisionstore):
1693
 
        """Create a KnitPackRevisionStore on repo with revisionstore.
1694
 
 
1695
 
        This will store its state in the Repository, use the
1696
 
        indices to provide a KnitGraphIndex,
1697
 
        and at the end of transactions write new indices.
1698
 
        """
1699
 
        KnitRevisionStore.__init__(self, revisionstore.versioned_file_store)
1700
 
        self.repo = repo
1701
 
        self._serializer = revisionstore._serializer
1702
 
        self.transport = transport
1703
 
 
1704
 
    def get_revision_file(self, transaction):
1705
 
        """Get the revision versioned file object."""
1706
 
        if getattr(self.repo, '_revision_knit', None) is not None:
1707
 
            return self.repo._revision_knit
1708
 
        self.repo._pack_collection.ensure_loaded()
1709
 
        add_callback = self.repo._pack_collection.revision_index.add_callback
1710
 
        # setup knit specific objects
1711
 
        knit_index = KnitGraphIndex(
1712
 
            self.repo._pack_collection.revision_index.combined_index,
1713
 
            add_callback=add_callback)
1714
 
        self.repo._revision_knit = knit.KnitVersionedFile(
1715
 
            'revisions', self.transport.clone('..'),
1716
 
            self.repo.control_files._file_mode,
1717
 
            create=False,
1718
 
            index=knit_index, delta=False, factory=knit.KnitPlainFactory(),
1719
 
            access_method=self.repo._pack_collection.revision_index.knit_access)
1720
 
        return self.repo._revision_knit
1721
 
 
1722
 
    def get_signature_file(self, transaction):
1723
 
        """Get the signature versioned file object."""
1724
 
        if getattr(self.repo, '_signature_knit', None) is not None:
1725
 
            return self.repo._signature_knit
1726
 
        self.repo._pack_collection.ensure_loaded()
1727
 
        add_callback = self.repo._pack_collection.signature_index.add_callback
1728
 
        # setup knit specific objects
1729
 
        knit_index = KnitGraphIndex(
1730
 
            self.repo._pack_collection.signature_index.combined_index,
1731
 
            add_callback=add_callback, parents=False)
1732
 
        self.repo._signature_knit = knit.KnitVersionedFile(
1733
 
            'signatures', self.transport.clone('..'),
1734
 
            self.repo.control_files._file_mode,
1735
 
            create=False,
1736
 
            index=knit_index, delta=False, factory=knit.KnitPlainFactory(),
1737
 
            access_method=self.repo._pack_collection.signature_index.knit_access)
1738
 
        return self.repo._signature_knit
1739
 
 
1740
 
 
1741
 
class KnitPackTextStore(VersionedFileStore):
1742
 
    """Presents a TextStore abstraction on top of packs.
1743
 
 
1744
 
    This class works by replacing the original VersionedFileStore.
1745
 
    We need to do this because the KnitPackRevisionStore is less
1746
 
    isolated in its layering - it uses services from the repo and shares them
1747
 
    with all the data written in a single write group.
1748
 
    """
1749
 
 
1750
 
    def __init__(self, repo, transport, weavestore):
1751
 
        """Create a KnitPackTextStore on repo with weavestore.
1752
 
 
1753
 
        This will store its state in the Repository, use the
1754
 
        indices FileNames to provide a KnitGraphIndex,
1755
 
        and at the end of transactions write new indices.
1756
 
        """
1757
 
        # don't call base class constructor - it's not suitable.
1758
 
        # no transient data stored in the transaction
1759
 
        # cache.
1760
 
        self._precious = False
1761
 
        self.repo = repo
1762
 
        self.transport = transport
1763
 
        self.weavestore = weavestore
1764
 
        # XXX for check() which isn't updated yet
1765
 
        self._transport = weavestore._transport
1766
 
 
1767
 
    def get_weave_or_empty(self, file_id, transaction):
1768
 
        """Get a 'Knit' backed by the .tix indices.
1769
 
 
1770
 
        The transaction parameter is ignored.
1771
 
        """
1772
 
        self.repo._pack_collection.ensure_loaded()
1773
 
        add_callback = self.repo._pack_collection.text_index.add_callback
1774
 
        # setup knit specific objects
1775
 
        file_id_index = GraphIndexPrefixAdapter(
1776
 
            self.repo._pack_collection.text_index.combined_index,
1777
 
            (file_id, ), 1, add_nodes_callback=add_callback)
1778
 
        knit_index = KnitGraphIndex(file_id_index,
1779
 
            add_callback=file_id_index.add_nodes,
1780
 
            deltas=True, parents=True)
1781
 
        return knit.KnitVersionedFile('text:' + file_id,
1782
 
            self.transport.clone('..'),
1783
 
            None,
1784
 
            index=knit_index,
1785
 
            access_method=self.repo._pack_collection.text_index.knit_access,
1786
 
            factory=knit.KnitPlainFactory())
1787
 
 
1788
 
    get_weave = get_weave_or_empty
1789
 
 
1790
 
    def __iter__(self):
1791
 
        """Generate a list of the fileids inserted, for use by check."""
1792
 
        self.repo._pack_collection.ensure_loaded()
1793
 
        ids = set()
1794
 
        for index, key, value, refs in \
1795
 
            self.repo._pack_collection.text_index.combined_index.iter_all_entries():
1796
 
            ids.add(key[0])
1797
 
        return iter(ids)
1798
 
 
1799
 
 
1800
 
class InventoryKnitThunk(object):
1801
 
    """An object to manage thunking get_inventory_weave to pack based knits."""
1802
 
 
1803
 
    def __init__(self, repo, transport):
1804
 
        """Create an InventoryKnitThunk for repo at transport.
1805
 
 
1806
 
        This will store its state in the Repository, use the
1807
 
        indices FileNames to provide a KnitGraphIndex,
1808
 
        and at the end of transactions write a new index..
1809
 
        """
1810
 
        self.repo = repo
1811
 
        self.transport = transport
1812
 
 
1813
 
    def get_weave(self):
1814
 
        """Get a 'Knit' that contains inventory data."""
1815
 
        self.repo._pack_collection.ensure_loaded()
1816
 
        add_callback = self.repo._pack_collection.inventory_index.add_callback
1817
 
        # setup knit specific objects
1818
 
        knit_index = KnitGraphIndex(
1819
 
            self.repo._pack_collection.inventory_index.combined_index,
1820
 
            add_callback=add_callback, deltas=True, parents=True)
1821
 
        return knit.KnitVersionedFile(
1822
 
            'inventory', self.transport.clone('..'),
1823
 
            self.repo.control_files._file_mode,
1824
 
            create=False,
1825
 
            index=knit_index, delta=True, factory=knit.KnitPlainFactory(),
1826
 
            access_method=self.repo._pack_collection.inventory_index.knit_access)
1827
 
 
1828
 
 
1829
 
class KnitPackRepository(KnitRepository):
1830
 
    """Experimental graph-knit using repository."""
1831
 
 
1832
 
    def __init__(self, _format, a_bzrdir, control_files, _revision_store,
1833
 
        control_store, text_store, _commit_builder_class, _serializer):
1834
 
        KnitRepository.__init__(self, _format, a_bzrdir, control_files,
1835
 
            _revision_store, control_store, text_store, _commit_builder_class,
1836
 
            _serializer)
1837
 
        index_transport = control_files._transport.clone('indices')
1838
 
        self._pack_collection = RepositoryPackCollection(self, control_files._transport,
1839
 
            index_transport,
1840
 
            control_files._transport.clone('upload'),
1841
 
            control_files._transport.clone('packs'))
1842
 
        self._revision_store = KnitPackRevisionStore(self, index_transport, self._revision_store)
1843
 
        self.weave_store = KnitPackTextStore(self, index_transport, self.weave_store)
1844
 
        self._inv_thunk = InventoryKnitThunk(self, index_transport)
1845
 
        # True when the repository object is 'write locked' (as opposed to the
1846
 
        # physical lock only taken out around changes to the pack-names list.) 
1847
 
        # Another way to represent this would be a decorator around the control
1848
 
        # files object that presents logical locks as physical ones - if this
1849
 
        # gets ugly consider that alternative design. RBC 20071011
1850
 
        self._write_lock_count = 0
1851
 
        self._transaction = None
1852
 
        # for tests
1853
 
        self._reconcile_does_inventory_gc = True
 
1648
        self._remove_resumed_pack_indices()
 
1649
        return tokens
 
1650
 
 
1651
    def _resume_write_group(self, tokens):
 
1652
        for token in tokens:
 
1653
            self._resume_pack(token)
 
1654
 
 
1655
 
 
1656
class PackRepository(MetaDirVersionedFileRepository):
 
1657
    """Repository with knit objects stored inside pack containers.
 
1658
 
 
1659
    The layering for a KnitPackRepository is:
 
1660
 
 
1661
    Graph        |  HPSS    | Repository public layer |
 
1662
    ===================================================
 
1663
    Tuple based apis below, string based, and key based apis above
 
1664
    ---------------------------------------------------
 
1665
    VersionedFiles
 
1666
      Provides .texts, .revisions etc
 
1667
      This adapts the N-tuple keys to physical knit records which only have a
 
1668
      single string identifier (for historical reasons), which in older formats
 
1669
      was always the revision_id, and in the mapped code for packs is always
 
1670
      the last element of key tuples.
 
1671
    ---------------------------------------------------
 
1672
    GraphIndex
 
1673
      A separate GraphIndex is used for each of the
 
1674
      texts/inventories/revisions/signatures contained within each individual
 
1675
      pack file. The GraphIndex layer works in N-tuples and is unaware of any
 
1676
      semantic value.
 
1677
    ===================================================
 
1678
 
 
1679
    """
 
1680
 
 
1681
    # These attributes are inherited from the Repository base class. Setting
 
1682
    # them to None ensures that if the constructor is changed to not initialize
 
1683
    # them, or a subclass fails to call the constructor, that an error will
 
1684
    # occur rather than the system working but generating incorrect data.
 
1685
    _commit_builder_class = None
 
1686
    _serializer = None
 
1687
 
 
1688
    def __init__(self, _format, a_bzrdir, control_files, _commit_builder_class,
 
1689
        _serializer):
 
1690
        MetaDirRepository.__init__(self, _format, a_bzrdir, control_files)
 
1691
        self._commit_builder_class = _commit_builder_class
 
1692
        self._serializer = _serializer
1854
1693
        self._reconcile_fixes_text_parents = True
1855
 
        self._reconcile_backsup_inventory = False
 
1694
        if self._format.supports_external_lookups:
 
1695
            self._unstacked_provider = graph.CachingParentsProvider(
 
1696
                self._make_parents_provider_unstacked())
 
1697
        else:
 
1698
            self._unstacked_provider = graph.CachingParentsProvider(self)
 
1699
        self._unstacked_provider.disable_cache()
 
1700
 
 
1701
    @needs_read_lock
 
1702
    def _all_revision_ids(self):
 
1703
        """See Repository.all_revision_ids()."""
 
1704
        return [key[0] for key in self.revisions.keys()]
1856
1705
 
1857
1706
    def _abort_write_group(self):
 
1707
        self.revisions._index._key_dependencies.clear()
1858
1708
        self._pack_collection._abort_write_group()
1859
1709
 
1860
 
    def _find_inconsistent_revision_parents(self):
1861
 
        """Find revisions with incorrectly cached parents.
1862
 
 
1863
 
        :returns: an iterator yielding tuples of (revison-id, parents-in-index,
1864
 
            parents-in-revision).
1865
 
        """
1866
 
        if not self.is_locked():
1867
 
            raise errors.ObjectNotLocked(self)
1868
 
        pb = ui.ui_factory.nested_progress_bar()
1869
 
        result = []
1870
 
        try:
1871
 
            revision_nodes = self._pack_collection.revision_index \
1872
 
                .combined_index.iter_all_entries()
1873
 
            index_positions = []
1874
 
            # Get the cached index values for all revisions, and also the location
1875
 
            # in each index of the revision text so we can perform linear IO.
1876
 
            for index, key, value, refs in revision_nodes:
1877
 
                pos, length = value[1:].split(' ')
1878
 
                index_positions.append((index, int(pos), key[0],
1879
 
                    tuple(parent[0] for parent in refs[0])))
1880
 
                pb.update("Reading revision index.", 0, 0)
1881
 
            index_positions.sort()
1882
 
            batch_count = len(index_positions) / 1000 + 1
1883
 
            pb.update("Checking cached revision graph.", 0, batch_count)
1884
 
            for offset in xrange(batch_count):
1885
 
                pb.update("Checking cached revision graph.", offset)
1886
 
                to_query = index_positions[offset * 1000:(offset + 1) * 1000]
1887
 
                if not to_query:
1888
 
                    break
1889
 
                rev_ids = [item[2] for item in to_query]
1890
 
                revs = self.get_revisions(rev_ids)
1891
 
                for revision, item in zip(revs, to_query):
1892
 
                    index_parents = item[3]
1893
 
                    rev_parents = tuple(revision.parent_ids)
1894
 
                    if index_parents != rev_parents:
1895
 
                        result.append((revision.revision_id, index_parents, rev_parents))
1896
 
        finally:
1897
 
            pb.finished()
1898
 
        return result
1899
 
 
1900
 
    @symbol_versioning.deprecated_method(symbol_versioning.one_one)
1901
 
    def get_parents(self, revision_ids):
1902
 
        """See graph._StackedParentsProvider.get_parents."""
1903
 
        parent_map = self.get_parent_map(revision_ids)
1904
 
        return [parent_map.get(r, None) for r in revision_ids]
1905
 
 
1906
 
    def get_parent_map(self, keys):
1907
 
        """See graph._StackedParentsProvider.get_parent_map
1908
 
 
1909
 
        This implementation accesses the combined revision index to provide
1910
 
        answers.
1911
 
        """
1912
 
        self._pack_collection.ensure_loaded()
1913
 
        index = self._pack_collection.revision_index.combined_index
1914
 
        keys = set(keys)
1915
 
        if None in keys:
1916
 
            raise ValueError('get_parent_map(None) is not valid')
1917
 
        if _mod_revision.NULL_REVISION in keys:
1918
 
            keys.discard(_mod_revision.NULL_REVISION)
1919
 
            found_parents = {_mod_revision.NULL_REVISION:()}
1920
 
        else:
1921
 
            found_parents = {}
1922
 
        search_keys = set((revision_id,) for revision_id in keys)
1923
 
        for index, key, value, refs in index.iter_entries(search_keys):
1924
 
            parents = refs[0]
1925
 
            if not parents:
1926
 
                parents = (_mod_revision.NULL_REVISION,)
1927
 
            else:
1928
 
                parents = tuple(parent[0] for parent in parents)
1929
 
            found_parents[key[0]] = parents
1930
 
        return found_parents
1931
 
 
1932
 
    def has_revisions(self, revision_ids):
1933
 
        """See Repository.has_revisions()."""
1934
 
        revision_ids = set(revision_ids)
1935
 
        result = revision_ids.intersection(
1936
 
            set([None, _mod_revision.NULL_REVISION]))
1937
 
        revision_ids.difference_update(result)
1938
 
        index = self._pack_collection.revision_index.combined_index
1939
 
        keys = [(revision_id,) for revision_id in revision_ids]
1940
 
        result.update(node[1][0] for node in index.iter_entries(keys))
1941
 
        return result
1942
 
 
1943
1710
    def _make_parents_provider(self):
1944
 
        return graph.CachingParentsProvider(self)
 
1711
        if not self._format.supports_external_lookups:
 
1712
            return self._unstacked_provider
 
1713
        return graph.StackedParentsProvider(_LazyListJoin(
 
1714
            [self._unstacked_provider], self._fallback_repositories))
1945
1715
 
1946
1716
    def _refresh_data(self):
1947
 
        if self._write_lock_count == 1 or (
1948
 
            self.control_files._lock_count == 1 and
1949
 
            self.control_files._lock_mode == 'r'):
1950
 
            # forget what names there are
1951
 
            self._pack_collection.reset()
1952
 
            # XXX: Better to do an in-memory merge when acquiring a new lock -
1953
 
            # factor out code from _save_pack_names.
1954
 
            self._pack_collection.ensure_loaded()
 
1717
        if not self.is_locked():
 
1718
            return
 
1719
        self._pack_collection.reload_pack_names()
 
1720
        self._unstacked_provider.disable_cache()
 
1721
        self._unstacked_provider.enable_cache()
1955
1722
 
1956
1723
    def _start_write_group(self):
1957
1724
        self._pack_collection._start_write_group()
1958
1725
 
1959
1726
    def _commit_write_group(self):
1960
 
        return self._pack_collection._commit_write_group()
1961
 
 
1962
 
    def get_inventory_weave(self):
1963
 
        return self._inv_thunk.get_weave()
 
1727
        hint = self._pack_collection._commit_write_group()
 
1728
        self.revisions._index._key_dependencies.clear()
 
1729
        # The commit may have added keys that were previously cached as
 
1730
        # missing, so reset the cache.
 
1731
        self._unstacked_provider.disable_cache()
 
1732
        self._unstacked_provider.enable_cache()
 
1733
        return hint
 
1734
 
 
1735
    def suspend_write_group(self):
 
1736
        # XXX check self._write_group is self.get_transaction()?
 
1737
        tokens = self._pack_collection._suspend_write_group()
 
1738
        self.revisions._index._key_dependencies.clear()
 
1739
        self._write_group = None
 
1740
        return tokens
 
1741
 
 
1742
    def _resume_write_group(self, tokens):
 
1743
        self._start_write_group()
 
1744
        try:
 
1745
            self._pack_collection._resume_write_group(tokens)
 
1746
        except errors.UnresumableWriteGroup:
 
1747
            self._abort_write_group()
 
1748
            raise
 
1749
        for pack in self._pack_collection._resumed_packs:
 
1750
            self.revisions._index.scan_unvalidated_index(pack.revision_index)
1964
1751
 
1965
1752
    def get_transaction(self):
1966
1753
        if self._write_lock_count:
1975
1762
        return self._write_lock_count
1976
1763
 
1977
1764
    def lock_write(self, token=None):
1978
 
        if not self._write_lock_count and self.is_locked():
 
1765
        """Lock the repository for writes.
 
1766
 
 
1767
        :return: A bzrlib.repository.RepositoryWriteLockResult.
 
1768
        """
 
1769
        locked = self.is_locked()
 
1770
        if not self._write_lock_count and locked:
1979
1771
            raise errors.ReadOnlyError(self)
1980
1772
        self._write_lock_count += 1
1981
1773
        if self._write_lock_count == 1:
1982
 
            from bzrlib import transactions
1983
1774
            self._transaction = transactions.WriteTransaction()
1984
 
        self._refresh_data()
 
1775
        if not locked:
 
1776
            if 'relock' in debug.debug_flags and self._prev_lock == 'w':
 
1777
                note('%r was write locked again', self)
 
1778
            self._prev_lock = 'w'
 
1779
            self._unstacked_provider.enable_cache()
 
1780
            for repo in self._fallback_repositories:
 
1781
                # Writes don't affect fallback repos
 
1782
                repo.lock_read()
 
1783
            self._refresh_data()
 
1784
        return RepositoryWriteLockResult(self.unlock, None)
1985
1785
 
1986
1786
    def lock_read(self):
 
1787
        """Lock the repository for reads.
 
1788
 
 
1789
        :return: A bzrlib.lock.LogicalLockResult.
 
1790
        """
 
1791
        locked = self.is_locked()
1987
1792
        if self._write_lock_count:
1988
1793
            self._write_lock_count += 1
1989
1794
        else:
1990
1795
            self.control_files.lock_read()
1991
 
        self._refresh_data()
 
1796
        if not locked:
 
1797
            if 'relock' in debug.debug_flags and self._prev_lock == 'r':
 
1798
                note('%r was read locked again', self)
 
1799
            self._prev_lock = 'r'
 
1800
            self._unstacked_provider.enable_cache()
 
1801
            for repo in self._fallback_repositories:
 
1802
                repo.lock_read()
 
1803
            self._refresh_data()
 
1804
        return LogicalLockResult(self.unlock)
1992
1805
 
1993
1806
    def leave_lock_in_place(self):
1994
1807
        # not supported - raise an error
1999
1812
        raise NotImplementedError(self.dont_leave_lock_in_place)
2000
1813
 
2001
1814
    @needs_write_lock
2002
 
    def pack(self):
 
1815
    def pack(self, hint=None, clean_obsolete_packs=False):
2003
1816
        """Compress the data within the repository.
2004
1817
 
2005
1818
        This will pack all the data to a single pack. In future it may
2006
1819
        recompress deltas or do other such expensive operations.
2007
1820
        """
2008
 
        self._pack_collection.pack()
 
1821
        self._pack_collection.pack(hint=hint, clean_obsolete_packs=clean_obsolete_packs)
2009
1822
 
2010
1823
    @needs_write_lock
2011
1824
    def reconcile(self, other=None, thorough=False):
2015
1828
        reconciler.reconcile()
2016
1829
        return reconciler
2017
1830
 
 
1831
    def _reconcile_pack(self, collection, packs, extension, revs, pb):
 
1832
        raise NotImplementedError(self._reconcile_pack)
 
1833
 
 
1834
    @only_raises(errors.LockNotHeld, errors.LockBroken)
2018
1835
    def unlock(self):
2019
1836
        if self._write_lock_count == 1 and self._write_group is not None:
2020
1837
            self.abort_write_group()
 
1838
            self._unstacked_provider.disable_cache()
2021
1839
            self._transaction = None
2022
1840
            self._write_lock_count = 0
2023
1841
            raise errors.BzrError(
2032
1850
        else:
2033
1851
            self.control_files.unlock()
2034
1852
 
2035
 
 
2036
 
class RepositoryFormatPack(MetaDirRepositoryFormat):
 
1853
        if not self.is_locked():
 
1854
            self._unstacked_provider.disable_cache()
 
1855
            for repo in self._fallback_repositories:
 
1856
                repo.unlock()
 
1857
 
 
1858
 
 
1859
class RepositoryFormatPack(MetaDirVersionedFileRepositoryFormat):
2037
1860
    """Format logic for pack structured repositories.
2038
1861
 
2039
1862
    This repository format has:
2058
1881
    # Set this attribute in derived clases to control the _serializer that the
2059
1882
    # repository objects will have passed to their constructor.
2060
1883
    _serializer = None
 
1884
    # Packs are not confused by ghosts.
 
1885
    supports_ghosts = True
2061
1886
    # External references are not supported in pack repositories yet.
2062
1887
    supports_external_lookups = False
2063
 
 
2064
 
    def _get_control_store(self, repo_transport, control_files):
2065
 
        """Return the control store for this repository."""
2066
 
        return VersionedFileStore(
2067
 
            repo_transport,
2068
 
            prefixed=False,
2069
 
            file_mode=control_files._file_mode,
2070
 
            versionedfile_class=knit.make_file_knit,
2071
 
            versionedfile_kwargs={'factory': knit.KnitPlainFactory()},
2072
 
            )
2073
 
 
2074
 
    def _get_revision_store(self, repo_transport, control_files):
2075
 
        """See RepositoryFormat._get_revision_store()."""
2076
 
        versioned_file_store = VersionedFileStore(
2077
 
            repo_transport,
2078
 
            file_mode=control_files._file_mode,
2079
 
            prefixed=False,
2080
 
            precious=True,
2081
 
            versionedfile_class=knit.make_file_knit,
2082
 
            versionedfile_kwargs={'delta': False,
2083
 
                                  'factory': knit.KnitPlainFactory(),
2084
 
                                 },
2085
 
            escaped=True,
2086
 
            )
2087
 
        return KnitRevisionStore(versioned_file_store)
2088
 
 
2089
 
    def _get_text_store(self, transport, control_files):
2090
 
        """See RepositoryFormat._get_text_store()."""
2091
 
        return self._get_versioned_file_store('knits',
2092
 
                                  transport,
2093
 
                                  control_files,
2094
 
                                  versionedfile_class=knit.make_file_knit,
2095
 
                                  versionedfile_kwargs={
2096
 
                                      'create_parent_dir': True,
2097
 
                                      'delay_create': True,
2098
 
                                      'dir_mode': control_files._dir_mode,
2099
 
                                  },
2100
 
                                  escaped=True)
 
1888
    # Most pack formats do not use chk lookups.
 
1889
    supports_chks = False
 
1890
    # What index classes to use
 
1891
    index_builder_class = None
 
1892
    index_class = None
 
1893
    _fetch_uses_deltas = True
 
1894
    fast_deltas = False
 
1895
    supports_funky_characters = True
 
1896
    revision_graph_can_have_wrong_parents = True
2101
1897
 
2102
1898
    def initialize(self, a_bzrdir, shared=False):
2103
1899
        """Create a pack based repository.
2109
1905
        """
2110
1906
        mutter('creating repository in %s.', a_bzrdir.transport.base)
2111
1907
        dirs = ['indices', 'obsolete_packs', 'packs', 'upload']
2112
 
        builder = GraphIndexBuilder()
 
1908
        builder = self.index_builder_class()
2113
1909
        files = [('pack-names', builder.finish())]
2114
1910
        utf8_files = [('format', self.get_format_string())]
2115
 
        
 
1911
 
2116
1912
        self._upload_blank_content(a_bzrdir, dirs, files, utf8_files, shared)
2117
 
        return self.open(a_bzrdir=a_bzrdir, _found=True)
 
1913
        repository = self.open(a_bzrdir=a_bzrdir, _found=True)
 
1914
        self._run_post_repo_init_hooks(repository, a_bzrdir, shared)
 
1915
        return repository
2118
1916
 
2119
1917
    def open(self, a_bzrdir, _found=False, _override_transport=None):
2120
1918
        """See RepositoryFormat.open().
2121
 
        
 
1919
 
2122
1920
        :param _override_transport: INTERNAL USE ONLY. Allows opening the
2123
1921
                                    repository at a slightly different url
2124
1922
                                    than normal. I.e. during 'upgrade'.
2125
1923
        """
2126
1924
        if not _found:
2127
 
            format = RepositoryFormat.find_format(a_bzrdir)
 
1925
            format = RepositoryFormatMetaDir.find_format(a_bzrdir)
2128
1926
        if _override_transport is not None:
2129
1927
            repo_transport = _override_transport
2130
1928
        else:
2131
1929
            repo_transport = a_bzrdir.get_repository_transport(None)
2132
1930
        control_files = lockable_files.LockableFiles(repo_transport,
2133
1931
                                'lock', lockdir.LockDir)
2134
 
        text_store = self._get_text_store(repo_transport, control_files)
2135
 
        control_store = self._get_control_store(repo_transport, control_files)
2136
 
        _revision_store = self._get_revision_store(repo_transport, control_files)
2137
1932
        return self.repository_class(_format=self,
2138
1933
                              a_bzrdir=a_bzrdir,
2139
1934
                              control_files=control_files,
2140
 
                              _revision_store=_revision_store,
2141
 
                              control_store=control_store,
2142
 
                              text_store=text_store,
2143
1935
                              _commit_builder_class=self._commit_builder_class,
2144
1936
                              _serializer=self._serializer)
2145
1937
 
2146
1938
 
2147
 
class RepositoryFormatKnitPack1(RepositoryFormatPack):
2148
 
    """A no-subtrees parameterized Pack repository.
2149
 
 
2150
 
    This format was introduced in 0.92.
2151
 
    """
2152
 
 
2153
 
    repository_class = KnitPackRepository
2154
 
    _commit_builder_class = PackCommitBuilder
2155
 
    _serializer = xml5.serializer_v5
2156
 
 
2157
 
    def _get_matching_bzrdir(self):
2158
 
        return bzrdir.format_registry.make_bzrdir('pack-0.92')
2159
 
 
2160
 
    def _ignore_setting_bzrdir(self, format):
2161
 
        pass
2162
 
 
2163
 
    _matchingbzrdir = property(_get_matching_bzrdir, _ignore_setting_bzrdir)
2164
 
 
2165
 
    def get_format_string(self):
2166
 
        """See RepositoryFormat.get_format_string()."""
2167
 
        return "Bazaar pack repository format 1 (needs bzr 0.92)\n"
2168
 
 
2169
 
    def get_format_description(self):
2170
 
        """See RepositoryFormat.get_format_description()."""
2171
 
        return "Packs containing knits without subtree support"
2172
 
 
2173
 
    def check_conversion_target(self, target_format):
2174
 
        pass
2175
 
 
2176
 
 
2177
 
class RepositoryFormatKnitPack3(RepositoryFormatPack):
2178
 
    """A subtrees parameterized Pack repository.
2179
 
 
2180
 
    This repository format uses the xml7 serializer to get:
2181
 
     - support for recording full info about the tree root
2182
 
     - support for recording tree-references
2183
 
 
2184
 
    This format was introduced in 0.92.
2185
 
    """
2186
 
 
2187
 
    repository_class = KnitPackRepository
2188
 
    _commit_builder_class = PackRootCommitBuilder
2189
 
    rich_root_data = True
2190
 
    supports_tree_reference = True
2191
 
    _serializer = xml7.serializer_v7
2192
 
 
2193
 
    def _get_matching_bzrdir(self):
2194
 
        return bzrdir.format_registry.make_bzrdir(
2195
 
            'pack-0.92-subtree')
2196
 
 
2197
 
    def _ignore_setting_bzrdir(self, format):
2198
 
        pass
2199
 
 
2200
 
    _matchingbzrdir = property(_get_matching_bzrdir, _ignore_setting_bzrdir)
2201
 
 
2202
 
    def check_conversion_target(self, target_format):
2203
 
        if not target_format.rich_root_data:
2204
 
            raise errors.BadConversionTarget(
2205
 
                'Does not support rich root data.', target_format)
2206
 
        if not getattr(target_format, 'supports_tree_reference', False):
2207
 
            raise errors.BadConversionTarget(
2208
 
                'Does not support nested trees', target_format)
2209
 
            
2210
 
    def get_format_string(self):
2211
 
        """See RepositoryFormat.get_format_string()."""
2212
 
        return "Bazaar pack repository format 1 with subtree support (needs bzr 0.92)\n"
2213
 
 
2214
 
    def get_format_description(self):
2215
 
        """See RepositoryFormat.get_format_description()."""
2216
 
        return "Packs containing knits with subtree support\n"
2217
 
 
2218
 
 
2219
 
class RepositoryFormatKnitPack4(RepositoryFormatPack):
2220
 
    """A rich-root, no subtrees parameterized Pack repository.
2221
 
 
2222
 
    This repository format uses the xml6 serializer to get:
2223
 
     - support for recording full info about the tree root
2224
 
 
2225
 
    This format was introduced in 1.0.
2226
 
    """
2227
 
 
2228
 
    repository_class = KnitPackRepository
2229
 
    _commit_builder_class = PackRootCommitBuilder
2230
 
    rich_root_data = True
2231
 
    supports_tree_reference = False
2232
 
    _serializer = xml6.serializer_v6
2233
 
 
2234
 
    def _get_matching_bzrdir(self):
2235
 
        return bzrdir.format_registry.make_bzrdir(
2236
 
            'rich-root-pack')
2237
 
 
2238
 
    def _ignore_setting_bzrdir(self, format):
2239
 
        pass
2240
 
 
2241
 
    _matchingbzrdir = property(_get_matching_bzrdir, _ignore_setting_bzrdir)
2242
 
 
2243
 
    def check_conversion_target(self, target_format):
2244
 
        if not target_format.rich_root_data:
2245
 
            raise errors.BadConversionTarget(
2246
 
                'Does not support rich root data.', target_format)
2247
 
 
2248
 
    def get_format_string(self):
2249
 
        """See RepositoryFormat.get_format_string()."""
2250
 
        return ("Bazaar pack repository format 1 with rich root"
2251
 
                " (needs bzr 1.0)\n")
2252
 
 
2253
 
    def get_format_description(self):
2254
 
        """See RepositoryFormat.get_format_description()."""
2255
 
        return "Packs containing knits with rich root support\n"
2256
 
 
2257
 
 
2258
 
class RepositoryFormatPackDevelopment0(RepositoryFormatPack):
2259
 
    """A no-subtrees development repository.
2260
 
 
2261
 
    This format should be retained until the second release after bzr 1.0.
2262
 
 
2263
 
    No changes to the disk behaviour from pack-0.92.
2264
 
    """
2265
 
 
2266
 
    repository_class = KnitPackRepository
2267
 
    _commit_builder_class = PackCommitBuilder
2268
 
    _serializer = xml5.serializer_v5
2269
 
 
2270
 
    def _get_matching_bzrdir(self):
2271
 
        return bzrdir.format_registry.make_bzrdir('development0')
2272
 
 
2273
 
    def _ignore_setting_bzrdir(self, format):
2274
 
        pass
2275
 
 
2276
 
    _matchingbzrdir = property(_get_matching_bzrdir, _ignore_setting_bzrdir)
2277
 
 
2278
 
    def get_format_string(self):
2279
 
        """See RepositoryFormat.get_format_string()."""
2280
 
        return "Bazaar development format 0 (needs bzr.dev from before 1.3)\n"
2281
 
 
2282
 
    def get_format_description(self):
2283
 
        """See RepositoryFormat.get_format_description()."""
2284
 
        return ("Development repository format, currently the same as "
2285
 
            "pack-0.92\n")
2286
 
 
2287
 
    def check_conversion_target(self, target_format):
2288
 
        pass
2289
 
 
2290
 
 
2291
 
class RepositoryFormatPackDevelopment0Subtree(RepositoryFormatPack):
2292
 
    """A subtrees development repository.
2293
 
 
2294
 
    This format should be retained until the second release after bzr 1.0.
2295
 
 
2296
 
    No changes to the disk behaviour from pack-0.92-subtree.
2297
 
    """
2298
 
 
2299
 
    repository_class = KnitPackRepository
2300
 
    _commit_builder_class = PackRootCommitBuilder
2301
 
    rich_root_data = True
2302
 
    supports_tree_reference = True
2303
 
    _serializer = xml7.serializer_v7
2304
 
 
2305
 
    def _get_matching_bzrdir(self):
2306
 
        return bzrdir.format_registry.make_bzrdir(
2307
 
            'development0-subtree')
2308
 
 
2309
 
    def _ignore_setting_bzrdir(self, format):
2310
 
        pass
2311
 
 
2312
 
    _matchingbzrdir = property(_get_matching_bzrdir, _ignore_setting_bzrdir)
2313
 
 
2314
 
    def check_conversion_target(self, target_format):
2315
 
        if not target_format.rich_root_data:
2316
 
            raise errors.BadConversionTarget(
2317
 
                'Does not support rich root data.', target_format)
2318
 
        if not getattr(target_format, 'supports_tree_reference', False):
2319
 
            raise errors.BadConversionTarget(
2320
 
                'Does not support nested trees', target_format)
2321
 
            
2322
 
    def get_format_string(self):
2323
 
        """See RepositoryFormat.get_format_string()."""
2324
 
        return ("Bazaar development format 0 with subtree support "
2325
 
            "(needs bzr.dev from before 1.3)\n")
2326
 
 
2327
 
    def get_format_description(self):
2328
 
        """See RepositoryFormat.get_format_description()."""
2329
 
        return ("Development repository format, currently the same as "
2330
 
            "pack-0.92-subtree\n")
 
1939
class RetryPackOperations(errors.RetryWithNewPacks):
 
1940
    """Raised when we are packing and we find a missing file.
 
1941
 
 
1942
    Meant as a signaling exception, to tell the RepositoryPackCollection.pack
 
1943
    code it should try again.
 
1944
    """
 
1945
 
 
1946
    internal_error = True
 
1947
 
 
1948
    _fmt = ("Pack files have changed, reload and try pack again."
 
1949
            " context: %(context)s %(orig_error)s")
 
1950
 
 
1951
 
 
1952
class _DirectPackAccess(object):
 
1953
    """Access to data in one or more packs with less translation."""
 
1954
 
 
1955
    def __init__(self, index_to_packs, reload_func=None, flush_func=None):
 
1956
        """Create a _DirectPackAccess object.
 
1957
 
 
1958
        :param index_to_packs: A dict mapping index objects to the transport
 
1959
            and file names for obtaining data.
 
1960
        :param reload_func: A function to call if we determine that the pack
 
1961
            files have moved and we need to reload our caches. See
 
1962
            bzrlib.repo_fmt.pack_repo.AggregateIndex for more details.
 
1963
        """
 
1964
        self._container_writer = None
 
1965
        self._write_index = None
 
1966
        self._indices = index_to_packs
 
1967
        self._reload_func = reload_func
 
1968
        self._flush_func = flush_func
 
1969
 
 
1970
    def add_raw_records(self, key_sizes, raw_data):
 
1971
        """Add raw knit bytes to a storage area.
 
1972
 
 
1973
        The data is spooled to the container writer in one bytes-record per
 
1974
        raw data item.
 
1975
 
 
1976
        :param sizes: An iterable of tuples containing the key and size of each
 
1977
            raw data segment.
 
1978
        :param raw_data: A bytestring containing the data.
 
1979
        :return: A list of memos to retrieve the record later. Each memo is an
 
1980
            opaque index memo. For _DirectPackAccess the memo is (index, pos,
 
1981
            length), where the index field is the write_index object supplied
 
1982
            to the PackAccess object.
 
1983
        """
 
1984
        if type(raw_data) is not str:
 
1985
            raise AssertionError(
 
1986
                'data must be plain bytes was %s' % type(raw_data))
 
1987
        result = []
 
1988
        offset = 0
 
1989
        for key, size in key_sizes:
 
1990
            p_offset, p_length = self._container_writer.add_bytes_record(
 
1991
                raw_data[offset:offset+size], [])
 
1992
            offset += size
 
1993
            result.append((self._write_index, p_offset, p_length))
 
1994
        return result
 
1995
 
 
1996
    def flush(self):
 
1997
        """Flush pending writes on this access object.
 
1998
 
 
1999
        This will flush any buffered writes to a NewPack.
 
2000
        """
 
2001
        if self._flush_func is not None:
 
2002
            self._flush_func()
 
2003
 
 
2004
    def get_raw_records(self, memos_for_retrieval):
 
2005
        """Get the raw bytes for a records.
 
2006
 
 
2007
        :param memos_for_retrieval: An iterable containing the (index, pos,
 
2008
            length) memo for retrieving the bytes. The Pack access method
 
2009
            looks up the pack to use for a given record in its index_to_pack
 
2010
            map.
 
2011
        :return: An iterator over the bytes of the records.
 
2012
        """
 
2013
        # first pass, group into same-index requests
 
2014
        request_lists = []
 
2015
        current_index = None
 
2016
        for (index, offset, length) in memos_for_retrieval:
 
2017
            if current_index == index:
 
2018
                current_list.append((offset, length))
 
2019
            else:
 
2020
                if current_index is not None:
 
2021
                    request_lists.append((current_index, current_list))
 
2022
                current_index = index
 
2023
                current_list = [(offset, length)]
 
2024
        # handle the last entry
 
2025
        if current_index is not None:
 
2026
            request_lists.append((current_index, current_list))
 
2027
        for index, offsets in request_lists:
 
2028
            try:
 
2029
                transport, path = self._indices[index]
 
2030
            except KeyError:
 
2031
                # A KeyError here indicates that someone has triggered an index
 
2032
                # reload, and this index has gone missing, we need to start
 
2033
                # over.
 
2034
                if self._reload_func is None:
 
2035
                    # If we don't have a _reload_func there is nothing that can
 
2036
                    # be done
 
2037
                    raise
 
2038
                raise errors.RetryWithNewPacks(index,
 
2039
                                               reload_occurred=True,
 
2040
                                               exc_info=sys.exc_info())
 
2041
            try:
 
2042
                reader = pack.make_readv_reader(transport, path, offsets)
 
2043
                for names, read_func in reader.iter_records():
 
2044
                    yield read_func(None)
 
2045
            except errors.NoSuchFile:
 
2046
                # A NoSuchFile error indicates that a pack file has gone
 
2047
                # missing on disk, we need to trigger a reload, and start over.
 
2048
                if self._reload_func is None:
 
2049
                    raise
 
2050
                raise errors.RetryWithNewPacks(transport.abspath(path),
 
2051
                                               reload_occurred=False,
 
2052
                                               exc_info=sys.exc_info())
 
2053
 
 
2054
    def set_writer(self, writer, index, transport_packname):
 
2055
        """Set a writer to use for adding data."""
 
2056
        if index is not None:
 
2057
            self._indices[index] = transport_packname
 
2058
        self._container_writer = writer
 
2059
        self._write_index = index
 
2060
 
 
2061
    def reload_or_raise(self, retry_exc):
 
2062
        """Try calling the reload function, or re-raise the original exception.
 
2063
 
 
2064
        This should be called after _DirectPackAccess raises a
 
2065
        RetryWithNewPacks exception. This function will handle the common logic
 
2066
        of determining when the error is fatal versus being temporary.
 
2067
        It will also make sure that the original exception is raised, rather
 
2068
        than the RetryWithNewPacks exception.
 
2069
 
 
2070
        If this function returns, then the calling function should retry
 
2071
        whatever operation was being performed. Otherwise an exception will
 
2072
        be raised.
 
2073
 
 
2074
        :param retry_exc: A RetryWithNewPacks exception.
 
2075
        """
 
2076
        is_error = False
 
2077
        if self._reload_func is None:
 
2078
            is_error = True
 
2079
        elif not self._reload_func():
 
2080
            # The reload claimed that nothing changed
 
2081
            if not retry_exc.reload_occurred:
 
2082
                # If there wasn't an earlier reload, then we really were
 
2083
                # expecting to find changes. We didn't find them, so this is a
 
2084
                # hard error
 
2085
                is_error = True
 
2086
        if is_error:
 
2087
            exc_class, exc_value, exc_traceback = retry_exc.exc_info
 
2088
            raise exc_class, exc_value, exc_traceback
 
2089
 
2331
2090
 
2332
2091