1
# Copyright (C) 2006-2010 Canonical Ltd
1
# Copyright (C) 2006 Canonical Ltd
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
46
47
from bzrlib.smart import client, protocol, request, vfs
47
48
from bzrlib.transport import ssh
50
#usually already imported, and getting IllegalScoperReplacer on it here.
49
51
from bzrlib import osutils
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
53
# We must not read any more than 64k at a time so we don't risk "no buffer
54
# space available" errors on some platforms. Windows in particular is likely
55
# to give error 10053 or 10055 if we read more than 64k from a socket.
56
_MAX_READ_SIZE = 64 * 1024
57
59
def _get_protocol_factory_for_bytes(bytes):
58
60
"""Determine the right protocol factory for 'bytes'.
274
276
def _serve_one_request_unguarded(self, protocol):
275
277
while protocol.next_read_size():
276
278
# 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)
279
# than _MAX_READ_SIZE ready, the socket wil just return a short
280
# read immediately rather than block.
281
bytes = self.read_bytes(_MAX_READ_SIZE)
281
283
self.finished = True
285
287
self._push_back(protocol.unused_data)
287
289
def _read_bytes(self, desired_count):
288
return osutils.read_bytes_from_socket(
289
self.socket, self._report_activity)
290
return _read_bytes_from_socket(
291
self.socket.recv, desired_count, self._report_activity)
291
293
def terminate_due_to_error(self):
292
294
# TODO: This should log to a server log file, but no such thing
293
295
# exists yet. Andrew Bennetts 2006-09-29.
296
osutils.until_no_eintr(self.socket.close)
295
297
self.finished = True
297
299
def _write_out(self, bytes):
298
300
tstart = osutils.timer_func()
299
301
osutils.send_all(self.socket, bytes, self._report_activity)
300
302
if 'hpss' in debug.debug_flags:
301
thread_id = thread.get_ident()
303
cur_thread = threading.currentThread()
304
thread_id = getattr(cur_thread, 'ident', None)
305
if thread_id is None:
306
thread_id = cur_thread.getName()
302
307
trace.mutter('%12s: [%s] %d bytes to the socket in %.3fs'
303
308
% ('wrote', thread_id, len(bytes),
304
309
osutils.timer_func() - tstart))
332
337
bytes_to_read = protocol.next_read_size()
333
338
if bytes_to_read == 0:
334
339
# Finished serving this request.
340
osutils.until_no_eintr(self._out.flush)
337
342
bytes = self.read_bytes(bytes_to_read)
339
344
# Connection has been closed.
340
345
self.finished = True
346
osutils.until_no_eintr(self._out.flush)
343
348
protocol.accept_bytes(bytes)
345
350
def _read_bytes(self, desired_count):
346
return self._in.read(desired_count)
351
return osutils.until_no_eintr(self._in.read, desired_count)
348
353
def terminate_due_to_error(self):
349
354
# TODO: This should log to a server log file, but no such thing
350
355
# exists yet. Andrew Bennetts 2006-09-29.
356
osutils.until_no_eintr(self._out.close)
352
357
self.finished = True
354
359
def _write_out(self, bytes):
355
self._out.write(bytes)
360
osutils.until_no_eintr(self._out.write, bytes)
358
363
class SmartClientMediumRequest(object):
607
612
# which is newer than a previously supplied older-than version.
608
613
# This indicates that some smart verb call is not guarded
609
614
# appropriately (it should simply not have been tried).
615
raise AssertionError(
611
616
"_remember_remote_is_before(%r) called, but "
612
617
"_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))
618
% (version_tuple, self._remote_version_is_before))
620
619
self._remote_version_is_before = version_tuple
622
621
def protocol_version(self):
715
714
"""A client medium using simple pipes.
717
716
This client does not manage the pipes: it assumes they will always be open.
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.
724
719
def __init__(self, readable_pipe, writeable_pipe, base):
729
724
def _accept_bytes(self, bytes):
730
725
"""See SmartClientStreamMedium.accept_bytes."""
731
self._writeable_pipe.write(bytes)
726
osutils.until_no_eintr(self._writeable_pipe.write, bytes)
732
727
self._report_activity(len(bytes), 'write')
734
729
def _flush(self):
735
730
"""See SmartClientStreamMedium._flush()."""
736
self._writeable_pipe.flush()
731
osutils.until_no_eintr(self._writeable_pipe.flush)
738
733
def _read_bytes(self, count):
739
734
"""See SmartClientStreamMedium._read_bytes."""
757
752
self._password = password
758
753
self._port = port
759
754
self._username = username
760
# for the benefit of progress making a short description of this
762
self._scheme = 'bzr+ssh'
763
755
# SmartClientStreamMedium stores the repr of this object in its
764
756
# _DebugCounter so we have to store all the values used in our repr
765
757
# method before calling the super init.
769
761
self._vendor = vendor
770
762
self._write_to = None
771
763
self._bzr_remote_path = bzr_remote_path
764
# for the benefit of progress making a short description of this
766
self._scheme = 'bzr+ssh'
773
768
def __repr__(self):
774
if self._port is None:
777
maybe_port = ':%s' % self._port
778
return "%s(%s://%s@%s%s/)" % (
769
return "%s(connected=%r, username=%r, host=%r, port=%r)" % (
779
770
self.__class__.__name__,
785
776
def _accept_bytes(self, bytes):
786
777
"""See SmartClientStreamMedium.accept_bytes."""
787
778
self._ensure_connection()
788
self._write_to.write(bytes)
779
osutils.until_no_eintr(self._write_to.write, bytes)
789
780
self._report_activity(len(bytes), 'write')
791
782
def disconnect(self):
792
783
"""See SmartClientMedium.disconnect()."""
793
784
if not self._connected:
795
self._read_from.close()
796
self._write_to.close()
786
osutils.until_no_eintr(self._read_from.close)
787
osutils.until_no_eintr(self._write_to.close)
797
788
self._ssh_connection.close()
798
789
self._connected = False
822
813
if not self._connected:
823
814
raise errors.MediumNotConnected(self)
824
815
bytes_to_read = min(count, _MAX_READ_SIZE)
825
bytes = self._read_from.read(bytes_to_read)
816
bytes = osutils.until_no_eintr(self._read_from.read, bytes_to_read)
826
817
self._report_activity(len(bytes), 'read')
906
897
"""See SmartClientMedium.read_bytes."""
907
898
if not self._connected:
908
899
raise errors.MediumNotConnected(self)
909
return osutils.read_bytes_from_socket(
910
self._socket, self._report_activity)
900
return _read_bytes_from_socket(
901
self._socket.recv, count, self._report_activity)
913
904
class SmartClientStreamMediumRequest(SmartClientMediumRequest):
950
941
self._medium._flush()
944
def _read_bytes_from_socket(sock, desired_count, report_activity):
945
# We ignore the desired_count because on sockets it's more efficient to
946
# read large chunks (of _MAX_READ_SIZE bytes) at a time.
948
bytes = osutils.until_no_eintr(sock, _MAX_READ_SIZE)
949
except socket.error, e:
950
if len(e.args) and e.args[0] in (errno.ECONNRESET, 10054):
951
# The connection was closed by the other side. Callers expect an
952
# empty string to signal end-of-stream.
957
report_activity(len(bytes), 'read')
added
removed
bzrlib/smart/medium.py
