1
=== modified file 'bzrlib/help_topics/en/debug-flags.txt'
2
--- bzrlib/help_topics/en/debug-flags.txt	2010-01-05 04:30:07 +0000
3
+++ bzrlib/help_topics/en/debug-flags.txt	2012-09-11 12:31:20 +0000
4
@@ -24,6 +24,8 @@
5
24
-Dindex           Trace major index operations.
24
-Dindex           Trace major index operations.
6
25
-Dknit            Trace knit operations.
25
-Dknit            Trace knit operations.
7
26
-Dlock            Trace when lockdir locks are taken or released.
26
-Dlock            Trace when lockdir locks are taken or released.
8
27
-Dnoretry         If a connection is reset, fail immediately rather than
9
28
                  retrying the request.
10
27
-Dprogress        Trace progress bar operations.
29
-Dprogress        Trace progress bar operations.
11
28
-Dmerge           Emit information for debugging merges.
30
-Dmerge           Emit information for debugging merges.
12
29
-Dno_apport       Don't use apport to report crashes.
31
-Dno_apport       Don't use apport to report crashes.
13
30
32
14
=== modified file 'bzrlib/osutils.py'
15
--- bzrlib/osutils.py	2010-05-27 04:00:01 +0000
16
+++ bzrlib/osutils.py	2012-09-11 12:31:20 +0000
17
@@ -40,6 +40,7 @@
18
40
    rmtree,
40
    rmtree,
19
41
    )
41
    )
