← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/remote.py

  • Committer: Martin Pool
  • Date: 2010-09-09 00:32:58 UTC
  • mfrom: (5348.1.2 deprecation)
  • mto: This revision was merged to the branch mainline in revision 5415.
  • Revision ID: mbp@sourcefrog.net-20100909003258-ioyugglvmu32yvff
merge news

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/remote.py
1
 
# Copyright (C) 2006, 2007, 2008, 2009 Canonical Ltd
 
1
# Copyright (C) 2006-2010 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
21
21
    branch,
22
22
    bzrdir,
23
23
    config,
 
24
    controldir,
24
25
    debug,
25
26
    errors,
26
27
    graph,
27
28
    lock,
28
29
    lockdir,
29
30
    repository,
 
31
    repository as _mod_repository,
30
32
    revision,
31
33
    revision as _mod_revision,
 
34
    static_tuple,
32
35
    symbol_versioning,
33
36
)
34
 
from bzrlib.branch import BranchReferenceFormat
 
37
from bzrlib.branch import BranchReferenceFormat, BranchWriteLockResult
35
38
from bzrlib.bzrdir import BzrDir, RemoteBzrDirFormat
36
 
from bzrlib.decorators import needs_read_lock, needs_write_lock
 
39
from bzrlib.decorators import needs_read_lock, needs_write_lock, only_raises
37
40
from bzrlib.errors import (
38
41
    NoSuchRevision,
39
42
    SmartProtocolError,
41
44
from bzrlib.lockable_files import LockableFiles
42
45
from bzrlib.smart import client, vfs, repository as smart_repo
43
46
from bzrlib.revision import ensure_null, NULL_REVISION
 
47
from bzrlib.repository import RepositoryWriteLockResult
44
48
from bzrlib.trace import mutter, note, warning
45
49
 
46
50
 
89
93
class RemoteBzrDir(BzrDir, _RpcHelper):
90
94
    """Control directory on a remote server, accessed via bzr:// or similar."""
91
95
 
92
 
    def __init__(self, transport, format, _client=None):
 
96
    def __init__(self, transport, format, _client=None, _force_probe=False):
93
97
        """Construct a RemoteBzrDir.
94
98
 
95
99
        :param _client: Private parameter for testing. Disables probing and the
99
103
        # this object holds a delegated bzrdir that uses file-level operations
100
104
        # to talk to the other side
101
105
        self._real_bzrdir = None
 
106
        self._has_working_tree = None
102
107
        # 1-shot cache for the call pattern 'create_branch; open_branch' - see
103
108
        # create_branch for details.
104
109
        self._next_open_branch_result = None
108
113
            self._client = client._SmartClient(medium)
109
114
        else:
110
115
            self._client = _client
111
 
            return
112
 
 
 
116
            if not _force_probe:
 
117
                return
 
118
 
 
119
        self._probe_bzrdir()
 
120
 
 
121
    def __repr__(self):
 
122
        return '%s(%r)' % (self.__class__.__name__, self._client)
 
123
 
 
124
    def _probe_bzrdir(self):
 
125
        medium = self._client._medium
113
126
        path = self._path_for_remote_call(self._client)
 
127
        if medium._is_remote_before((2, 1)):
 
128
            self._rpc_open(path)
 
129
            return
 
130
        try:
 
131
            self._rpc_open_2_1(path)
 
132
            return
 
133
        except errors.UnknownSmartMethod:
 
134
            medium._remember_remote_is_before((2, 1))
 
135
            self._rpc_open(path)
 
136
 
 
137
    def _rpc_open_2_1(self, path):
 
138
        response = self._call('BzrDir.open_2.1', path)
 
139
        if response == ('no',):
 
140
            raise errors.NotBranchError(path=self.root_transport.base)
 
141
        elif response[0] == 'yes':
 
142
            if response[1] == 'yes':
 
143
                self._has_working_tree = True
 
144
            elif response[1] == 'no':
 
145
                self._has_working_tree = False
 
146
            else:
 
147
                raise errors.UnexpectedSmartServerResponse(response)
 
148
        else:
 
149
            raise errors.UnexpectedSmartServerResponse(response)
 
150
 
 
151
    def _rpc_open(self, path):
114
152
        response = self._call('BzrDir.open', path)
115
153
        if response not in [('yes',), ('no',)]:
116
154
            raise errors.UnexpectedSmartServerResponse(response)
117
155
        if response == ('no',):
118
 
            raise errors.NotBranchError(path=transport.base)
 
156
            raise errors.NotBranchError(path=self.root_transport.base)
119
157
 
120
158
    def _ensure_real(self):
121
159
        """Ensure that there is a _real_bzrdir set.
123
161
        Used before calls to self._real_bzrdir.
124
162
        """
125
163
        if not self._real_bzrdir:
 
164
            if 'hpssvfs' in debug.debug_flags:
 
165
                import traceback
 
166
                warning('VFS BzrDir access triggered\n%s',
 
167
                    ''.join(traceback.format_stack()))
126
168
            self._real_bzrdir = BzrDir.open_from_transport(
127
169
                self.root_transport, _server_formats=False)
128
170
            self._format._network_name = \
204
246
        self._ensure_real()
205
247
        self._real_bzrdir.destroy_repository()
206
248
 
207
 
    def create_branch(self):
 
249
    def create_branch(self, name=None):
208
250
        # as per meta1 formats - just delegate to the format object which may
209
251
        # be parameterised.
210
 
        real_branch = self._format.get_branch_format().initialize(self)
 
252
        real_branch = self._format.get_branch_format().initialize(self,
 
253
            name=name)
211
254
        if not isinstance(real_branch, RemoteBranch):
212
 
            result = RemoteBranch(self, self.find_repository(), real_branch)
 
255
            result = RemoteBranch(self, self.find_repository(), real_branch,
 
256
                                  name=name)
213
257
        else:
214
258
            result = real_branch
215
259
        # BzrDir.clone_on_transport() uses the result of create_branch but does
221
265
        self._next_open_branch_result = result
222
266
        return result
223
267
 
224
 
    def destroy_branch(self):
 
268
    def destroy_branch(self, name=None):
225
269
        """See BzrDir.destroy_branch"""
226
270
        self._ensure_real()
227
 
        self._real_bzrdir.destroy_branch()
 
271
        self._real_bzrdir.destroy_branch(name=name)
228
272
        self._next_open_branch_result = None
229
273
 
230
274
    def create_workingtree(self, revision_id=None, from_branch=None):
231
275
        raise errors.NotLocalUrl(self.transport.base)
232
276
 
233
 
    def find_branch_format(self):
 
277
    def find_branch_format(self, name=None):
234
278
        """Find the branch 'format' for this bzrdir.
235
279
 
236
280
        This might be a synthetic object for e.g. RemoteBranch and SVN.
237
281
        """
238
 
        b = self.open_branch()
 
282
        b = self.open_branch(name=name)
239
283
        return b._format
240
284
 
241
 
    def get_branch_reference(self):
 
285
    def get_branch_reference(self, name=None):
242
286
        """See BzrDir.get_branch_reference()."""
 
287
        if name is not None:
 
288
            # XXX JRV20100304: Support opening colocated branches
 
289
            raise errors.NoColocatedBranchSupport(self)
243
290
        response = self._get_branch_reference()
244
291
        if response[0] == 'ref':
245
292
            return response[1]
249
296
    def _get_branch_reference(self):
250
297
        path = self._path_for_remote_call(self._client)
251
298
        medium = self._client._medium
252
 
        if not medium._is_remote_before((1, 13)):
 
299
        candidate_calls = [
 
300
            ('BzrDir.open_branchV3', (2, 1)),
 
301
            ('BzrDir.open_branchV2', (1, 13)),
 
302
            ('BzrDir.open_branch', None),
 
303
            ]
 
304
        for verb, required_version in candidate_calls:
 
305
            if required_version and medium._is_remote_before(required_version):
 
306
                continue
253
307
            try:
254
 
                response = self._call('BzrDir.open_branchV2', path)
255
 
                if response[0] not in ('ref', 'branch'):
256
 
                    raise errors.UnexpectedSmartServerResponse(response)
257
 
                return response
 
308
                response = self._call(verb, path)
258
309
            except errors.UnknownSmartMethod:
259
 
                medium._remember_remote_is_before((1, 13))
260
 
        response = self._call('BzrDir.open_branch', path)
261
 
        if response[0] != 'ok':
 
310
                if required_version is None:
 
311
                    raise
 
312
                medium._remember_remote_is_before(required_version)
 
313
            else:
 
314
                break
 
315
        if verb == 'BzrDir.open_branch':
 
316
            if response[0] != 'ok':
 
317
                raise errors.UnexpectedSmartServerResponse(response)
 
318
            if response[1] != '':
 
319
                return ('ref', response[1])
 
320
            else:
 
321
                return ('branch', '')
 
322
        if response[0] not in ('ref', 'branch'):
262
323
            raise errors.UnexpectedSmartServerResponse(response)
263
 
        if response[1] != '':
264
 
            return ('ref', response[1])
265
 
        else:
266
 
            return ('branch', '')
 
324
        return response
267
325
 
268
 
    def _get_tree_branch(self):
 
326
    def _get_tree_branch(self, name=None):
269
327
        """See BzrDir._get_tree_branch()."""
270
 
        return None, self.open_branch()
 
328
        return None, self.open_branch(name=name)
271
329
 
272
 
    def open_branch(self, _unsupported=False, ignore_fallbacks=False):
273
 
        if _unsupported:
 
330
    def open_branch(self, name=None, unsupported=False,
 
331
                    ignore_fallbacks=False):
 
332
        if unsupported:
274
333
            raise NotImplementedError('unsupported flag support not implemented yet.')
275
334
        if self._next_open_branch_result is not None:
276
335
            # See create_branch for details.
281
340
        if response[0] == 'ref':
282
341
            # a branch reference, use the existing BranchReference logic.
283
342
            format = BranchReferenceFormat()
284
 
            return format.open(self, _found=True, location=response[1],
285
 
                ignore_fallbacks=ignore_fallbacks)
 
343
            return format.open(self, name=name, _found=True,
 
344
                location=response[1], ignore_fallbacks=ignore_fallbacks)
286
345
        branch_format_name = response[1]
287
346
        if not branch_format_name:
288
347
            branch_format_name = None
289
348
        format = RemoteBranchFormat(network_name=branch_format_name)
290
349
        return RemoteBranch(self, self.find_repository(), format=format,
291
 
            setup_stacking=not ignore_fallbacks)
 
350
            setup_stacking=not ignore_fallbacks, name=name)
292
351
 
293
352
    def _open_repo_v1(self, path):
294
353
        verb = 'BzrDir.find_repository'
355
414
        else:
356
415
            raise errors.NoRepositoryPresent(self)
357
416
 
 
417
    def has_workingtree(self):
 
418
        if self._has_working_tree is None:
 
419
            self._ensure_real()
 
420
            self._has_working_tree = self._real_bzrdir.has_workingtree()
 
421
        return self._has_working_tree
 
422
 
358
423
    def open_workingtree(self, recommend_upgrade=True):
359
 
        self._ensure_real()
360
 
        if self._real_bzrdir.has_workingtree():
 
424
        if self.has_workingtree():
361
425
            raise errors.NotLocalUrl(self.root_transport)
362
426
        else:
363
427
            raise errors.NoWorkingTree(self.root_transport.base)
366
430
        """Return the path to be used for this bzrdir in a remote call."""
367
431
        return client.remote_path_from_transport(self.root_transport)
368
432
 
369
 
    def get_branch_transport(self, branch_format):
 
433
    def get_branch_transport(self, branch_format, name=None):
370
434
        self._ensure_real()
371
 
        return self._real_bzrdir.get_branch_transport(branch_format)
 
435
        return self._real_bzrdir.get_branch_transport(branch_format, name=name)
372
436
 
373
437
    def get_repository_transport(self, repository_format):
374
438
        self._ensure_real()
426
490
        self._custom_format = None
427
491
        self._network_name = None
428
492
        self._creating_bzrdir = None
 
493
        self._supports_chks = None
429
494
        self._supports_external_lookups = None
430
495
        self._supports_tree_reference = None
431
496
        self._rich_root_data = None
432
497
 
 
498
    def __repr__(self):
 
499
        return "%s(_network_name=%r)" % (self.__class__.__name__,
 
500
            self._network_name)
 
501
 
433
502
    @property
434
503
    def fast_deltas(self):
435
504
        self._ensure_real()
443
512
        return self._rich_root_data
444
513
 
445
514
    @property
 
515
    def supports_chks(self):
 
516
        if self._supports_chks is None:
 
517
            self._ensure_real()
 
518
            self._supports_chks = self._custom_format.supports_chks
 
519
        return self._supports_chks
 
520
 
 
521
    @property
446
522
    def supports_external_lookups(self):
447
523
        if self._supports_external_lookups is None:
448
524
            self._ensure_real()
549
625
        return self._custom_format._fetch_reconcile
550
626
 
551
627
    def get_format_description(self):
552
 
        return 'bzr remote repository'
 
628
        self._ensure_real()
 
629
        return 'Remote: ' + self._custom_format.get_format_description()
553
630
 
554
631
    def __eq__(self, other):
555
632
        return self.__class__ is other.__class__
556
633
 
557
 
    def check_conversion_target(self, target_format):
558
 
        if self.rich_root_data and not target_format.rich_root_data:
559
 
            raise errors.BadConversionTarget(
560
 
                'Does not support rich root data.', target_format)
561
 
        if (self.supports_tree_reference and
562
 
            not getattr(target_format, 'supports_tree_reference', False)):
563
 
            raise errors.BadConversionTarget(
564
 
                'Does not support nested trees', target_format)
565
 
 
566
634
    def network_name(self):
567
635
        if self._network_name:
568
636
            return self._network_name
580
648
        return self._custom_format._serializer
581
649
 
582
650
 
583
 
class RemoteRepository(_RpcHelper):
 
651
class RemoteRepository(_RpcHelper, lock._RelockDebugMixin,
 
652
    controldir.ControlComponent):
584
653
    """Repository accessed over rpc.
585
654
 
586
655
    For the moment most operations are performed using local transport-backed
629
698
        # Additional places to query for data.
630
699
        self._fallback_repositories = []
631
700
 
 
701
    @property
 
702
    def user_transport(self):
 
703
        return self.bzrdir.user_transport
 
704
 
 
705
    @property
 
706
    def control_transport(self):
 
707
        # XXX: Normally you shouldn't directly get at the remote repository
 
708
        # transport, but I'm not sure it's worth making this method
 
709
        # optional -- mbp 2010-04-21
 
710
        return self.bzrdir.get_repository_transport(None)
 
711
        
632
712
    def __str__(self):
633
713
        return "%s(%s)" % (self.__class__.__name__, self.base)
634
714
 
820
900
    def _has_same_fallbacks(self, other_repo):
821
901
        """Returns true if the repositories have the same fallbacks."""
822
902
        # XXX: copied from Repository; it should be unified into a base class
823
 
        # <https://bugs.edge.launchpad.net/bzr/+bug/401622>
 
903
        # <https://bugs.launchpad.net/bzr/+bug/401622>
824
904
        my_fb = self._fallback_repositories
825
905
        other_fb = other_repo._fallback_repositories
826
906
        if len(my_fb) != len(other_fb):
842
922
        parents_provider = self._make_parents_provider(other_repository)
843
923
        return graph.Graph(parents_provider)
844
924
 
 
925
    @needs_read_lock
 
926
    def get_known_graph_ancestry(self, revision_ids):
 
927
        """Return the known graph for a set of revision ids and their ancestors.
 
928
        """
 
929
        st = static_tuple.StaticTuple
 
930
        revision_keys = [st(r_id).intern() for r_id in revision_ids]
 
931
        known_graph = self.revisions.get_known_graph_ancestry(revision_keys)
 
932
        return graph.GraphThunkIdsToKeys(known_graph)
 
933
 
845
934
    def gather_stats(self, revid=None, committers=None):
846
935
        """See Repository.gather_stats()."""
847
936
        path = self.bzrdir._path_for_remote_call(self._client)
907
996
    def is_write_locked(self):
908
997
        return self._lock_mode == 'w'
909
998
 
 
999
    def _warn_if_deprecated(self, branch=None):
 
1000
        # If we have a real repository, the check will be done there, if we
 
1001
        # don't the check will be done remotely.
 
1002
        pass
 
1003
 
910
1004
    def lock_read(self):
 
1005
        """Lock the repository for read operations.
 
1006
 
 
1007
        :return: A bzrlib.lock.LogicalLockResult.
 
1008
        """
911
1009
        # wrong eventually - want a local lock cache context
912
1010
        if not self._lock_mode:
 
1011
            self._note_lock('r')
913
1012
            self._lock_mode = 'r'
914
1013
            self._lock_count = 1
915
1014
            self._unstacked_provider.enable_cache(cache_misses=True)
919
1018
                repo.lock_read()
920
1019
        else:
921
1020
            self._lock_count += 1
 
1021
        return lock.LogicalLockResult(self.unlock)
922
1022
 
923
1023
    def _remote_lock_write(self, token):
924
1024
        path = self.bzrdir._path_for_remote_call(self._client)
935
1035
 
936
1036
    def lock_write(self, token=None, _skip_rpc=False):
937
1037
        if not self._lock_mode:
 
1038
            self._note_lock('w')
938
1039
            if _skip_rpc:
939
1040
                if self._lock_token is not None:
940
1041
                    if token != self._lock_token:
963
1064
            raise errors.ReadOnlyError(self)
964
1065
        else:
965
1066
            self._lock_count += 1
966
 
        return self._lock_token or None
 
1067
        return RepositoryWriteLockResult(self.unlock, self._lock_token or None)
967
1068
 
968
1069
    def leave_lock_in_place(self):
969
1070
        if not self._lock_token:
1043
1144
        else:
1044
1145
            raise errors.UnexpectedSmartServerResponse(response)
1045
1146
 
 
1147
    @only_raises(errors.LockNotHeld, errors.LockBroken)
1046
1148
    def unlock(self):
1047
1149
        if not self._lock_count:
1048
1150
            return lock.cant_unlock_not_held(self)
1148
1250
            # state, so always add a lock here. If a caller passes us a locked
1149
1251
            # repository, they are responsible for unlocking it later.
1150
1252
            repository.lock_read()
 
1253
        self._check_fallback_repository(repository)
1151
1254
        self._fallback_repositories.append(repository)
1152
1255
        # If self._real_repository was parameterised already (e.g. because a
1153
1256
        # _real_branch had its get_stacked_on_url method called), then the
1154
1257
        # repository to be added may already be in the _real_repositories list.
1155
1258
        if self._real_repository is not None:
1156
 
            fallback_locations = [repo.bzrdir.root_transport.base for repo in
 
1259
            fallback_locations = [repo.user_url for repo in
1157
1260
                self._real_repository._fallback_repositories]
1158
 
            if repository.bzrdir.root_transport.base not in fallback_locations:
 
1261
            if repository.user_url not in fallback_locations:
1159
1262
                self._real_repository.add_fallback_repository(repository)
1160
1263
 
 
1264
    def _check_fallback_repository(self, repository):
 
1265
        """Check that this repository can fallback to repository safely.
 
1266
 
 
1267
        Raise an error if not.
 
1268
 
 
1269
        :param repository: A repository to fallback to.
 
1270
        """
 
1271
        return _mod_repository.InterRepository._assert_same_model(
 
1272
            self, repository)
 
1273
 
1161
1274
    def add_inventory(self, revid, inv, parents):
1162
1275
        self._ensure_real()
1163
1276
        return self._real_repository.add_inventory(revid, inv, parents)
1164
1277
 
1165
1278
    def add_inventory_by_delta(self, basis_revision_id, delta, new_revision_id,
1166
 
                               parents):
 
1279
            parents, basis_inv=None, propagate_caches=False):
1167
1280
        self._ensure_real()
1168
1281
        return self._real_repository.add_inventory_by_delta(basis_revision_id,
1169
 
            delta, new_revision_id, parents)
 
1282
            delta, new_revision_id, parents, basis_inv=basis_inv,
 
1283
            propagate_caches=propagate_caches)
