/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
 
25
25
"""
26
26
 
27
27
import os
28
 
import socket
29
28
import sys
30
29
import urllib
31
30
 
32
31
from bzrlib.lazy_import import lazy_import
33
32
lazy_import(globals(), """
 
33
import atexit
 
34
import socket
 
35
import thread
 
36
import weakref
 
37
 
34
38
from bzrlib import (
 
39
    debug,
35
40
    errors,
36
 
    osutils,
37
41
    symbol_versioning,
 
42
    trace,
 
43
    ui,
38
44
    urlutils,
39
45
    )
40
 
from bzrlib.smart import protocol
 
46
from bzrlib.smart import client, protocol, request, vfs
41
47
from bzrlib.transport import ssh
42
48
""")
43
 
 
44
 
 
45
 
# We must not read any more than 64k at a time so we don't risk "no buffer
46
 
# space available" errors on some platforms.  Windows in particular is likely
47
 
# to give error 10053 or 10055 if we read more than 64k from a socket.
48
 
_MAX_READ_SIZE = 64 * 1024
49
 
 
 
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
50
56
 
51
57
def _get_protocol_factory_for_bytes(bytes):
52
58
    """Determine the right protocol factory for 'bytes'.
82
88
 
83
89
def _get_line(read_bytes_func):
84
90
    """Read bytes using read_bytes_func until a newline byte.
85
 
    
 
91
 
86
92
    This isn't particularly efficient, so should only be used when the
87
93
    expected size of the line is quite short.
88
 
    
 
94
 
89
95
    :returns: a tuple of two strs: (line, excess)
90
96
    """
91
97
    newline_pos = -1
107
113
 
108
114
    def __init__(self):
109
115
        self._push_back_buffer = None
110
 
        
 
116
 
111
117
    def _push_back(self, bytes):
112
118
        """Return unused bytes to the medium, because they belong to the next
113
119
        request(s).
147
153
 
148
154
    def _get_line(self):
149
155
        """Read bytes from this request's response until a newline byte.
150
 
        
 
156
 
151
157
        This isn't particularly efficient, so should only be used when the
152
158
        expected size of the line is quite short.
153
159
 
156
162
        line, excess = _get_line(self.read_bytes)
157
163
        self._push_back(excess)
158
164
        return line
159
 
 
 
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
 
160
178
 
161
179
class SmartServerStreamMedium(SmartMedium):
162
180
    """Handles smart commands coming over a stream.
167
185
    One instance is created for each connected client; it can serve multiple
168
186
    requests in the lifetime of the connection.
169
187
 
170
 
    The server passes requests through to an underlying backing transport, 
 
188
    The server passes requests through to an underlying backing transport,
171
189
    which will typically be a LocalTransport looking at the server's filesystem.
172
190
 
173
191
    :ivar _push_back_buffer: a str of bytes that have been read from the stream
218
236
 
219
237
    def _serve_one_request(self, protocol):
220
238
        """Read one request from input, process, send back a response.
221
 
        
 
239
 
222
240
        :param protocol: a SmartServerRequestProtocol.
223
241
        """
224
242
        try:
256
274
    def _serve_one_request_unguarded(self, protocol):
257
275
        while protocol.next_read_size():
258
276
            # We can safely try to read large chunks.  If there is less data
259
 
            # than _MAX_READ_SIZE ready, the socket wil just return a short
260
 
            # read immediately rather than block.
261
 
            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)
262
280
            if bytes == '':
263
281
                self.finished = True
264
282
                return
265
283
            protocol.accept_bytes(bytes)
266
 
        
 
284
 
267
285
        self._push_back(protocol.unused_data)
268
286
 
269
287
    def _read_bytes(self, desired_count):
270
 
        # We ignore the desired_count because on sockets it's more efficient to
271
 
        # read large chunks (of _MAX_READ_SIZE bytes) at a time.
272
 
        return self.socket.recv(_MAX_READ_SIZE)
 
288
        return osutils.read_bytes_from_socket(
 
289
            self.socket, self._report_activity)
273
290
 
274
291
    def terminate_due_to_error(self):
275
292
        # TODO: This should log to a server log file, but no such thing
278
295
        self.finished = True
279
296
 
280
297
    def _write_out(self, bytes):
281
 
        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))
282
305
 
283
306
 
284
307
class SmartServerPipeStreamMedium(SmartServerStreamMedium):
345
368
    request.finished_reading()
346
369
 
347
370
    It is up to the individual SmartClientMedium whether multiple concurrent
348
 
    requests can exist. See SmartClientMedium.get_request to obtain instances 
349
 
    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
350
373
    details on concurrency and pipelining.
351
374
    """
