diff -r 4aa6233961e5 Doc/library/multiprocessing.rst --- a/Doc/library/multiprocessing.rst Wed Sep 07 04:03:02 2016 +0300 +++ b/Doc/library/multiprocessing.rst Wed Sep 07 14:52:42 2016 -0500 @@ -1682,7 +1682,9 @@ of processes. Objects of this type are returned by :func:`multiprocessing.Manager`. - It also supports creation of shared lists and dictionaries. + Its methods create and return :ref:`multiprocessing-proxy_objects` for a + number of commonly used data types to be synchronized across processes. + This notably includes shared lists and dictionaries. .. method:: Barrier(parties[, action[, timeout]]) @@ -1745,30 +1747,12 @@ dict(mapping) dict(sequence) - Create a shared ``dict`` object and return a proxy for it. + Create a shared :class:`dict` object and return a proxy for it. .. method:: list() list(sequence) - Create a shared ``list`` object and return a proxy for it. - - .. note:: - - Modifications to mutable values or items in dict and list proxies will not - be propagated through the manager, because the proxy has no way of knowing - when its values or items are modified. To modify such an item, you can - re-assign the modified object to the container proxy:: - - # create a list proxy and append a mutable object (a dictionary) - lproxy = manager.list() - lproxy.append({}) - # now mutate the dictionary - d = lproxy[0] - d['a'] = 1 - d['b'] = 2 - # at this point, the changes to d are not yet synced, but by - # reassigning the dictionary, the proxy is notified of the change - lproxy[0] = d + Create a shared :class:`list` object and return a proxy for it. .. class:: Namespace @@ -1881,6 +1865,8 @@ >>> s = m.get_server() >>> s.serve_forever() +.. _multiprocessing-proxy_objects: + Proxy Objects ~~~~~~~~~~~~~ @@ -1890,8 +1876,7 @@ A proxy object has methods which invoke corresponding methods of its referent (although not every method of the referent will necessarily be available through -the proxy). A proxy can usually be used in most of the same ways that its -referent can: +the proxy). In this way, a proxy can be used just like its referent can: .. doctest:: @@ -1912,9 +1897,9 @@ the proxy. An important feature of proxy objects is that they are picklable so they can be -passed between processes. Note, however, that if a proxy is sent to the -corresponding manager's process then unpickling it will produce the referent -itself. This means, for example, that one shared object can contain a second: +passed between processes. As such, a referent can contain +:ref:`multiprocessing-proxy_objects`. This permits a nesting of these managed +lists, dicts, and other :ref:`multiprocessing-proxy_objects`: .. doctest:: @@ -1922,10 +1907,46 @@ >>> b = manager.list() >>> a.append(b) # referent of a now contains referent of b >>> print(a, b) - [[]] [] + [] [] >>> b.append('hello') - >>> print(a, b) - [['hello']] ['hello'] + >>> print(a[0], b) + ['hello'] ['hello'] + +Similarly, dict and list proxies may be nested inside one another:: + + >>> l_outer = manager.list([ manager.dict() for i in range(2) ]) + >>> d_first_inner = l_outer[0] + >>> d_first_inner['a'] = 1 + >>> d_first_inner['b'] = 2 + >>> l_outer[1]['c'] = 3 + >>> l_outer[1]['z'] = 26 + >>> print(l_outer[0]) + {'a': 1, 'b': 2} + >>> print(l_outer[1]) + {'c': 3, 'z': 26} + +If standard (non-proxy) :class:`list` or :class:`dict` objects are contained +in a referent, modifications to those mutable values will not be propagated +through the manager because the proxy has no way of knowing when the values +contained within are modified. However, storing a value in a container proxy +(which triggers a ``__setitem__`` on the proxy object) does propagate through +the manager and so to effectively modify such an item, one could re-assign the +modified value to the container proxy:: + + # create a list proxy and append a mutable object (a dictionary) + lproxy = manager.list() + lproxy.append({}) + # now mutate the dictionary + d = lproxy[0] + d['a'] = 1 + d['b'] = 2 + # at this point, the changes to d are not yet synced, but by + # updating the dictionary, the proxy is notified of the change + lproxy[0] = d + +This approach is perhaps less convenient than employing nested +:ref:`multiprocessing-proxy_objects` for most use cases but also +demonstrates a level of control over the synchronization. .. note:: diff -r 4aa6233961e5 Lib/multiprocessing/managers.py --- a/Lib/multiprocessing/managers.py Wed Sep 07 04:03:02 2016 +0300 +++ b/Lib/multiprocessing/managers.py Wed Sep 07 14:52:42 2016 -0500 @@ -142,7 +142,8 @@ self.id_to_obj = {'0': (None, ())} self.id_to_refcount = {} - self.mutex = threading.RLock() + self.id_to_local_proxy_obj = {} + self.mutex = threading.Lock() def serve_forever(self): ''' @@ -227,7 +228,14 @@ methodname = obj = None request = recv() ident, methodname, args, kwds = request - obj, exposed, gettypeid = id_to_obj[ident] + 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 as second_ke: + raise ke if methodname not in exposed: raise AttributeError( @@ -308,7 +316,7 @@ ''' with self.mutex: result = [] - keys = list(self.id_to_obj.keys()) + keys = list(self.id_to_refcount.keys()) keys.sort() for ident in keys: if ident != '0': @@ -321,7 +329,8 @@ ''' Number of shared objects ''' - return len(self.id_to_obj) - 1 # don't count ident='0' + # Doesn't use (len(self.id_to_obj) - 1) as we shouldn't count ident='0' + return len(self.id_to_refcount) def shutdown(self, c): ''' @@ -363,13 +372,9 @@ self.id_to_obj[ident] = (obj, set(exposed), method_to_typeid) if ident not in self.id_to_refcount: self.id_to_refcount[ident] = 0 - # increment the reference count immediately, to avoid - # this object being garbage collected before a Proxy - # object for it can be created. The caller of create() - # is responsible for doing a decref once the Proxy object - # has been created. - self.incref(c, ident) - return ident, tuple(exposed) + + self.incref(c, ident) + return ident, tuple(exposed) def get_methods(self, c, token): ''' @@ -387,15 +392,42 @@ def incref(self, c, ident): with self.mutex: - self.id_to_refcount[ident] += 1 + try: + self.id_to_refcount[ident] += 1 + except KeyError as ke: + # If no external references exist but an internal (to the + # manager) still does and a new external reference is created + # from it, restore the manager's tracking of it from the + # previously stashed internal ref. + if ident in self.id_to_local_proxy_obj: + self.id_to_refcount[ident] = 1 + self.id_to_obj[ident] = \ + self.id_to_local_proxy_obj[ident] + obj, exposed, gettypeid = self.id_to_obj[ident] + util.debug('Server re-enabled tracking & INCREF %r', ident) + else: + raise ke def decref(self, c, ident): + if ident not in self.id_to_refcount and \ + ident in self.id_to_local_proxy_obj: + util.debug('Server DECREF skipping %r', ident) + return + with self.mutex: assert self.id_to_refcount[ident] >= 1 self.id_to_refcount[ident] -= 1 if self.id_to_refcount[ident] == 0: - del self.id_to_obj[ident], self.id_to_refcount[ident] - util.debug('disposing of obj with id %r', ident) + del self.id_to_refcount[ident] + + if ident not in self.id_to_refcount: + # Two-step process in case the object turns out to contain other + # proxy objects (e.g. a managed list of managed lists). + self.id_to_obj[ident] = (None, (), None) # thread-safe + util.debug('disposing of obj with id %r', ident) + with self.mutex: + del self.id_to_obj[ident] + # # Class to represent state of a manager @@ -658,7 +690,7 @@ _mutex = util.ForkAwareThreadLock() def __init__(self, token, serializer, manager=None, - authkey=None, exposed=None, incref=True): + authkey=None, exposed=None, incref=True, manager_owned=False): with BaseProxy._mutex: tls_idset = BaseProxy._address_to_local.get(token.address, None) if tls_idset is None: @@ -680,6 +712,12 @@ self._serializer = serializer self._Client = listener_client[serializer][1] + # Should be set to True only when a proxy object is being created + # on the manager server; primary use case: nested proxy objects. + # RebuildProxy detects when a proxy is being created on the manager + # and sets this value appropriately. + self._owned_by_manager = manager_owned + if authkey is not None: self._authkey = process.AuthenticationString(authkey) elif self._manager is not None: @@ -738,6 +776,10 @@ return self._callmethod('#GETVALUE') def _incref(self): + if self._owned_by_manager: + util.debug('owned_by_manager skipped INCREF of %r', self._token.id) + return + conn = self._Client(self._token.address, authkey=self._authkey) dispatch(conn, None, 'incref', (self._id,)) util.debug('INCREF %r', self._token.id) @@ -822,19 +864,19 @@ def RebuildProxy(func, token, serializer, kwds): ''' Function used for unpickling proxy objects. - - If possible the shared object is returned, or otherwise a proxy for it. ''' server = getattr(process.current_process(), '_manager_server', None) - if server and server.address == token.address: - return server.id_to_obj[token.id][0] - else: - incref = ( - kwds.pop('incref', True) and - not getattr(process.current_process(), '_inheriting', False) - ) - return func(token, serializer, incref=incref, **kwds) + util.debug('Rebuild a proxy owned by manager, token=%r', token) + kwds['manager_owned'] = True + if token.id not in server.id_to_local_proxy_obj: + server.id_to_local_proxy_obj[token.id] = \ + server.id_to_obj[token.id] + incref = ( + kwds.pop('incref', True) and + not getattr(process.current_process(), '_inheriting', False) + ) + return func(token, serializer, incref=incref, **kwds) # # Functions to create proxies and proxy types diff -r 4aa6233961e5 Lib/test/_test_multiprocessing.py --- a/Lib/test/_test_multiprocessing.py Wed Sep 07 04:03:02 2016 +0300 +++ b/Lib/test/_test_multiprocessing.py Wed Sep 07 14:52:42 2016 -0500 @@ -1628,13 +1628,33 @@ d = [a, b] e = self.list(d) self.assertEqual( - e[:], + [element[:] for element in e], [[0, 1, 2, 3, 4, 5, 6, 7, 8, 9], [0, 1, 2, 3, 4, 0, 1, 2, 3, 4]] ) f = self.list([a]) a.append('hello') - self.assertEqual(f[:], [[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 'hello']]) + self.assertEqual(f[0][:], [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 'hello']) + + def test_list_proxy_in_list(self): + a = self.list([self.list(range(3)) for _i in range(3)]) + self.assertEqual([inner[:] for inner in a], [[0, 1, 2]] * 3) + + a[0][-1] = 55 + self.assertEqual(a[0][:], [0, 1, 55]) + for i in range(1, 3): + self.assertEqual(a[i][:], [0, 1, 2]) + + self.assertEqual(a[1].pop(), 2) + self.assertEqual(len(a[1]), 2) + for i in range(0, 3, 2): + self.assertEqual(len(a[i]), 3) + + del a + + b = self.list() + b.append(b) + del b def test_dict(self): d = self.dict() @@ -1646,6 +1666,47 @@ self.assertEqual(sorted(d.values()), [chr(i) for i in indices]) self.assertEqual(sorted(d.items()), [(i, chr(i)) for i in indices]) + def test_dict_proxy_nested(self): + pets = self.dict(ferrets=2, hamsters=4) + supplies = self.dict(water=10, feed=3) + d = self.dict(pets=pets, supplies=supplies) + + self.assertEqual(supplies['water'], 10) + self.assertEqual(d['supplies']['water'], 10) + + d['supplies']['blankets'] = 5 + self.assertEqual(supplies['blankets'], 5) + self.assertEqual(d['supplies']['blankets'], 5) + + d['supplies']['water'] = 7 + self.assertEqual(supplies['water'], 7) + self.assertEqual(d['supplies']['water'], 7) + + del pets + del supplies + self.assertEqual(d['pets']['ferrets'], 2) + d['supplies']['blankets'] = 11 + self.assertEqual(d['supplies']['blankets'], 11) + + pets = d['pets'] + supplies = d['supplies'] + supplies['water'] = 7 + self.assertEqual(supplies['water'], 7) + self.assertEqual(d['supplies']['water'], 7) + + l = self.list([pets, supplies]) + l[0]['marmots'] = 1 + self.assertEqual(pets['marmots'], 1) + self.assertEqual(l[0]['marmots'], 1) + + del pets + del supplies + self.assertEqual(l[0]['marmots'], 1) + + outer = self.list([[88, 99], l]) + self.assertIsInstance(outer[0], list) # Not a ListProxy + self.assertEqual(outer[-1][-1]['feed'], 3) + def test_namespace(self): n = self.Namespace() n.name = 'Bob' @@ -3856,6 +3917,7 @@ time.sleep(t) t *= 2 gc.collect() # do garbage collection + t = 0.01 if cls.manager._number_of_objects() != 0: # This is not really an error since some tests do not # ensure that all processes which hold a reference to a