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

Ответ 1

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

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

Ответ 2

Вы можете запускать подпрограммы параллельно на 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

Ответ 3

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

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