Fix gevent.os.nb_write/read not always closing the IO event the opened, which...

Fix gevent.os.nb_write/read not always closing the IO event the opened, which is a problem on libuv.

CHANGES.rst

@@ -13,6 +13,10 @@
 - Fix embedded uses of :func:`gevent.Greenlet.spawn`, especially under
   uwsgi. Reported in :issue:`1212` by Kunal Gangakhedkar.
 
+- Fix :func:`gevent.os.nb_write` and :func:`gevent.os.nb_read` not
+  always closing the IO event they opened in the event of an
+  exception. This would be a problem especially for libuv.
+
 1.3.0 (2018-05-11)
 ==================
 

src/gevent/os.py

@@ -74,61 +74,76 @@ if fcntl:
     __extensions__ += ['make_nonblocking', 'nb_read', 'nb_write']
 
     def make_nonblocking(fd):
-        """Put the file descriptor *fd* into non-blocking mode if possible.
+        """Put the file descriptor *fd* into non-blocking mode if
+        possible.
 
-        :return: A boolean value that evaluates to True if successful."""
+        :return: A boolean value that evaluates to True if successful.
+        """
         flags = fcntl.fcntl(fd, fcntl.F_GETFL, 0)
         if not bool(flags & os.O_NONBLOCK):
             fcntl.fcntl(fd, fcntl.F_SETFL, flags | os.O_NONBLOCK)
             return True
 
     def nb_read(fd, n):
-        """Read up to `n` bytes from file descriptor `fd`. Return a string
-        containing the bytes read. If end-of-file is reached, an empty string
-        is returned.
+        """
+        Read up to *n* bytes from file descriptor *fd*. Return a
+        byte string containing the bytes read, which may be shorter than
+        *n*. If end-of-file is reached, an empty string is returned.
 
         The descriptor must be in non-blocking mode.
         """
-        hub, event = None, None
-        while True:
-            try:
-                result = _read(fd, n)
-                if event is not None:
-                    event.close()
-                return result
-            except OSError as e:
-                if e.errno not in ignored_errors:
-                    raise
-                if not PY3:
-                    sys.exc_clear()
-            if hub is None:
-                hub = get_hub()
-                event = hub.loop.io(fd, 1)
-            hub.wait(event)
+        hub = None
+        event = None
+        try:
+            while 1:
+                try:
+                    result = _read(fd, n)
+                    return result
+                except OSError as e:
+                    if e.errno not in ignored_errors:
+                        raise
+                    if not PY3:
+                        sys.exc_clear()
+                if hub is None:
+                    hub = get_hub()
+                    event = hub.loop.io(fd, 1)
+                hub.wait(event)
+        finally:
+            if event is not None:
+                event.close()
+                event = None
+                hub = None
 
 
     def nb_write(fd, buf):
-        """Write bytes from buffer `buf` to file descriptor `fd`. Return the
-        number of bytes written.
+        """
+        Write some number of bytes from buffer *buf* to file
+        descriptor *fd*. Return the number of bytes written, which may
+        be less than the length of *buf*.
 
         The file descriptor must be in non-blocking mode.
         """
-        hub, event = None, None
-        while True:
-            try:
-                result = _write(fd, buf)
-                if event is not None:
-                    event.close()
-                return result
-            except OSError as e:
-                if e.errno not in ignored_errors:
-                    raise
-                if not PY3:
-                    sys.exc_clear()
-            if hub is None:
-                hub = get_hub()
-                event = hub.loop.io(fd, 2)
-            hub.wait(event)
+        hub = None
+        event = None
+        try:
+            while 1:
+                try:
+                    result = _write(fd, buf)
+                    return result
+                except OSError as e:
+                    if e.errno not in ignored_errors:
+                        raise
+                    if not PY3:
+                        sys.exc_clear()
+                if hub is None:
+                    hub = get_hub()
+                    event = hub.loop.io(fd, 2)
+                hub.wait(event)
+        finally:
+            if event is not None:
+                event.close()
+                event = None
+                hub = None
 
 
 def tp_read(fd, n):

src/greentest/test__os.py

