← 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: 2012-03-09 16:48:55 UTC
  • mto: (6437.23.24 2.5)
  • mto: This revision was merged to the branch mainline in revision 6499.
  • Revision ID: v.ladeuil+lp@free.fr-20120309164855-htdn25hp7x65mmir
https://launchpad.net/bugs/940164
Rely on sphinx for texinfo doc generation

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
 
24
24
bzrlib/transport/smart/__init__.py.
25
25
"""
26
26
 
 
27
from __future__ import absolute_import
 
28
 
 
29
import errno
27
30
import os
28
 
import socket
29
31
import sys
30
 
import urllib
 
32
import time
31
33
 
 
34
import bzrlib
32
35
from bzrlib.lazy_import import lazy_import
33
36
lazy_import(globals(), """
 
37
import select
 
38
import socket
 
39
import thread
 
40
import weakref
 
41
 
34
42
from bzrlib import (
 
43
    debug,
35
44
    errors,
36
 
    osutils,
37
 
    symbol_versioning,
 
45
    trace,
 
46
    transport,
 
47
    ui,
38
48
    urlutils,
39
49
    )
40
 
from bzrlib.smart import protocol
 
50
from bzrlib.i18n import gettext
 
51
from bzrlib.smart import client, protocol, request, signals, vfs
41
52
from bzrlib.transport import ssh
42
53
""")
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
 
 
 
54
from bzrlib import osutils
 
55
 
 
56
# Throughout this module buffer size parameters are either limited to be at
 
57
# most _MAX_READ_SIZE, or are ignored and _MAX_READ_SIZE is used instead.
 
58
# For this module's purposes, MAX_SOCKET_CHUNK is a reasonable size for reads
 
59
# from non-sockets as well.
 
60
_MAX_READ_SIZE = osutils.MAX_SOCKET_CHUNK
50
61
 
51
62
def _get_protocol_factory_for_bytes(bytes):
52
63
    """Determine the right protocol factory for 'bytes'.
82
93
 
83
94
def _get_line(read_bytes_func):
84
95
    """Read bytes using read_bytes_func until a newline byte.
85
 
    
 
96
 
86
97
    This isn't particularly efficient, so should only be used when the
87
98
    expected size of the line is quite short.
88
 
    
 
99
 
89
100
    :returns: a tuple of two strs: (line, excess)
90
101
    """
91
102
    newline_pos = -1
107
118
 
108
119
    def __init__(self):
109
120
        self._push_back_buffer = None
110
 
        
 
121
 
111
122
    def _push_back(self, bytes):
112
123
        """Return unused bytes to the medium, because they belong to the next
113
124
        request(s).
147
158
 
148
159
    def _get_line(self):
149
160
        """Read bytes from this request's response until a newline byte.
150
 
        
 
161
 
151
162
        This isn't particularly efficient, so should only be used when the
152
163
        expected size of the line is quite short.
153
164
 
156
167
        line, excess = _get_line(self.read_bytes)
157
168
        self._push_back(excess)
158
169
        return line
159
 
 
 
170
 
 
171
    def _report_activity(self, bytes, direction):
 
172
        """Notify that this medium has activity.
 
173
 
 
174
        Implementations should call this from all methods that actually do IO.
 
175
        Be careful that it's not called twice, if one method is implemented on
 
176
        top of another.
 
177
 
 
178
        :param bytes: Number of bytes read or written.
 
179
        :param direction: 'read' or 'write' or None.
 
180
        """
 
181
        ui.ui_factory.report_transport_activity(self, bytes, direction)
 
182
 
 
183
 
 
184
_bad_file_descriptor = (errno.EBADF,)
 
185
if sys.platform == 'win32':
 
186
    # Given on Windows if you pass a closed socket to select.select. Probably
 
187
    # also given if you pass a file handle to select.
 
188
    WSAENOTSOCK = 10038
 
189
    _bad_file_descriptor += (WSAENOTSOCK,)
 
190
 
160
191
 
161
192
class SmartServerStreamMedium(SmartMedium):
162
193
    """Handles smart commands coming over a stream.
167
198
    One instance is created for each connected client; it can serve multiple
168
199
    requests in the lifetime of the connection.
169
200
 
170
 
    The server passes requests through to an underlying backing transport, 
 
201
    The server passes requests through to an underlying backing transport,
171
202
    which will typically be a LocalTransport looking at the server's filesystem.
172
203
 
173
204
    :ivar _push_back_buffer: a str of bytes that have been read from the stream
176
207
        the stream.  See also the _push_back method.
177
208
    """
178
209
 
179
 
    def __init__(self, backing_transport, root_client_path='/'):
 
210
    _timer = time.time
 
211
 
 
212
    def __init__(self, backing_transport, root_client_path='/', timeout=None):
