Pythonでの並行処理：マルチコアを活かす

並行処理の基礎：マルチコアを最大限に活かすために

現代のコンピュータは、複数の処理コア（マルチコア）を搭載し、複雑なタスクを効率的にこなせるようになりました。しかし、その性能を最大限に引き出すには、並行処理の理解が不可欠です。本記事では、並行処理の基本概念から、PythonでマルチコアCPUの性能をフルに活用するための具体的な方法までを、徹底的に解説します。

なぜ並行処理が重要なのか？

シングルコアCPUの時代、プログラムはタスクを順番に処理していました。しかし、マルチコアCPUの登場により、複数のタスクを「同時に」実行できるようになりました。これにより、処理速度が飛躍的に向上し、より複雑な処理も現実的な時間で完了できるようになりました。

例えば、動画編集、科学計算、機械学習など、CPUに大きな負荷がかかる処理では、並行処理を活用することで、処理時間を大幅に短縮できます。Webサーバーであれば、多数の同時リクエストを効率的に処理し、Webサイトの応答速度を向上させることが可能です。

並行処理 vs 並列処理：違いを理解する

並行処理（Concurrency）とは、複数のタスクを「同時に進行しているように見せる」技術です。Webブラウザで複数のタブを開いて同時にWebサイトを閲覧したり、音楽を聴きながらドキュメントを作成したりするのも、並行処理の一例です。シングルコアCPUでも実現可能です。

一方、並列処理（Parallelism）とは、複数のタスクを「実際に同時に」実行する技術です。これは、マルチコアCPUのような複数の処理ユニットを持つ環境でのみ実現可能です。各コアが独立してタスクを実行するため、CPUの利用率が向上し、処理時間が大幅に短縮されます。

マルチコアCPUのメリット

マルチコアCPUを活用することで、以下のメリットが得られます。

• 処理速度の向上: 複数のコアが並列にタスクを実行するため、処理時間が短縮されます。
• 応答性の向上: バックグラウンドで重い処理を実行しても、UIの操作がスムーズに行えます。
• リソースの有効活用: CPUの利用率が向上し、システム全体のパフォーマンスが向上します。

Pythonで並行処理を学ぶ：本記事の構成

本記事では、Pythonで並行処理を実現するための主要な3つの方法、threading、multiprocessing、asyncioについて、以下の構成で解説します。

1. threading: 手軽に並行処理を始めるための基本モジュール。GIL（Global Interpreter Lock）の制約と、その回避策について解説します。
2. multiprocessing: GILの制約を受けない、真の並列処理を実現するためのモジュール。プロセス間通信（IPC）の方法や、オーバーヘッドについても解説します。
3. asyncio: シングルスレッドで効率的な並行処理を実現するためのライブラリ。I/Oバウンドなタスクに最適な理由を解説します。
4. 実践：ケーススタディと最適な並行処理の選択: 具体的なケーススタディを通して、各手法の最適な選択肢を解説し、パフォーマンスを最大化する方法を提案します。

さあ、Pythonで並行処理の世界へ飛び込み、マルチコアCPUのパワーを最大限に引き出しましょう！

threading：手軽な並行処理、GILの壁

Pythonで並行処理を始めるなら、threadingモジュールが手軽な選択肢の一つです。threadingを使うと、あたかも複数のタスクが同時に実行されているかのように見せかけることができます。これは、プログラムの中で複数の「スレッド」を生成し、それぞれのスレッドに異なる処理を割り当てることで実現されます。

threadingの基本：スレッドの作成と実行

threadingモジュールを使うと、簡単にスレッドを作成し、実行することができます。以下は、threadingを使った簡単なコード例です。

“`python
import threading
import time

def task(name):
print(f”スレッド{name}: 開始”)
time.sleep(2) # 2秒間スリープ
print(f”スレッド{name}: 終了”)

# スレッドの作成と実行
thread1 = threading.Thread(target=task, args=(“A”,))
thread2 = threading.Thread(target=task, args=(“B”,))

thread1.start()
thread2.start()

thread1.join()
thread2.join()

print(“すべてのスレッドが終了しました”)
“`

この例では、task関数を2つのスレッドで実行しています。thread1.start()とthread2.start()でスレッドを開始し、thread1.join()とthread2.join()でスレッドの終了を待ちます。これにより、task関数が（見かけ上）並行に実行されます。

実行結果

“`
スレッドA: 開始
スレッドB: 開始
スレッドA: 終了
スレッドB: 終了
すべてのスレッドが終了しました
“`

GIL（Global Interpreter Lock）とは？

threadingモジュールは手軽で便利な反面、GIL（Global Interpreter Lock）という制約があります。GILとは、CPythonインタープリタ（標準的なPythonの実装）が持つ仕組みで、一度に一つのスレッドしかPythonバイトコードを実行できないようにするものです。

なぜGILが存在するのか？

GILは、CPythonのメモリ管理を簡素化し、スレッドセーフを確保するために導入されました。しかし、その代償として、CPUバウンドなタスクにおける並列処理の恩恵を受けられないというデメリットがあります。

