/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/smart/protocol.py

• browse files at revision 4869.2.9
• compare with another revision
• compare with revision 1663
• download diff
• download tarball
• view history from revision 4869.2.9

Show diffs side-by-side

added

removed

bzrlib/smart/protocol.py

1

 

# Copyright (C) 2006, 2007 Canonical Ltd

 

1

# Copyright (C) 2006, 2007, 2008, 2009 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

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:

1062

1066

class _ProtocolThreeEncoder(object):

1063

1067

 

1064

1068

    response_marker = request_marker = MESSAGE_VERSION_THREE

 

1069

    BUFFER_SIZE = 1024*1024 # 1 MiB buffer before flushing

1065

1070

 

1066

1071

    def __init__(self, write_func):

1067

1072

        self._buf = []

 

1073

        self._buf_len = 0

1068

1074

        self._real_write_func = write_func

1069

1075

 

1070

1076

    def _write_func(self, bytes):

 

1077

        # TODO: It is probably more appropriate to use sum(map(len, _buf))

 

1078

        #       for total number of bytes to write, rather than buffer based on

 

1079

        #       the number of write() calls

 

1080

        # TODO: Another possibility would be to turn this into an async model.

 

1081

        #       Where we let another thread know that we have some bytes if

 

1082

        #       they want it, but we don't actually block for it

 

1083

        #       Note that osutils.send_all always sends 64kB chunks anyway, so

 

1084

        #       we might just push out smaller bits at a time?

1071

1085

        self._buf.append(bytes)

1072

 

        if len(self._buf) > 100:

 

1086

        self._buf_len += len(bytes)

 

1087

        if self._buf_len > self.BUFFER_SIZE:

1073

1088

            self.flush()

1074

1089

 

1075

1090

    def flush(self):

1076

1091

        if self._buf:

1077

1092

            self._real_write_func(''.join(self._buf))

1078

1093

            del self._buf[:]

 

1094

            self._buf_len = 0

1079

1095

 

1080

1096

    def _serialise_offsets(self, offsets):

1081

1097

        """Serialise a readv offset list."""

1130

1146

        _ProtocolThreeEncoder.__init__(self, write_func)

1131

1147

        self.response_sent = False

1132

1148

        self._headers = {'Software version': bzrlib.__version__}

 

1149

        if 'hpss' in debug.debug_flags:

 

1150

            self._thread_id = threading.currentThread().get_ident()

 

1151

            self._response_start_time = None

 

1152

 

 

1153

    def _trace(self, action, message, extra_bytes=None, include_time=False):

 

1154

        if self._response_start_time is None:

 

1155

            self._response_start_time = osutils.timer_func()

 

1156

        if include_time:

 

1157

            t = '%5.3fs ' % (time.clock() - self._response_start_time)

 

1158

        else:

 

1159

            t = ''

 

1160

        if extra_bytes is None:

 

1161

            extra = ''

 

1162

        else:

 

1163

            extra = ' ' + repr(extra_bytes[:40])

 

1164

            if len(extra) > 33:

 

1165

                extra = extra[:29] + extra[-1] + '...'

 

1166

        mutter('%12s: [%s] %s%s%s'

 

1167

               % (action, self._thread_id, t, message, extra))

1133

1168

 

1134

1169

    def send_error(self, exception):

1135

1170

        if self.response_sent:

1141

1176

                ('UnknownMethod', exception.verb))

1142

1177

            self.send_response(failure)

1143

1178

            return

 

1179

        if 'hpss' in debug.debug_flags:

 

1180

            self._trace('error', str(exception))

1144

1181

        self.response_sent = True

1145

1182

        self._write_protocol_version()

1146

1183

        self._write_headers(self._headers)

1160

1197

            self._write_success_status()

1161

1198

        else:

1162

1199

            self._write_error_status()

 

1200

        if 'hpss' in debug.debug_flags:

 

1201

            self._trace('response', repr(response.args))

1163

1202

        self._write_structure(response.args)

1164

1203

        if response.body is not None:

1165

1204

            self._write_prefixed_body(response.body)

 

1205

            if 'hpss' in debug.debug_flags:

 

1206

                self._trace('body', '%d bytes' % (len(response.body),),

 

1207

                            response.body, include_time=True)

1166

1208

        elif response.body_stream is not None:

 

1209

            count = num_bytes = 0

 

1210

            first_chunk = None

1167

1211

            for exc_info, chunk in _iter_with_errors(response.body_stream):

 

1212

                count += 1

1168

1213

                if exc_info is not None:

1169

1214

                    self._write_error_status()

1170

1215

                    error_struct = request._translate_error(exc_info[1])

1175

1220

                        self._write_error_status()

1176

1221

                        self._write_structure(chunk.args)

1177

1222

                        break

 

1223

                    num_bytes += len(chunk)

 

1224

                    if first_chunk is None:

 

1225

                        first_chunk = chunk

1178

1226

                    self._write_prefixed_body(chunk)

 

1227

                    if 'hpssdetail' in debug.debug_flags:

 

1228

                        # Not worth timing separately, as _write_func is

 

1229

                        # actually buffered

 

1230

                        self._trace('body chunk',

 

1231

                                    '%d bytes' % (len(chunk),),

 

1232

                                    chunk, suppress_time=True)

 

1233

            if 'hpss' in debug.debug_flags:

 

1234

                self._trace('body stream',

 

1235

                            '%d bytes %d chunks' % (num_bytes, count),

 

1236

                            first_chunk)

1179

1237

        self._write_end()

 

1238

        if 'hpss' in debug.debug_flags:

 

1239

            self._trace('response end', '', include_time=True)

1180

1240

 

1181

1241

 

1182

1242

def _iter_with_errors(iterable):

1234

1294

            base = getattr(self._medium_request._medium, 'base', None)

1235

1295

            if base is not None:

1236

1296

                mutter('             (to %s)', base)

1237

 

            self._request_start_time = time.time()

 

1297

            self._request_start_time = osutils.timer_func()

1238

1298

        self._write_protocol_version()

1239

1299

        self._write_headers(self._headers)

1240

1300

        self._write_structure(args)

1252

1312

            if path is not None:

1253

1313

                mutter('                  (to %s)', path)

1254

1314

            mutter('              %d bytes', len(body))

1255

 

            self._request_start_time = time.time()

 

1315

            self._request_start_time = osutils.timer_func()

1256

1316

        self._write_protocol_version()

1257

1317

        self._write_headers(self._headers)

1258

1318

        self._write_structure(args)

1271

1331

            path = getattr(self._medium_request._medium, '_path', None)

1272

1332

            if path is not None:

1273

1333

                mutter('                  (to %s)', path)

1274

 

            self._request_start_time = time.time()

 

1334

            self._request_start_time = osutils.timer_func()

1275

1335

        self._write_protocol_version()

1276

1336

        self._write_headers(self._headers)

1277

1337

        self._write_structure(args)

1288

1348

            path = getattr(self._medium_request._medium, '_path', None)

1289

1349

            if path is not None:

1290

1350

                mutter('                  (to %s)', path)

1291

 

            self._request_start_time = time.time()

 

1351

            self._request_start_time = osutils.timer_func()

1292

1352

        self._write_protocol_version()

1293

1353

        self._write_headers(self._headers)

1294

1354

        self._write_structure(args)

• Older »

Loggerhead is a web-based interface for Breezy
Version: 2.0.2.dev0