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):
77
self._backing_transport = backing_transport
78
self.excess_buffer = ''
79
self._finished = False
81
self.has_dispatched = False
83
self._body_decoder = None
84
self._write_func = write_func
86
def accept_bytes(self, bytes):
87
"""Take bytes, and advance the internal state machine appropriately.
89
:param bytes: must be a byte string
91
assert isinstance(bytes, str)
92
self.in_buffer += bytes
93
if not self.has_dispatched:
94
if '\n' not in self.in_buffer:
97
self.has_dispatched = True
99
first_line, self.in_buffer = self.in_buffer.split('\n', 1)
101
req_args = _decode_tuple(first_line)
102
self.request = request.SmartServerRequestHandler(
103
self._backing_transport, commands=request.request_handlers)
104
self.request.dispatch_command(req_args[0], req_args[1:])
105
if self.request.finished_reading:
107
self.excess_buffer = self.in_buffer
109
self._send_response(self.request.response)
110
except KeyboardInterrupt:
112
except Exception, exception:
113
# everything else: pass to client, flush, and quit
114
log_exception_quietly()
115
self._send_response(request.FailedSmartServerResponse(
116
('error', str(exception))))
119
if self.has_dispatched:
121
# nothing to do.XXX: this routine should be a single state
123
self.excess_buffer += self.in_buffer
126
if self._body_decoder is None:
127
self._body_decoder = LengthPrefixedBodyDecoder()
128
self._body_decoder.accept_bytes(self.in_buffer)
129
self.in_buffer = self._body_decoder.unused_data
130
body_data = self._body_decoder.read_pending_data()
131
self.request.accept_body(body_data)
132
if self._body_decoder.finished_reading:
133
self.request.end_of_body()
134
assert self.request.finished_reading, \
135
"no more body, request not finished"
136
if self.request.response is not None:
137
self._send_response(self.request.response)
138
self.excess_buffer = self.in_buffer
141
assert not self.request.finished_reading, \
142
"no response and we have finished reading."
144
def _send_response(self, response):
145
"""Send a smart server response down the output stream."""
146
assert not self._finished, 'response already sent'
149
self._finished = True
150
self._write_protocol_version()
151
self._write_success_or_failure_prefix(response)
152
self._write_func(_encode_tuple(args))
154
assert isinstance(body, str), 'body must be a str'
155
bytes = self._encode_bulk_data(body)
156
self._write_func(bytes)
158
def _write_protocol_version(self):
159
"""Write any prefixes this protocol requires.
161
Version one doesn't send protocol versions.
164
def _write_success_or_failure_prefix(self, response):
165
"""Write the protocol specific success/failure prefix.
167
For SmartServerRequestProtocolOne this is omitted but we
168
call is_successful to ensure that the response is valid.
170
response.is_successful()
172
def next_read_size(self):
175
if self._body_decoder is None:
178
return self._body_decoder.next_read_size()
181
class SmartServerRequestProtocolTwo(SmartServerRequestProtocolOne):
182
r"""Version two of the server side of the smart protocol.
184
This prefixes responses with the value of RESPONSE_VERSION_TWO.
187
def _write_success_or_failure_prefix(self, response):
188
"""Write the protocol specific success/failure prefix."""
189
if response.is_successful():
190
self._write_func('success\n')
192
self._write_func('failed\n')
194
def _write_protocol_version(self):
195
r"""Write any prefixes this protocol requires.
197
Version two sends the value of RESPONSE_VERSION_TWO.
199
self._write_func(RESPONSE_VERSION_TWO)
201
def _send_response(self, response):
202
"""Send a smart server response down the output stream."""
203
assert not self._finished, 'response already sent'
204
self._finished = True
205
self._write_protocol_version()
206
self._write_success_or_failure_prefix(response)
207
self._write_func(_encode_tuple(response.args))
208
if response.body is not None:
209
assert isinstance(response.body, str), 'body must be a str'
210
assert response.body_stream is None, (
211
'body_stream and body cannot both be set')
212
bytes = self._encode_bulk_data(response.body)
213
self._write_func(bytes)
214
elif response.body_stream is not None:
215
_send_stream(response.body_stream, self._write_func)
218
def _send_stream(stream, write_func):
219
write_func('chunked\n')
220
_send_chunks(stream, write_func)
224
def _send_chunks(stream, write_func):
226
if isinstance(chunk, str):
227
bytes = "%x\n%s" % (len(chunk), chunk)
229
elif isinstance(chunk, request.FailedSmartServerResponse):
231
_send_chunks(chunk.args, write_func)
234
raise errors.BzrError(
235
'Chunks must be str or FailedSmartServerResponse, got %r'
239
class _StatefulDecoder(object):
242
self.finished_reading = False
243
self.unused_data = ''
244
self.bytes_left = None
246
def accept_bytes(self, bytes):
247
"""Decode as much of bytes as possible.
249
If 'bytes' contains too much data it will be appended to
252
finished_reading will be set when no more data is required. Further
253
data will be appended to self.unused_data.
255
# accept_bytes is allowed to change the state
256
current_state = self.state_accept
257
self.state_accept(bytes)
258
while current_state != self.state_accept:
259
current_state = self.state_accept
260
self.state_accept('')
263
class ChunkedBodyDecoder(_StatefulDecoder):
264
"""Decoder for chunked body data.
266
This is very similar the HTTP's chunked encoding. See the description of
267
streamed body data in `doc/developers/network-protocol.txt` for details.
271
_StatefulDecoder.__init__(self)
272
self.state_accept = self._state_accept_expecting_header
274
self.chunk_in_progress = None
275
self.chunks = collections.deque()
277
self.error_in_progress = None
279
def next_read_size(self):
280
# Note: the shortest possible chunk is 2 bytes: '0\n', and the
281
# end-of-body marker is 4 bytes: 'END\n'.
282
if self.state_accept == self._state_accept_reading_chunk:
283
# We're expecting more chunk content. So we're expecting at least
284
# the rest of this chunk plus an END chunk.
285
return self.bytes_left + 4
286
elif self.state_accept == self._state_accept_expecting_length:
287
if self._in_buffer == '':
288
# We're expecting a chunk length. There's at least two bytes
289
# left: a digit plus '\n'.
292
# We're in the middle of reading a chunk length. So there's at
293
# least one byte left, the '\n' that terminates the length.
295
elif self.state_accept == self._state_accept_reading_unused:
297
elif self.state_accept == self._state_accept_expecting_header:
298
return max(0, len('chunked\n') - len(self._in_buffer))
300
raise AssertionError("Impossible state: %r" % (self.state_accept,))
302
def read_next_chunk(self):
304
return self.chunks.popleft()
308
def _extract_line(self):
309
pos = self._in_buffer.find('\n')
311
# We haven't read a complete length prefix yet, so there's nothing
314
line = self._in_buffer[:pos]
315
# Trim the prefix (including '\n' delimiter) from the _in_buffer.
316
self._in_buffer = self._in_buffer[pos+1:]
320
self.unused_data = self._in_buffer
321
self._in_buffer = None
322
self.state_accept = self._state_accept_reading_unused
324
error_args = tuple(self.error_in_progress)
325
self.chunks.append(request.FailedSmartServerResponse(error_args))
326
self.error_in_progress = None
327
self.finished_reading = True
329
def _state_accept_expecting_header(self, bytes):
330
self._in_buffer += bytes
331
prefix = self._extract_line()
333
# We haven't read a complete length prefix yet, so there's nothing
336
elif prefix == 'chunked':
337
self.state_accept = self._state_accept_expecting_length
339
raise errors.SmartProtocolError(
340
'Bad chunked body header: "%s"' % (prefix,))
342
def _state_accept_expecting_length(self, bytes):
343
self._in_buffer += bytes
344
prefix = self._extract_line()
346
# We haven't read a complete length prefix yet, so there's nothing
349
elif prefix == 'ERR':
351
self.error_in_progress = []
352
self._state_accept_expecting_length('')
354
elif prefix == 'END':
355
# We've read the end-of-body marker.
356
# Any further bytes are unused data, including the bytes left in
361
self.bytes_left = int(prefix, 16)
362
self.chunk_in_progress = ''
363
self.state_accept = self._state_accept_reading_chunk
365
def _state_accept_reading_chunk(self, bytes):
366
self._in_buffer += bytes
367
in_buffer_len = len(self._in_buffer)
368
self.chunk_in_progress += self._in_buffer[:self.bytes_left]
369
self._in_buffer = self._in_buffer[self.bytes_left:]
370
self.bytes_left -= in_buffer_len
371
if self.bytes_left <= 0:
372
# Finished with chunk
373
self.bytes_left = None
375
self.error_in_progress.append(self.chunk_in_progress)
377
self.chunks.append(self.chunk_in_progress)
378
self.chunk_in_progress = None
379
self.state_accept = self._state_accept_expecting_length
381
def _state_accept_reading_unused(self, bytes):
382
self.unused_data += bytes
385
class LengthPrefixedBodyDecoder(_StatefulDecoder):
386
"""Decodes the length-prefixed bulk data."""
389
_StatefulDecoder.__init__(self)
390
self.state_accept = self._state_accept_expecting_length
391
self.state_read = self._state_read_no_data
393
self._trailer_buffer = ''
395
def next_read_size(self):
396
if self.bytes_left is not None:
397
# Ideally we want to read all the remainder of the body and the
399
return self.bytes_left + 5
400
elif self.state_accept == self._state_accept_reading_trailer:
401
# Just the trailer left
402
return 5 - len(self._trailer_buffer)
403
elif self.state_accept == self._state_accept_expecting_length:
404
# There's still at least 6 bytes left ('\n' to end the length, plus
408
# Reading excess data. Either way, 1 byte at a time is fine.
411
def read_pending_data(self):
412
"""Return any pending data that has been decoded."""
413
return self.state_read()
415
def _state_accept_expecting_length(self, bytes):
416
self._in_buffer += bytes
417
pos = self._in_buffer.find('\n')
420
self.bytes_left = int(self._in_buffer[:pos])
421
self._in_buffer = self._in_buffer[pos+1:]
422
self.bytes_left -= len(self._in_buffer)
423
self.state_accept = self._state_accept_reading_body
424
self.state_read = self._state_read_in_buffer
426
def _state_accept_reading_body(self, bytes):
427
self._in_buffer += bytes
428
self.bytes_left -= len(bytes)
429
if self.bytes_left <= 0:
431
if self.bytes_left != 0:
432
self._trailer_buffer = self._in_buffer[self.bytes_left:]
433
self._in_buffer = self._in_buffer[:self.bytes_left]
434
self.bytes_left = None
435
self.state_accept = self._state_accept_reading_trailer
437
def _state_accept_reading_trailer(self, bytes):
438
self._trailer_buffer += bytes
439
# TODO: what if the trailer does not match "done\n"? Should this raise
440
# a ProtocolViolation exception?
441
if self._trailer_buffer.startswith('done\n'):
442
self.unused_data = self._trailer_buffer[len('done\n'):]
443
self.state_accept = self._state_accept_reading_unused
444
self.finished_reading = True
446
def _state_accept_reading_unused(self, bytes):
447
self.unused_data += bytes
449
def _state_read_no_data(self):
452
def _state_read_in_buffer(self):
453
result = self._in_buffer
458
class SmartClientRequestProtocolOne(SmartProtocolBase):
459
"""The client-side protocol for smart version 1."""
461
def __init__(self, request):
462
"""Construct a SmartClientRequestProtocolOne.
464
:param request: A SmartClientMediumRequest to serialise onto and
467
self._request = request
468
self._body_buffer = None
469
self._request_start_time = None
471
def call(self, *args):
472
if 'hpss' in debug.debug_flags:
473
mutter('hpss call: %s', repr(args)[1:-1])
474
if getattr(self._request._medium, 'base', None) is not None:
475
mutter(' (to %s)', self._request._medium.base)
476
self._request_start_time = time.time()
477
self._write_args(args)
478
self._request.finished_writing()
480
def call_with_body_bytes(self, args, body):
481
"""Make a remote call of args with body bytes 'body'.
483
After calling this, call read_response_tuple to find the result out.
485
if 'hpss' in debug.debug_flags:
486
mutter('hpss call w/body: %s (%r...)', repr(args)[1:-1], body[:20])
487
if getattr(self._request._medium, '_path', None) is not None:
488
mutter(' (to %s)', self._request._medium._path)
489
mutter(' %d bytes', len(body))
490
self._request_start_time = time.time()
491
if 'hpssdetail' in debug.debug_flags:
492
mutter('hpss body content: %s', body)
493
self._write_args(args)
494
bytes = self._encode_bulk_data(body)
495
self._request.accept_bytes(bytes)
496
self._request.finished_writing()
498
def call_with_body_readv_array(self, args, body):
499
"""Make a remote call with a readv array.
501
The body is encoded with one line per readv offset pair. The numbers in
502
each pair are separated by a comma, and no trailing \n is emitted.
504
if 'hpss' in debug.debug_flags:
505
mutter('hpss call w/readv: %s', repr(args)[1:-1])
506
if getattr(self._request._medium, '_path', None) is not None:
507
mutter(' (to %s)', self._request._medium._path)
508
self._request_start_time = time.time()
509
self._write_args(args)
510
readv_bytes = self._serialise_offsets(body)
511
bytes = self._encode_bulk_data(readv_bytes)
512
self._request.accept_bytes(bytes)
513
self._request.finished_writing()
514
if 'hpss' in debug.debug_flags:
515
mutter(' %d bytes in readv request', len(readv_bytes))
517
def cancel_read_body(self):
518
"""After expecting a body, a response code may indicate one otherwise.
520
This method lets the domain client inform the protocol that no body
521
will be transmitted. This is a terminal method: after calling it the
522
protocol is not able to be used further.
524
self._request.finished_reading()
526
def read_response_tuple(self, expect_body=False):
527
"""Read a response tuple from the wire.
529
This should only be called once.
531
result = self._recv_tuple()
532
if 'hpss' in debug.debug_flags:
533
if self._request_start_time is not None:
534
mutter(' result: %6.3fs %s',
535
time.time() - self._request_start_time,
537
self._request_start_time = None
539
mutter(' result: %s', repr(result)[1:-1])
541
self._request.finished_reading()
544
def read_body_bytes(self, count=-1):
545
"""Read bytes from the body, decoding into a byte stream.
547
We read all bytes at once to ensure we've checked the trailer for
548
errors, and then feed the buffer back as read_body_bytes is called.
550
if self._body_buffer is not None:
551
return self._body_buffer.read(count)
552
_body_decoder = LengthPrefixedBodyDecoder()
554
# Read no more than 64k at a time so that we don't risk error 10055 (no
555
# buffer space available) on Windows.
557
while not _body_decoder.finished_reading:
558
bytes_wanted = min(_body_decoder.next_read_size(), max_read)
559
bytes = self._request.read_bytes(bytes_wanted)
560
_body_decoder.accept_bytes(bytes)
561
self._request.finished_reading()
562
self._body_buffer = StringIO(_body_decoder.read_pending_data())
563
# XXX: TODO check the trailer result.
564
if 'hpss' in debug.debug_flags:
565
mutter(' %d body bytes read',
566
len(self._body_buffer.getvalue()))
567
return self._body_buffer.read(count)
569
def _recv_tuple(self):
570
"""Receive a tuple from the medium request."""
571
return _decode_tuple(self._recv_line())
573
def _recv_line(self):
574
"""Read an entire line from the medium request."""
576
while not line or line[-1] != '\n':
577
# TODO: this is inefficient - but tuples are short.
578
new_char = self._request.read_bytes(1)
580
# end of file encountered reading from server
581
raise errors.ConnectionReset(
582
"please check connectivity and permissions",
583
"(and try -Dhpss if further diagnosis is required)")
587
def query_version(self):
588
"""Return protocol version number of the server."""
590
resp = self.read_response_tuple()
591
if resp == ('ok', '1'):
593
elif resp == ('ok', '2'):
596
raise errors.SmartProtocolError("bad response %r" % (resp,))
598
def _write_args(self, args):
599
self._write_protocol_version()
600
bytes = _encode_tuple(args)
601
self._request.accept_bytes(bytes)
603
def _write_protocol_version(self):
604
"""Write any prefixes this protocol requires.
606
Version one doesn't send protocol versions.
610
class SmartClientRequestProtocolTwo(SmartClientRequestProtocolOne):
611
"""Version two of the client side of the smart protocol.
613
This prefixes the request with the value of REQUEST_VERSION_TWO.
616
def read_response_tuple(self, expect_body=False):
617
"""Read a response tuple from the wire.
619
This should only be called once.
621
version = self._request.read_line()
622
if version != RESPONSE_VERSION_TWO:
623
raise errors.SmartProtocolError('bad protocol marker %r' % version)
624
response_status = self._recv_line()
625
if response_status not in ('success\n', 'failed\n'):
626
raise errors.SmartProtocolError(
627
'bad protocol status %r' % response_status)
628
self.response_status = response_status == 'success\n'
629
return SmartClientRequestProtocolOne.read_response_tuple(self, expect_body)
631
def _write_protocol_version(self):
632
"""Write any prefixes this protocol requires.
634
Version two sends the value of REQUEST_VERSION_TWO.
636
self._request.accept_bytes(REQUEST_VERSION_TWO)
638
def read_streamed_body(self):
639
"""Read bytes from the body, decoding into a byte stream.
641
# Read no more than 64k at a time so that we don't risk error 10055 (no
642
# buffer space available) on Windows.
644
_body_decoder = ChunkedBodyDecoder()
645
while not _body_decoder.finished_reading:
646
bytes_wanted = min(_body_decoder.next_read_size(), max_read)
647
bytes = self._request.read_bytes(bytes_wanted)
648
_body_decoder.accept_bytes(bytes)
649
for body_bytes in iter(_body_decoder.read_next_chunk, None):
650
if 'hpss' in debug.debug_flags:
651
mutter(' %d byte chunk read',
654
self._request.finished_reading()
added
removed
bzrlib/smart/protocol.py
