major backward-incompatible changes to core; it is now libevent-2 based

- libevent1.4 currently does not compile
- interface is not compatible with previous version
- core no longer initializes the event base (Hub does it)
- core.event and core.http accept event_base
- core.event clears 'callback' and 'arg' after callback is executed or it was cancelled (unless it was persistent). this helps to avoid creating GC cycles
- core: read_event/write_event/timer/active_event/.. are gone (but now available as methods on event_base, with different interface, however)

gevent/__init__.py

@@ -44,7 +44,6 @@ core.EV_WRITE   = 0x04
 core.EV_SIGNAL  = 0x08
 core.EV_PERSIST = 0x10
 
-from gevent.core import reinit
 from gevent.greenlet import Greenlet, joinall, killall
 spawn = Greenlet.spawn
 spawn_later = Greenlet.spawn_later
@@ -57,3 +56,8 @@ try:
     from gevent.hub import fork
 except ImportError:
     __all__.remove('fork')
+
+
+def reinit():
+    from gevent.hub import get_hub
+    return get_hub().reinit()

gevent/core.pyx

-#
-# event.pyx
-#
-# libevent Python bindings
-#
-# Copyright (c) 2004 Dug Song <dugsong@monkey.org>
-# Copyright (c) 2003 Martin Murray <murrayma@citi.umich.edu>
-# Copyright (c) 2009-2010 Denis Bilenko <denis.bilenko@gmail.com>
-#
+# Copyright (c) 2009-2010 Denis Bilenko and gevent contributors. See LICENSE for details.
 
 """Wrappers around libevent API.
 
@@ -15,27 +7,20 @@ specific event on a file handle, file descriptor, or signal occurs,
 or after a given time has passed. It also provides wrappers around
 structures and functions from libevent-dns and libevent-http.
 
-This module does not work with the greenlets. A callback passed
+This module does not work with greenlets. A callback passed
 to a method from this module will be executed in the event loop,
 which is running in the :class:`Hub <gevent.hub.Hub>` greenlet.
-Therefore it must not use any synchronous gevent API,
-that is, the functions that switch to the Hub. It's OK to call asynchronous
-stuff like :func:`gevent.spawn`, :meth:`Event.set <gevent.event.Event.set` or
-:meth:`Queue.put_nowait <gevent.queue.Queue.put_nowait>`.
+Therefore it must not use any synchronous gevent API, that is, blocking
+functions that need to switch to the Hub. It's OK to call asynchronous
+stuff like :func:`gevent.spawn`, :meth:`Event.set <gevent.event.Event.set`
+or :meth:`Queue.put_nowait <gevent.queue.Queue.put_nowait>`.
 
-The code is based on pyevent_.
+This module is very similar to pyevent_. In fact, it grew out of pyevent_ source code.
+However it is currently more up to date with regard to libevent API.
 
 .. _pyevent: http://code.google.com/p/pyevent/
 """
 
-__author__ = ( 'Dug Song <dugsong@monkey.org>',
-               'Martin Murray <mmurray@monkey.org>' )
-__copyright__ = ( 'Copyright (c) 2004 Dug Song',
-                  'Copyright (c) 2003 Martin Murray' )
-__license__ = 'BSD'
-__url__ = 'http://monkey.org/~dugsong/pyevent/'
-__version__ = '0.4+'
-
 __all__ = ['event', 'read_event', 'write_event', 'timer', 'signal', 'active_event',
            'init', 'dispatch', 'loop', 'get_version', 'get_method', 'get_header_version']
 # note, that .pxi files append stuff to __all__
@@ -45,6 +30,11 @@ import traceback
 from pprint import pformat
 import weakref
 
+cimport levent
+
+import _socket
+gaierror = _socket.gaierror
+
 
 cdef extern from "sys/types.h":
     ctypedef unsigned char u_char
@@ -55,62 +45,24 @@ cdef extern from "Python.h":
     object PyString_FromStringAndSize(char *v, int len)
     object PyString_FromString(char *v)
     int    PyObject_AsCharBuffer(object obj, char **buffer, int *buffer_len)
+    void   PyOS_snprintf(void*, size_t, char*, ...)
 
 cdef extern from "frameobject.h":
     ctypedef struct PyThreadState:
         void* exc_type
         void* exc_value
         void* exc_traceback
-
     PyThreadState* PyThreadState_GET()
 
-ctypedef void (*event_handler)(int fd, short evtype, void *arg)
+cdef extern from "stdio.h":
+    void* memset(void*, int, size_t)
 
-ctypedef void* event_base
-
-cdef extern from "libevent.h":
-
-    # event.h:
-    struct timeval:
-        unsigned int tv_sec
-        unsigned int tv_usec
-
-    struct event_t "event":
-        int   ev_fd
-        short ev_events
-        int   ev_flags
-        void *ev_arg
-
-    void* event_init()
-    int event_reinit(void *base)
-    char* event_get_version()
-    char* event_get_method()
-    void event_set(event_t *ev, int fd, short event, event_handler handler, void *arg)
-    void evtimer_set(event_t *ev, event_handler handler, void *arg)
-    int  event_add(event_t *ev, timeval *tv)
-    int  event_del(event_t *ev)
-    int  event_dispatch() nogil
-    int  event_loop(int loop) nogil
-    int  event_pending(event_t *ev, short, timeval *tv)
-    void event_active(event_t *ev, int res, short ncalls)
-
-    int EVLOOP_ONCE
-    int EVLOOP_NONBLOCK
-    char* _EVENT_VERSION
-
-    int EV_TIMEOUT
-    int EV_READ
-    int EV_WRITE
-    int EV_SIGNAL
-    int EV_PERSIST
-
-    int EVLIST_TIMEOUT
-    int EVLIST_INSERTED
-    int EVLIST_SIGNAL
-    int EVLIST_ACTIVE
-    int EVLIST_INTERNAL
-    int EVLIST_INIT
+cdef extern from "socketmodule.h":
+    char* get_gaierror(int)
+    object makesockaddr(int sockfd, void *, int addrlen, int proto)
 
+ctypedef void (*event_handler)(int fd, short evtype, void *arg)
+ctypedef void (*event_log_cb)(int severity, char *msg)
 
 cdef extern from "string.h":
     char* strerror(int errnum)
@@ -119,8 +71,291 @@ cdef extern from "errno.h":
     int errno
 
 
