グローバルインタプリタロック（GIL）のメカニズム
CPython実装におけるGIL（Global Interpreter Lock）は、本質的にはインタプリタレベルのミューテックスロックです。このロックの存在により、同一プロセス内の複数のネイティブスレッドが同時にPythonバイトコードを実行することが制限されます。すなわち、Pythonのマルチスレッドは真の並列処理（Parallelism）ではなく、コンテキストスイッチングによる並行処理（Concurrency）を実現します。
すべてのスレッドは、自身のタスクを実行するためにCPythonインタプリタのコードにアクセスする必要があります。インタプリタの内部状態（特にメモリ管理とガベージコレクションの機構）はスレッドセーフに設計されていないため、データ競合を防ぐ唯一の手段として、インタプリタへのアクセス権をGILによって排他制御しています。
GILとユーザー定義ロック（Lock）の役割の違い
アプリケーション開発において、threading.Lock のようなユーザー定義のロックが依然として必要になる理由を理解するには、GILとLockの保護対象の違いを把握する必要があります。
- GIL: インタプリタの内部状態を保護し、バイトコード実行の排他制御を行います。
- Lock: アプリケーションレベルの共有データ（変数、リスト、辞書など）に対する読み書きの整合性を保証します。
あるスレッドがGILを取得して実行を開始し、その後ユーザー定義のLockを獲得したとします。このスレッドがまだLockを解放していない段階で、OSのスケジューリングによってGILが別のスレッドに渡ることがあります。次のスレッドはGILを取得しますが、目的のデータにアクセスするためのLockが解放されていないため、ブロックされGILを即座に手放します。これにより、共有データへのアクセスは結果的に直列化されます。
join() メソッドも処理を直列化しますが、これはスレッドのライフサイクル全体を待機させるものです。対してLockは、共有リソースを操作する「クリティカルセクション」のみを保護するため、システム全体のスループットを維持しつつデータ安全性を確保できます。
マルチスレッドとマルチプロセスの最適な使い分け
Pythonの並行処理モデルを選択する際は、タスクの性質を「CPUバウンド（計算密集型）」と「I/Oバウンド（I/O密集型）」に分類して検討します。
- CPUバウンド: 複雑な数学的計算や画像処理など、CPUリソースを常に消費するタスク。マルチコア環境では、GILの制約を受けない
multiprocessingを使用してプロセスレベルの並列処理を行うのが最適です。 - I/Oバウンド: ネットワークリクエスト、データベースクエリ、ファイル読み書きなど、外部リソースの応答を待つ時間が長いタスク。待機中にGILが解放されるため、リソースオーバーヘッドの少ない
threadingが効果的です。
import time
import multiprocessing
import threading
def cpu_heavy_task(identifier):
"""CPUリソースを消費する計算処理"""
result = sum(i * i for i in range(5 * 10**6))
return result
def io_blocking_task(identifier):
"""外部リソースへのアクセスをシミュレート"""
time.sleep(0.5)
def execute_benchmark(target_func, use_multiprocessing=False):
start_time = time.perf_counter()
workers = []
WorkerClass = multiprocessing.Process if use_multiprocessing else threading.Thread
for i in range(4):
worker = WorkerClass(target=target_func, args=(i,))
workers.append(worker)
worker.start()
for worker in workers:
worker.join()
elapsed = time.perf_counter() - start_time
mode = "マルチプロセス" if use_multiprocessing else "マルチスレッド"
print(f"[{mode}] 完了時間: {elapsed:.3f}秒")
if __name__ == '__main__':
print("--- CPUバウンドタスクのテスト ---")
execute_benchmark(cpu_heavy_task, use_multiprocessing=False) # 遅い
execute_benchmark(cpu_heavy_task, use_multiprocessing=True) # 速い
print("\n--- I/Oバウンドタスクのテスト ---")
execute_benchmark(io_blocking_task, use_multiprocessing=False) # 速い
execute_benchmark(io_blocking_task, use_multiprocessing=True) # プロセス生成のオーバーヘッドでわずかに遅い
デッドロックと再帰的ロック（RLock）
複数のスレッドが、互いにが保持しているリソースの解放を永久に待ち続ける状態をデッドロックと呼びます。これは通常、複数のロックを異なる順序で取得しようとした場合に発生します。
Pythonでは、同一スレッド内で同じロックを複数回取得する必要がある場合に備え、threading.RLock（再帰的ロック）を提供しています。RLockは内部に「所有者スレッドID」と「再帰カウンター」を保持しており、同じスレッドからの acquire() 呼び出しをカウンターを増やして許可し、release() が同じ回数呼ばれてカウンターがゼロになった時点で、他のスレッドにロックを譲渡します。
import threading
import time
class SecureVault:
def __init__(self):
# 通常のLock()を使用すると、同じスレッド内での再帰呼び出しでデッドロックが発生する
self._lock = threading.RLock()
self._items = []
def add_item(self, item):
with self._lock:
self._items.append(item)
self._audit_log(f"Added: {item}")
def _audit_log(self, action):
# 同一スレッド内で再度ロックを要求する
with self._lock:
print(f"[監査ログ] {action} - 現在のアイテム数: {len(self._items)}")
def process_batch(self, items):
with self._lock:
for item in items:
self.add_item(item)
vault = SecureVault()
vault.process_batch(["Gold", "Silver", "Diamond"])
セマフォ（Semaphore）による同時接続数の制御
セマフォは、内部カウンターによって管理される高度なロック機構です。acquire() が呼ばれるとカウンターがデクリメントされ、release() でインクリメントされます。カウンターがゼロに達すると、それ以上の acquire() 呼び出しは、他のスレッドが release() を実行するまでブロックされます。データベース接続プールやAPIレート制限の実装に適しています。
import threading
import time
import random
class DatabaseConnectionPool:
def __init__(self, max_connections):
self.semaphore = threading.Semaphore(max_connections)
def execute_query(self, query_id):
self.semaphore.acquire()
try:
print(f"DB接続獲得: クエリID {query_id} の実行を開始")
time.sleep(random.uniform(0.5, 1.5)) # クエリ実行時間をシミュレート
finally:
print(f"DB接続解放: クエリID {query_id} が完了")
self.semaphore.release()
pool = DatabaseConnectionPool(max_connections=2)
threads = [threading.Thread(target=pool.execute_query, args=(i,)) for i in range(5)]
for t in threads:
t.start()
Eventオブジェクトを活用したスレッド間シグナリング
threading.Event は、スレッド間でシンプルな状態フラグを共有するためのプリミティブです。内部のフラグはデフォルトでFalseに設定されており、set() でTrueに、clear() でFalseに戻せます。wait() メソッドを呼び出したスレッドは、フラグがTrueになるまで実行を一時停止します。
import threading
import time
def initialize_system(ready_signal):
print("システム初期化プロセスを開始...")
time.sleep(2) # 初期化処理
print("システム準備完了。ワーカーを起動します。")
ready_signal.set()
def worker_node(node_id, ready_signal):
print(f"ノード {node_id}: システム準備完了を待機中...")
ready_signal.wait()
print(f"ノード {node_id}: リクエストの受信を開始")
system_ready = threading.Event()
# 初期化スレッド
init_thread = threading.Thread(target=initialize_system, args=(system_ready,))
init_thread.start()
# ワーカースレッド群
for i in range(3):
threading.Thread(target=worker_node, args=(i, system_ready)).start()
スレッドセーフなキュー（queueモジュール）
マルチスレッド環境でデータを安全に受け渡すには、独自にLockを実装するよりも queue モジュールを使用するのが標準的です。このモジュールは内部で必要なロック処理を自動的に行います。
- Queue: 最初に追加された要素が最初に取り出される（FIFO）標準的なキュー。
- LifoQueue: 最後に追加された要素が最初に取り出される（LIFO）、スタック構造。
- PriorityQueue: 要素に設定された優先度（数値が小さいほど高い）に基づいて取り出されるキュー。
import queue
# 1. FIFOキュー（タスクの順番処理）
task_queue = queue.Queue()
task_queue.put("Task_Alpha")
task_queue.put("Task_Beta")
print(f"FIFO取得: {task_queue.get()}") # Task_Alpha
# 2. LIFOキュー（アンドゥ操作や最新ログの処理）
history_stack = queue.LifoQueue()
history_stack.put("State_1")
history_stack.put("State_2")
print(f"LIFO取得: {history_stack.get()}") # State_2
# 3. 優先度キュー（緊急度に応じたタスク処理）
# タプル (優先度, データ) の形式で投入
alert_queue = queue.PriorityQueue()
alert_queue.put((3, "低優先度: 定期ログ"))
alert_queue.put((1, "最高優先度: サーバーダウン"))
alert_queue.put((2, "中優先度: メモリ警告"))
print(f"優先度取得: {alert_queue.get()}") # (1, '最高優先度: サーバーダウン')