~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/smart/medium.py

  • Committer: Andrew Bennetts
  • Date: 2009-10-15 02:53:30 UTC
  • mto: This revision was merged to the branch mainline in revision 4775.
  • Revision ID: andrew.bennetts@canonical.com-20091015025330-9cwu80sdkttu7v58
Add OperationWithCleanups helper, use it to make commit.py simpler and more robust.

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2006,2007 Canonical Ltd
 
1
# Copyright (C) 2006 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
12
12
#
13
13
# You should have received a copy of the GNU General Public License
14
14
# along with this program; if not, write to the Free Software
15
 
# Foundation, Inc., 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
import errno
27
28
import os
28
29
import socket
29
30
import sys
30
 
from bzrlib import errors
31
 
from bzrlib.smart.protocol import SmartServerRequestProtocolOne
32
 
 
33
 
try:
34
 
    from bzrlib.transport import ssh
35
 
except errors.ParamikoNotPresent:
36
 
    # no paramiko.  SmartSSHClientMedium will break.
37
 
    pass
38
 
 
39
 
 
40
 
class SmartServerStreamMedium(object):
 
31
import urllib
 
32
 
 
33
from bzrlib.lazy_import import lazy_import
 
34
lazy_import(globals(), """
 
35
import atexit
 
36
import weakref
 
37
from bzrlib import (
 
38
    debug,
 
39
    errors,
 
40
    symbol_versioning,
 
41
    trace,
 
42
    ui,
 
43
    urlutils,
 
44
    )
 
45
from bzrlib.smart import client, protocol, request, vfs
 
46
from bzrlib.transport import ssh
 
47
""")
 
48
#usually already imported, and getting IllegalScoperReplacer on it here.
 
49
from bzrlib import osutils
 
50
 
 
51
# We must not read any more than 64k at a time so we don't risk "no buffer
 
52
# space available" errors on some platforms.  Windows in particular is likely
 
53
# to give error 10053 or 10055 if we read more than 64k from a socket.
 
54
_MAX_READ_SIZE = 64 * 1024
 
55
 
 
56
 
 
57
def _get_protocol_factory_for_bytes(bytes):
 
58
    """Determine the right protocol factory for 'bytes'.
 
59
 
 
60
    This will return an appropriate protocol factory depending on the version
 
61
    of the protocol being used, as determined by inspecting the given bytes.
 
62
    The bytes should have at least one newline byte (i.e. be a whole line),
 
63
    otherwise it's possible that a request will be incorrectly identified as
 
64
    version 1.
 
65
 
 
66
    Typical use would be::
 
67
 
 
68
         factory, unused_bytes = _get_protocol_factory_for_bytes(bytes)
 
69
         server_protocol = factory(transport, write_func, root_client_path)
 
70
         server_protocol.accept_bytes(unused_bytes)
 
71
 
 
72
    :param bytes: a str of bytes of the start of the request.
 
73
    :returns: 2-tuple of (protocol_factory, unused_bytes).  protocol_factory is
 
74
        a callable that takes three args: transport, write_func,
 
75
        root_client_path.  unused_bytes are any bytes that were not part of a
 
76
        protocol version marker.
 
77
    """
 
78
    if bytes.startswith(protocol.MESSAGE_VERSION_THREE):
 
79
        protocol_factory = protocol.build_server_protocol_three
 
80
        bytes = bytes[len(protocol.MESSAGE_VERSION_THREE):]
 
81
    elif bytes.startswith(protocol.REQUEST_VERSION_TWO):
 
82
        protocol_factory = protocol.SmartServerRequestProtocolTwo
 
83
        bytes = bytes[len(protocol.REQUEST_VERSION_TWO):]
 
84
    else:
 
85
        protocol_factory = protocol.SmartServerRequestProtocolOne
 
86
    return protocol_factory, bytes
 
87
 
 
88
 
 
89
def _get_line(read_bytes_func):
 
90
    """Read bytes using read_bytes_func until a newline byte.
 
91
 
 
92
    This isn't particularly efficient, so should only be used when the
 
93
    expected size of the line is quite short.
 
94
 
 
95
    :returns: a tuple of two strs: (line, excess)
 
96
    """
 
97
    newline_pos = -1
 
98
    bytes = ''
 
99
    while newline_pos == -1:
 
100
        new_bytes = read_bytes_func(1)
 
101
        bytes += new_bytes
 
102
        if new_bytes == '':
 
103
            # Ran out of bytes before receiving a complete line.
 
104
            return bytes, ''
 
105
        newline_pos = bytes.find('\n')
 
106
    line = bytes[:newline_pos+1]
 
107
    excess = bytes[newline_pos+1:]
 
108
    return line, excess
 
109
 
 
110
 
 
111
class SmartMedium(object):
 
112
    """Base class for smart protocol media, both client- and server-side."""
 
