← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/smart/medium.py

  • Committer: John Arbash Meinel
  • Date: 2009-06-02 19:56:24 UTC
  • mto: This revision was merged to the branch mainline in revision 4469.
  • Revision ID: john@arbash-meinel.com-20090602195624-utljsyz0qgmq63lg
Add a chunks_to_gzip function.
This allows the _record_to_data code to build up a list of chunks,
rather than requiring a single string.
It should be ~ the same performance when using a single string, since
we are only adding a for() loop over the chunks and an if check.
We could possibly just remove the if check and not worry about adding
some empty strings in there.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/smart/medium.py
1
 
# Copyright (C) 2006-2011 Canonical Ltd
 
1
# Copyright (C) 2006 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
27
28
import os
 
29
import socket
28
30
import sys
29
31
import urllib
30
32
 
31
 
import bzrlib
32
33
from bzrlib.lazy_import import lazy_import
33
34
lazy_import(globals(), """
34
 
import socket
35
 
import thread
 
35
import atexit
36
36
import weakref
37
 
 
38
37
from bzrlib import (
39
38
    debug,
40
39
    errors,
 
40
    osutils,
 
41
    symbol_versioning,
41
42
    trace,
42
43
    ui,
43
44
    urlutils,
45
46
from bzrlib.smart import client, protocol, request, vfs
46
47
from bzrlib.transport import ssh
47
48
""")
48
 
from bzrlib import osutils
49
 
 
50
 
# Throughout this module buffer size parameters are either limited to be at
51
 
# most _MAX_READ_SIZE, or are ignored and _MAX_READ_SIZE is used instead.
52
 
# For this module's purposes, MAX_SOCKET_CHUNK is a reasonable size for reads
53
 
# from non-sockets as well.
54
 
_MAX_READ_SIZE = osutils.MAX_SOCKET_CHUNK
 
49
 
 
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
 
55
56
 
56
57
def _get_protocol_factory_for_bytes(bytes):
57
58
    """Determine the right protocol factory for 'bytes'.
273
274
    def _serve_one_request_unguarded(self, protocol):
274
275
        while protocol.next_read_size():
275
276
            # We can safely try to read large chunks.  If there is less data
276
 
            # than MAX_SOCKET_CHUNK ready, the socket will just return a
277
 
            # short read immediately rather than block.
278
 
            bytes = self.read_bytes(osutils.MAX_SOCKET_CHUNK)
 
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)
279
280
            if bytes == '':
280
281
                self.finished = True
281
282
                return
284
285
        self._push_back(protocol.unused_data)
285
286
 
286
287
    def _read_bytes(self, desired_count):
287
 
        return osutils.read_bytes_from_socket(
288
 
            self.socket, self._report_activity)
 
288
        # We ignore the desired_count because on sockets it's more efficient to
 
289
        # read large chunks (of _MAX_READ_SIZE bytes) at a time.
 
290
        bytes = osutils.until_no_eintr(self.socket.recv, _MAX_READ_SIZE)
 
291
        self._report_activity(len(bytes), 'read')
 
292
        return bytes
289
293
 
290
294
    def terminate_due_to_error(self):
291
295
        # TODO: This should log to a server log file, but no such thing
294
298
        self.finished = True
295
299
 
296
300
    def _write_out(self, bytes):
297
 
        tstart = osutils.timer_func()
298
301
        osutils.send_all(self.socket, bytes, self._report_activity)
299
 
        if 'hpss' in debug.debug_flags:
300
 
            thread_id = thread.get_ident()
301
 
            trace.mutter('%12s: [%s] %d bytes to the socket in %.3fs'
302
 
                         % ('wrote', thread_id, len(bytes),
303
 
                            osutils.timer_func() - tstart))
304
302
 
305
303
 
306
304
class SmartServerPipeStreamMedium(SmartServerStreamMedium):
477
475
        if not line.endswith('\n'):
478
476
            # end of file encountered reading from server
479
477
            raise errors.ConnectionReset(
480
 
                "Unexpected end of message. Please check connectivity "
481
 
                "and permissions, and report a bug if problems persist.")
 
478
                "please check connectivity and permissions")
482
479
        return line
483
480
 
484
481
    def _read_line(self):
493
490
class _DebugCounter(object):
494
491
    """An object that counts the HPSS calls made to each client medium.
495
492
 
496
 
    When a medium is garbage-collected, or failing that when
497
 
    bzrlib.global_state exits, the total number of calls made on that medium
498
 
    are reported via trace.note.
 
