Python並行・並列処理を整理する (入門)

## 概要
大量データに関するワークロードを、モノリシックに解決する場合、並行・並列処理の知識は活用できます
但し、一般にPythonにはGILが存在するため、正しい挙動を抑えておきたい所です

## 内容
### Pythonにおける並行処理・並列処理
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2266010/1930f807-d02a-17ac-6c9b-388648d94f27.png)

1. プロセス = OSが管理する実行単位 (CPUコア毎)
2. スレッド = プロセス内の軽量な実行単位 (1CPUコア内)

上記前提を踏まえると、以下となります
1. 並行処理 = 1つのプロセスに複数のスレッドが存在します (PythonはGILにより基本1プロセス制約です)
2. 並列処理 = 複数のプロセスを並行して進めます (複数プロセス活用できます)


**MMU (Memory Management Unit)に関して**
プロセスから参照するメモリは`MMU`により制御された仮想メモリを参照します。`MMU`によりプロセス間ではメモリ競合は起きません
(1プロセス ≒ 1CPUコアであり ≠ 1CPUなので、有難い制約です)
これは並列処理に貢献します

一方、並行処理では同一`MMU`内のメモリを複数スレッドが参照する為、競合が発生するケースがあります
誤った解釈として、PythonはGILなので、競合は起きないという誤解がありますが
スケジューラにより、実行スレッドが切り替わる際、前後にて同一のメモリに対する操作を行うケースなどは、`race condition`は発生し得ます
スレッドは所謂`mutex`として機能しないことに注意が必要です

**データ共有に関して**
並列処理を行う際、メインプロセスからサブプロセスを生成し、データを共有します
その際、内部的には必要なオブジェクトを`pickle`化 (serialization)して渡します
並列処理を扱う条件として、`pickle`化可能なオブジェクトのみを渡す必要があります
この処理は重く、オーバヘッドが大きい為、場合によっては並列化しない方が速いケースも存在する事に注意が必要です

一方、共有メモリを用いて受け渡す方法もあります
この場合、共有メモリ内の参照を渡すことで、上記オーバヘッドへの対策とします


### 標準ライブラリ対応表
|||並行|並列|備考|導入ver|
|-|-|-|-|-|-|
|1|~~thread~~|o|-|`concurrent.futures`の方が優れているので、覚えなくても良い|1.x|
|2|~~threading~~|o|-|(同上)|2.7|
|3|**concurrent.futures.ThreadPoolExecutor**|o|-|futureパターンを踏襲。比較的よく見かける並行処理|3.2|
|4|**concurrent.futures.ProcessPoolExectuor**|-|o|比較的よく見かける並列処理<br/> (単体では共有メモリが使えない)|3.2|
|5|multiprocessing.Process|-|o|共有メモリが使える<br/>生成コストは低いが、実行する関数毎にオブジェクトが作成されるので、メモリを消費しやすい|2.6|
|6|multiprocessing.Pool|-|o|共有メモリが使える<br/>生成コストは重いが、worker数毎にタスクを順次処理、オブジェクト作成し、終わればQueueにある次のタスク開始するので、オブジェクト数が膨張してメモリを消費しない|2.6|
|7|asyncio|o|-|Node.js的な非同期IO|3.4|

- 個人的な所感としては、<ins>項番3,4の`concurrent.futures`がデファクト</ins>かと思います。IFがシンプルで扱いやすいですし、単純な操作であれば、並行・並列、共に対応できます
  - 但し、弱点として並列処理において、<ins>単体では共有メモリが使えない</ins>ので、利用する場合は項番5,6の`multiprocessing`ライブラリと組み合わせる必要があります
<br/>

- 項番7の`asyncio`は比較的新しいです。非同期IOはサーバアプリケーションにおけるマルチスレッドにおけるC10K問題への対策として有名になったシングルスレッドでの制御方式であり、<ins>処理をスレッド数で分散するのではなく、時間軸で分散する</ins>概念となります
  - 例えば、`Uvicorn -> (ASGI) -> FastAPI`の様な構成では`asyncio`を介して非同期で動作するのが前提なので、サーバサイドにおける並行処理に関する開発では必要な知識となります

