multiprocessingのcsvへの書き込みについて

Question

閲覧ありがとうございます。
プログラムの処理を早くできればと思い、並列処理をプログラムに組み込んでみたのですがcsvファイルに書き込む際に思った通りに動作せず困っております。
キャリアとしてまだ半年ほどで、初歩的なところを勘違いしているかもしれません。

目的は、関数内で生成した値をnp配列により結合していき、最終的にまとまった生成値をcsvファイルに書き込むことです。
こちらの都合で申し訳ないのですが、multiprocessingを使用する際にmapを使用せずにfor文の繰り返しでプログラムができればと考えております。

そのまま関数を実行するとcsvファイルに書き込まれるのに対して
並行処理を組み込んで実行するとcsvファイルに書き込みがされません。

プログラムは以下のように作成しました。
ご教授いただければ幸いです。
よろしくお願いします。

import numpy as np
import multiprocessing

a = np.array([0])
b = np.array([0])

def test():
    global a,b
    a = np.random.randn(2)
    b = np.r_[b,a]

if __name__ == "__main__":  
    for i in range (10):
        p = multiprocessing.Process(target=test)
        p.start()
        #test()
    np.savetxt('sample.csv',b)

Answer 1

以下の問題点があります。

複数のスレッドで global 変数経由でデータの共有はできない。

スレッドを作成した際、スレッドごとに別のコンテキストが作成されるため、global 変数経由でデータを共有することはできないです。
(スレッドで実行した関数が参照しているグローバル変数 b と親スレッドのグローバル変数 b は別物)

Python multiprocessing global variable updates not returned to parent

複数スレッドでデータの共有をしたい場合は、multiprocessing.Queue を使います。

numpy.random はスレッドセーフではない

numpy.random はスレッドセーフではないので、標準ライブラリの random を使ったほうがいいです。

Differences between numpy.random and random.random in Python

スレッドの終了を待つ

Process.join で待たないと、スレッドがデータを生成するより前に csv の書き込みの処理が実行される可能性があるので、すべてのスレッドが完了するのを待ちましょう。

[t.join() for t in threads]

修正したコード

import random
from multiprocessing import Process, Queue

import numpy as np

def test(queue):
    a = random.randint(0, 10)
    queue.put(a)

if __name__ == "__main__":
    queue = Queue()

    threads = []
    for i in range (10):
        t = Process(target=test, args=(queue,))
        t.start()
        threads.append(t)

    # 全部のスレッドが完了するまで待機
    [t.join() for t in threads]

    # キューのデータを numpy 配列に変換する。
    data = []
    while not queue.empty():
        data.append(queue.get())
    data = np.array(data)
    print(data)

    # ファイルに書き込む。
    np.savetxt('sample.csv', data)

    with open('sample.csv') as f:
        print(f.read())

[4 2 8 1 0 8 9 6 7 9]
4.000000000000000000e+00
2.000000000000000000e+00
8.000000000000000000e+00
1.000000000000000000e+00
0.000000000000000000e+00
8.000000000000000000e+00
9.000000000000000000e+00
6.000000000000000000e+00
7.000000000000000000e+00
9.000000000000000000e+00

Comments

[tsujitsuji]

回答ありがとうございます。
複数のスレッドではglobal変数でデータの共有ができないこと、numpy.randomを使用しないほうが良いこと、またスレッドの終了を待たないと先に次の処理が実行されてしまうことを理解しました。
ずっと悩んでおりましたので質問してよかったです。
これで研究も進みそうです。
ありがとうございました。