/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/repofmt/pack_repo.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2008-01-11 07:06:36 UTC
  • mfrom: (3172.2.1 remote-fetch-formats)
  • Revision ID: pqm@pqm.ubuntu.com-20080111070636-jhozu5eo7wvh1k0o
(andrew) Enable use of smart revision streaming between repos with
        compatible models.

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2007-2010 Canonical Ltd
 
1
# Copyright (C) 2005, 2006, 2007 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., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
16
 
 
17
 
import re
18
 
import sys
 
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
19
16
 
20
17
from bzrlib.lazy_import import lazy_import
21
18
lazy_import(globals(), """
22
19
from itertools import izip
 
20
import math
 
21
import md5
23
22
import time
24
23
 
25
24
from bzrlib import (
26
 
    chk_map,
27
 
    cleanup,
28
 
    debug,
29
 
    graph,
30
 
    osutils,
31
 
    pack,
32
 
    transactions,
33
 
    ui,
34
 
    xml5,
35
 
    xml6,
36
 
    xml7,
37
 
    )
 
25
        debug,
 
26
        graph,
 
27
        pack,
 
28
        ui,
 
29
        )
38
30
from bzrlib.index import (
 
31
    GraphIndex,
 
32
    GraphIndexBuilder,
 
33
    InMemoryGraphIndex,
39
34
    CombinedGraphIndex,
40
35
    GraphIndexPrefixAdapter,
41
36
    )
42
 
from bzrlib.knit import (
43
 
    KnitPlainFactory,
44
 
    KnitVersionedFiles,
45
 
    _KnitGraphIndex,
46
 
    _DirectPackAccess,
47
 
    )
 
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
48
41
from bzrlib import tsort
49
42
""")
50
43
from bzrlib import (
51
44
    bzrdir,
 
45
    deprecated_graph,
52
46
    errors,
 
47
    knit,
53
48
    lockable_files,
54
49
    lockdir,
55
 
    revision as _mod_revision,
 
50
    osutils,
 
51
    symbol_versioning,
 
52
    transactions,
 
53
    xml5,
 
54
    xml6,
 
55
    xml7,
56
56
    )
57
57
 
58
 
from bzrlib.decorators import needs_write_lock, only_raises
59
 
from bzrlib.btree_index import (
60
 
    BTreeGraphIndex,
61
 
    BTreeBuilder,
62
 
    )
63
 
from bzrlib.index import (
64
 
    GraphIndex,
65
 
    InMemoryGraphIndex,
66
 
    )
67
 
from bzrlib.lock import LogicalLockResult
 
58
from bzrlib.decorators import needs_read_lock, needs_write_lock
68
59
from bzrlib.repofmt.knitrepo import KnitRepository
69
60
from bzrlib.repository import (
70
61
    CommitBuilder,
 
62
    MetaDirRepository,
71
63
    MetaDirRepositoryFormat,
72
 
    RepositoryFormat,
73
 
    RepositoryWriteLockResult,
74
64
    RootCommitBuilder,
75
 
    StreamSource,
76
 
    )
77
 
from bzrlib.trace import (
78
 
    mutter,
79
 
    note,
80
 
    warning,
81
 
    )
 
65
    )
 
66
import bzrlib.revision as _mod_revision
 
67
from bzrlib.store.revision.knit import KnitRevisionStore
 
68
from bzrlib.store.versioned import VersionedFileStore
 
69
from bzrlib.trace import mutter, note, warning
82
70
 
83
71
 
84
72
class PackCommitBuilder(CommitBuilder):
85
73
    """A subclass of CommitBuilder to add texts with pack semantics.
86
 
 
 
74
    
87
75
    Specifically this uses one knit object rather than one knit object per
88
76
    added text, reducing memory and object pressure.
89
77
    """
97
85
        self._file_graph = graph.Graph(
98
86
            repository._pack_collection.text_index.combined_index)
99
87
 
 
88
    def _add_text_to_weave(self, file_id, new_lines, parents, nostore_sha):
 
89
        return self.repository._pack_collection._add_text_to_weave(file_id,
 
90
            self._new_revision_id, new_lines, parents, nostore_sha,
 
91
            self.random_revid)
 
92
 
100
93
    def _heads(self, file_id, revision_ids):
101
94
        keys = [(file_id, revision_id) for revision_id in revision_ids]
102
95
        return set([key[1] for key in self._file_graph.heads(keys)])
104
97
 
105
98
class PackRootCommitBuilder(RootCommitBuilder):
106
99
    """A subclass of RootCommitBuilder to add texts with pack semantics.
107
 
 
 
100
    
108
101
    Specifically this uses one knit object rather than one knit object per
109
102
    added text, reducing memory and object pressure.
110
103
    """
118
111
        self._file_graph = graph.Graph(
119
112
            repository._pack_collection.text_index.combined_index)
120
113
 
 
114
    def _add_text_to_weave(self, file_id, new_lines, parents, nostore_sha):
 
115
        return self.repository._pack_collection._add_text_to_weave(file_id,
 
116
            self._new_revision_id, new_lines, parents, nostore_sha,
 
117
            self.random_revid)
 
118
 
121
119
    def _heads(self, file_id, revision_ids):
122
120
        keys = [(file_id, revision_id) for revision_id in revision_ids]
123
121
        return set([key[1] for key in self._file_graph.heads(keys)])
130
128
    ExistingPack and NewPack are used.
131
129
    """
132
130
 
133
 
    # A map of index 'type' to the file extension and position in the
134
 
    # index_sizes array.
135
 
    index_definitions = {
136
 
        'chk': ('.cix', 4),
137
 
        'revision': ('.rix', 0),
138
 
        'inventory': ('.iix', 1),
139
 
        'text': ('.tix', 2),
140
 
        'signature': ('.six', 3),
141
 
        }
142
 
 
143
131
    def __init__(self, revision_index, inventory_index, text_index,
144
 
        signature_index, chk_index=None):
 
132
        signature_index):
145
133
        """Create a pack instance.
146
134
 
147
135
        :param revision_index: A GraphIndex for determining what revisions are
152
140
        :param text_index: A GraphIndex for determining what file texts
153
141
            are present in the pack and accessing the locations of their
154
142
            texts/deltas (via (fileid, revisionid) tuples).
155
 
        :param signature_index: A GraphIndex for determining what signatures are
 
143
        :param revision_index: A GraphIndex for determining what signatures are
156
144
            present in the Pack and accessing the locations of their texts.
157
 
        :param chk_index: A GraphIndex for accessing content by CHK, if the
158
 
            pack has one.
159
145
        """
160
146
        self.revision_index = revision_index
161
147
        self.inventory_index = inventory_index
162
148
        self.text_index = text_index
163
149
        self.signature_index = signature_index
164
 
        self.chk_index = chk_index
165
150
 
166
151
    def access_tuple(self):
167
152
        """Return a tuple (transport, name) for the pack content."""
168
153
        return self.pack_transport, self.file_name()
169
154
 
170
 
    def _check_references(self):
171
 
        """Make sure our external references are present.
172
 
 
173
 
        Packs are allowed to have deltas whose base is not in the pack, but it
174
 
        must be present somewhere in this collection.  It is not allowed to
175
 
        have deltas based on a fallback repository.
