スレッドとプロセス

スレッド（thread）とプロセス（process）の簡単な例を作ってみた。スレッドでは、スレッド間の通信は容易であるが、複数のスレッドは、各スレッドの処理が他のスレッドの処理と部分的に順番に交代で実行されるので、スレッドの数に応じて処理時間が長くなる。これに対して、プロセスは複数のCPUコアで同時に実行されるので、プロセスの管理処理の時間の増加はあるが、プロセス数に比例して処理時間が増えるということはない。これらを確認するためのスレッドのプログラム例とプロセスのプログラム例を作成してみた。

スレッドでは、スレッドとの通信が容易であるので、入出力機器の監視を行いながら処理を行うときに便利である。入出力機器の監視に処理の待ち時間がある場合は、スレッドとして処理を行えば、他の処理を並行して行うことができるので、待ち時間の影響を抑えることができる。

プロセスの場合は、複数のプロセスが複数のCPU（コア）によって処理が行われるので、長時間掛かる計算を複数のプロセスに分割して行うことにより、計算時間の短縮が可能になる。

スレッドの例

マルチスレッド（multi-thread）の処理時間を確認するスクリプト例と、スレッド間のデータのやり取りを行うスクリプト例の２つを用意した。まず、処理時間の比較例を示し、次にスレッド間の通信例を示す。なお、Windows 10の場合、実行はPython IDLEでも可能であるが、端末画面においてコマンドpythonによって実行する方が、出力が見やすい。

まず、処理時間の比較の基準とするため、以下のスクリプトを用意した。

def calc_sum(d, nm):

sum_v = 0.0

i = 0

for v in d:

sum_v += v

i += 1

if i % 1_000_000 == 0:

print(nm, i)

return sum_v

n = 10_000_000

data = ss.norm.rvs(size = n)

start_t = time.time()

sum_v = calc_sum(data, '0:')

end_t = time.time()

print('time = ', end_t - start_t)

print('result: ', sum_v, '\n')

このスクリプトの実行で、以下の出力を得る。

0: 1000000

0: 2000000

0: 3000000

0: 4000000

0: 5000000

0: 6000000

0: 7000000

0: 8000000

0: 9000000

0: 10000000

time = 2.648660898208618

result: -3262.383797592117

処理時間は、約2.6秒である。

上のスクリプトにおける関数calc_sumの実行をスレッドで行うため、以下のスクリプトを用意した。

class Calc_Sum(Thread):

def __init__(self, d, td_name):

super().__init__()

self.d = d

self.my_name = td_name

def run(self):

self.sum_v = calc_sum(self.d, self.my_name)

# 1 thread

threads = [Calc_Sum(data, 'id-0')]

start_t = time.time()

for thread in threads:

thread.start()

for thread in threads:

thread.join()

end_t = time.time()

print('time =', end_t - start_t)

for t in threads:

print('result:', t.sum_v)

print('\n')

スレッドは、クラス型Threadを継承したクラスとして用意し、実行処理内容はメソッドrunに記述する。このメソッドrunは、メソッドstartの呼び出しで実行される。スレッドの実行終了はメソッドjoinの呼び出しでチェックできる。メソッドjoinは、メソッドの終了を待つものである。

スレッドは１本である。実行時の出力は以下のようになった。

id-0 1000000

id-0 2000000

id-0 3000000

id-0 4000000

id-0 5000000

id-0 6000000

id-0 7000000

id-0 8000000

id-0 9000000

id-0 10000000

time = 2.708895683288574

result: -3262.383797592117

処理時間は約2.7秒である。先のスレッドでない場合は約2.6秒であったことに注意。スレッドとして実行するための処理時間の増加が見られる。

次に、スレッドを増やして４本のスレッドを実行するスクリプトを以下のように用意した。

# 4 threads

threads = [Calc_Sum(data, f'ID-{i+1}') for i in range(4)]

start_t = time.time()

for thread in threads:

thread.start()

for thread in threads:

thread.join()

end_t = time.time()

print('time =', end_t - start_t)

for t in threads:

