← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/transport/remote.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2009-08-27 02:27:19 UTC
  • mfrom: (4634.3.19 gc-batching)
  • Revision ID: pqm@pqm.ubuntu.com-20090827022719-bl2yoqhpj3fcfczu
(andrew) Fix #402657: 2a fetch over dumb transport reads one group at
        a time.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/transport/remote.py
12
12
#
13
13
# You should have received a copy of the GNU General Public License
14
14
# along with this program; if not, write to the Free Software
15
 
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
16
16
 
17
17
"""RemoteTransport client for the smart-server.
18
18
 
28
28
    config,
29
29
    debug,
30
30
    errors,
 
31
    remote,
31
32
    trace,
32
33
    transport,
33
34
    urlutils,
34
35
    )
35
36
from bzrlib.smart import client, medium
36
 
from bzrlib.symbol_versioning import (deprecated_method, one_four)
 
37
from bzrlib.symbol_versioning import (
 
38
    deprecated_method,
 
39
    )
37
40
 
38
41
 
39
42
class _SmartStat(object):
43
46
        self.st_mode = mode
44
47
 
45
48
 
46
 
class RemoteTransport(transport.ConnectedTransport, medium.SmartClientMedium):
 
49
class RemoteTransport(transport.ConnectedTransport):
47
50
    """Connection to a smart server.
48
51
 
49
52
    The connection holds references to the medium that can be used to send
51
54
 
52
55
    The connection has a notion of the current directory to which it's
53
56
    connected; this is incorporated in filenames passed to the server.
54
 
    
55
 
    This supports some higher-level RPC operations and can also be treated 
 
57
 
 
58
    This supports some higher-level RPC operations and can also be treated
56
59
    like a Transport to do file-like operations.
57
60
 
58
61
    The connection can be made over a tcp socket, an ssh pipe or a series of
60
63
    RemoteTCPTransport, etc.
61
64
    """
62
65
 
 
66
    # When making a readv request, cap it at requesting 5MB of data
 
67
    _max_readv_bytes = 5*1024*1024
 
68
 
63
69
    # IMPORTANT FOR IMPLEMENTORS: RemoteTransport MUST NOT be given encoding
64
70
    # responsibilities: Put those on SmartClient or similar. This is vital for
65
71
    # the ability to support multiple versions of the smart protocol over time:
66
 
    # RemoteTransport is an adapter from the Transport object model to the 
 
72
    # RemoteTransport is an adapter from the Transport object model to the
67
73
    # SmartClient model, not an encoder.
68
74
 
69
75
    # FIXME: the medium parameter should be private, only the tests requires
86
92
            should only be used for testing purposes; normally this is
87
93
            determined from the medium.
88
94
        """
89
 
        super(RemoteTransport, self).__init__(url,
90
 
                                              _from_transport=_from_transport)
 
95
        super(RemoteTransport, self).__init__(
 
96
            url, _from_transport=_from_transport)
91
97
 
92
98
        # The medium is the connection, except when we need to share it with
93
99
        # other objects (RemoteBzrDir, RemoteRepository etc). In these cases
94
100
        # what we want to share is really the shared connection.
95
101
 
96
 
        if _from_transport is None:
 
102
        if (_from_transport is not None
 
103
            and isinstance(_from_transport, RemoteTransport)):
 
104
            _client = _from_transport._client
 
105
        elif _from_transport is None:
97
106
            # If no _from_transport is specified, we need to intialize the
98
107
            # shared medium.
99
108
            credentials = None
129
138
        # No credentials
130
139
        return None, None
131
140
 
 
141
    def _report_activity(self, bytes, direction):
 
142
        """See Transport._report_activity.
 
143
 
 
144
        Does nothing; the smart medium will report activity triggered by a
 
145
        RemoteTransport.
 
146
        """
 
147
        pass
 
148
 
132
149
    def is_readonly(self):
133
150
        """Smart server transport can do read/write file operations."""
134
151
        try:
143
160
        elif resp == ('no', ):
144
161
            return False
145
162
        else:
146
 
            self._translate_error(resp)
147
 
        raise errors.UnexpectedSmartServerResponse(resp)
 
163
            raise errors.UnexpectedSmartServerResponse(resp)