113
 
 
114
    def __init__(self):
 
115
        self._push_back_buffer = None
 
116
 
 
117
    def _push_back(self, bytes):
 
118
        """Return unused bytes to the medium, because they belong to the next
 
119
        request(s).
 
120
 
 
121
        This sets the _push_back_buffer to the given bytes.
 
122
        """
 
123
        if self._push_back_buffer is not None:
 
124
            raise AssertionError(
 
125
                "_push_back called when self._push_back_buffer is %r"
 
126
                % (self._push_back_buffer,))
 
127
        if bytes == '':
 
128
            return
 
129
        self._push_back_buffer = bytes
 
130
 
 
131
    def _get_push_back_buffer(self):
 
132
        if self._push_back_buffer == '':
 
133
            raise AssertionError(
 
134
                '%s._push_back_buffer should never be the empty string, '
 
135
                'which can be confused with EOF' % (self,))
 
136
        bytes = self._push_back_buffer
 
137
        self._push_back_buffer = None
 
138
        return bytes
 
139
 
 
140
    def read_bytes(self, desired_count):
 
141
        """Read some bytes from this medium.
 
142
 
 
143
        :returns: some bytes, possibly more or less than the number requested
 
144
            in 'desired_count' depending on the medium.
 
145
        """
 
146
        if self._push_back_buffer is not None:
 
147
            return self._get_push_back_buffer()
 
148
        bytes_to_read = min(desired_count, _MAX_READ_SIZE)
 
149
        return self._read_bytes(bytes_to_read)
 
150
 
 
151
    def _read_bytes(self, count):
 
152
        raise NotImplementedError(self._read_bytes)
 
153
 
 
154
    def _get_line(self):
 
155
        """Read bytes from this request's response until a newline byte.
 
156
 
 
157
        This isn't particularly efficient, so should only be used when the
 
158
        expected size of the line is quite short.
 
159
 
 
160
        :returns: a string of bytes ending in a newline (byte 0x0A).
 
161
        """
 
162
        line, excess = _get_line(self.read_bytes)
 
163
        self._push_back(excess)
 
164
        return line
 
165
 
 
166
    def _report_activity(self, bytes, direction):
 
167
        """Notify that this medium has activity.
 
168
 
 
169
        Implementations should call this from all methods that actually do IO.
 
170
        Be careful that it's not called twice, if one method is implemented on
 
171
        top of another.
 
172
 
 
173
        :param bytes: Number of bytes read or written.
 
174
        :param direction: 'read' or 'write' or None.
 
175
        """
 
176
        ui.ui_factory.report_transport_activity(self, bytes, direction)
 
177
 
 
178
 
 
179
class SmartServerStreamMedium(SmartMedium):
41
180
    """Handles smart commands coming over a stream.
42
181
 
43
182
    The stream may be a pipe connected to sshd, or a tcp socket, or an
46
185
    One instance is created for each connected client; it can serve multiple
47
186
    requests in the lifetime of the connection.
48
187
 
49
 
    The server passes requests through to an underlying backing transport, 
 
188
    The server passes requests through to an underlying backing transport,
50
189
    which will typically be a LocalTransport looking at the server's filesystem.
 
190
 
 
191
    :ivar _push_back_buffer: a str of bytes that have been read from the stream
 
192
        but not used yet, or None if there are no buffered bytes.  Subclasses
 
193
        should make sure to exhaust this buffer before reading more bytes from
 
194
        the stream.  See also the _push_back method.
51
195
    """
52
196
 
53
 
    def __init__(self, backing_transport):
 
197
    def __init__(self, backing_transport, root_client_path='/'):
54
198
        """Construct new server.
55
199
 
56
200
        :param backing_transport: Transport for the directory served.
57
201
        """
58
202
        # backing_transport could be passed to serve instead of __init__
59
203
        self.backing_transport = backing_transport
 
204
        self.root_client_path = root_client_path
60
205
        self.finished = False
 
206
        SmartMedium.__init__(self)
61
207
 
62
208
    def serve(self):
63
209
        """Serve requests until the client disconnects."""
66
212
        from sys import stderr
67
213
        try:
68
214
            while not self.finished:
69
 
                protocol = SmartServerRequestProtocolOne(self.backing_transport,
70
 
                                                         self._write_out)
71
 
                self._serve_one_request(protocol)
 
215
                server_protocol = self._build_protocol()
 
216
                self._serve_one_request(server_protocol)
72
217
        except Exception, e:
73
218
            stderr.write("%s terminating on exception %s\n" % (self, e))
74
219
            raise
75
220
 
 
221
    def _build_protocol(self):
 
222
        """Identifies the version of the incoming request, and returns an
 
223
        a protocol object that can interpret it.
 
224
 
 
225
        If more bytes than the version prefix of the request are read, they will
 
226
        be fed into the protocol before it is returned.
 
227
 
 
228
        :returns: a SmartServerRequestProtocol.
 
229
        """
 
