Pythonでconcurrent.futuresを使った並列タスク実行

[concurrent.futures]:http://docs.python.jp/3/library/concurrent.futures.html
[Executor]:http://docs.python.jp/3/library/concurrent.futures.html#executor-objects
[ThreadPoolExecutor]:http://docs.python.jp/3/library/concurrent.futures.html#threadpoolexecutor
[ProcessPoolExecutor]:http://docs.python.jp/3/library/concurrent.futures.html#processpoolexecutor
[threading]:http://docs.python.jp/3/library/threading.html
[multiprocessing]:http://docs.python.jp/3/library/multiprocessing.html
[submit]:http://docs.python.jp/3/library/concurrent.futures.html#concurrent.futures.Executor.submit
[map]:http://docs.python.jp/3/library/concurrent.futures.html#concurrent.futures.Executor.map
[Future]:http://docs.python.jp/3/library/concurrent.futures.html#future-objects

[gist]:https://gist.github.com/tag1216/40b75346fd4ffdbfba22a55905094b0e
[01_thread.py]:https://gist.github.com/tag1216/40b75346fd4ffdbfba22a55905094b0e#file-01_thread-py
[02_future.py]:https://gist.github.com/tag1216/40b75346fd4ffdbfba22a55905094b0e#file-02_future-py
[03_map.py]:https://gist.github.com/tag1216/40b75346fd4ffdbfba22a55905094b0e#file-03_map-py
[04_process.py]:https://gist.github.com/tag1216/40b75346fd4ffdbfba22a55905094b0e#file-04_process-py

# concurrent.futuresモジュールの概要

Python3.2で追加された[concurrent.futures]モジュールは、複数の処理を並列実行するための機能を提供します。

Pythonには他に[threading]と[multiprocessing]というモジュールがありますが、これらが１つのスレッド・プロセスを扱うのに対して、[concurrent.futures]モジュールは複数のスレッド・プロセスを扱うことを目的としています。

## Executor

[concurrent.futures]モジュールには抽象クラスとして[Executor]クラスがあり、実装クラスとして2つのクラスが提供されています。
並列タスクを実行するにはこの２つのうちどちらかを使用します。

- [ThreadPoolExecutor]
    - スレッドを使って並列タスクを実行します。
    - ネットワークアクセスなどCPUに負荷がかからない処理の並列実行に適しています。
- [ProcessPoolExecutor]
    - プロセスを使って並列タスクを実行します。
    - CPUに負荷がかかる計算処理などの並列実行に適しています。

## max_workers

Executorのコンストラクタでは同時に実行可能なタスクの最大数を引数max_workersで指定します。
同時実行可能数より多いタスクの実行を要求すると、タスクはキューに追加されて他のタスクの終了を待ってから順次実行します。

## タスクを実行するメソッドsubmitとmap

Executorには並列タスクを実行する以下のメソッドがあります。

- [submit]
    - １つのタスクを実行キューに追加します。
    - 実行中のタスクがmax_workers未満であれば追加されたタスクは即実行が開始されます。
    - 戻り値の[Future]オブジェクトでタスクのキャンセルや実行結果の取得を行います。
- [map]
    - 実行タスクをiteratorで渡します。
    - 戻り値はタスクの実行結果を取得するためのgeneratorです。

# サンプル
[concurrent.futures]を使ったサンプルです。Python3.6で書いてますが、3.2以降の環境なら少し直せば動かせると思います。

- [01_thread.py]
    - ThreadPoolExecutorを使ったシンプルな例
- [02_future.py]
    - submitの結果を受け取る例
- [03_map.py]
    - mapでタスクを一括追加/結果を一括取得する例
- [04_process.py]
    - ProcessPoolExecutorで重い処理を実行する例

## ThreadPoolExecutorを使ったシンプルな例

[01_thread.py]
2つのスレッドで5つのタスクを実行します。タスクは1秒間スリープするだけの処理です。

```py3:
def task(v):
    getLogger().info("%s start", v)
    time.sleep(1.0)
    getLogger().info("%s end", v)

def main():
    init_logger()
    getLogger().info("main start")
    with ThreadPoolExecutor(max_workers=2, thread_name_prefix="thread") as executor:
        for i in range(5):
            executor.submit(task, i)
        getLogger().info("submit end")
    getLogger().info("main end")
```

```shell-session
[2017-04-02 12:01:39,747] [MainThread] main start
[2017-04-02 12:01:39,748] [thread_0] 0 start
[2017-04-02 12:01:39,749] [thread_1] 1 start
[2017-04-02 12:01:39,750] [MainThread] submit end
[2017-04-02 12:01:40,755] [thread_0] 0 end
[2017-04-02 12:01:40,755] [thread_0] 2 start
[2017-04-02 12:01:40,756] [thread_1] 1 end
[2017-04-02 12:01:40,756] [thread_1] 3 start
[2017-04-02 12:01:41,761] [thread_0] 2 end
[2017-04-02 12:01:41,761] [thread_0] 4 start
[2017-04-02 12:01:41,761] [thread_1] 3 end
[2017-04-02 12:01:42,764] [thread_0] 4 end
[2017-04-02 12:01:42,765] [MainThread] main end
```

