/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: Martin von Gagern
  • Date: 2010-04-20 08:47:38 UTC
  • mfrom: (5167 +trunk)
  • mto: This revision was merged to the branch mainline in revision 5195.
  • Revision ID: martin.vgagern@gmx.net-20100420084738-ygymnqmdllzrhpfn
merge trunk

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):
 
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).
515
610
            raise AssertionError(
516
611
                "_remember_remote_is_before(%r) called, but "
517
612
                "_remember_remote_is_before(%r) was called previously."
555
650
 
556
651
    def disconnect(self):
557
652
        """If this medium maintains a persistent connection, close it.
558
 
        
 
653
 
559
654
        The default implementation does nothing.
560
655
        """
561
 
        
 
656
 
562
657
    def remote_path_from_transport(self, transport):
563
658
        """Convert transport into a path suitable for using in a request.
564
 
        
 
659
 
565
660
        Note that the resulting remote path doesn't encode the host name or
566
661
        anything but path, so it is only safe to use it in requests sent over
567
662
        the medium from the matching transport.
595
690
 
596
691
    def _flush(self):
597
692
        """Flush the output stream.
598
 
        
 
693
 
599
694
        This method is used by the SmartClientStreamMediumRequest to ensure that
600
695
        all data for a request is sent, to avoid long timeouts or deadlocks.
601
696
        """
612
707
 
613
708
class SmartSimplePipesClientMedium(SmartClientStreamMedium):
614
709
    """A client medium using simple pipes.
615
 
    
 
710
 
616
711
    This client does not manage the pipes: it assumes they will always be open.
 
712
 
 
713
    Note that if readable_pipe.read might raise IOError or OSError with errno
 
714
    of EINTR, it must be safe to retry the read.  Plain CPython fileobjects
 
715
    (such as used for sys.stdin) are safe.
617
716
    """
618
717
 
619
718
    def __init__(self, readable_pipe, writeable_pipe, base):
624
723
    def _accept_bytes(self, bytes):
625
724
        """See SmartClientStreamMedium.accept_bytes."""
626
725
        self._writeable_pipe.write(bytes)
 
726
        self._report_activity(len(bytes), 'write')
627
727
 
628
728
    def _flush(self):
629
729
        """See SmartClientStreamMedium._flush()."""
631
731
 
632
732
    def _read_bytes(self, count):
633
733
        """See SmartClientStreamMedium._read_bytes."""
634
 
        return self._readable_pipe.read(count)
 
734
        bytes = osutils.until_no_eintr(self._readable_pipe.read, count)
 
735
        self._report_activity(len(bytes), 'read')
 
736
        return bytes
635
737
 
636
738
 
637
739
class SmartSSHClientMedium(SmartClientStreamMedium):
638
740
    """A client medium using SSH."""
639
 
    
 
741
 
640
742
    def __init__(self, host, port=None, username=None, password=None,
641
743
            base=None, vendor=None, bzr_remote_path=None):
642
744
        """Creates a client that will connect on the first use.
643
 
        
 
745
 
644
746
        :param vendor: An optional override for the ssh vendor to use. See
645
747
            bzrlib.transport.ssh for details on ssh vendors.
646
748
        """
647
 
        SmartClientStreamMedium.__init__(self, base)
648
749
        self._connected = False
649
750
        self._host = host
650
751
        self._password = password
651
752
        self._port = port
652
753
        self._username = username
 
754
        # for the benefit of progress making a short description of this
 
755
        # transport
 
756
        self._scheme = 'bzr+ssh'
 
757
        # SmartClientStreamMedium stores the repr of this object in its
 
758
        # _DebugCounter so we have to store all the values used in our repr
 
759
        # method before calling the super init.
 
760
        SmartClientStreamMedium.__init__(self, base)
653
761
        self._read_from = None
654
762
        self._ssh_connection = None
655
763
        self._vendor = vendor
656
764
        self._write_to = None
657
765
        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')
 
766
 
 
767
    def __repr__(self):
 
768
        if self._port is None:
 
769
            maybe_port = ''
 
770
        else:
 
771
            maybe_port = ':%s' % self._port
 
772
        return "%s(%s://%s@%s%s/)" % (
 
773
            self.__class__.__name__,
 
774
            self._scheme,
 
775
            self._username,
 
776
            self._host,
 
777
            maybe_port)
663
778
 
664
779
    def _accept_bytes(self, bytes):
665
780
        """See SmartClientStreamMedium.accept_bytes."""
666
781
        self._ensure_connection()
667
782
        self._write_to.write(bytes)
 
783
        self._report_activity(len(bytes), 'write')
668
784
 
669
785
    def disconnect(self):
670
786
        """See SmartClientMedium.disconnect()."""
700
816
        if not self._connected:
701
817
            raise errors.MediumNotConnected(self)
702
818
        bytes_to_read = min(count, _MAX_READ_SIZE)
703
 
        return self._read_from.read(bytes_to_read)
 
819
        bytes = self._read_from.read(bytes_to_read)
 
820
        self._report_activity(len(bytes), 'read')
 
821
        return bytes
704
822
 
705
823
 
706
824
# Port 4155 is the default port for bzr://, registered with IANA.
710
828
 
711
829
class SmartTCPClientMedium(SmartClientStreamMedium):
712
830
    """A client medium using TCP."""
713
 
    
 
831
 
714
832
    def __init__(self, host, port, base):
715
833
        """Creates a client that will connect on the first use."""
716
834
        SmartClientStreamMedium.__init__(self, base)
722
840
    def _accept_bytes(self, bytes):
723
841
        """See SmartClientMedium.accept_bytes."""
724
842
        self._ensure_connection()
725
 
        osutils.send_all(self._socket, bytes)
 
843
        osutils.send_all(self._socket, bytes, self._report_activity)
726
844
 
727
845
    def disconnect(self):
728
846
        """See SmartClientMedium.disconnect()."""
741
859
        else:
742
860
            port = int(self._port)
743
861
        try:
744
 
            sockaddrs = socket.getaddrinfo(self._host, port, socket.AF_UNSPEC, 
 
862
            sockaddrs = socket.getaddrinfo(self._host, port, socket.AF_UNSPEC,
745
863
                socket.SOCK_STREAM, 0, 0)
746
864
        except socket.gaierror, (err_num, err_msg):
747
865
            raise errors.ConnectionError("failed to lookup %s:%d: %s" %
751
869
        for (family, socktype, proto, canonname, sockaddr) in sockaddrs:
752
870
            try:
753
871
                self._socket = socket.socket(family, socktype, proto)
754
 
                self._socket.setsockopt(socket.IPPROTO_TCP, 
 
872
                self._socket.setsockopt(socket.IPPROTO_TCP,
755
873
                                        socket.TCP_NODELAY, 1)
756
874
                self._socket.connect(sockaddr)
757
875
            except socket.error, err:
773
891
 
774
892
    def _flush(self):
775
893
        """See SmartClientStreamMedium._flush().
776
 
        
777
 
        For TCP we do no flushing. We may want to turn off TCP_NODELAY and 
 
894
 
 
895
        For TCP we do no flushing. We may want to turn off TCP_NODELAY and
778
896
        add a means to do a flush, but that can be done in the future.
779
897
        """
780
898
 
782
900
        """See SmartClientMedium.read_bytes."""
783
901
        if not self._connected:
784
902
            raise errors.MediumNotConnected(self)
785
 
        # We ignore the desired_count because on sockets it's more efficient to
786
 
        # read large chunks (of _MAX_READ_SIZE bytes) at a time.
787
 
        return self._socket.recv(_MAX_READ_SIZE)
 
903
        return osutils.read_bytes_from_socket(
 
904
            self._socket, self._report_activity)
788
905
 
789
906
 
790
907
class SmartClientStreamMediumRequest(SmartClientMediumRequest):
803
920
 
804
921
    def _accept_bytes(self, bytes):
805
922
        """See SmartClientMediumRequest._accept_bytes.
806
 
        
 
923
 
807
924
        This forwards to self._medium._accept_bytes because we are operating
808
925
        on the mediums stream.
809
926
        """
812
929
    def _finished_reading(self):
813
930
        """See SmartClientMediumRequest._finished_reading.
814
931
 
815
 
        This clears the _current_request on self._medium to allow a new 
 
932
        This clears the _current_request on self._medium to allow a new
816
933
        request to be created.
817
934
        """
818
935
        if self._medium._current_request is not self:
819
936
            raise AssertionError()
820
937
        self._medium._current_request = None
821
 
        
 
938
 
822
939
    def _finished_writing(self):
823
940
        """See SmartClientMediumRequest._finished_writing.
824
941
 
826
943
        """
827
944
        self._medium._flush()
828
945
 
 
946