← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/smart/medium.py

  • Committer: Jelmer Vernooij
  • Date: 2012-01-27 19:05:43 UTC
  • mto: This revision was merged to the branch mainline in revision 6450.
  • Revision ID: jelmer@samba.org-20120127190543-vk350mv4a0c7aug2
Fix weave test.

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
 
31
import sys
 
32
import time
 
33
 
 
34
import bzrlib
 
35
from bzrlib.lazy_import import lazy_import
 
36
lazy_import(globals(), """
 
37
import select
28
38
import socket
29
 
import sys
30
 
import urllib
 
39
import thread
 
40
import weakref
31
41
 
32
42
from bzrlib import (
 
43
    debug,
33
44
    errors,
34
 
    osutils,
35
 
    symbol_versioning,
 
45
    trace,
 
46
    transport,
 
47
    ui,
36
48
    urlutils,
37
49
    )
38
 
from bzrlib.smart.protocol import (
39
 
    MESSAGE_VERSION_THREE,
40
 
    REQUEST_VERSION_TWO,
41
 
    SmartClientRequestProtocolOne,
42
 
    SmartServerRequestProtocolOne,
43
 
    SmartServerRequestProtocolTwo,
44
 
    build_server_protocol_three
45
 
    )
 
50
from bzrlib.i18n import gettext
 
51
from bzrlib.smart import client, protocol, request, signals, vfs
46
52
from bzrlib.transport import ssh
 
53
""")
 
54
from bzrlib import osutils
47
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
48
61
 
49
62
def _get_protocol_factory_for_bytes(bytes):
50
63
    """Determine the right protocol factory for 'bytes'.
67
80
        root_client_path.  unused_bytes are any bytes that were not part of a
68
81
        protocol version marker.
69
82
    """
70
 
    if bytes.startswith(MESSAGE_VERSION_THREE):
71
 
        protocol_factory = build_server_protocol_three
72
 
        bytes = bytes[len(MESSAGE_VERSION_THREE):]
73
 
    elif bytes.startswith(REQUEST_VERSION_TWO):
74
 
        protocol_factory = SmartServerRequestProtocolTwo
75
 
        bytes = bytes[len(REQUEST_VERSION_TWO):]
 
83
    if bytes.startswith(protocol.MESSAGE_VERSION_THREE):
 
84
        protocol_factory = protocol.build_server_protocol_three
 
85
        bytes = bytes[len(protocol.MESSAGE_VERSION_THREE):]
 
86
    elif bytes.startswith(protocol.REQUEST_VERSION_TWO):
 
87
        protocol_factory = protocol.SmartServerRequestProtocolTwo
 
88
        bytes = bytes[len(protocol.REQUEST_VERSION_TWO):]
76
89
    else:
77
 
        protocol_factory = SmartServerRequestProtocolOne
 
90
        protocol_factory = protocol.SmartServerRequestProtocolOne
78
91
    return protocol_factory, bytes
79
92
 
80
93
 
81
 
class SmartServerStreamMedium(object):
 
94
def _get_line(read_bytes_func):
 
95
    """Read bytes using read_bytes_func until a newline byte.
 
96
 
 
97
    This isn't particularly efficient, so should only be used when the
 
98
    expected size of the line is quite short.
 
99
 
 
100
    :returns: a tuple of two strs: (line, excess)
 
101
    """
 
102
    newline_pos = -1
 
103
    bytes = ''
 
104
    while newline_pos == -1:
 
105
        new_bytes = read_bytes_func(1)
 
106
        bytes += new_bytes
 
107
        if new_bytes == '':
 
108
            # Ran out of bytes before receiving a complete line.
 
109
            return bytes, ''
 
110
        newline_pos = bytes.find('\n')
 
111
    line = bytes[:newline_pos+1]
 
112
    excess = bytes[newline_pos+1:]
 
113
    return line, excess
 
114
 
 
115
 
 
116
class SmartMedium(object):
 
117
    """Base class for smart protocol media, both client- and server-side."""
 
118
 
 
119
    def __init__(self):
 
120
        self._push_back_buffer = None
 
121
 
 
122
    def _push_back(self, bytes):
 
123
        """Return unused bytes to the medium, because they belong to the next
 
124
        request(s).
 
125
 
 
126
        This sets the _push_back_buffer to the given bytes.
 
127
        """
 
128
        if self._push_back_buffer is not None:
 
