Pythonでサンプルコードから学ぶマルチプロセス：CPUバウンドタスク、I/Oバウンドタスク、キュー、共有メモリ

Pythonにおけるマルチプロセスとマルチスレッド

• マルチスレッド: 1つのプロセス内で複数のスレッドを実行する。
• マルチプロセス: 複数の独立したプロセスを実行する。

それぞれ異なる利点と欠点があり、状況に応じて使い分ける必要があります。

マルチスレッドの利点と欠点

利点:

• 処理速度が速い。
• 少ないメモリで動作する。
• 共有メモリが容易。

欠点:

• GIL (Global Interpreter Lock) の存在により、真の並列処理ができない。
• スレッド間のデータ競合が発生する可能性がある。

マルチプロセスの利点と欠点

利点:

• GILの影響を受けずに真の並列処理が可能。
• スレッド間のデータ競合が発生しない。

欠点:

• 処理速度がマルチスレッドよりも遅い。
• 多くのメモリを使用する。
• 共有メモリが難しい。

マルチプロセスにおけるロック

マルチプロセスでは、複数の独立したプロセスが同時にメモリにアクセスするため、データ競合が発生する可能性があります。これを防ぐために、ロックと呼ばれる機構を用います。

ロックは、共有リソースへのアクセスを制御するものです。プロセスがリソースを使用する前にロックを取得し、使用後はロックを解放する必要があります。

異なるロックの種類

Pythonのマルチプロセスモジュールでは、いくつかの種類のロックを提供しています。

• Lock: 排他的ロック。一度に1つのプロセスだけがリソースにアクセスできる。
• RLock: 再帰ロック。同じプロセスが複数回ロックを取得できる。
• Semaphore: セマフォ。リソースへのアクセス許可数を制御できる。

マルチプロセスにおけるロックの例

以下は、Lockを使って共有リソースへのアクセスを制御する例です。

import multiprocessing

# 共有リソース
counter = 0

def worker(lock):
    # ロックを取得
    lock.acquire()

    # 共有リソースを操作
    global counter
    counter += 1

    # ロックを解放
    lock.release()

if __name__ == '__main__':
    # ロックを作成
    lock = multiprocessing.Lock()

    # 複数のプロセスを起動
    processes = [multiprocessing.Process(target=worker, args=(lock,)) for _ in range(4)]
    for p in processes:
        p.start()

    # すべてのプロセスが終了するまで待機
    for p in processes:
        p.join()

    # 共有リソースの最終的な値
    print(counter)

この例では、4つのプロセスが同時にcounter変数をインクリメントしようとします。しかし、Lockを使うことで、一度に1つのプロセスしかcounter変数にアクセスできないため、データ競合が発生することはありません。

Pythonにおけるマルチプロセスとマルチスレッド、そしてマルチプロセスにおけるロックについて解説しました。

マルチタスク処理を行う際には、それぞれの方法の利点と欠点を理解し、状況に応じて使い分けることが重要です。また、マルチプロセスでは、データ競合を防ぐためにロックを使用する必要があります。

マルチプロセスを使ったサンプルコード

CPU バウンドタスクの並列処理

import multiprocessing

def worker(num):
    # CPU バウンドな処理
    for i in range(1000000):
        pass

if __name__ == '__main__':
    # CPU コア数分のプロセスを起動
    num_workers = multiprocessing.cpu_count()
    processes = [multiprocessing.Process(target=worker, args=(i,)) for i in range(num_workers)]
    for p in processes:
        p.start()

    # すべてのプロセスが終了するまで待機
    for p in processes:
        p.join()

I/O バウンドタスクの並列処理

import multiprocessing
import requests

def worker(urls):
    # I/O バウンドな処理
    for url in urls:
        requests.get(url)

if __name__ == '__main__':
    # URL のリスト
    urls = ["https://www.google.com", "https://www.yahoo.co.jp", "https://www.bing.com"]

    # 複数のプロセスで処理
    num_workers = multiprocessing.cpu_count()
    chunk_size = len(urls) // num_workers
    processes = [multiprocessing.Process(target=worker, args=(urls[i * chunk_size:(i + 1) * chunk_size],)) for i in range(num_workers)]
    for p in processes:
        p.start()

    # すべてのプロセスが終了するまで待機
    for p in processes:
        p.join()

このコードは、I/O バウンドな処理を複数のプロセスで並列処理する例です。URL のリストを分割し、それぞれ異なるプロセスで処理することで、処理時間を短縮することができます。

