ProcessPoolExecutor and Lock in Python

Question

I am trying to use concurrent.futures.ProcessPoolExecutor with Locks, but I'm getting a run time error. (I'm working on Windows if that's relevant)

Here's my code:

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()

Here's the error I get:

    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

What's weird is that if I write the same code with multiprocessing.Process it all works fine:

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()

This works and I get:

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

score 25 · Accepted Answer · answered Feb 14 '16 at 17:09

25

You need to use a Manager and use a Manager.Lock() instead:

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()

Result:

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

answered Feb 14 '16 at 17:09

Nolen Royalty

18,415
4
40
50

31

It would be good to explain _why_ using `Manager.Lock()` solves the issue. – blazs Apr 26 '18 at 10:27
4

This is because `Manager.Lock()` can be pickled and `multiprocessing.Lock()` can't be. – ap14 Dec 09 '20 at 13:09

score 0 · Answer 2 · answered Oct 28 '20 at 06:03

I experimented the code it's working as expected. My understanding is Manager.Lock() returns the handle to acquire(i.e. multiprocessing.managers.AcquirerProxy). when it is used along with key word "with", It actually locks all the processors except the current one so that the piece of code within the "with" scope acts as in the single processing.

ProcessPoolExecutor and Lock in Python

2 Answers2

Linked