← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/smart/medium.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2010-09-01 06:45:57 UTC
  • mfrom: (5247.2.41 more-ignored-exceptions)
  • Revision ID: pqm@pqm.ubuntu.com-20100901064557-qsxmjmp195ozbluf
(vila) Catch EPIPE when shutting down test servers. (Vincent Ladeuil)

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-2010 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
24
24
bzrlib/transport/smart/__init__.py.
25
25
"""
26
26
 
27
 
import errno
28
27
import os
29
 
import socket
30
28
import sys
31
29
import urllib
32
30
 
 
31
import bzrlib
33
32
from bzrlib.lazy_import import lazy_import
34
33
lazy_import(globals(), """
35
 
import atexit
 
34
import socket
 
35
import thread
36
36
import weakref
 
37
 
37
38
from bzrlib import (
38
39
    debug,
39
40
    errors,
45
46
from bzrlib.smart import client, protocol, request, vfs
46
47
from bzrlib.transport import ssh
47
48
""")
48
 
#usually already imported, and getting IllegalScoperReplacer on it here.
49
49
from bzrlib import osutils
50
50
 
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
55
 
 
 
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
56
 
57
57
def _get_protocol_factory_for_bytes(bytes):
58
58
    """Determine the right protocol factory for 'bytes'.
274
274
    def _serve_one_request_unguarded(self, protocol):
275
275
        while protocol.next_read_size():
276
276
            # We can safely try to read large chunks.  If there is less data
277
 
            # than _MAX_READ_SIZE ready, the socket wil just return a short
278
 
            # read immediately rather than block.
279
 
            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)
280
280
            if bytes == '':
281
281
                self.finished = True
282
282
                return
285
285
        self._push_back(protocol.unused_data)
286
286
 
287
287
    def _read_bytes(self, desired_count):
288
 
        return _read_bytes_from_socket(
289
 
            self.socket.recv, desired_count, self._report_activity)
 
288
        return osutils.read_bytes_from_socket(
 
289
            self.socket, self._report_activity)
290
290
 
291
291
    def terminate_due_to_error(self):
292
292
        # TODO: This should log to a server log file, but no such thing
293
293
        # exists yet.  Andrew Bennetts 2006-09-29.
294
 
        osutils.until_no_eintr(self.socket.close)
 
294
        self.socket.close()
295
295
        self.finished = True
296
296
 
297
297
    def _write_out(self, bytes):
 
298
        tstart = osutils.timer_func()
298
299
        osutils.send_all(self.socket, bytes, self._report_activity)
 
300
        if 'hpss' in debug.debug_flags:
 
301
            thread_id = thread.get_ident()
 
302
            trace.mutter('%12s: [%s] %d bytes to the socket in %.3fs'
 
303
                         % ('wrote', thread_id, len(bytes),
 
304
                            osutils.timer_func() - tstart))
299
305
 
300
306
 
301
307
class SmartServerPipeStreamMedium(SmartServerStreamMedium):
326
332
            bytes_to_read = protocol.next_read_size()
327
333
            if bytes_to_read == 0:
328
334
                # Finished serving this request.
329
 
                osutils.until_no_eintr(self._out.flush)
 
335
                self._out.flush()
330
336
                return
331
337
            bytes = self.read_bytes(bytes_to_read)
332
338
            if bytes == '':
333
339
                # Connection has been closed.
334
340
                self.finished = True
335
 
                osutils.until_no_eintr(self._out.flush)
 
341
                self._out.flush()
336
342
                return
337
343
            protocol.accept_bytes(bytes)
338
344
 
339
345
    def _read_bytes(self, desired_count):
340
 
        return osutils.until_no_eintr(self._in.read, desired_count)
 
346
        return self._in.read(desired_count)
341
347
 
342
348
    def terminate_due_to_error(self):
343
349
        # TODO: This should log to a server log file, but no such thing
344
350
        # exists yet.  Andrew Bennetts 2006-09-29.
345
 
        osutils.until_no_eintr(self._out.close)
 
