← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/smart/medium.py

  • Committer: John Arbash Meinel
  • Date: 2008-09-09 15:09:12 UTC
  • mto: This revision was merged to the branch mainline in revision 3699.
  • Revision ID: john@arbash-meinel.com-20080909150912-wyttm8he1zsls2ck
Use the right timing function on win32

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
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., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  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
28
29
import sys
29
30
import urllib
30
31
 
31
32
from bzrlib.lazy_import import lazy_import
32
33
lazy_import(globals(), """
33
 
import atexit
34
 
import socket
35
 
import thread
36
 
import weakref
37
 
 
38
34
from bzrlib import (
39
 
    debug,
40
35
    errors,
 
36
    osutils,
41
37
    symbol_versioning,
42
 
    trace,
43
 
    ui,
44
38
    urlutils,
45
39
    )
46
 
from bzrlib.smart import client, protocol, request, vfs
 
40
from bzrlib.smart import protocol
47
41
from bzrlib.transport import ssh
48
42
""")
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
 
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
 
56
50
 
57
51
def _get_protocol_factory_for_bytes(bytes):
58
52
    """Determine the right protocol factory for 'bytes'.
88
82
 
89
83
def _get_line(read_bytes_func):
90
84
    """Read bytes using read_bytes_func until a newline byte.
91
 
 
 
85
    
92
86
    This isn't particularly efficient, so should only be used when the
93
87
    expected size of the line is quite short.
94
 
 
 
88
    
95
89
    :returns: a tuple of two strs: (line, excess)
96
90
    """
97
91
    newline_pos = -1
113
107
 
114
108
    def __init__(self):
115
109
        self._push_back_buffer = None
116
 
 
 
110
        
117
111
    def _push_back(self, bytes):
118
112
        """Return unused bytes to the medium, because they belong to the next
119
113
        request(s).
153
147
 
154
148
    def _get_line(self):
155
149
        """Read bytes from this request's response until a newline byte.
156
 
 
 
150
        
157
151
        This isn't particularly efficient, so should only be used when the
158
152
        expected size of the line is quite short.
159
153
 
162
156
        line, excess = _get_line(self.read_bytes)
163
157
        self._push_back(excess)
164
158
        return line
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
 
 
 
159
 
178
160
 
179
161
class SmartServerStreamMedium(SmartMedium):
180
162
    """Handles smart commands coming over a stream.
185
167
    One instance is created for each connected client; it can serve multiple
186
168
    requests in the lifetime of the connection.
187
169
 
188
 
    The server passes requests through to an underlying backing transport,
 
170
    The server passes requests through to an underlying backing transport, 
189
171
    which will typically be a LocalTransport looking at the server's filesystem.
190
172
 
191
173
    :ivar _push_back_buffer: a str of bytes that have been read from the stream
236
218
 
237
219
    def _serve_one_request(self, protocol):
238
220
        """Read one request from input, process, send back a response.
239
 
 
 
221
        
240
222
        :param protocol: a SmartServerRequestProtocol.
241
223
        """
242
224
        try:
274
256
    def _serve_one_request_unguarded(self, protocol):
275
257
        while protocol.next_read_size():
276
258
            # 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)
 
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)
280
262
            if bytes == '':
281
263
                self.finished = True
282
264
                return
283
265
            protocol.accept_bytes(bytes)
284
 
 
 
266
        
285
267
        self._push_back(protocol.unused_data)
286
268
 
287
269
    def _read_bytes(self, desired_count):
288
 
        return osutils.read_bytes_from_socket(
289
 
            self.socket, self._report_activity)
 
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)
290
273
 
291
274
    def terminate_due_to_error(self):
292
275
        # TODO: This should log to a server log file, but no such thing
295
278
        self.finished = True
296
279
 
297
280
    def _write_out(self, 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))
 
281
        osutils.send_all(self.socket, bytes)
305
282
 
306
283
 
307
284
class SmartServerPipeStreamMedium(SmartServerStreamMedium):
368
345
    request.finished_reading()
369
346
 
370
347
    It is up to the individual SmartClientMedium whether multiple concurrent
371
 
    requests can exist. See SmartClientMedium.get_request to obtain instances
372
 
    of SmartClientMediumRequest, and the concrete Medium you are using for
 
348
    requests can exist. See SmartClientMedium.get_request to obtain instances 
 
349
    of SmartClientMediumRequest, and the concrete Medium you are using for 
373
350
    details on concurrency and pipelining.
374
351
    """
375
352
 
384
361
    def accept_bytes(self, bytes):
385
362
        """Accept bytes for inclusion in this request.
386
363
 
387
 
        This method may not be called after finished_writing() has been
 
364
        This method may not be be called after finished_writing() has been
388
365
        called.  It depends upon the Medium whether or not the bytes will be