-cdef extern from "libevent.h":
-    event_base* current_base
+cdef class event_base:
+
+    cdef void* _ptr
+    cdef object _dns
+
+    def __init__(self, size_t ptr=0):
+        if ptr:
+            self._ptr = <void*>ptr
+        else:
+            self._ptr = levent.event_base_new()
+            if not self._ptr:
+                if errno:
+                    raise IOError(errno, strerror(errno))
+                else:
+                    raise IOError("event_base_new returned NULL")
+
+    def __dealloc__(self):
+        self.free()
+
+    property ptr:
+
+        def __get__(self):
+            return <size_t>self._ptr
+
+    property dns:
+
+        def __get__(self):
+            if self._dns is None:
+                self._dns = evdns_base.new(self)
+            return self._dns
+
+    def dispatch(self):
+        return levent.event_base_dispatch(self._ptr)
+
+    def reinit(self):
+        cdef int result = levent.event_reinit(self._ptr)
+        if result != 0:
+            raise IOError('event_reinit failed with %s' % result)
+
+    def get_method(self):
+        return levent.event_base_get_method(self._ptr)
+
+    def get_info(self):
+        return 'libevent-%s/%s' % (get_version(), self.get_method())
+
+    def free(self):
+        if self._ptr:
+            levent.event_base_free(self._ptr)
+            self._ptr = NULL
+        if self._dns is not None:
+            self._dns.free()
+            self._dns = None
+
+    def read_event(self, int handle, persist=False):
+        cdef short evtype = levent.EV_READ
+        if persist:
+            evtype = evtype | levent.EV_PERSIST
+        cdef event ev = event(evtype, handle, self)
+        return ev
+
+    def write_event(self, int handle, persist=False):
+        cdef short evtype = levent.EV_WRITE
+        if persist:
+            evtype = evtype | levent.EV_PERSIST
+        cdef event ev = event(evtype, handle, base=self)
+        return ev
+
+    def readwrite_event(self, int handle, persist=False):
+        cdef short evtype = levent.EV_READ | levent.EV_WRITE
+        if persist:
+            evtype = evtype | levent.EV_PERSIST
+        cdef event ev = event(evtype, handle, base=self)
+        return ev
+
+    def signal(self, int signalnum, callback, *args, **kwargs):
+        cdef event ev = simple_event(levent.EV_SIGNAL|levent.EV_PERSIST, signalnum, base=self)
+        ev.add(None, callback, *args, **kwargs)
+        return ev
+
+    def timer(self, seconds=None, callback=None, *args, **kwargs):
+        cdef event ev = simple_event(0, -1, base=self)
+        if callback is not None:
+            ev.add(seconds, callback, *args, **kwargs)
+        return ev
+
+    def active_event(self, callback, *args, **kwargs):
+        cdef event ev = simple_event(0, -1, base=self)
+        ev.active(callback, *args, **kwargs)
+        return ev
+
+
+cdef class evdns_base:
+
+    cdef void* _ptr
+    cdef public event_base base
+
+    def __init__(self, event_base base, size_t ptr=0):
+        self.base = base
+        self._ptr = <void*>ptr
+
+    @classmethod
+    def new(cls, object base, int init=1):
+        cdef void* ptr = levent.evdns_base_new((<event_base?>base)._ptr, <int>init)
+        if not ptr:
+            if errno:
+                raise IOError(errno, strerror(errno))
+            else:
+                raise IOError("evdns_base_new returned NULL")
+        return cls(base, <size_t>ptr)
+
+    property ptr:
+
+        def __get__(self):
+            return <size_t>self._ptr
+
+    def free(self, int fail_requests=1):
+        if self._ptr:
+            self.base = None
+            levent.evdns_base_free(self._ptr, fail_requests)
+            self._ptr = NULL
+
+    def add_nameserver(self, char* address):
+        """Add a nameserver.
+
+        This function parses a n IPv4 or IPv6 address from a string and adds it as a
+        nameserver.  It supports the following formats:
+        - [IPv6Address]:port
+        - [IPv6Address]
+        - IPv6Address
+        - IPv4Address:port
+        - IPv4Address
+
+        If no port is specified, it defaults to 53."""
+        cdef int result = levent.evdns_base_nameserver_ip_add(self._ptr, address)
+        if result:
+            if errno:
+                raise IOError(errno, strerror(errno))
+            else:
+                raise IOError("evdns_base_nameserver_ip_add returned %r" % result)
+
+    def count_nameservers(self):
+        return levent.evdns_base_count_nameservers(self._ptr)
+
+    def set_option(self, char* option, object val):
+        """Set the value of a configuration option.
+
+        The available configuration options are (as of libevent-2.0.8-rc):
+
+        ndots, timeout, max-timeouts, max-inflight, attempts, randomize-case,
+        bind-to, initial-probe-timeout, getaddrinfo-allow-skew."""
+        # XXX auto add colon fix it: In versions before Libevent 2.0.3-alpha, the option name needed to end with a colon.
+        cdef char* c_val
+        if isinstance(val, str):
+            c_val = val
+        elif isinstance(val, (int, long, float)):
+            val = str(val)
+            c_val = val
+        elif isinstance(val, unicode):
+            val = val.encode('ascii')
+            c_val = val
+        else:
+            raise TypeError('Expected a string or a number: %r' % (val, ))
+        cdef int result = levent.evdns_base_set_option(self._ptr, option, c_val)
+        if result:
+            if errno:
+                raise IOError(errno, strerror(errno))
+            else:
+                raise IOError("evdns_base_set_option returned %r" % result)
+
+    def getaddrinfo(self, callback, host, port, int family=0, int socktype=0, int proto=0, int flags=0, arg=None):
+        # evdns and socket module do not match flags
+        cdef char* nodename = NULL
+        cdef char* servname = NULL
+        cdef char pbuf[30]
+        cdef levent.evutil_addrinfo hints
+        if host is None:
+            pass
+        elif isinstance(host, unicode):
+            host = host.encode('idna')
+            nodename = host
+        elif isinstance(host, str):
+            nodename = host
+        else:
+            raise TypeError("getaddrinfo() first argument must be string or None")
+        if port is None:
+            pass
+        elif isinstance(port, (int, long)):
+            PyOS_snprintf(pbuf, sizeof(pbuf), "%ld", <long>port)
+            servname = pbuf
+        else:
+            servname = <char*?>port  # check that it raises TypeError
+        memset(&hints, 0, sizeof(hints))
+        hints.ai_family = family
+        hints.ai_socktype = socktype
+        hints.ai_protocol = proto
+        hints.ai_flags = flags
+        cdef object param = [callback, arg]
+        Py_INCREF(param)
+        cdef void* c_request = levent.evdns_getaddrinfo(self._ptr, nodename, servname, &hints, __getaddrinfo_handler, <void*>param)
+        if c_request:
+            request = getaddrinfo_request(self.base, <size_t>c_request)
+            param.append(request)
+            return request
+
+
+cdef void __getaddrinfo_handler(int code, levent.evutil_addrinfo* res, void* c_param):
+    cdef object callback
+    cdef object arg
+    cdef object request = None
+    cdef object param = <object>c_param
+    Py_DECREF(param)
+    cdef list result
+    cdef char* canonname
+    try:
+        if len(param) == 2:
+            callback, arg = param
+        else:
+            callback, arg, request = param
+
+        if request is not None:
+            if request.base is None:  # cancelled?
+                request.detach()
+                return
+            request.detach()
+
+        if code:
+            callback(None, gaierror(code, get_gaierror(code)), arg)
+        else:
+            result = []
+            while res:
+                if res.ai_canonname:
+                    canonname = res.ai_canonname
+                else:
+                    canonname = ''
+                result.append((res.ai_family,
+                               res.ai_socktype,
+                               res.ai_protocol,
+                               canonname,
+                               makesockaddr(-1, res.ai_addr, res.ai_addrlen, res.ai_protocol)))
+                res = res.ai_next
+            callback(result, None, arg)
+    except:
+        traceback.print_exc()
+        try:
+            sys.stderr.write('Failed to execute callback %r\n\n' % (callback, ))
+        except:
+            traceback.print_exc()
+        sys.exc_clear()
+
+
+cdef class getaddrinfo_request:
+
+    cdef public event_base base
+    cdef void* _ptr
+
+    def __init__(self, event_base base, size_t ptr=0):
+        self.base = base
+        self._ptr = <void*>ptr
+
+    property ptr:
+
+        def __get__(self):
+            return <size_t>self._ptr
+
+    def __str__(self):
+        return '<%s ptr=%x>' % (self.__class__.__name__, self.ptr)
+
+    def detach(self):
+        self.base = None
+        self._ptr = NULL
+
+    def _cancel(self):
+        if self._ptr and self.base is not None:
+            levent.evdns_getaddrinfo_cancel(self._ptr)
+            # getaddrinfo_handler will be called immediatelly, with EVUTIL_EAI_CANCEL argument
+
+    def cancel(self):
+        if self._ptr and self.base is not None:
+            self.base.active_event(self._cancel)
+            self.base = None
+
+    @property
+    def pending(self):
+        if self._ptr:
+            return self.base is not None
 
 
 cdef void __event_handler(int fd, short evtype, void *arg) with gil:
@@ -135,7 +370,7 @@ cdef void __event_handler(int fd, short evtype, void *arg) with gil:
             traceback.print_exc()
         sys.exc_clear()
     finally:
-        if not event_pending(&self.ev, EV_READ|EV_WRITE|EV_SIGNAL|EV_TIMEOUT, NULL):
+        if not levent.event_pending(&self.ev, levent.EV_READ|levent.EV_WRITE|levent.EV_SIGNAL|levent.EV_TIMEOUT, NULL):
             self._delref()
 
 
