← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/smart/medium.py

(jam) Add bzrlib.tests.permute_for_extension to simplify extension
        testing.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/smart/medium.py
1
 
# Copyright (C) 2006-2010 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
33
from bzrlib.lazy_import import lazy_import
32
34
lazy_import(globals(), """
33
35
import atexit
34
 
import socket
35
36
import thread
36
37
import weakref
37
38
 
46
47
from bzrlib.smart import client, protocol, request, vfs
47
48
from bzrlib.transport import ssh
48
49
""")
 
50
#usually already imported, and getting IllegalScoperReplacer on it here.
49
51
from bzrlib import osutils
50
52
 
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
 
53
# We must not read any more than 64k at a time so we don't risk "no buffer
 
54
# space available" errors on some platforms.  Windows in particular is likely
 
55
# to give error 10053 or 10055 if we read more than 64k from a socket.
 
56
_MAX_READ_SIZE = 64 * 1024
 
57
 
56
58
 
57
59
def _get_protocol_factory_for_bytes(bytes):
58
60
    """Determine the right protocol factory for 'bytes'.
274
276
    def _serve_one_request_unguarded(self, protocol):
275
277
        while protocol.next_read_size():
276
278
            # We can safely try to read large chunks.  If there is less data
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)
 
279
            # than _MAX_READ_SIZE ready, the socket wil just return a short
 
280
            # read immediately rather than block.
 
281
            bytes = self.read_bytes(_MAX_READ_SIZE)
280
282
            if bytes == '':
281
283
                self.finished = True
282
284
                return
285
287
        self._push_back(protocol.unused_data)
286
288
 
287
289
    def _read_bytes(self, desired_count):
288
 
        return osutils.read_bytes_from_socket(
289
 
            self.socket, self._report_activity)
 
290
        return _read_bytes_from_socket(
 
291
            self.socket.recv, desired_count, self._report_activity)
290
292
 
291
293
    def terminate_due_to_error(self):
292
294
        # TODO: This should log to a server log file, but no such thing
293
295
        # exists yet.  Andrew Bennetts 2006-09-29.
294
 
        self.socket.close()
 
296
        osutils.until_no_eintr(self.socket.close)
295
297
        self.finished = True
296
298
 
297
299
    def _write_out(self, bytes):
332
334
            bytes_to_read = protocol.next_read_size()
333
335
            if bytes_to_read == 0:
334
336
                # Finished serving this request.
335
 
                self._out.flush()
 
337
                osutils.until_no_eintr(self._out.flush)
336
338
                return
337
339
            bytes = self.read_bytes(bytes_to_read)
338
340
            if bytes == '':
339
341
                # Connection has been closed.
340
342
                self.finished = True
341
 
                self._out.flush()
 
343
                osutils.until_no_eintr(self._out.flush)
342
344
                return
343
345
            protocol.accept_bytes(bytes)
344
346
 
345
347
    def _read_bytes(self, desired_count):
346
 
        return self._in.read(desired_count)
 
348
        return osutils.until_no_eintr(self._in.read, desired_count)
347
349
 
348
350
    def terminate_due_to_error(self):
349
351
        # TODO: This should log to a server log file, but no such thing
350
352
        # exists yet.  Andrew Bennetts 2006-09-29.
351
 
        self._out.close()
 
353
        osutils.until_no_eintr(self._out.close)
352
354
        self.finished = True
353
355
 
354
356
    def _write_out(self, bytes):
355
 
        self._out.write(bytes)
 
357
        osutils.until_no_eintr(self._out.write, bytes)
356
358
 
357
359
 
358
360
class SmartClientMediumRequest(object):
607
609
            # which is newer than a previously supplied older-than version.
608
610
            # This indicates that some smart verb call is not guarded
609
611
            # appropriately (it should simply not have been tried).
610
 
            trace.mutter(
 
612
            raise AssertionError(
611
613
                "_remember_remote_is_before(%r) called, but "
612
614
                "_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))
619
 
            return
 
615
                % (version_tuple, self._remote_version_is_before))
620
616
        self._remote_version_is_before = version_tuple
621
617
 
622
618
    def protocol_version(self):
715
711
    """A client medium using simple pipes.
716
712
 
717
713
    This client does not manage the pipes: it assumes they will always be open.
718
 
 
719
 
    Note that if readable_pipe.read might raise IOError or OSError with errno
720
 
    of EINTR, it must be safe to retry the read.  Plain CPython fileobjects
721
 
    (such as used for sys.stdin) are safe.