## submitで結果を受け取る

[02_future.py]

[submit]の結果を受け取る例です。

```py3
def task(v):
    getLogger().info("%s start", v)
    time.sleep(1.0)
    getLogger().info("%s end", v)
    return v * 2

def main():
    init_logger()
    getLogger().info("main start")
    with ThreadPoolExecutor(max_workers=2, thread_name_prefix="thread") as executor:
        futures = []
        for i in range(5):
            futures.append(executor.submit(task, i))
        getLogger().info("submit end")
        getLogger().info([f.result() for f in futures])
    getLogger().info("main end")
```

```shell-session
[2017-04-02 12:08:23,853] [MainThread] main start
[2017-04-02 12:08:23,854] [thread_0] 0 start
[2017-04-02 12:08:23,855] [thread_1] 1 start
[2017-04-02 12:08:23,856] [MainThread] submit end
[2017-04-02 12:08:24,856] [thread_0] 0 end
[2017-04-02 12:08:24,856] [thread_0] 2 start
[2017-04-02 12:08:24,857] [thread_1] 1 end
[2017-04-02 12:08:24,857] [thread_1] 3 start
[2017-04-02 12:08:25,863] [thread_0] 2 end
[2017-04-02 12:08:25,864] [thread_0] 4 start
[2017-04-02 12:08:25,864] [thread_1] 3 end
[2017-04-02 12:08:26,867] [thread_0] 4 end
[2017-04-02 12:08:26,868] [MainThread] [0, 2, 4, 6, 8]
[2017-04-02 12:08:26,868] [MainThread] main end
```

## mapでタスクを一括追加/結果を一括取得する例

[03_map.py]
タスクを一括で処理する場合は[submit]より[map]を使った方が楽に書けます。

```py3
def task(v):
    getLogger().info("%s start", v)
    time.sleep(1.0)
    getLogger().info("%s end", v)
    return v * 2

def main():
    init_logger()
    getLogger().info("main start")
    with ThreadPoolExecutor(max_workers=2, thread_name_prefix="thread") as executor:
        results = executor.map(task, range(5))
        getLogger().info("map end")
    getLogger().info(list(results))
    getLogger().info("main end")
```

```shell-session
[2017-04-02 12:10:03,997] [MainThread] main start
[2017-04-02 12:10:03,998] [thread_0] 0 start
[2017-04-02 12:10:04,000] [thread_1] 1 start
[2017-04-02 12:10:04,000] [MainThread] map end
[2017-04-02 12:10:05,005] [thread_0] 0 end
[2017-04-02 12:10:05,006] [thread_0] 2 start
[2017-04-02 12:10:05,006] [thread_1] 1 end
[2017-04-02 12:10:05,006] [thread_1] 3 start
[2017-04-02 12:10:06,007] [thread_0] 2 end
[2017-04-02 12:10:06,007] [thread_0] 4 start
[2017-04-02 12:10:06,007] [thread_1] 3 end
[2017-04-02 12:10:07,014] [thread_0] 4 end
[2017-04-02 12:10:07,014] [MainThread] [0, 2, 4, 6, 8]
[2017-04-02 12:10:07,014] [MainThread] main end
```

## ProcessPoolExecutorで重い処理を実行する例

[04_process.py]
最後にProcessPoolExecutorを使った例です。
パラメータを変えてパフォーマンスの違いを見てみます。

```py3
def task(params):
    (v, num_calc) = params
    a = float(v)
    for _ in range(num_calc):
        a = pow(a, a)
    return a

def main():
    init_logger()

    if len(sys.argv) != 5:
        print("usage: 05_process.py max_workers chunk_size num_tasks num_calc")
        sys.exit(1)
    (max_workers, chunk_size, num_tasks, num_calc) = map(int, sys.argv[1:])

    start = time()

    with ProcessPoolExecutor(max_workers=max_workers) as executor:
        params = map(lambda _: (random(), num_calc), range(num_tasks))
        results = executor.map(task, params, chunksize=chunk_size)
    getLogger().info(sum(results))

    getLogger().info("{:.3f}".format(time() - start))
```

### パラメータの説明

- max_workers
    - 同時に実行する最大プロセス数
- chunk_size
    - プロセスに一度に渡すタスク数？
- num_tasks
    - タスクの実行回数
- num_calc
    - 1タスク内で実行する計算の回数

### 実行環境

- CPU:Intel Core i5(2.6 GHz)
- コア数:2

### 実行結果

上の４つが大きいタスク、下が小さいタスクを大量に実行した結果です。
小さいタスクを大量に実行する場合にはchunk_sizeの指定が重要なようです。

|max_workers|chunk_size|num_tasks|num_calc|実行時間(sec)|
|---:|---:|---:|---:|---:|
|1|  1|   100|100,000|1.954|
|2|  1|   100|100,000|1.042|
|1| 10|   100|100,000|1.922|
|2| 10|   100|100,000|1.071|
|1|  1|10,000|  1,000|3.295|
|2|  1|10,000|  1,000|3.423|
|1| 10|10,000|  1,000|2.272|
|2| 10|10,000|  1,000|1.279|
|1|100|10,000|  1,000|2.126|
|2|100|10,000|  1,000|1.090|