1170
1284
 
1171
1285
    def add_revision(self, rev_id, rev, inv=None, config=None):
1172
1286
        self._ensure_real()
1178
1292
        self._ensure_real()
1179
1293
        return self._real_repository.get_inventory(revision_id)
1180
1294
 
1181
 
    def iter_inventories(self, revision_ids):
 
1295
    def iter_inventories(self, revision_ids, ordering=None):
1182
1296
        self._ensure_real()
1183
 
        return self._real_repository.iter_inventories(revision_ids)
 
1297
        return self._real_repository.iter_inventories(revision_ids, ordering)
1184
1298
 
1185
1299
    @needs_read_lock
1186
1300
    def get_revision(self, revision_id):
1202
1316
        return self._real_repository.make_working_trees()
1203
1317
 
1204
1318
    def refresh_data(self):
1205
 
        """Re-read any data needed to to synchronise with disk.
 
1319
        """Re-read any data needed to synchronise with disk.
1206
1320
 
1207
1321
        This method is intended to be called after another repository instance
1208
1322
        (such as one used by a smart server) has inserted data into the
1209
 
        repository. It may not be called during a write group, but may be
1210
 
        called at any other time.
 
1323
        repository. On all repositories this will work outside of write groups.
 
1324
        Some repository formats (pack and newer for bzrlib native formats)
 
