主键没查到为什么INSERT还在等把InnoDB锁画在10和20之间前面那篇行锁实验里A锁住已有的记录B修改同一行就等待。这个好理解这行正被另一个事务占着。现在把条件改一下。表里只有主键10和20A查15返回空集。15这一行都不存在B再插入15总该能继续吧先不背“间隙锁”这个名字。我们把这条空查询留下再开一个窗口看看究竟发生了什么。留着A的事务去B插入一条记录本次实测是MySQL8.0.46、InnoDBA显式使用REPEATABLE READ可重复读后面简称RR。id是单列主键没有其他索引或外键。下面在自己的临时实例操作数据库名已存在时换一个新名字。任意会话先建表CREATE DATABASE gap_lab; USE gap_lab; CREATE TABLE account ( id INT PRIMARY KEY, balance INT NOT NULL ) ENGINEInnoDB; INSERT INTO account VALUES (10,100), (20,200);A执行到SELECT就停下先别提交USE gap_lab; SELECT CONNECTION_ID(); SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ; START TRANSACTION; SELECT * FROM account WHERE id15 FOR UPDATE;屏幕上是空集。注意这里有FOR UPDATE并且事务还没结束。B也连同一个实例执行USE gap_lab; SELECT CONNECTION_ID(); SET SESSION innodb_lock_wait_timeout30; START TRANSACTION; INSERT INTO account VALUES (15,150);这次INSERT没有马上返回。先别急着在B重复输入去第三个会话C看一眼。C需要有读取performance_schema的权限SELECT t.PROCESSLIST_ID, l.INDEX_NAME, l.LOCK_MODE, l.LOCK_STATUS, l.LOCK_DATA FROM performance_schema.data_locks AS l JOIN performance_schema.threads AS t ON t.THREAD_IDl.THREAD_ID WHERE l.OBJECT_SCHEMAgap_lab AND l.OBJECT_NAMEaccount ORDER BY t.PROCESSLIST_ID, l.LOCK_DATA;按刚才记下的连接ID区分A和B。本次关键的两条记录是下面这样省略了表级意向锁A PRIMARY X,GAP GRANTED 20 B PRIMARY X,GAP,INSERT_INTENTION WAITING 20两个LOCK_DATA都是20。明明查的是15为什么这里却写20先把15要放的位置画出来主键按顺序排列现在只有两个位置放了记录已有记录 已有记录 10 20 |---------------| ↑ 15要插在这里查15时找不到记录但可以确定它应该在10与20之间。A持有的X,GAP描述的是这个间隙在这次输出里它附着在右端索引记录20上。所以不能把LOCK_DATA20直接读成“20这条业务记录被独占了”。还得一起看LOCK_MODE这里有GAP。A查询15空集事务未结束 10 [ 暂时不让新记录插进来 ] 20 ↑ B想把15放进这里 ↓ 等A结束如果只挡住15这个值那么插入12应该没问题。可以重新做一轮先在A执行COMMIT让B的INSERT返回B执行ROLLBACK撤销刚插入的15使表回到只有10和20。接着A重新开始同样的事务、同样查询15B把插入值换成12。结果12也等待。19同样等待换成25在A还没提交时就能完成。每轮结束都在B回滚以免上一轮改变下一轮的间隙。现在范围就看出来了挡住的是10与20之间的新插入不只是查询条件里写出的那个15。25不落在这里也不是“整张表都锁住了”。为什么会顺带影响12可以回到那张主键顺序图当前索引里没有12或15的记录二者要进入的是同一个相邻记录间隙。这里的锁以这个索引间隙为单位不是在给业务条件中的每个整数单独建一把锁。MySQL对间隙锁的说明把它用于阻止向相应索引间隙插入间隙锁之间的兼容规则也不能照搬记录共享锁、排他锁的直觉。把查询改回10会有什么区别还是每轮先把两边事务收好恢复10和20。A查询的条件改成START TRANSACTION; SELECT * FROM account WHERE id10 FOR UPDATE;这次能拿到已有记录10,100。观察会话看到的关键记录是A PRIMARY X,REC_NOT_GAP GRANTED 10此时B插入15可以完成B更新10则会等待。查询10确实找到这一行 [10被锁住] --- 15可插入 --- 20 ↑ B更新10要等“唯一等值查询只锁一条记录”这句话不能省掉“确实定位到已有记录”等前提。这里用到的是完整单列主键复合唯一索引只给部分列、二级非唯一索引、范围查询都不能直接套这个图。不加FOR UPDATE呢换隔离级别呢A恢复REPEATABLE READ开启事务只运行普通SELECTSTART TRANSACTION; SELECT * FROM account WHERE id15;同样是空集这次B插入15可以完成。本例普通SELECT是非锁定一致性读不能看见一个SELECT就认为它把扫描范围锁住了。再结束双方事务、恢复数据。A在新事务开始前改成READ COMMITTED读已提交简称RCSET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED; START TRANSACTION; SELECT * FROM account WHERE id15 FOR UPDATE;B插入15也能完成。注意修改隔离级别的那句放在START TRANSACTION之前不是在原事务中途切换。把这几轮放在一起读才能知道是哪一个条件起了作用A未结束时做了什么B尝试什么本次结果RR锁定读缺失的15插入12、15或19等待RR锁定读缺失的15插入25完成RR锁定读已有的10插入15完成RR锁定读已有的10更新10等待RR普通读取缺失的15插入15完成RC锁定读缺失的15插入15完成RC下针对搜索、扫描通常不采用同样的间隙锁机制但外键检查、重复键检查还有例外。本文这张简单表的结果不等于“RC绝不会出现间隙锁”。相应规则见MySQL隔离级别说明。查询空集以后要不要一直占着这个事务这个问题在“先查有没有再决定插入”的代码里很实际。查询没有找到数据不代表它没有影响别人事务如果在这里等用户操作、调用远程接口其他落在同一间隙的插入就可能一起等着。反过来间隙锁也不是“业务查重已经万无一失”的凭据。多个事务先查空、再同时插入仍要考虑等待和死锁最后防止重复数据的约束应由唯一索引等机制承担并处理实际返回的错误。排查一次INSERT卡住时可以按这次的顺序走先确认它真的在等锁再找到持锁事务的SQL、隔离级别和未提交时间最后结合索引顺序画出它要插的位置。先别因为SELECT返回空集就把那个会话排除掉。附自动实验怎样避免“好像卡住了”的判断人工实验适合看过程自动程序则需要明确的先后顺序A收到SELECT完成标记后才启动B预期等待的分支必须在data_lock_waits中观察到B对A的等待关系才让A提交。只用sleep等一秒无法区分锁等待与机器慢。预期不等待的分支则要求A仍保持事务时B已经收到SQL完成标记。每轮回滚B确认表还是10,100与20,200再开始下一轮。下面是完整运行器。它新建临时数据目录只使用Unix socket不连接原有数据库。需要Linux、Python3和MySQL 8.0的二进制发行目录通过BLOG_MYSQL_ROOT指定解包根目录其中应包含usr/sbin/mysqld、usr/bin/mysql以及相应动态库。若你使用系统安装的MySQL前面的三个会话SQL更适合直接复现。保存为verify_gap.py普通用户运行BLOG_MYSQL_ROOT/path/to/mysql-package-root python3 verify_gap.py结果写在程序旁的results.json每个运行目录保留服务日志与客户端错误日志。临时实例关闭后保留实验目录方便检查没有自动删除已有目录。Run only in an isolated MySQL instance; never connect to a users database. import getpass import json import os from pathlib import Path import select import subprocess import tempfile import time base Path(__file__).resolve().parent package Path(os.environ.get(BLOG_MYSQL_ROOT, /mnt/d/blog-ops/experiments/2026-10-03-candidates/mysql-packages/root)) env dict(os.environ, LD_LIBRARY_PATHstr(package / usr/lib/x86_64-linux-gnu)) run Path(tempfile.mkdtemp(prefixgap-, dirbase)) (run / data).mkdir() sock Path(/dev/shm) / (run.name .sock) args [str(package / usr/sbin/mysqld), --no-defaults, --basedir str(package / usr), --datadir str(run / data), --lc-messages-dir str(package / usr/share/mysql), --user getpass.getuser(), --secure-file-privNULL, --innodb-buffer-pool-size67108864, --innodb-redo-log-capacity16777216, --log-error str(run / server.log)] subprocess.run(args [--initialize-insecure], envenv, checkTrue, capture_outputTrue, timeout90) client [str(package / usr/bin/mysql), --no-defaults, --protocolSOCKET, --socket str(sock), -u, root, --batch, --skip-column-names, --unbuffered, --connect-timeout2] server subprocess.Popen(args [--skip-networking, --mysqlx0, --socket str(sock), --pid-file str(run / mysql.pid), --tmpdir str(run)], envenv, stdoutsubprocess.DEVNULL, stderrsubprocess.DEVNULL) sessions [] record {passed: False, run: str(run), cases: []} observer_sql SELECT t.PROCESSLIST_ID, l.INDEX_NAME, l.LOCK_MODE, l.LOCK_STATUS, l.LOCK_DATA FROM performance_schema.data_locks AS l JOIN performance_schema.threads AS t ON t.THREAD_IDl.THREAD_ID WHERE l.OBJECT_SCHEMAgap_lab AND l.OBJECT_NAMEaccount ORDER BY t.PROCESSLIST_ID, l.LOCK_DATA; class Session: def __init__(self, name): self.err open(run / (name .stderr), wb) self.p subprocess.Popen(client, envenv, stdinsubprocess.PIPE, stdoutsubprocess.PIPE, stderrself.err) self.buffer b self.counter 0 sessions.append(self) def send(self, text): self.counter 1 self.marker __done_%d__ % self.counter self.p.stdin.write((text \nSELECT %s;\n % self.marker).encode()) self.p.stdin.flush() def receive(self): lines [] deadline time.monotonic() 15 while time.monotonic() deadline: while b\n in self.buffer: line, self.buffer self.buffer.split(b\n, 1) value line.decode().rstrip(\r) if value self.marker: return lines lines.append(value) if select.select([self.p.stdout], [], [], .1)[0]: data os.read(self.p.stdout.fileno(), 65536) if not data: raise RuntimeError(MySQL client ended; see self.err.name) self.buffer data raise TimeoutError(SQL did not reach its completion marker) def sql(self, text): self.send(text) return self.receive() def close(self): if self.p.poll() is None: self.p.stdin.close() try: self.p.wait(timeout5) except subprocess.TimeoutExpired: self.p.kill() self.p.wait() self.p.stdout.close() self.err.close() try: for _ in range(200): ping subprocess.run(client, inputSELECT VERSION();, textTrue, envenv, capture_outputTrue, timeout3) if ping.returncode 0: record[version] ping.stdout.strip() break if server.poll() is not None: raise RuntimeError((run / server.log).read_text()) time.sleep(.1) else: raise TimeoutError(server startup) a, b, c (Session(n) for n in [A, B, C]) c.sql(CREATE DATABASE gap_lab; USE gap_lab; CREATE TABLE account(id INT PRIMARY KEY,balance INT NOT NULL) ENGINEInnoDB; INSERT INTO account VALUES(10,100),(20,200);) for s in (a, b): s.sql(USE gap_lab; SET SESSION innodb_lock_wait_timeout30;) ids [int(s.sql(SELECT CONNECTION_ID();)[0]) for s in (a, b, c)] record[connectionIds] ids # Same schema; change one condition at a time. cases [ (rr-missing-15, REPEATABLE READ, 15, True, INSERT INTO account VALUES(15,150);, True), (rr-missing-12, REPEATABLE READ, 15, True, INSERT INTO account VALUES(12,120);, True), (rr-missing-19, REPEATABLE READ, 15, True, INSERT INTO account VALUES(19,190);, True), (rr-outside-25, REPEATABLE READ, 15, True, INSERT INTO account VALUES(25,250);, False), (rc-missing-15, READ COMMITTED, 15, True, INSERT INTO account VALUES(15,150);, False), (rr-existing-insert, REPEATABLE READ, 10, True, INSERT INTO account VALUES(15,150);, False), (rr-existing-update, REPEATABLE READ, 10, True, UPDATE account SET balance101 WHERE id10;, True), (rr-ordinary, REPEATABLE READ, 15, False, INSERT INTO account VALUES(15,150);, False), ] for name, isolation, key, locking, action, should_wait in cases: a.sql(SET SESSION TRANSACTION ISOLATION LEVEL isolation ; START TRANSACTION;) selection fSELECT * FROM account WHERE id{key} ( FOR UPDATE; if locking else ;) selected a.sql(selection) assert selected ([10\t100] if key 10 else []), selected locks c.sql(SELECT INDEX_NAME,LOCK_TYPE,LOCK_MODE,LOCK_STATUS,LOCK_DATA FROM performance_schema.data_locks WHERE OBJECT_SCHEMAgap_lab AND THREAD_ID(SELECT THREAD_ID FROM performance_schema.threads fWHERE PROCESSLIST_ID{ids[0]}) ORDER BY INDEX_NAME,LOCK_DATA;) record_locks [line for line in locks if line.startswith(PRIMARY\tRECORD\t)] if isolation REPEATABLE READ and locking: assert record_locks ([PRIMARY\tRECORD\tX,GAP\tGRANTED\t20] if key 15 else [PRIMARY\tRECORD\tX,REC_NOT_GAP\tGRANTED\t10]), locks else: assert not record_locks, locks b.sql(START TRANSACTION;) b.send(action) waits [] if should_wait: deadline time.monotonic() 8 while time.monotonic() deadline: waits c.sql(SELECT r.LOCK_MODE,r.LOCK_STATUS,r.LOCK_DATA, bl.LOCK_MODE,bl.LOCK_STATUS,bl.LOCK_DATA FROM performance_schema.data_lock_waits w JOIN performance_schema.data_locks r ON r.ENGINE_LOCK_IDw.REQUESTING_ENGINE_LOCK_ID AND r.ENGINEw.ENGINE JOIN performance_schema.data_locks bl ON bl.ENGINE_LOCK_IDw.BLOCKING_ENGINE_LOCK_ID AND bl.ENGINEw.ENGINE JOIN performance_schema.threads t ON t.THREAD_IDw.REQUESTING_THREAD_ID fWHERE t.PROCESSLIST_ID{ids[1]};) if waits: break time.sleep(.05) assert waits, name did not enter an observed lock wait assert b.p.poll() is None observed c.sql(observer_sql) assert any(line.startswith(str(ids[0])\tPRIMARY\t) for line in observed) assert any(line.startswith(str(ids[1])\tPRIMARY\t) and \tWAITING\t in line for line in observed), observed a.sql(COMMIT;) assert b.receive() [] else: assert b.receive() [] # A still holds its transaction here. observed c.sql(observer_sql) a.sql(COMMIT;) b.sql(ROLLBACK;) assert c.sql(SELECT * FROM account ORDER BY id;) [10\t100, 20\t200] record[cases].append({id: name, isolation: isolation, select: selection, selected: selected, action: action, waitObserved: bool(waits), locks: locks, waits: waits, observerRows: observed, completedAfterACommit: should_wait}) record[passed] True finally: for s in sessions: s.close() server.terminate() try: server.wait(timeout30) except subprocess.TimeoutExpired: server.kill() server.wait() record[serverStopped] server.poll() is not None record[clientsStopped] all(s.p.poll() is not None for s in sessions) record[socketRemoved] not sock.exists() (base / results.json).write_text(json.dumps(record, ensure_asciiFalse, indent2)) print(RESULTS str(base / results.json)) 
长河君 · 2026/11/5 2:00:43 · 阅读 1.2k 主键没查到，为什么INSERT还在等？把InnoDB锁画在10和20之间
主键没查到为什么INSERT还在等把InnoDB锁画在10和20之间前面那篇行锁实验里A锁住已有的记录B修改同一行就等待。这个好理解这行正被另一个事务占着。现在把条件改一下。表里只有主键10和20A查15返回空集。15这一行都不存在B再插入15总该能继续吧先不背“间隙锁”这个名字。我们把这条空查询留下再开一个窗口看看究竟发生了什么。留着A的事务去B插入一条记录本次实测是MySQL8.0.46、InnoDBA显式使用REPEATABLE READ可重复读后面简称RR。id是单列主键没有其他索引或外键。下面在自己的临时实例操作数据库名已存在时换一个新名字。任意会话先建表CREATE DATABASE gap_lab; USE gap_lab; CREATE TABLE account ( id INT PRIMARY KEY, balance INT NOT NULL ) ENGINEInnoDB; INSERT INTO account VALUES (10,100), (20,200);A执行到SELECT就停下先别提交USE gap_lab; SELECT CONNECTION_ID(); SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ; START TRANSACTION; SELECT * FROM account WHERE id15 FOR UPDATE;屏幕上是空集。注意这里有FOR UPDATE并且事务还没结束。B也连同一个实例执行USE gap_lab; SELECT CONNECTION_ID(); SET SESSION innodb_lock_wait_timeout30; START TRANSACTION; INSERT INTO account VALUES (15,150);这次INSERT没有马上返回。先别急着在B重复输入去第三个会话C看一眼。C需要有读取performance_schema的权限SELECT t.PROCESSLIST_ID, l.INDEX_NAME, l.LOCK_MODE, l.LOCK_STATUS, l.LOCK_DATA FROM performance_schema.data_locks AS l JOIN performance_schema.threads AS t ON t.THREAD_IDl.THREAD_ID WHERE l.OBJECT_SCHEMAgap_lab AND l.OBJECT_NAMEaccount ORDER BY t.PROCESSLIST_ID, l.LOCK_DATA;按刚才记下的连接ID区分A和B。本次关键的两条记录是下面这样省略了表级意向锁A PRIMARY X,GAP GRANTED 20 B PRIMARY X,GAP,INSERT_INTENTION WAITING 20两个LOCK_DATA都是20。明明查的是15为什么这里却写20先把15要放的位置画出来主键按顺序排列现在只有两个位置放了记录已有记录 已有记录 10 20 |---------------| ↑ 15要插在这里查15时找不到记录但可以确定它应该在10与20之间。A持有的X,GAP描述的是这个间隙在这次输出里它附着在右端索引记录20上。所以不能把LOCK_DATA20直接读成“20这条业务记录被独占了”。还得一起看LOCK_MODE这里有GAP。A查询15空集事务未结束 10 [ 暂时不让新记录插进来 ] 20 ↑ B想把15放进这里 ↓ 等A结束如果只挡住15这个值那么插入12应该没问题。可以重新做一轮先在A执行COMMIT让B的INSERT返回B执行ROLLBACK撤销刚插入的15使表回到只有10和20。接着A重新开始同样的事务、同样查询15B把插入值换成12。结果12也等待。19同样等待换成25在A还没提交时就能完成。每轮结束都在B回滚以免上一轮改变下一轮的间隙。现在范围就看出来了挡住的是10与20之间的新插入不只是查询条件里写出的那个15。25不落在这里也不是“整张表都锁住了”。为什么会顺带影响12可以回到那张主键顺序图当前索引里没有12或15的记录二者要进入的是同一个相邻记录间隙。这里的锁以这个索引间隙为单位不是在给业务条件中的每个整数单独建一把锁。MySQL对间隙锁的说明把它用于阻止向相应索引间隙插入间隙锁之间的兼容规则也不能照搬记录共享锁、排他锁的直觉。把查询改回10会有什么区别还是每轮先把两边事务收好恢复10和20。A查询的条件改成START TRANSACTION; SELECT * FROM account WHERE id10 FOR UPDATE;这次能拿到已有记录10,100。观察会话看到的关键记录是A PRIMARY X,REC_NOT_GAP GRANTED 10此时B插入15可以完成B更新10则会等待。查询10确实找到这一行 [10被锁住] --- 15可插入 --- 20 ↑ B更新10要等“唯一等值查询只锁一条记录”这句话不能省掉“确实定位到已有记录”等前提。这里用到的是完整单列主键复合唯一索引只给部分列、二级非唯一索引、范围查询都不能直接套这个图。不加FOR UPDATE呢换隔离级别呢A恢复REPEATABLE READ开启事务只运行普通SELECTSTART TRANSACTION; SELECT * FROM account WHERE id15;同样是空集这次B插入15可以完成。本例普通SELECT是非锁定一致性读不能看见一个SELECT就认为它把扫描范围锁住了。再结束双方事务、恢复数据。A在新事务开始前改成READ COMMITTED读已提交简称RCSET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED; START TRANSACTION; SELECT * FROM account WHERE id15 FOR UPDATE;B插入15也能完成。注意修改隔离级别的那句放在START TRANSACTION之前不是在原事务中途切换。把这几轮放在一起读才能知道是哪一个条件起了作用A未结束时做了什么B尝试什么本次结果RR锁定读缺失的15插入12、15或19等待RR锁定读缺失的15插入25完成RR锁定读已有的10插入15完成RR锁定读已有的10更新10等待RR普通读取缺失的15插入15完成RC锁定读缺失的15插入15完成RC下针对搜索、扫描通常不采用同样的间隙锁机制但外键检查、重复键检查还有例外。本文这张简单表的结果不等于“RC绝不会出现间隙锁”。相应规则见MySQL隔离级别说明。查询空集以后要不要一直占着这个事务这个问题在“先查有没有再决定插入”的代码里很实际。查询没有找到数据不代表它没有影响别人事务如果在这里等用户操作、调用远程接口其他落在同一间隙的插入就可能一起等着。反过来间隙锁也不是“业务查重已经万无一失”的凭据。多个事务先查空、再同时插入仍要考虑等待和死锁最后防止重复数据的约束应由唯一索引等机制承担并处理实际返回的错误。排查一次INSERT卡住时可以按这次的顺序走先确认它真的在等锁再找到持锁事务的SQL、隔离级别和未提交时间最后结合索引顺序画出它要插的位置。先别因为SELECT返回空集就把那个会话排除掉。附自动实验怎样避免“好像卡住了”的判断人工实验适合看过程自动程序则需要明确的先后顺序A收到SELECT完成标记后才启动B预期等待的分支必须在data_lock_waits中观察到B对A的等待关系才让A提交。只用sleep等一秒无法区分锁等待与机器慢。预期不等待的分支则要求A仍保持事务时B已经收到SQL完成标记。每轮回滚B确认表还是10,100与20,200再开始下一轮。下面是完整运行器。它新建临时数据目录只使用Unix socket不连接原有数据库。需要Linux、Python3和MySQL 8.0的二进制发行目录通过BLOG_MYSQL_ROOT指定解包根目录其中应包含usr/sbin/mysqld、usr/bin/mysql以及相应动态库。若你使用系统安装的MySQL前面的三个会话SQL更适合直接复现。保存为verify_gap.py普通用户运行BLOG_MYSQL_ROOT/path/to/mysql-package-root python3 verify_gap.py结果写在程序旁的results.json每个运行目录保留服务日志与客户端错误日志。临时实例关闭后保留实验目录方便检查没有自动删除已有目录。Run only in an isolated MySQL instance; never connect to a users database. import getpass import json import os from pathlib import Path import select import subprocess import tempfile import time base Path(__file__).resolve().parent package Path(os.environ.get(BLOG_MYSQL_ROOT, /mnt/d/blog-ops/experiments/2026-10-03-candidates/mysql-packages/root)) env dict(os.environ, LD_LIBRARY_PATHstr(package / usr/lib/x86_64-linux-gnu)) run Path(tempfile.mkdtemp(prefixgap-, dirbase)) (run / data).mkdir() sock Path(/dev/shm) / (run.name .sock) args [str(package / usr/sbin/mysqld), --no-defaults, --basedir str(package / usr), --datadir str(run / data), --lc-messages-dir str(package / usr/share/mysql), --user getpass.getuser(), --secure-file-privNULL, --innodb-buffer-pool-size67108864, --innodb-redo-log-capacity16777216, --log-error str(run / server.log)] subprocess.run(args [--initialize-insecure], envenv, checkTrue, capture_outputTrue, timeout90) client [str(package / usr/bin/mysql), --no-defaults, --protocolSOCKET, --socket str(sock), -u, root, --batch, --skip-column-names, --unbuffered, --connect-timeout2] server subprocess.Popen(args [--skip-networking, --mysqlx0, --socket str(sock), --pid-file str(run / mysql.pid), --tmpdir str(run)], envenv, stdoutsubprocess.DEVNULL, stderrsubprocess.DEVNULL) sessions [] record {passed: False, run: str(run), cases: []} observer_sql SELECT t.PROCESSLIST_ID, l.INDEX_NAME, l.LOCK_MODE, l.LOCK_STATUS, l.LOCK_DATA FROM performance_schema.data_locks AS l JOIN performance_schema.threads AS t ON t.THREAD_IDl.THREAD_ID WHERE l.OBJECT_SCHEMAgap_lab AND l.OBJECT_NAMEaccount ORDER BY t.PROCESSLIST_ID, l.LOCK_DATA; class Session: def __init__(self, name): self.err open(run / (name .stderr), wb) self.p subprocess.Popen(client, envenv, stdinsubprocess.PIPE, stdoutsubprocess.PIPE, stderrself.err) self.buffer b self.counter 0 sessions.append(self) def send(self, text): self.counter 1 self.marker __done_%d__ % self.counter self.p.stdin.write((text \nSELECT %s;\n % self.marker).encode()) self.p.stdin.flush() def receive(self): lines [] deadline time.monotonic() 15 while time.monotonic() deadline: while b\n in self.buffer: line, self.buffer self.buffer.split(b\n, 1) value line.decode().rstrip(\r) if value self.marker: return lines lines.append(value) if select.select([self.p.stdout], [], [], .1)[0]: data os.read(self.p.stdout.fileno(), 65536) if not data: raise RuntimeError(MySQL client ended; see self.err.name) self.buffer data raise TimeoutError(SQL did not reach its completion marker) def sql(self, text): self.send(text) return self.receive() def close(self): if self.p.poll() is None: self.p.stdin.close() try: self.p.wait(timeout5) except subprocess.TimeoutExpired: self.p.kill() self.p.wait() self.p.stdout.close() self.err.close() try: for _ in range(200): ping subprocess.run(client, inputSELECT VERSION();, textTrue, envenv, capture_outputTrue, timeout3) if ping.returncode 0: record[version] ping.stdout.strip() break if server.poll() is not None: raise RuntimeError((run / server.log).read_text()) time.sleep(.1) else: raise TimeoutError(server startup) a, b, c (Session(n) for n in [A, B, C]) c.sql(CREATE DATABASE gap_lab; USE gap_lab; CREATE TABLE account(id INT PRIMARY KEY,balance INT NOT NULL) ENGINEInnoDB; INSERT INTO account VALUES(10,100),(20,200);) for s in (a, b): s.sql(USE gap_lab; SET SESSION innodb_lock_wait_timeout30;) ids [int(s.sql(SELECT CONNECTION_ID();)[0]) for s in (a, b, c)] record[connectionIds] ids # Same schema; change one condition at a time. cases [ (rr-missing-15, REPEATABLE READ, 15, True, INSERT INTO account VALUES(15,150);, True), (rr-missing-12, REPEATABLE READ, 15, True, INSERT INTO account VALUES(12,120);, True), (rr-missing-19, REPEATABLE READ, 15, True, INSERT INTO account VALUES(19,190);, True), (rr-outside-25, REPEATABLE READ, 15, True, INSERT INTO account VALUES(25,250);, False), (rc-missing-15, READ COMMITTED, 15, True, INSERT INTO account VALUES(15,150);, False), (rr-existing-insert, REPEATABLE READ, 10, True, INSERT INTO account VALUES(15,150);, False), (rr-existing-update, REPEATABLE READ, 10, True, UPDATE account SET balance101 WHERE id10;, True), (rr-ordinary, REPEATABLE READ, 15, False, INSERT INTO account VALUES(15,150);, False), ] for name, isolation, key, locking, action, should_wait in cases: a.sql(SET SESSION TRANSACTION ISOLATION LEVEL isolation ; START TRANSACTION;) selection fSELECT * FROM account WHERE id{key} ( FOR UPDATE; if locking else ;) selected a.sql(selection) assert selected ([10\t100] if key 10 else []), selected locks c.sql(SELECT INDEX_NAME,LOCK_TYPE,LOCK_MODE,LOCK_STATUS,LOCK_DATA FROM performance_schema.data_locks WHERE OBJECT_SCHEMAgap_lab AND THREAD_ID(SELECT THREAD_ID FROM performance_schema.threads fWHERE PROCESSLIST_ID{ids[0]}) ORDER BY INDEX_NAME,LOCK_DATA;) record_locks [line for line in locks if line.startswith(PRIMARY\tRECORD\t)] if isolation REPEATABLE READ and locking: assert record_locks ([PRIMARY\tRECORD\tX,GAP\tGRANTED\t20] if key 15 else [PRIMARY\tRECORD\tX,REC_NOT_GAP\tGRANTED\t10]), locks else: assert not record_locks, locks b.sql(START TRANSACTION;) b.send(action) waits [] if should_wait: deadline time.monotonic() 8 while time.monotonic() deadline: waits c.sql(SELECT r.LOCK_MODE,r.LOCK_STATUS,r.LOCK_DATA, bl.LOCK_MODE,bl.LOCK_STATUS,bl.LOCK_DATA FROM performance_schema.data_lock_waits w JOIN performance_schema.data_locks r ON r.ENGINE_LOCK_IDw.REQUESTING_ENGINE_LOCK_ID AND r.ENGINEw.ENGINE JOIN performance_schema.data_locks bl ON bl.ENGINE_LOCK_IDw.BLOCKING_ENGINE_LOCK_ID AND bl.ENGINEw.ENGINE JOIN performance_schema.threads t ON t.THREAD_IDw.REQUESTING_THREAD_ID fWHERE t.PROCESSLIST_ID{ids[1]};) if waits: break time.sleep(.05) assert waits, name did not enter an observed lock wait assert b.p.poll() is None observed c.sql(observer_sql) assert any(line.startswith(str(ids[0])\tPRIMARY\t) for line in observed) assert any(line.startswith(str(ids[1])\tPRIMARY\t) and \tWAITING\t in line for line in observed), observed a.sql(COMMIT;) assert b.receive() [] else: assert b.receive() [] # A still holds its transaction here. observed c.sql(observer_sql) a.sql(COMMIT;) b.sql(ROLLBACK;) assert c.sql(SELECT * FROM account ORDER BY id;) [10\t100, 20\t200] record[cases].append({id: name, isolation: isolation, select: selection, selected: selected, action: action, waitObserved: bool(waits), locks: locks, waits: waits, observerRows: observed, completedAfterACommit: should_wait}) record[passed] True finally: for s in sessions: s.close() server.terminate() try: server.wait(timeout30) except subprocess.TimeoutExpired: server.kill() server.wait() record[serverStopped] server.poll() is not None record[clientsStopped] all(s.p.poll() is not None for s in sessions) record[socketRemoved] not sock.exists() (base / results.json).write_text(json.dumps(record, ensure_asciiFalse, indent2)) print(RESULTS str(base / results.json))