493
    When a medium is garbage-collected, or failing that when atexit functions
 
494
    are run, the total number of calls made on that medium are reported via
 
495
    trace.note.
499
496
    """
500
497
 
501
498
    def __init__(self):
502
499
        self.counts = weakref.WeakKeyDictionary()
503
500
        client._SmartClient.hooks.install_named_hook(
504
501
            'call', self.increment_call_count, 'hpss call counter')
505
 
        bzrlib.global_state.cleanups.add_cleanup(self.flush_all)
 
502
        atexit.register(self.flush_all)
506
503
 
507
504
    def track(self, medium):
508
505
        """Start tracking calls made to a medium.
524
521
        # Increment the count in the WeakKeyDictionary
525
522
        value = self.counts[params.medium]
526
523
        value['count'] += 1
527
 
        try:
528
 
            request_method = request.request_handlers.get(params.method)
529
 
        except KeyError:
530
 
            # A method we don't know about doesn't count as a VFS method.
531
 
            return
 
524
        request_method = request.request_handlers.get(params.method)
532
525
        if issubclass(request_method, vfs.VfsRequest):
533
526
            value['vfs_count'] += 1
534
527
 
606
599
            # which is newer than a previously supplied older-than version.
607
600
            # This indicates that some smart verb call is not guarded
608
601
            # appropriately (it should simply not have been tried).
609
 
            trace.mutter(
 
602
            raise AssertionError(
610
603
                "_remember_remote_is_before(%r) called, but "
611
604
                "_remember_remote_is_before(%r) was called previously."
612
 
                , version_tuple, self._remote_version_is_before)
613
 
            if 'hpss' in debug.debug_flags:
614
 
                ui.ui_factory.show_warning(
615
 
                    "_remember_remote_is_before(%r) called, but "
616
 
                    "_remember_remote_is_before(%r) was called previously."
617
 
                    % (version_tuple, self._remote_version_is_before))
618
 
            return
 
605
                % (version_tuple, self._remote_version_is_before))
619
606
        self._remote_version_is_before = version_tuple
620
607
 
621
608
    def protocol_version(self):
732
719
 
733
720
    def _read_bytes(self, count):
734
721
        """See SmartClientStreamMedium._read_bytes."""
735
 
        bytes_to_read = min(count, _MAX_READ_SIZE)
736
 
        bytes = self._readable_pipe.read(bytes_to_read)
 
722
        bytes = self._readable_pipe.read(count)
737
723
        self._report_activity(len(bytes), 'read')
738
724
        return bytes
739
725
 
740
726
 
741
 
class SSHParams(object):
742
 
    """A set of parameters for starting a remote bzr via SSH."""
 
727
class SmartSSHClientMedium(SmartClientStreamMedium):
 
728
    """A client medium using SSH."""
743
729
 
744
730
    def __init__(self, host, port=None, username=None, password=None,
745
 
            bzr_remote_path='bzr'):
746
 
        self.host = host
747
 
        self.port = port
748
 
        self.username = username
749
 
        self.password = password
750
 
        self.bzr_remote_path = bzr_remote_path
751
 
 
752
 
 
753
 
class SmartSSHClientMedium(SmartClientStreamMedium):
754
 
    """A client medium using SSH.
755
 
    
756
 
    It delegates IO to a SmartClientSocketMedium or
757
 
    SmartClientAlreadyConnectedSocketMedium (depending on platform).
758
 
    """
759
 
 
760
 
    def __init__(self, base, ssh_params, vendor=None):
 
731
            base=None, vendor=None, bzr_remote_path=None):
761
732
        """Creates a client that will connect on the first use.
762
733
 
763
 
        :param ssh_params: A SSHParams instance.
764
734
        :param vendor: An optional override for the ssh vendor to use. See
765
735
            bzrlib.transport.ssh for details on ssh vendors.
766
736
        """
767
 
        self._real_medium = None
768
 
        self._ssh_params = ssh_params
769
 
        # for the benefit of progress making a short description of this
770
 
        # transport
771
 
        self._scheme = 'bzr+ssh'
 
737
        self._connected = False
 
738
        self._host = host
 
739
        self._password = password
 
740
        self._port = port
 
741
        self._username = username
772
742
        # SmartClientStreamMedium stores the repr of this object in its
773
743
        # _DebugCounter so we have to store all the values used in our repr
774
744
        # method before calling the super init.
775
745
        SmartClientStreamMedium.__init__(self, base)
 
746
        self._read_from = None
 
