Pythonマルチプロセシング
Poolの利用
現代のPCはマルチプロセシングがサポートされている。Pythonにおけるマルチプロセシングの簡単な例を、複数のフォームをプロセスに対応して表示する例と、計算を分割して計算時間の短縮を図る例を用意してみた。計算の分割例は、本サイトの後半で示す。まず、複数のフォームをマルチプロセッシングで表示する例について示す。
複数のフォームの表示をマルチプロセッシングで行う例をリスト１に示す。リスト１のスクリプトでは、リスト２の３つの関数を呼び出して同時に実行している。３つの関数それぞれはtkinterパッケージのTkクラスで作成したウィンドウを表示するものである（図１）。.
図１
これらのウィンドウは10秒後に消えて、プログラムの実行終了となる。
マルチプロセシングは次のコードによって実行されている。
pool = Pool()
Results = pool.map(func_r,
params)
このコードの実行は、次の条件が満たされていることが必要である。
__name__ == '__main__'
上の条件を無視して実行すると、プログラムはハングアップする。
マルチプロセシングで呼び出される関数func_rは、リスト２で宣言されている。マルチプロセシングで呼び出される関数は、引数の個数が１個である。この１個の引数を項目とするリストをmapの実引数として設定すると、リストの各項目に対応して関数func_rのプロセスが実行される。リスト２の関数func_rでは、リストの項目である辞書のキーIDの値に応じて、呼び出す関数が選ばれている。
マルチプロセシングの関数mapの関数値は、マルチプロセシングで実行された各関数の関数値を項目とするリストである（図２）。
図２
リスト１および２のファイルは、圧縮ファイルにまとめた。
from multiprocessing.pool import Pool
from submodule import *
print('Started...')
params_t = {'ID': 0, 'r':0, 'g':0,
'b':0}
params = []
for iproc in range(3):
p_t = params_t.copy()
p_t['ID'] = iproc + 1
if iproc == 0:
p_t['r'] = 255
elif iproc == 1:
p_t['g'] = 255
else:
p_t['b'] = 255
params.append(p_t)
if __name__ == '__main__':
pool = Pool()
Results = pool.map(func_r,
params)
print(Results)
from tkinter import *
import time
def rgb2str(r, g, b):
return
'#{0:0>2x}{1:0>2x}{2:0>2x}'.format(r,g,b)
def func1(r, g, b):
root = Tk()
c = Canvas(root, width =
100, height = 200, background = rgb2str(0, 255, 255))
c.pack()
for i in range(10):
c.create_rectangle(0, 0, 100, 200, outline = rgb2str(r, g, b), fill =
rgb2str(r, g, b))
c.update()
r =
255 - r
g =
255 - g
b =
255 - b
time.sleep(1)
root.destroy()
return 'func1'
def func2(r, g, b):
root = Tk()
c = Canvas(root, width =
200, height = 100, background = rgb2str(0, 255, 255))
c.pack()
for i in range(10):
c.create_rectangle(0, 0, 200, 100, outline = rgb2str(r, g, b), fill =
rgb2str(r, g, b))
c.update()
r =
255 - r
g =
255 - g
b =
255 - b
time.sleep(1)
root.destroy()
return 'func2'
def func3(r, g, b):
root = Tk()
c = Canvas(root, width =
150, height = 150, background = rgb2str(0, 255, 255))
c.pack()
for i in range(10):
c.create_rectangle(0, 0, 150, 150, outline = rgb2str(r, g, b), fill =
rgb2str(r, g, b))
c.update()
r =
255 - r
g =
255 - g
b =
255 - b
time.sleep(1)
root.destroy()
return 'func3'
def func_r( params ):
results = None
if params['ID'] == 1:
results = func1(params['r'], params['g'], params['b'])
elif params['ID'] == 2:
results = func2(params['r'], params['g'], params['b'])
else:
results = func3(params['r'], params['g'], params['b'])
return results, params['ID']
大きな計算は分割して並列実行すると計算時間の短縮が期待できる。
計算時間の掛かる例として、２項係数およびフィボナッチ数を再帰的にそのまま計算する方法を取り上げる。
リストA.1のスクリプトでは、2項係数とフィボナッチ数は次のスクリプトで計算されている。なお、リストA.1と後で説明するリストA.2のスクリプトファイルは、圧縮ファイルNumSmpls.zipとしてまとめた。ダウンロード・展開して、自由に利用されたい。
def bincoeff(n, k): # Binomial coefficient
if n==k or k==0:
return 1
else:
return bincoeff(n-1, k-1) + bincoeff(n-1, k)
def Fib(n): # Fibonacci number
if n==0 or n==1:
return 1
else:
return Fib(n-1) + Fib(n-2)
関数の実行例のコードは
n = 27
k = 15
v1 =
time.perf_counter()
v = bincoeff(n, k)
v2 = time.perf_counter()
print(f'v = {v}, n = {n}, k = {k}')
print('time =', v2-v1)
n = 36
v1 = time.perf_counter()
v = Fib(n)
v2 = time.perf_counter()
print(f'v = {v}, n = {n}')
print('time =', v2-v1)
であり、実行結果の出力は、以下の通りである。
v = 17383860, n = 27, k = 15
time = 4.410504400002537
v = 24157817, n = 36
time = 4.842723500012653
２項係数の計算に約4.4秒、フィボナッチ数の計算に約4.8秒掛かっている。これは、使用PCに依存する値である。
なお、スクリプトの実行部分は、次のif文内であることに注意。
if __name__
== '__main__':
これは、マルチプロセッシングでは、各プロセスに対応してPythonインタープリターが起動するので、実行部分は名前が__main__のプロセスのみに限定して、実行の重なりを防ぐためである。最初に起動されたPythonインタープリターによる実行プロセスに名前__main__が与えられる。
２つの関数の計算を、次のコードにより同時に行っている。
Results = pool.map(ck_func, params)
パラメータリストparamsの各要素に対応して関数ck_funcが別個のプロセスとして実行される。各関数の戻り値は、paramsのリストに対応した順序でResultsのリストの要素として得ることができる。
結果は、以下の通りである。
Results =
[(17383860, 27, 15), (24157817, 36)]
time = 5.719632100022864
計算時間は、約5.7秒で、上の2つの関数の計算時間の和、約９秒より短い。２つの関数の計算が同時並行に行われているが、1つ1つの関数の計算時間、約4.4秒、および約4.8秒より長いのは、プロセスの並列実行のための制御時間であると思われる。
リストA.1では、2つの関数の並列処理例であるが、もっと多くの関数の並列処理例をリストA.2に示す。
from multiprocessing.pool import Pool
import time
def bincoeff(n, k): # Binomial coefficient
if n==k or k==0:
return 1
else:
return bincoeff(n-1, k-1) + bincoeff(n-1, k)
def Fib(n): # Fibonacci number
if n==0 or n==1:
return 1
else:
return Fib(n-1) + Fib(n-2)
def ck_func(param):
"""
パラメータparamに対応して関数がプロセスとして実行される
"""
if param[0] == 'B':
v =
bincoeff(param[1], param[2])
return v, param[1], param[2]
if param[0] == 'F':
v =
Fib(param[1])
return v, param[1]
else:
return None
if __name__ == '__main__':
n = 27
k = 15
v1 =
time.perf_counter()
v = bincoeff(n, k)
v2 = time.perf_counter()
print(f'v = {v}, n = {n}, k = {k}')
print('time =', v2-v1)
n = 36
v1 = time.perf_counter()
v = Fib(n)
v2 = time.perf_counter()
print(f'v = {v}, n = {n}')
print('time =', v2-v1)
params = [['B', 27, 15],
['F', 36]] # パラメータのリスト
pool = Pool()
# マルチプロセッシング用のプールの作成
v1 = time.perf_counter()
#
# パラメータのリストparamsの各要素に対して
# ck_funcがプロセスとして実行される
# 返り値Resultsには、パラメータのリストparamsの各パラメータに
# 対応して実行されたck_funcの返り値がリストの要素として返される
#
Results = pool.map(ck_func,
params)
v2 = time.perf_counter()
print('Results =\n',
Results)
print('time =', v2-v1)
from multiprocessing.pool import Pool
import time
def bincoeff(n, k):
if n==k or k==0:
return 1
else:
return bincoeff(n-1, k-1) + bincoeff(n-1, k)
def bins(param):
v = bincoeff(param[0],
param[1])
return v, param
if __name__ == '__main__':
#
# ２項係数25_C_iを計算するためのパラメータリストの作成
#
params = []
for i in range(26):
params.append([25, i])
#
# ２項係数を順番に計算してリストb_vsに格納
#
b_vs = []
v1 = time.perf_counter()
for p in params:
v =
bincoeff(p[0], p[1])
b_vs.append([v, p])
v2 = time.perf_counter()
print('Values...\n', b_vs)
print('time =', v2-v1)
pool = Pool()
# マルチプロセッシング用プールの作成
v1 = time.perf_counter()
Results = pool.map(bins,
params) # パラメータリストのパラメータに対応して
v2 =
time.perf_counter()
# 関数binsをプロセスとして実行
print('Results...\n',
Results) # Resultsには各プロセスの返り値が
print('time =',v2-v1)
# パラメータリストparamsの順序に対応して
# 返される。
リストA.2では、まず次のスクリプトにより2項係数
の値を順番に求めている。
v1 = time.perf_counter()
for p in params:
v =
bincoeff(p[0], p[1])
b_vs.append([v, p])
v2 = time.perf_counter()
print('Values...\n', b_vs)
print('time =', v2-v1)
実行結果は、以下の通りである。
Values...
[[1, [25, 0]], [25, [25, 1]], [300, [25,
2]], [2300, [25, 3]], [12650, [25, 4]], [53130, [25, 5]], [177100, [25, 6]],
[480700, [25, 7]], [1081575, [25, 8]], [2042975, [25, 9]], [3268760, [25, 10]],
[4457400, [25, 11]], [5200300, [25, 12]], [5200300, [25, 13]], [4457400, [25,
14]], [3268760, [25, 15]], [2042975, [25, 16]], [1081575, [25, 17]], [480700,
[25, 18]], [177100, [25, 19]], [53130, [25, 20]], [12650, [25, 21]], [2300,
[25, 22]], [300, [25, 23]], [25, [25, 24]], [1, [25, 25]]]
time = 8.562252600007923
約8.6秒掛かっている。
これらの2項係数の計算を、次のコードで並列に行ってみる。
v1 = time.perf_counter()
Results = pool.map(bins,
params) # パラメータリストのパラメータに対応して
v2 =
time.perf_counter()
# 関数binsをプロセスとして実行
print('Results...\n',
Results) # Resultsには各プロセスの返り値が
print('time =',v2-v1)
# パラメータリストparamsの順序に対応して
# 返される。
結果は、以下のようになる。
Results...
[(1, [25, 0]), (25, [25, 1]), (300, [25,
2]), (2300, [25, 3]), (12650, [25, 4]), (53130, [25, 5]), (177100, [25, 6]),
(480700, [25, 7]), (1081575, [25, 8]), (2042975, [25, 9]), (3268760, [25, 10]),
(4457400, [25, 11]), (5200300, [25, 12]), (5200300, [25, 13]), (4457400, [25,
14]), (3268760, [25, 15]), (2042975, [25, 16]), (1081575, [25, 17]), (480700,
[25, 18]), (177100, [25, 19]), (53130, [25, 20]), (12650, [25, 21]), (2300,
[25, 22]), (300, [25, 23]), (25, [25, 24]), (1, [25, 25])]
time = 2.226495699986117
実行時間は、約2.2秒で、順番に行ったときの実行時間約8.6秒の1／4である。
複数のプロセスを平行に実行した場合、全体の計算時間は最も時間の掛かるプロセスにより決まる。また、上の例の場合、26個のプロセスが実行されるので、PCに実装されているコア数を越えている。そのため、すべてのプロセスを同時に実行することは不可能であり、各プロセスは実装コア数の制限内で順番に実行されることになる。また、これらのプロセスの実行の制御のための時間も掛かる。これらの事情により、処理時間が1／4という結果が得られたと思われる。
PCに搭載されているコア数は、次のように関数cpu_count()で知ることができる。
>>> import multiprocessing as
mp
>>> mp.cpu_count()
8
>>>
コア数を考慮して、計算の分割を工夫すればよい。
また、プロセスの実行時間が数分以上になるときは、計算がどこまで進んでいるか知りたくなる。このときは、プロセス内ではprint()文などは使えないが、各プロセスごとにフォームを表示して処理状況を示すことができる。各プロセスからフォームを表示する方法は、本サイトの前半で説明しているが、コルーチンの場合の例を示すウェブサイトも参考になると思われる。
なお、Pythonによるデータ分析の解説書＜岡本安晴「いまさら聞けないPythonでデータ分析」丸善出版＞も用意している。