230
        bytes = self._get_line()
 
231
        protocol_factory, unused_bytes = _get_protocol_factory_for_bytes(bytes)
 
232
        protocol = protocol_factory(
 
233
            self.backing_transport, self._write_out, self.root_client_path)
 
234
        protocol.accept_bytes(unused_bytes)
 
235
        return protocol
 
236
 
76
237
    def _serve_one_request(self, protocol):
77
238
        """Read one request from input, process, send back a response.
78
 
        
 
239
 
79
240
        :param protocol: a SmartServerRequestProtocol.
80
241
        """
81
242
        try:
89
250
        """Called when an unhandled exception from the protocol occurs."""
90
251
        raise NotImplementedError(self.terminate_due_to_error)
91
252
 
 
253
    def _read_bytes(self, desired_count):
 
254
        """Get some bytes from the medium.
 
255
 
 
256
        :param desired_count: number of bytes we want to read.
 
257
        """
 
258
        raise NotImplementedError(self._read_bytes)
 
259
 
92
260
 
93
261
class SmartServerSocketStreamMedium(SmartServerStreamMedium):
94
262
 
95
 
    def __init__(self, sock, backing_transport):
 
263
    def __init__(self, sock, backing_transport, root_client_path='/'):
96
264
        """Constructor.
97
265
 
98
266
        :param sock: the socket the server will read from.  It will be put
99
267
            into blocking mode.
100
268
        """
101
 
        SmartServerStreamMedium.__init__(self, backing_transport)
102
 
        self.push_back = ''
 
269
        SmartServerStreamMedium.__init__(
 
270
            self, backing_transport, root_client_path=root_client_path)
103
271
        sock.setblocking(True)
104
272
        self.socket = sock
105
273
 
106
274
    def _serve_one_request_unguarded(self, protocol):
107
275
        while protocol.next_read_size():
108
 
            if self.push_back:
109
 
                protocol.accept_bytes(self.push_back)
110
 
                self.push_back = ''
111
 
            else:
112
 
                bytes = self.socket.recv(4096)
113
 
                if bytes == '':
114
 
                    self.finished = True
115
 
                    return
116
 
                protocol.accept_bytes(bytes)
117
 
        
118
 
        self.push_back = protocol.excess_buffer
119
 
    
 
276
            # We can safely try to read large chunks.  If there is less data
 
277
            # than _MAX_READ_SIZE ready, the socket wil just return a short
 
278
            # read immediately rather than block.
 
279
            bytes = self.read_bytes(_MAX_READ_SIZE)
 
280
            if bytes == '':
 
281
                self.finished = True
 
282
                return
 
283
            protocol.accept_bytes(bytes)
 
284
 
 
285
        self._push_back(protocol.unused_data)
 
286
 
 
287
    def _read_bytes(self, desired_count):
 
288
        return _read_bytes_from_socket(
 
289
            self.socket.recv, desired_count, self._report_activity)
 
290
 
120
291
    def terminate_due_to_error(self):
121
 
        """Called when an unhandled exception from the protocol occurs."""
122
292
        # TODO: This should log to a server log file, but no such thing
123
293
        # exists yet.  Andrew Bennetts 2006-09-29.
124
 
        self.socket.close()
 
294
        osutils.until_no_eintr(self.socket.close)
125
295
        self.finished = True
126
296
 
127
297
    def _write_out(self, bytes):
128
 
        self.socket.sendall(bytes)
 
298
        osutils.send_all(self.socket, bytes, self._report_activity)
129
299
 
130
300
 
131
301
class SmartServerPipeStreamMedium(SmartServerStreamMedium):
150
320
 
151
321
    def _serve_one_request_unguarded(self, protocol):
152
322
        while True:
 
323
            # We need to be careful not to read past the end of the current
 
324
            # request, or else the read from the pipe will block, so we use
 
325
            # protocol.next_read_size().
153
326
            bytes_to_read = protocol.next_read_size()
154
327
            if bytes_to_read == 0:
155
328
                # Finished serving this request.
156
 
                self._out.flush()
 
329
                osutils.until_no_eintr(self._out.flush)
157
330
                return
158
 
            bytes = self._in.read(bytes_to_read)
 
331
            bytes = self.read_bytes(bytes_to_read)
159
332
            if bytes == '':
160
333
                # Connection has been closed.
161
334
                self.finished = True
162
 
                self._out.flush()
 
335
                osutils.until_no_eintr(self._out.flush)
163
336
                return
164
337
            protocol.accept_bytes(bytes)
165
338
 
 
339
    def _read_bytes(self, desired_count):
 
340
        return osutils.until_no_eintr(self._in.read, desired_count)
 