148
164
 
149
165
    def get_smart_client(self):
150
166
        return self._get_connection()
152
168
    def get_smart_medium(self):
153
169
        return self._get_connection()
154
170
 
155
 
    @deprecated_method(one_four)
156
 
    def get_shared_medium(self):
157
 
        return self._get_shared_connection()
158
 
 
159
171
    def _remote_path(self, relpath):
160
172
        """Returns the Unicode version of the absolute path for relpath."""
161
173
        return self._combine_paths(self._path, relpath)
162
174
 
163
175
    def _call(self, method, *args):
164
 
        try:
165
 
            resp = self._call2(method, *args)
166
 
        except errors.ErrorFromSmartServer, err:
167
 
            self._translate_error(err.error_tuple)
168
 
        self._translate_error(resp)
 
176
        resp = self._call2(method, *args)
 
177
        self._ensure_ok(resp)
169
178
 
170
179
    def _call2(self, method, *args):
171
180
        """Call a method on the remote server."""
172
181
        try:
173
182
            return self._client.call(method, *args)
174
183
        except errors.ErrorFromSmartServer, err:
175
 
            self._translate_error(err.error_tuple)
 
184
            # The first argument, if present, is always a path.
 
185
            if args:
 
186
                context = {'relpath': args[0]}
 
187
            else:
 
188
                context = {}
 
189
            self._translate_error(err, **context)
176
190
 
177
191
    def _call_with_body_bytes(self, method, args, body):
178
192
        """Call a method on the remote server with body bytes."""
179
193
        try:
180
194
            return self._client.call_with_body_bytes(method, args, body)
181
195
        except errors.ErrorFromSmartServer, err:
182
 
            self._translate_error(err.error_tuple)
 
196
            # The first argument, if present, is always a path.
 
197
            if args:
 
198
                context = {'relpath': args[0]}
 
199
            else:
 
200
                context = {}
 
201
            self._translate_error(err, **context)
183
202
 
184
203
    def has(self, relpath):
185
204
        """Indicate whether a remote file of the given name exists or not.
192
211
        elif resp == ('no', ):
193
212
            return False
194
213
        else:
195
 
            self._translate_error(resp)
 
214
            raise errors.UnexpectedSmartServerResponse(resp)
196
215
 
197
216
    def get(self, relpath):
198
217
        """Return file-like object reading the contents of a remote file.
199
 
        
 
218
 
200
219
        :see: Transport.get_bytes()/get_file()
