@testing.requires.threading_with_mock
    @testing.requires.timing_intensive
    def test_waiters_handled(self):
        """test that threads waiting for connections are
        handled when the pool is replaced.

        """
        mutex = threading.Lock()
        dbapi = MockDBAPI()

        def creator():
            mutex.acquire()
            try:
                return dbapi.connect()
            finally:
                mutex.release()

        success = []
        for timeout in (None, 30):
            for max_overflow in (0, -1, 3):
                p = pool.QueuePool(creator=creator,
                                   pool_size=2, timeout=timeout,
                                   max_overflow=max_overflow)

                def waiter(p, timeout, max_overflow):
                    success_key = (timeout, max_overflow)
                    conn = p.connect()
                    success.append(success_key)
                    time.sleep(.1)
                    conn.close()

                c1 = p.connect()
                c2 = p.connect()

                threads = []
                for i in range(2):
                    t = threading.Thread(target=waiter,
                                    args=(p, timeout, max_overflow))
                    t.daemon = True
                    t.start()
                    threads.append(t)

                # this sleep makes sure that the
                # two waiter threads hit upon wait()
                # inside the queue, before we invalidate the other
                # two conns
                time.sleep(.2)
                p._invalidate(c2)

                for t in threads:
                    t.join(join_timeout)

        eq_(len(success), 12, "successes: %s" % success)
dataset/ETHPy150Open zzzeek/sqlalchemy/test/engine/test_pool.py/QueuePoolTest.test_waiters_handled/original