Merge pull request #1488 from gevent/issue1487

Make Semaphores fair.

CHANGES.rst

@@ -36,6 +36,13 @@
   Using spin locks is not recommended, but may have been done in code
   written for threads, especially on Python 3. See :issue:`1464`.
 
+- Fix Semaphore (and monkey-patched threading locks) to be fair. This
+  eliminates the rare potential for starvation of greenlets. As part
+  of this change, the low-level method ``rawlink`` of Semaphore,
+  Event, and AsyncResult now always remove the link object when
+  calling it, so ``unlink`` can sometimes be optimized out. See
+  :issue:`1487`.
+
 1.5a2 (2019-10-21)
 ==================
 

src/gevent/__abstract_linkable.pxd

@@ -39,7 +39,7 @@ cdef class AbstractLinkable(object):
    cdef readonly SwitchOutGreenletWithLoop hub
 
    cdef _notifier
-   cdef set _links
+   cdef list _links
    cdef bint _notify_all
 
    cpdef rawlink(self, callback)
@@ -47,6 +47,7 @@ cdef class AbstractLinkable(object):
    cpdef unlink(self, callback)
 
    cdef _check_and_notify(self)
+   @cython.nonecheck(False)
    cpdef _notify_links(self)
    cdef _wait_core(self, timeout, catch=*)
    cdef _wait_return_value(self, waited, wait_success)

src/gevent/_abstract_linkable.py

@@ -26,40 +26,44 @@ class AbstractLinkable(object):
     # Encapsulates the standard parts of the linking and notifying
     # protocol common to both repeatable events (Event, Semaphore) and
     # one-time events (AsyncResult).
-
-    __slots__ = ('hub', '_links', '_notifier', '_notify_all', '__weakref__')
+    #
+    # TODO: As of gevent 1.5, we use the same datastructures and almost
+    # the same algorithm as Greenlet. See about unifying them more.
+
+    __slots__ = (
+        'hub',
+        '_links',
+        '_notifier',
+        '_notify_all',
+        '__weakref__'
+    )
 
     def __init__(self):
         # Before this implementation, AsyncResult and Semaphore
         # maintained the order of notifications, but Event did not.
 
-        # In gevent 1.3, before Semaphore extended this class,
-        # that was changed to not maintain the order. It was done because
+        # In gevent 1.3, before Semaphore extended this class, that
+        # was changed to not maintain the order. It was done because
         # Event guaranteed to only call callbacks once (a set) but
-        # AsyncResult had no such guarantees.
-
-        # Semaphore likes to maintain order of callbacks, though,
-        # so when it was added we went back to a list implementation
-        # for storing callbacks. But we want to preserve the unique callback
-        # property, so we manually check.
-
-        # We generally don't expect to have so many waiters (for any of those
-        # objects) that testing membership and removing is a bottleneck.
-
-        # In PyPy 2.6.1 with Cython 0.23, `cdef public` or `cdef
-        # readonly` or simply `cdef` attributes of type `object` can appear to leak if
-        # a Python subclass is used (this is visible simply
-        # instantiating this subclass if _links=[]). Our _links and
-        # _notifier are such attributes, and gevent.thread subclasses
-        # this class. Thus, we carefully manage the lifetime of the
-        # objects we put in these attributes so that, in the normal
-        # case of a semaphore used correctly (deallocated when it's not
-        # locked and no one is waiting), the leak goes away (because
-        # these objects are back to None). This can also be solved on PyPy
-        # by simply not declaring these objects in the pxd file, but that doesn't work for
-        # CPython ("No attribute...")
-        # See https://github.com/gevent/gevent/issues/660
-        self._links = set()
+        # AsyncResult had no such guarantees. When Semaphore was
+        # changed to extend this class, it lost its ordering
+        # guarantees. Unfortunately, that made it unfair. There are
+        # rare cases that this can starve a greenlet
+        # (https://github.com/gevent/gevent/issues/1487) and maybe
+        # even lead to deadlock (not tested).
+
+        # So in gevent 1.5 we go back to maintaining order. But it's
+        # still important not to make duplicate calls, and it's also
+        # important to avoid O(n^2) behaviour that can result from
+        # naive use of a simple list due to the need to handle removed
+        # links in the _notify_links loop. Cython has special support for
+        # built-in sets, lists, and dicts, but not ordereddict. Rather than
+        # use two data structures, or a dict({link: order}), we simply use a
+        # list and remove objects as we go, keeping track of them so as not to
+        # have duplicates called. This makes `unlink` O(n), but we can avoid
+        # calling it in the common case in _wait_core (even so, the number of
+        # waiters should usually be pretty small)
+        self._links = []
         self._notifier = None
         # This is conceptually a class attribute, defined here for ease of access in
         # cython. If it's true, when notifiers fire, all existing callbacks are called.
