一覧DB-08-3 · トランザクションと同時実行

同時更新とロック競合

数量10からAが1個、Bが2個を引く操作を、まず上書きで失敗させ、版番号条件と読み直しで修正します。SQLiteの書込みロック競合を確認した後、行ロックの循環待ちをPythonの待機表で表し、取消しと取得順の統一を考えます。

説明

二つの処理が同じ数量を読み、それぞれ古い値から計算して上書きすると、一方の変更が失われることがあります。読んだ版番号を更新条件に含める方法や、書込みのロックを先に取る方法で、競合を検出できます。

1数量と版番号を二接続で読む

A・Bは同じファイルを開き、数量10・版1を読みます。

import sqlite3, tempfile
from pathlib import Path
tmp = tempfile.TemporaryDirectory()
path = Path(tmp.name)/'conflict.db'
a = sqlite3.connect(path, isolation_level=None, timeout=0)
b = sqlite3.connect(path, isolation_level=None, timeout=0)
a.executescript('CREATE TABLE stock(id INTEGER PRIMARY KEY,

qty INTEGER,version INTEGER数量と変更の版

); INSERT INTO stock VALUES(1,10,1);')

qa,vaAが読んだ値

 = a.execute('SELECT qty,version FROM stock WHERE id=1').fetchone()

qb,vbBが読んだ値

 = b.execute('SELECT qty,version FROM stock WHERE id=1').fetchone()
print('A:', qa,va,'B:',qb,vb)

両方とも同じ10と1を持っています。ここでは各SELECTが独立して終わっており、長い読み取りトランザクションはありません。次の更新で、この読み取った値を使います。

2古い値で上書きすると減算が失われる

Aは10-1、Bは10-2を計算して順に保存します。

a.execute('UPDATE stock SET qty=?,version=version+1 WHERE id=1',(qa-1,)Aの古い値から計算

)
b.execute('UPDATE stock SET qty=?,version=version+1 WHERE id=1',

(qb-2,)Bも古い値から計算

)
print(a.execute('SELECT qty,version FROM stock').fetchone())

数量は8、版は3です。Aの1個減算後の9を、Bが古い10から計算した8で上書きしました。本来の10-1-2=7ではなく、一方の変更が結果へ反映されていません。

3同じ初期値から版条件付きで更新する

10・版1へ戻し、Aの更新に読取り時の版を条件として付けます。