176
 
        (See <https://bugs.launchpad.net/bzr/+bug/288751>)
177
 
        """
178
 
        missing_items = {}
179
 
        for (index_name, external_refs, index) in [
180
 
            ('texts',
181
 
                self._get_external_refs(self.text_index),
182
 
                self._pack_collection.text_index.combined_index),
183
 
            ('inventories',
184
 
                self._get_external_refs(self.inventory_index),
185
 
                self._pack_collection.inventory_index.combined_index),
186
 
            ]:
187
 
            missing = external_refs.difference(
188
 
                k for (idx, k, v, r) in
189
 
                index.iter_entries(external_refs))
190
 
            if missing:
191
 
                missing_items[index_name] = sorted(list(missing))
192
 
        if missing_items:
193
 
            from pprint import pformat
194
 
            raise errors.BzrCheckError(
195
 
                "Newly created pack file %r has delta references to "
196
 
                "items not in its repository:\n%s"
197
 
                % (self, pformat(missing_items)))
198
 
 
199
155
    def file_name(self):
200
156
        """Get the file name for the pack on disk."""
201
157
        return self.name + '.pack'
203
159
    def get_revision_count(self):
204
160
        return self.revision_index.key_count()
205
161
 
206
 
    def index_name(self, index_type, name):
207
 
        """Get the disk name of an index type for pack name 'name'."""
208
 
        return name + Pack.index_definitions[index_type][0]
209
 
 
210
 
    def index_offset(self, index_type):
211
 
        """Get the position in a index_size array for a given index type."""
212
 
        return Pack.index_definitions[index_type][1]
213
 
 
214
162
    def inventory_index_name(self, name):
215
163
        """The inv index is the name + .iix."""
216
164
        return self.index_name('inventory', name)
227
175
        """The text index is the name + .tix."""
228
176
        return self.index_name('text', name)
229
177
 
230
 
    def _replace_index_with_readonly(self, index_type):
231
 
        unlimited_cache = False
232
 
        if index_type == 'chk':
233
 
            unlimited_cache = True
234
 
        setattr(self, index_type + '_index',
235
 
            self.index_class(self.index_transport,
236
 
                self.index_name(index_type, self.name),
237
 
                self.index_sizes[self.index_offset(index_type)],
238
 
                unlimited_cache=unlimited_cache))
 
178
    def _external_compression_parents_of_texts(self):
 
179
        keys = set()
 
180
        refs = set()
 
181
        for node in self.text_index.iter_all_entries():
 
182
            keys.add(node[1])
 
183
            refs.update(node[3][1])
 
184
        return refs - keys
239
185
 
240
186
 
241
187
class ExistingPack(Pack):
242
188
    """An in memory proxy for an existing .pack and its disk indices."""
243
189
 
244
190
    def __init__(self, pack_transport, name, revision_index, inventory_index,
245
 
        text_index, signature_index, chk_index=None):
 
191
        text_index, signature_index):
246
192
        """Create an ExistingPack object.
247
193
 
248
194
        :param pack_transport: The transport where the pack file resides.
249
195
        :param name: The name of the pack on disk in the pack_transport.
250
196
        """
251
197
        Pack.__init__(self, revision_index, inventory_index, text_index,
252
 
            signature_index, chk_index)
 
198
            signature_index)
253
199
        self.name = name
254
200
        self.pack_transport = pack_transport
255
 
        if None in (revision_index, inventory_index, text_index,
256
 
                signature_index, name, pack_transport):
257
 
            raise AssertionError()
 
201
        assert None not in (revision_index, inventory_index, text_index,
 
202
            signature_index, name, pack_transport)
258
203
 
259
204
    def __eq__(self, other):
260
205
        return self.__dict__ == other.__dict__
263
208
        return not self.__eq__(other)
264
209
 
265
210
    def __repr__(self):
266
 
        return "<%s.%s object at 0x%x, %s, %s" % (
267
 
            self.__class__.__module__, self.__class__.__name__, id(self),
268
 
            self.pack_transport, self.name)
269
 
 
270
 
 
271
 
class ResumedPack(ExistingPack):
272
 
 
273
 
    def __init__(self, name, revision_index, inventory_index, text_index,
274
 
        signature_index, upload_transport, pack_transport, index_transport,
275
 
        pack_collection, chk_index=None):
276
 
        """Create a ResumedPack object."""
277
 
        ExistingPack.__init__(self, pack_transport, name, revision_index,
278
 
            inventory_index, text_index, signature_index,
279
 
            chk_index=chk_index)
280
 
        self.upload_transport = upload_transport
281
 
        self.index_transport = index_transport
282
 
        self.index_sizes = [None, None, None, None]
283
 
        indices = [
284
 
            ('revision', revision_index),
285
 
            ('inventory', inventory_index),
286
 
            ('text', text_index),
287
 
            ('signature', signature_index),
288
 
            ]
289
 
        if chk_index is not None:
290
 
            indices.append(('chk', chk_index))
291
 
            self.index_sizes.append(None)
292
 
        for index_type, index in indices:
293
 
            offset = self.index_offset(index_type)
294
 
            self.index_sizes[offset] = index._size
295
 
        self.index_class = pack_collection._index_class
296
 
        self._pack_collection = pack_collection
297
 
        self._state = 'resumed'
298
 
        # XXX: perhaps check that the .pack file exists?
299
 
 
300
 
    def access_tuple(self):
301
 
        if self._state == 'finished':
302
 
            return Pack.access_tuple(self)
303
 
        elif self._state == 'resumed':
304
 
            return self.upload_transport, self.file_name()
305
 
        else:
306
 
            raise AssertionError(self._state)
307
 
 
308
 
    def abort(self):
309
 
        self.upload_transport.delete(self.file_name())
310
 
        indices = [self.revision_index, self.inventory_index, self.text_index,
311
 
            self.signature_index]
312
 
        if self.chk_index is not None:
313
 
            indices.append(self.chk_index)
314
 
        for index in indices:
315
 
            index._transport.delete(index._name)
316
 
 
317
 
    def finish(self):
318
 
        self._check_references()
319
 
        index_types = ['revision', 'inventory', 'text', 'signature']
320
 
        if self.chk_index is not None:
321
 
            index_types.append('chk')
322
 
        for index_type in index_types:
323
 
            old_name = self.index_name(index_type, self.name)
324
 
            new_name = '../indices/' + old_name
325
 
            self.upload_transport.rename(old_name, new_name)
326
 
            self._replace_index_with_readonly(index_type)
327
 
        new_name = '../packs/' + self.file_name()
328
 
        self.upload_transport.rename(self.file_name(), new_name)
329
 
        self._state = 'finished'
330
 
 
331
 
    def _get_external_refs(self, index):
332
 
        """Return compression parents for this index that are not present.
333
 
 
334
 
        This returns any compression parents that are referenced by this index,
335
 
        which are not contained *in* this index. They may be present elsewhere.
336
 
        """
337
 
        return index.external_references(1)
 
211
        return "<bzrlib.repofmt.pack_repo.Pack object at 0x%x, %s, %s" % (
 
212
            id(self), self.transport, self.name)
338
213
 
339
214
 
340
215
class NewPack(Pack):
341
216
    """An in memory proxy for a pack which is being created."""
342
217
 
343
 
    def __init__(self, pack_collection, upload_suffix='', file_mode=None):
 
218
    # A map of index 'type' to the file extension and position in the
 
219
    # index_sizes array.
 
220
    index_definitions = {
 
221
        'revision': ('.rix', 0),
 
222
        'inventory': ('.iix', 1),
 
223
        'text': ('.tix', 2),
 
224
        'signature': ('.six', 3),
 
225
        }
 
226
 
 
227
    def __init__(self, upload_transport, index_transport, pack_transport,
 
228
        upload_suffix='', file_mode=None):
344
229
        """Create a NewPack instance.
345
230
 
346
 
        :param pack_collection: A PackCollection into which this is being inserted.
 
231
        :param upload_transport: A writable transport for the pack to be
 
232
            incrementally uploaded to.
 
233
        :param index_transport: A writable transport for the pack's indices to
 
234
            be written to when the pack is finished.
 
235
        :param pack_transport: A writable transport for the pack to be renamed
 
236
            to when the upload is complete. This *must* be the same as
 
237
            upload_transport.clone('../packs').
347
238
        :param upload_suffix: An optional suffix to be given to any temporary
348
239
            files created during the pack creation. e.g '.autopack'
349
 
        :param file_mode: Unix permissions for newly created file.
 
240
        :param file_mode: An optional file mode to create the new files with.
350
241
        """
351
242
        # The relative locations of the packs are constrained, but all are
352
243
        # passed in because the caller has them, so as to avoid object churn.
353
 
        index_builder_class = pack_collection._index_builder_class
354
 
        if pack_collection.chk_index is not None:
355
 
            chk_index = index_builder_class(reference_lists=0)
356
 
        else:
357
 
            chk_index = None
358
244
        Pack.__init__(self,
359
245
            # Revisions: parents list, no text compression.
360
 
            index_builder_class(reference_lists=1),
 
246
            InMemoryGraphIndex(reference_lists=1),
361
247
            # Inventory: We want to map compression only, but currently the
362
248
            # knit code hasn't been updated enough to understand that, so we
363
249
            # have a regular 2-list index giving parents and compression
364
250
            # source.
365
 
            index_builder_class(reference_lists=2),
 
251
            InMemoryGraphIndex(reference_lists=2),
366
252
            # Texts: compression and per file graph, for all fileids - so two
367
253
            # reference lists and two elements in the key tuple.
368
 
            index_builder_class(reference_lists=2, key_elements=2),
 
254
            InMemoryGraphIndex(reference_lists=2, key_elements=2),
369
255
            # Signatures: Just blobs to store, no compression, no parents
370
256
            # listing.
371
 
            index_builder_class(reference_lists=0),
372
 
            # CHK based storage - just blobs, no compression or parents.
373
 
            chk_index=chk_index
 
257
            InMemoryGraphIndex(reference_lists=0),
374
258
            )
375
 
        self._pack_collection = pack_collection
376
 
        # When we make readonly indices, we need this.
377
 
        self.index_class = pack_collection._index_class
378
259
        # where should the new pack be opened
379
 
        self.upload_transport = pack_collection._upload_transport
 
260
        self.upload_transport = upload_transport
380
261
        # where are indices written out to
381
 
        self.index_transport = pack_collection._index_transport
 
262
        self.index_transport = index_transport
382
263
        # where is the pack renamed to when it is finished?
383
 
        self.pack_transport = pack_collection._pack_transport
 
264
        self.pack_transport = pack_transport
384
265
        # What file mode to upload the pack and indices with.
385
266
        self._file_mode = file_mode
386
267
        # tracks the content written to the .pack file.
387
 
        self._hash = osutils.md5()
388
 
        # a tuple with the length in bytes of the indices, once the pack
389
 
        # is finalised. (rev, inv, text, sigs, chk_if_in_use)
 
268
        self._hash = md5.new()
 
269
        # a four-tuple with the length in bytes of the indices, once the pack
 
270
        # is finalised. (rev, inv, text, sigs)
390
271
        self.index_sizes = None
391
272
        # How much data to cache when writing packs. Note that this is not
392
273
        # synchronised with reads, because it's not in the transport layer, so
394
275
        # under creation.
395
276
        self._cache_limit = 0
396
277
        # the temporary pack file name.
397
 
        self.random_name = osutils.rand_chars(20) + upload_suffix
 
278
        self.random_name = rand_chars(20) + upload_suffix
398
279
        # when was this pack started ?
399
280
        self.start_time = time.time()
400
281
        # open an output stream for the data added to the pack.
404
285
            mutter('%s: create_pack: pack stream open: %s%s t+%6.3fs',
405
286
                time.ctime(), self.upload_transport.base, self.random_name,
406
287
                time.time() - self.start_time)
407
 
        # A list of byte sequences to be written to the new pack, and the
408
 
        # aggregate size of them.  Stored as a list rather than separate
 
288
        # A list of byte sequences to be written to the new pack, and the 
 
289
        # aggregate size of them.  Stored as a list rather than separate 
409
290
        # variables so that the _write_data closure below can update them.
410
291
        self._buffer = [[], 0]
411
 
        # create a callable for adding data
 
292
        # create a callable for adding data 
412
293
        #
413
294
        # robertc says- this is a closure rather than a method on the object
414
295
        # so that the variables are locals, and faster than accessing object
430
311
        self._writer.begin()
431
312
        # what state is the pack in? (open, finished, aborted)
432
313
        self._state = 'open'
433
 
        # no name until we finish writing the content
434
 
        self.name = None
435
314
 
436
315
    def abort(self):
437
316
        """Cancel creating this pack."""
443
322
 
444
323
    def access_tuple(self):
445
324
        """Return a tuple (transport, name) for the pack content."""
 
325
        assert self._state in ('open', 'finished')
446
326
        if self._state == 'finished':
447
327
            return Pack.access_tuple(self)
448
 
        elif self._state == 'open':
 
328
        else:
449
329
            return self.upload_transport, self.random_name
450
 
        else:
451
 
            raise AssertionError(self._state)
452
330
 
453
331
    def data_inserted(self):
454
332
        """True if data has been added to this pack."""
455
333
        return bool(self.get_revision_count() or
456
334
            self.inventory_index.key_count() or
457
335
            self.text_index.key_count() or
458
 
            self.signature_index.key_count() or
459
 
            (self.chk_index is not None and self.chk_index.key_count()))
460
 
 
461
 
    def finish_content(self):
462
 
        if self.name is not None:
463
 
            return
464
 
        self._writer.end()
465
 
        if self._buffer[1]:
466
 
            self._write_data('', flush=True)
467
 
        self.name = self._hash.hexdigest()
468
 
 
469
 
    def finish(self, suspend=False):
 
336
            self.signature_index.key_count())
 
337
 
 
338
    def finish(self):
470
339
        """Finish the new pack.
471
340
 
472
341
        This:
477
346
         - stores the index size tuple for the pack in the index_sizes
478
347
           attribute.
479
348
        """
480
 
        self.finish_content()
481
 
        if not suspend:
482
 
            self._check_references()
 
349
        self._writer.end()
 
350
        if self._buffer[1]:
 
351
            self._write_data('', flush=True)
 
352
        self.name = self._hash.hexdigest()
483
353
        # write indices
484
354
        # XXX: It'd be better to write them all to temporary names, then
485
355
        # rename them all into place, so that the window when only some are
486
356
        # visible is smaller.  On the other hand none will be seen until
487
357
        # they're in the names list.
488
358
        self.index_sizes = [None, None, None, None]
489
 
        self._write_index('revision', self.revision_index, 'revision', suspend)
490
 
        self._write_index('inventory', self.inventory_index, 'inventory',
491
 
            suspend)
492
 
        self._write_index('text', self.text_index, 'file texts', suspend)
 
359
        self._write_index('revision', self.revision_index, 'revision')
 
360
        self._write_index('inventory', self.inventory_index, 'inventory')
 
361
        self._write_index('text', self.text_index, 'file texts')
493
362
        self._write_index('signature', self.signature_index,
494
 
            'revision signatures', suspend)
495
 
        if self.chk_index is not None:
496
 
            self.index_sizes.append(None)
497
 
            self._write_index('chk', self.chk_index,
498
 
                'content hash bytes', suspend)
 
363
            'revision signatures')
499
364
        self.write_stream.close()
500
365
        # Note that this will clobber an existing pack with the same name,
501
366
        # without checking for hash collisions. While this is undesirable this
508
373
        #  - try for HASH.pack
509
374
        #  - try for temporary-name
510
375
        #  - refresh the pack-list to see if the pack is now absent
511
 
        new_name = self.name + '.pack'
512
 
        if not suspend:
513
 
            new_name = '../packs/' + new_name
514
 
        self.upload_transport.rename(self.random_name, new_name)
 
376
        self.upload_transport.rename(self.random_name,
 
377
                '../packs/' + self.name + '.pack')
515
378
        self._state = 'finished'
516
379
        if 'pack' in debug.debug_flags:
517
380
            # XXX: size might be interesting?
518
 
            mutter('%s: create_pack: pack finished: %s%s->%s t+%6.3fs',
 
381
            mutter('%s: create_pack: pack renamed into place: %s%s->%s%s t+%6.3fs',
519
382
                time.ctime(), self.upload_transport.base, self.random_name,
520
 
                new_name, time.time() - self.start_time)
 
383
                self.pack_transport, self.name,
 
384
                time.time() - self.start_time)
521
385
 
522
386
    def flush(self):
523
387
        """Flush any current data."""
527
391
            self._hash.update(bytes)
528
392
            self._buffer[:] = [[], 0]
529
393
 
530
 
    def _get_external_refs(self, index):
531
 
        return index._external_references()
 
394
    def index_name(self, index_type, name):
 
395
        """Get the disk name of an index type for pack name 'name'."""
 
396
        return name + NewPack.index_definitions[index_type][0]
 
397
 
 
398
    def index_offset(self, index_type):
 
399
        """Get the position in a index_size array for a given index type."""
 
400
        return NewPack.index_definitions[index_type][1]
 
401
 
 
402
    def _replace_index_with_readonly(self, index_type):
 
403
        setattr(self, index_type + '_index',
 
404
            GraphIndex(self.index_transport,
 
405
                self.index_name(index_type, self.name),
 
406
                self.index_sizes[self.index_offset(index_type)]))
532
407
 
533
408
    def set_write_cache_size(self, size):
534
409
        self._cache_limit = size
535
410
 
536
 
    def _write_index(self, index_type, index, label, suspend=False):
 
411
    def _write_index(self, index_type, index, label):
537
412
        """Write out an index.
538
413
 
539
414
        :param index_type: The type of index to write - e.g. 'revision'.
541
416
        :param label: What label to give the index e.g. 'revision'.
542
417
        """
543
418
        index_name = self.index_name(index_type, self.name)
544
 
        if suspend:
545
 
            transport = self.upload_transport
546
 
        else:
547
 
            transport = self.index_transport
548
 
        self.index_sizes[self.index_offset(index_type)] = transport.put_file(
549
 
            index_name, index.finish(), mode=self._file_mode)
 
419
        self.index_sizes[self.index_offset(index_type)] = \
 
420
            self.index_transport.put_file(index_name, index.finish(),
 
421
            mode=self._file_mode)
550
422
        if 'pack' in debug.debug_flags:
551
423
            # XXX: size might be interesting?
552
424
            mutter('%s: create_pack: wrote %s index: %s%s t+%6.3fs',
553
425
                time.ctime(), label, self.upload_transport.base,
554
426
                self.random_name, time.time() - self.start_time)
555
 
        # Replace the writable index on this object with a readonly,
 
427
        # Replace the writable index on this object with a readonly, 
556
428
        # presently unloaded index. We should alter
557
429
        # the index layer to make its finish() error if add_node is
558
430
        # subsequently used. RBC
567
439
    such as 'revision index'.
568
440
 
569
441
    A CombinedIndex provides an index on a single key space built up
570
 
    from several on-disk indices.  The AggregateIndex builds on this
 
442
    from several on-disk indices.  The AggregateIndex builds on this 
571
443
    to provide a knit access layer, and allows having up to one writable
572
444
    index within the collection.
573
445
    """
574
446
    # XXX: Probably 'can be written to' could/should be separated from 'acts
575
447
    # like a knit index' -- mbp 20071024
576
448
 
577
 
    def __init__(self, reload_func=None, flush_func=None):
578
 
        """Create an AggregateIndex.
579
 
 
580
 
        :param reload_func: A function to call if we find we are missing an
581
 
            index. Should have the form reload_func() => True if the list of
582
 
            active pack files has changed.
583
 
        """
584
 
        self._reload_func = reload_func
 
449
    def __init__(self):
 
450
        """Create an AggregateIndex."""
585
451
        self.index_to_pack = {}
586
 
        self.combined_index = CombinedGraphIndex([], reload_func=reload_func)
587
 
        self.data_access = _DirectPackAccess(self.index_to_pack,
588
 
                                             reload_func=reload_func,
589
 
                                             flush_func=flush_func)
 
452
        self.combined_index = CombinedGraphIndex([])
 
453
        self.knit_access = _PackAccess(self.index_to_pack)
 
454
 
 
455
    def replace_indices(self, index_to_pack, indices):
 
456
        """Replace the current mappings with fresh ones.
 
457
 
 
458
        This should probably not be used eventually, rather incremental add and
 
459
        removal of indices. It has been added during refactoring of existing
 
460
        code.
 
461
 
 
462
        :param index_to_pack: A mapping from index objects to
 
463
            (transport, name) tuples for the pack file data.
 
464
        :param indices: A list of indices.
 
465
        """
 
466
        # refresh the revision pack map dict without replacing the instance.
 
467
        self.index_to_pack.clear()
 
468
        self.index_to_pack.update(index_to_pack)
 
469
        # XXX: API break - clearly a 'replace' method would be good?
 
470
        self.combined_index._indices[:] = indices
 
471
        # the current add nodes callback for the current writable index if
 
472
        # there is one.
590
473
        self.add_callback = None
591
474
 
592
475
    def add_index(self, index, pack):
594
477
 
595
478
        Future searches on the aggregate index will seach this new index
596
479
        before all previously inserted indices.
597
 
 
 
480
        
598
481
        :param index: An Index for the pack.
599
482
        :param pack: A Pack instance.
600
483
        """
601
484
        # expose it to the index map
602
485
        self.index_to_pack[index] = pack.access_tuple()
603
486
        # put it at the front of the linear index list
604
 
        self.combined_index.insert_index(0, index, pack.name)
 
487
        self.combined_index.insert_index(0, index)
605
488
 
606
489
    def add_writable_index(self, index, pack):
607
490
        """Add an index which is able to have data added to it.
608
491
 
609
492
        There can be at most one writable index at any time.  Any
610
493
        modifications made to the knit are put into this index.
611
 
 
 
494
        
612
495
        :param index: An index from the pack parameter.
613
496
        :param pack: A Pack instance.
614
497
        """
615
 
        if self.add_callback is not None:
616
 
            raise AssertionError(
617
 
                "%s already has a writable index through %s" % \
618
 
                (self, self.add_callback))
 
498
        assert self.add_callback is None, \
 
499
            "%s already has a writable index through %s" % \
 
500
            (self, self.add_callback)
619
501
        # allow writing: queue writes to a new index
620
502
        self.add_index(index, pack)
621
503
        # Updates the index to packs mapping as a side effect,
622
 
        self.data_access.set_writer(pack._writer, index, pack.access_tuple())
 
504
        self.knit_access.set_writer(pack._writer, index, pack.access_tuple())
623
505
        self.add_callback = index.add_nodes
624
506
 
625
507
    def clear(self):
626
508
        """Reset all the aggregate data to nothing."""
627
 
        self.data_access.set_writer(None, None, (None, None))
 
509
        self.knit_access.set_writer(None, None, (None, None))
628
510
        self.index_to_pack.clear()
629
511
        del self.combined_index._indices[:]
630
 
        del self.combined_index._index_names[:]
631
512
        self.add_callback = None
632
513
 
633
 
    def remove_index(self, index):
 
514
    def remove_index(self, index, pack):
634
515
        """Remove index from the indices used to answer queries.
635
 
 
 
516
        
636
517
        :param index: An index from the pack parameter.
 
518
        :param pack: A Pack instance.
637
519
        """
638
520
        del self.index_to_pack[index]
639
 
        pos = self.combined_index._indices.index(index)
640
 
        del self.combined_index._indices[pos]
641
 
        del self.combined_index._index_names[pos]
 
521
        self.combined_index._indices.remove(index)
642
522
        if (self.add_callback is not None and
643
523
            getattr(index, 'add_nodes', None) == self.add_callback):
644
524
            self.add_callback = None
645
 
            self.data_access.set_writer(None, None, (None, None))
 
525
            self.knit_access.set_writer(None, None, (None, None))
646
526
 
647
527
 
648
528
class Packer(object):
649
529
    """Create a pack from packs."""
650
530
 
651
 
    def __init__(self, pack_collection, packs, suffix, revision_ids=None,
652
 
                 reload_func=None):
 
531
    def __init__(self, pack_collection, packs, suffix, revision_ids=None):
653
532
        """Create a Packer.
654
533
 
655
534
        :param pack_collection: A RepositoryPackCollection object where the
657
536
        :param packs: The packs to combine.
658
537
        :param suffix: The suffix to use on the temporary files for the pack.
659
538
        :param revision_ids: Revision ids to limit the pack to.
660
 
        :param reload_func: A function to call if a pack file/index goes
661
 
            missing. The side effect of calling this function should be to
662
 
            update self.packs. See also AggregateIndex
663
539
        """
664
540
        self.packs = packs
665
541
        self.suffix = suffix
667
543
        # The pack object we are creating.
668
544
        self.new_pack = None
669
545
        self._pack_collection = pack_collection
670
 
        self._reload_func = reload_func
671
546
        # The index layer keys for the revisions being copied. None for 'all
672
547
        # objects'.
673
548
        self._revision_keys = None
679
554
    def _extra_init(self):
680
555
        """A template hook to allow extending the constructor trivially."""
681
556
 
682
 
    def _pack_map_and_index_list(self, index_attribute):
683
 
        """Convert a list of packs to an index pack map and index list.
684
 
 
685
 
        :param index_attribute: The attribute that the desired index is found
686
 
            on.
687
 
        :return: A tuple (map, list) where map contains the dict from
688
 
            index:pack_tuple, and list contains the indices in the preferred
689
 
            access order.
690
 
        """
691
 
        indices = []
692
 
        pack_map = {}
693
 
        for pack_obj in self.packs:
694
 
            index = getattr(pack_obj, index_attribute)
695
 
            indices.append(index)
696
 
            pack_map[index] = pack_obj
697
 
        return pack_map, indices
698
 
 
699
 
    def _index_contents(self, indices, key_filter=None):
700
 
        """Get an iterable of the index contents from a pack_map.
701
 
 
702
 
        :param indices: The list of indices to query
703
 
        :param key_filter: An optional filter to limit the keys returned.
704
 
        """
705
 
        all_index = CombinedGraphIndex(indices)
706
 
        if key_filter is None:
707
 
            return all_index.iter_all_entries()
708
 
        else:
709
 
            return all_index.iter_entries(key_filter)
710
 
 
711
557
    def pack(self, pb=None):
712
558
        """Create a new pack by reading data from other packs.
713
559
 
714
560
        This does little more than a bulk copy of data. One key difference
715
561
        is that data with the same item key across multiple packs is elided
716
562
        from the output. The new pack is written into the current pack store
717
 
        along with its indices, and the name added to the pack names. The
 
563
        along with its indices, and the name added to the pack names. The 
718
564
        source packs are not altered and are not required to be in the current
719
565
        pack collection.
720
566
 
727
573
        # XXX: - duplicate code warning with start_write_group; fix before
728
574
        #      considering 'done'.
729
575
        if self._pack_collection._new_pack is not None:
730
 
            raise errors.BzrError('call to %s.pack() while another pack is'
731
 
                                  ' being written.'
732
 
                                  % (self.__class__.__name__,))
 
576
            raise errors.BzrError('call to create_pack_from_packs while '
 
577
                'another pack is being written.')
733
578
        if self.revision_ids is not None:
734
579
            if len(self.revision_ids) == 0:
735
580
                # silly fetch request.
736
581
                return None
737
582
            else:
738
583
                self.revision_ids = frozenset(self.revision_ids)
739
 
                self.revision_keys = frozenset((revid,) for revid in
740
 
                    self.revision_ids)
741
584
        if pb is None:
742
585
            self.pb = ui.ui_factory.nested_progress_bar()
743
586
        else:
750
593
 
751
594
    def open_pack(self):
752
595
        """Open a pack for the pack we are creating."""
753
 
        new_pack = self._pack_collection.pack_factory(self._pack_collection,
754
 
                upload_suffix=self.suffix,
755
 
                file_mode=self._pack_collection.repo.bzrdir._get_file_mode())
756
 
        # We know that we will process all nodes in order, and don't need to
757
 
        # query, so don't combine any indices spilled to disk until we are done
758
 
        new_pack.revision_index.set_optimize(combine_backing_indices=False)
759
 
        new_pack.inventory_index.set_optimize(combine_backing_indices=False)
760
 
        new_pack.text_index.set_optimize(combine_backing_indices=False)
761
 
        new_pack.signature_index.set_optimize(combine_backing_indices=False)
762
 
        return new_pack
763
 
 
764
 
    def _update_pack_order(self, entries, index_to_pack_map):
765
 
        """Determine how we want our packs to be ordered.
766
 
 
767
 
        This changes the sort order of the self.packs list so that packs unused
768
 
        by 'entries' will be at the end of the list, so that future requests
769
 
        can avoid probing them.  Used packs will be at the front of the
770
 
        self.packs list, in the order of their first use in 'entries'.
771
 
 
772
 
        :param entries: A list of (index, ...) tuples
773
 
        :param index_to_pack_map: A mapping from index objects to pack objects.
774
 
        """
775
 
        packs = []
776
 
        seen_indexes = set()
777
 
        for entry in entries:
778
 
            index = entry[0]
779
 
            if index not in seen_indexes:
780
 
                packs.append(index_to_pack_map[index])
781
 
                seen_indexes.add(index)
782
 
        if len(packs) == len(self.packs):
783
 
            if 'pack' in debug.debug_flags:
784
 
                mutter('Not changing pack list, all packs used.')
785
 
            return
786
 
        seen_packs = set(packs)
787
 
        for pack in self.packs:
788
 
            if pack not in seen_packs:
789
 
                packs.append(pack)
790
 
                seen_packs.add(pack)
791
 
        if 'pack' in debug.debug_flags:
792
 
            old_names = [p.access_tuple()[1] for p in self.packs]
793
 
            new_names = [p.access_tuple()[1] for p in packs]
794
 
            mutter('Reordering packs\nfrom: %s\n  to: %s',
795
 
                   old_names, new_names)
796
 
        self.packs = packs
 
596
        return NewPack(self._pack_collection._upload_transport,
 
597
            self._pack_collection._index_transport,
 
598
            self._pack_collection._pack_transport, upload_suffix=self.suffix,
 
599
            file_mode=self._pack_collection.repo.control_files._file_mode)
797
600
 
798
601
    def _copy_revision_texts(self):
799
602
        """Copy revision data to the new pack."""
803
606
        else:
804
607
            revision_keys = None
805
608
        # select revision keys
806
 
        revision_index_map, revision_indices = self._pack_map_and_index_list(
807
 
            'revision_index')
808
 
        revision_nodes = self._index_contents(revision_indices, revision_keys)
809
 
        revision_nodes = list(revision_nodes)
810
 
        self._update_pack_order(revision_nodes, revision_index_map)
 
609
        revision_index_map = self._pack_collection._packs_list_to_pack_map_and_index_list(
 
610
            self.packs, 'revision_index')[0]
 
611
        revision_nodes = self._pack_collection._index_contents(revision_index_map, revision_keys)
811
612
        # copy revision keys and adjust values
812
613
        self.pb.update("Copying revision texts", 1)
813
614
        total_items, readv_group_iter = self._revision_node_readv(revision_nodes)
833
634
        # querying for keys here could introduce a bug where an inventory item
834
635
        # is missed, so do not change it to query separately without cross
835
636
        # checking like the text key check below.
836
 
        inventory_index_map, inventory_indices = self._pack_map_and_index_list(
837
 
            'inventory_index')
838
 
        inv_nodes = self._index_contents(inventory_indices, inv_keys)
 
637
        inventory_index_map = self._pack_collection._packs_list_to_pack_map_and_index_list(
 
638
            self.packs, 'inventory_index')[0]
 
639
        inv_nodes = self._pack_collection._index_contents(inventory_index_map, inv_keys)
839
640
        # copy inventory keys and adjust values
840
641
        # XXX: Should be a helper function to allow different inv representation
841
642
        # at this point.
842
643
        self.pb.update("Copying inventory texts", 2)
843
644
        total_items, readv_group_iter = self._least_readv_node_readv(inv_nodes)
844
 
        # Only grab the output lines if we will be processing them
845
 
        output_lines = bool(self.revision_ids)
846
645
        inv_lines = self._copy_nodes_graph(inventory_index_map,
847
646
            self.new_pack._writer, self.new_pack.inventory_index,
848
 
            readv_group_iter, total_items, output_lines=output_lines)
 
647
            readv_group_iter, total_items, output_lines=True)