以下は、`concurrent.futures`をベースに検証を進めます

### 速度検証 (プロファイリング)
並列・並行処理を導入することで解決したい課題は主に2種類に分割できます
1. IO Bound = 処理の実行時間がI/Oに依存
2. CPU Bound = 処理の実行時間がCPUの計算速度に依存

<ins>IO Boundには並行処理、CPU Boundには並列処理</ins>が向いているとされています

例えば
```python
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor, as_completed
from line_profiler import LineProfiler
import time
```

以下の様に`IO Bound`と`CPU Bound`の処理を用意したとします
```python
def IOBoundedTask(i):
    print(f"IO Bound: {i}")
    time.sleep(1) # 擬似的にsleepとする
    return i

def CPUBoundedTask(i):
    print(f"CPU Bound: {i}")
    result = sum(range(10**8)) # CPUを消費する擬似的な処理
    return result
```

`concurrent.futures`を利用する場合の並行・並列処理の例です
```python
def concurrent_task(task, num_workers):
    with ThreadPoolExecutor(max_workers=num_workers) as executor:
        futures = []
        for i in range(4):
            futures.append(executor.submit(task, i))
        results = [future.result() for future in as_completed(futures)]
        print(len(results))

def parallel_task(task, num_workers):
    with ProcessPoolExecutor(max_workers=num_workers) as executor:
        futures = []
        for i in range(4):
            futures.append(executor.submit(task, i))
        results = [future.result() for future in as_completed(futures)]
        print(len(results))
```

以下の様に実験条件を設定し、プロファイリングを行います

```python
def main():
    # 2x
    concurrent_task(IOBoundedTask, num_workers=2)
    parallel_task(CPUBoundedTask, num_workers=2)

    # 4x
    concurrent_task(IOBoundedTask, num_workers=4)
    parallel_task(CPUBoundedTask, num_workers=4)

    # ineffective use
    parallel_task(IOBoundedTask, num_workers=4)
    concurrent_task(CPUBoundedTask, num_workers=4)


# https://stackoverflow.com/questions/24374288/where-to-put-freeze-support-in-a-python-script
if __name__ == '__main__':
    profiler = LineProfiler()
    profiler.add_function(main)
    profiler.runcall(main)
    profiler.print_stats()
```
![スクリーンショット 2024-08-04 11.12.57.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2266010/b9e6e60f-1290-f95e-f1f5-11f3aceb1809.png)
`Timer unit`はナノ秒なので、Timeと掛け合わせて考えると

**2x, 4x**
- 4xの方が、`concurrent`, `parallel`共に実行時間は削減されています (効果的な使い方)


**ineffective use**
- `concurrent`における`CPUBoundedTask`は明確に意味がなく、%Timeから見ても明らかに効果は薄いです
- `parallel`は`pickle化`のオーバヘッドが少なく、意外と時間は削減できましたが
大量I/Oを並列実行する際などを想定し、検証条件を変更すると、`concurrent`で`IOBoundedTask`を行うケースとの差は明確になると思います

### 参考
書籍
[Effective Python](https://www.oreilly.co.jp/books/9784873119175/)

Web
[Pythonの並行処理/並列処理を学ぼう](https://selfs-ryo.com/detail/python_concurrent_and_parallel#cpuio)
[Pythonの並列処理・並行処理をしっかり調べてみた](https://qiita.com/simonritchie/items/1ce3914eb5444d2157ac#multiprocessing%E3%81%AEprocess%E3%81%A8pool%E3%81%AE%E3%81%A9%E3%81%A1%E3%82%89%E3%82%92%E4%BD%BF%E3%81%86%E3%81%B9%E3%81%8D%E3%81%8B)
[Node.jsの非同期I/Oについて調べてみた
](https://techblog.recruit.co.jp/article-1115/)

### 総括
雑ですが、整理してみました
並行処理に関しては、競合はバグの温床となり、機能要件に影響するので、正しい理解が必要だと思いました
並列処理に関しては、別のアプローチとして、インフラ側で分散構成を取る回避策もあります。但し、保守性の観点ではどちらも良し悪しあると思います

<ins>以上</ins>