Rev 4912: Just backport the implementation of osutils.send_all to bzr-2.1. in http://bazaar.launchpad.net/~jameinel/bzr/2.1-client-reconnect-819604

[John Arbash Meinel]

At http://bazaar.launchpad.net/~jameinel/bzr/2.1-client-reconnect-819604

------------------------------------------------------------
revno: 4912
revision-id: john at arbash-meinel.com-20120911074559-p0pre5od6hkpu082
parent: john at arbash-meinel.com-20120911073914-pfhzozat7lk3iqb6
committer: John Arbash Meinel <john at arbash-meinel.com>
branch nick: 2.1-client-reconnect-819604
timestamp: Tue 2012-09-11 11:45:59 +0400
message:
  Just backport the implementation of osutils.send_all to bzr-2.1.
  We know it works well in newer versions of bzr for a long time.
-------------- next part --------------
=== modified file 'bzrlib/osutils.py'
--- a/bzrlib/osutils.py	2010-05-27 04:00:01 +0000
+++ b/bzrlib/osutils.py	2012-09-11 07:45:59 +0000
@@ -40,6 +40,7 @@
     rmtree,
     )
 import signal
+import socket
 import subprocess
 import tempfile
 from tempfile import (
@@ -1929,6 +1930,20 @@
         return socket.gethostname().decode(get_user_encoding())
 
 
+# We must not read/write any more than 64k at a time from/to a socket so we
+# don't risk "no buffer space available" errors on some platforms.  Windows in
+# particular is likely to throw WSAECONNABORTED or WSAENOBUFS if given too much
+# data at once.
+MAX_SOCKET_CHUNK = 64 * 1024
+
+_end_of_stream_errors = [errno.ECONNRESET, errno.EPIPE, errno.EINVAL]
+for _eno in ['WSAECONNRESET', 'WSAECONNABORTED']:
+    _eno = getattr(errno, _eno, None)
+    if _eno is not None:
+        _end_of_stream_errors.append(_eno)
+del _eno
+
+
 def recv_all(socket, bytes):
     """Receive an exact number of bytes.
 
@@ -1948,21 +1963,33 @@
     return b
 
 
-def send_all(socket, bytes, report_activity=None):
+def send_all(sock, bytes, report_activity=None):
     """Send all bytes on a socket.
 
-    Regular socket.sendall() can give socket error 10053 on Windows.  This
-    implementation sends no more than 64k at a time, which avoids this problem.
+    Breaks large blocks in smaller chunks to avoid buffering limitations on
+    some platforms, and catches EINTR which may be thrown if the send is
+    interrupted by a signal.
+
+    This is preferred to socket.sendall(), because it avoids portability bugs
+    and provides activity reporting.
 
     :param report_activity: Call this as bytes are read, see
         Transport._report_activity
     """
-    chunk_size = 2**16
-    for pos in xrange(0, len(bytes), chunk_size):
-        block = bytes[pos:pos+chunk_size]
-        if report_activity is not None:
-            report_activity(len(block), 'write')
-        until_no_eintr(socket.sendall, block)
+    sent_total = 0
+    byte_count = len(bytes)
+    while sent_total < byte_count:
+        try:
+            sent = sock.send(buffer(bytes, sent_total, MAX_SOCKET_CHUNK))
+        except (socket.error, IOError), e:
+            if e.args[0] in _end_of_stream_errors:
+                raise errors.ConnectionReset(
+                    "Error trying to write to socket", e)
+            if e.args[0] != errno.EINTR:
+                raise
+        else:
+            sent_total += sent
+            report_activity(sent, 'write')
 
 
 def dereference_path(path):

=== modified file 'bzrlib/tests/test_osutils.py'
--- a/bzrlib/tests/test_osutils.py	2010-11-30 20:42:42 +0000
+++ b/bzrlib/tests/test_osutils.py	2012-09-11 07:45:59 +0000
@@ -801,6 +801,27 @@
         self.assertEqual(None, osutils.safe_file_id(None))
 
 
+class TestSendAll(tests.TestCase):
+
+    def test_send_with_disconnected_socket(self):
+        class DisconnectedSocket(object):
+            def __init__(self, err):
+                self.err = err
+            def send(self, content):
+                raise self.err
+            def close(self):
+                pass
+        # All of these should be treated as ConnectionReset
+        errs = []
+        for err_cls in (IOError, socket.error):
+            for errnum in osutils._end_of_stream_errors:
+                errs.append(err_cls(errnum))
+        for err in errs:
+            sock = DisconnectedSocket(err)
+            self.assertRaises(errors.ConnectionReset,
+                osutils.send_all, sock, 'some more content')
+
+
 class TestWin32Funcs(tests.TestCase):
     """Test that _win32 versions of os utilities return appropriate paths."""