#!/usr/bin/env python3 """Real MySQL experiments in brand-new disposable schemas; no shared data is reset. Optional isolated client dependencies: PyMySQL + sqlparse. See README. """ import argparse from concurrent.futures import ThreadPoolExecutor from contextlib import closing from datetime import datetime import importlib.util import json import os from pathlib import Path import re import time import uuid import pymysql import sqlparse HERE = Path(__file__).resolve().parent def execute(conn, query, args=None): with conn.cursor() as cur: cur.execute(query, args) rows = cur.fetchall() if cur.description else [] columns = [c[0] for c in cur.description] if cur.description else [] return {'columns': columns, 'rows': rows, 'affected': cur.rowcount} def fixture(conn, filename): for query in sqlparse.split((HERE / filename).read_text()): clean = sqlparse.format(query, strip_comments=True).strip() if not clean or re.match(r'^(CREATE DATABASE|USE\s)', clean, re.I): continue execute(conn, clean) def plans(conn, schema): fixture(conn, 'init.sql') results = [] for query in sqlparse.split((HERE / 'queries.sql').read_text()): clean = sqlparse.format(query, strip_comments=True).strip() if not clean or re.match(r'^USE\s', clean, re.I): continue result = execute(conn, clean) results.append({'sql': clean, **result}) assert execute(conn, "SELECT COUNT(*) FROM orders WHERE user_id=1234 AND status='PAID'")['rows'] == ((0,),) assert execute(conn, "SELECT COUNT(*) FROM orders WHERE user_id=1233 AND status='PAID'")['rows'] == ((10,),) offset = execute(conn, 'SELECT id FROM orders ORDER BY id LIMIT 90000,20')['rows'] keyset = execute(conn, 'SELECT id FROM orders WHERE id>90000 ORDER BY id LIMIT 20')['rows'] assert offset == keyset == tuple((i,) for i in range(90001, 90021)) return {'schema': schema, 'cases': results, 'offset_keyset_same_ids': [r[0] for r in offset]} def transactions(connect, schema): def open_tx(isolation='REPEATABLE READ'): conn = connect(schema) execute(conn, 'SET SESSION TRANSACTION ISOLATION LEVEL ' + isolation) execute(conn, 'SET SESSION innodb_lock_wait_timeout=8') return conn with closing(connect(schema)) as observer: fixture(observer, 'setup.sql') def reset(): execute(observer, 'UPDATE ds_account SET balance=100,version=1') execute(observer, 'UPDATE ds_doctor SET on_call=1') def lock_edges(): return execute(observer, '''SELECT r.OBJECT_NAME,r.INDEX_NAME,r.LOCK_MODE,r.LOCK_DATA, b.LOCK_MODE,b.LOCK_DATA FROM performance_schema.data_lock_waits w JOIN performance_schema.data_locks r ON r.ENGINE=w.ENGINE AND r.ENGINE_LOCK_ID=w.REQUESTING_ENGINE_LOCK_ID JOIN performance_schema.data_locks b ON b.ENGINE=w.ENGINE AND b.ENGINE_LOCK_ID=w.BLOCKING_ENGINE_LOCK_ID WHERE r.OBJECT_SCHEMA=%s AND b.OBJECT_SCHEMA=%s''', (schema,schema))['rows'] def wait_edge(): deadline = time.monotonic() + 4 while time.monotonic() < deadline: edges = lock_edges() if edges: return edges time.sleep(.03) raise AssertionError('Expected a real data_lock_waits edge before releasing blocker') result = {} for isolation in ['REPEATABLE READ', 'READ COMMITTED']: reset() with closing(open_tx(isolation)) as a, closing(open_tx()) as b: a.begin() first = execute(a,'SELECT balance FROM ds_account WHERE id=1')['rows'][0][0] execute(b,'UPDATE ds_account SET balance=80,version=2 WHERE id=1') second = execute(a,'SELECT balance FROM ds_account WHERE id=1')['rows'][0][0] current = execute(a,'SELECT balance FROM ds_account WHERE id=1 FOR UPDATE')['rows'][0][0] execute(a,'UPDATE ds_account SET balance=balance-10 WHERE id=1') own = execute(a,'SELECT balance FROM ds_account WHERE id=1')['rows'][0][0] a.rollback() assert [first,second,current,own] == ([100,100,80,70] if isolation=='REPEATABLE READ' else [100,80,80,70]) result[isolation] = [first,second,current,own] reset() with closing(open_tx()) as a, closing(open_tx()) as b: a.begin() execute(b,'UPDATE ds_account SET balance=80,version=2 WHERE id=1') value = execute(a,'SELECT balance FROM ds_account WHERE id=1')['rows'][0][0] assert value == 80 result['begin_before_first_read'] = value a.rollback() with ThreadPoolExecutor(max_workers=1) as pool: with closing(open_tx()) as a, closing(open_tx()) as b: a.begin(); execute(a,'SELECT * FROM ds_lock WHERE id=15 FOR UPDATE') b.begin() future = pool.submit(execute,b,'INSERT INTO ds_lock VALUES(15)') edges = wait_edge() a.rollback() assert future.result(timeout=10)['affected'] == 1 b.rollback() result['unique_miss_gap_wait'] = edges a.begin();execute(a,'SELECT * FROM ds_lock WHERE id=20 FOR UPDATE') b.begin();assert execute(b,'INSERT INTO ds_lock VALUES(15)')['affected'] == 1 assert lock_edges() == () a.rollback();b.rollback() result['unique_hit_allows_other_gap'] = True reset() with closing(open_tx('READ COMMITTED')) as a, closing(open_tx('READ COMMITTED')) as b: a.begin();b.begin() query='UPDATE ds_account SET balance=balance-80,version=version+1 WHERE id=1 AND balance>=80' assert execute(a,query)['affected'] == 1 future=pool.submit(execute,b,query) edges=wait_edge();a.commit() affected=future.result(timeout=10)['affected'];b.commit() balance=execute(observer,'SELECT balance FROM ds_account WHERE id=1')['rows'][0][0] assert (affected,balance)==(0,20) result['conditional_wait_then_recheck']={'affected':affected,'balance':balance,'edges':edges} reset() with closing(open_tx()) as a, closing(open_tx()) as b: a.begin();b.begin() execute(a,'UPDATE ds_account SET balance=balance+1 WHERE id=1') execute(b,'UPDATE ds_account SET balance=balance+1 WHERE id=2') def attempted(conn, query): try:return {'status':'ok','affected':execute(conn,query)['affected']} except pymysql.err.OperationalError as exc: if exc.args[0] != 1213:raise return {'status':'victim','code':1213} future=pool.submit(attempted,a,'UPDATE ds_account SET balance=balance+1 WHERE id=2') edges=wait_edge() b_result=attempted(b,'UPDATE ds_account SET balance=balance+1 WHERE id=1') a_result=future.result(timeout=10) assert sorted([a_result['status'],b_result['status']]) == ['ok','victim'] a.rollback();b.rollback() result['deadlock']={'A':a_result,'B':b_result,'pre_cycle_edges':edges} status=execute(observer,'SHOW ENGINE INNODB STATUS')['rows'][0][2] # Save only this experiment's latest deadlock section, never unrelated global history. section=status.split('LATEST DETECTED DEADLOCK',1)[-1].split('TRANSACTIONS',1)[0] result['deadlock']['own_status_section']=section if schema in section else 'not captured: latest global deadlock was replaced' reset() with closing(open_tx()) as a, closing(open_tx()) as b: a.begin();b.begin() assert execute(a,'SELECT SUM(on_call) FROM ds_doctor')['rows'][0][0] == 2 assert execute(b,'SELECT SUM(on_call) FROM ds_doctor')['rows'][0][0] == 2 execute(a,'UPDATE ds_doctor SET on_call=0 WHERE id=1');a.commit() execute(b,'UPDATE ds_doctor SET on_call=0 WHERE id=2');b.commit() remaining=execute(observer,'SELECT SUM(on_call) FROM ds_doctor')['rows'][0][0] assert remaining == 0 result['write_skew_remaining']=remaining reset() with closing(open_tx()) as a, closing(open_tx()) as b: for conn,doctor in [(a,1),(b,2)]: conn.begin();execute(conn,'SELECT shift_id FROM ds_guard WHERE shift_id=1 FOR UPDATE') rows=execute(conn,'SELECT id,on_call FROM ds_doctor ORDER BY id FOR UPDATE')['rows'] if sum(r[1] for r in rows)>1: execute(conn,'UPDATE ds_doctor SET on_call=0 WHERE id=%s',(doctor,)) conn.commit() remaining=execute(observer,'SELECT SUM(on_call) FROM ds_doctor')['rows'][0][0] assert remaining == 1 result['guard_current_check_remaining']=remaining return result def main(): parser=argparse.ArgumentParser(description=__doc__) parser.add_argument('--host', default='127.0.0.1') parser.add_argument('--port', type=int, default=3306) parser.add_argument('--user', default='root') parser.add_argument('--mode', choices=['plans','tx','practice','all'], default='all') args=parser.parse_args() def connect(schema=None): return pymysql.connect(host=args.host,port=args.port,user=args.user,password=os.environ.get('MYSQL_PWD',''), database=schema,autocommit=True,connect_timeout=3,read_timeout=15,write_timeout=15) prefix='study_t16_'+datetime.now().strftime('%Y%m%d')+'_'+uuid.uuid4().hex[:8] created=[] result={'kind':'real-mysql','schema_prefix':prefix} with closing(connect()) as admin: result['runtime']=execute(admin,'''SELECT VERSION() AS version,@@global.transaction_isolation AS global_isolation, @@session.autocommit AS client_autocommit,@@innodb_page_size AS page_size, @@innodb_deadlock_detect AS deadlock_detect,@@innodb_rollback_on_timeout AS rollback_on_timeout, @@innodb_flush_log_at_trx_commit AS flush_log_at_commit,@@sync_binlog AS sync_binlog''') try: for mode in ['plans','tx','practice']: if args.mode not in ('all',mode):continue schema=prefix+'_'+mode assert re.fullmatch(r'study_t16_\d{8}_[0-9a-f]{8}_(plans|tx|practice)',schema) # CREATE without IF NOT EXISTS refuses collisions; cleanup only tracks newly created schemas. execute(admin,'CREATE DATABASE `'+schema+'` CHARACTER SET utf8mb4') created.append(schema) if mode=='tx':result['tx']=transactions(connect,schema) elif mode=='plans': with closing(connect(schema)) as conn:result['plans']=plans(conn,schema) else: spec=importlib.util.spec_from_file_location('practice_verifier',HERE/'verify_practice.py') module=importlib.util.module_from_spec(spec);spec.loader.exec_module(module) # The optional MySQL adapter is added by the SQL content slice. with closing(connect(schema)) as conn:result['practice']=module.verify_mysql(conn) finally: for schema in reversed(created):execute(admin,'DROP DATABASE `'+schema+'`') result['cleaned_schemas']=created print(json.dumps(result,ensure_ascii=False,indent=2,default=str)) if __name__=='__main__':main()