← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/smart/medium.py

  • Committer: Danny van Heumen
  • Date: 2010-03-09 21:42:11 UTC
  • mto: (4634.139.5 2.0)
  • mto: This revision was merged to the branch mainline in revision 5160.
  • Revision ID: danny@dannyvanheumen.nl-20100309214211-iqh42x6qcikgd9p3
Reverted now-useless TODO list.

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
 
import thread
36
36
import weakref
37
 
 
38
37
from bzrlib import (
39
38
    debug,
40
39
    errors,
46
45
from bzrlib.smart import client, protocol, request, vfs
47
46
from bzrlib.transport import ssh
48
47
""")
 
48
#usually already imported, and getting IllegalScoperReplacer on it here.
49
49
from bzrlib import osutils
50
50
 
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
 
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
 
56
56
 
57
57
def _get_protocol_factory_for_bytes(bytes):
58
58
    """Determine the right protocol factory for 'bytes'.
274
274
    def _serve_one_request_unguarded(self, protocol):
275
275
        while protocol.next_read_size():
276
276
            # 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)
 
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)
280
280
            if bytes == '':
281
281
                self.finished = True
282
282
                return
285
285
        self._push_back(protocol.unused_data)
286
286
 
287
287
    def _read_bytes(self, desired_count):
288
 
        return osutils.read_bytes_from_socket(
289
 
            self.socket, self._report_activity)
 
288
        return _read_bytes_from_socket(
 
289
            self.socket.recv, desired_count, self._report_activity)
290
290
 
291
291
    def terminate_due_to_error(self):
292
292
        # TODO: This should log to a server log file, but no such thing
293
293
        # exists yet.  Andrew Bennetts 2006-09-29.
294
 
        self.socket.close()
 
294
        osutils.until_no_eintr(self.socket.close)
295
295
        self.finished = True
296
296
 
297
297
    def _write_out(self, bytes):
298
 
        tstart = osutils.timer_func()
299
298
        osutils.send_all(self.socket, bytes, self._report_activity)
300
 
        if 'hpss' in debug.debug_flags:
301
 
            thread_id = thread.get_ident()
302
 
            trace.mutter('%12s: [%s] %d bytes to the socket in %.3fs'
303
 
                         % ('wrote', thread_id, len(bytes),
304
 
                            osutils.timer_func() - tstart))
305
299
 
306
300
 
307
301
class SmartServerPipeStreamMedium(SmartServerStreamMedium):
332
326
            bytes_to_read = protocol.next_read_size()
333
327
            if bytes_to_read == 0:
334
328
                # Finished serving this request.
335
 
                self._out.flush()
 
329
                osutils.until_no_eintr(self._out.flush)
336
330
                return
337
331
            bytes = self.read_bytes(bytes_to_read)
338
332
            if bytes == '':
339
333
                # Connection has been closed.
340
334
                self.finished = True
341
 
                self._out.flush()
 
335
                osutils.until_no_eintr(self._out.flush)
342
336
                return
343
337
            protocol.accept_bytes(bytes)
344
338
 
345
339
    def _read_bytes(self, desired_count):
346
 
        return self._in.read(desired_count)
 
340
        return osutils.until_no_eintr(self._in.read, desired_count)
347
341
 
348
342
    def terminate_due_to_error(self):
349
343
        # TODO: This should log to a server log file, but no such thing
350
344
        # exists yet.  Andrew Bennetts 2006-09-29.
351
 
        self._out.close()
 
345
        osutils.until_no_eintr(self._out.close)
352
346
        self.finished = True
353
347
 
354
348
    def _write_out(self, bytes):
355
 
        self._out.write(bytes)
 
349
        osutils.until_no_eintr(self._out.write, bytes)
356
350
 
357
351
 
358
352
class SmartClientMediumRequest(object):
607
601
            # which is newer than a previously supplied older-than version.
608
602
            # This indicates that some smart verb call is not guarded
609
603
            # appropriately (it should simply not have been tried).