@@ -147,20 +382,17 @@ cdef class event:
     - *callback* -- user callback with ``(event, evtype)`` prototype
     - *arg*      -- optional object, which will be made available as :attr:`arg` property.
     """
-    cdef event_t ev
+    cdef levent.event ev
     cdef public object callback
     cdef public object arg
     cdef int _incref  # 1 if we already INCREFed this object once (because libevent references it)
 
-    def __init__(self, short evtype, int handle, callback, arg=None):
-        self.callback = callback
-        self.arg = arg
+    def __init__(self, short evtype, int handle, event_base base=None):
         self._incref = 0
         cdef void* c_self = <void*>self
-        if evtype == 0 and not handle:
-            evtimer_set(&self.ev, __event_handler, c_self)
-        else:
-            event_set(&self.ev, handle, evtype, __event_handler, c_self)
+        levent.event_set(&self.ev, handle, evtype, __event_handler, c_self)
+        if base is not None:
+            levent.event_base_set((<event_base?>base)._ptr, &self.ev)
 
     cdef _addref(self):
         if self._incref <= 0:
@@ -171,12 +403,14 @@ cdef class event:
         if self._incref > 0:
             Py_DECREF(self)
             self._incref -= 1
+        self.callback = None
+        self.arg = None
 
     property pending:
         """Return True if the event is still scheduled to run."""
 
         def __get__(self):
-            return event_pending(&self.ev, EV_TIMEOUT|EV_SIGNAL|EV_READ|EV_WRITE, NULL)
+            return levent.event_pending(&self.ev, levent.EV_TIMEOUT|levent.EV_SIGNAL|levent.EV_READ|levent.EV_WRITE, NULL)
 
     property fd:
 
@@ -194,8 +428,11 @@ cdef class event:
             result = []
             cdef short events = self.ev.ev_events
             cdef short c_event
-            for (event, txt) in ((EV_TIMEOUT, 'TIMEOUT'), (EV_READ, 'READ'), (EV_WRITE, 'WRITE'),
-                                 (EV_SIGNAL, 'SIGNAL'), (EV_PERSIST, 'PERSIST')):
+            for (event, txt) in ((levent.EV_TIMEOUT, 'TIMEOUT'),
+                                 (levent.EV_READ, 'READ'),
+                                 (levent.EV_WRITE, 'WRITE'),
+                                 (levent.EV_SIGNAL, 'SIGNAL'),
+                                 (levent.EV_PERSIST, 'PERSIST')):
                 c_event = event
                 if events & c_event:
                     result.append(txt)
@@ -216,8 +453,12 @@ cdef class event:
             result = []
             cdef int flags = self.ev.ev_flags
             cdef int c_flag
-            for (flag, txt) in ((EVLIST_TIMEOUT, 'TIMEOUT'), (EVLIST_INSERTED, 'INSERTED'), (EVLIST_SIGNAL, 'SIGNAL'),
-                                (EVLIST_ACTIVE, 'ACTIVE'), (EVLIST_INTERNAL, 'INTERNAL'), (EVLIST_INIT, 'INIT')):
+            for (flag, txt) in ((levent.EVLIST_TIMEOUT, 'TIMEOUT'),
+                                (levent.EVLIST_INSERTED, 'INSERTED'),
+                                (levent.EVLIST_SIGNAL, 'SIGNAL'),
+                                (levent.EVLIST_ACTIVE, 'ACTIVE'),
+                                (levent.EVLIST_INTERNAL, 'INTERNAL'),
+                                (levent.EVLIST_INIT, 'INIT')):
                 c_flag = flag
                 if flags & c_flag:
                     result.append(txt)
@@ -227,16 +468,15 @@ cdef class event:
                 result.append(hex(flags))
             return '|'.join(result)
 
-    def add(self, timeout=None):
-        """Add event to be executed after an optional *timeout* - number of seconds
-        after which the event will be executed."""
-        global errno
-        cdef timeval tv
+    def add(self, timeout, callback, arg=None):
+        cdef levent.timeval tv
         cdef double c_timeout
         cdef int result
-        errno = 0  # event_add sometime does not set errno
+        if not self.ev.ev_base:
+            # libevent starting with 2.0.7 actually does check for this condition, so we should not
+            raise AssertionError('ev_base is not set')
         if timeout is None:
-            result = event_add(&self.ev, NULL)
+            result = levent.event_add(&self.ev, NULL)
         else:
             c_timeout = <double>timeout
             if c_timeout < 0.0:
@@ -244,19 +484,30 @@ cdef class event:
             else:
                 tv.tv_sec = <long>c_timeout
                 tv.tv_usec = <unsigned int>((c_timeout - <double>tv.tv_sec) * 1000000.0)
-                result = event_add(&self.ev, &tv)
+                result = levent.event_add(&self.ev, &tv)
         if result < 0:
             if errno:
                 raise IOError(errno, strerror(errno))
             else:
                 raise IOError("event_add(fileno=%s) returned %s" % (self.fd, result))
+        self.callback = callback
+        self.arg = arg
+        self._addref()
+
+    def active(self, callback, arg):
+        levent.event_active(&self.ev, levent.EV_TIMEOUT, 1)
+        self.callback = callback
+        self.arg = arg
         self._addref()
 
     def cancel(self):
         """Remove event from the event queue."""
         cdef int result
-        if event_pending(&self.ev, EV_TIMEOUT|EV_SIGNAL|EV_READ|EV_WRITE, NULL):
-            result = event_del(&self.ev)
+        # setting self.callback and self.arg to None to avoid refcounting cycles
+        self.callback = None
+        self.arg = None
+        if levent.event_pending(&self.ev, levent.EV_TIMEOUT|levent.EV_SIGNAL|levent.EV_READ|levent.EV_WRITE, NULL):
+            result = levent.event_del(&self.ev)
             if result < 0:
                 return result
             self._delref()
@@ -295,39 +546,6 @@ cdef class event:
         self.cancel()
 
 
-cdef class read_event(event):
-    """Create a new scheduled event with evtype=EV_READ"""
-
-    def __init__(self, int handle, callback, timeout=None, arg=None, persist=False):
-        cdef short evtype = EV_READ
-        if persist:
-            evtype = evtype | EV_PERSIST
-        event.__init__(self, evtype, handle, callback, arg)
-        self.add(timeout)
-
-
-cdef class write_event(event):
-    """Create a new scheduled event with evtype=EV_WRITE"""
-
-    def __init__(self, int handle, callback, timeout=None, arg=None, persist=False):
-        cdef short evtype = EV_WRITE
-        if persist:
-            evtype = evtype | EV_PERSIST
-        event.__init__(self, evtype, handle, callback, arg)
-        self.add(timeout)
-
-
-class readwrite_event(event):
-    """Create a new scheduled event with evtype=EV_READ|EV_WRITE"""
-
-    def __init__(self, int handle, callback, timeout=None, arg=None, persist=False):
-        cdef short evtype = EV_READ|EV_WRITE
-        if persist:
-            evtype = evtype | EV_PERSIST
-        event.__init__(self, evtype, handle, callback, arg)
-        self.add(timeout)
-
-
 cdef void __simple_handler(int fd, short evtype, void *arg) with gil:
     cdef event self = <event>arg
     try:
@@ -341,125 +559,35 @@ cdef void __simple_handler(int fd, short evtype, void *arg) with gil:
             traceback.print_exc()
         sys.exc_clear()
     finally:
-        if not event_pending(&self.ev, EV_READ|EV_WRITE|EV_SIGNAL|EV_TIMEOUT, NULL):
+        if not levent.event_pending(&self.ev, levent.EV_READ|levent.EV_WRITE|levent.EV_SIGNAL|levent.EV_TIMEOUT, NULL):
             self._delref()
 
 
-cdef class timer(event):
-    """Create a new scheduled timer"""
-
-    def __init__(self, float seconds, callback, *args, **kwargs):
-        self.callback = callback
-        self.arg = (args, kwargs)
-        evtimer_set(&self.ev, __simple_handler, <void*>self)
-        self.add(seconds)
-
-
-cdef class signal(event):
-    """Create a new persistent signal event"""
-
-    def __init__(self, int signalnum, callback, *args, **kwargs):
-        self.callback = callback
-        self.arg = (args, kwargs)
-        event_set(&self.ev, signalnum, EV_SIGNAL|EV_PERSIST, __simple_handler, <void*>self)
-        self.add()
-
-
-cdef class active_event(event):
-    """An event that is scheduled to run in the current loop iteration"""
-
-    def __init__(self, callback, *args, **kwargs):
-        self.callback = callback
-        self.arg = (args, kwargs)
-        evtimer_set(&self.ev, __simple_handler, <void*>self)
-        self._addref()
-        event_active(&self.ev, EV_TIMEOUT, 1)
-
-    def add(self, timeout=None):
-        raise NotImplementedError
+cdef class simple_event(event):
 
+    def __init__(self, short evtype, int handle, base=None):
+        self._incref = 0
+        cdef void* c_self = <void*>self
+        levent.event_set(&self.ev, handle, evtype, __simple_handler, c_self)
+        if base is not None:
+            levent.event_base_set((<event_base?>base)._ptr, &self.ev)
 
-def init():
-    """Initialize event queue."""
-    event_init()
-
+    def add(self, seconds, callback, *args, **kwargs):
+        return event.add(self, seconds, callback, (args, kwargs))
 
-def dispatch():
-    """Dispatch all events on the event queue.
-    Returns 0 on success, and 1 if no events are registered.
-    May raise IOError.
-    """
-    cdef int ret
-    with nogil:
-        ret = event_dispatch()
-    if ret < 0:
-        raise IOError(errno, strerror(errno))
-    return ret
-
-
-def loop(nonblock=False):
-    """Dispatch all pending events on queue in a single pass.
-    Returns 0 on success, and 1 if no events are registered.
-    May raise IOError.
-    """
-    cdef int flags, ret
-    flags = EVLOOP_ONCE
-    if nonblock:
-        flags = EVLOOP_ONCE|EVLOOP_NONBLOCK
-    with nogil:
-        ret = event_loop(flags)
-    if ret < 0:
-        raise IOError(errno, strerror(errno))
-    return ret
+    def active(self, callback, *args, **kwargs):
+        return event.active(self, callback, (args, kwargs))
 
 
 def get_version():
     """Wrapper for :meth:`event_get_version`"""
-    return event_get_version()
-
+    return levent.event_get_version()
 
-def get_method():
-    """Wrapper for :meth:`event_get_method`"""
-    return event_get_method()
-
-
-cdef extern from *:
-    cdef void emit_ifdef "#if defined(_EVENT_VERSION) //" ()
-    cdef void emit_else  "#else //" ()
-    cdef void emit_endif "#endif //" ()
-
-
-# _EVENT_VERSION is available since libevent 1.4.0-beta
 
 def get_header_version():
     """Return _EVENT_VERSION"""
-    emit_ifdef()
-    return _EVENT_VERSION
-    emit_endif()
+    return levent._EVENT_VERSION
 
-# event_reinit is available since libevent 1.4.1-beta,
-# but I cannot check for existence of a function here, can I?
-# so I'm going to use _EVENT_VERSION as an indicator of event_reinit presence
-# which will work in every version other than 1.4.0-beta
-
-def reinit():
-    """Wrapper for :meth:`event_reinit`."""
-    emit_ifdef()
-    return event_reinit(current_base)
-    emit_endif()
-
-include "evdns.pxi"
-
-# XXX - make sure event queue is always initialized.
-init()
-
-if get_version() != get_header_version() and get_header_version() is not None and get_version() != '1.3.99-trunk':
-    import warnings
-    msg = "libevent version mismatch: system version is %r but this gevent is compiled against %r" % (get_version(), get_header_version())
-    warnings.warn(msg, UserWarning, stacklevel=2)
-
-include "evbuffer.pxi"
-include "evhttp.pxi"
 
 def set_exc_info(object typ, object value, object tb):
     cdef PyThreadState* tstate = PyThreadState_GET()
@@ -476,3 +604,13 @@ def set_exc_info(object typ, object value, object tb):
     tstate.exc_value = <void *>value
     tstate.exc_traceback = <void *>tb
 
+
+#include "evdns.pxi"
+include "evbuffer.pxi"
+include "evhttp.pxi"
+
+
+if get_version() != get_header_version() and get_header_version() is not None and get_version() != '1.3.99-trunk':
+    import warnings
+    msg = "libevent version mismatch: system version is %r but this gevent is compiled against %r" % (get_version(), get_header_version())
+    warnings.warn(msg, UserWarning, stacklevel=2)

gevent/coros.py

@@ -3,7 +3,6 @@
 
 import sys
 import traceback
-from gevent.core import active_event
 from gevent.hub import get_hub, getcurrent
 from gevent.timeout import Timeout
 
@@ -35,7 +34,7 @@ class Semaphore(object):
     def release(self):
         self.counter += 1
         if self._links and self.counter > 0 and self._notifier is None:
-            self._notifier = active_event(self._notify_links, list(self._links))
+            self._notifier = get_hub().reactor.active_event(self._notify_links, list(self._links))
 
     def _notify_links(self, links):
         try:
@@ -64,7 +63,7 @@ class Semaphore(object):
             raise TypeError('Expected callable: %r' % (callback, ))
         self._links.append(callback)
         if self.counter > 0 and self._notifier is None:
-            self._notifier = active_event(self._notify_links, list(self._links))
+            self._notifier = get_hub().reactor.active_event(self._notify_links, list(self._links))
 
     def unlink(self, callback):
         """Remove the callback set by :meth:`rawlink`"""

gevent/dns.py

@@ -20,10 +20,6 @@ __all__ = ['DNSError',
            'QUERY_NO_SEARCH']
 
 
-# move from here into Hub.__init__ (once event_init() is move here as well)
-core.dns_init()
-
-
 class DNSError(gaierror):
     """A subclass of :class:`socket.gaierror` used by :mod:`evdns` functions to report errors.
 

