← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/smart/protocol.py

  • Committer: John Arbash Meinel
  • Date: 2009-03-06 20:42:40 UTC
  • mto: This revision was merged to the branch mainline in revision 4088.
  • Revision ID: john@arbash-meinel.com-20090306204240-mzjavv31z3gu1x7i
Fix a small bug in setup.py when an extension fails to build

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/smart/protocol.py
1
 
# Copyright (C) 2006, 2007, 2008, 2009 Canonical Ltd
 
1
# Copyright (C) 2006, 2007 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
12
12
#
13
13
# You should have received a copy of the GNU General Public License
14
14
# along with this program; if not, write to the Free Software
15
 
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
16
16
 
17
17
"""Wire-level encoding and decoding of requests and responses for the smart
18
18
client and server.
22
22
from cStringIO import StringIO
23
23
import struct
24
24
import sys
25
 
import thread
26
 
import threading
27
25
import time
28
26
 
29
27
import bzrlib
30
 
from bzrlib import (
31
 
    debug,
32
 
    errors,
33
 
    osutils,
34
 
    )
 
28
from bzrlib import debug
 
29
from bzrlib import errors
35
30
from bzrlib.smart import message, request
36
31
from bzrlib.trace import log_exception_quietly, mutter
37
 
from bzrlib.bencode import bdecode_as_tuple, bencode
 
32
from bzrlib.util.bencode import bdecode_as_tuple, bencode
38
33
 
39
34
 
40
35
# Protocol version strings.  These are sent as prefixes of bzr requests and
119
114
class SmartServerRequestProtocolOne(SmartProtocolBase):
120
115
    """Server-side encoding and decoding logic for smart version 1."""
121
116
 
122
 
    def __init__(self, backing_transport, write_func, root_client_path='/',
123
 
            jail_root=None):
 
117
    def __init__(self, backing_transport, write_func, root_client_path='/'):
124
118
        self._backing_transport = backing_transport
125
119
        self._root_client_path = root_client_path
126
 
        self._jail_root = jail_root
127
120
        self.unused_data = ''
128
121
        self._finished = False
129
122
        self.in_buffer = ''
151
144
                req_args = _decode_tuple(first_line)
152
145
                self.request = request.SmartServerRequestHandler(
153
146
                    self._backing_transport, commands=request.request_handlers,
154
 
                    root_client_path=self._root_client_path,
155
 
                    jail_root=self._jail_root)
156
 
                self.request.args_received(req_args)
 
147
                    root_client_path=self._root_client_path)
 
148
                self.request.dispatch_command(req_args[0], req_args[1:])
157
149
                if self.request.finished_reading:
158
150
                    # trivial request
159
151
                    self.unused_data = self.in_buffer
620
612
            mutter('hpss call:   %s', repr(args)[1:-1])
621
613
            if getattr(self._request._medium, 'base', None) is not None:
622
614
                mutter('             (to %s)', self._request._medium.base)
623
 
            self._request_start_time = osutils.timer_func()
 
615
            self._request_start_time = time.time()
624
616
        self._write_args(args)
625
617
        self._request.finished_writing()
626
618
        self._last_verb = args[0]
635
627
            if getattr(self._request._medium, '_path', None) is not None:
636
628
                mutter('                  (to %s)', self._request._medium._path)
637
629
            mutter('              %d bytes', len(body))
638
 
            self._request_start_time = osutils.timer_func()
 
630
            self._request_start_time = time.time()
639
631
            if 'hpssdetail' in debug.debug_flags:
640
632
                mutter('hpss body content: %s', body)
641
633
        self._write_args(args)
654
646
            mutter('hpss call w/readv: %s', repr(args)[1:-1])
655
647
            if getattr(self._request._medium, '_path', None) is not None:
656
648
                mutter('                  (to %s)', self._request._medium._path)
657
 
            self._request_start_time = osutils.timer_func()
 
649
            self._request_start_time = time.time()
658
650
        self._write_args(args)
659
651
        readv_bytes = self._serialise_offsets(body)
660
652
        bytes = self._encode_bulk_data(readv_bytes)
686
678
        if 'hpss' in debug.debug_flags:
687
679
            if self._request_start_time is not None:
688
680
                mutter('   result:   %6.3fs  %s',
689
 
                       osutils.timer_func() - self._request_start_time,
 
681
                       time.time() - self._request_start_time,
690
682
                       repr(result)[1:-1])
691
683
                self._request_start_time = None
692
684
            else:
866
858
 
867
859
 
868
860
def build_server_protocol_three(backing_transport, write_func,
869
 
                                root_client_path, jail_root=None):
 
861
                                root_client_path):
870
862
    request_handler = request.SmartServerRequestHandler(
871
863
        backing_transport, commands=request.request_handlers,
872
 
        root_client_path=root_client_path, jail_root=jail_root)
 
864
        root_client_path=root_client_path)
873
865
    responder = ProtocolThreeResponder(write_func)
874
866
    message_handler = message.ConventionalRequestHandler(request_handler, responder)
875
867
    return ProtocolThreeDecoder(message_handler)
905
897
            # We do *not* set self.decoding_failed here.  The message handler
906
898
            # has raised an error, but the decoder is still able to parse bytes
907
899
            # and determine when this message ends.
908
 
            if not isinstance(exception.exc_value, errors.UnknownSmartMethod):
909
 
                log_exception_quietly()
 
900
            log_exception_quietly()
910
901
            self.message_handler.protocol_error(exception.exc_value)
911
902
            # The state machine is ready to continue decoding, but the
912
903
            # exception has interrupted the loop that runs the state machine.
1045
1036
            raise errors.SmartMessageHandlerError(sys.exc_info())
1046
1037
 
1047
1038
    def _state_accept_reading_unused(self):
1048
 
        self.unused_data += self._get_in_buffer()
 
1039
        self.unused_data = self._get_in_buffer()
1049
1040
        self._set_in_buffer(None)
1050
1041
 
1051
1042
    def next_read_size(self):
1067
1058
class _ProtocolThreeEncoder(object):
1068
1059
 
1069
1060
    response_marker = request_marker = MESSAGE_VERSION_THREE
1070
 
    BUFFER_SIZE = 1024*1024 # 1 MiB buffer before flushing
1071
1061
 
1072
1062
    def __init__(self, write_func):
1073
1063
        self._buf = []
1074
 
        self._buf_len = 0
1075
1064
        self._real_write_func = write_func
1076
1065
 
1077
1066
    def _write_func(self, bytes):
1078
 
        # TODO: It is probably more appropriate to use sum(map(len, _buf))
1079
 
        #       for total number of bytes to write, rather than buffer based on
1080
 
        #       the number of write() calls
1081
 
        # TODO: Another possibility would be to turn this into an async model.
1082
 
        #       Where we let another thread know that we have some bytes if
1083
 
        #       they want it, but we don't actually block for it
1084
 
        #       Note that osutils.send_all always sends 64kB chunks anyway, so
1085
 
        #       we might just push out smaller bits at a time?
1086
1067
        self._buf.append(bytes)
1087
 
        self._buf_len += len(bytes)
1088
 
        if self._buf_len > self.BUFFER_SIZE:
 
1068
        if len(self._buf) > 100:
1089
1069
            self.flush()
1090
1070
 
1091
1071
    def flush(self):
1092
1072
        if self._buf:
1093
1073
            self._real_write_func(''.join(self._buf))
1094
1074
            del self._buf[:]
1095
 
            self._buf_len = 0
1096
1075
 
1097
1076
    def _serialise_offsets(self, offsets):
1098
1077
        """Serialise a readv offset list."""
1147
1126
        _ProtocolThreeEncoder.__init__(self, write_func)
1148
1127
        self.response_sent = False
1149
1128
        self._headers = {'Software version': bzrlib.__version__}
1150
 
        if 'hpss' in debug.debug_flags:
1151
 
            self._thread_id = thread.get_ident()
1152
 
            self._response_start_time = None
1153
 
 
1154
 
    def _trace(self, action, message, extra_bytes=None, include_time=False):
1155
 
        if self._response_start_time is None:
1156
 
            self._response_start_time = osutils.timer_func()
1157
 
        if include_time:
1158
 
            t = '%5.3fs ' % (time.clock() - self._response_start_time)
1159
 
        else:
1160
 
            t = ''
1161
 
        if extra_bytes is None:
1162
 
            extra = ''
1163
 
        else:
1164
 
            extra = ' ' + repr(extra_bytes[:40])
1165
 
            if len(extra) > 33:
1166
 
                extra = extra[:29] + extra[-1] + '...'
1167
 
        mutter('%12s: [%s] %s%s%s'
1168
 
               % (action, self._thread_id, t, message, extra))
1169
1129
 
1170
1130
    def send_error(self, exception):
1171
1131
        if self.response_sent:
1177
1137
                ('UnknownMethod', exception.verb))
1178
1138
            self.send_response(failure)
1179
1139
            return
1180
 
        if 'hpss' in debug.debug_flags:
1181
 
            self._trace('error', str(exception))
1182
1140
        self.response_sent = True
1183
1141
        self._write_protocol_version()
1184
1142
        self._write_headers(self._headers)
1198
1156
            self._write_success_status()
1199
1157
        else:
1200
1158
            self._write_error_status()
1201
 
        if 'hpss' in debug.debug_flags:
1202
 
            self._trace('response', repr(response.args))
1203
1159
        self._write_structure(response.args)
1204
1160
        if response.body is not None:
1205
1161
            self._write_prefixed_body(response.body)
1206
 
            if 'hpss' in debug.debug_flags:
1207
 
                self._trace('body', '%d bytes' % (len(response.body),),
1208
 
                            response.body, include_time=True)
1209
1162
        elif response.body_stream is not None:
1210
 
            count = num_bytes = 0
1211
 
            first_chunk = None
1212
1163
            for exc_info, chunk in _iter_with_errors(response.body_stream):
1213
 
                count += 1
1214
1164
                if exc_info is not None:
1215
1165
                    self._write_error_status()
1216
1166
                    error_struct = request._translate_error(exc_info[1])
1221
1171
                        self._write_error_status()
1222
1172
                        self._write_structure(chunk.args)
1223
1173
                        break
1224
 
                    num_bytes += len(chunk)
1225
 
                    if first_chunk is None:
1226
 
                        first_chunk = chunk
1227
1174
                    self._write_prefixed_body(chunk)
1228
 
                    if 'hpssdetail' in debug.debug_flags:
1229
 
                        # Not worth timing separately, as _write_func is
1230
 
                        # actually buffered
1231
 
                        self._trace('body chunk',
1232
 
                                    '%d bytes' % (len(chunk),),
1233
 
                                    chunk, suppress_time=True)
1234
 
            if 'hpss' in debug.debug_flags:
1235
 
                self._trace('body stream',
1236
 
                            '%d bytes %d chunks' % (num_bytes, count),
1237
 
                            first_chunk)
1238
1175
        self._write_end()
1239
 
        if 'hpss' in debug.debug_flags:
1240
 
            self._trace('response end', '', include_time=True)
1241
1176
 
1242
1177
 
1243
1178
def _iter_with_errors(iterable):
1273
1208
        except (KeyboardInterrupt, SystemExit):
1274
1209
            raise
1275
1210
        except Exception:
1276
 
            mutter('_iter_with_errors caught error')
1277
 
            log_exception_quietly()
1278
1211
            yield sys.exc_info(), None
1279
1212
            return
1280
1213
 
1295
1228
            base = getattr(self._medium_request._medium, 'base', None)
1296
1229
            if base is not None:
1297
1230
                mutter('             (to %s)', base)
1298
 
            self._request_start_time = osutils.timer_func()
 
1231
            self._request_start_time = time.time()
1299
1232
        self._write_protocol_version()
1300
1233
        self._write_headers(self._headers)
1301
1234
        self._write_structure(args)
1313
1246
            if path is not None:
1314
1247
                mutter('                  (to %s)', path)
1315
1248
            mutter('              %d bytes', len(body))
1316
 
            self._request_start_time = osutils.timer_func()
 
1249
            self._request_start_time = time.time()
1317
1250
        self._write_protocol_version()
1318
1251
        self._write_headers(self._headers)
1319
1252
        self._write_structure(args)
1332
1265
            path = getattr(self._medium_request._medium, '_path', None)
1333
1266
            if path is not None:
1334
1267
                mutter('                  (to %s)', path)
1335
 
            self._request_start_time = osutils.timer_func()
 
1268
            self._request_start_time = time.time()
1336
1269
        self._write_protocol_version()
1337
1270
        self._write_headers(self._headers)
1338
1271
        self._write_structure(args)
1349
1282
            path = getattr(self._medium_request._medium, '_path', None)
1350
1283
            if path is not None:
1351
1284
                mutter('                  (to %s)', path)
1352
 
            self._request_start_time = osutils.timer_func()
 
1285
            self._request_start_time = time.time()
1353
1286
        self._write_protocol_version()
1354
1287
        self._write_headers(self._headers)
1355
1288
        self._write_structure(args)