129
            raise AssertionError(
 
130
                "_push_back called when self._push_back_buffer is %r"
 
131
                % (self._push_back_buffer,))
 
132
        if bytes == '':
 
133
            return
 
134
        self._push_back_buffer = bytes
 
135
 
 
136
    def _get_push_back_buffer(self):
 
137
        if self._push_back_buffer == '':
 
138
            raise AssertionError(
 
139
                '%s._push_back_buffer should never be the empty string, '
 
140
                'which can be confused with EOF' % (self,))
 
141
        bytes = self._push_back_buffer
 
142
        self._push_back_buffer = None
 
143
        return bytes
 
144
 
 
145
    def read_bytes(self, desired_count):
 
146
        """Read some bytes from this medium.
 
147
 
 
148
        :returns: some bytes, possibly more or less than the number requested
 
149
            in 'desired_count' depending on the medium.
 
150
        """
 
151
        if self._push_back_buffer is not None:
 
152
            return self._get_push_back_buffer()
 
153
        bytes_to_read = min(desired_count, _MAX_READ_SIZE)
 
154
        return self._read_bytes(bytes_to_read)
 
155
 
 
156
    def _read_bytes(self, count):
 
157
        raise NotImplementedError(self._read_bytes)
 
158
 
 
159
    def _get_line(self):
 
160
        """Read bytes from this request's response until a newline byte.
 
161
 
 
162
        This isn't particularly efficient, so should only be used when the
 
163
        expected size of the line is quite short.
 
164
 
 
165
        :returns: a string of bytes ending in a newline (byte 0x0A).
 
166
        """
 
167
        line, excess = _get_line(self.read_bytes)
 
168
        self._push_back(excess)
 
169
        return line
 
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
 
 
191
 
 
192
class SmartServerStreamMedium(SmartMedium):
82
193
    """Handles smart commands coming over a stream.
83
194
 
84
195
    The stream may be a pipe connected to sshd, or a tcp socket, or an
87
198
    One instance is created for each connected client; it can serve multiple
88
199
    requests in the lifetime of the connection.
89
200
 
90
 
    The server passes requests through to an underlying backing transport, 
 
201
    The server passes requests through to an underlying backing transport,
91
202
    which will typically be a LocalTransport looking at the server's filesystem.
92
203
 
93
204
    :ivar _push_back_buffer: a str of bytes that have been read from the stream
96
207
        the stream.  See also the _push_back method.
97
208
    """
98
209
 
99
 
    def __init__(self, backing_transport, root_client_path='/'):
 
210
    _timer = time.time
 
211
 
 
212
    def __init__(self, backing_transport, root_client_path='/', timeout=None):
100
213
        """Construct new server.
101
214
 
102
215
        :param backing_transport: Transport for the directory served.
105
218
        self.backing_transport = backing_transport
106
219
        self.root_client_path = root_client_path
107
220
        self.finished = False
108
 
        self._push_back_buffer = None
109
 
 
110
 
    def _push_back(self, bytes):
111
 
        """Return unused bytes to the medium, because they belong to the next
112
 
        request(s).
113
 
 
114
 
        This sets the _push_back_buffer to the given bytes.
115
 
        """
116
 
        if self._push_back_buffer is not None:
117
 
            raise AssertionError(
118
 
                "_push_back called when self._push_back_buffer is %r"
119
 
                % (self._push_back_buffer,))
120
 
        if bytes == '':
121
 
            return
122
 
        self._push_back_buffer = bytes
123
 
 
124
 
    def _get_push_back_buffer(self):
125
 
        if self._push_back_buffer == '':
126
 
            raise AssertionError(
127
 
                '%s._push_back_buffer should never be the empty string, '
128
 
                'which can be confused with EOF' % (self,))
129
 
        bytes = self._push_back_buffer
130
 
        self._push_back_buffer = None
131
 
        return bytes
 
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)
 
225
        SmartMedium.__init__(self)
132
226
 
133
227
    def serve(self):
134
228
        """Serve requests until the client disconnects."""
139
233
            while not self.finished:
140
234
                server_protocol = self._build_protocol()
141
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
142
242
        except Exception, e:
143
243
            stderr.write("%s terminating on exception %s\n" % (self, e))
144
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)
145
269
 
146
270
    def _build_protocol(self):
147
271
        """Identifies the version of the incoming request, and returns an
152
276
 
153
277
        :returns: a SmartServerRequestProtocol.