180
213
        """Construct new server.
181
214
 
182
215
        :param backing_transport: Transport for the directory served.
185
218
        self.backing_transport = backing_transport
186
219
        self.root_client_path = root_client_path
187
220
        self.finished = False
 
221
        if timeout is None:
 
222
            raise AssertionError('You must supply a timeout.')
 
223
        self._client_timeout = timeout
 
224
        self._client_poll_timeout = min(timeout / 10.0, 1.0)
188
225
        SmartMedium.__init__(self)
189
226
 
190
227
    def serve(self):
196
233
            while not self.finished:
197
234
                server_protocol = self._build_protocol()
198
235
                self._serve_one_request(server_protocol)
 
236
        except errors.ConnectionTimeout, e:
 
237
            trace.note('%s' % (e,))
 
238
            trace.log_exception_quietly()
 
239
            self._disconnect_client()
 
240
            # We reported it, no reason to make a big fuss.
 
241
            return
199
242
        except Exception, e:
200
243
            stderr.write("%s terminating on exception %s\n" % (self, e))
201
244
            raise
 
245
        self._disconnect_client()
 
246
 
 
247
    def _stop_gracefully(self):
 
248
        """When we finish this message, stop looking for more."""
 
249
        trace.mutter('Stopping %s' % (self,))
 
250
        self.finished = True
 
251
 
 
252
    def _disconnect_client(self):
 
253
        """Close the current connection. We stopped due to a timeout/etc."""
 
254
        # The default implementation is a no-op, because that is all we used to
 
255
        # do when disconnecting from a client. I suppose we never had the
 
256
        # *server* initiate a disconnect, before
 
257
 
 
258
    def _wait_for_bytes_with_timeout(self, timeout_seconds):
 
259
        """Wait for more bytes to be read, but timeout if none available.
 
260
 
 
261
        This allows us to detect idle connections, and stop trying to read from
 
262
        them, without setting the socket itself to non-blocking. This also
 
263
        allows us to specify when we watch for idle timeouts.
 
264
 
 
265
        :return: Did we timeout? (True if we timed out, False if there is data
 
266
            to be read)
 
267
        """
 
268
        raise NotImplementedError(self._wait_for_bytes_with_timeout)
202
269
 
203
270
    def _build_protocol(self):
204
271
        """Identifies the version of the incoming request, and returns an
209
276
 
210
277
        :returns: a SmartServerRequestProtocol.
211
278
        """
 
279
        self._wait_for_bytes_with_timeout(self._client_timeout)
 
280
        if self.finished:
 
281
            # We're stopping, so don't try to do any more work
 
282
            return None
212
283
        bytes = self._get_line()
213
284
        protocol_factory, unused_bytes = _get_protocol_factory_for_bytes(bytes)
214
285
        protocol = protocol_factory(
216
287
        protocol.accept_bytes(unused_bytes)
217
288
        return protocol
218
289
 
 
290
    def _wait_on_descriptor(self, fd, timeout_seconds):
 
291
        """select() on a file descriptor, waiting for nonblocking read()
 
292
 
 
293
        This will raise a ConnectionTimeout exception if we do not get a
 
294
        readable handle before timeout_seconds.
 
295
        :return: None
 
296
        """
 
297
        t_end = self._timer() + timeout_seconds
 
298
        poll_timeout = min(timeout_seconds, self._client_poll_timeout)
 
299
        rs = xs = None
 
300
        while not rs and not xs and self._timer() < t_end:
 
301
            if self.finished:
 
302
                return
 
303
            try:
 
304
                rs, _, xs = select.select([fd], [], [fd], poll_timeout)
 
305
            except (select.error, socket.error) as e:
 
306
                err = getattr(e, 'errno', None)
 
307
                if err is None and getattr(e, 'args', None) is not None:
 
308
                    # select.error doesn't have 'errno', it just has args[0]
 
309
                    err = e.args[0]
 
310
                if err in _bad_file_descriptor:
 
311
                    return # Not a socket indicates read() will fail
 
312
                elif err == errno.EINTR:
 
313
                    # Interrupted, keep looping.
 
314
                    continue
 
315
                raise
 
316
        if rs or xs:
 
317
            return
 
318
        raise errors.ConnectionTimeout('disconnecting client after %.1f seconds'
 
319
                                       % (timeout_seconds,))
 
320
 
219
321
    def _serve_one_request(self, protocol):
220
322
        """Read one request from input, process, send back a response.
221
 
        
 
323
 
222
324
        :param protocol: a SmartServerRequestProtocol.
223
325
        """
 
326
        if protocol is None:
 
327
            return
224
328
        try:
225
329
            self._serve_one_request_unguarded(protocol)
226
330
        except KeyboardInterrupt:
242
346
 
243
347
class SmartServerSocketStreamMedium(SmartServerStreamMedium):
244
348
 
245
 
    def __init__(self, sock, backing_transport, root_client_path='/'):
 
349
    def __init__(self, sock, backing_transport, root_client_path='/',
 
350
                 timeout=None):
246
351
        """Constructor.
