← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/smart/medium.py

  • Committer: Tarmac
  • Author(s): Vincent Ladeuil
  • Date: 2017-01-30 14:42:05 UTC
  • mfrom: (6620.1.1 trunk)
  • Revision ID: tarmac-20170130144205-r8fh2xpmiuxyozpv
Merge  2.7 into trunk including fix for bug #1657238 [r=vila]

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/smart/medium.py
1
 
# Copyright (C) 2006 Canonical Ltd
 
1
# Copyright (C) 2006-2011 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
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
"""The 'medium' layer for the smart servers and clients.
18
18
 
24
24
bzrlib/transport/smart/__init__.py.
25
25
"""
26
26
 
 
27
from __future__ import absolute_import
 
28
 
27
29
import errno
28
30
import os
29
 
import socket
30
31
import sys
31
 
import urllib
 
32
import time
32
33
 
 
34
import bzrlib
33
35
from bzrlib.lazy_import import lazy_import
34
36
lazy_import(globals(), """
35
 
import atexit
 
37
import select
 
38
import socket
 
39
import thread
36
40
import weakref
 
41
 
37
42
from bzrlib import (
38
43
    debug,
39
44
    errors,
40
 
    osutils,
41
 
    symbol_versioning,
42
45
    trace,
 
46
    transport,
 
47
    ui,
43
48
    urlutils,
44
49
    )
45
 
from bzrlib.smart import client, protocol
 
50
from bzrlib.i18n import gettext
 
51
from bzrlib.smart import client, protocol, request, signals, vfs
46
52
from bzrlib.transport import ssh
47
53
""")
48
 
 
49
 
 
50
 
# We must not read any more than 64k at a time so we don't risk "no buffer
51
 
# space available" errors on some platforms.  Windows in particular is likely
52
 
# to give error 10053 or 10055 if we read more than 64k from a socket.
53
 
_MAX_READ_SIZE = 64 * 1024
54
 
 
 
54
from bzrlib import osutils
 
55
 
 
56
# Throughout this module buffer size parameters are either limited to be at
 
57
# most _MAX_READ_SIZE, or are ignored and _MAX_READ_SIZE is used instead.
 
58
# For this module's purposes, MAX_SOCKET_CHUNK is a reasonable size for reads
 
59
# from non-sockets as well.
 
60
_MAX_READ_SIZE = osutils.MAX_SOCKET_CHUNK
55
61
 
56
62
def _get_protocol_factory_for_bytes(bytes):
57
63
    """Determine the right protocol factory for 'bytes'.
87
93
 
88
94
def _get_line(read_bytes_func):
89
95
    """Read bytes using read_bytes_func until a newline byte.
90
 
    
 
96
 
91
97
    This isn't particularly efficient, so should only be used when the
92
98
    expected size of the line is quite short.
93
 
    
 
99
 
94
100
    :returns: a tuple of two strs: (line, excess)
95
101
    """
96
102
    newline_pos = -1
112
118
 
113
119
    def __init__(self):
114
120
        self._push_back_buffer = None
115
 
        
 
121
 
116
122
    def _push_back(self, bytes):
117
123
        """Return unused bytes to the medium, because they belong to the next
118
124
        request(s).
152
158
 
153
159
    def _get_line(self):
154
160
        """Read bytes from this request's response until a newline byte.
155
 
        
 
161
 
156
162
        This isn't particularly efficient, so should only be used when the
157
163
        expected size of the line is quite short.
158
164
 
161
167
        line, excess = _get_line(self.read_bytes)
162
168
        self._push_back(excess)
163
169
        return line
164
 
 
 
170
 
 
171
    def _report_activity(self, bytes, direction):
 
172
        """Notify that this medium has activity.
 
173
 
 
174
        Implementations should call this from all methods that actually do IO.
 
175
        Be careful that it's not called twice, if one method is implemented on
 
176
        top of another.
 
177
 
 
178
        :param bytes: Number of bytes read or written.
 
179
        :param direction: 'read' or 'write' or None.
 
180
        """
 
181
        ui.ui_factory.report_transport_activity(self, bytes, direction)
 
182
 
 
183
 
 
184
_bad_file_descriptor = (errno.EBADF,)
 
185
if sys.platform == 'win32':
 
186
    # Given on Windows if you pass a closed socket to select.select. Probably
 
187
    # also given if you pass a file handle to select.
 
188
    WSAENOTSOCK = 10038
 
189
    _bad_file_descriptor += (WSAENOTSOCK,)
 
190
 
165
191
 
166
192
class SmartServerStreamMedium(SmartMedium):
167
193
    """Handles smart commands coming over a stream.
172
198
    One instance is created for each connected client; it can serve multiple
173
199
    requests in the lifetime of the connection.
174
200
 
175
 
    The server passes requests through to an underlying backing transport, 
 
201
    The server passes requests through to an underlying backing transport,
176
202
    which will typically be a LocalTransport looking at the server's filesystem.
177
203
 
178
204
    :ivar _push_back_buffer: a str of bytes that have been read from the stream
181
207
        the stream.  See also the _push_back method.
182
208
    """
183
209
 
184
 
    def __init__(self, backing_transport, root_client_path='/'):
 
210
    _timer = time.time
 
211
 
 
212
    def __init__(self, backing_transport, root_client_path='/', timeout=None):
185
213
        """Construct new server.
