← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/smart/medium.py

  • Committer: John Arbash Meinel
  • Date: 2009-03-27 22:29:55 UTC
  • mto: (3735.39.2 clean)
  • mto: This revision was merged to the branch mainline in revision 4280.
  • Revision ID: john@arbash-meinel.com-20090327222955-utifmfm888zerixt
Implement apply_delta_to_source which doesn't have to malloc another string.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/smart/medium.py
1
 
# Copyright (C) 2006-2010 Canonical Ltd
 
1
# Copyright (C) 2006 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
24
24
bzrlib/transport/smart/__init__.py.
25
25
"""
26
26
 
 
27
import errno
27
28
import os
 
29
import socket
28
30
import sys
29
31
import urllib
30
32
 
31
33
from bzrlib.lazy_import import lazy_import
32
34
lazy_import(globals(), """
33
35
import atexit
34
 
import socket
35
 
import thread
36
36
import weakref
37
 
 
38
37
from bzrlib import (
39
38
    debug,
40
39
    errors,
 
40
    osutils,
41
41
    symbol_versioning,
42
42
    trace,
43
43
    ui,
44
44
    urlutils,
45
45
    )
46
 
from bzrlib.smart import client, protocol, request, vfs
 
46
from bzrlib.smart import client, protocol
47
47
from bzrlib.transport import ssh
48
48
""")
49
 
from bzrlib import osutils
50
 
 
51
 
# Throughout this module buffer size parameters are either limited to be at
52
 
# most _MAX_READ_SIZE, or are ignored and _MAX_READ_SIZE is used instead.
53
 
# For this module's purposes, MAX_SOCKET_CHUNK is a reasonable size for reads
54
 
# from non-sockets as well.
55
 
_MAX_READ_SIZE = osutils.MAX_SOCKET_CHUNK
 
49
 
 
50
 
 
51
# We must not read any more than 64k at a time so we don't risk "no buffer
 
52
# space available" errors on some platforms.  Windows in particular is likely
 
53
# to give error 10053 or 10055 if we read more than 64k from a socket.
 
54
_MAX_READ_SIZE = 64 * 1024
 
55
 
56
56
 
57
57
def _get_protocol_factory_for_bytes(bytes):
58
58
    """Determine the right protocol factory for 'bytes'.
274
274
    def _serve_one_request_unguarded(self, protocol):
275
275
        while protocol.next_read_size():
276
276
            # We can safely try to read large chunks.  If there is less data
277
 
            # than MAX_SOCKET_CHUNK ready, the socket will just return a
278
 
            # short read immediately rather than block.
279
 
            bytes = self.read_bytes(osutils.MAX_SOCKET_CHUNK)
 
277
            # than _MAX_READ_SIZE ready, the socket wil just return a short
 
278
            # read immediately rather than block.
 
279
            bytes = self.read_bytes(_MAX_READ_SIZE)
280
280
            if bytes == '':
281
281
                self.finished = True
282
282
                return
285
285
        self._push_back(protocol.unused_data)
286
286
 
287
287
    def _read_bytes(self, desired_count):
288
 
        return osutils.read_bytes_from_socket(
289
 
            self.socket, self._report_activity)
 
288
        # We ignore the desired_count because on sockets it's more efficient to
 
289
        # read large chunks (of _MAX_READ_SIZE bytes) at a time.
 
290
        bytes = osutils.until_no_eintr(self.socket.recv, _MAX_READ_SIZE)
 
291
        self._report_activity(len(bytes), 'read')
 
292
        return bytes
290
293
 
291
294
    def terminate_due_to_error(self):
292
295
        # TODO: This should log to a server log file, but no such thing
295
298
        self.finished = True
296
299
 
297
300
    def _write_out(self, bytes):
298
 
        tstart = osutils.timer_func()
299
301
        osutils.send_all(self.socket, bytes, self._report_activity)
300
 
        if 'hpss' in debug.debug_flags:
301
 
            thread_id = thread.get_ident()
302
 
            trace.mutter('%12s: [%s] %d bytes to the socket in %.3fs'
303
 
                         % ('wrote', thread_id, len(bytes),
304
 
                            osutils.timer_func() - tstart))
305
302
 
306
303
 
307
304
class SmartServerPipeStreamMedium(SmartServerStreamMedium):
384
381
    def accept_bytes(self, bytes):
385
382
        """Accept bytes for inclusion in this request.
386
383
 
387
 
        This method may not be called after finished_writing() has been
 
384
        This method may not be be called after finished_writing() has been
388
385
        called.  It depends upon the Medium whether or not the bytes will be
389
386
        immediately transmitted. Message based Mediums will tend to buffer the
390
387
        bytes until finished_writing() is called.
478
475
        if not line.endswith('\n'):
479
476
            # end of file encountered reading from server
480
477
            raise errors.ConnectionReset(
481
 
                "Unexpected end of message. Please check connectivity "
482
 
                "and permissions, and report a bug if problems persist.")
 
478
                "please check connectivity and permissions")
483
479
        return line
484
480
 
485
481
    def _read_line(self):
513
509
        """
514
510
        medium_repr = repr(medium)
515
511
        # Add this medium to the WeakKeyDictionary
516
 
        self.counts[medium] = dict(count=0, vfs_count=0,
517
 
                                   medium_repr=medium_repr)
 
512
        self.counts[medium] = [0, medium_repr]
518
513
        # Weakref callbacks are fired in reverse order of their association
519
514
        # with the referenced object.  So we add a weakref *after* adding to
520
515
        # the WeakKeyDict so that we can report the value from it before the