gevent/event.py

@@ -3,7 +3,6 @@
 
 import sys
 import traceback
-from gevent import core
 from gevent.hub import get_hub, getcurrent, _NONE
 from gevent.timeout import Timeout
 
@@ -40,7 +39,7 @@ class Event(object):
         self._flag = True
         if self._links:
             # schedule a job to notify the links already set
-            core.active_event(self._notify_links, list(self._links))
+            get_hub().reactor.active_event(self._notify_links, list(self._links))
 
     def clear(self):
         """Reset the internal flag to false.
@@ -92,7 +91,7 @@ class Event(object):
             raise TypeError('Expected callable: %r' % (callback, ))
         self._links.append(callback)
         if self._flag:
-            core.active_event(self._notify_links, list(self._links))  # XXX just pass [callback]
+            get_hub().reactor.active_event(self._notify_links, list(self._links))  # XXX just pass [callback]
 
     def unlink(self, callback):
         """Remove the callback set by :meth:`rawlink`"""
@@ -178,7 +177,7 @@ class AsyncResult(object):
         self.value = value
         self._exception = None
         if self._links and self._notifier is None:
-            self._notifier = core.active_event(self._notify_links)
+            self._notifier = get_hub().reactor.active_event(self._notify_links)
 
     def set_exception(self, exception):
         """Store the exception. Wake up the waiters.
@@ -188,7 +187,7 @@ class AsyncResult(object):
         """
         self._exception = exception
         if self._links and self._notifier is None:
-            self._notifier = core.active_event(self._notify_links)
+            self._notifier = get_hub().reactor.active_event(self._notify_links)
 
     def get(self, block=True, timeout=None):
         """Return the stored value or raise the exception.
@@ -293,7 +292,7 @@ class AsyncResult(object):
             raise TypeError('Expected callable: %r' % (callback, ))
         self._links.add(callback)
         if self.ready() and self._notifier is None:
-            self._notifier = core.active_event(self._notify_links)
+            self._notifier = get_hub().reactor.active_event(self._notify_links)
 
     def unlink(self, callback):
         """Remove the callback set by :meth:`rawlink`"""

gevent/evhttp.pxi

@@ -54,7 +54,7 @@ cdef extern from "libevent.h":
     # evhttp
     ctypedef void (*evhttp_handler)(evhttp_request *, void *arg)
 
-    evhttp*   evhttp_new(event_base *base)
+    evhttp*   evhttp_new(void *base)
     int       evhttp_bind_socket(evhttp *http, char* address, int port)
     int       evhttp_accept_socket(evhttp *http, int fd)
     void      evhttp_free(evhttp* http)
@@ -511,21 +511,27 @@ cdef class http:
     cdef public object default_response_headers
     cdef public dict _requests
 
-    def __init__(self, object handle, object default_response_headers=None):
+    def __init__(self, object handle, object default_response_headers=None, event_base base=None):
         self.handle = handle
         if default_response_headers is None:
             self.default_response_headers = []
         else:
             self.default_response_headers = default_response_headers
-        self._requests = {} # maps connection id to WeakKeyDictionary which holds requests
-        self.__obj = evhttp_new(current_base)
+        self._requests = {}  # maps connection id to WeakKeyDictionary which holds requests
+        if base is None:
+            if levent.current_base:
+                self.__obj = evhttp_new(levent.current_base)
+            else:
+                raise ValueError('Please provide event_base')
+        else:
+            self.__obj = evhttp_new(base._ptr)
         evhttp_set_gencb(self.__obj, _http_cb_handler, <void *>self)
 
     def __dealloc__(self):
         if self.__obj != NULL:
             evhttp_set_gencb(self.__obj, _http_cb_reply_error, NULL)
             evhttp_free(self.__obj)
-        self.__obj = NULL
+            self.__obj = NULL
 
     property _obj:
 

gevent/greenlet.py

@@ -2,7 +2,6 @@
 
 import sys
 import traceback
-from gevent import core
 from gevent.hub import greenlet, getcurrent, get_hub, GreenletExit, Waiter, kill
 from gevent.timeout import Timeout
 