849
648
        if self.revision_ids:
850
649
            self._process_inventory_lines(inv_lines)
851
650
        else:
857
656
                time.ctime(), self._pack_collection._upload_transport.base,
858
657
                self.new_pack.random_name,
859
658
                self.new_pack.inventory_index.key_count(),
860
 
                time.time() - self.new_pack.start_time)
 
659
                time.time() - new_pack.start_time)
861
660
 
862
661
    def _copy_text_texts(self):
863
662
        # select text keys
875
674
            if missing_text_keys:
876
675
                # TODO: raise a specific error that can handle many missing
877
676
                # keys.
878
 
                mutter("missing keys during fetch: %r", missing_text_keys)
879
677
                a_missing_key = missing_text_keys.pop()
880
678
                raise errors.RevisionNotPresent(a_missing_key[1],
881
679
                    a_missing_key[0])
886
684
            self.new_pack.text_index, readv_group_iter, total_items))
887
685
        self._log_copied_texts()
888
686
 
 
687
    def _check_references(self):
 
688
        """Make sure our external refereneces are present."""
 
689
        external_refs = self.new_pack._external_compression_parents_of_texts()
 
690
        if external_refs:
 
691
            index = self._pack_collection.text_index.combined_index
 
692
            found_items = list(index.iter_entries(external_refs))
 
693
            if len(found_items) != len(external_refs):
 
694
                found_keys = set(k for idx, k, refs, value in found_items)
 
695
                missing_items = external_refs - found_keys
 
696
                missing_file_id, missing_revision_id = missing_items.pop()
 
697
                raise errors.RevisionNotPresent(missing_revision_id,
 
698
                                                missing_file_id)
 
699
 
889
700
    def _create_pack_from_packs(self):
890
701
        self.pb.update("Opening pack", 0, 5)
891
702
        self.new_pack = self.open_pack()
909
720
        self._copy_text_texts()
910
721
        # select signature keys
911
722
        signature_filter = self._revision_keys # same keyspace
912
 
        signature_index_map, signature_indices = self._pack_map_and_index_list(
913
 
            'signature_index')
914
 
        signature_nodes = self._index_contents(signature_indices,
 
723
        signature_index_map = self._pack_collection._packs_list_to_pack_map_and_index_list(
 
724
            self.packs, 'signature_index')[0]
 
725
        signature_nodes = self._pack_collection._index_contents(signature_index_map,
915
726
            signature_filter)
916
727
        # copy signature keys and adjust values
917
728
        self.pb.update("Copying signature texts", 4)
922
733
                time.ctime(), self._pack_collection._upload_transport.base, new_pack.random_name,
923
734
                new_pack.signature_index.key_count(),
924
735
                time.time() - new_pack.start_time)
925
 
        # copy chk contents
926
 
        # NB XXX: how to check CHK references are present? perhaps by yielding
927
 
        # the items? How should that interact with stacked repos?
928
 
        if new_pack.chk_index is not None:
929
 
            self._copy_chks()
930
 
            if 'pack' in debug.debug_flags:
931
 
                mutter('%s: create_pack: chk content copied: %s%s %d items t+%6.3fs',
932
 
                    time.ctime(), self._pack_collection._upload_transport.base,
933
 
                    new_pack.random_name,
934
 
                    new_pack.chk_index.key_count(),
935
 
                    time.time() - new_pack.start_time)
936
 
        new_pack._check_references()
 
736
        self._check_references()
937
737
        if not self._use_pack(new_pack):
938
738
            new_pack.abort()
939
739
            return None
942
742
        self._pack_collection.allocate(new_pack)
943
743
        return new_pack
944
744
 
945
 
    def _copy_chks(self, refs=None):
946
 
        # XXX: Todo, recursive follow-pointers facility when fetching some
947
 
        # revisions only.
948
 
        chk_index_map, chk_indices = self._pack_map_and_index_list(
949
 
            'chk_index')
950
 
        chk_nodes = self._index_contents(chk_indices, refs)
951
 
        new_refs = set()
952
 
        # TODO: This isn't strictly tasteful as we are accessing some private
953
 
        #       variables (_serializer). Perhaps a better way would be to have
954
 
        #       Repository._deserialise_chk_node()
955
 
        search_key_func = chk_map.search_key_registry.get(
956
 
            self._pack_collection.repo._serializer.search_key_name)
957
 
        def accumlate_refs(lines):
958
 
            # XXX: move to a generic location
959
 
            # Yay mismatch:
960
 
            bytes = ''.join(lines)
961
 
            node = chk_map._deserialise(bytes, ("unknown",), search_key_func)
962
 
            new_refs.update(node.refs())
963
 
        self._copy_nodes(chk_nodes, chk_index_map, self.new_pack._writer,
964
 
            self.new_pack.chk_index, output_lines=accumlate_refs)
965
 
        return new_refs
966
 
 
967
 
    def _copy_nodes(self, nodes, index_map, writer, write_index,
968
 
        output_lines=None):
969
 
        """Copy knit nodes between packs with no graph references.
970
 
 
971
 
        :param output_lines: Output full texts of copied items.
972
 
        """
 
745
    def _copy_nodes(self, nodes, index_map, writer, write_index):
 
746
        """Copy knit nodes between packs with no graph references."""
973
747
        pb = ui.ui_factory.nested_progress_bar()
974
748
        try:
975
749
            return self._do_copy_nodes(nodes, index_map, writer,
976
 
                write_index, pb, output_lines=output_lines)
 
750
                write_index, pb)
977
751
        finally:
978
752
            pb.finished()
979
753
 
980
 
    def _do_copy_nodes(self, nodes, index_map, writer, write_index, pb,
981
 
        output_lines=None):
 
754
    def _do_copy_nodes(self, nodes, index_map, writer, write_index, pb):
982
755
        # for record verification
983
 
        knit = KnitVersionedFiles(None, None)
 
756
        knit_data = _KnitData(None)
984
757
        # plan a readv on each source pack:
985
758
        # group by pack
986
759
        nodes = sorted(nodes)
987
760
        # how to map this into knit.py - or knit.py into this?
988
761
        # we don't want the typical knit logic, we want grouping by pack
989
 
        # at this point - perhaps a helper library for the following code
 
762
        # at this point - perhaps a helper library for the following code 
990
763
        # duplication points?
991
764
        request_groups = {}
992
765
        for index, key, value in nodes:
1005
778
            # linear scan up the pack
1006
779
            pack_readv_requests.sort()
1007
780
            # copy the data
1008
 
            pack_obj = index_map[index]
1009
 
            transport, path = pack_obj.access_tuple()
1010
 
            try:
1011
 
                reader = pack.make_readv_reader(transport, path,
1012
 
                    [offset[0:2] for offset in pack_readv_requests])
1013
 
            except errors.NoSuchFile:
1014
 
                if self._reload_func is not None:
1015
 
                    self._reload_func()
1016
 
                raise
 
781
            transport, path = index_map[index]
 
782
            reader = pack.make_readv_reader(transport, path,
 
783
                [offset[0:2] for offset in pack_readv_requests])
1017
784
            for (names, read_func), (_1, _2, (key, eol_flag)) in \
1018
785
                izip(reader.iter_records(), pack_readv_requests):
1019
786
                raw_data = read_func(None)
1020
787
                # check the header only
1021
 
                if output_lines is not None:
1022
 
                    output_lines(knit._parse_record(key[-1], raw_data)[0])
1023
 
                else:
1024
 
                    df, _ = knit._parse_record_header(key, raw_data)
1025
 
                    df.close()
 
788
                df, _ = knit_data._parse_record_header(key[-1], raw_data)
 
789
                df.close()
1026
790
                pos, size = writer.add_bytes_record(raw_data, names)
1027
791
                write_index.add_node(key, eol_flag + "%d %d" % (pos, size))
1028
792
                pb.update("Copied record", record_index)
1050
814
    def _do_copy_nodes_graph(self, index_map, writer, write_index,
1051
815
        output_lines, pb, readv_group_iter, total_items):
1052
816
        # for record verification
1053
 
        knit = KnitVersionedFiles(None, None)
 
817
        knit_data = _KnitData(None)
1054
818
        # for line extraction when requested (inventories only)
1055
819
        if output_lines:
1056
 
            factory = KnitPlainFactory()
 
820
            factory = knit.KnitPlainFactory()
1057
821
        record_index = 0
1058
822
        pb.update("Copied record", record_index, total_items)
1059
823
        for index, readv_vector, node_vector in readv_group_iter:
1060
824
            # copy the data
1061
 
            pack_obj = index_map[index]
1062
 
            transport, path = pack_obj.access_tuple()
1063
 
            try:
1064
 
                reader = pack.make_readv_reader(transport, path, readv_vector)
1065
 
            except errors.NoSuchFile:
1066
 
                if self._reload_func is not None:
1067
 
                    self._reload_func()
1068
 
                raise
 
825
            transport, path = index_map[index]
 
826
            reader = pack.make_readv_reader(transport, path, readv_vector)
1069
827
            for (names, read_func), (key, eol_flag, references) in \
1070
828
                izip(reader.iter_records(), node_vector):
1071
829
                raw_data = read_func(None)
 
830
                version_id = key[-1]
1072
831
                if output_lines:
1073
832
                    # read the entire thing
1074
 
                    content, _ = knit._parse_record(key[-1], raw_data)
 
833
                    content, _ = knit_data._parse_record(version_id, raw_data)
1075
834
                    if len(references[-1]) == 0:
1076
835
                        line_iterator = factory.get_fulltext_content(content)
1077
836
                    else:
1078
837
                        line_iterator = factory.get_linedelta_content(content)
1079
838
                    for line in line_iterator:
1080
 
                        yield line, key
 
839
                        yield line, version_id
1081
840
                else:
1082
841
                    # check the header only
1083
 
                    df, _ = knit._parse_record_header(key, raw_data)
 
842
                    df, _ = knit_data._parse_record_header(version_id, raw_data)
1084
843
                    df.close()
1085
844
                pos, size = writer.add_bytes_record(raw_data, names)
1086
845
                write_index.add_node(key, eol_flag + "%d %d" % (pos, size), references)
1088
847
                record_index += 1
1089
848
 
1090
849
    def _get_text_nodes(self):
1091
 
        text_index_map, text_indices = self._pack_map_and_index_list(
1092
 
            'text_index')
1093
 
        return text_index_map, self._index_contents(text_indices,
 
850
        text_index_map = self._pack_collection._packs_list_to_pack_map_and_index_list(
 
851
            self.packs, 'text_index')[0]
 
852
        return text_index_map, self._pack_collection._index_contents(text_index_map,
1094
853
            self._text_filter)
1095
854
 
1096
855
    def _least_readv_node_readv(self, nodes):
1097
856
        """Generate request groups for nodes using the least readv's.
1098
 
 
 
857
        
1099
858
        :param nodes: An iterable of graph index nodes.
1100
859
        :return: Total node count and an iterator of the data needed to perform
1101
860
            readvs to obtain the data for nodes. Each item yielded by the
1102
861
            iterator is a tuple with:
1103
862
            index, readv_vector, node_vector. readv_vector is a list ready to
1104
863
            hand to the transport readv method, and node_vector is a list of
1105
 
            (key, eol_flag, references) for the node retrieved by the
 
864
            (key, eol_flag, references) for the the node retrieved by the
1106
865
            matching readv_vector.
1107
866
        """
1108
867
        # group by pack so we do one readv per pack
1142
901
        """Use up the inv_lines generator and setup a text key filter."""
1143
902
        repo = self._pack_collection.repo
1144
903
        fileid_revisions = repo._find_file_ids_from_xml_inventory_lines(
1145
 
            inv_lines, self.revision_keys)
 
904
            inv_lines, self.revision_ids)
1146
905
        text_filter = []
1147
906
        for fileid, file_revids in fileid_revisions.iteritems():
1148
907
            text_filter.extend([(fileid, file_revid) for file_revid in file_revids])
1199
958
        # TODO: combine requests in the same index that are in ascending order.
1200
959
        return total, requests
1201
960
 
1202
 
    def open_pack(self):
1203
 
        """Open a pack for the pack we are creating."""
1204
 
        new_pack = super(OptimisingPacker, self).open_pack()
1205
 
        # Turn on the optimization flags for all the index builders.
1206
 
        new_pack.revision_index.set_optimize(for_size=True)
1207
 
        new_pack.inventory_index.set_optimize(for_size=True)
1208
 
        new_pack.text_index.set_optimize(for_size=True)
1209
 
        new_pack.signature_index.set_optimize(for_size=True)
1210
 
        return new_pack
1211
 
 
1212
961
 
1213
962
class ReconcilePacker(Packer):
1214
963
    """A packer which regenerates indices etc as it copies.
1215
 
 
 
964
    
1216
965
    This is used by ``bzr reconcile`` to cause parent text pointers to be
1217
966
    regenerated.
1218
967
    """
1241
990
        # 1) generate the ideal index
1242
991
        repo = self._pack_collection.repo
1243
992
        ancestors = dict([(key[0], tuple(ref[0] for ref in refs[0])) for
1244
 
            _1, key, _2, refs in
 
993
            _1, key, _2, refs in 
1245
994
            self.new_pack.revision_index.iter_all_entries()])
1246
995
        ideal_index = repo._generate_text_key_index(self._text_refs, ancestors)
1247
996
        # 2) generate a text_nodes list that contains all the deltas that can
1253
1002
        text_index_map, text_nodes = self._get_text_nodes()
1254
1003
        for node in text_nodes:
1255
1004
            # 0 - index
1256
 
            # 1 - key
 
1005
            # 1 - key 
1257
1006
            # 2 - value
1258
1007
            # 3 - refs
1259
1008
            try:
1299
1048
        # space (we only topo sort the revisions, which is smaller).
1300
1049
        topo_order = tsort.topo_sort(ancestors)
1301
1050
        rev_order = dict(zip(topo_order, range(len(topo_order))))
1302
 
        bad_texts.sort(key=lambda key:rev_order.get(key[0][1], 0))
 
1051
        bad_texts.sort(key=lambda key:rev_order[key[0][1]])
1303
1052
        transaction = repo.get_transaction()
1304
1053
        file_id_index = GraphIndexPrefixAdapter(
1305
1054
            self.new_pack.text_index,
1306
1055
            ('blank', ), 1,
1307
1056
            add_nodes_callback=self.new_pack.text_index.add_nodes)
1308
 
        data_access = _DirectPackAccess(
1309
 
                {self.new_pack.text_index:self.new_pack.access_tuple()})
1310
 
        data_access.set_writer(self.new_pack._writer, self.new_pack.text_index,
1311
 
            self.new_pack.access_tuple())
1312
 
        output_texts = KnitVersionedFiles(
1313
 
            _KnitGraphIndex(self.new_pack.text_index,
1314
 
                add_callback=self.new_pack.text_index.add_nodes,
1315
 
                deltas=True, parents=True, is_locked=repo.is_locked),
1316
 
            data_access=data_access, max_delta_chain=200)
 
1057
        knit_index = KnitGraphIndex(file_id_index,
 
1058
            add_callback=file_id_index.add_nodes,
 
1059
            deltas=True, parents=True)
 
1060
        output_knit = knit.KnitVersionedFile('reconcile-texts',
 
1061
            self._pack_collection.transport,
 
1062
            None,
 
1063
            index=knit_index,
 
1064
            access_method=_PackAccess(
 
1065
                {self.new_pack.text_index:self.new_pack.access_tuple()},
 
1066
                (self.new_pack._writer, self.new_pack.text_index)),
 
1067
            factory=knit.KnitPlainFactory())
1317
1068
        for key, parent_keys in bad_texts:
1318
1069
            # We refer to the new pack to delta data being output.
1319
1070
            # A possible improvement would be to catch errors on short reads
1326
1077
                    raise errors.BzrError('Mismatched key parent %r:%r' %
1327
1078
                        (key, parent_keys))
