← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/smart/medium.py

  • Committer: Vincent Ladeuil
  • Date: 2011-08-12 09:49:24 UTC
  • mfrom: (6015.9.10 2.4)
  • mto: This revision was merged to the branch mainline in revision 6066.
  • Revision ID: v.ladeuil+lp@free.fr-20110812094924-knc5s0g7vs31a2f1
Merge 2.4 into 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-2011 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
 
 
31
import bzrlib
32
32
from bzrlib.lazy_import import lazy_import
33
33
lazy_import(globals(), """
 
34
import socket
 
35
import thread
 
36
import weakref
 
37
 
34
38
from bzrlib import (
 
39
    debug,
35
40
    errors,
36
 
    osutils,
37
 
    symbol_versioning,
 
41
    trace,
 
42
    ui,
38
43
    urlutils,
39
44
    )
40
 
from bzrlib.smart import protocol
 
45
from bzrlib.smart import client, protocol, request, vfs
41
46
from bzrlib.transport import ssh
42
47
""")
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
 
 
 
48
from bzrlib import osutils
 
49
 
 
50
# Throughout this module buffer size parameters are either limited to be at
 
51
# most _MAX_READ_SIZE, or are ignored and _MAX_READ_SIZE is used instead.
 
52
# For this module's purposes, MAX_SOCKET_CHUNK is a reasonable size for reads
 
53
# from non-sockets as well.
 
54
_MAX_READ_SIZE = osutils.MAX_SOCKET_CHUNK
50
55
 
51
56
def _get_protocol_factory_for_bytes(bytes):
52
57
    """Determine the right protocol factory for 'bytes'.
82
87
 
83
88
def _get_line(read_bytes_func):
84
89
    """Read bytes using read_bytes_func until a newline byte.
85
 
    
 
90
 
86
91
    This isn't particularly efficient, so should only be used when the
87
92
    expected size of the line is quite short.
88
 
    
 
93
 
89
94
    :returns: a tuple of two strs: (line, excess)
90
95
    """
91
96
    newline_pos = -1
107
112
 
108
113
    def __init__(self):
109
114
        self._push_back_buffer = None
110
 
        
 
115
 
111
116
    def _push_back(self, bytes):
112
117
        """Return unused bytes to the medium, because they belong to the next
113
118
        request(s).
147
152
 
148
153
    def _get_line(self):
149
154
        """Read bytes from this request's response until a newline byte.
150
 
        
 
155
 
151
156
        This isn't particularly efficient, so should only be used when the
152
157
        expected size of the line is quite short.
153
158
 
156
161
        line, excess = _get_line(self.read_bytes)
157
162
        self._push_back(excess)
158
163
        return line
159
 
 
 
164
 
 
165
    def _report_activity(self, bytes, direction):
 
166
        """Notify that this medium has activity.
 
167
 
 
168
        Implementations should call this from all methods that actually do IO.
 
169
        Be careful that it's not called twice, if one method is implemented on
 
170
        top of another.
 
171
 
 
172
        :param bytes: Number of bytes read or written.
 
173
        :param direction: 'read' or 'write' or None.
 
174
        """
 
175
        ui.ui_factory.report_transport_activity(self, bytes, direction)
 
176
 
160
177
 
161
178
class SmartServerStreamMedium(SmartMedium):
162
179
    """Handles smart commands coming over a stream.
167
184
    One instance is created for each connected client; it can serve multiple
168
185
    requests in the lifetime of the connection.
169
186
 
170
 
    The server passes requests through to an underlying backing transport, 
 
187
    The server passes requests through to an underlying backing transport,
171
188
    which will typically be a LocalTransport looking at the server's filesystem.
172
189
 
173
190
    :ivar _push_back_buffer: a str of bytes that have been read from the stream
218
235
 
219
236
    def _serve_one_request(self, protocol):
