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

180

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

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

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

Мермос
источник
1
"это как функция класса"? Можете ли вы опубликовать код, который на самом деле получает фактическую ошибку. Без реального кода мы можем только догадываться, что вы делаете неправильно.
S.Lott
Как общее замечание, существуют модули маринования, более мощные, чем стандартный модуль маринования Python (например, модуль picloud, упомянутый в этом ответе ).
Клаус Се
1
У меня была похожая проблема с замыканиями внутри IPython.Parallel, но там вы могли обойти эту проблему, помещая объекты в узлы. Это кажется довольно раздражающим, чтобы обойти эту проблему с многопроцессорностью.
Алекс С
Вот calculateэто пригодны для консервирования, так что кажется , что это может быть решена путем 1) создания функционального объекта с конструктором , который копирует в calculateэкземпляр , а затем 2) , проходящей экземпляр этой функции объекта Pool«ы mapметода. Нет?
rd11
1
@ math Я не верю, что какие-либо "недавние изменения" в Python окажут какую-либо помощь. Некоторые ограничения multiprocessingмодуля связаны с его целью кроссплатформенной реализации и отсутствием fork(2)системного вызова в Windows. Если вам не нужна поддержка Win32, возможно, существует более простой обходной путь на основе процессов. Или, если вы готовы использовать потоки вместо процессов, вы можете заменить их from multiprocessing import Poolна from multiprocessing.pool import ThreadPool as Pool.
Ая

Ответы:

69

Меня также раздражали ограничения на то, какие функции можно использовать 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))
mrule
источник
1
Это сработало очень хорошо для меня, спасибо. Я обнаружил одно слабое место: я попытался использовать parmap для некоторых функций, которые обходили defaultdict, и снова получил PicklingError. Я не нашел решения для этого, я просто переработал свой код, чтобы не использовать defaultdict.
без
2
Это не работает в Python 2.7.2 (по умолчанию, 12 июня 2011, 15:08:59) [MSC v.1500 32 бит (Intel)] на win32
ubershmekel
3
Это работает на Python 2.7.3 августа 1,2012, 05:14:39. Это не работает для гигантских итераций -> это вызывает ошибку OSErr: [Errno 24] Слишком много открытых файлов из-за количества каналов, которые он открывает.
Eiyrioü von Kauyf
Это решение порождает процесс для каждого рабочего элемента. Решение "Клаус се" ниже является более эффективным.
ypnos
У меня похожий вопрос
Асмита Поддар
85

Я не мог использовать коды, опубликованные до сих пор, потому что коды, использующие «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]))
Клаус Се
источник
2
Как бы вы получили индикатор выполнения для правильной работы с этой parmapфункцией?
Shokburner
2
Вопрос - я использовал это решение, но заметил, что процессы python, которые я породил, остались активными в памяти. Любая быстрая мысль о том, как убить их, когда ваш parmap выходит?
CompEcon
1
@ klaus-se Я знаю, что мы не можем просто сказать спасибо в комментариях, но ваш ответ слишком ценен для меня, я не удержался. Я хотел бы дать вам больше, чем одну репутацию ...
Дештоп
2
Передача @greole (None, None)в качестве последнего элемента указывает на funто, что он достиг конца последовательности элементов для каждого процесса.
aganders3
4
@deshtop: вы можете получить награду, если у вас достаточно репутации :-)
Марк
57

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