1328
1079
                parents.append(parent_key[1])
1329
 
            text_lines = osutils.split_lines(repo.texts.get_record_stream(
1330
 
                [key], 'unordered', True).next().get_bytes_as('fulltext'))
1331
 
            output_texts.add_lines(key, parent_keys, text_lines,
1332
 
                random_id=True, check_content=False)
 
1080
            source_weave = repo.weave_store.get_weave(key[0], transaction)
 
1081
            text_lines = source_weave.get_lines(key[1])
 
1082
            # adapt the 'knit' to the current file_id.
 
1083
            file_id_index = GraphIndexPrefixAdapter(
 
1084
                self.new_pack.text_index,
 
1085
                (key[0], ), 1,
 
1086
                add_nodes_callback=self.new_pack.text_index.add_nodes)
 
1087
            knit_index._graph_index = file_id_index
 
1088
            knit_index._add_callback = file_id_index.add_nodes
 
1089
            output_knit.add_lines_with_ghosts(
 
1090
                key[1], parents, text_lines, random_id=True, check_content=False)
1333
1091
        # 5) check that nothing inserted has a reference outside the keyspace.
1334
 
        missing_text_keys = self.new_pack.text_index._external_references()
 
1092
        missing_text_keys = self.new_pack._external_compression_parents_of_texts()
1335
1093
        if missing_text_keys:
1336
 
            raise errors.BzrCheckError('Reference to missing compression parents %r'
1337
 
                % (missing_text_keys,))
 
1094
            raise errors.BzrError('Reference to missing compression parents %r'
 
1095
                % (refs - keys,))
1338
1096
        self._log_copied_texts()
1339
1097
 
1340
1098
    def _use_pack(self, new_pack):
1353
1111
 
1354
1112
 
1355
1113
class RepositoryPackCollection(object):
1356
 
    """Management of packs within a repository.
1357
 
 
1358
 
    :ivar _names: map of {pack_name: (index_size,)}
1359
 
    """
1360
 
 
1361
 
    pack_factory = NewPack
1362
 
    resumed_pack_factory = ResumedPack
 
1114
    """Management of packs within a repository."""
1363
1115
 
1364
1116
    def __init__(self, repo, transport, index_transport, upload_transport,
1365
 
                 pack_transport, index_builder_class, index_class,
1366
 
                 use_chk_index):
 
1117
                 pack_transport):
1367
1118
        """Create a new RepositoryPackCollection.
1368
1119
 
1369
 
        :param transport: Addresses the repository base directory
 
1120
        :param transport: Addresses the repository base directory 
1370
1121
            (typically .bzr/repository/).
1371
1122
        :param index_transport: Addresses the directory containing indices.
1372
1123
        :param upload_transport: Addresses the directory into which packs are written
1373
1124
            while they're being created.
1374
1125
        :param pack_transport: Addresses the directory of existing complete packs.
1375
 
        :param index_builder_class: The index builder class to use.
1376
 
        :param index_class: The index class to use.
1377
 
        :param use_chk_index: Whether to setup and manage a CHK index.
1378
1126
        """
1379
 
        # XXX: This should call self.reset()
1380
1127
        self.repo = repo
1381
1128
        self.transport = transport
1382
1129
        self._index_transport = index_transport
1383
1130
        self._upload_transport = upload_transport
1384
1131
        self._pack_transport = pack_transport
1385
 
        self._index_builder_class = index_builder_class
1386
 
        self._index_class = index_class
1387
 
        self._suffix_offsets = {'.rix': 0, '.iix': 1, '.tix': 2, '.six': 3,
1388
 
            '.cix': 4}
 
1132
        self._suffix_offsets = {'.rix': 0, '.iix': 1, '.tix': 2, '.six': 3}
1389
1133
        self.packs = []
1390
1134
        # name:Pack mapping
1391
 
        self._names = None
1392
1135
        self._packs_by_name = {}
1393
1136
        # the previous pack-names content
1394
1137
        self._packs_at_load = None
1395
1138
        # when a pack is being created by this object, the state of that pack.
1396
1139
        self._new_pack = None
1397
1140
        # aggregated revision index data
1398
 
        flush = self._flush_new_pack
1399
 
        self.revision_index = AggregateIndex(self.reload_pack_names, flush)
1400
 
        self.inventory_index = AggregateIndex(self.reload_pack_names, flush)
1401
 
        self.text_index = AggregateIndex(self.reload_pack_names, flush)
1402
 
        self.signature_index = AggregateIndex(self.reload_pack_names, flush)
1403
 
        all_indices = [self.revision_index, self.inventory_index,
1404
 
                self.text_index, self.signature_index]
1405
 
        if use_chk_index:
1406
 
            self.chk_index = AggregateIndex(self.reload_pack_names, flush)
1407
 
            all_indices.append(self.chk_index)
1408
 
        else:
1409
 
            # used to determine if we're using a chk_index elsewhere.
1410
 
            self.chk_index = None
1411
 
        # Tell all the CombinedGraphIndex objects about each other, so they can
1412
 
        # share hints about which pack names to search first.
1413
 
        all_combined = [agg_idx.combined_index for agg_idx in all_indices]
1414
 
        for combined_idx in all_combined:
1415
 
            combined_idx.set_sibling_indices(
1416
 
                set(all_combined).difference([combined_idx]))
1417
 
        # resumed packs
1418
 
        self._resumed_packs = []
1419
 
 
1420
 
    def __repr__(self):
1421
 
        return '%s(%r)' % (self.__class__.__name__, self.repo)
 
1141
        self.revision_index = AggregateIndex()
 
1142
        self.inventory_index = AggregateIndex()
 
1143
        self.text_index = AggregateIndex()
 
1144
        self.signature_index = AggregateIndex()
1422
1145
 
1423
1146
    def add_pack_to_memory(self, pack):
1424
1147
        """Make a Pack object available to the repository to satisfy queries.
1425
 
 
 
1148
        
1426
1149
        :param pack: A Pack object.
1427
1150
        """
1428
 
        if pack.name in self._packs_by_name:
1429
 
            raise AssertionError(
1430
 
                'pack %s already in _packs_by_name' % (pack.name,))
 
1151
        assert pack.name not in self._packs_by_name
1431
1152
        self.packs.append(pack)
1432
1153
        self._packs_by_name[pack.name] = pack
1433
1154
        self.revision_index.add_index(pack.revision_index, pack)
1434
1155
        self.inventory_index.add_index(pack.inventory_index, pack)
1435
1156
        self.text_index.add_index(pack.text_index, pack)
1436
1157
        self.signature_index.add_index(pack.signature_index, pack)
1437
 
        if self.chk_index is not None:
1438
 
            self.chk_index.add_index(pack.chk_index, pack)
 
1158
        
 
1159
    def _add_text_to_weave(self, file_id, revision_id, new_lines, parents,
 
1160
        nostore_sha, random_revid):
 
1161
        file_id_index = GraphIndexPrefixAdapter(
 
1162
            self.text_index.combined_index,
 
1163
            (file_id, ), 1,
 
1164
            add_nodes_callback=self.text_index.add_callback)
 
1165
        self.repo._text_knit._index._graph_index = file_id_index
 
1166
        self.repo._text_knit._index._add_callback = file_id_index.add_nodes
 
1167
        return self.repo._text_knit.add_lines_with_ghosts(
 
1168
            revision_id, parents, new_lines, nostore_sha=nostore_sha,
 
1169
            random_id=random_revid, check_content=False)[0:2]
1439
1170
 
1440
1171
    def all_packs(self):
1441
1172
        """Return a list of all the Pack objects this repository has.
1451
1182
 
1452
1183
    def autopack(self):
1453
1184
        """Pack the pack collection incrementally.
1454
 
 
 
1185
        
1455
1186
        This will not attempt global reorganisation or recompression,
1456
1187
        rather it will just ensure that the total number of packs does
1457
1188
        not grow without bound. It uses the _max_pack_count method to
1463
1194
        in synchronisation with certain steps. Otherwise the names collection
1464
1195
        is not flushed.
1465
1196
 
1466
 
        :return: Something evaluating true if packing took place.
 
1197
        :return: True if packing took place.
1467
1198
        """
1468
 
        while True:
1469
 
            try:
1470
 
                return self._do_autopack()
1471
 
            except errors.RetryAutopack:
1472
 
                # If we get a RetryAutopack exception, we should abort the
1473
 
                # current action, and retry.
1474
 
                pass
1475
 
 
1476
 
    def _do_autopack(self):
1477
1199
        # XXX: Should not be needed when the management of indices is sane.
1478
1200
        total_revisions = self.revision_index.combined_index.key_count()
1479
1201
        total_packs = len(self._names)
1480
1202
        if self._max_pack_count(total_revisions) >= total_packs:
1481
 
            return None
 
1203
            return False
 
1204
        # XXX: the following may want to be a class, to pack with a given
 
1205
        # policy.
 
1206
        mutter('Auto-packing repository %s, which has %d pack files, '
 
1207
            'containing %d revisions into %d packs.', self, total_packs,
 
1208
            total_revisions, self._max_pack_count(total_revisions))
1482
1209
        # determine which packs need changing
1483
1210
        pack_distribution = self.pack_distribution(total_revisions)
1484
1211
        existing_packs = []
1492
1219
                # group their data with the relevant commit, and that may
1493
1220
                # involve rewriting ancient history - which autopack tries to
1494
1221
                # avoid. Alternatively we could not group the data but treat
1495
 
                # each of these as having a single revision, and thus add
 
1222
                # each of these as having a single revision, and thus add 
1496
1223
                # one revision for each to the total revision count, to get
1497
1224
                # a matching distribution.
1498
1225
                continue
1499
1226
            existing_packs.append((revision_count, pack))
1500
1227
        pack_operations = self.plan_autopack_combinations(
1501
1228
            existing_packs, pack_distribution)
1502
 
        num_new_packs = len(pack_operations)
1503
 
        num_old_packs = sum([len(po[1]) for po in pack_operations])
1504
 
        num_revs_affected = sum([po[0] for po in pack_operations])
1505
 
        mutter('Auto-packing repository %s, which has %d pack files, '
1506
 
            'containing %d revisions. Packing %d files into %d affecting %d'
1507
 
            ' revisions', self, total_packs, total_revisions, num_old_packs,
1508
 
            num_new_packs, num_revs_affected)
1509
 
        result = self._execute_pack_operations(pack_operations,
1510
 
                                      reload_func=self._restart_autopack)
1511
 
        mutter('Auto-packing repository %s completed', self)
1512
 
        return result
 
1229
        self._execute_pack_operations(pack_operations)
 
1230
        return True
1513
1231
 
1514
 
    def _execute_pack_operations(self, pack_operations, _packer_class=Packer,
1515
 
                                 reload_func=None):
 
1232
    def _execute_pack_operations(self, pack_operations, _packer_class=Packer):
1516
1233
        """Execute a series of pack operations.
1517
1234
 
1518
1235
        :param pack_operations: A list of [revision_count, packs_to_combine].
1519
1236
        :param _packer_class: The class of packer to use (default: Packer).
1520
 
        :return: The new pack names.
 
1237
        :return: None.
1521
1238
        """
1522
1239
        for revision_count, packs in pack_operations:
1523
1240
            # we may have no-ops from the setup logic
1524
1241
            if len(packs) == 0:
1525
1242
                continue
1526
 
            packer = _packer_class(self, packs, '.autopack',
1527
 
                                   reload_func=reload_func)
1528
 
            try:
1529
 
                packer.pack()
1530
 
            except errors.RetryWithNewPacks:
1531
 
                # An exception is propagating out of this context, make sure
1532
 
                # this packer has cleaned up. Packer() doesn't set its new_pack
1533
 
                # state into the RepositoryPackCollection object, so we only
1534
 
                # have access to it directly here.
1535
 
                if packer.new_pack is not None:
1536
 
                    packer.new_pack.abort()
1537
 
                raise
 
1243
            _packer_class(self, packs, '.autopack').pack()
1538
1244
            for pack in packs:
1539
1245
                self._remove_pack_from_memory(pack)
1540
1246
        # record the newly available packs and stop advertising the old
1541
1247
        # packs
1542
 
        to_be_obsoleted = []
1543
 
        for _, packs in pack_operations:
1544
 
            to_be_obsoleted.extend(packs)
1545
 
        result = self._save_pack_names(clear_obsolete_packs=True,
1546
 
                                       obsolete_packs=to_be_obsoleted)
1547
 
        return result
1548
 
 
1549
 
    def _flush_new_pack(self):
1550
 
        if self._new_pack is not None:
1551
 
            self._new_pack.flush()
 
1248
        self._save_pack_names(clear_obsolete_packs=True)
 
1249
        # Move the old packs out of the way now they are no longer referenced.
 
1250
        for revision_count, packs in pack_operations:
 
1251
            self._obsolete_packs(packs)
1552
1252
 
1553
1253
    def lock_names(self):
1554
1254
        """Acquire the mutex around the pack-names index.
1555
 
 
 
1255
        
1556
1256
        This cannot be used in the middle of a read-only transaction on the
1557
1257
        repository.
1558
1258
        """
1559
1259
        self.repo.control_files.lock_write()
1560
1260
 
1561
 
    def _already_packed(self):
1562
 
        """Is the collection already packed?"""
1563
 
        return not (self.repo._format.pack_compresses or (len(self._names) > 1))
1564
 
 
1565
 
    def pack(self, hint=None, clean_obsolete_packs=False):
 
1261
    def pack(self):
1566
1262
        """Pack the pack collection totally."""
1567
1263
        self.ensure_loaded()
1568
1264
        total_packs = len(self._names)
1569
 
        if self._already_packed():
 
1265
        if total_packs < 2:
 
1266
            # This is arguably wrong because we might not be optimal, but for
 
1267
            # now lets leave it in. (e.g. reconcile -> one pack. But not
 
1268
            # optimal.
1570
1269
            return
1571
1270
        total_revisions = self.revision_index.combined_index.key_count()
1572
1271
        # XXX: the following may want to be a class, to pack with a given
1573
1272
        # policy.
1574
1273
        mutter('Packing repository %s, which has %d pack files, '
1575
 
            'containing %d revisions with hint %r.', self, total_packs,
1576
 
            total_revisions, hint)
 
1274
            'containing %d revisions into 1 packs.', self, total_packs,
 
1275
            total_revisions)
1577
1276
        # determine which packs need changing
 
1277
        pack_distribution = [1]
1578
1278
        pack_operations = [[0, []]]
1579
1279
        for pack in self.all_packs():
1580
 
            if hint is None or pack.name in hint:
1581
 
                # Either no hint was provided (so we are packing everything),
1582
 
                # or this pack was included in the hint.
1583
 
                pack_operations[-1][0] += pack.get_revision_count()
1584
 
                pack_operations[-1][1].append(pack)
 
1280
            pack_operations[-1][0] += pack.get_revision_count()
 
1281
            pack_operations[-1][1].append(pack)
1585
1282
        self._execute_pack_operations(pack_operations, OptimisingPacker)
1586
1283
 
1587
 
        if clean_obsolete_packs:
1588
 
            self._clear_obsolete_packs()
1589
 
 
1590
1284
    def plan_autopack_combinations(self, existing_packs, pack_distribution):
1591
1285
        """Plan a pack operation.
1592
1286
 
1602
1296
        # plan out what packs to keep, and what to reorganise
1603
1297
        while len(existing_packs):
1604
1298
            # take the largest pack, and if its less than the head of the
1605
 
            # distribution chart we will include its contents in the new pack
1606
 
            # for that position. If its larger, we remove its size from the
 
1299
            # distribution chart we will include its contents in the new pack for
 
1300
            # that position. If its larger, we remove its size from the
1607
1301
            # distribution chart
1608
1302
            next_pack_rev_count, next_pack = existing_packs.pop(0)
1609
1303
            if next_pack_rev_count >= pack_distribution[0]:
1626
1320
                    # this pack is used up, shift left.
1627
1321
                    del pack_distribution[0]
1628
1322
                    pack_operations.append([0, []])
1629
 
        # Now that we know which pack files we want to move, shove them all
1630
 
        # into a single pack file.
1631
 
        final_rev_count = 0
1632
 
        final_pack_list = []
1633
 
        for num_revs, pack_files in pack_operations:
1634
 
            final_rev_count += num_revs
1635
 
            final_pack_list.extend(pack_files)
1636
 
        if len(final_pack_list) == 1:
1637
 
            raise AssertionError('We somehow generated an autopack with a'
1638
 
                ' single pack file being moved.')
1639
 
            return []
1640
 
        return [[final_rev_count, final_pack_list]]
 
1323
        
 
1324
        return pack_operations
1641
1325
 
1642
1326
    def ensure_loaded(self):
1643
 
        """Ensure we have read names from disk.
1644
 
 
1645
 
        :return: True if the disk names had not been previously read.
1646
 
        """
1647
1327
        # NB: if you see an assertion error here, its probably access against
1648
1328
        # an unlocked repo. Naughty.
1649
1329
        if not self.repo.is_locked():
1655
1335
                name = key[0]
1656
1336
                self._names[name] = self._parse_index_sizes(value)
1657
1337
                self._packs_at_load.add((key, value))
1658
 
            result = True
1659
 
        else:
1660
 
            result = False
1661
1338
        # populate all the metadata.
1662
1339
        self.all_packs()
1663
 
        return result
1664
1340
 
1665
1341
    def _parse_index_sizes(self, value):
1666
1342
        """Parse a string of index sizes."""
1679
1355
            inv_index = self._make_index(name, '.iix')
1680
1356
            txt_index = self._make_index(name, '.tix')
1681
1357
            sig_index = self._make_index(name, '.six')
1682
 
            if self.chk_index is not None:
1683
 
                chk_index = self._make_index(name, '.cix', unlimited_cache=True)
1684
 
            else:
1685
 
                chk_index = None
1686
1358
            result = ExistingPack(self._pack_transport, name, rev_index,
1687
 
                inv_index, txt_index, sig_index, chk_index)
 
1359
                inv_index, txt_index, sig_index)
1688
1360
            self.add_pack_to_memory(result)
1689
1361
            return result
1690
1362
 
1691
 
    def _resume_pack(self, name):
1692
 
        """Get a suspended Pack object by name.
1693
 
 
1694
 
        :param name: The name of the pack - e.g. '123456'
1695
 
        :return: A Pack object.
1696
 
        """
1697
 
        if not re.match('[a-f0-9]{32}', name):
1698
 
            # Tokens should be md5sums of the suspended pack file, i.e. 32 hex