@@ -242,12 +241,12 @@ class Greenlet(greenlet):
     def start(self):
         """Schedule the greenlet to run in this loop iteration"""
         assert not self.started, 'Greenlet already started'
-        self._start_event = core.active_event(self.switch)
+        self._start_event = get_hub().reactor.active_event(self.switch)
 
     def start_later(self, seconds):
         """Schedule the greenlet to run in the future loop iteration *seconds* later"""
         assert not self.started, 'Greenlet already started'
-        self._start_event = core.timer(seconds, self.switch)
+        self._start_event = get_hub().reactor.timer(seconds, self.switch)
 
     @classmethod
     def spawn(cls, *args, **kwargs):
@@ -302,7 +301,7 @@ class Greenlet(greenlet):
             self._start_event = None
         if not self.dead:
             waiter = Waiter()
-            core.active_event(_kill, self, exception, waiter)
+            get_hub().reactor.active_event(_kill, self, exception, waiter)
             if block:
                 waiter.get()
                 self.join(timeout)
@@ -375,7 +374,7 @@ class Greenlet(greenlet):
         self._exception = None
         self.value = result
         if self._links and self._notifier is None:
-            self._notifier = core.active_event(self._notify_links)
+            self._notifier = get_hub().reactor.active_event(self._notify_links)
 
     def _report_error(self, exc_info):
         exception = exc_info[1]
@@ -389,7 +388,7 @@ class Greenlet(greenlet):
         self._exception = exception
 
         if self._links and self._notifier is None:
-            self._notifier = core.active_event(self._notify_links)
+            self._notifier = get_hub().reactor.active_event(self._notify_links)
 
         info = str(self) + ' failed with '
         try:
@@ -421,7 +420,7 @@ class Greenlet(greenlet):
             raise TypeError('Expected callable: %r' % (callback, ))
         self._links.add(callback)
         if self.ready() and self._notifier is None:
-            self._notifier = core.active_event(self._notify_links)
+            self._notifier = get_hub().reactor.active_event(self._notify_links)
 
     def link(self, receiver=None, GreenletLink=GreenletLink, SpawnedLink=SpawnedLink):
         """Link greenlet's completion to callable or another greenlet.
@@ -546,7 +545,7 @@ def _killall(greenlets, exception):
 def killall(greenlets, exception=GreenletExit, block=True, timeout=None):
     if block:
         waiter = Waiter()
-        core.active_event(_killall3, greenlets, exception, waiter)
+        get_hub().reactor.active_event(_killall3, greenlets, exception, waiter)
         if block:
             t = Timeout.start_new(timeout)
             try:
@@ -556,7 +555,7 @@ def killall(greenlets, exception=GreenletExit, block=True, timeout=None):
             finally:
                 t.cancel()
     else:
-        core.active_event(_killall, greenlets, exception)
+        get_hub().reactor.active_event(_killall, greenlets, exception)
 
 
 class LinkedExited(Exception):

gevent/http.py

 # Copyright (c) 2009-2010 Denis Bilenko. See LICENSE for details.
 from gevent import core
 from gevent.baseserver import BaseServer
+from gevent.hub import get_hub
 
 
 __all__ = ['HTTPServer']
@@ -38,7 +39,8 @@ class HTTPServer(BaseServer):
         request.send_reply(503, 'Service Unavailable', msg)
 
     def start_accepting(self):
-        self.http = core.http(self._on_request, self.default_response_headers)
+        reactor = get_hub().reactor
+        self.http = core.http(self._on_request, self.default_response_headers, base=reactor)
         self.http.accept(self.socket.fileno())
 
     def stop_accepting(self):

gevent/hub.py

@@ -56,11 +56,11 @@ def _switch_helper(function, args, kwargs):
 def spawn_raw(function, *args, **kwargs):
     if kwargs:
         g = greenlet(_switch_helper, get_hub())
-        core.active_event(g.switch, function, args, kwargs)
+        get_hub().reactor.active_event(g.switch, function, args, kwargs)
         return g
     else:
         g = greenlet(function, get_hub())
-        core.active_event(g.switch, *args)
+        get_hub().reactor.active_event(g.switch, *args)
         return g
 
 
@@ -74,7 +74,7 @@ def sleep(seconds=0):
     unique_mark = object()
     if not seconds >= 0:
         raise IOError(22, 'Invalid argument')
-    timer = core.timer(seconds, getcurrent().switch, unique_mark)
+    timer = get_hub().reactor.timer(seconds, getcurrent().switch, unique_mark)
     try:
         switch_result = get_hub().switch()
         assert switch_result is unique_mark, 'Invalid switch into sleep(): %r' % (switch_result, )
@@ -91,25 +91,25 @@ def kill(greenlet, exception=GreenletExit):
     so you have to use this function.
     """
     if not greenlet.dead:
-        core.active_event(greenlet.throw, exception)
+        get_hub().reactor.active_event(greenlet.throw, exception)
 
 
 def _wrap_signal_handler(handler, args, kwargs):
     try:
         handler(*args, **kwargs)
     except:
-        core.active_event(MAIN.throw, *sys.exc_info())
+        get_hub().reactor.active_event(MAIN.throw, *sys.exc_info())
 
 
 def signal(signalnum, handler, *args, **kwargs):
-    return core.signal(signalnum, lambda: spawn_raw(_wrap_signal_handler, handler, args, kwargs))
+    return get_hub().reactor.signal(signalnum, lambda: spawn_raw(_wrap_signal_handler, handler, args, kwargs))
 
 
 if _original_fork is not None:
 
     def fork():
         result = _original_fork()
-        core.reinit()
+        get_hub().reactor.reinit()
         return result
 
 
@@ -121,19 +121,31 @@ def shutdown():
         hub.shutdown()
 
 
+def get_hub_class():
+    """Return the type of hub to use for the current thread.
+
+    If there's no type of hub for the current thread yet, 'gevent.hub.Hub' is used.
+    """
+    global _threadlocal
+    try:
+        hubtype = _threadlocal.Hub
+    except AttributeError:
+        hubtype = None
+    if hubtype is None:
+        hubtype = _threadlocal.Hub = Hub
+    return hubtype
+
+
 def get_hub():
+    """Return the hub for the current thread.
+
+    If hub does not exists in the current thread, the new one is created with call to :meth:`get_hub_class`.
+    """
     global _threadlocal
     try:
         return _threadlocal.hub
     except AttributeError:
-        try:
-            hubtype = _threadlocal.Hub
-        except AttributeError:
-            # do not pretend to support multiple threads because it's not implemented properly by core.pyx
-            # this may change in the future, although currently I don't have a strong need for this
-            raise NotImplementedError('gevent is only usable from a single thread')
-        if hubtype is None:
-            hubtype = Hub
+        hubtype = get_hub_class()
         hub = _threadlocal.hub = hubtype()
         return hub
 
