Skip to content

Instantly share code, notes, and snippets.

@jmlon
Created September 20, 2022 14:13
Show Gist options
  • Save jmlon/d12506a71bd70ca56c83cc506b7e41f0 to your computer and use it in GitHub Desktop.
Save jmlon/d12506a71bd70ca56c83cc506b7e41f0 to your computer and use it in GitHub Desktop.
# pip install mysql-connector-python
import threading
import time
import mysql.connector
HOST = 'localhost'
USER = 'root'
PASSWORD = 'my-secret-pw'
SLEEP = 2
CONNECTIONS = 10
err_count=0
def openConnection():
mydb = mysql.connector.connect(
host=HOST,
user=USER,
password=PASSWORD
)
return mydb
def closeConnection(conn):
conn.close()
def runQuery(conn):
mycursor = conn.cursor()
mycursor.execute("SELECT 1+1")
total = 0
for x in mycursor:
total += x[0]
return total
def worker(id):
global err_count
try:
c = openConnection()
start_time = time.time()
t = runQuery(c)
end_time = time.time()
print(f"{id} time: {end_time-start_time:.6f}, total={t}")
time.sleep(SLEEP)
closeConnection(c)
except mysql.connector.errors.DatabaseError:
err_count += 1
if __name__=="__main__":
print('MySQL Connection stress test')
hilos = []
err_count=0
for i in range(CONNECTIONS):
h = threading.Thread( target=worker, args=(len(hilos),) )
hilos.append(h)
h.start()
for h in hilos:
h.join()
print(f'Connection errors: {err_count}')
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment