← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/fetch.py

  • Committer: Martin Pool
  • Date: 2009-03-03 03:01:49 UTC
  • mfrom: (4070 +trunk)
  • mto: This revision was merged to the branch mainline in revision 4073.
  • Revision ID: mbp@sourcefrog.net-20090303030149-8p8o8hszdtqa7w8f
merge trunk

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/fetch.py
36
36
import bzrlib.errors as errors
37
37
from bzrlib.errors import InstallFailed
38
38
from bzrlib.progress import ProgressPhase
39
 
from bzrlib.revision import is_null, NULL_REVISION
40
 
from bzrlib.symbol_versioning import (deprecated_function,
41
 
        deprecated_method,
42
 
        )
 
39
from bzrlib.revision import NULL_REVISION
43
40
from bzrlib.tsort import topo_sort
44
41
from bzrlib.trace import mutter
45
42
import bzrlib.ui
46
 
from bzrlib.versionedfile import filter_absent, FulltextContentFactory
 
43
from bzrlib.versionedfile import FulltextContentFactory
47
44
 
48
45
# TODO: Avoid repeatedly opening weaves so many times.
49
46
 
69
66
    last_revision
70
67
        if set, try to limit to the data this revision references.
71
68
 
72
 
    after running:
73
 
    count_copied -- number of revisions copied
74
 
 
75
69
    This should not be used directly, it's essential a object to encapsulate
76
70
    the logic in InterRepository.fetch().
77
71
    """
85
79
            exists to facilitate a hack done in InterPackRepo.fetch.  We would
86
80
            like to remove this parameter.
87
81
        """
88
 
        # result variables.
89
 
        self.failed_revisions = []
90
 
        self.count_copied = 0
91
82
        if to_repository.has_same_location(from_repository):
92
83
            # repository.fetch should be taking care of this case.
93
84
            raise errors.BzrError('RepoFetcher run '
95
86
                    '%r and %r' % (to_repository, from_repository))
96
87
        self.to_repository = to_repository
97
88
        self.from_repository = from_repository
 
89
        self.sink = to_repository._get_sink()
98
90
        # must not mutate self._last_revision as its potentially a shared instance
99
91
        self._last_revision = last_revision
100
92
        self.find_ghosts = find_ghosts
106
98
            self.nested_pb = None
107
99
        self.from_repository.lock_read()
108
100
        try:
109
 
            self.to_repository.lock_write()
110
101
            try:
111
 
                self.to_repository.start_write_group()
112
 
                try:
113
 
                    self.__fetch()
114
 
                except:
115
 
                    self.to_repository.abort_write_group(suppress_errors=True)
116
 
                    raise
117
 
                else:
118
 
                    self.to_repository.commit_write_group()
 
102
                self.__fetch()
119
103
            finally:
120
 
                try:
121
 
                    if self.nested_pb is not None:
122
 
                        self.nested_pb.finished()
123
 
                finally:
124
 
                    self.to_repository.unlock()
 
104
                if self.nested_pb is not None:
 
105
                    self.nested_pb.finished()
125
106
        finally:
126
107
            self.from_repository.unlock()
127
108
 
128
109
    def __fetch(self):
129
110
        """Primary worker function.
130
111
 
131
 
        This initialises all the needed variables, and then fetches the 
 
112
        This initialises all the needed variables, and then fetches the
132
113
        requested revisions, finally clearing the progress bar.
133
114
        """
 
115
        # Roughly this is what we're aiming for fetch to become:
 
116
        #
 
117
        # missing = self.sink.insert_stream(self.source.get_stream(search))
 
118
        # if missing:
 
119
        #     missing = self.sink.insert_stream(self.source.get_items(missing))
 
120
        # assert not missing
134
121
        self.count_total = 0
135
122
        self.file_ids_names = {}
136
123
        pp = ProgressPhase('Transferring', 4, self.pb)
139
126
            search = self._revids_to_fetch()
140
127
            if search is None:
