Python: _pickle.PicklingError: невозможно рассолить <функция <лямбда>>

Python не может обрабатывать лямбда-функции. Вместо этого вы должны определить функцию и вместо этого передать имя функции. Вот как вы можете подойти к этому:

import itertools
import random
import time
from multiprocessing import Pool


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


def foo(x):
    return add_nums(x[0], x[1])


if __name__ == "__main__":
    listLen = 1000

    # Create a list of random numbers to do pairwise additions of
    myList = [random.choice(range(1000)) for i in range(listLen)]
    # Create a list of all pairwise combinations of the indices of the list
    index_combns = [
        (myList[i[0]], myList[i[1]])
        for i in itertools.combinations(range(len(myList)), 2)
    ]

    # Do the pairwise operation without multiprocessing
    start_time = time.time()
    sums_no_mp = [*map(foo, index_combns)]
    end_time = time.time() - start_time
    print(f"Process took {end_time} seconds with no MP")

    # Do the pairwise operations with multiprocessing
    start_time = time.time()
    pool = Pool(processes=64)
    sums_mp = pool.map(foo, index_combns)
    end_time = time.time() - start_time
    print(f"Process took {end_time} seconds with MP")

    pool.close()
    pool.join()

Я изменил index_combns, чтобы также извлечь значение из myList на месте, потому что myList не будет доступен из foo, а передача нескольких копий myList увеличит сложность вашего скрипта.

Запуск этой печати:

Process took 0.053926944732666016 seconds with no MP
Process took 0.4799039363861084 seconds with MP

Я не совсем уверен, что Зачем (хотя тщательное чтение многопроцессорные документы, вероятно, дало бы ответ), но в многопроцессорной обработке python задействован процесс травления, когда дочерним процессам передаются определенные вещи. Хотя я ожидал, что лямбды будут унаследованы, а не переданы через рассола, я думаю, что это не то, что происходит.

После обсуждения в комментариях рассмотрим примерно такой подход:

import time
from multiprocessing import Pool
import itertools
import numpy as np
from multiprocessing import shared_memory

def add_mats(a, b):
    #time.sleep(0.00001)
    return (a + b)

# Helper for mp version
def add_mats_shared(shm_name, array_shape, array_dtype, i1, i2):
    shm = shared_memory.SharedMemory(name=shm_name)
    stacked = np.ndarray(array_shape, dtype=array_dtype, buffer=shm.buf)
    a = stacked[i1]
    b = stacked[i2]
    result = add_mats(a, b)
    shm.close()
    return result

if __name__ == "__main__":
    class Timer:
        def __init__(self):
            self.start = None
            self.stop  = None
            self.delta = None

        def __enter__(self):
            self.start = time.time()
            return self

        def __exit__(self, *exc_args):
            self.stop = time.time()
            self.delta = self.stop - self.start

    arrays = [np.random.rand(5,5) for _ in range(50)]
    index_combns = list(itertools.combinations(range(len(arrays)),2))

    # Helper for non-mp version
    def add_mats_pair(ij_pair):
        i, j = ij_pair
        a = arrays[i]
        b = arrays[j]
        return add_mats(a, b)

    with Timer() as t:
        # Do the pairwise operation without multiprocessing
        sums_no_mp = list(map(add_mats_pair, index_combns))

    print(f"Process took {t.delta} seconds with no MP")


    with Timer() as t:
        # Stack arrays and copy result into shared memory
        stacked = np.stack(arrays)
        shm = shared_memory.SharedMemory(create=True, size=stacked.nbytes)
        shm_arr = np.ndarray(stacked.shape, dtype=stacked.dtype, buffer=shm.buf)
        shm_arr[:] = stacked[:]

        with Pool(processes=32) as pool:
            processes = [pool.apply_async(add_mats_shared, (
                shm.name,
                stacked.shape,
                stacked.dtype,
                i,
                j,
            )) for (i,j) in index_combns]
            sums_mp = [p.get() for p in processes]

        shm.close()
        shm.unlink()

    print(f"Process took {t.delta} seconds with MP")

    for i in range(len(sums_no_mp)):
        assert (sums_no_mp[i] == sums_mp[i]).all()

    print("Results match.")