+from __future__ import print_function, absolute_import, division
+
 import sys
-from greentest import six
 from os import pipe
+
+
 import gevent
 from gevent import os
 from greentest import TestCase, main, LARGE_TIMEOUT
 from gevent import Greenlet, joinall
+
+from greentest import mock
+from greentest import six
 from greentest.skipping import skipOnLibuvOnPyPyOnWin
 
 
@@ -15,11 +21,8 @@ class TestOS_tp(TestCase):
     def pipe(self):
         return pipe()
 
-    def read(self, *args):
-        return os.tp_read(*args)
-
-    def write(self, *args):
-        return os.tp_write(*args)
+    read = staticmethod(os.tp_read)
+    write = staticmethod(os.tp_write)
 
     @skipOnLibuvOnPyPyOnWin("Sometimes times out")
     def _test_if_pipe_blocks(self, buffer_class):
@@ -68,17 +71,79 @@ if hasattr(os, 'make_nonblocking'):
 
     class TestOS_nb(TestOS_tp):
 
+        read = staticmethod(os.nb_read)
+        write = staticmethod(os.nb_write)
+
         def pipe(self):
             r, w = super(TestOS_nb, self).pipe()
             os.make_nonblocking(r)
             os.make_nonblocking(w)
             return r, w
 
-        def read(self, *args):
-            return os.nb_read(*args)
-
-        def write(self, *args):
-            return os.nb_write(*args)
+        def _make_ignored_oserror(self):
+            import errno
+            ignored_oserror = OSError()
+            ignored_oserror.errno = errno.EINTR
+            return ignored_oserror
+
+
+        def _check_hub_event_closed(self, mock_get_hub, fd, event):
+            mock_get_hub.assert_called_once_with()
+            hub = mock_get_hub.return_value
+            io = hub.loop.io
+            io.assert_called_once_with(fd, event)
+
+            event = io.return_value
+            event.close.assert_called_once_with()
+
+        def _test_event_closed_on_normal_io(self, nb_func, nb_arg,
+                                            mock_io, mock_get_hub, event):
+            mock_io.side_effect = [self._make_ignored_oserror(), 42]
+
+            fd = 100
+            result = nb_func(fd, nb_arg)
+            self.assertEqual(result, 42)
+
+            self._check_hub_event_closed(mock_get_hub, fd, event)
+
+        def _test_event_closed_on_io_error(self, nb_func, nb_arg,
+                                           mock_io, mock_get_hub, event):
+            mock_io.side_effect = [self._make_ignored_oserror(), ValueError()]
+
+            fd = 100
+
+            with self.assertRaises(ValueError):
+                nb_func(fd, nb_arg)
+
+            self._check_hub_event_closed(mock_get_hub, fd, event)
+
+        @mock.patch('gevent.os.get_hub')
+        @mock.patch('gevent.os._write')
+        def test_event_closed_on_write(self, mock_write, mock_get_hub):
+            self._test_event_closed_on_normal_io(os.nb_write, b'buf',
+                                                 mock_write, mock_get_hub,
+                                                 2)
+
+        @mock.patch('gevent.os.get_hub')
+        @mock.patch('gevent.os._write')
+        def test_event_closed_on_write_error(self, mock_write, mock_get_hub):
+            self._test_event_closed_on_io_error(os.nb_write, b'buf',
+                                                mock_write, mock_get_hub,
+                                                2)
+
+        @mock.patch('gevent.os.get_hub')
+        @mock.patch('gevent.os._read')
+        def test_event_closed_on_read(self, mock_read, mock_get_hub):
+            self._test_event_closed_on_normal_io(os.nb_read, b'buf',
+                                                 mock_read, mock_get_hub,
+                                                 1)
+
+        @mock.patch('gevent.os.get_hub')
+        @mock.patch('gevent.os._read')
+        def test_event_closed_on_read_error(self, mock_read, mock_get_hub):
+            self._test_event_closed_on_io_error(os.nb_read, b'buf',
+                                                mock_read, mock_get_hub,
+                                                1)
 
 
 if hasattr(os, 'fork_and_watch'):