154
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
155
283
        bytes = self._get_line()
156
284
        protocol_factory, unused_bytes = _get_protocol_factory_for_bytes(bytes)
157
285
        protocol = protocol_factory(
159
287
        protocol.accept_bytes(unused_bytes)
160
288
        return protocol
161
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
 
162
321
    def _serve_one_request(self, protocol):
163
322
        """Read one request from input, process, send back a response.
164
 
        
 
323
 
165
324
        :param protocol: a SmartServerRequestProtocol.
166
325
        """
 
326
        if protocol is None:
 
327
            return
167
328
        try:
168
329
            self._serve_one_request_unguarded(protocol)
169
330
        except KeyboardInterrupt:
175
336
        """Called when an unhandled exception from the protocol occurs."""
176
337
        raise NotImplementedError(self.terminate_due_to_error)
177
338
 
178
 
    def _get_bytes(self, desired_count):
 
339
    def _read_bytes(self, desired_count):
179
340
        """Get some bytes from the medium.
180
341
 
181
342
        :param desired_count: number of bytes we want to read.
182
343
        """
183
 
        raise NotImplementedError(self._get_bytes)
184
 
 
185
 
    def _get_line(self):
186
 
        """Read bytes from this request's response until a newline byte.
187
 
        
188
 
        This isn't particularly efficient, so should only be used when the
189
 
        expected size of the line is quite short.
190
 
 
191
 
        :returns: a string of bytes ending in a newline (byte 0x0A).
192
 
        """
193
 
        newline_pos = -1
194
 
        bytes = ''
195
 
        while newline_pos == -1:
196
 
            new_bytes = self._get_bytes(1)
197
 
            bytes += new_bytes
198
 
            if new_bytes == '':
199
 
                # Ran out of bytes before receiving a complete line.
200
 
                return bytes
201
 
            newline_pos = bytes.find('\n')
202
 
        line = bytes[:newline_pos+1]
203
 
        self._push_back(bytes[newline_pos+1:])
204
 
        return line
205
 
 
 
344
        raise NotImplementedError(self._read_bytes)
 
345
 
206
346
 
207
347
class SmartServerSocketStreamMedium(SmartServerStreamMedium):
208
348
 
209
 
    def __init__(self, sock, backing_transport, root_client_path='/'):
 
349
    def __init__(self, sock, backing_transport, root_client_path='/',
 
350
                 timeout=None):
210
351
        """Constructor.
211
352
 
212
353
        :param sock: the socket the server will read from.  It will be put
213
354
            into blocking mode.
214
355
        """
215
356
        SmartServerStreamMedium.__init__(
216
 
            self, backing_transport, root_client_path=root_client_path)
 
357
            self, backing_transport, root_client_path=root_client_path,
 
358
            timeout=timeout)
217
359
        sock.setblocking(True)
218
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)
219
373
 
220
374
    def _serve_one_request_unguarded(self, protocol):
221
375
        while protocol.next_read_size():
222
 
            bytes = self._get_bytes(4096)
 
376
            # We can safely try to read large chunks.  If there is less data
 
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)
223
380
            if bytes == '':
224
381
                self.finished = True
225
382
                return
226
383
            protocol.accept_bytes(bytes)
227
 
        
 
384
 
228
385
        self._push_back(protocol.unused_data)
229
386
 
230
 
    def _get_bytes(self, desired_count):
231
 
        if self._push_back_buffer is not None:
232
 
            return self._get_push_back_buffer()
233
 
        # We ignore the desired_count because on sockets it's more efficient to
234
 
        # read 4k at a time.
235
 
        return self.socket.recv(4096)
236
 
    
 
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
 
 
403
    def _read_bytes(self, desired_count):
 
404
        return osutils.read_bytes_from_socket(
 
405
            self.socket, self._report_activity)
 
406
 
237
407
    def terminate_due_to_error(self):
238
408
        # TODO: This should log to a server log file, but no such thing
239
409
        # exists yet.  Andrew Bennetts 2006-09-29.
241
411
        self.finished = True
242
412
 
243
413
    def _write_out(self, bytes):
244
 
        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))
245
421
 
246
422
 
247
423
class SmartServerPipeStreamMedium(SmartServerStreamMedium):
248
424
 
249
 
    def __init__(self, in_file, out_file, backing_transport):
 
425
    def __init__(self, in_file, out_file, backing_transport, timeout=None):
