ProcessPoolExecutor 和 Lock in Python
ProcessPoolExecutor and Lock in Python
我正在尝试将 concurrent.futures.ProcessPoolExecutor
与锁一起使用,但出现 运行 时间错误。
(如果相关的话,我正在研究 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
您需要使用 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
我试验了它按预期工作的代码。
我的理解是Manager.Lock() returns获取句柄(即multiprocessing.managers.AcquirerProxy)。当它与关键字 "with" 一起使用时,它实际上锁定了除当前处理器之外的所有处理器,以便 "with" 范围内的代码片段就像在单个处理中一样.
我正在尝试将 concurrent.futures.ProcessPoolExecutor
与锁一起使用,但出现 运行 时间错误。
(如果相关的话,我正在研究 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
您需要使用 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
我试验了它按预期工作的代码。 我的理解是Manager.Lock() returns获取句柄(即multiprocessing.managers.AcquirerProxy)。当它与关键字 "with" 一起使用时,它实际上锁定了除当前处理器之外的所有处理器,以便 "with" 范围内的代码片段就像在单个处理中一样.