1325
        support refresh_data inside write groups. If called inside a write
 
1326
        group on a repository that does not support refreshing in a write group
 
1327
        IsInWriteGroupError will be raised.
1211
1328
        """
1212
 
        if self.is_in_write_group():
1213
 
            raise errors.InternalBzrError(
1214
 
                "May not refresh_data while in a write group.")
1215
1329
        if self._real_repository is not None:
1216
1330
            self._real_repository.refresh_data()
1217
1331
 
1431
1545
        return self._real_repository.get_signature_text(revision_id)
1432
1546
 
1433
1547
    @needs_read_lock
1434
 
    def get_inventory_xml(self, revision_id):
1435
 
        self._ensure_real()
1436
 
        return self._real_repository.get_inventory_xml(revision_id)
1437
 
 
1438
 
    def deserialise_inventory(self, revision_id, xml):
1439
 
        self._ensure_real()
1440
 
        return self._real_repository.deserialise_inventory(revision_id, xml)
 
1548
    def _get_inventory_xml(self, revision_id):
 
1549
        self._ensure_real()
 
1550
        return self._real_repository._get_inventory_xml(revision_id)
1441
1551
 
1442
1552
    def reconcile(self, other=None, thorough=False):
1443
1553
        self._ensure_real()
1519
1629
        return self._real_repository.inventories
1520
1630
 
1521
1631
    @needs_write_lock
1522
 
    def pack(self, hint=None):
 
1632
    def pack(self, hint=None, clean_obsolete_packs=False):
1523
1633
        """Compress the data within the repository.