キューを使った通信

import multiprocessing
import queue

def producer(q):
    # データを生成
    for i in range(10):
        q.put(i)

def consumer(q):
    # データを消費
    while True:
        data = q.get()
        if data is None:
            break
        print(data)

if __name__ == '__main__':
    # キューを作成
    q = queue.Queue()

    # プロデューサーとコンシューマーを起動
    producer_process = multiprocessing.Process(target=producer, args=(q,))
    consumer_process = multiprocessing.Process(target=consumer, args=(q,))
    producer_process.start()
    consumer_process.start()

    # プロデューサーが終了するまで待機
    producer_process.join()

    # キューに終了通知を送信
    q.put(None)

    # コンシューマーが終了するまで待機
    consumer_process.join()

このコードは、キューを使ってプロセス間通信を行う例です。プロデューサーはデータをキューに生成し、コンシューマーはキューからデータを取り出して処理します。

共有メモリを使った通信

import multiprocessing
import array

def worker(shared_array):
    # 共有メモリへのアクセス
    for i in range(10):
        shared_array[i] += 1

if __name__ == '__main__':
    # 共有メモリを作成
    shared_array = array.array('i', [0] * 10)

    # 複数のプロセスで共有メモリへアクセス
    num_workers = multiprocessing.cpu_count()
    processes = [multiprocessing.Process(target=worker, args=(shared_array,)) for i in range(num_workers)]
    for p in processes:
        p.start()

    # すべてのプロセスが終了するまで待機
    for p in processes:
        p.join()

    # 共有メモリの内容
    print(shared_array)

このコードは、共有メモリを使ってプロセス間通信を行う例です。複数のプロセスが同じメモリ領域にアクセスすることで、データを共有することができます。

その他のサンプルコード

• [Pythonのmultiprocessingの使い方を現役エンジニアが解説【初心者向け】](https://m

マルチプロセスとマルチスレッド以外での並列処理方法

非同期処理は、複数の処理を同時に開始し、それぞれが完了するのを待たずに次の処理を開始する方法です。Pythonでは、asyncioモジュールを使って非同期処理を行うことができます。

import asyncio

async def worker(num):
    # 非同期処理
    await asyncio.sleep(1)
    print(num)

async def main():
    # 複数の非同期処理を同時に開始
    tasks = [asyncio.create_task(worker(i)) for i in range(10)]
    await asyncio.gather(*tasks)

if __name__ == '__main__':
    asyncio.run(main())

このコードは、asyncioモジュールを使って非同期処理を行う例です。10個の非同期処理を同時に開始し、それぞれが完了するのを待たずに次の処理を開始します。

GPU を使った並列処理

GPU は、CPU よりも多くの処理を並列処理できる能力を持っています。Pythonでは、PyTorchやTensorFlowなどのライブラリを使って、GPU を使った並列処理を行うことができます。

import torch

def worker(num):
    # GPU を使った処理
    x = torch.rand(1000, 1000)
    y = torch.rand(1000, 1000)
    z = torch.mm(x, y)

if __name__ == '__main__':
    # GPU を使用
    device = torch.device("cuda")

    # 複数の処理をGPU で実行
    num_workers = torch.cuda.device_count()
    processes = [multiprocessing.Process(target=worker, args=(i,)) for i in range(num_workers)]
    for p in processes:
        p.start()

    # すべてのプロセスが終了するまで待機
    for p in processes:
        p.join()

このコードは、PyTorchを使ってGPU を使った並列処理を行う例です。10個の処理をGPU で実行し、CPU で実行するよりも高速に処理することができます。

分散処理は、複数のコンピュータで処理を分散させる方法です。Pythonでは、DaskやRayなどのライブラリを使って、分散処理を行うことができます。

import dask

def worker(data):
    # 分散処理
    return data.sum()

if __name__ == '__main__':
    # 分散処理を実行
    dask.distributed.Client()
    results = dask.compute(worker, data=[1, 2, 3, 4, 5])
    print(results)

このコードは、Daskを使って分散処理を行う例です。5つのデータを5つのコンピュータに分散させ、処理結果を集計します。

マルチプロセスとマルチスレッド以外にも、非同期処理、GPU を使った並列処理、分散処理など、さまざまな方法で並列処理を行うことができます。

それぞれの方法には利点と欠点があり、状況に応じて使い分けることが重要です。