/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
 
35
import thread
36
36
import weakref
 
37
 
37
38
from bzrlib import (
38
39
    debug,
39
40
    errors,
45
46
from bzrlib.smart import client, protocol, request, vfs
46
47
from bzrlib.transport import ssh
47
48
""")
48
 
#usually already imported, and getting IllegalScoperReplacer on it here.
49
49
from bzrlib import osutils
50
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
 
 
 
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
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_READ_SIZE ready, the socket wil just return a short
278
 
            # read immediately rather than block.
279
 
            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)
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 _read_bytes_from_socket(
289
 
            self.socket.recv, desired_count, self._report_activity)
 
288
        return osutils.read_bytes_from_socket(
 
289
            self.socket, 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
 
        osutils.until_no_eintr(self.socket.close)
 
294
        self.socket.close()
295
295
        self.finished = True
296
296
 
297
297
    def _write_out(self, bytes):
 
298
        tstart = osutils.timer_func()
298
299
        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))
299
305
 
300
306
 
301
307
class SmartServerPipeStreamMedium(SmartServerStreamMedium):
326
332
            bytes_to_read = protocol.next_read_size()
327
333
            if bytes_to_read == 0:
328
334
                # Finished serving this request.
329
 
                osutils.until_no_eintr(self._out.flush)
 
335
                self._out.flush()
330
336
                return
331
337
            bytes = self.read_bytes(bytes_to_read)
332
338
            if bytes == '':
333
339
                # Connection has been closed.
334
340
                self.finished = True
335
 
                osutils.until_no_eintr(self._out.flush)
 
341
                self._out.flush()
336
342
                return
337
343
            protocol.accept_bytes(bytes)
338
344
 
339
345
    def _read_bytes(self, desired_count):
340
 
        return osutils.until_no_eintr(self._in.read, desired_count)
 
346
        return self._in.read(desired_count)
341
347
 
342
348
    def terminate_due_to_error(self):
343
349
        # TODO: This should log to a server log file, but no such thing
344
350
        # exists yet.  Andrew Bennetts 2006-09-29.
345
 
        osutils.until_no_eintr(self._out.close)
 
351
        self._out.close()
346
352
        self.finished = True
347
353
 
348
354
    def _write_out(self, bytes):
349
 
        osutils.until_no_eintr(self._out.write, bytes)
 
355
        self._out.write(bytes)
350
356
 
351
357
 
352
358
class SmartClientMediumRequest(object):
601
607
            # which is newer than a previously supplied older-than version.
602
608
            # This indicates that some smart verb call is not guarded
603
609
            # appropriately (it should simply not have been tried).
604
 
            raise AssertionError(
 
610
            trace.mutter(
605
611
                "_remember_remote_is_before(%r) called, but "
606
612
                "_remember_remote_is_before(%r) was called previously."
607
 
                % (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
608
620
        self._remote_version_is_before = version_tuple
609
621
 
610
622
    def protocol_version(self):
703
715
    """A client medium using simple pipes.
704
716
 
705
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.
706
722
    """
707
723
 
708
724
    def __init__(self, readable_pipe, writeable_pipe, base):
712
728
 
713
729
    def _accept_bytes(self, bytes):
714
730
        """See SmartClientStreamMedium.accept_bytes."""
715
 
        osutils.until_no_eintr(self._writeable_pipe.write, bytes)
 
731
        self._writeable_pipe.write(bytes)
716
732
        self._report_activity(len(bytes), 'write')
717
733
 
718
734
    def _flush(self):
719
735
        """See SmartClientStreamMedium._flush()."""
720
 
        osutils.until_no_eintr(self._writeable_pipe.flush)
 
736
        self._writeable_pipe.flush()
721
737
 
722
738
    def _read_bytes(self, count):
723
739
        """See SmartClientStreamMedium._read_bytes."""
741
757
        self._password = password
742
758
        self._port = port
743
759
        self._username = username
 
760
        # for the benefit of progress making a short description of this
 
761
        # transport
 
762
        self._scheme = 'bzr+ssh'
744
763
        # SmartClientStreamMedium stores the repr of this object in its
745
764
        # _DebugCounter so we have to store all the values used in our repr
746
765
        # method before calling the super init.
750
769
        self._vendor = vendor
751
770
        self._write_to = None
752
771
        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'
756
772
 
757
773
    def __repr__(self):
758
 
        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/)" % (
759
779
            self.__class__.__name__,
760
 
            self._connected,
 
780
            self._scheme,
761
781
            self._username,
762
782
            self._host,
763
 
            self._port)
 
783
            maybe_port)
764
784
 
765
785
    def _accept_bytes(self, bytes):
766
786
        """See SmartClientStreamMedium.accept_bytes."""
767
787
        self._ensure_connection()
768
 
        osutils.until_no_eintr(self._write_to.write, bytes)
 
788
        self._write_to.write(bytes)
769
789
        self._report_activity(len(bytes), 'write')
770
790
 
771
791
    def disconnect(self):
772
792
        """See SmartClientMedium.disconnect()."""
773
793
        if not self._connected:
774
794
            return
775
 
        osutils.until_no_eintr(self._read_from.close)
776
 
        osutils.until_no_eintr(self._write_to.close)
 
795
        self._read_from.close()
 
796
        self._write_to.close()
777
797
        self._ssh_connection.close()
778
798
        self._connected = False
779
799
 
802
822
        if not self._connected:
803
823
            raise errors.MediumNotConnected(self)
804
824
        bytes_to_read = min(count, _MAX_READ_SIZE)
805
 
        bytes = osutils.until_no_eintr(self._read_from.read, bytes_to_read)
 
825
        bytes = self._read_from.read(bytes_to_read)
806
826
        self._report_activity(len(bytes), 'read')
807
827
        return bytes
808
828
 
832
852
        """See SmartClientMedium.disconnect()."""
833
853
        if not self._connected:
834
854
            return
835
 
        osutils.until_no_eintr(self._socket.close)
 
855
        self._socket.close()
836
856
        self._socket = None
837
857
        self._connected = False
838
858
 
886
906
        """See SmartClientMedium.read_bytes."""
887
907
        if not self._connected:
888
908
            raise errors.MediumNotConnected(self)
889
 
        return _read_bytes_from_socket(
890
 
            self._socket.recv, count, self._report_activity)
 
909
        return osutils.read_bytes_from_socket(
 
910
            self._socket, self._report_activity)
891
911
 
892
912
 
893
913
class SmartClientStreamMediumRequest(SmartClientMediumRequest):
930
950
        self._medium._flush()
931
951
 
932
952
 
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