20
42
import signal
42
import signal
21
43
import socket
22
43
import subprocess
44
import subprocess
23
44
import tempfile
45
import tempfile
24
45
from tempfile import (
46
from tempfile import (
25
@@ -1929,6 +1930,20 @@
26
1929
        return socket.gethostname().decode(get_user_encoding())
1930
        return socket.gethostname().decode(get_user_encoding())
27
1930
1931
28
1931
1932
29
1933
# We must not read/write any more than 64k at a time from/to a socket so we
30
1934
# don't risk "no buffer space available" errors on some platforms.  Windows in
31
1935
# particular is likely to throw WSAECONNABORTED or WSAENOBUFS if given too much
32
1936
# data at once.
33
1937
MAX_SOCKET_CHUNK = 64 * 1024
34
1938
35
1939
_end_of_stream_errors = [errno.ECONNRESET, errno.EPIPE, errno.EINVAL]
36
1940
for _eno in ['WSAECONNRESET', 'WSAECONNABORTED']:
37
1941
    _eno = getattr(errno, _eno, None)
38
1942
    if _eno is not None:
39
1943
        _end_of_stream_errors.append(_eno)
40
1944
del _eno
41
1945
42
1946
43
1932
def recv_all(socket, bytes):
1947
def recv_all(socket, bytes):
44
1933
    """Receive an exact number of bytes.
1948
    """Receive an exact number of bytes.
45
1934
1949
46
@@ -1948,21 +1963,37 @@
47
1948
    return b
1963
    return b
48
1949
1964
49
1950
1965
51
1951
def send_all(socket, bytes, report_activity=None):
1966
def send_all(sock, bytes, report_activity=None):
52
1952
    """Send all bytes on a socket.
1967
    """Send all bytes on a socket.
53
1953
1968
56
1954
    Regular socket.sendall() can give socket error 10053 on Windows.  This
1969
    Breaks large blocks in smaller chunks to avoid buffering limitations on
57
1955
    implementation sends no more than 64k at a time, which avoids this problem.
1970
    some platforms, and catches EINTR which may be thrown if the send is
58
1971
    interrupted by a signal.
59
1972
60
1973
    This is preferred to socket.sendall(), because it avoids portability bugs
61
1974
    and provides activity reporting.
62
1956
1975
63
1957
    :param report_activity: Call this as bytes are read, see
1976
    :param report_activity: Call this as bytes are read, see
64
1958
        Transport._report_activity
1977
        Transport._report_activity
65
1959
    """
1978
    """
72
1960
    chunk_size = 2**16
1979
    sent_total = 0
73
1961
    for pos in xrange(0, len(bytes), chunk_size):
1980
    byte_count = len(bytes)
74
1962
        block = bytes[pos:pos+chunk_size]
1981
    while sent_total < byte_count:
75
1963
        if report_activity is not None:
1982
        try:
76
1964
            report_activity(len(block), 'write')
1983
            sent = sock.send(buffer(bytes, sent_total, MAX_SOCKET_CHUNK))
77
1965
        until_no_eintr(socket.sendall, block)
1984
        except (socket.error, IOError), e:
78
1985
            if e.args[0] in _end_of_stream_errors:
79
1986
                raise errors.ConnectionReset(
80
1987
                    "Error trying to write to socket", e)
81
1988
            if e.args[0] != errno.EINTR:
82
1989
                raise
83
1990
        else:
84
1991
            if sent == 0:
85
1992
                raise errors.ConnectionReset('Sending to %s returned 0 bytes'
86
1993
                                             % (sock,))
87
1994
            sent_total += sent
88
1995
            if report_activity is not None:
89
1996
                report_activity(sent, 'write')
90
1966
1997
91
1967
1998
92
1968
def dereference_path(path):
1999
def dereference_path(path):
93
1969
2000
94
=== modified file 'bzrlib/smart/client.py'
95
--- bzrlib/smart/client.py	2010-02-17 17:11:16 +0000
96
+++ bzrlib/smart/client.py	2012-09-11 12:31:20 +0000
97
@@ -16,10 +16,11 @@
98
16
16
99
17
import bzrlib
17
import bzrlib
100
18
from bzrlib.smart import message, protocol
18
from bzrlib.smart import message, protocol
101
19
from bzrlib.trace import warning
102
20
from bzrlib import (
19
from bzrlib import (
103
20
    debug,
104
21
    errors,
21
    errors,
105
22
    hooks,
22
    hooks,
106
23
    trace,
107
23
    )
24
    )
108
24
25
109
25
26
110
@@ -39,93 +40,12 @@
111
39
    def __repr__(self):
40
    def __repr__(self):
112
40
        return '%s(%r)' % (self.__class__.__name__, self._medium)
41
        return '%s(%r)' % (self.__class__.__name__, self._medium)
113
41
42
114
42
    def _send_request(self, protocol_version, method, args, body=None,
115
43
                      readv_body=None, body_stream=None):
116
44
        encoder, response_handler = self._construct_protocol(
117
45
            protocol_version)
118
46
        encoder.set_headers(self._headers)
119
47
        if body is not None:
120
48
            if readv_body is not None:
121
49
                raise AssertionError(
122
50
                    "body and readv_body are mutually exclusive.")
123
51
            if body_stream is not None:
124
52
                raise AssertionError(
125
53
                    "body and body_stream are mutually exclusive.")
126
54
            encoder.call_with_body_bytes((method, ) + args, body)
127
55
        elif readv_body is not None:
128
56
            if body_stream is not None:
129
57
                raise AssertionError(
130
58
                    "readv_body and body_stream are mutually exclusive.")
131
59
            encoder.call_with_body_readv_array((method, ) + args, readv_body)
132
60
        elif body_stream is not None:
133
61
            encoder.call_with_body_stream((method, ) + args, body_stream)
134
62
        else:
135
63
            encoder.call(method, *args)
136
64
        return response_handler
137
65
138
66
    def _run_call_hooks(self, method, args, body, readv_body):
139
67
        if not _SmartClient.hooks['call']:
140
68
            return
141
69
        params = CallHookParams(method, args, body, readv_body, self._medium)
142
70
        for hook in _SmartClient.hooks['call']:
143
71
            hook(params)
144
72
145
73
    def _call_and_read_response(self, method, args, body=None, readv_body=None,
43
    def _call_and_read_response(self, method, args, body=None, readv_body=None,
146
74
            body_stream=None, expect_response_body=True):
44
            body_stream=None, expect_response_body=True):
201
75
        self._run_call_hooks(method, args, body, readv_body)
45
        request = _SmartClientRequest(self, method, args, body=body,
202
76
        if self._medium._protocol_version is not None:
46
            readv_body=readv_body, body_stream=body_stream,
203
77
            response_handler = self._send_request(
47
            expect_response_body=expect_response_body)
204
78
                self._medium._protocol_version, method, args, body=body,
48
        return request.call_and_read_response()
151
79
                readv_body=readv_body, body_stream=body_stream)
152
80
            return (response_handler.read_response_tuple(
153
81
                        expect_body=expect_response_body),
154
82
                    response_handler)
155
83
        else:
156
84
            for protocol_version in [3, 2]:
157
85
                if protocol_version == 2:
158
86
                    # If v3 doesn't work, the remote side is older than 1.6.
159
87
                    self._medium._remember_remote_is_before((1, 6))
160
88
                response_handler = self._send_request(
161
89
                    protocol_version, method, args, body=body,
162
90
                    readv_body=readv_body, body_stream=body_stream)
163
91
                try:
164
92
                    response_tuple = response_handler.read_response_tuple(
165
93
                        expect_body=expect_response_body)
166
94
                except errors.UnexpectedProtocolVersionMarker, err:
167
95
                    # TODO: We could recover from this without disconnecting if
168
96
                    # we recognise the protocol version.
169
97
                    warning(
170
98
                        'Server does not understand Bazaar network protocol %d,'
171
99
                        ' reconnecting.  (Upgrade the server to avoid this.)'
172
100
                        % (protocol_version,))
173
101
                    self._medium.disconnect()
174
102
                    continue
175
103
                except errors.ErrorFromSmartServer:
176
104
                    # If we received an error reply from the server, then it
177
105
                    # must be ok with this protocol version.
178
106
                    self._medium._protocol_version = protocol_version
179
107
                    raise
180
108
                else:
181
109
                    self._medium._protocol_version = protocol_version
182
110
                    return response_tuple, response_handler
183
111
            raise errors.SmartProtocolError(
184
112
                'Server is not a Bazaar server: ' + str(err))
185
113
186
114
    def _construct_protocol(self, version):
187
115
        request = self._medium.get_request()
188
116
        if version == 3:
189
117
            request_encoder = protocol.ProtocolThreeRequester(request)
190
118
            response_handler = message.ConventionalResponseHandler()
191
119
            response_proto = protocol.ProtocolThreeDecoder(
192
120
                response_handler, expect_version_marker=True)
193
121
            response_handler.setProtoAndMediumRequest(response_proto, request)
194
122
        elif version == 2:
195
123
            request_encoder = protocol.SmartClientRequestProtocolTwo(request)
196
124
            response_handler = request_encoder
197
125
        else:
198
126
            request_encoder = protocol.SmartClientRequestProtocolOne(request)
199
127
            response_handler = request_encoder
200
128
        return request_encoder, response_handler
205
129
49
206
130
    def call(self, method, *args):
50
    def call(self, method, *args):
207
131
        """Call a method on the remote server."""
51
        """Call a method on the remote server."""
208
@@ -191,6 +111,170 @@
209
191
        return self._medium.remote_path_from_transport(transport)
111
        return self._medium.remote_path_from_transport(transport)
210
192
112
211
193
113
212
114
class _SmartClientRequest(object):
213
115
    """Encapsulate the logic for a single request.
214
116
215
117
    This class handles things like reconnecting and sending the request a
216
118
    second time when the connection is reset in the middle. It also handles the
217
119
    multiple requests that get made if we don't know what protocol the server
218
120
    supports yet.
219
121
220
122
    Generally, you build up one of these objects, passing in the arguments that
221
123
    you want to send to the server, and then use 'call_and_read_response' to
222
124
    get the response from the server.
223
125
    """
224
126
225
127
    def __init__(self, client, method, args, body=None, readv_body=None,
226
128
                 body_stream=None, expect_response_body=True):
227
129
        self.client = client
228
130
        self.method = method
229
131
        self.args = args
230
132
        self.body = body
231
133
        self.readv_body = readv_body
232
134
        self.body_stream = body_stream
233
135
        self.expect_response_body = expect_response_body
234
136
235
137
    def call_and_read_response(self):
236
138
        """Send the request to the server, and read the initial response.
237
139
238
140
        This doesn't read all of the body content of the response, instead it
239
141
        returns (response_tuple, response_handler). response_tuple is the 'ok',
240
142
        or 'error' information, and 'response_handler' can be used to get the
241
143
        content stream out.
242
144
        """
243
145
        self._run_call_hooks()
244
146
        protocol_version = self.client._medium._protocol_version
245
147
        if protocol_version is None:
246
148
            return self._call_determining_protocol_version()
247
149
        else:
248
150
            return self._call(protocol_version)
249
151
250
152
    def _run_call_hooks(self):
251
153
        if not _SmartClient.hooks['call']:
252
154
            return
253
155
        params = CallHookParams(self.method, self.args, self.body,
254
156
                                self.readv_body, self.client._medium)
255
157
        for hook in _SmartClient.hooks['call']:
256
158
            hook(params)
257
159
258
160
    def _call(self, protocol_version):
259
161
        """We know the protocol version.
260
162
261
163
        So this just sends the request, and then reads the response. This is
262
164
        where the code will be to retry requests if the connection is closed.
263
165
        """
264
166
        response_handler = self._send(protocol_version)
265
167
        response_tuple = response_handler.read_response_tuple(
266
168
            expect_body=self.expect_response_body)
267
169
        return (response_tuple, response_handler)
268
170
269
171
    def _call_determining_protocol_version(self):
270
172
        """Determine what protocol the remote server supports.
271
173
272
174
        We do this by placing a request in the most recent protocol, and
273
175
        handling the UnexpectedProtocolVersionMarker from the server.
274
176
        """
275
177
        for protocol_version in [3, 2]:
276
178
            if protocol_version == 2:
277
179
                # If v3 doesn't work, the remote side is older than 1.6.
278
180
                self.client._medium._remember_remote_is_before((1, 6))
279
181
            try:
280
182
                response_tuple, response_handler = self._call(protocol_version)
281
183
            except errors.UnexpectedProtocolVersionMarker, err:
282
184
                # TODO: We could recover from this without disconnecting if
283
185
                # we recognise the protocol version.
284
186
                trace.warning(
285
187
                    'Server does not understand Bazaar network protocol %d,'
286
188
                    ' reconnecting.  (Upgrade the server to avoid this.)'
287
189
                    % (protocol_version,))
288
190
                self.client._medium.disconnect()
289
191
                continue
290
192
            except errors.ErrorFromSmartServer:
291
193
                # If we received an error reply from the server, then it
292
194
                # must be ok with this protocol version.
293
195
                self.client._medium._protocol_version = protocol_version
294
196
                raise
295
197
            else:
296
198
                self.client._medium._protocol_version = protocol_version
297
199
                return response_tuple, response_handler
298
200
        raise errors.SmartProtocolError(
299
201
            'Server is not a Bazaar server: ' + str(err))
300
202
301
203
    def _construct_protocol(self, version):
302
204
        """Build the encoding stack for a given protocol version."""
303
205
        request = self.client._medium.get_request()
304
206
        if version == 3:
305
207
            request_encoder = protocol.ProtocolThreeRequester(request)
306
208
            response_handler = message.ConventionalResponseHandler()
307
209
            response_proto = protocol.ProtocolThreeDecoder(
308
210
                response_handler, expect_version_marker=True)
309
211
            response_handler.setProtoAndMediumRequest(response_proto, request)
310
212
        elif version == 2:
311
213
            request_encoder = protocol.SmartClientRequestProtocolTwo(request)
312
214
            response_handler = request_encoder
313
215
        else:
314
216
            request_encoder = protocol.SmartClientRequestProtocolOne(request)
315
217
            response_handler = request_encoder
316
218
        return request_encoder, response_handler
317
219
318
220
    def _send(self, protocol_version):
319
221
        """Encode the request, and send it to the server.
320
222
321
223
        This will retry a request if we get a ConnectionReset while sending the
322
224
        request to the server. (Unless we have a body_stream that we have
323
225
        already started consuming, since we can't restart body_streams)
324
226
325
227
        :return: response_handler as defined by _construct_protocol
326
228
        """
327
229
        encoder, response_handler = self._construct_protocol(protocol_version)
328
230
        try:
329
231
            self._send_no_retry(encoder)
330
232
        except errors.ConnectionReset, e:
331
233
            # If we fail during the _send_no_retry phase, then we can
332
234
            # be confident that the server did not get our request, because we
333
235
            # haven't started waiting for the reply yet. So try the request
334
236
            # again. We only issue a single retry, because if the connection
335
237
            # really is down, there is no reason to loop endlessly.
336
238
337
239
            # Connection is dead, so close our end of it.
338
240
            self.client._medium.reset()
339
241
            if (('noretry' in debug.debug_flags)
340
242
                or self.body_stream is not None):
341
243
                # We can't restart a body_stream that has been partially
342
244
                # consumed, so we don't retry.
343
245
                raise
344
246
            trace.warning('ConnectionReset calling %r, retrying'
345
247
                          % (self.method,))
346
248
            trace.log_exception_quietly()
347
249
            encoder, response_handler = self._construct_protocol(
348
250
                protocol_version)
349
251
            self._send_no_retry(encoder)
350
252
        return response_handler
351
253
352
254
    def _send_no_retry(self, encoder):
353
255
        """Just encode the request and try to send it."""
354
256
        encoder.set_headers(self.client._headers)
355
257
        if self.body is not None:
356
258
            if self.readv_body is not None:
357
259
                raise AssertionError(
358
260
                    "body and readv_body are mutually exclusive.")
359
261
            if self.body_stream is not None:
360
262
                raise AssertionError(
361
263
                    "body and body_stream are mutually exclusive.")
362
264
            encoder.call_with_body_bytes((self.method, ) + self.args, self.body)
363
265
        elif self.readv_body is not None:
364
266
            if self.body_stream is not None:
365
267
                raise AssertionError(
366
268
                    "readv_body and body_stream are mutually exclusive.")
367
269
            encoder.call_with_body_readv_array((self.method, ) + self.args,
368
270
                                               self.readv_body)
369
271
        elif self.body_stream is not None:
370
272
            encoder.call_with_body_stream((self.method, ) + self.args,
371
273
                                          self.body_stream)
372
274
        else:
373
275
            encoder.call(self.method, *self.args)
374
276
375
277
376
194
class SmartClientHooks(hooks.Hooks):
278
class SmartClientHooks(hooks.Hooks):
377
195
279
378
196
    def __init__(self):
280
    def __init__(self):
379
197
281
380
=== modified file 'bzrlib/smart/medium.py'
381
--- bzrlib/smart/medium.py	2010-04-27 06:40:37 +0000
382
+++ bzrlib/smart/medium.py	2012-09-11 12:31:20 +0000
383
@@ -712,6 +712,14 @@
384
712
        """
712
        """
385
713
        return SmartClientStreamMediumRequest(self)
713
        return SmartClientStreamMediumRequest(self)
386
714
714
387
715
    def reset(self):
388
716
        """We have been disconnected, reset current state.
389
717
390
718
        This resets things like _current_request and connected state.
391
719
        """
392
720
        self.disconnect()
393
721
        self._current_request = None
394
722
395
715
723
396
716
class SmartSimplePipesClientMedium(SmartClientStreamMedium):
724
class SmartSimplePipesClientMedium(SmartClientStreamMedium):
397
717
    """A client medium using simple pipes.
725
    """A client medium using simple pipes.
398
@@ -726,11 +734,21 @@
399
726
734
400
727
    def _accept_bytes(self, bytes):
735
    def _accept_bytes(self, bytes):
401
728
        """See SmartClientStreamMedium.accept_bytes."""
736
        """See SmartClientStreamMedium.accept_bytes."""
403
729
        osutils.until_no_eintr(self._writeable_pipe.write, bytes)
737
        try:
404
738
            osutils.until_no_eintr(self._writeable_pipe.write, bytes)
405
739
        except IOError, e:
406
740
            if e.errno in (errno.EINVAL, errno.EPIPE):
407
741
                raise errors.ConnectionReset(
408
742
                    "Error trying to write to subprocess:\n%s"
409
743
                    % (e,))
410
744
            raise
411
730
        self._report_activity(len(bytes), 'write')
745
        self._report_activity(len(bytes), 'write')
412
731
746
413
732
    def _flush(self):
747
    def _flush(self):
414
733
        """See SmartClientStreamMedium._flush()."""
748
        """See SmartClientStreamMedium._flush()."""
415
749
        # Note: If flush were to fail, we'd like to raise ConnectionReset, etc.
416
750
        #       However, testing shows that even when the child process is
417
751
        #       gone, this doesn't error.
418
734
        osutils.until_no_eintr(self._writeable_pipe.flush)
752
        osutils.until_no_eintr(self._writeable_pipe.flush)
419
735
753
420
736
    def _read_bytes(self, count):
754
    def _read_bytes(self, count):
421
@@ -741,7 +759,10 @@
422
741
759
423
742
760
424
743
class SmartSSHClientMedium(SmartClientStreamMedium):
761
class SmartSSHClientMedium(SmartClientStreamMedium):
426
744
    """A client medium using SSH."""
762
    """A client medium using SSH.
427
763
428
764
    It delegates IO to a SmartSimplePipesClientMedium.
429
765
    """
430
745
766
431
746
    def __init__(self, host, port=None, username=None, password=None,
767
    def __init__(self, host, port=None, username=None, password=None,
432
747
            base=None, vendor=None, bzr_remote_path=None):
768
            base=None, vendor=None, bzr_remote_path=None):
433
@@ -750,11 +771,11 @@
434
750
        :param vendor: An optional override for the ssh vendor to use. See
771
        :param vendor: An optional override for the ssh vendor to use. See
435
751
            bzrlib.transport.ssh for details on ssh vendors.
772
            bzrlib.transport.ssh for details on ssh vendors.
436
752
        """
773
        """
437
753
        self._connected = False
438
754
        self._host = host
774
        self._host = host
439
755
        self._password = password
775
        self._password = password
440
756
        self._port = port
776
        self._port = port
441
757
        self._username = username
777
        self._username = username
442
778
        self._real_medium = None
443
758
        # for the benefit of progress making a short description of this
779
        # for the benefit of progress making a short description of this
444
759
        # transport
780
        # transport
445
760
        self._scheme = 'bzr+ssh'
781
        self._scheme = 'bzr+ssh'
446
@@ -762,10 +783,8 @@
447
762
        # _DebugCounter so we have to store all the values used in our repr
783
        # _DebugCounter so we have to store all the values used in our repr
448
763
        # method before calling the super init.
784
        # method before calling the super init.
449
764
        SmartClientStreamMedium.__init__(self, base)
785
        SmartClientStreamMedium.__init__(self, base)
450
765
        self._read_from = None
451
766
        self._ssh_connection = None
786
        self._ssh_connection = None
452
767
        self._vendor = vendor
787
        self._vendor = vendor
453
768
        self._write_to = None
454
769
        self._bzr_remote_path = bzr_remote_path
788
        self._bzr_remote_path = bzr_remote_path
455
770
789
456
771
    def __repr__(self):
790
    def __repr__(self):
457
@@ -783,21 +802,20 @@
458
783
    def _accept_bytes(self, bytes):
802
    def _accept_bytes(self, bytes):
459
784
        """See SmartClientStreamMedium.accept_bytes."""
803
        """See SmartClientStreamMedium.accept_bytes."""
460
785
        self._ensure_connection()
804
        self._ensure_connection()
463
786
        osutils.until_no_eintr(self._write_to.write, bytes)
805
        self._real_medium.accept_bytes(bytes)
462
787
        self._report_activity(len(bytes), 'write')
464
788
806
465
789
    def disconnect(self):
807
    def disconnect(self):
466
790
        """See SmartClientMedium.disconnect()."""
808
        """See SmartClientMedium.disconnect()."""
473
791
        if not self._connected:
809
        if self._real_medium is not None:
474
792
            return
810
            self._real_medium.disconnect()
475
793
        osutils.until_no_eintr(self._read_from.close)
811
            self._real_medium = None
476
794
        osutils.until_no_eintr(self._write_to.close)
812
        if self._ssh_connection is not None:
477
795
        self._ssh_connection.close()
813
            self._ssh_connection.close()
478
796
        self._connected = False
814
            self._ssh_connection = None
479
797
815
480
798
    def _ensure_connection(self):
816
    def _ensure_connection(self):
481
799
        """Connect this medium if not already connected."""
817
        """Connect this medium if not already connected."""
483
800
        if self._connected:
818
        if self._real_medium is not None:
484
801
            return
819
            return
485
802
        if self._vendor is None:
820
        if self._vendor is None:
486
803
            vendor = ssh._get_ssh_vendor()
821
            vendor = ssh._get_ssh_vendor()
487
@@ -807,22 +825,19 @@
488
807
                self._password, self._host, self._port,
825
                self._password, self._host, self._port,
489
808
                command=[self._bzr_remote_path, 'serve', '--inet',
826
                command=[self._bzr_remote_path, 'serve', '--inet',
490
809
                         '--directory=/', '--allow-writes'])
827
                         '--directory=/', '--allow-writes'])
