Merge lp:~spiv/bzr/no_until_no_eintr into lp:bzr

[Andrew Bennetts]

(andrew) Either correctly handle EINTR or don't handle it at all. (#496813)

This change removes the unsafe uses of until_no_eintr, based on the bug reports and patches from the very patient Martin <gz>. Unlike Martin <gz>'s original patch it doesn't remove the until_no_eintr helper entirely, because it is safe (and useful) to use for reading from a pipe.

(In other changes we've already reduced the number of signal handlers we install that will trigger EINTR, but it can still occur, so, when practical, we may as well handle it correctly rather than fail.)

The most alarming diff hunk is:

- return '\x01'.join(args) + '\n'
+ joined = '\x01'.join(args) + '\n'
+ if type(joined) is unicode:
+ # XXX: We should fix things so this never happens! -AJB, 04032010.
+ mutter('response args contain unicode, should be only bytes: %r',
+ joined)
+ joined = joined.encode('ascii')
+ return joined

This is a latent bug in the HPSS protocol v1 (and maybe v2) code now exposed by the change to use buffer() in send_all. This change explicitly does the convert-to-ASCII that was being done anyway. v3 is unaffected.

The diff applies cleanly to 2.1 (aside from the NEWS entry, of course), and I'll submit a backport for review once this proposal is accepted.

[Martin Pool (mbp)]

nice

review: Approve

[Martin Pool (mbp)]

If Martin <gz> gets the chance to review it that would be good, otherwise I suggest you merge.

[Martin Packman (gz)]

Interesting on the encode tuple unicode issue, I didn't see any issues there in testing. Nit, you've got the ISO 8601 basic format wrong, it's big endian.

Having thought about this a bit more, I'm not mad about a bzrlib.smart.medium focused change. The module is already wrappers around wrappers around wrappers, and it's too hard to work out what's going on. Case in point - as far as I can tell, SmartSSHClientMedium still uses get_filelike_channels which uses an unsafe wrapper for sockets pre-2.7 in _ParamikoSSHConnection and non-EINTR-protected pipes in SSHSubprocess. This means we're not covering codepaths like the one bug 341535 highlighted.

I've been waiting for magic bullet inspiration on a clean fix here, but nothing's struck yet. Perhaps a different approach, like unifying on one lowest-common-denominator wrapper for both sockets and pipe fds and using those throughout would be clearer and mean less copy-code to get subtly wrong in certain versions.

review: Needs Information

[Andrew Bennetts (spiv)]

Martin [gz] wrote:
[...]
> I've been waiting for magic bullet inspiration on a clean fix here, but
> nothing's struck yet. Perhaps a different approach, like unifying on one
> lowest-common-denominator wrapper for both sockets and pipe fds and using
> those throughout would be clearer and mean less copy-code to get subtly wrong
> in certain versions.

I've been wondering about that approach too. It might simplify transport
activity reporting too.

-Andrew.

[Andrew Bennetts (spiv)]

Martin [gz] wrote:
> Interesting on the encode tuple unicode issue, I didn't see any issues there
> in testing. Nit, you've got the ISO 8601 basic format wrong, it's big endian.

Heh, fixed :)

> Having thought about this a bit more, I'm not mad about a bzrlib.smart.medium
> focused change. The module is already wrappers around wrappers around
> wrappers, and it's too hard to work out what's going on. Case in point - as
> far as I can tell, SmartSSHClientMedium still uses get_filelike_channels which
> uses an unsafe wrapper for sockets pre-2.7 in _ParamikoSSHConnection and
> non-EINTR-protected pipes in SSHSubprocess. This means we're not covering
> codepaths like the one bug 341535 highlighted.

That's true. However I've just sent this patch to PQM anyway because I think
it's still a clear improvement. Here's my reasoning:

  - I think regressing 341535 is better than risking silent corruption of HPSS data
    (and certainly better than a confusing error caused by corrupted HPSS data),
  - Our SIGWINCH handler uses siginterrupt to avoid triggering EINTR, and
  - bzrlib's only other signal handler is SIGQUIT, which is a developer tool not
    a production feature.

I'll reopen 341535, of course.

Thanks for another thoughtful review!

