← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/fetch.py

  • Committer: Jelmer Vernooij
  • Date: 2009-02-23 20:55:58 UTC
  • mfrom: (4034 +trunk)
  • mto: This revision was merged to the branch mainline in revision 4053.
  • Revision ID: jelmer@samba.org-20090223205558-1cx2k4w1zgs8r5qa
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
95
92
                    '%r and %r' % (to_repository, from_repository))
96
93
        self.to_repository = to_repository
97
94
        self.from_repository = from_repository
 
95
        self.sink = to_repository._get_sink()
98
96
        # must not mutate self._last_revision as its potentially a shared instance
99
97
        self._last_revision = last_revision
100
98
        self.find_ghosts = find_ghosts
128
126
    def __fetch(self):
129
127
        """Primary worker function.
130
128
 
131
 
        This initialises all the needed variables, and then fetches the 
 
129
        This initialises all the needed variables, and then fetches the
132
130
        requested revisions, finally clearing the progress bar.
133
131
        """
 
132
        # Roughly this is what we're aiming for fetch to become:
 
133
        #
 
134
        # missing = self.sink.insert_stream(self.source.get_stream(search))
 
135
        # if missing:
 
136
        #     missing = self.sink.insert_stream(self.source.get_items(missing))
 
137
        # assert not missing
134
138
        self.count_total = 0
135
139
        self.file_ids_names = {}
136
140
        pp = ProgressPhase('Transferring', 4, self.pb)
139
143
            search = self._revids_to_fetch()
140
144
            if search is None:
141
145
                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)
 
146
            self._fetch_everything_for_search(search, pp)
147
147
        finally:
148
148
            self.pb.clear()
149
149
 
157
157
        # item_keys_introduced_by should have a richer API than it does at the
158
158
        # moment, so that it can feed the progress information back to this
159
159
        # function?
 
160
        self.pb = bzrlib.ui.ui_factory.nested_progress_bar()
 
161
        try:
 
162
            from_format = self.from_repository._format
 
163
            stream = self.get_stream(search, pp)
 
164
            missing_keys = self.sink.insert_stream(stream, from_format)
 
165
            if missing_keys:
 
166
                stream = self.get_stream_for_missing_keys(missing_keys)
 
167
                missing_keys = self.sink.insert_stream(stream, from_format)
 
168
            if missing_keys:
 
169
                raise AssertionError(
 
170
                    "second push failed to complete a fetch %r." % (
 
171
                        missing_keys,))
 
172
            self.sink.finished()
 
173
        finally:
 
174
            if self.pb is not None:
 
175
                self.pb.finished()
 
176
 
 
177
    def get_stream(self, search, pp):
160
178
        phase = 'file'
161
 
        pb = bzrlib.ui.ui_factory.nested_progress_bar()
162
 
        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()
207
 
        finally:
208
 
            if pb is not None:
209
 
                pb.finished()
 
179
        revs = search.get_keys()
 
180
        graph = self.from_repository.get_graph()
 
181
        revs = list(graph.iter_topo_order(revs))
 
182
        data_to_fetch = self.from_repository.item_keys_introduced_by(
 
183
            revs, self.pb)
 
184
        text_keys = []
 
185
        for knit_kind, file_id, revisions in data_to_fetch:
 
186
            if knit_kind != phase:
 
187
                phase = knit_kind
 
188
                # Make a new progress bar for this phase
 
189
                self.pb.finished()
 
190
                pp.next_phase()
 
191
                self.pb = bzrlib.ui.ui_factory.nested_progress_bar()
 
192
            if knit_kind == "file":
 
193
                # Accumulate file texts
 
194
                text_keys.extend([(file_id, revision) for revision in
 
195
                    revisions])
 
196
            elif knit_kind == "inventory":
 
197
                # Now copy the file texts.
 
198
                to_texts = self.to_repository.texts
 
199
                from_texts = self.from_repository.texts
 
200
                yield ('texts', from_texts.get_record_stream(
 
201
                    text_keys, self.to_repository._fetch_order,
 
202
                    not self.to_repository._fetch_uses_deltas))
 
203
                # Cause an error if a text occurs after we have done the
 
204
                # copy.
 
205
                text_keys = None
 
206
                # Before we process the inventory we generate the root
 
207
                # texts (if necessary) so that the inventories references
 
208
                # will be valid.
 
209
                for _ in self._generate_root_texts(revs):
 
210
                    yield _
 
211
                # NB: This currently reopens the inventory weave in source;
 
212
                # using a single stream interface instead would avoid this.
 
213
                self.pb.update("fetch inventory", 0, 1)
 
214
                from_weave = self.from_repository.inventories
 
215
                # we fetch only the referenced inventories because we do not
 
216
                # know for unselected inventories whether all their required
 
217
                # texts are present in the other repository - it could be
 
218
                # corrupt.
 
219
                yield ('inventories', from_weave.get_record_stream(
 
220
                    [(rev_id,) for rev_id in revs],
 
221
                    self.inventory_fetch_order(),
 
222
                    not self.delta_on_metadata()))
 
223
            elif knit_kind == "signatures":
 
224
                # Nothing to do here; this will be taken care of when
 
225
                # _fetch_revision_texts happens.
 
226
                pass
 
227
            elif knit_kind == "revisions":
 
228
                for _ in self._fetch_revision_texts(revs, self.pb):
 
229
                    yield _
 
230
            else:
 