220
237
        """Read one request from input, process, send back a response.
221
 
        
 
238
 
222
239
        :param protocol: a SmartServerRequestProtocol.
223
240
        """
224
241
        try:
256
273
    def _serve_one_request_unguarded(self, protocol):
257
274
        while protocol.next_read_size():
258
275
            # 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)
 
276
            # than MAX_SOCKET_CHUNK ready, the socket will just return a
 
277
            # short read immediately rather than block.
 
278
            bytes = self.read_bytes(osutils.MAX_SOCKET_CHUNK)
262
279
            if bytes == '':
263
280
                self.finished = True
264
281
                return
265
282
            protocol.accept_bytes(bytes)
266
 
        
 
283
 
267
284
        self._push_back(protocol.unused_data)
268
285
 
269
286
    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)
 
287
        return osutils.read_bytes_from_socket(
 
288
            self.socket, self._report_activity)
273
289
 
274
290
    def terminate_due_to_error(self):
275
291
        # TODO: This should log to a server log file, but no such thing
278
294
        self.finished = True
279
295
 
280
296
    def _write_out(self, bytes):
281
 
        osutils.send_all(self.socket, bytes)
 
297
        tstart = osutils.timer_func()
 
298
        osutils.send_all(self.socket, bytes, self._report_activity)
 
299
        if 'hpss' in debug.debug_flags:
 
300
            thread_id = thread.get_ident()
 
301
            trace.mutter('%12s: [%s] %d bytes to the socket in %.3fs'
 
302
                         % ('wrote', thread_id, len(bytes),
 
303
                            osutils.timer_func() - tstart))
282
304
 
283
305
 
284
306
class SmartServerPipeStreamMedium(SmartServerStreamMedium):
345
367
    request.finished_reading()
346
368
 
347
369
    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 
 
370
    requests can exist. See SmartClientMedium.get_request to obtain instances
 
371
    of SmartClientMediumRequest, and the concrete Medium you are using for
350
372
    details on concurrency and pipelining.
351
373
    """
352
374
 
361
383
    def accept_bytes(self, bytes):
362
384
        """Accept bytes for inclusion in this request.
363
385
 
364
 
        This method may not be be called after finished_writing() has been
 
386
        This method may not be called after finished_writing() has been
365
387
        called.  It depends upon the Medium whether or not the bytes will be
366
388
        immediately transmitted. Message based Mediums will tend to buffer the
367
389
        bytes until finished_writing() is called.
398
420
    def _finished_reading(self):
399
421
        """Helper for finished_reading.
400
422
 
401
 
        finished_reading checks the state of the request to determine if 
 
423
        finished_reading checks the state of the request to determine if
402
424
        finished_reading is allowed, and if it is hands off to _finished_reading
403
425
        to perform the action.
404
426
        """
418
440
    def _finished_writing(self):
419
441
        """Helper for finished_writing.
420
442
 
421
 
        finished_writing checks the state of the request to determine if 
 
443
        finished_writing checks the state of the request to determine if
422
444
        finished_writing is allowed, and if it is hands off to _finished_writing
423
445
        to perform the action.
424
446
        """
444
466
        read_bytes checks the state of the request to determing if bytes
445
467
        should be read. After that it hands off to _read_bytes to do the
446
468
        actual read.
447
 
        
 
469
 
448
470
        By default this forwards to self._medium.read_bytes because we are
449
471
        operating on the medium's stream.
450
472
        """
455
477
        if not line.endswith('\n'):
456
478
            # end of file encountered reading from server
457
479
            raise errors.ConnectionReset(
458
 
                "please check connectivity and permissions",
459
 
                "(and try -Dhpss if further diagnosis is required)")
 
480
                "Unexpected end of message. Please check connectivity "
 
481
                "and permissions, and report a bug if problems persist.")
460
482
        return line
461
483
 
462
484
    def _read_line(self):
463
485
        """Helper for SmartClientMediumRequest.read_line.
464
 
        
 
486
 
