426 lines
13 KiB
Python
426 lines
13 KiB
Python
|
###
|
||
|
|
# This file is test__semaphore.py only for organization purposes.
|
||
|
|
# The public API,
|
||
|
|
# and the *only* correct place to import Semaphore --- even in tests ---
|
||
|
|
# is ``gevent.lock``, never ``gevent._semaphore``.
|
||
|
|
##
|
||
|
|
from __future__ import print_function
|
||
|
|
from __future__ import absolute_import
|
||
|
|
|
||
|
|
import weakref
|
||
|
|
|
||
|
|
import gevent
|
||
|
|
import gevent.exceptions
|
||
|
|
from gevent.lock import Semaphore
|
||
|
|
from gevent.lock import BoundedSemaphore
|
||
|
|
|
||
|
|
import gevent.testing as greentest
|
||
|
|
from gevent.testing import timing
|
||
|
|
|
||
|
|
class TestSemaphore(greentest.TestCase):
|
||
|
|
|
||
|
|
# issue 39
|
||
|
|
def test_acquire_returns_false_after_timeout(self):
|
||
|
|
s = Semaphore(value=0)
|
||
|
|
result = s.acquire(timeout=0.01)
|
||
|
|
assert result is False, repr(result)
|
||
|
|
|
||
|
|
def test_release_twice(self):
|
||
|
|
s = Semaphore()
|
||
|
|
result = []
|
||
|
|
s.rawlink(lambda s: result.append('a'))
|
||
|
|
s.release()
|
||
|
|
s.rawlink(lambda s: result.append('b'))
|
||
|
|
s.release()
|
||
|
|
gevent.sleep(0.001)
|
||
|
|
# The order, though, is not guaranteed.
|
||
|
|
self.assertEqual(sorted(result), ['a', 'b'])
|
||
|
|
|
||
|
|
def test_semaphore_weakref(self):
|
||
|
|
s = Semaphore()
|
||
|
|
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
|
||
|
|
# was implemented with Cython.
|
||
|
|
class X(object):
|
||
|
|
def __init__(self):
|
||
|
|
self.s = Semaphore()
|
||
|
|
|
||
|
|
def __del__(self):
|
||
|
|
self.s.acquire()
|
||
|
|
|
||
|
|
X()
|
||
|
|
import gc
|
||
|
|
gc.collect()
|
||
|
|
gc.collect()
|
||
|
|
|
||
|
|
|
||
|
|
def test_rawlink_on_unacquired_runs_notifiers(self):
|
||
|
|
# https://github.com/gevent/gevent/issues/1287
|
||
|
|
|
||
|
|
# Rawlinking a ready semaphore should fire immediately,
|
||
|
|
# not raise LoopExit
|
||
|
|
s = Semaphore()
|
||
|
|
gevent.wait([s])
|
||
|
|
|
||
|
|
|
||
|
|
class TestSemaphoreMultiThread(greentest.TestCase):
|
||
|
|
# Tests that the object can be acquired correctly across
|
||
|
|
# multiple threads.
|
||
|
|
# Used as a base class.
|
||
|
|
|
||
|
|
# See https://github.com/gevent/gevent/issues/1437
|
||
|
|
|
||
|
|
def _getTargetClass(self):
|
||
|
|
return Semaphore
|
||
|
|
|
||
|
|
def _makeOne(self):
|
||
|
|
# Create an object that is associated with the current hub. If
|
||
|
|
# we don't do this now, it gets initialized lazily the first
|
||
|
|
# time it would have to block, which, in the event of threads,
|
||
|
|
# would be from an arbitrary thread.
|
||
|
|
return self._getTargetClass()(1)
|
||
|
|
|
||
|
|
def _makeThreadMain(self, thread_running, thread_acquired, sem,
|
||
|
|
acquired, exc_info,
|
||
|
|
**thread_acquire_kwargs):
|
||
|
|
from gevent._hub_local import get_hub_if_exists
|
||
|
|
import sys
|
||
|
|
|
||
|
|
def thread_main():
|
||
|
|
thread_running.set()
|
||
|
|
try:
|
||
|
|
acquired.append(
|
||
|
|
sem.acquire(**thread_acquire_kwargs)
|
||
|
|
)
|
||
|
|
except:
|
||
|
|
exc_info[:] = sys.exc_info()
|
||
|
|
raise # Print
|
||
|
|
finally:
|
||
|
|
hub = get_hub_if_exists()
|
||
|
|
if hub is not None:
|
||
|
|
hub.join()
|
||
|
|
hub.destroy(destroy_loop=True)
|
||
|
|
thread_acquired.set()
|
||
|
|
return thread_main
|
||
|
|
|
||
|
|
IDLE_ITERATIONS = 5
|
||
|
|
|
||
|
|
def _do_test_acquire_in_one_then_another(self,
|
||
|
|
release=True,
|
||
|
|
require_thread_acquired_to_finish=False,
|
||
|
|
**thread_acquire_kwargs):
|
||
|
|
from gevent import monkey
|
||
|
|
self.assertFalse(monkey.is_module_patched('threading'))
|
||
|
|
|
||
|
|
import threading
|
||
|
|
thread_running = threading.Event()
|
||
|
|
thread_acquired = threading.Event()
|
||
|
|
|
||
|
|
sem = self._makeOne()
|
||
|
|
# Make future acquires block
|
||
|
|
sem.acquire()
|
||
|
|
|
||
|
|
exc_info = []
|
||
|
|
acquired = []
|
||
|
|
|
||
|
|
t = threading.Thread(target=self._makeThreadMain(
|
||
|
|
thread_running, thread_acquired, sem,
|
||
|
|
acquired, exc_info,
|
||
|
|
**thread_acquire_kwargs
|
||
|
|
))
|
||
|
|
t.daemon = True
|
||
|
|
t.start()
|
||
|
|
thread_running.wait(10) # implausibly large time
|
||
|
|
if release:
|
||
|
|
sem.release()
|
||
|
|
# Spin the loop to be sure the release gets through.
|
||
|
|
# (Release schedules the notifier to run, and when the
|
||
|
|
# notifier run it sends the async notification to the
|
||
|
|
# other thread. Depending on exactly where we are in the
|
||
|
|
# event loop, and the limit to the number of callbacks
|
||
|
|
# that get run (including time-based) the notifier may or
|
||
|
|
# may not be immediately ready to run, so this can take up
|
||
|
|
# to two iterations.)
|
||
|
|
for _ in range(self.IDLE_ITERATIONS):
|
||
|
|
gevent.idle()
|
||
|
|
if thread_acquired.wait(timing.LARGE_TICK):
|
||
|
|
break
|
||
|
|
|
||
|
|
self.assertEqual(acquired, [True])
|
||
|
|
|
||
|
|
if not release and thread_acquire_kwargs.get("timeout"):
|
||
|
|
# Spin the loop to be sure that the timeout has a chance to
|
||
|
|
# process. Interleave this with something that drops the GIL
|
||
|
|
# so the background thread has a chance to notice that.
|
||
|
|
for _ in range(self.IDLE_ITERATIONS):
|
||
|
|
gevent.idle()
|
||
|
|
if thread_acquired.wait(timing.LARGE_TICK):
|
||
|
|
break
|
||
|
|
thread_acquired.wait(timing.LARGE_TICK * 5)
|
||
|
|
|
||
|
|
if require_thread_acquired_to_finish:
|
||
|
|
self.assertTrue(thread_acquired.is_set())
|
||
|
|
try:
|
||
|
|
self.assertEqual(exc_info, [])
|
||
|
|
finally:
|
||
|
|
exc_info = None
|
||
|
|
|
||
|
|
return sem, acquired
|
||
|
|
|
||
|
|
def test_acquire_in_one_then_another(self):
|
||
|
|
self._do_test_acquire_in_one_then_another(release=True)
|
||
|
|
|
||
|
|
def test_acquire_in_one_then_another_timed(self):
|
||
|
|
sem, acquired_in_thread = self._do_test_acquire_in_one_then_another(
|
||
|
|
release=False,
|
||
|
|
require_thread_acquired_to_finish=True,
|
||
|
|
timeout=timing.SMALLEST_RELIABLE_DELAY)
|
||
|
|
self.assertEqual([False], acquired_in_thread)
|
||
|
|
# This doesn't, of course, notify anything, because
|
||
|
|
# the waiter has given up.
|
||
|
|
sem.release()
|
||
|
|
notifier = getattr(sem, '_notifier', None)
|
||
|
|
self.assertIsNone(notifier)
|
||
|
|
|
||
|
|
def test_acquire_in_one_wait_greenlet_wait_thread_gives_up(self):
|
||
|
|
# The waiter in the thread both arrives and gives up while
|
||
|
|
# the notifier is already running...or at least, that's what
|
||
|
|
# we'd like to arrange, but the _notify_links function doesn't
|
||
|
|
# drop the GIL/object lock, so the other thread is stuck and doesn't
|
||
|
|
# actually get to call into the acquire method.
|
||
|
|
|
||
|
|
from gevent import monkey
|
||
|
|
self.assertFalse(monkey.is_module_patched('threading'))
|
||
|
|
|
||
|
|
import threading
|
||
|
|
|
||
|
|
sem = self._makeOne()
|
||
|
|
# Make future acquires block
|
||
|
|
sem.acquire()
|
||
|
|
|
||
|
|
def greenlet_one():
|
||
|
|
ack = sem.acquire()
|
||
|
|
# We're running in the notifier function right now. It switched to
|
||
|
|
# us.
|
||
|
|
thread.start()
|
||
|
|
gevent.sleep(timing.LARGE_TICK)
|
||
|
|
return ack
|
||
|
|
|
||
|
|
exc_info = []
|
||
|
|
acquired = []
|
||
|
|
|
||
|
|
glet = gevent.spawn(greenlet_one)
|
||
|
|
thread = threading.Thread(target=self._makeThreadMain(
|
||
|
|
threading.Event(), threading.Event(),
|
||
|
|
sem,
|
||
|
|
acquired, exc_info,
|
||
|
|
timeout=timing.LARGE_TICK
|
||
|
|
))
|
||
|
|
thread.daemon = True
|
||
|
|
gevent.idle()
|
||
|
|
sem.release()
|
||
|
|
glet.join()
|
||
|
|
for _ in range(3):
|
||
|
|
gevent.idle()
|
||
|
|
thread.join(timing.LARGE_TICK)
|
||
|
|
|
||
|
|
self.assertEqual(glet.value, True)
|
||
|
|
self.assertEqual([], exc_info)
|
||
|
|
self.assertEqual([False], acquired)
|
||
|
|
self.assertTrue(glet.dead, glet)
|
||
|
|
glet = None
|
||
|
|
|
||
|
|
def assertOneHasNoHub(self, sem):
|
||
|
|
self.assertIsNone(sem.hub, sem)
|
||
|
|
|
||
|
|
@greentest.skipOnPyPyOnWindows("Flaky there; can't reproduce elsewhere")
|
||
|
|
def test_dueling_threads(self, acquire_args=(), create_hub=None):
|
||
|
|
# pylint:disable=too-many-locals,too-many-statements
|
||
|
|
|
||
|
|
# Threads doing nothing but acquiring and releasing locks, without
|
||
|
|
# having any other greenlets to switch to.
|
||
|
|
# https://github.com/gevent/gevent/issues/1698
|
||
|
|
from gevent import monkey
|
||
|
|
from gevent._hub_local import get_hub_if_exists
|
||
|
|
|
||
|
|
self.assertFalse(monkey.is_module_patched('threading'))
|
||
|
|
|
||
|
|
import threading
|
||
|
|
from time import sleep as native_sleep
|
||
|
|
|
||
|
|
sem = self._makeOne()
|
||
|
|
self.assertOneHasNoHub(sem)
|
||
|
|
count = 10000
|
||
|
|
results = [-1, -1]
|
||
|
|
run = True
|
||
|
|
def do_it(ix):
|
||
|
|
if create_hub:
|
||
|
|
gevent.get_hub()
|
||
|
|
|
||
|
|
try:
|
||
|
|
for i in range(count):
|
||
|
|
if not run:
|
||
|
|
break
|
||
|
|
|
||
|
|
acquired = sem.acquire(*acquire_args)
|
||
|
|
assert acquire_args or acquired
|
||
|
|
if acquired:
|
||
|
|
sem.release()
|
||
|
|
results[ix] = i
|
||
|
|
if not create_hub:
|
||
|
|
# We don't artificially create the hub.
|
||
|
|
self.assertIsNone(
|
||
|
|
get_hub_if_exists(),
|
||
|
|
(get_hub_if_exists(), ix, i)
|
||
|
|
)
|
||
|
|
if create_hub and i % 10 == 0:
|
||
|
|
gevent.sleep(timing.SMALLEST_RELIABLE_DELAY)
|
||
|
|
elif i % 100 == 0:
|
||
|
|
native_sleep(timing.SMALLEST_RELIABLE_DELAY)
|
||
|
|
except Exception as ex: # pylint:disable=broad-except
|
||
|
|
import traceback; traceback.print_exc()
|
||
|
|
results[ix] = str(ex)
|
||
|
|
ex = None
|
||
|
|
finally:
|
||
|
|
hub = get_hub_if_exists()
|
||
|
|
if hub is not None:
|
||
|
|
hub.join()
|
||
|
|
hub.destroy(destroy_loop=True)
|
||
|
|
|
||
|
|
t1 = threading.Thread(target=do_it, args=(0,))
|
||
|
|
t1.daemon = True
|
||
|
|
t2 = threading.Thread(target=do_it, args=(1,))
|
||
|
|
t2.daemon = True
|
||
|
|
t1.start()
|
||
|
|
t2.start()
|
||
|
|
|
||
|
|
t1.join(1)
|
||
|
|
t2.join(1)
|
||
|
|
|
||
|
|
while t1.is_alive() or t2.is_alive():
|
||
|
|
cur = list(results)
|
||
|
|
t1.join(7)
|
||
|
|
t2.join(7)
|
||
|
|
if cur == results:
|
||
|
|
# Hmm, after two seconds, no progress
|
||
|
|
run = False
|
||
|
|
break
|
||
|
|
|
||
|
|
self.assertEqual(results, [count - 1, count - 1])
|
||
|
|
|
||
|
|
def test_dueling_threads_timeout(self):
|
||
|
|
self.test_dueling_threads((True, 4))
|
||
|
|
|
||
|
|
def test_dueling_threads_with_hub(self):
|
||
|
|
self.test_dueling_threads(create_hub=True)
|
||
|
|
|
||
|
|
|
||
|
|
# XXX: Need a test with multiple greenlets in a non-primary
|
||
|
|
# thread. Things should work, just very slowly; instead of moving through
|
||
|
|
# greenlet.switch(), they'll be moving with async watchers.
|
||
|
|
|
||
|
|
class TestBoundedSemaphoreMultiThread(TestSemaphoreMultiThread):
|
||
|
|
|
||
|
|
def _getTargetClass(self):
|
||
|
|
return BoundedSemaphore
|
||
|
|
|
||
|
|
@greentest.skipOnPurePython("Needs C extension")
|
||
|
|
class TestCExt(greentest.TestCase):
|
||
|
|
|
||
|
|
def test_c_extension(self):
|
||
|
|
self.assertEqual(Semaphore.__module__,
|
||
|
|
'gevent._gevent_c_semaphore')
|
||
|
|
|
||
|
|
|
||
|
|
class SwitchWithFixedHash(object):
|
||
|
|
# Replaces greenlet.switch with a callable object
|
||
|
|
# with a hash code we control. This only matters if
|
||
|
|
# we're hashing this somewhere (which we used to), but
|
||
|
|
# that doesn't preserve order, so we don't do
|
||
|
|
# that anymore.
|
||
|
|
|
||
|
|
def __init__(self, greenlet, hashcode):
|
||
|
|
self.switch = greenlet.switch
|
||
|
|
self.hashcode = hashcode
|
||
|
|
|
||
|
|
def __hash__(self):
|
||
|
|
raise AssertionError
|
||
|
|
|
||
|
|
def __eq__(self, other):
|
||
|
|
raise AssertionError
|
||
|
|
|
||
|
|
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: # pragma: no cover
|
||
|
|
return
|
||
|
|
g = FirstG.spawn(acquire_then_spawn, sem, should_quit)
|
||
|
|
g.join()
|
||
|
|
|
||
|
|
class TestSemaphoreFair(greentest.TestCase):
|
||
|
|
|
||
|
|
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)
|
||
|
|
|
||
|
|
sem.release()
|
||
|
|
keep_going1.kill()
|
||
|
|
keep_going2.kill()
|
||
|
|
exiting.kill()
|
||
|
|
|
||
|
|
gevent.idle()
|
||
|
|
|
||
|
|
if __name__ == '__main__':
|
||
|
|
greentest.main()
|