Многопроцессорность: как использовать Pool.map для функции, определенной в классе?

Когда я запускаю что-то вроде:

from multiprocessing import Pool

p = Pool(5)
def f(x):
     return x*x

p.map(f, [1,2,3])

это работает отлично. Однако, помещая это как функцию класса:

class calculate(object):
    def run(self):
        def f(x):
            return x*x

        p = Pool()
        return p.map(f, [1,2,3])

cl = calculate()
print cl.run()

Дает мне следующую ошибку:

Exception in thread Thread-1:
Traceback (most recent call last):
  File "/sw/lib/python2.6/threading.py", line 532, in __bootstrap_inner
    self.run()
  File "/sw/lib/python2.6/threading.py", line 484, in run
    self.__target(*self.__args, **self.__kwargs)
  File "/sw/lib/python2.6/multiprocessing/pool.py", line 225, in _handle_tasks
    put(task)
PicklingError: Can't pickle <type 'function'>: attribute lookup __builtin__.function failed

Я видел сообщение от Алекса Мартелли, посвященное той же проблеме, но оно не было достаточно явным.

Меня также раздражали ограничения на то, какие функции можно использовать pool.map. Я написал следующее, чтобы обойти это. Похоже, работает, даже для рекурсивного использования parmap.

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]
    proc = [Process(target=spawn(f), args=(c, x)) for x, (p, c) in izip(X, pipe)]
    [p.start() for p in proc]
    [p.join() for p in proc]
    return [p.recv() for (p, c) in pipe]

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

Я не мог использовать коды, опубликованные до сих пор, потому что коды, использующие «multiprocessing.Pool», не работают с лямбда-выражениями, а коды, не использующие «multiprocessing.Pool», порождают столько процессов, сколько есть рабочих элементов.

Я адаптировал код st, он порождает предопределенное количество рабочих и выполняет итерацию по списку ввода только при наличии незанятого рабочего. Я также включил режим "daemon" для рабочих sttrl-c работает, как ожидалось.

import multiprocessing


def fun(f, q_in, q_out):
    while True:
        i, x = q_in.get()
        if i is None:
            break
        q_out.put((i, f(x)))


def parmap(f, X, nprocs=multiprocessing.cpu_count()):
    q_in = multiprocessing.Queue(1)
    q_out = multiprocessing.Queue()

    proc = [multiprocessing.Process(target=fun, args=(f, q_in, q_out))
            for _ in range(nprocs)]
    for p in proc:
        p.daemon = True
        p.start()

    sent = [q_in.put((i, x)) for i, x in enumerate(X)]
    [q_in.put((None, None)) for _ in range(nprocs)]
    res = [q_out.get() for _ in range(len(sent))]

    [p.join() for p in proc]

    return [x for i, x in sorted(res)]


if __name__ == '__main__':
    print(parmap(lambda i: i * 2, [1, 2, 3, 4, 6, 7, 8]))

Многопроцессорная обработка и травление нарушены и ограничены, если вы не выйдете за пределы стандартной библиотеки.

Если вы используете вилку multiprocessingназывается pathos.multiprocesssing, вы можете напрямую использовать классы и методы класса в многопроцессорных - х mapфункциях. Это потому, что dillиспользуется вместо pickleили cPickle, и dillможет сериализовать почти все в Python.

pathos.multiprocessingтакже предоставляет асинхронную функцию отображения ... и может mapфункционировать с несколькими аргументами (например map(math.pow, [1,2,3], [4,5,6]))

Смотрите обсуждения: что могут делать мультипроцессор и укроп вместе?

и: http://matthewrocklin.com/blog/work/2013/12/05/Parallelism-and-Serialization

Он даже обрабатывает код, который вы написали изначально, без изменений и от интерпретатора. Зачем делать что-то еще более хрупкое и специфичное для одного случая?

>>> from pathos.multiprocessing import ProcessingPool as Pool
>>> class calculate(object):
...  def run(self):
...   def f(x):
...    return x*x
...   p = Pool()
...   return p.map(f, [1,2,3])
... 
>>> cl = calculate()
>>> print cl.run()
[1, 4, 9]

Получить код здесь: https://github.com/uqfoundation/pathos

И, просто чтобы показать немного больше, что он может сделать:

>>> from pathos.multiprocessing import ProcessingPool as Pool
>>> 
>>> p = Pool(4)
>>> 
>>> def add(x,y):
...   return x+y
... 
>>> x = [0,1,2,3]
>>> y = [4,5,6,7]
>>> 
>>> p.map(add, x, y)
[4, 6, 8, 10]
>>> 
>>> class Test(object):
...   def plus(self, x, y): 
...     return x+y
... 
>>> t = Test()
>>> 
>>> p.map(Test.plus, [t]*4, x, y)
[4, 6, 8, 10]
>>> 
>>> res = p.amap(t.plus, x, y)
>>> res.get()
[4, 6, 8, 10]

Насколько я знаю, в настоящее время нет решения вашей проблемы: функция, которую вы предоставляете, map()должна быть доступна через импорт вашего модуля. Вот почему работает код Роберта: функцию f()можно получить, импортировав следующий код:

def f(x):
    return x*x

class Calculate(object):
    def run(self):
        p = Pool()
        return p.map(f, [1,2,3])

if __name__ == '__main__':
    cl = Calculate()
    print cl.run()

Я фактически добавил «основной» раздел, потому что он следует рекомендациям для платформы Windows («Убедитесь, что основной модуль может быть безопасно импортирован новым интерпретатором Python, не вызывая непреднамеренных побочных эффектов»).

Я также добавил заглавную букву перед Calculateтем, чтобы следовать PEP 8 . :)

Решение от mrule правильное, но имеет ошибку: если дочерний элемент отправляет обратно большой объем данных, он может заполнить буфер канала, блокируя дочерний pipe.send(), в то время как родительский объект ожидает, пока дочерний объект завершит работу pipe.join(). Решение состоит в том, чтобы прочитать данные о ребенке перед его join()использованием. Кроме того, ребенок должен закрыть родительский конец трубы, чтобы предотвратить тупик. Код ниже исправляет это. Также имейте в виду, что это parmapсоздает один процесс на элемент в X. Более продвинутым решением является multiprocessing.cpu_count()разделение Xна несколько частей, а затем объединение результатов перед возвратом. Я оставляю это как упражнение для читателя, чтобы не испортить краткость милого ответа от mrule. ;)

from multiprocessing import Process, Pipe
from itertools import izip

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

def parmap(f,X):
    pipe=[Pipe() for x in X]
    proc=[Process(target=spawn(f),args=(p,c,x)) for x,(p,c) in izip(X,pipe)]
    [p.start() for p in proc]
    ret = [p.recv() for (p,c) in pipe]
    [p.join() for p in proc]
    return ret

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

Я тоже боролся с этим. У меня были функции в качестве членов данных класса, в качестве упрощенного примера:

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)  

Мне нужно было использовать функцию self.f в вызове Pool.map () из того же класса, и self.f не принимает кортеж в качестве аргумента. Поскольку эта функция была встроена в класс, мне не было понятно, как написать тип оболочки, предложенный другими ответами.

Я решил эту проблему с помощью другой оболочки, которая использует кортеж / список, где первый элемент - это функция, а остальные элементы - аргументы этой функции, называемые eval_func_tuple (f_args). Используя это, проблемная строка может быть заменена возвращением 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 может быть сделана более эффективной для случая с большим количеством процессов, чем с числом доступных процессоров. Я копирую отредактированную версию ниже. Это мой первый пост, и я не был уверен, стоит ли мне напрямую редактировать исходный ответ. Я также переименовал некоторые переменные.

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))         

Я взял ответ Клауса Се и Агандерса3 и сделал документированный модуль, который будет более читабельным и хранится в одном файле. Вы можете просто добавить его в свой проект. У этого даже есть дополнительный индикатор выполнения!