GILの何が問題なのか？

GILの存在が、CPUバウンドなタスク（計算処理が中心のタスク）において、threadingを使った並行処理の効果を大きく制限します。例えば、画像を処理するようなCPUを酷使するタスクを複数のスレッドで実行しても、実際には一つのコアしか使用されないため、処理速度はほとんど向上しません。

GILの回避策：マルチコアを活かす道

GILの制約を回避し、マルチコアCPUの性能を最大限に引き出すためには、いくつかの方法があります。

1. multiprocessingモジュールを使う: multiprocessingモジュールを使うと、複数のプロセスを生成し、各プロセスが独立したメモリ空間を持つため、GILの制約を受けずに真の並列処理を実現できます。CPUバウンドなタスクには、multiprocessingが適しています。
2. I/Oバウンドなタスクにthreadingを使う: I/Oバウンドなタスク（ネットワーク通信、ファイルアクセスなど）では、スレッドがI/O待ちの間はGILが解放されるため、threadingでもある程度の並行性を実現できます。I/O待ちの間に別のスレッドが実行されるため、全体の処理時間を短縮できます。
3. NumPyなどのC拡張ライブラリを使う: NumPyなどのCで記述された拡張ライブラリは、内部でGILを解放して並列処理を行うものがあります。これらのライブラリを使うことで、GILの制約を受けずに高速な処理が可能です。
4. JythonやIronPythonなどのGILを持たないPython実装を使う: CPython以外のPython実装（Jython、IronPythonなど）は、GILを持たないため、CPUバウンドなタスクでも並列処理の効果を期待できます。

threadingの適切な使い方：まとめ

threadingモジュールは、手軽に並行処理を実装できる便利なツールですが、GILの制約があることを理解しておく必要があります。I/Oバウンドなタスクにはthreading、CPUバウンドなタスクにはmultiprocessingというように、タスクの特性に応じて適切なモジュールを選択することが、PythonでマルチコアCPUの性能を最大限に引き出すための鍵となります。

次のセクションでは、GILの制約を受けないmultiprocessingについて解説します。

multiprocessing：真の並列処理、オーバーヘッド

threadingで並行処理を試してみたけど、思ったより速くならない…」と感じたことはありませんか？それは、PythonのGIL（Global Interpreter Lock）という制約が原因かもしれません。GILは、一度に一つのスレッドしかPythonバイトコードを実行できないようにする仕組みです。そのため、CPUバウンドな処理では、threadingを使っても複数のコアをフルに活用できず、並列処理の効果を十分に得られないのです。

そこで登場するのがmultiprocessingモジュールです。multiprocessingは、複数のプロセスを生成することで、GILの制約を回避し、真の並列処理を実現します。各プロセスは独立したメモリ空間を持つため、複数のCPUコア上で同時にPythonコードを実行できます。

multiprocessingの基本：プロセスの生成と実行

multiprocessingモジュールを使うと、簡単にプロセスを生成し、実行することができます。以下は、multiprocessingを使った簡単なコード例です。

“`python
import multiprocessing

def worker(num):
“””各プロセスで実行される関数”””
print(f’Worker {num}: プロセスIDは {multiprocessing.current_process().pid}’)

if __name__ == ‘__main__’:
processes = []
for i in range(multiprocessing.cpu_count()): # CPUのコア数だけプロセスを生成
p = multiprocessing.Process(target=worker, args=(i,))
processes.append(p)
p.start()

for p in processes:
p.join()
“`

上記の例では、multiprocessing.cpu_count()でCPUのコア数を取得し、その数だけプロセスを生成しています。各プロセスはworker関数を実行し、自身のプロセスIDを表示します。

実行結果

“`
Worker 0: プロセスIDは 1234
Worker 1: プロセスIDは 1235
Worker 2: プロセスIDは 1236
Worker 3: プロセスIDは 1237
“`

threadingとの違い：メモリ空間とGIL

threadingとmultiprocessingの主な違いは、以下の通りです。

• 並列処理の実現方法: threadingは1つのプロセス内で複数のスレッドを管理しますが、multiprocessingは複数のプロセスを生成します。
• GILの影響: threadingはGILの影響を受けますが、multiprocessingは受けません。
• メモリ空間: threadingはメモリを共有しますが、multiprocessingは共有しません。
• オーバーヘッド: プロセスの生成には、スレッドの生成よりも大きなオーバーヘッドがかかります。

これらの違いから、CPUバウンドなタスクにはmultiprocessing、I/Oバウンドなタスクにはthreadingまたはasyncioが適していると言えます。

プロセス間通信（IPC）の方法：Queueを使ってデータを共有する

multiprocessingでは、各プロセスが独立したメモリ空間を持つため、プロセス間でデータを共有するためには、特別な仕組みが必要です。これをプロセス間通信（IPC: Inter-Process Communication）と呼びます。

multiprocessingモジュールには、様々なIPCの手段が用意されています。ここでは、最も基本的なIPCの手段であるQueueを使った例を紹介します。