494
810
        self._read_from, self._write_to = \
828
        read_from, write_to = self._ssh_connection.get_filelike_channels()
495
811
            self._ssh_connection.get_filelike_channels()
829
        self._real_medium = SmartSimplePipesClientMedium(
496
812
        self._connected = True
830
            read_from, write_to, self.base)
497
813
831
498
814
    def _flush(self):
832
    def _flush(self):
499
815
        """See SmartClientStreamMedium._flush()."""
833
        """See SmartClientStreamMedium._flush()."""
501
816
        self._write_to.flush()
834
        self._real_medium._flush()
502
817
835
503
818
    def _read_bytes(self, count):
836
    def _read_bytes(self, count):
504
819
        """See SmartClientStreamMedium.read_bytes."""
837
        """See SmartClientStreamMedium.read_bytes."""
506
820
        if not self._connected:
838
        if self._real_medium is None:
507
821
            raise errors.MediumNotConnected(self)
839
            raise errors.MediumNotConnected(self)
512
822
        bytes_to_read = min(count, _MAX_READ_SIZE)
840
        return self._real_medium.read_bytes(count)
509
823
        bytes = osutils.until_no_eintr(self._read_from.read, bytes_to_read)
510
824
        self._report_activity(len(bytes), 'read')
511
825
        return bytes