465
487
        By default this forwards to self._medium._get_line because we are
466
488
        operating on the medium's stream.
467
489
        """
468
490
        return self._medium._get_line()
469
491
 
470
492
 
 
493
class _DebugCounter(object):
 
494
    """An object that counts the HPSS calls made to each client medium.
 
495
 
 
496
    When a medium is garbage-collected, or failing that when
 
497
    bzrlib.global_state exits, the total number of calls made on that medium
 
498
    are reported via trace.note.
 
499
    """
 
500
 
 
501
    def __init__(self):
 
502
        self.counts = weakref.WeakKeyDictionary()
 
503
        client._SmartClient.hooks.install_named_hook(
 
504
            'call', self.increment_call_count, 'hpss call counter')
 
505
        bzrlib.global_state.cleanups.add_cleanup(self.flush_all)
 
506
 
 
507
    def track(self, medium):
 
508
        """Start tracking calls made to a medium.
 
509
 
 
510
        This only keeps a weakref to the medium, so shouldn't affect the
 
511
        medium's lifetime.
 
512
        """
 
513
        medium_repr = repr(medium)
 
514
        # Add this medium to the WeakKeyDictionary
 
515
        self.counts[medium] = dict(count=0, vfs_count=0,
 
516
                                   medium_repr=medium_repr)
 
517
        # Weakref callbacks are fired in reverse order of their association
 
518
        # with the referenced object.  So we add a weakref *after* adding to
 
519
        # the WeakKeyDict so that we can report the value from it before the
 
520
        # entry is removed by the WeakKeyDict's own callback.
 
521
        ref = weakref.ref(medium, self.done)
 
522
 
 
523
    def increment_call_count(self, params):
 
524
        # Increment the count in the WeakKeyDictionary
 
525
        value = self.counts[params.medium]
 
526
        value['count'] += 1
 
527
        try:
 
528
            request_method = request.request_handlers.get(params.method)
 
529
        except KeyError:
 
530
            # A method we don't know about doesn't count as a VFS method.
 
531
            return
 
532
        if issubclass(request_method, vfs.VfsRequest):
 
533
            value['vfs_count'] += 1
 
534
 
 
535
    def done(self, ref):
 
536
        value = self.counts[ref]
 
537
        count, vfs_count, medium_repr = (
 
538
            value['count'], value['vfs_count'], value['medium_repr'])
 
539
        # In case this callback is invoked for the same ref twice (by the
 
540
        # weakref callback and by the atexit function), set the call count back
 
541
        # to 0 so this item won't be reported twice.
 
542
        value['count'] = 0
 
543
        value['vfs_count'] = 0
 
544
        if count != 0:
 
545
            trace.note('HPSS calls: %d (%d vfs) %s',
 
546
                       count, vfs_count, medium_repr)
 
547
 
 
548
    def flush_all(self):
 
549
        for ref in list(self.counts.keys()):
 
550
            self.done(ref)
 
551
 
 
552
_debug_counter = None
 
553
 
 
554
 
471
555
class SmartClientMedium(SmartMedium):
472
556
    """Smart client is a medium for sending smart protocol requests over."""
473
557
 
482
566
        # _remote_version_is_before tracks the bzr version the remote side
483
567
        # can be based on what we've seen so far.
484
568
        self._remote_version_is_before = None
 
569
        # Install debug hook function if debug flag is set.
 
570
        if 'hpss' in debug.debug_flags:
 
571
            global _debug_counter
 
572
            if _debug_counter is None:
 
573
                _debug_counter = _DebugCounter()
 
574
            _debug_counter.track(self)
485
575
 
486
576
    def _is_remote_before(self, version_tuple):
487
577
        """Is it possible the remote side supports RPCs for a given version?
