1
# Copyright (C) 2006, 2007 Canonical Ltd
3
# This program is free software; you can redistribute it and/or modify
4
# it under the terms of the GNU General Public License as published by
5
# the Free Software Foundation; either version 2 of the License, or
6
# (at your option) any later version.
8
# This program is distributed in the hope that it will be useful,
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11
# GNU General Public License for more details.
13
# You should have received a copy of the GNU General Public License
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
17
"""Wire-level encoding and decoding of requests and responses for the smart
22
from cStringIO import StringIO
25
from bzrlib import debug
26
from bzrlib import errors
27
from bzrlib.smart import request
28
from bzrlib.trace import log_exception_quietly, mutter
31
# Protocol version strings. These are sent as prefixes of bzr requests and
32
# responses to identify the protocol version being used. (There are no version
33
# one strings because that version doesn't send any).
34
REQUEST_VERSION_TWO = 'bzr request 2\n'
35
RESPONSE_VERSION_TWO = 'bzr response 2\n'
38
def _recv_tuple(from_file):
39
req_line = from_file.readline()
40
return _decode_tuple(req_line)
43
def _decode_tuple(req_line):
44
if req_line == None or req_line == '':
46
if req_line[-1] != '\n':
47
raise errors.SmartProtocolError("request %r not terminated" % req_line)
48
return tuple(req_line[:-1].split('\x01'))
51
def _encode_tuple(args):
52
"""Encode the tuple args to a bytestream."""
53
return '\x01'.join(args) + '\n'
56
class SmartProtocolBase(object):
57
"""Methods common to client and server"""
59
# TODO: this only actually accomodates a single block; possibly should
60
# support multiple chunks?
61
def _encode_bulk_data(self, body):
62
"""Encode body as a bulk data chunk."""
63
return ''.join(('%d\n' % len(body), body, 'done\n'))
65
def _serialise_offsets(self, offsets):
66
"""Serialise a readv offset list."""
68
for start, length in offsets:
69
txt.append('%d,%d' % (start, length))
73
class SmartServerRequestProtocolOne(SmartProtocolBase):
74
"""Server-side encoding and decoding logic for smart version 1."""
76
def __init__(self, backing_transport, write_func, root_client_path='/'):
77
self._backing_transport = backing_transport
78
self._root_client_path = root_client_path
79
self.excess_buffer = ''
80
self._finished = False
82
self.has_dispatched = False
84
self._body_decoder = None
85
self._write_func = write_func
87
def accept_bytes(self, bytes):
88
"""Take bytes, and advance the internal state machine appropriately.
90
:param bytes: must be a byte string
92
assert isinstance(bytes, str)
93
self.in_buffer += bytes
94
if not self.has_dispatched:
95
if '\n' not in self.in_buffer:
98
self.has_dispatched = True
100
first_line, self.in_buffer = self.in_buffer.split('\n', 1)
102
req_args = _decode_tuple(first_line)
103
self.request = request.SmartServerRequestHandler(
104
self._backing_transport, commands=request.request_handlers,
105
root_client_path=self._root_client_path)
106
self.request.dispatch_command(req_args[0], req_args[1:])
107
if self.request.finished_reading:
109
self.excess_buffer = self.in_buffer
111
self._send_response(self.request.response)
112
except KeyboardInterrupt:
114
except Exception, exception:
115
# everything else: pass to client, flush, and quit
116
log_exception_quietly()
117
self._send_response(request.FailedSmartServerResponse(
118
('error', str(exception))))
121
if self.has_dispatched:
123
# nothing to do.XXX: this routine should be a single state
125
self.excess_buffer += self.in_buffer
128
if self._body_decoder is None:
129
self._body_decoder = LengthPrefixedBodyDecoder()
130
self._body_decoder.accept_bytes(self.in_buffer)
131
self.in_buffer = self._body_decoder.unused_data
132
body_data = self._body_decoder.read_pending_data()
133
self.request.accept_body(body_data)
134
if self._body_decoder.finished_reading:
135
self.request.end_of_body()
136
assert self.request.finished_reading, \
137
"no more body, request not finished"
138
if self.request.response is not None:
139
self._send_response(self.request.response)
140
self.excess_buffer = self.in_buffer
143
assert not self.request.finished_reading, \
144
"no response and we have finished reading."
146
def _send_response(self, response):
147
"""Send a smart server response down the output stream."""
148
assert not self._finished, 'response already sent'
151
self._finished = True
152
self._write_protocol_version()
153
self._write_success_or_failure_prefix(response)
154
self._write_func(_encode_tuple(args))
156
assert isinstance(body, str), 'body must be a str'
157
bytes = self._encode_bulk_data(body)
158
self._write_func(bytes)
160
def _write_protocol_version(self):
161
"""Write any prefixes this protocol requires.
163
Version one doesn't send protocol versions.
166
def _write_success_or_failure_prefix(self, response):
167
"""Write the protocol specific success/failure prefix.
169
For SmartServerRequestProtocolOne this is omitted but we
170
call is_successful to ensure that the response is valid.
172
response.is_successful()
174
def next_read_size(self):
177
if self._body_decoder is None:
180
return self._body_decoder.next_read_size()
183
class SmartServerRequestProtocolTwo(SmartServerRequestProtocolOne):
184
r"""Version two of the server side of the smart protocol.
186
This prefixes responses with the value of RESPONSE_VERSION_TWO.
189
def _write_success_or_failure_prefix(self, response):
190
"""Write the protocol specific success/failure prefix."""
191
if response.is_successful():
192
self._write_func('success\n')
194
self._write_func('failed\n')
196
def _write_protocol_version(self):
197
r"""Write any prefixes this protocol requires.
199
Version two sends the value of RESPONSE_VERSION_TWO.
201
self._write_func(RESPONSE_VERSION_TWO)
203
def _send_response(self, response):
204
"""Send a smart server response down the output stream."""
205
assert not self._finished, 'response already sent'
206
self._finished = True
207
self._write_protocol_version()
208
self._write_success_or_failure_prefix(response)
209
self._write_func(_encode_tuple(response.args))
210
if response.body is not None:
211
assert isinstance(response.body, str), 'body must be a str'
212
assert response.body_stream is None, (
213
'body_stream and body cannot both be set')
214
bytes = self._encode_bulk_data(response.body)
215
self._write_func(bytes)
216
elif response.body_stream is not None:
217
_send_stream(response.body_stream, self._write_func)
220
def _send_stream(stream, write_func):
221
write_func('chunked\n')
222
_send_chunks(stream, write_func)
226
def _send_chunks(stream, write_func):
228
if isinstance(chunk, str):
229
bytes = "%x\n%s" % (len(chunk), chunk)
231
elif isinstance(chunk, request.FailedSmartServerResponse):
233
_send_chunks(chunk.args, write_func)
236
raise errors.BzrError(
237
'Chunks must be str or FailedSmartServerResponse, got %r'
241
class _StatefulDecoder(object):
244
self.finished_reading = False
245
self.unused_data = ''
246
self.bytes_left = None
248
def accept_bytes(self, bytes):
249
"""Decode as much of bytes as possible.
251
If 'bytes' contains too much data it will be appended to
254
finished_reading will be set when no more data is required. Further
255
data will be appended to self.unused_data.
257
# accept_bytes is allowed to change the state
258
current_state = self.state_accept
259
self.state_accept(bytes)
260
while current_state != self.state_accept:
261
current_state = self.state_accept
262
self.state_accept('')
265
class ChunkedBodyDecoder(_StatefulDecoder):
266
"""Decoder for chunked body data.
268
This is very similar the HTTP's chunked encoding. See the description of
269
streamed body data in `doc/developers/network-protocol.txt` for details.
273
_StatefulDecoder.__init__(self)
274
self.state_accept = self._state_accept_expecting_header
276
self.chunk_in_progress = None
277
self.chunks = collections.deque()
279
self.error_in_progress = None
281
def next_read_size(self):
282
# Note: the shortest possible chunk is 2 bytes: '0\n', and the
283
# end-of-body marker is 4 bytes: 'END\n'.
284
if self.state_accept == self._state_accept_reading_chunk:
285
# We're expecting more chunk content. So we're expecting at least
286
# the rest of this chunk plus an END chunk.
287
return self.bytes_left + 4
288
elif self.state_accept == self._state_accept_expecting_length:
289
if self._in_buffer == '':
290
# We're expecting a chunk length. There's at least two bytes
291
# left: a digit plus '\n'.
294
# We're in the middle of reading a chunk length. So there's at
295
# least one byte left, the '\n' that terminates the length.
297
elif self.state_accept == self._state_accept_reading_unused:
299
elif self.state_accept == self._state_accept_expecting_header:
300
return max(0, len('chunked\n') - len(self._in_buffer))
302
raise AssertionError("Impossible state: %r" % (self.state_accept,))
304
def read_next_chunk(self):
306
return self.chunks.popleft()
310
def _extract_line(self):
311
pos = self._in_buffer.find('\n')
313
# We haven't read a complete length prefix yet, so there's nothing
316
line = self._in_buffer[:pos]
317
# Trim the prefix (including '\n' delimiter) from the _in_buffer.
318
self._in_buffer = self._in_buffer[pos+1:]
322
self.unused_data = self._in_buffer
323
self._in_buffer = None
324
self.state_accept = self._state_accept_reading_unused
326
error_args = tuple(self.error_in_progress)
327
self.chunks.append(request.FailedSmartServerResponse(error_args))
328
self.error_in_progress = None
329
self.finished_reading = True
331
def _state_accept_expecting_header(self, bytes):
332
self._in_buffer += bytes
333
prefix = self._extract_line()
335
# We haven't read a complete length prefix yet, so there's nothing
338
elif prefix == 'chunked':
339
self.state_accept = self._state_accept_expecting_length
341
raise errors.SmartProtocolError(
342
'Bad chunked body header: "%s"' % (prefix,))
344
def _state_accept_expecting_length(self, bytes):
345
self._in_buffer += bytes
346
prefix = self._extract_line()
348
# We haven't read a complete length prefix yet, so there's nothing
351
elif prefix == 'ERR':
353
self.error_in_progress = []
354
self._state_accept_expecting_length('')
356
elif prefix == 'END':
357
# We've read the end-of-body marker.
358
# Any further bytes are unused data, including the bytes left in
363
self.bytes_left = int(prefix, 16)
364
self.chunk_in_progress = ''
365
self.state_accept = self._state_accept_reading_chunk
367
def _state_accept_reading_chunk(self, bytes):
368
self._in_buffer += bytes
369
in_buffer_len = len(self._in_buffer)
370
self.chunk_in_progress += self._in_buffer[:self.bytes_left]
371
self._in_buffer = self._in_buffer[self.bytes_left:]
372
self.bytes_left -= in_buffer_len
373
if self.bytes_left <= 0:
374
# Finished with chunk
375
self.bytes_left = None
377
self.error_in_progress.append(self.chunk_in_progress)
379
self.chunks.append(self.chunk_in_progress)
380
self.chunk_in_progress = None
381
self.state_accept = self._state_accept_expecting_length
383
def _state_accept_reading_unused(self, bytes):
384
self.unused_data += bytes
387
class LengthPrefixedBodyDecoder(_StatefulDecoder):
388
"""Decodes the length-prefixed bulk data."""
391
_StatefulDecoder.__init__(self)
392
self.state_accept = self._state_accept_expecting_length
393
self.state_read = self._state_read_no_data
395
self._trailer_buffer = ''
397
def next_read_size(self):
398
if self.bytes_left is not None:
399
# Ideally we want to read all the remainder of the body and the
401
return self.bytes_left + 5
402
elif self.state_accept == self._state_accept_reading_trailer:
403
# Just the trailer left
404
return 5 - len(self._trailer_buffer)
405
elif self.state_accept == self._state_accept_expecting_length:
406
# There's still at least 6 bytes left ('\n' to end the length, plus
410
# Reading excess data. Either way, 1 byte at a time is fine.
413
def read_pending_data(self):
414
"""Return any pending data that has been decoded."""
415
return self.state_read()
417
def _state_accept_expecting_length(self, bytes):
418
self._in_buffer += bytes
419
pos = self._in_buffer.find('\n')
422
self.bytes_left = int(self._in_buffer[:pos])
423
self._in_buffer = self._in_buffer[pos+1:]
424
self.bytes_left -= len(self._in_buffer)
425
self.state_accept = self._state_accept_reading_body
426
self.state_read = self._state_read_in_buffer
428
def _state_accept_reading_body(self, bytes):
429
self._in_buffer += bytes
430
self.bytes_left -= len(bytes)
431
if self.bytes_left <= 0:
433
if self.bytes_left != 0:
434
self._trailer_buffer = self._in_buffer[self.bytes_left:]
435
self._in_buffer = self._in_buffer[:self.bytes_left]
436
self.bytes_left = None
437
self.state_accept = self._state_accept_reading_trailer
439
def _state_accept_reading_trailer(self, bytes):
440
self._trailer_buffer += bytes
441
# TODO: what if the trailer does not match "done\n"? Should this raise
442
# a ProtocolViolation exception?
443
if self._trailer_buffer.startswith('done\n'):
444
self.unused_data = self._trailer_buffer[len('done\n'):]
445
self.state_accept = self._state_accept_reading_unused
446
self.finished_reading = True
448
def _state_accept_reading_unused(self, bytes):
449
self.unused_data += bytes
451
def _state_read_no_data(self):
454
def _state_read_in_buffer(self):
455
result = self._in_buffer
460
class SmartClientRequestProtocolOne(SmartProtocolBase):
461
"""The client-side protocol for smart version 1."""
463
def __init__(self, request):
464
"""Construct a SmartClientRequestProtocolOne.
466
:param request: A SmartClientMediumRequest to serialise onto and
469
self._request = request
470
self._body_buffer = None
471
self._request_start_time = None
472
self._last_verb = None
474
def call(self, *args):
475
if 'hpss' in debug.debug_flags:
476
mutter('hpss call: %s', repr(args)[1:-1])
477
if getattr(self._request._medium, 'base', None) is not None:
478
mutter(' (to %s)', self._request._medium.base)
479
self._request_start_time = time.time()
480
self._write_args(args)
481
self._request.finished_writing()
482
self._last_verb = args[0]
484
def call_with_body_bytes(self, args, body):
485
"""Make a remote call of args with body bytes 'body'.
487
After calling this, call read_response_tuple to find the result out.
489
if 'hpss' in debug.debug_flags:
490
mutter('hpss call w/body: %s (%r...)', repr(args)[1:-1], body[:20])
491
if getattr(self._request._medium, '_path', None) is not None:
492
mutter(' (to %s)', self._request._medium._path)
493
mutter(' %d bytes', len(body))
494
self._request_start_time = time.time()
495
if 'hpssdetail' in debug.debug_flags:
496
mutter('hpss body content: %s', body)
497
self._write_args(args)
498
bytes = self._encode_bulk_data(body)
499
self._request.accept_bytes(bytes)
500
self._request.finished_writing()
501
self._last_verb = args[0]
503
def call_with_body_readv_array(self, args, body):
504
"""Make a remote call with a readv array.
506
The body is encoded with one line per readv offset pair. The numbers in
507
each pair are separated by a comma, and no trailing \n is emitted.
509
if 'hpss' in debug.debug_flags:
510
mutter('hpss call w/readv: %s', repr(args)[1:-1])
511
if getattr(self._request._medium, '_path', None) is not None:
512
mutter(' (to %s)', self._request._medium._path)
513
self._request_start_time = time.time()
514
self._write_args(args)
515
readv_bytes = self._serialise_offsets(body)
516
bytes = self._encode_bulk_data(readv_bytes)
517
self._request.accept_bytes(bytes)
518
self._request.finished_writing()
519
if 'hpss' in debug.debug_flags:
520
mutter(' %d bytes in readv request', len(readv_bytes))
521
self._last_verb = args[0]
523
def cancel_read_body(self):
524
"""After expecting a body, a response code may indicate one otherwise.
526
This method lets the domain client inform the protocol that no body
527
will be transmitted. This is a terminal method: after calling it the
528
protocol is not able to be used further.
530
self._request.finished_reading()
532
def read_response_tuple(self, expect_body=False):
533
"""Read a response tuple from the wire.
535
This should only be called once.
537
result = self._recv_tuple()
538
if 'hpss' in debug.debug_flags:
539
if self._request_start_time is not None:
540
mutter(' result: %6.3fs %s',
541
time.time() - self._request_start_time,
543
self._request_start_time = None
545
mutter(' result: %s', repr(result)[1:-1])
546
self._response_is_unknown_method(result)
548
self._request.finished_reading()
551
def _response_is_unknown_method(self, result_tuple):
552
"""Raise UnexpectedSmartServerResponse if the response is an 'unknonwn
553
method' response to the request.
555
:param response: The response from a smart client call_expecting_body
557
:param verb: The verb used in that call.
558
:raises: UnexpectedSmartServerResponse
560
if (result_tuple == ('error', "Generic bzr smart protocol error: "
561
"bad request '%s'" % self._last_verb) or
562
result_tuple == ('error', "Generic bzr smart protocol error: "
563
"bad request u'%s'" % self._last_verb)):
564
# The response will have no body, so we've finished reading.
565
self._request.finished_reading()
566
raise errors.UnknownSmartMethod(self._last_verb)
568
def read_body_bytes(self, count=-1):
569
"""Read bytes from the body, decoding into a byte stream.
571
We read all bytes at once to ensure we've checked the trailer for
572
errors, and then feed the buffer back as read_body_bytes is called.
574
if self._body_buffer is not None:
575
return self._body_buffer.read(count)
576
_body_decoder = LengthPrefixedBodyDecoder()
578
# Read no more than 64k at a time so that we don't risk error 10055 (no
579
# buffer space available) on Windows.
581
while not _body_decoder.finished_reading:
582
bytes_wanted = min(_body_decoder.next_read_size(), max_read)
583
bytes = self._request.read_bytes(bytes_wanted)
584
_body_decoder.accept_bytes(bytes)
585
self._request.finished_reading()
586
self._body_buffer = StringIO(_body_decoder.read_pending_data())
587
# XXX: TODO check the trailer result.
588
if 'hpss' in debug.debug_flags:
589
mutter(' %d body bytes read',
590
len(self._body_buffer.getvalue()))
591
return self._body_buffer.read(count)
593
def _recv_tuple(self):
594
"""Receive a tuple from the medium request."""
595
return _decode_tuple(self._recv_line())
597
def _recv_line(self):
598
"""Read an entire line from the medium request."""
600
while not line or line[-1] != '\n':
601
# TODO: this is inefficient - but tuples are short.
602
new_char = self._request.read_bytes(1)
604
# end of file encountered reading from server
605
raise errors.ConnectionReset(
606
"please check connectivity and permissions",
607
"(and try -Dhpss if further diagnosis is required)")
611
def query_version(self):
612
"""Return protocol version number of the server."""
614
resp = self.read_response_tuple()
615
if resp == ('ok', '1'):
617
elif resp == ('ok', '2'):
620
raise errors.SmartProtocolError("bad response %r" % (resp,))
622
def _write_args(self, args):
623
self._write_protocol_version()
624
bytes = _encode_tuple(args)
625
self._request.accept_bytes(bytes)
627
def _write_protocol_version(self):
628
"""Write any prefixes this protocol requires.
630
Version one doesn't send protocol versions.
634
class SmartClientRequestProtocolTwo(SmartClientRequestProtocolOne):
635
"""Version two of the client side of the smart protocol.
637
This prefixes the request with the value of REQUEST_VERSION_TWO.
640
def read_response_tuple(self, expect_body=False):
641
"""Read a response tuple from the wire.
643
This should only be called once.
645
version = self._request.read_line()
646
if version != RESPONSE_VERSION_TWO:
647
raise errors.SmartProtocolError('bad protocol marker %r' % version)
648
response_status = self._recv_line()
649
if response_status not in ('success\n', 'failed\n'):
650
raise errors.SmartProtocolError(
651
'bad protocol status %r' % response_status)
652
self.response_status = response_status == 'success\n'
653
return SmartClientRequestProtocolOne.read_response_tuple(self, expect_body)
655
def _write_protocol_version(self):
656
"""Write any prefixes this protocol requires.
658
Version two sends the value of REQUEST_VERSION_TWO.
660
self._request.accept_bytes(REQUEST_VERSION_TWO)
662
def read_streamed_body(self):
663
"""Read bytes from the body, decoding into a byte stream.
665
# Read no more than 64k at a time so that we don't risk error 10055 (no
666
# buffer space available) on Windows.
668
_body_decoder = ChunkedBodyDecoder()
669
while not _body_decoder.finished_reading:
670
bytes_wanted = min(_body_decoder.next_read_size(), max_read)
671
bytes = self._request.read_bytes(bytes_wanted)
672
_body_decoder.accept_bytes(bytes)
673
for body_bytes in iter(_body_decoder.read_next_chunk, None):
674
if 'hpss' in debug.debug_flags:
675
mutter(' %d byte chunk read',
678
self._request.finished_reading()
added
removed
bzrlib/smart/protocol.py
