+++ /dev/null
-import gc
-import os
-import random
-
-import eventlet
-from eventlet import hubs, greenpool, event, pools
-from eventlet.support import greenlets as greenlet, six
-import tests
-
-
-def passthru(a):
- eventlet.sleep(0.01)
- return a
-
-
-def passthru2(a, b):
- eventlet.sleep(0.01)
- return a, b
-
-
-def raiser(exc):
- raise exc
-
-
-class GreenPool(tests.LimitedTestCase):
- def test_spawn(self):
- p = greenpool.GreenPool(4)
- waiters = []
- for i in range(10):
- waiters.append(p.spawn(passthru, i))
- results = [waiter.wait() for waiter in waiters]
- self.assertEqual(results, list(range(10)))
-
- def test_spawn_n(self):
- p = greenpool.GreenPool(4)
- results_closure = []
-
- def do_something(a):
- eventlet.sleep(0.01)
- results_closure.append(a)
-
- for i in range(10):
- p.spawn(do_something, i)
- p.waitall()
- self.assertEqual(results_closure, list(range(10)))
-
- def test_waiting(self):
- pool = greenpool.GreenPool(1)
- done = event.Event()
-
- def consume():
- done.wait()
-
- def waiter(pool):
- gt = pool.spawn(consume)
- gt.wait()
-
- waiters = []
- self.assertEqual(pool.running(), 0)
- waiters.append(eventlet.spawn(waiter, pool))
- eventlet.sleep(0)
- self.assertEqual(pool.waiting(), 0)
- waiters.append(eventlet.spawn(waiter, pool))
- eventlet.sleep(0)
- self.assertEqual(pool.waiting(), 1)
- waiters.append(eventlet.spawn(waiter, pool))
- eventlet.sleep(0)
- self.assertEqual(pool.waiting(), 2)
- self.assertEqual(pool.running(), 1)
- done.send(None)
- for w in waiters:
- w.wait()
- self.assertEqual(pool.waiting(), 0)
- self.assertEqual(pool.running(), 0)
-
- def test_multiple_coros(self):
- evt = event.Event()
- results = []
-
- def producer():
- results.append('prod')
- evt.send()
-
- def consumer():
- results.append('cons1')
- evt.wait()
- results.append('cons2')
-
- pool = greenpool.GreenPool(2)
- done = pool.spawn(consumer)
- pool.spawn_n(producer)
- done.wait()
- self.assertEqual(['cons1', 'prod', 'cons2'], results)
-
- def test_timer_cancel(self):
- # this test verifies that local timers are not fired
- # outside of the context of the spawn
- timer_fired = []
-
- def fire_timer():
- timer_fired.append(True)
-
- def some_work():
- hubs.get_hub().schedule_call_local(0, fire_timer)
-
- pool = greenpool.GreenPool(2)
- worker = pool.spawn(some_work)
- worker.wait()
- eventlet.sleep(0)
- eventlet.sleep(0)
- self.assertEqual(timer_fired, [])
-
- def test_reentrant(self):
- pool = greenpool.GreenPool(1)
-
- def reenter():
- waiter = pool.spawn(lambda a: a, 'reenter')
- self.assertEqual('reenter', waiter.wait())
-
- outer_waiter = pool.spawn(reenter)
- outer_waiter.wait()
-
- evt = event.Event()
-
- def reenter_async():
- pool.spawn_n(lambda a: a, 'reenter')
- evt.send('done')
-
- pool.spawn_n(reenter_async)
- self.assertEqual('done', evt.wait())
-
- def assert_pool_has_free(self, pool, num_free):
- self.assertEqual(pool.free(), num_free)
-
- def wait_long_time(e):
- e.wait()
-
- timer = eventlet.Timeout(1)
- try:
- evt = event.Event()
- for x in six.moves.range(num_free):
- pool.spawn(wait_long_time, evt)
- # if the pool has fewer free than we expect,
- # then we'll hit the timeout error
- finally:
- timer.cancel()
-
- # if the runtime error is not raised it means the pool had
- # some unexpected free items
- timer = eventlet.Timeout(0, RuntimeError)
- try:
- self.assertRaises(RuntimeError, pool.spawn, wait_long_time, evt)
- finally:
- timer.cancel()
-
- # clean up by causing all the wait_long_time functions to return
- evt.send(None)
- eventlet.sleep(0)
- eventlet.sleep(0)
-
- def test_resize(self):
- pool = greenpool.GreenPool(2)
- evt = event.Event()
-
- def wait_long_time(e):
- e.wait()
-
- pool.spawn(wait_long_time, evt)
- pool.spawn(wait_long_time, evt)
- self.assertEqual(pool.free(), 0)
- self.assertEqual(pool.running(), 2)
- self.assert_pool_has_free(pool, 0)
-
- # verify that the pool discards excess items put into it
- pool.resize(1)
-
- # cause the wait_long_time functions to return, which will
- # trigger puts to the pool
- evt.send(None)
- eventlet.sleep(0)
- eventlet.sleep(0)
-
- self.assertEqual(pool.free(), 1)
- self.assertEqual(pool.running(), 0)
- self.assert_pool_has_free(pool, 1)
-
- # resize larger and assert that there are more free items
- pool.resize(2)
- self.assertEqual(pool.free(), 2)
- self.assertEqual(pool.running(), 0)
- self.assert_pool_has_free(pool, 2)
-
- def test_pool_smash(self):
- # The premise is that a coroutine in a Pool tries to get a token out
- # of a token pool but times out before getting the token. We verify
- # that neither pool is adversely affected by this situation.
- pool = greenpool.GreenPool(1)
- tp = pools.TokenPool(max_size=1)
- tp.get() # empty out the pool
-
- def do_receive(tp):
- timer = eventlet.Timeout(0, RuntimeError())
- try:
- tp.get()
- self.fail("Shouldn't have recieved anything from the pool")
- except RuntimeError:
- return 'timed out'
- else:
- timer.cancel()
-
- # the spawn makes the token pool expect that coroutine, but then
- # immediately cuts bait
- e1 = pool.spawn(do_receive, tp)
- self.assertEqual(e1.wait(), 'timed out')
-
- # the pool can get some random item back
- def send_wakeup(tp):
- tp.put('wakeup')
- gt = eventlet.spawn(send_wakeup, tp)
-
- # now we ask the pool to run something else, which should not
- # be affected by the previous send at all
- def resume():
- return 'resumed'
- e2 = pool.spawn(resume)
- self.assertEqual(e2.wait(), 'resumed')
-
- # we should be able to get out the thing we put in there, too
- self.assertEqual(tp.get(), 'wakeup')
- gt.wait()
-
- def test_spawn_n_2(self):
- p = greenpool.GreenPool(2)
- self.assertEqual(p.free(), 2)
- r = []
-
- def foo(a):
- r.append(a)
-
- gt = p.spawn(foo, 1)
- self.assertEqual(p.free(), 1)
- gt.wait()
- self.assertEqual(r, [1])
- eventlet.sleep(0)
- self.assertEqual(p.free(), 2)
-
- # Once the pool is exhausted, spawning forces a yield.
- p.spawn_n(foo, 2)
- self.assertEqual(1, p.free())
- self.assertEqual(r, [1])
-
- p.spawn_n(foo, 3)
- self.assertEqual(0, p.free())
- self.assertEqual(r, [1])
-
- p.spawn_n(foo, 4)
- self.assertEqual(set(r), set([1, 2, 3]))
- eventlet.sleep(0)
- self.assertEqual(set(r), set([1, 2, 3, 4]))
-
- def test_exceptions(self):
- p = greenpool.GreenPool(2)
- for m in (p.spawn, p.spawn_n):
- self.assert_pool_has_free(p, 2)
- m(raiser, RuntimeError())
- self.assert_pool_has_free(p, 1)
- p.waitall()
- self.assert_pool_has_free(p, 2)
- m(raiser, greenlet.GreenletExit)
- self.assert_pool_has_free(p, 1)
- p.waitall()
- self.assert_pool_has_free(p, 2)
-
- def test_imap(self):
- p = greenpool.GreenPool(4)
- result_list = list(p.imap(passthru, range(10)))
- self.assertEqual(result_list, list(range(10)))
-
- def test_empty_imap(self):
- p = greenpool.GreenPool(4)
- result_iter = p.imap(passthru, [])
- self.assertRaises(StopIteration, result_iter.next)
-
- def test_imap_nonefunc(self):
- p = greenpool.GreenPool(4)
- result_list = list(p.imap(None, range(10)))
- self.assertEqual(result_list, [(x,) for x in range(10)])
-
- def test_imap_multi_args(self):
- p = greenpool.GreenPool(4)
- result_list = list(p.imap(passthru2, range(10), range(10, 20)))
- self.assertEqual(result_list, list(zip(range(10), range(10, 20))))
-
- def test_imap_raises(self):
- # testing the case where the function raises an exception;
- # both that the caller sees that exception, and that the iterator
- # continues to be usable to get the rest of the items
- p = greenpool.GreenPool(4)
-
- def raiser(item):
- if item == 1 or item == 7:
- raise RuntimeError("intentional error")
- else:
- return item
-
- it = p.imap(raiser, range(10))
- results = []
- while True:
- try:
- results.append(six.next(it))
- except RuntimeError:
- results.append('r')
- except StopIteration:
- break
- self.assertEqual(results, [0, 'r', 2, 3, 4, 5, 6, 'r', 8, 9])
-
- def test_starmap(self):
- p = greenpool.GreenPool(4)
- result_list = list(p.starmap(passthru, [(x,) for x in range(10)]))
- self.assertEqual(result_list, list(range(10)))
-
- def test_waitall_on_nothing(self):
- p = greenpool.GreenPool()
- p.waitall()
-
- def test_recursive_waitall(self):
- p = greenpool.GreenPool()
- gt = p.spawn(p.waitall)
- self.assertRaises(AssertionError, gt.wait)
-
-
-class GreenPile(tests.LimitedTestCase):
- def test_pile(self):
- p = greenpool.GreenPile(4)
- for i in range(10):
- p.spawn(passthru, i)
- result_list = list(p)
- self.assertEqual(result_list, list(range(10)))
-
- def test_pile_spawn_times_out(self):
- p = greenpool.GreenPile(4)
- for i in range(4):
- p.spawn(passthru, i)
- # now it should be full and this should time out
- eventlet.Timeout(0)
- self.assertRaises(eventlet.Timeout, p.spawn, passthru, "time out")
- # verify that the spawn breakage didn't interrupt the sequence
- # and terminates properly
- for i in range(4, 10):
- p.spawn(passthru, i)
- self.assertEqual(list(p), list(range(10)))
-
- def test_constructing_from_pool(self):
- pool = greenpool.GreenPool(2)
- pile1 = greenpool.GreenPile(pool)
- pile2 = greenpool.GreenPile(pool)
-
- def bunch_of_work(pile, unique):
- for i in range(10):
- pile.spawn(passthru, i + unique)
-
- eventlet.spawn(bunch_of_work, pile1, 0)
- eventlet.spawn(bunch_of_work, pile2, 100)
- eventlet.sleep(0)
- self.assertEqual(list(pile2), list(range(100, 110)))
- self.assertEqual(list(pile1), list(range(10)))
-
-
-class StressException(Exception):
- pass
-
-r = random.Random(0)
-
-
-def pressure(arg):
- while r.random() < 0.5:
- eventlet.sleep(r.random() * 0.001)
- if r.random() < 0.8:
- return arg
- else:
- raise StressException(arg)
-
-
-def passthru(arg):
- while r.random() < 0.5:
- eventlet.sleep(r.random() * 0.001)
- return arg
-
-
-class Stress(tests.LimitedTestCase):
- # tests will take extra-long
- TEST_TIMEOUT = 60
-
- @tests.skip_unless(os.environ.get('RUN_STRESS_TESTS') == 'YES')
- def spawn_order_check(self, concurrency):
- # checks that piles are strictly ordered
- p = greenpool.GreenPile(concurrency)
-
- def makework(count, unique):
- for i in six.moves.range(count):
- token = (unique, i)
- p.spawn(pressure, token)
-
- iters = 1000
- eventlet.spawn(makework, iters, 1)
- eventlet.spawn(makework, iters, 2)
- eventlet.spawn(makework, iters, 3)
- p.spawn(pressure, (0, 0))
- latest = [-1] * 4
- received = 0
- it = iter(p)
- while True:
- try:
- i = six.next(it)
- except StressException as exc:
- i = exc.args[0]
- except StopIteration:
- break
- received += 1
- if received % 5 == 0:
- eventlet.sleep(0.0001)
- unique, order = i
- assert latest[unique] < order
- latest[unique] = order
- for l in latest[1:]:
- self.assertEqual(l, iters - 1)
-
- @tests.skip_unless(os.environ.get('RUN_STRESS_TESTS') == 'YES')
- def test_ordering_5(self):
- self.spawn_order_check(5)
-
- @tests.skip_unless(os.environ.get('RUN_STRESS_TESTS') == 'YES')
- def test_ordering_50(self):
- self.spawn_order_check(50)
-
- def imap_memory_check(self, concurrency):
- # checks that imap is strictly
- # ordered and consumes a constant amount of memory
- p = greenpool.GreenPool(concurrency)
- count = 1000
- it = p.imap(passthru, six.moves.range(count))
- latest = -1
- while True:
- try:
- i = six.next(it)
- except StopIteration:
- break
-
- if latest == -1:
- gc.collect()
- initial_obj_count = len(gc.get_objects())
- assert i > latest
- latest = i
- if latest % 5 == 0:
- eventlet.sleep(0.001)
- if latest % 10 == 0:
- gc.collect()
- objs_created = len(gc.get_objects()) - initial_obj_count
- assert objs_created < 25 * concurrency, objs_created
- # make sure we got to the end
- self.assertEqual(latest, count - 1)
-
- @tests.skip_unless(os.environ.get('RUN_STRESS_TESTS') == 'YES')
- def test_imap_50(self):
- self.imap_memory_check(50)
-
- @tests.skip_unless(os.environ.get('RUN_STRESS_TESTS') == 'YES')
- def test_imap_500(self):
- self.imap_memory_check(500)
-
- @tests.skip_unless(os.environ.get('RUN_STRESS_TESTS') == 'YES')
- def test_with_intpool(self):
- class IntPool(pools.Pool):
- def create(self):
- self.current_integer = getattr(self, 'current_integer', 0) + 1
- return self.current_integer
-
- def subtest(intpool_size, pool_size, num_executes):
- def run(int_pool):
- token = int_pool.get()
- eventlet.sleep(0.0001)
- int_pool.put(token)
- return token
-
- int_pool = IntPool(max_size=intpool_size)
- pool = greenpool.GreenPool(pool_size)
- for ix in six.moves.range(num_executes):
- pool.spawn(run, int_pool)
- pool.waitall()
-
- subtest(4, 7, 7)
- subtest(50, 75, 100)
- for isize in (10, 20, 30, 40, 50):
- for psize in (5, 25, 35, 50):
- subtest(isize, psize, psize)
Code Review / packages / trusty / python-eventlet.git / blobdiff