341
 
166
342
    def terminate_due_to_error(self):
167
343
        # TODO: This should log to a server log file, but no such thing
168
344
        # exists yet.  Andrew Bennetts 2006-09-29.
169
 
        self._out.close()
 
345
        osutils.until_no_eintr(self._out.close)
170
346
        self.finished = True
171
347
 
172
348
    def _write_out(self, bytes):
173
 
        self._out.write(bytes)
 
349
        osutils.until_no_eintr(self._out.write, bytes)
174
350
 
175
351
 
176
352
class SmartClientMediumRequest(object):
186
362
    request.finished_reading()
187
363
 
188
364
    It is up to the individual SmartClientMedium whether multiple concurrent
189
 
    requests can exist. See SmartClientMedium.get_request to obtain instances 
190
 
    of SmartClientMediumRequest, and the concrete Medium you are using for 
 
365
    requests can exist. See SmartClientMedium.get_request to obtain instances
 
366
    of SmartClientMediumRequest, and the concrete Medium you are using for
191
367
    details on concurrency and pipelining.
192
368
    """
193
369
 
202
378
    def accept_bytes(self, bytes):
203
379
        """Accept bytes for inclusion in this request.
204
380
 
205
 
        This method may not be be called after finished_writing() has been
 
381
        This method may not be called after finished_writing() has been
206
382
        called.  It depends upon the Medium whether or not the bytes will be
207
383
        immediately transmitted. Message based Mediums will tend to buffer the
208
384
        bytes until finished_writing() is called.
239
415
    def _finished_reading(self):
240
416
        """Helper for finished_reading.
241
417
 
242
 
        finished_reading checks the state of the request to determine if 
 
418
        finished_reading checks the state of the request to determine if
243
419
        finished_reading is allowed, and if it is hands off to _finished_reading
244
420
        to perform the action.
245
421
        """
259
435
    def _finished_writing(self):
260
436
        """Helper for finished_writing.
261
437
 
262
 
        finished_writing checks the state of the request to determine if 
 
438
        finished_writing checks the state of the request to determine if
263
439
        finished_writing is allowed, and if it is hands off to _finished_writing
264
440
        to perform the action.
265
441
        """
270
446
 
271
447
        This method will block and wait for count bytes to be read. It may not
272
448
        be invoked until finished_writing() has been called - this is to ensure
273
 
        a message-based approach to requests, for compatability with message
 
449
        a message-based approach to requests, for compatibility with message
274
450
        based mediums like HTTP.
275
451
        """
276
452
        if self._state == "writing":
280
456
        return self._read_bytes(count)
281
457
 
282
458
    def _read_bytes(self, count):
283
 
        """Helper for read_bytes.
 
459
        """Helper for SmartClientMediumRequest.read_bytes.
284
460
 
285
461
        read_bytes checks the state of the request to determing if bytes
286
462
        should be read. After that it hands off to _read_bytes to do the
287
463
        actual read.
288
 
        """
289
 
        raise NotImplementedError(self._read_bytes)
290
 
 
291
 
 
292
 
class SmartClientMedium(object):
 
464
 
 
465
        By default this forwards to self._medium.read_bytes because we are
 
466
        operating on the medium's stream.
 
467
        """
 
468
        return self._medium.read_bytes(count)
 
469
 
 
470
    def read_line(self):
 
471
        line = self._read_line()
 
472
        if not line.endswith('\n'):
 
473
            # end of file encountered reading from server
 
474
            raise errors.ConnectionReset(
 
475
                "Unexpected end of message. Please check connectivity "
 
476
                "and permissions, and report a bug if problems persist.")
 
477
        return line
 
478
 
 
479
    def _read_line(self):
 
480
        """Helper for SmartClientMediumRequest.read_line.
 
481
 
 
482
        By default this forwards to self._medium._get_line because we are
 
483
        operating on the medium's stream.
 
484
        """
 
485
        return self._medium._get_line()
 
486
 
 
487
 
 
488
class _DebugCounter(object):
 
489
    """An object that counts the HPSS calls made to each client medium.
 
490
 
 
491
    When a medium is garbage-collected, or failing that when atexit functions
 
492
    are run, the total number of calls made on that medium are reported via
 
493
    trace.note.
 
494
    """
 
495
 
 
496
    def __init__(self):
 
497
        self.counts = weakref.WeakKeyDictionary()
 
498
        client._SmartClient.hooks.install_named_hook(
 
499
            'call', self.increment_call_count, 'hpss call counter')
 
500
        atexit.register(self.flush_all)
 
501
 
 
502
    def track(self, medium):
 
503
        """Start tracking calls made to a medium.
 
504
 
 
505
        This only keeps a weakref to the medium, so shouldn't affect the
 
506
        medium's lifetime.
 