201
220
        """
202
221
        return StringIO(self.get_bytes(relpath))
206
225
        try:
207
226
            resp, response_handler = self._client.call_expecting_body('get', remote)
208
227
        except errors.ErrorFromSmartServer, err:
209
 
            self._translate_error(err.error_tuple, relpath)
 
228
            self._translate_error(err, relpath)
210
229
        if resp != ('ok', ):
211
230
            response_handler.cancel_read_body()
212
231
            raise errors.UnexpectedSmartServerResponse(resp)
221
240
    def mkdir(self, relpath, mode=None):
222
241
        resp = self._call2('mkdir', self._remote_path(relpath),
223
242
            self._serialise_optional_mode(mode))
224
 
        self._translate_error(resp)
225
243
 
226
244
    def open_write_stream(self, relpath, mode=None):
227
245
        """See Transport.open_write_stream."""
243
261
        resp = self._call_with_body_bytes('put',
244
262
            (self._remote_path(relpath), self._serialise_optional_mode(mode)),
245
263
            upload_contents)
246
 
        self._translate_error(resp)
 
264
        self._ensure_ok(resp)
247
265
        return len(upload_contents)
248
266
 
249
267
    def put_bytes_non_atomic(self, relpath, bytes, mode=None,
260
278
            (self._remote_path(relpath), self._serialise_optional_mode(mode),
261
279
             create_parent_str, self._serialise_optional_mode(dir_mode)),
262
280
            bytes)
263
 
        self._translate_error(resp)
 
281
        self._ensure_ok(resp)
264
282
 
265
283
    def put_file(self, relpath, upload_file, mode=None):
266
284
        # its not ideal to seek back, but currently put_non_atomic_file depends
282
300
 
283
301
    def append_file(self, relpath, from_file, mode=None):
284
302
        return self.append_bytes(relpath, from_file.read(), mode)
285
 
        
 
303
 
286
304
    def append_bytes(self, relpath, bytes, mode=None):
287
305
        resp = self._call_with_body_bytes(
288
306
            'append',
290
308
            bytes)
291
309
        if resp[0] == 'appended':
292
310
            return int(resp[1])
293
 
        self._translate_error(resp)
 
311
        raise errors.UnexpectedSmartServerResponse(resp)
294
312
 
295
313
    def delete(self, relpath):
296
314
        resp = self._call2('delete', self._remote_path(relpath))
297
 
        self._translate_error(resp)
 
315
        self._ensure_ok(resp)
298
316
 
299
317
    def external_url(self):
300
318
        """See bzrlib.transport.Transport.external_url."""
304
322
    def recommended_page_size(self):
305
323
        """Return the recommended page size for this transport."""
306
324
        return 64 * 1024
307
 
        
 
325
 
308
326
    def _readv(self, relpath, offsets):
309
327
        if not offsets:
310
328
            return
314
332
        sorted_offsets = sorted(offsets)
315
333
        coalesced = list(self._coalesce_offsets(sorted_offsets,
316
334
                               limit=self._max_readv_combine,
317
 
                               fudge_factor=self._bytes_to_read_before_seek))
318
 
 
319
 
        try:
320
 
            result = self._client.call_with_body_readv_array(
321
 
                ('readv', self._remote_path(relpath),),
322
 
                [(c.start, c.length) for c in coalesced])
323
 
            resp, response_handler = result
324
 
        except errors.ErrorFromSmartServer, err:
325
 
            self._translate_error(err.error_tuple)
326
 
 
327
 
        if resp[0] != 'readv':
328
 
            # This should raise an exception
329
 
            response_handler.cancel_read_body()
330
 
            raise errors.UnexpectedSmartServerResponse(resp)
331
 
 
332
 
        return self._handle_response(offsets, coalesced, response_handler)
333
 
 
334
 
    def _handle_response(self, offsets, coalesced, response_handler):
335
 
        # turn the list of offsets into a stack
 
335
                               fudge_factor=self._bytes_to_read_before_seek,
 
336
                               max_size=self._max_readv_bytes))
 
337
 
 
338
        # now that we've coallesced things, avoid making enormous requests
 
339
        requests = []
 
340
        cur_request = []
 
341
        cur_len = 0
 
342
        for c in coalesced:
 
343
            if c.length + cur_len > self._max_readv_bytes:
 
344
                requests.append(cur_request)
 
345
                cur_request = [c]
 
346
                cur_len = c.length
 
347
                continue
 
348
            cur_request.append(c)
 
349
            cur_len += c.length
 
350
        if cur_request:
 
351
            requests.append(cur_request)
 
352
        if 'hpss' in debug.debug_flags:
 
353
            trace.mutter('%s.readv %s offsets => %s coalesced'
 
354
                         ' => %s requests (%s)',
 
355
                         self.__class__.__name__, len(offsets), len(coalesced),
 
356
                         len(requests), sum(map(len, requests)))
 
357
        # Cache the results, but only until they have been fulfilled
 
358
        data_map = {}
 
359
        # turn the list of offsets into a single stack to iterate
336
360
        offset_stack = iter(offsets)
337
 
        cur_offset_and_size = offset_stack.next()
 
361
        # using a list so it can be modified when passing down and coming back
 
362
        next_offset = [offset_stack.next()]
 
363
        for cur_request in requests:
 
364
            try:
 
365
                result = self._client.call_with_body_readv_array(
 
366
                    ('readv', self._remote_path(relpath),),
 
367
                    [(c.start, c.length) for c in cur_request])
 
368
                resp, response_handler = result
 
369
            except errors.ErrorFromSmartServer, err:
 
370
                self._translate_error(err, relpath)
 
371
 
 
372
            if resp[0] != 'readv':
 
373
                # This should raise an exception
 
374
                response_handler.cancel_read_body()
 
375
                raise errors.UnexpectedSmartServerResponse(resp)
 
376
 
 
377
            for res in self._handle_response(offset_stack, cur_request,
 
378
                                             response_handler,
 
379
                                             data_map,
 
380
                                             next_offset):
 
381
                yield res
 
382
 
 
383
    def _handle_response(self, offset_stack, coalesced, response_handler,
 
384
                         data_map, next_offset):
 
385
        cur_offset_and_size = next_offset[0]
338
386
        # FIXME: this should know how many bytes are needed, for clarity.
339
387
        data = response_handler.read_body_bytes()
340
 
        # Cache the results, but only until they have been fulfilled
341
 
        data_map = {}
342
388
        data_offset = 0
343
389
        for c_offset in coalesced:
344
390
            if len(data) < c_offset.length:
357
403
                #       not have a real string.
358
404
                if key == cur_offset_and_size:
359
405
                    yield cur_offset_and_size[0], this_data
360
 
                    cur_offset_and_size = offset_stack.next()
 
406
                    cur_offset_and_size = next_offset[0] = offset_stack.next()
361
407
                else:
362
408
                    data_map[key] = this_data
363
409
            data_offset += c_offset.length
366
412
            while cur_offset_and_size in data_map:
367
413
                this_data = data_map.pop(cur_offset_and_size)
368
414
                yield cur_offset_and_size[0], this_data
369
 
                cur_offset_and_size = offset_stack.next()
 
415
                cur_offset_and_size = next_offset[0] = offset_stack.next()
370
416
 
371
417
    def rename(self, rel_from, rel_to):
372
418
        self._call('rename',
381
427
    def rmdir(self, relpath):
382
428
        resp = self._call('rmdir', self._remote_path(relpath))
383
429
 
384
 
    def _translate_error(self, resp, orig_path=None):
385
 
        """Raise an exception from a response"""
386
 
        if resp is None:
387
 
            what = None
388
 
        else:
389
 
            what = resp[0]
390
 
        if what == 'ok':
391
 
            return
392
 
        elif what == 'NoSuchFile':
393
 
            if orig_path is not None:
394
 
                error_path = orig_path
395
 
            else:
396
 
                error_path = resp[1]
397
 
            raise errors.NoSuchFile(error_path)
398
 
        elif what == 'error':
399
 
            raise errors.SmartProtocolError(unicode(resp[1]))
400
 
        elif what == 'FileExists':
401
 
            raise errors.FileExists(resp[1])
402
 
        elif what == 'DirectoryNotEmpty':
403
 
            raise errors.DirectoryNotEmpty(resp[1])
404
 
        elif what == 'ShortReadvError':
405
 
            raise errors.ShortReadvError(resp[1], int(resp[2]),
406
 
                                         int(resp[3]), int(resp[4]))
407
 
        elif what in ('UnicodeEncodeError', 'UnicodeDecodeError'):
408
 
            encoding = str(resp[1]) # encoding must always be a string
409
 
            val = resp[2]
410
 
            start = int(resp[3])
411
 
            end = int(resp[4])
412
 
            reason = str(resp[5]) # reason must always be a string
413
 
            if val.startswith('u:'):
414
 
                val = val[2:].decode('utf-8')
415
 
            elif val.startswith('s:'):
416
 
                val = val[2:].decode('base64')
417
 
            if what == 'UnicodeDecodeError':
418
 
                raise UnicodeDecodeError(encoding, val, start, end, reason)
419
 
            elif what == 'UnicodeEncodeError':
420
 
                raise UnicodeEncodeError(encoding, val, start, end, reason)
421
 
        elif what == "ReadOnlyError":
422
 
            raise errors.TransportNotPossible('readonly transport')
423
 
        elif what == "ReadError":
424
 
            if orig_path is not None:
425
 
                error_path = orig_path
426
 
            else:
427
 
                error_path = resp[1]
428
 
            raise errors.ReadError(error_path)
429
 
        elif what == "PermissionDenied":
430
 
            if orig_path is not None:
431
 
                error_path = orig_path
432
 
            else:
433
 
                error_path = resp[1]
434
 
            raise errors.PermissionDenied(error_path)
435
 
        else:
436
 
            raise errors.SmartProtocolError('unexpected smart server error: %r' % (resp,))
 
430
    def _ensure_ok(self, resp):
 
431
        if resp[0] != 'ok':
 
432
            raise errors.UnexpectedSmartServerResponse(resp)
 
433
 
 
434
    def _translate_error(self, err, relpath=None):
 
435
        remote._translate_error(err, path=relpath)
437
436
 
438
437
    def disconnect(self):
439
438
        self.get_smart_medium().disconnect()
442
441
        resp = self._call2('stat', self._remote_path(relpath))
443
442
        if resp[0] == 'stat':
444
443
            return _SmartStat(int(resp[1]), int(resp[2], 8))
445
 
        else:
446
 
            self._translate_error(resp)
 
444
        raise errors.UnexpectedSmartServerResponse(resp)
447
445
 
448
446
    ## def lock_read(self, relpath):
449
447
    ##     """Lock the given file for shared (read) access.