141
128
                return
142
 
            if getattr(self, '_fetch_everything_for_search', None) is not None:
143
 
                self._fetch_everything_for_search(search, pp)
144
 
            else:
145
 
                # backward compatibility
146
 
                self._fetch_everything_for_revisions(search.get_keys, pp)
 
129
            self._fetch_everything_for_search(search, pp)
147
130
        finally:
148
131
            self.pb.clear()
149
132
 
157
140
        # item_keys_introduced_by should have a richer API than it does at the
158
141
        # moment, so that it can feed the progress information back to this
159
142
        # function?
160
 
        phase = 'file'
161
 
        pb = bzrlib.ui.ui_factory.nested_progress_bar()
 
143
        if (self.from_repository._format.rich_root_data and
 
144
            not self.to_repository._format.rich_root_data):
 
145
            raise errors.IncompatibleRepositories(
 
146
                self.from_repository, self.to_repository,
 
147
                "different rich-root support")
 
148
        self.pb = bzrlib.ui.ui_factory.nested_progress_bar()
162
149
        try:
163
 
            revs = search.get_keys()
164
 
            graph = self.from_repository.get_graph()
165
 
            revs = list(graph.iter_topo_order(revs))
166
 
            data_to_fetch = self.from_repository.item_keys_introduced_by(revs,
167
 
                                                                         pb)
168
 
            text_keys = []
169
 
            for knit_kind, file_id, revisions in data_to_fetch:
170
 
                if knit_kind != phase:
171
 
                    phase = knit_kind
172
 
                    # Make a new progress bar for this phase
173
 
                    pb.finished()
174
 
                    pp.next_phase()
175
 
                    pb = bzrlib.ui.ui_factory.nested_progress_bar()
176
 
                if knit_kind == "file":
177
 
                    # Accumulate file texts
178
 
                    text_keys.extend([(file_id, revision) for revision in
179
 
                        revisions])
180
 
                elif knit_kind == "inventory":
181
 
                    # Now copy the file texts.
182
 
                    to_texts = self.to_repository.texts
183
 
                    from_texts = self.from_repository.texts
184
 
                    to_texts.insert_record_stream(from_texts.get_record_stream(
185
 
                        text_keys, self.to_repository._fetch_order,
186
 
                        not self.to_repository._fetch_uses_deltas))
187
 
                    # Cause an error if a text occurs after we have done the
188
 
                    # copy.
189
 
                    text_keys = None
190
 
                    # Before we process the inventory we generate the root
191
 
                    # texts (if necessary) so that the inventories references
192
 
                    # will be valid.
193
 
                    self._generate_root_texts(revs)
194
 
                    # NB: This currently reopens the inventory weave in source;
195
 
                    # using a single stream interface instead would avoid this.
196
 
                    self._fetch_inventory_weave(revs, pb)
197
 
                elif knit_kind == "signatures":
198
 
                    # Nothing to do here; this will be taken care of when
199
 
                    # _fetch_revision_texts happens.
200
 
                    pass
201
 
                elif knit_kind == "revisions":
202
 
                    self._fetch_revision_texts(revs, pb)
203
 
                else:
204
 
                    raise AssertionError("Unknown knit kind %r" % knit_kind)
205
 
            if self.to_repository._fetch_reconcile:
206
 
                self.to_repository.reconcile()
 
150
            source = self.from_repository._get_source(
 
151
                self.to_repository._format)
 
152
            stream = source.get_stream(search)
 
153
            from_format = self.from_repository._format
 
154
            resume_tokens, missing_keys = self.sink.insert_stream(
 
155
                stream, from_format, [])
 
156
            if missing_keys:
 
157
                stream = source.get_stream_for_missing_keys(missing_keys)
 
158
                resume_tokens, missing_keys = self.sink.insert_stream(
 
159
                    stream, from_format, resume_tokens)
 
160
            if missing_keys:
 
