#!/usr/bin/env python3 """Measure SQLite writer contention in a disposable WAL database.""" import argparse import json import platform import sqlite3 import tempfile import threading import time from datetime import datetime, timezone from pathlib import Path def milliseconds(started): return round((time.monotonic() - started) * 1000, 1) def main(): parser = argparse.ArgumentParser() parser.add_argument('--output', required=True) args = parser.parse_args() output = Path(args.output) if output.exists(): raise SystemExit('Existing evidence will not be overwritten.') with tempfile.TemporaryDirectory(prefix='punchblog-sqlite-lock-') as directory: database = Path(directory) / 'lab.sqlite3' setup = sqlite3.connect(database) journal_mode = setup.execute('PRAGMA journal_mode=WAL').fetchone()[0] setup.execute('CREATE TABLE events (id INTEGER PRIMARY KEY, label TEXT NOT NULL)') setup.execute("INSERT INTO events(label) VALUES ('baseline')") setup.commit() setup.close() def hold_writer(label, ready, release, failures): connection = sqlite3.connect(database, timeout=1.0) try: connection.execute('BEGIN IMMEDIATE') connection.execute('INSERT INTO events(label) VALUES (?)', (label,)) ready.set() if not release.wait(5): raise RuntimeError('Writer release timed out') connection.commit() except Exception as error: failures.append(repr(error)) ready.set() finally: connection.close() def start_holder(label): ready = threading.Event() release = threading.Event() failures = [] thread = threading.Thread( target=hold_writer, args=(label, ready, release, failures), daemon=True, ) thread.start() if not ready.wait(5): raise RuntimeError('Writer did not acquire the lock') if failures: raise RuntimeError(failures[0]) return thread, release, failures holder, release, failures = start_holder('holder-zero') reader = sqlite3.connect(database, timeout=0) started = time.monotonic() visible_labels = [row[0] for row in reader.execute('SELECT label FROM events ORDER BY id')] reader_ms = milliseconds(started) reader.close() contender = sqlite3.connect(database, timeout=0) started = time.monotonic() try: contender.execute("INSERT INTO events(label) VALUES ('zero-contender')") contender.commit() zero_result = 'unexpected success' zero_error = '' except sqlite3.OperationalError as error: zero_result = 'database is locked' zero_error = str(error) contender.rollback() zero_ms = milliseconds(started) contender.close() release.set() holder.join(5) assert not holder.is_alive() and not failures assert visible_labels == ['baseline'] assert zero_error == 'database is locked' holder, release, failures = start_holder('holder-short') contender = sqlite3.connect(database, timeout=0.2) started = time.monotonic() try: contender.execute("INSERT INTO events(label) VALUES ('short-contender')") contender.commit() short_result = 'unexpected success' short_error = '' except sqlite3.OperationalError as error: short_result = 'database is locked' short_error = str(error) contender.rollback() short_ms = milliseconds(started) contender.close() release.set() holder.join(5) assert not holder.is_alive() and not failures assert short_error == 'database is locked' holder, release, failures = start_holder('holder-wait') timer = threading.Timer(0.45, release.set) timer.start() contender = sqlite3.connect(database, timeout=1.5) started = time.monotonic() contender.execute("INSERT INTO events(label) VALUES ('waiting-contender')") contender.commit() waited_ms = milliseconds(started) contender.close() timer.join() holder.join(5) assert not holder.is_alive() and not failures check = sqlite3.connect(f'file:{database}?mode=ro', uri=True) final_labels = [row[0] for row in check.execute('SELECT label FROM events ORDER BY id')] integrity = check.execute('PRAGMA integrity_check').fetchone()[0] check.close() assert zero_result == short_result == 'database is locked' assert final_labels == [ 'baseline', 'holder-zero', 'holder-short', 'holder-wait', 'waiting-contender' ] assert integrity == 'ok' record = { 'schema_version': 1, 'observed_at': datetime.now(timezone.utc).isoformat(), 'environment': { 'os': platform.system(), 'python': platform.python_version(), 'sqlite': sqlite3.sqlite_version, 'journal_mode': journal_mode, }, 'method': '임시 SQLite DB를 WAL 모드로 만들고 한 연결이 BEGIN IMMEDIATE 쓰기 트랜잭션을 보유한 동안 다른 연결의 읽기와 쓰기를 실행했다. 운영 DB와 네트워크는 사용하지 않았다.', 'limits': '단일 호스트·단일 프로세스의 짧은 합성 실험이다. 동시 writer가 많은 부하, 네트워크 파일시스템, 긴 트랜잭션의 원인, 애플리케이션 재시도 정책은 측정하지 않았다.', 'observations': { 'reader_during_writer': { 'result': 'success', 'elapsed_ms': reader_ms, 'visible_labels': visible_labels, 'uncommitted_holder_visible': 'holder-zero' in visible_labels, }, 'writer_timeout_zero': { 'timeout_seconds': 0, 'result': zero_result, 'elapsed_ms': zero_ms, }, 'writer_timeout_short': { 'timeout_seconds': 0.2, 'result': short_result, 'elapsed_ms': short_ms, }, 'writer_waits_for_release': { 'timeout_seconds': 1.5, 'holder_release_after_ms': 450, 'result': 'commit success', 'elapsed_ms': waited_ms, }, 'final': { 'integrity_check': integrity, 'row_count': len(final_labels), 'labels': final_labels, }, }, 'tables': { 'contention': { 'headers': ['상황', '설정', '결과', '경과 시간'], 'rows': [ ['writer가 미커밋 상태일 때 읽기', 'timeout=0초', '성공, 커밋된 1행만 조회', f'{reader_ms}ms'], ['다른 writer가 잠금 보유', 'timeout=0초', zero_result, f'{zero_ms}ms'], ['다른 writer가 잠금 보유', 'timeout=0.2초', short_result, f'{short_ms}ms'], ['450ms 뒤 기존 writer 해제', 'timeout=1.5초', 'commit success', f'{waited_ms}ms'], ], } }, } output.parent.mkdir(parents=True, exist_ok=True) with output.open('x') as stream: json.dump(record, stream, ensure_ascii=False, indent=2) stream.write('\n') print(json.dumps(record['tables'], ensure_ascii=False)) if __name__ == '__main__': main()