Fix calling gevent.wait() twice on a set Event.

The detection of when to de-queue an existing notifier was wrong, leading us to de-queue it while it was running and then promptly re-queue one.

docs/changes/1540.bugfix

+Using ``gevent.wait`` on an ``Event`` more than once, when that Event
+is already set, could previously raise an AssertionError.
+
+As part of this, exceptions raised in the main greenlet will now
+include a more complete traceback from the failing greenlet.

src/gevent/_abstract_linkable.py

@@ -101,11 +101,14 @@ class AbstractLinkable(object):
         except ValueError:
             pass
 
-        if not self._links and self._notifier is not None:
-            # If we currently have one queued, de-queue it.
+        if not self._links and self._notifier is not None and self._notifier.pending:
+            # If we currently have one queued, but not running, de-queue it.
             # This will break a reference cycle.
             # (self._notifier -> self._notify_links -> self)
-            # But we can't set it to None in case it was actually running.
+            # If it's actually running, though, (and we're here as a result of callbacks)
+            # we don't want to change it; it needs to finish what its doing
+            # so we don't attempt to start a fresh one or swap it out from underneath the
+            # _notify_links method.
             self._notifier.stop()
 
     def _check_and_notify(self):
@@ -170,7 +173,6 @@ class AbstractLinkable(object):
 
         try:
             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.

src/gevent/event.py

@@ -79,8 +79,11 @@ class Event(AbstractLinkable): # pylint:disable=undefined-variable
         self._flag = False
 
     def __str__(self):
-        return '<%s %s _links[%s]>' % (self.__class__.__name__, (self._flag and 'set') or 'clear',
-                                       self.linkcount())
+        return '<%s %s _links[%s]>' % (
+            self.__class__.__name__,
+            'set' if self._flag else 'clear',
+            self.linkcount()
+        )
 
     def is_set(self):
         """Return true if and only if the internal flag is true."""

src/gevent/hub.py

@@ -531,19 +531,23 @@ class Hub(WaitOperationsGreenlet):
         if not issubclass(type, self.NOT_ERROR):
             self.print_exception(context, type, value, tb)
         if context is None or issubclass(type, self.SYSTEM_ERROR):
-            self.handle_system_error(type, value)
+            self.handle_system_error(type, value, tb)
 
-    def handle_system_error(self, type, value):
+    def handle_system_error(self, type, value, tb=None):
         """
         Called from `handle_error` when the exception type is determined
         to be a :attr:`system error <SYSTEM_ERROR>`.
 
         System errors cause the exception to be raised in the main
         greenlet (the parent of this hub).
+
+        .. versionchanged:: NEXT
+           Allow passing the traceback to associate with the
+           exception if it is rethrown into the main greenlet.
         """
         current = getcurrent()
         if current is self or current is self.parent or self.loop is None:
-            self.parent.throw(type, value)
+            self.parent.throw(type, value, tb)
         else:
             # in case system error was handled and life goes on
             # switch back to this greenlet as well
@@ -553,7 +557,7 @@ class Hub(WaitOperationsGreenlet):
             except: # pylint:disable=bare-except
                 traceback.print_exc(file=self.exception_stream)
             try:
-                self.parent.throw(type, value)
+                self.parent.throw(type, value, tb)
             finally:
                 if cb is not None:
                     cb.stop()

src/gevent/libev/corecext.pyx

@@ -469,7 +469,12 @@ cdef public class loop [object PyGeventLoopObject, type PyGeventLoop_Type]:
                 cb = self._callbacks.popleft()
 
                 libev.ev_unref(self._ptr)
-                gevent_call(self, cb) # XXX: Why is this a C callback, not cython?
+                # On entry, this will set cb.callback to None,
+                # changing cb.pending from True to False; on exit,
+                # this will set cb.args to None, changing bool(cb)
+                # from True to False.
+                # XXX: Why is this a C callback, not cython?
+                gevent_call(self, cb)
                 count -= 1
 
                 if count == 0 and self._callbacks.head is not None:

src/gevent/tests/test__event.py

@@ -272,6 +272,17 @@ class TestEventBasics(greentest.TestCase):
         check.stop()
         check.close()
 
+    def test_gevent_wait_twice_when_already_set(self):
+        event = Event()
+        event.set()
+        # First one works fine.
+        result = gevent.wait([event])
+        self.assertEqual(result, [event])
+        # Second one used to fail with an AssertionError,
+        # now it passes
+        result = gevent.wait([event])
+        self.assertEqual(result, [event])
+
 
 del AbstractGenericGetTestCase
 del AbstractGenericWaitTestCase