494
494
class _DebugCounter(object):
495
495
"""An object that counts the HPSS calls made to each client medium.
497
When a medium is garbage-collected, or failing that when
498
bzrlib.global_state exits, the total number of calls made on that medium
499
are reported via trace.note.
497
When a medium is garbage-collected, or failing that when atexit functions
498
are run, the total number of calls made on that medium are reported via
502
502
def __init__(self):
503
503
self.counts = weakref.WeakKeyDictionary()
504
504
client._SmartClient.hooks.install_named_hook(
505
505
'call', self.increment_call_count, 'hpss call counter')
506
bzrlib.global_state.cleanups.add_cleanup(self.flush_all)
506
atexit.register(self.flush_all)
508
508
def track(self, medium):
509
509
"""Start tracking calls made to a medium.
607
607
# which is newer than a previously supplied older-than version.
608
608
# This indicates that some smart verb call is not guarded
609
609
# appropriately (it should simply not have been tried).
610
raise AssertionError(
611
611
"_remember_remote_is_before(%r) called, but "
612
612
"_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))
613
% (version_tuple, self._remote_version_is_before))
620
614
self._remote_version_is_before = version_tuple
622
616
def protocol_version(self):
734
732
def _read_bytes(self, count):
735
733
"""See SmartClientStreamMedium._read_bytes."""
736
bytes_to_read = min(count, _MAX_READ_SIZE)
737
bytes = self._readable_pipe.read(bytes_to_read)
734
bytes = osutils.until_no_eintr(self._readable_pipe.read, count)
738
735
self._report_activity(len(bytes), 'read')
742
class SSHParams(object):
743
"""A set of parameters for starting a remote bzr via SSH."""
739
class SmartSSHClientMedium(SmartClientStreamMedium):
740
"""A client medium using SSH."""
745
742
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):
743
base=None, vendor=None, bzr_remote_path=None):
762
744
"""Creates a client that will connect on the first use.
764
:param ssh_params: A SSHParams instance.
765
746
:param vendor: An optional override for the ssh vendor to use. See
766
747
bzrlib.transport.ssh for details on ssh vendors.
768
self._real_medium = None
769
self._ssh_params = ssh_params
749
self._connected = False
751
self._password = password
753
self._username = username
770
754
# for the benefit of progress making a short description of this
772
756
self._scheme = 'bzr+ssh'
774
758
# _DebugCounter so we have to store all the values used in our repr
775
759
# method before calling the super init.
776
760
SmartClientStreamMedium.__init__(self, base)
761
self._read_from = None
762
self._ssh_connection = None
777
763
self._vendor = vendor
778
self._ssh_connection = None
764
self._write_to = None
765
self._bzr_remote_path = bzr_remote_path
780
767
def __repr__(self):
781
if self._ssh_params.port is None:
768
if self._port is None:
784
maybe_port = ':%s' % self._ssh_params.port
771
maybe_port = ':%s' % self._port
785
772
return "%s(%s://%s@%s%s/)" % (
786
773
self.__class__.__name__,
788
self._ssh_params.username,
789
self._ssh_params.host,
792
779
def _accept_bytes(self, bytes):
793
780
"""See SmartClientStreamMedium.accept_bytes."""
794
781
self._ensure_connection()
795
self._real_medium.accept_bytes(bytes)
782
self._write_to.write(bytes)
783
self._report_activity(len(bytes), 'write')
797
785
def disconnect(self):
798
786
"""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
787
if not self._connected:
789
self._read_from.close()
790
self._write_to.close()
791
self._ssh_connection.close()
792
self._connected = False
806
794
def _ensure_connection(self):
807
795
"""Connect this medium if not already connected."""
808
if self._real_medium is not None:
810
798
if self._vendor is None:
811
799
vendor = ssh._get_ssh_vendor()
813
801
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',
802
self._ssh_connection = vendor.connect_ssh(self._username,
803
self._password, self._host, self._port,
804
command=[self._bzr_remote_path, 'serve', '--inet',
818
805
'--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))
806
self._read_from, self._write_to = \
807
self._ssh_connection.get_filelike_channels()
808
self._connected = True
832
810
def _flush(self):
833
811
"""See SmartClientStreamMedium._flush()."""
834
self._real_medium._flush()
812
self._write_to.flush()
836
814
def _read_bytes(self, count):
837
815
"""See SmartClientStreamMedium.read_bytes."""
838
if self._real_medium is None:
816
if not self._connected:
839
817
raise errors.MediumNotConnected(self)
840
return self._real_medium.read_bytes(count)
818
bytes_to_read = min(count, _MAX_READ_SIZE)
819
bytes = self._read_from.read(bytes_to_read)
820
self._report_activity(len(bytes), 'read')
843
824
# Port 4155 is the default port for bzr://, registered with IANA.
845
826
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.
829
class SmartTCPClientMedium(SmartClientStreamMedium):
830
"""A client medium using TCP."""
854
def __init__(self, base):
832
def __init__(self, host, port, base):
833
"""Creates a client that will connect on the first use."""
855
834
SmartClientStreamMedium.__init__(self, base)
835
self._connected = False
856
838
self._socket = None
857
self._connected = False
859
840
def _accept_bytes(self, bytes):
860
841
"""See SmartClientMedium.accept_bytes."""
861
842
self._ensure_connection()
862
843
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
845
def disconnect(self):
884
846
"""See SmartClientMedium.disconnect()."""
885
847
if not self._connected:
937
889
(self._host, port, err_msg))
938
890
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.
893
"""See SmartClientStreamMedium._flush().
895
For TCP we do no flushing. We may want to turn off TCP_NODELAY and
896
add a means to do a flush, but that can be done in the future.
899
def _read_bytes(self, count):
900
"""See SmartClientMedium.read_bytes."""
901
if not self._connected:
902
raise errors.MediumNotConnected(self)
903
return osutils.read_bytes_from_socket(
904
self._socket, self._report_activity)
958
907
class SmartClientStreamMediumRequest(SmartClientMediumRequest):
added
removed
bzrlib/smart/medium.py
