[python] 멀티 프로세싱 : 클래스에 정의 된 함수에서 Pool.map을 사용하는 방법?

나는 또한 이것으로 고투했다. 간단한 예제로 클래스의 데이터 멤버로서의 기능을 가졌습니다.

from multiprocessing import Pool
import itertools
pool = Pool()
class Example(object):
    def __init__(self, my_add):
        self.f = my_add
    def add_lists(self, list1, list2):
        # Needed to do something like this (the following line won't work)
        return pool.map(self.f,list1,list2)  

동일한 클래스 내에서 Pool.map () 호출에서 self.f 함수를 사용해야했고 self.f는 튜플을 인수로 사용하지 않았습니다. 이 함수는 클래스에 포함되었으므로 다른 답변이 제안한 래퍼 유형을 작성하는 방법이 명확하지 않았습니다.

첫 번째 요소가 함수이고 나머지 요소가 eval_func_tuple (f_args)라는 해당 함수의 인수 인 튜플 /리스트를 사용하는 다른 래퍼를 사용하여이 문제를 해결했습니다. 이를 사용하여 문제가있는 줄을 return pool.map (eval_func_tuple, itertools.izip (itertools.repeat (self.f), list1, list2))로 바꿀 수 있습니다. 전체 코드는 다음과 같습니다.

파일 : util.py

def add(a, b): return a+b

def eval_func_tuple(f_args):
    """Takes a tuple of a function and args, evaluates and returns result"""
    return f_args[0](*f_args[1:])  

파일 : main.py

from multiprocessing import Pool
import itertools
import util

pool = Pool()
class Example(object):
    def __init__(self, my_add):
        self.f = my_add
    def add_lists(self, list1, list2):
        # The following line will now work
        return pool.map(util.eval_func_tuple,
            itertools.izip(itertools.repeat(self.f), list1, list2))

if __name__ == '__main__':
    myExample = Example(util.add)
    list1 = [1, 2, 3]
    list2 = [10, 20, 30]
    print myExample.add_lists(list1, list2)  

main.py를 실행하면 [11, 22, 33]이 나타납니다. 예를 들어 eval_func_tuple을 수정하여 키워드 인수를 사용할 수도 있습니다.

또 다른 참고로, 다른 답변에서, “parmap”기능은 사용 가능한 CPU 수보다 많은 프로세스의 경우 더 효율적으로 만들 수 있습니다. 아래에서 수정 된 버전을 복사하고 있습니다. 이것은 첫 번째 게시물이며 원래 답변을 직접 수정해야하는지 잘 모르겠습니다. 또한 일부 변수의 이름을 바꿨습니다.

from multiprocessing import Process, Pipe
from itertools import izip

def spawn(f):
    def fun(pipe,x):
        pipe.send(f(x))
        pipe.close()
    return fun

def parmap(f,X):
    pipe=[Pipe() for x in X]
    processes=[Process(target=spawn(f),args=(c,x)) for x,(p,c) in izip(X,pipe)]
    numProcesses = len(processes)
    processNum = 0
    outputList = []
    while processNum < numProcesses:
        endProcessNum = min(processNum+multiprocessing.cpu_count(), numProcesses)
        for proc in processes[processNum:endProcessNum]:
            proc.start()
        for proc in processes[processNum:endProcessNum]:
            proc.join()
        for proc,c in pipe[processNum:endProcessNum]:
            outputList.append(proc.recv())
        processNum = endProcessNum
    return outputList

if __name__ == '__main__':
    print parmap(lambda x:x**x,range(1,5))         

답변

"""
The ``processes`` module provides some convenience functions
for using parallel processes in python.

Adapted from http://stackoverflow.com/a/16071616/287297

Example usage:

    print prll_map(lambda i: i * 2, [1, 2, 3, 4, 6, 7, 8], 32, verbose=True)

Comments:

"It spawns a predefined amount of workers and only iterates through the input list
 if there exists an idle worker. I also enabled the "daemon" mode for the workers so
 that KeyboardInterupt works as expected."

Pitfalls: all the stdouts are sent back to the parent stdout, intertwined.

Alternatively, use this fork of multiprocessing:
https://github.com/uqfoundation/multiprocess
"""

# Modules #
import multiprocessing
from tqdm import tqdm

################################################################################
def apply_function(func_to_apply, queue_in, queue_out):
    while not queue_in.empty():
        num, obj = queue_in.get()
        queue_out.put((num, func_to_apply(obj)))

################################################################################
def prll_map(func_to_apply, items, cpus=None, verbose=False):
    # Number of processes to use #
    if cpus is None: cpus = min(multiprocessing.cpu_count(), 32)
    # Create queues #
    q_in  = multiprocessing.Queue()
    q_out = multiprocessing.Queue()
    # Process list #
    new_proc  = lambda t,a: multiprocessing.Process(target=t, args=a)
    processes = [new_proc(apply_function, (func_to_apply, q_in, q_out)) for x in range(cpus)]
    # Put all the items (objects) in the queue #
    sent = [q_in.put((i, x)) for i, x in enumerate(items)]
    # Start them all #
    for proc in processes:
        proc.daemon = True
        proc.start()
    # Display progress bar or not #
    if verbose:
        results = [q_out.get() for x in tqdm(range(len(sent)))]
    else:
        results = [q_out.get() for x in range(len(sent))]
    # Wait for them to finish #
    for proc in processes: proc.join()
    # Return results #
    return [x for i, x in sorted(results)]

################################################################################
def test():
    def slow_square(x):
        import time
        time.sleep(2)
        return x**2
    objs    = range(20)
    squares = prll_map(slow_square, objs, 4, verbose=True)
    print "Result: %s" % squares