yeah, sorry I get this entirely now, ticket is http://www.sqlalchemy.org/trac/ticket/2880 and here’s a test, has nothing to do with gevent. the patch should be fine and I’ll see if i can get it into 0.8.4 as well.
from sqlalchemy.pool import QueuePool
from sqlalchemy.testing.mock import Mock, call
import threading
import time
dbapi = Mock()
def hanging_dbapi():
time.sleep(5)
return dbapi.connect()
def fast_dbapi():
return dbapi.connect()
def failing_dbapi():
time.sleep(5)
raise Exception("connection failed")
creator = threading.local()
def create():
return creator.mock_connector()
def run_test(name, should_hang, should_fail):
print("run test: %s %s %s" % (name, should_hang, should_fail))
if should_fail:
creator.mock_connector = failing_dbapi
elif should_hang:
creator.mock_connector = hanging_dbapi
else:
creator.mock_connector = fast_dbapi
conn = pool.connect()
print("connected: %s" % name)
conn.operation(name)
time.sleep(3)
conn.close()
pool = QueuePool(creator=create, pool_size=2, max_overflow=3)
success_one = threading.Thread(target=run_test, args=("success_one", False,
False))
success_two = threading.Thread(target=run_test, args=("success_two", False,
False))
overflow_one = threading.Thread(target=run_test, args=("overflow_one", True,
False))
overflow_two = threading.Thread(target=run_test, args=("overflow_two", False,
False))
overflow_three = threading.Thread(target=run_test, args=("overflow_three",
False, False))
success_one.start()
time.sleep(.5)
success_two.start()
time.sleep(.5)
overflow_one.start()
time.sleep(.5)
overflow_two.start()
time.sleep(.5)
overflow_three.start()
time.sleep(.5)
overflow_one.join(timeout=10)
assert \
dbapi.connect().operation.mock_calls == \
[call("success_one"), call("success_two"),
call("overflow_two"), call("overflow_three"), call("overflow_one")],\
dbapi.connect().operation.mock_calls
signature.asc
Description: Message signed with OpenPGP using GPGMail