@@ -144,13 +156,19 @@ class Hub(greenlet):
     It is created automatically by :func:`get_hub`.
     """
 
-    def __init__(self):
+    reactor_class = core.event_base
+
+    def __init__(self, reactor=None):
         greenlet.__init__(self)
         self.keyboard_interrupt_signal = None
+        if reactor is None:
+            self.reactor = self.reactor_class()
+        else:
+            self.reactor = reactor
 
     def switch(self):
         cur = getcurrent()
-        assert cur is not self, 'Cannot switch to MAINLOOP from MAINLOOP'
+        assert cur is not self, 'Impossible to call blocking function in the event loop callback'
         exc_info = sys.exc_info()
         try:
             sys.exc_clear()
@@ -166,16 +184,16 @@ class Hub(greenlet):
 
     def run(self):
         global _threadlocal
-        assert self is getcurrent(), 'Do not call run() directly'
+        assert self is getcurrent(), 'Do not call Hub.run() directly'
         try:
-            self.keyboard_interrupt_signal = signal(2, core.active_event, MAIN.throw, KeyboardInterrupt)
+            self.keyboard_interrupt_signal = signal(2, self.reactor.active_event, MAIN.throw, KeyboardInterrupt)
         except IOError:
             pass  # no signal() on windows
         try:
             loop_count = 0
             while True:
                 try:
-                    result = core.dispatch()
+                    result = self.reactor.dispatch()
                 except IOError, ex:
                     loop_count += 1
                     if loop_count > 15:
@@ -196,7 +214,9 @@ class Hub(greenlet):
         if self.keyboard_interrupt_signal is not None:
             self.keyboard_interrupt_signal.cancel()
             self.keyboard_interrupt_signal = None
-        core.dns_shutdown()
+        dns = self.reactor.dns
+        if dns is not None:
+            dns.free()
         if not self or self.dead:
             if _threadlocal.__dict__.get('hub') is self:
                 _threadlocal.__dict__.pop('hub')
@@ -231,7 +251,7 @@ class Waiter(object):
     The :meth:`get` method must be called from a greenlet other than :class:`Hub`.
 
         >>> result = Waiter()
-        >>> _ = core.timer(0.1, result.switch, 'hello from Waiter')
+        >>> _ = get_hub().reactor.timer(0.1, result.switch, 'hello from Waiter')
         >>> result.get() # blocks for 0.1 seconds
         'hello from Waiter'
 
@@ -239,7 +259,7 @@ class Waiter(object):
     :class:`Waiter` stores the value.
 
         >>> result = Waiter()
-        >>> _ = core.timer(0.1, result.switch, 'hi from Waiter')
+        >>> _ = get_hub().reactor.timer(0.1, result.switch, 'hi from Waiter')
         >>> sleep(0.2)
         >>> result.get() # returns immediatelly without blocking
         'hi from Waiter'

gevent/levent.pxd

+ctypedef void (*event_handler)(int fd, short evtype, void *arg)
+ctypedef void (*event_log_cb)(int severity, char *msg)
+
+cdef extern from "libevent.h":
+
+    # event.h:
+    struct timeval:
+        unsigned int tv_sec
+        unsigned int tv_usec
+
+    struct event:
+        void* ev_base
+        int   ev_fd
+        short ev_events
+        int   ev_flags
+        void *ev_arg
+    
+    char* event_get_version()
+
+    int EV_TIMEOUT
+    int EV_READ
+    int EV_WRITE
+    int EV_SIGNAL
+    int EV_PERSIST
+
+    int EVLIST_TIMEOUT
+    int EVLIST_INSERTED
+    int EVLIST_SIGNAL
+    int EVLIST_ACTIVE
+    int EVLIST_INTERNAL
+    int EVLIST_INIT
+
+    int DNS_ERR_NONE
+    int DNS_ERR_FORMAT
+    int DNS_ERR_SERVERFAILED
+    int DNS_ERR_NOTEXIST
+    int DNS_ERR_NOTIMPL
+    int DNS_ERR_REFUSED
+    int DNS_ERR_TRUNCATED
+    int DNS_ERR_UNKNOWN
+    int DNS_ERR_TIMEOUT
+    int DNS_ERR_SHUTDOWN
+
+    int DNS_IPv4_A
+    int DNS_PTR
+    int DNS_IPv6_AAAA
+
+    int DNS_QUERY_NO_SEARCH
+
+    void* event_base_new()
+    int   event_reinit(void *base)
+    int   event_base_dispatch(void*) nogil
+    char* event_base_get_method(void*)
+    void  event_base_free(void*)
+    int   event_base_set(void *, event*)
+
+    void event_set(event *ev, int fd, short event, event_handler handler, void *arg)
+    int  event_add(event *ev, timeval *tv)
+    int  event_del(event *ev)
+    int  event_pending(event *ev, short, timeval *tv)
+    void event_active(event *ev, int res, short ncalls)
+
+    int EVLOOP_ONCE
+    int EVLOOP_NONBLOCK
+    char* _EVENT_VERSION
+    
+    struct evutil_addrinfo:
+        int   ai_flags            # AI_PASSIVE, AI_CANONNAME, AI_NUMERICHOST
+        int   ai_family           # PF_xxx
+        int   ai_socktype         # SOCK_xxx
+        int   ai_protocol         # 0 or IPPROTO_xxx for IPv4 and IPv6
+        size_t ai_addrlen         # length of ai_addr
+        char *ai_canonname        # canonical name for nodename
+        void *ai_addr             # binary address
+        evutil_addrinfo *ai_next  # next structure in linked list
+
+    ctypedef void (*evdns_callback_type)(int result, char t, int count, int ttl, void *addrs, void *arg)
+    ctypedef void (*evdns_getaddrinfo_cb)(int result, evutil_addrinfo *res, void *arg)
+
+    void* evdns_base_new(void *event_base, int initialize_nameservers)
+
+    void* current_base
+    void evdns_base_free(void *dns_base, int fail_requests)
+    char* evdns_err_to_string(int err)
+    int evdns_base_nameserver_ip_add(void *dns_base, char *ip_as_string)
+    int evdns_base_count_nameservers(void *base)
+    void *evdns_base_resolve_ipv4(void *dns_base, char *name, int flags, evdns_callback_type callback, void *ptr)
+    void *evdns_base_resolve_ipv6(void *dns_base, char *name, int flags, evdns_callback_type callback, void *ptr)
+    void *evdns_base_resolve_reverse(void *dns_base, void *, int flags, evdns_callback_type callback, void *ptr)
+    void *evdns_base_resolve_reverse_ipv6(void *dns_base, void *, int flags, evdns_callback_type callback, void *ptr)
+    void  evdns_cancel_request(void *dns_base, void *req)
+    int   evdns_base_set_option(void *dns_base, char *option, char *val)
+
+    void *evdns_getaddrinfo(void *dns_base, char *nodename, char *servname, evutil_addrinfo *hints_in, evdns_getaddrinfo_cb cb, void *arg)
+    void evdns_getaddrinfo_cancel(void*)
+
+    int EVUTIL_EAI_CANCEL

gevent/libevent.h

@@ -20,6 +20,7 @@
 #include "event2/buffer_compat.h"
 #include "event2/dns.h"
 #include "event2/dns_compat.h"
+#include "event2/util.h"
 
 #define EVBUFFER_DRAIN evbuffer_drain
 #define EVHTTP_SET_CB  evhttp_set_cb

gevent/queue.py

@@ -34,7 +34,6 @@ from Queue import Full, Empty
 
 from gevent.timeout import Timeout
 from gevent.hub import get_hub, Waiter, getcurrent, _NONE
-from gevent import core
 
 
 __all__ = ['Queue', 'PriorityQueue', 'LifoQueue', 'JoinableQueue']
@@ -240,7 +239,7 @@ class Queue(object):
 
     def _schedule_unlock(self):
         if self._event_unlock is None:
-            self._event_unlock = core.active_event(self._unlock)
+            self._event_unlock = get_hub().reactor.active_event(self._unlock)
             # QQQ re-activate event (with event_active libevent call) instead of creating a new one each time
 
     def __iter__(self):

gevent/select.py

 # Copyright (c) 2009-2010 Denis Bilenko. See LICENSE for details.
 
-from gevent import core
 from gevent.timeout import Timeout
 from gevent.event import Event
+from gevent import core
+from gevent.hub import get_hub
 
 __implements__ = ['select']
 __all__ = ['error'] + __implements__
@@ -36,12 +37,12 @@ class SelectResult(object):
         if evtype & core.EV_READ:
             self.read.append(event.arg)
             if self.timer is None:
-                self.timer = core.timer(0, self.event.set)
+                self.timer = get_hub().reactor.timer(0, self.event.set)
         elif evtype & core.EV_WRITE:
             self.write.append(event.arg)
             if self.timer is None:
-                self.timer = core.timer(0, self.event.set)
-        # using core.timer(0, ...) to let other active events call update() before Event.wait() returns
+                self.timer = get_hub().reactor.timer(0, self.event.set)
+        # using timer(0, ...) to let other active events call update() before Event.wait() returns
 
 
 def select(rlist, wlist, xlist, timeout=None):
@@ -54,10 +55,15 @@ def select(rlist, wlist, xlist, timeout=None):
     result = SelectResult()
     try:
         try:
+            reactor = get_hub().reactor
             for readfd in rlist:
-                allevents.append(core.read_event(get_fileno(readfd), result.update, arg=readfd))
+                event = reactor.read_event(get_fileno(readfd))
+                event.add(None, result.update, arg=readfd)
+                allevents.append(event)
             for writefd in wlist:
-                allevents.append(core.write_event(get_fileno(writefd), result.update, arg=writefd))
+                event = reactor.write_event(get_fileno(writefd))
+                event.add(None, result.update, arg=writefd)
+                allevents.append(event)
         except IOError, ex:
             raise error(*ex.args)
         result.event.wait(timeout=timeout)

gevent/server.py

@@ -4,8 +4,8 @@ import sys
 import errno
 import traceback
 from gevent import socket
-from gevent import core
 from gevent.baseserver import BaseServer
+from gevent.hub import get_hub
 
 
 __all__ = ['StreamServer']
@@ -88,7 +88,8 @@ class StreamServer(BaseServer):
 
     def start_accepting(self):
         if self._accept_event is None:
-            self._accept_event = core.read_event(self.socket.fileno(), self._do_accept, persist=True)
+            self._accept_event = get_hub().reactor.read_event(self.socket.fileno(), persist=True)
+            self._accept_event.add(None, self._do_accept)
 
     def _start_accepting_if_started(self, _event=None):
         if self.started:
@@ -139,7 +140,7 @@ class StreamServer(BaseServer):
             traceback.print_exc()
         if self.delay >= 0:
             self.stop_accepting()
-            self._start_accepting_timer = core.timer(self.delay, self.start_accepting)
+            self._start_accepting_timer = get_hub().reactor.timer(self.delay, self._start_accepting_if_started)
             self.delay = min(self.max_delay, self.delay * 2)
 
     def is_fatal_error(self, ex):

gevent/socket.py

@@ -75,7 +75,6 @@ __imports__ = ['error',
 
 import sys
 import time
-import random
 import re
 
 is_windows = sys.platform == 'win32'
@@ -123,18 +122,9 @@ for name in __socket__.__all__:
 
 del name, value
 
-if 'inet_ntop' not in globals():
-    # inet_ntop is required by our implementation of getaddrinfo
-
-    def inet_ntop(address_family, packed_ip):
-        if address_family == AF_INET:
-            return inet_ntoa(packed_ip)
-        # XXX: ipv6 won't work on windows
-        raise NotImplementedError('inet_ntop() is not available on this platform')
-
 # XXX: implement blocking functions that are not yet implemented
 
-from gevent.hub import getcurrent, get_hub
+from gevent.hub import getcurrent, get_hub, Waiter
 from gevent import core
 
 _ip4_re = re.compile('^[\d\.]+$')
@@ -148,29 +138,37 @@ def _wait_helper(ev, evtype):
         current.switch(ev)
 
 
-def wait_read(fileno, timeout=None, timeout_exc=timeout('timed out'), event=None):
-    """Block the current greenlet until *fileno* is ready to read.
+def wait(event, timeout=None, timeout_exc=timeout('timed out')):
+    """Block the current greenlet until *event* is ready.
 
     If *timeout* is non-negative, then *timeout_exc* is raised after *timeout* second has passed.
     By default *timeout_exc* is ``socket.timeout('timed out')``.
 
     If :func:`cancel_wait` is called, raise ``socket.error(EBADF, 'File descriptor was closed in another greenlet')``.
     """