513
826
841
514
827
842
515
828
# Port 4155 is the default port for bzr://, registered with IANA.
843
# Port 4155 is the default port for bzr://, registered with IANA.
516
@@ -948,13 +963,17 @@
517
948
        self._medium._flush()
963
        self._medium._flush()
518
949
964
519
950
965
520
966
WSAECONNABORTED = 10053
521
967
WSAECONNRESET = 10054
522
968
523
951
def _read_bytes_from_socket(sock, desired_count, report_activity):
969
def _read_bytes_from_socket(sock, desired_count, report_activity):
524
952
    # We ignore the desired_count because on sockets it's more efficient to
970
    # We ignore the desired_count because on sockets it's more efficient to
525
953
    # read large chunks (of _MAX_READ_SIZE bytes) at a time.
971
    # read large chunks (of _MAX_READ_SIZE bytes) at a time.
526
954
    try:
972
    try:
527
955
        bytes = osutils.until_no_eintr(sock, _MAX_READ_SIZE)
973
        bytes = osutils.until_no_eintr(sock, _MAX_READ_SIZE)
528
956
    except socket.error, e:
974
    except socket.error, e:
530
957
        if len(e.args) and e.args[0] in (errno.ECONNRESET, 10054):
975
        if len(e.args) and e.args[0] in (errno.ECONNRESET, WSAECONNABORTED,
531
976
                                         WSAECONNRESET):
532
958
            # The connection was closed by the other side.  Callers expect an
977
            # The connection was closed by the other side.  Callers expect an
533
959
            # empty string to signal end-of-stream.
978
            # empty string to signal end-of-stream.
534
960
            bytes = ''
979
            bytes = ''
535
961
980
536
=== modified file 'bzrlib/smart/protocol.py'
537
--- bzrlib/smart/protocol.py	2010-02-17 17:11:16 +0000
538
+++ bzrlib/smart/protocol.py	2012-09-11 12:31:20 +0000
539
@@ -1075,9 +1075,6 @@
540
1075
        self._real_write_func = write_func