351
        self._out.close()
346
352
        self.finished = True
347
353
 
348
354
    def _write_out(self, bytes):
349
 
        osutils.until_no_eintr(self._out.write, bytes)
 
355
        self._out.write(bytes)
350
356
 
351
357
 
352
358
class SmartClientMediumRequest(object):
488
494
class _DebugCounter(object):
489
495
    """An object that counts the HPSS calls made to each client medium.
490
496
 
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
493
 
    trace.note.
 
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.
494
500
    """
495
501
 
496
502
    def __init__(self):
497
503
        self.counts = weakref.WeakKeyDictionary()
498
504
        client._SmartClient.hooks.install_named_hook(
499
505
            'call', self.increment_call_count, 'hpss call counter')
500
 
        atexit.register(self.flush_all)
 
506
        bzrlib.global_state.cleanups.add_cleanup(self.flush_all)
501
507
 
502
508
    def track(self, medium):
503
509
        """Start tracking calls made to a medium.
601
607
            # which is newer than a previously supplied older-than version.
602
608
            # This indicates that some smart verb call is not guarded
603
609
            # appropriately (it should simply not have been tried).
604
 
            raise AssertionError(
 
610
            trace.mutter(
605
611
                "_remember_remote_is_before(%r) called, but "
606
612
                "_remember_remote_is_before(%r) was called previously."
607
 
                % (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))
 
619
            return
608
620
        self._remote_version_is_before = version_tuple
609
621
 
610
622
    def protocol_version(self):
712
724
 
713
725
    def _accept_bytes(self, bytes):
714
726
        """See SmartClientStreamMedium.accept_bytes."""
715
 
        osutils.until_no_eintr(self._writeable_pipe.write, bytes)
 
727
        self._writeable_pipe.write(bytes)
716
728
        self._report_activity(len(bytes), 'write')
717
729
 
718
730
    def _flush(self):
719
731
        """See SmartClientStreamMedium._flush()."""
720
 
        osutils.until_no_eintr(self._writeable_pipe.flush)
 
732
        self._writeable_pipe.flush()
721
733
 
722
734
    def _read_bytes(self, count):
723
735
        """See SmartClientStreamMedium._read_bytes."""
724
 
        bytes = osutils.until_no_eintr(self._readable_pipe.read, count)
 
736
        bytes_to_read = min(count, _MAX_READ_SIZE)
 
737
        bytes = self._readable_pipe.read(bytes_to_read)
725
738
        self._report_activity(len(bytes), 'read')
726
739
        return bytes
727
740
 
728
741
 
 
742
class SSHParams(object):
 
743
    """A set of parameters for starting a remote bzr via SSH."""
 
744
 
 
745
    def __init__(self, host, port=None, username=None, password=None,
 
746
            bzr_remote_path='bzr'):
 
747
        self.host = host
 
748
        self.port = port
 
749
        self.username = username
 
750
        self.password = password
 
751
        self.bzr_remote_path = bzr_remote_path
 
752
 
 
753
 
729
754
class SmartSSHClientMedium(SmartClientStreamMedium):
730
 
    """A client medium using SSH."""
 
755
    """A client medium using SSH.
 
756
    
 
757
    It delegates IO to a SmartClientSocketMedium or
 
758
    SmartClientAlreadyConnectedSocketMedium (depending on platform).
 
759
    """
731
760
 
732
 
    def __init__(self, host, port=None, username=None, password=None,
733
 
            base=None, vendor=None, bzr_remote_path=None):
 
761
    def __init__(self, base, ssh_params, vendor=None):
734
762
        """Creates a client that will connect on the first use.
735
763
 
 
764
        :param ssh_params: A SSHParams instance.
736
765
        :param vendor: An optional override for the ssh vendor to use. See
737
766
            bzrlib.transport.ssh for details on ssh vendors.
