はじめに

トランザクションやらコミットのこと無知な状態でマルチスレッドでsqliteやろうとしたらデッドロックしたのでメモ

threading.Event()でトランザクション前にスレッドを待機させる

import threading
import random
import time

# event3の中でevent1→event2の順でイベントが実行される
# event1はランダム秒後に開始する、event3の終了を待つ,終了したらevent3_endをclear
# event2はevent1が開始するまで待機状態
# event3はevent1が開始したらevent1をclearにしevent2を待つ
# event3でevent2が開始したらevent2をclearにしevent3の終了を告げる
# 上記繰り返し

test_event1_begin = threading.Event()
test_event2_begin = threading.Event()
test_event3_end = threading.Event()


def test_sql1():
    global test_event3_end
    while True:
        interval = random.uniform(1, 3)
        print(f'evet1　{interval}sec 後に開始')
        time.sleep(interval)
        # event1を開始
        test_event1_begin.set()
        # event3が終了するまで待機
        test_event3_end.wait()
        test_event3_end.clear()


def test_sql2():
    global test_event1_end, test_event2_end
    while True:
        time.sleep(0.001)
        # event1を待機
        test_event1_begin.wait()
        interval = random.uniform(1, 3)
        print(f'evet2　{interval}sec 後に開始', end='\n')
        time.sleep(interval)
        # event2を開始
        test_event2_begin.set()


def test_sql3():
    global test_event1_end, test_event2_end, test_event3_end
    while True:
        time.sleep(0.001)
        test_event1_begin.wait()
        test_event1_begin.clear()
        test_event2_begin.wait()
        test_event2_begin.clear()
        print('event3終了')
        time.sleep(1)
        test_event3_end.set()


def main():
    thread_list = []
    tasks = [test_sql1, test_sql2, test_sql3]
    for task in tasks:
        th = threading.Thread(target=task, args=())
        th.start()
        thread_list.append(th)
    for th in thread_list:
        th.join()


if __name__ == '__main__':
    main()

evet1　1.198280231761349sec 後に開始
evet2　1.606435444539516sec 後に開始
event3終了
evet1　2.7041704830942663sec 後に開始
evet2　1.2936549765944632sec 後に開始
event3終了
evet1　1.3772512376025186sec 後に開始
evet2　1.7498730389596326sec 後に開始
event3終了

結論

上記のように別スレッドからEventフラグを使ってスレッド管理するのもありですが、トランザクション処理は単一のスレッドで行ったほうが管理しやすいかなと思いました。パフォーマンスもこっちのほうがいいのでは？

# 以下、DBに反映させたいデータ
q1 = []
q2 = []
q3 = []

def th1():
    while True:
        # q1キューにデータを入れる処理
        q1.append('any')

def th2():
    while True:
        # q2キューにデータを入れる処理
        q2.append('any')

def th3():
    while True:
        # q3キューにデータを入れる処理
        q3.append('any')

def any_transaction():
    global q1,q2,q3
    while True:
        time.sleep(0.001)
        # キューにデータがあれば順次する
        if q1:
            # q1をdbに入れる処理
            pass
        if q1:
            # q1をdbに入れる処理
            pass
        if q1:
            # q1をdbに入れる処理
            pass