From 46234611411f75294acce37e8f3a56b409786c40 Mon Sep 17 00:00:00 2001 From: Ryan Williams Date: Sat, 20 Feb 2010 14:29:02 -0500 Subject: [PATCH] Documented the hell out of semaphore, noticed that BoundedSemaphore had the wrong semantics so created a new BoundedSemaphore, renaming the old to CappedSemaphore, to correct for that. --- doc/modules/semaphore.rst | 9 +- eventlet/semaphore.py | 136 ++++++++++++++++++++++-- tests/semaphore_test.py | 4 +- tests/stdlib/test_thread__boundedsem.py | 2 +- 4 files changed, 136 insertions(+), 15 deletions(-) diff --git a/doc/modules/semaphore.rst b/doc/modules/semaphore.rst index 7d92ee4..7146571 100644 --- a/doc/modules/semaphore.rst +++ b/doc/modules/semaphore.rst @@ -1,6 +1,11 @@ :mod:`semaphore` -- Semaphore classes ================================================== -.. automodule:: eventlet.semaphore +.. autoclass:: eventlet.semaphore.Semaphore :members: - :undoc-members: + +.. autoclass:: eventlet.semaphore.BoundedSemaphore + :members: + +.. autoclass:: eventlet.semaphore.CappedSemaphore + :members: \ No newline at end of file diff --git a/eventlet/semaphore.py b/eventlet/semaphore.py index 3072a97..b9356a8 100644 --- a/eventlet/semaphore.py +++ b/eventlet/semaphore.py @@ -5,12 +5,25 @@ 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 coroutine until *count* becomes + *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. """ - def __init__(self, count=0): - self.counter = count + 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): @@ -22,13 +35,31 @@ class Semaphore(object): 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): - # for consistency with BoundedSemaphore + """ Returns False; for consistency with :class:`~eventlet.semaphore.CappedSemaphore`.""" return False 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 if self.counter <= 0: @@ -45,7 +76,12 @@ class Semaphore(object): self.acquire() def release(self, blocking=True): - # `blocking' parameter is for consistency with BoundedSemaphore and is ignored + """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) @@ -61,22 +97,69 @@ class Semaphore(object): @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(object): - """A bounded semaphore. +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 coroutine until count becomes nonzero + *count* is zero suspends the calling greenthread until count becomes nonzero again. Attempting to :meth:`release` after *count* has reached *limit* - suspends the calling coroutine until *count* becomes less than *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'") @@ -92,12 +175,31 @@ class BoundedSemaphore(object): 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): + 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() @@ -114,6 +216,12 @@ class BoundedSemaphore(object): 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() @@ -128,4 +236,12 @@ class BoundedSemaphore(object): @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 \ No newline at end of file diff --git a/tests/semaphore_test.py b/tests/semaphore_test.py index 131c0be..64153ed 100644 --- a/tests/semaphore_test.py +++ b/tests/semaphore_test.py @@ -5,7 +5,7 @@ from tests import LimitedTestCase class TestSemaphore(LimitedTestCase): def test_bounded(self): - sem = semaphore.BoundedSemaphore(2, limit=3) + sem = semaphore.CappedSemaphore(2, limit=3) self.assertEqual(sem.acquire(), True) self.assertEqual(sem.acquire(), True) gt1 = eventlet.spawn(sem.release) @@ -21,7 +21,7 @@ class TestSemaphore(LimitedTestCase): gt2.wait() def test_bounded_with_zero_limit(self): - sem = semaphore.BoundedSemaphore(0, 0) + sem = semaphore.CappedSemaphore(0, 0) gt = eventlet.spawn(sem.acquire) sem.release() gt.wait() diff --git a/tests/stdlib/test_thread__boundedsem.py b/tests/stdlib/test_thread__boundedsem.py index c530c61..5408e68 100644 --- a/tests/stdlib/test_thread__boundedsem.py +++ b/tests/stdlib/test_thread__boundedsem.py @@ -8,7 +8,7 @@ def allocate_lock(): original_allocate_lock = thread.allocate_lock thread.allocate_lock = allocate_lock original_LockType = thread.LockType -thread.LockType = coros.BoundedSemaphore +thread.LockType = coros.CappedSemaphore try: import os.path