Named multiprocessing lock

user_1576254 0
Viewed 1034

My goal is to lock a directory from multiple processes that will attempt to do git operations on it. I need to create the lock objects after the processes have branched based on a unique string identifier. By reading the docs, I have not found any solution in the multiprocessing module.

Here is the code demonstrating the problem:

from multiprocessing import Lock, Process
import os
import time


class NamedResource:

    def __init__(self, identifier: str):
        self.identifier = identifier
        self.lock = Lock()

def work_on(name: str):
    nr = NamedResource(name)
    with nr.lock:
        print(time.time() % 100 // 1, os.getpid(), f"working on {nr.identifier}")
        time.sleep(5)
        print(time.time() % 100 // 1,os.getpid(), f"done with {nr.identifier}")

res = ['a', 'b']
processes = []

for r in res:
    p1 = Process(target=work_on, args=(r,))
    p1.start()

    p2 = Process(target=work_on, args=(r,))
    p2.start()

    processes += [p1, p2]

for p in processes:
    p.join()

obviously lock objects are different, and they do not prevent the access - as can be seen from the output:

67.0 48689 working on b
67.0 48690 working on b
67.0 48688 working on a
67.0 48687 working on a
72.0 48689 done with b
72.0 48690 done with b
72.0 48688 done with a
72.0 48687 done with a

correct output would be having only one process work on each of [a, b], but two processes working at a time (because there are two different resources that do not interfere).

In the end I have implemented this using writing a hidden file to the directory I want to limit access to, and checking the presence of this file based on the name in another process. This is not very fast and creates a race condition in the short time between the check and writing the file. Is there a better way?

python
1 Answers

a single Lock object can only be entered once. the problem with your code is that you're just creating lots of independent locks, hence processes don't exclude each other as they're all locking/unlocking their own locks

about the nicest way I can get things going is by using a custom Manager to expose a single NamedResource:

from collections import defaultdict
from multiprocessing import Lock, Process
from multiprocessing.managers import SyncManager

class NamedResource:
    def __init__(self):
        # each client process is served by a seperate thread,
        # this lock is used to serialize access to our state
        self.lock = Lock()
        self.locks = defaultdict(Lock)

    def acquire(self, name):
        with self.lock:
            lock = self.locks[name]
        lock.acquire()

    def release(self, name):
        with self.lock:
            lock = self.locks[name]
        lock.release()

class MyManager(SyncManager):
    pass

MyManager.register('NamedResource', NamedResource)

we can change your work_on to use this like this:

def work_on(name: str, locker: NamedResource):
    locker.acquire(name)
    try:
        print(f"{time.time() % 60:.3f} {os.getpid()} working on {name}")
        time.sleep(0.3)
        print(f"{time.time() % 60:.3f} {os.getpid()} done with {name}")
    finally:
        locker.release(name)

you could make a nice context manager to wrap this acquire & release up if you want

we then put this all together by doing:

res = ['a', 'b']
processes = []

with MyManager() as manager:
    locker = manager.NamedResource()

    for r in res:
        p1 = Process(target=work_on, args=(r, locker))
        p1.start()

        p2 = Process(target=work_on, args=(r, locker))
        p2.start()

        processes += [p1, p2]

    for p in processes:
        p.join()

if you wanted that nice context manager I mentioned above, I'd suggest something like:

from contextlib import contextmanager
from multiprocessing.managers import BaseProxy

class NamedResourceProxy(BaseProxy):
    _exposed_ = ('acquire', 'release')

    def acquire(self, name: str):
        return self._callmethod('acquire', (name,))

    def release(self, name: str):
        return self._callmethod('release', (name,))

    @contextmanager
    def use(self, name: str):
        self.acquire(name)
        try:
            yield None
        finally:
            self.release(name)

and change the call to register to something like: MyManager.register('NamedResource', NamedResource, NamedResourceProxy)

you work_on could then be something like:

def work_on(name: str, locker: NamedResource):
    with locker.use(name):
        print(f"{time.time() % 60:.3f} {os.getpid()} working on {name}")
        time.sleep(0.3)
        print(f"{time.time() % 60:.3f} {os.getpid()} done with {name}")
user_13583080
Related