507
        """
 
508
        medium_repr = repr(medium)
 
509
        # Add this medium to the WeakKeyDictionary
 
510
        self.counts[medium] = dict(count=0, vfs_count=0,
 
511
                                   medium_repr=medium_repr)
 
512
        # Weakref callbacks are fired in reverse order of their association
 
513
        # with the referenced object.  So we add a weakref *after* adding to
 
514
        # the WeakKeyDict so that we can report the value from it before the
 
515
        # entry is removed by the WeakKeyDict's own callback.
 
516
        ref = weakref.ref(medium, self.done)
 
517
 
 
518
    def increment_call_count(self, params):
 
519
        # Increment the count in the WeakKeyDictionary
 
520
        value = self.counts[params.medium]
 
521
        value['count'] += 1
 
522
        try:
 
523
            request_method = request.request_handlers.get(params.method)
 
524
        except KeyError:
 
525
            # A method we don't know about doesn't count as a VFS method.
 
526
            return
 
527
        if issubclass(request_method, vfs.VfsRequest):
 
528
            value['vfs_count'] += 1
 
529
 
 
530
    def done(self, ref):
 
531
        value = self.counts[ref]
 
532
        count, vfs_count, medium_repr = (
 
533
            value['count'], value['vfs_count'], value['medium_repr'])
 
534
        # In case this callback is invoked for the same ref twice (by the
 
535
        # weakref callback and by the atexit function), set the call count back
 
536
        # to 0 so this item won't be reported twice.
 
537
        value['count'] = 0
 
538
        value['vfs_count'] = 0
 
539
        if count != 0:
 
540
            trace.note('HPSS calls: %d (%d vfs) %s',
 
541
                       count, vfs_count, medium_repr)
 
542
 
 
543
    def flush_all(self):
 
544
        for ref in list(self.counts.keys()):
 
545
            self.done(ref)
 
546
 
 
547
_debug_counter = None
 
548
 
 
549
 
 
550
class SmartClientMedium(SmartMedium):
293
551
    """Smart client is a medium for sending smart protocol requests over."""
294
552
 
 
553
    def __init__(self, base):
 
554
        super(SmartClientMedium, self).__init__()
 
555
        self.base = base
 
556
        self._protocol_version_error = None
 
557
        self._protocol_version = None
 
558
        self._done_hello = False
 
559
        # Be optimistic: we assume the remote end can accept new remote
 
560
        # requests until we get an error saying otherwise.
 
561
        # _remote_version_is_before tracks the bzr version the remote side
 
562
        # can be based on what we've seen so far.
 
563
        self._remote_version_is_before = None
 
564
        # Install debug hook function if debug flag is set.
 
565
        if 'hpss' in debug.debug_flags:
 
566
            global _debug_counter
 
567
            if _debug_counter is None:
 
568
                _debug_counter = _DebugCounter()
 
569
            _debug_counter.track(self)
 
570
 
 
571
    def _is_remote_before(self, version_tuple):
 
572
        """Is it possible the remote side supports RPCs for a given version?
 
573
 
 
574
        Typical use::
 
575
 
 
576
            needed_version = (1, 2)
 
577
            if medium._is_remote_before(needed_version):
 
578
                fallback_to_pre_1_2_rpc()
 
579
            else:
 
580
                try:
 
581
                    do_1_2_rpc()
 
582
                except UnknownSmartMethod:
 
583
                    medium._remember_remote_is_before(needed_version)
 
584
                    fallback_to_pre_1_2_rpc()
 
585
 
 
586
        :seealso: _remember_remote_is_before
 
587
        """
 
588
        if self._remote_version_is_before is None:
 
589
            # So far, the remote side seems to support everything
 
590
            return False
 
591
        return version_tuple >= self._remote_version_is_before
 
592
 
 
593
    def _remember_remote_is_before(self, version_tuple):
 
594
        """Tell this medium that the remote side is older the given version.
 
595
 
 
596
        :seealso: _is_remote_before
 
597
        """
 
598
        if (self._remote_version_is_before is not None and
 
599
            version_tuple > self._remote_version_is_before):
 
600
            # We have been told that the remote side is older than some version
 
601
            # which is newer than a previously supplied older-than version.
 
602
            # This indicates that some smart verb call is not guarded
 
603
            # appropriately (it should simply not have been tried).
 
604
            raise AssertionError(
 
605
                "_remember_remote_is_before(%r) called, but "
 
606
                "_remember_remote_is_before(%r) was called previously."
 
607
                % (version_tuple, self._remote_version_is_before))
 
608
        self._remote_version_is_before = version_tuple
 
609
 
 
610
    def protocol_version(self):
 
611
        """Find out if 'hello' smart request works."""
 
612
        if self._protocol_version_error is not None:
 
613
            raise self._protocol_version_error
 
614
        if not self._done_hello:
 
615
            try:
 
616
                medium_request = self.get_request()
 
617
                # Send a 'hello' request in protocol version one, for maximum
 
618
                # backwards compatibility.
 
619
                client_protocol = protocol.SmartClientRequestProtocolOne(medium_request)
 
620
                client_protocol.query_version()
 
621
                self._done_hello = True
 
622
            except errors.SmartProtocolError, e:
 
623
                # Cache the error, just like we would cache a successful
 
624
                # result.
 
625
                self._protocol_version_error = e
 
626
                raise
 
627
        return '2'
 
628
 
 
629
    def should_probe(self):
 
630
        """Should RemoteBzrDirFormat.probe_transport send a smart request on
 