print(f'{t.my_name}: results = {t.sum_v}')

このスクリプトを実行すると、以下のような出力になる。

ID-3 1000000

ID-2 1000000

ID-1 1000000

ID-4 1000000

ID-3 2000000

ID-1 2000000

ID-2 2000000

　　・

ID-4 8000000

ID-1 9000000

ID-2 10000000

ID-4 9000000

ID-1 10000000

ID-4 10000000

time = 11.90782356262207

ID-1: results = -3262.383797592117

ID-2: results = -3262.383797592117

ID-3: results = -3262.383797592117

ID-4: results = -3262.383797592117

計算の途中経過の出力から４本のスレッドが並行して実行されていることが分かるが、各スレッドの実行は部分的に進み、各瞬間には１つのスレッドのみが実行されている。このことは、処理時間が約11.9秒とスレッドが１本の場合の約４倍になっていることで確認できる。

上のスクリプト例の完全なスクリプトリストをリスト１に示す。

リスト１

import scipy.stats as ss

from threading import Thread

import time

def calc_sum(d, nm):

sum_v = 0.0

i = 0

for v in d:

sum_v += v

i += 1

if i % 1_000_000 == 0:

print(nm, i)

return sum_v

n = 10_000_000

data = ss.norm.rvs(size = n)

start_t = time.time()

sum_v = calc_sum(data, '0:')

end_t = time.time()

print('time = ', end_t - start_t)

print('result: ', sum_v, '\n')

class Calc_Sum(Thread):

def __init__(self, d, td_name):

super().__init__()

self.d = d

self.my_name = td_name

def run(self):

self.sum_v = calc_sum(self.d, self.my_name)

# 1 thread

threads = [Calc_Sum(data, 'id-0')]

start_t = time.time()

for thread in threads:

thread.start()

for thread in threads:

thread.join()

end_t = time.time()

print('time =', end_t - start_t)

for t in threads:

print('result:', t.sum_v)

print('\n')

# 4 threads

threads = [Calc_Sum(data, f'ID-{i+1}') for i in range(4)]

start_t = time.time()

for thread in threads:

thread.start()

for thread in threads:

thread.join()

end_t = time.time()

print('time =', end_t - start_t)

for t in threads:

print(f'{t.my_name}: results = {t.sum_v}')

実行は、リスト１のファイル名がcheck_thread.pyのときは、pythonコマンドを

PS D:\ThreadPool\Prgs> python check_thread.py

と実行すればよい。

スレッド間の通信

計数のスレッドを２つ走らせ、それらのスレッドを監視するスレッドを１つ用意する。

計数用スレッドは以下のように用意した。

class Counting(Thread):

def __init__(self, ck):

super().__init__()

self.ck = ck # Threadの状態

self.id = ck # ThreadのID

self.a = None # 他のThread

self.ps = 0 # printの使用状態

self.counter = 0 # カウンター

def run(self):

while self.ck != 0:

self.counter += self.ck

while self.a.ps == 1: # 他のThreadのprint使用の空きを待つ

pass

self.ps = 1

if self.ck < 0:

print('{0:>10s}'.format(' '),end = '')

print(f'{self.id}: {self.counter}')

self.ps = 0

s = time.time()

while time.time() - s < 1:

pass

スレッドの状態を表すアトリビュートckは、係数アトリビュートcounterに加えられる値であるが、このckが0になるとスレッドの実行は修了する。

他のスレッドをアトリビュートaで表し、そのスレッドの状態のチェックができるようにする。

アトリビュートpsは、画面出力printの使用状態を表す。１ならば、使用中である。

計数用スレッドの管理用スレッドを以下のように用意した。

class Man(Thread):

def __init__(self, c1, c2):

super().__init__()

self.c1 = c1

self.c2 = c2

c1.a = c2

c2.a = c1

def run(self):

s = time.time()

while time.time() - s < 10.1:

pass

# ckの値を0に設定してThreadの終了

self.c1.ck = 0

self.c2.ck = 0