Он использует multiprocessing.shared_memory для совместного использования одного массива размерностей numpy (Н+1) (вместо списка массивов размерностей Н) между хост-процессом и дочерними процессами.

Другие вещи, которые отличаются, но не имеют значения:

• Pool используется в качестве диспетчера контекста, чтобы избежать явного закрытия и присоединения к нему.
• Timer — это простой контекстный менеджер для временных блоков кода.
• Некоторые числа были скорректированы случайным образом
• pool.map заменен на звонки pool.apply_async

pool.map тоже подойдет, но вам нужно создать список аргументов перед вызовом .map и распаковать его в рабочей функции, например:

with Pool(processes=32) as pool:
    args = [(
        shm.name,
        stacked.shape,
        stacked.dtype,
        i,
        j,
    ) for (i,j) in index_combns]
    sums_mp = pool.map(add_mats_shared, args)

# and 

# Helper for mp version
def add_mats_shared(args):
    shm_name, array_shape, array_dtype, i1, i2 = args
    shm = shared_memory.SharedMemory(name=shm_name)
    ....

Q :
_{" ... trying to learn about multiprocessing in order to solve a problem at work. "}

А:
единственный самый важный опыт, который нужно изучить
, это то, насколько БОЛЬШИМ являются РАСХОДЫ-из-(процесса)-ИНСТАНЦИЯ(s),
все остальные дополнительные накладные расходы
(но ни в коем случае не незначительные, тем более в увеличение масштабов проблемы )
детали по сравнению с этой огромной и главной.

_{Before the answer is read-through and completely understood, here is a Live GUI-interactive simulator of how much we will have to pay to start using more than 1 stream of process-flow orchestration ( costs vary - lower for threads, larger for MPI-based distributed operations, highest for multiprocessing-processes, as used in Python Interpreter, where N-many copies of the main Python Interpreter process get first copied (RAM-allocated and O/S scheduler spawned - as 2022-Q2 still reports issues if less expensive backends try to avoid this cost, yet at problems with deadlocking on wrong-shared or ill-copied or forgotten to copy some already blocking MUTEX-es and similar internalities - so that even the full-copy is not always safe in 2022 - not having met them in person does not mean these do not still exist, as documented by countless professionals - a story about a pool of sharks is a good place to start from )}

Инвентаризация проблем:

а) травление лямбд ^{( and many other SER/DES blockers )}

легко — достаточно conda install dill и import dill as pickle как умеет укроп, годами травить их — кредит @MikeMcKearns и вашему коду не нужно рефакторить использование простого интерфейса pickle.dumps()-вызова. Таким образом, использование pathos.multiprocess по умолчанию используется для внутреннего использования dill, и этой уже давно известной слабости multiprocessing SER/DES удается избежать.

б) убийцы производительности

- multiprocessing.Pool.map() это скорее End-to-End анти-шаблон производительности здесь - Расходы..., если мы начнем не пренебрегать ими, покажите, сколько CPU-тактов и заблокированных физических-RAM-вводов/выводов платится за столько процессов- экземпляров ( 60+ ), которые, наконец, «занимают» почти все физические ядра ЦП, но оставляют, таким образом, почти нулевое пространство для действительно высокой производительности numpy-родных многоядерных вычислений основной проблемы (для которой ожидалось, что максимальная производительность будет форсирован, не так ли? )

- просто переместите ползунок п в симуляторе на что-то меньшее, чем 100% (без [SERIAL]-части выполнения задачи, что хорошо в теории, но никогда не выполнимо на практике, даже запуск программы является чистым [SERIAL], по замыслу )

