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
17
17
"""The 'medium' layer for the smart servers and clients.
32
33
from bzrlib.lazy_import import lazy_import
33
34
lazy_import(globals(), """
34
37
from bzrlib import (
40
from bzrlib.smart import protocol
45
from bzrlib.smart import client, protocol, request, vfs
41
46
from bzrlib.transport import ssh
48
#usually already imported, and getting IllegalScoperReplacer on it here.
49
from bzrlib import osutils
45
51
# We must not read any more than 64k at a time so we don't risk "no buffer
46
52
# space available" errors on some platforms. Windows in particular is likely
156
162
line, excess = _get_line(self.read_bytes)
157
163
self._push_back(excess)
166
def _report_activity(self, bytes, direction):
167
"""Notify that this medium has activity.
169
Implementations should call this from all methods that actually do IO.
170
Be careful that it's not called twice, if one method is implemented on
173
:param bytes: Number of bytes read or written.
174
:param direction: 'read' or 'write' or None.
176
ui.ui_factory.report_transport_activity(self, bytes, direction)
161
179
class SmartServerStreamMedium(SmartMedium):
162
180
"""Handles smart commands coming over a stream.
263
281
self.finished = True
265
283
protocol.accept_bytes(bytes)
267
285
self._push_back(protocol.unused_data)
269
287
def _read_bytes(self, desired_count):
270
# We ignore the desired_count because on sockets it's more efficient to
271
# read large chunks (of _MAX_READ_SIZE bytes) at a time.
272
return self.socket.recv(_MAX_READ_SIZE)
288
return _read_bytes_from_socket(
289
self.socket.recv, desired_count, self._report_activity)
274
291
def terminate_due_to_error(self):
275
292
# TODO: This should log to a server log file, but no such thing
345
362
request.finished_reading()
347
364
It is up to the individual SmartClientMedium whether multiple concurrent
348
requests can exist. See SmartClientMedium.get_request to obtain instances
349
of SmartClientMediumRequest, and the concrete Medium you are using for
365
requests can exist. See SmartClientMedium.get_request to obtain instances
366
of SmartClientMediumRequest, and the concrete Medium you are using for
350
367
details on concurrency and pipelining.
361
378
def accept_bytes(self, bytes):
362
379
"""Accept bytes for inclusion in this request.
364
This method may not be be called after finished_writing() has been
381
This method may not be called after finished_writing() has been
365
382
called. It depends upon the Medium whether or not the bytes will be
366
383
immediately transmitted. Message based Mediums will tend to buffer the
367
384
bytes until finished_writing() is called.
455
472
if not line.endswith('\n'):
456
473
# end of file encountered reading from server
457
474
raise errors.ConnectionReset(
458
"please check connectivity and permissions",
459
"(and try -Dhpss if further diagnosis is required)")
475
"Unexpected end of message. Please check connectivity "
476
"and permissions, and report a bug if problems persist.")
462
479
def _read_line(self):
463
480
"""Helper for SmartClientMediumRequest.read_line.
465
482
By default this forwards to self._medium._get_line because we are
466
483
operating on the medium's stream.
468
485
return self._medium._get_line()
488
class _DebugCounter(object):
489
"""An object that counts the HPSS calls made to each client medium.
491
When a medium is garbage-collected, or failing that when atexit functions
492
are run, the total number of calls made on that medium are reported via
497
self.counts = weakref.WeakKeyDictionary()
498
client._SmartClient.hooks.install_named_hook(
499
'call', self.increment_call_count, 'hpss call counter')
500
atexit.register(self.flush_all)
502
def track(self, medium):
503
"""Start tracking calls made to a medium.
505
This only keeps a weakref to the medium, so shouldn't affect the
508
medium_repr = repr(medium)
509
# Add this medium to the WeakKeyDictionary
510
self.counts[medium] = dict(count=0, vfs_count=0,
511
medium_repr=medium_repr)
512
# Weakref callbacks are fired in reverse order of their association
513
# with the referenced object. So we add a weakref *after* adding to
514
# the WeakKeyDict so that we can report the value from it before the
515
# entry is removed by the WeakKeyDict's own callback.
516
ref = weakref.ref(medium, self.done)
518
def increment_call_count(self, params):
519
# Increment the count in the WeakKeyDictionary
520
value = self.counts[params.medium]
523
request_method = request.request_handlers.get(params.method)
525
# A method we don't know about doesn't count as a VFS method.
527
if issubclass(request_method, vfs.VfsRequest):
528
value['vfs_count'] += 1
531
value = self.counts[ref]
532
count, vfs_count, medium_repr = (
533
value['count'], value['vfs_count'], value['medium_repr'])
534
# In case this callback is invoked for the same ref twice (by the
535
# weakref callback and by the atexit function), set the call count back
536
# to 0 so this item won't be reported twice.
538
value['vfs_count'] = 0
540
trace.note('HPSS calls: %d (%d vfs) %s',
541
count, vfs_count, medium_repr)
544
for ref in list(self.counts.keys()):
547
_debug_counter = None
471
550
class SmartClientMedium(SmartMedium):
472
551
"""Smart client is a medium for sending smart protocol requests over."""
482
561
# _remote_version_is_before tracks the bzr version the remote side
483
562
# can be based on what we've seen so far.
484
563
self._remote_version_is_before = None
564
# Install debug hook function if debug flag is set.
565
if 'hpss' in debug.debug_flags:
566
global _debug_counter
567
if _debug_counter is None:
568
_debug_counter = _DebugCounter()
569
_debug_counter.track(self)
486
571
def _is_remote_before(self, version_tuple):
487
572
"""Is it possible the remote side supports RPCs for a given version?
513
598
if (self._remote_version_is_before is not None and
514
599
version_tuple > self._remote_version_is_before):
600
# We have been told that the remote side is older than some version
601
# which is newer than a previously supplied older-than version.
602
# This indicates that some smart verb call is not guarded
603
# appropriately (it should simply not have been tried).
515
604
raise AssertionError(
516
605
"_remember_remote_is_before(%r) called, but "
517
606
"_remember_remote_is_before(%r) was called previously."
556
645
def disconnect(self):
557
646
"""If this medium maintains a persistent connection, close it.
559
648
The default implementation does nothing.
562
651
def remote_path_from_transport(self, transport):
563
652
"""Convert transport into a path suitable for using in a request.
565
654
Note that the resulting remote path doesn't encode the host name or
566
655
anything but path, so it is only safe to use it in requests sent over
567
656
the medium from the matching transport.
632
722
def _read_bytes(self, count):
633
723
"""See SmartClientStreamMedium._read_bytes."""
634
return self._readable_pipe.read(count)
724
bytes = self._readable_pipe.read(count)
725
self._report_activity(len(bytes), 'read')
637
729
class SmartSSHClientMedium(SmartClientStreamMedium):
638
730
"""A client medium using SSH."""
640
732
def __init__(self, host, port=None, username=None, password=None,
641
733
base=None, vendor=None, bzr_remote_path=None):
642
734
"""Creates a client that will connect on the first use.
644
736
:param vendor: An optional override for the ssh vendor to use. See
645
737
bzrlib.transport.ssh for details on ssh vendors.
647
SmartClientStreamMedium.__init__(self, base)
648
739
self._connected = False
649
740
self._host = host
650
741
self._password = password
651
742
self._port = port
652
743
self._username = username
744
# SmartClientStreamMedium stores the repr of this object in its
745
# _DebugCounter so we have to store all the values used in our repr
746
# method before calling the super init.
747
SmartClientStreamMedium.__init__(self, base)
653
748
self._read_from = None
654
749
self._ssh_connection = None
655
750
self._vendor = vendor
656
751
self._write_to = None
657
752
self._bzr_remote_path = bzr_remote_path
658
if self._bzr_remote_path is None:
659
symbol_versioning.warn(
660
'bzr_remote_path is required as of bzr 0.92',
661
DeprecationWarning, stacklevel=2)
662
self._bzr_remote_path = os.environ.get('BZR_REMOTE_PATH', 'bzr')
753
# for the benefit of progress making a short description of this
755
self._scheme = 'bzr+ssh'
758
return "%s(connected=%r, username=%r, host=%r, port=%r)" % (
759
self.__class__.__name__,
664
765
def _accept_bytes(self, bytes):
665
766
"""See SmartClientStreamMedium.accept_bytes."""
666
767
self._ensure_connection()
667
768
self._write_to.write(bytes)
769
self._report_activity(len(bytes), 'write')
669
771
def disconnect(self):
670
772
"""See SmartClientMedium.disconnect()."""
700
802
if not self._connected:
701
803
raise errors.MediumNotConnected(self)
702
804
bytes_to_read = min(count, _MAX_READ_SIZE)
703
return self._read_from.read(bytes_to_read)
805
bytes = self._read_from.read(bytes_to_read)
806
self._report_activity(len(bytes), 'read')
706
810
# Port 4155 is the default port for bzr://, registered with IANA.
707
BZR_DEFAULT_INTERFACE = '0.0.0.0'
811
BZR_DEFAULT_INTERFACE = None
708
812
BZR_DEFAULT_PORT = 4155
711
815
class SmartTCPClientMedium(SmartClientStreamMedium):
712
816
"""A client medium using TCP."""
714
818
def __init__(self, host, port, base):
715
819
"""Creates a client that will connect on the first use."""
716
820
SmartClientStreamMedium.__init__(self, base)
736
840
"""Connect this medium if not already connected."""
737
841
if self._connected:
739
self._socket = socket.socket()
740
self._socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
741
843
if self._port is None:
742
844
port = BZR_DEFAULT_PORT
744
846
port = int(self._port)
746
self._socket.connect((self._host, port))
747
except socket.error, err:
848
sockaddrs = socket.getaddrinfo(self._host, port, socket.AF_UNSPEC,
849
socket.SOCK_STREAM, 0, 0)
850
except socket.gaierror, (err_num, err_msg):
851
raise errors.ConnectionError("failed to lookup %s:%d: %s" %
852
(self._host, port, err_msg))
853
# Initialize err in case there are no addresses returned:
854
err = socket.error("no address found for %s" % self._host)
855
for (family, socktype, proto, canonname, sockaddr) in sockaddrs:
857
self._socket = socket.socket(family, socktype, proto)
858
self._socket.setsockopt(socket.IPPROTO_TCP,
859
socket.TCP_NODELAY, 1)
860
self._socket.connect(sockaddr)
861
except socket.error, err:
862
if self._socket is not None:
867
if self._socket is None:
748
868
# socket errors either have a (string) or (errno, string) as their
750
870
if type(err.args) is str:
766
886
"""See SmartClientMedium.read_bytes."""
767
887
if not self._connected:
768
888
raise errors.MediumNotConnected(self)
769
# We ignore the desired_count because on sockets it's more efficient to
770
# read large chunks (of _MAX_READ_SIZE bytes) at a time.
771
return self._socket.recv(_MAX_READ_SIZE)
889
return _read_bytes_from_socket(
890
self._socket.recv, count, self._report_activity)
774
893
class SmartClientStreamMediumRequest(SmartClientMediumRequest):
796
915
def _finished_reading(self):
797
916
"""See SmartClientMediumRequest._finished_reading.
799
This clears the _current_request on self._medium to allow a new
918
This clears the _current_request on self._medium to allow a new
800
919
request to be created.
802
921
if self._medium._current_request is not self:
803
922
raise AssertionError()
804
923
self._medium._current_request = None
806
925
def _finished_writing(self):
807
926
"""See SmartClientMediumRequest._finished_writing.
811
930
self._medium._flush()
933
def _read_bytes_from_socket(sock, desired_count, report_activity):
934
# We ignore the desired_count because on sockets it's more efficient to
935
# read large chunks (of _MAX_READ_SIZE bytes) at a time.
937
bytes = osutils.until_no_eintr(sock, _MAX_READ_SIZE)
938
except socket.error, e:
939
if len(e.args) and e.args[0] in (errno.ECONNRESET, 10054):
940
# The connection was closed by the other side. Callers expect an
941
# empty string to signal end-of-stream.
946
report_activity(len(bytes), 'read')
added
removed
bzrlib/smart/medium.py
