« back to all changes in this revision
Viewing changes to bzrlib/smart/client.py
- browse files at revision 6206.1.1
- compare with another revision
- compare with revision 6006.4.4
- download diff
- download tarball
- view history from revision 6206.1.1
- Committer: John Arbash Meinel
- Date: 2011-10-10 14:55:12 UTC
- mfrom: (6015.49.2 2.4-client-reconnect-819604)
- mto: This revision was merged to the branch mainline in revision 6321.
- Revision ID: john@arbash-meinel.com-20111010145512-zuzpcmhyz34n0xjc
Merge the fix for bug #819604 into trunk, resolve conflicts.
- files modified:
- bzrlib/help_topics/en/debug-flags.txt
added
removed
bzrlib/smart/client.py
14
14
# along with this program; if not, write to the Free Software
15
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
16
16
17
from bzrlib import lazy_import
18
lazy_import.lazy_import(globals(), """
19
from bzrlib.smart import request as _mod_request
20
""")
21
17
22
import bzrlib
18
23
from bzrlib.smart import message, protocol
19
from bzrlib.trace import warning
20
24
from bzrlib import (
25
debug,
21
26
errors,
22
27
hooks,
28
trace,
23
29
)
24
30
25
31
39
45
def __repr__(self):
40
46
return '%s(%r)' % (self.__class__.__name__, self._medium)
41
47
42
def _send_request(self, protocol_version, method, args, body=None,
43
readv_body=None, body_stream=None):
44
encoder, response_handler = self._construct_protocol(
45
protocol_version)
46
encoder.set_headers(self._headers)
47
if body is not None:
48
if readv_body is not None:
49
raise AssertionError(
50
"body and readv_body are mutually exclusive.")
51
if body_stream is not None:
52
raise AssertionError(
53
"body and body_stream are mutually exclusive.")
54
encoder.call_with_body_bytes((method, ) + args, body)
55
elif readv_body is not None:
56
if body_stream is not None:
57
raise AssertionError(
58
"readv_body and body_stream are mutually exclusive.")
59
encoder.call_with_body_readv_array((method, ) + args, readv_body)
60
elif body_stream is not None:
61
encoder.call_with_body_stream((method, ) + args, body_stream)
62
else:
63
encoder.call(method, *args)
64
return response_handler
65
66
def _run_call_hooks(self, method, args, body, readv_body):
67
if not _SmartClient.hooks['call']:
68
return
69
params = CallHookParams(method, args, body, readv_body, self._medium)
70
for hook in _SmartClient.hooks['call']:
71
hook(params)
72
73
48
def _call_and_read_response(self, method, args, body=None, readv_body=None,
74
49
body_stream=None, expect_response_body=True):
75
self._run_call_hooks(method, args, body, readv_body)
76
if self._medium._protocol_version is not None:
77
response_handler = self._send_request(
78
self._medium._protocol_version, method, args, body=body,
79
readv_body=readv_body, body_stream=body_stream)
80
return (response_handler.read_response_tuple(
81
expect_body=expect_response_body),
82
response_handler)
83
else:
84
for protocol_version in [3, 2]:
85
if protocol_version == 2:
86
# If v3 doesn't work, the remote side is older than 1.6.
87
self._medium._remember_remote_is_before((1, 6))
88
response_handler = self._send_request(
89
protocol_version, method, args, body=body,
90
readv_body=readv_body, body_stream=body_stream)
91
try:
92
response_tuple = response_handler.read_response_tuple(
93
expect_body=expect_response_body)
94
except errors.UnexpectedProtocolVersionMarker, err:
95
# TODO: We could recover from this without disconnecting if
96
# we recognise the protocol version.
97
warning(
98
'Server does not understand Bazaar network protocol %d,'
99
' reconnecting. (Upgrade the server to avoid this.)'
100
% (protocol_version,))
101
self._medium.disconnect()
102
continue
103
except errors.ErrorFromSmartServer:
104
# If we received an error reply from the server, then it
105
# must be ok with this protocol version.
106
self._medium._protocol_version = protocol_version
107
raise
108
else:
109
self._medium._protocol_version = protocol_version
110
return response_tuple, response_handler
111
raise errors.SmartProtocolError(
112
'Server is not a Bazaar server: ' + str(err))
113
114
def _construct_protocol(self, version):
115
request = self._medium.get_request()
116
if version == 3:
117
request_encoder = protocol.ProtocolThreeRequester(request)
118
response_handler = message.ConventionalResponseHandler()
119
response_proto = protocol.ProtocolThreeDecoder(
120
response_handler, expect_version_marker=True)
121
response_handler.setProtoAndMediumRequest(response_proto, request)
122
elif version == 2:
123
request_encoder = protocol.SmartClientRequestProtocolTwo(request)
124
response_handler = request_encoder
125
else:
126
request_encoder = protocol.SmartClientRequestProtocolOne(request)
127
response_handler = request_encoder
128
return request_encoder, response_handler
50
request = _SmartClientRequest(self, method, args, body=body,
51
readv_body=readv_body, body_stream=body_stream,
52
expect_response_body=expect_response_body)
53
return request.call_and_read_response()
129
54
130
55
def call(self, method, *args):
131
56
"""Call a method on the remote server."""
191
116
return self._medium.remote_path_from_transport(transport)
192
117
193
118
119
class _SmartClientRequest(object):
120
"""Encapsulate the logic for a single request.
121
122
This class handles things like reconnecting and sending the request a
123
second time when the connection is reset in the middle. It also handles the
124
multiple requests that get made if we don't know what protocol the server
125
supports yet.
126
127
Generally, you build up one of these objects, passing in the arguments that
128
you want to send to the server, and then use 'call_and_read_response' to
129
get the response from the server.
130
"""
131
132
def __init__(self, client, method, args, body=None, readv_body=None,
133
body_stream=None, expect_response_body=True):
134
self.client = client
135
self.method = method
136
self.args = args
137
self.body = body
138
self.readv_body = readv_body
139
self.body_stream = body_stream
140
self.expect_response_body = expect_response_body
141
142
def call_and_read_response(self):
143
"""Send the request to the server, and read the initial response.
144
145
This doesn't read all of the body content of the response, instead it
146
returns (response_tuple, response_handler). response_tuple is the 'ok',
147
or 'error' information, and 'response_handler' can be used to get the
148
content stream out.
149
"""
150
self._run_call_hooks()
151
protocol_version = self.client._medium._protocol_version
152
if protocol_version is None:
153
return self._call_determining_protocol_version()
154
else:
155
return self._call(protocol_version)
156
157
def _is_safe_to_send_twice(self):
158
"""Check if the current method is re-entrant safe."""
159
if self.body_stream is not None or 'noretry' in debug.debug_flags:
160
# We can't restart a body stream that has already been consumed.
161
return False
162
request_type = _mod_request.request_handlers.get_info(self.method)
163
if request_type in ('read', 'idem', 'semi'):
164
return True
165
# If we have gotten this far, 'stream' cannot be retried, because we
166
# already consumed the local stream.
167
if request_type in ('semivfs', 'mutate', 'stream'):
168
return False
169
trace.mutter('Unknown request type: %s for method %s'
170
% (request_type, self.method))
171
return False
172
173
def _run_call_hooks(self):
174
if not _SmartClient.hooks['call']:
175
return
176
params = CallHookParams(self.method, self.args, self.body,
177
self.readv_body, self.client._medium)
178
for hook in _SmartClient.hooks['call']:
179
hook(params)
180
181
def _call(self, protocol_version):
182
"""We know the protocol version.
183
184
So this just sends the request, and then reads the response. This is
185
where the code will be to retry requests if the connection is closed.
186
"""
187
response_handler = self._send(protocol_version)
188
try:
189
response_tuple = response_handler.read_response_tuple(
190
expect_body=self.expect_response_body)
191
except errors.ConnectionReset, e:
192
self.client._medium.reset()
193
if not self._is_safe_to_send_twice():
194
raise
195
trace.warning('ConnectionReset reading response for %r, retrying'
196
% (self.method,))
197
trace.log_exception_quietly()
198
encoder, response_handler = self._construct_protocol(
199
protocol_version)
200
self._send_no_retry(encoder)
201
response_tuple = response_handler.read_response_tuple(
202
expect_body=self.expect_response_body)
203
return (response_tuple, response_handler)
204
205
def _call_determining_protocol_version(self):
206
"""Determine what protocol the remote server supports.
207
208
We do this by placing a request in the most recent protocol, and
209
handling the UnexpectedProtocolVersionMarker from the server.
210
"""
211
for protocol_version in [3, 2]:
212
if protocol_version == 2:
213
# If v3 doesn't work, the remote side is older than 1.6.
214
self.client._medium._remember_remote_is_before((1, 6))
215
try:
216
response_tuple, response_handler = self._call(protocol_version)
217
except errors.UnexpectedProtocolVersionMarker, err:
218
# TODO: We could recover from this without disconnecting if
219
# we recognise the protocol version.
220
trace.warning(
221
'Server does not understand Bazaar network protocol %d,'
222
' reconnecting. (Upgrade the server to avoid this.)'
223
% (protocol_version,))
224
self.client._medium.disconnect()
225
continue
226
except errors.ErrorFromSmartServer:
227
# If we received an error reply from the server, then it
228
# must be ok with this protocol version.
229
self.client._medium._protocol_version = protocol_version
230
raise
231
else:
232
self.client._medium._protocol_version = protocol_version
233
return response_tuple, response_handler
234
raise errors.SmartProtocolError(
235
'Server is not a Bazaar server: ' + str(err))
236
237
def _construct_protocol(self, version):
238
"""Build the encoding stack for a given protocol version."""
239
request = self.client._medium.get_request()
240
if version == 3:
241
request_encoder = protocol.ProtocolThreeRequester(request)
242
response_handler = message.ConventionalResponseHandler()
243
response_proto = protocol.ProtocolThreeDecoder(
244
response_handler, expect_version_marker=True)
245
response_handler.setProtoAndMediumRequest(response_proto, request)
246
elif version == 2:
247
request_encoder = protocol.SmartClientRequestProtocolTwo(request)
248
response_handler = request_encoder
249
else:
250
request_encoder = protocol.SmartClientRequestProtocolOne(request)
251
response_handler = request_encoder
252
return request_encoder, response_handler
253
254
def _send(self, protocol_version):
255
"""Encode the request, and send it to the server.
256
257
This will retry a request if we get a ConnectionReset while sending the
258
request to the server. (Unless we have a body_stream that we have
259
already started consuming, since we can't restart body_streams)
260
261
:return: response_handler as defined by _construct_protocol
262
"""
263
encoder, response_handler = self._construct_protocol(protocol_version)
264
try:
265
self._send_no_retry(encoder)
266
except errors.ConnectionReset, e:
267
# If we fail during the _send_no_retry phase, then we can
268
# be confident that the server did not get our request, because we
269
# haven't started waiting for the reply yet. So try the request
270
# again. We only issue a single retry, because if the connection
271
# really is down, there is no reason to loop endlessly.
272
273
# Connection is dead, so close our end of it.
274
self.client._medium.reset()
275
if (('noretry' in debug.debug_flags)
276
or (self.body_stream is not None
277
and encoder.body_stream_started)):
278
# We can't restart a body_stream that has been partially
279
# consumed, so we don't retry.
280
# Note: We don't have to worry about
281
# SmartClientRequestProtocolOne or Two, because they don't
282
# support client-side body streams.
283
raise
284
trace.warning('ConnectionReset calling %r, retrying'
285
% (self.method,))
286
trace.log_exception_quietly()
287
encoder, response_handler = self._construct_protocol(
288
protocol_version)
289
self._send_no_retry(encoder)
290
return response_handler
291
292
def _send_no_retry(self, encoder):
293
"""Just encode the request and try to send it."""
294
encoder.set_headers(self.client._headers)
295
if self.body is not None:
296
if self.readv_body is not None:
297
raise AssertionError(
298
"body and readv_body are mutually exclusive.")
299
if self.body_stream is not None:
300
raise AssertionError(
301
"body and body_stream are mutually exclusive.")
302
encoder.call_with_body_bytes((self.method, ) + self.args, self.body)
303
elif self.readv_body is not None:
304
if self.body_stream is not None:
305
raise AssertionError(
306
"readv_body and body_stream are mutually exclusive.")
307
encoder.call_with_body_readv_array((self.method, ) + self.args,
308
self.readv_body)
309
elif self.body_stream is not None:
310
encoder.call_with_body_stream((self.method, ) + self.args,
311
self.body_stream)
312
else:
313
encoder.call(self.method, *self.args)
314
315
194
316
class SmartClientHooks(hooks.Hooks):
195
317
196
318
def __init__(self):
Loggerhead is a web-based interface for Breezy
Version: 2.0.1