/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/smart/medium.py

  • Committer: Robert Collins
  • Date: 2010-05-06 11:08:10 UTC
  • mto: This revision was merged to the branch mainline in revision 5223.
  • Revision ID: robertc@robertcollins.net-20100506110810-h3j07fh5gmw54s25
Cleaner matcher matching revised unlocking protocol.

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
 
33
31
from bzrlib.lazy_import import lazy_import
34
32
lazy_import(globals(), """
35
33
import atexit
 
34
import socket
36
35
import thread
37
36
import weakref
38
37
 
47
46
from bzrlib.smart import client, protocol, request, vfs
48
47
from bzrlib.transport import ssh
49
48
""")
50
 
#usually already imported, and getting IllegalScoperReplacer on it here.
51
49
from bzrlib import osutils
52
50
 
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
 
 
 
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
58
56
 
59
57
def _get_protocol_factory_for_bytes(bytes):
60
58
    """Determine the right protocol factory for 'bytes'.
276
274
    def _serve_one_request_unguarded(self, protocol):
277
275
        while protocol.next_read_size():
278
276
            # We can safely try to read large chunks.  If there is less data
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)
 
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)
282
280
            if bytes == '':
283
281
                self.finished = True
284
282
                return
287
285
        self._push_back(protocol.unused_data)
288
286
 
289
287
    def _read_bytes(self, desired_count):
290
 
        return _read_bytes_from_socket(
291
 
            self.socket.recv, desired_count, self._report_activity)
 
288
        return osutils.read_bytes_from_socket(
 
289
            self.socket, self._report_activity)
292
290
 
293
291
    def terminate_due_to_error(self):
294
292
        # TODO: This should log to a server log file, but no such thing
295
293
        # exists yet.  Andrew Bennetts 2006-09-29.
296
 
        osutils.until_no_eintr(self.socket.close)
 
294
        self.socket.close()
297
295
        self.finished = True
298
296
 
299
297
    def _write_out(self, bytes):
334
332
            bytes_to_read = protocol.next_read_size()
335
333
            if bytes_to_read == 0:
336
334
                # Finished serving this request.
337
 
                osutils.until_no_eintr(self._out.flush)
 
335
                self._out.flush()
338
336
                return
339
337
            bytes = self.read_bytes(bytes_to_read)
340
338
            if bytes == '':
341
339
                # Connection has been closed.
342
340
                self.finished = True
343
 
                osutils.until_no_eintr(self._out.flush)
 
341
                self._out.flush()
344
342
                return
345
343
            protocol.accept_bytes(bytes)
346
344
 
347
345
    def _read_bytes(self, desired_count):
348
 
        return osutils.until_no_eintr(self._in.read, desired_count)
 
346
        return self._in.read(desired_count)
349
347
 
350
348
    def terminate_due_to_error(self):
351
349
        # TODO: This should log to a server log file, but no such thing
352
350
        # exists yet.  Andrew Bennetts 2006-09-29.
353
 
        osutils.until_no_eintr(self._out.close)
 
351
        self._out.close()
354
352
        self.finished = True
355
353
 
356
354
    def _write_out(self, bytes):
357
 
        osutils.until_no_eintr(self._out.write, bytes)
 
355
        self._out.write(bytes)
358
356
 
359
357
 
360
358
class SmartClientMediumRequest(object):
609
607
            # which is newer than a previously supplied older-than version.
610
608
            # This indicates that some smart verb call is not guarded
611
609
            # appropriately (it should simply not have been tried).
612
 
            raise AssertionError(
 
610
            trace.mutter(
613
611
                "_remember_remote_is_before(%r) called, but "
614
612
                "_remember_remote_is_before(%r) was called previously."
615
 
                % (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
616
620
        self._remote_version_is_before = version_tuple
617
621
 
618
622
    def protocol_version(self):
711
715
    """A client medium using simple pipes.
712
716
 
713
717
    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.
