← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/smart/protocol.py

Merge shell-like-tests into description

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/smart/protocol.py
22
22
from cStringIO import StringIO
23
23
import struct
24
24
import sys
 
25
import threading
25
26
import time
26
27
 
27
28
import bzrlib
28
 
from bzrlib import debug
29
 
from bzrlib import errors
 
29
from bzrlib import (
 
30
    debug,
 
31
    errors,
 
32
    osutils,
 
33
    )
30
34
from bzrlib.smart import message, request
31
35
from bzrlib.trace import log_exception_quietly, mutter
32
36
from bzrlib.bencode import bdecode_as_tuple, bencode
615
619
            mutter('hpss call:   %s', repr(args)[1:-1])
616
620
            if getattr(self._request._medium, 'base', None) is not None:
617
621
                mutter('             (to %s)', self._request._medium.base)
618
 
            self._request_start_time = time.time()
 
622
            self._request_start_time = osutils.timer_func()
619
623
        self._write_args(args)
620
624
        self._request.finished_writing()
621
625
        self._last_verb = args[0]
630
634
            if getattr(self._request._medium, '_path', None) is not None:
631
635
                mutter('                  (to %s)', self._request._medium._path)
632
636
            mutter('              %d bytes', len(body))
633
 
            self._request_start_time = time.time()
 
637
            self._request_start_time = osutils.timer_func()
634
638
            if 'hpssdetail' in debug.debug_flags:
635
639
                mutter('hpss body content: %s', body)
636
640
        self._write_args(args)
649
653
            mutter('hpss call w/readv: %s', repr(args)[1:-1])
650
654
            if getattr(self._request._medium, '_path', None) is not None:
651
655
                mutter('                  (to %s)', self._request._medium._path)
652
 
            self._request_start_time = time.time()
 
656
            self._request_start_time = osutils.timer_func()
653
657
        self._write_args(args)
654
658
        readv_bytes = self._serialise_offsets(body)
655
659
        bytes = self._encode_bulk_data(readv_bytes)
681
685
        if 'hpss' in debug.debug_flags:
682
686
            if self._request_start_time is not None:
683
687
                mutter('   result:   %6.3fs  %s',
684
 
                       time.time() - self._request_start_time,
 
688
                       osutils.timer_func() - self._request_start_time,
685
689
                       repr(result)[1:-1])
686
690
                self._request_start_time = None
687
691
            else:
1068
1072
        self._real_write_func = write_func
1069
1073
 
1070
1074
    def _write_func(self, bytes):
 
1075
        # TODO: It is probably more appropriate to use sum(map(len, _buf))
 
1076
        #       for total number of bytes to write, rather than buffer based on
 
1077
        #       the number of write() calls
 
1078
        # TODO: Another possibility would be to turn this into an async model.
 
1079
        #       Where we let another thread know that we have some bytes if
 
1080
        #       they want it, but we don't actually block for it
 
1081
        #       Note that osutils.send_all always sends 64kB chunks anyway, so
 
1082
        #       we might just push out smaller bits at a time?
1071
1083
        self._buf.append(bytes)
1072
1084
        if len(self._buf) > 100:
1073
1085
            self.flush()
1130
1142
        _ProtocolThreeEncoder.__init__(self, write_func)
1131
1143
        self.response_sent = False
1132
1144
        self._headers = {'Software version': bzrlib.__version__}
 
1145
        if 'hpss' in debug.debug_flags:
 
1146
            self._thread_id = threading.currentThread().get_ident()
 
1147
            self._response_start_time = None
 
1148
 
 
1149
    def _trace(self, action, message, extra_bytes=None, include_time=False):
 
1150
        if self._response_start_time is None:
 
1151
            self._response_start_time = osutils.timer_func()
 
1152
        if include_time:
 
1153
            t = '%5.3fs ' % (time.clock() - self._response_start_time)
 
1154
        else:
 
1155
            t = ''
 
1156
        if extra_bytes is None:
 
1157
            extra = ''
 
1158
        else:
 
1159
            extra = ' ' + repr(extra_bytes[:40])
 
1160
            if len(extra) > 33:
 
1161
                extra = extra[:29] + extra[-1] + '...'
 