@@ -95,13 +99,15 @@ class AbstractLinkable(object):
         """
         if not callable(callback):
             raise TypeError('Expected callable: %r' % (callback, ))
-
-        self._links.add(callback)
+        self._links.append(callback)
         self._check_and_notify()
 
     def unlink(self, callback):
         """Remove the callback set by :meth:`rawlink`"""
-        self._links.discard(callback)
+        try:
+            self._links.remove(callback)
+        except ValueError:
+            pass
 
         if not self._links and self._notifier is not None:
             # If we currently have one queued, de-queue it.
@@ -110,37 +116,42 @@ 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).
         notifier = self._notifier
-        # 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 will want to
-        # notify everyone that the status was once true, even though not it
-        # may not be anymore.
-        todo = set(self._links)
+        # Early links are allowed to remove later links, and links
+        # are allowed to add more links.
+        #
+        # 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
+        # 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:
-            for link in todo:
-                if not self._notify_all and not self.ready():
+            while self._links: # remember this can be mutated
+                if only_while_ready and not self.ready():
                     break
 
-                if link not in self._links:
-                    # Been removed already by some previous link. OK, fine.
+                link = self._links.pop(0) # Cython optimizes using list internals
+                id_link = id(link)
+                if id_link in done:
                     continue
+                done.add(id_link)
                 try:
                     link(self)
                 except: # pylint:disable=bare-except
-                    # We're running in the hub, so getcurrent() returns
-                    # a hub.
-                    self.hub.handle_error((link, self), *sys.exc_info()) # pylint:disable=undefined-variable
-                finally:
-                    if getattr(link, 'auto_unlink', None):
-                        # This attribute can avoid having to keep a reference to the function
-                        # *in* the function, which is a cycle
-                        self.unlink(link)
+                    # 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
         finally:
             # We should not have created a new notifier even if callbacks
             # released us because we loop through *all* of our links on the
@@ -148,10 +159,10 @@ class AbstractLinkable(object):
             assert self._notifier is notifier
             self._notifier = None
 
-        # Our set of active links changed, and we were told to stop on the first
-        # time we went unready. See if we're ready, and if so, go around
-        # again.
-        if not self._notify_all and todo != self._links:
+        # Now we may be ready or not ready. If we're ready, which
+        # could have happened during the last link we called, then we
+        # must have more links than we started with. We need to schedule the
+        # wakeup.
         self._check_and_notify()
 
     def _wait_core(self, timeout, catch=Timeout):
@@ -161,7 +172,6 @@ class AbstractLinkable(object):
         # Returns a true value if the wait succeeded without timing out.
         switch = getcurrent().switch # pylint:disable=undefined-variable
         self.rawlink(switch)
-        try:
         with Timeout._start_new_or_dummy(timeout) as timer:
             try:
                 if self.hub is None:
@@ -169,15 +179,18 @@ class AbstractLinkable(object):
                 result = self.hub.switch()
                 if result is not self: # pragma: no cover
                     raise InvalidSwitchError('Invalid switch into Event.wait(): %r' % (result, ))
+                # If we got here, we were automatically unlinked already.
                 return True
             except catch as ex:
+                self.unlink(switch)
                 if ex is not timer:
                     raise
                 # test_set_and_clear and test_timeout in test_threading
                 # rely on the exact return values, not just truthish-ness
                 return False
-        finally:
+            except:
                 self.unlink(switch)
+                raise
 
     def _wait_return_value(self, waited, wait_success):
         # pylint:disable=unused-argument

src/gevent/_ffi/callback.py

@@ -5,19 +5,13 @@ __all__ = [
 ]
 
 
-# For times when *args is captured but often not passed (empty),
-# we can avoid keeping the new tuple that was created for *args
-# around by using a constant.
-_NOARGS = ()
-
-
 class callback(object):
 
     __slots__ = ('callback', 'args')
 
     def __init__(self, cb, args):
         self.callback = cb
-        self.args = args or _NOARGS
+        self.args = args
 
     def stop(self):
         self.callback = None

src/gevent/_semaphore.py

@@ -36,6 +36,12 @@ class Semaphore(AbstractLinkable): # pylint:disable=undefined-variable
 
         The order in which waiters are awakened is not specified. It was not
         specified previously, but usually went in FIFO order.
+    .. versionchanged:: 1.5a3
+       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.
     """
 
     def __init__(self, value=1):

src/gevent/baseserver.py

@@ -166,6 +166,7 @@ class BaseServer(object):
             raise TypeError("'handle' must be provided")
 
     def _start_accepting_if_started(self, _event=None):
+        print("Begin accepting. Already started?", self.started)
         if self.started:
             self.start_accepting()
 
@@ -209,6 +210,8 @@ class BaseServer(object):
         for _ in xrange(self.max_accept):
             if self.full():
                 self.stop_accepting()
+                if self.pool is not None:
+                    self.pool._semaphore.rawlink(self._start_accepting_if_started)
                 return
             try:
                 args = self.do_read()

src/gevent/event.py

@@ -41,6 +41,12 @@ class Event(AbstractLinkable): # pylint:disable=undefined-variable
         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.
+
+    .. versionchanged:: 1.5a3
+       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.
     """
 
     __slots__ = ('_flag',)
@@ -181,6 +187,11 @@ class AsyncResult(AbstractLinkable): # pylint:disable=undefined-variable
     .. versionchanged:: 1.1
        Callbacks :meth:`linked <rawlink>` to this object are required to be hashable, and duplicates are
        merged.
+    .. versionchanged:: 1.5a3
+       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.
     """
 
     __slots__ = ('_value', '_exc_info', '_imap_task_index')

