Make waiters on Events that arrive while the event is ready and notifying...

Make waiters on Events that arrive while the event is ready and notifying greenlets that had to block run after they do.

Fixes #1520

docs/changes/1520.feature

+Waiters on Event and Semaphore objects that call ``wait()`` or
+``acquire()``, respectively, that find the Event already set, or the
+Semaphore available, no longer "cut in line" and run before any
+previously scheduled greenlets. They now run in the order in which
+they arrived, just as waiters that had to block in those methods do.

src/gevent/__abstract_linkable.pxd

@@ -42,13 +42,19 @@ cdef class AbstractLinkable(object):
    cdef list _links
    cdef bint _notify_all
 
+   cpdef linkcount(self)
    cpdef rawlink(self, callback)
    cpdef bint ready(self)
    cpdef unlink(self, callback)
 
    cdef _check_and_notify(self)
+
+   @cython.nonecheck(False)
+   cdef _notify_link_list(self, list links)
+
    @cython.nonecheck(False)
-   cpdef _notify_links(self)
+   cpdef _notify_links(self, list arrived_while_waiting)
+
    cdef _wait_core(self, timeout, catch=*)
    cdef _wait_return_value(self, waited, wait_success)
    cdef _wait(self, timeout=*)

src/gevent/_abstract_linkable.py

@@ -81,14 +81,6 @@ class AbstractLinkable(object):
         # Instances must define this
         raise NotImplementedError
 
-    def _check_and_notify(self):
-        # If this object is ready to be notified, begin the process.
-        if self.ready() and self._links and not self._notifier:
-            if self.hub is None:
-                self.hub = get_hub()
-
-            self._notifier = self.hub.loop.run_callback(self._notify_links)
-
     def rawlink(self, callback):
         """
         Register a callback to call when this object is ready.
@@ -116,45 +108,77 @@ class AbstractLinkable(object):
             # But we can't set it to None in case it was actually running.
             self._notifier.stop()
 
-    def _notify_links(self):
-        # We release self._notifier here. We are called by it
-        # at the end of the loop, and it is now false in a boolean way (as soon
-        # as this method returns).
+    def _check_and_notify(self):
+        # If this object is ready to be notified, begin the process.
+        if self.ready() and self._links and not self._notifier:
+            if self.hub is None:
+                self.hub = get_hub()
+
+            self._notifier = self.hub.loop.run_callback(self._notify_links, [])
+
+    def _notify_link_list(self, links):
+        # The core of the _notify_links method to notify
+        # links in order. Lets the ``links`` list be mutated,
+        # and only notifies up to the last item in the list, in case
+        # objects are added to it.
+        only_while_ready = not self._notify_all
+        final_link = links[-1]
+        done = set() # of ids
+        while links: # remember this can be mutated
+            if only_while_ready and not self.ready():
+                break
+
+            link = links.pop(0) # Cython optimizes using list internals
+            id_link = id(link)
+            if id_link not in done:
+                # XXX: JAM: What was I thinking? This doesn't make much sense,
+                # there's a good chance `link` will be deallocated, and its id() will
+                # be free to be reused.
+                done.add(id_link)
+                try:
+                    link(self)
+                except: # pylint:disable=bare-except
+                    # We're running in the hub, errors must not escape.
+                    self.hub.handle_error((link, self), *sys.exc_info())
+
+            if link is final_link:
+                break
+
+    def _notify_links(self, arrived_while_waiting):
+        # ``arrived_while_waiting`` is a list of greenlet.switch methods
+        # to call. These were objects that called wait() while we were processing,
+        # and which would have run *before* those that had actually waited
+        # and blocked. Instead of returning True immediately, we add them to this
+        # list so they wait their turn.
+
+        # We release self._notifier here when done invoking links.
+        # The object itself becomes false in a boolean way as soon
+        # as this method returns.
         notifier = self._notifier
         # Early links are allowed to remove later links, and links
-        # are allowed to add more links.
+        # are allowed to add more links, thus we must not
+        # make a copy of our the ``_links`` list, we must traverse it and
+        # mutate in place.
         #
         # We were ready() at the time this callback was scheduled; we
         # may not be anymore, and that status may change during
         # callback processing. Some of our subclasses (Event) will
         # want to notify everyone who was registered when the status
         # became true that it was once true, even though it may not be
-        # anymore. In that case, we must not keep notifying anyone that's
+        # any more. In that case, we must not keep notifying anyone that's
         # newly added after that, even if we go ready again.
-        final_link = self._links[-1]
-        only_while_ready = not self._notify_all
-        done = set() # of ids
+
         try:
-            while self._links: # remember this can be mutated
-                if only_while_ready and not self.ready():
-                    break
-
-                link = self._links.pop(0) # Cython optimizes using list internals
-
-                id_link = id(link)
-                if id_link not in done:
-                    # XXX: JAM: What was I thinking? This doesn't make much sense,
-                    # there's a good chance `link` will be deallocated, and its id() will
-                    # be free to be reused.
-                    done.add(id_link)
-                    try:
-                        link(self)
-                    except: # pylint:disable=bare-except
-                        # We're running in the hub, errors must not escape.
-                        self.hub.handle_error((link, self), *sys.exc_info())
-
-                if link is final_link:
-                    break
+            self._notify_link_list(self._links)
+
+            # Now, those that arrived after we had begun the notification
+            # process. Follow the same rules, stop with those that are
+            # added so far to prevent starvation.
+            if arrived_while_waiting:
+                self._notify_link_list(arrived_while_waiting)
+
+                # Anything left needs to go back on the main list.
+                self._links.extend(arrived_while_waiting)
         finally:
             # We should not have created a new notifier even if callbacks
             # released us because we loop through *all* of our links on the
@@ -203,7 +227,17 @@ class AbstractLinkable(object):
 
     def _wait(self, timeout=None):
         if self.ready():
-            return self._wait_return_value(False, False)
+            result = self._wait_return_value(False, False) # pylint:disable=assignment-from-none
+            if self._notifier:
+                # We're already notifying waiters; one of them must have run
+                # and switched to us.
+                switch = getcurrent().switch # pylint:disable=undefined-variable
+                self._notifier.args[0].append(switch)
+                switch_result = self.hub.switch()
+                if switch_result is not self: # pragma: no cover
+                    raise InvalidSwitchError('Invalid switch into Event.wait(): %r' % (result, ))
+
+            return result
 
         gotit = self._wait_core(timeout)
         return self._wait_return_value(True, gotit)

src/gevent/event.py

@@ -22,31 +22,54 @@ def _get_linkable():
 locals()['AbstractLinkable'] = _get_linkable()
 del _get_linkable
 
-# Sadly, something about the way we have to "import" AbstractLinkable
-# breaks pylint's inference of slots, even though they're declared
-# right here.
-# pylint:disable=assigning-non-slot
 
 class Event(AbstractLinkable): # pylint:disable=undefined-variable
-    """A synchronization primitive that allows one greenlet to wake up one or more others.
-    It has the same interface as :class:`threading.Event` but works across greenlets.
-
-    An event object manages an internal flag that can be set to true with the
-    :meth:`set` method and reset to false with the :meth:`clear` method. The :meth:`wait` method
-    blocks until the flag is true.
+    """
+    A synchronization primitive that allows one greenlet to wake up
+    one or more others. It has the same interface as
+    :class:`threading.Event` but works across greenlets.
+
+    An event object manages an internal flag that can be set to true
+    with the :meth:`set` method and reset to false with the
+    :meth:`clear` method. The :meth:`wait` method blocks until the
+    flag is true; as soon as the flag is set to true, all greenlets
+    that are currently blocked in a call to :meth:`wait` will be scheduled
+    to awaken.
+
+    Note that the flag may be cleared and set many times before
+    any individual greenlet runs; all the greenlet can know for sure is that the
+    flag was set *at least once* while it was waiting.
+    If the greenlet cares whether the flag is still
+    set, it must check with :meth:`ready` and possibly call back into
+    :meth:`wait` again.
 
     .. note::
-        The order and timing in which waiting greenlets are awakened is not determined.
-        As an implementation note, in gevent 1.1 and 1.0, waiting greenlets are awakened in a
-        undetermined order sometime *after* the current greenlet yields to the event loop.  Other greenlets
-        (those not waiting to be awakened) may run between the current greenlet yielding and
-        the waiting greenlets being awakened. These details may change in the future.
+
+        The exact order and timing in which waiting greenlets are awakened is not determined.
+
+        Once the event is set, other greenlets may run before any waiting greenlets
+        are awakened.
+
+        While the code here will awaken greenlets in the order in which they
+        waited, each such greenlet that runs may in turn cause other greenlets
+        to run.
+
+        These details may change in the future.
 
     .. versionchanged:: 1.5a3
-       Waiting greenlets are now awakened in the order in which they waited.
+
+        Waiting greenlets are now awakened in
+        the order in which they waited.
+
     .. versionchanged:: 1.5a3
-       The low-level ``rawlink`` method (most users won't use this) now automatically
-       unlinks waiters before calling them.
+
+        The low-level ``rawlink`` method (most users won't use this) now
+        automatically unlinks waiters before calling them.
+
+    .. versionchanged:: NEXT
+
+        Callers to ``wait`` that find the event already set will now run
+        after any other waiters that had to block. See :issue:`1520`.
     """
 
     __slots__ = ('_flag',)