512
602
        """
513
603
        if (self._remote_version_is_before is not None and
514
604
            version_tuple > self._remote_version_is_before):
515
 
            raise AssertionError(
 
605
            # We have been told that the remote side is older than some version
 
606
            # which is newer than a previously supplied older-than version.
 
607
            # This indicates that some smart verb call is not guarded
 
608
            # appropriately (it should simply not have been tried).
 
609
            trace.mutter(
516
610
                "_remember_remote_is_before(%r) called, but "
517
611
                "_remember_remote_is_before(%r) was called previously."
518
 
                % (version_tuple, self._remote_version_is_before))
 
612
                , version_tuple, self._remote_version_is_before)
 
613
            if 'hpss' in debug.debug_flags:
 
614
                ui.ui_factory.show_warning(
 
615
                    "_remember_remote_is_before(%r) called, but "
 
616
                    "_remember_remote_is_before(%r) was called previously."
 
617
                    % (version_tuple, self._remote_version_is_before))
 
618
            return
519
619
        self._remote_version_is_before = version_tuple
520
620
 
521
621
    def protocol_version(self):
555
655
 
556
656
    def disconnect(self):
557
657
        """If this medium maintains a persistent connection, close it.
558
 
        
 
658
 
559
659
        The default implementation does nothing.
560
660
        """
561
 
        
 
661
 
562
662
    def remote_path_from_transport(self, transport):
563
663
        """Convert transport into a path suitable for using in a request.
564
 
        
 
664
 
565
665
        Note that the resulting remote path doesn't encode the host name or
566
666
        anything but path, so it is only safe to use it in requests sent over
567
667
        the medium from the matching transport.
595
695
 
596
696
    def _flush(self):
597
697
        """Flush the output stream.
598
 
        
 
698
 
599
699
        This method is used by the SmartClientStreamMediumRequest to ensure that
600
700
        all data for a request is sent, to avoid long timeouts or deadlocks.
601
701
        """
612
712
 
613
713
class SmartSimplePipesClientMedium(SmartClientStreamMedium):
614
714
    """A client medium using simple pipes.
615
 
    
 
715
 
616
716
    This client does not manage the pipes: it assumes they will always be open.
617
717
    """
618
718
 
624
724
    def _accept_bytes(self, bytes):
625
725
        """See SmartClientStreamMedium.accept_bytes."""
626
726
        self._writeable_pipe.write(bytes)
 
727
        self._report_activity(len(bytes), 'write')
627
728
 
628
729
    def _flush(self):
629
730
        """See SmartClientStreamMedium._flush()."""
631
732
 
632
733
    def _read_bytes(self, count):
633
734
        """See SmartClientStreamMedium._read_bytes."""
634
 
        return self._readable_pipe.read(count)
 
735
        bytes_to_read = min(count, _MAX_READ_SIZE)
 
736
        bytes = self._readable_pipe.read(bytes_to_read)
 
737
        self._report_activity(len(bytes), 'read')
 
738
        return bytes
 
739
 
 
740
 
 
741
class SSHParams(object):
 
742
    """A set of parameters for starting a remote bzr via SSH."""
 
743
 
 
744
    def __init__(self, host, port=None, username=None, password=None,
 
745
            bzr_remote_path='bzr'):
 
746
        self.host = host
 
747
        self.port = port
 
748
        self.username = username
 
749
        self.password = password
 
750
        self.bzr_remote_path = bzr_remote_path
635
751
 
636
752
 
637
753
class SmartSSHClientMedium(SmartClientStreamMedium):
638
 
    """A client medium using SSH."""
 
754
    """A client medium using SSH.
639
755
    
640
 
    def __init__(self, host, port=None, username=None, password=None,
641
 
            base=None, vendor=None, bzr_remote_path=None):
 
756
    It delegates IO to a SmartClientSocketMedium or
 
757
    SmartClientAlreadyConnectedSocketMedium (depending on platform).
 
758
    """
 
759
 
 
760
    def __init__(self, base, ssh_params, vendor=None):
