← 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: 2009-06-22 17:11:20 UTC
  • mfrom: (4398.8.10 1.16-commit-fulltext)
  • Revision ID: pqm@pqm.ubuntu.com-20090622171120-fuxez9ylfqpxynqn
(jam) Add VF._add_text and reduce memory overhead during commit (see
        bug #109114)

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
        return _read_bytes_from_socket(
 
289
            self.socket.recv, desired_count, self._report_activity)
289
290
 
290
291
    def terminate_due_to_error(self):
291
292
        # TODO: This should log to a server log file, but no such thing
294
295
        self.finished = True
295
296
 
296
297
    def _write_out(self, bytes):
297
 
        tstart = osutils.timer_func()
298
298
        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
299
 
305
300
 
306
301
class SmartServerPipeStreamMedium(SmartServerStreamMedium):
477
472
        if not line.endswith('\n'):
478
473
            # end of file encountered reading from server
479
474
            raise errors.ConnectionReset(
480
 
                "Unexpected end of message. Please check connectivity "
481
 
                "and permissions, and report a bug if problems persist.")
 
475
                "please check connectivity and permissions")
482
476
        return line
483
477
 
484
478
    def _read_line(self):
493
487
class _DebugCounter(object):
494
488
    """An object that counts the HPSS calls made to each client medium.
495
489
 
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.
 
490
    When a medium is garbage-collected, or failing that when atexit functions
 
491
    are run, the total number of calls made on that medium are reported via
 
492
    trace.note.
499
493
    """
500
494
 
501
495
    def __init__(self):
502
496
        self.counts = weakref.WeakKeyDictionary()
503
497
        client._SmartClient.hooks.install_named_hook(
504
498
            'call', self.increment_call_count, 'hpss call counter')
505
 
        bzrlib.global_state.cleanups.add_cleanup(self.flush_all)
 
499
        atexit.register(self.flush_all)
506
500
 
507
501
    def track(self, medium):
508
502
        """Start tracking calls made to a medium.
524
518
        # Increment the count in the WeakKeyDictionary
525
519
        value = self.counts[params.medium]
526
520
        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
 
521
        request_method = request.request_handlers.get(params.method)
532
522
        if issubclass(request_method, vfs.VfsRequest):
533
523
            value['vfs_count'] += 1
534
524
 
606
596
            # which is newer than a previously supplied older-than version.
607
597
            # This indicates that some smart verb call is not guarded
608
598
            # appropriately (it should simply not have been tried).
609
 
            trace.mutter(
 
599
            raise AssertionError(
610
600
                "_remember_remote_is_before(%r) called, but "
611
601
                "_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
 
602
                % (version_tuple, self._remote_version_is_before))
619
603
        self._remote_version_is_before = version_tuple
620
604
 
621
605
    def protocol_version(self):
732
716
 
733
717
    def _read_bytes(self, count):
734
718
        """See SmartClientStreamMedium._read_bytes."""
735
 
        bytes_to_read = min(count, _MAX_READ_SIZE)
736
 
        bytes = self._readable_pipe.read(bytes_to_read)
 
719
        bytes = self._readable_pipe.read(count)
737
720
        self._report_activity(len(bytes), 'read')
738
721
        return bytes
739
722
 
740
723
 
741
 
class SSHParams(object):
742
 
    """A set of parameters for starting a remote bzr via SSH."""
 
724
class SmartSSHClientMedium(SmartClientStreamMedium):
 
725
    """A client medium using SSH."""
743
726
 
744
727
    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):
 
728
            base=None, vendor=None, bzr_remote_path=None):
761
729
        """Creates a client that will connect on the first use.
762
730
 
763
 
        :param ssh_params: A SSHParams instance.
764
731
        :param vendor: An optional override for the ssh vendor to use. See
765
732
            bzrlib.transport.ssh for details on ssh vendors.
766
733
        """
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'
 
734
        self._connected = False
 
735
        self._host = host
 
736
        self._password = password
 
737
        self._port = port
 
738
        self._username = username
772
739
        # SmartClientStreamMedium stores the repr of this object in its
773
740
        # _DebugCounter so we have to store all the values used in our repr
774
741
        # method before calling the super init.
775
742
        SmartClientStreamMedium.__init__(self, base)
 
743
        self._read_from = None
 
744
        self._ssh_connection = None
776
745
        self._vendor = vendor
777
 
        self._ssh_connection = None
 
746
        self._write_to = None
 
747
        self._bzr_remote_path = bzr_remote_path
 
748
        # for the benefit of progress making a short description of this
 
749
        # transport
 
750
        self._scheme = 'bzr+ssh'
778
751
 
779
752
    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/)" % (
 
753
        return "%s(connected=%r, username=%r, host=%r, port=%r)" % (
785
754
            self.__class__.__name__,
786
 
            self._scheme,
787
 
            self._ssh_params.username,
788
 
            self._ssh_params.host,
789
 
            maybe_port)
 
755
            self._connected,
 
756
            self._username,
 
757
            self._host,
 
758
            self._port)
790
759
 
791
760
    def _accept_bytes(self, bytes):
792
761
        """See SmartClientStreamMedium.accept_bytes."""
793
762
        self._ensure_connection()
794
 
        self._real_medium.accept_bytes(bytes)
 
763
        self._write_to.write(bytes)
 
764
        self._report_activity(len(bytes), 'write')
795
765
 
796
766
    def disconnect(self):
797
767
        """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
 