465
463
        resp = self._call2('list_dir', self._remote_path(relpath))
466
464
        if resp[0] == 'names':
467
465
            return [name.encode('ascii') for name in resp[1:]]
468
 
        else:
469
 
            self._translate_error(resp)
 
466
        raise errors.UnexpectedSmartServerResponse(resp)
470
467
 
471
468
    def iter_files_recursive(self):
472
469
        resp = self._call2('iter_files_recursive', self._remote_path(''))
473
470
        if resp[0] == 'names':
474
471
            return resp[1:]
475
 
        else:
476
 
            self._translate_error(resp)
 
472
        raise errors.UnexpectedSmartServerResponse(resp)
477
473
 
478
474
 
479
475
class RemoteTCPTransport(RemoteTransport):
480
476
    """Connection to smart server over plain tcp.
481
 
    
 
477
 
482
478
    This is essentially just a factory to get 'RemoteTransport(url,
483
479
        SmartTCPClientMedium).
484
480
    """
512
508
    """
513
509
 
514
510
    def _build_medium(self):
515
 
        # ssh will prompt the user for a password if needed and if none is
516
 
        # provided but it will not give it back, so no credentials can be
517
 
        # stored.
518
511
        location_config = config.LocationConfig(self.base)
519
512
        bzr_remote_path = location_config.get_bzr_remote_path()
 
513
        user = self._user
 
514
        if user is None:
 
515
            auth = config.AuthenticationConfig()
 
516
            user = auth.get_user('ssh', self._host, self._port)
520
517
        client_medium = medium.SmartSSHClientMedium(self._host, self._port,
521
 
            self._user, self._password, self.base,
 
518
            user, self._password, self.base,
522
519
            bzr_remote_path=bzr_remote_path)
523
 
        return client_medium, None
 
520
        return client_medium, (user, self._password)
524
521
 
525
522
 
526
523
class RemoteHTTPTransport(RemoteTransport):
527
524
    """Just a way to connect between a bzr+http:// url and http://.