1524
1634
 
1525
1635
        This is not currently implemented within the smart server.
1526
1636
        """
1527
1637
        self._ensure_real()
1528
 
        return self._real_repository.pack(hint=hint)
 
1638
        return self._real_repository.pack(hint=hint, clean_obsolete_packs=clean_obsolete_packs)
1529
1639
 
1530
1640
    @property
1531
1641
    def revisions(self):
1682
1792
    def insert_stream(self, stream, src_format, resume_tokens):
1683
1793
        target = self.target_repo
1684
1794
        target._unstacked_provider.missing_keys.clear()
 
1795
        candidate_calls = [('Repository.insert_stream_1.19', (1, 19))]
1685
1796
        if target._lock_token:
1686
 
            verb = 'Repository.insert_stream_locked'
1687
 
            extra_args = (target._lock_token or '',)
1688
 
            required_version = (1, 14)
 
1797
            candidate_calls.append(('Repository.insert_stream_locked', (1, 14)))
 
1798
            lock_args = (target._lock_token or '',)
1689
1799
        else:
1690
 
            verb = 'Repository.insert_stream'
1691
 
            extra_args = ()
1692
 
            required_version = (1, 13)
 
1800
            candidate_calls.append(('Repository.insert_stream', (1, 13)))
 
1801
            lock_args = ()
1693
1802
        client = target._client
1694
1803
        medium = client._medium
1695
 
        if medium._is_remote_before(required_version):
1696
 
            # No possible way this can work.
1697
 
            return self._insert_real(stream, src_format, resume_tokens)
1698
1804
        path = target.bzrdir._path_for_remote_call(client)
1699
 
        if not resume_tokens:
1700
 
            # XXX: Ugly but important for correctness, *will* be fixed during
1701
 
            # 1.13 cycle. Pushing a stream that is interrupted results in a
1702
 
            # fallback to the _real_repositories sink *with a partial stream*.
1703
 
            # Thats bad because we insert less data than bzr expected. To avoid
1704
 
            # this we do a trial push to make sure the verb is accessible, and
1705
 
            # do not fallback when actually pushing the stream. A cleanup patch
1706
 
            # is going to look at rewinding/restarting the stream/partial
1707
 
            # buffering etc.
 
1805
        # Probe for the verb to use with an empty stream before sending the
 
1806
        # real stream to it.  We do this both to avoid the risk of sending a
 
1807
        # large request that is then rejected, and because we don't want to
 
1808
        # implement a way to buffer, rewind, or restart the stream.
 
1809
        found_verb = False
 
1810
        for verb, required_version in candidate_calls:
 
1811
            if medium._is_remote_before(required_version):
 
1812
                continue
 
1813
            if resume_tokens:
 
1814
                # We've already done the probing (and set _is_remote_before) on
 
1815
                # a previous insert.
 
1816
                found_verb = True
 
1817
                break
1708
1818
            byte_stream = smart_repo._stream_to_byte_stream([], src_format)
1709
1819
            try:
1710
1820
                response = client.call_with_body_stream(
1711
 
                    (verb, path, '') + extra_args, byte_stream)
 
1821
                    (verb, path, '') + lock_args, byte_stream)
1712
1822
            except errors.UnknownSmartMethod:
1713
1823
                medium._remember_remote_is_before(required_version)
1714
 
                return self._insert_real(stream, src_format, resume_tokens)
 
1824
            else:
 
1825
                found_verb = True
 
1826
                break
 
1827
        if not found_verb:
 
1828
            # Have to use VFS.
 
1829
            return self._insert_real(stream, src_format, resume_tokens)
 
1830
        self._last_inv_record = None
 
1831
        self._last_substream = None
 
1832
        if required_version < (1, 19):
 
1833
            # Remote side doesn't support inventory deltas.  Wrap the stream to
 
1834
            # make sure we don't send any.  If the stream contains inventory
 
1835
            # deltas we'll interrupt the smart insert_stream request and
 
1836
            # fallback to VFS.
 
1837
            stream = self._stop_stream_if_inventory_delta(stream)
1715
1838
        byte_stream = smart_repo._stream_to_byte_stream(
1716
1839
            stream, src_format)
1717
1840
        resume_tokens = ' '.join(resume_tokens)
1718
1841
        response = client.call_with_body_stream(
1719
 
            (verb, path, resume_tokens) + extra_args, byte_stream)
 
1842
            (verb, path, resume_tokens) + lock_args, byte_stream)
1720
1843
        if response[0][0] not in ('ok', 'missing-basis'):
1721
1844
            raise errors.UnexpectedSmartServerResponse(response)
 
1845
        if self._last_substream is not None:
 
1846
            # The stream included an inventory-delta record, but the remote
 
1847
            # side isn't new enough to support them.  So we need to send the
 
1848
            # rest of the stream via VFS.
 
1849
            self.target_repo.refresh_data()
 
1850
            return self._resume_stream_with_vfs(response, src_format)
1722
1851
        if response[0][0] == 'missing-basis':
1723
1852
            tokens, missing_keys = bencode.bdecode_as_tuple(response[0][1])
1724
1853
            resume_tokens = tokens
1727
1856
            self.target_repo.refresh_data()
1728
1857
            return [], set()
1729
1858
 
 
1859
    def _resume_stream_with_vfs(self, response, src_format):
 
1860
        """Resume sending a stream via VFS, first resending the record and
 