247
352
 
248
353
        :param sock: the socket the server will read from.  It will be put
249
354
            into blocking mode.
250
355
        """
251
356
        SmartServerStreamMedium.__init__(
252
 
            self, backing_transport, root_client_path=root_client_path)
 
357
            self, backing_transport, root_client_path=root_client_path,
 
358
            timeout=timeout)
253
359
        sock.setblocking(True)
254
360
        self.socket = sock
 
361
        # Get the getpeername now, as we might be closed later when we care.
 
362
        try:
 
363
            self._client_info = sock.getpeername()
 
364
        except socket.error:
 
365
            self._client_info = '<unknown>'
 
366
 
 
367
    def __str__(self):
 
368
        return '%s(client=%s)' % (self.__class__.__name__, self._client_info)
 
369
 
 
370
    def __repr__(self):
 
371
        return '%s.%s(client=%s)' % (self.__module__, self.__class__.__name__,
 
372
            self._client_info)
255
373
 
256
374
    def _serve_one_request_unguarded(self, protocol):
257
375
        while protocol.next_read_size():
258
376
            # 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)
 
377
            # than MAX_SOCKET_CHUNK ready, the socket will just return a
 
378
            # short read immediately rather than block.
 
379
            bytes = self.read_bytes(osutils.MAX_SOCKET_CHUNK)
262
380
            if bytes == '':
263
381
                self.finished = True
264
382
                return
265
383
            protocol.accept_bytes(bytes)
266
 
        
 
384
 
267
385
        self._push_back(protocol.unused_data)
268
386
 
 
387
    def _disconnect_client(self):
 
388
        """Close the current connection. We stopped due to a timeout/etc."""
 
389
        self.socket.close()
 
390
 
 
391
    def _wait_for_bytes_with_timeout(self, timeout_seconds):
 
392
        """Wait for more bytes to be read, but timeout if none available.
 
393
 
 
394
        This allows us to detect idle connections, and stop trying to read from
 
395
        them, without setting the socket itself to non-blocking. This also
 
396
        allows us to specify when we watch for idle timeouts.
 
397
 
 
398
        :return: None, this will raise ConnectionTimeout if we time out before
 
399
            data is available.
 
400
        """
 
401
        return self._wait_on_descriptor(self.socket, timeout_seconds)
 
402
 
269
403
    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)
 
404
        return osutils.read_bytes_from_socket(
 
405
            self.socket, self._report_activity)
273
406
 
274
407
    def terminate_due_to_error(self):
275
408
        # TODO: This should log to a server log file, but no such thing
278
411
        self.finished = True
279
412
 
280
413
    def _write_out(self, bytes):
281
 
        osutils.send_all(self.socket, bytes)
 
414
        tstart = osutils.timer_func()
 
415
        osutils.send_all(self.socket, bytes, self._report_activity)
 
416
        if 'hpss' in debug.debug_flags:
 
417
            thread_id = thread.get_ident()
 
418
            trace.mutter('%12s: [%s] %d bytes to the socket in %.3fs'
 
419
                         % ('wrote', thread_id, len(bytes),
 
420
                            osutils.timer_func() - tstart))
282
421
 
283
422
 
284
423
class SmartServerPipeStreamMedium(SmartServerStreamMedium):
285
424
 
286
 
    def __init__(self, in_file, out_file, backing_transport):
 
425
    def __init__(self, in_file, out_file, backing_transport, timeout=None):
287
426
        """Construct new server.
288
427
 
289
428
        :param in_file: Python file from which requests can be read.
290
429
        :param out_file: Python file to write responses.
291
430
        :param backing_transport: Transport for the directory served.
292
431
        """
293
 
        SmartServerStreamMedium.__init__(self, backing_transport)
 
432
        SmartServerStreamMedium.__init__(self, backing_transport,
 
433
            timeout=timeout)
294
434
        if sys.platform == 'win32':
295
435
            # force binary mode for files
296
436
            import msvcrt
301
441
        self._in = in_file
302
442
        self._out = out_file
303
443
 
 
444
    def serve(self):
 
445
        """See SmartServerStreamMedium.serve"""
 
446
        # This is the regular serve, except it adds signal trapping for soft
 
447
        # shutdown.
 
448
        stop_gracefully = self._stop_gracefully
 
449
        signals.register_on_hangup(id(self), stop_gracefully)
 
450
        try:
 
451
            return super(SmartServerPipeStreamMedium, self).serve()
 
452
        finally:
 
453
            signals.unregister_on_hangup(id(self))
 
454
 
304
455
    def _serve_one_request_unguarded(self, protocol):
305
456
        while True:
306
457
            # We need to be careful not to read past the end of the current
319
470
                return
320
471
            protocol.accept_bytes(bytes)
321
472
 
 
473
    def _disconnect_client(self):
 
474
        self._in.close()
 
475
        self._out.flush()
 
476
        self._out.close()
 
477
 
 
478
    def _wait_for_bytes_with_timeout(self, timeout_seconds):
 
479
        """Wait for more bytes to be read, but timeout if none available.
 
