Merge pull request #1061 from gevent/close_io_watcher

Explicitly close IO watchers 

.travis.yml

@@ -12,6 +12,7 @@ env:
       # first group of parallel runs (4) as posible
       - TASK=test-py27-libuv
       - TASK=test-py36-libuv
+      - TASK=test-pypy-libuv
       - TASK=test-py27-noembed
       - TASK=test-pypy
       - TASK=test-py36

Makefile

@@ -187,9 +187,11 @@ test-py36-libuv: $(PY36)
 	GEVENT_CORE_CFFI_ONLY=libuv make test-py36
 
 test-pypy: $(PYPY)
-	ls $(BUILD_RUNTIMES)/versions/pypy590/bin/
 	PYTHON=$(PYPY) PATH=$(BUILD_RUNTIMES)/versions/pypy590/bin:$(PATH) make develop toxtest
 
+test-pypy-libuv: $(PYPY)
+	GEVENT_CORE_CFFI_ONLY=libuv make test-pypy
+
 test-pypy3: $(PYPY3)
 	PYTHON=$(PYPY3) PATH=$(BUILD_RUNTIMES)/versions/pypy3.5_590/bin:$(PATH) make develop toxtest
 

appveyor.yml

@@ -10,8 +10,10 @@ environment:
     # Pre-installed Python versions, which Appveyor may upgrade to
     # a later point release.
 
-    # We're not quite ready for PyPy+libuv, it doesn't even
-    # work correctly on Posix.
+    # We're not quite ready for PyPy+libuv.
+    # It works correctly on POSIX (linux and darwin),
+    # but has some strange errors and many timeouts on Windows.
+    # Most recent build: https://ci.appveyor.com/project/denik/gevent/build/1.0.1174/job/cv63181yj3ebb9cs
     # - PYTHON: "C:\\pypy2-v5.9.0-win32"
     #   PYTHON_ID: "pypy"
     #   PYTHON_EXE: pypy

src/gevent/_ffi/loop.py

@@ -144,7 +144,7 @@ class _Callbacks(object):
                 # The normal, expected scenario when we find the watcher still
                 # in the keepaliveset is that it is still active at the event loop
                 # level, so we don't expect that python_stop gets called.
-                _dbg("The watcher has not stopped itself, possibly still active", the_watcher)
+                #_dbg("The watcher has not stopped itself, possibly still active", the_watcher)
                 return 1
             return 2 # it stopped itself
 
@@ -296,10 +296,11 @@ class AbstractLoop(object):
 
 
     @classmethod
-    def __make_watcher_ref_callback(cls, typ, active_watchers, ffi_watcher):
+    def __make_watcher_ref_callback(cls, typ, active_watchers, ffi_watcher, debug):
         # separate method to make sure we have no ref to the watcher
         def callback(_):
             active_watchers.pop(ffi_watcher)
+            _dbg("Python weakref callback closing", debug)
             typ._watcher_ffi_close(ffi_watcher)
 
         return callback
@@ -309,7 +310,8 @@ class AbstractLoop(object):
                                                      self.__make_watcher_ref_callback(
                                                          type(python_watcher),
                                                          self._active_watchers,
-                                                         ffi_watcher))
+                                                         ffi_watcher,
+                                                         repr(python_watcher)))
 
     def _init_loop_and_aux_watchers(self, flags=None, default=None):
 

src/gevent/_ffi/watcher.py

@@ -176,20 +176,13 @@ class watcher(object):
     _handle = None # FFI object to self. This is a GC cycle. See _watcher_create
     _watcher = None
 
-    # Do we create the native resources when this class is created?
-    # If so, we call _watcher_full_init from the constructor.
-    # Otherwise, it must be called before we are started.
-    # If a subclass sets this to false, they must make that call.
-    # Currently unused. Experimental functionality for libuv.
-    _watcher_init_on_init = True
+    _watcher_registers_with_loop_on_create = True
 
     def __init__(self, _loop, ref=True, priority=None, args=_NOARGS):
         self.loop = _loop
         self.__init_priority = priority
         self.__init_args = args
         self.__init_ref = ref
