ThreadPoolExecutor внутри ProcessPoolExecutor

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

class Particle:
    def __init__(self, i):
        self.i = i
        self.fitness = None
    def getfitness(self):
        self.fitness = 2 * self.i

Теперь нам есть во что поиграть. Далее некоторые константы:

MAX_PROCESSES = 3
MAX_THREADS = 2 # per process
CHUNKSIZE = 100

Поиграйте с ними по вкусу. CHUNKSIZE будет объяснено позже.

Первый сюрприз для вас — это то, что делает моя рабочая функция самого низкого уровня. Это потому, что вы слишком оптимистичны:

Поскольку побочные эффекты вызова p.getFitness сохраняются в каждой частице, мне не нужно беспокоиться о возврате от futures.ProcessPoolExecutor().

Увы, ничего, сделанное в рабочем процессе, не может повлиять на Particle экземпляров в вашей основной программе. Рабочий процесс работает с копиями Particle экземпляров, либо с помощью реализации копирования при записи fork(), либо потому, что он работает с копией, созданной в результате распаковки Particle pickle, переданного между процессами.

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

def thread_worker(p):
    p.getfitness()
    return (p.i, p.fitness)

Учитывая это, легко распределить список Particle по потокам и вернуть список (particle_id, fitness) результатов:

def proc_worker(ps):
    import concurrent.futures as cf
    with cf.ThreadPoolExecutor(max_workers=MAX_THREADS) as e:
        result = list(e.map(thread_worker, ps))
    return result

Заметки:

1. Это функция, которую будет выполнять каждый рабочий процесс.
2. Я использую Python 3, поэтому используйте list(), чтобы заставить e.map() материализовать все результаты в списке.
3. Как упоминалось в комментарии, в CPython распределение задач, связанных с ЦП, по потокам медленнее, чем выполнение их всех в одном потоке.

Остается только написать код, чтобы распределить список Particle по процессам и получить результаты. Это чертовски легко сделать с multiprocessing, поэтому я собираюсь использовать его. Я понятия не имею, может ли это сделать concurrent.futures (учитывая, что мы также смешиваем потоки), но мне все равно. Но поскольку я даю вам рабочий код, вы можете поиграть с ним и сообщить ;-)

if __name__ == "__main__":
    import multiprocessing

    particles = [Particle(i) for i in range(100000)]
    # Note the code below relies on that particles[i].i == i
    assert all(particles[i].i == i for i in range(len(particles)))

    pool = multiprocessing.Pool(MAX_PROCESSES)
    for result_list in pool.imap_unordered(proc_worker,
                      (particles[i: i+CHUNKSIZE]
                       for i in range(0, len(particles), CHUNKSIZE))):
        for i, fitness in result_list:
            particles[i].fitness = fitness

    pool.close()
    pool.join()

    assert all(p.fitness == 2*p.i for p in particles)

Заметки:

1. Я разбиваю список Particles на куски "вручную". Вот для чего CHUNKSIZE. Это потому, что рабочему процессу нужен список из Particle для работы, и, в свою очередь, это то, что нужно функции futures map(). Хорошей идеей является разделение работы независимо от того, что вы получаете, чтобы получить реальную отдачу в обмен на межпроцессные накладные расходы для каждого вызова.
2. imap_unordered() не дает никаких гарантий относительно порядка, в котором возвращаются результаты. Это дает реализации больше свободы для организации работы максимально эффективно. И нас не волнует порядок здесь, так что все в порядке.
3. Обратите внимание, что цикл получает результаты (particle_id, fitness) и соответствующим образом изменяет экземпляры Particle. Возможно, ваш настоящий .getfitness вносит другие мутации в экземпляры Particle - не могу предположить. Несмотря на это, основная программа никогда не увидит никаких мутаций, сделанных в воркерах "по волшебству" - вы должны явно организовать это. Вместо этого вы можете вернуть (particle_id, particle_instance) пар и заменить Particle экземпляров в основной программе. Тогда они будут отражать все мутации, сделанные в рабочих процессах.

Веселиться :-)

Фьючерсы полностью вниз

Оказывается, заменить multiprocessing было очень просто. Вот изменения. Это также (как упоминалось ранее) заменяет исходные экземпляры Particle, чтобы зафиксировать все мутации. Однако здесь есть компромисс: для травления экземпляра требуется «намного больше» байтов, чем для травления одного результата «пригодности». Больше сетевого трафика. Выбрать свой яд ;-)