480
 
 
481
        This allows us to detect idle connections, and stop trying to read from
 
482
        them, without setting the socket itself to non-blocking. This also
 
483
        allows us to specify when we watch for idle timeouts.
 
484
 
 
485
        :return: None, this will raise ConnectionTimeout if we time out before
 
486
            data is available.
 
487
        """
 
488
        if (getattr(self._in, 'fileno', None) is None
 
489
            or sys.platform == 'win32'):
 
490
            # You can't select() file descriptors on Windows.
 
491
            return
 
492
        return self._wait_on_descriptor(self._in, timeout_seconds)
 
493
 
322
494
    def _read_bytes(self, desired_count):
323
495
        return self._in.read(desired_count)
324
496
 
345
517
    request.finished_reading()
346
518
 
347
519
    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 
 
520
    requests can exist. See SmartClientMedium.get_request to obtain instances
 
521
    of SmartClientMediumRequest, and the concrete Medium you are using for
350
522
    details on concurrency and pipelining.
351
523
    """
352
524
 
361
533
    def accept_bytes(self, bytes):
362
534
        """Accept bytes for inclusion in this request.
363
535
 
364
 
        This method may not be be called after finished_writing() has been
 
536
        This method may not be called after finished_writing() has been
365
537
        called.  It depends upon the Medium whether or not the bytes will be
366
538
        immediately transmitted. Message based Mediums will tend to buffer the
367
539
        bytes until finished_writing() is called.
398
570
    def _finished_reading(self):
399
571
        """Helper for finished_reading.
400
572
 
401
 
        finished_reading checks the state of the request to determine if 
 
573
        finished_reading checks the state of the request to determine if
402
574
        finished_reading is allowed, and if it is hands off to _finished_reading
403
575
        to perform the action.
404
576
        """
418
590
    def _finished_writing(self):
419
591
        """Helper for finished_writing.
420
592
 
421
 
        finished_writing checks the state of the request to determine if 
 
593
        finished_writing checks the state of the request to determine if
422
594
        finished_writing is allowed, and if it is hands off to _finished_writing
423
595
        to perform the action.
424
596
        """
444
616
        read_bytes checks the state of the request to determing if bytes
445
617
        should be read. After that it hands off to _read_bytes to do the
446
618
        actual read.
447
 
        
 
619
 
448
620
        By default this forwards to self._medium.read_bytes because we are
449
621
        operating on the medium's stream.
450
622
        """
455
627
        if not line.endswith('\n'):
456
628
            # end of file encountered reading from server
457
629
            raise errors.ConnectionReset(
458
 
                "please check connectivity and permissions",
459
 
                "(and try -Dhpss if further diagnosis is required)")
 
630
                "Unexpected end of message. Please check connectivity "
 
631
                "and permissions, and report a bug if problems persist.")
460
632
        return line
461
633
 
462
634
    def _read_line(self):
463
635
        """Helper for SmartClientMediumRequest.read_line.
464
 
        
 
636
 
465
637
        By default this forwards to self._medium._get_line because we are
466
638
        operating on the medium's stream.
467
639
        """
468
640
        return self._medium._get_line()
469
641
 
470
642
 
 
643
class _VfsRefuser(object):
 
644
    """An object that refuses all VFS requests.
 
645
 
 
646
    """
 
647
 
 
648
    def __init__(self):
 
649
        client._SmartClient.hooks.install_named_hook(
 
650
            'call', self.check_vfs, 'vfs refuser')
 
651
 
 
652
    def check_vfs(self, params):
 
653
        try:
 
654
            request_method = request.request_handlers.get(params.method)
 
655
        except KeyError:
 
656
            # A method we don't know about doesn't count as a VFS method.
 
657
            return
 
658
        if issubclass(request_method, vfs.VfsRequest):
 
659
            raise errors.HpssVfsRequestNotAllowed(params.method, params.args)
 
660
 
 
661
 
 
662
class _DebugCounter(object):
 
663
    """An object that counts the HPSS calls made to each client medium.
 
664
 
 
665
    When a medium is garbage-collected, or failing that when
 
666
    bzrlib.global_state exits, the total number of calls made on that medium
 
667
    are reported via trace.note.
 
668
    """
 
669
 
 
670
    def __init__(self):
 
671
        self.counts = weakref.WeakKeyDictionary()
 
672
        client._SmartClient.hooks.install_named_hook(
 
673
            'call', self.increment_call_count, 'hpss call counter')
 
674
        bzrlib.global_state.cleanups.add_cleanup(self.flush_all)
 
