← 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-02-11 04:02:41 UTC
  • mfrom: (5017.2.2 tariff)
  • Revision ID: pqm@pqm.ubuntu.com-20100211040241-w6n021dz0uus341n
(mbp) add import-tariff tests

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."""
785
777
    def _accept_bytes(self, bytes):
786
778
        """See SmartClientStreamMedium.accept_bytes."""
787
779
        self._ensure_connection()
788
 
        self._write_to.write(bytes)
 
780
        osutils.until_no_eintr(self._write_to.write, bytes)
789
781
        self._report_activity(len(bytes), 'write')
790
782
 
791
783
    def disconnect(self):
792
784
        """See SmartClientMedium.disconnect()."""
793
785
        if not self._connected:
794
786
            return
795
 
        self._read_from.close()
796
 
        self._write_to.close()
 
787
        osutils.until_no_eintr(self._read_from.close)
 
788
        osutils.until_no_eintr(self._write_to.close)
797
789
        self._ssh_connection.close()
798
790
        self._connected = False
799
791
 
822
814
        if not self._connected:
823
815
            raise errors.MediumNotConnected(self)
824
816
        bytes_to_read = min(count, _MAX_READ_SIZE)
825
 
        bytes = self._read_from.read(bytes_to_read)
 
817
        bytes = osutils.until_no_eintr(self._read_from.read, bytes_to_read)
826
818
        self._report_activity(len(bytes), 'read')
827
819
        return bytes
828
820
 
852
844
        """See SmartClientMedium.disconnect()."""
853
845
        if not self._connected:
854
846
            return
855
 
        self._socket.close()
 
847
        osutils.until_no_eintr(self._socket.close)
856
848
        self._socket = None
857
849
        self._connected = False
858
850
 
906
898
        """See SmartClientMedium.read_bytes."""
907
899
        if not self._connected:
908
900
            raise errors.MediumNotConnected(self)
909
 
        return osutils.read_bytes_from_socket(
910
 
            self._socket, self._report_activity)
 
901
        return _read_bytes_from_socket(
 
902
            self._socket.recv, count, self._report_activity)
911
903
 
912
904
 
913
905
class SmartClientStreamMediumRequest(SmartClientMediumRequest):
950
942
        self._medium._flush()
951
943
 
952
944
 
 
945
def _read_bytes_from_socket(sock, desired_count, report_activity):
 
946
    # We ignore the desired_count because on sockets it's more efficient to
 
947
    # read large chunks (of _MAX_READ_SIZE bytes) at a time.
 
948
    try:
 
949
        bytes = osutils.until_no_eintr(sock, _MAX_READ_SIZE)
 
950
    except socket.error, e:
 
951
        if len(e.args) and e.args[0] in (errno.ECONNRESET, 10054):
 
952
            # The connection was closed by the other side.  Callers expect an
 
953
            # empty string to signal end-of-stream.
 
954
            bytes = ''
 
955
        else:
 
956
            raise
 
957
    else:
 
958
        report_activity(len(bytes), 'read')
 
959
    return bytes
 
960