161
                raise AssertionError(
 
162
                    "second push failed to complete a fetch %r." % (
 
163
                        missing_keys,))
 
164
            if resume_tokens:
 
165
                raise AssertionError(
 
166
                    "second push failed to commit the fetch %r." % (
 
167
                        resume_tokens,))
 
168
            self.sink.finished()
207
169
        finally:
208
 
            if pb is not None:
209
 
                pb.finished()
210
 
        self.count_copied += len(revs)
211
 
        
 
170
            if self.pb is not None:
 
171
                self.pb.finished()
 
172
 
212
173
    def _revids_to_fetch(self):
213
174
        """Determines the exact revisions needed from self.from_repository to
214
175
        install self._last_revision in self.to_repository.
229
190
        except errors.NoSuchRevision, e:
230
191
            raise InstallFailed([self._last_revision])
231
192
 
232
 
    def _fetch_inventory_weave(self, revs, pb):
233
 
        pb.update("fetch inventory", 0, 2)
234
 
        to_weave = self.to_repository.inventories
235
 
        # just merge, this is optimisable and its means we don't
236
 
        # copy unreferenced data such as not-needed inventories.
237
 
        pb.update("fetch inventory", 1, 3)
238
 
        from_weave = self.from_repository.inventories
239
 
        pb.update("fetch inventory", 2, 3)
240
 
        # we fetch only the referenced inventories because we do not
241
 
        # know for unselected inventories whether all their required
242
 
        # texts are present in the other repository - it could be
243
 
        # corrupt.
244
 
        to_weave.insert_record_stream(from_weave.get_record_stream(
245
 
            [(rev_id,) for rev_id in revs],
246
 
            self.to_repository._fetch_order,
247
 
            not self.to_repository._fetch_uses_deltas))
248
 
 
249
 
    def _fetch_revision_texts(self, revs, pb):
250
 
        # fetch signatures first and then the revision texts
251
 
        # may need to be a InterRevisionStore call here.
252
 
        to_sf = self.to_repository.signatures
253
 
        from_sf = self.from_repository.signatures
254
 
        # A missing signature is just skipped.
255
 
        to_sf.insert_record_stream(filter_absent(from_sf.get_record_stream(
256
 
            [(rev_id,) for rev_id in revs],
257
 
            self.to_repository._fetch_order,
258
 
            not self.to_repository._fetch_uses_deltas)))
259
 
        self._fetch_just_revision_texts(revs)
260
 
 
261
 
    def _fetch_just_revision_texts(self, version_ids):
262
 
        to_rf = self.to_repository.revisions
263
 
        from_rf = self.from_repository.revisions
264
 
        # If a revision has a delta, this is actually expanded inside the
265
 
        # insert_record_stream code now, which is an alternate fix for
266
 
        # bug #261339
267
 
        to_rf.insert_record_stream(from_rf.get_record_stream(
268
 
            [(rev_id,) for rev_id in version_ids],
269
 
            self.to_repository._fetch_order,
270
 
            not self.to_repository._fetch_uses_deltas))
271
 
 
272
 
    def _generate_root_texts(self, revs):
273
 
        """This will be called by __fetch between fetching weave texts and
274
 
        fetching the inventory weave.
275
 
 
276
 
        Subclasses should override this if they need to generate root texts
277
 
        after fetching weave texts.
278
 
        """
279
 
        pass
280
 
 
281
193
 
282
194
class Inter1and2Helper(object):
283
195
    """Helper for operations that convert data from model 1 and 2
284
 
    
 
196
 
285
197
    This is for use by fetchers and converters.
286
198
    """
287
199
 
288
 
    def __init__(self, source, target):
 
200
    def __init__(self, source):
289
201
        """Constructor.
290
202
 
291
203
        :param source: The repository data comes from
292
 
        :param target: The repository data goes to
293
204
        """
294
205
        self.source = source
295
 
        self.target = target
296
206
 
297
207
    def iter_rev_trees(self, revs):
298
208
        """Iterate through RevisionTrees efficiently.