186
214
 
187
215
        :param backing_transport: Transport for the directory served.
190
218
        self.backing_transport = backing_transport
191
219
        self.root_client_path = root_client_path
192
220
        self.finished = False
 
221
        if timeout is None:
 
222
            raise AssertionError('You must supply a timeout.')
 
223
        self._client_timeout = timeout
 
224
        self._client_poll_timeout = min(timeout / 10.0, 1.0)
193
225
        SmartMedium.__init__(self)
194
226
 
195
227
    def serve(self):
201
233
            while not self.finished:
202
234
                server_protocol = self._build_protocol()
203
235
                self._serve_one_request(server_protocol)
 
236
        except errors.ConnectionTimeout, e:
 
237
            trace.note('%s' % (e,))
 
238
            trace.log_exception_quietly()
 
239
            self._disconnect_client()
 
240
            # We reported it, no reason to make a big fuss.
 
241
            return
204
242
        except Exception, e:
205
243
            stderr.write("%s terminating on exception %s\n" % (self, e))
206
244
            raise
 
245
        self._disconnect_client()
 
246
 
 
247
    def _stop_gracefully(self):
 
248
        """When we finish this message, stop looking for more."""
 
249
        trace.mutter('Stopping %s' % (self,))
 
250
        self.finished = True
 
251
 
 
252
    def _disconnect_client(self):
 
253
        """Close the current connection. We stopped due to a timeout/etc."""
 
254
        # The default implementation is a no-op, because that is all we used to
 
255
        # do when disconnecting from a client. I suppose we never had the
 
256
        # *server* initiate a disconnect, before
 
257
 
 
258
    def _wait_for_bytes_with_timeout(self, timeout_seconds):
 
259
        """Wait for more bytes to be read, but timeout if none available.
 
260
 
 
261
        This allows us to detect idle connections, and stop trying to read from
 
262
        them, without setting the socket itself to non-blocking. This also
 
263
        allows us to specify when we watch for idle timeouts.
 
264
 
 
265
        :return: Did we timeout? (True if we timed out, False if there is data
 
266
            to be read)
 
267
        """
 
268
        raise NotImplementedError(self._wait_for_bytes_with_timeout)
207
269
 
208
270
    def _build_protocol(self):
209
271
        """Identifies the version of the incoming request, and returns an
214
276
 
215
277
        :returns: a SmartServerRequestProtocol.
216
278
        """
 
279
        self._wait_for_bytes_with_timeout(self._client_timeout)
 
280
        if self.finished:
 
281
            # We're stopping, so don't try to do any more work
 
282
            return None
217
283
        bytes = self._get_line()
218
284
        protocol_factory, unused_bytes = _get_protocol_factory_for_bytes(bytes)
219
285
        protocol = protocol_factory(
221
287
        protocol.accept_bytes(unused_bytes)
222
288
        return protocol
223
289
 
 
290
    def _wait_on_descriptor(self, fd, timeout_seconds):
 
291
        """select() on a file descriptor, waiting for nonblocking read()
 
292
 
 
293
        This will raise a ConnectionTimeout exception if we do not get a
 
294
        readable handle before timeout_seconds.
 
295
        :return: None
 
296
        """
 
297
        t_end = self._timer() + timeout_seconds
 
298
        poll_timeout = min(timeout_seconds, self._client_poll_timeout)
 
299
        rs = xs = None
 
300
        while not rs and not xs and self._timer() < t_end:
 
301
            if self.finished:
 
302
                return
 
303
            try:
 
304
                rs, _, xs = select.select([fd], [], [fd], poll_timeout)
 
305
            except (select.error, socket.error) as e:
 
306
                err = getattr(e, 'errno', None)
 
307
                if err is None and getattr(e, 'args', None) is not None:
 
308
                    # select.error doesn't have 'errno', it just has args[0]
 
309
                    err = e.args[0]
 
310
                if err in _bad_file_descriptor:
 
311
                    return # Not a socket indicates read() will fail
 
312
                elif err == errno.EINTR:
 
313
                    # Interrupted, keep looping.
 
314
                    continue
 
315
                raise
 
316
        if rs or xs:
 
317
            return
 
318
        raise errors.ConnectionTimeout('disconnecting client after %.1f seconds'
 
319
                                       % (timeout_seconds,))
 
320
 
224
321
    def _serve_one_request(self, protocol):
225
322
        """Read one request from input, process, send back a response.
226
 
        
 
323
 
227
324
        :param protocol: a SmartServerRequestProtocol.
228
325
        """
 
326
        if protocol is None:
 
327
            return
229
328
        try:
230
329
            self._serve_one_request_unguarded(protocol)
231
330
        except KeyboardInterrupt:
247
346
 
248
347
class SmartServerSocketStreamMedium(SmartServerStreamMedium):
249
348
 
250
 
    def __init__(self, sock, backing_transport, root_client_path='/'):
 
349
    def __init__(self, sock, backing_transport, root_client_path='/',
 
350
                 timeout=None):
251
351
        """Constructor.
252
352
 
253
353
        :param sock: the socket the server will read from.  It will be put
254
354
            into blocking mode.
255
355
        """
256
356
        SmartServerStreamMedium.__init__(
257
 
            self, backing_transport, root_client_path=root_client_path)
 
357
            self, backing_transport, root_client_path=root_client_path,
 
358
            timeout=timeout)