524
519
    def increment_call_count(self, params):
525
520
        # Increment the count in the WeakKeyDictionary
526
521
        value = self.counts[params.medium]
527
 
        value['count'] += 1
528
 
        try:
529
 
            request_method = request.request_handlers.get(params.method)
530
 
        except KeyError:
531
 
            # A method we don't know about doesn't count as a VFS method.
532
 
            return
533
 
        if issubclass(request_method, vfs.VfsRequest):
534
 
            value['vfs_count'] += 1
 
522
        value[0] += 1
535
523
 
536
524
    def done(self, ref):
537
525
        value = self.counts[ref]
538
 
        count, vfs_count, medium_repr = (
539
 
            value['count'], value['vfs_count'], value['medium_repr'])
 
526
        count, medium_repr = value
540
527
        # In case this callback is invoked for the same ref twice (by the
541
528
        # weakref callback and by the atexit function), set the call count back
542
529
        # to 0 so this item won't be reported twice.
543
 
        value['count'] = 0
544
 
        value['vfs_count'] = 0
 
530
        value[0] = 0
545
531
        if count != 0:
546
 
            trace.note('HPSS calls: %d (%d vfs) %s',
547
 
                       count, vfs_count, medium_repr)
 
532
            trace.note('HPSS calls: %d %s', count, medium_repr)
548
533
 
549
534
    def flush_all(self):
550
535
        for ref in list(self.counts.keys()):
607
592
            # which is newer than a previously supplied older-than version.
608
593
            # This indicates that some smart verb call is not guarded
609
594
            # appropriately (it should simply not have been tried).
610
 
            trace.mutter(
 
595
            raise AssertionError(
611
596
                "_remember_remote_is_before(%r) called, but "
612
597
                "_remember_remote_is_before(%r) was called previously."
613
 
                , version_tuple, self._remote_version_is_before)
614
 
            if 'hpss' in debug.debug_flags:
615
 
                ui.ui_factory.show_warning(
616
 
                    "_remember_remote_is_before(%r) called, but "
617
 
                    "_remember_remote_is_before(%r) was called previously."
618
 
                    % (version_tuple, self._remote_version_is_before))
619
 
            return
 
598
                % (version_tuple, self._remote_version_is_before))
620
599
        self._remote_version_is_before = version_tuple
621
600
 
622
601
    def protocol_version(self):
715
694
    """A client medium using simple pipes.
716
695
 
717
696
    This client does not manage the pipes: it assumes they will always be open.
718
 
 
719
 
    Note that if readable_pipe.read might raise IOError or OSError with errno
720
 
    of EINTR, it must be safe to retry the read.  Plain CPython fileobjects
721
 
    (such as used for sys.stdin) are safe.
722
697
    """
723
698
 
724
699
    def __init__(self, readable_pipe, writeable_pipe, base):
737
712
 
738
713
    def _read_bytes(self, count):
739
714
        """See SmartClientStreamMedium._read_bytes."""
740
 
        bytes = osutils.until_no_eintr(self._readable_pipe.read, count)
 
715
        bytes = self._readable_pipe.read(count)
741
716
        self._report_activity(len(bytes), 'read')
742
717
        return bytes
743
718
 
757
732
        self._password = password
758
733
        self._port = port
759
734
        self._username = username
760
 
        # for the benefit of progress making a short description of this
761
 
        # transport
762
 
        self._scheme = 'bzr+ssh'
763
735
        # SmartClientStreamMedium stores the repr of this object in its
764
736
        # _DebugCounter so we have to store all the values used in our repr
765
737
        # method before calling the super init.
769
741
        self._vendor = vendor
770
742
        self._write_to = None
771
743
        self._bzr_remote_path = bzr_remote_path
 
744
        # for the benefit of progress making a short description of this
 
745
        # transport
 
746
        self._scheme = 'bzr+ssh'
772
747
 
773
748
    def __repr__(self):
774
 
        if self._port is None:
775
 
            maybe_port = ''
776
 
        else:
777
 
            maybe_port = ':%s' % self._port
778
 
        return "%s(%s://%s@%s%s/)" % (
 
749
        return "%s(connected=%r, username=%r, host=%r, port=%r)" % (
779
750
            self.__class__.__name__,
780
 
            self._scheme,
 
751
            self._connected,
781
752
            self._username,
782
753
            self._host,
783
 
            maybe_port)
 
754
            self._port)
784
755
 
785
756
    def _accept_bytes(self, bytes):
786
757
        """See SmartClientStreamMedium.accept_bytes."""
906
877
        """See SmartClientMedium.read_bytes."""
907
878
        if not self._connected:
908
879
            raise errors.MediumNotConnected(self)
909
 
        return osutils.read_bytes_from_socket(
910
 
            self._socket, self._report_activity)
 
880
        # We ignore the desired_count because on sockets it's more efficient to
 
881
        # read large chunks (of _MAX_READ_SIZE bytes) at a time.
 
882
        try:
 
883
            bytes = osutils.until_no_eintr(self._socket.recv, _MAX_READ_SIZE)
 
884
        except socket.error, e:
 
885
            if len(e.args) and e.args[0] == errno.ECONNRESET:
 
886
                # Callers expect an empty string in that case
 
887
                return ''
 
888
            else:
 
889
                raise
 
890
        else:
 
891
            self._report_activity(len(bytes), 'read')
 
892
            return bytes
911
893
 
912
894
 
913
895
class SmartClientStreamMediumRequest(SmartClientMediumRequest):
949
931
        """
950
932
        self._medium._flush()
951
933
 
952