675
 
 
676
    def track(self, medium):
 
677
        """Start tracking calls made to a medium.
 
678
 
 
679
        This only keeps a weakref to the medium, so shouldn't affect the
 
680
        medium's lifetime.
 
681
        """
 
682
        medium_repr = repr(medium)
 
683
        # Add this medium to the WeakKeyDictionary
 
684
        self.counts[medium] = dict(count=0, vfs_count=0,
 
685
                                   medium_repr=medium_repr)
 
686
        # Weakref callbacks are fired in reverse order of their association
 
687
        # with the referenced object.  So we add a weakref *after* adding to
 
688
        # the WeakKeyDict so that we can report the value from it before the
 
689
        # entry is removed by the WeakKeyDict's own callback.
 
690
        ref = weakref.ref(medium, self.done)
 
691
 
 
692
    def increment_call_count(self, params):
 
693
        # Increment the count in the WeakKeyDictionary
 
694
        value = self.counts[params.medium]
 
695
        value['count'] += 1
 
696
        try:
 
697
            request_method = request.request_handlers.get(params.method)
 
698
        except KeyError:
 
699
            # A method we don't know about doesn't count as a VFS method.
 
700
            return
 
701
        if issubclass(request_method, vfs.VfsRequest):
 
702
            value['vfs_count'] += 1
 
703
 
 
704
    def done(self, ref):
 
705
        value = self.counts[ref]
 
706
        count, vfs_count, medium_repr = (
 
707
            value['count'], value['vfs_count'], value['medium_repr'])
 
708
        # In case this callback is invoked for the same ref twice (by the
 
709
        # weakref callback and by the atexit function), set the call count back
 
710
        # to 0 so this item won't be reported twice.
 
711
        value['count'] = 0
 
712
        value['vfs_count'] = 0
 
713
        if count != 0:
 
714
            trace.note(gettext('HPSS calls: {0} ({1} vfs) {2}').format(
 
715
                       count, vfs_count, medium_repr))
 
716
 
 
717
    def flush_all(self):
 
718
        for ref in list(self.counts.keys()):
 
719
            self.done(ref)
 
720
 
 
721
_debug_counter = None
 
722
_vfs_refuser = None
 
723
 
 
724
 
471
725
class SmartClientMedium(SmartMedium):
472
726
    """Smart client is a medium for sending smart protocol requests over."""
473
727
 
482
736
        # _remote_version_is_before tracks the bzr version the remote side
483
737
        # can be based on what we've seen so far.
484
738
        self._remote_version_is_before = None
 
739
        # Install debug hook function if debug flag is set.
 
740
        if 'hpss' in debug.debug_flags:
 
741
            global _debug_counter
 
742
            if _debug_counter is None:
 
743
                _debug_counter = _DebugCounter()
 
744
            _debug_counter.track(self)
 
745
        if 'hpss_client_no_vfs' in debug.debug_flags:
 
746
            global _vfs_refuser
 
747
            if _vfs_refuser is None:
 
748
                _vfs_refuser = _VfsRefuser()
485
749
 
486
750
    def _is_remote_before(self, version_tuple):
487
751
        """Is it possible the remote side supports RPCs for a given version?
512
776
        """
513
777
        if (self._remote_version_is_before is not None and
514
778
            version_tuple > self._remote_version_is_before):
515
 
            raise AssertionError(
 
779
            # We have been told that the remote side is older than some version
 
780
            # which is newer than a previously supplied older-than version.
 
781
            # This indicates that some smart verb call is not guarded
 
782
            # appropriately (it should simply not have been tried).
 
783
            trace.mutter(
516
784
                "_remember_remote_is_before(%r) called, but "
517
785
                "_remember_remote_is_before(%r) was called previously."
518
 
                % (version_tuple, self._remote_version_is_before))
 
786
                , version_tuple, self._remote_version_is_before)
 
787
            if 'hpss' in debug.debug_flags:
 
788
                ui.ui_factory.show_warning(
 
789
                    "_remember_remote_is_before(%r) called, but "
 
790
                    "_remember_remote_is_before(%r) was called previously."
 
791
                    % (version_tuple, self._remote_version_is_before))
 
792
            return
519
793
        self._remote_version_is_before = version_tuple
520
794
 
521
795
    def protocol_version(self):
555
829
 
556
830
    def disconnect(self):
557
831
        """If this medium maintains a persistent connection, close it.
558
 
        
 
832
 
559
833
        The default implementation does nothing.
560
834
        """
561
 
        
 
835
 
562
836
    def remote_path_from_transport(self, transport):
563
837
        """Convert transport into a path suitable for using in a request.
564
 
        
 
838
 
565
839
        Note that the resulting remote path doesn't encode the host name or
566
840
        anything but path, so it is only safe to use it in requests sent over
567
841
        the medium from the matching transport.
568
842
        """