Для возврата измененного экземпляра требуется просто заменить последнюю строку thread_worker(), например:

return (p.i, p)

Затем замените весь блок «main» следующим:

def update_fitness():
    import concurrent.futures as cf
    with cf.ProcessPoolExecutor(max_workers=MAX_PROCESSES) as e:
        for result_list in e.map(proc_worker,
                      (particles[i: i+CHUNKSIZE]
                       for i in range(0, len(particles), CHUNKSIZE))):
            for i, p in result_list:
                particles[i] = p

if __name__ == "__main__":
    particles = [Particle(i) for i in range(500000)]
    assert all(particles[i].i == i for i in range(len(particles)))

    update_fitness()

    assert all(particles[i].i == i for i in range(len(particles)))
    assert all(p.fitness == 2*p.i for p in particles)

Код очень похож на танец multiprocessor. Лично я бы использовал версию multiprocessing, потому что imap_unordered ценна. Это проблема упрощенных интерфейсов: они часто покупают простоту за счет сокрытия полезных возможностей.

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

Если добавление потоков увеличивает вашу производительность, следующий вопрос заключается в том, можно ли добиться большей производительности с помощью ручной балансировки нагрузки или автоматической. Под «сделанным вручную» я подразумеваю тщательное разделение рабочей нагрузки на части одинаковой вычислительной сложности и создание нового процессора задач для каждой части — ваше оригинальное, но сомнительное решение. По автоматическому созданию пула процессов/потоков и связи в рабочей очереди для новых задач, к которым вы стремитесь. На мой взгляд, первый подход является одним из парадигм Apache Hadoop, второй реализуется обработчиками очереди работ, такими как Celery. Первый подход может страдать от того, что некоторые фрагменты задач работают медленнее и выполняются, в то время как другие завершаются, второй добавляет накладные расходы на коммутацию и ожидание задачи, и это вторая точка тестов производительности, которые необходимо выполнить.

Наконец, если вы хотите иметь статическую коллекцию процессов с многопоточностью внутри, AFAIK, вы не можете добиться этого с помощью concurrent.futures как есть, и вам придется немного изменить его. Я не знаю, существуют ли решения для этой задачи, но поскольку concurrent — это решение на чистом питоне (без кода на языке C), это можно легко сделать. Рабочий процессор определяется в _adjust_process_count рутине класса ProcessPoolExecutor, а его подклассы и переопределение с помощью многопоточного подхода довольно просты, вам просто нужно указать свой собственный _process_worker на основе concurrent.features.thread

Оригинал ProcessPoolExecutor._adjust_process_count для справки:

def _adjust_process_count(self):
    for _ in range(len(self._processes), self._max_workers):
        p = multiprocessing.Process(
                target=_process_worker,
                args=(self._call_queue,
                      self._result_queue))
        p.start()
        self._processes[p.pid] = p

Это обобщенный ответ, в котором используется пакет threadedprocess, который реализует ThreadedProcesPoolExecutor, позволяя комбинированное использование потока пул внутри пула процессов. Ниже приведена служебная функция общего назначения, которая ее использует:

import concurrent.futures
import logging
from typing import Callable, Iterable, Optional

import threadedprocess

log = logging.getLogger(__name__)


def concurrently_execute(fn: Callable, fn_args: Iterable, max_processes: Optional[int] = None, max_threads_per_process: Optional[int] = None) -> None:
    """Execute the given callable concurrently using multiple threads and/or processes."""
    # Ref: https://stackoverflow.com/a/57999709/
    if max_processes == 1:
        executor = concurrent.futures.ThreadPoolExecutor(max_workers=max_threads_per_process)
    elif max_threads_per_process == 1:
        executor = concurrent.futures.ProcessPoolExecutor(max_workers=max_processes)  # type: ignore
    else:
        executor = threadedprocess.ThreadedProcessPoolExecutor(max_processes=max_processes, max_threads=max_threads_per_process)

    if max_processes and max_threads_per_process:
        max_workers = max_processes * max_threads_per_process
        log.info("Using %s with %s processes and %s threads per process, i.e. with %s workers.", executor.__class__.__name__, max_processes, max_threads_per_process, max_workers)

    with executor:
        futures = [executor.submit(fn, *fn_args_cur) for fn_args_cur in fn_args]

    for future in concurrent.futures.as_completed(futures):
        future.result()  # Raises exception if it occurred in process worker.