642
761
        """Creates a client that will connect on the first use.
643
 
        
 
762
 
 
763
        :param ssh_params: A SSHParams instance.
644
764
        :param vendor: An optional override for the ssh vendor to use. See
645
765
            bzrlib.transport.ssh for details on ssh vendors.
646
766
        """
 
767
        self._real_medium = None
 
768
        self._ssh_params = ssh_params
 
769
        # for the benefit of progress making a short description of this
 
770
        # transport
 
771
        self._scheme = 'bzr+ssh'
 
772
        # SmartClientStreamMedium stores the repr of this object in its
 
773
        # _DebugCounter so we have to store all the values used in our repr
 
774
        # method before calling the super init.
647
775
        SmartClientStreamMedium.__init__(self, base)
648
 
        self._connected = False
649
 
        self._host = host
650
 
        self._password = password
651
 
        self._port = port
652
 
        self._username = username
653
 
        self._read_from = None
 
776
        self._vendor = vendor
654
777
        self._ssh_connection = None
655
 
        self._vendor = vendor
656
 
        self._write_to = None
657
 
        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')
 
778
 
 
779
    def __repr__(self):
 
780
        if self._ssh_params.port is None:
 
781
            maybe_port = ''
 
782
        else:
 
783
            maybe_port = ':%s' % self._ssh_params.port
 
784
        return "%s(%s://%s@%s%s/)" % (
 
785
            self.__class__.__name__,
 
786
            self._scheme,
 
787
            self._ssh_params.username,
 
788
            self._ssh_params.host,
 
789
            maybe_port)
663
790
 
664
791
    def _accept_bytes(self, bytes):
665
792
        """See SmartClientStreamMedium.accept_bytes."""
666
793
        self._ensure_connection()
667
 
        self._write_to.write(bytes)
 
794
        self._real_medium.accept_bytes(bytes)
668
795
 
669
796
    def disconnect(self):
670
797
        """See SmartClientMedium.disconnect()."""
671
 
        if not self._connected:
672
 
            return
673
 
        self._read_from.close()
674
 
        self._write_to.close()
675
 
        self._ssh_connection.close()
676
 
        self._connected = False
 
798
        if self._real_medium is not None:
 
799
            self._real_medium.disconnect()
 
800
            self._real_medium = None
 
801
        if self._ssh_connection is not None:
 
802
            self._ssh_connection.close()
 
803
            self._ssh_connection = None
677
804
 
678
805
    def _ensure_connection(self):
679
806
        """Connect this medium if not already connected."""
680
 
        if self._connected:
 
807
        if self._real_medium is not None:
681
808
            return
682
809
        if self._vendor is None:
683
810
            vendor = ssh._get_ssh_vendor()
684
811
        else:
685
812
            vendor = self._vendor
