/brz/remove-bazaar

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

« back to all changes in this revision

Viewing changes to bzrlib/fetch.py

  • Committer: Andrew Bennetts
  • Date: 2009-03-10 04:57:28 UTC
  • mfrom: (4102 +trunk)
  • mto: This revision was merged to the branch mainline in revision 4105.
  • Revision ID: andrew.bennetts@canonical.com-20090310045728-ty6iymi18247i2pl
Merge bzr.dev

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
    """
78
72
 
79
 
    def __init__(self, to_repository, from_repository, last_revision=None, pb=None,
80
 
        find_ghosts=True):
 
73
    def __init__(self, to_repository, from_repository, last_revision=None,
 
74
        pb=None, find_ghosts=True, fetch_spec=None):
81
75
        """Create a repo fetcher.
82
76
 
83
77
        :param find_ghosts: If True search the entire history for ghosts.
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
 
92
        self._fetch_spec = fetch_spec
100
93
        self.find_ghosts = find_ghosts
101
94
        if pb is None:
102
95
            self.pb = bzrlib.ui.ui_factory.nested_progress_bar()
106
99
            self.nested_pb = None
107
100
        self.from_repository.lock_read()
108
101
        try:
109
 
            self.to_repository.lock_write()
110
102
            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()
 
103
                self.__fetch()
119
104
            finally:
120
 
                try:
121
 
                    if self.nested_pb is not None:
122
 
                        self.nested_pb.finished()
123
 
                finally:
124
 
                    self.to_repository.unlock()
 
105
                if self.nested_pb is not None:
 
106
                    self.nested_pb.finished()
125
107
        finally:
126
108
            self.from_repository.unlock()
127
109
 
128
110
    def __fetch(self):
129
111
        """Primary worker function.
130
112
 
131
 
        This initialises all the needed variables, and then fetches the 
 
113
        This initialises all the needed variables, and then fetches the
132
114
        requested revisions, finally clearing the progress bar.
133
115
        """
 
116
        # Roughly this is what we're aiming for fetch to become:
 
117
        #
 
118
        # missing = self.sink.insert_stream(self.source.get_stream(search))
 
119
        # if missing:
 
120
        #     missing = self.sink.insert_stream(self.source.get_items(missing))
 
121
        # assert not missing
134
122
        self.count_total = 0
135
123
        self.file_ids_names = {}
136
124
        pp = ProgressPhase('Transferring', 4, self.pb)
139
127
            search = self._revids_to_fetch()
140
128
            if search is None:
141
129
                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)
 
130
            self._fetch_everything_for_search(search, pp)
147
131
        finally:
148
132
            self.pb.clear()
149
133
 
157
141
        # item_keys_introduced_by should have a richer API than it does at the
158
142
        # moment, so that it can feed the progress information back to this
159
143
        # function?
160
 
        phase = 'file'
161
 
        pb = bzrlib.ui.ui_factory.nested_progress_bar()
 
144
        if (self.from_repository._format.rich_root_data and
 
145
            not self.to_repository._format.rich_root_data):
 
146
            raise errors.IncompatibleRepositories(
 
147
                self.from_repository, self.to_repository,
 
148
                "different rich-root support")
 
149
        self.pb = bzrlib.ui.ui_factory.nested_progress_bar()
162
150
        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()
 
151
            source = self.from_repository._get_source(
 
152
                self.to_repository._format)
 
153
            stream = source.get_stream(search)
 
154
            from_format = self.from_repository._format
 
155
            resume_tokens, missing_keys = self.sink.insert_stream(
 
156
                stream, from_format, [])
 
157
            if missing_keys:
 
158
                stream = source.get_stream_for_missing_keys(missing_keys)
 
159
                resume_tokens, missing_keys = self.sink.insert_stream(
 
160
                    stream, from_format, resume_tokens)
 
161
            if missing_keys:
 
162
                raise AssertionError(
 
163
                    "second push failed to complete a fetch %r." % (
 
164
                        missing_keys,))
 
165
            if resume_tokens:
 
166
                raise AssertionError(
 
167
                    "second push failed to commit the fetch %r." % (
 
168
                        resume_tokens,))
 
169
            self.sink.finished()
207
170
        finally:
208
 
            if pb is not None:
209
 
                pb.finished()
210
 
        self.count_copied += len(revs)
211
 
        
 
171
            if self.pb is not None:
 
172
                self.pb.finished()
 
173
 
212
174
    def _revids_to_fetch(self):
213
175
        """Determines the exact revisions needed from self.from_repository to
214
176
        install self._last_revision in self.to_repository.
215
177
 
216
178
        If no revisions need to be fetched, then this just returns None.
217
179
        """
 
180
        if self._fetch_spec is not None:
 
181
            return self._fetch_spec
218
182
        mutter('fetch up to rev {%s}', self._last_revision)
219
183
        if self._last_revision is NULL_REVISION:
220
184
            # explicit limit of no revisions needed
229
193
        except errors.NoSuchRevision, e:
230
194
            raise InstallFailed([self._last_revision])
231
195
 
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
196
 
282
197
class Inter1and2Helper(object):
283
198
    """Helper for operations that convert data from model 1 and 2
284
 
    
 
199
 
285
200
    This is for use by fetchers and converters.
286
201
    """
287
202
 
288
 
    def __init__(self, source, target):
 
203
    def __init__(self, source):
289
204
        """Constructor.
290
205
 
291
206
        :param source: The repository data comes from
292
 
        :param target: The repository data goes to
293
207
        """
294
208
        self.source = source
295
 
        self.target = target
296
209
 
297
210
    def iter_rev_trees(self, revs):
298
211
        """Iterate through RevisionTrees efficiently.
338
251
 
339
252
        :param revs: the revisions to include
340
253
        """
341
 
        to_texts = self.target.texts
342
254
        graph = self.source.get_graph()
343
255
        parent_map = graph.get_parent_map(revs)
344
256
        rev_order = topo_sort(parent_map)
368
280
                    if parent != NULL_REVISION and
369
281
                        rev_id_to_root_id.get(parent, root_id) == root_id)
370
282
                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)
 
283
        return [('texts', yield_roots())]