Skip to content
Back to top

# Source code for gevent._semaphore

    # cython: auto_pickle=False,embedsignature=True,always_allow_keywords=False
    ###
    # This file is ``gevent._semaphore`` so that it can be compiled by Cython
    # individually. However, this is not the place to import from. Everyone,
    # gevent internal code included, must import from ``gevent.lock``.
    # The only exception are .pxd files which need access to the
    # C code; the PURE_PYTHON things that have to happen and which are
    # handled in ``gevent.lock``, do not apply to them.
    ###
    from __future__ import print_function, absolute_import, division

    __all__ = [
        'Semaphore',
        'BoundedSemaphore',
    ]

    from time import sleep as _native_sleep

    from gevent._compat import monotonic
    from gevent.exceptions import InvalidThreadUseError
    from gevent.exceptions import LoopExit
    from gevent.timeout import Timeout

    def _get_linkable():
        x = __import__('gevent._abstract_linkable')
        return x._abstract_linkable.AbstractLinkable
    locals()['AbstractLinkable'] = _get_linkable()
    del _get_linkable

    from gevent._hub_local import get_hub_if_exists
    from gevent._hub_local import get_hub
    from gevent.hub import spawn_raw

    class _LockReleaseLink(object):
        __slots__ = (
            'lock',
        )

        def __init__(self, lock):
            self.lock = lock

        def __call__(self, _):
            self.lock.release()

    _UNSET = object()
    _MULTI = object()

    class Semaphore(AbstractLinkable): # pylint:disable=undefined-variable
        """
        Semaphore(value=1) -> Semaphore

        .. seealso:: :class:`BoundedSemaphore` for a safer version that prevents
           some classes of bugs. If unsure, most users should opt for `BoundedSemaphore`.

        A semaphore manages a counter representing the number of `release`
        calls minus the number of `acquire` calls, plus an initial value.
        The `acquire` method blocks if necessary until it can return
        without making the counter negative. A semaphore does not track ownership
        by greenlets; any greenlet can call `release`, whether or not it has previously
        called `acquire`.

        If not given, ``value`` defaults to 1.

        The semaphore is a context manager and can be used in ``with`` statements.

        This Semaphore's ``__exit__`` method does not call the trace function
        on CPython, but does under PyPy.

        .. versionchanged:: 1.4.0
            Document that the order in which waiters are awakened is not specified. It was not
            specified previously, but due to CPython implementation quirks 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.
        .. versionchanged:: 20.12.0
           Improved support for multi-threaded usage. When multi-threaded usage is detected,
           instances will no longer create the thread's hub if it's not present.

        .. versionchanged:: 24.2.1
           Uses Python 3 native lock timeouts for cross-thread operations instead
           of spinning.
        """

        __slots__ = (
            'counter',
            # long integer, signed (Py2) or unsigned (Py3); see comments
            # in the .pxd file for why we store as Python object. Set to ``_UNSET``
            # initially. Set to the ident of the first thread that
            # acquires us. If we later see a different thread ident, set
            # to ``_MULTI``.
            '_multithreaded',
        )

        def __init__(self, value=1, hub=None):
            self.counter = value
            if self.counter < 0: # Do the check after Cython native int conversion
                raise ValueError("semaphore initial value must be >= 0")
            super(Semaphore, self).__init__(hub)
            self._notify_all = False
            self._multithreaded = _UNSET

        def __str__(self):
            return '<%s at 0x%x counter=%s _links[%s]>' % (
                self.__class__.__name__,
                id(self),
                self.counter,
                self.linkcount()
            )

        def locked(self):
            """
            Return a boolean indicating whether the semaphore can be
            acquired (`False` if the semaphore *can* be acquired). Most
            useful with binary semaphores (those with an initial value of 1).

            :rtype: bool
            """
            return self.counter <= 0

        def release(self):
            """
            Release the semaphore, notifying any waiters if needed. There
            is no return value.

            .. note::

                This can be used to over-release the semaphore.
                (Release more times than it has been acquired or was initially
                created with.)

                This is usually a sign of a bug, but under some circumstances it can be
                used deliberately, for example, to model the arrival of additional
                resources.

            :rtype: None
            """
            self.counter += 1
            self._check_and_notify()
            return self.counter

        def ready(self):
            """
            Return a boolean indicating whether the semaphore can be
            acquired (`True` if the semaphore can be acquired).

            :rtype: bool
            """
            return self.counter > 0

        def _start_notify(self):
            self._check_and_notify()

        def _wait_return_value(self, waited, wait_success):
            if waited:
                return wait_success
            # We didn't even wait, we must be good to go.
            # XXX: This is probably dead code, we're careful not to go into the wait
            # state if we don't expect to need to
            return True

        def wait(self, timeout=None):
            """
            Wait until it is possible to acquire this semaphore, or until the optional
            *timeout* elapses.

            .. note:: If this semaphore was initialized with a *value* of 0,
               this method will block forever if no timeout is given.

            :keyword float timeout: If given, specifies the maximum amount of seconds
               this method will block.
            :return: A number indicating how many times the semaphore can be acquired
                before blocking. *This could be 0,* if other waiters acquired
                the semaphore.
            :rtype: int
            """
            if self.counter > 0:
                return self.counter

            self._wait(timeout) # return value irrelevant, whether we got it or got a timeout
            return self.counter

        def acquire(self, blocking=True, timeout=None):
            """
            acquire(blocking=True, timeout=None) -> bool

            Acquire the semaphore.

            .. note:: If this semaphore was initialized with a *value* of 0,
               this method will block forever (unless a timeout is given or blocking is
               set to false).

            :keyword bool blocking: If True (the default), this function will block
               until the semaphore is acquired.
            :keyword float timeout: If given, and *blocking* is true,
               specifies the maximum amount of seconds
               this method will block.
            :return: A `bool` indicating whether the semaphore was acquired.
               If ``blocking`` is True and ``timeout`` is None (the default), then
               (so long as this semaphore was initialized with a size greater than 0)
               this will always return True. If a timeout was given, and it expired before
               the semaphore was acquired, False will be returned. (Note that this can still
               raise a ``Timeout`` exception, if some other caller had already started a timer.)
            """
            # pylint:disable=too-many-return-statements,too-many-branches
            # Sadly, the body of this method is rather complicated.
            if self._multithreaded is _UNSET:
                self._multithreaded = self._get_thread_ident()
            elif self._multithreaded != self._get_thread_ident():
                self._multithreaded = _MULTI

            # We conceptually now belong to the hub of the thread that
            # called this, whether or not we have to block. Note that we
            # cannot force it to be created yet, because Semaphore is used
            # by importlib.ModuleLock which is used when importing the hub
            # itself! This also checks for cross-thread issues.
            invalid_thread_use = None
            try:
                self._capture_hub(False)
            except InvalidThreadUseError as e:
                # My hub belongs to some other thread. We didn't release the GIL/object lock
                # by raising the exception, so we know this is still true.
                invalid_thread_use = e.args
                e = None
                if not self.counter and blocking:
                    # We would need to block. So coordinate with the main hub.
                    return self.__acquire_from_other_thread(invalid_thread_use, blocking, timeout)

            if self.counter > 0:
                self.counter -= 1
                return True

            if not blocking:
                return False

            if self._multithreaded is not _MULTI and self.hub is None: # pylint:disable=access-member-before-definition
                self.hub = get_hub() # pylint:disable=attribute-defined-outside-init

            if self.hub is None and not invalid_thread_use:
                # Someone else is holding us. There's not a hub here,
                # nor is there a hub in that thread. We'll need to use regular locks.
                # This will be unfair to yet a third thread that tries to use us with greenlets.
                return self.__acquire_from_other_thread(
                    (None, None, self._getcurrent(), "NoHubs"),
                    blocking,
                    timeout
                )

            # self._wait may drop both the GIL and the _lock_lock.
            # By the time we regain control, both have been reacquired.
            try:
                success = self._wait(timeout)
            except LoopExit as ex:
                args = ex.args
                ex = None
                if self.counter:
                    success = True
                else:
                    # Avoid using ex.hub property to keep holding the GIL
                    if len(args) == 3 and args[1].main_hub:
                        # The main hub, meaning the main thread. We probably can do nothing with this.
                        raise
                    return self.__acquire_from_other_thread(
                        (self.hub, get_hub_if_exists(), self._getcurrent(), "LoopExit"),
                        blocking,
                        timeout)

            if not success:
                assert timeout is not None
                # Our timer expired.
                return False

            # Neither our timer or another one expired, so we blocked until
            # awoke. Therefore, the counter is ours
            assert self.counter > 0, (self.counter, blocking, timeout, success,)
            self.counter -= 1
            return True

        _py3k_acquire = acquire # PyPy needs this; it must be static for Cython

        def __enter__(self):
            self.acquire()

        def __exit__(self, t, v, tb):
            self.release()

        def _handle_unswitched_notifications(self, unswitched):
            # If we fail to switch to a greenlet in another thread to send
            # a notification, just re-queue it, in the hopes that the
            # other thread will eventually run notifications itself.
            #
            # We CANNOT do what the ``super()`` does and actually allow
            # this notification to get run sometime in the future by
            # scheduling a callback in the other thread. The algorithm
            # that we use to handle cross-thread locking/unlocking was
            # designed before the schedule-a-callback mechanism was
            # implemented. If we allow this to be run as a callback, we
            # can find ourself the victim of ``InvalidSwitchError`` (or
            # worse, silent corruption) because the switch can come at an
            # unexpected time: *after* the destination thread has already
            # acquired the lock.
            #
            # This manifests in a fairly reliable test failure,
            # ``gevent.tests.test__semaphore``
            # ``TestSemaphoreMultiThread.test_dueling_threads_with_hub``,
            # but ONLY when running in PURE_PYTHON mode.
            #
            # TODO: Maybe we can rewrite that part of the algorithm to be friendly to
            # running the callbacks?
            self._links.extend(unswitched)

        def __add_link(self, link):
            if not self._notifier:
                self.rawlink(link)
            else:
                self._notifier.args[0].append(link)

        def __acquire_from_other_thread(self, ex_args, blocking, timeout):
            assert blocking
            # Some other hub owns this object. We must ask it to wake us
            # up. In general, we can't use a Python-level ``Lock`` because
            #
            # (1) it doesn't support a timeout on all platforms; and
            # (2) we don't want to block this hub from running.
            #
            # So we need to do so in a way that cooperates with *two*
            # hubs. That's what an async watcher is built for.
            #
            # Of course, if we don't actually have two hubs, then we must find some other
            # solution. That involves using a lock.

            # We have to take an action that drops the GIL and drops the object lock
            # to allow the main thread (the thread for our hub) to advance.
            owning_hub = ex_args[0]
            hub_for_this_thread = ex_args[1]
            current_greenlet = ex_args[2]

            if owning_hub is None and hub_for_this_thread is None:
                return self.__acquire_without_hubs(timeout)

            if hub_for_this_thread is None:
                # Probably a background worker thread. We don't want to create
                # the hub if not needed, and since it didn't exist there are no
                # other greenlets that we could yield to anyway, so there's nothing
                # to block and no reason to try to avoid blocking, so using a native
                # lock is the simplest way to go.
                return self.__acquire_using_other_hub(owning_hub, timeout)

            # We have a hub we don't want to block. Use an async watcher
            # and ask the next releaser of this object to wake us up.
            return self.__acquire_using_two_hubs(hub_for_this_thread,
                                                 current_greenlet,
                                                 timeout)

        def __acquire_using_two_hubs(self,
                                     hub_for_this_thread,
                                     current_greenlet,
                                     timeout):
            # Allocating and starting the watcher *could* release the GIL.
            # with the libev corcext, allocating won't, but starting briefly will.
            # With other backends, allocating might, and starting might also.
            # So...
            watcher = hub_for_this_thread.loop.async_()
            send = watcher.send_ignoring_arg
            watcher.start(current_greenlet.switch, self)
            try:
                with Timeout._start_new_or_dummy(timeout) as timer:
                    # ... now that we're back holding the GIL, we need to verify our
                    # state.
                    try:
                        while 1:
                            if self.counter > 0:
                                self.counter -= 1
                                assert self.counter >= 0, (self,)
                                return True

                            self.__add_link(send)

                            # Releases the object lock
                            self._switch_to_hub(hub_for_this_thread)
                            # We waited and got notified. We should be ready now, so a non-blocking
                            # acquire() should succeed. But sometimes we get spurious notifications?
                            # It's not entirely clear how. So we need to loop until we get it, or until
                            # the timer expires
                            result = self.acquire(0)
                            if result:
                                return result
                    except Timeout as tex:
                        if tex is not timer:
                            raise
                        return False
            finally:
                self._quiet_unlink_all(send)
                watcher.stop()
                watcher.close()

        def __acquire_from_other_thread_cb(self, results, blocking, timeout, thread_lock):
            try:
                result = self.acquire(blocking, timeout)
                results.append(result)
            finally:
                thread_lock.release()
            return result

        def __acquire_using_other_hub(self, owning_hub, timeout):
            assert owning_hub is not get_hub_if_exists()
            thread_lock = self._allocate_lock()
            thread_lock.acquire()
            results = []

            owning_hub.loop.run_callback_threadsafe(
                spawn_raw,
                self.__acquire_from_other_thread_cb,
                results,
                1,       # blocking,
                timeout, # timeout,
                thread_lock)

            # We MUST use a blocking acquire here, or at least be sure we keep going
            # until we acquire it. If we timed out waiting here,
            # just before the callback runs, then we would be out of sync.
            self.__spin_on_native_lock(thread_lock, None)
            return results[0]

        def __acquire_without_hubs(self, timeout):
            thread_lock = self._allocate_lock()
            thread_lock.acquire()
            absolute_expiration = 0
            begin = 0
            if timeout:
                absolute_expiration = monotonic() + timeout

            # Cython won't compile a lambda here
            link = _LockReleaseLink(thread_lock)
            while 1:
                self.__add_link(link)
                if absolute_expiration:
                    begin = monotonic()

                got_native = self.__spin_on_native_lock(thread_lock, timeout)
                self._quiet_unlink_all(link)
                if got_native:
                    if self.acquire(0):
                        return True
                if absolute_expiration:
                    now = monotonic()
                    if now >= absolute_expiration:
                        return False
                    duration = now - begin
                    timeout -= duration
                    if timeout <= 0:
                        return False

        def __spin_on_native_lock(self, thread_lock, timeout):
            self._drop_lock_for_switch_out()
            try:
                # Unlike Python 2, Python 3 thread locks
                # can be interrupted when blocking, with or
                # without a timeout. Python 2 didn't even
                # support a timeout for non -blocking.
                if timeout:
                    return thread_lock.acquire(True, timeout)

                return thread_lock.acquire()
            finally:
                self._acquire_lock_for_switch_in()

    class BoundedSemaphore(Semaphore):
        """
        BoundedSemaphore(value=1) -> BoundedSemaphore

        A bounded semaphore checks to make sure its current value doesn't
        exceed its initial value. If it does, :class:`ValueError` is
        raised. In most situations semaphores are used to guard resources
        with limited capacity. If the semaphore is released too many times
        it's a sign of a bug.

        If not given, *value* defaults to 1.
        """

        __slots__ = (
            '_initial_value',
        )

        #: For monkey-patching, allow changing the class of error we raise
        _OVER_RELEASE_ERROR = ValueError

        def __init__(self, *args, **kwargs):
            Semaphore.__init__(self, *args, **kwargs)
            self._initial_value = self.counter

        def release(self):
            """
            Like :meth:`Semaphore.release`, but raises :class:`ValueError`
            if the semaphore is being over-released.
            """
            if self.counter >= self._initial_value:
                raise self._OVER_RELEASE_ERROR("Semaphore released too many times")
            counter = Semaphore.release(self)
            # When we are absolutely certain that no one holds this semaphore,
            # release our hub and go back to floating. This assists in cross-thread
            # uses.
            if counter == self._initial_value:
                self.hub = None # pylint:disable=attribute-defined-outside-init
            return counter

        def _at_fork_reinit(self):
            super(BoundedSemaphore, self)._at_fork_reinit()
            self.counter = self._initial_value

    # By building the semaphore with Cython under PyPy, we get
    # atomic operations (specifically, exiting/releasing), at the
    # cost of some speed (one trivial semaphore micro-benchmark put the pure-python version
    # at around 1s and the compiled version at around 4s). Some clever subclassing
    # and having only the bare minimum be in cython might help reduce that penalty.
    # NOTE: You must use version 0.23.4 or later to avoid a memory leak.
    # https://mail.python.org/pipermail/cython-devel/2015-October/004571.html
    # However, that's all for naught on up to and including PyPy 4.0.1 which
    # have some serious crashing bugs with GC interacting with cython.
    # It hasn't been tested since then, and PURE_PYTHON is assumed to be true
    # for PyPy in all cases anyway, so this does nothing.

    from gevent._util import import_c_accel
    import_c_accel(globals(), 'gevent.__semaphore')