Merge branch 'master' of https://github.com/3ddi/gevent

gevent/corecffi.py

@@ -295,6 +295,8 @@ libev.vfd_free = lambda fd: None
 @ffi.callback("int(void* handle, int revents)")
 def _python_callback(handle, revents):
     watcher = ffi.from_handle(handle)
+    if len(watcher.args) > 0 and watcher.args[0] == GEVENT_CORE_EVENTS:
+        watcher.args = (revents, ) + watcher.args[1:]
     try:
         watcher.callback(*watcher.args)
     except:
@@ -1013,6 +1015,11 @@ class io(watcher):
             raise ValueError('illegal event mask: %r' % events)
         watcher.__init__(self, loop, ref=ref, priority=priority, args=(fd, events))
 
+    def start(self, callback, *args, **kwargs):
+        if kwargs.get('pass_events'):
+            args = (GEVENT_CORE_EVENTS, ) + args
+        super(io, self).start(callback, *args)
+
     def _get_fd(self):
         return libev.vfd_get(self._watcher.fd)
 

gevent/monkey.py

@@ -248,7 +248,6 @@ def patch_select(aggressive=True):
         # since these are blocking we're removing them here. This makes some other
         # modules (e.g. asyncore)  non-blocking, as they use select that we provide
         # when none of these are available.
-        remove_item(select, 'poll')
         remove_item(select, 'epoll')
         remove_item(select, 'kqueue')
         remove_item(select, 'kevent')
@@ -268,9 +267,8 @@ def patch_select(aggressive=True):
 
         if aggressive:
             # If `selectors` had already been imported before we removed
-            # select.poll|epoll|kqueue, these may have been defined in terms
+            # select.epoll|kqueue, these may have been defined in terms
             # of those functions. They'll fail at runtime.
-            remove_item(selectors, 'PollSelector')
             remove_item(selectors, 'EpollSelector')
             remove_item(selectors, 'KqueueSelector')
             selectors.DefaultSelector = selectors.SelectSelector

gevent/select.py

@@ -3,11 +3,19 @@ from __future__ import absolute_import
 from gevent.event import Event
 from gevent.hub import get_hub
 from gevent.hub import integer_types
+from select import POLLIN, POLLOUT
+
+try:
+    from select import poll as original_poll
+    __implements__ = ['select', 'poll']
+except ImportError:
+    original_poll = None
+    __implements__ = ['select']
 
-__implements__ = ['select']
 __all__ = ['error'] + __implements__
 
 import select as __select__
+
 error = __select__.error
 
 
@@ -16,14 +24,13 @@ def get_fileno(obj):
         fileno_f = obj.fileno
     except AttributeError:
         if not isinstance(obj, integer_types):
-            raise TypeError('argument must be an int, or have a fileno() method: %r' % (obj, ))
+            raise TypeError('argument must be an int, or have a fileno() method: %r' % (obj,))
         return obj
     else:
         return fileno_f()
 
 
 class SelectResult(object):
-
     __slots__ = ['read', 'write', 'event']
 
     def __init__(self):
@@ -69,3 +76,59 @@ def select(rlist, wlist, xlist, timeout=None):
     finally:
         for watcher in watchers:
             watcher.stop()
+
+if original_poll is not None:
+    class PollResult(object):
+        __slots__ = ['events', 'event']
+
+        def __init__(self):
+            self.events = set()
+            self.event = Event()
+
+        def add_event(self, events, fd):
+            result_flags = 0
+            result_flags |= POLLIN if events & 1 else 0
+            result_flags |= POLLOUT if events & 2 else 0
+            self.events.add((fd, result_flags))
+            self.event.set()
+
+    class poll(object):
+        def __init__(self):
+            self.fds = {}
+            self.loop = get_hub().loop
+            # Using the original poll to handle immediate response polling.
+            # Using the ev based way, we receive a signal on the first descriptor.
+            self.poll_obj = original_poll()
+
+        def register(self, fd, eventmask=POLLIN | POLLOUT):
+            flags = 0
+            flags |= 1 if eventmask & POLLIN else 0
+            flags |= 2 if eventmask & POLLOUT else 0
+            watcher = self.loop.io(get_fileno(fd), flags)
+            watcher.priority = self.loop.MAXPRI
+            self.fds[fd] = watcher
+            self.poll_obj.register(fd, eventmask)
+
+        def modify(self, fd, eventmask):
+            self.register(fd, eventmask)
+            self.poll_obj.modify(fd, eventmask)
+
+        def poll(self, timeout=None):
+            results = self.poll_obj.poll(0)
+            if results:
+                return results
+            result = PollResult()
+            try:
+                for fd in self.fds:
+                    self.fds[fd].start(result.add_event, get_fileno(fd), pass_events=True)
+                if timeout is not None and -1 < timeout:
+                    timeout /= 1000.0
+                result.event.wait(timeout=timeout)
+                return list(result.events)
+            finally:
+                for fd in self.fds:
+                    self.fds[fd].stop()
+
+        def unregister(self, fd):
+            self.fds.pop(fd, None)
+            self.poll_obj.unregister(fd)

greentest/test__select.py

@@ -23,6 +23,18 @@ if sys.platform != 'win32':
                 os.close(r)
                 os.close(w)
 
+    class TestPollRead(greentest.GenericWaitTestCase):
+        def wait(self, timeout):
+            r, w = os.pipe()
+            try:
+                poll = select.poll()
+                poll.register(r)
+                poll.poll(timeout * 1000)
+                poll.unregister(r)
+            finally:
+                os.close(r)
+                os.close(w)
+
 
 class TestSelectTypes(greentest.TestCase):