1699
 
            # digits.
1700
 
            raise errors.UnresumableWriteGroup(
1701
 
                self.repo, [name], 'Malformed write group token')
1702
 
        try:
1703
 
            rev_index = self._make_index(name, '.rix', resume=True)
1704
 
            inv_index = self._make_index(name, '.iix', resume=True)
1705
 
            txt_index = self._make_index(name, '.tix', resume=True)
1706
 
            sig_index = self._make_index(name, '.six', resume=True)
1707
 
            if self.chk_index is not None:
1708
 
                chk_index = self._make_index(name, '.cix', resume=True,
1709
 
                                             unlimited_cache=True)
1710
 
            else:
1711
 
                chk_index = None
1712
 
            result = self.resumed_pack_factory(name, rev_index, inv_index,
1713
 
                txt_index, sig_index, self._upload_transport,
1714
 
                self._pack_transport, self._index_transport, self,
1715
 
                chk_index=chk_index)
1716
 
        except errors.NoSuchFile, e:
1717
 
            raise errors.UnresumableWriteGroup(self.repo, [name], str(e))
1718
 
        self.add_pack_to_memory(result)
1719
 
        self._resumed_packs.append(result)
1720
 
        return result
1721
 
 
1722
1363
    def allocate(self, a_new_pack):
1723
1364
        """Allocate name in the list of packs.
1724
1365
 
1734
1375
 
1735
1376
    def _iter_disk_pack_index(self):
1736
1377
        """Iterate over the contents of the pack-names index.
1737
 
 
 
1378
        
1738
1379
        This is used when loading the list from disk, and before writing to
1739
1380
        detect updates from others during our write operation.
1740
1381
        :return: An iterator of the index contents.
1741
1382
        """
1742
 
        return self._index_class(self.transport, 'pack-names', None
 
1383
        return GraphIndex(self.transport, 'pack-names', None
1743
1384
                ).iter_all_entries()
1744
1385
 
1745
 
    def _make_index(self, name, suffix, resume=False, unlimited_cache=False):
 
1386
    def _make_index(self, name, suffix):
1746
1387
        size_offset = self._suffix_offsets[suffix]
1747
1388
        index_name = name + suffix
1748
 
        if resume:
1749
 
            transport = self._upload_transport
1750
 
            index_size = transport.stat(index_name).st_size
1751
 
        else:
1752
 
            transport = self._index_transport
1753
 
            index_size = self._names[name][size_offset]
1754
 
        return self._index_class(transport, index_name, index_size,
1755
 
                                 unlimited_cache=unlimited_cache)
 
1389
        index_size = self._names[name][size_offset]
 
1390
        return GraphIndex(
 
1391
            self._index_transport, index_name, index_size)
1756
1392
 
1757
1393
    def _max_pack_count(self, total_revisions):
1758
1394
        """Return the maximum number of packs to use for total revisions.
1759
 
 
 
1395
        
1760
1396
        :param total_revisions: The total number of revisions in the
1761
1397
            repository.
1762
1398
        """
1786
1422
        :param return: None.
1787
1423
        """
1788
1424
        for pack in packs:
1789
 
            try:
1790
 
                pack.pack_transport.rename(pack.file_name(),
1791
 
                    '../obsolete_packs/' + pack.file_name())
1792
 
            except (errors.PathError, errors.TransportError), e:
1793
 
                # TODO: Should these be warnings or mutters?
1794
 
                mutter("couldn't rename obsolete pack, skipping it:\n%s"
1795
 
                       % (e,))
 
1425
            pack.pack_transport.rename(pack.file_name(),
 
1426
                '../obsolete_packs/' + pack.file_name())
1796
1427
            # TODO: Probably needs to know all possible indices for this pack
1797
1428
            # - or maybe list the directory and move all indices matching this
1798
1429
            # name whether we recognize it or not?
1799
 
            suffixes = ['.iix', '.six', '.tix', '.rix']
1800
 
            if self.chk_index is not None:
1801
 
                suffixes.append('.cix')
1802
 
            for suffix in suffixes:
1803
 
                try:
1804
 
                    self._index_transport.rename(pack.name + suffix,
1805
 
                        '../obsolete_packs/' + pack.name + suffix)
1806
 
                except (errors.PathError, errors.TransportError), e:
1807
 
                    mutter("couldn't rename obsolete index, skipping it:\n%s"
1808
 
                           % (e,))
 
1430
            for suffix in ('.iix', '.six', '.tix', '.rix'):
 
1431
                self._index_transport.rename(pack.name + suffix,
 
1432
                    '../obsolete_packs/' + pack.name + suffix)
1809
1433
 
1810
1434
    def pack_distribution(self, total_revisions):
1811
1435
        """Generate a list of the number of revisions to put in each pack.
1829
1453
 
1830
1454
    def _remove_pack_from_memory(self, pack):
1831
1455
        """Remove pack from the packs accessed by this repository.
1832
 
 
 
1456
        
1833
1457
        Only affects memory state, until self._save_pack_names() is invoked.
1834
1458
        """
1835
1459
        self._names.pop(pack.name)
1836
1460
        self._packs_by_name.pop(pack.name)
1837
1461
        self._remove_pack_indices(pack)
1838
 
        self.packs.remove(pack)
1839
1462
 
1840
 
    def _remove_pack_indices(self, pack, ignore_missing=False):
1841
 
        """Remove the indices for pack from the aggregated indices.
1842
 
        
1843
 
        :param ignore_missing: Suppress KeyErrors from calling remove_index.
1844
 
        """
1845
 
        for index_type in Pack.index_definitions.keys():
1846
 
            attr_name = index_type + '_index'
1847
 
            aggregate_index = getattr(self, attr_name)
1848
 
            if aggregate_index is not None:
1849
 
                pack_index = getattr(pack, attr_name)
1850
 
                try:
1851
 
                    aggregate_index.remove_index(pack_index)
1852
 
                except KeyError:
1853
 
                    if ignore_missing:
1854
 
                        continue
1855
 
                    raise
 
1463
    def _remove_pack_indices(self, pack):
 
1464
        """Remove the indices for pack from the aggregated indices."""
 
1465
        self.revision_index.remove_index(pack.revision_index, pack)
 
1466
        self.inventory_index.remove_index(pack.inventory_index, pack)
 
1467
        self.text_index.remove_index(pack.text_index, pack)
 
1468
        self.signature_index.remove_index(pack.signature_index, pack)
1856
1469
 
1857
1470
    def reset(self):
1858
1471
        """Clear all cached data."""
1859
1472
        # cached revision data
 
1473
        self.repo._revision_knit = None
1860
1474
        self.revision_index.clear()
1861
1475
        # cached signature data
 
1476
        self.repo._signature_knit = None
1862
1477
        self.signature_index.clear()
1863
1478
        # cached file text data
1864
1479
        self.text_index.clear()
 
1480
        self.repo._text_knit = None
1865
1481
        # cached inventory data
1866
1482
        self.inventory_index.clear()
1867
 
        # cached chk data
1868
 
        if self.chk_index is not None:
1869
 
            self.chk_index.clear()
1870
1483
        # remove the open pack
1871
1484
        self._new_pack = None
1872
1485
        # information about packs.
1875
1488
        self._packs_by_name = {}
1876
1489
        self._packs_at_load = None
1877
1490
 
 
1491
    def _make_index_map(self, index_suffix):
 
1492
        """Return information on existing indices.
 
1493
 
 
1494
        :param suffix: Index suffix added to pack name.
 
1495
 
 
1496
        :returns: (pack_map, indices) where indices is a list of GraphIndex 
 
1497
        objects, and pack_map is a mapping from those objects to the 
 
1498
        pack tuple they describe.
 
1499
        """
 
1500
        # TODO: stop using this; it creates new indices unnecessarily.
 
1501
        self.ensure_loaded()
 
1502
        suffix_map = {'.rix': 'revision_index',
 
1503
            '.six': 'signature_index',
 
1504
            '.iix': 'inventory_index',
 
1505
            '.tix': 'text_index',
 
1506
        }
 
1507
        return self._packs_list_to_pack_map_and_index_list(self.all_packs(),
 
1508
            suffix_map[index_suffix])
 
1509
 
 
1510
    def _packs_list_to_pack_map_and_index_list(self, packs, index_attribute):
 
1511
        """Convert a list of packs to an index pack map and index list.
 
1512
 
 
1513
        :param packs: The packs list to process.
 
1514
        :param index_attribute: The attribute that the desired index is found
 
1515
            on.
 
1516
        :return: A tuple (map, list) where map contains the dict from
 
1517
            index:pack_tuple, and lsit contains the indices in the same order
 
1518
            as the packs list.
 
1519
        """
 
1520
        indices = []
 
1521
        pack_map = {}
 
1522
        for pack in packs:
 
1523
            index = getattr(pack, index_attribute)
 
1524
            indices.append(index)
 
1525
            pack_map[index] = (pack.pack_transport, pack.file_name())
 
1526
        return pack_map, indices
 
1527
 
 
1528
    def _index_contents(self, pack_map, key_filter=None):
 
1529
        """Get an iterable of the index contents from a pack_map.
 
1530
 
 
1531
        :param pack_map: A map from indices to pack details.
 
1532
        :param key_filter: An optional filter to limit the
 
1533
            keys returned.
 
1534
        """
 
1535
        indices = [index for index in pack_map.iterkeys()]
 
1536
        all_index = CombinedGraphIndex(indices)
 
1537
        if key_filter is None:
 
1538
            return all_index.iter_all_entries()
 
1539
        else:
 
1540
            return all_index.iter_entries(key_filter)
 
1541
 
1878
1542
    def _unlock_names(self):
1879
1543
        """Release the mutex around the pack-names index."""
1880
1544
        self.repo.control_files.unlock()
1881
1545
 
1882
 
    def _diff_pack_names(self):
1883
 
        """Read the pack names from disk, and compare it to the one in memory.
1884
 
 
1885
 
        :return: (disk_nodes, deleted_nodes, new_nodes)
1886
 
            disk_nodes    The final set of nodes that should be referenced
1887
 
            deleted_nodes Nodes which have been removed from when we started
1888
 
            new_nodes     Nodes that are newly introduced
1889
 
        """
1890
 
        # load the disk nodes across
1891
 
        disk_nodes = set()
1892
 
        for index, key, value in self._iter_disk_pack_index():
1893
 
            disk_nodes.add((key, value))
1894
 
        orig_disk_nodes = set(disk_nodes)
1895
 
 
1896
 
        # do a two-way diff against our original content
1897
 
        current_nodes = set()
1898
 
        for name, sizes in self._names.iteritems():
1899
 
            current_nodes.add(
1900
 
                ((name, ), ' '.join(str(size) for size in sizes)))
1901
 
 
1902
 
        # Packs no longer present in the repository, which were present when we
1903
 
        # locked the repository
1904
 
        deleted_nodes = self._packs_at_load - current_nodes
1905
 
        # Packs which this process is adding
1906
 
        new_nodes = current_nodes - self._packs_at_load
1907
 
 
1908
 
        # Update the disk_nodes set to include the ones we are adding, and
1909
 
        # remove the ones which were removed by someone else
1910
 
        disk_nodes.difference_update(deleted_nodes)
1911
 
        disk_nodes.update(new_nodes)
1912
 
 
1913
 
        return disk_nodes, deleted_nodes, new_nodes, orig_disk_nodes
1914
 
 
1915
 
    def _syncronize_pack_names_from_disk_nodes(self, disk_nodes):
1916
 
        """Given the correct set of pack files, update our saved info.
1917
 
 
1918
 
        :return: (removed, added, modified)
1919
 
            removed     pack names removed from self._names
1920
 
            added       pack names added to self._names
1921
 
            modified    pack names that had changed value
1922
 
        """
1923
 
        removed = []
1924
 
        added = []
1925
 
        modified = []
1926
 
        ## self._packs_at_load = disk_nodes
 
1546
    def _save_pack_names(self, clear_obsolete_packs=False):
 
1547
        """Save the list of packs.
 
1548
 
 
1549
        This will take out the mutex around the pack names list for the
 
1550
        duration of the method call. If concurrent updates have been made, a
 
1551
        three-way merge between the current list and the current in memory list
 
1552
        is performed.
 
1553
 
 
1554
        :param clear_obsolete_packs: If True, clear out the contents of the
 
1555
            obsolete_packs directory.
 
1556
        """
 
1557
        self.lock_names()
 
1558
        try:
 
1559
            builder = GraphIndexBuilder()
 
1560
            # load the disk nodes across
 
1561
            disk_nodes = set()
 
1562
            for index, key, value in self._iter_disk_pack_index():
 
1563
                disk_nodes.add((key, value))
 
1564
            # do a two-way diff against our original content
 
1565
            current_nodes = set()
 
1566
            for name, sizes in self._names.iteritems():
 
1567
                current_nodes.add(
 
1568
                    ((name, ), ' '.join(str(size) for size in sizes)))
 
1569
            deleted_nodes = self._packs_at_load - current_nodes
 
1570
            new_nodes = current_nodes - self._packs_at_load
 
1571
            disk_nodes.difference_update(deleted_nodes)
 
1572
            disk_nodes.update(new_nodes)
 
1573
            # TODO: handle same-name, index-size-changes here - 
 
1574
            # e.g. use the value from disk, not ours, *unless* we're the one
 
1575
            # changing it.
 
1576
            for key, value in disk_nodes:
 
1577
                builder.add_node(key, value)
 
1578
            self.transport.put_file('pack-names', builder.finish(),
 
1579
                mode=self.repo.control_files._file_mode)
 
1580
            # move the baseline forward
 
1581
            self._packs_at_load = disk_nodes
 
1582
            # now clear out the obsolete packs directory
 
1583
            if clear_obsolete_packs:
 
1584
                self.transport.clone('obsolete_packs').delete_multi(
 
1585
                    self.transport.list_dir('obsolete_packs'))
 
1586
        finally:
 
1587
            self._unlock_names()
 
1588
        # synchronise the memory packs list with what we just wrote:
1927
1589
        new_names = dict(disk_nodes)
1928
1590
        # drop no longer present nodes
1929
1591
        for pack in self.all_packs():
1930
1592
            if (pack.name,) not in new_names:
1931
 
                removed.append(pack.name)
1932
1593
                self._remove_pack_from_memory(pack)
1933
1594
        # add new nodes/refresh existing ones
1934
1595
        for key, value in disk_nodes:
1948
1609
                    self._remove_pack_from_memory(self.get_pack_by_name(name))
1949
1610
                    self._names[name] = sizes
1950
1611
                    self.get_pack_by_name(name)
1951
 
                    modified.append(name)
1952
1612
            else:
1953
1613
                # new
1954
1614
                self._names[name] = sizes
1955
1615
                self.get_pack_by_name(name)
1956
 
                added.append(name)
1957
 
        return removed, added, modified
1958
 
 
1959
 
    def _save_pack_names(self, clear_obsolete_packs=False, obsolete_packs=None):
1960
 
        """Save the list of packs.
1961
 
 
1962
 
        This will take out the mutex around the pack names list for the
1963
 
        duration of the method call. If concurrent updates have been made, a
1964
 
        three-way merge between the current list and the current in memory list
1965
 
        is performed.
1966
 
 
1967
 
        :param clear_obsolete_packs: If True, clear out the contents of the
1968
 
            obsolete_packs directory.
1969
 
        :param obsolete_packs: Packs that are obsolete once the new pack-names
1970
 
            file has been written.
1971
 
        :return: A list of the names saved that were not previously on disk.
1972
 
        """
1973
 
        already_obsolete = []
1974
 
        self.lock_names()
1975
 
        try:
1976
 
            builder = self._index_builder_class()
1977
 
            (disk_nodes, deleted_nodes, new_nodes,
1978
 
             orig_disk_nodes) = self._diff_pack_names()
1979
 
            # TODO: handle same-name, index-size-changes here -
1980
 
            # e.g. use the value from disk, not ours, *unless* we're the one
1981
 
            # changing it.
1982
 
            for key, value in disk_nodes:
1983
 
                builder.add_node(key, value)
1984
 
            self.transport.put_file('pack-names', builder.finish(),
1985
 
                mode=self.repo.bzrdir._get_file_mode())
1986
 
            self._packs_at_load = disk_nodes
1987
 
            if clear_obsolete_packs:
1988
 
                to_preserve = None
1989
 
                if obsolete_packs:
1990
 
                    to_preserve = set([o.name for o in obsolete_packs])
1991
 
                already_obsolete = self._clear_obsolete_packs(to_preserve)
1992
 
        finally:
1993
 
            self._unlock_names()
1994
 
        # synchronise the memory packs list with what we just wrote:
1995
 
        self._syncronize_pack_names_from_disk_nodes(disk_nodes)
1996
 
        if obsolete_packs:
1997
 
            # TODO: We could add one more condition here. "if o.name not in
1998
 
            #       orig_disk_nodes and o != the new_pack we haven't written to
1999
 
            #       disk yet. However, the new pack object is not easily
2000
 
            #       accessible here (it would have to be passed through the
2001
 
            #       autopacking code, etc.)
2002
 
            obsolete_packs = [o for o in obsolete_packs
2003
 
                              if o.name not in already_obsolete]
2004
 
            self._obsolete_packs(obsolete_packs)
2005
 
        return [new_node[0][0] for new_node in new_nodes]
2006
 
 
2007
 
    def reload_pack_names(self):
2008
 
        """Sync our pack listing with what is present in the repository.
2009
 
 
2010
 
        This should be called when we find out that something we thought was
2011
 
        present is now missing. This happens when another process re-packs the
2012
 
        repository, etc.
2013
 
 
2014
 
        :return: True if the in-memory list of packs has been altered at all.
2015
 
        """
2016
 
        # The ensure_loaded call is to handle the case where the first call
2017
 
        # made involving the collection was to reload_pack_names, where we 
2018
 
        # don't have a view of disk contents. Its a bit of a bandaid, and
2019
 
        # causes two reads of pack-names, but its a rare corner case not struck
2020
 
        # with regular push/pull etc.
2021
 
        first_read = self.ensure_loaded()
2022
 
        if first_read:
2023
 
            return True
2024
 
        # out the new value.
2025
 
        (disk_nodes, deleted_nodes, new_nodes,
2026
 
         orig_disk_nodes) = self._diff_pack_names()
2027
 
        # _packs_at_load is meant to be the explicit list of names in
2028
 
        # 'pack-names' at then start. As such, it should not contain any
2029
 
        # pending names that haven't been written out yet.
2030
 
        self._packs_at_load = orig_disk_nodes
2031
 
        (removed, added,
2032
 
         modified) = self._syncronize_pack_names_from_disk_nodes(disk_nodes)
2033
 
        if removed or added or modified:
2034
 
            return True
2035
 
        return False
2036
 
 
2037
 
    def _restart_autopack(self):
2038
 
        """Reload the pack names list, and restart the autopack code."""
2039
 
        if not self.reload_pack_names():
2040
 
            # Re-raise the original exception, because something went missing
2041
 
            # and a restart didn't find it
2042
 
            raise
2043
 
        raise errors.RetryAutopack(self.repo, False, sys.exc_info())
2044
 
 
2045
 
    def _clear_obsolete_packs(self, preserve=None):
2046
 
        """Delete everything from the obsolete-packs directory.
