python - Multiprocessing managers and custom classes

Question

I do not know why, but I get this strange error whenever I try to pass to the method of a shared object shared custom class object. Python version: 3.6.3

Code:

from multiprocessing.managers import SyncManager

class MyManager(SyncManager): pass
class MyClass: pass

class Wrapper:
    def set(self, ent):
        self.ent = ent

MyManager.register('MyClass', MyClass)
MyManager.register('Wrapper', Wrapper)

if __name__ == '__main__':
    manager = MyManager()
    manager.start()

    try:
        obj = manager.MyClass()
        lst = manager.list([1,2,3])

        collection = manager.Wrapper()
        collection.set(lst) # executed fine
        collection.set(obj) # raises error
    except Exception as e:
        raise

Error:

---------------------------------------------------------------------------
Traceback (most recent call last):
  File "D:Program FilesPython363libmultiprocessingmanagers.py", line 228, in serve_client
    request = recv()
  File "D:Program FilesPython363libmultiprocessingconnection.py", line 251, in recv
    return _ForkingPickler.loads(buf.getbuffer())
  File "D:Program FilesPython363libmultiprocessingmanagers.py", line 881, in RebuildProxy
    return func(token, serializer, incref=incref, **kwds)
TypeError: AutoProxy() got an unexpected keyword argument 'manager_owned'
---------------------------------------------------------------------------

What's the problem here?

Answer 1

