【Python】並列処理の高速化について

Question

前提・実現したいこと

以前、こちらの記事で質問させていただいたのですが補足でご指摘いただいた通りに粒度を大きくしてみたのですが全然高速化できておりません、、、

ですので、何処がボトルネックになっているかご指摘と改善出来るならアドバイス頂けるとありがたいです。

個人的には、並列化した時の通信時間よりforの方が時間がかかっている感じがするのでこのような結果になったのではないかなぁと推測しております。

元のソースコード

import concurrent.futures
import matplotlib.pyplot as plt
import numpy as np
import cv2, sys, os

img =  cv2.cvtColor(cv2.imread(sys.argv[1]), cv2.COLOR_BGR2RGB)
useCPU = 1

def main():
    """
    メイン関数
    """
    try:
        useCPU = int( input("使用するCPUのコアを入力してください[ 1 ~ {0} ] : ".format(os.cpu_count())) )
    except:
        useCPU = os.cpu_count()
    if useCPU > os.cpu_count():
        useCPU = os.cpu_count()
    mulchProcess(useCPU=useCPU)
    plt.imshow(img)
    plt.show()

def changeToGray( number: int, width: np.ndarray ):
    """
    並列化する処理 ( グレースケールに変換する )
    @param  number (int)       : このプロセスの番号
    @param  width (np.ndarray) : 横１行の配列[ [R, G, B], ・・・・ ,[R, G, B] ]
    @return number (int)       : このプロセスの番号
    @return width (np.ndarray) : 引数で受け取った配列をグレースケールに変換した配列
    """
    for pixel in width:
        gray = int(pixel[0]*0.3) + int(pixel[1]*0.59) + int(pixel[2]*0.11)
        pixel[0] = gray
        pixel[1] = gray
        pixel[2] = gray
    return number, width

def mulchProcess(useCPU: int):
    """
    マルチコアでプロセスを生成して実行させる処理
    @param  useCPU (int)  : 使用するCPUのコア数
    """
    with concurrent.futures.ProcessPoolExecutor(max_workers=useCPU) as executer:
        fs = [ executer.submit(changeToGray, i, width) for width, i in zip( img, range(len(img)) ) ]
        for future in concurrent.futures.as_completed(fs):
            line_number = future.result()[0]
            gray_width  = future.result()[1]
            img[line_number] = gray_width

if __name__ == '__main__':
    main()

直したソースコード

import concurrent.futures
import matplotlib.pyplot as plt
import numpy as np
import cv2, sys, os

img = cv2.cvtColor(cv2.imread(sys.argv[1]), cv2.COLOR_BGR2RGB)
useCPU = 1
step = 1

def main():
    """
    メイン関数
    """
    try:
        useCPU = int( input("使用するCPUのコアを入力してください[ 1 ~ {0} ] : ".format(os.cpu_count())) )
    except:
        useCPU = os.cpu_count()
    if useCPU > os.cpu_count():
        useCPU = os.cpu_count()
    step = int( len(img) / useCPU )
    mulchProcess(useCPU=useCPU, step= step)
    plt.imshow(img)
    plt.show()

def changeToGray( number: int, length: int ):
    """
    並列化する処理
    @param  number      (int) : 画像の処理対象範囲の先頭の添字
    @param  length      (int) : 対象範囲の長さ
    @return number      (int) : 画像の処理対象範囲の先頭の添字
    @return part_height (int) : 処理後の画像の配列
    """
    endPioint = number + length  if number + length < len(img)  else  len(img) - 1
    part_height = img[ number : endPioint-1 ]
    for width in part_height:
        for pixel in width:
            gray = int(pixel[0]*0.3) + int(pixel[1]*0.59) + int(pixel[2]*0.11)
            pixel[0] = gray
            pixel[1] = gray
            pixel[2] = gray
    return number, part_height

def mulchProcess(useCPU: int, step: int):
    """
    マルチコアでプロセスを生成して実行させる処理
    @param  useCPU (int)  : 使用するCPUのコア数
    @param  step   (int)  : 画像の高さをコア数で割った数
    """
    index_list = [ i for i in range(0, len(img), step)  if i < len(img) ]
    with concurrent.futures.ProcessPoolExecutor(max_workers=useCPU) as executer:
        fs = [ executer.submit(changeToGray, i, step) for i in index_list ]
        for future in concurrent.futures.as_completed(fs):
            line_number = future.result()[0]
            part_height  = future.result()[1]
            for i, height in zip( range(line_number, line_number+len(part_height)), part_height ):
                img[i] = height