1861
        substream that couldn't be sent via an insert_stream verb.
 
1862
        """
 
1863
        if response[0][0] == 'missing-basis':
 
1864
            tokens, missing_keys = bencode.bdecode_as_tuple(response[0][1])
 
1865
            # Ignore missing_keys, we haven't finished inserting yet
 
1866
        else:
 
1867
            tokens = []
 
1868
        def resume_substream():
 
1869
            # Yield the substream that was interrupted.
 
1870
            for record in self._last_substream:
 
1871
                yield record
 
1872
            self._last_substream = None
 
1873
        def resume_stream():
 
1874
            # Finish sending the interrupted substream
 
1875
            yield ('inventory-deltas', resume_substream())
 
1876
            # Then simply continue sending the rest of the stream.
 
1877
            for substream_kind, substream in self._last_stream:
 
1878
                yield substream_kind, substream
 
1879
        return self._insert_real(resume_stream(), src_format, tokens)
 
1880
 
 
1881
    def _stop_stream_if_inventory_delta(self, stream):
 
1882
        """Normally this just lets the original stream pass-through unchanged.
 
1883
 
 
1884
        However if any 'inventory-deltas' substream occurs it will stop
 
1885
        streaming, and store the interrupted substream and stream in
 
1886
        self._last_substream and self._last_stream so that the stream can be
 
1887
        resumed by _resume_stream_with_vfs.
 
1888
        """
 
1889
                    
 
1890
        stream_iter = iter(stream)
 
1891
        for substream_kind, substream in stream_iter:
 
1892
            if substream_kind == 'inventory-deltas':
 
1893
                self._last_substream = substream
 
1894
                self._last_stream = stream_iter
 
1895
                return
 
1896
            else:
 
1897
                yield substream_kind, substream
 
1898
            
1730
1899
 
1731
1900
class RemoteStreamSource(repository.StreamSource):
1732
1901
    """Stream data from a remote server."""
1747
1916
            sources.append(repo)
1748
1917
        return self.missing_parents_chain(search, sources)
1749
1918
 
 
1919
    def get_stream_for_missing_keys(self, missing_keys):
 
1920
        self.from_repository._ensure_real()
 
1921
        real_repo = self.from_repository._real_repository
 
1922
        real_source = real_repo._get_source(self.to_format)
 
1923
        return real_source.get_stream_for_missing_keys(missing_keys)
 
1924
 
1750
1925
    def _real_stream(self, repo, search):
1751
1926
        """Get a stream for search from repo.
1752
1927
        