1075
        self._real_write_func = write_func
541
1076
1076
542
1077
    def _write_func(self, bytes):
1077
    def _write_func(self, bytes):
543
1078
        # TODO: It is probably more appropriate to use sum(map(len, _buf))
544
1079
        #       for total number of bytes to write, rather than buffer based on
545
1080
        #       the number of write() calls
546
1081
        # TODO: Another possibility would be to turn this into an async model.
1078
        # TODO: Another possibility would be to turn this into an async model.
547
1082
        #       Where we let another thread know that we have some bytes if
1079
        #       Where we let another thread know that we have some bytes if
548
1083
        #       they want it, but we don't actually block for it
1080
        #       they want it, but we don't actually block for it
549
1084
1081
550
=== modified file 'bzrlib/tests/test_osutils.py'
551
--- bzrlib/tests/test_osutils.py	2010-11-30 20:42:42 +0000
552
+++ bzrlib/tests/test_osutils.py	2012-09-11 12:31:20 +0000
553
@@ -801,6 +801,45 @@
554
801
        self.assertEqual(None, osutils.safe_file_id(None))
801
        self.assertEqual(None, osutils.safe_file_id(None))
555
802
802
556
803
803
557
804
class TestSendAll(tests.TestCase):
558
805
559
806
    def test_send_with_disconnected_socket(self):
560
807
        class DisconnectedSocket(object):
561
808
            def __init__(self, err):
562
809
                self.err = err
563
810
            def send(self, content):
564
811
                raise self.err
565
812
            def close(self):
566
813
                pass
567
814
        # All of these should be treated as ConnectionReset
568
815
        errs = []
569
816
        for err_cls in (IOError, socket.error):
570
817
            for errnum in osutils._end_of_stream_errors:
571
818
                errs.append(err_cls(errnum))
572
819
        for err in errs:
573
820
            sock = DisconnectedSocket(err)
574
821
            self.assertRaises(errors.ConnectionReset,
575
822
                osutils.send_all, sock, 'some more content')
576
823
577
824
    def test_send_with_no_progress(self):
578
825
        # See https://bugs.launchpad.net/bzr/+bug/1047309
579
826
        # It seems that paramiko can get into a state where it doesn't error,
580
827
        # but it returns 0 bytes sent for requests over and over again.
581
828
        class NoSendingSocket(object):
582
829
            def __init__(self):
583
830
                self.call_count = 0
584
831
            def send(self, bytes):
585
832
                self.call_count += 1
586
833
                if self.call_count > 100:
587
834
                    # Prevent the test suite from hanging
588
835
                    raise RuntimeError('too many calls')
589
836
                return 0
590
837
        sock = NoSendingSocket()
591
838
        self.assertRaises(errors.ConnectionReset,
592
839
                          osutils.send_all, sock, 'content')
593
840
        self.assertEqual(1, sock.call_count)
594
841
595
842
596
804
class TestWin32Funcs(tests.TestCase):
843
class TestWin32Funcs(tests.TestCase):
597
805
    """Test that _win32 versions of os utilities return appropriate paths."""
844
    """Test that _win32 versions of os utilities return appropriate paths."""