714
722
    """
715
723
 
716
724
    def __init__(self, readable_pipe, writeable_pipe, base):
720
728
 
721
729
    def _accept_bytes(self, bytes):
722
730
        """See SmartClientStreamMedium.accept_bytes."""
723
 
        osutils.until_no_eintr(self._writeable_pipe.write, bytes)
 
731
        self._writeable_pipe.write(bytes)
724
732
        self._report_activity(len(bytes), 'write')
725
733
 
726
734
    def _flush(self):
727
735
        """See SmartClientStreamMedium._flush()."""
728
 
        osutils.until_no_eintr(self._writeable_pipe.flush)
 
736
        self._writeable_pipe.flush()
729
737
 
730
738
    def _read_bytes(self, count):
731
739
        """See SmartClientStreamMedium._read_bytes."""
749
757
        self._password = password
750
758
        self._port = port
751
759
        self._username = username
 
760
        # for the benefit of progress making a short description of this
 
761
        # transport
 
762
        self._scheme = 'bzr+ssh'
752
763
        # SmartClientStreamMedium stores the repr of this object in its
753
764
        # _DebugCounter so we have to store all the values used in our repr
754
765
        # method before calling the super init.
758
769
        self._vendor = vendor
759
770
        self._write_to = None
760
771
        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'
764
772
 
765
773
    def __repr__(self):
766
 
        return "%s(connected=%r, username=%r, host=%r, port=%r)" % (
 
774
        if self._port is None:
 
775
            maybe_port = ''
 
776
        else:
 
777
            maybe_port = ':%s' % self._port
 
778
        return "%s(%s://%s@%s%s/)" % (
767
779
            self.__class__.__name__,
768
 
            self._connected,
 
780
            self._scheme,
769
781
            self._username,
770
782
            self._host,
771
 
            self._port)
 
783
            maybe_port)
772
784
 
773
785
    def _accept_bytes(self, bytes):
774
786
        """See SmartClientStreamMedium.accept_bytes."""
775
787
        self._ensure_connection()
776
 
        osutils.until_no_eintr(self._write_to.write, bytes)
 
788
        self._write_to.write(bytes)
777
789
        self._report_activity(len(bytes), 'write')
778
790
 
779
791
    def disconnect(self):
780
792
        """See SmartClientMedium.disconnect()."""
781
793
        if not self._connected:
782
794
            return
783
 
        osutils.until_no_eintr(self._read_from.close)
784
 
        osutils.until_no_eintr(self._write_to.close)
 
795
        self._read_from.close()
 
796
        self._write_to.close()
785
797
        self._ssh_connection.close()
786
798
        self._connected = False
787
799
 
810
822
        if not self._connected:
811
823
            raise errors.MediumNotConnected(self)
812
824
        bytes_to_read = min(count, _MAX_READ_SIZE)
813
 
        bytes = osutils.until_no_eintr(self._read_from.read, bytes_to_read)
 
825
        bytes = self._read_from.read(bytes_to_read)
814
826
        self._report_activity(len(bytes), 'read')
815
827
        return bytes
816
828
 
840
852
        """See SmartClientMedium.disconnect()."""
841
853
        if not self._connected:
842
854
            return
843
 
        osutils.until_no_eintr(self._socket.close)
 
855
        self._socket.close()
844
856
        self._socket = None
845
857
        self._connected = False
846
858
 
894
906
        """See SmartClientMedium.read_bytes."""
895
907
        if not self._connected:
896
908
            raise errors.MediumNotConnected(self)
897
 
        return _read_bytes_from_socket(
898
 
            self._socket.recv, count, self._report_activity)
 
909
        return osutils.read_bytes_from_socket(
 
910
            self._socket, self._report_activity)
899
911
 
900
912
 
901
913
class SmartClientStreamMediumRequest(SmartClientMediumRequest):
938
950
        self._medium._flush()
939
951
 
940
952
 
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