631
        this medium?
 
632
 
 
633
        Some transports are unambiguously smart-only; there's no need to check
 
634
        if the transport is able to carry smart requests, because that's all
 
635
        it is for.  In those cases, this method should return False.
 
636
 
 
637
        But some HTTP transports can sometimes fail to carry smart requests,
 
638
        but still be usuable for accessing remote bzrdirs via plain file
 
639
        accesses.  So for those transports, their media should return True here
 
640
        so that RemoteBzrDirFormat can determine if it is appropriate for that
 
641
        transport.
 
642
        """
 
643
        return False
 
644
 
295
645
    def disconnect(self):
296
646
        """If this medium maintains a persistent connection, close it.
297
 
        
 
647
 
298
648
        The default implementation does nothing.
299
649
        """
300
 
        
 
650
 
 
651
    def remote_path_from_transport(self, transport):
 
652
        """Convert transport into a path suitable for using in a request.
 
653
 
 
654
        Note that the resulting remote path doesn't encode the host name or
 
655
        anything but path, so it is only safe to use it in requests sent over
 
656
        the medium from the matching transport.
 
657
        """
 
658
        medium_base = urlutils.join(self.base, '/')
 
659
        rel_url = urlutils.relative_url(medium_base, transport.base)
 
660
        return urllib.unquote(rel_url)
 
661
 
301
662
 
302
663
class SmartClientStreamMedium(SmartClientMedium):
303
664
    """Stream based medium common class.
308
669
    receive bytes.
309
670
    """
310
671
 
311
 
    def __init__(self):
 
672
    def __init__(self, base):
 
673
        SmartClientMedium.__init__(self, base)
312
674
        self._current_request = None
313
675
 
314
676
    def accept_bytes(self, bytes):
322
684
 
323
685
    def _flush(self):
324
686
        """Flush the output stream.
325
 
        
 
687
 
326
688
        This method is used by the SmartClientStreamMediumRequest to ensure that
327
689
        all data for a request is sent, to avoid long timeouts or deadlocks.
328
690
        """
336
698
        """
337
699
        return SmartClientStreamMediumRequest(self)
338
700
 
339
 
    def read_bytes(self, count):
340
 
        return self._read_bytes(count)
341
 
 
342
701
 
343
702
class SmartSimplePipesClientMedium(SmartClientStreamMedium):
344
703
    """A client medium using simple pipes.
345
 
    
 
704
 
346
705
    This client does not manage the pipes: it assumes they will always be open.
347
706
    """
348
707
 
349
 
    def __init__(self, readable_pipe, writeable_pipe):
350
 
        SmartClientStreamMedium.__init__(self)
 
708
    def __init__(self, readable_pipe, writeable_pipe, base):
 
709
        SmartClientStreamMedium.__init__(self, base)
351
710
        self._readable_pipe = readable_pipe
352
711
        self._writeable_pipe = writeable_pipe
353
712
 
354
713
    def _accept_bytes(self, bytes):
355
714
        """See SmartClientStreamMedium.accept_bytes."""
356
 
        self._writeable_pipe.write(bytes)
 
715
        osutils.until_no_eintr(self._writeable_pipe.write, bytes)
 
716
        self._report_activity(len(bytes), 'write')
357
717
 
358
718
    def _flush(self):
359
719
        """See SmartClientStreamMedium._flush()."""
360
 
        self._writeable_pipe.flush()
 
720
        osutils.until_no_eintr(self._writeable_pipe.flush)
361
721
 
362
722
    def _read_bytes(self, count):
363
723
        """See SmartClientStreamMedium._read_bytes."""
364
 
        return self._readable_pipe.read(count)
 
724
        bytes = osutils.until_no_eintr(self._readable_pipe.read, count)
 
725
        self._report_activity(len(bytes), 'read')
 
726
        return bytes
365
727
 
366
728
 
367
729
class SmartSSHClientMedium(SmartClientStreamMedium):
368
730
    """A client medium using SSH."""
369
 
    
 
731
 
370
732
    def __init__(self, host, port=None, username=None, password=None,
371
 
            vendor=None):
 