src/gevent/tests/test__semaphore.py

-import gevent.testing as greentest
+from __future__ import print_function
+from __future__ import absolute_import
+
+import weakref
+
 import gevent
+import gevent.exceptions
 from gevent.lock import Semaphore
 from gevent.thread import allocate_lock
-import weakref
+
+import gevent.testing as greentest
+
 try:
     from _thread import allocate_lock as std_allocate_lock
 except ImportError: # Py2
@@ -34,6 +41,7 @@ class TestSemaphore(greentest.TestCase):
         r = weakref.ref(s)
         self.assertEqual(s, r())
 
+    @greentest.ignores_leakcheck
     def test_semaphore_in_class_with_del(self):
         # Issue #704. This used to crash the process
         # under PyPy through at least 4.0.1 if the Semaphore
@@ -50,7 +58,6 @@ class TestSemaphore(greentest.TestCase):
         gc.collect()
         gc.collect()
 
-    test_semaphore_in_class_with_del.ignore_leakcheck = True
 
     def test_rawlink_on_unacquired_runs_notifiers(self):
         # https://github.com/gevent/gevent/issues/1287
@@ -87,5 +94,81 @@ class TestCExt(greentest.TestCase):
                          'gevent.__semaphore')
 
 
+class SwitchWithFixedHash(object):
+    # Replaces greenlet.switch with a callable object
+    # with a hash code we control.
+
+    def __init__(self, greenlet, hashcode):
+        self.switch = greenlet.switch
+        self.hashcode = hashcode
+
+    def __hash__(self):
+        return self.hashcode
+
+    def __eq__(self, other):
+        return self is other
+
+    def __call__(self, *args, **kwargs):
+        return self.switch(*args, **kwargs)
+
+    def __repr__(self):
+        return repr(self.switch)
+
+class FirstG(gevent.Greenlet):
+    # A greenlet whose switch method will have a low hashcode.
+
+    hashcode = 10
+
+    def __init__(self, *args, **kwargs):
+        gevent.Greenlet.__init__(self, *args, **kwargs)
+        self.switch = SwitchWithFixedHash(self, self.hashcode)
+
+
+class LastG(FirstG):
+    # A greenlet whose switch method will have a high hashcode.
+    hashcode = 12
+
+
+def acquire_then_exit(sem, should_quit):
+    sem.acquire()
+    should_quit.append(True)
+
+
+def acquire_then_spawn(sem, should_quit):
+    if should_quit:
+        return
+    sem.acquire()
+    g = FirstG.spawn(release_then_spawn, sem, should_quit)
+    g.join()
+
+def release_then_spawn(sem, should_quit):
+    sem.release()
+    if should_quit:
+        return
+    g = FirstG.spawn(acquire_then_spawn, sem, should_quit)
+    g.join()
+
+class TestSemaphoreFair(greentest.TestCase):
+
+    @greentest.ignores_leakcheck
+    def test_fair_or_hangs(self):
+        # If the lock isn't fair, this hangs, spinning between
+        # the last two greenlets.
+        # See https://github.com/gevent/gevent/issues/1487
+        sem = Semaphore()
+        should_quit = []
+
+        keep_going1 = FirstG.spawn(acquire_then_spawn, sem, should_quit)
+        keep_going2 = FirstG.spawn(acquire_then_spawn, sem, should_quit)
+        exiting = LastG.spawn(acquire_then_exit, sem, should_quit)
+
+        with self.assertRaises(gevent.exceptions.LoopExit):
+            gevent.joinall([keep_going1, keep_going2, exiting])
+
+        self.assertTrue(exiting.dead, exiting)
+        self.assertTrue(keep_going2.dead, keep_going2)
+        self.assertFalse(keep_going1.dead, keep_going1)
+
+
 if __name__ == '__main__':
     greentest.main()

src/gevent/threadpool.py

@@ -449,13 +449,11 @@ else:
                 fn(future)
             except Exception: # pylint: disable=broad-except
                 future.hub.print_exception((fn, future), *sys.exc_info())
-        cbwrap.auto_unlink = True
         return cbwrap
 
     def _wrap(future, fn):
         def f(_):
             fn(future)
-        f.auto_unlink = True
         return f
 
     class _FutureProxy(object):
@@ -490,8 +488,6 @@ else:
                 else:
                     w.add_exception(self)
 
-        __when_done.auto_unlink = True
-
         @property
         def _state(self):
             if self.done():

src/gevent/util.py

@@ -427,7 +427,12 @@ class GreenletTree(object):
             tree.child_multidata(pprint.pformat(tree_locals))
 
         self.__render_locals(tree)
+        try:
             self.__render_children(tree)
+        except RuntimeError:
+            # If the tree is exceptionally deep, we can hit the recursion error.
+            # Usually it's several levels down so we can make a print call.
+            print("When rendering children", *sys.exc_info())
         return tree.lines
 
     def __render_children(self, tree):