1758
1933
        """
1759
1934
        source = repo._get_source(self.to_format)
1760
1935
        if isinstance(source, RemoteStreamSource):
1761
 
            return repository.StreamSource.get_stream(source, search)
 
1936
            repo._ensure_real()
 
1937
            source = repo._real_repository._get_source(self.to_format)
1762
1938
        return source.get_stream(search)
1763
1939
 
1764
1940
    def _get_stream(self, repo, search):
1781
1957
            return self._real_stream(repo, search)
1782
1958
        client = repo._client
1783
1959
        medium = client._medium
1784
 
        if medium._is_remote_before((1, 13)):
1785
 
            # streaming was added in 1.13
1786
 
            return self._real_stream(repo, search)
1787
1960
        path = repo.bzrdir._path_for_remote_call(client)
1788
 
        try:
1789
 
            search_bytes = repo._serialise_search_result(search)
1790
 
            response = repo._call_with_body_bytes_expecting_body(
1791
 
                'Repository.get_stream',
1792
 
                (path, self.to_format.network_name()), search_bytes)
1793
 
            response_tuple, response_handler = response
1794
 
        except errors.UnknownSmartMethod:
1795
 
            medium._remember_remote_is_before((1,13))
 
1961
        search_bytes = repo._serialise_search_result(search)
 
1962
        args = (path, self.to_format.network_name())
 
1963
        candidate_verbs = [
 
1964
            ('Repository.get_stream_1.19', (1, 19)),
 
1965
            ('Repository.get_stream', (1, 13))]
 
1966
        found_verb = False
 
1967
        for verb, version in candidate_verbs:
 
1968
            if medium._is_remote_before(version):
 
1969
                continue
 
1970
            try:
 
1971
                response = repo._call_with_body_bytes_expecting_body(
 
1972
                    verb, args, search_bytes)
 
1973
            except errors.UnknownSmartMethod:
 
1974
                medium._remember_remote_is_before(version)
 
1975
            else:
 
1976
                response_tuple, response_handler = response
 
1977
                found_verb = True
 
1978
                break
 
1979
        if not found_verb:
1796
1980
            return self._real_stream(repo, search)
1797
1981
        if response_tuple[0] != 'ok':
1798
1982
            raise errors.UnexpectedSmartServerResponse(response_tuple)
1799
1983
        byte_stream = response_handler.read_streamed_body()
1800
 
        src_format, stream = smart_repo._byte_stream_to_stream(byte_stream)
 
1984
        src_format, stream = smart_repo._byte_stream_to_stream(byte_stream,
 
1985
            self._record_counter)
1801
1986
        if src_format.network_name() != repo._format.network_name():
1802
1987
            raise AssertionError(
1803
1988
                "Mismatched RemoteRepository and stream src %r, %r" % (
1810
1995
        :param search: The overall search to satisfy with streams.
1811
1996
        :param sources: A list of Repository objects to query.
1812
1997
        """
1813
 
        self.serialiser = self.to_format._serializer
 
1998
        self.from_serialiser = self.from_repository._format._serializer
1814
1999
        self.seen_revs = set()
1815
2000
        self.referenced_revs = set()
1816
2001
        # If there are heads in the search, or the key count is > 0, we are not
1833
2018
    def missing_parents_rev_handler(self, substream):
1834
2019
        for content in substream:
1835
2020
            revision_bytes = content.get_bytes_as('fulltext')
1836
 
            revision = self.serialiser.read_revision_from_string(revision_bytes)
 
2021
            revision = self.from_serialiser.read_revision_from_string(
 
2022
                revision_bytes)
1837
2023
            self.seen_revs.add(content.key[-1])
1838
2024
            self.referenced_revs.update(revision.parent_ids)
1839
2025
            yield content
1878
2064
                self._network_name)
1879
2065
 
1880
2066
    def get_format_description(self):
1881
 
        return 'Remote BZR Branch'
 
2067
        self._ensure_real()
 
2068
        return 'Remote: ' + self._custom_format.get_format_description()
1882
2069
 
1883
2070
    def network_name(self):
1884
2071
        return self._network_name
1885
2072
 
1886
 
    def open(self, a_bzrdir, ignore_fallbacks=False):
1887
 
        return a_bzrdir.open_branch(ignore_fallbacks=ignore_fallbacks)
 
2073
    def open(self, a_bzrdir, name=None, ignore_fallbacks=False):
 
2074
        return a_bzrdir.open_branch(name=name, 
 
2075
            ignore_fallbacks=ignore_fallbacks)
1888
2076
 
1889
 
    def _vfs_initialize(self, a_bzrdir):
 
2077
    def _vfs_initialize(self, a_bzrdir, name):
1890
2078
        # Initialisation when using a local bzrdir object, or a non-vfs init
1891
2079
        # method is not available on the server.
1892
2080
        # self._custom_format is always set - the start of initialize ensures
1893
2081
        # that.
1894
2082
        if isinstance(a_bzrdir, RemoteBzrDir):
1895
2083
            a_bzrdir._ensure_real()
1896
 
            result = self._custom_format.initialize(a_bzrdir._real_bzrdir)
 
2084
            result = self._custom_format.initialize(a_bzrdir._real_bzrdir,
 
2085
                name)
1897
2086
        else:
1898
2087
            # We assume the bzrdir is parameterised; it may not be.
1899
 
            result = self._custom_format.initialize(a_bzrdir)
 
2088
            result = self._custom_format.initialize(a_bzrdir, name)
1900
2089
        if (isinstance(a_bzrdir, RemoteBzrDir) and
1901
2090
            not isinstance(result, RemoteBranch)):
1902
 
            result = RemoteBranch(a_bzrdir, a_bzrdir.find_repository(), result)
 
2091
            result = RemoteBranch(a_bzrdir, a_bzrdir.find_repository(), result,
 
2092
                                  name=name)
1903
2093
        return result
1904
2094
 
1905
 
    def initialize(self, a_bzrdir):
 
2095
    def initialize(self, a_bzrdir, name=None):
1906
2096
        # 1) get the network name to use.
1907
2097
        if self._custom_format:
1908
2098
            network_name = self._custom_format.network_name()
1914
2104
            network_name = reference_format.network_name()
1915
2105
        # Being asked to create on a non RemoteBzrDir:
1916
2106
        if not isinstance(a_bzrdir, RemoteBzrDir):
1917
 
            return self._vfs_initialize(a_bzrdir)
 
2107
            return self._vfs_initialize(a_bzrdir, name=name)
1918
2108
        medium = a_bzrdir._client._medium
1919
2109
        if medium._is_remote_before((1, 13)):