722
714
    """
723
715
 
724
716
    def __init__(self, readable_pipe, writeable_pipe, base):
728
720
 
729
721
    def _accept_bytes(self, bytes):
730
722
        """See SmartClientStreamMedium.accept_bytes."""
731
 
        self._writeable_pipe.write(bytes)
 
723
        osutils.until_no_eintr(self._writeable_pipe.write, bytes)
732
724
        self._report_activity(len(bytes), 'write')
733
725
 
734
726
    def _flush(self):
735
727
        """See SmartClientStreamMedium._flush()."""
736
 
        self._writeable_pipe.flush()
 
728
        osutils.until_no_eintr(self._writeable_pipe.flush)
737
729
 
738
730
    def _read_bytes(self, count):
739
731
        """See SmartClientStreamMedium._read_bytes."""
757
749
        self._password = password
758
750
        self._port = port
759
751
        self._username = username
760
 
        # for the benefit of progress making a short description of this
761
 
        # transport
762
 
        self._scheme = 'bzr+ssh'
763
752
        # SmartClientStreamMedium stores the repr of this object in its
764
753
        # _DebugCounter so we have to store all the values used in our repr
765
754
        # method before calling the super init.
769
758
        self._vendor = vendor
770
759
        self._write_to = None
771
760
        self._bzr_remote_path = bzr_remote_path
 
761
        # for the benefit of progress making a short description of this
 
762
        # transport
 
763
        self._scheme = 'bzr+ssh'
772
764
 
773
765
    def __repr__(self):
774
 
        if self._port is None:
775
 
            maybe_port = ''
776
 
        else:
777
 
            maybe_port = ':%s' % self._port
778
 
        return "%s(%s://%s@%s%s/)" % (
 
766
        return "%s(connected=%r, username=%r, host=%r, port=%r)" % (
779
767
            self.__class__.__name__,
780
 
            self._scheme,
 
768
            self._connected,
781
769
            self._username,
782
770
            self._host,
783
 
            maybe_port)
 
771
            self._port)
784
772
 
785
773
    def _accept_bytes(self, bytes):
786
774
        """See SmartClientStreamMedium.accept_bytes."""
787
775
        self._ensure_connection()
788
 
        self._write_to.write(bytes)
 
776
        osutils.until_no_eintr(self._write_to.write, bytes)
789
777
        self._report_activity(len(bytes), 'write')
790
778
 
791
779
    def disconnect(self):
792
780
        """See SmartClientMedium.disconnect()."""
793
781
        if not self._connected:
794
782
            return
795
 
        self._read_from.close()
796
 
        self._write_to.close()
 
783
        osutils.until_no_eintr(self._read_from.close)
 
784
        osutils.until_no_eintr(self._write_to.close)
797
785
        self._ssh_connection.close()
798
786
        self._connected = False
799
787
 
822
810
        if not self._connected:
823
811
            raise errors.MediumNotConnected(self)
824
812
        bytes_to_read = min(count, _MAX_READ_SIZE)
825
 
        bytes = self._read_from.read(bytes_to_read)
 
813
        bytes = osutils.until_no_eintr(self._read_from.read, bytes_to_read)
826
814
        self._report_activity(len(bytes), 'read')
827
815
        return bytes
828
816
 
852
840
        """See SmartClientMedium.disconnect()."""
853
841
        if not self._connected:
854
842
            return
855
 
        self._socket.close()
 
843
        osutils.until_no_eintr(self._socket.close)
856
844
        self._socket = None
857
845
        self._connected = False
858
846
 
906
894
        """See SmartClientMedium.read_bytes."""
907
895
        if not self._connected:
908
896
            raise errors.MediumNotConnected(self)
909
 
        return osutils.read_bytes_from_socket(
910
 
            self._socket, self._report_activity)
 
897
        return _read_bytes_from_socket(
 
898
            self._socket.recv, count, self._report_activity)
911
899
 
912
900
 
913
901
class SmartClientStreamMediumRequest(SmartClientMediumRequest):
950
938
        self._medium._flush()
951
939
 
952
940
 
 
941
def _read_bytes_from_socket(sock, desired_count, report_activity):
 
942
    # We ignore the desired_count because on sockets it's more efficient to
 
943
    # read large chunks (of _MAX_READ_SIZE bytes) at a time.
 
944
    try:
 
945
        bytes = osutils.until_no_eintr(sock, _MAX_READ_SIZE)
 
946
    except socket.error, e:
 
947
        if len(e.args) and e.args[0] in (errno.ECONNRESET, 10054):
 
948
            # The connection was closed by the other side.  Callers expect an
 
949
            # empty string to signal end-of-stream.
 
950
            bytes = ''
 
951
        else:
 
952
            raise
 
953
    else:
 
954
        report_activity(len(bytes), 'read')
 
955
    return bytes
 
956