610
 
            trace.mutter(
 
604
            raise AssertionError(
611
605
                "_remember_remote_is_before(%r) called, but "
612
606
                "_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
 
607
                % (version_tuple, self._remote_version_is_before))
620
608
        self._remote_version_is_before = version_tuple
621
609
 
622
610
    def protocol_version(self):
715
703
    """A client medium using simple pipes.
716
704
 
717
705
    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
706
    """
723
707
 
724
708
    def __init__(self, readable_pipe, writeable_pipe, base):
728
712
 
729
713
    def _accept_bytes(self, bytes):
730
714
        """See SmartClientStreamMedium.accept_bytes."""
731
 
        self._writeable_pipe.write(bytes)
 
715
        osutils.until_no_eintr(self._writeable_pipe.write, bytes)
732
716
        self._report_activity(len(bytes), 'write')
733
717
 
734
718
    def _flush(self):
735
719
        """See SmartClientStreamMedium._flush()."""
736
 
        self._writeable_pipe.flush()
 
720
        osutils.until_no_eintr(self._writeable_pipe.flush)
737
721
 
738
722
    def _read_bytes(self, count):
739
723
        """See SmartClientStreamMedium._read_bytes."""
757
741
        self._password = password
758
742
        self._port = port
759
743
        self._username = username
760
 
        # for the benefit of progress making a short description of this
761
 
        # transport
762
 
        self._scheme = 'bzr+ssh'
763
744
        # SmartClientStreamMedium stores the repr of this object in its
764
745
        # _DebugCounter so we have to store all the values used in our repr
765
746
        # method before calling the super init.
769
750
        self._vendor = vendor
770
751
        self._write_to = None
771
752
        self._bzr_remote_path = bzr_remote_path
 
753
        # for the benefit of progress making a short description of this
 
754
        # transport
 
755
        self._scheme = 'bzr+ssh'
772
756
 
773
757
    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/)" % (
 
758
        return "%s(connected=%r, username=%r, host=%r, port=%r)" % (
779
759
            self.__class__.__name__,
780
 
            self._scheme,
 
760
            self._connected,
781
761
            self._username,
782
762
            self._host,
783
 
            maybe_port)
 
763
            self._port)
784
764
 
785
765
    def _accept_bytes(self, bytes):
786
766
        """See SmartClientStreamMedium.accept_bytes."""
787
767
        self._ensure_connection()
788
 
        self._write_to.write(bytes)
 
768
        osutils.until_no_eintr(self._write_to.write, bytes)
789
769
        self._report_activity(len(bytes), 'write')
790
770
 
791
771
    def disconnect(self):
792
772
        """See SmartClientMedium.disconnect()."""
793
773
        if not self._connected:
794
774
            return
795
 
        self._read_from.close()
796
 
        self._write_to.close()
 
775
        osutils.until_no_eintr(self._read_from.close)
 
776
        osutils.until_no_eintr(self._write_to.close)
797
777
        self._ssh_connection.close()
798
778
        self._connected = False
799
779
 
822
802
        if not self._connected:
823
803
            raise errors.MediumNotConnected(self)
824
804
        bytes_to_read = min(count, _MAX_READ_SIZE)
825
 
        bytes = self._read_from.read(bytes_to_read)
 
805
        bytes = osutils.until_no_eintr(self._read_from.read, bytes_to_read)
826
806
        self._report_activity(len(bytes), 'read')
827
807
        return bytes
828
808
 
852
832
        """See SmartClientMedium.disconnect()."""
853
833
        if not self._connected:
854
834
            return
855
 
        self._socket.close()
 
835
        osutils.until_no_eintr(self._socket.close)
856
836
        self._socket = None
857
837
        self._connected = False
858
838
 
906
886
        """See SmartClientMedium.read_bytes."""
907
887
        if not self._connected:
908
888
            raise errors.MediumNotConnected(self)
909
 
        return osutils.read_bytes_from_socket(
910
 
            self._socket, self._report_activity)
 
889
        return _read_bytes_from_socket(
 
890
            self._socket.recv, count, self._report_activity)
911
891
 
912
892
 
913
893
class SmartClientStreamMediumRequest(SmartClientMediumRequest):
950
930
        self._medium._flush()
951
931
 
952
932
 
 
933
def _read_bytes_from_socket(sock, desired_count, report_activity):
 
934
    # We ignore the desired_count because on sockets it's more efficient to
 
935
    # read large chunks (of _MAX_READ_SIZE bytes) at a time.
 
936
    try:
 
937
        bytes = osutils.until_no_eintr(sock, _MAX_READ_SIZE)
 
938
    except socket.error, e:
 
939
        if len(e.args) and e.args[0] in (errno.ECONNRESET, 10054):
 
940
            # The connection was closed by the other side.  Callers expect an
 
941
            # empty string to signal end-of-stream.
 
942
            bytes = ''
 
943
        else:
 
944
            raise
 
945
    else:
 
946
        report_activity(len(bytes), 'read')
 
947
    return bytes
 
948