1920
 
            return self._vfs_initialize(a_bzrdir)
 
2110
            return self._vfs_initialize(a_bzrdir, name=name)
1921
2111
        # Creating on a remote bzr dir.
1922
2112
        # 2) try direct creation via RPC
1923
2113
        path = a_bzrdir._path_for_remote_call(a_bzrdir._client)
 
2114
        if name is not None:
 
2115
            # XXX JRV20100304: Support creating colocated branches
 
2116
            raise errors.NoColocatedBranchSupport(self)
1924
2117
        verb = 'BzrDir.create_branch'
1925
2118
        try:
1926
2119
            response = a_bzrdir._call(verb, path, network_name)
1927
2120
        except errors.UnknownSmartMethod:
1928
2121
            # Fallback - use vfs methods
1929
2122
            medium._remember_remote_is_before((1, 13))
1930
 
            return self._vfs_initialize(a_bzrdir)
 
2123
            return self._vfs_initialize(a_bzrdir, name=name)
1931
2124
        if response[0] != 'ok':
1932
2125
            raise errors.UnexpectedSmartServerResponse(response)
1933
2126
        # Turn the response into a RemoteRepository object.
1941
2134
                a_bzrdir._client)
1942
2135
        remote_repo = RemoteRepository(repo_bzrdir, repo_format)
1943
2136
        remote_branch = RemoteBranch(a_bzrdir, remote_repo,
1944
 
            format=format, setup_stacking=False)
 
2137
            format=format, setup_stacking=False, name=name)
1945
2138
        # XXX: We know this is a new branch, so it must have revno 0, revid
1946
2139
        # NULL_REVISION. Creating the branch locked would make this be unable
1947
2140
        # to be wrong; here its simply very unlikely to be wrong. RBC 20090225
1967
2160
        return self._custom_format.supports_set_append_revisions_only()
1968
2161
 
1969
2162
 
1970
 
class RemoteBranch(branch.Branch, _RpcHelper):
 
2163
class RemoteBranch(branch.Branch, _RpcHelper, lock._RelockDebugMixin):
1971
2164
    """Branch stored on a server accessed by HPSS RPC.
1972
2165
 
1973
2166
    At the moment most operations are mapped down to simple file operations.
1974
2167
    """
1975
2168
 
1976
2169
    def __init__(self, remote_bzrdir, remote_repository, real_branch=None,
1977
 
        _client=None, format=None, setup_stacking=True):
 
2170
        _client=None, format=None, setup_stacking=True, name=None):
1978
2171
        """Create a RemoteBranch instance.
1979
2172
 
1980
2173
        :param real_branch: An optional local implementation of the branch
1986
2179
        :param setup_stacking: If True make an RPC call to determine the
1987
2180
            stacked (or not) status of the branch. If False assume the branch
1988
2181
            is not stacked.
 
2182
        :param name: Colocated branch name
1989
2183
        """
1990
2184
        # We intentionally don't call the parent class's __init__, because it
1991
2185
        # will try to assign to self.tags, which is a property in this subclass.
2010
2204
            self._real_branch = None
2011
2205
        # Fill out expected attributes of branch for bzrlib API users.
2012
2206
        self._clear_cached_state()
2013
 
        self.base = self.bzrdir.root_transport.base
 
2207
        # TODO: deprecate self.base in favor of user_url
 
2208
        self.base = self.bzrdir.user_url
 
2209
        self._name = name
2014
2210
        self._control_files = None
2015
2211
        self._lock_mode = None
2016
2212
        self._lock_token = None
2027
2223
                    self._real_branch._format.network_name()
2028
2224
        else:
2029
2225
            self._format = format
 
2226
        # when we do _ensure_real we may need to pass ignore_fallbacks to the
 
2227
        # branch.open_branch method.
 
2228
        self._real_ignore_fallbacks = not setup_stacking
2030
2229
        if not self._format._network_name:
2031
2230
            # Did not get from open_branchV2 - old server.
2032
2231
            self._ensure_real()
2077
2276
                raise AssertionError('smart server vfs must be enabled '
2078
2277
                    'to use vfs implementation')
2079
2278
            self.bzrdir._ensure_real()
2080
 
            self._real_branch = self.bzrdir._real_bzrdir.open_branch()
 
2279
            self._real_branch = self.bzrdir._real_bzrdir.open_branch(
 
2280
                ignore_fallbacks=self._real_ignore_fallbacks, name=self._name)
2081
2281
            if self.repository._real_repository is None:
2082
2282
                # Give the remote repository the matching real repo.
2083
2283
                real_repo = self._real_branch.repository
2187
2387
        medium = self._client._medium
2188
2388
        if medium._is_remote_before((1, 18)):
2189
2389
            self._vfs_set_tags_bytes(bytes)
 
2390
            return
2190
2391
        try:
2191
2392
            args = (
2192
2393
                self._remote_path(), self._lock_token, self._repo_lock_token)
2197
2398
            self._vfs_set_tags_bytes(bytes)
2198
2399
 
2199
2400
    def lock_read(self):
 
2401
        """Lock the branch for read operations.
 
2402
 
 
2403
        :return: A bzrlib.lock.LogicalLockResult.
 
2404
        """
2200
2405
        self.repository.lock_read()
2201
2406
        if not self._lock_mode:
 
2407
            self._note_lock('r')
2202
2408
            self._lock_mode = 'r'
2203
2409
            self._lock_count = 1
2204
2410
            if self._real_branch is not None:
2205
2411
                self._real_branch.lock_read()
2206
2412
        else:
2207
2413
            self._lock_count += 1
 
2414
        return lock.LogicalLockResult(self.unlock)
2208
2415
 
2209
2416
    def _remote_lock_write(self, token):
2210
2417
        if token is None:
2211
2418
            branch_token = repo_token = ''
2212
2419
        else:
2213
2420
            branch_token = token
2214
 
            repo_token = self.repository.lock_write()
 
2421
            repo_token = self.repository.lock_write().repository_token
2215
2422
            self.repository.unlock()
2216
2423
        err_context = {'token': token}
2217
 
        response = self._call(
2218
 
            'Branch.lock_write', self._remote_path(), branch_token,
2219
 
            repo_token or '', **err_context)
 
2424
        try:
 
2425
            response = self._call(
 
2426
                'Branch.lock_write', self._remote_path(), branch_token,
 
2427
                repo_token or '', **err_context)
 