258
359
        sock.setblocking(True)
259
360
        self.socket = sock
 
361
        # Get the getpeername now, as we might be closed later when we care.
 
362
        try:
 
363
            self._client_info = sock.getpeername()
 
364
        except socket.error:
 
365
            self._client_info = '<unknown>'
 
366
 
 
367
    def __str__(self):
 
368
        return '%s(client=%s)' % (self.__class__.__name__, self._client_info)
 
369
 
 
370
    def __repr__(self):
 
371
        return '%s.%s(client=%s)' % (self.__module__, self.__class__.__name__,
 
372
            self._client_info)
260
373
 
261
374
    def _serve_one_request_unguarded(self, protocol):
262
375
        while protocol.next_read_size():
263
376
            # We can safely try to read large chunks.  If there is less data
264
 
            # than _MAX_READ_SIZE ready, the socket wil just return a short
265
 
            # read immediately rather than block.
266
 
            bytes = self.read_bytes(_MAX_READ_SIZE)
 
377
            # than MAX_SOCKET_CHUNK ready, the socket will just return a
 
378
            # short read immediately rather than block.
 
379
            bytes = self.read_bytes(osutils.MAX_SOCKET_CHUNK)
267
380
            if bytes == '':
268
381
                self.finished = True
269
382
                return
270
383
            protocol.accept_bytes(bytes)
271
 
        
 
384
 
272
385
        self._push_back(protocol.unused_data)
273
386
 
 
387
    def _disconnect_client(self):
 
388
        """Close the current connection. We stopped due to a timeout/etc."""
 
389
        self.socket.close()
 
390
 
 
391
    def _wait_for_bytes_with_timeout(self, timeout_seconds):
 
392
        """Wait for more bytes to be read, but timeout if none available.
 
393
 
 
394
        This allows us to detect idle connections, and stop trying to read from
 
395
        them, without setting the socket itself to non-blocking. This also
 
396
        allows us to specify when we watch for idle timeouts.
 
397
 
 
398
        :return: None, this will raise ConnectionTimeout if we time out before
 
399
            data is available.
 
400
        """
 
401
        return self._wait_on_descriptor(self.socket, timeout_seconds)
 
402
 
274
403
    def _read_bytes(self, desired_count):
275
 
        # We ignore the desired_count because on sockets it's more efficient to
276
 
        # read large chunks (of _MAX_READ_SIZE bytes) at a time.
277
 
        return self.socket.recv(_MAX_READ_SIZE)
 
404
        return osutils.read_bytes_from_socket(
 
405
            self.socket, self._report_activity)
278
406
 
279
407
    def terminate_due_to_error(self):
280
408
        # TODO: This should log to a server log file, but no such thing
283
411
        self.finished = True
284
412
 
285
413
    def _write_out(self, bytes):
286
 
        osutils.send_all(self.socket, bytes)
 
414
        tstart = osutils.timer_func()
 
415
        osutils.send_all(self.socket, bytes, self._report_activity)
 
416
        if 'hpss' in debug.debug_flags:
 
417
            thread_id = thread.get_ident()
 
418
            trace.mutter('%12s: [%s] %d bytes to the socket in %.3fs'
 
419
                         % ('wrote', thread_id, len(bytes),
 
420
                            osutils.timer_func() - tstart))
287
421
 
288
422
 
289
423
class SmartServerPipeStreamMedium(SmartServerStreamMedium):
290
424
 
291
 
    def __init__(self, in_file, out_file, backing_transport):
 
425
    def __init__(self, in_file, out_file, backing_transport, timeout=None):
292
426
        """Construct new server.
293
427
 
294
428
        :param in_file: Python file from which requests can be read.
295
429
        :param out_file: Python file to write responses.
296
430
        :param backing_transport: Transport for the directory served.
297
431
        """
298
 
        SmartServerStreamMedium.__init__(self, backing_transport)
 
432
        SmartServerStreamMedium.__init__(self, backing_transport,
 
433
            timeout=timeout)
299
434
        if sys.platform == 'win32':
300
435
            # force binary mode for files
301
436
            import msvcrt
306
441
        self._in = in_file
307
442
        self._out = out_file
308
443
 
 
444
    def serve(self):
 
445
        """See SmartServerStreamMedium.serve"""
 
446
        # This is the regular serve, except it adds signal trapping for soft
 
447
        # shutdown.
 
448
        stop_gracefully = self._stop_gracefully
 
449
        signals.register_on_hangup(id(self), stop_gracefully)
 
450
        try:
 
451
            return super(SmartServerPipeStreamMedium, self).serve()
 
452
        finally:
 
453
            signals.unregister_on_hangup(id(self))
 
454
 
309
455
    def _serve_one_request_unguarded(self, protocol):
310
456
        while True:
311
457
            # We need to be careful not to read past the end of the current
324
470
                return
325
471
            protocol.accept_bytes(bytes)
326
472
 
 
473
    def _disconnect_client(self):
 
474
        self._in.close()
 
