/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
12
12
#
13
13
# You should have received a copy of the GNU General Public License
14
14
# along with this program; if not, write to the Free Software
15
 
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
16
16
 
17
17
"""The 'medium' layer for the smart servers and clients.
18
18
 
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
    ui,
43
44
    urlutils,
44
45
    )
45
 
from bzrlib.smart import client, protocol
 
46
from bzrlib.smart import client, protocol, request, vfs
46
47
from bzrlib.transport import ssh
47
48
""")
48
 
 
49
 
 
50
 
# We must not read any more than 64k at a time so we don't risk "no buffer
51
 
# space available" errors on some platforms.  Windows in particular is likely
52
 
# to give error 10053 or 10055 if we read more than 64k from a socket.
53
 
_MAX_READ_SIZE = 64 * 1024
54
 
 
 
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
55
56
 
56
57
def _get_protocol_factory_for_bytes(bytes):
57
58
    """Determine the right protocol factory for 'bytes'.
87
88
 
88
89
def _get_line(read_bytes_func):
89
90
    """Read bytes using read_bytes_func until a newline byte.
90
 
    
 
91
 
91
92
    This isn't particularly efficient, so should only be used when the
92
93
    expected size of the line is quite short.
93
 
    
 
94
 
94
95
    :returns: a tuple of two strs: (line, excess)
95
96
    """
96
97
    newline_pos = -1
112
113
 
113
114
    def __init__(self):
114
115
        self._push_back_buffer = None
115
 
        
 
116
 
116
117
    def _push_back(self, bytes):
117
118
        """Return unused bytes to the medium, because they belong to the next
118
119
        request(s).
152
153
 
153
154
    def _get_line(self):
154
155
        """Read bytes from this request's response until a newline byte.
155
 
        
 
156
 
156
157
        This isn't particularly efficient, so should only be used when the
157
158
        expected size of the line is quite short.
158
159
 
161
162
        line, excess = _get_line(self.read_bytes)
162
163
        self._push_back(excess)
163
164
        return line
164
 
 
 
165
 
 
166
    def _report_activity(self, bytes, direction):
 
167
        """Notify that this medium has activity.
 
168
 
 
169
        Implementations should call this from all methods that actually do IO.
 
170
        Be careful that it's not called twice, if one method is implemented on
 
171
        top of another.
 
172
 
 
173
        :param bytes: Number of bytes read or written.
 
174
        :param direction: 'read' or 'write' or None.
 
175
        """
 
176
        ui.ui_factory.report_transport_activity(self, bytes, direction)
 
177
 
165
178
 
166
179
class SmartServerStreamMedium(SmartMedium):
167
180
    """Handles smart commands coming over a stream.
172
185
    One instance is created for each connected client; it can serve multiple
173
186
    requests in the lifetime of the connection.
174
187
 
175
 
    The server passes requests through to an underlying backing transport, 
 
188
    The server passes requests through to an underlying backing transport,
176
189
    which will typically be a LocalTransport looking at the server's filesystem.
177
190
 
178
191
    :ivar _push_back_buffer: a str of bytes that have been read from the stream
223
236
 
224
237
    def _serve_one_request(self, protocol):
225
238
        """Read one request from input, process, send back a response.
226
 
        
 
239
 
227
240
        :param protocol: a SmartServerRequestProtocol.
228
241
        """
229
242
        try:
261
274
    def _serve_one_request_unguarded(self, protocol):
262
275
        while protocol.next_read_size():
263
276
            # We can safely try to read large chunks.  If there is less data
264
 
            # than _MAX_READ_SIZE ready, the socket wil just return a short
265
 
            # read immediately rather than block.
266
 
            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)
267
280
            if bytes == '':
268
281
                self.finished = True
269
282
                return
270
283
            protocol.accept_bytes(bytes)
271
 
        
 
284
 
272
285
        self._push_back(protocol.unused_data)
273
286
 
274
287
    def _read_bytes(self, desired_count):
275
 
        # We ignore the desired_count because on sockets it's more efficient to
276
 
        # read large chunks (of _MAX_READ_SIZE bytes) at a time.
277
 
        return self.socket.recv(_MAX_READ_SIZE)
 
288
        return osutils.read_bytes_from_socket(
 
289
            self.socket, self._report_activity)
278
290
 
279
291
    def terminate_due_to_error(self):
280
292
        # TODO: This should log to a server log file, but no such thing
283
295
        self.finished = True
284
296
 
285
297
    def _write_out(self, bytes):
286
 
        osutils.send_all(self.socket, bytes)
 
298
        tstart = osutils.timer_func()
 
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))
287
305
 