389
366
        immediately transmitted. Message based Mediums will tend to buffer the
390
367
        bytes until finished_writing() is called.
421
398
    def _finished_reading(self):
422
399
        """Helper for finished_reading.
423
400
 
424
 
        finished_reading checks the state of the request to determine if
 
401
        finished_reading checks the state of the request to determine if 
425
402
        finished_reading is allowed, and if it is hands off to _finished_reading
426
403
        to perform the action.
427
404
        """
441
418
    def _finished_writing(self):
442
419
        """Helper for finished_writing.
443
420
 
444
 
        finished_writing checks the state of the request to determine if
 
421
        finished_writing checks the state of the request to determine if 
445
422
        finished_writing is allowed, and if it is hands off to _finished_writing
446
423
        to perform the action.
447
424
        """
467
444
        read_bytes checks the state of the request to determing if bytes
468
445
        should be read. After that it hands off to _read_bytes to do the
469
446
        actual read.
470
 
 
 
447
        
471
448
        By default this forwards to self._medium.read_bytes because we are
472
449
        operating on the medium's stream.
473
450
        """
478
455
        if not line.endswith('\n'):
479
456
            # end of file encountered reading from server
480
457
            raise errors.ConnectionReset(
481
 
                "Unexpected end of message. Please check connectivity "
482
 
                "and permissions, and report a bug if problems persist.")
 
458
                "please check connectivity and permissions",
 
459
                "(and try -Dhpss if further diagnosis is required)")
483
460
        return line
484
461
 
485
462
    def _read_line(self):
486
463
        """Helper for SmartClientMediumRequest.read_line.
487
 
 
 
464
        
488
465
        By default this forwards to self._medium._get_line because we are
489
466
        operating on the medium's stream.
490
467
        """
491
468
        return self._medium._get_line()
492
469
 
493
470
 
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
 
 
556
471
class SmartClientMedium(SmartMedium):
557
472
    """Smart client is a medium for sending smart protocol requests over."""
558
473
 
567
482
        # _remote_version_is_before tracks the bzr version the remote side
568
483
        # can be based on what we've seen so far.
569
484
        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)
576
485
 
577
486
    def _is_remote_before(self, version_tuple):
578
487
        """Is it possible the remote side supports RPCs for a given version?
603
512
        """
604
513
        if (self._remote_version_is_before is not None and
605
514
            version_tuple > self._remote_version_is_before):
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(
 
515
            raise AssertionError(
611
516
                "_remember_remote_is_before(%r) called, but "
612
517
                "_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
 
518
                % (version_tuple, self._remote_version_is_before))
620
519
        self._remote_version_is_before = version_tuple
621
520
 
622
521
    def protocol_version(self):
656
555
 
657
556
    def disconnect(self):
658
557
        """If this medium maintains a persistent connection, close it.
659
 
 
 
558
        
660
559
        The default implementation does nothing.
661
560
        """
662
 
 
 
561
        
663
562
    def remote_path_from_transport(self, transport):
664
563
        """Convert transport into a path suitable for using in a request.
665
 
 
 
564
        
666
565
        Note that the resulting remote path doesn't encode the host name or
667
566
        anything but path, so it is only safe to use it in requests sent over
668
567
        the medium from the matching transport.
696
595
 
697
596
    def _flush(self):
698
597
        """Flush the output stream.
699
 
 
 
598
        
700
599
        This method is used by the SmartClientStreamMediumRequest to ensure that
701
600
        all data for a request is sent, to avoid long timeouts or deadlocks.
702
601
        """
713
612
 
714
613
class SmartSimplePipesClientMedium(SmartClientStreamMedium):
715
614
    """A client medium using simple pipes.
716
 
 
 
615
    
717
616
    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
617
    """
723
618
 
724
619
    def __init__(self, readable_pipe, writeable_pipe, base):
729
624
    def _accept_bytes(self, bytes):
730
625
        """See SmartClientStreamMedium.accept_bytes."""
731
626
        self._writeable_pipe.write(bytes)
732
 
        self._report_activity(len(bytes), 'write')
733
627
 
734
628
    def _flush(self):
735
629
        """See SmartClientStreamMedium._flush()."""
737
631
 
738
632
    def _read_bytes(self, count):
739
633
        """See SmartClientStreamMedium._read_bytes."""
740
 
        bytes = osutils.until_no_eintr(self._readable_pipe.read, count)
741
 
        self._report_activity(len(bytes), 'read')
742
 
        return bytes
 
634
        return self._readable_pipe.read(count)
743
635
 
744
636
 
745
637
class SmartSSHClientMedium(SmartClientStreamMedium):
746
638
    """A client medium using SSH."""
747
 
 
 
639
    
748
640
    def __init__(self, host, port=None, username=None, password=None,
749
641
            base=None, vendor=None, bzr_remote_path=None):
