1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
|
# load test of connection pool
import testenv; testenv.configure_for_tests()
import thread, time
from sqlalchemy import *
import sqlalchemy.pool as pool
from testlib import testing
db = create_engine(testing.db.url, pool_timeout=30, echo_pool=True)
metadata = MetaData(db)
users_table = Table('users', metadata,
Column('user_id', Integer, primary_key=True),
Column('user_name', String(40)),
Column('password', String(10)))
metadata.drop_all()
metadata.create_all()
users_table.insert().execute([{'user_name':'user#%d' % i, 'password':'pw#%d' % i} for i in range(1000)])
def runfast():
while True:
c = db.pool.connect()
time.sleep(.5)
c.close()
# result = users_table.select(limit=100).execute()
# d = {}
# for row in result:
# for col in row.keys():
# d[col] = row[col]
# time.sleep(.005)
# result.close()
print "runfast cycle complete"
#thread.start_new_thread(runslow, ())
for x in xrange(0,50):
thread.start_new_thread(runfast, ())
time.sleep(100)
|