if __name__ == '__main__':
    main()

試したこと

横1列ごとのピクセルで処理していたのを「横*（高さ/使用するコア数）」ごとに処理をさせるようにして通信時間を使用するコア数に削減させたつもりです....

補足情報（FW/ツールのバージョンなど）

[実行環境]
MacBook Pro (15-inch, 2016)
プロセッサ : 2.6 GHz Intel Core i7 ( 4コア8スレッド )
メモリ : 16 GB 2133 MHz LPDDR3
Python 3.6.4

[実行時間]
使用画像 : 5000px*5025px (15.1MB)
修正前のコード : 44.79758310317993秒
修正後のコード : 45.401297092437744秒

Comments

[mkgrei]

ボトルネックとおっしゃるので、とりあえず今の処理でどこでどれほどの時間をかけているのかの情報を追記していただけませんか？

[reishisu]

申し訳ございません、何処でどれほどの時間とは変えた部分の実行速度でよろしいでしょうか？

[退会済みユーザー]

全部ですよ

[退会済みユーザー]

それぞれの行にどのくらいの時間がかかっているかです

[reishisu]

では、mulchProcessより前はほとんど処理が同じなので入ってから終わるまでの実行時間を追記させて頂きますね。

[reishisu]

使用画像とコードごとの実行時間について追記させていただきました。

[umyu]

@reishisuさんへ 前の質問を見る限りではグレースケール化したいだけじゃないですよね。実際のコードとかけ離れたコードをチューニングしても無意味なのでグレースケール以外にもしている実際の処理を載せてくださいな

[reishisu]

実際にはこれだけでグレースケール化をしておりますが、、、前の質問ではまだプロトタイプ版で尚且つ質問の趣旨が違うので確か前のコードと少し違うかもしれないですがほとんど時間計測の処理と標準出力の部分を削除しているだけですよ。

[reishisu]

あとは、前の質問でもあった通りスレッドの部分は必要がないとのことでしたので実際にも使わないのと、こちらで投稿するときに要らないところまで書くと見辛いので削除しております。

[reishisu]

回答になっておりますか？もし、別のところでしたらどちらの部分か教えて頂けるとありがたいです。

[umyu]

前も似たようなコメントをしたのですが、ProcessPoolExecutorを使わずともcv2.cvtColor(img, cv2.COLOR_RGB2GRAY)を使うとグレースケール化は0.046秒で終わるんですよね。

[umyu]

ようするに、グレースケール化以外にも処理をしたいから、ProcessPoolExecutorを使うという話を仰られてたじゃないですか。でもこれはグレースケール部分のコードしかないので、グレースケール化以外部分がボトルネックにもしもなっていても、このコードからはわかんないという話です。

[reishisu]

確かに、それを使えば早いのはOpenCV使ってるのでそれに関しては100の承知ですが。今回はそれを踏まえた上で並列処理を利用してどれだけ逐次処理処理と比べて高速化できるかという所をやろうとしております....

[umyu]

例えば、numpyなら1バイト単位にループで演算しなくてもベクトル演算が使えると思いますし

[umyu]

@reishisuさんへ　なんとなくやりたいことが理解できました。コメント欄お邪魔して失礼致しました。

Answer 1

基本的に numpy の配列に対する演算は numpy で完結させたほうが速いです。

提示の「直したソースコード」は動かないので**「元のソースコード」**で、
changeToGray を numpy で完結する処理に変更した例を示します。

def changeToGray( number: int, width: np.ndarray ):
    """
    並列化する処理 ( グレースケールに変換する )
    @param  number (int)       : このプロセスの番号
    @param  width (np.ndarray) : 横１行の配列[ [R, G, B], ・・・・ ,[R, G, B] ]
    @return number (int)       : このプロセスの番号
    @return width (np.ndarray) : 引数で受け取った配列をグレースケールに変換した配列
    """
    return number, np.tile((width * [0.3, 0.59, 0.11]).astype(np.int).sum(axis=1), (3, 1)).T