750
642
        """Creates a client that will connect on the first use.
751
 
 
 
643
        
752
644
        :param vendor: An optional override for the ssh vendor to use. See
753
645
            bzrlib.transport.ssh for details on ssh vendors.
754
646
        """
 
647
        SmartClientStreamMedium.__init__(self, base)
755
648
        self._connected = False
756
649
        self._host = host
757
650
        self._password = password
758
651
        self._port = port
759
652
        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)
767
653
        self._read_from = None
768
654
        self._ssh_connection = None
769
655
        self._vendor = vendor
770
656
        self._write_to = None
771
657
        self._bzr_remote_path = bzr_remote_path
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)
 
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')
784
663
 
785
664
    def _accept_bytes(self, bytes):
786
665
        """See SmartClientStreamMedium.accept_bytes."""
787
666
        self._ensure_connection()
788
667
        self._write_to.write(bytes)
789
 
        self._report_activity(len(bytes), 'write')
790
668
 
791
669
    def disconnect(self):
792
670
        """See SmartClientMedium.disconnect()."""
822
700
        if not self._connected:
823
701
            raise errors.MediumNotConnected(self)
824
702
        bytes_to_read = min(count, _MAX_READ_SIZE)
825
 
        bytes = self._read_from.read(bytes_to_read)
826
 
        self._report_activity(len(bytes), 'read')
827
 
        return bytes
 
703
        return self._read_from.read(bytes_to_read)
828
704
 
829
705
 
830
706
# Port 4155 is the default port for bzr://, registered with IANA.
831
 
BZR_DEFAULT_INTERFACE = None
 
707
BZR_DEFAULT_INTERFACE = '0.0.0.0'
832
708
BZR_DEFAULT_PORT = 4155
833
709
 
834
710
 
835
711
class SmartTCPClientMedium(SmartClientStreamMedium):
836
712
    """A client medium using TCP."""
837
 
 
 
713
    
838
714
    def __init__(self, host, port, base):
839
715
        """Creates a client that will connect on the first use."""
840
716
        SmartClientStreamMedium.__init__(self, base)
846
722
    def _accept_bytes(self, bytes):
847
723
        """See SmartClientMedium.accept_bytes."""
848
724
        self._ensure_connection()
849
 
        osutils.send_all(self._socket, bytes, self._report_activity)
 
725
        osutils.send_all(self._socket, bytes)
850
726
 
851
727
    def disconnect(self):
852
728
        """See SmartClientMedium.disconnect()."""
860
736
        """Connect this medium if not already connected."""
861
737
        if self._connected:
862
738
            return
 
739
        self._socket = socket.socket()
 
740
        self._socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
863
741
        if self._port is None:
864
742
            port = BZR_DEFAULT_PORT
865
743
        else:
866
744
            port = int(self._port)
867
745
        try:
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:
 
746
            self._socket.connect((self._host, port))
 
747
        except socket.error, err:
888
748
            # socket errors either have a (string) or (errno, string) as their
889
749
            # args.
890
750
            if type(err.args) is str:
897
757
 
898
758
    def _flush(self):
899
759
        """See SmartClientStreamMedium._flush().
900
 
 
901
 
        For TCP we do no flushing. We may want to turn off TCP_NODELAY and
 
760
        
 
761
        For TCP we do no flushing. We may want to turn off TCP_NODELAY and 
902
762
        add a means to do a flush, but that can be done in the future.
903
763
        """
904
764
 
906
766
        """See SmartClientMedium.read_bytes."""
907
767
        if not self._connected:
908
768
            raise errors.MediumNotConnected(self)
909
 
        return osutils.read_bytes_from_socket(
910
 
            self._socket, self._report_activity)
 
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)
911
772
 
912
773
 
913
774
class SmartClientStreamMediumRequest(SmartClientMediumRequest):
926
787
 
927
788
    def _accept_bytes(self, bytes):
928
789
        """See SmartClientMediumRequest._accept_bytes.
929
 
 
 
790
        
930
791
        This forwards to self._medium._accept_bytes because we are operating
931
792
        on the mediums stream.
932
793
        """
935
796
    def _finished_reading(self):
936
797
        """See SmartClientMediumRequest._finished_reading.
937
798
 
938
 
        This clears the _current_request on self._medium to allow a new
 
799
        This clears the _current_request on self._medium to allow a new 
939
800
        request to be created.
940
801
        """
941
802
        if self._medium._current_request is not self:
942
803
            raise AssertionError()
943
804
        self._medium._current_request = None
944
 
 
 
805
        
945
806
    def _finished_writing(self):
946
807
        """See SmartClientMediumRequest._finished_writing.
947
808
 
949
810
        """
950
811
        self._medium._flush()
951
812
 
952