first implementation of the event loop itself for libuv.

src/gevent/_ffi/callback.py

@@ -15,8 +15,8 @@ class callback(object):
 
     __slots__ = ('callback', 'args')
 
-    def __init__(self, callback, args):
-        self.callback = callback
+    def __init__(self, cb, args):
+        self.callback = cb
         self.args = args or _NOARGS
 
     def stop(self):

src/gevent/_ffi/loop.py

@@ -106,8 +106,7 @@ class _Callbacks(object):
                 return 0
             return 1 # It stopped itself
 
-    def python_handle_error(self, handle, revents):
-        print("HANDLING ERROR", handle, revents)
+    def python_handle_error(self, handle, _revents):
         try:
             watcher = self.ffi.from_handle(handle)
             exc_info = watcher._exc_info
@@ -118,12 +117,13 @@ class _Callbacks(object):
             # made sure that the watcher object was put in loop._keepaliveset,
             # what about not stopping the watcher? Looks like a possible
             # memory leak?
-            if revents & (libev.EV_READ | libev.EV_WRITE):
-                try:
-                    watcher.stop()
-                except: # pylint:disable=bare-except
-                    watcher.loop.handle_error(watcher, *sys.exc_info())
-                return # pylint:disable=lost-exception
+            # XXX: This used to do "if revents & (libev.EV_READ | libev.EV_WRITE)"
+            # before stopping. Why?
+            try:
+                watcher.stop()
+            except: # pylint:disable=bare-except
+                watcher.loop.handle_error(watcher, *sys.exc_info())
+            return # pylint:disable=lost-exception
 
     def python_stop(self, handle):
         watcher = self.ffi.from_handle(handle)
@@ -177,9 +177,10 @@ class AbstractLoop(object):
     _PREPARE_POINTER = None
     _PREPARE_CALLBACK_SIG = None
 
-    def __init__(self, ffi, lib, flags=None, default=None):
+    def __init__(self, ffi, lib, watchers, flags=None, default=None):
         self._ffi = ffi
         self._lib = lib
+        self._watchers = watchers
         self._in_callback = False
         self._callbacks = []
         self._keepaliveset = set()
@@ -314,7 +315,7 @@ class AbstractLoop(object):
 
     @property
     def WatcherType(self):
-        return watcher
+        return self._watchers.watcher
 
     @property
     def MAXPRI(self):
@@ -398,46 +399,46 @@ class AbstractLoop(object):
 
     @property
     def backend(self):
-        return None
+        return "<default backend>"
 
     @property
     def pendingcnt(self):
         return 0
 
     def io(self, fd, events, ref=True, priority=None):
-        return io(self, fd, events, ref, priority)
+        return self._watchers.io(self, fd, events, ref, priority)
 
     def timer(self, after, repeat=0.0, ref=True, priority=None):
-        return timer(self, after, repeat, ref, priority)
+        return self._watchers.timer(self, after, repeat, ref, priority)
 
     def signal(self, signum, ref=True, priority=None):
-        return signal(self, signum, ref, priority)
+        return self._watchers.signal(self, signum, ref, priority)
 
     def idle(self, ref=True, priority=None):
-        return idle(self, ref, priority)
+        return self._watchers.idle(self, ref, priority)
 
     def prepare(self, ref=True, priority=None):
-        return prepare(self, ref, priority)
+        return self._watchers.prepare(self, ref, priority)
 
     def check(self, ref=True, priority=None):
-        return check(self, ref, priority)
+        return self._watchers.check(self, ref, priority)
 
     def fork(self, ref=True, priority=None):
-        return fork(self, ref, priority)
+        return self._watchers.fork(self, ref, priority)
 
     def async(self, ref=True, priority=None):
-        return async(self, ref, priority)
+        return self._watchers.async(self, ref, priority)
 
     if sys.platform != "win32":
 
         def child(self, pid, trace=0, ref=True):
-            return child(self, pid, trace, ref)
+            return self._watchers.child(self, pid, trace, ref)
 
         def install_sigchld(self):
             pass
 
     def stat(self, path, interval=0.0, ref=True, priority=None):
