Python использует блокировку между процессами

Я пытаюсь использовать частичную функцию, чтобы pool.map() мог нацеливать на функцию с несколькими параметрами (в данном случае объект Lock()).

Вот пример кода (взятый из ответа на предыдущий вопрос):

from functools import partial

def target(lock, iterable_item):
    for item in items:
        # Do cool stuff
        if (... some condition here ...):
            lock.acquire()
            # Write to stdout or logfile, etc.
            lock.release()

def main():
    iterable = [1, 2, 3, 4, 5]
    pool = multiprocessing.Pool()
    l = multiprocessing.Lock()
    func = partial(target, l)
    pool.map(func, iterable)
    pool.close()
    pool.join()

Однако, когда я запускаю этот код, я получаю ошибку:

Runtime Error: Lock objects should only be shared between processes through inheritance.

Что мне здесь не хватает? Как я могу совместно использовать блокировку между моими подпроцессами?

Ответы

Ответ 1

Извините, я должен был поймать это в ответе на ваш другой вопрос. Вы не можете передавать обычные multiprocessing.Lock объекты в методы Pool, потому что их нельзя травить. Есть два способа обойти это. Один из них - создать Manager() и передать Manager.Lock():

def main():
    iterable = [1, 2, 3, 4, 5]
    pool = multiprocessing.Pool()
    m = multiprocessing.Manager()
    l = m.Lock()
    func = partial(target, l)
    pool.map(func, iterable)
    pool.close()
    pool.join()

Это немного тяжеловес; с помощью Manager требуется нерест другого процесса для размещения сервера Manager. И все вызовы acquire/release блокировки должны быть отправлены на этот сервер через IPC.

Другой вариант - передать обычный multiprocessing.Lock() в момент создания пула, используя initializer kwarg. Это сделает ваш экземпляр блокировки глобальным во всех дочерних рабочих:

def target(iterable_item):
    for item in items:
        # Do cool stuff
        if (... some condition here ...):
            lock.acquire()
            # Write to stdout or logfile, etc.
            lock.release()
def init(l):
    global lock
    lock = l

def main():
    iterable = [1, 2, 3, 4, 5]
    l = multiprocessing.Lock()
    pool = multiprocessing.Pool(initializer=init, initargs=(l,))
    pool.map(target, iterable)
    pool.close()
    pool.join()

Второе решение имеет побочный эффект, который больше не требует partial.