Share RLock between multiple instances of Python with multiprocessing

Question

Consider this MWE:

from multiprocessing import RLock

class TheSetup:
    def __init__(self):
        self._hardware_lock = RLock()
    
    def hold_hardware(self):
        return self._hardware_lock
    
    def do_thing_with_hardware(self):
        with self._hardware_lock:
            print('Doing thing...')
    
    def do_other_thing_with_hardware(self):
        with self._hardware_lock:
            print('Doing other thing...')

if __name__=='__main__':
    from multiprocessing.managers import BaseManager
    
    class TheSetupManager(BaseManager):
        pass
    
    the_setup = TheSetup()
    
    TheSetupManager.register('get_the_setup', callable=lambda:the_setup)
    m = TheSetupManager(address=('', 50000), authkey=b'abracadabra')
    s = m.get_server()
    print('Serving...')
    s.serve_forever()

When this script is executed, it instantiates the_setup and serves it. Then I want clients to be able to do things like this from other scripts:

from multiprocessing.managers import BaseManager

class TheSetup(BaseManager):
    pass

TheSetup.register('get_the_setup')
m = TheSetup(address=('', 50000), authkey=b'abracadabra')
m.connect()
the_setup = m.get_the_setup()

with the_setup.hold_hardware(): # Removing this `with` it works fine, but I cannot guarantee to the client that nobody else will use this hardware in the meantime.
    the_setup.do_thing_with_hardware()
    the_setup.do_other_thing_with_hardware()

However, I get RuntimeError: RLock objects should only be shared between processes through inheritance. If the with the_setup.hold_hardware(): is removed, it works fine but then I cannot guarantee that the hardware wasn't used by someone else in the middle.

Is it possible to do what I want? I.e. having the_setup running 24/7 and allowing interaction with it at any time from other Python instances. How?

Answer 1

Update : I have included a patch for multiprocessing.managers to assimilate RLocks seamlessly within. Scroll below to the next section.

Multiprocessing does not like it when you pass objects used for synchronization inside other such objects. That means you cannot put semaphores, locks, queues, pipes inside other queues and pipes (those offered through multiprocessing library). When you create a manager using BaseManager, it uses pipes internally to establish communication between different processes and the manager process. So when you do the_setup.hold_hardware(), you are essentially attempting to pass an Rlock through a pipe, which as discussed, does not work.

Why workarounds don't work

