24
24
bzrlib/transport/smart/__init__.py.
33
32
from bzrlib.lazy_import import lazy_import
34
33
lazy_import(globals(), """
37
38
from bzrlib import (
45
from bzrlib.smart import client, protocol
46
from bzrlib.smart import client, protocol, request, vfs
46
47
from bzrlib.transport import ssh
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
49
from bzrlib import osutils
51
# Throughout this module buffer size parameters are either limited to be at
52
# most _MAX_READ_SIZE, or are ignored and _MAX_READ_SIZE is used instead.
53
# For this module's purposes, MAX_SOCKET_CHUNK is a reasonable size for reads
54
# from non-sockets as well.
55
_MAX_READ_SIZE = osutils.MAX_SOCKET_CHUNK
56
57
def _get_protocol_factory_for_bytes(bytes):
57
58
"""Determine the right protocol factory for 'bytes'.
261
274
def _serve_one_request_unguarded(self, protocol):
262
275
while protocol.next_read_size():
263
276
# 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)
277
# than MAX_SOCKET_CHUNK ready, the socket will just return a
278
# short read immediately rather than block.
279
bytes = self.read_bytes(osutils.MAX_SOCKET_CHUNK)
268
281
self.finished = True
270
283
protocol.accept_bytes(bytes)
272
285
self._push_back(protocol.unused_data)
274
287
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 osutils.until_no_eintr(self.socket.recv, _MAX_READ_SIZE)
288
return osutils.read_bytes_from_socket(
289
self.socket, self._report_activity)
279
291
def terminate_due_to_error(self):
280
292
# TODO: This should log to a server log file, but no such thing
505
524
def increment_call_count(self, params):
506
525
# Increment the count in the WeakKeyDictionary
507
526
value = self.counts[params.medium]
529
request_method = request.request_handlers.get(params.method)
531
# A method we don't know about doesn't count as a VFS method.
533
if issubclass(request_method, vfs.VfsRequest):
534
value['vfs_count'] += 1
510
536
def done(self, ref):
511
537
value = self.counts[ref]
512
count, medium_repr = value
538
count, vfs_count, medium_repr = (
539
value['count'], value['vfs_count'], value['medium_repr'])
513
540
# In case this callback is invoked for the same ref twice (by the
514
541
# weakref callback and by the atexit function), set the call count back
515
542
# to 0 so this item won't be reported twice.
544
value['vfs_count'] = 0
518
trace.note('HPSS calls: %d %s', count, medium_repr)
546
trace.note('HPSS calls: %d (%d vfs) %s',
547
count, vfs_count, medium_repr)
520
549
def flush_all(self):
521
550
for ref in list(self.counts.keys()):
524
553
_debug_counter = None
527
556
class SmartClientMedium(SmartMedium):
528
557
"""Smart client is a medium for sending smart protocol requests over."""
575
604
if (self._remote_version_is_before is not None and
576
605
version_tuple > self._remote_version_is_before):
577
raise AssertionError(
606
# We have been told that the remote side is older than some version
607
# which is newer than a previously supplied older-than version.
608
# This indicates that some smart verb call is not guarded
609
# appropriately (it should simply not have been tried).
578
611
"_remember_remote_is_before(%r) called, but "
579
612
"_remember_remote_is_before(%r) was called previously."
580
% (version_tuple, self._remote_version_is_before))
613
, version_tuple, self._remote_version_is_before)
614
if 'hpss' in debug.debug_flags:
615
ui.ui_factory.show_warning(
616
"_remember_remote_is_before(%r) called, but "
617
"_remember_remote_is_before(%r) was called previously."
618
% (version_tuple, self._remote_version_is_before))
581
620
self._remote_version_is_before = version_tuple
583
622
def protocol_version(self):
694
734
def _read_bytes(self, count):
695
735
"""See SmartClientStreamMedium._read_bytes."""
696
return self._readable_pipe.read(count)
736
bytes_to_read = min(count, _MAX_READ_SIZE)
737
bytes = self._readable_pipe.read(bytes_to_read)
738
self._report_activity(len(bytes), 'read')
742
class SSHParams(object):
743
"""A set of parameters for starting a remote bzr via SSH."""
745
def __init__(self, host, port=None, username=None, password=None,
746
bzr_remote_path='bzr'):
749
self.username = username
750
self.password = password
751
self.bzr_remote_path = bzr_remote_path
699
754
class SmartSSHClientMedium(SmartClientStreamMedium):
700
"""A client medium using SSH."""
755
"""A client medium using SSH.
702
def __init__(self, host, port=None, username=None, password=None,
703
base=None, vendor=None, bzr_remote_path=None):
757
It delegates IO to a SmartClientSocketMedium or
758
SmartClientAlreadyConnectedSocketMedium (depending on platform).
761
def __init__(self, base, ssh_params, vendor=None):
704
762
"""Creates a client that will connect on the first use.
764
:param ssh_params: A SSHParams instance.
706
765
:param vendor: An optional override for the ssh vendor to use. See
707
766
bzrlib.transport.ssh for details on ssh vendors.
768
self._real_medium = None
769
self._ssh_params = ssh_params
770
# for the benefit of progress making a short description of this
772
self._scheme = 'bzr+ssh'
773
# SmartClientStreamMedium stores the repr of this object in its
774
# _DebugCounter so we have to store all the values used in our repr
775
# method before calling the super init.
709
776
SmartClientStreamMedium.__init__(self, base)
710
self._connected = False
712
self._password = password
714
self._username = username
715
self._read_from = None
777
self._vendor = vendor
716
778
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')
781
if self._ssh_params.port is None:
784
maybe_port = ':%s' % self._ssh_params.port
785
return "%s(%s://%s@%s%s/)" % (
786
self.__class__.__name__,
788
self._ssh_params.username,
789
self._ssh_params.host,
726
792
def _accept_bytes(self, bytes):
727
793
"""See SmartClientStreamMedium.accept_bytes."""
728
794
self._ensure_connection()
729
self._write_to.write(bytes)
795
self._real_medium.accept_bytes(bytes)
731
797
def disconnect(self):
732
798
"""See SmartClientMedium.disconnect()."""
733
if not self._connected:
735
self._read_from.close()
736
self._write_to.close()
737
self._ssh_connection.close()
738
self._connected = False
799
if self._real_medium is not None:
800
self._real_medium.disconnect()
801
self._real_medium = None
802
if self._ssh_connection is not None:
803
self._ssh_connection.close()
804
self._ssh_connection = None
740
806
def _ensure_connection(self):
741
807
"""Connect this medium if not already connected."""
808
if self._real_medium is not None:
744
810
if self._vendor is None:
745
811
vendor = ssh._get_ssh_vendor()
747
813
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',
814
self._ssh_connection = vendor.connect_ssh(self._ssh_params.username,
815
self._ssh_params.password, self._ssh_params.host,
816
self._ssh_params.port,
817
command=[self._ssh_params.bzr_remote_path, 'serve', '--inet',
751
818
'--directory=/', '--allow-writes'])
752
self._read_from, self._write_to = \
753
self._ssh_connection.get_filelike_channels()
754
self._connected = True
819
io_kind, io_object = self._ssh_connection.get_sock_or_pipes()
820
if io_kind == 'socket':
821
self._real_medium = SmartClientAlreadyConnectedSocketMedium(
822
self.base, io_object)
823
elif io_kind == 'pipes':
824
read_from, write_to = io_object
825
self._real_medium = SmartSimplePipesClientMedium(
826
read_from, write_to, self.base)
828
raise AssertionError(
829
"Unexpected io_kind %r from %r"
830
% (io_kind, self._ssh_connection))
756
832
def _flush(self):
757
833
"""See SmartClientStreamMedium._flush()."""
758
self._write_to.flush()
834
self._real_medium._flush()
760
836
def _read_bytes(self, count):
761
837
"""See SmartClientStreamMedium.read_bytes."""
762
if not self._connected:
838
if self._real_medium is None:
763
839
raise errors.MediumNotConnected(self)
764
bytes_to_read = min(count, _MAX_READ_SIZE)
765
return self._read_from.read(bytes_to_read)
840
return self._real_medium.read_bytes(count)
768
843
# Port 4155 is the default port for bzr://, registered with IANA.
770
845
BZR_DEFAULT_PORT = 4155
773
class SmartTCPClientMedium(SmartClientStreamMedium):
774
"""A client medium using TCP."""
848
class SmartClientSocketMedium(SmartClientStreamMedium):
849
"""A client medium using a socket.
851
This class isn't usable directly. Use one of its subclasses instead.
854
def __init__(self, base):
855
SmartClientStreamMedium.__init__(self, base)
857
self._connected = False
859
def _accept_bytes(self, bytes):
860
"""See SmartClientMedium.accept_bytes."""
861
self._ensure_connection()
862
osutils.send_all(self._socket, bytes, self._report_activity)
864
def _ensure_connection(self):
865
"""Connect this medium if not already connected."""
866
raise NotImplementedError(self._ensure_connection)
869
"""See SmartClientStreamMedium._flush().
871
For sockets we do no flushing. For TCP sockets we may want to turn off
872
TCP_NODELAY and add a means to do a flush, but that can be done in the
876
def _read_bytes(self, count):
877
"""See SmartClientMedium.read_bytes."""
878
if not self._connected:
879
raise errors.MediumNotConnected(self)
880
return osutils.read_bytes_from_socket(
881
self._socket, self._report_activity)
883
def disconnect(self):
884
"""See SmartClientMedium.disconnect()."""
885
if not self._connected:
889
self._connected = False
892
class SmartTCPClientMedium(SmartClientSocketMedium):
893
"""A client medium that creates a TCP connection."""
776
895
def __init__(self, host, port, base):
777
896
"""Creates a client that will connect on the first use."""
778
SmartClientStreamMedium.__init__(self, base)
779
self._connected = False
897
SmartClientSocketMedium.__init__(self, base)
780
898
self._host = host
781
899
self._port = port
784
def _accept_bytes(self, bytes):
785
"""See SmartClientMedium.accept_bytes."""
786
self._ensure_connection()
787
osutils.send_all(self._socket, bytes)
789
def disconnect(self):
790
"""See SmartClientMedium.disconnect()."""
791
if not self._connected:
795
self._connected = False
797
901
def _ensure_connection(self):
798
902
"""Connect this medium if not already connected."""
833
937
(self._host, port, err_msg))
834
938
self._connected = True
837
"""See SmartClientStreamMedium._flush().
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.
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.
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
941
class SmartClientAlreadyConnectedSocketMedium(SmartClientSocketMedium):
942
"""A client medium for an already connected socket.
944
Note that this class will assume it "owns" the socket, so it will close it
945
when its disconnect method is called.
948
def __init__(self, base, sock):
949
SmartClientSocketMedium.__init__(self, base)
951
self._connected = True
953
def _ensure_connection(self):
954
# Already connected, by definition! So nothing to do.
859
958
class SmartClientStreamMediumRequest(SmartClientMediumRequest):
added
removed
bzrlib/smart/medium.py