- просто переместите ползунок О_verhead в симуляторе на что-нибудь выше простого нуля (выражая относительную дополнительную стоимость создания одного из процессов Н_CPUcores, как число процентов, по отношению к такому номеру части [PARALLEL]-раздела инструкций - математически "плотная" работа имеет много таких "полезных"-инструкций и может , предполагая, что никакие другие убийцы производительности не выпрыгивают из коробки, могут потратить некоторую разумную сумму «дополнительных» затрат, чтобы породить некоторое количество одновременных или параллельных операций (фактическое число зависит только от фактического экономия затрат, а не от того, сколько CPU-ядра присутствуют, тем более по нашим "пожеланиям" или схоластическим или еще хуже копипаст-"советам" ). Наоборот, математически «мелкая» работа почти всегда имеет «ускорение» << 1 (замедление огромный), так как почти нет шансов оправдать известные дополнительные затраты (оплата на процесс-инстанциациях, данные SER/xfer /DES перемещается внутрь (параметры) и назад (результаты)

- далее переместите ползунок О_verhead в симуляторе к самому правому краю == 1. Это показывает случай, когда фактические накладные расходы (потеря времени), порождающие процесс, по-прежнему не превышают <= 1 % всех последующих инструкций, связанных с вычислениями, которые будут выполняться для «полезной» части. работы, которая будет вычисляться внутри такого порожденного экземпляра процесса. Таким образом, даже такой коэффициент пропорции 1:100 (выполнение в 100 раз больше «полезной» работы, чем потерянное процессорное время, сожженное для организации такого количества копий и заставляющего O/S-планировщика организовывать их параллельное выполнение внутри доступной системной виртуальной памяти) имеет уже все предупреждения видны на графике прогрессии Speedup-деградации - просто поиграйте немного с О_verhead-слайдером в симуляторе, прежде чем трогать остальные...

- избежать греха «совместного использования» (если целью является производительность) - опять же, затраты на эксплуатацию такой оркестровки между несколькими процессами интерпретатора Python, теперь независимыми, влекут за собой дополнительные дополнительные затраты, которые никогда не оправдываются повышением производительности, поскольку борьба за занимая общие ресурсы (ядра ЦП, каналы физической ОЗУ-ввода-вывода) только опустошает коэффициенты повторного использования кэш-памяти процессора, переключатели контекста процесса, управляемые планировщиком O/S, и все это еще больше снижает итоговую -Конец производительности (чего мы не хотим, не так ли?)

в) повышение работоспособности

- уважайте факты о фактические затраты любой вычислительной операции
- избегайте "поверхностных" шагов вычислений,
- максимизируйте то, что так дорого обходится в набор распределенных-процессов (если такая необходимость остается),
- избегайте все операции добавления служебных данных (например, добавление локальных временных переменных, где встроенные операции позволяют вместо хранения частичных результатов)
и
- попробуйте перейти к использованию сверхпроизводительного, дружественного к кэшу и оптимизированного, родного numpy-векторизованного многоядерного и пошагового возможности трюков, не блокируемые предварительно перегруженными ядрами ЦП за счет планирования такого количества (~ 60) копий процесса Python Interpreter, каждая из которых пытается вызвать numpy-код, таким образом, не имея свободных ядер для фактического размещения такого высокопроизводительного кэша. -удобные для повторного использования векторные вычисления (здесь мы получаем или теряем большую часть производительности, а не в медленных последовательных итераторах, не в порождении 60+ основанных на процессах полных копий «__main__» интерпретатора Python, прежде чем выполнять один кусок полезной работы на наших больших данных, дорого РА M-распределено и физически скопировано более 60 раз)

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

и как только вы уже работаете на переднем крае производительности, установите gc.disable() прежде чем вы создадите интерпретатор Python в N-множестве реплицированных копий, чтобы не ждать спонтанной сборки мусора при достижении максимальной производительности