-        return stat(self, path, interval, ref, priority)
+        return self._watchers.stat(self, path, interval, ref, priority)
 
     def callback(self, priority=None):
         return callback(self, priority)

src/gevent/_ffi/watcher.py

@@ -2,6 +2,7 @@
 Useful base classes for watchers. The available
 watchers will depend on the specific event loop.
 """
+# pylint:disable=not-callable
 from __future__ import absolute_import, print_function
 import sys
 import os
@@ -136,6 +137,7 @@ class watcher(object):
             import pdb; pdb.set_trace()
 
             raise
+
     def _watcher_ffi_set_priority(self, priority):
         pass
 

src/gevent/libev/corecffi.py

 # pylint: disable=too-many-lines, protected-access, redefined-outer-name, not-callable
 from __future__ import absolute_import, print_function
 import sys
-import os
-import traceback
-import signal as signalmodule
-
 
 __all__ = [
     'get_version',
@@ -188,23 +184,22 @@ def time():
 _default_loop_destroyed = False
 
 
-def _loop_callback(*args, **kwargs):
-    return ffi.callback(*args, **kwargs)
-
 from gevent._ffi.loop import AbstractLoop
 
-from gevent.libev.watcher import watcher
-from gevent.libev.watcher import io
-from gevent.libev.watcher import timer
-from gevent.libev.watcher import signal
-from gevent.libev.watcher import idle
-from gevent.libev.watcher import prepare
-from gevent.libev.watcher import check
-from gevent.libev.watcher import fork
-from gevent.libev.watcher import async
-from gevent.libev.watcher import child
-from gevent.libev.watcher import stat
-from gevent.libev.watcher import _events_to_str # exported
+# from gevent.libev.watcher import watcher
+# from gevent.libev.watcher import io
+# from gevent.libev.watcher import timer
+# from gevent.libev.watcher import signal
+# from gevent.libev.watcher import idle
+# from gevent.libev.watcher import prepare
+# from gevent.libev.watcher import check
+# from gevent.libev.watcher import fork
+# from gevent.libev.watcher import async
+# from gevent.libev.watcher import child
+# from gevent.libev.watcher import stat
+
+from gevent.libev import watcher as _watchers
+_events_to_str = _watchers._events_to_str # exported
 
 class loop(AbstractLoop):
     # pylint:disable=too-many-public-methods
@@ -220,7 +215,7 @@ class loop(AbstractLoop):
     _TIMER_POINTER = 'struct ev_timer *'
 
     def __init__(self, flags=None, default=None):
-        AbstractLoop.__init__(self, ffi, libev, flags, default)
+        AbstractLoop.__init__(self, ffi, libev, _watchers, flags, default)
 
     def _init_loop(self, flags, default):
         c_flags = _flags_to_int(flags)
@@ -285,11 +280,6 @@ class loop(AbstractLoop):
                 _default_loop_destroyed = True
             libev.ev_loop_destroy(ptr)
 
-
-    @property
-    def WatcherType(self):
-        return watcher
-
     @property
     def MAXPRI(self):
         return libev.EV_MAXPRI
@@ -363,35 +353,8 @@ class loop(AbstractLoop):
     def pendingcnt(self):
         return libev.ev_pending_count(self._ptr)
 
-    def io(self, fd, events, ref=True, priority=None):
-        return io(self, fd, events, ref, priority)
-
-    def timer(self, after, repeat=0.0, ref=True, priority=None):
-        return timer(self, after, repeat, ref, priority)
-
-    def signal(self, signum, ref=True, priority=None):
-        return signal(self, signum, ref, priority)
-
-    def idle(self, ref=True, priority=None):
-        return idle(self, ref, priority)
-
-    def prepare(self, ref=True, priority=None):
-        return prepare(self, ref, priority)
-
-    def check(self, ref=True, priority=None):
-        return check(self, ref, priority)
-
-    def fork(self, ref=True, priority=None):
-        return fork(self, ref, priority)
-
-    def async(self, ref=True, priority=None):
-        return async(self, ref, priority)
-
     if sys.platform != "win32":
 
-        def child(self, pid, trace=0, ref=True):
-            return child(self, pid, trace, ref)
-
         def install_sigchld(self):
             libev.gevent_install_sigchld_handler()
 

src/gevent/libev/watcher.py

@@ -28,7 +28,7 @@ else:
 # The CFFI implementation does none of these things, and so
 # is possibly NOT FUNCTIONALLY CORRECT on Win32
 #####
-
+_NOARGS = ()
 _events = [(libev.EV_READ, 'READ'),
            (libev.EV_WRITE, 'WRITE'),
            (libev.EV__IOFDSET, '_IOFDSET'),
@@ -148,15 +148,6 @@ class io(_base.IoMixin, watcher):
 
     EVENT_MASK = libev.EV__IOFDSET | libev.EV_READ | libev.EV_WRITE
 
-    def __init__(self, loop, fd, events, ref=True, priority=None):
-        # XXX: Win32: Need to vfd_open the fd and free the old one?
-        # XXX: Win32: Need a destructor to free the old fd?
-        if fd < 0:
-            raise ValueError('fd must be non-negative: %r' % fd)
-        if events & ~(libev.EV__IOFDSET | libev.EV_READ | libev.EV_WRITE):
-            raise ValueError('illegal event mask: %r' % events)
-        watcher.__init__(self, loop, ref=ref, priority=priority, args=(fd, events))
-
     def _get_fd(self):
         return vfd_get(self._watcher.fd)
 

src/gevent/libuv/_corecffi_cdef.c

@@ -160,22 +160,25 @@ typedef void (*uv_fs_poll_cb)(uv_fs_poll_t* handle, int status, const uv_stat_t*
 
 // loop functions
 uv_loop_t *uv_default_loop();
+uv_loop_t* uv_loop_new(); // not documented
 int uv_loop_init(uv_loop_t* loop);
 int uv_loop_alive(const uv_loop_t *loop);
 int uv_loop_close(uv_loop_t* loop);
 int uv_run(uv_loop_t *, uv_run_mode mode);
+int uv_backend_fd(const uv_loop_t* loop);
+uint64_t uv_now(const uv_loop_t* loop);
 void uv_stop(uv_loop_t *);
 void uv_walk(uv_loop_t *loop, uv_walk_cb walk_cb, void *arg);
 
 // handle functions
 // uv_handle_t is the base type for all libuv handle types.
 
-void uv_ref(uv_handle_t *);
-void uv_unref(uv_handle_t *);
-int uv_has_ref(const uv_handle_t *);
-void uv_close(uv_handle_t *handle, uv_close_cb close_cb);
-int uv_is_active(const uv_handle_t *handle);
-int uv_is_closing(const uv_handle_t *handle);
+void uv_ref(void *);
+void uv_unref(void *);
+int uv_has_ref(void *);
+void uv_close(void *handle, uv_close_cb close_cb);
+int uv_is_active(void *handle);
+int uv_is_closing(void *handle);
 
 // idle functions
 // Idle handles will run the given callback once per loop iteration, right
@@ -277,7 +280,7 @@ int uv_fs_poll_stop(uv_fs_poll_t*);
 static int (*python_callback)(void* handle, int revents);
 static void (*python_handle_error)(void* handle, int revents);
 static void (*python_stop)(void* handle);
-
+static void (*gevent_noop)(void* handle);
 /*
  * We use a single C callback for every watcher type that shares the same signature, which in turn calls the
  * Python callbacks. The uv_handle_t pointer type can be used for every watcher type

src/gevent/libuv/_corecffi_source.c

@@ -4,6 +4,10 @@ static int (*python_callback)(void* handle, int revents);
 static void (*python_handle_error)(void* handle, int revents);
 static void (*python_stop)(void* handle);
 
+static void _gevent_noop(void*handle) {}
+
+static void (*gevent_noop)(void* handle) = &_gevent_noop;
+
 static void _gevent_generic_callback1(uv_handle_t* watcher, int arg)
 {
     void* handle = watcher->data;