1162
        mutter('%12s: [%s] %s%s%s'
 
1163
               % (action, self._thread_id, t, message, extra))
1133
1164
 
1134
1165
    def send_error(self, exception):
1135
1166
        if self.response_sent:
1141
1172
                ('UnknownMethod', exception.verb))
1142
1173
            self.send_response(failure)
1143
1174
            return
 
1175
        if 'hpss' in debug.debug_flags:
 
1176
            self._trace('error', str(exception))
1144
1177
        self.response_sent = True
1145
1178
        self._write_protocol_version()
1146
1179
        self._write_headers(self._headers)
1160
1193
            self._write_success_status()
1161
1194
        else:
1162
1195
            self._write_error_status()
 
1196
        if 'hpss' in debug.debug_flags:
 
1197
            self._trace('response', repr(response.args))
1163
1198
        self._write_structure(response.args)
1164
1199
        if response.body is not None:
1165
1200
            self._write_prefixed_body(response.body)
 
1201
            if 'hpss' in debug.debug_flags:
 
1202
                self._trace('body', '%d bytes' % (len(response.body),),
 
1203
                            response.body, include_time=True)
1166
1204
        elif response.body_stream is not None:
 
1205
            count = num_bytes = 0
 
1206
            first_chunk = None
1167
1207
            for exc_info, chunk in _iter_with_errors(response.body_stream):
 
1208
                count += 1
1168
1209
                if exc_info is not None:
1169
1210
                    self._write_error_status()
1170
1211
                    error_struct = request._translate_error(exc_info[1])
1175
1216
                        self._write_error_status()
1176
1217
                        self._write_structure(chunk.args)
1177
1218
                        break
 
1219
                    num_bytes += len(chunk)
 
1220
                    if first_chunk is None:
 
1221
                        first_chunk = chunk
1178
1222
                    self._write_prefixed_body(chunk)
 
1223
                    if 'hpssdetail' in debug.debug_flags:
 
1224
                        # Not worth timing separately, as _write_func is
 
1225
                        # actually buffered
 
1226
                        self._trace('body chunk',
 
1227
                                    '%d bytes' % (len(chunk),),
 
1228
                                    chunk, suppress_time=True)
 
1229
            if 'hpss' in debug.debug_flags:
 
1230
                self._trace('body stream',
 
1231
                            '%d bytes %d chunks' % (num_bytes, count),
 
1232
                            first_chunk)
1179
1233
        self._write_end()
 
1234
        if 'hpss' in debug.debug_flags:
 
1235
            self._trace('response end', '', include_time=True)
1180
1236
 
1181
1237
 
1182
1238
def _iter_with_errors(iterable):
1234
1290
            base = getattr(self._medium_request._medium, 'base', None)
1235
1291
            if base is not None:
1236
1292
                mutter('             (to %s)', base)
1237
 
            self._request_start_time = time.time()
 
1293
            self._request_start_time = osutils.timer_func()
1238
1294
        self._write_protocol_version()
1239
1295
        self._write_headers(self._headers)
1240
1296
        self._write_structure(args)
1252
1308
            if path is not None:
1253
1309
                mutter('                  (to %s)', path)
1254
1310
            mutter('              %d bytes', len(body))
1255
 
            self._request_start_time = time.time()
 
1311
            self._request_start_time = osutils.timer_func()
1256
1312
        self._write_protocol_version()
1257
1313
        self._write_headers(self._headers)
1258
1314
        self._write_structure(args)
1271
1327
            path = getattr(self._medium_request._medium, '_path', None)
1272
1328
            if path is not None:
1273
1329
                mutter('                  (to %s)', path)
1274
 
            self._request_start_time = time.time()
 
1330
            self._request_start_time = osutils.timer_func()
1275
1331
        self._write_protocol_version()
1276
1332
        self._write_headers(self._headers)
1277
1333
        self._write_structure(args)
1288
1344
            path = getattr(self._medium_request._medium, '_path', None)
1289
1345
            if path is not None:
1290
1346
                mutter('                  (to %s)', path)
1291
 
            self._request_start_time = time.time()
 
1347
            self._request_start_time = osutils.timer_func()
1292
1348
        self._write_protocol_version()
1293
1349
        self._write_headers(self._headers)
1294
1350
        self._write_structure(args)