475
        self._out.flush()
 
476
        self._out.close()
 
477
 
 
478
    def _wait_for_bytes_with_timeout(self, timeout_seconds):
 
479
        """Wait for more bytes to be read, but timeout if none available.
 
480
 
 
481
        This allows us to detect idle connections, and stop trying to read from
 
482
        them, without setting the socket itself to non-blocking. This also
 
483
        allows us to specify when we watch for idle timeouts.
 
484
 
 
485
        :return: None, this will raise ConnectionTimeout if we time out before
 
486
            data is available.
 
487
        """
 
488
        if (getattr(self._in, 'fileno', None) is None
 
489
            or sys.platform == 'win32'):
 
490
            # You can't select() file descriptors on Windows.
 
491
            return
 
492
        return self._wait_on_descriptor(self._in, timeout_seconds)
 
493
 
327
494
    def _read_bytes(self, desired_count):
328
495
        return self._in.read(desired_count)
329
496
 
350
517
    request.finished_reading()
351
518
 
352
519
    It is up to the individual SmartClientMedium whether multiple concurrent
353
 
    requests can exist. See SmartClientMedium.get_request to obtain instances 
354
 
    of SmartClientMediumRequest, and the concrete Medium you are using for 
 
520
    requests can exist. See SmartClientMedium.get_request to obtain instances
 
521
    of SmartClientMediumRequest, and the concrete Medium you are using for
355
522
    details on concurrency and pipelining.
356
523
    """
357
524
 
366
533
    def accept_bytes(self, bytes):
367
534
        """Accept bytes for inclusion in this request.
368
535
 
369
 
        This method may not be be called after finished_writing() has been
 
536
        This method may not be called after finished_writing() has been
370
537
        called.  It depends upon the Medium whether or not the bytes will be
371
538
        immediately transmitted. Message based Mediums will tend to buffer the
372
539
        bytes until finished_writing() is called.
403
570
    def _finished_reading(self):
404
571
        """Helper for finished_reading.
405
572
 
406
 
        finished_reading checks the state of the request to determine if 
 
573
        finished_reading checks the state of the request to determine if
407
574
        finished_reading is allowed, and if it is hands off to _finished_reading
408
575
        to perform the action.
409
576
        """
423
590
    def _finished_writing(self):
424
591
        """Helper for finished_writing.
425
592
 
426
 
        finished_writing checks the state of the request to determine if 
 
593
        finished_writing checks the state of the request to determine if
427
594
        finished_writing is allowed, and if it is hands off to _finished_writing
428
595
        to perform the action.
429
596
        """
449
616
        read_bytes checks the state of the request to determing if bytes
450
617
        should be read. After that it hands off to _read_bytes to do the
451
618
        actual read.
452
 
        
 
619
 
453
620
        By default this forwards to self._medium.read_bytes because we are
454
621
        operating on the medium's stream.
455
622
        """
460
627
        if not line.endswith('\n'):
461
628
            # end of file encountered reading from server
462
629
            raise errors.ConnectionReset(
463
 
                "please check connectivity and permissions",
464
 
                "(and try -Dhpss if further diagnosis is required)")
 
630
                "Unexpected end of message. Please check connectivity "
 
631
                "and permissions, and report a bug if problems persist.")
465
632
        return line
466
633
 
467
634
    def _read_line(self):
468
635
        """Helper for SmartClientMediumRequest.read_line.
469
 
        
 
636
 
470
637
        By default this forwards to self._medium._get_line because we are
471
638
        operating on the medium's stream.
472
639
        """
473
640
        return self._medium._get_line()
474
641
 
475
642
 
 
643
class _VfsRefuser(object):
 
644
    """An object that refuses all VFS requests.
 
645
 
 
646
    """
 
647
 
 
648
    def __init__(self):
 
649
        client._SmartClient.hooks.install_named_hook(
 
650
            'call', self.check_vfs, 'vfs refuser')
 
651
 
 
652
    def check_vfs(self, params):
 
653
        try:
 
654
            request_method = request.request_handlers.get(params.method)
 
655
        except KeyError:
 
656
            # A method we don't know about doesn't count as a VFS method.
 
657
            return
 
658
        if issubclass(request_method, vfs.VfsRequest):
 
659
            raise errors.HpssVfsRequestNotAllowed(params.method, params.args)
 
660
 
 
661
 
476
662
class _DebugCounter(object):
477
663
    """An object that counts the HPSS calls made to each client medium.
478
664
 
479
 
    When a medium is garbage-collected, or failing that when atexit functions
480
 
    are run, the total number of calls made on that medium are reported via
481
 
    trace.note.
 
665
    When a medium is garbage-collected, or failing that when
 
666
    bzrlib.global_state exits, the total number of calls made on that medium
 
667
    are reported via trace.note.
482
668
    """
483
669
 
484
670
    def __init__(self):
485
671
        self.counts = weakref.WeakKeyDictionary()
486
672
        client._SmartClient.hooks.install_named_hook(
487
673
            'call', self.increment_call_count, 'hpss call counter')
488
 
        atexit.register(self.flush_all)
 
674
        bzrlib.global_state.cleanups.add_cleanup(self.flush_all)
489
675
 
490
676
    def track(self, medium):
491
677
        """Start tracking calls made to a medium.