288
306
 
289
307
class SmartServerPipeStreamMedium(SmartServerStreamMedium):
350
368
    request.finished_reading()
351
369
 
352
370
    It is up to the individual SmartClientMedium whether multiple concurrent
353
 
    requests can exist. See SmartClientMedium.get_request to obtain instances 
354
 
    of SmartClientMediumRequest, and the concrete Medium you are using for 
 
371
    requests can exist. See SmartClientMedium.get_request to obtain instances
 
372
    of SmartClientMediumRequest, and the concrete Medium you are using for
355
373
    details on concurrency and pipelining.
356
374
    """
357
375
 
366
384
    def accept_bytes(self, bytes):
367
385
        """Accept bytes for inclusion in this request.
368
386
 
369
 
        This method may not be be called after finished_writing() has been
 
387
        This method may not be called after finished_writing() has been
370
388
        called.  It depends upon the Medium whether or not the bytes will be
371
389
        immediately transmitted. Message based Mediums will tend to buffer the
372
390
        bytes until finished_writing() is called.
403
421
    def _finished_reading(self):
404
422
        """Helper for finished_reading.
405
423
 
406
 
        finished_reading checks the state of the request to determine if 
 
424
        finished_reading checks the state of the request to determine if
407
425
        finished_reading is allowed, and if it is hands off to _finished_reading
408
426
        to perform the action.
409
427
        """
423
441
    def _finished_writing(self):
424
442
        """Helper for finished_writing.
425
443
 
426
 
        finished_writing checks the state of the request to determine if 
 
444
        finished_writing checks the state of the request to determine if
427
445
        finished_writing is allowed, and if it is hands off to _finished_writing
428
446
        to perform the action.
429
447
        """
449
467
        read_bytes checks the state of the request to determing if bytes
450
468
        should be read. After that it hands off to _read_bytes to do the
451
469
        actual read.
452
 
        
 
470
 
453
471
        By default this forwards to self._medium.read_bytes because we are
454
472
        operating on the medium's stream.
455
473
        """
460
478
        if not line.endswith('\n'):
461
479
            # end of file encountered reading from server
462
480
            raise errors.ConnectionReset(
463
 
                "please check connectivity and permissions",
464
 
                "(and try -Dhpss if further diagnosis is required)")
 
481
                "Unexpected end of message. Please check connectivity "
 
482
                "and permissions, and report a bug if problems persist.")
465
483
        return line
466
484
 
467
485
    def _read_line(self):
468
486
        """Helper for SmartClientMediumRequest.read_line.
469
 
        
 
487
 
470
488
        By default this forwards to self._medium._get_line because we are
471
489
        operating on the medium's stream.
472
490
        """
495
513
        """
496
514
        medium_repr = repr(medium)
497
515
        # Add this medium to the WeakKeyDictionary
498
 
        self.counts[medium] = [0, medium_repr]
 
516
        self.counts[medium] = dict(count=0, vfs_count=0,
 
517
                                   medium_repr=medium_repr)
499
518
        # Weakref callbacks are fired in reverse order of their association
500
519
        # with the referenced object.  So we add a weakref *after* adding to
501
520
        # the WeakKeyDict so that we can report the value from it before the
505
524
    def increment_call_count(self, params):
506
525
        # Increment the count in the WeakKeyDictionary
507
526
        value = self.counts[params.medium]
508
 
        value[0] += 1
 
527
        value['count'] += 1
 
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
 
533
        if issubclass(request_method, vfs.VfsRequest):
 
534
            value['vfs_count'] += 1
509
535
 
510
536
    def done(self, ref):
511
537
        value = self.counts[ref]
512
 
        count, medium_repr = value
 
538
        count, vfs_count, medium_repr = (
 
539
            value['count'], value['vfs_count'], value['medium_repr'])
513
540
        # In case this callback is invoked for the same ref twice (by the
514
541
        # weakref callback and by the atexit function), set the call count back
515
542
        # to 0 so this item won't be reported twice.
516
 
        value[0] = 0
 
543
        value['count'] = 0
 
544
        value['vfs_count'] = 0
517
545
        if count != 0:
518
 
            trace.note('HPSS calls: %d %s', count, medium_repr)
519
 
        
 
546
            trace.note('HPSS calls: %d (%d vfs) %s',
 
547
                       count, vfs_count, medium_repr)
 
548
 
520
549
    def flush_all(self):
521
550
        for ref in list(self.counts.keys()):
522
551
            self.done(ref)
523
552
 
524
553
_debug_counter = None
525
 
  
526
 
  
 
554
 
 
555
 