所定の時間（10.1秒）が経つと、係数スレッドのアトリビュートckを０に設定して終了させる。

上の２種類のスレッドを実行するスクリプト例をリスト１aのように用意した。

リスト１a

import scipy.stats as ss

from threading import Thread

import time

class Counting(Thread):

def __init__(self, ck):

super().__init__()

self.ck = ck # Threadの状態

self.id = ck # ThreadのID

self.a = None # 他のThread

self.ps = 0 # printの使用状態

self.counter = 0 # カウンター

def run(self):

while self.ck != 0:

self.counter += self.ck

while self.a.ps == 1: # 他のThreadのprint使用の空きを待つ

pass

self.ps = 1

if self.ck < 0:

print('{0:>10s}'.format(' '),end = '')

print(f'{self.id}: {self.counter}')

self.ps = 0

s = time.time()

while time.time() - s < 1:

pass

# Threadの監視用Thread

class Man(Thread):

def __init__(self, c1, c2):

super().__init__()

self.c1 = c1

self.c2 = c2

c1.a = c2

c2.a = c1

def run(self):

s = time.time()

while time.time() - s < 10:

pass

# ckの値を0に設定してThreadの終了

self.c1.ck = 0

self.c2.ck = 0

thread1 = Counting(1)

thread2 = Counting(-1)

thread1.a = thread2

thread2.a = thread1

threadM = Man(thread1, thread2)

thread1.start()

thread2.start()

threadM.start()

thread1.join()

thread2.join()

threadM.join()

print('Threadの終了.')

リスト１aのファイル名がcheck_1_thread.pyであるときの実行例を以下に示す。

PS D:\ThreadPool\Prgs> python .\check_1_thread.py

1: 1

-1: -1

1: 2

-1: -2

1: 3

-1: -3

1: 4

-1: -4

1: 5

-1: -5

1: 6

-1: -6

1: 7

-1: -7

1: 8

-1: -8

1: 9

-1: -9

1: 10

-1: -10

Threadの終了.

プロセスの例

マルチプロセス（multi-process）の処理時間を、以下の手順で確認した。なお、マルチプロセスのプログラムは、Windows 10では、Python IDLEでの実行ではトラブルが生じたが、端末画面においてコマンドpythonによる実行では問題なかった。

マルチプロセスの実行は、クラス型Poolを用いて行う方法を用いるが、このときマルチプロセスの実行は次のif文

if __name__ == '__main__':

のブロック（スイート：suite）内で行う必要がある。ここで、変数__name__はこのスクリプトを実行しているプログラムの名前であり、メインの実行プログラムのみ名前が’__main__’である。マルチプロセシングでメイン以外の実行プロセスから読み込まれているときは’__main__’以外の名前が与えられる。従って、上記のif文のブロックは、最初のメインのプログラムとして実行されている場合のみ処理の対象となる。

また、input関数の実行も、このif文のブロック内で行うが、Poolとinputについてはこのウェブサイトで説明している。

マルチプロセシングの実行時間測定の基準を得るために、まず、次のスクリプトを用意する。

def calc_sum(i_d):

d = i_d[1]

sum_v = 0.0

print(f'{i_d[0]} started.!')

for v in d:

sum_v += v

print(f'{i_d[0]} ended.')

return i_d[0], sum_v

if __name__ == '__main__':

n = 30_000_000

data = ss.norm.rvs(size = n)

start_t = time.time()

_, sum_v = calc_sum((0, data))

end_t = time.time()

print('time =', end_t - start_t)

print('result =', sum_v)

関数calc_sumの宣言が、if文

if __name__ == '__main__':

の前（スイート外）で行われていることに注意。このことにより、メイン以外のプロセスからこの関数宣言を読むことができる。

また、スクリプトによる処理、例えば、

data = ss.norm.rvs(size = n)

は、if文のスイート内に書かれているが、これにより上記の実行はメインのプログラムでのみ行われることになる。

上のスクリプトの実行時の出力は、以下のようになる。

0 started.!

0 ended.

time = 4.180828332901001

result = -493.9780580970286