250
426
        """Construct new server.
251
427
 
252
428
        :param in_file: Python file from which requests can be read.
253
429
        :param out_file: Python file to write responses.
254
430
        :param backing_transport: Transport for the directory served.
255
431
        """
256
 
        SmartServerStreamMedium.__init__(self, backing_transport)
 
432
        SmartServerStreamMedium.__init__(self, backing_transport,
 
433
            timeout=timeout)
257
434
        if sys.platform == 'win32':
258
435
            # force binary mode for files
259
436
            import msvcrt
264
441
        self._in = in_file
265
442
        self._out = out_file
266
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
 
267
455
    def _serve_one_request_unguarded(self, protocol):
268
456
        while True:
 
457
            # We need to be careful not to read past the end of the current
 
458
            # request, or else the read from the pipe will block, so we use
 
459
            # protocol.next_read_size().
269
460
            bytes_to_read = protocol.next_read_size()
270
461
            if bytes_to_read == 0:
271
462
                # Finished serving this request.
272
463
                self._out.flush()
273
464
                return
274
 
            bytes = self._get_bytes(bytes_to_read)
 
465
            bytes = self.read_bytes(bytes_to_read)
275
466
            if bytes == '':
276
467
                # Connection has been closed.
277
468
                self.finished = True
279
470
                return
280
471
            protocol.accept_bytes(bytes)
281
472
 
282
 
    def _get_bytes(self, desired_count):
283
 
        if self._push_back_buffer is not None:
284
 
            return self._get_push_back_buffer()
 
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
 
 
494
    def _read_bytes(self, desired_count):
285
495
        return self._in.read(desired_count)
286
496
 
287
497
    def terminate_due_to_error(self):
307
517
    request.finished_reading()
308
518
 
309
519
    It is up to the individual SmartClientMedium whether multiple concurrent
310
 
    requests can exist. See SmartClientMedium.get_request to obtain instances 
311
 
    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
312
522
    details on concurrency and pipelining.
313
523
    """
314
524
 
323
533
    def accept_bytes(self, bytes):
324
534
        """Accept bytes for inclusion in this request.
325
535
 
326
 
        This method may not be be called after finished_writing() has been
 
536
        This method may not be called after finished_writing() has been
327
537
        called.  It depends upon the Medium whether or not the bytes will be
328
538
        immediately transmitted. Message based Mediums will tend to buffer the
329
539
        bytes until finished_writing() is called.
360
570
    def _finished_reading(self):
361
571
        """Helper for finished_reading.
362
572
 
363
 
        finished_reading checks the state of the request to determine if 
 
573
        finished_reading checks the state of the request to determine if
364
574
        finished_reading is allowed, and if it is hands off to _finished_reading
365
575
        to perform the action.
366
576
        """
380
590
    def _finished_writing(self):
381
591
        """Helper for finished_writing.
382
592
 
383
 
        finished_writing checks the state of the request to determine if 
 
593
        finished_writing checks the state of the request to determine if
384
594
        finished_writing is allowed, and if it is hands off to _finished_writing
385
595
        to perform the action.
386
596
        """
401
611
        return self._read_bytes(count)
402
612
 
403
613
    def _read_bytes(self, count):
404
 
        """Helper for read_bytes.
 
614
        """Helper for SmartClientMediumRequest.read_bytes.
405
615
 
406
616
        read_bytes checks the state of the request to determing if bytes
407
617
        should be read. After that it hands off to _read_bytes to do the
408
618
        actual read.
 
619
 
 
620
        By default this forwards to self._medium.read_bytes because we are
 
621
        operating on the medium's stream.
409
622
        """
410
 
        raise NotImplementedError(self._read_bytes)
 
623
        return self._medium.read_bytes(count)
411
624
 
412
625
    def read_line(self):
413
 
        """Read bytes from this request's response until a newline byte.
414
 
        
415
 
        This isn't particularly efficient, so should only be used when the
416
 
        expected size of the line is quite short.
417
 
 
418
 
        :returns: a string of bytes ending in a newline (byte 0x0A).