352
375
 
361
384
    def accept_bytes(self, bytes):
362
385
        """Accept bytes for inclusion in this request.
363
386
 
364
 
        This method may not be be called after finished_writing() has been
 
387
        This method may not be called after finished_writing() has been
365
388
        called.  It depends upon the Medium whether or not the bytes will be
366
389
        immediately transmitted. Message based Mediums will tend to buffer the
367
390
        bytes until finished_writing() is called.
398
421
    def _finished_reading(self):
399
422
        """Helper for finished_reading.
400
423
 
401
 
        finished_reading checks the state of the request to determine if 
 
424
        finished_reading checks the state of the request to determine if
402
425
        finished_reading is allowed, and if it is hands off to _finished_reading
403
426
        to perform the action.
404
427
        """
418
441
    def _finished_writing(self):
419
442
        """Helper for finished_writing.
420
443
 
421
 
        finished_writing checks the state of the request to determine if 
 
444
        finished_writing checks the state of the request to determine if
422
445
        finished_writing is allowed, and if it is hands off to _finished_writing
423
446
        to perform the action.
424
447
        """
444
467
        read_bytes checks the state of the request to determing if bytes
445
468
        should be read. After that it hands off to _read_bytes to do the
446
469
        actual read.
447
 
        
 
470
 
448
471
        By default this forwards to self._medium.read_bytes because we are
449
472
        operating on the medium's stream.
450
473
        """
455
478
        if not line.endswith('\n'):
456
479
            # end of file encountered reading from server
457
480
            raise errors.ConnectionReset(
458
 
                "please check connectivity and permissions",
459
 
                "(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.")
460
483
        return line
461
484
 
462
485
    def _read_line(self):
463
486
        """Helper for SmartClientMediumRequest.read_line.
464
 
        
 
487
 
465
488
        By default this forwards to self._medium._get_line because we are
466
489
        operating on the medium's stream.
467
490
        """
468
491
        return self._medium._get_line()
469
492
 
470
493
 
 
494
class _DebugCounter(object):
 
495
    """An object that counts the HPSS calls made to each client medium.
 
496
 
 
497
    When a medium is garbage-collected, or failing that when atexit functions
 
498
    are run, the total number of calls made on that medium are reported via
 
499
    trace.note.
 
500
    """
 
501
 
 
502
    def __init__(self):
 
503
        self.counts = weakref.WeakKeyDictionary()
 
504
        client._SmartClient.hooks.install_named_hook(
 
505
            'call', self.increment_call_count, 'hpss call counter')
 
506
        atexit.register(self.flush_all)
 
507
 
 
508
    def track(self, medium):
 
509
        """Start tracking calls made to a medium.
 
510
 
 
511
        This only keeps a weakref to the medium, so shouldn't affect the
 
512
        medium's lifetime.
 
513
        """
 
514
        medium_repr = repr(medium)
 
515
        # Add this medium to the WeakKeyDictionary
 
516
        self.counts[medium] = dict(count=0, vfs_count=0,
 
517
                                   medium_repr=medium_repr)
 
518
        # Weakref callbacks are fired in reverse order of their association
 
519
        # with the referenced object.  So we add a weakref *after* adding to
 
520
        # the WeakKeyDict so that we can report the value from it before the
 
521
        # entry is removed by the WeakKeyDict's own callback.
 
522
        ref = weakref.ref(medium, self.done)
 
523
 
 
524
    def increment_call_count(self, params):
 
525
        # Increment the count in the WeakKeyDictionary
 
526
        value = self.counts[params.medium]
 
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
 
535
 
 
536
    def done(self, ref):
 
537
        value = self.counts[ref]
 
538
        count, vfs_count, medium_repr = (
 
539
            value['count'], value['vfs_count'], value['medium_repr'])
 
540
        # In case this callback is invoked for the same ref twice (by the
 
541
        # weakref callback and by the atexit function), set the call count back
 
542
        # to 0 so this item won't be reported twice.
 
543
        value['count'] = 0
 
544
        value['vfs_count'] = 0
 
545
        if count != 0:
 
546
            trace.note('HPSS calls: %d (%d vfs) %s',
 
547
                       count, vfs_count, medium_repr)
 
548
 
 
549
    def flush_all(self):
 
550
        for ref in list(self.counts.keys()):
 