527
556
class SmartClientMedium(SmartMedium):
528
557
    """Smart client is a medium for sending smart protocol requests over."""
529
558
 
574
603
        """
575
604
        if (self._remote_version_is_before is not None and
576
605
            version_tuple > self._remote_version_is_before):
577
 
            raise AssertionError(
 
606
            # We have been told that the remote side is older than some version
 
607
            # which is newer than a previously supplied older-than version.
 
608
            # This indicates that some smart verb call is not guarded
 
609
            # appropriately (it should simply not have been tried).
 
610
            trace.mutter(
578
611
                "_remember_remote_is_before(%r) called, but "
579
612
                "_remember_remote_is_before(%r) was called previously."
580
 
                % (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
581
620
        self._remote_version_is_before = version_tuple
582
621
 
583
622
    def protocol_version(self):
617
656
 
618
657
    def disconnect(self):
619
658
        """If this medium maintains a persistent connection, close it.
620
 
        
 
659
 
621
660
        The default implementation does nothing.
622
661
        """
623
 
        
 
662
 
624
663
    def remote_path_from_transport(self, transport):
625
664
        """Convert transport into a path suitable for using in a request.
626
 
        
 
665
 
627
666
        Note that the resulting remote path doesn't encode the host name or
628
667
        anything but path, so it is only safe to use it in requests sent over
629
668
        the medium from the matching transport.
657
696
 
658
697
    def _flush(self):
659
698
        """Flush the output stream.
660
 
        
 
699
 
661
700
        This method is used by the SmartClientStreamMediumRequest to ensure that
662
701
        all data for a request is sent, to avoid long timeouts or deadlocks.
663
702
        """
674
713
 
675
714
class SmartSimplePipesClientMedium(SmartClientStreamMedium):
676
715
    """A client medium using simple pipes.
677
 
    
 
716
 
678
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.
679
722
    """
680
723
 
681
724
    def __init__(self, readable_pipe, writeable_pipe, base):
686
729
    def _accept_bytes(self, bytes):
687
730
        """See SmartClientStreamMedium.accept_bytes."""
688
731
        self._writeable_pipe.write(bytes)
 
732
        self._report_activity(len(bytes), 'write')
689
733
 
690
734
    def _flush(self):
691
735
        """See SmartClientStreamMedium._flush()."""
693
737
 
694
738
    def _read_bytes(self, count):
695
739
        """See SmartClientStreamMedium._read_bytes."""
696
 
        return self._readable_pipe.read(count)
 
740
        bytes = osutils.until_no_eintr(self._readable_pipe.read, count)
 
741
        self._report_activity(len(bytes), 'read')
 
742
        return bytes
697
743
 
698
744
 
699
745
class SmartSSHClientMedium(SmartClientStreamMedium):
700
746
    """A client medium using SSH."""
701
 
    
 
747
 
702
748
    def __init__(self, host, port=None, username=None, password=None,
703
749
            base=None, vendor=None, bzr_remote_path=None):
704
750
        """Creates a client that will connect on the first use.
705
 
        
 
751
 
706
752
        :param vendor: An optional override for the ssh vendor to use. See
707
753
            bzrlib.transport.ssh for details on ssh vendors.
708
754
        """
709
 
        SmartClientStreamMedium.__init__(self, base)
710
755
        self._connected = False
711
756
        self._host = host
712
757
        self._password = password
713
758
        self._port = port
714
759
        self._username = username
 
760
        # for the benefit of progress making a short description of this
 
761
        # transport
 
762
        self._scheme = 'bzr+ssh'
 
763
        # SmartClientStreamMedium stores the repr of this object in its
 
764
        # _DebugCounter so we have to store all the values used in our repr
 
765
        # method before calling the super init.
 
766
        SmartClientStreamMedium.__init__(self, base)
715
767
        self._read_from = None
716
768
        self._ssh_connection = None
717
769
        self._vendor = vendor
718
770
        self._write_to = None
719
771
        self._bzr_remote_path = bzr_remote_path
720
 
        if self._bzr_remote_path is None:
721
 
            symbol_versioning.warn(
722
 
                'bzr_remote_path is required as of bzr 0.92',
723
 
                DeprecationWarning, stacklevel=2)
724
 
            self._bzr_remote_path = os.environ.get('BZR_REMOTE_PATH', 'bzr')
 
772
 
 
773
    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/)" % (
 
779
            self.__class__.__name__,
 
780
            self._scheme,
 
781
            self._username,
 
782
            self._host,
 
783
            maybe_port)
725
784
 
726
785
    def _accept_bytes(self, bytes):
727
786
        """See SmartClientStreamMedium.accept_bytes."""
