~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/repofmt/pack_repo.py

  • Committer: Matthew Revell
  • Date: 2008-03-05 14:11:35 UTC
  • mto: This revision was merged to the branch mainline in revision 3263.
  • Revision ID: matthew.revell@canonical.com-20080305141135-96jf8j24b9crru6v
Added an admin-guide directory to doc/en and outlined headings for the admin guide in index.txt

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