環境はだいぶ違いますが、Windows 10 (Core i5)、4 コア(論理プロセッサの最大)で比較したところ
5472x3648 ピクセルの画像は、約 130 秒が 5 秒に、
1920x1200 ピクセルの画像は、約 9 秒が 2 秒に短縮しました。

---

[以降追記]
まずは、NumPy についてです。

NumPy - Wikipedia

目的
Pythonは動的型付け言語であるため、プログラムを柔軟に記述できる一方で、純粋にPythonのみを使って数値計算を行うと、ほとんどの場合C言語やJavaなどの静的型付き言語で書いたコードに比べて大幅に計算時間がかかる。そこでNumPyは、Pythonに対して型付きの多次元配列オブジェクト (numpy.ndarray) と、その配列に対する多数の演算関数や操作関数を提供することにより、この問題を解決しようとしている。NumPyの内部はC言語 (およびFortran)によって実装されているため非常に高速に動作する。したがって、目的の処理を、大きな多次元配列（ベクトル・行列など）に対する演算として記述できれば（ベクトル化できれば）、計算時間の大半はPythonではなくC言語によるネイティブコードで実行されるようになり大幅に高速化する。

したがって、以下のように配列の 1 要素ごとに Python で演算を行っていたのでは NumPy を活かすことができません。

def changeToGray( number: int, width: np.ndarray ):
    for pixel in width:
        gray = int(pixel[0]*0.3) + int(pixel[1]*0.59) + int(pixel[2]*0.11)
        pixel[0] = gray
        pixel[1] = gray
        pixel[2] = gray
    return number, width

np.tile((width * [0.3, 0.59, 0.11]).astype(np.int).sum(axis=1), (3, 1)).T は、
“配列に対する多数の演算関数や操作関数”を使用するように置き換えたものです。

NumPy では + や * などの演算子は、配列同士の演算ができるように定義されています。
以下のように、配列 a と b に対して、a + b を行った結果は、同じ位置の要素同士を加算したものになります。

>>> a = np.arange(6).reshape(2, 3)
>>> a
array([[0, 1, 2],
       [3, 4, 5]])

>>> b = np.ones((2, 3))
>>> b
array([[1., 1., 1.],
       [1., 1., 1.]])

>>> a + b
array([[1., 2., 3.],
       [4., 5., 6.]])

また、「ブロードキャスト」といって、長さが異なる配列の演算は、不足している要素を、ルールにしたがって自動的に補完してくれます。詳しくは「NumPy ブロードキャスト」などで検索してください。

>>> c = np.array([3, 5, 7])
>>> c
array([3, 5, 7])

>>> a + c
array([[ 3,  6,  9],
       [ 6,  9, 12]])

>>> d = np.array([5])
>>> d
array([5])

>>> a + d
array([[ 5,  6,  7],
       [ 8,  9, 10]])

つまり、width * [0.3, 0.59, 0.11] は、以下の処理と等価です。

for pixel in width:
    gray = pixel[0]*0.3 + pixel[1]*0.59 + pixel[2]*0.11

この結果に対して、.astype(np.int) は配列の int 変換を、.sum(axis=1) は 1 次元の要素同士の合計(つまりR+G+B)を求めています。
そして、その合計を np.tile によって、元の長さに戻しています。

>>> e = np.tile(np.arange(5), (3, 1))
>>> e
array([[0, 1, 2, 3, 4],
       [0, 1, 2, 3, 4],
       [0, 1, 2, 3, 4]])

ただし、このままだと、0次元目と1次元目が入れ替わってしまっているため、.T で転置しています。

>>> e.T
array([[0, 0, 0],
       [1, 1, 1],
       [2, 2, 2],
       [3, 3, 3],
       [4, 4, 4]])

Comments

[reishisu]

大変申し訳ありません、引数や名前を色々修正できておりませんでした。。。
現在は修正して実行の確認もOKです。。。。

