/brz/remove-bazaar

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

« back to all changes in this revision

Viewing changes to bzrlib/repofmt/pack_repo.py

  • Committer: Andrew Bennetts
  • Date: 2008-03-12 20:13:07 UTC
  • mfrom: (3267 +trunk)
  • mto: This revision was merged to the branch mainline in revision 3756.
  • Revision ID: andrew.bennetts@canonical.com-20080312201307-ngd5bynt2nvhnlb7
Merge from bzr.dev.

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