“`python
from multiprocessing import Process, Queue

def producer(queue):
“””データを生成してキューに入れる”””
for i in range(5):
queue.put(i)
print(f’Producer: {i}をキューに入れました’)

def consumer(queue):
“””キューからデータを取り出して処理する”””
while True:
item = queue.get()
if item is None:
break
print(f’Consumer: {item}をキューから取り出しました’)

if __name__ == ‘__main__’:
queue = Queue()
p = Process(target=producer, args=(queue,))
c = Process(target=consumer, args=(queue,))
p.start()
c.start()
p.join()
queue.put(None) # キューに終了の合図を送る
c.join()
“`

この例では、Queueを使って、producerプロセスが生成したデータをconsumerプロセスが処理しています。producerプロセスは、キューにデータを入れ終わったら、Noneをキューに入れて、consumerプロセスに終了を知らせます。

実行結果

“`
Producer: 0をキューに入れました
Producer: 1をキューに入れました
Producer: 2をキューに入れました
Producer: 3をキューに入れました
Producer: 4をキューに入れました
Consumer: 0をキューから取り出しました
Consumer: 1をキューから取り出しました
Consumer: 2をキューから取り出しました
Consumer: 3をキューから取り出しました
Consumer: 4をキューから取り出しました
“`

オーバーヘッド：プロセスの生成とIPCのコスト

multiprocessingは、GILの制約を回避できる強力なツールですが、オーバーヘッドも伴います。

• プロセスの生成: プロセスの生成には、スレッドの生成よりも時間がかかります。
• プロセス間通信: プロセス間通信には、データのシリアライズ・デシリアライズのオーバーヘッドがかかります。プロセス間でデータをやり取りする際には、データをネットワークを通じて送受信できる形式に変換する必要があります。この変換処理をシリアライズと呼び、受信側で元のデータ形式に戻す処理をデシリアライズと呼びます。

そのため、タスクの処理時間が短い場合や、プロセス間通信の頻度が高い場合は、multiprocessingのオーバーヘッドが無視できなくなる可能性があります。そのような場合は、threadingやasyncioなど、他の並行処理モデルを検討する方が良いでしょう。

Pool：プロセス管理を簡単にする

Poolを使うことで、プロセス管理を簡単にすることができます。

“`python
from multiprocessing import Pool

def square(x):
“””引数の2乗を計算する”””
return x * x

if __name__ == ‘__main__’:
with Pool(processes=4) as pool:
numbers = [1, 2, 3, 4, 5]
results = pool.map(square, numbers)
print(f’Results: {results}’)
“`

この例では、Poolを使って4つのプロセスを生成し、numbersリストの各要素に対してsquare関数を並列に実行しています。pool.map()を使うことで、複数の引数を並列に処理し、結果をリストとして取得できます。

実行結果

“`
Results: [1, 4, 9, 16, 25]
“`

multiprocessingの適切な使い方：まとめ

multiprocessingは、CPUバウンドなタスクを並列化するための強力なツールです。しかし、オーバーヘッドも考慮して、適切な並行処理モデルを選択することが重要です。

次のセクションでは、I/Oバウンドなタスクに最適なasyncioについて解説します。

asyncio：非同期処理、I/Oバウンドなタスクに最適

Pythonで効率的な並行処理を実現するもう一つの強力な武器が、asyncioライブラリです。特に、ネットワーク通信やファイルアクセスなど、I/O待ち時間が長いタスク（I/Oバウンドなタスク）において、その真価を発揮します。

asyncioの基本：シングルスレッドで並行処理を実現する

asyncioは、シングルスレッドで複数のタスクをあたかも同時に実行しているかのように見せる仕組みを提供します。ここで重要なのは、スレッドやプロセスを新たに生成するのではなく、イベントループと呼ばれる機構がタスクの実行を効率的に管理する点です。

async/await構文：非同期処理をシンプルに記述する

asyncioの中核となるのが、asyncとawaitという2つのキーワードです。

• async def: async defで定義された関数はコルーチンと呼ばれ、非同期的に実行できる関数であることを示します。
• await: awaitキーワードは、I/O処理などの完了を待つ際に使用します。awaitを実行すると、イベントループは一時的にコルーチンの実行を中断し、その間に他のタスクを実行します。I/O処理が完了すると、awaitで中断されたコルーチンが処理を再開します。

以下は、asyncioとaiohttpライブラリを使って、非同期的にHTTPリクエストを送信する例です。

“`python
import asyncio
import aiohttp

async def fetch_url(session, url):
async with session.get(url) as response:
return await response.text()

async def main():
async with aiohttp.ClientSession() as session:
html = await fetch_url(session, ‘https://www.example.com’)
print(html[:50]) # 最初の50文字を表示

if __name__ == ‘__main__’:
asyncio.run(main())
“`

上記の例では、fetch_url関数がコルーチンとして定義され、aiohttpライブラリを使って非同期的にHTTPリクエストを送信しています。await response.text()の部分で、レスポンスの受信を待つ間、イベントループは他のタスクに処理を移譲できます。

実行結果

“`