関数calc_sum((0, data))の実行時間が、約4.2秒であることが分かる。

まず、同じ処理をプロセスとして実行すると、処理時間がどうなるかを調べるため、以下のスクリプトを用意した。

pool = Pool()

# 1 process

params = [(0, data)]

print()

start_t = time.time()

results = pool.map(calc_sum, params)

end_t = time.time()

print('time =', end_t - start_t)

print('result =',results)

プロセスはPool型のオブジェクトのmapメソッドにより実行することができる。上のスクリプトでは、mapメソッドが

results = pool.map(calc_sum, params)

と呼び出されている。メソッドmapの第1引数にプロセスとして実行する関数名を置き、第２引数に関数のパラメータのリストを置く。リスト内の各パラメータに対応して第１引数で指定された関数がプロセスとして実行される。プロセスとして実行された関数の戻り値は、メソッドmapの戻り値であるリストの要素として返される。

実行時の出力は以下のようである。

0 started.!

0 ended.

time = 6.0625505447387695

result = [(0, -493.9780580970286)]

実行時間が、約6.1秒となり、先の単に関数の実行の場合の時間約4.2秒より長くなっている。これは、プロセスとしての処理が加わったためである。

次に、プロセスの数を４本にして実行するスクリプトを以下のように用意した。

# 4 processes

params = [(i+1, data) for i in range(4)]

print()

start_t = time.time()

results = pool.map(calc_sum, params)

end_t =time.time()

print('time =', end_t - start_t)

print('results:')

for r in results:

print(r)

実行時の出力は以下のようである。

1 started.!

2 started.!

3 started.!

4 started.!

1 ended.

2 ended.

4 ended.

3 ended.

time = 7.732580661773682

results:

(1, -493.9780580970286)

(2, -493.9780580970286)

(3, -493.9780580970286)

(4, -493.9780580970286)

プロセス数が１本であるときの処理時間は約6.1秒であったが、４本にしたときの上の実行結果では約7.7秒になっている。１本のときの処理時間約６秒の４倍である24秒より短く、１本のときの処理時間約6.1秒に近い。４本のプロセスが複数のCPU（Core）で同時に処理されていることが分かる。6.1秒より少し長いのは、プロセスの管理のための時間と思われる。

上のプロセスのスクリプト例を１つの完全なスクリプトとしてリスト２に示す。

リスト２

import scipy.stats as ss

from multiprocessing.pool import Pool

import time

import copy

def calc_sum(i_d):

d = i_d[1]

sum_v = 0.0

print(f'{i_d[0]} started.!')

for v in d:

sum_v += v

print(f'{i_d[0]} ended.')

return i_d[0], sum_v

if __name__ == '__main__':

n = 30_000_000

data = ss.norm.rvs(size = n)

start_t = time.time()

_, sum_v = calc_sum((0, data))

end_t = time.time()

print('time =', end_t - start_t)

print('result =', sum_v)

pool = Pool()

# 1 process

params = [(0, data)]

print()

start_t = time.time()

results = pool.map(calc_sum, params)

end_t = time.time()

print('time =', end_t - start_t)

print('result =',results)

# 4 processes

params = [(i+1, data) for i in range(4)]

print()

start_t = time.time()

results = pool.map(calc_sum, params)

end_t =time.time()

print('time =', end_t - start_t)

print('results:')

for r in results:

print(r)

リスト２を実行すると以下のような出力を得る。

0 started.!

0 ended.

time = 4.180828332901001

result = -493.9780580970286

0 started.!

0 ended.

time = 6.0625505447387695

result = [(0, -493.9780580970286)]

1 started.!

2 started.!

3 started.!

4 started.!

1 ended.

2 ended.

4 ended.

3 ended.

time = 7.732580661773682

results:

(1, -493.9780580970286)

(2, -493.9780580970286)

(3, -493.9780580970286)

(4, -493.9780580970286)

リスト２のスクリプトのファイル名がcheck_pool.pyのときは、次のコマンドpythonで実行することができる。

PS D:\ThreadPool\Prgs> python check_pool.py