2047
 
 
2048
 
        :return: A list of pack identifiers (the filename without '.pack') that
2049
 
            were found in obsolete_packs.
2050
 
        """
2051
 
        found = []
2052
 
        obsolete_pack_transport = self.transport.clone('obsolete_packs')
2053
 
        if preserve is None:
2054
 
            preserve = set()
2055
 
        for filename in obsolete_pack_transport.list_dir('.'):
2056
 
            name, ext = osutils.splitext(filename)
2057
 
            if ext == '.pack':
2058
 
                found.append(name)
2059
 
            if name in preserve:
2060
 
                continue
2061
 
            try:
2062
 
                obsolete_pack_transport.delete(filename)
2063
 
            except (errors.PathError, errors.TransportError), e:
2064
 
                warning("couldn't delete obsolete pack, skipping it:\n%s"
2065
 
                        % (e,))
2066
 
        return found
2067
1616
 
2068
1617
    def _start_write_group(self):
2069
1618
        # Do not permit preparation for writing if we're not in a 'write lock'.
2070
1619
        if not self.repo.is_write_locked():
2071
1620
            raise errors.NotWriteLocked(self)
2072
 
        self._new_pack = self.pack_factory(self, upload_suffix='.pack',
2073
 
            file_mode=self.repo.bzrdir._get_file_mode())
 
1621
        self._new_pack = NewPack(self._upload_transport, self._index_transport,
 
1622
            self._pack_transport, upload_suffix='.pack',
 
1623
            file_mode=self.repo.control_files._file_mode)
2074
1624
        # allow writing: queue writes to a new index
2075
1625
        self.revision_index.add_writable_index(self._new_pack.revision_index,
2076
1626
            self._new_pack)
2078
1628
            self._new_pack)
2079
1629
        self.text_index.add_writable_index(self._new_pack.text_index,
2080
1630
            self._new_pack)
2081
 
        self._new_pack.text_index.set_optimize(combine_backing_indices=False)
2082
1631
        self.signature_index.add_writable_index(self._new_pack.signature_index,
2083
1632
            self._new_pack)
2084
 
        if self.chk_index is not None:
2085
 
            self.chk_index.add_writable_index(self._new_pack.chk_index,
2086
 
                self._new_pack)
2087
 
            self.repo.chk_bytes._index._add_callback = self.chk_index.add_callback
2088
 
            self._new_pack.chk_index.set_optimize(combine_backing_indices=False)
2089
1633
 
2090
 
        self.repo.inventories._index._add_callback = self.inventory_index.add_callback
2091
 
        self.repo.revisions._index._add_callback = self.revision_index.add_callback
2092
 
        self.repo.signatures._index._add_callback = self.signature_index.add_callback
2093
 
        self.repo.texts._index._add_callback = self.text_index.add_callback
 
1634
        # reused revision and signature knits may need updating
 
1635
        #
 
1636
        # "Hysterical raisins. client code in bzrlib grabs those knits outside
 
1637
        # of write groups and then mutates it inside the write group."
 
1638
        if self.repo._revision_knit is not None:
 
1639
            self.repo._revision_knit._index._add_callback = \
 
1640
                self.revision_index.add_callback
 
1641
        if self.repo._signature_knit is not None:
 
1642
            self.repo._signature_knit._index._add_callback = \
 
1643
                self.signature_index.add_callback
 
1644
        # create a reused knit object for text addition in commit.
 
1645
        self.repo._text_knit = self.repo.weave_store.get_weave_or_empty(
 
1646
            'all-texts', None)
2094
1647
 
2095
1648
    def _abort_write_group(self):
2096
1649
        # FIXME: just drop the transient index.
2097
1650
        # forget what names there are
2098
1651
        if self._new_pack is not None:
2099
 
            operation = cleanup.OperationWithCleanups(self._new_pack.abort)
2100
 
            operation.add_cleanup(setattr, self, '_new_pack', None)
2101
 
            # If we aborted while in the middle of finishing the write
2102
 
            # group, _remove_pack_indices could fail because the indexes are
2103
 
            # already gone.  But they're not there we shouldn't fail in this
2104
 
            # case, so we pass ignore_missing=True.
2105
 
            operation.add_cleanup(self._remove_pack_indices, self._new_pack,
2106
 
                ignore_missing=True)
2107
 
            operation.run_simple()
2108
 
        for resumed_pack in self._resumed_packs:
2109
 
            operation = cleanup.OperationWithCleanups(resumed_pack.abort)
2110
 
            # See comment in previous finally block.
2111
 
            operation.add_cleanup(self._remove_pack_indices, resumed_pack,
2112
 
                ignore_missing=True)
2113
 
            operation.run_simple()
2114
 
        del self._resumed_packs[:]
2115
 
 
2116
 
    def _remove_resumed_pack_indices(self):
2117
 
        for resumed_pack in self._resumed_packs:
2118
 
            self._remove_pack_indices(resumed_pack)
2119
 
        del self._resumed_packs[:]
2120
 
 
2121
 
    def _check_new_inventories(self):
2122
 
        """Detect missing inventories in this write group.
2123
 
 
2124
 
        :returns: list of strs, summarising any problems found.  If the list is
2125
 
            empty no problems were found.
2126
 
        """
2127
 
        # The base implementation does no checks.  GCRepositoryPackCollection
2128
 
        # overrides this.
2129
 
        return []
2130
 
        
 
1652
            self._new_pack.abort()
 
1653
            self._remove_pack_indices(self._new_pack)
 
1654
            self._new_pack = None
 
1655
        self.repo._text_knit = None
 
1656
 
2131
1657
    def _commit_write_group(self):
2132
 
        all_missing = set()
2133
 
        for prefix, versioned_file in (
2134
 
                ('revisions', self.repo.revisions),
2135
 
                ('inventories', self.repo.inventories),
2136
 
                ('texts', self.repo.texts),
2137
 
                ('signatures', self.repo.signatures),
2138
 
                ):
2139
 
            missing = versioned_file.get_missing_compression_parent_keys()
2140
 
            all_missing.update([(prefix,) + key for key in missing])
2141
 
        if all_missing:
2142
 
            raise errors.BzrCheckError(
2143
 
                "Repository %s has missing compression parent(s) %r "
2144
 
                 % (self.repo, sorted(all_missing)))
2145
 
        problems = self._check_new_inventories()
2146
 
        if problems:
2147
 
            problems_summary = '\n'.join(problems)
2148
 
            raise errors.BzrCheckError(
2149
 
                "Cannot add revision(s) to repository: " + problems_summary)
2150
1658
        self._remove_pack_indices(self._new_pack)
2151
 
        any_new_content = False
2152
1659
        if self._new_pack.data_inserted():
2153
1660
            # get all the data to disk and read to use
2154
1661
            self._new_pack.finish()
2155
1662
            self.allocate(self._new_pack)
2156
1663
            self._new_pack = None
2157
 
            any_new_content = True
2158
 
        else:
2159
 
            self._new_pack.abort()
2160
 
            self._new_pack = None
2161
 
        for resumed_pack in self._resumed_packs:
2162
 
            # XXX: this is a pretty ugly way to turn the resumed pack into a
2163
 
            # properly committed pack.
2164
 
            self._names[resumed_pack.name] = None
2165
 
            self._remove_pack_from_memory(resumed_pack)
2166
 
            resumed_pack.finish()
2167
 
            self.allocate(resumed_pack)
2168
 
            any_new_content = True
2169
 
        del self._resumed_packs[:]
2170
 
        if any_new_content:
2171
 
            result = self.autopack()
2172
 
            if not result:
 
1664
            if not self.autopack():
2173
1665
                # when autopack takes no steps, the names list is still
2174
1666
                # unsaved.
2175
 
                return self._save_pack_names()
2176
 
            return result
2177
 
        return []
2178
 
 
2179
 
    def _suspend_write_group(self):
2180
 
        tokens = [pack.name for pack in self._resumed_packs]
2181
 
        self._remove_pack_indices(self._new_pack)
2182
 
        if self._new_pack.data_inserted():
2183
 
            # get all the data to disk and read to use
2184
 
            self._new_pack.finish(suspend=True)
2185
 
            tokens.append(self._new_pack.name)
2186
 
            self._new_pack = None
 
1667
                self._save_pack_names()
2187
1668
        else:
2188
1669
            self._new_pack.abort()
2189
1670
            self._new_pack = None
2190
 
        self._remove_resumed_pack_indices()
2191
 
        return tokens
2192
 
 
2193
 
    def _resume_write_group(self, tokens):
2194
 
        for token in tokens:
2195
 
            self._resume_pack(token)
 
1671
        self.repo._text_knit = None
 
1672
 
 
1673
 
 
1674
class KnitPackRevisionStore(KnitRevisionStore):
 
1675
    """An object to adapt access from RevisionStore's to use KnitPacks.
 
1676
 
 
1677
    This class works by replacing the original RevisionStore.
 
1678
    We need to do this because the KnitPackRevisionStore is less
 
1679
    isolated in its layering - it uses services from the repo.
 
1680
    """
 
1681
 
 
1682
    def __init__(self, repo, transport, revisionstore):
 
1683
        """Create a KnitPackRevisionStore on repo with revisionstore.
 
1684
 
 
1685
        This will store its state in the Repository, use the
 
1686
        indices to provide a KnitGraphIndex,
 
1687
        and at the end of transactions write new indices.
 
1688
        """
 
1689
        KnitRevisionStore.__init__(self, revisionstore.versioned_file_store)
 
1690
        self.repo = repo
 
1691
        self._serializer = revisionstore._serializer
 
1692
        self.transport = transport
 
1693
 
 
1694
    def get_revision_file(self, transaction):
 
1695
        """Get the revision versioned file object."""
 
1696
        if getattr(self.repo, '_revision_knit', None) is not None:
 
1697
            return self.repo._revision_knit
 
1698
        self.repo._pack_collection.ensure_loaded()
 
1699
        add_callback = self.repo._pack_collection.revision_index.add_callback
 
1700
        # setup knit specific objects
 
1701
        knit_index = KnitGraphIndex(
 
1702
            self.repo._pack_collection.revision_index.combined_index,
 
1703
            add_callback=add_callback)
 
1704
        self.repo._revision_knit = knit.KnitVersionedFile(
 
1705
            'revisions', self.transport.clone('..'),
 
1706
            self.repo.control_files._file_mode,
 
1707
            create=False, access_mode=self.repo._access_mode(),
 
1708
            index=knit_index, delta=False, factory=knit.KnitPlainFactory(),
 
1709
            access_method=self.repo._pack_collection.revision_index.knit_access)
 
1710
        return self.repo._revision_knit
 
1711
 
 
1712
    def get_signature_file(self, transaction):
 
1713
        """Get the signature versioned file object."""
 
1714
        if getattr(self.repo, '_signature_knit', None) is not None:
 
1715
            return self.repo._signature_knit
 
1716
        self.repo._pack_collection.ensure_loaded()
 
1717
        add_callback = self.repo._pack_collection.signature_index.add_callback
 
1718
        # setup knit specific objects
 
1719
        knit_index = KnitGraphIndex(
 
1720
            self.repo._pack_collection.signature_index.combined_index,
 
1721
            add_callback=add_callback, parents=False)
 
1722
        self.repo._signature_knit = knit.KnitVersionedFile(
 
1723
            'signatures', self.transport.clone('..'),
 
1724
            self.repo.control_files._file_mode,
 
1725
            create=False, access_mode=self.repo._access_mode(),
 
1726
            index=knit_index, delta=False, factory=knit.KnitPlainFactory(),
 
1727
            access_method=self.repo._pack_collection.signature_index.knit_access)
 
1728
        return self.repo._signature_knit
 
1729
 
 
1730
 
 
1731
class KnitPackTextStore(VersionedFileStore):
 
1732
    """Presents a TextStore abstraction on top of packs.
 
1733
 
 
1734
    This class works by replacing the original VersionedFileStore.
 
1735
    We need to do this because the KnitPackRevisionStore is less
 
1736
    isolated in its layering - it uses services from the repo and shares them
 
1737
    with all the data written in a single write group.
 
1738
    """
 
1739
 
 
1740
    def __init__(self, repo, transport, weavestore):
 
1741
        """Create a KnitPackTextStore on repo with weavestore.
 
1742
 
 
1743
        This will store its state in the Repository, use the
 
1744
        indices FileNames to provide a KnitGraphIndex,
 
1745
        and at the end of transactions write new indices.
 
1746
        """
 
1747
        # don't call base class constructor - it's not suitable.
 
1748
        # no transient data stored in the transaction
 
1749
        # cache.
 
1750
        self._precious = False
 
1751
        self.repo = repo
 
1752
        self.transport = transport
 
1753
        self.weavestore = weavestore
 
1754
        # XXX for check() which isn't updated yet
 
1755
        self._transport = weavestore._transport
 
1756
 
 
1757
    def get_weave_or_empty(self, file_id, transaction):
 
1758
        """Get a 'Knit' backed by the .tix indices.
 
1759
 
 
1760
        The transaction parameter is ignored.
 
1761
        """
 
1762
        self.repo._pack_collection.ensure_loaded()
 
1763
        add_callback = self.repo._pack_collection.text_index.add_callback
 
1764
        # setup knit specific objects
 
1765
        file_id_index = GraphIndexPrefixAdapter(
 
1766
            self.repo._pack_collection.text_index.combined_index,
 
1767
            (file_id, ), 1, add_nodes_callback=add_callback)
 
1768
        knit_index = KnitGraphIndex(file_id_index,
 
1769
            add_callback=file_id_index.add_nodes,
 
1770
            deltas=True, parents=True)
 
1771
        return knit.KnitVersionedFile('text:' + file_id,
 
1772
            self.transport.clone('..'),
 
1773
            None,
 
1774
            index=knit_index,
 
1775
            access_method=self.repo._pack_collection.text_index.knit_access,
 
1776
            factory=knit.KnitPlainFactory())
 
1777
 
 
1778
    get_weave = get_weave_or_empty
 
1779
 
 
1780
    def __iter__(self):
 
1781
        """Generate a list of the fileids inserted, for use by check."""
 
1782
        self.repo._pack_collection.ensure_loaded()
 
1783
        ids = set()
 
1784
        for index, key, value, refs in \
 
1785
            self.repo._pack_collection.text_index.combined_index.iter_all_entries():
 
1786
            ids.add(key[0])
 
1787
        return iter(ids)
 
1788
 
 
1789
 
 
1790
class InventoryKnitThunk(object):
 
1791
    """An object to manage thunking get_inventory_weave to pack based knits."""
 
1792
 
 
1793
    def __init__(self, repo, transport):
 
1794
        """Create an InventoryKnitThunk for repo at transport.
 
1795
 
 
1796
        This will store its state in the Repository, use the
 
1797
        indices FileNames to provide a KnitGraphIndex,
 
1798
        and at the end of transactions write a new index..
 
1799
        """
 
1800
        self.repo = repo
 
1801
        self.transport = transport
 
1802
 
 
1803
    def get_weave(self):
 
1804
        """Get a 'Knit' that contains inventory data."""
 
1805
        self.repo._pack_collection.ensure_loaded()
 
1806
        add_callback = self.repo._pack_collection.inventory_index.add_callback
 
1807
        # setup knit specific objects
 
1808
        knit_index = KnitGraphIndex(
 
1809
            self.repo._pack_collection.inventory_index.combined_index,
 
1810
            add_callback=add_callback, deltas=True, parents=True)
 
1811
        return knit.KnitVersionedFile(
 
1812
            'inventory', self.transport.clone('..'),
 
1813
            self.repo.control_files._file_mode,
 
1814
            create=False, access_mode=self.repo._access_mode(),
 
1815
            index=knit_index, delta=True, factory=knit.KnitPlainFactory(),
 
1816
            access_method=self.repo._pack_collection.inventory_index.knit_access)
2196
1817
 
2197
1818
 
2198
1819
class KnitPackRepository(KnitRepository):
2199
 
    """Repository with knit objects stored inside pack containers.
2200
 
 
2201
 
    The layering for a KnitPackRepository is:
2202
 
 
2203
 
    Graph        |  HPSS    | Repository public layer |
2204
 
    ===================================================
2205
 
    Tuple based apis below, string based, and key based apis above
2206
 
    ---------------------------------------------------
2207
 
    KnitVersionedFiles
2208
 
      Provides .texts, .revisions etc
2209
 
      This adapts the N-tuple keys to physical knit records which only have a
2210
 
      single string identifier (for historical reasons), which in older formats
2211
 
      was always the revision_id, and in the mapped code for packs is always
2212
 
      the last element of key tuples.
2213
 
    ---------------------------------------------------
2214
 
    GraphIndex
2215
 
      A separate GraphIndex is used for each of the
2216
 
      texts/inventories/revisions/signatures contained within each individual
2217
 
      pack file. The GraphIndex layer works in N-tuples and is unaware of any
2218
 
      semantic value.
2219
 
    ===================================================
