~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/repofmt/pack_repo.py

  • Committer: Vincent Ladeuil
  • Date: 2007-12-21 12:20:33 UTC
  • mto: (3146.3.1 179368) (3156.2.1 trunk)
  • mto: This revision was merged to the branch mainline in revision 3158.
  • Revision ID: v.ladeuil+lp@free.fr-20071221122033-42bc21re0zj4kqbg
Merge back test_http_implementations.pc into test_http.py.

* bzrlib/tests/test_http.py: 
Merge test_http_implementations.py now that we have rewritten
load_tests. That should reduce the noise in the final proposed
patch.

* bzrlib/tests/http_server.py:
(TestingHTTPRequestHandler.log_message): Ghaaa, don't over spell-check.

Show diffs side-by-side

added added

removed removed

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