a.execute('UPDATE stock SET qty=10,version=1 WHERE id=1')
qa,va = a.execute('SELECT qty,version FROM stock WHERE id=1').fetchone()
qb,vb = b.execute('SELECT qty,version FROM stock WHERE id=1').fetchone()
changed = a.execute('UPDATE stock SET qty=?,version=version+1 WHERE id=1

AND version=?読み取った版が現在も一致する条件',(qa-1,va)Aの更新値と期待する版).rowcount変更対象の件数


print('A changed:',changed)

Aは一行を変更し、数量9・版2になります。WHERE version=?で、読んだ時の版1が今も変わっていないことを更新時に検査しています。

4Bの古い版による更新を拒む

Bは版1を条件に更新します。

changed = b.execute('UPDATE stock SET qty=?,version=version+1 WHERE id=1 AND version=?',(qb-2,vb)Bが持つ古い版1

).rowcount

print('B changed:'零件を競合として確認

,changed)
print(b.execute('SELECT qty,version FROM stock').fetchone())

Bの変更件数は0で、数量9・版2が残ります。SQLエラーではありませんが、版の不一致により更新できませんでした。件数を見ずに成功扱いすると、競合を取りこぼします。

5最新の値を読み直して再計算する

Bが数量9・版2を取得し、2個の減算をやり直します。

qb,vb = b.execute('SELECT qty,version最新値を取得

 FROM stock WHERE id=1').fetchone()
changed = b.execute('UPDATE stock SET qty=?,version=version+1 WHERE id=1 AND version=?',

(qb-2,vb)最新値と最新の版で再計算

).rowcount
print('B retry:',changed,b.execute('SELECT qty,version FROM stock').fetchone())

一行が変更され、数量7・版3です。古い8を再送したのではなく、新しい9を基に7を計算しました。再試行回数を制限し、外部への通知などを二重に行わない設計も合わせて考えます。

6Aが書込みロックを先に取る

AをBEGIN IMMEDIATEし、Bも同じ開始を試します。

a.execute('BEGIN IMMEDIATE')Aが書込みを開始


try:

b.execute('BEGIN IMMEDIATE')Bが同じ権利を要求


except sqlite3.OperationalError as error:
    print(

error.sqlite_errorname競合のエラーコード

+':',error)
print('A active:',a.in_transaction,'B active:',b.in_transaction)

BはSQLITE_BUSY、Aは処理中、Bは開始できていません。SQLiteの書込みは一度に一接続なので競合します。これは待機関係が循環したデッドロックの証拠ではなく、既に書込みを持つAとの競合です。

7Aの終了後にBが再試行する

Aを確定してから、Bが一個の減算を行います。

a.execute('COMMIT')先の保持者が終了

b.execute('BEGIN IMMEDIATE')Bが取得をやり直す


b.execute('UPDATE stock SET

qty=qty-1取得後に現在値から更新

,version=version+1 WHERE id=1')
b.execute('COMMIT')
print(b.execute('SELECT qty,version FROM stock').fetchone())

数量6・版4です。ロックの取得に失敗した場合でも、保持者が終了した後なら進めます。待ち時間や再試行は、競合原因が解消する条件と回数の上限を含めて設計します。

8二資源の循環待ちを表す

行ごとの排他ロックを持つ場合を、資源の所有表と待機表で表します。

owners = {'row_x':'A','row_y':'B'}現在のロック保持者

waiting = {'A':'row_y','B':'row_x'}次に欲しい資源


wait_for = {tx:

owners[resource]待つ相手の処理へ変換

 for tx,resource in waiting.items()}
print('owners:',owners)
print('wait_for:',wait_for)

AはBのrow_yを待ち、BはAのrow_xを待つので、A→B→Aの輪ができています。これは行ロックの循環を説明する待機モデルです。先ほどのSQLiteの一書込み競合と区別して、どの資源を誰が保持するかを明示しています。

9待機先をたどって循環を検出する

Aから待機先をたどり、再び同じ処理へ到達するか調べます。

path, current = [], 'A'
while current in wait_for and

current not in path既に通った処理への到達を検出

:
    path.append(current)

current = wait_for[current]保持者をたどる

cycle =循環部分を取り出す

 path[path.index(current):]+[current] if current in path else []
print('cycle:',cycle)

A・B・Aの循環が見つかります。実際のDBはデッドロックを検出すると、一方のトランザクションを中止するなどして循環を解消します。どの処理が選ばれるかに依存したアプリケーションにせず、中止された処理を最初から再試行できるようにします。

10一方を取り消して資源を解放する

Bを取り消した後、Aが必要な資源を得て終了する順序を作ります。

del waiting['B']

del owners['row_y']中止したBの資源を解放


owners['row_y']='A'
del waiting['A']
print('A continues:',owners,'waiting:',waiting)

owners.clear()Aの終了で解放

owners['row_x']='B'Bが統一した順で再取得


owners['row_y']='B'
print('B retries:',owners)

Bの取消しでrow_yを解放するとAが進め、Aの終了後にBをやり直せます。複数資源を常にrow_x→row_yの順で取れば、この逆順取得による循環を避けられます。取消し後はロックだけでなく、トランザクションの変更も戻った状態から再開します。

まとめ

読取り後の上書きは、版番号条件と変更件数で競合を検出できます。

再試行では最新値を読み直して再計算し、古い計算結果をそのまま適用しません。

単なるロック競合と循環待ちを分け、取消し範囲と資源取得順を設計します。

操作履歴と解説を振り返れます。

Python / SQLite

import sqlite3, tempfile
from pathlib import Path
tmp = tempfile.TemporaryDirectory()
path = Path(tmp.name)/'conflict.db'
a = sqlite3.connect(path, isolation_level=None, timeout=0)
b = sqlite3.connect(path, isolation_level=None, timeout=0)
a.executescript('CREATE TABLE stock(id INTEGER PRIMARY KEY,qty INTEGER,version INTEGER); INSERT INTO stock VALUES(1,10,1);')
qa,va = a.execute('SELECT qty,version FROM stock WHERE id=1').fetchone()
qb,vb = b.execute('SELECT qty,version FROM stock WHERE id=1').fetchone()
print('A:', qa,va,'B:',qb,vb)

A: 10 1 B: 10 1

a.execute('UPDATE stock SET qty=?,version=version+1 WHERE id=1',(qa-1,))
b.execute('UPDATE stock SET qty=?,version=version+1 WHERE id=1',(qb-2,))
print(a.execute('SELECT qty,version FROM stock').fetchone())

(8, 3)

a.execute('UPDATE stock SET qty=10,version=1 WHERE id=1')
qa,va = a.execute('SELECT qty,version FROM stock WHERE id=1').fetchone()
qb,vb = b.execute('SELECT qty,version FROM stock WHERE id=1').fetchone()
changed = a.execute('UPDATE stock SET qty=?,version=version+1 WHERE id=1 AND version=?',(qa-1,va)).rowcount
print('A changed:',changed)

A changed: 1

changed = b.execute('UPDATE stock SET qty=?,version=version+1 WHERE id=1 AND version=?',(qb-2,vb)).rowcount
print('B changed:',changed)
print(b.execute('SELECT qty,version FROM stock').fetchone())

B changed: 0
(9, 2)

qb,vb = b.execute('SELECT qty,version FROM stock WHERE id=1').fetchone()
changed = b.execute('UPDATE stock SET qty=?,version=version+1 WHERE id=1 AND version=?',(qb-2,vb)).rowcount
print('B retry:',changed,b.execute('SELECT qty,version FROM stock').fetchone())

B retry: 1 (7, 3)

a.execute('BEGIN IMMEDIATE')
try:
    b.execute('BEGIN IMMEDIATE')
except sqlite3.OperationalError as error:
    print(error.sqlite_errorname+':',error)
print('A active:',a.in_transaction,'B active:',b.in_transaction)

SQLITE_BUSY: database is locked
A active: True B active: False

a.execute('COMMIT')
b.execute('BEGIN IMMEDIATE')
b.execute('UPDATE stock SET qty=qty-1,version=version+1 WHERE id=1')
b.execute('COMMIT')
print(b.execute('SELECT qty,version FROM stock').fetchone())

(6, 4)

owners = {'row_x':'A','row_y':'B'}
waiting = {'A':'row_y','B':'row_x'}
wait_for = {tx:owners[resource] for tx,resource in waiting.items()}
print('owners:',owners)
print('wait_for:',wait_for)

owners: {'row_x': 'A', 'row_y': 'B'}
wait_for: {'A': 'B', 'B': 'A'}

path, current = [], 'A'
while current in wait_for and current not in path:
    path.append(current)
    current = wait_for[current]
cycle = path[path.index(current):]+[current] if current in path else []
print('cycle:',cycle)

cycle: ['A', 'B', 'A']

del waiting['B']
del owners['row_y']
owners['row_y']='A'
del waiting['A']
print('A continues:',owners,'waiting:',waiting)
owners.clear()
owners['row_x']='B'
owners['row_y']='B'
print('B retries:',owners)

A continues: {'row_x': 'A', 'row_y': 'A'} waiting: {}
B retries: {'row_x': 'B', 'row_y': 'B'}

演習完了