733
            base=None, vendor=None, bzr_remote_path=None):
372
734
        """Creates a client that will connect on the first use.
373
 
        
 
735
 
374
736
        :param vendor: An optional override for the ssh vendor to use. See
375
737
            bzrlib.transport.ssh for details on ssh vendors.
376
738
        """
377
 
        SmartClientStreamMedium.__init__(self)
378
739
        self._connected = False
379
740
        self._host = host
380
741
        self._password = password
381
742
        self._port = port
382
743
        self._username = username
 
744
        # SmartClientStreamMedium stores the repr of this object in its
 
745
        # _DebugCounter so we have to store all the values used in our repr
 
746
        # method before calling the super init.
 
747
        SmartClientStreamMedium.__init__(self, base)
383
748
        self._read_from = None
384
749
        self._ssh_connection = None
385
750
        self._vendor = vendor
386
751
        self._write_to = None
 
752
        self._bzr_remote_path = bzr_remote_path
 
753
        # for the benefit of progress making a short description of this
 
754
        # transport
 
755
        self._scheme = 'bzr+ssh'
 
756
 
 
757
    def __repr__(self):
 
758
        return "%s(connected=%r, username=%r, host=%r, port=%r)" % (
 
759
            self.__class__.__name__,
 
760
            self._connected,
 
761
            self._username,
 
762
            self._host,
 
763
            self._port)
387
764
 
388
765
    def _accept_bytes(self, bytes):
389
766
        """See SmartClientStreamMedium.accept_bytes."""
390
767
        self._ensure_connection()
391
 
        self._write_to.write(bytes)
 
768
        osutils.until_no_eintr(self._write_to.write, bytes)
 
769
        self._report_activity(len(bytes), 'write')
392
770
 
393
771
    def disconnect(self):
394
772
        """See SmartClientMedium.disconnect()."""
395
773
        if not self._connected:
396
774
            return
397
 
        self._read_from.close()
398
 
        self._write_to.close()
 
775
        osutils.until_no_eintr(self._read_from.close)
 
776
        osutils.until_no_eintr(self._write_to.close)
399
777
        self._ssh_connection.close()
400
778
        self._connected = False
401
779
 
403
781
        """Connect this medium if not already connected."""
404
782
        if self._connected:
405
783
            return
406
 
        executable = os.environ.get('BZR_REMOTE_PATH', 'bzr')
407
784
        if self._vendor is None:
408
785
            vendor = ssh._get_ssh_vendor()
409
786
        else:
410
787
            vendor = self._vendor
411
788
        self._ssh_connection = vendor.connect_ssh(self._username,
412
789
                self._password, self._host, self._port,
413
 
                command=[executable, 'serve', '--inet', '--directory=/',
414
 
                         '--allow-writes'])
 
790
                command=[self._bzr_remote_path, 'serve', '--inet',
 
791
                         '--directory=/', '--allow-writes'])
415
792
        self._read_from, self._write_to = \
416
793
            self._ssh_connection.get_filelike_channels()
417
794
        self._connected = True
424
801
        """See SmartClientStreamMedium.read_bytes."""
425
802
        if not self._connected:
426
803
            raise errors.MediumNotConnected(self)
427
 
        return self._read_from.read(count)
 
804
        bytes_to_read = min(count, _MAX_READ_SIZE)
 
805
        bytes = osutils.until_no_eintr(self._read_from.read, bytes_to_read)
 
806
        self._report_activity(len(bytes), 'read')
 
807
        return bytes
 
808
 
 
809
 
 
810
# Port 4155 is the default port for bzr://, registered with IANA.
 
811
BZR_DEFAULT_INTERFACE = None
 
812
BZR_DEFAULT_PORT = 4155
428
813
 
429
814
 
430
815
class SmartTCPClientMedium(SmartClientStreamMedium):
431
816
    """A client medium using TCP."""
432
 
    
433
 
    def __init__(self, host, port):
 
817
 
 
818
    def __init__(self, host, port, base):
434
819
        """Creates a client that will connect on the first use."""
435
 
        SmartClientStreamMedium.__init__(self)
 
820
        SmartClientStreamMedium.__init__(self, base)
436
821
        self._connected = False
437
822
        self._host = host
438
823
        self._port = port
441
826
    def _accept_bytes(self, bytes):
442
827
        """See SmartClientMedium.accept_bytes."""
443
828
        self._ensure_connection()
444
 
        self._socket.sendall(bytes)
 
829
        osutils.send_all(self._socket, bytes, self._report_activity)
445
830
 
446
831
    def disconnect(self):
447
832
        """See SmartClientMedium.disconnect()."""
448
833
        if not self._connected:
449
834
            return
450
 
        self._socket.close()
 
835
        osutils.until_no_eintr(self._socket.close)
451
836
        self._socket = None
452
837
        self._connected = False
453
838
 