528
 
    
 
525
 
529
526
    This connection operates slightly differently than the RemoteSSHTransport.
530
527
    It uses a plain http:// transport underneath, which defines what remote
531
528
    .bzr/smart URL we are connected to. From there, all paths that are sent are
580
577
                                   _from_transport=self,
581
578
                                   http_transport=self._http_transport)
582
579
 
 
580
    def _redirected_to(self, source, target):
 
581
        """See transport._redirected_to"""
 
582
        redirected = self._http_transport._redirected_to(source, target)
 
583
        if (redirected is not None
 
584
            and isinstance(redirected, type(self._http_transport))):
 
585
            return RemoteHTTPTransport('bzr+' + redirected.external_url(),
 
586
                                       http_transport=redirected)
 
587
        else:
 
588
            # Either None or a transport for a different protocol
 
589
            return redirected
 
590
 
 
591
 
 
592
class HintingSSHTransport(transport.Transport):
 
593
    """Simple transport that handles ssh:// and points out bzr+ssh://."""
 
594
 
 
595
    def __init__(self, url):
 
596
        raise errors.UnsupportedProtocol(url,
 
597
            'bzr supports bzr+ssh to operate over ssh, use "bzr+%s".' % url)
 
598
 
583
599
 
584
600
def get_test_permutations():
585
601
    """Return (transport, server) permutations for testing."""