-class Tpool(ProcessBase):
- @skip_with_pyevent
- def test_tpool_size(self):
- expected = "40"
- normal = "20"
- new_mod = """from eventlet import tpool
-import eventlet
-import time
-current = [0]
-highwater = [0]
-def count():
- current[0] += 1
- time.sleep(0.1)
- if current[0] > highwater[0]:
- highwater[0] = current[0]
- current[0] -= 1
-expected = %s
-normal = %s
-p = eventlet.GreenPool()
-for i in range(expected*2):
- p.spawn(tpool.execute, count)
-p.waitall()
-assert highwater[0] > 20, "Highwater %%s <= %%s" %% (highwater[0], normal)
-"""
- os.environ['EVENTLET_THREADPOOL_SIZE'] = expected
- try:
- self.write_to_tempfile("newmod", new_mod % (expected, normal))
- output, lines = self.launch_subprocess('newmod.py')
- self.assertEqual(len(lines), 1, lines)
- finally:
- del os.environ['EVENTLET_THREADPOOL_SIZE']