419
 
        """
420
 
        # XXX: this duplicates SmartClientRequestProtocolOne._recv_tuple
421
 
        line = ''
422
 
        while not line or line[-1] != '\n':
423
 
            new_char = self.read_bytes(1)
424
 
            line += new_char
425
 
            if new_char == '':
426
 
                # end of file encountered reading from server
427
 
                raise errors.ConnectionReset(
428
 
                    "please check connectivity and permissions",
429
 
                    "(and try -Dhpss if further diagnosis is required)")
 
626
        line = self._read_line()
 
627
        if not line.endswith('\n'):
 
628
            # end of file encountered reading from server
 
629
            raise errors.ConnectionReset(
 
630
                "Unexpected end of message. Please check connectivity "
 
631
                "and permissions, and report a bug if problems persist.")
430
632
        return line
431
633
 
432
 
 
433
 
class SmartClientMedium(object):
 
634
    def _read_line(self):
 
635
        """Helper for SmartClientMediumRequest.read_line.
 
636
 
 
637
        By default this forwards to self._medium._get_line because we are
 
638
        operating on the medium's stream.
 
639
        """
 
640
        return self._medium._get_line()
 
641
 
 
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
 
 
725
class SmartClientMedium(SmartMedium):
434
726
    """Smart client is a medium for sending smart protocol requests over."""
435
727
 
436
728
    def __init__(self, base):
439
731
        self._protocol_version_error = None
440
732
        self._protocol_version = None
441
733
        self._done_hello = False
 
734
        # Be optimistic: we assume the remote end can accept new remote
 
735
        # requests until we get an error saying otherwise.
 
736
        # _remote_version_is_before tracks the bzr version the remote side
 
737
        # can be based on what we've seen so far.
 
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()
 
749
 
 
750
    def _is_remote_before(self, version_tuple):
 
751
        """Is it possible the remote side supports RPCs for a given version?
 
752
 
 
753
        Typical use::
 
754
 
 
755
            needed_version = (1, 2)
 
756
            if medium._is_remote_before(needed_version):
 
757
                fallback_to_pre_1_2_rpc()
 
758
            else:
 
759
                try:
 
760
                    do_1_2_rpc()
 
761
                except UnknownSmartMethod:
 
762
                    medium._remember_remote_is_before(needed_version)
 
763
                    fallback_to_pre_1_2_rpc()
 
764
 
 
765
        :seealso: _remember_remote_is_before
 
766
        """
 
767
        if self._remote_version_is_before is None:
 
768
            # So far, the remote side seems to support everything
 
769
            return False
 
770
        return version_tuple >= self._remote_version_is_before
 
771
 
 
772
    def _remember_remote_is_before(self, version_tuple):
 
773
        """Tell this medium that the remote side is older the given version.
 
774
 
 
775
        :seealso: _is_remote_before
 
776
        """
 
777
        if (self._remote_version_is_before is not None and
 
778
            version_tuple > self._remote_version_is_before):
 
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(
 
784
                "_remember_remote_is_before(%r) called, but "
 
785
                "_remember_remote_is_before(%r) was called previously."
 
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
 
793
        self._remote_version_is_before = version_tuple
442
794
 
443
795
    def protocol_version(self):
444
796
        """Find out if 'hello' smart request works."""
449
801
                medium_request = self.get_request()
450
802
                # Send a 'hello' request in protocol version one, for maximum
451
803
                # backwards compatibility.
452
 
                client_protocol = SmartClientRequestProtocolOne(medium_request)
 
804
                client_protocol = protocol.SmartClientRequestProtocolOne(medium_request)
453
805
                client_protocol.query_version()
454
806
                self._done_hello = True
455
807
            except errors.SmartProtocolError, e:
477
829
 
478
830
    def disconnect(self):
479
831
        """If this medium maintains a persistent connection, close it.
480
 
        
 
832
 
481
833
        The default implementation does nothing.
482
834
        """
483
 
        
 
835
 
484
836
    def remote_path_from_transport(self, transport):
485
837
        """Convert transport into a path suitable for using in a request.
486
 
        
 
838
 
487
839
        Note that the resulting remote path doesn't encode the host name or
488
840
        anything but path, so it is only safe to use it in requests sent over
489
841
        the medium from the matching transport.
490
842
        """
491
843
        medium_base = urlutils.join(self.base, '/')
492
844
        rel_url = urlutils.relative_url(medium_base, transport.base)
493
 
        return urllib.unquote(rel_url)
 
845
        return urlutils.unquote(rel_url)
494
846
 
495
847
 
496
848
class SmartClientStreamMedium(SmartClientMedium):
505
857
    def __init__(self, base):
506
858
        SmartClientMedium.__init__(self, base)
507
859
        self._current_request = None
508
 
        # Be optimistic: we assume the remote end can accept new remote
509
 
        # requests until we get an error saying otherwise.  (1.2 adds some
510
 
        # requests that send bodies, which confuses older servers.)
511
 
        self._remote_is_at_least_1_2 = True
512
860
 
513
861
    def accept_bytes(self, bytes):
514
862
        self._accept_bytes(bytes)
521
869
 
522
870
    def _flush(self):
523
871
        """Flush the output stream.
