24
24
bzrlib/transport/smart/__init__.py.
32
33
from bzrlib.lazy_import import lazy_import
33
34
lazy_import(globals(), """
38
37
from bzrlib import (
46
from bzrlib.smart import client, protocol, request, vfs
46
from bzrlib.smart import client, protocol
47
47
from bzrlib.transport import ssh
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
51
# We must not read any more than 64k at a time so we don't risk "no buffer
52
# space available" errors on some platforms. Windows in particular is likely
53
# to give error 10053 or 10055 if we read more than 64k from a socket.
54
_MAX_READ_SIZE = 64 * 1024
57
57
def _get_protocol_factory_for_bytes(bytes):
58
58
"""Determine the right protocol factory for 'bytes'.
524
519
def increment_call_count(self, params):
525
520
# Increment the count in the WeakKeyDictionary
526
521
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
536
524
def done(self, ref):
537
525
value = self.counts[ref]
538
count, vfs_count, medium_repr = (
539
value['count'], value['vfs_count'], value['medium_repr'])
526
count, medium_repr = value
540
527
# In case this callback is invoked for the same ref twice (by the
541
528
# weakref callback and by the atexit function), set the call count back
542
529
# to 0 so this item won't be reported twice.
544
value['vfs_count'] = 0
546
trace.note('HPSS calls: %d (%d vfs) %s',
547
count, vfs_count, medium_repr)
532
trace.note('HPSS calls: %d %s', count, medium_repr)
549
534
def flush_all(self):
550
535
for ref in list(self.counts.keys()):
607
592
# which is newer than a previously supplied older-than version.
608
593
# This indicates that some smart verb call is not guarded
609
594
# appropriately (it should simply not have been tried).
595
raise AssertionError(
611
596
"_remember_remote_is_before(%r) called, but "
612
597
"_remember_remote_is_before(%r) was called previously."
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))
598
% (version_tuple, self._remote_version_is_before))
620
599
self._remote_version_is_before = version_tuple
622
601
def protocol_version(self):
734
713
def _read_bytes(self, count):
735
714
"""See SmartClientStreamMedium._read_bytes."""
736
bytes_to_read = min(count, _MAX_READ_SIZE)
737
bytes = self._readable_pipe.read(bytes_to_read)
715
bytes = self._readable_pipe.read(count)
738
716
self._report_activity(len(bytes), 'read')
742
class SSHParams(object):
743
"""A set of parameters for starting a remote bzr via SSH."""
720
class SmartSSHClientMedium(SmartClientStreamMedium):
721
"""A client medium using SSH."""
745
723
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
754
class SmartSSHClientMedium(SmartClientStreamMedium):
755
"""A client medium using SSH.
757
It delegates IO to a SmartClientSocketMedium or
758
SmartClientAlreadyConnectedSocketMedium (depending on platform).
761
def __init__(self, base, ssh_params, vendor=None):
724
base=None, vendor=None, bzr_remote_path=None):
762
725
"""Creates a client that will connect on the first use.
764
:param ssh_params: A SSHParams instance.
765
727
:param vendor: An optional override for the ssh vendor to use. See
766
728
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'
730
self._connected = False
732
self._password = password
734
self._username = username
773
735
# SmartClientStreamMedium stores the repr of this object in its
774
736
# _DebugCounter so we have to store all the values used in our repr
775
737
# method before calling the super init.
776
738
SmartClientStreamMedium.__init__(self, base)
739
self._read_from = None
740
self._ssh_connection = None
777
741
self._vendor = vendor
778
self._ssh_connection = None
742
self._write_to = None
743
self._bzr_remote_path = bzr_remote_path
744
# for the benefit of progress making a short description of this
746
self._scheme = 'bzr+ssh'
780
748
def __repr__(self):
781
if self._ssh_params.port is None:
784
maybe_port = ':%s' % self._ssh_params.port
785
return "%s(%s://%s@%s%s/)" % (
749
return "%s(connected=%r, username=%r, host=%r, port=%r)" % (
786
750
self.__class__.__name__,
788
self._ssh_params.username,
789
self._ssh_params.host,
792
756
def _accept_bytes(self, bytes):
793
757
"""See SmartClientStreamMedium.accept_bytes."""
794
758
self._ensure_connection()
795
self._real_medium.accept_bytes(bytes)
759
self._write_to.write(bytes)
760
self._report_activity(len(bytes), 'write')
797
762
def disconnect(self):
798
763
"""See SmartClientMedium.disconnect()."""
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
764
if not self._connected:
766
self._read_from.close()
767
self._write_to.close()
768
self._ssh_connection.close()
769
self._connected = False
806
771
def _ensure_connection(self):
807
772
"""Connect this medium if not already connected."""
808
if self._real_medium is not None:
810
775
if self._vendor is None:
811
776
vendor = ssh._get_ssh_vendor()
813
778
vendor = self._vendor
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',
779
self._ssh_connection = vendor.connect_ssh(self._username,
780
self._password, self._host, self._port,
781
command=[self._bzr_remote_path, 'serve', '--inet',
818
782
'--directory=/', '--allow-writes'])
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))
783
self._read_from, self._write_to = \
784
self._ssh_connection.get_filelike_channels()
785
self._connected = True
832
787
def _flush(self):
833
788
"""See SmartClientStreamMedium._flush()."""
834
self._real_medium._flush()
789
self._write_to.flush()
836
791
def _read_bytes(self, count):
837
792
"""See SmartClientStreamMedium.read_bytes."""
838
if self._real_medium is None:
793
if not self._connected:
839
794
raise errors.MediumNotConnected(self)
840
return self._real_medium.read_bytes(count)
795
bytes_to_read = min(count, _MAX_READ_SIZE)
796
bytes = self._read_from.read(bytes_to_read)
797
self._report_activity(len(bytes), 'read')
843
801
# Port 4155 is the default port for bzr://, registered with IANA.
845
803
BZR_DEFAULT_PORT = 4155
848
class SmartClientSocketMedium(SmartClientStreamMedium):
849
"""A client medium using a socket.
851
This class isn't usable directly. Use one of its subclasses instead.
806
class SmartTCPClientMedium(SmartClientStreamMedium):
807
"""A client medium using TCP."""
854
def __init__(self, base):
809
def __init__(self, host, port, base):
810
"""Creates a client that will connect on the first use."""
855
811
SmartClientStreamMedium.__init__(self, base)
812
self._connected = False
856
815
self._socket = None
857
self._connected = False
859
817
def _accept_bytes(self, bytes):
860
818
"""See SmartClientMedium.accept_bytes."""
861
819
self._ensure_connection()
862
820
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
822
def disconnect(self):
884
823
"""See SmartClientMedium.disconnect()."""
885
824
if not self._connected:
937
866
(self._host, port, err_msg))
938
867
self._connected = True
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.
870
"""See SmartClientStreamMedium._flush().
872
For TCP we do no flushing. We may want to turn off TCP_NODELAY and
873
add a means to do a flush, but that can be done in the future.
876
def _read_bytes(self, count):
877
"""See SmartClientMedium.read_bytes."""
878
if not self._connected:
879
raise errors.MediumNotConnected(self)
880
# We ignore the desired_count because on sockets it's more efficient to
881
# read large chunks (of _MAX_READ_SIZE bytes) at a time.
883
bytes = osutils.until_no_eintr(self._socket.recv, _MAX_READ_SIZE)
884
except socket.error, e:
885
if len(e.args) and e.args[0] == errno.ECONNRESET:
886
# Callers expect an empty string in that case
891
self._report_activity(len(bytes), 'read')
958
895
class SmartClientStreamMediumRequest(SmartClientMediumRequest):
added
removed
bzrlib/smart/medium.py