455
840
        """Connect this medium if not already connected."""
456
841
        if self._connected:
457
842
            return
458
 
        self._socket = socket.socket()
459
 
        self._socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
460
 
        result = self._socket.connect_ex((self._host, int(self._port)))
461
 
        if result:
 
843
        if self._port is None:
 
844
            port = BZR_DEFAULT_PORT
 
845
        else:
 
846
            port = int(self._port)
 
847
        try:
 
848
            sockaddrs = socket.getaddrinfo(self._host, port, socket.AF_UNSPEC,
 
849
                socket.SOCK_STREAM, 0, 0)
 
850
        except socket.gaierror, (err_num, err_msg):
 
851
            raise errors.ConnectionError("failed to lookup %s:%d: %s" %
 
852
                    (self._host, port, err_msg))
 
853
        # Initialize err in case there are no addresses returned:
 
854
        err = socket.error("no address found for %s" % self._host)
 
855
        for (family, socktype, proto, canonname, sockaddr) in sockaddrs:
 
856
            try:
 
857
                self._socket = socket.socket(family, socktype, proto)
 
858
                self._socket.setsockopt(socket.IPPROTO_TCP,
 
859
                                        socket.TCP_NODELAY, 1)
 
860
                self._socket.connect(sockaddr)
 
861
            except socket.error, err:
 
862
                if self._socket is not None:
 
863
                    self._socket.close()
 
864
                self._socket = None
 
865
                continue
 
866
            break
 
867
        if self._socket is None:
 
868
            # socket errors either have a (string) or (errno, string) as their
 
869
            # args.
 
870
            if type(err.args) is str:
 
871
                err_msg = err.args
 
872
            else:
 
873
                err_msg = err.args[1]
462
874
            raise errors.ConnectionError("failed to connect to %s:%d: %s" %
463
 
                    (self._host, self._port, os.strerror(result)))
 
875
                    (self._host, port, err_msg))
464
876
        self._connected = True
465
877
 
466
878
    def _flush(self):
467
879
        """See SmartClientStreamMedium._flush().
468
 
        
469
 
        For TCP we do no flushing. We may want to turn off TCP_NODELAY and 
 
880
 
 
881
        For TCP we do no flushing. We may want to turn off TCP_NODELAY and
470
882
        add a means to do a flush, but that can be done in the future.
471
883
        """
472
884
 
474
886
        """See SmartClientMedium.read_bytes."""
475
887
        if not self._connected:
476
888
            raise errors.MediumNotConnected(self)
477
 
        return self._socket.recv(count)
 
889
        return _read_bytes_from_socket(
 
890
            self._socket.recv, count, self._report_activity)
478
891
 
479
892
 
480
893
class SmartClientStreamMediumRequest(SmartClientMediumRequest):
493
906
 
494
907
    def _accept_bytes(self, bytes):
495
908
        """See SmartClientMediumRequest._accept_bytes.
496
 
        
 
909
 
497
910
        This forwards to self._medium._accept_bytes because we are operating
498
911
        on the mediums stream.
499
912
        """
502
915
    def _finished_reading(self):
503
916
        """See SmartClientMediumRequest._finished_reading.
504
917
 
505
 
        This clears the _current_request on self._medium to allow a new 
 
918
        This clears the _current_request on self._medium to allow a new
506
919
        request to be created.
507
920
        """
508
 
        assert self._medium._current_request is self
 
921
        if self._medium._current_request is not self:
 
922
            raise AssertionError()
509
923
        self._medium._current_request = None
510
 
        
 
924
 
511
925
    def _finished_writing(self):
512
926
        """See SmartClientMediumRequest._finished_writing.
513
927
 
515
929
        """
516
930
        self._medium._flush()
517
931
 
518
 
    def _read_bytes(self, count):
519
 
        """See SmartClientMediumRequest._read_bytes.
520
 
        
521
 
        This forwards to self._medium._read_bytes because we are operating
522
 
        on the mediums stream.
523
 
        """
524
 
        return self._medium._read_bytes(count)
525
932
 
 
933
def _read_bytes_from_socket(sock, desired_count, report_activity):
 
934
    # We ignore the desired_count because on sockets it's more efficient to
 
935
    # read large chunks (of _MAX_READ_SIZE bytes) at a time.
 
936
    try:
 
937
        bytes = osutils.until_no_eintr(sock, _MAX_READ_SIZE)
 
938
    except socket.error, e:
 
939
        if len(e.args) and e.args[0] in (errno.ECONNRESET, 10054):
 
940
            # The connection was closed by the other side.  Callers expect an
 
941
            # empty string to signal end-of-stream.
 
942
            bytes = ''
 
943
        else:
 
944
            raise
 
945
    else:
 
946
        report_activity(len(bytes), 'read')
 
947
    return bytes
526
948