/brz/remove-bazaar

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

« back to all changes in this revision

Viewing changes to bzrlib/repofmt/pack_repo.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2007-12-14 08:47:17 UTC
  • mfrom: (3053.4.10 integrate-1.0)
  • Revision ID: pqm@pqm.ubuntu.com-20071214084717-xu119wq2x8l2w8lj
Merge 1.0final back to trunk and bump to 1.1dev

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