multiprocessing.Manager nested shared objects doesn't work with Queue

user_1621041 0
Viewed 1750

Python docs of the multiprocessing module state:

Changed in version 3.6: Shared objects are capable of being nested. For example, a shared container object such as a shared list can contain other shared objects which will all be managed and synchronized by the SyncManager.

This does work with list and dict. However, if I try to create a shared Queue inside a shared dict, I get an error:

>>> from multiprocessing import Manager
>>> m = Manager()
>>> d = m.dict()
>>> d['a'] = m.list()
>>> d['b'] = m.dict()
>>> d['c'] = m.Queue()
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "<string>", line 2, in __setitem__
  File "/usr/lib/python3.6/multiprocessing/managers.py", line 772, in _callmethod
    raise convert_to_error(kind, result)
multiprocessing.managers.RemoteError: 
---------------------------------------------------------------------------
Traceback (most recent call last):
  File "/usr/lib/python3.6/multiprocessing/managers.py", line 228, in serve_client
    request = recv()
  File "/usr/lib/python3.6/multiprocessing/connection.py", line 251, in recv
    return _ForkingPickler.loads(buf.getbuffer())
  File "/usr/lib/python3.6/multiprocessing/managers.py", line 881, in RebuildProxy
    return func(token, serializer, incref=incref, **kwds)
TypeError: AutoProxy() got an unexpected keyword argument 'manager_owned'
---------------------------------------------------------------------------

Seems like https://hg.python.org/cpython/rev/39e7307f9aee is the changeset which introduced nested shared objects.

memory-management queue process multiprocessing python
2 Answers

The error is caused by AutoProxy currently not handling all BaseProxy arguments. There's a pull request which has not been merged yet. You either need to monkey-patch AutoProxy, or you look into multiprocessing.managers.py and apply the changes in the patch here directly to your source code.

It's really important to fix both lines in the patch to prevent a memory leak in the server process. The manager_owned-flag is used to let the BaseProxy code know, when to skip a reference increment for a proxy the manager owns himself (through nesting).

user_90594200

If you don't want to patch the underlying python library, you can apply this patch in your own code using the following.

I've copied the change from the pull request reference by @Darkonaut and made package name modifications so it works outside the original package. This is placed at the module level of any module that uses multiprocessing.managers.

Note that the solution I referenced in comments in @Darkonaut's answer produced seg faults in my own testing, but this solution did not.

import multiprocessing.managers

def AutoProxy(token, serializer, manager=None, authkey=None,
              exposed=None, incref=True, manager_owned=False):
    '''
    Return an auto-proxy for `token`
    '''
    _Client = multiprocessing.managers.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 = multiprocessing.process.current_process().authkey

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

multiprocessing.managers.AutoProxy = AutoProxy
user_47908710
Related