551
            self.done(ref)
 
552
 
 
553
_debug_counter = None
 
554
 
 
555
 
471
556
class SmartClientMedium(SmartMedium):
472
557
    """Smart client is a medium for sending smart protocol requests over."""
473
558
 
482
567
        # _remote_version_is_before tracks the bzr version the remote side
483
568
        # can be based on what we've seen so far.
484
569
        self._remote_version_is_before = None
 
570
        # Install debug hook function if debug flag is set.
 
571
        if 'hpss' in debug.debug_flags:
 
572
            global _debug_counter
 
573
            if _debug_counter is None:
 
574
                _debug_counter = _DebugCounter()
 
575
            _debug_counter.track(self)
485
576
 
486
577
    def _is_remote_before(self, version_tuple):
487
578
        """Is it possible the remote side supports RPCs for a given version?
512
603
        """
513
604
        if (self._remote_version_is_before is not None and
514
605
            version_tuple > self._remote_version_is_before):
515
 
            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(
516
611
                "_remember_remote_is_before(%r) called, but "
517
612
                "_remember_remote_is_before(%r) was called previously."
518
 
                % (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
519
620
        self._remote_version_is_before = version_tuple
520
621
 
521
622
    def protocol_version(self):
555
656
 
556
657
    def disconnect(self):
557
658
        """If this medium maintains a persistent connection, close it.
558
 
        
 
659
 
559
660
        The default implementation does nothing.
560
661
        """
561
 
        
 
662
 
562
663
    def remote_path_from_transport(self, transport):
563
664
        """Convert transport into a path suitable for using in a request.
564
 
        
 
665
 
565
666
        Note that the resulting remote path doesn't encode the host name or
566
667
        anything but path, so it is only safe to use it in requests sent over
567
668
        the medium from the matching transport.
595
696
 
596
697
    def _flush(self):
597
698
        """Flush the output stream.
598
 
        
 
699
 
599
700
        This method is used by the SmartClientStreamMediumRequest to ensure that
600
701
        all data for a request is sent, to avoid long timeouts or deadlocks.
601
702
        """
612
713
 
613
714
class SmartSimplePipesClientMedium(SmartClientStreamMedium):
614
715
    """A client medium using simple pipes.
615
 
    
 
716
 
616
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.
617
722
    """
618
723
 
619
724
    def __init__(self, readable_pipe, writeable_pipe, base):
624
729
    def _accept_bytes(self, bytes):
625
730
        """See SmartClientStreamMedium.accept_bytes."""
626
731
        self._writeable_pipe.write(bytes)
 
732
        self._report_activity(len(bytes), 'write')
627
733
 
628
734
    def _flush(self):
629
735
        """See SmartClientStreamMedium._flush()."""
631
737
 
632
738
    def _read_bytes(self, count):
633
739
        """See SmartClientStreamMedium._read_bytes."""
634
 
        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
635
743
 
636
744
 
637
745
class SmartSSHClientMedium(SmartClientStreamMedium):
638
746
    """A client medium using SSH."""
639
 
    
 
747
 
640
748
    def __init__(self, host, port=None, username=None, password=None,
641
749
            base=None, vendor=None, bzr_remote_path=None):
642
750
        """Creates a client that will connect on the first use.
643
 
        
 
751
 
644
752
        :param vendor: An optional override for the ssh vendor to use. See
645
753
            bzrlib.transport.ssh for details on ssh vendors.
646
754
        """
647
 
        SmartClientStreamMedium.__init__(self, base)
648
755
        self._connected = False
649
756
        self._host = host
650
757
        self._password = password
651
758
        self._port = port
652
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)
653
767
        self._read_from = None
654
768
        self._ssh_connection = None
655
769
        self._vendor = vendor
656
770
        self._write_to = None
657
771
        self._bzr_remote_path = bzr_remote_path
658
 
        if self._bzr_remote_path is None:
659
 
            symbol_versioning.warn(
660
 
                'bzr_remote_path is required as of bzr 0.92',
661
 
                DeprecationWarning, stacklevel=2)
662
 
            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)
663
784
 
664
785
    def _accept_bytes(self, bytes):
665
786
        """See SmartClientStreamMedium.accept_bytes."""
666
787
        self._ensure_connection()
667
788
        self._write_to.write(bytes)
 
789
        self._report_activity(len(bytes), 'write')
668
790
 
669
791
    def disconnect(self):
670
792
        """See SmartClientMedium.disconnect()."""
700
822
        if not self._connected:
701
823
            raise errors.MediumNotConnected(self)
702
824
        bytes_to_read = min(count, _MAX_READ_SIZE)
703
 
        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
704
828
 
705
829
 
706
830
# Port 4155 is the default port for bzr://, registered with IANA.
707
 
BZR_DEFAULT_INTERFACE = '0.0.0.0'
 
831
BZR_DEFAULT_INTERFACE = None
708
832
BZR_DEFAULT_PORT = 4155
709
833
 
710
834
 
711
835
class SmartTCPClientMedium(SmartClientStreamMedium):
712
836
    """A client medium using TCP."""
713
 
    
 
837
 
714
838
    def __init__(self, host, port, base):
715
839
        """Creates a client that will connect on the first use."""
716
840
        SmartClientStreamMedium.__init__(self, base)
722
846
    def _accept_bytes(self, bytes):
723
847
        """See SmartClientMedium.accept_bytes."""
724
848
        self._ensure_connection()
725
 
        osutils.send_all(self._socket, bytes)
 
849
        osutils.send_all(self._socket, bytes, self._report_activity)
726
850
 
727
851
    def disconnect(self):
728
852
        """See SmartClientMedium.disconnect()."""
736
860
        """Connect this medium if not already connected."""
737
861
        if self._connected:
738
862
            return
739
 
        self._socket = socket.socket()
740
 
        self._socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
741
863
        if self._port is None:
742
864
            port = BZR_DEFAULT_PORT
743
865
        else:
744
866
            port = int(self._port)
745
867
        try:
746
 
            self._socket.connect((self._host, port))
747
 
        except socket.error, err:
 
868
            sockaddrs = socket.getaddrinfo(self._host, port, socket.AF_UNSPEC,
 
869
                socket.SOCK_STREAM, 0, 0)
 
870
        except socket.gaierror, (err_num, err_msg):
 
871
            raise errors.ConnectionError("failed to lookup %s:%d: %s" %
 
872
                    (self._host, port, err_msg))
 
873
        # Initialize err in case there are no addresses returned:
 
874
        err = socket.error("no address found for %s" % self._host)
 
875
        for (family, socktype, proto, canonname, sockaddr) in sockaddrs:
 
876
            try:
 
877
                self._socket = socket.socket(family, socktype, proto)
 
878
                self._socket.setsockopt(socket.IPPROTO_TCP,
 
879
                                        socket.TCP_NODELAY, 1)
 
880
                self._socket.connect(sockaddr)
 
881
            except socket.error, err:
 
882
                if self._socket is not None:
 
883
                    self._socket.close()
 
884
                self._socket = None
 
885
                continue
 
886
            break
 
887
        if self._socket is None:
748
888
            # socket errors either have a (string) or (errno, string) as their
749
889
            # args.
750
890
            if type(err.args) is str:
757
897
 
758
898
    def _flush(self):
759
899
        """See SmartClientStreamMedium._flush().
760
 
        
761
 
        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
762
902
        add a means to do a flush, but that can be done in the future.
763
903
        """
764
904
 
766
906
        """See SmartClientMedium.read_bytes."""
767
907
        if not self._connected:
768
908
            raise errors.MediumNotConnected(self)
769
 
        # We ignore the desired_count because on sockets it's more efficient to
770
 
        # read large chunks (of _MAX_READ_SIZE bytes) at a time.
771
 
        return self._socket.recv(_MAX_READ_SIZE)
 
909
        return osutils.read_bytes_from_socket(
 
910
            self._socket, self._report_activity)
772
911
 
773
912
 
774
913
class SmartClientStreamMediumRequest(SmartClientMediumRequest):
787
926
 
788
927
    def _accept_bytes(self, bytes):
789
928
        """See SmartClientMediumRequest._accept_bytes.
790
 
        
 
929
 
791
930
        This forwards to self._medium._accept_bytes because we are operating
792
931
        on the mediums stream.
793
932
        """
796
935
    def _finished_reading(self):
797
936
        """See SmartClientMediumRequest._finished_reading.
798
937
 
799
 
        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
800
939
        request to be created.
801
940
        """
802
941
        if self._medium._current_request is not self:
803
942
            raise AssertionError()
804
943
        self._medium._current_request = None
805
 
        
 
944
 
806
945
    def _finished_writing(self):
807
946
        """See SmartClientMediumRequest._finished_writing.
808
947
 
810
949
        """
811
950
        self._medium._flush()
812
951
 
 
952