495
681
        """
496
682
        medium_repr = repr(medium)
497
683
        # Add this medium to the WeakKeyDictionary
498
 
        self.counts[medium] = [0, medium_repr]
 
684
        self.counts[medium] = dict(count=0, vfs_count=0,
 
685
                                   medium_repr=medium_repr)
499
686
        # Weakref callbacks are fired in reverse order of their association
500
687
        # with the referenced object.  So we add a weakref *after* adding to
501
688
        # the WeakKeyDict so that we can report the value from it before the
505
692
    def increment_call_count(self, params):
506
693
        # Increment the count in the WeakKeyDictionary
507
694
        value = self.counts[params.medium]
508
 
        value[0] += 1
 
695
        value['count'] += 1
 
696
        try:
 
697
            request_method = request.request_handlers.get(params.method)
 
698
        except KeyError:
 
699
            # A method we don't know about doesn't count as a VFS method.
 
700
            return
 
701
        if issubclass(request_method, vfs.VfsRequest):
 
702
            value['vfs_count'] += 1
509
703
 
510
704
    def done(self, ref):
511
705
        value = self.counts[ref]
512
 
        count, medium_repr = value
 
706
        count, vfs_count, medium_repr = (
 
707
            value['count'], value['vfs_count'], value['medium_repr'])
513
708
        # In case this callback is invoked for the same ref twice (by the
514
709
        # weakref callback and by the atexit function), set the call count back
515
710
        # to 0 so this item won't be reported twice.
516
 
        value[0] = 0
 
711
        value['count'] = 0
 
712
        value['vfs_count'] = 0
517
713
        if count != 0:
518
 
            trace.note('HPSS calls: %d %s', count, medium_repr)
519
 
        
 
714
            trace.note(gettext('HPSS calls: {0} ({1} vfs) {2}').format(
 
715
                       count, vfs_count, medium_repr))
 
716
 
520
717
    def flush_all(self):
521
718
        for ref in list(self.counts.keys()):
522
719
            self.done(ref)
523
720
 
524
721
_debug_counter = None
525
 
  
526
 
  
 
722
_vfs_refuser = None
 
723
 
 
724
 
527
725
class SmartClientMedium(SmartMedium):
528
726
    """Smart client is a medium for sending smart protocol requests over."""
529
727
 
544
742
            if _debug_counter is None:
545
743
                _debug_counter = _DebugCounter()
546
744
            _debug_counter.track(self)
 
745
        if 'hpss_client_no_vfs' in debug.debug_flags:
 
746
            global _vfs_refuser
 
747
            if _vfs_refuser is None:
 
748
                _vfs_refuser = _VfsRefuser()
547
749
 
548
750
    def _is_remote_before(self, version_tuple):
549
751
        """Is it possible the remote side supports RPCs for a given version?
574
776
        """
575
777
        if (self._remote_version_is_before is not None and
576
778
            version_tuple > self._remote_version_is_before):
577
 
            raise AssertionError(
 
779
            # We have been told that the remote side is older than some version
 
780
            # which is newer than a previously supplied older-than version.
 
781
            # This indicates that some smart verb call is not guarded
 
782
            # appropriately (it should simply not have been tried).
 
783
            trace.mutter(
578
784
                "_remember_remote_is_before(%r) called, but "
579
785
                "_remember_remote_is_before(%r) was called previously."
580
 
                % (version_tuple, self._remote_version_is_before))
 
786
                , version_tuple, self._remote_version_is_before)
 
787
            if 'hpss' in debug.debug_flags:
 
788
                ui.ui_factory.show_warning(
 
789
                    "_remember_remote_is_before(%r) called, but "
 
790
                    "_remember_remote_is_before(%r) was called previously."
 
791
                    % (version_tuple, self._remote_version_is_before))
 
792
            return
581
793
        self._remote_version_is_before = version_tuple
582
794
 
583
795
    def protocol_version(self):
617
829
 
618
830
    def disconnect(self):
619
831
        """If this medium maintains a persistent connection, close it.
620
 
        
 
832
 
621
833
        The default implementation does nothing.
622
834
        """
623
 
        
 
835
 
624
836
    def remote_path_from_transport(self, transport):
625
837
        """Convert transport into a path suitable for using in a request.
626
 
        
 
838
 
627
839
        Note that the resulting remote path doesn't encode the host name or
628
840
        anything but path, so it is only safe to use it in requests sent over
629
841
        the medium from the matching transport.
630
842
        """
631
843
        medium_base = urlutils.join(self.base, '/')
632
844
        rel_url = urlutils.relative_url(medium_base, transport.base)
633
 
        return urllib.unquote(rel_url)
 
845
        return urlutils.unquote(rel_url)
634
846
 
635
847
 
636
848
class SmartClientStreamMedium(SmartClientMedium):
657
869
 
658
870
    def _flush(self):
659
871
        """Flush the output stream.
660
 
        
 
872
 
661
873
        This method is used by the SmartClientStreamMediumRequest to ensure that
662
874
        all data for a request is sent, to avoid long timeouts or deadlocks.
663
875
        """
671
883
        """
672
884
        return SmartClientStreamMediumRequest(self)