@@ -107,11 +130,11 @@ class Event(AbstractLinkable): # pylint:disable=undefined-variable
 
     def wait(self, timeout=None):
         """
-        Block until the internal flag is true.
+        Block until this object is :meth:`ready`.
 
         If the internal flag is true on entry, return immediately. Otherwise,
         block until another thread (greenlet) calls :meth:`set` to set the flag to true,
-        or until the optional timeout occurs.
+        or until the optional *timeout* expires.
 
         When the *timeout* argument is present and not ``None``, it should be a
         floating point number specifying a timeout for the operation in seconds

src/gevent/tests/test__event.py

@@ -25,7 +25,7 @@ class TestEventWait(AbstractGenericWaitTestCase):
         str(Event())
 
 
-class TestWaitEvent(AbstractGenericWaitTestCase):
+class TestGeventWaitOnEvent(AbstractGenericWaitTestCase):
 
     def wait(self, timeout):
         gevent.wait([Event()], timeout=timeout)
@@ -246,6 +246,32 @@ class TestEventBasics(greentest.TestCase):
         del e
         del r
 
+    def test_wait_while_notifying(self):
+        # If someone calls wait() on an Event that is
+        # ready, and notifying other waiters, that new
+        # waiter still runs at the end, but this does not
+        # require a trip around the event loop.
+        # See https://github.com/gevent/gevent/issues/1520
+        event = Event()
+        results = []
+
+        def wait_then_append(arg):
+            event.wait()
+            results.append(arg)
+
+        gevent.spawn(wait_then_append, 1)
+        gevent.spawn(wait_then_append, 2)
+        gevent.idle()
+        self.assertEqual(2, event.linkcount())
+        check = gevent.get_hub().loop.check()
+        check.start(results.append, 4)
+        event.set()
+        wait_then_append(3)
+        self.assertEqual(results, [1, 2, 3])
+        # Note that the check event DID NOT run.
+        check.stop()
+        check.close()
+
 
 del AbstractGenericGetTestCase
 del AbstractGenericWaitTestCase