524
 
        
 
872
 
525
873
        This method is used by the SmartClientStreamMediumRequest to ensure that
526
874
        all data for a request is sent, to avoid long timeouts or deadlocks.
527
875
        """
535
883
        """
536
884
        return SmartClientStreamMediumRequest(self)
537
885
 
538
 
    def read_bytes(self, count):
539
 
        return self._read_bytes(count)
 
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
540
893
 
541
894
 
542
895
class SmartSimplePipesClientMedium(SmartClientStreamMedium):
543
896
    """A client medium using simple pipes.
544
 
    
 
897
 
545
898
    This client does not manage the pipes: it assumes they will always be open.
546
899
    """
547
900
 
552
905
 
553
906
    def _accept_bytes(self, bytes):
554
907
        """See SmartClientStreamMedium.accept_bytes."""
555
 
        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')
556
916
 
557
917
    def _flush(self):
558
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.
559
922
        self._writeable_pipe.flush()
560
923
 
561
924
    def _read_bytes(self, count):
562
925
        """See SmartClientStreamMedium._read_bytes."""
563
 
        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
564
942
 
565
943
 
566
944
class SmartSSHClientMedium(SmartClientStreamMedium):
567
 
    """A client medium using SSH."""
568
 
    
569
 
    def __init__(self, host, port=None, username=None, password=None,
570
 
            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):
571
952
        """Creates a client that will connect on the first use.
572
 
        
 
953
 
 
954
        :param ssh_params: A SSHParams instance.
573
955
        :param vendor: An optional override for the ssh vendor to use. See
574
956
            bzrlib.transport.ssh for details on ssh vendors.
575
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.
576
966
        SmartClientStreamMedium.__init__(self, base)
577
 
        self._connected = False
578
 
        self._host = host
579
 
        self._password = password
580
 
        self._port = port
581
 
        self._username = username
582
 
        self._read_from = None
 
967
        self._vendor = vendor
583
968
        self._ssh_connection = None
584
 
        self._vendor = vendor
585
 
        self._write_to = None
586
 
        self._bzr_remote_path = bzr_remote_path
587
 
        if self._bzr_remote_path is None:
588
 
            symbol_versioning.warn(
589
 
                'bzr_remote_path is required as of bzr 0.92',
590
 
                DeprecationWarning, stacklevel=2)
591
 
            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)
592
985
 
593
986
    def _accept_bytes(self, bytes):
594
987
        """See SmartClientStreamMedium.accept_bytes."""
595
988
        self._ensure_connection()
596
 
        self._write_to.write(bytes)
 
989
        self._real_medium.accept_bytes(bytes)
597
990
 
598
991
    def disconnect(self):
599
992
        """See SmartClientMedium.disconnect()."""
600
 
        if not self._connected:
601
 
            return
602
 
        self._read_from.close()
603
 
        self._write_to.close()
604
 
        self._ssh_connection.close()
605
 
        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
606
999
 
607
1000
    def _ensure_connection(self):
608
1001
        """Connect this medium if not already connected."""
609
 
        if self._connected:
 
1002
        if self._real_medium is not None:
610
1003
            return
611
1004
        if self._vendor is None:
612
1005
            vendor = ssh._get_ssh_vendor()
613
1006
        else:
614
1007
            vendor = self._vendor
615
 
        self._ssh_connection = vendor.connect_ssh(self._username,
616
 
                self._password, self._host, self._port,
617
 
                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',
618
1012
                         '--directory=/', '--allow-writes'])
619
 
        self._read_from, self._write_to = \
620
 
            self._ssh_connection.get_filelike_channels()
621
 
        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)
622
1027
 
623
1028
    def _flush(self):
624
1029
        """See SmartClientStreamMedium._flush()."""
625
 
        self._write_to.flush()
 
1030
        self._real_medium._flush()
626
1031
 
627
1032
    def _read_bytes(self, count):
628
1033
        """See SmartClientStreamMedium.read_bytes."""
629
 
        if not self._connected:
 
1034
        if self._real_medium is None:
630
1035
            raise errors.MediumNotConnected(self)
631
 
        return self._read_from.read(count)
 
1036
        return self._real_medium.read_bytes(count)
632
1037
 
633
1038
 
634
1039
# Port 4155 is the default port for bzr://, registered with IANA.
635
 
BZR_DEFAULT_INTERFACE = '0.0.0.0'
 
1040
BZR_DEFAULT_INTERFACE = None
636
1041
BZR_DEFAULT_PORT = 4155
637
1042
 
638
1043
 
639
 
class SmartTCPClientMedium(SmartClientStreamMedium):
640
 
    """A client medium using TCP."""
 
1044
class SmartClientSocketMedium(SmartClientStreamMedium):
 
1045
    """A client medium using a socket.