738
767
        """
739
 
        self._connected = False
740
 
        self._host = host
741
 
        self._password = password
742
 
        self._port = port
743
 
        self._username = username
 
768
        self._real_medium = None
 
769
        self._ssh_params = ssh_params
 
770
        # for the benefit of progress making a short description of this
 
771
        # transport
 
772
        self._scheme = 'bzr+ssh'
744
773
        # SmartClientStreamMedium stores the repr of this object in its
745
774
        # _DebugCounter so we have to store all the values used in our repr
746
775
        # method before calling the super init.
747
776
        SmartClientStreamMedium.__init__(self, base)
748
 
        self._read_from = None
 
777
        self._vendor = vendor
749
778
        self._ssh_connection = None
750
 
        self._vendor = vendor
751
 
        self._write_to = None
752
 
        self._bzr_remote_path = bzr_remote_path
753
 
        # for the benefit of progress making a short description of this
754
 
        # transport
755
 
        self._scheme = 'bzr+ssh'
756
779
 
757
780
    def __repr__(self):
758
 
        return "%s(connected=%r, username=%r, host=%r, port=%r)" % (
 
781
        if self._ssh_params.port is None:
 
782
            maybe_port = ''
 
783
        else:
 
784
            maybe_port = ':%s' % self._ssh_params.port
 
785
        return "%s(%s://%s@%s%s/)" % (
759
786
            self.__class__.__name__,
760
 
            self._connected,
761
 
            self._username,
762
 
            self._host,
763
 
            self._port)
 
787
            self._scheme,
 
788
            self._ssh_params.username,
 
789
            self._ssh_params.host,
 
790
            maybe_port)
764
791
 
765
792
    def _accept_bytes(self, bytes):
766
793
        """See SmartClientStreamMedium.accept_bytes."""
767
794
        self._ensure_connection()
768
 
        osutils.until_no_eintr(self._write_to.write, bytes)
769
 
        self._report_activity(len(bytes), 'write')
 
795
        self._real_medium.accept_bytes(bytes)
770
796
 
771
797
    def disconnect(self):
772
798
        """See SmartClientMedium.disconnect()."""
773
 
        if not self._connected:
774
 
            return
775
 
        osutils.until_no_eintr(self._read_from.close)
776
 
        osutils.until_no_eintr(self._write_to.close)
777
 
        self._ssh_connection.close()
778
 
        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
779
805
 
780
806
    def _ensure_connection(self):
781
807
        """Connect this medium if not already connected."""
782
 
        if self._connected:
 
808
        if self._real_medium is not None:
783
809
            return
784
810
        if self._vendor is None:
785
811
            vendor = ssh._get_ssh_vendor()
786
812
        else:
787
813
            vendor = self._vendor
788
 
        self._ssh_connection = vendor.connect_ssh(self._username,
789
 
                self._password, self._host, self._port,
790
 
                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',
791
818
                         '--directory=/', '--allow-writes'])
792
 
        self._read_from, self._write_to = \
793
 
            self._ssh_connection.get_filelike_channels()
794
 
        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)
 
827
        else:
 
828
            raise AssertionError(
 
829
                "Unexpected io_kind %r from %r"
 
830
                % (io_kind, self._ssh_connection))
795
831
 
796
832
    def _flush(self):
797
833
        """See SmartClientStreamMedium._flush()."""
798
 
        self._write_to.flush()
 
834
        self._real_medium._flush()
799
835
 
800
836
    def _read_bytes(self, count):
801
837
        """See SmartClientStreamMedium.read_bytes."""
802
 
        if not self._connected:
 
838
        if self._real_medium is None:
803
839
            raise errors.MediumNotConnected(self)
804
 
        bytes_to_read = min(count, _MAX_READ_SIZE)
805
 
        bytes = osutils.until_no_eintr(self._read_from.read, bytes_to_read)
806
 
        self._report_activity(len(bytes), 'read')
807
 
        return bytes
 
840
        return self._real_medium.read_bytes(count)
808
841
 
809
842
 
810
843
# Port 4155 is the default port for bzr://, registered with IANA.
812
845
BZR_DEFAULT_PORT = 4155
813
846
 
814
847
 
815
 
class SmartTCPClientMedium(SmartClientStreamMedium):
816
 
    """A client medium using TCP."""
 
848
class SmartClientSocketMedium(SmartClientStreamMedium):
 
849
    """A client medium using a socket.
 