2220
 
 
2221
 
    """
2222
 
 
2223
 
    def __init__(self, _format, a_bzrdir, control_files, _commit_builder_class,
2224
 
        _serializer):
 
1820
    """Experimental graph-knit using repository."""
 
1821
 
 
1822
    def __init__(self, _format, a_bzrdir, control_files, _revision_store,
 
1823
        control_store, text_store, _commit_builder_class, _serializer):
2225
1824
        KnitRepository.__init__(self, _format, a_bzrdir, control_files,
2226
 
            _commit_builder_class, _serializer)
2227
 
        index_transport = self._transport.clone('indices')
2228
 
        self._pack_collection = RepositoryPackCollection(self, self._transport,
 
1825
            _revision_store, control_store, text_store, _commit_builder_class,
 
1826
            _serializer)
 
1827
        index_transport = control_files._transport.clone('indices')
 
1828
        self._pack_collection = RepositoryPackCollection(self, control_files._transport,
2229
1829
            index_transport,
2230
 
            self._transport.clone('upload'),
2231
 
            self._transport.clone('packs'),
2232
 
            _format.index_builder_class,
2233
 
            _format.index_class,
2234
 
            use_chk_index=self._format.supports_chks,
2235
 
            )
2236
 
        self.inventories = KnitVersionedFiles(
2237
 
            _KnitGraphIndex(self._pack_collection.inventory_index.combined_index,
2238
 
                add_callback=self._pack_collection.inventory_index.add_callback,
2239
 
                deltas=True, parents=True, is_locked=self.is_locked),
2240
 
            data_access=self._pack_collection.inventory_index.data_access,
2241
 
            max_delta_chain=200)
2242
 
        self.revisions = KnitVersionedFiles(
2243
 
            _KnitGraphIndex(self._pack_collection.revision_index.combined_index,
2244
 
                add_callback=self._pack_collection.revision_index.add_callback,
2245
 
                deltas=False, parents=True, is_locked=self.is_locked,
2246
 
                track_external_parent_refs=True),
2247
 
            data_access=self._pack_collection.revision_index.data_access,
2248
 
            max_delta_chain=0)
2249
 
        self.signatures = KnitVersionedFiles(
2250
 
            _KnitGraphIndex(self._pack_collection.signature_index.combined_index,
2251
 
                add_callback=self._pack_collection.signature_index.add_callback,
2252
 
                deltas=False, parents=False, is_locked=self.is_locked),
2253
 
            data_access=self._pack_collection.signature_index.data_access,
2254
 
            max_delta_chain=0)
2255
 
        self.texts = KnitVersionedFiles(
2256
 
            _KnitGraphIndex(self._pack_collection.text_index.combined_index,
2257
 
                add_callback=self._pack_collection.text_index.add_callback,
2258
 
                deltas=True, parents=True, is_locked=self.is_locked),
2259
 
            data_access=self._pack_collection.text_index.data_access,
2260
 
            max_delta_chain=200)
2261
 
        if _format.supports_chks:
2262
 
            # No graph, no compression:- references from chks are between
2263
 
            # different objects not temporal versions of the same; and without
2264
 
            # some sort of temporal structure knit compression will just fail.
2265
 
            self.chk_bytes = KnitVersionedFiles(
2266
 
                _KnitGraphIndex(self._pack_collection.chk_index.combined_index,
2267
 
                    add_callback=self._pack_collection.chk_index.add_callback,
2268
 
                    deltas=False, parents=False, is_locked=self.is_locked),
2269
 
                data_access=self._pack_collection.chk_index.data_access,
2270
 
                max_delta_chain=0)
2271
 
        else:
2272
 
            self.chk_bytes = None
 
1830
            control_files._transport.clone('upload'),
 
1831
            control_files._transport.clone('packs'))
 
1832
        self._revision_store = KnitPackRevisionStore(self, index_transport, self._revision_store)
 
1833
        self.weave_store = KnitPackTextStore(self, index_transport, self.weave_store)
 
1834
        self._inv_thunk = InventoryKnitThunk(self, index_transport)
2273
1835
        # True when the repository object is 'write locked' (as opposed to the
2274
 
        # physical lock only taken out around changes to the pack-names list.)
 
1836
        # physical lock only taken out around changes to the pack-names list.) 
2275
1837
        # Another way to represent this would be a decorator around the control
2276
1838
        # files object that presents logical locks as physical ones - if this
2277
1839
        # gets ugly consider that alternative design. RBC 20071011
2282
1844
        self._reconcile_fixes_text_parents = True
2283
1845
        self._reconcile_backsup_inventory = False
2284
1846
 
2285
 
    def _warn_if_deprecated(self, branch=None):
2286
 
        # This class isn't deprecated, but one sub-format is
2287
 
        if isinstance(self._format, RepositoryFormatKnitPack5RichRootBroken):
2288
 
            super(KnitPackRepository, self)._warn_if_deprecated(branch)
2289
 
 
2290
1847
    def _abort_write_group(self):
2291
 
        self.revisions._index._key_dependencies.clear()
2292
1848
        self._pack_collection._abort_write_group()
2293
1849
 
2294
 
    def _get_source(self, to_format):
2295
 
        if to_format.network_name() == self._format.network_name():
2296
 
            return KnitPackStreamSource(self, to_format)
2297
 
        return super(KnitPackRepository, self)._get_source(to_format)
 
1850
    def _access_mode(self):
 
1851
        """Return 'w' or 'r' for depending on whether a write lock is active.
 
1852
        
 
1853
        This method is a helper for the Knit-thunking support objects.
 
1854
        """
 
1855
        if self.is_write_locked():
 
1856
            return 'w'
 
1857
        return 'r'
 
1858
 
 
1859
    def _find_inconsistent_revision_parents(self):
 
1860
        """Find revisions with incorrectly cached parents.
 
1861
 
 
1862
        :returns: an iterator yielding tuples of (revison-id, parents-in-index,
 
1863
            parents-in-revision).
 
1864
        """
 
1865
        if not self.is_locked():
 
1866
            raise errors.ObjectNotLocked(self)
 
1867
        pb = ui.ui_factory.nested_progress_bar()
 
1868
        result = []
 
1869
        try:
 
1870
            revision_nodes = self._pack_collection.revision_index \
 
1871
                .combined_index.iter_all_entries()
 
1872
            index_positions = []
 
1873
            # Get the cached index values for all revisions, and also the location
 
1874
            # in each index of the revision text so we can perform linear IO.
 
1875
            for index, key, value, refs in revision_nodes:
 
1876
                pos, length = value[1:].split(' ')
 
1877
                index_positions.append((index, int(pos), key[0],
 
1878
                    tuple(parent[0] for parent in refs[0])))
 
1879
                pb.update("Reading revision index.", 0, 0)
 
1880
            index_positions.sort()
 
1881
            batch_count = len(index_positions) / 1000 + 1
 
1882
            pb.update("Checking cached revision graph.", 0, batch_count)
 
1883
            for offset in xrange(batch_count):
 
1884
                pb.update("Checking cached revision graph.", offset)
 
1885
                to_query = index_positions[offset * 1000:(offset + 1) * 1000]
 
1886
                if not to_query:
 
1887
                    break
 
1888
                rev_ids = [item[2] for item in to_query]
 
1889
                revs = self.get_revisions(rev_ids)
 
1890
                for revision, item in zip(revs, to_query):
 
1891
                    index_parents = item[3]
 
1892
                    rev_parents = tuple(revision.parent_ids)
 
1893
                    if index_parents != rev_parents:
 
1894
                        result.append((revision.revision_id, index_parents, rev_parents))
 
1895
        finally:
 
1896
            pb.finished()
 
1897
        return result
 
1898
 
 
1899
    @symbol_versioning.deprecated_method(symbol_versioning.one_one)
 
1900
    def get_parents(self, revision_ids):
 
1901
        """See graph._StackedParentsProvider.get_parents."""
 
1902
        parent_map = self.get_parent_map(revision_ids)
 
1903
        return [parent_map.get(r, None) for r in revision_ids]
 
1904
 
 
1905
    def get_parent_map(self, keys):
 
1906
        """See graph._StackedParentsProvider.get_parent_map
 
1907
 
 
1908
        This implementation accesses the combined revision index to provide
 
1909
        answers.
 
1910
        """
 
1911
        self._pack_collection.ensure_loaded()
 
1912
        index = self._pack_collection.revision_index.combined_index
 
1913
        keys = set(keys)
 
1914
        if _mod_revision.NULL_REVISION in keys:
 
1915
            keys.discard(_mod_revision.NULL_REVISION)
 
1916
            found_parents = {_mod_revision.NULL_REVISION:()}
 
1917
        else:
 
1918
            found_parents = {}
 
1919
        search_keys = set((revision_id,) for revision_id in keys)
 
1920
        for index, key, value, refs in index.iter_entries(search_keys):
 
1921
            parents = refs[0]
 
1922
            if not parents:
 
1923
                parents = (_mod_revision.NULL_REVISION,)
 
1924
            else:
 
1925
                parents = tuple(parent[0] for parent in parents)
 
1926
            found_parents[key[0]] = parents
 
1927
        return found_parents
2298
1928
 
2299
1929
    def _make_parents_provider(self):
2300
1930
        return graph.CachingParentsProvider(self)
2301
1931
 
2302
1932
    def _refresh_data(self):
2303
 
        if not self.is_locked():
2304
 
            return
2305
 
        self._pack_collection.reload_pack_names()
 
1933
        if self._write_lock_count == 1 or (
 
1934
            self.control_files._lock_count == 1 and
 
1935
            self.control_files._lock_mode == 'r'):
 
1936
            # forget what names there are
 
1937
            self._pack_collection.reset()
 
1938
            # XXX: Better to do an in-memory merge when acquiring a new lock -
 
1939
            # factor out code from _save_pack_names.
 
1940
            self._pack_collection.ensure_loaded()
2306
1941
 
2307
1942
    def _start_write_group(self):
2308
1943
        self._pack_collection._start_write_group()
2309
1944
 
2310
1945
    def _commit_write_group(self):
2311
 
        hint = self._pack_collection._commit_write_group()
2312
 
        self.revisions._index._key_dependencies.clear()
2313
 
        return hint
2314
 
 
2315
 
    def suspend_write_group(self):
2316
 
        # XXX check self._write_group is self.get_transaction()?
2317
 
        tokens = self._pack_collection._suspend_write_group()
2318
 
        self.revisions._index._key_dependencies.clear()
2319
 
        self._write_group = None
2320
 
        return tokens
2321
 
 
2322
 
    def _resume_write_group(self, tokens):
2323
 
        self._start_write_group()
2324
 
        try:
2325
 
            self._pack_collection._resume_write_group(tokens)
2326
 
        except errors.UnresumableWriteGroup:
2327
 
            self._abort_write_group()
2328
 
            raise
2329
 
        for pack in self._pack_collection._resumed_packs:
2330
 
            self.revisions._index.scan_unvalidated_index(pack.revision_index)
 
1946
        return self._pack_collection._commit_write_group()
 
1947
 
 
1948
    def get_inventory_weave(self):
 
1949
        return self._inv_thunk.get_weave()
2331
1950
 
2332
1951
    def get_transaction(self):
2333
1952
        if self._write_lock_count:
2342
1961
        return self._write_lock_count
2343
1962
 
2344
1963
    def lock_write(self, token=None):
2345
 
        """Lock the repository for writes.
2346
 
 
2347
 
        :return: A bzrlib.repository.RepositoryWriteLockResult.
2348
 
        """
2349
 
        locked = self.is_locked()
2350
 
        if not self._write_lock_count and locked:
 
1964
        if not self._write_lock_count and self.is_locked():
2351
1965
            raise errors.ReadOnlyError(self)
2352
1966
        self._write_lock_count += 1
2353
1967
        if self._write_lock_count == 1:
 
1968
            from bzrlib import transactions
2354
1969
            self._transaction = transactions.WriteTransaction()
2355
 
        if not locked:
2356
 
            if 'relock' in debug.debug_flags and self._prev_lock == 'w':
2357
 
                note('%r was write locked again', self)
2358
 
            self._prev_lock = 'w'
2359
 
            for repo in self._fallback_repositories:
2360
 
                # Writes don't affect fallback repos
2361
 
                repo.lock_read()
2362
 
            self._refresh_data()
2363
 
        return RepositoryWriteLockResult(self.unlock, None)
 
1970
        self._refresh_data()
2364
1971
 
2365
1972
    def lock_read(self):
2366
 
        """Lock the repository for reads.
2367
 
 
2368
 
        :return: A bzrlib.lock.LogicalLockResult.
2369
 
        """
2370
 
        locked = self.is_locked()
2371
1973
        if self._write_lock_count:
2372
1974
            self._write_lock_count += 1
2373
1975
        else:
2374
1976
            self.control_files.lock_read()
2375
 
        if not locked:
2376
 
            if 'relock' in debug.debug_flags and self._prev_lock == 'r':
2377
 
                note('%r was read locked again', self)
2378
 
            self._prev_lock = 'r'
2379
 
            for repo in self._fallback_repositories:
2380
 
                repo.lock_read()
2381
 
            self._refresh_data()
2382
 
        return LogicalLockResult(self.unlock)
 
1977
        self._refresh_data()
2383
1978
 
2384
1979
    def leave_lock_in_place(self):
2385
1980
        # not supported - raise an error
2390
1985
        raise NotImplementedError(self.dont_leave_lock_in_place)
2391
1986
 
2392
1987
    @needs_write_lock
2393
 
    def pack(self, hint=None, clean_obsolete_packs=False):
 
1988
    def pack(self):
2394
1989
        """Compress the data within the repository.
2395
1990
 
2396
1991
        This will pack all the data to a single pack. In future it may
2397
1992
        recompress deltas or do other such expensive operations.
2398
1993
        """
2399
 
        self._pack_collection.pack(hint=hint, clean_obsolete_packs=clean_obsolete_packs)
 
1994
        self._pack_collection.pack()
2400
1995
 
2401
1996
    @needs_write_lock
2402
1997
    def reconcile(self, other=None, thorough=False):
2406
2001
        reconciler.reconcile()
2407
2002
        return reconciler
2408
2003
 
2409
 
    def _reconcile_pack(self, collection, packs, extension, revs, pb):
2410
 
        packer = ReconcilePacker(collection, packs, extension, revs)
2411
 
        return packer.pack(pb)
2412
 
 
2413
 
    @only_raises(errors.LockNotHeld, errors.LockBroken)
2414
2004
    def unlock(self):
2415
2005
        if self._write_lock_count == 1 and self._write_group is not None:
2416
2006
            self.abort_write_group()
2428
2018
        else:
2429
2019
            self.control_files.unlock()
2430
2020
 
2431
 
        if not self.is_locked():
2432
 
            for repo in self._fallback_repositories:
2433
 
                repo.unlock()
2434
 
 
2435
 
 
2436
 
class KnitPackStreamSource(StreamSource):
2437
 
    """A StreamSource used to transfer data between same-format KnitPack repos.
2438
 
 
2439
 
    This source assumes:
2440
 
        1) Same serialization format for all objects
2441
 
        2) Same root information
2442
 
        3) XML format inventories
2443
 
        4) Atomic inserts (so we can stream inventory texts before text
2444
 
           content)
2445
 
        5) No chk_bytes
2446
 
    """
2447
 
 
2448
 
    def __init__(self, from_repository, to_format):
2449
 
        super(KnitPackStreamSource, self).__init__(from_repository, to_format)
2450
 
        self._text_keys = None
2451
 
        self._text_fetch_order = 'unordered'
2452
 
 
2453
 
    def _get_filtered_inv_stream(self, revision_ids):
2454
 
        from_repo = self.from_repository
2455
 
        parent_ids = from_repo._find_parent_ids_of_revisions(revision_ids)
2456
 
        parent_keys = [(p,) for p in parent_ids]
2457
 
        find_text_keys = from_repo._find_text_key_references_from_xml_inventory_lines
2458
 
        parent_text_keys = set(find_text_keys(
2459
 
            from_repo._inventory_xml_lines_for_keys(parent_keys)))
2460
 
        content_text_keys = set()
2461
 
        knit = KnitVersionedFiles(None, None)
2462
 
        factory = KnitPlainFactory()
2463
 
        def find_text_keys_from_content(record):
2464
 
            if record.storage_kind not in ('knit-delta-gz', 'knit-ft-gz'):
2465
 
                raise ValueError("Unknown content storage kind for"
2466
 
                    " inventory text: %s" % (record.storage_kind,))
2467
 
            # It's a knit record, it has a _raw_record field (even if it was
2468
 
            # reconstituted from a network stream).
2469
 
            raw_data = record._raw_record
2470
 
            # read the entire thing
2471
 
            revision_id = record.key[-1]
2472
 
            content, _ = knit._parse_record(revision_id, raw_data)
2473
 
            if record.storage_kind == 'knit-delta-gz':
2474
 
                line_iterator = factory.get_linedelta_content(content)
2475
 
            elif record.storage_kind == 'knit-ft-gz':
2476
 
                line_iterator = factory.get_fulltext_content(content)
2477
 
            content_text_keys.update(find_text_keys(
2478
 
                [(line, revision_id) for line in line_iterator]))
2479
 
        revision_keys = [(r,) for r in revision_ids]
2480
 
        def _filtered_inv_stream():
2481
 
            source_vf = from_repo.inventories
2482
 
            stream = source_vf.get_record_stream(revision_keys,
2483
 
                                                 'unordered', False)
2484
 
            for record in stream:
2485
 
                if record.storage_kind == 'absent':
2486
 
                    raise errors.NoSuchRevision(from_repo, record.key)
2487
 
                find_text_keys_from_content(record)
2488
 
                yield record
2489
 
            self._text_keys = content_text_keys - parent_text_keys
2490
 
        return ('inventories', _filtered_inv_stream())
2491
 
 
2492
 
    def _get_text_stream(self):
2493
 
        # Note: We know we don't have to handle adding root keys, because both
2494
 
        # the source and target are the identical network name.
2495
 
        text_stream = self.from_repository.texts.get_record_stream(
2496
 
                        self._text_keys, self._text_fetch_order, False)
2497
 
        return ('texts', text_stream)
2498
 
 
2499
 
    def get_stream(self, search):
2500
 
        revision_ids = search.get_keys()
2501
 
        for stream_info in self._fetch_revision_texts(revision_ids):
2502
 
            yield stream_info
2503
 
        self._revision_keys = [(rev_id,) for rev_id in revision_ids]
2504
 
        yield self._get_filtered_inv_stream(revision_ids)
2505
 
        yield self._get_text_stream()
2506
 
 
2507
 
 
2508
2021
 
2509
2022
class RepositoryFormatPack(MetaDirRepositoryFormat):
2510
2023
    """Format logic for pack structured repositories.
2531
2044
    # Set this attribute in derived clases to control the _serializer that the
2532
2045
    # repository objects will have passed to their constructor.
2533
2046
    _serializer = None
2534
 
    # Packs are not confused by ghosts.
2535
 
    supports_ghosts = True
2536
 
    # External references are not supported in pack repositories yet.
2537
 
    supports_external_lookups = False
2538
 
    # Most pack formats do not use chk lookups.
2539
 
    supports_chks = False
2540
 
    # What index classes to use
2541
 
    index_builder_class = None
2542
 
    index_class = None
2543
 
    _fetch_uses_deltas = True