747
        self._ssh_connection = None
776
748
        self._vendor = vendor
777
 
        self._ssh_connection = None
 
749
        self._write_to = None
 
750
        self._bzr_remote_path = bzr_remote_path
 
751
        # for the benefit of progress making a short description of this
 
752
        # transport
 
753
        self._scheme = 'bzr+ssh'
778
754
 
779
755
    def __repr__(self):
780
 
        if self._ssh_params.port is None:
781
 
            maybe_port = ''
782
 
        else:
783
 
            maybe_port = ':%s' % self._ssh_params.port
784
 
        return "%s(%s://%s@%s%s/)" % (
 
756
        return "%s(connected=%r, username=%r, host=%r, port=%r)" % (
785
757
            self.__class__.__name__,
786
 
            self._scheme,
787
 
            self._ssh_params.username,
788
 
            self._ssh_params.host,
789
 
            maybe_port)
 
758
            self._connected,
 
759
            self._username,
 
760
            self._host,
 
761
            self._port)
790
762
 
791
763
    def _accept_bytes(self, bytes):
792
764
        """See SmartClientStreamMedium.accept_bytes."""
793
765
        self._ensure_connection()
794
 
        self._real_medium.accept_bytes(bytes)
 
766
        self._write_to.write(bytes)
 
767
        self._report_activity(len(bytes), 'write')
795
768
 
796
769
    def disconnect(self):
797
770
        """See SmartClientMedium.disconnect()."""
798
 
        if self._real_medium is not None:
799
 
            self._real_medium.disconnect()
800
 
            self._real_medium = None
801
 
        if self._ssh_connection is not None:
802
 
            self._ssh_connection.close()
803
 
            self._ssh_connection = None
 
771
        if not self._connected:
 
772
            return
 
773
        self._read_from.close()
 
774
        self._write_to.close()
 
775
        self._ssh_connection.close()
 
776
        self._connected = False
804
777
 
805
778
    def _ensure_connection(self):
806
779
        """Connect this medium if not already connected."""
807
 
        if self._real_medium is not None:
 
780
        if self._connected:
808
781
            return
809
782
        if self._vendor is None:
810
783
            vendor = ssh._get_ssh_vendor()
811
784
        else:
812
785
            vendor = self._vendor
