SQLite WALモードを最大限に活かす！並行性とデータ整合性のためのロック戦略
2025-05-31
通常のSQLiteのジャーナルモード（Rollback Journalモード）では、書き込みを行う際にデータベースファイル全体に排他ロックがかかるため、書き込み中は他の読み込みや書き込みがブロックされます。これに対し、WALモードは並行処理性能を向上させるために、異なるロックメカニズムを採用しています。
WALモードとは？
WALモードでは、データの変更を直接データベースファイル（.db）に書き込むのではなく、まずWALファイル（.wal）と呼ばれる別のファイルに追記します。そして、定期的にWALファイルの内容をデータベースファイルにマージする「チェックポイント処理」が行われます。これにより、書き込みと読み込みの並行処理が可能になります。
WALモードのロックの仕組み
WALモードでは、主に以下の種類のロックが協調して動作します。
-
- 用途
WALモードのデータベースに接続しているすべてのプロセス（読み書き、読み取り専用問わず）が、常にこのロックを保持します。これは、データベースがWALモードで動作していることを他のプロセスに知らせる役割も果たします。 - 特徴
複数のプロセスが同時に共有ロックを保持できます。これにより、複数の読み取り処理が同時に実行可能です。
- 用途
-
WAL_WRITE_LOCK (書き込みロック)
- 用途
WALファイルに新しいデータを追記するライター（書き込みを行うプロセス）が、このロックを排他的に取得します。 - 特徴
一度にこのロックを保持できるのは1つのプロセスのみです。これにより、WALファイルへの追記が順序立てて行われ、データの整合性が保たれます。他の書き込みプロセスは、このロックが解放されるまで待機します。
- 用途
-
WAL_CKPT_LOCK (チェックポイントロック)
- 用途
WALファイルの内容をデータベースファイルにマージするチェックポイント処理を行うプロセスが、このロックを排他的に取得します。 - 特徴
チェックポイント処理中は、このロックが保持されます。チェックポイント処理も一度に1つのプロセスしか実行できません。
- 用途
-
WAL_READ_LOCK(N) (読み取りロック)
- 用途
読み取りトランザクション中に、WALファイル内の特定の「読み取りマーク」に関連する共有ロックが取得されます。 - 特徴
読み取りプロセスは、特定の時点のデータベースのスナップショットを読み取るために、このロックを利用します。これにより、書き込み処理が進行中でも、読み取り処理は一貫性のあるデータを参照できます。
- 用途
-
SQLITE_LOCK_EXCLUSIVE (排他ロック)
- 用途
WALモードから通常のロールバックジャーナルモードへの切り替え時や、データベースの復旧処理、あるいはデータベースが完全に閉じられる際に、一時的に取得されることがあります。 - 特徴
データベース全体に対する排他ロックであり、このロックが取得されている間は他の全ての操作がブロックされます。
- 用途
WALモードの最大の利点は、読み込みと書き込みの並行処理が大幅に向上することです。
- 書き込みのシリアル化
WALファイルへの書き込み自体は依然としてシリアル化されます（一度に1つのライターのみがWAL_WRITE_LOCKを取得）。しかし、これは非常に短時間で完了するため、Rollback Journalモードのようにデータベースファイル全体が長時間ロックされることはありません。 - 読み取りと書き込みの同時実行
ライターがWALファイルに追記している間も、リーダーはデータベースファイルから古いバージョンのデータを読み取ることができます。これにより、「データベースがロックされました」といったエラーの発生頻度が大幅に減少します。
WALモードの一般的なエラーと問題
-
- 状況
WALモードは読み込みと書き込みの並行性を向上させますが、書き込み操作自体は依然として単一のライターのみが許可されます。つまり、複数のプロセスやスレッドが同時に書き込みを行おうとすると、SQLITE_BUSYエラーが発生することがあります。特に、トランザクションが長時間継続する場合に発生しやすくなります。 - WAL特有の状況
- ライターの競合
複数のプロセスが同時にWAL_WRITE_LOCKを取得しようとすると、競合が発生します。 - トランザクションのアップグレード
読み取り専用でトランザクションを開始した後、そのトランザクション内で書き込みを行おうとすると、SQLiteは内部的にトランザクションを書き込み可能にアップグレードしようとします。このとき、他の書き込みトランザクションがすでに存在すると、即座にSQLITE_BUSYエラーが発生することがあります（PRAGMA busy_timeoutの設定に関わらず）。
- ライターの競合
- 状況
-
WALファイル (
.wal) または共有メモリファイル (.shm) の肥大化/削除されない問題- 状況
WALモードでは、データベースファイル本体 (.db) の他に、変更ログを記録する.walファイルと、WALファイルのインデックスとして使用される.shmファイルが生成されます。通常、データベースへの最後の接続が閉じられたとき、または自動チェックポイント処理が完了したときにこれらのファイルは削除されるかリセットされます。しかし、以下のような場合にこれらのファイルが残り続けることがあります。- アプリケーションの異常終了
データベース接続が適切に閉じられずにアプリケーションがクラッシュした場合、WALファイルやSHMファイルが適切にクリーンアップされないことがあります。 - 常時アクティブなリーダー
チェックポイント処理が完了するためには、WALファイルにアクセスしているアクティブなリーダーが一時的にでも存在しない「リーダーギャップ」が必要です。もし、常にいずれかのプロセスがデータベースを読み込んでいる状態だと、チェックポイントが完了せず、WALファイルが際限なく肥大化する可能性があります。 - 自動チェックポイントの無効化/遅延
PRAGMA wal_autocheckpointの設定を変更したり、明示的にチェックポイントを無効にしたりしている場合。
- アプリケーションの異常終了
- 状況
-
データ整合性の問題（特にクラッシュ後の回復）
- 状況
WALモードでは、synchronousプラグマの設定が重要です。PRAGMA synchronous=NORMAL;(デフォルト) の場合、コミット時にWALファイルへの書き込みは確実にディスクに同期されますが、データベースファイル本体へのマージ（チェックポイント）は必ずしも同期されません。この設定はパフォーマンスが高いですが、システムクラッシュや電源喪失が発生した場合、最後にコミットされたデータの一部が失われる可能性があります（ただし、データベースが破損することはありません）。PRAGMA synchronous=FULL;(Rollback Journalモードのデフォルト) に設定すると、コミット時にWALファイルの内容が確実にディスクに書き込まれ、チェックポイント処理もより堅牢になります。これは最も安全な設定ですが、パフォーマンスが低下します。
- 状況
-
ネットワークファイルシステム上での問題
- 状況
SQLiteのWALモードは、共有メモリ（-shmファイル）を介したプロセス間通信に依存しているため、ネットワークファイルシステム (NFSなど) 上では正常に動作しません。同一ホスト上のプロセス間でのみ機能するように設計されています。
- 状況
-
SQLITE_BUSYエラーへの対処- PRAGMA busy_timeout の設定
最も基本的な対策です。SQLiteがロックされた場合に、操作をすぐに失敗させるのではなく、指定されたミリ秒間待機して再試行するように設定します。PRAGMA busy_timeout = 5000; -- 5秒間待機する例 - 書き込みトランザクションの短縮
SQLITE_BUSYは主に書き込みの競合によって発生するため、書き込みトランザクションをできるだけ短くすることが重要です。- 不必要な処理（例：ネットワーク呼び出し、複雑な計算）をトランザクション内で行わないようにします。
- 大きなバッチ処理は、小さなチャンクに分割して複数のトランザクションとして実行することを検討します。
- BEGIN IMMEDIATE の使用
書き込みトランザクションを開始することが確定している場合は、BEGIN IMMEDIATEを使用して、すぐに書き込みロックを取得するようにします。これにより、後からトランザクションタイプをアップグレードする際の競合を避けることができます。 - リード/ライト接続の分離
読み込みと書き込みを頻繁に行うアプリケーションでは、読み込み専用のデータベース接続と、書き込み専用のデータベース接続を分離することを検討します。これにより、読み込み処理が書き込みロックによってブロックされるのを防ぎやすくなります。 - アプリケーションレベルでのロック/キューイング
非常に高負荷な書き込みが予想される場合は、アプリケーションレベルでデータベースへの書き込みをシリアル化するためのキューやロックメカニズムを実装することも有効です。
- PRAGMA busy_timeout の設定
-
WAL/SHMファイルの肥大化/削除されない問題への対処
- PRAGMA wal_checkpoint(FULL); の明示的な実行
- アプリケーション終了時や、書き込みが落ち着いたタイミングで、明示的にチェックポイント処理を実行します。
- これにより、WALファイルの内容がデータベースファイルにマージされ、WALファイルがリセットまたは削除されます。
- 特に、常時アクティブなリーダーがいる環境でWALファイルが肥大化し続ける場合に有効です。チェックポイント処理中は、読み込みが一時的にブロックされる可能性がありますが、データの整合性を保つために重要です。
- アプリケーションの正常終了
データベース接続を常に適切に閉じるように、アプリケーションの終了処理を確認します。 - PRAGMA wal_autocheckpoint の確認
デフォルトで有効ですが、意図せず無効になっていないか、あるいは閾値が適切かを確認します。
- PRAGMA wal_checkpoint(FULL); の明示的な実行
-
データ整合性の問題への対処
- PRAGMA synchronous=FULL; の検討
最高のデータ堅牢性が求められる場合は、synchronous=FULLを使用します。ただし、パフォーマンスとのトレードオフがあります。 - 定期的なバックアップ
どのようなデータベース運用においても基本ですが、定期的なバックアップは必須です。WALモードでは、.db、.wal、.shmの3つのファイルをセットでバックアップする必要があります。
- PRAGMA synchronous=FULL; の検討
-
ネットワークファイルシステムでの使用禁止
- 根本的な解決
WALモードはネットワークファイルシステム上での使用をサポートしていません。NFSなどでSQLiteを使用する場合は、Rollback Journalモードに戻すか、ネットワーク経由でのデータベースアクセスを考慮した別のデータベース（PostgreSQL, MySQLなど）の使用を検討してください。
- 根本的な解決
- テスト環境での再現
問題が発生した場合は、できるだけシンプルなコードで問題を再現し、原因を特定することを目指します。 - SQLiteのバージョン
SQLiteのバージョンが古いと、WALモードの安定性やパフォーマンスが低い場合があります。常に最新の安定版を使用することをお勧めします。 - エラーメッセージの確認
SQLiteのエラーコード (SQLITE_BUSY,SQLITE_LOCKED,SQLITE_IOERRなど) を確認し、それぞれの意味を理解することがトラブルシューティングの第一歩です。
しかし、WALモードの特性を最大限に活かし、ロック関連のエラー（特にSQLITE_BUSY）を適切に処理するためのコーディングプラクティスは存在します。ここでは、Pythonを例に、これらのプラクティスを示します。他の言語でも基本的な考え方は同じです。
WALモードの有効化と接続
まず、データベースをWALモードで開く必要があります。
import sqlite3
import time
import threading
DATABASE_FILE = 'my_wal_database.db'
def get_db_connection():
conn = sqlite3.connect(DATABASE_FILE)
# WALモードを有効にする
conn.execute('PRAGMA journal_mode = WAL;')
# 通常、busy_timeoutはSQLITE_BUSYエラーを処理するために設定します。
# WALモードでも書き込みの競合が発生しうるため重要です。
conn.execute('PRAGMA busy_timeout = 5000;') # 5秒間待機
return conn
# データベースの初期化（初回のみ実行）
def initialize_database():
conn = get_db_connection()
try:
conn.execute('''
CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
content TEXT NOT NULL,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
);
''')
conn.commit()
except sqlite3.Error as e:
print(f"データベース初期化エラー: {e}")
finally:
conn.close()
if __name__ == '__main__':
initialize_database()
print(f"データベース '{DATABASE_FILE}' をWALモードで初期化しました。")
print(f"ジャーナルモード: {sqlite3.connect(DATABASE_FILE).execute('PRAGMA journal_mode;').fetchone()[0]}")
解説
PRAGMA busy_timeout = 5000;は、SQLITE_BUSYエラーが発生した場合に、SQLiteが5秒間再試行するように設定します。これは、WALモードでも複数の書き込みが同時に発生しようとした場合に、エラーではなく待機を促すために非常に重要です。PRAGMA journal_mode = WAL;がWALモードを有効にする最も重要な部分です。このプラグマは、データベース接続を開いた直後に実行する必要があります。
並行書き込みのシミュレーションと SQLITE_BUSY の処理
WALモードでは、同時に1つの書き込みトランザクションのみが可能です。複数のスレッド/プロセスが同時に書き込みを行おうとすると、SQLITE_BUSY エラーが発生する可能性があります。busy_timeout が設定されていれば、SQLiteは再試行を行いますが、タイムアウト後も取得できない場合はエラーになります。
import sqlite3
import time
import threading
DATABASE_FILE = 'my_wal_database.db'
def get_db_connection():
conn = sqlite3.connect(DATABASE_FILE)
conn.execute('PRAGMA journal_mode = WAL;')
conn.execute('PRAGMA busy_timeout = 5000;')
return conn
def writer_task(thread_id, num_writes):
conn = None
try:
conn = get_db_connection()
print(f"ライター {thread_id}: 接続確立")
for i in range(num_writes):
try:
# 明示的にトランザクションを開始し、コミットする
conn.execute('BEGIN IMMEDIATE;') # 書き込みロックをすぐに取得しようとする
message = f"メッセージ {i+1} from writer {thread_id}"
conn.execute('INSERT INTO messages (content) VALUES (?);', (message,))
conn.commit()
print(f"ライター {thread_id}: 成功 - '{message}'")
except sqlite3.OperationalError as e:
if "database is locked" in str(e):
print(f"ライター {thread_id}: ロック競合発生 (SQLITE_BUSY). 再試行します。")
# ここではbusy_timeoutが機能するはずですが、
# 非常に短い間隔で大量に書き込むと発生する可能性があります。
# アプリケーションによっては、より複雑なリトライロジックが必要になります。
conn.rollback() # ロールバックしてロックを解放
time.sleep(0.1) # 少し待ってから再試行
else:
print(f"ライター {thread_id}: その他のエラー - {e}")
conn.rollback()
break # エラー発生時は中断
except Exception as e:
print(f"ライター {thread_id}: 予期せぬエラー - {e}")
conn.rollback()
break
time.sleep(0.01) # 少し間隔を空ける
finally:
if conn:
conn.close()
print(f"ライター {thread_id}: 接続クローズ")
def reader_task(thread_id, num_reads):
conn = None
try:
conn = get_db_connection()
print(f"リーダー {thread_id}: 接続確立")
for i in range(num_reads):
try:
# 読み取りは共有ロックのみを必要とするため、通常ロックされにくい
cursor = conn.execute('SELECT COUNT(*) FROM messages;')
count = cursor.fetchone()[0]
print(f"リーダー {thread_id}: 現在のメッセージ数: {count}")
except sqlite3.OperationalError as e:
print(f"リーダー {thread_id}: 読み取りエラー - {e}")
time.sleep(0.05) # 少し間隔を空ける
finally:
if conn:
conn.close()
print(f"リーダー {thread_id}: 接続クローズ")
if __name__ == '__main__':
initialize_database() # データベースがWALモードであることを確認
# 複数のライターとリーダーを起動して競合をシミュレート
threads = []
num_writers = 3
num_readers = 5
writes_per_writer = 10
reads_per_reader = 20
print("\n--- スレッド開始 ---")
for i in range(num_writers):
thread = threading.Thread(target=writer_task, args=(i+1, writes_per_writer))
threads.append(thread)
thread.start()
for i in range(num_readers):
thread = threading.Thread(target=reader_task, args=(i+1, reads_per_reader))
threads.append(thread)
thread.start()
for thread in threads:
thread.join()
print("--- 全てのスレッド終了 ---")
# 最終的なメッセージ数を確認
conn = get_db_connection()
total_messages = conn.execute('SELECT COUNT(*) FROM messages;').fetchone()[0]
print(f"\n最終的なメッセージ総数: {total_messages}")
conn.close()
解説
- reader_task
- リーダーは基本的に
SQLITE_BUSYに遭遇しにくいことを示しています。これはWALモードの大きな利点です。リーダーはライターと並行して動作し、ライターがWALファイルに書き込んでいる間も、データベースファイルから古いスナップショットを読み取ることができます。
- リーダーは基本的に
- writer_task
conn.execute('BEGIN IMMEDIATE;'): これは重要です。BEGIN DEFERRED(デフォルト) では、最初の書き込み操作まで実際のロック取得が遅延されます。BEGIN IMMEDIATEは、トランザクション開始時にすぐに書き込みロック (WAL_WRITE_LOCK) の取得を試みます。これにより、ロックの競合をより早く検出し、busy_timeoutを利用して待機させることができます。try...except sqlite3.OperationalError:SQLITE_BUSY(PythonではOperationalErrorとして表現され、メッセージに "database is locked" が含まれる) が発生した場合の処理を示しています。ここでは、エラーを捕捉し、少し待ってから再試行する単純なロジックを実装しています。実際のアプリケーションでは、より洗練されたリトライ回数制限やバックオフ戦略を導入することもあります。conn.rollback(): エラーが発生した場合、トランザクションをロールバックして、取得していた可能性のあるロックを解放することが重要です。
チェックポイント処理の明示的な実行
WALファイル (.wal) の肥大化を防ぎ、データベースファイルにWALファイルの内容をマージするために、明示的なチェックポイント処理を行うことができます。これは、アプリケーションが終了する際や、書き込み操作が落ち着いたタイミングで実行すると良いでしょう。
import sqlite3
DATABASE_FILE = 'my_wal_database.db'
def get_db_connection():
conn = sqlite3.connect(DATABASE_FILE)
conn.execute('PRAGMA journal_mode = WAL;')
conn.execute('PRAGMA busy_timeout = 5000;')
return conn
def perform_checkpoint():
conn = None
try:
conn = get_db_connection()
print("チェックポイントを開始します...")
# FULL: WALファイルのすべての内容をデータベースファイルにマージします。
# SQLiteはWALファイルをリセットし、可能な限り縮小します。
conn.execute('PRAGMA wal_checkpoint(FULL);')
print("チェックポイントが完了しました。")
except sqlite3.Error as e:
print(f"チェックポイントエラー: {e}")
finally:
if conn:
conn.close()
if __name__ == '__main__':
# 既存のデータベースファイルが存在し、WALモードで開かれていることを前提とします。
# ここでは、前述のスクリプトで書き込みが行われた後を想定。
# チェックポイント前後のファイルサイズを確認
import os
db_size_before = os.path.getsize(DATABASE_FILE) if os.path.exists(DATABASE_FILE) else 0
wal_size_before = os.path.getsize(DATABASE_FILE + '-wal') if os.path.exists(DATABASE_FILE + '-wal') else 0
print(f"チェックポイント前: DBサイズ={db_size_before} bytes, WALサイズ={wal_size_before} bytes")
perform_checkpoint()
# チェックポイント後もファイルサイズを確認
db_size_after = os.path.getsize(DATABASE_FILE) if os.path.exists(DATABASE_FILE) else 0
wal_size_after = os.path.getsize(DATABASE_FILE + '-wal') if os.path.exists(DATABASE_FILE + '-wal') else 0
print(f"チェックポイント後: DBサイズ={db_size_after} bytes, WALサイズ={wal_size_after} bytes")
# WALファイルが非常に小さいか、削除されていることを確認
if wal_size_after < wal_size_before:
print("WALファイルは縮小またはリセットされました。")
if not os.path.exists(DATABASE_FILE + '-wal'):
print("WALファイルは削除されました。")
解説
- チェックポイント処理は、すべての読み取りリーダーが一時的にデータベースファイルから読み取るために停止する必要がある「リーダーギャップ」を必要とします。そのため、非常にアクティブなシステムでは、チェックポイントが完了するまでに時間がかかるか、失敗する可能性があります。
PRAGMA wal_checkpoint(FULL);：このプラグマは、WALファイルの内容をデータベースファイルに完全にマージするようSQLiteに指示します。これにより、WALファイルはサイズが縮小されるか、完全に空になります。
WALモードはSQLiteのパフォーマンスを向上させる強力な機能ですが、そのロックメカニズムを理解し、特にSQLITE_BUSY エラーを適切に処理するためのコーディングプラクティス（busy_timeout の設定、BEGIN IMMEDIATE の使用、リトライロジック、明示的なチェックポイント）を適用することが重要です。
SQLiteのWAL (Write-Ahead Logging) モードは、読み取りと書き込みの並行性を高めるための重要な機能です。ここでは、WALモードに関連するプログラミングの例を、PythonとNode.js (sqlite3ライブラリ) で示します。
WALモードの有効化
データベース接続を開いた後、PRAGMA journal_mode=WAL; コマンドを実行することでWALモードを有効にできます。この設定はデータベースファイルに永続的に保存されるため、一度設定すれば次回以降の接続で再度設定する必要はありません。
Pythonの例
import sqlite3
import os
import time
DATABASE_NAME = "my_wal_database.db"
def enable_wal_mode(conn):
"""WALモードを有効にする関数"""
try:
cursor = conn.cursor()
cursor.execute("PRAGMA journal_mode=WAL;")
result = cursor.fetchone()
print(f"ジャーナルモード設定結果: {result[0]}")
if result[0] != 'wal':
print("警告: WALモードへの切り替えに失敗した可能性があります。")
except sqlite3.Error as e:
print(f"WALモードの有効化中にエラーが発生しました: {e}")
def setup_database(conn):
"""データベースのテーブルをセットアップする関数"""
try:
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
email TEXT UNIQUE
);
""")
conn.commit()
print("テーブル 'users' がセットアップされました。")
except sqlite3.Error as e:
print(f"データベースのセットアップ中にエラーが発生しました: {e}")
if __name__ == "__main__":
# 既存のデータベースファイルを削除（テスト用）
if os.path.exists(DATABASE_NAME):
os.remove(DATABASE_NAME)
if os.path.exists(f"{DATABASE_NAME}-wal"):
os.remove(f"{DATABASE_NAME}-wal")
if os.path.exists(f"{DATABASE_NAME}-shm"):
os.remove(f"{DATABASE_NAME}-shm")
# データベース接続を開く
conn = sqlite3.connect(DATABASE_NAME)
# WALモードを有効にする
enable_wal_mode(conn)
# データベースのセットアップ
setup_database(conn)
conn.close()
print(f"{DATABASE_NAME} がWALモードで初期化されました。")
# データベースファイルとそのジャーナルファイルが存在することを確認
print(f"データベースファイル: {os.path.exists(DATABASE_NAME)}")
print(f"WALファイル: {os.path.exists(f'{DATABASE_NAME}-wal')}")
print(f"SHMファイル: {os.path.exists(f'{DATABASE_NAME}-shm')}")
Node.jsの例 (better-sqlite3ライブラリを使用)
better-sqlite3 は同期的なAPIを提供し、WALモードとの相性が良いとされています。
const Database = require('better-sqlite3');
const fs = require('fs');
const DATABASE_NAME = "my_wal_database.db";
// 既存のデータベースファイルを削除（テスト用）
if (fs.existsSync(DATABASE_NAME)) fs.unlinkSync(DATABASE_NAME);
if (fs.existsSync(`${DATABASE_NAME}-wal`)) fs.unlinkSync(`${DATABASE_NAME}-wal`);
if (fs.existsSync(`${DATABASE_NAME}-shm`)) fs.unlinkSync(`${DATABASE_NAME}-shm`);
try {
const db = new Database(DATABASE_NAME);
// WALモードを有効にする
db.pragma('journal_mode = WAL');
console.log(`ジャーナルモード設定結果: ${db.pragma('journal_mode', { simple: true })}`);
// データベースのテーブルをセットアップ
db.exec(`
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
email TEXT UNIQUE
);
`);
console.log("テーブル 'users' がセットアップされました。");
db.close();
console.log(`${DATABASE_NAME} がWALモードで初期化されました。`);
// データベースファイルとそのジャーナルファイルが存在することを確認
console.log(`データベースファイル: ${fs.existsSync(DATABASE_NAME)}`);
console.log(`WALファイル: ${fs.existsSync(`${DATABASE_NAME}-wal`)}`);
console.log(`SHMファイル: ${fs.existsSync(`${DATABASE_NAME}-shm`)}`);
} catch (e) {
console.error(`エラーが発生しました: ${e.message}`);
}
複数の接続による読み書きの並行処理のシミュレーション
WALモードの最大の利点は、複数の読み込み処理と1つの書き込み処理が同時に行えることです。SQLITE_BUSY エラーを避けるために、PRAGMA busy_timeout を設定することも重要です。
Pythonの例
import sqlite3
import threading
import time
import os
DATABASE_NAME = "my_wal_database.db"
# データベースをWALモードで初期化（前述のコードを実行しておくか、この関数で初期化）
def initialize_db_for_wal():
if os.path.exists(DATABASE_NAME):
os.remove(DATABASE_NAME)
if os.path.exists(f"{DATABASE_NAME}-wal"):
os.remove(f"{DATABASE_NAME}-wal")
if os.path.exists(f"{DATABASE_NAME}-shm"):
os.remove(f"{DATABASE_NAME}-shm")
conn = sqlite3.connect(DATABASE_NAME)
conn.execute("PRAGMA journal_mode=WAL;")
conn.execute("PRAGMA busy_timeout=5000;") # 5秒のタイムアウトを設定
conn.execute("""
CREATE TABLE IF NOT EXISTS data (
id INTEGER PRIMARY KEY AUTOINCREMENT,
value TEXT NOT NULL,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
);
""")
conn.commit()
conn.close()
print("データベースがWALモードで初期化されました。")
def writer_thread():
"""書き込みを行うスレッド"""
conn = sqlite3.connect(DATABASE_NAME)
conn.execute("PRAGMA busy_timeout=5000;") # 5秒のタイムアウトを設定
try:
for i in range(10):
value = f"データ {i}"
conn.execute("INSERT INTO data (value) VALUES (?)", (value,))
conn.commit()
print(f"Writer: データを挿入しました: '{value}'")
time.sleep(0.1) # 少し待機して他の操作を許可
except sqlite3.OperationalError as e:
print(f"Writerエラー: {e}")
finally:
conn.close()
def reader_thread(thread_id):
"""読み込みを行うスレッド"""
conn = sqlite3.connect(DATABASE_NAME)
conn.execute("PRAGMA busy_timeout=5000;") # 5秒のタイムアウトを設定
try:
for i in range(15): # 書き込みよりも多く読み込みを試みる
cursor = conn.cursor()
cursor.execute("SELECT id, value, timestamp FROM data ORDER BY id DESC LIMIT 5;")
rows = cursor.fetchall()
print(f"Reader {thread_id}: 読み込み結果 ({len(rows)}件): {rows}")
time.sleep(0.05) # 少し待機
except sqlite3.OperationalError as e:
print(f"Reader {thread_id}エラー: {e}")
finally:
conn.close()
if __name__ == "__main__":
initialize_db_for_wal()
writer = threading.Thread(target=writer_thread)
readers = [threading.Thread(target=reader_thread, args=(i,)) for i in range(3)] # 3つのリーダー
writer.start()
for r in readers:
r.start()
writer.join()
for r in readers:
r.join()
print("すべてのスレッドが終了しました。")
# 最終的なデータを確認
conn = sqlite3.connect(DATABASE_NAME)
cursor = conn.cursor()
cursor.execute("SELECT COUNT(*) FROM data;")
count = cursor.fetchone()[0]
print(f"最終的なデータ件数: {count}")
conn.close()
# WALファイルとSHMファイルがまだ存在するか確認（自動チェックポイントによる削除は後続の接続で起こる可能性がある）
print(f"WALファイルが残っているか: {os.path.exists(f'{DATABASE_NAME}-wal')}")
print(f"SHMファイルが残っているか: {os.path.exists(f'{DATABASE_NAME}-shm')}")
Node.jsの例 (better-sqlite3ライブラリを使用)
Node.jsのイベントループの性質上、スレッドではなく非同期関数やプロセス間通信で並行性をシミュレートします。ここでは、シンプルに非同期処理で複数のデータベース操作を行います。
const Database = require('better-sqlite3');
const fs = require('fs');
const DATABASE_NAME = "my_wal_database.db";
// データベースをWALモードで初期化
function initializeDbForWal() {
if (fs.existsSync(DATABASE_NAME)) fs.unlinkSync(DATABASE_NAME);
if (fs.existsSync(`${DATABASE_NAME}-wal`)) fs.unlinkSync(`${DATABASE_NAME}-wal`);
if (fs.existsSync(`${DATABASE_NAME}-shm`)) fs.unlinkSync(`${DATABASE_NAME}-shm`);
const db = new Database(DATABASE_NAME);
db.pragma('journal_mode = WAL');
db.pragma('busy_timeout = 5000'); // 5秒のタイムアウトを設定
db.exec(`
CREATE TABLE IF NOT EXISTS data (
id INTEGER PRIMARY KEY AUTOINCREMENT,
value TEXT NOT NULL,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
);
`);
db.close();
console.log("データベースがWALモードで初期化されました。");
}
async function writerProcess() {
const db = new Database(DATABASE_NAME);
db.pragma('busy_timeout = 5000'); // 5秒のタイムアウトを設定
try {
for (let i = 0; i < 10; i++) {
const value = `データ ${i}`;
db.prepare("INSERT INTO data (value) VALUES (?)").run(value);
console.log(`Writer: データを挿入しました: '${value}'`);
await new Promise(resolve => setTimeout(resolve, 100)); // 少し待機
}
} catch (e) {
console.error(`Writerエラー: ${e.message}`);
} finally {
db.close();
}
}
async function readerProcess(readerId) {
const db = new Database(DATABASE_NAME);
db.pragma('busy_timeout = 5000'); // 5秒のタイムアウトを設定
try {
for (let i = 0; i < 15; i++) {
const rows = db.prepare("SELECT id, value, timestamp FROM data ORDER BY id DESC LIMIT 5;").all();
console.log(`Reader ${readerId}: 読み込み結果 (${rows.length}件):`, rows);
await new Promise(resolve => setTimeout(resolve, 50)); // 少し待機
}
} catch (e) {
console.error(`Reader ${readerId}エラー: ${e.message}`);
} finally {
db.close();
}
}
async function runSimulation() {
initializeDbForWal();
const writerPromise = writerProcess();
const readerPromises = [];
for (let i = 0; i < 3; i++) {
readerPromises.push(readerProcess(i));
}
await Promise.all([writerPromise, ...readerPromises]);
console.log("すべてのプロセスが終了しました。");
// 最終的なデータを確認
const db = new Database(DATABASE_NAME);
const count = db.prepare("SELECT COUNT(*) FROM data;").get()['COUNT(*)'];
console.log(`最終的なデータ件数: ${count}`);
db.close();
// WALファイルとSHMファイルがまだ存在するか確認
console.log(`WALファイルが残っているか: ${fs.existsSync(`${DATABASE_NAME}-wal`)}`);
console.log(`SHMファイルが残っているか: ${fs.existsSync(`${DATABASE_NAME}-shm`)}`);
}
runSimulation();
WALファイルが肥大化するのを防ぐため、または変更をすぐにデータベース本体にマージしたい場合に、明示的にチェックポイントを実行できます。
Pythonの例
import sqlite3
import os
import time
DATABASE_NAME = "my_wal_database_ckpt.db"
def initialize_db_for_wal_and_insert_data():
if os.path.exists(DATABASE_NAME):
os.remove(DATABASE_NAME)
if os.path.exists(f"{DATABASE_NAME}-wal"):
os.remove(f"{DATABASE_NAME}-wal")
if os.path.exists(f"{DATABASE_NAME}-shm"):
os.remove(f"{DATABASE_NAME}-shm")
conn = sqlite3.connect(DATABASE_NAME)
conn.execute("PRAGMA journal_mode=WAL;")
conn.execute("""
CREATE TABLE IF NOT EXISTS logs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
message TEXT NOT NULL,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
);
""")
conn.commit()
# データを挿入
for i in range(100):
conn.execute("INSERT INTO logs (message) VALUES (?)", (f"ログメッセージ {i}",))
conn.commit()
conn.close()
print("データベースがWALモードで初期化され、データが挿入されました。")
def perform_checkpoint():
"""明示的にチェックポイントを実行する関数"""
print("\nチェックポイントを実行します...")
conn = sqlite3.connect(DATABASE_NAME)
# WALモードのチェックポイントは、複数のモードがあります。
# - FULL: 全てのWALエントリをDBにマージし、WALファイルをクリア（リーダーがいない場合に成功）
# - TRUNCATE: FULLと同様だが、WALファイルを切り詰めることを試みる
# - RESTART: 新しいWALファイルセッションを開始
# - PASSIVE: アクティブなリーダーをブロックせずに可能な限り多くのWALエントリをマージ
try:
# FULLチェックポイントを試行
conn.execute("PRAGMA wal_checkpoint(FULL);")
print("チェックポイント(FULL)が実行されました。")
except sqlite3.OperationalError as e:
print(f"チェックポイントエラー: {e} (他の接続がアクティブな可能性があります)")
finally:
conn.close()
if __name__ == "__main__":
initialize_db_for_wal_and_insert_data()
print(f"初期状態 - WALファイルが存在するか: {os.path.exists(f'{DATABASE_NAME}-wal')}")
# チェックポイントを実行
perform_checkpoint()
# チェックポイント後のWALファイルの状態を確認
print(f"チェックポイント後 - WALファイルが存在するか: {os.path.exists(f'{DATABASE_NAME}-wal')}")
# WALファイルが完全に削除されるには、最後の接続が閉じられるか、他のプロセスが存在しないリーダーギャップが必要な場合があります。
# この例では、一度DBを閉じてから再度開くと、WALファイルが削除される可能性があります。
conn = sqlite3.connect(DATABASE_NAME)
conn.close() # 接続を閉じることで自動チェックポイントがトリガーされることもあります
print(f"接続を閉じた後 - WALファイルが存在するか: {os.path.exists(f'{DATABASE_NAME}-wal')}")
const Database = require('better-sqlite3');
const fs = require('fs');
const DATABASE_NAME = "my_wal_database_ckpt.db";
function initializeDbForWalAndInsertData() {
if (fs.existsSync(DATABASE_NAME)) fs.unlinkSync(DATABASE_NAME);
if (fs.existsSync(`${DATABASE_NAME}-wal`)) fs.unlinkSync(`${DATABASE_NAME}-wal`);
if (fs.existsSync(`${DATABASE_NAME}-shm`)) fs.unlinkSync(`${DATABASE_NAME}-shm`);
const db = new Database(DATABASE_NAME);
db.pragma('journal_mode = WAL');
db.exec(`
CREATE TABLE IF NOT EXISTS logs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
message TEXT NOT NULL,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP
);
`);
// データを挿入
for (let i = 0; i < 100; i++) {
db.prepare("INSERT INTO logs (message) VALUES (?)").run(`ログメッセージ ${i}`);
}
db.close();
console.log("データベースがWALモードで初期化され、データが挿入されました。");
}
function performCheckpoint() {
console.log("\nチェックポイントを実行します...");
let db;
try {
db = new Database(DATABASE_NAME);
// PRAGMA wal_checkpoint() は結果を返すため、.all() を使う
const result = db.pragma('wal_checkpoint(FULL)');
// resultは通常 [{ result: 0 }] のような形式
console.log(`チェックポイント(FULL)が実行されました。結果:`, result);
} catch (e) {
console.error(`チェックポイントエラー: ${e.message} (他の接続がアクティブな可能性があります)`);
} finally {
if (db) db.close();
}
}
if (require.main === module) {
initializeDbForWalAndInsertData();
console.log(`初期状態 - WALファイルが存在するか: ${fs.existsSync(`${DATABASE_NAME}-wal`)}`);
performCheckpoint();
console.log(`チェックポイント後 - WALファイルが存在するか: ${fs.existsSync(`${DATABASE_NAME}-wal`)}`);
// 接続を閉じることで自動チェックポイントがトリガーされることもあります
const db = new Database(DATABASE_NAME);
db.close();
console.log(`接続を閉じた後 - WALファイルが存在するか: ${fs.existsSync(`${DATABASE_NAME}-wal`)}`);
}
ここでは、WALモードのロックに関連する問題への代替・補完的なアプローチをいくつか紹介します。
PRAGMA busy_timeout の適切な利用
これはWALモードのロック問題に対する最も基本的な解決策であり、代替というよりは「必須の補完」です。
- 考慮事項
タイムアウト値を過度に大きくすると、アプリケーションの応答性が低下する可能性があります。逆に小さすぎると、頻繁にエラーが発生します。ワークロードに合わせて適切な値を設定することが重要です。 - プログラミングの例 (Python)
import sqlite3 conn = sqlite3.connect('my_database.db') conn.execute("PRAGMA journal_mode=WAL;") conn.execute("PRAGMA busy_timeout = 5000;") # 5秒間待機 # これ以降の操作でロック競合が発生した場合、最大5秒待機する
トランザクションの短縮化と最適化
ロックの保持時間を短くすることは、競合を減らす最も効果的な方法です。
- プログラミングの例 (Python)
import sqlite3 def batch_insert(conn, data_list): cursor = conn.cursor() # 長いトランザクションを避けるため、小さなチャンクでコミット batch_size = 100 for i in range(0, len(data_list), batch_size): try: # 明示的にBEGIN IMMEDIATEで書き込みロックを早く取得 cursor.execute("BEGIN IMMEDIATE;") for item in data_list[i : i + batch_size]: cursor.execute("INSERT INTO my_table (value) VALUES (?)", (item,)) conn.commit() # コミットでロックを解放 print(f"Batch {i//batch_size + 1} committed.") except sqlite3.OperationalError as e: print(f"Batch insert error: {e}. Retrying or handling...") conn.rollback() # エラー時はロールバック # 必要に応じてリトライロジックを追加 time.sleep(0.01) # 少し待機して他のスレッド/プロセスに機会を与える # 複数スレッド/プロセスからの呼び出しを想定
リード/ライト接続の分離とキューイング
- プログラミングの例 (Python with queue module)
import sqlite3 import threading import queue import time # 書き込みリクエストを保持するキュー write_queue = queue.Queue() def writer_process(db_path, q): conn = sqlite3.connect(db_path) conn.execute("PRAGMA journal_mode=WAL;") conn.execute("PRAGMA busy_timeout = 5000;") try: while True: task = q.get() if task is None: # 終了シグナル break sql, params = task try: conn.execute(sql, params) conn.commit() print(f"Writer: コミットしました: {sql} {params}") except sqlite3.OperationalError as e: print(f"Writerエラー: {e}") conn.rollback() finally: q.task_done() finally: conn.close() def reader_process(db_path, reader_id): conn = sqlite3.connect(db_path) conn.execute("PRAGMA journal_mode=WAL;") conn.execute("PRAGMA busy_timeout = 5000;") try: for _ in range(10): # 複数回読み込みを試行 try: cursor = conn.cursor() cursor.execute("SELECT COUNT(*) FROM my_table") count = cursor.fetchone()[0] print(f"Reader {reader_id}: 現在のデータ件数: {count}") except sqlite3.OperationalError as e: print(f"Reader {reader_id}エラー: {e}") time.sleep(0.5) finally: conn.close() if __name__ == "__main__": db_file = 'isolated_rw.db' # DB初期化 conn_init = sqlite3.connect(db_file) conn_init.execute("PRAGMA journal_mode=WAL;") conn_init.execute("CREATE TABLE IF NOT EXISTS my_table (id INTEGER PRIMARY KEY, value TEXT);") conn_init.commit() conn_init.close() writer_thread = threading.Thread(target=writer_process, args=(db_file, write_queue)) reader_threads = [threading.Thread(target=reader_process, args=(db_file, i)) for i in range(2)] writer_thread.start() for r_thread in reader_threads: r_thread.start() # 書き込みリクエストをキューに追加 for i in range(5): write_queue.put(("INSERT INTO my_table (value) VALUES (?)", (f"Data {i}",))) time.sleep(0.1) # 終了シグナルを送信 write_queue.put(None) writer_thread.join() for r_thread in reader_threads: r_thread.join() print("シミュレーション終了") - 利点
読み取りプロセスは、書き込みロックにほとんど影響されず、自由に読み取りを行うことができます。書き込みはシリアル化されるため、ロック競合が内部で管理されます。
アプリケーションレベルでの排他制御（ロック）
- プログラミングの例 (Python with threading.Lock)
import sqlite3 import threading import time db_lock = threading.Lock() # アプリケーションレベルのロック def write_data(db_path, data): with db_lock: # ロックを取得 conn = sqlite3.connect(db_path) conn.execute("PRAGMA journal_mode=WAL;") conn.execute("PRAGMA busy_timeout = 5000;") try: conn.execute("INSERT INTO shared_table (value) VALUES (?)", (data,)) conn.commit() print(f"データ '{data}' を書き込みました。") except sqlite3.OperationalError as e: print(f"書き込みエラー: {e}") finally: conn.close() time.sleep(0.1) # ロック解放後、少し待機 def read_data(db_path): with db_lock: # ロックを取得（読み込みもシリアル化される） conn = sqlite3.connect(db_path) conn.execute("PRAGMA journal_mode=WAL;") conn.execute("PRAGMA busy_timeout = 5000;") try: cursor = conn.execute("SELECT value FROM shared_table ORDER BY id DESC LIMIT 1;") row = cursor.fetchone() print(f"読み込みデータ: {row[0] if row else 'なし'}") except sqlite3.OperationalError as e: print(f"読み込みエラー: {e}") finally: conn.close() time.sleep(0.05) # ロック解放後、少し待機 if __name__ == "__main__": db_file = 'app_locked.db' # DB初期化 conn_init = sqlite3.connect(db_file) conn_init.execute("PRAGMA journal_mode=WAL;") conn_init.execute("CREATE TABLE IF NOT EXISTS shared_table (id INTEGER PRIMARY KEY, value TEXT);") conn_init.commit() conn_init.close() threads = [] for i in range(5): t = threading.Thread(target=write_data, args=(db_file, f"Value {i}")) threads.append(t) t = threading.Thread(target=read_data, args=(db_file,)) threads.append(t) for t in threads: t.start() for t in threads: t.join() print("シミュレーション終了") - 考慮事項
デッドロックのリスクやパフォーマンスオーバーヘッドが増加する可能性があります。慎重な設計が必要です。 - 利点
WALモードのロックとは独立して、より粒度の細かい、あるいはアプリケーションロジックに合わせた排他制御が可能です。
- 考慮事項
学習コスト、運用コスト、データ移行、既存アプリケーションへの影響などを考慮する必要があります。 - 例
- PostgreSQL/MySQL
複数の書き込みクライアントからの同時アクセス、レプリケーション、クラスタリングなど、より堅牢な機能を提供します。 - NoSQLデータベース
Key-Valueストア、ドキュメントデータベースなど、特定のユースケースに最適化された並行処理モデルを持つものがあります。 - 分散SQLite（例：dqlite, rqlite）
これらはSQLiteの上に分散システムを構築し、複数のノードでWALモードの機能を拡張しようとする試みです。
- PostgreSQL/MySQL