-    if event is None:
-        event = core.read_event(fileno, _wait_helper, timeout, (getcurrent(), timeout_exc))
-    else:
-        assert event.callback == _wait_helper, event.callback
-        assert event.arg is None, 'This event is already used by another greenlet: %r' % (event.arg, )
-        event.arg = (getcurrent(), timeout_exc)
-        event.add(timeout)
+    assert event.arg is None, 'This event is already used by another greenlet: %r' % (event.arg, )
+    event.add(timeout, _wait_helper, (getcurrent(), timeout_exc))
     try:
         switch_result = get_hub().switch()
-        assert event is switch_result, 'Invalid switch into wait_read(): %r' % (switch_result, )
+        assert event is switch_result, 'Invalid switch into wait(%r): %r' % (event, switch_result, )
     finally:
         event.cancel()
         event.arg = None
 
 
+def wait_read(fileno, timeout=None, timeout_exc=timeout('timed out'), event=None):
+    """Block the current greenlet until *fileno* is ready to read.
+
+    If *timeout* is non-negative, then *timeout_exc* is raised after *timeout* second has passed.
+    By default *timeout_exc* is ``socket.timeout('timed out')``.
+
+    If :func:`cancel_wait` is called, raise ``socket.error(EBADF, 'File descriptor was closed in another greenlet')``.
+    """
+    if event is None:
+        event = get_hub().reactor.read_event(fileno)
+    return wait(event, timeout, timeout_exc)
+
+
 def wait_write(fileno, timeout=None, timeout_exc=timeout('timed out'), event=None):
     """Block the current greenlet until *fileno* is ready to write.
 
@@ -180,18 +178,8 @@ def wait_write(fileno, timeout=None, timeout_exc=timeout('timed out'), event=Non
     If :func:`cancel_wait` is called, raise ``socket.error(EBADF, 'File descriptor was closed in another greenlet')``.
     """
     if event is None:
-        event = core.write_event(fileno, _wait_helper, timeout, (getcurrent(), timeout_exc))
-    else:
-        assert event.callback == _wait_helper, event.callback
-        assert event.arg is None, 'This event is already used by another greenlet: %r' % (event.arg, )
-        event.arg = (getcurrent(), timeout_exc)
-        event.add(timeout)
-    try:
-        switch_result = get_hub().switch()
-        assert event is switch_result, 'Invalid switch into wait_write(): %r' % (switch_result, )
-    finally:
-        event.arg = None
-        event.cancel()
+        event = get_hub().reactor.write_event(fileno)
+    return wait(event, timeout, timeout_exc)
 
 
 def wait_readwrite(fileno, timeout=None, timeout_exc=timeout('timed out'), event=None):
