_multiprocessing.SemLock не реализован при работе на AWS Lambda

Насколько я могу судить, многопроцессорность не будет работать на AWS Lambda, поскольку отсутствует среда выполнения/контейнер /dev/shm — см. https://forums.aws.amazon.com/thread.jspa?threadID=219962 (может потребоваться авторизация).

Нет слов (что я могу найти) о том, изменит ли / когда Amazon это. Я также посмотрел на другие библиотеки, например. https://pythonhosted.org/joblib/parallel.html вернется к /tmp (которое, как мы знаем, СУЩЕСТВУЕТ), если он не может найти /dev/shm, но на самом деле это не решает проблему.

Вы можете запускать процедуры параллельно на AWS Lambda, используя модуль многопроцессорности Python, но вы не можете использовать пулы или очереди, как указано в других ответах. Рабочим решением является использование Process и Pipe, как описано в этой статье https://aws.amazon.com/blogs/compute/parallel-processing-in-python-with-aws-lambda/

Хотя статья определенно помогла мне найти решение (описано ниже), есть несколько моментов, о которых следует знать. Во-первых, решение, основанное на процессах и конвейерах, работает не так быстро, как встроенная функция карты в пуле, хотя я заметил почти линейное ускорение, когда увеличил доступную память/ресурсы ЦП в моей функции Lambda. Во-вторых, при разработке функций многопроцессорной обработки таким образом необходимо выполнить определенные действия по управлению. Я подозреваю, что это, по крайней мере частично, почему мое решение медленнее, чем встроенные методы. Если у кого-то есть предложения по ускорению, буду рад их услышать! Наконец, хотя в статье отмечается, что многопроцессорность полезна для разгрузки асинхронных процессов, существуют и другие причины для использования многопроцессорности, такие как множество интенсивных математических операций, что я и пытался сделать. В конце концов, я был достаточно доволен улучшением производительности, так как это было намного лучше, чем последовательное выполнение!

Код:

# Python 3.6
from multiprocessing import Pipe, Process

def myWorkFunc(data, connection):
    result = None

    # Do some work and store it in result

    if result:
        connection.send([result])
    else:
        connection.send([None])


def myPipedMultiProcessFunc():

    # Get number of available logical cores
    plimit = multiprocessing.cpu_count()

    # Setup management variables
    results = []
    parent_conns = []
    processes = []
    pcount = 0
    pactive = []
    i = 0

    for data in iterable:
        # Create the pipe for parent-child process communication
        parent_conn, child_conn = Pipe()
        # create the process, pass data to be operated on and connection
        process = Process(target=myWorkFunc, args=(data, child_conn,))
        parent_conns.append(parent_conn)
        process.start()
        pcount += 1

        if pcount == plimit: # There is not currently room for another process
            # Wait until there are results in the Pipes
            finishedConns = multiprocessing.connection.wait(parent_conns)
            # Collect the results and remove the connection as processing
            # the connection again will lead to errors
            for conn in finishedConns:
                results.append(conn.recv()[0])
                parent_conns.remove(conn)
                # Decrement pcount so we can add a new process
                pcount -= 1

    # Ensure all remaining active processes have their results collected
    for conn in parent_conns:
        results.append(conn.recv()[0])
        conn.close()

    # Process results as needed

Я столкнулся с той же проблемой. Это код, который у меня был ранее, который отлично работал на моей локальной машине:

import concurrent.futures


class Concurrent:

    @staticmethod
    def execute_concurrently(function, kwargs_list):
        results = []
        with concurrent.futures.ProcessPoolExecutor() as executor:
            for _, result in zip(kwargs_list, executor.map(function, kwargs_list)):
                results.append(result)
        return results

И я заменил его на этот:

import concurrent.futures


class Concurrent:

    @staticmethod
    def execute_concurrently(function, kwargs_list):
        results = []
        with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
            futures = [executor.submit(function, kwargs) for kwargs in kwargs_list]
        for future in concurrent.futures.as_completed(futures):
            results.append(future.result())
        return results

Работает как шарм.

Взято из этот запрос на включение

multiprocessing.Pool и multiprocessing.Queue изначально не поддерживаются (из-за проблемы с SemLock), но multiprocessing.Process и multiprocessing.Pipe и т. д. работают правильно в AWSLambda.

Это должно позволить вам создать обходное решение, вручную создавая/разветвляя процессы и используя multiprocessing.Pipe для связи между родительским и дочерним процессами. надеюсь, это поможет

Вы можете обновить и использовать среду выполнения AWS Lambda Python 3.7. Работает на меня!

AWS Lambda поддерживает несколько языков благодаря использованию сред выполнения. Вы выбираете среду выполнения при создании функции, и вы можете изменить среды выполнения, обновив конфигурацию вашей функции. Базовая среда выполнения предоставляет дополнительные библиотеки и переменные среды, доступ к которым можно получить из кода функции.