728
787
        self._ensure_connection()
729
788
        self._write_to.write(bytes)
 
789
        self._report_activity(len(bytes), 'write')
730
790
 
731
791
    def disconnect(self):
732
792
        """See SmartClientMedium.disconnect()."""
762
822
        if not self._connected:
763
823
            raise errors.MediumNotConnected(self)
764
824
        bytes_to_read = min(count, _MAX_READ_SIZE)
765
 
        return self._read_from.read(bytes_to_read)
 
825
        bytes = self._read_from.read(bytes_to_read)
 
826
        self._report_activity(len(bytes), 'read')
 
827
        return bytes
766
828
 
767
829
 
768
830
# Port 4155 is the default port for bzr://, registered with IANA.
772
834
 
773
835
class SmartTCPClientMedium(SmartClientStreamMedium):
774
836
    """A client medium using TCP."""
775
 
    
 
837
 
776
838
    def __init__(self, host, port, base):
777
839
        """Creates a client that will connect on the first use."""
778
840
        SmartClientStreamMedium.__init__(self, base)
784
846
    def _accept_bytes(self, bytes):
785
847
        """See SmartClientMedium.accept_bytes."""
786
848
        self._ensure_connection()
787
 
        osutils.send_all(self._socket, bytes)
 
849
        osutils.send_all(self._socket, bytes, self._report_activity)
788
850
 
789
851
    def disconnect(self):
790
852
        """See SmartClientMedium.disconnect()."""
803
865
        else:
804
866
            port = int(self._port)
805
867
        try:
806
 
            sockaddrs = socket.getaddrinfo(self._host, port, socket.AF_UNSPEC, 
 
868
            sockaddrs = socket.getaddrinfo(self._host, port, socket.AF_UNSPEC,
807
869
                socket.SOCK_STREAM, 0, 0)
808
870
        except socket.gaierror, (err_num, err_msg):
809
871
            raise errors.ConnectionError("failed to lookup %s:%d: %s" %
813
875
        for (family, socktype, proto, canonname, sockaddr) in sockaddrs:
814
876
            try:
815
877
                self._socket = socket.socket(family, socktype, proto)
816
 
                self._socket.setsockopt(socket.IPPROTO_TCP, 
 
878
                self._socket.setsockopt(socket.IPPROTO_TCP,
817
879
                                        socket.TCP_NODELAY, 1)
818
880
                self._socket.connect(sockaddr)
819
881
            except socket.error, err:
835
897
 
836
898
    def _flush(self):
837
899
        """See SmartClientStreamMedium._flush().
838
 
        
839
 
        For TCP we do no flushing. We may want to turn off TCP_NODELAY and 
 
900
 
 
901
        For TCP we do no flushing. We may want to turn off TCP_NODELAY and
840
902
        add a means to do a flush, but that can be done in the future.
841
903
        """
842
904
 
844
906
        """See SmartClientMedium.read_bytes."""
845
907
        if not self._connected:
846
908
            raise errors.MediumNotConnected(self)
847
 
        # We ignore the desired_count because on sockets it's more efficient to
848
 
        # read large chunks (of _MAX_READ_SIZE bytes) at a time.
849
 
        try:
850
 
            return self._socket.recv(_MAX_READ_SIZE)
851
 
        except socket.error, e:
852
 
            if len(e.args) and e.args[0] == errno.ECONNRESET:
853
 
                # Callers expect an empty string in that case
854
 
                return ''
855
 
            else:
856
 
                raise
 
909
        return osutils.read_bytes_from_socket(
 
910
            self._socket, self._report_activity)
857
911
 
858
912
 
859
913
class SmartClientStreamMediumRequest(SmartClientMediumRequest):
872
926
 
873
927
    def _accept_bytes(self, bytes):
874
928
        """See SmartClientMediumRequest._accept_bytes.
875
 
        
 
929
 
876
930
        This forwards to self._medium._accept_bytes because we are operating
877
931
        on the mediums stream.
878
932
        """
881
935
    def _finished_reading(self):
882
936
        """See SmartClientMediumRequest._finished_reading.
883
937
 
884
 
        This clears the _current_request on self._medium to allow a new 
 
938
        This clears the _current_request on self._medium to allow a new
885
939
        request to be created.
886
940
        """
887
941
        if self._medium._current_request is not self:
888
942
            raise AssertionError()
889
943
        self._medium._current_request = None
890
 
        
 
944
 
891
945
    def _finished_writing(self):
892
946
        """See SmartClientMediumRequest._finished_writing.
893
947
 
895
949
        """
896
950
        self._medium._flush()
897
951
 
 
952