338
248
 
339
249
        :param revs: the revisions to include
340
250
        """
341
 
        to_texts = self.target.texts
342
251
        graph = self.source.get_graph()
343
252
        parent_map = graph.get_parent_map(revs)
344
253
        rev_order = topo_sort(parent_map)
368
277
                    if parent != NULL_REVISION and
369
278
                        rev_id_to_root_id.get(parent, root_id) == root_id)
370
279
                yield FulltextContentFactory(key, parent_keys, None, '')
371
 
        to_texts.insert_record_stream(yield_roots())
372
 
 
373
 
    def regenerate_inventory(self, revs):
374
 
        """Generate a new inventory versionedfile in target, convertin data.
375
 
        
376
 
        The inventory is retrieved from the source, (deserializing it), and
377
 
        stored in the target (reserializing it in a different format).
378
 
        :param revs: The revisions to include
379
 
        """
380
 
        for tree in self.iter_rev_trees(revs):
381
 
            parents = tree.get_parent_ids()
382
 
            self.target.add_inventory(tree.get_revision_id(), tree.inventory,
383
 
                                      parents)
384
 
 
385
 
    def fetch_revisions(self, revision_ids):
386
 
        # TODO: should this batch them up rather than requesting 10,000
387
 
        #       revisions at once?
388
 
        for revision in self.source.get_revisions(revision_ids):
389
 
            self.target.add_revision(revision.revision_id, revision)
390
 
 
391
 
 
392
 
class Model1toKnit2Fetcher(RepoFetcher):
393
 
    """Fetch from a Model1 repository into a Knit2 repository
394
 
    """
395
 
    def __init__(self, to_repository, from_repository, last_revision=None,
396
 
                 pb=None, find_ghosts=True):
397
 
        self.helper = Inter1and2Helper(from_repository, to_repository)
398
 
        RepoFetcher.__init__(self, to_repository, from_repository,
399
 
            last_revision, pb, find_ghosts)
400
 
 
401
 
    def _generate_root_texts(self, revs):
402
 
        self.helper.generate_root_texts(revs)
403
 
 
404
 
    def _fetch_inventory_weave(self, revs, pb):
405
 
        self.helper.regenerate_inventory(revs)
406
 
 
407
 
    def _fetch_revision_texts(self, revs, pb):
408
 
        """Fetch revision object texts"""
409
 
        count = 0
410
 
        total = len(revs)
411
 
        for rev in revs:
412
 
            pb.update('copying revisions', count, total)
413
 
            try:
414
 
                sig_text = self.from_repository.get_signature_text(rev)
415
 
                self.to_repository.add_signature_text(rev, sig_text)
416
 
            except errors.NoSuchRevision:
417
 
                # not signed.
418
 
                pass
419
 
            self._copy_revision(rev)
420
 
            count += 1
421
 
 
422
 
    def _copy_revision(self, rev):
423
 
        self.helper.fetch_revisions([rev])
424
 
 
425
 
 
426
 
class Knit1to2Fetcher(RepoFetcher):
427
 
    """Fetch from a Knit1 repository into a Knit2 repository"""
428
 
 
429
 
    def __init__(self, to_repository, from_repository, last_revision=None,
430
 
                 pb=None, find_ghosts=True):
431
 
        self.helper = Inter1and2Helper(from_repository, to_repository)
432
 
        RepoFetcher.__init__(self, to_repository, from_repository,
433
 
            last_revision, pb, find_ghosts)
434
 
 
435
 
    def _generate_root_texts(self, revs):
436
 
        self.helper.generate_root_texts(revs)
437
 
 
438
 
    def _fetch_inventory_weave(self, revs, pb):
439
 
        self.helper.regenerate_inventory(revs)
440
 
 
441
 
    def _fetch_just_revision_texts(self, version_ids):
442
 
        self.helper.fetch_revisions(version_ids)
 
280
        return [('texts', yield_roots())]