-
-        if self._watcher_init_on_init:
         self._watcher_full_init()
 
     def _watcher_full_init(self):
@@ -226,7 +219,12 @@ class watcher(object):
         self._watcher = self._watcher_new()
         # This call takes care of calling _watcher_ffi_close when
         # self goes away, making sure self._watcher stays alive
-        # that long
+        # that long.
+
+        # XXX: All watchers should go to a model like libuv's
+        # IO watcher that gets explicitly closed so that we can always
+        # have control over when this gets done.
+        if self._watcher_registers_with_loop_on_create:
             self.loop._register_watcher(self, self._watcher)
 
         self._watcher.data = self._handle
@@ -401,6 +399,7 @@ class IoMixin(object):
             raise ValueError('fd must be non-negative: %r' % fd)
         if events & ~self.EVENT_MASK:
             raise ValueError('illegal event mask: %r' % events)
+        self._fd = fd
         super(IoMixin, self).__init__(loop, ref=ref, priority=priority,
                                       args=_args or (fd, events))
 
@@ -410,6 +409,11 @@ class IoMixin(object):
             args = (GEVENT_CORE_EVENTS, ) + args
         super(IoMixin, self).start(callback, *args)
 
+    def close(self):
+        pass
+
+    def _format(self):
+        return ' fd=%d' % self._fd
 
 class TimerMixin(object):
     _watcher_type = 'timer'

src/gevent/_socket2.py

@@ -208,9 +208,11 @@ class socket(object):
 
         if self._read_event is not None:
             self.hub.cancel_wait(self._read_event, cancel_wait_ex)
+            self._read_event.close()
             self._read_event = None
         if self._write_event is not None:
             self.hub.cancel_wait(self._write_event, cancel_wait_ex)
+            self._write_event.close()
             self._write_event = None
         s = self._sock
         self._sock = _closedsocket()

src/gevent/_socket3.py

@@ -252,9 +252,11 @@ class socket(object):
 
         if self._read_event is not None:
             self.hub.cancel_wait(self._read_event, cancel_wait_ex)
+            self._read_event.close()
             self._read_event = None
         if self._write_event is not None:
             self.hub.cancel_wait(self._write_event, cancel_wait_ex)
+            self._write_event.close()
             self._write_event = None
         _ss.close(self._sock)
 

src/gevent/_socketcommon.py

@@ -179,7 +179,10 @@ def wait_read(fileno, timeout=None, timeout_exc=_NONE):
     .. seealso:: :func:`cancel_wait`
      """
     io = get_hub().loop.io(fileno, 1)
+    try:
         return wait(io, timeout, timeout_exc)
+    finally:
+        io.close()
 
 
 def wait_write(fileno, timeout=None, timeout_exc=_NONE, event=_NONE):
@@ -196,7 +199,10 @@ def wait_write(fileno, timeout=None, timeout_exc=_NONE, event=_NONE):
     """
     # pylint:disable=unused-argument
     io = get_hub().loop.io(fileno, 2)
+    try:
         return wait(io, timeout, timeout_exc)
+    finally:
+        io.close()
 
 
 def wait_readwrite(fileno, timeout=None, timeout_exc=_NONE, event=_NONE):
