sqlalchemy-high-concurrency-issue.py

"""
$ python sqlalchemy-high-concurrency-issue.py
...
$ echo "select * from steal;" | mysql -uscott -ptiger test
user_id money
89b65422-36ed-48db-ba25-68c3b27eedcc    2
$
$ python -O sqlalchemy-high-concurrency-issue.py
...
"""
import multiprocessing, time, random

##########################################################################################
# Fuzzing is a technique for amplifying race condition errors to make them more visible


def fuzz():
    if not __debug__:  # enable: `$ python -O ...`
        time.sleep(random.randint(0, 500) / 1_000)  # 0 ~ 0.5 seconds


###########################################################################################

counter = 0  # memory not share cross processes

import warnings
import uuid
USER_ID = uuid.uuid4()
SQLALCHEMY_DATABASE_URI = "mysql+pymysql://scott:tiger@localhost:3306/test?charset=utf8mb4"  # database(mysql) store

from sqlalchemy import Column, CHAR, Integer, create_engine
from sqlalchemy import exc as sa_exc
from sqlalchemy.orm import declarative_base, Session
Base = declarative_base()

class Steal(Base):
    __tablename__ = "steal"
    user_id = Column(CHAR(36), nullable=False, primary_key=True)
    money = Column(Integer, nullable=False)


def maintainer_manually_setup_database():
    engine = create_engine(SQLALCHEMY_DATABASE_URI, future=True, echo=False)
    Base.metadata.drop_all(engine)
    Base.metadata.create_all(engine)
    with Session(engine) as session:
        session.add(Steal(user_id=USER_ID, money=0))
        session.commit()

def plus_one():
    with warnings.catch_warnings():
        warnings.simplefilter("ignore", category=sa_exc.Base20DeprecationWarning)
        engine = create_engine(SQLALCHEMY_DATABASE_URI, future=True, echo=False)
        with Session(engine) as session:
            db_obj: Steal = session.query(Steal).get(USER_ID)
            oldmoney = db_obj.money
            fuzz()
            newmoney = oldmoney + 1
            fuzz()
            db_obj.money = newmoney
            session.add(db_obj)
            fuzz()
            session.commit()
            print('The money is %d' % newmoney, end='')
            print()
            print('---------------', end='')

maintainer_manually_setup_database()


print('Starting up')
fuzz()
thrs = [multiprocessing.Process(target=plus_one) for _ in range(10)]
for t in thrs:
    t.start()
    # t.join()  # Uncomment to Apply Synchronization
    fuzz()
print('Finishing up')
fuzz()
print(f"{counter = }")
fuzz()
for t in thrs:
    t.join()
print("End of all processes!")
print(f"{counter = }")  # `$ echo "select * from steal;" | mysql -uscott -ptiger test`