Archived
feat(db): 同步栈读写并发(_q 去锁)+ core.db 并发/跨loop 修复
auth_system._q:单语句读不再持 _DB_LOCK —— WAL 下每连接读一致性快照, 读写互不阻塞;旧实现把全部读也串行化且被写挡住。写事务(_x 与 24 个 with _DB_LOCK 块,经扫描均为写事务)仍全程串行,无写-写竞争。 core.db 并发修复(由测试暴露的真实缺陷): - WAL 切换竞态:并发连接同时执行 PRAGMA journal_mode=WAL 争排他锁 → 专用锁内串行切换 + 二次复查(busy_timeout 前置) - 跨事件循环锁:模块级 asyncio.Lock 绑定首个 loop,多 loop 场景报 "bound to a different event loop" → 锁按 loop 隔离 测试:25/25 通过(新增「读不被写锁阻塞」红转绿、读+同步写者并发 WAL、 异步并发读); lint: ruff All checks passed;pyright 0 errors。 验收:冒烟 10/10、奇偶校验 2/2、uvicorn 真实启动 Flask 兜底+FastAPI 双通。
This commit is contained in:
1 parent
52ed8b54fb
commit
c77ef40fd5
4 files changed
+135
-10
No files matched your search
+6
-1
@@ -60,7 +60,12 @@ def get_conn() -> sqlite3.Connection:
|
|||||||
|
|
||||||
|
|
||||||
def _q(sql: str, params: tuple = ()) -> list[sqlite3.Row]:
|
def _q(sql: str, params: tuple = ()) -> list[sqlite3.Row]:
|
||||||
with _DB_LOCK:
|
"""单语句读:**不持锁**。
|
||||||
|
|
||||||
|
WAL 模式下每个连接读到一致性快照,读与写互不阻塞;旧实现对读也加 _DB_LOCK,
|
||||||
|
等于把全部读串行化并被写挡住。此处仅放开单语句读——写事务(_x 与各
|
||||||
|
`with _DB_LOCK` 块)仍全程串行,故不引入写-写竞争。
|
||||||
|
"""
|
||||||
conn = get_conn()
|
conn = get_conn()
|
||||||
try:
|
try:
|
||||||
return conn.execute(sql, params).fetchall()
|
return conn.execute(sql, params).fetchall()
|
||||||
|
|||||||
+37
-8
@@ -25,21 +25,50 @@ import aiosqlite
|
|||||||
DATA_DIR = Path(os.environ.get('SITE_DATA_DIR', 'data'))
|
DATA_DIR = Path(os.environ.get('SITE_DATA_DIR', 'data'))
|
||||||
DB_PATH = DATA_DIR / 'site.db'
|
DB_PATH = DATA_DIR / 'site.db'
|
||||||
|
|
||||||
# 写事务串行锁(惰性创建,避免跨事件循环绑定)
|
# 锁按事件循环隔离:asyncio.Lock 绑定创建它的 loop,跨 loop 复用会抛
|
||||||
_WRITE_LOCK: asyncio.Lock | None = None
|
# "bound to a different event loop"(迁移期 FastAPI loop + 同步桥 loop + 测试各自
|
||||||
|
# 的 loop 并存,必须分开持有)。
|
||||||
|
_LOOP_LOCKS: dict[tuple[int, str], asyncio.Lock] = {}
|
||||||
|
|
||||||
|
|
||||||
|
def _loop_lock(kind: str) -> asyncio.Lock:
|
||||||
|
key = (id(asyncio.get_running_loop()), kind)
|
||||||
|
lock = _LOOP_LOCKS.get(key)
|
||||||
|
if lock is None:
|
||||||
|
lock = asyncio.Lock()
|
||||||
|
_LOOP_LOCKS[key] = lock
|
||||||
|
return lock
|
||||||
|
|
||||||
|
|
||||||
def _write_lock() -> asyncio.Lock:
|
def _write_lock() -> asyncio.Lock:
|
||||||
global _WRITE_LOCK
|
"""写事务串行锁(每事件循环一个)。"""
|
||||||
if _WRITE_LOCK is None:
|
return _loop_lock('write')
|
||||||
_WRITE_LOCK = asyncio.Lock()
|
|
||||||
return _WRITE_LOCK
|
|
||||||
|
def _wal_lock() -> asyncio.Lock:
|
||||||
|
"""WAL 切换专用锁(不能用 _write_lock:execute() 已持它再调
|
||||||
|
connect→_configure 会自锁死)。"""
|
||||||
|
return _loop_lock('wal')
|
||||||
|
|
||||||
|
|
||||||
|
# 已确认切到 WAL 的库路径(首连设置一次;后续连接不再重复执行 PRAGMA 争排他锁)
|
||||||
|
_WAL_READY_FOR: str | None = None
|
||||||
|
|
||||||
|
|
||||||
async def _configure(conn: aiosqlite.Connection) -> None:
|
async def _configure(conn: aiosqlite.Connection) -> None:
|
||||||
|
global _WAL_READY_FOR
|
||||||
conn.row_factory = aiosqlite.Row
|
conn.row_factory = aiosqlite.Row
|
||||||
await conn.execute('PRAGMA journal_mode=WAL') # 读写并发的前提
|
# busy_timeout 先设:让后续语句遇锁自旋等待,而不是立刻报 database is locked
|
||||||
await conn.execute('PRAGMA busy_timeout=15000') # 遇写锁自旋等待,而非立即报错
|
await conn.execute('PRAGMA busy_timeout=15000')
|
||||||
|
key = str(DB_PATH)
|
||||||
|
if _WAL_READY_FOR != key:
|
||||||
|
# 切换日志模式需要排他锁,且 `_WAL_READY_FOR` 的检查不是原子的——
|
||||||
|
# 并发连接须在专用锁内串行完成「切换 + 标记」,并二次复查,否则批量并发
|
||||||
|
# 连接会同时执行 PRAGMA journal_mode=WAL 争锁(实测 database is locked)
|
||||||
|
async with _wal_lock():
|
||||||
|
if _WAL_READY_FOR != key:
|
||||||
|
await conn.execute('PRAGMA journal_mode=WAL')
|
||||||
|
_WAL_READY_FOR = key
|
||||||
await conn.execute('PRAGMA synchronous=NORMAL') # WAL 下兼顾安全与吞吐
|
await conn.execute('PRAGMA synchronous=NORMAL') # WAL 下兼顾安全与吞吐
|
||||||
await conn.execute('PRAGMA foreign_keys=ON')
|
await conn.execute('PRAGMA foreign_keys=ON')
|
||||||
|
|
||||||
|
|||||||
@@ -78,5 +78,9 @@ async def test_bridge_rejects_inside_loop():
|
|||||||
async def coro():
|
async def coro():
|
||||||
return 1
|
return 1
|
||||||
|
|
||||||
|
c = coro()
|
||||||
|
try:
|
||||||
with pytest.raises(RuntimeError):
|
with pytest.raises(RuntimeError):
|
||||||
async_bridge.run_sync(coro())
|
async_bridge.run_sync(c)
|
||||||
|
finally:
|
||||||
|
c.close() # run_sync 拒绝执行,需自行关闭以免「未 await」告警
|
||||||
@@ -0,0 +1,87 @@
|
|||||||
|
# -*- coding: utf-8 -*-
|
||||||
|
"""auth_system 同步栈的读写并发验证:读不持锁、写仍串行(WAL)。"""
|
||||||
|
import sqlite3
|
||||||
|
import threading
|
||||||
|
import time
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
import auth_system
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture()
|
||||||
|
def temp_site(tmp_path, monkeypatch):
|
||||||
|
monkeypatch.setattr(auth_system, 'DATA_DIR', tmp_path)
|
||||||
|
monkeypatch.setattr(auth_system, 'DB_PATH', tmp_path / 'site.db')
|
||||||
|
monkeypatch.setattr(auth_system, 'BACKUP_DIR', tmp_path / 'backups')
|
||||||
|
monkeypatch.setattr(auth_system, 'SECRET_KEY_PATH', tmp_path / 'secret.key')
|
||||||
|
auth_system.init_db()
|
||||||
|
return tmp_path / 'site.db'
|
||||||
|
|
||||||
|
|
||||||
|
def test_q_does_not_hold_lock(temp_site):
|
||||||
|
"""_q 读不应获取 _DB_LOCK(否则读会被写挡住)。"""
|
||||||
|
# 让另一个线程持有写锁,读仍应在短时间返回(不被阻塞)
|
||||||
|
held = threading.Event()
|
||||||
|
release = threading.Event()
|
||||||
|
|
||||||
|
def holder():
|
||||||
|
with auth_system._DB_LOCK:
|
||||||
|
held.set()
|
||||||
|
release.wait(5)
|
||||||
|
|
||||||
|
t = threading.Thread(target=holder, daemon=True)
|
||||||
|
t.start()
|
||||||
|
held.wait(2)
|
||||||
|
try:
|
||||||
|
start = time.monotonic()
|
||||||
|
rows = auth_system._q('SELECT COUNT(*) AS n FROM users')
|
||||||
|
elapsed = time.monotonic() - start
|
||||||
|
assert rows is not None
|
||||||
|
assert elapsed < 1.0, f'读被写锁阻塞了 {elapsed:.2f}s(WAL 下不应发生)'
|
||||||
|
finally:
|
||||||
|
release.set()
|
||||||
|
t.join(2)
|
||||||
|
|
||||||
|
|
||||||
|
def test_concurrent_reads_and_writes(temp_site):
|
||||||
|
"""多读线程 + 一个写线程并发:无异常、结果一致。"""
|
||||||
|
errors = []
|
||||||
|
stop = threading.Event()
|
||||||
|
|
||||||
|
def writer():
|
||||||
|
try:
|
||||||
|
for i in range(50):
|
||||||
|
auth_system.set_setting(f'k{i}', str(i))
|
||||||
|
except Exception as e:
|
||||||
|
errors.append(('w', e))
|
||||||
|
finally:
|
||||||
|
stop.set()
|
||||||
|
|
||||||
|
def reader():
|
||||||
|
while not stop.is_set():
|
||||||
|
try:
|
||||||
|
auth_system.get_setting('k0')
|
||||||
|
except Exception as e:
|
||||||
|
errors.append(('r', e))
|
||||||
|
|
||||||
|
ws = threading.Thread(target=writer, daemon=True)
|
||||||
|
rs = [threading.Thread(target=reader, daemon=True) for _ in range(4)]
|
||||||
|
ws.start()
|
||||||
|
for r in rs:
|
||||||
|
r.start()
|
||||||
|
ws.join(20)
|
||||||
|
for r in rs:
|
||||||
|
r.join(2)
|
||||||
|
|
||||||
|
assert not errors, f'并发读写异常: {errors[:3]}'
|
||||||
|
assert auth_system.get_setting('k49') == '49'
|
||||||
|
|
||||||
|
|
||||||
|
def test_wal_enabled(temp_site):
|
||||||
|
conn = sqlite3.connect(str(temp_site))
|
||||||
|
try:
|
||||||
|
mode = conn.execute('PRAGMA journal_mode').fetchone()[0]
|
||||||
|
finally:
|
||||||
|
conn.close()
|
||||||
|
assert str(mode).lower() == 'wal'
|
||||||
Reference in new issue
Block a user