813
 
        self._ssh_connection = vendor.connect_ssh(self._ssh_params.username,
814
 
                self._ssh_params.password, self._ssh_params.host,
815
 
                self._ssh_params.port,
816
 
                command=[self._ssh_params.bzr_remote_path, 'serve', '--inet',
 
786
        self._ssh_connection = vendor.connect_ssh(self._username,
 
787
                self._password, self._host, self._port,
 
788
                command=[self._bzr_remote_path, 'serve', '--inet',
817
789
                         '--directory=/', '--allow-writes'])
818
 
        io_kind, io_object = self._ssh_connection.get_sock_or_pipes()
819
 
        if io_kind == 'socket':
820
 
            self._real_medium = SmartClientAlreadyConnectedSocketMedium(
821
 
                self.base, io_object)
822
 
        elif io_kind == 'pipes':
823
 
            read_from, write_to = io_object
824
 
            self._real_medium = SmartSimplePipesClientMedium(
825
 
                read_from, write_to, self.base)
826
 
        else:
827
 
            raise AssertionError(
828
 
                "Unexpected io_kind %r from %r"
829
 
                % (io_kind, self._ssh_connection))
 
790
        self._read_from, self._write_to = \
 
791
            self._ssh_connection.get_filelike_channels()
 
792
        self._connected = True
830
793
 
831
794
    def _flush(self):
832
795
        """See SmartClientStreamMedium._flush()."""
833
 
        self._real_medium._flush()
 
796
        self._write_to.flush()
834
797
 
835
798
    def _read_bytes(self, count):
836
799
        """See SmartClientStreamMedium.read_bytes."""
837
 
        if self._real_medium is None:
 
800
        if not self._connected:
838
801
            raise errors.MediumNotConnected(self)
839
 
        return self._real_medium.read_bytes(count)
 
802
        bytes_to_read = min(count, _MAX_READ_SIZE)
 
803
        bytes = self._read_from.read(bytes_to_read)
 
804
        self._report_activity(len(bytes), 'read')
 
805
        return bytes
840
806
 
841
807
 
842
808
# Port 4155 is the default port for bzr://, registered with IANA.
844
810
BZR_DEFAULT_PORT = 4155
845
811
 
846
812
 
847
 
class SmartClientSocketMedium(SmartClientStreamMedium):
848
 
    """A client medium using a socket.
849
 
    
850
 
    This class isn't usable directly.  Use one of its subclasses instead.
851
 
    """
 
813
class SmartTCPClientMedium(SmartClientStreamMedium):
 
814
    """A client medium using TCP."""
852
815
 
853
 
    def __init__(self, base):
 
816
    def __init__(self, host, port, base):
 
817
        """Creates a client that will connect on the first use."""
854
818
        SmartClientStreamMedium.__init__(self, base)
 
819
        self._connected = False
 
820
        self._host = host
 
821
        self._port = port
855
822
        self._socket = None
856
 
        self._connected = False
857
823
 
858
824
    def _accept_bytes(self, bytes):
859
825
        """See SmartClientMedium.accept_bytes."""
860
826
        self._ensure_connection()
861
827
        osutils.send_all(self._socket, bytes, self._report_activity)
862
828
 
863
 
    def _ensure_connection(self):
864
 
        """Connect this medium if not already connected."""
865
 
        raise NotImplementedError(self._ensure_connection)
866
 
 
867
 
    def _flush(self):
868
 
        """See SmartClientStreamMedium._flush().
869
 
 
870
 
        For sockets we do no flushing. For TCP sockets we may want to turn off
871
 
        TCP_NODELAY and add a means to do a flush, but that can be done in the
872
 
        future.
873
 
        """
874
 
 
875
 
    def _read_bytes(self, count):
876
 
        """See SmartClientMedium.read_bytes."""
877
 
        if not self._connected:
878
 
            raise errors.MediumNotConnected(self)
879
 
        return osutils.read_bytes_from_socket(
880
 
            self._socket, self._report_activity)
881
 
 
882
829
    def disconnect(self):
883
830
        """See SmartClientMedium.disconnect()."""
884
831
        if not self._connected:
887
834
        self._socket = None
888
835
        self._connected = False
889
836
 
890
 
 
891
 
class SmartTCPClientMedium(SmartClientSocketMedium):
892
 
    """A client medium that creates a TCP connection."""
893
 
 
894
 
    def __init__(self, host, port, base):
895
 
        """Creates a client that will connect on the first use."""
896
 
        SmartClientSocketMedium.__init__(self, base)
897
 
        self._host = host
898
 
        self._port = port
899
 
 
900
837
    def _ensure_connection(self):
901
838
        """Connect this medium if not already connected."""
902
839
        if self._connected:
936
873
                    (self._host, port, err_msg))
937
874
        self._connected = True
938
875
 
939
 
 
940
 
class SmartClientAlreadyConnectedSocketMedium(SmartClientSocketMedium):
941
 
    """A client medium for an already connected socket.
942
 
    
943
 
    Note that this class will assume it "owns" the socket, so it will close it
944
 
    when its disconnect method is called.
945
 
    """
946
 
 
947
 
    def __init__(self, base, sock):
948
 
        SmartClientSocketMedium.__init__(self, base)
949
 
        self._socket = sock
950
 
        self._connected = True
951
 
 
952
 
    def _ensure_connection(self):
953
 
        # Already connected, by definition!  So nothing to do.
954
 
        pass
 
876
    def _flush(self):
 
877
        """See SmartClientStreamMedium._flush().
 
878
 
 
879
        For TCP we do no flushing. We may want to turn off TCP_NODELAY and
 
880
        add a means to do a flush, but that can be done in the future.
 
881
        """
 
882
 
 
883
    def _read_bytes(self, count):
 
884
        """See SmartClientMedium.read_bytes."""
 
885
        if not self._connected:
 
886
            raise errors.MediumNotConnected(self)
 
887
        # We ignore the desired_count because on sockets it's more efficient to
 
888
        # read large chunks (of _MAX_READ_SIZE bytes) at a time.
 
889
        try:
 
890
            bytes = osutils.until_no_eintr(self._socket.recv, _MAX_READ_SIZE)
 
891
        except socket.error, e:
 
892
            if len(e.args) and e.args[0] == errno.ECONNRESET:
 
893
                # Callers expect an empty string in that case
 
894
                return ''
 
895
            else:
 
896
                raise
 
897
        else:
 
898
            self._report_activity(len(bytes), 'read')
 
899
            return bytes
955
900
 
956
901
 
957
902
class SmartClientStreamMediumRequest(SmartClientMediumRequest):
993
938
        """
994
939
        self._medium._flush()
995
940
 
996