ファッ！これはとても凄いですね！！
私の環境では8コアで8000×8000の処理が５秒で終わりました(*_*;

お手数ではございますが私の理解力が足りないのでどうか戻り値の部分解説いただけないでしょうか＞＜

[reishisu]

ちなみに修正した「直したソースコード」のところを
for width in part_height:
width = np.tile((width * [0.3, 0.59, 0.11]).astype(np.int).sum(axis=1), (3, 1)).T
# for pixel in width:
# gray = int(pixel[0]*0.3) + int(pixel[1]*0.59) + int(pixel[2]*0.11)
# pixel[0] = gray
# pixel[1] = gray
# pixel[2] = gray
のように修正したら、何も実行されずに終了しました、、、

[copepoda]

回答に解説を追記しました。

「直したソースコード」の方は、width 経由で part_height の中を更新する必要があるので、
width[...] = np.tile((width * [0.3, 0.59, 0.11]).astype(np.int).sum(axis=1), (3, 1)).T
です。
width 自体を置き換えるのではなく、width[...] で中身を置き換えます。

[reishisu]

丁寧な解説ありがとうございました！！
Python始めたばかりでなんでわざわざ関数呼び出しなんてしてるんだろうと思っていたらNumpyはCで実装されていたんですね＞＜
Numpyはとても奥深いのを知ったのと線形代数をプログラミングで使うの初めてでちょっと慣れるまで時間がかかりそうです笑

疑問なのですが、Pythonって動的型付け言語なのでwidthで出来そうなのですが何故width[...]じゃ無いとできないのでしょうか？？

[reishisu]

あ、あとフォロー返していただきありがとうございます！！

[reishisu]

度々質問ばかりで申し訳ないですが、解説を読ませていただいて「直したコード」のところにあるmulchProcess関数の最後のところで

for i, height in zip( range(line_number, line_number+len(part_height)), part_height ):
img[i] = height

と書いており、line_numberで画像の高さの初めの位置を保存しており戻り値の配列をそのまま先頭から代入しております。

img[ line_number : line_number+len(part_height) ] = part_height

forを消してこのように出来ればと考えておりますが、シンタックスエラー出てしまいます。
ですので、どの様に実装すれば良いか教えて頂けないでしょうか....?

[reishisu]

大変申し訳ありません、カッコの数が多いだけでした。

おかげさまで8000×8000の画像が「元のコード」でも5秒でかなり高速化されていたのに「直したコード」で実行するとさらに1.7秒まで大幅に短縮することが出来ました！！！

本当にありがとうございました！

Answer 2

測定したい時間は以下のものです。

def changeToGray( number: int, width: np.ndarray ):
    ココから
    本来のコード
    ココまで
    return number, part_height

---

普通に並列処理させたら、普通に速くなりました。

「直したソースコード」が誤っていますが、実行すべきコードを実行できていますか？

---

私の手元では以下のコードのuseCPUを変えると正常にスケールします。

import concurrent.futures
import numpy as np
import time

img = np.ones(shape=(500,5000,3))

def main():
    useCPU = 4
    step = int(len(img) / useCPU )
    mulchProcess(useCPU=useCPU, step=step)

def changeToGray(number: int, length: int):
    s = time.time()
    endPioint = number + length  if number + length < len(img)  else  len(img) - 1
    part_height = img[number:endPioint-1]
    for width in part_height:
        for pixel in width:
            gray = int(pixel[0]*0.3) + int(pixel[1]*0.59) + int(pixel[2]*0.11)
            pixel[0] = gray
            pixel[1] = gray
            pixel[2] = gray
    t = time.time() - s
    return number, part_height, t

def mulchProcess(useCPU: int, step: int):
    index_list = [i for i in range(0, len(img), step)]
    with concurrent.futures.ProcessPoolExecutor(max_workers=useCPU) as executer:
        fs = [executer.submit(changeToGray, i, step) for i in index_list]
        for future in concurrent.futures.as_completed(fs):
            line_number = future.result()[0]
            part_height  = future.result()[1]
            t = future.result()[2]
            print(t)
            for i, height in zip(range(line_number, line_number+len(part_height)), part_height):
                img[i] = height

s = time.time()
main()
print(time.time()-s)

時間は以下のようになります。
手元のPCは4coreなので、8個使おうとしても一度に4つずつしか実行できません。
それらがキューに乗せられて順番に実行されるわけではなく、スイッチしながら実行されるので、プロセスあたり2倍の時間がかかり、結局全体の時間は同じになります。
むしろスイッチした分だけオーバーヘッドがあって全体の実行時間が長くなっています。

useCPU = 1
3.6020491123199463
total 4.26886773109436

useCPU = 2
1.7662358283996582
1.7804999351501465
total 2.166551113128662

useCPU = 4
0.9876649379730225
0.9972729682922363
1.0097663402557373
1.0141608715057373
total 1.2486088275909424

useCPU = 8
0.9724259376525879
0.9717621803283691
0.973222017288208
0.97617506980896
0.9814469814300537
0.9806389808654785
0.9841761589050293
0.9821760654449463
0.015022039413452148
total 1.2429370880126953

Comments

[reishisu]

大変申し訳ありません、引数や名前を色々修正できておりませんでした。。。
修正しましたので、実行結果を貼らせて頂きます＞＜
ーー　8コア　ーー
finish time : 41.220848083496094
finish time : 3.910064697265625e-05
finish time : 41.38852524757385
finish time : 41.44024705886841
finish time : 41.493447065353394
finish time : 41.56655478477478
finish time : 41.57504391670227
finish time : 41.58701515197754
finish time : 41.611690044403076

[reishisu]

今気が付いたのですが、何故か２行目の結果がおかしい事を確認しました....
5000*5000では謎の行が出て来て4000*3000では正しく出力されるので現在原因を調査中でございます

Answer 3

何処がボトルネックになっているか

a,numpyはループでピクセル単位に処理を行うとものすごーく時間がかかるので。
あまりテストできてませんが。

How can I convert an RGB image into grayscale in Python?を参考に。

from time import perf_counter

def changeToGray_beta( number: int, length :int):
    st_time = perf_counter()
    endPioint = number + length  if number + length < len(img)  else  len(img) - 1
    part_height = img[ number : endPioint-1 ]
    gray = np.ceil(np.dot(part_height[..., :3], [0.3, 0.59, 0.11]))
    # 1ch => 3ch
    part_height = np.stack((gray,) * 3, -1)
    # 戻り値に実行時間を追加
    return number, part_height, perf_counter() - st_time

-for future in concurrent.futures.as_completed(fs):
-    line_number = future.result()[0]
-    part_height  = future.result()[1]
+for future in concurrent.futures.as_completed(fs):
+    line_number, part_height, exe_time = future.result()
+    print(exe_time)

※numpyマニアな方なツッコミ待ちです。

b, あとはグレースケール変換部分をcv2.cvtColorを使ったこのような形にでも。

gray = cv2.cvtColor(part_height, cv2.COLOR_RGB2GRAY)

---

あと思いつく点としては、changeToGrayの処理で戻り値として3次元(RGB)データを返していますが。
グレースケールデータは1次元で十分なのでas_completed呼び出し側で一部の処理を行えば返す必要も無い気がします。
これに関しては受け渡しするpickleのデータ量とas_completed呼び出し側のCPU時間のトレードオフになるのではないかと、プロファイリングを複数回採ってみた方がよいと思います。

---

テスト中に気づいたのですが、質問文の元コードだと白色のグレースケール値が254になります。

こんな感じのハッシュ関数を作ってテストコードに入れておくのをお勧め致します。

def image_hash(img):
    from hashlib import sha384
    x = np.ascontiguousarray(img, dtype=np.uint8)
    print(sha384(x).hexdigest())

◇参考情報
Fast way to Hash Numpy objects for Caching

Comments

[mkgrei]

全く関係ない雑談ですが。

一枚が処理終わるまでの時間を減らさなければならないのでないならば、画像ごとに並列化した方がミスも少なくて良いと思います。

今回のようなケースでは、①並列化すると時間が短縮されるのは自明で、②pythonのマルチプロセスを試すのよい題材でもなく、③opencvライブラリの処理を使うと圧倒的に速く、④numpyのベルトル処理を使っても容易に速くなることを踏まえて、やりたいのことの意図を掴みかねているのですが。
段階的にコードの高速化を実現した、という実績が欲しいのでなければ何も想像力が働かない次第。

ちなみにnp.ceilのデフォルトだと整数型にならないので、場合によっては困るかも。プロットする時にたまに整数を必要とする。
.astype('i')だけでも良いような。±1は誤差でしょ。

マルチプロセスで変数をまるごと共有するとコピーが発生して遅いので、プロセスに渡す時点で画像を先に分割しておくと、少し高速化が望めるかも。
今回の場合、プロセス間でシェア変数にしてもバグらないはず。
ちょっとトリックが必要になりそうですが。

[umyu]

@mkgreiさんへ
.astype('i')の件、参考になりました。
指摘ありがとうございます。