@@ -214,7 +220,10 @@ def wait_readwrite(fileno, timeout=None, timeout_exc=_NONE, event=_NONE):
     """
     # pylint:disable=unused-argument
     io = get_hub().loop.io(fileno, 3)
+    try:
         return wait(io, timeout, timeout_exc)
+    finally:
+        io.close()
 
 #: The exception raised by default on a call to :func:`cancel_wait`
 class cancel_wait_ex(error): # pylint: disable=undefined-variable

src/gevent/ares.pyx

@@ -381,6 +381,7 @@ cdef public class channel [object PyGeventAresChannelObject, type PyGeventAresCh
             watcher.events = events
         else:
             watcher.stop()
+            watcher.close()
             self._watchers.pop(socket, None)
             if not self._watchers:
                 self._timer.stop()

src/gevent/libev/corecext.ppyx

@@ -808,6 +808,9 @@ cdef public class io(watcher) [object PyGeventIOObject, type PyGeventIO_Type]:
 
     PENDING
 
+    def close(self):
+        pass
+
 #ifdef _WIN32
 
     def __init__(self, loop loop, libev.vfd_socket_t fd, int events, ref=True, priority=None):

src/gevent/libuv/_corecffi_cdef.c

@@ -41,7 +41,9 @@ enum uv_poll_event {
 	UV_READABLE = 1,
 	UV_WRITABLE = 2,
 	/* new in 1.9 */
-	UV_DISCONNECT = 4
+	UV_DISCONNECT = 4,
+	/* new in 1.14.0 */
+	UV_PRIORITIZED = 8,
 };
 
 enum uv_fs_event {

src/gevent/libuv/loop.py

@@ -7,8 +7,9 @@ from __future__ import absolute_import, print_function
 import os
 from collections import defaultdict
 from collections import namedtuple
+from operator import delitem
 import signal
-from weakref import WeakValueDictionary
+
 
 from gevent._compat import PYPY
 from gevent._ffi.loop import AbstractLoop
@@ -81,7 +82,7 @@ class loop(AbstractLoop):
         AbstractLoop.__init__(self, ffi, libuv, _watchers, flags, default)
         self.__loop_pid = os.getpid()
         self._child_watchers = defaultdict(list)
-        self._io_watchers = WeakValueDictionary()
+        self._io_watchers = dict()
         self._fork_watchers = set()
         self._pid = os.getpid()
 
@@ -351,20 +352,20 @@ class loop(AbstractLoop):
 
     @gcBefore
     def io(self, fd, events, ref=True, priority=None):
-        # We don't keep a hard ref to the root object;
-        # the caller must keep the multiplexed watcher
-        # alive as long as its in use.
-
-        # We go to great pains to avoid GC cycles here, otherwise
-        # CPython tests (e.g., test_asyncore) fail on Windows.
-        # For PyPy, though, avoiding cycles isn't enough and we must
-        # do a GC to force cleaning up old objects.
+        # We rely on hard references here and explicit calls to
+        # close() on the returned object to correctly manage
+        # the watcher lifetimes.
 
         io_watchers = self._io_watchers
         try:
             io_watcher = io_watchers[fd]
+            assert io_watcher._multiplex_watchers, ("IO Watcher %s unclosed but should be dead" % io_watcher)
         except KeyError:
-            io_watcher = self._watchers.io(self, fd, self._watchers.io.EVENT_MASK)
+            # Start the watcher with just the events that we're interested in.
+            # as multiplexers are added, the real event mask will be updated to keep in sync.
+            # If we watch for too much, we get spurious wakeups and busy loops.
+            io_watcher = self._watchers.io(self, fd, 0)
             io_watchers[fd] = io_watcher
+            io_watcher._no_more_watchers = lambda: delitem(io_watchers, fd)
 
         return io_watcher.multiplex(events)

src/gevent/os.py

@@ -90,7 +90,10 @@ if fcntl:
         hub, event = None, None
         while True:
             try:
-                return _read(fd, n)
+                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
@@ -101,6 +104,7 @@ if fcntl:
                 event = hub.loop.io(fd, 1)
             hub.wait(event)
 
+
     def nb_write(fd, buf):
         """Write bytes from buffer `buf` to file descriptor `fd`. Return the
         number of bytes written.
@@ -110,7 +114,10 @@ if fcntl:
         hub, event = None, None
         while True:
             try:
-                return _write(fd, buf)
+                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

src/gevent/select.py

@@ -101,6 +101,7 @@ class SelectResult(object):
     def _closeall(self, watchers):
         for watcher in watchers:
             watcher.stop()
+            watcher.close()
         del watchers[:]
 
     def select(self, rlist, wlist, timeout):
@@ -208,6 +209,9 @@ if original_poll is not None:
                 # that. Should we raise an error?
 
             fileno = get_fileno(fd)
+            if fileno in self.fds:
+                self.fds[fileno].close()
+
             watcher = self.loop.io(fileno, flags)
             watcher.priority = self.loop.MAXPRI
             self.fds[fileno] = watcher