768
        if not self._connected:
 
769
            return
 
770
        self._read_from.close()
 
771
        self._write_to.close()
 
772
        self._ssh_connection.close()
 
773
        self._connected = False
804
774
 
805
775
    def _ensure_connection(self):
806
776
        """Connect this medium if not already connected."""
807
 
        if self._real_medium is not None:
 
777
        if self._connected:
808
778
            return
809
779
        if self._vendor is None:
810
780
            vendor = ssh._get_ssh_vendor()
811
781
        else:
812
782
            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',
 
783
        self._ssh_connection = vendor.connect_ssh(self._username,
 
784
                self._password, self._host, self._port,
 
785
                command=[self._bzr_remote_path, 'serve', '--inet',
817
786
                         '--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))
 
787
        self._read_from, self._write_to = \
 
788
            self._ssh_connection.get_filelike_channels()
 
789
        self._connected = True
830
790
 
831
791
    def _flush(self):
832
792
        """See SmartClientStreamMedium._flush()."""
833
 
        self._real_medium._flush()
 
793
        self._write_to.flush()
834
794
 
835
795
    def _read_bytes(self, count):
836
796
        """See SmartClientStreamMedium.read_bytes."""
837
 
        if self._real_medium is None:
 
797
        if not self._connected:
838
798
            raise errors.MediumNotConnected(self)
839
 
        return self._real_medium.read_bytes(count)
 
799
        bytes_to_read = min(count, _MAX_READ_SIZE)
 
800
        bytes = self._read_from.read(bytes_to_read)
 
801
        self._report_activity(len(bytes), 'read')
 
802
        return bytes
840
803
 
841
804
 
842
805
# Port 4155 is the default port for bzr://, registered with IANA.
844
807
BZR_DEFAULT_PORT = 4155
845
808
 
846
809
 
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
 
    """
 
810
class SmartTCPClientMedium(SmartClientStreamMedium):
 
811
    """A client medium using TCP."""
852
812
 
853
 
    def __init__(self, base):
 
813
    def __init__(self, host, port, base):
 
814
        """Creates a client that will connect on the first use."""
854
815
        SmartClientStreamMedium.__init__(self, base)
 
816
        self._connected = False
 
817
        self._host = host
 
818
        self._port = port
855
819
        self._socket = None
856
 
        self._connected = False
857
820
 
858
821
    def _accept_bytes(self, bytes):
859
822
        """See SmartClientMedium.accept_bytes."""
860
823
        self._ensure_connection()
861
824
        osutils.send_all(self._socket, bytes, self._report_activity)
862
825
 
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
826
    def disconnect(self):
883
827
        """See SmartClientMedium.disconnect()."""
884
828
        if not self._connected:
887
831
        self._socket = None
888
832
        self._connected = False
889
833
 
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
834
    def _ensure_connection(self):
901
835
        """Connect this medium if not already connected."""
902
836
        if self._connected:
936
870
                    (self._host, port, err_msg))
937
871
        self._connected = True
938
872
 
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
 
873
    def _flush(self):
 
874
        """See SmartClientStreamMedium._flush().
 
875
 
 
876
        For TCP we do no flushing. We may want to turn off TCP_NODELAY and
 
877
        add a means to do a flush, but that can be done in the future.
 
878
        """
 
879
 
 
880
    def _read_bytes(self, count):
 
881
        """See SmartClientMedium.read_bytes."""
 
882
        if not self._connected:
 
883
            raise errors.MediumNotConnected(self)
 
884
        return _read_bytes_from_socket(
 
885
            self._socket.recv, count, self._report_activity)
955
886
 
956
887
 
957
888
class SmartClientStreamMediumRequest(SmartClientMediumRequest):
994
925
        self._medium._flush()
995
926
 
996
927
 
 
928
def _read_bytes_from_socket(sock, desired_count, report_activity):
 
929
    # We ignore the desired_count because on sockets it's more efficient to
 
930
    # read large chunks (of _MAX_READ_SIZE bytes) at a time.
 
931
    try:
 
932
        bytes = osutils.until_no_eintr(sock, _MAX_READ_SIZE)
 
933
    except socket.error, e:
 
934
        if len(e.args) and e.args[0] in (errno.ECONNRESET, 10054):
 
935
            # The connection was closed by the other side.  Callers expect an
 
936
            # empty string to signal end-of-stream.
 
937
            bytes = ''
 
938
        else:
 
939
            raise
 
940
    else:
 
941
        report_activity(len(bytes), 'read')
 
942
    return bytes
 
943