auto_id_cache.py

import threading
import time

import mysql.connector
from mysql.connector import pooling

from collections import Counter

import sys

# Create a connection pool
dbconfig = {
    "database": "test",
    "user": "root",
    "password": "",
    "host": "127.0.0.1",
    "port": "4000"
}

try:
    pool = mysql.connector.pooling.MySQLConnectionPool(
        pool_name="mypool",
        pool_size=10,
        pool_reset_session=True,
        **dbconfig
    )
    # Create test table
    connection = pool.get_connection()
    cursor = connection.cursor()
    cursor.execute("drop table if exists t_cache")
    cursor.execute("create table t_cache(a int primary key auto_increment, b timestamp not null default now()) auto_id_cache 1000")
    #cursor.execute("create table t_cache(a int primary key auto_increment, b timestamp not null default now()) auto_id_cache 1")
    cursor.close()
    connection.close()
except mysql.connector.Error as err:
    sys.exit("Error: {}".format(err))
  
# test name
test_name = "auto_increment with auto_id_cache=1000"
# test_name = "auto_increment with auto_id_cache=1"

# run time in second
run_time = 60

counter = Counter()

def do_work(pool):
    try:
        connection = pool.get_connection() 
        if connection.is_connected():
            cursor = connection.cursor()
            cursor.execute("insert into t_cache (a) values (NULL)")
            cursor.close()
            connection.commit()
            connection.close()
    except mysql.connector.Error as err:
        sys.exit("Error: {}".format(err))

_run_ = True

def thread1(pool):
    while (_run_):
        do_work(pool)
        counter[1] += 1

def thread2(pool):
    while (_run_):
        do_work(pool)
        counter[2] += 1

def thread3(pool):
    while (_run_):
        do_work(pool)
        counter[3] += 1

def thread4(pool):
    while (_run_):
        do_work(pool)
        counter[4] += 1

# Create two threads
thread1 = threading.Thread(target=thread1, args=(pool,))
thread2 = threading.Thread(target=thread2, args=(pool,))
thread3 = threading.Thread(target=thread3, args=(pool,))
thread4 = threading.Thread(target=thread4, args=(pool,))

# Start the threads
thread1.start()
thread2.start()
thread3.start()
thread4.start()


time.sleep(run_time)

_run_ = False

# Wait for all threads to complete
thread1.join()
thread2.join()
thread3.join()
thread4.join()

print(test_name + " QPS: " + str(sum(counter.values())/run_time))