569
843
        medium_base = urlutils.join(self.base, '/')
570
844
        rel_url = urlutils.relative_url(medium_base, transport.base)
571
 
        return urllib.unquote(rel_url)
 
845
        return urlutils.unquote(rel_url)
572
846
 
573
847
 
574
848
class SmartClientStreamMedium(SmartClientMedium):
595
869
 
596
870
    def _flush(self):
597
871
        """Flush the output stream.
598
 
        
 
872
 
599
873
        This method is used by the SmartClientStreamMediumRequest to ensure that
600
874
        all data for a request is sent, to avoid long timeouts or deadlocks.
601
875
        """
609
883
        """
610
884
        return SmartClientStreamMediumRequest(self)
611
885
 
 
886
    def reset(self):
 
887
        """We have been disconnected, reset current state.
 
888
 
 
889
        This resets things like _current_request and connected state.
 
890
        """
 
891
        self.disconnect()
 
892
        self._current_request = None
 
893
 
612
894
 
613
895
class SmartSimplePipesClientMedium(SmartClientStreamMedium):
614
896
    """A client medium using simple pipes.
615
 
    
 
897
 
616
898
    This client does not manage the pipes: it assumes they will always be open.
617
899
    """
618
900
 
623
905
 
624
906
    def _accept_bytes(self, bytes):
625
907
        """See SmartClientStreamMedium.accept_bytes."""
626
 
        self._writeable_pipe.write(bytes)
 
908
        try:
 
909
            self._writeable_pipe.write(bytes)
 
910
        except IOError, e:
 
911
            if e.errno in (errno.EINVAL, errno.EPIPE):
 
912
                raise errors.ConnectionReset(
 
913
                    "Error trying to write to subprocess:\n%s" % (e,))
 
914
            raise
 
915
        self._report_activity(len(bytes), 'write')
627
916
 
628
917
    def _flush(self):
629
918
        """See SmartClientStreamMedium._flush()."""
 
919
        # Note: If flush were to fail, we'd like to raise ConnectionReset, etc.
 
920
        #       However, testing shows that even when the child process is
 
921
        #       gone, this doesn't error.
630
922
        self._writeable_pipe.flush()
631
923
 
632
924
    def _read_bytes(self, count):
633
925
        """See SmartClientStreamMedium._read_bytes."""
634
 
        return self._readable_pipe.read(count)
 
926
        bytes_to_read = min(count, _MAX_READ_SIZE)
 
927
        bytes = self._readable_pipe.read(bytes_to_read)
 
928
        self._report_activity(len(bytes), 'read')
 
929
        return bytes
 
930
 
 
931
 
 
932
class SSHParams(object):
 
933
    """A set of parameters for starting a remote bzr via SSH."""
 
934
 
 
935
    def __init__(self, host, port=None, username=None, password=None,
 
936
            bzr_remote_path='bzr'):
 
937
        self.host = host
 
938
        self.port = port
 
939
        self.username = username
 
940
        self.password = password
 
941
        self.bzr_remote_path = bzr_remote_path
635
942
 
636
943
 
637
944
class SmartSSHClientMedium(SmartClientStreamMedium):
638
 
    """A client medium using SSH."""
639
 
    
640
 
    def __init__(self, host, port=None, username=None, password=None,
641
 
            base=None, vendor=None, bzr_remote_path=None):
 
945
    """A client medium using SSH.
 
946
 
 
947
    It delegates IO to a SmartSimplePipesClientMedium or
 
948
    SmartClientAlreadyConnectedSocketMedium (depending on platform).
 
949
    """
 
950
 
 
951
    def __init__(self, base, ssh_params, vendor=None):
642
952
        """Creates a client that will connect on the first use.
643
 
        
 
953
 
 
954
        :param ssh_params: A SSHParams instance.
644
955
        :param vendor: An optional override for the ssh vendor to use. See
645
956
            bzrlib.transport.ssh for details on ssh vendors.
646
957
        """
 
958
        self._real_medium = None
 
959
        self._ssh_params = ssh_params
 
960
        # for the benefit of progress making a short description of this
 
961
        # transport
 
962
        self._scheme = 'bzr+ssh'
 
963
        # SmartClientStreamMedium stores the repr of this object in its
 
964
        # _DebugCounter so we have to store all the values used in our repr
 
965
        # method before calling the super init.
647
966
        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
 
967
        self._vendor = vendor
654
968
        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')
 
969
 
 
970
    def __repr__(self):
 
971
        if self._ssh_params.port is None:
 
972
            maybe_port = ''
 
973
        else:
 
974
            maybe_port = ':%s' % self._ssh_params.port
 
975
        if self._ssh_params.username is None:
 
976
            maybe_user = ''
 
977
        else:
 
978
            maybe_user = '%s@' % self._ssh_params.username
 
979
        return "%s(%s://%s%s%s/)" % (
 
980
            self.__class__.__name__,
 
981
            self._scheme,
 
982
            maybe_user,
 
983
            self._ssh_params.host,
 
984
            maybe_port)
663
985
 
664
986
    def _accept_bytes(self, bytes):
665
987
        """See SmartClientStreamMedium.accept_bytes."""
666
988
        self._ensure_connection()
667
 
        self._write_to.write(bytes)
 
989
        self._real_medium.accept_bytes(bytes)
668
990
 
669
991
    def disconnect(self):
670
992
        """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
 
