pytestの並行実行で理解するスレッドセーフと排他制御

はじめに

pytestで複数のテストケースを、並行して実行させたい場合があります。
テストを高速化したい、一気に負荷をかけたい、などの目的のほか、
並行処理で下記のような衝突が起きないかチェックしたい場合もあります。

• 処理順序が期待とおり制御できず、同期が取れていない
• 同じリソースへの操作が競合し、データ不整合がおきる

TL;DR

• pytest-parallelプラグインを利用して、複数テストケースを並列・並行実行できる
• 複数テストケースを並行実行させ、スレッドセーフ(thread-safe)かをチェックできる
• スレッドアンセーフ(thread-unsafe)なプログラムを、排他制御を用いてスレッドセーフにできる

関連用語のおさらい

プロセス?スレッド?

• プロセスとは(process)

  • 実行中のプログラムインスタンスを指す
  • 複数プロセスは同じメモリ領域を共有しない(プロセス間通信の共有メモリを除く)
  • 通常、プロセスの処理は、ほかのプロセスに影響しない(プロセス間で同じリソースを共有する場合を除く)
  • 複数プロセスが並列処理されることをマルチプロセスと呼ぶ

• スレッドとは(thread)

  • スレッドは、実行中プログラムのCPUにおける処理単位を指す
    • ある時点で1つのコアに割り当てられるのは1つ処理のみ
    • 例えば、4コア環境の場合は、同時に4スレッドまで処理可能
  • スレッドはプロセスに含まれる、プロセス内の同じメモリ領域を共有する <- ここが競合の元
  • プロセス内に複数スレッドを持つ場合、マルチスレッドと呼ぶ

並列?並行?

• 並列とは(parallel)
  • 複数タスク(図示では三つ)が同時進行
  • どの時点においても進行中タスクが複数
  • 物理CPUの観点から、複数コアで同時並列実行している

• 並行とは(concurrent)
  • 複数タスク(図示では三つ)が交互に進行
  • どの時点においても進行中タスクは1つのみ(長いスパンでは、複数タスク同時進行に見える?)
  • 物理CPUの観点から、1つのコアでタスクの切替(context switch)による疑似的な並列実行

pytest-parallelプラグインを利用して、複数テストケースを並列・並行実行できる

• インストール

$ pip install pytest-parallel

• 並列実行例(マルチプロセス)

# 並列実行数(プロセス数)は4
$ pytest --workers 4

# 並列実行数(プロセス数)はCPUコア数分
$ pytest --workers auto

• 並行実行例(マルチスレッド)

# 並行実行数(スレッド数)は4
$ pytest --tests-per-worker 4

# 並列実行数(スレッド数)はCPUコア数分
$ pytest --tests-per-worker auto

複数テストケースを並行実行させ、スレッドセーフかチェックできる

• スレッドアンセーフのコード例

test_thread_unsafe.py

# グローバル変数は、プロセス間では共有されない、同じプロセス内のスレッド間では共有される
counter = 0

# カウンタ1は、グローバル変数counterを100万までインクリメント
def test_counter1():
    global counter
    counter = 0
    for _ in range(1000000):
        counter += 1
    # 期待値は、100万
    assert counter == 1000000

# カウンタ2は、グローバル変数counterを100万までインクリメント
def test_counter2():
    global counter
    counter = 0
    for _ in range(1000000):
        counter += 1
    # 期待値は、100万
    assert counter == 1000000

• まず、2プロセスで並列実行

結果、プロセス間でグローバル変数counterを共有しないため、
二つのテストケースはお互い影響せず、両方ともに成功します。

$ pytest --workers 2 test_thread_unsafe.py
============================= test session starts ============================= 
platform linux -- Python 3.9.7, pytest-7.1.2, pluggy-1.0.0
rootdir: /home/zhao/thread-safe
plugins: forked-1.4.0, parallel-0.1.1
collected 2 items
pytest-parallel: 2 workers (processes), 1 test per worker (thread)
....
============================= 2 passed in 0.12s =============================

• つぎ、2スレッドで並行実行

結果、二つのテストケースでグローバル変数counterを共有するため、
counter値がほかのスレッドにより不意に更新されるなど衝突が起きます。
二つのテストケースともに、期待値を得られていません。

$ pytest --tests-per-worker 2 test_thread_unsafe.py
============================= test session starts =============================
platform linux -- Python 3.9.7, pytest-7.1.2, pluggy-1.0.0
rootdir: /home/zhao/thread-safe
plugins: forked-1.4.0, parallel-0.1.1
collected 2 items
pytest-parallel: 1 worker (process), 2 tests per worker (threads)
FFFF
============================= FAILURES =============================
__________________________ test_counter2 __________________________

    def test_counter2():
        global counter
        counter = 0
        for _ in range(1000000):
            counter += 1
        # 期待値は、100万
>       assert counter == 1000000
E       assert 1428819 == 1000000

test_thread_unsafe.py:20: AssertionError
__________________________ test_counter1 __________________________

    def test_counter1():
        global counter
        counter = 0
        for _ in range(1000000):
            counter += 1
        # 期待値は、100万
>       assert counter == 1000000
E       assert 1343031 == 1000000

test_thread_unsafe.py:11: AssertionError
============================= short test summary info =============================
FAILED test_thread_unsafe.py::test_counter2 - assert 1428819 == 1000000
FAILED test_thread_unsafe.py::test_counter1 - assert 1343031 == 1000000
============================= 2 failed in 0.20s =============================

スレッドアンセーフなプログラムを、排他制御を用いてスレッドセーフにできる

二つのテストケース間で衝突が起きないようにしたいです。
排他制御を追加し、counter操作中はロックをかけます。

• スレッドセーフに修正したコード例

test_thread_safe.py

import threading

lock = threading.Lock()

# グローバル変数は、プロセス間では共有されない、同じプロセス内のスレッド間では共有される
counter = 0

# カウンタ1は、グローバル変数counterを100万までインクリメント
def test_counter1():
    global counter
    lock.acquire()  # 排他制御開始
    counter = 0
    for _ in range(1000000):
        counter += 1
    # 期待値は、100万
    assert counter == 1000000
    lock.release()  # 排他制御解除

# カウンタ2は、グローバル変数counterを100万までインクリメント
def test_counter2():
    global counter
    lock.acquire()  # 排他制御開始
    counter = 0
    for _ in range(1000000):
        counter += 1
    # 期待値は、100万
    assert counter == 1000000
    lock.release()  # 排他制御解除

• 再度、2スレッドで並行実行

結果、グローバル変数counterの操作時に排他制御がかかり、
二つのテストケース間で衝突がおきず、両方成功しました。

$ pytest --tests-per-worker 2 test_thread_safe.py
============================= test session starts =============================
platform linux -- Python 3.9.7, pytest-7.1.2, pluggy-1.0.0
rootdir: /home/zhao/thread-safe
plugins: forked-1.4.0, parallel-0.1.1
collected 2 items
pytest-parallel: 1 worker (process), 2 tests per worker (threads)
....
============================= 2 passed in 0.17s =============================

おわりに

pytestを並行実行させ、スレッドセーフかチェックしてみました。
テストケース作成時は、並列実行・並行実行も考慮したほうがよさそうです。