686
 
        self._ssh_connection = vendor.connect_ssh(self._username,
687
 
                self._password, self._host, self._port,
688
 
                command=[self._bzr_remote_path, 'serve', '--inet',
 
813
        self._ssh_connection = vendor.connect_ssh(self._ssh_params.username,
 
814
                self._ssh_params.password, self._ssh_params.host,
 
815
                self._ssh_params.port,
 
816
                command=[self._ssh_params.bzr_remote_path, 'serve', '--inet',
689
817
                         '--directory=/', '--allow-writes'])
690
 
        self._read_from, self._write_to = \
691
 
            self._ssh_connection.get_filelike_channels()
692
 
        self._connected = True
 
818
        io_kind, io_object = self._ssh_connection.get_sock_or_pipes()
 
819
        if io_kind == 'socket':
 
820
            self._real_medium = SmartClientAlreadyConnectedSocketMedium(
 
821
                self.base, io_object)
 
822
        elif io_kind == 'pipes':
 
823
            read_from, write_to = io_object
 
824
            self._real_medium = SmartSimplePipesClientMedium(
 
825
                read_from, write_to, self.base)
 
826
        else:
 
827
            raise AssertionError(
 
828
                "Unexpected io_kind %r from %r"
 
829
                % (io_kind, self._ssh_connection))
693
830
 
694
831
    def _flush(self):
695
832
        """See SmartClientStreamMedium._flush()."""
696
 
        self._write_to.flush()
 
833
        self._real_medium._flush()
697
834
 
698
835
    def _read_bytes(self, count):
699
836
        """See SmartClientStreamMedium.read_bytes."""
700
 
        if not self._connected:
 
837
        if self._real_medium is None:
701
838
            raise errors.MediumNotConnected(self)
702
 
        bytes_to_read = min(count, _MAX_READ_SIZE)
703
 
        return self._read_from.read(bytes_to_read)
 
839
        return self._real_medium.read_bytes(count)
704
840
 
705
841
 
706
842
# Port 4155 is the default port for bzr://, registered with IANA.
707
 
BZR_DEFAULT_INTERFACE = '0.0.0.0'
 
843
BZR_DEFAULT_INTERFACE = None
708
844
BZR_DEFAULT_PORT = 4155
709
845
 
710
846
 
711
 
class SmartTCPClientMedium(SmartClientStreamMedium):
712
 
    """A client medium using TCP."""
 
847
class SmartClientSocketMedium(SmartClientStreamMedium):
 
848
    """A client medium using a socket.
713
849
    
 
850
    This class isn't usable directly.  Use one of its subclasses instead.
 
851
    """
 
852
 
 
853
    def __init__(self, base):
 
854
        SmartClientStreamMedium.__init__(self, base)
 
855
        self._socket = None
 
856
        self._connected = False
 
857
 
 
858
    def _accept_bytes(self, bytes):
 
859
        """See SmartClientMedium.accept_bytes."""
 
860
        self._ensure_connection()
 
861
        osutils.send_all(self._socket, bytes, self._report_activity)
 
862
 
 
863
    def _ensure_connection(self):
 
864
        """Connect this medium if not already connected."""
 
865
        raise NotImplementedError(self._ensure_connection)
 
866
 
 
867
    def _flush(self):
 
868
        """See SmartClientStreamMedium._flush().
 
869
 
 
870
        For sockets we do no flushing. For TCP sockets we may want to turn off
 
871
        TCP_NODELAY and add a means to do a flush, but that can be done in the
 
872
        future.
 
873
        """
 
874
 
 
875
    def _read_bytes(self, count):
 
876
        """See SmartClientMedium.read_bytes."""
 
877
        if not self._connected:
 
878
            raise errors.MediumNotConnected(self)
 
879
        return osutils.read_bytes_from_socket(
 
880
            self._socket, self._report_activity)
 
881
 
 
882
    def disconnect(self):
 
883
        """See SmartClientMedium.disconnect()."""
 
884
        if not self._connected:
 
885
            return
 
886
        self._socket.close()
 
887
        self._socket = None
 
888
        self._connected = False
 
889
 
 
890
 
 
891
class SmartTCPClientMedium(SmartClientSocketMedium):
 
892
    """A client medium that creates a TCP connection."""
 
893
 
714
894
    def __init__(self, host, port, base):
715
895
        """Creates a client that will connect on the first use."""
716
 
        SmartClientStreamMedium.__init__(self, base)
717
 
        self._connected = False
 
896
        SmartClientSocketMedium.__init__(self, base)
718
897
        self._host = host
719
898
        self._port = port
720
 
        self._socket = None
721
 
 
722
 
    def _accept_bytes(self, bytes):
723
 
        """See SmartClientMedium.accept_bytes."""
724
 
        self._ensure_connection()
725
 
        osutils.send_all(self._socket, bytes)
726
 
 
727
 
    def disconnect(self):
728
 
        """See SmartClientMedium.disconnect()."""
729
 
        if not self._connected:
730
 
            return
731
 
        self._socket.close()
732
 
        self._socket = None
733
 
        self._connected = False
734
899
 
735
900
    def _ensure_connection(self):
736
901
        """Connect this medium if not already connected."""
737
902
        if self._connected:
738
903
            return
739
 
        self._socket = socket.socket()
740
 
        self._socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
741
904
        if self._port is None:
742
905
            port = BZR_DEFAULT_PORT
743
906
        else:
744
907
            port = int(self._port)
745
908
        try:
746
 
            self._socket.connect((self._host, port))
747
 
        except socket.error, err:
 
909
            sockaddrs = socket.getaddrinfo(self._host, port, socket.AF_UNSPEC,
 
910
                socket.SOCK_STREAM, 0, 0)
 
911
        except socket.gaierror, (err_num, err_msg):
 
912
            raise errors.ConnectionError("failed to lookup %s:%d: %s" %
 
913
                    (self._host, port, err_msg))
 
914
        # Initialize err in case there are no addresses returned:
 
915
        err = socket.error("no address found for %s" % self._host)
 
916
        for (family, socktype, proto, canonname, sockaddr) in sockaddrs:
 
917
            try:
 
918
                self._socket = socket.socket(family, socktype, proto)
 
919
                self._socket.setsockopt(socket.IPPROTO_TCP,
 
920
                                        socket.TCP_NODELAY, 1)
 
921
                self._socket.connect(sockaddr)
 
922
            except socket.error, err:
 
923
                if self._socket is not None:
 
924
                    self._socket.close()
 
925
                self._socket = None
 
926
                continue
 
927
            break
 
928
        if self._socket is None:
748
929
            # socket errors either have a (string) or (errno, string) as their
749
930
            # args.
750
931
            if type(err.args) is str:
755
936
                    (self._host, port, err_msg))
756
937
        self._connected = True
757
938
 
758
 
    def _flush(self):
759
 
        """See SmartClientStreamMedium._flush().
760
 
        
761
 
        For TCP we do no flushing. We may want to turn off TCP_NODELAY and 
762
 
        add a means to do a flush, but that can be done in the future.
763
 
        """
764
 
 
765
 
    def _read_bytes(self, count):
766
 
        """See SmartClientMedium.read_bytes."""
767
 
        if not self._connected:
768
 
            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)
 