Если вы используете вилку 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]
Майк Маккернс
источник
1
У pathos.multiprocessing также есть асинхронная карта ( amap), которая позволяет использовать индикаторы выполнения и другое асинхронное программирование.
Майк Маккернс
Мне нравится pathos.multiprocessing, который может служить практически полной заменой непараллельной карты, наслаждаясь многопроцессорностью. У меня есть простая оболочка pathos.multiprocessing.map, так что она более эффективна для памяти при обработке большой структуры данных только для чтения на нескольких ядрах, см. Этот репозиторий git .
Fashandge
Кажется интересным, но это не установить. Это сообщение, которое дает пип:Could not find a version that satisfies the requirement pp==1.5.7-pathos (from pathos)
xApple
1
Да. Я не выпускал некоторое время, поскольку я разделял функциональность на отдельные пакеты, а также конвертировал в 2/3 совместимый код. Многое из вышеперечисленного было модульным, в multiprocessкотором совместимо 2/3. См stackoverflow.com/questions/27873093/... и pypi.python.org/pypi/multiprocess .
Майк Маккернс
3
@xApple: Как продолжение, pathosу него есть новый стабильный выпуск, а также совместимы с 2.x и 3.x.
Майк Маккернс
40

Насколько я знаю, в настоящее время нет решения вашей проблемы: функция, которую вы предоставляете, 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 . :)

Эрик О Лебиго
источник
18

Решение от 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))
Боб Макелрат
источник
Как вы выбираете количество процессов?
patapouf_ai
Однако он умирает довольно быстро из-за ошибки OSError: [Errno 24] Too many open files. Я думаю, что должны быть какие-то ограничения на количество процессов, чтобы он работал правильно ...
patapouf_ai
13

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

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))         
Brandt
источник
8

Я взял ответ Клауса Се и Агандерса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 и функция тестирования

xApple
источник
одна проблема с вашим индикатором выполнения ... Индикатор показывает только, насколько неэффективно рабочая нагрузка была распределена между процессорами. Если рабочая нагрузка идеально разделена, то все процессоры будут работать join()одновременно, и вы просто получите 100%завершение на tqdmдисплее. Единственный раз, когда это будет полезно, если у каждого процессора есть предвзятая рабочая нагрузка
Александр Макфарлейн
1
двигаться, tqdm()чтобы обернуть линию: result = [q_out.get() for _ in tqdm(sent)]и это работает намного лучше - большие усилия, хотя действительно ценю это так +1
Александр Макфарлейн
Спасибо за этот совет, я попробую и затем обновлю ответ!
xApple
Ответ обновляется, и индикатор выполнения работает намного лучше!
xApple
8

Я знаю, что об этом спрашивали более 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)
ночная Сова
источник
7

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

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()
Роберт
источник
15
спасибо, но я нахожу это немного грязным, чтобы определить функцию вне класса. Класс должен объединять все, что ему нужно для достижения поставленной задачи.
Mermoz
3
@Memoz: «Класс должен объединять все, что ему нужно» Правда? Я не могу найти много примеров этого. Большинство классов зависят от других классов или функций. Зачем называть зависимость класса «грязной»? Что не так с зависимостью?
S.Lott
Ну, функция не должна изменять существующие данные класса - потому что она будет изменять версию в другом процессе - так что это может быть статический метод. Вы можете сортировать статический метод: stackoverflow.com/questions/1914261/… Или, для чего-то такого тривиального, вы можете использовать лямбду.
Роберт
6

Я знаю, что этот вопрос задавался 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

TornaxO7
источник
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 , надеюсь, что обходные пути там помогут.

aganders3
источник
1

Вы можете запустить свой код без каких-либо проблем, если вы каким-то образом вручную проигнорируете 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, и он работает.

эмир
источник
0

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

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
CpILL
источник
0
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)
ShikharDua
источник
0

Вот мое решение, которое я считаю немного менее хакерским, чем большинство других здесь. Это похоже на ответ 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)
Эрленд Ауне
источник
0

С 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()
Боб Бэксли
источник
0

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

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к своей функции, поскольку я должен получить доступ к атрибутам и функциям моего класса через эту функцию. Это работает для меня. Исправления и предложения всегда приветствуются.

Мухаммед Хасан
источник
0

Вот пример, который я написал для использования многопроцессорного пула в 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самом деле кажется худшим для меня. Опробуйте ваш кейс и сравните его с машиной, на которой вы планируете его запускать. Также читайте о пулах процессов . Ура!

radtek
источник