673
885
 
 
886
    def reset(self):
 
887
        """We have been disconnected, reset current state.
 
888
 
 
889
        This resets things like _current_request and connected state.
 
890
        """
 
891
        self.disconnect()
 
892
        self._current_request = None
 
893
 
674
894
 
675
895
class SmartSimplePipesClientMedium(SmartClientStreamMedium):
676
896
    """A client medium using simple pipes.
677
 
    
 
897
 
678
898
    This client does not manage the pipes: it assumes they will always be open.
679
899
    """
680
900
 
685
905
 
686
906
    def _accept_bytes(self, bytes):
687
907
        """See SmartClientStreamMedium.accept_bytes."""
688
 
        self._writeable_pipe.write(bytes)
 
908
        try:
 
909
            self._writeable_pipe.write(bytes)
 
910
        except IOError, e:
 
911
            if e.errno in (errno.EINVAL, errno.EPIPE):
 
912
                raise errors.ConnectionReset(
 
913
                    "Error trying to write to subprocess", e)
 
914
            raise
 
915
        self._report_activity(len(bytes), 'write')
689
916
 
690
917
    def _flush(self):
691
918
        """See SmartClientStreamMedium._flush()."""
 
919
        # Note: If flush were to fail, we'd like to raise ConnectionReset, etc.
 
920
        #       However, testing shows that even when the child process is
 
921
        #       gone, this doesn't error.
692
922
        self._writeable_pipe.flush()
693
923
 
694
924
    def _read_bytes(self, count):
695
925
        """See SmartClientStreamMedium._read_bytes."""
696
 
        return self._readable_pipe.read(count)
 
926
        bytes_to_read = min(count, _MAX_READ_SIZE)
 
927
        bytes = self._readable_pipe.read(bytes_to_read)
 
928
        self._report_activity(len(bytes), 'read')
 
929
        return bytes
 
930
 
 
931
 
 
932
class SSHParams(object):
 
933
    """A set of parameters for starting a remote bzr via SSH."""
 
934
 
 
935
    def __init__(self, host, port=None, username=None, password=None,
 
936
            bzr_remote_path='bzr'):
 
937
        self.host = host
 
938
        self.port = port
 
939
        self.username = username
 
940
        self.password = password
 
941
        self.bzr_remote_path = bzr_remote_path
697
942
 
698
943
 
699
944
class SmartSSHClientMedium(SmartClientStreamMedium):
700
 
    """A client medium using SSH."""
701
 
    
702
 
    def __init__(self, host, port=None, username=None, password=None,
703
 
            base=None, vendor=None, bzr_remote_path=None):
 
945
    """A client medium using SSH.
 
946
 
 
947
    It delegates IO to a SmartSimplePipesClientMedium or
 
948
    SmartClientAlreadyConnectedSocketMedium (depending on platform).
 
949
    """
 
950
 
 
951
    def __init__(self, base, ssh_params, vendor=None):
704
952
        """Creates a client that will connect on the first use.
705
 
        
 
953
 
 
954
        :param ssh_params: A SSHParams instance.
706
955
        :param vendor: An optional override for the ssh vendor to use. See
707
956
            bzrlib.transport.ssh for details on ssh vendors.
708
957
        """
 
958
        self._real_medium = None
 
959
        self._ssh_params = ssh_params
 
960
        # for the benefit of progress making a short description of this
 
961
        # transport
 
962
        self._scheme = 'bzr+ssh'
 
963
        # SmartClientStreamMedium stores the repr of this object in its
 
964
        # _DebugCounter so we have to store all the values used in our repr
 
965
        # method before calling the super init.
709
966
        SmartClientStreamMedium.__init__(self, base)
710
 
        self._connected = False
711
 
        self._host = host
712
 
        self._password = password
713
 
        self._port = port
714
 
        self._username = username
715
 
        self._read_from = None
 
967
        self._vendor = vendor
716
968
        self._ssh_connection = None
717
 
        self._vendor = vendor
718
 
        self._write_to = None
719
 
        self._bzr_remote_path = bzr_remote_path
720
 
        if self._bzr_remote_path is None:
721
 
            symbol_versioning.warn(
722
 
                'bzr_remote_path is required as of bzr 0.92',
723
 
                DeprecationWarning, stacklevel=2)
724
 
            self._bzr_remote_path = os.environ.get('BZR_REMOTE_PATH', 'bzr')
 
969
 
 
970
    def __repr__(self):
 
971
        if self._ssh_params.port is None:
 
972
            maybe_port = ''
 
973
        else:
 
974
            maybe_port = ':%s' % self._ssh_params.port
 
975
        if self._ssh_params.username is None:
 
976
            maybe_user = ''
 
977
        else:
 
978
            maybe_user = '%s@' % self._ssh_params.username
 
979
        return "%s(%s://%s%s%s/)" % (
 
980
            self.__class__.__name__,
 
981
            self._scheme,
 
982
            maybe_user,
 
983
            self._ssh_params.host,
 
984
            maybe_port)
725
985
 
726
986
    def _accept_bytes(self, bytes):
727
987
        """See SmartClientStreamMedium.accept_bytes."""
728
988
        self._ensure_connection()
729
 
        self._write_to.write(bytes)
 
989
        self._real_medium.accept_bytes(bytes)