993
        if self._real_medium is not None:
 
994
            self._real_medium.disconnect()
 
995
            self._real_medium = None
 
996
        if self._ssh_connection is not None:
 
997
            self._ssh_connection.close()
 
998
            self._ssh_connection = None
677
999
 
678
1000
    def _ensure_connection(self):
679
1001
        """Connect this medium if not already connected."""
680
 
        if self._connected:
 
1002
        if self._real_medium is not None:
681
1003
            return
682
1004
        if self._vendor is None:
683
1005
            vendor = ssh._get_ssh_vendor()
684
1006
        else:
685
1007
            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',
 
1008
        self._ssh_connection = vendor.connect_ssh(self._ssh_params.username,
 
1009
                self._ssh_params.password, self._ssh_params.host,
 
1010
                self._ssh_params.port,
 
1011
                command=[self._ssh_params.bzr_remote_path, 'serve', '--inet',
689
1012
                         '--directory=/', '--allow-writes'])
690
 
        self._read_from, self._write_to = \
691
 
            self._ssh_connection.get_filelike_channels()
692
 
        self._connected = True
 
1013
        io_kind, io_object = self._ssh_connection.get_sock_or_pipes()
 
1014
        if io_kind == 'socket':
 
1015
            self._real_medium = SmartClientAlreadyConnectedSocketMedium(
 
1016
                self.base, io_object)
 
1017
        elif io_kind == 'pipes':
 
1018
            read_from, write_to = io_object
 
1019
            self._real_medium = SmartSimplePipesClientMedium(
 
1020
                read_from, write_to, self.base)
 
1021
        else:
 
1022
            raise AssertionError(
 
1023
                "Unexpected io_kind %r from %r"
 
1024
                % (io_kind, self._ssh_connection))
 
1025
        for hook in transport.Transport.hooks["post_connect"]:
 
1026
            hook(self)
693
1027
 
694
1028
    def _flush(self):
695
1029
        """See SmartClientStreamMedium._flush()."""
696
 
        self._write_to.flush()
 
1030
        self._real_medium._flush()
697
1031
 
698
1032
    def _read_bytes(self, count):
699
1033
        """See SmartClientStreamMedium.read_bytes."""
700
 
        if not self._connected:
 
1034
        if self._real_medium is None:
701
1035
            raise errors.MediumNotConnected(self)
702
 
        bytes_to_read = min(count, _MAX_READ_SIZE)
703
 
        return self._read_from.read(bytes_to_read)
 
1036
        return self._real_medium.read_bytes(count)
704
1037
 
705
1038
 
706
1039
# Port 4155 is the default port for bzr://, registered with IANA.
707
 
BZR_DEFAULT_INTERFACE = '0.0.0.0'
 
1040
BZR_DEFAULT_INTERFACE = None
708
1041
BZR_DEFAULT_PORT = 4155
709
1042
 
710
1043
 
711
 
class SmartTCPClientMedium(SmartClientStreamMedium):
712
 
    """A client medium using TCP."""
 
1044
class SmartClientSocketMedium(SmartClientStreamMedium):
 
1045
    """A client medium using a socket.
713
1046
    
 
1047
    This class isn't usable directly.  Use one of its subclasses instead.
 
1048
    """
 
1049
 
 
1050
    def __init__(self, base):
 
1051
        SmartClientStreamMedium.__init__(self, base)
 
1052
        self._socket = None
 
1053
        self._connected = False
 
1054
 
 
1055
    def _accept_bytes(self, bytes):
 
1056
        """See SmartClientMedium.accept_bytes."""
 
1057
        self._ensure_connection()
 
1058
        osutils.send_all(self._socket, bytes, self._report_activity)
 
1059
 
 
1060
    def _ensure_connection(self):
 
1061
        """Connect this medium if not already connected."""
 
1062
        raise NotImplementedError(self._ensure_connection)
 
1063
 
 
1064
    def _flush(self):
 
1065
        """See SmartClientStreamMedium._flush().
 
1066
 
 
1067
        For sockets we do no flushing. For TCP sockets we may want to turn off
 
1068
        TCP_NODELAY and add a means to do a flush, but that can be done in the
 
1069
        future.
 
1070
        """
 
1071
 
 
1072
    def _read_bytes(self, count):
 
1073
        """See SmartClientMedium.read_bytes."""
 