598
806
845
599
807
846
600
=== modified file 'bzrlib/tests/test_smart_transport.py'
601
--- bzrlib/tests/test_smart_transport.py	2010-02-17 17:11:16 +0000
602
+++ bzrlib/tests/test_smart_transport.py	2012-09-11 12:31:20 +0000
603
@@ -18,13 +18,17 @@
604
18
18
605
19
# all of this deals with byte strings so this is safe
19
# all of this deals with byte strings so this is safe
606
20
from cStringIO import StringIO
20
from cStringIO import StringIO
607
21
import errno
608
21
import os
22
import os
609
22
import socket
23
import socket
610
24
import subprocess
611
25
import sys
612
23
import threading
26
import threading
613
24
27
614
25
import bzrlib
28
import bzrlib
615
26
from bzrlib import (
29
from bzrlib import (
616
27
        bzrdir,
30
        bzrdir,
617
31
        debug,
618
28
        errors,
32
        errors,
619
29
        osutils,
33
        osutils,
620
30
        tests,
34
        tests,
621
@@ -49,6 +53,29 @@
622
49
from bzrlib.transport.http import SmartClientHTTPMediumRequest
53
from bzrlib.transport.http import SmartClientHTTPMediumRequest
623
50
54
624
51
55
625
56
def create_file_pipes():
626
57
    r, w = os.pipe()
627
58
    # These must be opened without buffering, or we get undefined results
628
59
    rf = os.fdopen(r, 'rb', 0)
629
60
    wf = os.fdopen(w, 'wb', 0)
630
61
    return rf, wf
631
62
632
63
633
64
def portable_socket_pair():
634
65
    """Return a pair of TCP sockets connected to each other.
635
66
636
67
    Unlike socket.socketpair, this should work on Windows.
637
68
    """
638
69
    listen_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
639
70
    listen_sock.bind(('127.0.0.1', 0))
640
71
    listen_sock.listen(1)
641
72
    client_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
642
73
    client_sock.connect(listen_sock.getsockname())
643
74
    server_sock, addr = listen_sock.accept()
644
75
    listen_sock.close()
645
76
    return server_sock, client_sock
646
77
647
78
648
52
class StringIOSSHVendor(object):
79
class StringIOSSHVendor(object):
649
53
    """A SSH vendor that uses StringIO to buffer writes and answer reads."""
80
    """A SSH vendor that uses StringIO to buffer writes and answer reads."""
650
54
81
651
@@ -63,6 +90,27 @@
652
63
        return StringIOSSHConnection(self)
90
        return StringIOSSHConnection(self)
653
64
91
654
65
92
655
93
class FirstRejectedStringIOSSHVendor(StringIOSSHVendor):
656
94
    """The first connection will be considered closed.
657
95
658
96
    The second connection will succeed normally.
659
97
    """
660
98
661
99
    def __init__(self, read_from, write_to, fail_at_write=True):
662
100
        super(FirstRejectedStringIOSSHVendor, self).__init__(read_from,
663
101
            write_to)
664
102
        self.fail_at_write = fail_at_write
665
103
        self._first = True
666
104
667
105
    def connect_ssh(self, username, password, host, port, command):
668
106
        self.calls.append(('connect_ssh', username, password, host, port,
669
107
            command))
670
108
        if self._first:
671
109
            self._first = False
672
110
            return ClosedSSHConnection(self)
673
111
        return StringIOSSHConnection(self)
674
112
675
113
676
66
class StringIOSSHConnection(object):
114
class StringIOSSHConnection(object):
677
67
    """A SSH connection that uses StringIO to buffer writes and answer reads."""
115
    """A SSH connection that uses StringIO to buffer writes and answer reads."""
678
68
116
679
@@ -71,11 +119,36 @@
680
71
119
681
72
    def close(self):
120
    def close(self):
682
73
        self.vendor.calls.append(('close', ))
121
        self.vendor.calls.append(('close', ))
683
122
        self.vendor.read_from.close()
684
123
        self.vendor.write_to.close()
685
74
124
686
75
    def get_filelike_channels(self):
125
    def get_filelike_channels(self):
687
76
        return self.vendor.read_from, self.vendor.write_to
126
        return self.vendor.read_from, self.vendor.write_to
688
77
127
689
78
128
690
129
class ClosedSSHConnection(object):
691
130
    """An SSH connection that just has closed channels."""
692
131
693
132
    def __init__(self, vendor):
694
133
        self.vendor = vendor
695
134
696
135
    def close(self):
697
136
        self.vendor.calls.append(('close', ))
698
137
699
138
    def get_filelike_channels(self):
700
139
        # We create matching pipes, and then close the ssh side
701
140
        bzr_read, ssh_write = create_file_pipes()
702
141
        # We always fail when bzr goes to read
703
142
        ssh_write.close()
704
143
        if self.vendor.fail_at_write:
705
144
            # If set, we'll also fail when bzr goes to write
706
145
            ssh_read, bzr_write = create_file_pipes()
707
146
            ssh_read.close()
708
147
        else:
709
148
            bzr_write = self.vendor.write_to
710
149
        return bzr_read, bzr_write
711
150
712
151
713
79
class _InvalidHostnameFeature(tests.Feature):
152
class _InvalidHostnameFeature(tests.Feature):
714
80
    """Does 'non_existent.invalid' fail to resolve?
153
    """Does 'non_existent.invalid' fail to resolve?
715
81
154
716
@@ -171,6 +244,91 @@
717
171
        client_medium._accept_bytes('abc')
244
        client_medium._accept_bytes('abc')
718
172
        self.assertEqual('abc', output.getvalue())
245
        self.assertEqual('abc', output.getvalue())
719
173
246
720
247
    def test_simple_pipes__accept_bytes_subprocess_closed(self):
721
248
        # It is unfortunate that we have to use Popen for this. However,
722
249
        # os.pipe() does not behave the same as subprocess.Popen().
723
250
        # On Windows, if you use os.pipe() and close the write side,
724
251
        # read.read() hangs. On Linux, read.read() returns the empty string.
725
252
        p = subprocess.Popen([sys.executable, '-c',
726
253
            'import sys\n'
727
254
            'sys.stdout.write(sys.stdin.read(4))\n'
728
255
            'sys.stdout.close()\n'],
729
256
            stdout=subprocess.PIPE, stdin=subprocess.PIPE)
730
257
        client_medium = medium.SmartSimplePipesClientMedium(
731
258
            p.stdout, p.stdin, 'base')
732
259
        client_medium._accept_bytes('abc\n')
733
260
        self.assertEqual('abc', client_medium._read_bytes(3))
734
261
        p.wait()
735
262
        # While writing to the underlying pipe,
736
263
        #   Windows py2.6.6 we get IOError(EINVAL)
737
264
        #   Lucid py2.6.5, we get IOError(EPIPE)
738
265
        # In both cases, it should be wrapped to ConnectionReset
739
266
        self.assertRaises(errors.ConnectionReset,
740
267
                          client_medium._accept_bytes, 'more')
741
268
742
269
    def test_simple_pipes__accept_bytes_pipe_closed(self):
743
270
        child_read, client_write = create_file_pipes()
744
271
        client_medium = medium.SmartSimplePipesClientMedium(
745
272
            None, client_write, 'base')
746
273
        client_medium._accept_bytes('abc\n')
747
274
        self.assertEqual('abc\n', child_read.read(4))
748
275
        # While writing to the underlying pipe,
749
276
        #   Windows py2.6.6 we get IOError(EINVAL)
750
277
        #   Lucid py2.6.5, we get IOError(EPIPE)
751
278
        # In both cases, it should be wrapped to ConnectionReset
752
279
        child_read.close()
753
280
        self.assertRaises(errors.ConnectionReset,
754
281
                          client_medium._accept_bytes, 'more')
755
282
756
283
    def test_simple_pipes__flush_pipe_closed(self):
757
284
        child_read, client_write = create_file_pipes()
758
285
        client_medium = medium.SmartSimplePipesClientMedium(
759
286
            None, client_write, 'base')
760
287
        client_medium._accept_bytes('abc\n')
761
288
        child_read.close()
762
289
        # Even though the pipe is closed, flush on the write side seems to be a
763
290
        # no-op, rather than a failure.
764
291
        client_medium._flush()
765
292
766
293
    def test_simple_pipes__flush_subprocess_closed(self):
767
294
        p = subprocess.Popen([sys.executable, '-c',
768
295
            'import sys\n'
769
296
            'sys.stdout.write(sys.stdin.read(4))\n'
770
297
            'sys.stdout.close()\n'],
771
298
            stdout=subprocess.PIPE, stdin=subprocess.PIPE)
772
299
        client_medium = medium.SmartSimplePipesClientMedium(
773
300
            p.stdout, p.stdin, 'base')
774
301
        client_medium._accept_bytes('abc\n')
775
302
        p.wait()
776
303
        # Even though the child process is dead, flush seems to be a no-op.
777
304
        client_medium._flush()
778
305
779
306
    def test_simple_pipes__read_bytes_pipe_closed(self):
780
307
        child_read, client_write = create_file_pipes()
781
308
        client_medium = medium.SmartSimplePipesClientMedium(
782
309
            child_read, client_write, 'base')
783
310
        client_medium._accept_bytes('abc\n')
784
311
        client_write.close()
785
312
        self.assertEqual('abc\n', client_medium._read_bytes(4))
786
313
        self.assertEqual('', client_medium._read_bytes(4))
787
314
788
315
    def test_simple_pipes__read_bytes_subprocess_closed(self):
789
316
        p = subprocess.Popen([sys.executable, '-c',
790
317
            'import sys\n'
791
318
            'if sys.platform == "win32":\n'
792
319
            '    import msvcrt, os\n'
793
320
            '    msvcrt.setmode(sys.stdout.fileno(), os.O_BINARY)\n'
794
321
            '    msvcrt.setmode(sys.stdin.fileno(), os.O_BINARY)\n'
795
322
            'sys.stdout.write(sys.stdin.read(4))\n'
796
323
            'sys.stdout.close()\n'],
797
324
            stdout=subprocess.PIPE, stdin=subprocess.PIPE)
798
325
        client_medium = medium.SmartSimplePipesClientMedium(
799
326
            p.stdout, p.stdin, 'base')
800
327
        client_medium._accept_bytes('abc\n')
801
328
        p.wait()
802
329
        self.assertEqual('abc\n', client_medium._read_bytes(4))
803
330
        self.assertEqual('', client_medium._read_bytes(4))
804
331
805
174
    def test_simple_pipes_client_disconnect_does_nothing(self):
332
    def test_simple_pipes_client_disconnect_does_nothing(self):
806
175
        # calling disconnect does nothing.
333
        # calling disconnect does nothing.
807
176
        input = StringIO()
334
        input = StringIO()
808
@@ -556,6 +714,28 @@
809
556
        request.finished_reading()
714
        request.finished_reading()
810
557
        self.assertRaises(errors.ReadingCompleted, request.read_bytes, None)
715
        self.assertRaises(errors.ReadingCompleted, request.read_bytes, None)
811
558
716
812
717
    def test_reset(self):
813
718
        server_sock, client_sock = portable_socket_pair()
814
719
        # TODO: Use SmartClientAlreadyConnectedSocketMedium for the versions of
815
720
        #       bzr where it exists.
816
721
        client_medium = medium.SmartTCPClientMedium(None, None, None)
817
722
        client_medium._socket = client_sock
818
723
        client_medium._connected = True
819
724
        req = client_medium.get_request()
820
725
        self.assertRaises(errors.TooManyConcurrentRequests,
821
726
            client_medium.get_request)
822
727
        client_medium.reset()
823
728
        # The stream should be reset, marked as disconnected, though ready for
824
729
        # us to make a new request
825
730
        self.assertFalse(client_medium._connected)
826
731
        self.assertIs(None, client_medium._socket)
827
732
        try:
828
733
            self.assertEqual('', client_sock.recv(1))
829
734
        except socket.error, e:
830
735
            if e.errno not in (errno.EBADF,):
831
736
                raise
832
737
        req = client_medium.get_request()
833
738
834
559
739
835
560
class RemoteTransportTests(TestCaseWithSmartMedium):
740
class RemoteTransportTests(TestCaseWithSmartMedium):
836
561
741
837
@@ -609,20 +789,6 @@
838
609
        super(TestSmartServerStreamMedium, self).setUp()
789
        super(TestSmartServerStreamMedium, self).setUp()
839
610
        self._captureVar('BZR_NO_SMART_VFS', None)
790
        self._captureVar('BZR_NO_SMART_VFS', None)
840
611
791
841
612
    def portable_socket_pair(self):
842
613
        """Return a pair of TCP sockets connected to each other.
843
614
844
615
        Unlike socket.socketpair, this should work on Windows.
845
616
        """
846
617
        listen_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
847
618
        listen_sock.bind(('127.0.0.1', 0))
848
619
        listen_sock.listen(1)
849
620
        client_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
850
621
        client_sock.connect(listen_sock.getsockname())
851
622
        server_sock, addr = listen_sock.accept()
852
623
        listen_sock.close()
853
624
        return server_sock, client_sock
854
625
855
626
    def test_smart_query_version(self):
792
    def test_smart_query_version(self):
856
627
        """Feed a canned query version to a server"""
793
        """Feed a canned query version to a server"""
857
628
        # wire-to-wire, using the whole stack
794
        # wire-to-wire, using the whole stack
858
@@ -687,7 +853,7 @@
859
687
853
860
688
    def test_socket_stream_with_bulk_data(self):
854
    def test_socket_stream_with_bulk_data(self):
861
689
        sample_request_bytes = 'command\n9\nbulk datadone\n'
855
        sample_request_bytes = 'command\n9\nbulk datadone\n'
863
690
        server_sock, client_sock = self.portable_socket_pair()
856
        server_sock, client_sock = portable_socket_pair()
864
691
        server = medium.SmartServerSocketStreamMedium(
857
        server = medium.SmartServerSocketStreamMedium(
865
692
            server_sock, None)
858
            server_sock, None)
866
693
        sample_protocol = SampleRequest(expected_bytes=sample_request_bytes)
859
        sample_protocol = SampleRequest(expected_bytes=sample_request_bytes)
867
@@ -706,7 +872,7 @@
868
706
        self.assertTrue(server.finished)
872
        self.assertTrue(server.finished)
869
707
873
870
708
    def test_socket_stream_shutdown_detection(self):
874
    def test_socket_stream_shutdown_detection(self):
872
709
        server_sock, client_sock = self.portable_socket_pair()
875
        server_sock, client_sock = portable_socket_pair()
873
710
        client_sock.close()
876
        client_sock.close()
874
711
        server = medium.SmartServerSocketStreamMedium(
877
        server = medium.SmartServerSocketStreamMedium(
875
712
            server_sock, None)
878
            server_sock, None)
876
@@ -726,7 +892,7 @@
877
726
        rest_of_request_bytes = 'lo\n'
892
        rest_of_request_bytes = 'lo\n'
878
727
        expected_response = (
893
        expected_response = (
879
728
            protocol.RESPONSE_VERSION_TWO + 'success\nok\x012\n')
894
            protocol.RESPONSE_VERSION_TWO + 'success\nok\x012\n')
881
729
        server_sock, client_sock = self.portable_socket_pair()
895
        server_sock, client_sock = portable_socket_pair()
882
730
        server = medium.SmartServerSocketStreamMedium(
896
        server = medium.SmartServerSocketStreamMedium(
883
731
            server_sock, None)
897
            server_sock, None)
884
732
        client_sock.sendall(incomplete_request_bytes)
898
        client_sock.sendall(incomplete_request_bytes)
885
@@ -802,7 +968,7 @@
886
802
        # _serve_one_request should still process both of them as if they had
968
        # _serve_one_request should still process both of them as if they had
887
803
        # been received separately.
969
        # been received separately.
888
804
        sample_request_bytes = 'command\n'
970
        sample_request_bytes = 'command\n'
890
805
        server_sock, client_sock = self.portable_socket_pair()
971
        server_sock, client_sock = portable_socket_pair()
891
806
        server = medium.SmartServerSocketStreamMedium(
972
        server = medium.SmartServerSocketStreamMedium(
892
807
            server_sock, None)
973
            server_sock, None)
893
808
        first_protocol = SampleRequest(expected_bytes=sample_request_bytes)
974
        first_protocol = SampleRequest(expected_bytes=sample_request_bytes)
894
@@ -839,7 +1005,7 @@
895
839
        self.assertTrue(server.finished)
1005
        self.assertTrue(server.finished)
896
840
1006
897
841
    def test_socket_stream_error_handling(self):
1007
    def test_socket_stream_error_handling(self):
899
842
        server_sock, client_sock = self.portable_socket_pair()
1008
        server_sock, client_sock = portable_socket_pair()
900
843
        server = medium.SmartServerSocketStreamMedium(
1009
        server = medium.SmartServerSocketStreamMedium(
901
844
            server_sock, None)
1010
            server_sock, None)
902
845
        fake_protocol = ErrorRaisingProtocol(Exception('boom'))
1011
        fake_protocol = ErrorRaisingProtocol(Exception('boom'))
903
@@ -860,7 +1026,7 @@
904
860
        self.assertEqual('', from_server.getvalue())
1026
        self.assertEqual('', from_server.getvalue())
905
861
1027
906
862
    def test_socket_stream_keyboard_interrupt_handling(self):
1028
    def test_socket_stream_keyboard_interrupt_handling(self):
908
863
        server_sock, client_sock = self.portable_socket_pair()
1029
        server_sock, client_sock = portable_socket_pair()
909
864
        server = medium.SmartServerSocketStreamMedium(
1030
        server = medium.SmartServerSocketStreamMedium(
910
865
            server_sock, None)
1031
            server_sock, None)
911
866
        fake_protocol = ErrorRaisingProtocol(KeyboardInterrupt('boom'))
1032
        fake_protocol = ErrorRaisingProtocol(KeyboardInterrupt('boom'))
912
@@ -877,7 +1043,7 @@
913
877
        return server._build_protocol()
1043
        return server._build_protocol()
914
878
1044
915
879
    def build_protocol_socket(self, bytes):
1045
    def build_protocol_socket(self, bytes):
917
880
        server_sock, client_sock = self.portable_socket_pair()
1046
        server_sock, client_sock = portable_socket_pair()
918
881
        server = medium.SmartServerSocketStreamMedium(
1047
        server = medium.SmartServerSocketStreamMedium(
919
882
            server_sock, None)
1048
            server_sock, None)
920
883
        client_sock.sendall(bytes)
1049
        client_sock.sendall(bytes)
921
@@ -3214,6 +3380,114 @@
922
3214
        # encoder.
3380
        # encoder.
923
3215
3381
924
3216
3382
925
3383
class Test_SmartClientRequest(tests.TestCase):
926
3384
927
3385
    def make_client_with_failing_medium(self, fail_at_write=True):
928
3386
        response = StringIO()
929
3387
        output = StringIO()
930
3388
        vendor = FirstRejectedStringIOSSHVendor(response, output,
931
3389
                    fail_at_write=fail_at_write)
932
3390
        client_medium = medium.SmartSSHClientMedium(
933
3391
            'a host', 'a port', 'a user', 'a pass', 'base', vendor,
934
3392
            'bzr')
935
3393
        smart_client = client._SmartClient(client_medium, headers={})
936
3394
        return output, vendor, smart_client
937
3395
938
3396
    def test__send_no_retry_pipes(self):
939
3397
        client_read, server_write = create_file_pipes()
940
3398
        server_read, client_write = create_file_pipes()
941
3399
        client_medium = medium.SmartSimplePipesClientMedium(client_read,
942
3400
            client_write, base='/')
943
3401
        smart_client = client._SmartClient(client_medium)
944
3402
        smart_request = client._SmartClientRequest(smart_client,
945
3403
            'hello', ())
946
3404
        # Close the server side
947
3405
        server_read.close()
948
3406
        encoder, response_handler = smart_request._construct_protocol(3)
949
3407
        self.assertRaises(errors.ConnectionReset,
950
3408
            smart_request._send_no_retry, encoder)
951
3409
952
3410
    def test__send_read_response_sockets(self):
953
3411
        listen_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
954
3412
        listen_sock.bind(('127.0.0.1', 0))
955
3413
        listen_sock.listen(1)
956
3414
        host, port = listen_sock.getsockname()
957
3415
        client_medium = medium.SmartTCPClientMedium(host, port, '/')
958
3416
        client_medium._ensure_connection()
959
3417
        smart_client = client._SmartClient(client_medium)
960
3418
        smart_request = client._SmartClientRequest(smart_client, 'hello', ())
961
3419
        # Accept the connection, but don't actually talk to the client.
962
3420
        server_sock, _ = listen_sock.accept()
963
3421
        server_sock.close()
964
3422
        # Sockets buffer and don't really notice that the server has closed the
965
3423
        # connection until we try to read again.
966
3424
        handler = smart_request._send(3)
967
3425
        self.assertRaises(errors.ConnectionReset,
968
3426
            handler.read_response_tuple, expect_body=False)
969
3427
970
3428
    def test__send_retries_on_write(self):
971
3429
        output, vendor, smart_client = self.make_client_with_failing_medium()
972
3430
        smart_request = client._SmartClientRequest(smart_client, 'hello', ())
973
3431
        handler = smart_request._send(3)
974
3432
        self.assertEqual('bzr message 3 (bzr 1.6)\n' # protocol
975
3433
                         '\x00\x00\x00\x02de'   # empty headers
976
3434
                         's\x00\x00\x00\tl5:helloee',
977
3435
                         output.getvalue())
978
3436
        self.assertEqual(
979
3437
            [('connect_ssh', 'a user', 'a pass', 'a host', 'a port',
980
3438
              ['bzr', 'serve', '--inet', '--directory=/', '--allow-writes']),
981
3439
             ('close',),
982
3440
             ('connect_ssh', 'a user', 'a pass', 'a host', 'a port',
983
3441
              ['bzr', 'serve', '--inet', '--directory=/', '--allow-writes']),
984
3442
            ],
985
3443
            vendor.calls)
986
3444
987
3445
    def test__send_doesnt_retry_read_failure(self):
988
3446
        output, vendor, smart_client = self.make_client_with_failing_medium(
989
3447
            fail_at_write=False)
990
3448
        smart_request = client._SmartClientRequest(smart_client, 'hello', ())
991
3449
        handler = smart_request._send(3)
992
3450
        self.assertEqual('bzr message 3 (bzr 1.6)\n' # protocol
993
3451
                         '\x00\x00\x00\x02de'   # empty headers
994
3452
                         's\x00\x00\x00\tl5:helloee',
995
3453
                         output.getvalue())
996
3454
        self.assertEqual(
997
3455
            [('connect_ssh', 'a user', 'a pass', 'a host', 'a port',
998
3456
              ['bzr', 'serve', '--inet', '--directory=/', '--allow-writes']),
999
3457
            ],
1000
3458
            vendor.calls)
1001
3459
        self.assertRaises(errors.ConnectionReset, handler.read_response_tuple)
1002
3460
1003
3461
    def test__send_doesnt_retry_body_stream(self):
1004
3462
        # We don't know how much of body_stream would get iterated as part of
1005
3463
        # _send before it failed to actually send the request, so we
1006
3464
        # just always fail in this condition.
1007
3465
        output, vendor, smart_client = self.make_client_with_failing_medium()
1008
3466
        smart_request = client._SmartClientRequest(smart_client, 'hello', (),
1009
3467
            body_stream=['a', 'b'])
1010
3468
        self.assertRaises(errors.ConnectionReset, smart_request._send, 3)
1011
3469
        # We got one connect, but it fails, so we disconnect, but we don't
1012
3470
        # retry it
1013
3471
        self.assertEqual(
1014
3472
            [('connect_ssh', 'a user', 'a pass', 'a host', 'a port',
1015
3473
              ['bzr', 'serve', '--inet', '--directory=/', '--allow-writes']),
1016
3474
             ('close',),
1017
3475
            ],
1018
3476
            vendor.calls)
1019
3477
1020
3478
    def test__send_disabled_retry(self):
1021
3479
        debug.debug_flags.add('noretry')
1022
3480
        output, vendor, smart_client = self.make_client_with_failing_medium()
1023
3481
        smart_request = client._SmartClientRequest(smart_client, 'hello', ())
1024
3482
        self.assertRaises(errors.ConnectionReset, smart_request._send, 3)
1025
3483
        self.assertEqual(
1026
3484
            [('connect_ssh', 'a user', 'a pass', 'a host', 'a port',
1027
3485
              ['bzr', 'serve', '--inet', '--directory=/', '--allow-writes']),
1028
3486
             ('close',),
1029
3487
            ],
1030
3488
            vendor.calls)
1031
3489
1032
3490
1033
3217
class LengthPrefixedBodyDecoder(tests.TestCase):
3491
class LengthPrefixedBodyDecoder(tests.TestCase):
1034
3218
3492
1035
3219
    # XXX: TODO: make accept_reading_trailer invoke translate_response or
3493
    # XXX: TODO: make accept_reading_trailer invoke translate_response or
Reviewer	Review Type	Date Requested	Status
bzr-core		2011-10-07	Pending
Review via email: mp+78597@code.launchpad.net