Python中的ProcessPoolExecutor和Lock

Question

我试图在使用concurrent.futures.ProcessPoolExecutor与Locks，但我遇到了运行时错误。(如果相关的话，我是在Windows上工作)

这是我的代码:

import multiprocessing
from concurrent.futures import ProcessPoolExecutor

import time


def f(i, lock):
    with lock:
        print(i, 'hello')
        time.sleep(1)
        print(i, 'world')


def main():
    lock = multiprocessing.Lock()
    pool = ProcessPoolExecutor()
    futures = [pool.submit(f, num, lock) for num in range(3)]
    for future in futures:
        future.result()


if __name__ == '__main__':
    main()

这是我收到的错误信息：

    Traceback (most recent call last):
  File "F:\WinPython-64bit-3.4.3.2\python-3.4.3.amd64\Lib\multiprocessing\queues.py", line 242, in _feed
    obj = ForkingPickler.dumps(obj)
  File "F:\WinPython-64bit-3.4.3.2\python-3.4.3.amd64\Lib\multiprocessing\reduction.py", line 50, in dumps
    cls(buf, protocol).dump(obj)
  File "F:\WinPython-64bit-3.4.3.2\python-3.4.3.amd64\Lib\multiprocessing\synchronize.py", line 102, in __getstate__
    context.assert_spawning(self)
  File "F:\WinPython-64bit-3.4.3.2\python-3.4.3.amd64\Lib\multiprocessing\context.py", line 347, in assert_spawning
    ' through inheritance' % type(obj).__name__
RuntimeError: Lock objects should only be shared between processes through inheritance

奇怪的是，如果我使用multiprocessing.Process编写相同的代码，一切都可以正常工作：

import multiprocessing

import time


def f(i, lock):
    with lock:
        print(i, 'hello')
        time.sleep(1)
        print(i, 'world')


def main():
    lock = multiprocessing.Lock()
    processes = [multiprocessing.Process(target=f, args=(i, lock)) for i in range(3)]
    for process in processes:
        process.start()
    for process in processes:
        process.join()



if __name__ == '__main__':
    main()

这个有效，并且我得到了：

1 hello
1 world
0 hello
0 world
2 hello
2 world

- David Sternlicht

Answer 1

您需要使用一个Manager并使用Manager.Lock()：

import multiprocessing
from concurrent.futures import ProcessPoolExecutor

import time

def f(i, lock):
    with lock:
        print(i, 'hello')
        time.sleep(1)
        print(i, 'world')

def main():
    pool = ProcessPoolExecutor()
    m = multiprocessing.Manager()
    lock = m.Lock()
    futures = [pool.submit(f, num, lock) for num in range(3)]
    for future in futures:
        future.result()


if __name__ == '__main__':
    main()

结果：

% python locks.py
0 hello
0 world
1 hello
1 world
2 hello
2 world

- Nolen Royalty

---

使用Manager.Lock()解决问题的原因应该被解释清楚。 - blazs

这是因为 Manager.Lock() 可以被序列化，而 multiprocessing.Lock() 不能。 - ap14

Answer 2

我尝试了代码，它的运行结果符合预期。

我理解 Manager.Lock() 返回可用于获取锁（即 multiprocessing.managers.AcquirerProxy）的句柄。当与关键字"with"一起使用时，它实际上会锁定除当前处理器外的所有处理器，以便"with"作用域内的代码片段表现为单个处理过程。

- Prabhakaran Thirugnanam