730
990
 
731
991
    def disconnect(self):
732
992
        """See SmartClientMedium.disconnect()."""
733
 
        if not self._connected:
734
 
            return
735
 
        self._read_from.close()
736
 
        self._write_to.close()
737
 
        self._ssh_connection.close()
738
 
        self._connected = False
 
993
        if self._real_medium is not None:
 
994
            self._real_medium.disconnect()
 
995
            self._real_medium = None
 
996
        if self._ssh_connection is not None:
 
997
            self._ssh_connection.close()
 
998
            self._ssh_connection = None
739
999
 
740
1000
    def _ensure_connection(self):
741
1001
        """Connect this medium if not already connected."""
742
 
        if self._connected:
 
1002
        if self._real_medium is not None:
743
1003
            return
744
1004
        if self._vendor is None:
745
1005
            vendor = ssh._get_ssh_vendor()
746
1006
        else:
747
1007
            vendor = self._vendor
748
 
        self._ssh_connection = vendor.connect_ssh(self._username,
749
 
                self._password, self._host, self._port,
750
 
                command=[self._bzr_remote_path, 'serve', '--inet',
 
1008
        self._ssh_connection = vendor.connect_ssh(self._ssh_params.username,
 
1009
                self._ssh_params.password, self._ssh_params.host,
 
1010
                self._ssh_params.port,
 
1011
                command=[self._ssh_params.bzr_remote_path, 'serve', '--inet',
751
1012
                         '--directory=/', '--allow-writes'])
752
 
        self._read_from, self._write_to = \
753
 
            self._ssh_connection.get_filelike_channels()
754
 
        self._connected = True
 
1013
        io_kind, io_object = self._ssh_connection.get_sock_or_pipes()
 
1014
        if io_kind == 'socket':
 
1015
            self._real_medium = SmartClientAlreadyConnectedSocketMedium(
 
1016
                self.base, io_object)
 
1017
        elif io_kind == 'pipes':
 
1018
            read_from, write_to = io_object
 
1019
            self._real_medium = SmartSimplePipesClientMedium(
 
1020
                read_from, write_to, self.base)
 
1021
        else:
 
1022
            raise AssertionError(
 
1023
                "Unexpected io_kind %r from %r"
 
1024
                % (io_kind, self._ssh_connection))
 
1025
        for hook in transport.Transport.hooks["post_connect"]:
 
1026
            hook(self)
755
1027
 
756
1028
    def _flush(self):
757
1029
        """See SmartClientStreamMedium._flush()."""
758
 
        self._write_to.flush()
 
1030
        self._real_medium._flush()
759
1031
 
760
1032
    def _read_bytes(self, count):
761
1033
        """See SmartClientStreamMedium.read_bytes."""
762
 
        if not self._connected:
 
1034
        if self._real_medium is None:
763
1035
            raise errors.MediumNotConnected(self)
764
 
        bytes_to_read = min(count, _MAX_READ_SIZE)
765
 
        return self._read_from.read(bytes_to_read)
 
1036
        return self._real_medium.read_bytes(count)
766
1037
 
767
1038
 
768
1039
# Port 4155 is the default port for bzr://, registered with IANA.
770
1041
BZR_DEFAULT_PORT = 4155
771
1042
 
772
1043
 
773
 
class SmartTCPClientMedium(SmartClientStreamMedium):
774
 
    """A client medium using TCP."""
775
 
    
 
1044
class SmartClientSocketMedium(SmartClientStreamMedium):
 
1045
    """A client medium using a socket.
 
1046
 
 
1047
    This class isn't usable directly.  Use one of its subclasses instead.
 
1048
    """
 
1049
 
 
1050
    def __init__(self, base):
 
1051
        SmartClientStreamMedium.__init__(self, base)
 
1052
        self._socket = None
 
1053
        self._connected = False
 
1054
 
 
1055
    def _accept_bytes(self, bytes):
 
1056
        """See SmartClientMedium.accept_bytes."""
 
1057
        self._ensure_connection()
 
1058
        osutils.send_all(self._socket, bytes, self._report_activity)
 
1059
 
 
1060
    def _ensure_connection(self):
 
1061
        """Connect this medium if not already connected."""
 
1062
        raise NotImplementedError(self._ensure_connection)
 
1063
 
 
1064
    def _flush(self):
 
1065
        """See SmartClientStreamMedium._flush().
 
1066
 
 
1067
        For sockets we do no flushing. For TCP sockets we may want to turn off
 
1068
        TCP_NODELAY and add a means to do a flush, but that can be done in the
 
1069
        future.
 
1070
        """
 
1071
 
 
1072
    def _read_bytes(self, count):
 
1073
        """See SmartClientMedium.read_bytes."""
 
1074
        if not self._connected:
 
1075
            raise errors.MediumNotConnected(self)
 
1076
        return osutils.read_bytes_from_socket(
 
1077
            self._socket, self._report_activity)
 
1078
 
 
1079
    def disconnect(self):
 
1080
        """See SmartClientMedium.disconnect()."""
 
1081
        if not self._connected:
 
1082
            return
 
1083
        self._socket.close()
 
1084
        self._socket = None
 
1085
        self._connected = False
 
1086
 
 
1087
 
 
1088
class SmartTCPClientMedium(SmartClientSocketMedium):
 
1089
    """A client medium that creates a TCP connection."""
 
1090
 
776
1091
    def __init__(self, host, port, base):
777
1092
        """Creates a client that will connect on the first use."""
778
 
        SmartClientStreamMedium.__init__(self, base)
779
 
        self._connected = False
 
1093
        SmartClientSocketMedium.__init__(self, base)
780
1094
        self._host = host
781
1095
        self._port = port
782
 
        self._socket = None
783
 
 
784
 
    def _accept_bytes(self, bytes):
785
 
        """See SmartClientMedium.accept_bytes."""
786
 
        self._ensure_connection()
787
 
        osutils.send_all(self._socket, bytes)
788
 
 
789
 
    def disconnect(self):
790
 
        """See SmartClientMedium.disconnect()."""
791
 
        if not self._connected:
792
 
            return
793
 
        self._socket.close()
794
 
        self._socket = None
795
 
        self._connected = False
796
1096
 
797
1097
    def _ensure_connection(self):
798
1098
        """Connect this medium if not already connected."""
803
1103
        else:
804
1104
            port = int(self._port)
805
1105
        try:
806
 
            sockaddrs = socket.getaddrinfo(self._host, port, socket.AF_UNSPEC, 
 
1106
            sockaddrs = socket.getaddrinfo(self._host, port, socket.AF_UNSPEC,
807
1107
                socket.SOCK_STREAM, 0, 0)
808
1108
        except socket.gaierror, (err_num, err_msg):
809
1109
            raise errors.ConnectionError("failed to lookup %s:%d: %s" %
813
1113
        for (family, socktype, proto, canonname, sockaddr) in sockaddrs:
814
1114
            try:
815
1115
                self._socket = socket.socket(family, socktype, proto)
816
 
                self._socket.setsockopt(socket.IPPROTO_TCP, 
 
1116
                self._socket.setsockopt(socket.IPPROTO_TCP,
817
1117
                                        socket.TCP_NODELAY, 1)
818
1118
                self._socket.connect(sockaddr)
819
1119
            except socket.error, err:
832
1132
            raise errors.ConnectionError("failed to connect to %s:%d: %s" %
833
1133
                    (self._host, port, err_msg))
834
1134
        self._connected = True
835
 
 
836
 
    def _flush(self):
837
 
        """See SmartClientStreamMedium._flush().
838
 
        
839
 
        For TCP we do no flushing. We may want to turn off TCP_NODELAY and 
840
 
        add a means to do a flush, but that can be done in the future.
841
 
        """
842
 
 
843
 
    def _read_bytes(self, count):
844
 
        """See SmartClientMedium.read_bytes."""
845
 
        if not self._connected:
846
 
            raise errors.MediumNotConnected(self)
847
 
        # We ignore the desired_count because on sockets it's more efficient to
848
 
        # read large chunks (of _MAX_READ_SIZE bytes) at a time.
849
 
        try:
850
 
            return self._socket.recv(_MAX_READ_SIZE)
851
 
        except socket.error, e:
852
 
            if len(e.args) and e.args[0] == errno.ECONNRESET:
853
 
                # Callers expect an empty string in that case
854
 
                return ''
855
 
            else:
856
 
                raise
 
1135
        for hook in transport.Transport.hooks["post_connect"]:
 
1136
            hook(self)
 
1137
 
 
1138
 
 
1139
class SmartClientAlreadyConnectedSocketMedium(SmartClientSocketMedium):
 
1140
    """A client medium for an already connected socket.
 
1141
    
 
1142
    Note that this class will assume it "owns" the socket, so it will close it
 
1143
    when its disconnect method is called.
 
1144
    """
 
1145
 
 
1146
    def __init__(self, base, sock):
 
1147
        SmartClientSocketMedium.__init__(self, base)
 
1148
        self._socket = sock
 
1149
        self._connected = True
 
1150
 
 
1151
    def _ensure_connection(self):
 
1152
        # Already connected, by definition!  So nothing to do.
 
1153
        pass
857
1154
 
858
1155
 
859
1156
class SmartClientStreamMediumRequest(SmartClientMediumRequest):
872
1169
 
873
1170
    def _accept_bytes(self, bytes):
874
1171
        """See SmartClientMediumRequest._accept_bytes.
875
 
        
 
1172
 
876
1173
        This forwards to self._medium._accept_bytes because we are operating
877
1174
        on the mediums stream.
878
1175
        """
881
1178
    def _finished_reading(self):
882
1179
        """See SmartClientMediumRequest._finished_reading.
883
1180
 
884
 
        This clears the _current_request on self._medium to allow a new 
 
1181
        This clears the _current_request on self._medium to allow a new
885
1182
        request to be created.
886
1183
        """
887
1184
        if self._medium._current_request is not self:
888
1185
            raise AssertionError()
889
1186
        self._medium._current_request = None
890
 
        
 
1187
 
891
1188
    def _finished_writing(self):
892
1189
        """See SmartClientMediumRequest._finished_writing.
893
1190
 
894
1191
        This invokes self._medium._flush to ensure all bytes are transmitted.
895
1192
        """
896
1193
        self._medium._flush()
897