850
    
 
851
    This class isn't usable directly.  Use one of its subclasses instead.
 
852
    """
 
853
 
 
854
    def __init__(self, base):
 
855
        SmartClientStreamMedium.__init__(self, base)
 
856
        self._socket = None
 
857
        self._connected = False
 
858
 
 
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)
 
863
 
 
864
    def _ensure_connection(self):
 
865
        """Connect this medium if not already connected."""
 
866
        raise NotImplementedError(self._ensure_connection)
 
867
 
 
868
    def _flush(self):
 
869
        """See SmartClientStreamMedium._flush().
 
870
 
 
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
 
873
        future.
 
874
        """
 
875
 
 
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)
 
882
 
 
883
    def disconnect(self):
 
884
        """See SmartClientMedium.disconnect()."""
 
885
        if not self._connected:
 
886
            return
 
887
        self._socket.close()
 
888
        self._socket = None
 
889
        self._connected = False
 
890
 
 
891
 
 
892
class SmartTCPClientMedium(SmartClientSocketMedium):
 
893
    """A client medium that creates a TCP connection."""
817
894
 
818
895
    def __init__(self, host, port, base):
819
896
        """Creates a client that will connect on the first use."""
820
 
        SmartClientStreamMedium.__init__(self, base)
821
 
        self._connected = False
 
897
        SmartClientSocketMedium.__init__(self, base)
822
898
        self._host = host
823
899
        self._port = port
824
 
        self._socket = None
825
 
 
826
 
    def _accept_bytes(self, bytes):
827
 
        """See SmartClientMedium.accept_bytes."""
828
 
        self._ensure_connection()
829
 
        osutils.send_all(self._socket, bytes, self._report_activity)
830
 
 
831
 
    def disconnect(self):
832
 
        """See SmartClientMedium.disconnect()."""
833
 
        if not self._connected:
834
 
            return
835
 
        osutils.until_no_eintr(self._socket.close)
836
 
        self._socket = None
837
 
        self._connected = False
838
900
 
839
901
    def _ensure_connection(self):
840
902
        """Connect this medium if not already connected."""
875
937
                    (self._host, port, err_msg))
876
938
        self._connected = True
877
939
 
878
 
    def _flush(self):
879
 
        """See SmartClientStreamMedium._flush().
880
 
 
881
 
        For TCP we do no flushing. We may want to turn off TCP_NODELAY and
882
 
        add a means to do a flush, but that can be done in the future.
883
 
        """
884
 
 
885
 
    def _read_bytes(self, count):
886
 
        """See SmartClientMedium.read_bytes."""
887
 
        if not self._connected:
888
 
            raise errors.MediumNotConnected(self)
889
 
        return _read_bytes_from_socket(
890
 
            self._socket.recv, count, self._report_activity)
 
940
 
 
941
class SmartClientAlreadyConnectedSocketMedium(SmartClientSocketMedium):
 
942
    """A client medium for an already connected socket.
 
943
    
 
944
    Note that this class will assume it "owns" the socket, so it will close it
 
945
    when its disconnect method is called.
 
946
    """
 
947
 
 
948
    def __init__(self, base, sock):
 
949
        SmartClientSocketMedium.__init__(self, base)
 
950
        self._socket = sock
 
951
        self._connected = True
 
952
 
 
953
    def _ensure_connection(self):
 
954
        # Already connected, by definition!  So nothing to do.
 
955
        pass
891
956
 
892
957
 
893
958
class SmartClientStreamMediumRequest(SmartClientMediumRequest):
930
995
        self._medium._flush()
931
996
 
932
997
 
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.
936
 
    try:
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.
942
 
            bytes = ''
943
 
        else:
944
 
            raise
945
 
    else:
946
 
        report_activity(len(bytes), 'read')
947
 
    return bytes
948