=== modified file 'NEWS'
--- NEWS	2010-03-25 12:33:15 +0000
+++ NEWS	2010-03-26 04:50:44 +0000
@@ -86,6 +86,9 @@
 * Added docstring for ``Tree.iter_changes``
   (John Arbash Meinel, #304182)
 
+* Allow additional arguments to
+  ``RemoteRepository.add_inventory_by_delta()``. (Jelmer Vernooij, #532631)
+
 * Allow exporting a single file using ``bzr export``.
   (Michal Junák, #511987)
 
@@ -144,6 +147,13 @@
 * Fix stub sftp test server to call os.getcwdu().
   (Vincent Ladeuil, #526211, #526353)
 
+* Many IO operations that returned ``EINTR`` were retried even if it
+  wasn't safe to do so via careless use of ``until_no_eintr``.  Bazaar now
+  only retries operations that are safe to retry, and in some cases has
+  switched to operations that can be retried (e.g. ``sock.send`` rather than
+  ``sock.sendall``).
+  (Andrew Bennetts, Martin <[e-mail]>, #496813)
+
 * Path conflicts now support --take-this and --take-other even when a
   deletion is involved.
   (Vincent Ladeuil, #531967)
@@ -163,9 +173,6 @@
   the debugger won't kill the session.
   (Martin <[e-mail]>, #162502)
 
-* Allow additional arguments to
-  ``RemoteRepository.add_inventory_by_delta()``. (Jelmer Vernooij, #532631)
-
 * Tolerate patches with leading noise in ``bzr-handle-patch``.
   (Toshio Kuratomi, Martin Pool, #502076)
 
 
=== modified file 'bzrlib/osutils.py'
--- bzrlib/osutils.py	2010-03-25 17:04:08 +0000
+++ bzrlib/osutils.py	2010-03-26 04:50:44 +0000
@@ -40,6 +40,7 @@
     rmtree,
     )
 import signal
+import socket
 import subprocess
 import tempfile
 from tempfile import (
@@ -50,12 +51,16 @@
 from bzrlib import (
     cache_utf8,
     errors,
+    trace,
     win32utils,
-    trace,
     )
-
 """)
 
+from bzrlib.symbol_versioning import (
+    deprecated_function,
+    deprecated_in,
+    )
+
 # sha and md5 modules are deprecated in python2.6 but hashlib is available as
 # of 2.5
 if sys.version_info < (2, 5):
@@ -1957,40 +1962,82 @@
         return socket.gethostname().decode(get_user_encoding())
 
 
-def recv_all(socket, bytes):
+# 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
+
+def read_bytes_from_socket(sock, report_activity=None,
+        max_read_size=MAX_SOCKET_CHUNK):
+    """Read up to max_read_size of bytes from sock and notify of progress.
+
+    Translates "Connection reset by peer" into file-like EOF (return an
+    empty string rather than raise an error), and repeats the recv if
+    interrupted by a signal.
+    """
+    while 1:
+        try:
+            bytes = sock.recv(max_read_size)
+        except socket.error, e:
+            eno = e.args[0]
+            if eno == getattr(errno, "WSAECONNRESET", errno.ECONNRESET):
+                # The connection was closed by the other side.  Callers expect
+                # an empty string to signal end-of-stream.
+                return ""
+            elif eno == errno.EINTR:
+                # Retry the interrupted recv.
+                continue
+            raise
+        else:
+            if report_activity is not None:
+                report_activity(len(bytes), 'read')
+            return bytes
+
+
+def recv_all(socket, count):
     """Receive an exact number of bytes.
 
     Regular Socket.recv() may return less than the requested number of bytes,
-    dependning on what's in the OS buffer.  MSG_WAITALL is not available
+    depending on what's in the OS buffer.  MSG_WAITALL is not available
     on all platforms, but this should work everywhere.  This will return
     less than the requested amount if the remote end closes.
 
     This isn't optimized and is intended mostly for use in testing.
     """
     b = ''
-    while len(b) < bytes:
-        new = until_no_eintr(socket.recv, bytes - len(b))
+    while len(b) < count:
+        new = read_bytes_from_socket(socket, None, count - len(b))
         if new == '':
             break # eof
         b += new
     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, e:
+            if e.args[0] != errno.EINTR:
+                raise
+        else:
+            sent_total += sent
+            report_activity(sent, 'write')
 
 
 def dereference_path(path):
@@ -2067,7 +2114,18 @@
 
 
 def until_no_eintr(f, *a, **kw):
-    """Run f(*a, **kw), retrying if an EINTR error occurs."""
+    """Run f(*a, **kw), retrying if an EINTR error occurs.
+    
+    WARNING: you must be certain that it is safe to retry the call repeatedly
+    if EINTR does occur.  This is typically only true for low-level operations
+    like os.read.  If in any doubt, don't use this.
+
+    Keep in mind that this is not a complete solution to EINTR.  There is
+    probably code in the Python standard library and other dependencies that
+    may encounter EINTR if a signal arrives (and there is signal handler for
+    that signal).  So this function can reduce the impact for IO that bzrlib
+    directly controls, but it is not a complete solution.
+    """
     # Borrowed from Twisted's twisted.python.util.untilConcludes function.
     while True:
         try:
@@ -2077,6 +2135,7 @@
                 continue
             raise
 
+
 def re_compile_checked(re_string, flags=0, where=""):
     """Return a compiled re, or raise a sensible error.
 
 
=== modified file 'bzrlib/smart/medium.py'
--- bzrlib/smart/medium.py	2010-02-17 17:11:16 +0000
+++ bzrlib/smart/medium.py	2010-03-26 04:50:44 +0000
@@ -24,15 +24,14 @@
 bzrlib/transport/smart/__init__.py.
 """
 
-import errno
 import os
-import socket
 import sys
 import urllib
 
 from bzrlib.lazy_import import lazy_import
 lazy_import(globals(), """
 import atexit
+import socket
 import thread
 import weakref
 
@@ -47,14 +46,13 @@
 from bzrlib.smart import client, protocol, request, vfs
 from bzrlib.transport import ssh
 """)
-#usually already imported, and getting IllegalScoperReplacer on it here.
 from bzrlib import osutils
 
-# We must not read any more than 64k at a time so we don't risk "no buffer
-# space available" errors on some platforms.  Windows in particular is likely
-# to give error 10053 or 10055 if we read more than 64k from a socket.
-_MAX_READ_SIZE = 64 * 1024
-
+# Throughout this module buffer size parameters are either limited to be at
+# most _MAX_READ_SIZE, or are ignored and _MAX_READ_SIZE is used instead.
+# For this module's purposes, MAX_SOCKET_CHUNK is a reasonable size for reads
+# from non-sockets as well.
+_MAX_READ_SIZE = osutils.MAX_SOCKET_CHUNK
 
 def _get_protocol_factory_for_bytes(bytes):
     """Determine the right protocol factory for 'bytes'.
@@ -276,9 +274,9 @@
     def _serve_one_request_unguarded(self, protocol):
         while protocol.next_read_size():
             # We can safely try to read large chunks.  If there is less data
-            # than _MAX_READ_SIZE ready, the socket wil just return a short
-            # read immediately rather than block.
-            bytes = self.read_bytes(_MAX_READ_SIZE)
+            # than MAX_SOCKET_CHUNK ready, the socket will just return a
+            # short read immediately rather than block.
+            bytes = self.read_bytes(osutils.MAX_SOCKET_CHUNK)
             if bytes == '':
                 self.finished = True
                 return
@@ -287,13 +285,13 @@
         self._push_back(protocol.unused_data)
 
     def _read_bytes(self, desired_count):
-        return _read_bytes_from_socket(
-            self.socket.recv, desired_count, self._report_activity)
+        return osutils.read_bytes_from_socket(
+            self.socket, self._report_activity)
 
     def terminate_due_to_error(self):
         # TODO: This should log to a server log file, but no such thing
         # exists yet.  Andrew Bennetts 2006-09-29.
-        osutils.until_no_eintr(self.socket.close)
+        self.socket.close()
         self.finished = True
 
     def _write_out(self, bytes):
@@ -334,27 +332,27 @@
             bytes_to_read = protocol.next_read_size()
             if bytes_to_read == 0:
                 # Finished serving this request.
-                osutils.until_no_eintr(self._out.flush)
+                self._out.flush()
                 return
             bytes = self.read_bytes(bytes_to_read)
             if bytes == '':
                 # Connection has been closed.
                 self.finished = True
-                osutils.until_no_eintr(self._out.flush)
+                self._out.flush()
                 return
             protocol.accept_bytes(bytes)
 
     def _read_bytes(self, desired_count):
-        return osutils.until_no_eintr(self._in.read, desired_count)
+        return self._in.read(desired_count)
 
     def terminate_due_to_error(self):
         # TODO: This should log to a server log file, but no such thing
         # exists yet.  Andrew Bennetts 2006-09-29.
-        osutils.until_no_eintr(self._out.close)
+        self._out.close()
         self.finished = True
 
     def _write_out(self, bytes):
-        osutils.until_no_eintr(self._out.write, bytes)
+        self._out.write(bytes)
 
 
 class SmartClientMediumRequest(object):
@@ -711,6 +709,10 @@
     """A client medium using simple pipes.
 
     This client does not manage the pipes: it assumes they will always be open.
+
+    Note that if readable_pipe.read might raise IOError or OSError with errno
+    of EINTR, it must be safe to retry the read.  Plain CPython fileobjects
+    (such as used for sys.stdin) are safe.
     """
 
     def __init__(self, readable_pipe, writeable_pipe, base):
@@ -720,12 +722,12 @@
 
     def _accept_bytes(self, bytes):
         """See SmartClientStreamMedium.accept_bytes."""
-        osutils.until_no_eintr(self._writeable_pipe.write, bytes)
+        self._writeable_pipe.write(bytes)
         self._report_activity(len(bytes), 'write')
 
     def _flush(self):
         """See SmartClientStreamMedium._flush()."""
-        osutils.until_no_eintr(self._writeable_pipe.flush)
+        self._writeable_pipe.flush()
 
     def _read_bytes(self, count):
         """See SmartClientStreamMedium._read_bytes."""
@@ -777,15 +779,15 @@
     def _accept_bytes(self, bytes):
         """See SmartClientStreamMedium.accept_bytes."""
         self._ensure_connection()
-        osutils.until_no_eintr(self._write_to.write, bytes)
+        self._write_to.write(bytes)
         self._report_activity(len(bytes), 'write')
 
     def disconnect(self):
         """See SmartClientMedium.disconnect()."""
         if not self._connected:
             return
-        osutils.until_no_eintr(self._read_from.close)
-        osutils.until_no_eintr(self._write_to.close)
+        self._read_from.close()
+        self._write_to.close()
         self._ssh_connection.close()
         self._connected = False
 
@@ -814,7 +816,7 @@
         if not self._connected:
             raise errors.MediumNotConnected(self)
         bytes_to_read = min(count, _MAX_READ_SIZE)
-        bytes = osutils.until_no_eintr(self._read_from.read, bytes_to_read)
+        bytes = self._read_from.read(bytes_to_read)
         self._report_activity(len(bytes), 'read')
         return bytes
 
@@ -844,7 +846,7 @@
         """See SmartClientMedium.disconnect()."""
         if not self._connected:
             return
-        osutils.until_no_eintr(self._socket.close)
+        self._socket.close()
         self._socket = None
         self._connected = False
 
@@ -898,8 +900,8 @@
         """See SmartClientMedium.read_bytes."""
         if not self._connected:
             raise errors.MediumNotConnected(self)
-        return _read_bytes_from_socket(
-            self._socket.recv, count, self._report_activity)
+        return osutils.read_bytes_from_socket(
+            self._socket, self._report_activity)
 
 
 class SmartClientStreamMediumRequest(SmartClientMediumRequest):
@@ -942,19 +944,3 @@
         self._medium._flush()
 
 
-def _read_bytes_from_socket(sock, desired_count, report_activity):
-    # We ignore the desired_count because on sockets it's more efficient to
-    # read large chunks (of _MAX_READ_SIZE bytes) at a time.
-    try:
-        bytes = osutils.until_no_eintr(sock, _MAX_READ_SIZE)
-    except socket.error, e:
-        if len(e.args) and e.args[0] in (errno.ECONNRESET, 10054):
-            # The connection was closed by the other side.  Callers expect an
-            # empty string to signal end-of-stream.
-            bytes = ''
-        else:
-            raise
-    else:
-        report_activity(len(bytes), 'read')
-    return bytes
-
 
=== modified file 'bzrlib/smart/protocol.py'
--- bzrlib/smart/protocol.py	2010-02-17 17:11:16 +0000
+++ bzrlib/smart/protocol.py	2010-03-26 04:50:44 +0000
@@ -62,7 +62,13 @@
 
 def _encode_tuple(args):
     """Encode the tuple args to a bytestream."""
-    return '\x01'.join(args) + '\n'
+    joined = '\x01'.join(args) + '\n'
+    if type(joined) is unicode:
+        # XXX: We should fix things so this never happens!  -AJB, 20100304
+        mutter('response args contain unicode, should be only bytes: %r',
+               joined)
+        joined = joined.encode('ascii')
+    return joined
 
 
 class Requester(object):
 
=== modified file 'bzrlib/smart/server.py'
--- bzrlib/smart/server.py	2010-02-23 07:43:11 +0000
+++ bzrlib/smart/server.py	2010-03-26 04:50:44 +0000
@@ -138,6 +138,8 @@
                         if e.args[0] != errno.EBADF:
                             trace.warning("listening socket error: %s", e)
                     else:
+                        if self._should_terminate:
+                            break
                         self.serve_conn(conn, thread_name_suffix)
             except KeyboardInterrupt:
                 # dont log when CTRL-C'd.