Hopefully solve the test__os issue. It turned out to be that calling uv_close...

Hopefully solve the test__os issue. It turned out to be that calling uv_close at random GC times is a very bad idea under libuv. See the comment. Try enabling pypy+libuv on posix and windows; it's working on darwin for me.

.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: $(PY36)
+	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,13 +10,11 @@ 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.
-    # - PYTHON: "C:\\pypy2-v5.9.0-win32"
-    #   PYTHON_ID: "pypy"
-    #   PYTHON_EXE: pypy
-    #   PYTHON_VERSION: "2.7.x"
-    #   PYTHON_ARCH: "32"
+    - PYTHON: "C:\\pypy2-v5.9.0-win32"
+      PYTHON_ID: "pypy"
+      PYTHON_EXE: pypy
+      PYTHON_VERSION: "2.7.x"
+      PYTHON_ARCH: "32"
 
     - PYTHON: "C:\\Python36-x64"
       PYTHON_VERSION: "3.6.x" # currently 3.6.0

src/gevent/_ffi/loop.py

@@ -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,21 +176,14 @@ 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()
+        self._watcher_full_init()
 
     def _watcher_full_init(self):
         priority = self.__init_priority
@@ -226,8 +219,13 @@ 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
-        self.loop._register_watcher(self, self._watcher)
+        # 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))
 
@@ -413,6 +412,8 @@ class IoMixin(object):
     def close(self):
         pass
 
+    def _format(self):
+        return ' fd=%d' % self._fd
 
 class TimerMixin(object):
     _watcher_type = 'timer'

src/gevent/libuv/loop.py

@@ -352,14 +352,9 @@ 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:

src/gevent/libuv/watcher.py

@@ -4,7 +4,6 @@ from __future__ import absolute_import, print_function
 
 import functools
 import sys
-import weakref
 
 import gevent.libuv._corecffi as _corecffi # pylint:disable=no-name-in-module,import-error
 
@@ -208,12 +207,18 @@ class io(_base.IoMixin, watcher):
     # a native watcher until the object is started, and we shut it down
     # when the object is stopped.
 
-    # XXX: I was able to solve at least Windows test_ftplib.py issues with more of a
-    # careful use of io objects in socket.py, so delaying this entirely is at least
-    # temporarily on hold. Instead sticking with the _watcher_create
-    # function override for the moment.
+    # XXX: I was able to solve at least Windows test_ftplib.py issues
+    # with more of a careful use of io objects in socket.py, so
+    # delaying this entirely is at least temporarily on hold. Instead
+    # sticking with the _watcher_create function override for the
+    # moment.
 
-    #_watcher_init_on_init = False
+    # XXX: Note 2: Moving to a deterministic close model, which was necessary
+    # for PyPy, also seems to solve the Windows issues. So we're completely taking
+    # this object out of the loop's registration; we don't want GC callbacks and
+    # uv_close anywhere *near* this object.
+
+    _watcher_registers_with_loop_on_create = False
 
     EVENT_MASK = libuv.UV_READABLE | libuv.UV_WRITABLE | libuv.UV_DISCONNECT
 
@@ -225,26 +230,6 @@ class io(_base.IoMixin, watcher):
         self._events = events
         self._multiplex_watchers = []
 
-    def _watcher_create(self, ref):
-        super(io, self)._watcher_create(ref)
-        # Immediately break the GC cycle. We restore the cycle before
-        # we are started and break it again when we are stopped.
-
-        # On Windows is critical to be able to garbage collect these
-        # objects in a timely fashion so that they don't get reused
-        # for multiplexing completely different sockets. This is because
-        # uv_poll_init_socket does a lot of setup for the socket to make
-        # polling work. If get reused for another socket that has the same
-        # fileno, things break badly. (In theory this could be a problem
-        # on posix too, but in practice it isn't).
-
-        # TODO: We should probably generalize this to all
-        # ffi watchers. Avoiding GC cycles as much as possible
-        # is a good thing, and potentially allocating new handles
-        # as needed gets us better memory locality.
-        self._handle = None
-        self._watcher.data = ffi.NULL
-
     def _get_fd(self):
         return self._fd
 
