Применение метода к списку объектов параллельно с использованием многопроцессорной обработки

Ответ 1

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

Здесь полная программа, использующая этот подход, который печатает то, что вы хотите в Python 2, и независимо от ОС:

class MyClass():
    def __init__(self, input):
        self.input = input
        self.result = int

    def my_process(self, multiply_by, add_to):
        self.result = self.input * multiply_by
        self._my_sub_process(add_to)
        return self.result

    def _my_sub_process(self, add_to):
        self.result += add_to

import multiprocessing as mp
NUM_CORE = 4  # set to the number of cores you want to use

def worker(arg):
    obj, m, a = arg
    return obj.my_process(m, a)

if __name__ == "__main__":
    list_of_numbers = range(0, 5)
    list_of_objects = [MyClass(i) for i in list_of_numbers]

    pool = mp.Pool(NUM_CORE)
    list_of_results = pool.map(worker, ((obj, 100, 1) for obj in list_of_objects))
    pool.close()
    pool.join()

    print list_of_numbers
    print list_of_results

Магия магии

Я должен отметить, что есть много преимуществ для использования очень простого подхода, который я предлагаю. Помимо того, что он "просто работает" на Pythons 2 и 3, не требует изменений в ваших классах, и его легко понять, он также хорошо сочетается со всеми методами Pool.

Однако, если у вас есть несколько методов, которые вы хотите запускать параллельно, это может немного раздражать, чтобы написать крошечную рабочую функцию для каждого. Итак, вот крошечный бит "магии", чтобы черви вокруг этого. Измените worker() следующим образом:

def worker(arg):
    obj, methname = arg[:2]
    return getattr(obj, methname)(*arg[2:])

Теперь для любого количества методов существует достаточная одна рабочая функция с любым количеством аргументов. В вашем конкретном случае просто измените одну строку, чтобы она соответствовала:

list_of_results = pool.map(worker, ((obj, "my_process", 100, 1) for obj in list_of_objects))

Более или менее очевидные обобщения могут также использоваться для методов с аргументами ключевого слова. Но в реальной жизни я обычно придерживаюсь первоначального предложения. В какой-то момент питание к обобщениям приносит больше вреда, чем пользы. Опять же, мне нравятся очевидные вещи: -)

Ответ 2

Как правило, самым простым способом параллельного параллельного вычисления является метод map функции multiprocessing.Pool (или as_completed из concurrent.futures в Python 3).

Однако метод map применяет функцию, которая принимает только один аргумент для итерации данных с использованием нескольких процессов.

Таким образом, эта функция не может быть обычным методом, поскольку для этого требуется как минимум два аргумента; он должен также включать self! Однако это может быть статический метод. См. Также этот ответ для более подробного объяснения.

Ответ 3

Если ваш класс не "огромный", я думаю, что процесс ориентирован лучше. Предлагается пул в многопроцессорной обработке. Это учебник → https://docs.python.org/2/library/multiprocessing.html#using-a-pool-of-workers

Затем отделите add_to от my_process, так как они быстры, и вы можете подождать, используя конец последнего процесса.

def my_process(input, multiby):
    return xxxx
def add_to(result,a_list):
    xxx
p = Pool(5)
res = []
for i in range(10):
    res.append(p.apply_async(my_process, (i,5)))
p.join()  # wait for the end of the last process
for i in range(10):
    print res[i].get()

Ответ 4

Если вам не нужно полностью придерживаться модуля Multiprocessing, его можно легко достичь, используя concurrents.futures library

здесь пример кода:

from concurrent.futures.thread import ThreadPoolExecutor, wait

MAX_WORKERS = 20

class MyClass():
    def __init__(self, input):
        self.input = input
        self.result = int

    def my_process(self, multiply_by, add_to):
        self.result = self.input * multiply_by
        self._my_sub_process(add_to)
        return self.result

    def _my_sub_process(self, add_to):
        self.result += add_to

list_of_numbers = range(0, 5)
list_of_objects = [MyClass(i) for i in list_of_numbers]

With ThreadPoolExecutor(MAX_WORKERS) as executor:
    for obj in list_of_objects:
        executor.submit(obj.my_process, 100, 1).add_done_callback(on_finish)

def on_finish(future):
    result = future.result() # do stuff with your result

здесь исполнитель возвращает будущее для каждой заданной им задачи. имейте в виду, что если вы используете add_done_callback() завершенная задача из потока возвращается к основному потоку (который блокирует ваш основной поток), если вы действительно хотите true parallelism, тогда вы должны ждать будущих объектов отдельно, здесь фрагмент кода для этого.

futures = []
with ThreadPoolExecutor(MAX_WORKERS) as executor:
    for objin list_of_objects:
        futures.append(executor.submit(obj.my_process, 100, 1))
wait(futures)

for succeded, failed in futures:
    # work with your result here
    if succeded:
       print (succeeeded.result())
    if failed:
        print (failed.result())

надеюсь, что это поможет.

Ответ 5

Основываясь на ответе Python Multiprocessing - примените метод класса к списку объектов и вашему коду:

• добавить MyClass object в simulation object

  class simulation(multiprocessing.Process):
      def __init__(self, id, worker, *args, **kwargs):
          # must call this before anything else
          multiprocessing.Process.__init__(self)
          self.id = id
          self.worker = worker
          self.args = args
          self.kwargs = kwargs
          sys.stdout.write('[%d] created\n' % (self.id))
  

• запустите то, что вы хотите, в функции run

      def run(self):
          sys.stdout.write('[%d] running ...  process id: %s\n' % (self.id, os.getpid()))
          self.worker.my_process(*self.args, **self.kwargs)
          sys.stdout.write('[%d] completed\n' % (self.id))
  

Попробуйте следующее:

list_of_numbers = range(0, 5)
list_of_objects = [MyClass(i) for i in list_of_numbers]
list_of_sim = [simulation(id=k, worker=obj, multiply_by=100*k, add_to=10*k) \
    for k, obj in enumerate(list_of_objects)]  

for sim in list_of_sim:
    sim.start()