The most simplest fix one would think of using here would be to create a manager and use manager.Rlock. This uses threading.Rlock instead of the one available through multiprocessing (and therefore can be shared using queues/pipes), but works in a multiprocessing environment because access to it is synchronized though pipes (again, it's using a manager).

Hence, this code should at least execute:

server.py

from multiprocessing.managers import SyncManager
.
.
.
def __init__(self):
    manager = SyncManager(authkey=b'same')
    manager.start()
    self._hardware_lock = manager.RLock()

client.py

import multiprocessing
import os
from multiprocessing.managers import BaseManager

class TheSetup(BaseManager):
    pass

TheSetup.register('get_the_setup')
m = TheSetup(address=('localhost', 50000), authkey=b'abracadabra')
m.connect()

multiprocessing.current_process().authkey = b'same'  # Need to setup proper authentication

the_setup = m.get_the_setup()
print(the_setup)
with the_setup.hold_hardware(): # Removing this `with` it works fine, but I cannot guarantee to the client that nobody else will use this hardware in the meantime.
    print('in')
    the_setup.do_thing_with_hardware()
    the_setup.do_other_thing_with_hardware()

Note that we need to set the manager and the client's authkey to the same value otherwise there will be an authentication error when attempting to unpickle the lock.

But regardless, even though the code will run, it won't do what you think it should do. It will block when trying to run the_setup.do_thing_with_hardware(). This is because the Rlock is actually inside manager process, and when you get the lock using with the_setup.hold_hardware() you are actually getting a proxy of the lock instead (try doing print(type(the_setup.hold_hardware())) ). When you attempt to acquire the lock using the proxy, the appropriate function name is sent to the manager process and is executed on the managed object via threading. This defeats the whole purpose of Rlock and the implementation of Rlock inside managers is useless.

Patching multiprocessing to work with RLocks

If you really want to use RLocks inside managers, then you will need to make your own managers and proxies, by subclassing BaseManager and BaseProxy, to relay process identifiers (like pids) which can then be used to create RLocks. Consider the below "patch":

import time
import os, sys
from multiprocessing.managers import BaseProxy, BaseManager, Server, Token, RebuildProxy
from multiprocessing import process
from multiprocessing.context import get_spawning_popen, ProcessError
from _thread import allocate_lock
from traceback import format_exc

from multiprocessing.managers import util, dispatch, convert_to_error, listener_client, public_methods, State
import threading


class PIDProxy(BaseProxy):


    def _callmethod(self, methodname, args=(), kwds={}):

        try:
            conn = self._tls.connection
        except AttributeError:
            util.debug('thread %r does not own a connection',
                       threading.current_thread().name)
            self._connect()
            conn = self._tls.connection

        conn.send((self._id, methodname, args, kwds, str(os.getpid())))
        kind, result = conn.recv()

        if kind == '#RETURN':
            return result
        elif kind == '#PROXY':
            exposed, token = result
            proxytype = self._manager._registry[token.typeid][-1]
            token.address = self._token.address
            proxy = proxytype(
                token, self._serializer, manager=self._manager,
                authkey=self._authkey, exposed=exposed
            )
            conn = self._Client(token.address, authkey=self._authkey)
            dispatch(conn, None, 'decref', (token.id,))
            return proxy
        raise convert_to_error(kind, result)

    def __reduce__(self):
        kwds = {}
        if get_spawning_popen() is not None:
            kwds['authkey'] = self._authkey

        if getattr(self, '_isauto', False):
            kwds['exposed'] = self._exposed_
            return (RebuildProxy,
                    (MyAutoProxy, self._token, self._serializer, kwds))
        else:
            return (RebuildProxy,
                    (type(self), self._token, self._serializer, kwds))


class PIDServer(Server):

    def serve_client(self, conn):

        global pid_registry
        util.debug('starting server thread to service %r',
                   threading.current_thread().name)

        recv = conn.recv
        send = conn.send
        id_to_obj = self.id_to_obj

        while not self.stop_event.is_set():

            try:
                methodname = obj = None
                request = recv()
                ident, methodname, args, kwds, pid = request
                try:
                    obj, exposed, gettypeid = id_to_obj[ident]
                except KeyError as ke:
                    try:
                        obj, exposed, gettypeid = \
                            self.id_to_local_proxy_obj[ident]
                    except KeyError:
                        raise ke

                if methodname not in exposed:
                    raise AttributeError(
                        'method %r of %r object is not in exposed=%r' %
                        (methodname, type(obj), exposed)
                        )

                function = getattr(obj, methodname)

                try:
                    pid_registry.forwarded_for = pid
                    res = function(*args, **kwds)
                except Exception as e:
                    msg = ('#ERROR', e)
                else:
                    typeid = gettypeid and gettypeid.get(methodname, None)
                    if typeid:
                        rident, rexposed = self.create(conn, typeid, res)
                        token = Token(typeid, self.address, rident)
                        msg = ('#PROXY', (rexposed, token))
                    else:
                        msg = ('#RETURN', res)

            except AttributeError:
                if methodname is None:
                    msg = ('#TRACEBACK', format_exc())
                else:
                    try:
                        fallback_func = self.fallback_mapping[methodname]
                        result = fallback_func(
                            self, conn, ident, obj, *args, **kwds
                            )
                        msg = ('#RETURN', result)
                    except Exception:
                        msg = ('#TRACEBACK', format_exc())

            except EOFError:
                util.debug('got EOF -- exiting thread serving %r',
                           threading.current_thread().name)
                sys.exit(0)

            except Exception:
                msg = ('#TRACEBACK', format_exc())

            try:
                try:
                    send(msg)
                except Exception:
                    send(('#UNSERIALIZABLE', format_exc()))
            except Exception as e:
                util.info('exception in thread serving %r',
                        threading.current_thread().name)
                util.info(' ... message was %r', msg)
                util.info(' ... exception was %r', e)
                conn.close()
                sys.exit(1)


class ForwardPIDManager(BaseManager):
    _Server = PIDServer

    def get_server(self):

        if self._state.value != State.INITIAL:
            if self._state.value == State.STARTED:
                raise ProcessError("Already started server")
            elif self._state.value == State.SHUTDOWN:
                raise ProcessError("Manager has shut down")
            else:
                raise ProcessError(
                    "Unknown state {!r}".format(self._state.value))
        return self._Server(self._registry, self._address,
                      self._authkey, self._serializer)


def MyMakeProxyType(name, exposed, _cache={}):
    '''
    Return a proxy type whose methods are given by `exposed`
    '''
    exposed = tuple(exposed)
    try:
        return _cache[(name, exposed)]
    except KeyError:
        pass

    dic = {}

    for meth in exposed:
        exec('''def %s(self, /, *args, **kwds):
        return self._callmethod(%r, args, kwds)''' % (meth, meth), dic)

    ProxyType = type(name, (PIDProxy,), dic)
    ProxyType._exposed_ = exposed
    _cache[(name, exposed)] = ProxyType
    return ProxyType


def MyAutoProxy(token, serializer, manager=None, authkey=None,
              exposed=None, incref=True, manager_owned=False):

    _Client = listener_client[serializer][1]

    if exposed is None:
        conn = _Client(token.address, authkey=authkey)
        try:
            exposed = dispatch(conn, None, 'get_methods', (token,))
        finally:
            conn.close()

    if authkey is None and manager is not None:
        authkey = manager._authkey
    if authkey is None:
        authkey = process.current_process().authkey

    ProxyType = MyMakeProxyType('AutoProxy[%s]' % token.typeid, exposed)
    proxy = ProxyType(token, serializer, manager=manager, authkey=authkey,
                      incref=incref, manager_owned=manager_owned)
    proxy._isauto = True
    return proxy

def init():
    global pid_registry
    pid_registry = threading.local()

Over here, PIDProxy is identical to BaseProxy, except for the fact that apart from only sending the method name and arguments to be called on the managed object, it sends an additional identifier with the value str(os.getpid()) as well. Similarly, ForwardPIDManager is identical to BaseManager, except for the fact the it uses a modified subclass of multiprocessing.managers.Server rather than the default parent, to start the server process. PIDServer modifies it's parent to accept an extra variable (the process identifier) when unpacking requests from proxies and stores it's value inside the current thread's local storage (created after executing init inside the server process; further reading about thread local storage here). This happens before the requested function is executed, meaning that all methods of the managed object will have access to this storage. Lastly, MyAutoProxy and MyMakeProxyType override the default methods to create proxies using our subclasses instead.

All you need to do now is to use ForwardPIDManager instead of BaseManager, and specify the proxytype kwarg explicitly as MyAutoProxy. All managed objects will then have access to the pid of the process that called the function by doing

global pid_registry
pid = pid_registry.forwarded_for

from within the functions on the managed object.

Creating RLock

Using this, we can create our own implementation of RLocks, which closely mirrors that of threading.RLock, but uses this pid_registry.forwarded_for to verify owners instead of threading.get_ident:

class ManagerRLock:

    def __init__(self):
        self._block = allocate_lock()
        self._owner = None
        self._count = 0

    def acquire(self, blocking=True, timeout=-1):
        global pid_registry
        process = pid_registry.forwarded_for
        if self._owner == process:
            self._count += 1
            return True

        rc = self._block.acquire(blocking, timeout)

        if rc:
            self._owner = process
            self._count = 1
        return rc

    def release(self):
        global pid_registry
        process = pid_registry.forwarded_for
        if self._owner != process:
            raise RuntimeError("cannot release un-acquired lock")
        self._count = count = self._count - 1
        if not count:
            self._owner = None
            self._block.release()

    def __repr__(self):
        owner = self._owner
        return "<%s %s.%s object owner=%r count=%d at %s>" % (
            "locked" if self._block.locked() else "unlocked",
            self.__class__.__module__,
            self.__class__.__qualname__,
            owner,
            self._count,
            hex(id(self))
        )

    def __enter__(self):
        return self.acquire()

    def __exit__(self, exc_type, exc_val, exc_tb):
        return self.release()

Keep in mind that objects of ManagerRLock are not picklable, and therefore should not be passed around by value from manager to proxies. You can, however, expose it's methods (acquire, release) to proxies without risk (example in the below section). Also, these are meant to be created directly, therefore do not use another manager to create them.

Example implementation

Using our patch and ManagerRLock, our implementation of TheSetup class becomes like this:

class TheSetup:
    def __init__(self):
        self._hardware_lock = None

    def initialize_lock(self):
        self._hardware_lock = ManagerRLock()

    def hold_hardware(self):
        self._hardware_lock.acquire()

    def release_hardware(self):
        self._hardware_lock.release()

    def do_thing_with_hardware(self):

        with self._hardware_lock:
            print('Doing thing...')
            # Mimic work being done, helpful for testing locks
            time.sleep(7)
            print('done')

    def do_other_thing_with_hardware(self):
        with self._hardware_lock:
            print('Doing other thing...')

    def return_(self):
        return self


if __name__ == '__main__':
    
    # Execute init here since this the main module is the service process
    init()
    
    # You can initialize lock in the constructor itself, this is only here for consistency between 
    # example implementations
    the_setup = TheSetup()
    the_setup.initialize_lock()
    
    ForwardPIDManager.register('get_the_setup', lambda: the_setup, MyAutoProxy)
    m = ForwardPIDManager(address=('localhost', 50000), authkey=b'abracadabra')

    s = m.get_server()
    print('Serving...')
    s.serve_forever()

One important thing to notice here is that you are not exposing the whole lock, only its methods (look at hold_hardware and release_hardware). This is important because ManagerRLock just like multiprocessing.RLock, is unpicklable. One side effect of this is that you can't directly use context managers for the lock and you would have to do

the_setup.hold_hardware()
.
.
.
the_setup.release_hardware()

inside client.py instead of using a with block. However, if this bothers you can create a wrapper for the lock from within the client and use that instead (example given in the next section).

Lastly, also notice how we call init from inside the main module itself. This is because init must be executed in the server process, and since we are retrieving the server and calling serve_forever inside the main process itself, we must execute init there too. If instead you are using the .start() method of managers to create the server in another process, here is the equivalent code:

if __name__ == '__main__':
    the_setup = TheSetup()
    
    # We cannot use lambda since that's not picklable
    ForwardPIDManager.register('get_the_setup', callable=the_setup.return_, proxytype=MyAutoProxy)
    m = ForwardPIDManager(address=('localhost', 50000), authkey=b'abracadabra')
    
    # Start the server in another process, and tell it to run init as soon as it starts
    m.start(initializer=init)
    
    # Initialize lock after transferring the instance data to the manager process (ManagerRLock is not picklable)    
    proxy_setup = m.get_the_setup()
    proxy_setup.initialize_lock()
    
    # Wait for the server process to quit, only for demonstration, do not do this is in real code!
    m._process.join()

Final solution

Combine the whole patch provided above (classes PIDProxy, PIDServer, ForwardPIDManager, and functions init, MyMakeProxyType, MyAutoProxy), the ManagerRLock class, and code related to your implementation (class TheSetup along with the if __name__... block) into one single server.py file. Then, an example client.py which uses the RLock can be like below:

client.py

from server import ForwardPIDManager, MyAutoProxy


class TheSetup(ForwardPIDManager):
    pass


class LockWrapper:

    def __init__(self, proxy):
        self.proxy = proxy

    def __enter__(self):
        return self.proxy.hold_hardware()

    def __exit__(self, exc_type, exc_val, exc_tb):
        return self.proxy.release_hardware()


TheSetup.register('get_the_setup', proxytype=MyAutoProxy)
m = TheSetup(address=('localhost', 50000), authkey=b'abracadabra')
m.connect()

the_setup = m.get_the_setup()
the_setup_lock = LockWrapper(the_setup)

with the_setup_lock:

    the_setup.do_thing_with_hardware()
    the_setup.do_other_thing_with_hardware()

Notice the use of LockWrapper as a way to use the lock with a context manager.

A word about threads...

The solution relies on using threading.local, therefore, trying to access pid_registry.forwarded_for from inside another thread would fail. Hence, if you are using threading inside your shared class, then make sure you explicitly pass the pid to thread upon starting.

Additionally, ManagerRLock expects single-threaded processes (processes can be more than one) to access it. This means that if you are running multiple processes, which each are also multi-threaded, then using ManagerRLock might be unsafe (untested). However, if this is the case then you can trivially extend the patch by passing not only the pid, but the thread identifer as well (from inside PIDProxy) and store this inside pid_registry (from inside PIDServer). Then you will have access to the thread as well as the pid which sent the request to the manager inside ManagerRLock, and you can then decide the current owner of the lock based on both these variables.

Edit: A quick note about proxies, if you want to use your own (and not use MyAutoProxy), then you can do so, just make sure that the proxy subclasses PIDProxy.

Answer 2

I am posting this as an answer because this solves my current problem, however it is a quick'n dirty solution so I will not accept it and would be glad to hear of better approaches.

I implemented my own lock in which the actual "lock machinery" is written to disk, which is a resource shared by all processes (in a normal PC) and I am not in the need of high speed. This is the code for this lock:

from pathlib import Path
import time
import atexit
import datetime

SEPARATION_STRING = '<-->' # This just has to be a very weird string that will never appear in the name of whoever wants to acquire the lock.

def _check_name(name:str):
    if not isinstance(name, str):
        raise TypeError(f'`name` must be a string.')
    if name == SEPARATION_STRING:
        raise ValueError(f'`name` cannot be "{SEPARATION_STRING}", please use another.')

class NamedLock:
    """This class implements a named lock, i.e. you can lock it or not
    depending on your name. The lock 
    """
    def __init__(self):
        self._path_to_this_lock_file = Path.home()/Path(f'NamedLock_{datetime.datetime.now().strftime("%Y%m%d%H%M%S%f")}.txt')
        def release_lock_at_exit():
            if self._path_to_this_lock_file.is_file():
                self._path_to_this_lock_file.unlink()
        atexit.register(release_lock_at_exit)
    
    def locked(self):
        """Return `True` if the lock is acquired."""
        return self._path_to_this_lock_file.is_file()
    
    def locked_by(self, name:str):
        """Return `True` if the lock is acquired by `name`."""
        _check_name(name)
        if self.locked():
            return self._acquired_by()==name
        return False
    
    def _acquired_by(self):
        """Returns the name of the whoever acquires the lock, or `None` if
        it is not acquired by anyone."""
        if self.locked():
            with open(self._path_to_this_lock_file, 'r') as f:
                return f.read().split(SEPARATION_STRING)[0]
        else:
            return None
    
    def acquire(self, name:str):
        """Acquire the lock, or wait indefinitely until it can be acquired."""
        _check_name(name)
        if self.locked_by(name):
            with open(self._path_to_this_lock_file, 'a') as f:
                print(name, file=f, end=SEPARATION_STRING)
        else:
            while self.locked():
                time.sleep(.1)
            with open(self._path_to_this_lock_file, 'w') as f:
                print(name, file=f, end=SEPARATION_STRING)
    
    def release(self, name:str):
        """Release the lock (only one recursive acquisition level)."""
        _check_name(name)
        if self.locked_by(name):
            with open(self._path_to_this_lock_file, 'r') as f:
                locks = f.read().split(SEPARATION_STRING)
            locks = locks[:-2]
            if len(locks) == 0: # Release the lock.
                self._path_to_this_lock_file.unlink()
            else:
                with open(self._path_to_this_lock_file, 'w') as f:
                    print(SEPARATION_STRING.join(locks), file=f, end=SEPARATION_STRING)
    
    def __call__(self, name:str):
        _check_name(name)
        self._wants_to_acquire = name
        return self
    
    def __enter__(self):
        if not hasattr(self, '_wants_to_acquire'):
            raise RuntimeError(f'To use with a `with` statement you have to call the `NamedLock` object with a name, e.g.: `with my_named_lock("a name"):`.')
        self.acquire(f'{self._wants_to_acquire}')
        if hasattr(self, '_wants_to_acquire'): # We have to check because in the meantime someone else could have change this.
            delattr(self, '_wants_to_acquire')
        return self
    
    def __exit__(self, exc_type, exc_value, traceback):
        self.release(self._acquired_by())

Because I don't know how to automatically distinguish between each process, I made this a "named lock" so when someone wants to acquire a lock he gives his name, this allows to get a similar behavior to RLock where the recursion is allowed if the same name of the "lock holder" is given.

Then, TheSetup is modified like this:

from multiprocessing import RLock
from NamedLock import NamedLock

class TheSetup:
    def __init__(self):
        self._hardware_lock = NamedLock()
    
    def hold_hardware(self, who:str):
        return self._hardware_lock(who)
    
    def do_thing_with_hardware(self, who:str):
        with self._hardware_lock(who):
            print('Doing thing...')
            return 'Thing done :)'
    
    def do_other_thing_with_hardware(self, who:str):
        with self._hardware_lock(who):
            print('Doing other thing...')
            return 'Other thing done!'

if __name__=='__main__':
    from multiprocessing.managers import BaseManager
    
    class TheSetupManager(BaseManager):
        pass
    
    the_setup = TheSetup()
    
    TheSetupManager.register('get_the_setup', callable=lambda:the_setup)
    m = TheSetupManager(address=('', 50000), authkey=b'abracadabra')
    s = m.get_server()
    print('Serving...')
    s.serve_forever()

The drawback is that now each method has to ask for the name of whoever is calling it to know if he/she is allowed to execute this method or has to wait.

Then, an example of a client could be this:

from multiprocessing.managers import BaseManager
import time
import datetime
import os

MY_NAME = str(os.getpid())

class TheSetup(BaseManager):
    pass

TheSetup.register('get_the_setup')
m = TheSetup(address=('', 50000), authkey=b'abracadabra')
m.connect()
the_setup = m.get_the_setup()

with the_setup.hold_hardware(MY_NAME):
    for k in range(5):
        print(f'{datetime.datetime.now()}: {the_setup.do_thing_with_hardware(MY_NAME)}')
        print(f'{datetime.datetime.now()}: {the_setup.do_other_thing_with_hardware(MY_NAME)}')
        time.sleep(2)

print('Finished.')
while True:
    time.sleep(2)

and a different client could be:

from multiprocessing.managers import BaseManager
import os
import datetime

MY_NAME = str(os.getpid())

class TheSetup(BaseManager):
    pass

TheSetup.register('get_the_setup')
m = TheSetup(address=('', 50000), authkey=b'abracadabra')
m.connect()
the_setup = m.get_the_setup()

print(f'{datetime.datetime.now()}: {the_setup.do_thing_with_hardware(MY_NAME)}')
print(f'{datetime.datetime.now()}: {the_setup.do_other_thing_with_hardware(MY_NAME)}')

As seen, each client's name is actually their process ID. Now, client 1 can run his script which takes 10 seconds to execute but he can be sure that while his script is within the with the_setup.hold_hardware(MY_NAME): no other client can access to the hardware. I.e. if client 2 runs his script after one second, he will have to wait until client 1 has finished.

This answer is exactly what I want to do, with the exception of the need to pass the name of whoever calls each method (this I would like to automate).