641
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
 
642
1091
    def __init__(self, host, port, base):
643
1092
        """Creates a client that will connect on the first use."""
644
 
        SmartClientStreamMedium.__init__(self, base)
645
 
        self._connected = False
 
1093
        SmartClientSocketMedium.__init__(self, base)
646
1094
        self._host = host
647
1095
        self._port = port
648
 
        self._socket = None
649
 
 
650
 
    def _accept_bytes(self, bytes):
651
 
        """See SmartClientMedium.accept_bytes."""
652
 
        self._ensure_connection()
653
 
        osutils.send_all(self._socket, bytes)
654
 
 
655
 
    def disconnect(self):
656
 
        """See SmartClientMedium.disconnect()."""
657
 
        if not self._connected:
658
 
            return
659
 
        self._socket.close()
660
 
        self._socket = None
661
 
        self._connected = False
662
1096
 
663
1097
    def _ensure_connection(self):
664
1098
        """Connect this medium if not already connected."""
665
1099
        if self._connected:
666
1100
            return
667
 
        self._socket = socket.socket()
668
 
        self._socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
669
1101
        if self._port is None:
670
1102
            port = BZR_DEFAULT_PORT
671
1103
        else:
672
1104
            port = int(self._port)
673
1105
        try:
674
 
            self._socket.connect((self._host, port))
675
 
        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:
676
1126
            # socket errors either have a (string) or (errno, string) as their
677
1127
            # args.
678
1128
            if type(err.args) is str:
682
1132
            raise errors.ConnectionError("failed to connect to %s:%d: %s" %
683
1133
                    (self._host, port, err_msg))
684
1134
        self._connected = True
685
 
 
686
 
    def _flush(self):
687
 
        """See SmartClientStreamMedium._flush().
688
 
        
689
 
        For TCP we do no flushing. We may want to turn off TCP_NODELAY and 
690
 
        add a means to do a flush, but that can be done in the future.
691
 
        """
692
 
 
693
 
    def _read_bytes(self, count):
694
 
        """See SmartClientMedium.read_bytes."""
695
 
        if not self._connected:
696
 
            raise errors.MediumNotConnected(self)
697
 
        return self._socket.recv(count)
 
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
698
1154
 
699
1155
 
700
1156
class SmartClientStreamMediumRequest(SmartClientMediumRequest):
713
1169
 
714
1170
    def _accept_bytes(self, bytes):
715
1171
        """See SmartClientMediumRequest._accept_bytes.
716
 
        
 
1172
 
717
1173
        This forwards to self._medium._accept_bytes because we are operating
718
1174
        on the mediums stream.
719
1175
        """
722
1178
    def _finished_reading(self):
723
1179
        """See SmartClientMediumRequest._finished_reading.
724
1180
 
725
 
        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
726
1182
        request to be created.
727
1183
        """
728
1184
        if self._medium._current_request is not self:
729
1185
            raise AssertionError()
730
1186
        self._medium._current_request = None
731
 
        
 
1187
 
732
1188
    def _finished_writing(self):
733
1189
        """See SmartClientMediumRequest._finished_writing.
734
1190
 
735
1191
        This invokes self._medium._flush to ensure all bytes are transmitted.
736
1192
        """
737
1193
        self._medium._flush()
738
 
 
739
 
    def _read_bytes(self, count):
740
 
        """See SmartClientMediumRequest._read_bytes.
741
 
        
742
 
        This forwards to self._medium._read_bytes because we are operating
743
 
        on the mediums stream.
744
 
        """
745
 
        return self._medium._read_bytes(count)
746