I ran into this too, as noted, this is a bug in Python multiprocessing (see issue #30256) and the pull request that corrects this has not yet been merged. The pull request has since been superseded by another PR that makes the same change but adds a test as well.

Apart from manually patching your local installation, you have three other options:

• you could use the MakeProxyType() callable to specify your proxytype, without relying on the AutoProxy proxy generator,
• you could define a custom proxy class,
• you can patch the bug with a monkeypatch

I'll describe those options below, after explaining what AutoProxy does:

What's the point of the AutoProxy class

The multiprocessing Manager pattern gives access to shared values by putting the values all in the same, dedicated 'canonical values server' process. All other processes (clients) talk to the server through proxies that then pass messages back and forth with the server.

The server does need to know what methods are acceptable for the type of object, however, so clients can produce a proxy object with the same methods. This is what the AutoProxy object is for. Whenever a client needs a new instance of your registered class, the default proxy the client creates is an AutoProxy, which then asks the server to tell it what methods it can use.

Once it has the method names, it calls MakeProxyType to construct a new class and then creates an instance for that class to return.

All this is deferred until you actually need an instance of the proxied type, so in principle AutoProxy saves a little bit of memory if you are not using certain classes you have registered. It's very little memory, however, and the downside is that this process has to take place in each client process.

These proxy objects use reference counting to track when the server can remove the canonical value. It is that part that is broken in the AutoProxy callable; a new argument is passed to the proxy type to disable reference counting when the proxy object is being created in the server process rather than in a client but the AutoProxy type wasn't updated to support this.

So, how can you fix this? Here are those 3 options:

Use the MakeProxyType() callable

As mentioned, AutoProxy is really just a call (via the server) to get the public methods of the type, and a call to MakeProxyType(). You can just make these calls yourself, when registering.

So, instead of

from multiprocessing.managers import SyncManager
SyncManager.register("YourType", YourType)

use

from multiprocessing.managers import SyncManager, MakeProxyType, public_methods
#               arguments:    classname,  sequence of method names
YourTypeProxy = MakeProxyType("YourType", public_methods(YourType))
SyncManager.register("YourType", YourType, YourTypeProxy)

Feel free to inline the MakeProxyType() call there.

If you were using the exposed argument to SyncManager.register(), you should pass those names to MakeProxyType instead:

# SyncManager.register("YourType", YourType, exposed=("foo", "bar"))
# becomes
YourTypeProxy = MakeProxyType("YourType", ("foo", "bar"))
SyncManager.register("YourType", YourType, YourTypeProxy)

You'd have to do this for all the pre-registered types, too:

from multiprocessing.managers import SyncManager, AutoProxy, MakeProxyType, public_methods

registry = SyncManager._registry
for typeid, (callable, exposed, method_to_typeid, proxytype) in registry.items():
    if proxytype is not AutoProxy:
        continue
    create_method = hasattr(managers.SyncManager, typeid)
    if exposed is None:
        exposed = public_methods(callable) 
    SyncManager.register(
        typeid,
        callable=callable,
        exposed=exposed,
        method_to_typeid=method_to_typeid,
        proxytype=MakeProxyType(f"{typeid}Proxy", exposed),
        create_method=create_method,
    )

Create custom proxies

You could not rely on multiprocessing creating a proxy for you. You could just write your own. The proxy is used in all processes except for the special 'managed values' server process, and the proxy should pass messages back and forth. This is not an option for the already-registered types, of course, but I'm mentioning it here because for your own types this offers opportunities for optimisations.

Note that you should have methods for all interactions that need to go back to the 'canonical' value instance, so you'd need to use properties to handle normal attributes or add __getattr__, __setattr__ and __delattr__ methods as needed.

The advantage is that you can have very fine-grained control over what methods actually need to exchange data with the server process; in my specific example, my proxy class caches information that is immutable (the values would never change once the object was created), but were used often. That includes a flag value that controls if other methods would do something, so the proxy could just check the flag value and not talk to the server process if not set. Something like this:

class FooProxy(BaseProxy):
    # what methods the proxy is allowed to access through calls
    _exposed_ = ("__getattribute__", "expensive_method", "spam")

    @property
    def flag(self):
        try:
            v = self._flag
        except AttributeError:
            # ask for the value from the server, "realvalue.flag"
            # use __getattribute__ because it's an attribute, not a property
            v = self._flag = self._callmethod("__getattribute__", ("flag",))
        return flag

    def expensive_method(self, *args, **kwargs):
        if self.flag:   # cached locally!
            return self._callmethod("expensive_method", args, kwargs)

    def spam(self, *args, **kwargs):
        return self._callmethod("spam", args, kwargs

SyncManager.register("Foo", Foo, FooProxy)

Because MakeProxyType() returns a BaseProxy subclass, you can combine that class with a custom subclass, saving yourself having to write any methods that just consist of return self._callmethod(...):

# a base class with the methods generated for us. The second argument
# doubles as the 'permitted' names, stored as _exposed_
FooProxyBase = MakeProxyType(
    "FooProxyBase",
    ("__getattribute__", "expensive_method", "spam"),
)

class FooProxy(FooProxyBase):
    @property
    def flag(self):
        try:
            v = self._flag
        except AttributeError:
            # ask for the value from the server, "realvalue.flag"
            # use __getattribute__ because it's an attribute, not a property
            v = self._flag = self._callmethod("__getattribute__", ("flag",))
        return flag

    def expensive_method(self, *args, **kwargs):
        if self.flag:   # cached locally!
            return self._callmethod("expensive_method", args, kwargs)

    def spam(self, *args, **kwargs):
        return self._callmethod("spam", args, kwargs

SyncManager.register("Foo", Foo, FooProxy)

Again, this won't solve the issue with standard types nested inside other proxied values.

Apply a monkeypatch

I use this to fix the AutoProxy callable, this should automatically avoid patching when you are running a Python version where the fix has already been applied to the source code:

# Backport of https://github.com/python/cpython/pull/4819
# Improvements to the Manager / proxied shared values code
# broke handling of proxied objects without a custom proxy type,
# as the AutoProxy function was not updated.
#
# This code adds a wrapper to AutoProxy if it is missing the
# new argument.

import logging
from inspect import signature
from functools import wraps
from multiprocessing import managers


logger = logging.getLogger(__name__)
orig_AutoProxy = managers.AutoProxy


@wraps(managers.AutoProxy)
def AutoProxy(*args, incref=True, manager_owned=False, **kwargs):
    # Create the autoproxy without the manager_owned flag, then
    # update the flag on the generated instance. If the manager_owned flag
    # is set, `incref` is disabled, so set it to False here for the same
    # result.
    autoproxy_incref = False if manager_owned else incref
    proxy = orig_AutoProxy(*args, incref=autoproxy_incref, **kwargs)
    proxy._owned_by_manager = manager_owned
    return proxy


def apply():
    if "manager_owned" in signature(managers.AutoProxy).parameters:
        return

    logger.debug("Patching multiprocessing.managers.AutoProxy to add manager_owned")
    managers.AutoProxy = AutoProxy

    # re-register any types already registered to SyncManager without a custom
    # proxy type, as otherwise these would all be using the old unpatched AutoProxy
    SyncManager = managers.SyncManager
    registry = managers.SyncManager._registry
    for typeid, (callable, exposed, method_to_typeid, proxytype) in registry.items():
        if proxytype is not orig_AutoProxy:
            continue
        create_method = hasattr(managers.SyncManager, typeid)
        SyncManager.register(
            typeid,
            callable=callable,
            exposed=exposed,
            method_to_typeid=method_to_typeid,
            create_method=create_method,
        )

Import the above and call the apply() function to fix multiprocessing. Do so before you start the manager server!