2544
 
    fast_deltas = False
 
2047
 
 
2048
    def _get_control_store(self, repo_transport, control_files):
 
2049
        """Return the control store for this repository."""
 
2050
        return VersionedFileStore(
 
2051
            repo_transport,
 
2052
            prefixed=False,
 
2053
            file_mode=control_files._file_mode,
 
2054
            versionedfile_class=knit.KnitVersionedFile,
 
2055
            versionedfile_kwargs={'factory': knit.KnitPlainFactory()},
 
2056
            )
 
2057
 
 
2058
    def _get_revision_store(self, repo_transport, control_files):
 
2059
        """See RepositoryFormat._get_revision_store()."""
 
2060
        versioned_file_store = VersionedFileStore(
 
2061
            repo_transport,
 
2062
            file_mode=control_files._file_mode,
 
2063
            prefixed=False,
 
2064
            precious=True,
 
2065
            versionedfile_class=knit.KnitVersionedFile,
 
2066
            versionedfile_kwargs={'delta': False,
 
2067
                                  'factory': knit.KnitPlainFactory(),
 
2068
                                 },
 
2069
            escaped=True,
 
2070
            )
 
2071
        return KnitRevisionStore(versioned_file_store)
 
2072
 
 
2073
    def _get_text_store(self, transport, control_files):
 
2074
        """See RepositoryFormat._get_text_store()."""
 
2075
        return self._get_versioned_file_store('knits',
 
2076
                                  transport,
 
2077
                                  control_files,
 
2078
                                  versionedfile_class=knit.KnitVersionedFile,
 
2079
                                  versionedfile_kwargs={
 
2080
                                      'create_parent_dir': True,
 
2081
                                      'delay_create': True,
 
2082
                                      'dir_mode': control_files._dir_mode,
 
2083
                                  },
 
2084
                                  escaped=True)
2545
2085
 
2546
2086
    def initialize(self, a_bzrdir, shared=False):
2547
2087
        """Create a pack based repository.
2553
2093
        """
2554
2094
        mutter('creating repository in %s.', a_bzrdir.transport.base)
2555
2095
        dirs = ['indices', 'obsolete_packs', 'packs', 'upload']
2556
 
        builder = self.index_builder_class()
 
2096
        builder = GraphIndexBuilder()
2557
2097
        files = [('pack-names', builder.finish())]
2558
2098
        utf8_files = [('format', self.get_format_string())]
2559
 
 
 
2099
        
2560
2100
        self._upload_blank_content(a_bzrdir, dirs, files, utf8_files, shared)
2561
 
        repository = self.open(a_bzrdir=a_bzrdir, _found=True)
2562
 
        self._run_post_repo_init_hooks(repository, a_bzrdir, shared)
2563
 
        return repository
 
2101
        return self.open(a_bzrdir=a_bzrdir, _found=True)
2564
2102
 
2565
2103
    def open(self, a_bzrdir, _found=False, _override_transport=None):
2566
2104
        """See RepositoryFormat.open().
2567
 
 
 
2105
        
2568
2106
        :param _override_transport: INTERNAL USE ONLY. Allows opening the
2569
2107
                                    repository at a slightly different url
2570
2108
                                    than normal. I.e. during 'upgrade'.
2571
2109
        """
2572
2110
        if not _found:
2573
2111
            format = RepositoryFormat.find_format(a_bzrdir)
 
2112
            assert format.__class__ ==  self.__class__
2574
2113
        if _override_transport is not None:
2575
2114
            repo_transport = _override_transport
2576
2115
        else:
2577
2116
            repo_transport = a_bzrdir.get_repository_transport(None)
2578
2117
        control_files = lockable_files.LockableFiles(repo_transport,
2579
2118
                                'lock', lockdir.LockDir)
 
2119
        text_store = self._get_text_store(repo_transport, control_files)
 
2120
        control_store = self._get_control_store(repo_transport, control_files)
 
2121
        _revision_store = self._get_revision_store(repo_transport, control_files)
2580
2122
        return self.repository_class(_format=self,
2581
2123
                              a_bzrdir=a_bzrdir,
2582
2124
                              control_files=control_files,
 
2125
                              _revision_store=_revision_store,
 
2126
                              control_store=control_store,
 
2127
                              text_store=text_store,
2583
2128
                              _commit_builder_class=self._commit_builder_class,
2584
2129
                              _serializer=self._serializer)
2585
2130
 
2592
2137
 
2593
2138
    repository_class = KnitPackRepository
2594
2139
    _commit_builder_class = PackCommitBuilder
2595
 
    @property
2596
 
    def _serializer(self):
2597
 
        return xml5.serializer_v5
2598
 
    # What index classes to use
2599
 
    index_builder_class = InMemoryGraphIndex
2600
 
    index_class = GraphIndex
 
2140
    _serializer = xml5.serializer_v5
2601
2141
 
2602
2142
    def _get_matching_bzrdir(self):
2603
2143
        return bzrdir.format_registry.make_bzrdir('pack-0.92')
2615
2155
        """See RepositoryFormat.get_format_description()."""
2616
2156
        return "Packs containing knits without subtree support"
2617
2157
 
 
2158
    def check_conversion_target(self, target_format):
 
2159
        pass
 
2160
 
2618
2161
 
2619
2162
class RepositoryFormatKnitPack3(RepositoryFormatPack):
2620
2163
    """A subtrees parameterized Pack repository.
2629
2172
    repository_class = KnitPackRepository
2630
2173
    _commit_builder_class = PackRootCommitBuilder
2631
2174
    rich_root_data = True
2632
 
    experimental = True
2633
2175
    supports_tree_reference = True
2634
 
    @property
2635
 
    def _serializer(self):
2636
 
        return xml7.serializer_v7
2637
 
    # What index classes to use
2638
 
    index_builder_class = InMemoryGraphIndex
2639
 
    index_class = GraphIndex
 
2176
    _serializer = xml7.serializer_v7
2640
2177
 
2641
2178
    def _get_matching_bzrdir(self):
2642
2179
        return bzrdir.format_registry.make_bzrdir(
2647
2184
 
2648
2185
    _matchingbzrdir = property(_get_matching_bzrdir, _ignore_setting_bzrdir)
2649
2186
 
 
2187
    def check_conversion_target(self, target_format):
 
2188
        if not target_format.rich_root_data:
 
2189
            raise errors.BadConversionTarget(
 
2190
                'Does not support rich root data.', target_format)
 
2191
        if not getattr(target_format, 'supports_tree_reference', False):
 
2192
            raise errors.BadConversionTarget(
 
2193
                'Does not support nested trees', target_format)
 
2194
            
2650
2195
    def get_format_string(self):
2651
2196
        """See RepositoryFormat.get_format_string()."""
2652
2197
        return "Bazaar pack repository format 1 with subtree support (needs bzr 0.92)\n"
2669
2214
    _commit_builder_class = PackRootCommitBuilder
2670
2215
    rich_root_data = True
2671
2216
    supports_tree_reference = False
2672
 
    @property
2673
 
    def _serializer(self):
2674
 
        return xml6.serializer_v6
2675
 
    # What index classes to use
2676
 
    index_builder_class = InMemoryGraphIndex
2677
 
    index_class = GraphIndex
 
2217
    _serializer = xml6.serializer_v6
2678
2218
 
2679
2219
    def _get_matching_bzrdir(self):
2680
2220
        return bzrdir.format_registry.make_bzrdir(
2685
2225
 
2686
2226
    _matchingbzrdir = property(_get_matching_bzrdir, _ignore_setting_bzrdir)
2687
2227
 
 
2228
    def check_conversion_target(self, target_format):
 
2229
        if not target_format.rich_root_data:
 
2230
            raise errors.BadConversionTarget(
 
2231
                'Does not support rich root data.', target_format)
 
2232
 
2688
2233
    def get_format_string(self):
2689
2234
        """See RepositoryFormat.get_format_string()."""
2690
2235
        return ("Bazaar pack repository format 1 with rich root"
2693
2238
    def get_format_description(self):
2694
2239
        """See RepositoryFormat.get_format_description()."""
2695
2240
        return "Packs containing knits with rich root support\n"
2696
 
 
2697
 
 
2698
 
class RepositoryFormatKnitPack5(RepositoryFormatPack):
2699
 
    """Repository that supports external references to allow stacking.
2700
 
 
2701
 
    New in release 1.6.
2702
 
 
2703
 
    Supports external lookups, which results in non-truncated ghosts after
2704
 
    reconcile compared to pack-0.92 formats.
2705
 
    """
2706
 
 
2707
 
    repository_class = KnitPackRepository
2708
 
    _commit_builder_class = PackCommitBuilder
2709
 
    supports_external_lookups = True
2710
 
    # What index classes to use
2711
 
    index_builder_class = InMemoryGraphIndex
2712
 
    index_class = GraphIndex
2713
 
 
2714
 
    @property
2715
 
    def _serializer(self):
2716
 
        return xml5.serializer_v5
2717
 
 
2718
 
    def _get_matching_bzrdir(self):
2719
 
        return bzrdir.format_registry.make_bzrdir('1.6')
2720
 
 
2721
 
    def _ignore_setting_bzrdir(self, format):
2722
 
        pass
2723
 
 
2724
 
    _matchingbzrdir = property(_get_matching_bzrdir, _ignore_setting_bzrdir)
2725
 
 
2726
 
    def get_format_string(self):
2727
 
        """See RepositoryFormat.get_format_string()."""
2728
 
        return "Bazaar RepositoryFormatKnitPack5 (bzr 1.6)\n"
2729
 
 
2730
 
    def get_format_description(self):
2731
 
        """See RepositoryFormat.get_format_description()."""
2732
 
        return "Packs 5 (adds stacking support, requires bzr 1.6)"
2733
 
 
2734
 
 
2735
 
class RepositoryFormatKnitPack5RichRoot(RepositoryFormatPack):
2736
 
    """A repository with rich roots and stacking.
2737
 
 
2738
 
    New in release 1.6.1.
2739
 
 
2740
 
    Supports stacking on other repositories, allowing data to be accessed
2741
 
    without being stored locally.
2742
 
    """
2743
 
 
2744
 
    repository_class = KnitPackRepository
2745
 
    _commit_builder_class = PackRootCommitBuilder
2746
 
    rich_root_data = True
2747
 
    supports_tree_reference = False # no subtrees
2748
 
    supports_external_lookups = True
2749
 
    # What index classes to use
2750
 
    index_builder_class = InMemoryGraphIndex
2751
 
    index_class = GraphIndex
2752
 
 
2753
 
    @property
2754
 
    def _serializer(self):
2755
 
        return xml6.serializer_v6
2756
 
 
2757
 
    def _get_matching_bzrdir(self):
2758
 
        return bzrdir.format_registry.make_bzrdir(
2759
 
            '1.6.1-rich-root')
2760
 
 
2761
 
    def _ignore_setting_bzrdir(self, format):
2762
 
        pass
2763
 
 
2764
 
    _matchingbzrdir = property(_get_matching_bzrdir, _ignore_setting_bzrdir)
2765
 
 
2766
 
    def get_format_string(self):
2767
 
        """See RepositoryFormat.get_format_string()."""
2768
 
        return "Bazaar RepositoryFormatKnitPack5RichRoot (bzr 1.6.1)\n"
2769
 
 
2770
 
    def get_format_description(self):
2771
 
        return "Packs 5 rich-root (adds stacking support, requires bzr 1.6.1)"
2772
 
 
2773
 
 
2774
 
class RepositoryFormatKnitPack5RichRootBroken(RepositoryFormatPack):
2775
 
    """A repository with rich roots and external references.
2776
 
 
2777
 
    New in release 1.6.
2778
 
 
2779
 
    Supports external lookups, which results in non-truncated ghosts after
2780
 
    reconcile compared to pack-0.92 formats.
2781
 
 
2782
 
    This format was deprecated because the serializer it uses accidentally
2783
 
    supported subtrees, when the format was not intended to. This meant that
2784
 
    someone could accidentally fetch from an incorrect repository.
2785
 
    """
2786
 
 
2787
 
    repository_class = KnitPackRepository
2788
 
    _commit_builder_class = PackRootCommitBuilder
2789
 
    rich_root_data = True
2790
 
    supports_tree_reference = False # no subtrees
2791
 
 
2792
 
    supports_external_lookups = True
2793
 
    # What index classes to use
2794
 
    index_builder_class = InMemoryGraphIndex
2795
 
    index_class = GraphIndex
2796
 
 
2797
 
    @property
2798
 
    def _serializer(self):
2799
 
        return xml7.serializer_v7
2800
 
 
2801
 
    def _get_matching_bzrdir(self):
2802
 
        matching = bzrdir.format_registry.make_bzrdir(
2803
 
            '1.6.1-rich-root')
2804
 
        matching.repository_format = self
2805
 
        return matching
2806
 
 
2807
 
    def _ignore_setting_bzrdir(self, format):
2808
 
        pass
2809
 
 
2810
 
    _matchingbzrdir = property(_get_matching_bzrdir, _ignore_setting_bzrdir)
2811
 
 
2812
 
    def get_format_string(self):
2813
 
        """See RepositoryFormat.get_format_string()."""
2814
 
        return "Bazaar RepositoryFormatKnitPack5RichRoot (bzr 1.6)\n"
2815
 
 
2816
 
    def get_format_description(self):
2817
 
        return ("Packs 5 rich-root (adds stacking support, requires bzr 1.6)"
2818
 
                " (deprecated)")
2819
 
 
2820
 
 
2821
 
class RepositoryFormatKnitPack6(RepositoryFormatPack):
2822
 
    """A repository with stacking and btree indexes,
2823
 
    without rich roots or subtrees.
2824
 
 
2825
 
    This is equivalent to pack-1.6 with B+Tree indices.
2826
 
    """
2827
 
 
2828
 
    repository_class = KnitPackRepository
2829
 
    _commit_builder_class = PackCommitBuilder
2830
 
    supports_external_lookups = True
2831
 
    # What index classes to use
2832
 
    index_builder_class = BTreeBuilder
2833
 
    index_class = BTreeGraphIndex
2834
 
 
2835
 
    @property
2836
 
    def _serializer(self):
2837
 
        return xml5.serializer_v5
2838
 
 
2839
 
    def _get_matching_bzrdir(self):
2840
 
        return bzrdir.format_registry.make_bzrdir('1.9')
2841
 
 
2842
 
    def _ignore_setting_bzrdir(self, format):
2843
 
        pass
2844
 
 
2845
 
    _matchingbzrdir = property(_get_matching_bzrdir, _ignore_setting_bzrdir)
2846
 
 
2847
 
    def get_format_string(self):
2848
 
        """See RepositoryFormat.get_format_string()."""
2849
 
        return "Bazaar RepositoryFormatKnitPack6 (bzr 1.9)\n"
2850
 
 
2851
 
    def get_format_description(self):
2852
 
        """See RepositoryFormat.get_format_description()."""
2853
 
        return "Packs 6 (uses btree indexes, requires bzr 1.9)"
2854
 
 
2855
 
 
2856
 
class RepositoryFormatKnitPack6RichRoot(RepositoryFormatPack):
2857
 
    """A repository with rich roots, no subtrees, stacking and btree indexes.
2858
 
 
2859
 
    1.6-rich-root with B+Tree indices.
2860
 
    """
2861
 
 
2862
 
    repository_class = KnitPackRepository
2863
 
    _commit_builder_class = PackRootCommitBuilder
2864
 
    rich_root_data = True
2865
 
    supports_tree_reference = False # no subtrees
2866
 
    supports_external_lookups = True
2867
 
    # What index classes to use
2868
 
    index_builder_class = BTreeBuilder
2869
 
    index_class = BTreeGraphIndex
2870
 
 
2871
 
    @property
2872
 
    def _serializer(self):
2873
 
        return xml6.serializer_v6
2874
 
 
2875
 
    def _get_matching_bzrdir(self):
2876
 
        return bzrdir.format_registry.make_bzrdir(
2877
 
            '1.9-rich-root')
2878
 
 
2879
 
    def _ignore_setting_bzrdir(self, format):
2880
 
        pass
2881
 
 
2882
 
    _matchingbzrdir = property(_get_matching_bzrdir, _ignore_setting_bzrdir)
2883
 
 
2884
 
    def get_format_string(self):
2885
 
        """See RepositoryFormat.get_format_string()."""
2886
 
        return "Bazaar RepositoryFormatKnitPack6RichRoot (bzr 1.9)\n"
2887
 
 
2888
 
    def get_format_description(self):
2889
 
        return "Packs 6 rich-root (uses btree indexes, requires bzr 1.9)"
2890
 
 
2891
 
 
2892
 
class RepositoryFormatPackDevelopment2Subtree(RepositoryFormatPack):
2893
 
    """A subtrees development repository.
2894
 
 
2895
 
    This format should be retained until the second release after bzr 1.7.
2896
 
 
2897
 
    1.6.1-subtree[as it might have been] with B+Tree indices.
2898
 
 
2899
 
    This is [now] retained until we have a CHK based subtree format in
2900
 
    development.
2901
 
    """
2902
 
 
2903
 
    repository_class = KnitPackRepository
2904
 
    _commit_builder_class = PackRootCommitBuilder
2905
 
    rich_root_data = True
2906
 
    experimental = True
2907
 
    supports_tree_reference = True
2908
 
    supports_external_lookups = True
2909
 
    # What index classes to use
2910
 
    index_builder_class = BTreeBuilder
2911
 
    index_class = BTreeGraphIndex
2912
 
 
2913
 
    @property
2914
 
    def _serializer(self):
2915
 
        return xml7.serializer_v7
2916
 
 
2917
 
    def _get_matching_bzrdir(self):
2918
 
        return bzrdir.format_registry.make_bzrdir(
2919
 
            'development-subtree')
2920
 
 
2921
 
    def _ignore_setting_bzrdir(self, format):
2922
 
        pass
2923
 
 
2924
 
    _matchingbzrdir = property(_get_matching_bzrdir, _ignore_setting_bzrdir)
2925
 
 
2926
 
    def get_format_string(self):
2927
 
        """See RepositoryFormat.get_format_string()."""
2928
 
        return ("Bazaar development format 2 with subtree support "
2929
 
            "(needs bzr.dev from before 1.8)\n")
2930
 
 
2931
 
    def get_format_description(self):
2932
 
        """See RepositoryFormat.get_format_description()."""
2933
 
        return ("Development repository format, currently the same as "
2934
 
            "1.6.1-subtree with B+Tree indices.\n")
2935