/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,
40
 
    osutils,
41
41
    symbol_versioning,
42
42
    trace,
43
43
    ui,
46
46
from bzrlib.smart import client, protocol, request, vfs
47
47
from bzrlib.transport import ssh
48
48
""")
49
 
 
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
 
 
 
49
from bzrlib import osutils
 
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
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
 
        # We ignore the desired_count because on sockets it's more efficient to
289
 
        # read large chunks (of _MAX_READ_SIZE bytes) at a time.
290
 
        bytes = osutils.until_no_eintr(self.socket.recv, _MAX_READ_SIZE)
291
 
        self._report_activity(len(bytes), 'read')
292
 
        return bytes
 
288
        return osutils.read_bytes_from_socket(
 
289
            self.socket, self._report_activity)
293
290
 
294
291
    def terminate_due_to_error(self):
295
292
        # TODO: This should log to a server log file, but no such thing
298
295
        self.finished = True
299
296
 
300
297
    def _write_out(self, bytes):
 
298
        tstart = osutils.timer_func()
301
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))
302
305
 
303
306
 
304
307
class SmartServerPipeStreamMedium(SmartServerStreamMedium):
475
478
        if not line.endswith('\n'):
476
479
            # end of file encountered reading from server
477
480
            raise errors.ConnectionReset(
478
 
                "please check connectivity and permissions")
 
481
                "Unexpected end of message. Please check connectivity "
 
482
                "and permissions, and report a bug if problems persist.")
479
483
        return line
480
484
 
481
485
    def _read_line(self):
521
525
        # Increment the count in the WeakKeyDictionary
522
526
        value = self.counts[params.medium]
523
527
        value['count'] += 1
524
 
        request_method = request.request_handlers.get(params.method)
 
528
        try:
 
529
            request_method = request.request_handlers.get(params.method)
 
530
        except KeyError:
 
531
            # A method we don't know about doesn't count as a VFS method.
 
532
            return
525
533
        if issubclass(request_method, vfs.VfsRequest):
526
534
            value['vfs_count'] += 1
527
535
 
599
607
            # which is newer than a previously supplied older-than version.
600
608
            # This indicates that some smart verb call is not guarded
601
609
            # appropriately (it should simply not have been tried).
602
 
            raise AssertionError(
 
610
            trace.mutter(
603
611
                "_remember_remote_is_before(%r) called, but "
604
612
                "_remember_remote_is_before(%r) was called previously."
605
 
                % (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
606
620
        self._remote_version_is_before = version_tuple
607
621
 
608
622
    def protocol_version(self):
701
715
    """A client medium using simple pipes.
702
716
 
703
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.
704
722
    """
705
723
 
706
724
    def __init__(self, readable_pipe, writeable_pipe, base):
719
737
 
720
738
    def _read_bytes(self, count):
721
739
        """See SmartClientStreamMedium._read_bytes."""
722
 
        bytes = self._readable_pipe.read(count)
 
740
        bytes = osutils.until_no_eintr(self._readable_pipe.read, count)
723
741
        self._report_activity(len(bytes), 'read')
724
742
        return bytes
725
743
 
739
757
        self._password = password
740
758
        self._port = port
741
759
        self._username = username
 
760
        # for the benefit of progress making a short description of this
 
761
        # transport
 
762
        self._scheme = 'bzr+ssh'
742
763
        # SmartClientStreamMedium stores the repr of this object in its
743
764
        # _DebugCounter so we have to store all the values used in our repr
744
765
        # method before calling the super init.
748
769
        self._vendor = vendor
749
770
        self._write_to = None
750
771
        self._bzr_remote_path = bzr_remote_path
751
 
        # for the benefit of progress making a short description of this
752
 
        # transport
753
 
        self._scheme = 'bzr+ssh'
754
772
 
755
773
    def __repr__(self):
756
 
        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/)" % (
757
779
            self.__class__.__name__,
758
 
            self._connected,
 
780
            self._scheme,
759
781
            self._username,
760
782
            self._host,
761
 
            self._port)
 
783
            maybe_port)
762
784
 
763
785
    def _accept_bytes(self, bytes):
764
786
        """See SmartClientStreamMedium.accept_bytes."""
884
906
        """See SmartClientMedium.read_bytes."""
885
907
        if not self._connected:
886
908
            raise errors.MediumNotConnected(self)
887
 
        # We ignore the desired_count because on sockets it's more efficient to
888
 
        # read large chunks (of _MAX_READ_SIZE bytes) at a time.
889
 
        try:
890
 
            bytes = osutils.until_no_eintr(self._socket.recv, _MAX_READ_SIZE)
891
 
        except socket.error, e:
892
 
            if len(e.args) and e.args[0] == errno.ECONNRESET:
893
 
                # Callers expect an empty string in that case
894
 
                return ''
895
 
            else:
896
 
                raise
897
 
        else:
898
 
            self._report_activity(len(bytes), 'read')
899
 
            return bytes
 
909
        return osutils.read_bytes_from_socket(
 
910
            self._socket, self._report_activity)
900
911
 
901
912
 
902
913
class SmartClientStreamMediumRequest(SmartClientMediumRequest):
938
949
        """
939
950
        self._medium._flush()
940
951
 
 
952