+++ /dev/null
-from __future__ import with_statement
-from eventlet import greenthread
-from eventlet import hubs
-from eventlet.timeout import Timeout
-
-
-class Semaphore(object):
-
- """An unbounded semaphore.
- Optionally initialize with a resource *count*, then :meth:`acquire` and
- :meth:`release` resources as needed. Attempting to :meth:`acquire` when
- *count* is zero suspends the calling greenthread until *count* becomes
- nonzero again.
-
- This is API-compatible with :class:`threading.Semaphore`.
-
- It is a context manager, and thus can be used in a with block::
-
- sem = Semaphore(2)
- with sem:
- do_some_stuff()
-
- If not specified, *value* defaults to 1.
-
- It is possible to limit acquire time::
-
- sem = Semaphore()
- ok = sem.acquire(timeout=0.1)
- # True if acquired, False if timed out.
-
- """
-
- def __init__(self, value=1):
- self.counter = value
- if value < 0:
- raise ValueError("Semaphore must be initialized with a positive "
- "number, got %s" % value)
- self._waiters = set()
-
- def __repr__(self):
- params = (self.__class__.__name__, hex(id(self)),
- self.counter, len(self._waiters))
- return '<%s at %s c=%s _w[%s]>' % params
-
- def __str__(self):
- params = (self.__class__.__name__, self.counter, len(self._waiters))
- return '<%s c=%s _w[%s]>' % params
-
- def locked(self):
- """Returns true if a call to acquire would block.
- """
- return self.counter <= 0
-
- def bounded(self):
- """Returns False; for consistency with
- :class:`~eventlet.semaphore.CappedSemaphore`.
- """
- return False
-
- def acquire(self, blocking=True, timeout=None):
- """Acquire a semaphore.
-
- When invoked without arguments: if the internal counter is larger than
- zero on entry, decrement it by one and return immediately. If it is zero
- on entry, block, waiting until some other thread has called release() to
- make it larger than zero. This is done with proper interlocking so that
- if multiple acquire() calls are blocked, release() will wake exactly one
- of them up. The implementation may pick one at random, so the order in
- which blocked threads are awakened should not be relied on. There is no
- return value in this case.
-
- When invoked with blocking set to true, do the same thing as when called
- without arguments, and return true.
-
- When invoked with blocking set to false, do not block. If a call without
- an argument would block, return false immediately; otherwise, do the
- same thing as when called without arguments, and return true.
- """
- if not blocking and timeout is not None:
- raise ValueError("can't specify timeout for non-blocking acquire")
- if not blocking and self.locked():
- return False
- if self.counter <= 0:
- self._waiters.add(greenthread.getcurrent())
- try:
- if timeout is not None:
- ok = False
- with Timeout(timeout, False):
- while self.counter <= 0:
- hubs.get_hub().switch()
- ok = True
- if not ok:
- return False
- else:
- while self.counter <= 0:
- hubs.get_hub().switch()
- finally:
- self._waiters.discard(greenthread.getcurrent())
- self.counter -= 1
- return True
-
- def __enter__(self):
- self.acquire()
-
- def release(self, blocking=True):
- """Release a semaphore, incrementing the internal counter by one. When
- it was zero on entry and another thread is waiting for it to become
- larger than zero again, wake up that thread.
-
- The *blocking* argument is for consistency with CappedSemaphore and is
- ignored
- """
- self.counter += 1
- if self._waiters:
- hubs.get_hub().schedule_call_global(0, self._do_acquire)
- return True
-
- def _do_acquire(self):
- if self._waiters and self.counter > 0:
- waiter = self._waiters.pop()
- waiter.switch()
-
- def __exit__(self, typ, val, tb):
- self.release()
-
- @property
- def balance(self):
- """An integer value that represents how many new calls to
- :meth:`acquire` or :meth:`release` would be needed to get the counter to
- 0. If it is positive, then its value is the number of acquires that can
- happen before the next acquire would block. If it is negative, it is
- the negative of the number of releases that would be required in order
- to make the counter 0 again (one more release would push the counter to
- 1 and unblock acquirers). It takes into account how many greenthreads
- are currently blocking in :meth:`acquire`.
- """
- # positive means there are free items
- # zero means there are no free items but nobody has requested one
- # negative means there are requests for items, but no items
- return self.counter - len(self._waiters)
-
-
-class BoundedSemaphore(Semaphore):
-
- """A bounded semaphore checks to make sure its current value doesn't exceed
- its initial value. If it does, 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.
- """
-
- def __init__(self, value=1):
- super(BoundedSemaphore, self).__init__(value)
- self.original_counter = value
-
- def release(self, blocking=True):
- """Release a semaphore, incrementing the internal counter by one. If
- the counter would exceed the initial value, raises ValueError. When
- it was zero on entry and another thread is waiting for it to become
- larger than zero again, wake up that thread.
-
- The *blocking* argument is for consistency with :class:`CappedSemaphore`
- and is ignored
- """
- if self.counter >= self.original_counter:
- raise ValueError("Semaphore released too many times")
- return super(BoundedSemaphore, self).release(blocking)
-
-
-class CappedSemaphore(object):
-
- """A blockingly bounded semaphore.
-
- Optionally initialize with a resource *count*, then :meth:`acquire` and
- :meth:`release` resources as needed. Attempting to :meth:`acquire` when
- *count* is zero suspends the calling greenthread until count becomes nonzero
- again. Attempting to :meth:`release` after *count* has reached *limit*
- suspends the calling greenthread until *count* becomes less than *limit*
- again.
-
- This has the same API as :class:`threading.Semaphore`, though its
- semantics and behavior differ subtly due to the upper limit on calls
- to :meth:`release`. It is **not** compatible with
- :class:`threading.BoundedSemaphore` because it blocks when reaching *limit*
- instead of raising a ValueError.
-
- It is a context manager, and thus can be used in a with block::
-
- sem = CappedSemaphore(2)
- with sem:
- do_some_stuff()
- """
-
- def __init__(self, count, limit):
- if count < 0:
- raise ValueError("CappedSemaphore must be initialized with a "
- "positive number, got %s" % count)
- if count > limit:
- # accidentally, this also catches the case when limit is None
- raise ValueError("'count' cannot be more than 'limit'")
- self.lower_bound = Semaphore(count)
- self.upper_bound = Semaphore(limit - count)
-
- def __repr__(self):
- params = (self.__class__.__name__, hex(id(self)),
- self.balance, self.lower_bound, self.upper_bound)
- return '<%s at %s b=%s l=%s u=%s>' % params
-
- def __str__(self):
- params = (self.__class__.__name__, self.balance,
- self.lower_bound, self.upper_bound)
- return '<%s b=%s l=%s u=%s>' % params
-
- def locked(self):
- """Returns true if a call to acquire would block.
- """
- return self.lower_bound.locked()
-
- def bounded(self):
- """Returns true if a call to release would block.
- """
- return self.upper_bound.locked()
-
- def acquire(self, blocking=True):
- """Acquire a semaphore.
-
- When invoked without arguments: if the internal counter is larger than
- zero on entry, decrement it by one and return immediately. If it is zero
- on entry, block, waiting until some other thread has called release() to
- make it larger than zero. This is done with proper interlocking so that
- if multiple acquire() calls are blocked, release() will wake exactly one
- of them up. The implementation may pick one at random, so the order in
- which blocked threads are awakened should not be relied on. There is no
- return value in this case.
-
- When invoked with blocking set to true, do the same thing as when called
- without arguments, and return true.
-
- When invoked with blocking set to false, do not block. If a call without
- an argument would block, return false immediately; otherwise, do the
- same thing as when called without arguments, and return true.
- """
- if not blocking and self.locked():
- return False
- self.upper_bound.release()
- try:
- return self.lower_bound.acquire()
- except:
- self.upper_bound.counter -= 1
- # using counter directly means that it can be less than zero.
- # however I certainly don't need to wait here and I don't seem to have
- # a need to care about such inconsistency
- raise
-
- def __enter__(self):
- self.acquire()
-
- def release(self, blocking=True):
- """Release a semaphore. In this class, this behaves very much like
- an :meth:`acquire` but in the opposite direction.
-
- Imagine the docs of :meth:`acquire` here, but with every direction
- reversed. When calling this method, it will block if the internal
- counter is greater than or equal to *limit*.
- """
- if not blocking and self.bounded():
- return False
- self.lower_bound.release()
- try:
- return self.upper_bound.acquire()
- except:
- self.lower_bound.counter -= 1
- raise
-
- def __exit__(self, typ, val, tb):
- self.release()
-
- @property
- def balance(self):
- """An integer value that represents how many new calls to
- :meth:`acquire` or :meth:`release` would be needed to get the counter to
- 0. If it is positive, then its value is the number of acquires that can
- happen before the next acquire would block. If it is negative, it is
- the negative of the number of releases that would be required in order
- to make the counter 0 again (one more release would push the counter to
- 1 and unblock acquirers). It takes into account how many greenthreads
- are currently blocking in :meth:`acquire` and :meth:`release`.
- """
- return self.lower_bound.balance - self.upper_bound.balance
Code Review / packages / trusty / python-eventlet.git / blobdiff