概要

Python3でクラスを定義して、メソッド中にクラス変数を書き換える手続きを記載しつつ、それをマルチプロセスで処理したらクラス変数がどうなるのかの確認。

クラスの割当→並列処理→計算

プログラム

test_main.py

#!/usr/bin/env python
# vim:fileencoding=utf-8
#Author: Hamano TASUKE
#Created: 2016-01-05

import sys
from test_class import TestMultiProcess
argvs = sys.argv
argc = len(argvs)

def calculate_helper(class_name, method_name, *args, **kwargs):
    return getattr(class_name, method_name)(*args, **kwargs)

def main():

    alpha = 2
    beta = 3
    test = TestMultiProcess(alpha, beta)
    array = [[1,2,3], [2,3,4], [4,5,6,7],[1,2,3]]
    result = test.parallel_calculation(4, array)

    print("Multiprocess result")
    print(result)
    print("Class-inner variable alpha after calculation:%i" % test.get_alpha())
    print("Class-inner variable beta after calculation:%i" % test.get_beta())

if __name__ =='__main__':
    main()

test_class.py

#!/usr/bin/env python
# vim:fileencoding=utf-8
#Author: Hamano TASUKE
#Created: 2016-01-05

import numpy as np
import multiprocessing as mp
import itertools as itrt
from time import sleep

class TestMultiProcess:
    def __init__(self, alpha, beta):
        self.alpha = alpha
        self.beta = beta

    def calculate(self, a):
        sleep(5)
        print(a)
        a_np = np.array(a)
        a_np = a_np * 2
        self.alpha += len(a)
        self.beta += len(a)**2
        return a_np

    def parallel_calculation(self, core_num, array):
        pool = mp.Pool(core_num)
        args = itrt.izip(itrt.repeat(self), itrt.repeat('calculate'), array)
        result = pool.map(tomap, args)
        return result

    def get_alpha(self):
        return self.alpha

    def get_beta(self):
        return self.beta

def toapply(class_name, method_name, *args, **kwargs):
    return getattr(class_name, method_name)(*args, **kwargs)

def tomap(args):
    return getattr(args[0], args[1])(*args[2:])

結果

[2, 3, 4]
[4, 5, 6, 7]
[1, 2, 3]
[1, 2, 3]
Multiprocess result
[array([2, 4, 6]), array([4, 6, 8]), array([ 8, 10, 12, 14]), array([2, 4, 6])]
Class-inner variable alpha after calculation:2
Class-inner variable beta after calculation:3

メソッド内では変更したクラス変数が使用されますが、プロセスの外側ではクラス変数が変化しないままのようです。
pool.mapやpool.applyはクラスのメソッドインスタンスを受け付けないので、multiprocessing.Poolがエラーになる人へから拝借したヘルパー関数を使用しています。

並列処理→クラスの割当→計算

プログラム

test_main.py


#!/usr/bin/env python
# vim:fileencoding=utf-8
#Author: Hamano TASUKE
#Created: 2016-01-17

import sys
import multiprocessing as mp
from test_class import TestMultiProcess
def arg_wrapper(args):
    return args[0](*args[1:])

def assign_and_calculation(alpha, beta, array):
    test = TestMultiProcess(alpha, beta)
    calculate_result = test.calculate(array)
    return (test.get_alpha(), test.get_beta(), calculate_result)

def main():
    alpha = 2
    beta = 3
    array = [[1,2,3], [2,3,4], [4,5,6,7],[1,2,3]]
    argument = [(assign_and_calculation, alpha, beta, each) for each in array]

    pool = mp.Pool(4)
    result = pool.map(arg_wrapper, argument)

    print("Multiprocess result")
    print([each[2] for each in result])
    print("Class-inner variable alpha after calculation:%s" % str([each[0] for each in result]))
    print("Class-inner variable beta after calculation:%s" % str([each[1] for each in result]))

if __name__ =='__main__':
    main()

test_class.py

#!/usr/bin/env python
# vim:fileencoding=utf-8
#Author: Hamano TASUKE
#Created: 2016-01-17

import numpy as np
import itertools as itrt
from time import sleep

class TestMultiProcess:
    def __init__(self, alpha, beta):
        self.alpha = alpha
        self.beta = beta

    def calculate(self, a):
        sleep(5)
        print(a)
        a_np = np.array(a)
        a_np = a_np * 2
        self.alpha += len(a)
        self.beta += len(a)**2
        return a_np

    def get_alpha(self):
        return self.alpha

    def get_beta(self):
        return self.beta

結果

[4, 5, 6, 7]
[1, 2, 3]
[1, 2, 3]
[2, 3, 4]
Multiprocess result
[array([2, 4, 6]), array([4, 6, 8]), array([ 8, 10, 12, 14]), array([2, 4, 6])]
Class-inner variable alpha after calculation:[5, 5, 6, 5]
Class-inner variable beta after calculation:[12, 12, 19, 12]

インスタンス宣言に使う変数をヘルパー関数でラッピングすることで複数の引数を渡す事が可能です(Python で並列計算 (multiprocessing モジュール) | 複数の引数を取る関数を map() メソッドで並列に走らせるより)。これを利用して並列処理中でインスタンスを宣言することで、クラス内変数の計算も可能です(あまり使いみちないかも…)。

参考