939
 
 
940
class SmartClientAlreadyConnectedSocketMedium(SmartClientSocketMedium):
 
941
    """A client medium for an already connected socket.
 
942
    
 
943
    Note that this class will assume it "owns" the socket, so it will close it
 
944
    when its disconnect method is called.
 
945
    """
 
946
 
 
947
    def __init__(self, base, sock):
 
948
        SmartClientSocketMedium.__init__(self, base)
 
949
        self._socket = sock
 
950
        self._connected = True
 
951
 
 
952
    def _ensure_connection(self):
 
953
        # Already connected, by definition!  So nothing to do.
 
954
        pass
772
955
 
773
956
 
774
957
class SmartClientStreamMediumRequest(SmartClientMediumRequest):
787
970
 
788
971
    def _accept_bytes(self, bytes):
789
972
        """See SmartClientMediumRequest._accept_bytes.
790
 
        
 
973
 
791
974
        This forwards to self._medium._accept_bytes because we are operating
792
975
        on the mediums stream.
793
976
        """
796
979
    def _finished_reading(self):
797
980
        """See SmartClientMediumRequest._finished_reading.
798
981
 
799
 
        This clears the _current_request on self._medium to allow a new 
 
982
        This clears the _current_request on self._medium to allow a new
800
983
        request to be created.
801
984
        """
802
985
        if self._medium._current_request is not self:
803
986
            raise AssertionError()
804
987
        self._medium._current_request = None
805
 
        
 
988
 
806
989
    def _finished_writing(self):
807
990
        """See SmartClientMediumRequest._finished_writing.
808
991
 
810
993
        """
811
994
        self._medium._flush()
812
995
 
 
996