@@ -203,18 +191,8 @@ def wait_readwrite(fileno, timeout=None, timeout_exc=timeout('timed out'), event
     If :func:`cancel_wait` is called, raise ``socket.error(EBADF, 'File descriptor was closed in another greenlet')``.
     """
     if event is None:
-        event = core.readwrite_event(fileno, _wait_helper, timeout, (getcurrent(), timeout_exc))
-    else:
-        assert event.callback == _wait_helper, event.callback
-        assert event.arg is None, 'This event is already used by another greenlet: %r' % (event.arg, )
-        event.arg = (getcurrent(), timeout_exc)
-        event.add(timeout)
-    try:
-        switch_result = get_hub().switch()
-        assert event is switch_result, 'Invalid switch into wait_readwrite(): %r' % (switch_result, )
-    finally:
-        event.arg = None
-        event.cancel()
+        event = get_hub().reactor.readwrite_event(fileno)
+    return wait(event, timeout, timeout_exc)
 
 
 def __cancel_wait(event):
@@ -225,7 +203,7 @@ def __cancel_wait(event):
 
 
 def cancel_wait(event):
-    core.active_event(__cancel_wait, event)
+    get_hub().reactor.active_event(__cancel_wait, event)
 
 
 if sys.version_info[:2] <= (2, 4):
@@ -290,9 +268,10 @@ class socket(object):
                 self._sock = _sock
                 self.timeout = _socket.getdefaulttimeout()
         self._sock.setblocking(0)
-        self._read_event = core.event(core.EV_READ, self.fileno(), _wait_helper)
-        self._write_event = core.event(core.EV_WRITE, self.fileno(), _wait_helper)
-        self._rw_event = core.event(core.EV_READ | core.EV_WRITE, self.fileno(), _wait_helper)
+        reactor = get_hub().reactor
+        self._read_event = reactor.read_event(self.fileno())
+        self._write_event = reactor.write_event(self.fileno())
+        self._rw_event = reactor.readwrite_event(self.fileno())
 
     def __repr__(self):
         return '<%s at %s %s>' % (type(self).__name__, hex(id(self)), self._formatinfo())
@@ -334,7 +313,7 @@ class socket(object):
                 if ex[0] != EWOULDBLOCK or self.timeout == 0.0:
                     raise
                 sys.exc_clear()
-            wait_read(sock.fileno(), timeout=self.timeout, event=self._read_event)
+            wait(self._read_event, timeout=self.timeout)
         return socket(_sock=client_socket), address
 
     def close(self):
@@ -361,7 +340,7 @@ class socket(object):
                 if not result or result == EISCONN:
                     break
                 elif (result in (EWOULDBLOCK, EINPROGRESS, EALREADY)) or (result == EINVAL and is_windows):
-                    wait_readwrite(sock.fileno(), event=self._rw_event)
+                    wait(self._rw_event)
                 else:
                     raise error(result, strerror(result))
         else:
@@ -377,7 +356,7 @@ class socket(object):
                     timeleft = end - time.time()
                     if timeleft <= 0:
                         raise timeout('timed out')
-                    wait_readwrite(sock.fileno(), timeout=timeleft, event=self._rw_event)
+                    wait(self._rw_event, timeout=timeleft)
                 else:
                     raise error(result, strerror(result))
 
@@ -417,7 +396,7 @@ class socket(object):
                 # QQQ without clearing exc_info test__refcount.test_clean_exit fails
                 sys.exc_clear()
             try:
-                wait_read(sock.fileno(), timeout=self.timeout, event=self._read_event)
+                wait(self._read_event, timeout=self.timeout)
             except error, ex:
                 if ex[0] == EBADF:
                     return ''
@@ -432,7 +411,7 @@ class socket(object):
                 if ex[0] != EWOULDBLOCK or self.timeout == 0.0:
                     raise
                 sys.exc_clear()
-            wait_read(sock.fileno(), timeout=self.timeout, event=self._read_event)
+            wait(self._read_event, timeout=self.timeout)
 
     def recvfrom_into(self, *args):
         sock = self._sock
@@ -443,7 +422,7 @@ class socket(object):
                 if ex[0] != EWOULDBLOCK or self.timeout == 0.0:
                     raise
                 sys.exc_clear()
-            wait_read(sock.fileno(), timeout=self.timeout, event=self._read_event)
+            wait(self._read_event, timeout=self.timeout)
 
     def recv_into(self, *args):
         sock = self._sock
@@ -457,7 +436,7 @@ class socket(object):
                     raise
                 sys.exc_clear()
             try:
-                wait_read(sock.fileno(), timeout=self.timeout, event=self._read_event)
+                wait(self._read_event, timeout=self.timeout)
             except error, ex:
                 if ex[0] == EBADF:
                     return 0
@@ -474,7 +453,7 @@ class socket(object):
                 raise
             sys.exc_clear()
             try:
-                wait_write(sock.fileno(), timeout=timeout, event=self._write_event)
+                wait(self._write_event, timeout=timeout)
             except error, ex:
                 if ex[0] == EBADF:
                     return 0
@@ -515,7 +494,7 @@ class socket(object):
             if ex[0] != EWOULDBLOCK or timeout == 0.0:
                 raise
             sys.exc_clear()
-            wait_write(sock.fileno(), timeout=self.timeout, event=self._write_event)
+            wait(self._write_event, timeout=self.timeout)
             try:
                 return sock.sendto(*args)
             except error, ex2:
@@ -640,100 +619,39 @@ def create_connection(address, timeout=_GLOBAL_DEFAULT_TIMEOUT, source_address=N
     raise msg
 
 
-try:
-    from gevent.dns import resolve_ipv4, resolve_ipv6
-except Exception:
-    import traceback
-    traceback.print_exc()
-    __implements__.remove('gethostbyname')
-    __implements__.remove('getaddrinfo')
-else:
+def gethostbyname(host):
+    # artificial limitations that regular gethostbyname has
+    if host is None:
+        raise TypeError('gethostbyname() argument must be string')
+    host = host.encode('ascii')
+    res = getaddrinfo(host, 0)
+    if res:
+        return res[0][4][0]
 
-    def gethostbyname(hostname):
-        """:func:`socket.gethostbyname` implemented using :mod:`gevent.dns`.
-
-        Differs in the following ways:
-
-        * raises :class:`DNSError` (a subclass of :class:`socket.gaierror`) with dns error
-          codes instead of standard socket error codes
-        * does not support ``/etc/hosts`` but calls the original :func:`socket.gethostbyname`
-          if *hostname* has no dots
-        * does not iterate through all addresses, instead picks a random one each time
-        """
-        # TODO: this is supposed to iterate through all the addresses
-        # could use a global dict(hostname, iter)
-        # - fix these nasty hacks for localhost, ips, etc.
-        if not isinstance(hostname, str) or '.' not in hostname:
-            return _socket.gethostbyname(hostname)
-        if _ip4_re.match(hostname):
-            return hostname
-        if hostname == _socket.gethostname():
-            return _socket.gethostbyname(hostname)
-        _ttl, addrs = resolve_ipv4(hostname)
-        return inet_ntoa(random.choice(addrs))
-
-    def getaddrinfo(host, port, *args, **kwargs):
-        """*Some* approximation of :func:`socket.getaddrinfo` implemented using :mod:`gevent.dns`.
-
-        If *host* is not a string, does not has any dots or is a numeric IP address, then
-        the standard :func:`socket.getaddrinfo` is called.
-
-        Otherwise, calls either :func:`resolve_ipv4` or :func:`resolve_ipv6` and
-        formats the result the way :func:`socket.getaddrinfo` does it.
-
-        Differs in the following ways:
-
-        * raises :class:`DNSError` (a subclass of :class:`gaierror`) with libevent-dns error
-          codes instead of standard socket error codes
-        * IPv6 support is untested.
-        * AF_UNSPEC only tries IPv4
-        * only supports TCP, UDP, IP protocols
-        * port must be numeric, does not support string service names. see socket.getservbyname
-        * *flags* argument is ignored
-
-        Additionally, supports *evdns_flags* keyword arguments (default ``0``) that is passed
-        to :mod:`dns` functions.
-        """
-        family, socktype, proto, _flags = args + (None, ) * (4 - len(args))
-        if isinstance(host, unicode):
-            host = host.encode('idna')
-        if not isinstance(host, str) or '.' not in host or _ip4_re.match(host):
-            return _socket.getaddrinfo(host, port, *args)
-
-        evdns_flags = kwargs.pop('evdns_flags', 0)
-        if kwargs:
-            raise TypeError('Unsupported keyword arguments: %s' % (kwargs.keys(), ))
-
-        if family in (None, AF_INET, AF_UNSPEC):
-            family = AF_INET
-            # TODO: AF_UNSPEC means try both AF_INET and AF_INET6
-            _ttl, addrs = resolve_ipv4(host, evdns_flags)
-        elif family == AF_INET6:
-            _ttl, addrs = resolve_ipv6(host, evdns_flags)
-        else:
-            raise NotImplementedError('family is not among AF_UNSPEC/AF_INET/AF_INET6: %r' % (family, ))
-
-        socktype_proto = [(SOCK_STREAM, 6), (SOCK_DGRAM, 17), (SOCK_RAW, 0)]
-        if socktype:
-            socktype_proto = [(x, y) for (x, y) in socktype_proto if socktype == x]
-        if proto:
-            socktype_proto = [(x, y) for (x, y) in socktype_proto if proto == y]
-
-        result = []
-        for addr in addrs:
-            for socktype, proto in socktype_proto:
-                result.append((family, socktype, proto, '', (inet_ntop(family, addr), port)))
-        return result
-        # TODO libevent2 has getaddrinfo that is probably better than the hack above; should wrap that.
 
+def getaddrinfo(host, port, family=0, socktype=0, proto=0, flags=0):
+    dns = get_hub().reactor.dns
+    if dns is None:
+        return _socket.getaddrinfo(host, port, family, socktype, proto, flags)
+    else:
+        waiter = Waiter()
+        request = dns.getaddrinfo(waiter.switch_args, host, port, family, socktype, proto, flags)
+        try:
+            result, request_error, _arg = waiter.get()
+        except:
+            if request is not None:
+                request.cancel()
+            raise
+        if request_error is None:
+            return result
+        raise request_error
 
-_have_ssl = False
 
 try:
     from gevent.ssl import sslwrap_simple as ssl, SSLError as sslerror, SSLSocket as SSLType
     _have_ssl = True
 except ImportError:
-    pass
+    _have_ssl = False
 
 
 if sys.version_info[:2] <= (2, 5) and _have_ssl:

gevent/timeout.py

@@ -13,8 +13,7 @@ to arbitrary code.
     which no switches occur, :class:`Timeout` is powerless.
 """
 
-from gevent import core
-from gevent.hub import getcurrent, _NONE
+from gevent.hub import getcurrent, _NONE, get_hub
 
 __all__ = ['Timeout',
            'with_timeout']
@@ -86,17 +85,17 @@ class Timeout(BaseException):
     def __init__(self, seconds=None, exception=None):
         self.seconds = seconds
         self.exception = exception
-        self.timer = None
+        self.timer = get_hub().reactor.timer()
 
     def start(self):
         """Schedule the timeout."""
         assert not self.pending, '%r is already started; to restart it, cancel it first' % self
         if self.seconds is None:  # "fake" timeout (never expires)
-            self.timer = None
+            pass
         elif self.exception is None or self.exception is False:  # timeout that raises self
-            self.timer = core.timer(self.seconds, getcurrent().throw, self)
+            self.timer.add(self.seconds, getcurrent().throw, self)
         else:  # regular timeout with user-provided exception
-            self.timer = core.timer(self.seconds, getcurrent().throw, self.exception)
+            self.timer.add(self.seconds, getcurrent().throw, self.exception)
 
     @classmethod
     def start_new(cls, timeout=None, exception=None):
@@ -167,7 +166,7 @@ class Timeout(BaseException):
             return '%s second%s (%s)' % (self.seconds, suffix, self.exception)
 
     def __enter__(self):
-        if self.timer is None:
+        if not self.timer.pending:
             self.start()
         return self