@@ -243,6 +247,8 @@ if original_poll is not None:
                library. Previously gevent did nothing.
             """
             fileno = get_fileno(fd)
+            io = self.fds[fileno]
+            io.close()
             del self.fds[fileno]
 
 del original_poll

src/greentest/greentest.py

@@ -390,7 +390,7 @@ if (PY3 and PYPY) or (PYPY and WIN and LIBUV):
     # pypy3 is very slow right now,
     # as is PyPy2 on windows (which only has libuv)
     CI_TIMEOUT = 15
-if PYPY and WIN and LIBUV:
+if PYPY and LIBUV:
     # slow and flaky timeouts
     LOCAL_TIMEOUT = CI_TIMEOUT
 else:

src/greentest/patched_tests_setup.py

@@ -158,6 +158,14 @@ disabled_tests = [
     'test_subprocess.ProcessTestCase.test_zombie_fast_process_del',
     # relies on subprocess._active which we don't use
 
+    # Very slow, tries to open lots and lots of subprocess and files,
+    # tends to timeout on CI.
+    'test_subprocess.ProcessTestCase.test_no_leaking',
+
+    # This test is also very slow, and has been timing out on Travis
+    # since November of 2016 on Python 3, but now also seen on Python 2/Pypy.
+    'test_subprocess.ProcessTestCase.test_leaking_fds_on_error',
+
     'test_ssl.ThreadedTests.test_default_ciphers',
     'test_ssl.ThreadedTests.test_empty_cert',
     'test_ssl.ThreadedTests.test_malformed_cert',
@@ -558,8 +566,6 @@ if sys.version_info[0] == 3:
             'test_subprocess.ProcessTestCaseNoPoll.test_cwd_with_relative_arg',
             'test_subprocess.ProcessTestCase.test_cwd_with_relative_executable',
 
-            # This test tends to timeout, starting at the end of November 2016
-            'test_subprocess.ProcessTestCase.test_leaking_fds_on_error',
         ]
 
     wrapped_tests.update({

src/greentest/test__core.py

@@ -22,7 +22,7 @@ class TestWatchers(unittest.TestCase):
     def test_io(self):
         if sys.platform == 'win32':
             # libev raises IOError, libuv raises ValueError
-            Error = (IOError,ValueError)
+            Error = (IOError, ValueError)
             win32 = True
         else:
             Error = ValueError
@@ -47,6 +47,7 @@ class TestWatchers(unittest.TestCase):
                 self.assertEqual(core._events_to_str(io.events), 'WRITE|_IOFDSET')
             else:
                 self.assertEqual(core._events_to_str(io.events), 'WRITE')
+            io.close()
 
     def test_timer_constructor(self):
         with self.assertRaises(ValueError):

src/greentest/test__core_watcher.py

@@ -73,10 +73,10 @@ class Test(greentest.TestCase):
         loop = core.loop(default=False)
 
         # Watchers aren't reused once all outstanding
-        # refs go away
+        # refs go away BUT THEY MUST BE CLOSED
         tty_watcher = loop.io(1, core.WRITE)
         watcher_handle = tty_watcher._watcher if IS_CFFI else tty_watcher
-
+        tty_watcher.close()
         del tty_watcher
         # XXX: Note there is a cycle in the CFFI code
         # from watcher_handle._handle -> watcher_handle.
@@ -86,7 +86,7 @@ class Test(greentest.TestCase):
 
         tty_watcher = loop.io(1, core.WRITE)
         self.assertIsNot(tty_watcher._watcher if IS_CFFI else tty_watcher, watcher_handle)
-
+        tty_watcher.close()
         loop.destroy()
 
 def reset(watcher, lst):

src/greentest/test__os.py

@@ -67,7 +67,7 @@ if hasattr(os, 'make_nonblocking'):
     class TestOS_nb(TestOS_tp):
 
         def pipe(self):
-            r, w = pipe()
+            r, w = super(TestOS_nb, self).pipe()
             os.make_nonblocking(r)
             os.make_nonblocking(w)
             return r, w