231
                raise AssertionError("Unknown knit kind %r" % knit_kind)
210
232
        self.count_copied += len(revs)
211
 
        
 
233
 
 
234
    def get_stream_for_missing_keys(self, missing_keys):
 
235
        # missing keys can only occur when we are byte copying and not
 
236
        # translating (because translation means we don't send
 
237
        # unreconstructable deltas ever).
 
238
        keys = {}
 
239
        keys['texts'] = set()
 
240
        keys['revisions'] = set()
 
241
        keys['inventories'] = set()
 
242
        keys['signatures'] = set()
 
243
        for key in missing_keys:
 
244
            keys[key[0]].add(key[1:])
 
245
        if len(keys['revisions']):
 
246
            # If we allowed copying revisions at this point, we could end up
 
247
            # copying a revision without copying its required texts: a
 
248
            # violation of the requirements for repository integrity.
 
249
            raise AssertionError(
 
250
                'cannot copy revisions to fill in missing deltas %s' % (
 
251
                    keys['revisions'],))
 
252
        for substream_kind, keys in keys.iteritems():
 
253
            vf = getattr(self.from_repository, substream_kind)
 
254
            # Ask for full texts always so that we don't need more round trips
 
255
            # after this stream.
 
256
            stream = vf.get_record_stream(keys,
 
257
                self.to_repository._fetch_order, True)
 
258
            yield substream_kind, stream
 
259
 
212
260
    def _revids_to_fetch(self):
213
261
        """Determines the exact revisions needed from self.from_repository to
214
262
        install self._last_revision in self.to_repository.
229
277
        except errors.NoSuchRevision, e:
230
278
            raise InstallFailed([self._last_revision])
231
279
 
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
280
    def _fetch_revision_texts(self, revs, pb):
250
281
        # fetch signatures first and then the revision texts
251
282
        # may need to be a InterRevisionStore call here.
252
 
        to_sf = self.to_repository.signatures
253
283
        from_sf = self.from_repository.signatures
254
284
        # 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],
 
285
        keys = [(rev_id,) for rev_id in revs]
 
286
        signatures = filter_absent(from_sf.get_record_stream(
 
287
            keys,
257
288
            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
 
289
            not self.to_repository._fetch_uses_deltas))
264
290
        # If a revision has a delta, this is actually expanded inside the
265
291
        # insert_record_stream code now, which is an alternate fix for
266
292
        # bug #261339
267
 
        to_rf.insert_record_stream(from_rf.get_record_stream(
268
 
            [(rev_id,) for rev_id in version_ids],
 
293
        from_rf = self.from_repository.revisions
 
294
        revisions = from_rf.get_record_stream(
 
295
            keys,
269
296
            self.to_repository._fetch_order,
270
 
            not self.to_repository._fetch_uses_deltas))
 
297
            not self.delta_on_metadata())
 
298
        return [('signatures', signatures), ('revisions', revisions)]
271
299
 
272
300
    def _generate_root_texts(self, revs):
273
301
        """This will be called by __fetch between fetching weave texts and
276
304
        Subclasses should override this if they need to generate root texts
277
305
        after fetching weave texts.
278
306
        """
279
 
        pass
 
307
        return []
 
308
 
 
309
    def inventory_fetch_order(self):
 
310
        return self.to_repository._fetch_order
 
311
 
 
312
    def delta_on_metadata(self):
 
313
        src_serializer = self.from_repository._format._serializer
 
314
        target_serializer = self.to_repository._format._serializer
 
315
        return (self.to_repository._fetch_uses_deltas and
 
316
            src_serializer == target_serializer)
280
317
 
281
318
 
282
319
class Inter1and2Helper(object):
283
320
    """Helper for operations that convert data from model 1 and 2
284
 
    
 
321
 
285
322
    This is for use by fetchers and converters.
286
323
    """
287
324
 
288
 
    def __init__(self, source, target):
 
325
    def __init__(self, source):
289
326
        """Constructor.
290
327
 
291
328
        :param source: The repository data comes from
292
 
        :param target: The repository data goes to
293
329
        """
294
330
        self.source = source
295
 
        self.target = target
296
331
 
297
332
    def iter_rev_trees(self, revs):
298
333
        """Iterate through RevisionTrees efficiently.
338
373
 
339
374
        :param revs: the revisions to include
340
375
        """
341
 
        to_texts = self.target.texts
342
376
        graph = self.source.get_graph()
343
377
        parent_map = graph.get_parent_map(revs)
344
378
        rev_order = topo_sort(parent_map)
368
402
                    if parent != NULL_REVISION and
369
403
                        rev_id_to_root_id.get(parent, root_id) == root_id)
370
404
                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)
 
405
        return [('texts', yield_roots())]
390
406
 
391
407
 
392
408
class Model1toKnit2Fetcher(RepoFetcher):
394
410
    """
395
411
    def __init__(self, to_repository, from_repository, last_revision=None,
396
412
                 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)
 
413
        self.helper = Inter1and2Helper(from_repository)
 
414
        RepoFetcher.__init__(self, to_repository, from_repository,
 
415
            last_revision, pb, find_ghosts)
 
416
 
 
417
    def _generate_root_texts(self, revs):
 
418
        return self.helper.generate_root_texts(revs)
 
419
 
 
420
    def inventory_fetch_order(self):
 
421
        return 'topological'
 
422
 
 
423
Knit1to2Fetcher = Model1toKnit2Fetcher