pools_test.py 文件源码

def test_create_contention(self):
        creates = [0]

        def sleep_create():
            creates[0] += 1
            eventlet.sleep()
            return "slept"

        p = pools.Pool(max_size=4, create=sleep_create)

        def do_get():
            x = p.get()
            self.assertEqual(x, "slept")
            p.put(x)

        gp = eventlet.GreenPool()
        for i in six.moves.range(100):
            gp.spawn_n(do_get)
        gp.waitall()
        self.assertEqual(creates[0], 4)