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