"""
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

РЕДАКТИРОВАТЬ : Добавлено предложение @ alexander-mcfarlane и функция тестирования

Я знаю, что об этом спрашивали более 6 лет назад, но я просто хотел добавить свое решение, так как некоторые из приведенных выше предложений кажутся ужасно сложными, но мое решение на самом деле было очень простым.

Все, что мне нужно было сделать, это обернуть вызов pool.map () во вспомогательную функцию. Передача объекта класса вместе с аргументами для метода в виде кортежа, который выглядел примерно так.

def run_in_parallel(args):
    return args[0].method(args[1])

myclass = MyClass()
method_args = [1,2,3,4,5,6]
args_map = [ (myclass, arg) for arg in method_args ]
pool = Pool()
pool.map(run_in_parallel, args_map)

Функции, определенные в классах (даже внутри функций в классах), на самом деле не работают. Тем не менее, это работает:

def f(x):
    return x*x

class calculate(object):
    def run(self):
        p = Pool()
    return p.map(f, [1,2,3])

cl = calculate()
print cl.run()

Я знаю, что этот вопрос задавался 8 лет и 10 месяцев назад, но я хочу представить вам свое решение:

from multiprocessing import Pool

class Test:

    def __init__(self):
        self.main()

    @staticmethod
    def methodForMultiprocessing(x):
        print(x*x)

    def main(self):
        if __name__ == "__main__":
            p = Pool()
            p.map(Test.methodForMultiprocessing, list(range(1, 11)))
            p.close()

TestObject = Test()

Вам просто нужно превратить функцию класса в статический метод. Но это также возможно с помощью метода класса:

from multiprocessing import Pool

class Test:

    def __init__(self):
        self.main()

    @classmethod
    def methodForMultiprocessing(cls, x):
        print(x*x)

    def main(self):
        if __name__ == "__main__":
            p = Pool()
            p.map(Test.methodForMultiprocessing, list(range(1, 11)))
            p.close()

TestObject = Test()

Протестировано в Python 3.7.3

Я изменил метод Клауса Се, потому что, пока он работал для меня с небольшими списками, он зависал, когда число элементов было ~ 1000 или больше. Вместо того, чтобы выдвигать задания по одному с Noneусловием остановки, я загружаю очередь ввода сразу и просто позволяю процессам жевать ее, пока она не опустеет.

from multiprocessing import cpu_count, Queue, Process

def apply_func(f, q_in, q_out):
    while not q_in.empty():
        i, x = q_in.get()
        q_out.put((i, f(x)))

# map a function using a pool of processes
def parmap(f, X, nprocs = cpu_count()):
    q_in, q_out   = Queue(), Queue()
    proc = [Process(target=apply_func, args=(f, q_in, q_out)) for _ in range(nprocs)]
    sent = [q_in.put((i, x)) for i, x in enumerate(X)]
    [p.start() for p in proc]
    res = [q_out.get() for _ in sent]
    [p.join() for p in proc]

    return [x for i,x in sorted(res)]

Изменить: к сожалению, сейчас я сталкиваюсь с этой ошибкой в ​​моей системе: максимальный размер очереди многопроцессорной обработки составляет 32767 , надеюсь, что обходные пути там помогут.

Вы можете запустить свой код без каких-либо проблем, если вы каким-то образом вручную проигнорируете Poolобъект из списка объектов в классе, потому что это невозможно, pickleкак говорит ошибка. Вы можете сделать это с помощью __getstate__функции (смотрите здесь ) следующим образом. PoolОбъект будет пытаться найти __getstate__и __setstate__функцию и выполнять их , если он находит его при запуске map, и map_asyncт.д.:

class calculate(object):
    def __init__(self):
        self.p = Pool()
    def __getstate__(self):
        self_dict = self.__dict__.copy()
        del self_dict['p']
        return self_dict
    def __setstate__(self, state):
        self.__dict__.update(state)

    def f(self, x):
        return x*x
    def run(self):
        return self.p.map(self.f, [1,2,3])

Затем сделайте:

cl = calculate()
cl.run()

даст вам вывод:

[1, 4, 9]

Я протестировал приведенный выше код в Python 3.x, и он работает.

Я не уверен, что этот подход был принят, но работа, которую я использую:

from multiprocessing import Pool

t = None

def run(n):
    return t.f(n)

class Test(object):
    def __init__(self, number):
        self.number = number

    def f(self, x):
        print x * self.number

    def pool(self):
        pool = Pool(2)
        pool.map(run, range(10))

if __name__ == '__main__':
    t = Test(9)
    t.pool()
    pool = Pool(2)
    pool.map(run, range(10))

Вывод должен быть:

0
9
18
27
36
45
54
63
72
81
0
9
18
27
36
45
54
63
72
81

class Calculate(object):
  # Your instance method to be executed
  def f(self, x, y):
    return x*y

if __name__ == '__main__':
  inp_list = [1,2,3]
  y = 2
  cal_obj = Calculate()
  pool = Pool(2)
  results = pool.map(lambda x: cal_obj.f(x, y), inp_list)

Есть вероятность, что вы захотите применить эту функцию для каждого отдельного экземпляра класса. Тогда вот решение для этого также

class Calculate(object):
  # Your instance method to be executed
  def __init__(self, x):
    self.x = x

  def f(self, y):
    return self.x*y

if __name__ == '__main__':
  inp_list = [Calculate(i) for i in range(3)]
  y = 2
  pool = Pool(2)
  results = pool.map(lambda x: x.f(y), inp_list)

Вот мое решение, которое я считаю немного менее хакерским, чем большинство других здесь. Это похоже на ответ nightowl.

someclasses = [MyClass(), MyClass(), MyClass()]

def method_caller(some_object, some_method='the method'):
    return getattr(some_object, some_method)()

othermethod = partial(method_caller, some_method='othermethod')

with Pool(6) as pool:
    result = pool.map(othermethod, someclasses)

С http://www.rueckstiess.net/research/snippets/show/ca1d7d90 и http://qingkaikong.blogspot.com/2016/12/python-parallel-method-in-class.html

Мы можем создать внешнюю функцию и заполнить ее объектом класса self:

from joblib import Parallel, delayed
def unwrap_self(arg, **kwarg):
    return square_class.square_int(*arg, **kwarg)

class square_class:
    def square_int(self, i):
        return i * i

    def run(self, num):
        results = []
        results = Parallel(n_jobs= -1, backend="threading")\
            (delayed(unwrap_self)(i) for i in zip([self]*len(num), num))
        print(results)

ИЛИ без JobLib:

from multiprocessing import Pool
import time

def unwrap_self_f(arg, **kwarg):
    return C.f(*arg, **kwarg)

class C:
    def f(self, name):
        print 'hello %s,'%name
        time.sleep(5)
        print 'nice to meet you.'

    def run(self):
        pool = Pool(processes=2)
        names = ('frank', 'justin', 'osi', 'thomas')
        pool.map(unwrap_self_f, zip([self]*len(names), names))

if __name__ == '__main__':
    c = C()
    c.run()

Это может быть не очень хорошим решением, но в моем случае я решаю это так.

from multiprocessing import Pool

def foo1(data):
    self = data.get('slf')
    lst = data.get('lst')
    return sum(lst) + self.foo2()

class Foo(object):
    def __init__(self, a, b):
        self.a = a
        self.b = b

    def foo2(self):
        return self.a**self.b   

    def foo(self):
        p = Pool(5)
        lst = [1, 2, 3]
        result = p.map(foo1, (dict(slf=self, lst=lst),))
        return result

if __name__ == '__main__':
    print(Foo(2, 4).foo())

Я должен был перейти selfк своей функции, поскольку я должен получить доступ к атрибутам и функциям моего класса через эту функцию. Это работает для меня. Исправления и предложения всегда приветствуются.

Вот пример, который я написал для использования многопроцессорного пула в python3, в частности, для запуска тестов использовался python3.7.7. Я получил свои самые быстрые пробеги, используя imap_unordered. Просто включите ваш сценарий и попробуйте. Вы можете использовать timeitили просто time.time()выяснить, что работает лучше для вас.

import multiprocessing
import time

NUMBER_OF_PROCESSES = multiprocessing.cpu_count()
MP_FUNCTION = 'starmap'  # 'imap_unordered' or 'starmap' or 'apply_async'

def process_chunk(a_chunk):
    print(f"processig mp chunk {a_chunk}")
    return a_chunk


map_jobs = [1, 2, 3, 4]

result_sum = 0

s = time.time()
if MP_FUNCTION == 'imap_unordered':
    pool = multiprocessing.Pool(processes=NUMBER_OF_PROCESSES)
    for i in pool.imap_unordered(process_chunk, map_jobs):
        result_sum += i
elif MP_FUNCTION == 'starmap':
    pool = multiprocessing.Pool(processes=NUMBER_OF_PROCESSES)
    try:
        map_jobs = [(i, ) for i in map_jobs]
        result_sum = pool.starmap(process_chunk, map_jobs)
        result_sum = sum(result_sum)
    finally:
        pool.close()
        pool.join()
elif MP_FUNCTION == 'apply_async':
    with multiprocessing.Pool(processes=NUMBER_OF_PROCESSES) as pool:
        result_sum = [pool.apply_async(process_chunk, [i, ]).get() for i in map_jobs]
    result_sum = sum(result_sum)
print(f"result_sum is {result_sum}, took {time.time() - s}s")

В приведенном выше сценарии на imap_unorderedсамом деле кажется худшим для меня. Опробуйте ваш кейс и сравните его с машиной, на которой вы планируете его запускать. Также читайте о пулах процессов . Ура!