@@ -267,32 +252,10 @@ class io(_base.IoMixin, watcher):
             assert self._handle is not None
             self._watcher_start(self._watcher, self._events, self._watcher_callback)
 
-    # This is what we'd do if we set _watcher_init_on_init to False:
-    # def start(self, *args, **kwargs):
-    #    assert self._watcher is None
-    #    self._watcher_full_init()
-    #    super(io, self).start(*args, **kwargs)
-
-    # Along with disposing of self._watcher in _watcher_ffi_stop.
-    # In that method, it's possible that we might be started again right after this,
-    # in which case we will create a new set of FFI objects.
-    # TODO: Does anything leak in that case? Verify...
-
     def _watcher_ffi_start(self):
-        assert self._handle is None
-        self._handle = self._watcher.data = type(self).new_handle(self)
         _dbg("Starting watcher", self, "with events", self._events)
         self._watcher_start(self._watcher, self._events, self._watcher_callback)
 
-    def _watcher_ffi_stop(self):
-        # We may or may not have been started yet, so
-        # we may or may not have a handle; either way,
-        # drop it.
-        _dbg("Stopping io watcher", self)
-        self._handle = None
-        self._watcher.data = ffi.NULL
-        super(io, self)._watcher_ffi_stop()
-
     if sys.platform.startswith('win32'):
         # uv_poll can only handle sockets on Windows, but the plain
         # uv_poll_init we call on POSIX assumes that the fileno
@@ -333,11 +296,10 @@ class io(_base.IoMixin, watcher):
             self._events = events
 
             # References:
-            # These objects keep the original IO object alive;
+            # These objects must keep the original IO object alive;
             # the IO object SHOULD NOT keep these alive to avoid cycles
-            # When they all go away, the original IO object can go
-            # away. *Hopefully* that means that the FD they were opened for
-            # has also gone away.
+            # We MUST NOT rely on GC to clean up the IO objects, but the explicit
+            # calls to close(); see _multiplex_closed.
             self._watcher_ref = watcher
 
         events = property(
@@ -409,11 +371,7 @@ class io(_base.IoMixin, watcher):
 
     def multiplex(self, events):
         watcher = self._multiplexwatcher(events, self)
-        #watcher_ref = weakref.ref(watcher, self._multiplex_watchers.remove)
-        # We must not keep a hard ref to the returned object.
-        #self._multiplex_watchers.append(watcher_ref)
         self._multiplex_watchers.append(watcher)
-
         self._calc_and_update_events()
         return watcher
 
@@ -423,6 +381,18 @@ class io(_base.IoMixin, watcher):
             _dbg("IO Watcher", self, "has no more multiplexes")
             self.stop() # should already be stopped
             self._no_more_watchers()
+            # It is absolutely critical that we control when the call
+            # to uv_close() gets made. uv_close() of a uv_poll_t
+            # handle winds up calling uv__platform_invalidate_fd,
+            # which, as the name implies, destroys any outstanding
+            # events for the *fd* that haven't been delivered yet, and also removes
+            # the *fd* from the poll set. So if this happens later, at some
+            # non-deterministic time when (cyclic or otherwise) GC runs,
+            # *and* we've opened a new watcher for the fd, that watcher will
+            # suddenly and mysteriously stop seeing events. So we do this now;
+            # this method is smart enough not to close the handle twice.
+            self._watcher_ffi_close(self._watcher)
+            self._watcher = None
             del self._multiplex_watchers
         else:
             self._calc_and_update_events()
@@ -430,6 +400,8 @@ class io(_base.IoMixin, watcher):
                  self._multiplex_watchers)
 
     def _no_more_watchers(self):
+        # The loop sets this on an individual watcher to delete it from
+        # the active list where it keeps hard references.
         pass
 
     def _io_callback(self, events):

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