この記事のポイント

  • 排他制御で発生しやすい設計肥大の構造的原因を整理
  • ロック責務をレビューアーが読み解く具体的な判断軸を提示
  • デッドロック・粒度設計・一貫性保証を具体例で徹底整理

そもそも排他制御とは

マルチスレッドや並行処理において、同時に同じリソースへアクセスすると競合や矛盾が発生します。
これを防ぐ仕組みが排他制御（Mutual Exclusion）です。

Pythonでは主に threading.Lock() を用いて排他制御を行います。

基本的な排他制御例 
import threading

lock = threading.Lock()

def critical_section():
    with lock:
        # ここが排他区間
        pass

排他制御は強力ですが、安易に導入すると設計肥大に直結します。レビューでは「とりあえずロックを巻く」を抑制し、設計責務の整理が重要となります。

なぜこれをレビューするのか

ロック設計は設計崩壊の温床になりやすく、以下のような事象を頻発させます。

  • ロック対象が肥大化して凝集度が低下
  • 責務境界が消失し全体設計が把握困難に
  • デッドロック経路の複雑化
  • ロック解放漏れによるスタック

レビューではロック設計＝責務分離設計と捉え、構造的問題を発見します。

レビューアー視点

  • ロック範囲と責務範囲が一致しているか
  • ロック粒度が粗すぎ・細かすぎになっていないか
  • ロック取得順序は一貫しているか
  • 例外時にロック解放責任は保証されているか
  • 読み取り専用処理にまでロックを強制していないか

開発者視点

開発者は「状態を守る手段」としてのロックを利用しがちですが、レビューでは「設計を整理する補助線」としてのロック設計が求められます。

良い実装例

正常設計例：ロック責務分離 
import threading

class Inventory:
    def __init__(self):
        self._lock = threading.Lock()
        self._stock = {}

    def add_item(self, item_id: str, count: int):
        with self._lock:
            self._stock[item_id] = self._stock.get(item_id, 0) + count

    def remove_item(self, item_id: str, count: int) -> bool:
        with self._lock:
            current = self._stock.get(item_id, 0)
            if current < count:
                return False
            self._stock[item_id] = current - count
            return True

    def get_stock(self, item_id: str) -> int:
        with self._lock:
            return self._stock.get(item_id, 0)

良い理由

  • データ構造（在庫表）に対してロック責任が集中
  • 読み取りも統一的にロックを取得し一貫性確保
  • 外部API側にロック責任が露出しない
  • 粒度がInventory単位で適切に分離

レビュー観点

  • ロック範囲とクラス内部状態の責任境界は一致しているか
  • 複数リソースを跨ぐロック順序は存在しないか
  • 例外発生時にロックは常に解放保証されるか
  • 読み取り専用での無駄なロック強制はないか
  • 外部にロック露出して呼び出し順序依存を生んでいないか

良くない実装例: ケース1

問題例: 呼び出し側にロック責任が漏洩 
import threading

class Inventory:
    def __init__(self):
        self._lock = threading.Lock()
        self._stock = {}

    def _add_item(self, item_id: str, count: int):
        self._stock[item_id] = self._stock.get(item_id, 0) + count

inventory = Inventory()

def caller():
    inventory._lock.acquire()

@Reviewer
呼び出し側がロック責任を負っています。ロック設計はカプセル化すべきです。内部に閉じ込めましょう。
    try:
        inventory._add_item("itemA", 10)
    finally:
        inventory._lock.release()

問題点

  • ロックが外部に露出し責任境界が崩壊
  • 呼び出し順序依存を生む
  • 実装者全員がロック責任を負う設計になる

改善例

改善後: 内部ロックカプセル化 
class Inventory:
    def add_item(self, item_id: str, count: int):
        with self._lock:
            self._stock[item_id] = self._stock.get(item_id, 0) + count
設計補足
  • ロックは常に内部実装責任として管理
  • 呼び出し側は状態操作APIのみを意識するだけでよくなる

良くない実装例: ケース2

問題例: 粒度肥大とリーダーロック未分離 
import threading

class Inventory:
    def __init__(self):
        self._lock = threading.Lock()
        self._stock = {}

    def get_all_items(self):
        with self._lock:
            return self._stock.copy()

    def clear_all(self):
        with self._lock:
            self._stock.clear()

問題点

  • 読み取り処理で全体ロックを毎回取得
  • 集計処理が長くなると全処理ブロック化
  • リーダー/ライター分離設計が未整理

改善例

改善後: 読み取り分離方針 
import threading

class Inventory:
    def __init__(self):
        self._lock = threading.RLock()  # 今回は例示上RLock利用
        self._stock = {}

    def add_item(self, item_id: str, count: int):
        with self._lock:
            self._stock[item_id] = self._stock.get(item_id, 0) + count

    def snapshot(self):
        with self._lock:
            return dict(self._stock)  # 読み取り用スナップショット返却
設計補足
  • 読み取り専用はスナップショット戦略でロック長時間保持回避
  • 排他は状態変化操作時に集中設計
  • リーダー/ライター分離設計の典型整理法

良くない実装例: ケース3

問題例: 複数ロック間の取得順序崩壊 
import threading

class Account:
    def __init__(self, name, balance):
        self.name = name
        self.balance = balance
        self._lock = threading.Lock()

def transfer(a1: Account, a2: Account, amount: int):
    with a1._lock:

@Reviewer
複数ロックを順序指定せずに取得しています。相互待ちが発生する危険があります。ロック順序ルールを統一しましょう。
        with a2._lock:
            if a1.balance >= amount:
                a1.balance -= amount
                a2.balance += amount

問題点

  • ロック順序が呼び出し順序依存
  • 相互待ちによるデッドロック可能性

改善例

改善後: ロック順序統一設計 
def transfer(a1: Account, a2: Account, amount: int):
    first, second = sorted([a1, a2], key=lambda acc: acc.name)
    with first._lock:
        with second._lock:
            if a1.balance >= amount:
                a1.balance -= amount
                a2.balance += amount
設計補足
  • ロック取得順序を事前決定して統一
  • 競合時の待機順序崩壊を防止

PlantUML設計イメージ

UML Diagram
UML Diagram

観点チェックリスト

まとめ

排他制御設計の崩壊は、責務境界の曖昧化から静かに進行します。レビューアーは「ロック粒度の崩壊は設計の肥大」を常に読み取り、状態単位での責務集約設計・外部依存の排除・一貫した取得順序の3本柱でコード構造を健全に整理する支援役として立ち回ることが重要です。