2428
        except errors.LockContention, e:
 
2429
            # The LockContention from the server doesn't have any
 
2430
            # information about the lock_url. We re-raise LockContention
 
2431
            # with valid lock_url.
 
2432
            raise errors.LockContention('(remote lock)',
 
2433
                self.repository.base.split('.bzr/')[0])
2220
2434
        if response[0] != 'ok':
2221
2435
            raise errors.UnexpectedSmartServerResponse(response)
2222
2436
        ok, branch_token, repo_token = response
2224
2438
 
2225
2439
    def lock_write(self, token=None):
2226
2440
        if not self._lock_mode:
 
2441
            self._note_lock('w')
2227
2442
            # Lock the branch and repo in one remote call.
2228
2443
            remote_tokens = self._remote_lock_write(token)
2229
2444
            self._lock_token, self._repo_lock_token = remote_tokens
2242
2457
            self._lock_mode = 'w'
2243
2458
            self._lock_count = 1
2244
2459
        elif self._lock_mode == 'r':
2245
 
            raise errors.ReadOnlyTransaction
 
2460
            raise errors.ReadOnlyError(self)
2246
2461
        else:
2247
2462
            if token is not None:
2248
2463
                # A token was given to lock_write, and we're relocking, so
2253
2468
            self._lock_count += 1
2254
2469
            # Re-lock the repository too.
2255
2470
            self.repository.lock_write(self._repo_lock_token)
2256
 
        return self._lock_token or None
 
2471
        return BranchWriteLockResult(self.unlock, self._lock_token or None)
2257
2472
 
2258
2473
    def _unlock(self, branch_token, repo_token):
2259
2474
        err_context = {'token': str((branch_token, repo_token))}
2264
2479
            return
2265
2480
        raise errors.UnexpectedSmartServerResponse(response)
2266
2481
 
 
2482
    @only_raises(errors.LockNotHeld, errors.LockBroken)
2267
2483
    def unlock(self):
2268
2484
        try:
2269
2485
            self._lock_count -= 1
2309
2525
            raise NotImplementedError(self.dont_leave_lock_in_place)
2310
2526
        self._leave_lock = False
2311
2527
 
 
2528
    @needs_read_lock
2312
2529
    def get_rev_id(self, revno, history=None):
2313
2530
        if revno == 0:
2314
2531
            return _mod_revision.NULL_REVISION
2580
2797
        medium = self._branch._client._medium
2581
2798
        if medium._is_remote_before((1, 14)):
2582
2799
            return self._vfs_set_option(value, name, section)
 
2800
        if isinstance(value, dict):
 
2801
            if medium._is_remote_before((2, 2)):
 
2802
                return self._vfs_set_option(value, name, section)
 
2803
            return self._set_config_option_dict(value, name, section)
 
2804
        else:
 
2805
            return self._set_config_option(value, name, section)
 
2806
 
 
2807
    def _set_config_option(self, value, name, section):
2583
2808
        try:
2584
2809
            path = self._branch._remote_path()
2585
2810
            response = self._branch._client.call('Branch.set_config_option',
2586
2811
                path, self._branch._lock_token, self._branch._repo_lock_token,
2587
2812
                value.encode('utf8'), name, section or '')
2588
2813
        except errors.UnknownSmartMethod:
 
2814
            medium = self._branch._client._medium
2589
2815
            medium._remember_remote_is_before((1, 14))
2590
2816
            return self._vfs_set_option(value, name, section)
2591
2817
        if response != ():
2592
2818
            raise errors.UnexpectedSmartServerResponse(response)
2593
2819
 
 
2820
    def _serialize_option_dict(self, option_dict):
 
2821
        utf8_dict = {}
 
2822
        for key, value in option_dict.items():
 
2823
            if isinstance(key, unicode):
 
2824
                key = key.encode('utf8')
 
2825
            if isinstance(value, unicode):
 
2826
                value = value.encode('utf8')
 
2827
            utf8_dict[key] = value
 
2828
        return bencode.bencode(utf8_dict)
 
2829
 
 
2830
    def _set_config_option_dict(self, value, name, section):
 
2831
        try:
 
2832
            path = self._branch._remote_path()
 
2833
            serialised_dict = self._serialize_option_dict(value)
 
2834
            response = self._branch._client.call(
 
2835
                'Branch.set_config_option_dict',
 
2836
                path, self._branch._lock_token, self._branch._repo_lock_token,
 
2837
                serialised_dict, name, section or '')
 
2838
        except errors.UnknownSmartMethod:
 
2839
            medium = self._branch._client._medium
 
2840
            medium._remember_remote_is_before((2, 2))
 
2841
            return self._vfs_set_option(value, name, section)
 
2842
        if response != ():
 
2843
            raise errors.UnexpectedSmartServerResponse(response)
 
2844
 
2594
2845
    def _real_object(self):
2595
2846
        self._branch._ensure_real()
2596
2847
        return self._branch._real_branch
2679
2930
                    'Missing key %r in context %r', key_err.args[0], context)
2680
2931
                raise err
2681
2932
 
2682
 
    if err.error_verb == 'NoSuchRevision':
 
2933
    if err.error_verb == 'IncompatibleRepositories':
 
2934
        raise errors.IncompatibleRepositories(err.error_args[0],
 
2935
            err.error_args[1], err.error_args[2])
 
2936
    elif err.error_verb == 'NoSuchRevision':
2683
2937
        raise NoSuchRevision(find('branch'), err.error_args[0])
2684
2938
    elif err.error_verb == 'nosuchrevision':
2685
2939
        raise NoSuchRevision(find('repository'), err.error_args[0])
2686
 
    elif err.error_tuple == ('nobranch',):
2687
 
        raise errors.NotBranchError(path=find('bzrdir').root_transport.base)
 
2940
    elif err.error_verb == 'nobranch':
 
2941
        if len(err.error_args) >= 1:
 
2942
            extra = err.error_args[0]
 
2943
        else:
 
2944
            extra = None
 
2945
        raise errors.NotBranchError(path=find('bzrdir').root_transport.base,
 
2946
            detail=extra)
2688
2947
    elif err.error_verb == 'norepository':
2689
2948
        raise errors.NoRepositoryPresent(find('bzrdir'))
2690
2949
    elif err.error_verb == 'LockContention':