1074
        if not self._connected:
 
1075
            raise errors.MediumNotConnected(self)
 
1076
        return osutils.read_bytes_from_socket(
 
1077
            self._socket, self._report_activity)
 
1078
 
 
1079
    def disconnect(self):
 
1080
        """See SmartClientMedium.disconnect()."""
 
1081
        if not self._connected:
 
1082
            return
 
1083
        self._socket.close()
 
1084
        self._socket = None
 
1085
        self._connected = False
 
1086
 
 
1087
 
 
1088
class SmartTCPClientMedium(SmartClientSocketMedium):
 
1089
    """A client medium that creates a TCP connection."""
 
1090
 
714
1091
    def __init__(self, host, port, base):
715
1092
        """Creates a client that will connect on the first use."""
716
 
        SmartClientStreamMedium.__init__(self, base)
717
 
        self._connected = False
 
1093
        SmartClientSocketMedium.__init__(self, base)
718
1094
        self._host = host
719
1095
        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
1096
 
735
1097
    def _ensure_connection(self):
736
1098
        """Connect this medium if not already connected."""
737
1099
        if self._connected:
738
1100
            return
739
 
        self._socket = socket.socket()
740
 
        self._socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
741
1101
        if self._port is None:
742
1102
            port = BZR_DEFAULT_PORT
743
1103
        else:
744
1104
            port = int(self._port)
745
1105
        try:
746
 
            self._socket.connect((self._host, port))
747
 
        except socket.error, err:
 
1106
            sockaddrs = socket.getaddrinfo(self._host, port, socket.AF_UNSPEC,
 
1107
                socket.SOCK_STREAM, 0, 0)
 
1108
        except socket.gaierror, (err_num, err_msg):
 
1109
            raise errors.ConnectionError("failed to lookup %s:%d: %s" %
 
1110
                    (self._host, port, err_msg))
 
1111
        # Initialize err in case there are no addresses returned:
 
1112
        err = socket.error("no address found for %s" % self._host)
 
1113
        for (family, socktype, proto, canonname, sockaddr) in sockaddrs:
 
1114
            try:
 
1115
                self._socket = socket.socket(family, socktype, proto)
 
1116
                self._socket.setsockopt(socket.IPPROTO_TCP,
 
1117
                                        socket.TCP_NODELAY, 1)
 
1118
                self._socket.connect(sockaddr)
 
1119
            except socket.error, err:
 
1120
                if self._socket is not None:
 
1121
                    self._socket.close()
 
1122
                self._socket = None
 
1123
                continue
 
1124
            break
 
1125
        if self._socket is None:
748
1126
            # socket errors either have a (string) or (errno, string) as their
749
1127
            # args.
750
1128
            if type(err.args) is str:
754
1132
            raise errors.ConnectionError("failed to connect to %s:%d: %s" %
755
1133
                    (self._host, port, err_msg))
756
1134
        self._connected = True
757
 
 
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)
 
1135
        for hook in transport.Transport.hooks["post_connect"]:
 
1136
            hook(self)
 
1137
 
 
1138
 
 
1139
class SmartClientAlreadyConnectedSocketMedium(SmartClientSocketMedium):
 
1140
    """A client medium for an already connected socket.
 
1141
    
 
1142
    Note that this class will assume it "owns" the socket, so it will close it
 
1143
    when its disconnect method is called.
 
1144
    """
 
1145
 
 
1146
    def __init__(self, base, sock):
 
1147
        SmartClientSocketMedium.__init__(self, base)
 
1148
        self._socket = sock
 
1149
        self._connected = True
 
1150
 
 
1151
    def _ensure_connection(self):
 
1152
        # Already connected, by definition!  So nothing to do.
 
1153
        pass
772
1154
 
773
1155
 
774
1156
class SmartClientStreamMediumRequest(SmartClientMediumRequest):
787
1169
 
788
1170
    def _accept_bytes(self, bytes):
789
1171
        """See SmartClientMediumRequest._accept_bytes.
790
 
        
 
1172
 
791
1173
        This forwards to self._medium._accept_bytes because we are operating
792
1174
        on the mediums stream.
793
1175
        """
796
1178
    def _finished_reading(self):
797
1179
        """See SmartClientMediumRequest._finished_reading.
798
1180
 
799
 
        This clears the _current_request on self._medium to allow a new 
 
1181
        This clears the _current_request on self._medium to allow a new
800
1182
        request to be created.
801
1183
        """
802
1184
        if self._medium._current_request is not self:
803
1185
            raise AssertionError()
804
1186
        self._medium._current_request = None
805
 
        
 
1187
 
806
1188
    def _finished_writing(self):
807
1189
        """See SmartClientMediumRequest._finished_writing.
808
1190
 
809
1191
        This invokes self._medium._flush to ensure all bytes are transmitted.
810
1192
        """
811
1193
        self._medium._flush()
812