mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-09-26 18:29:57 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 10

bpo-30256: Add manager_owned keyword arg to AutoProxy (GH-16341)

Browse source 
Co-authored-by: Jordan Speicher <jordan@jspeicher.com>
This commit is contained in:
finefoot 2021-07-02 05:45:02 +02:00 • committed by GitHub
parent a7e251b07a
commit 85b920498b
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 15 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

4
Lib/multiprocessing/managers.py
View file

@ -967,7 +967,7 @@ def MakeProxyType(name, exposed, _cache={}):
def AutoProxy(token, serializer, manager=None, authkey=None, def AutoProxy(token, serializer, manager=None, authkey=None,
exposed=None, incref=True): exposed=None, incref=True, manager_owned=False):
''' '''
Return an auto-proxy for `token` Return an auto-proxy for `token`
''' '''
@ -987,7 +987,7 @@ def AutoProxy(token, serializer, manager=None, authkey=None,
ProxyType = MakeProxyType('AutoProxy[%s]' % token.typeid, exposed) ProxyType = MakeProxyType('AutoProxy[%s]' % token.typeid, exposed)
proxy = ProxyType(token, serializer, manager=manager, authkey=authkey, proxy = ProxyType(token, serializer, manager=manager, authkey=authkey,
incref=incref) incref=incref, manager_owned=manager_owned)
proxy._isauto = True proxy._isauto = True
return proxy return proxy

10
Lib/test/_test_multiprocessing.py
View file

@ -2286,6 +2286,16 @@ class _TestContainers(BaseTestCase):
self.assertIsInstance(outer[0], list) # Not a ListProxy self.assertIsInstance(outer[0], list) # Not a ListProxy
self.assertEqual(outer[-1][-1]['feed'], 3) self.assertEqual(outer[-1][-1]['feed'], 3)
def test_nested_queue(self):
a = self.list() # Test queue inside list
a.append(self.Queue())
a[0].put(123)
self.assertEqual(a[0].get(), 123)
b = self.dict() # Test queue inside dict
b[0] = self.Queue()
b[0].put(456)
self.assertEqual(b[0].get(), 456)
def test_namespace(self): def test_namespace(self):
n = self.Namespace() n = self.Namespace()
n.name = 'Bob' n.name = 'Bob'

1
Misc/NEWS.d/next/Library/2019-09-25-13-54-41.bpo-30256.wBkzox.rst Normal file
View file

@ -0,0 +1 @@
Pass multiprocessing BaseProxy argument `manager_owned` through AutoProxy

2
Misc/NEWS.d/next/Tests/2019-09-25-18-10-10.bpo-30256.A5i76Q.rst Normal file
View file

@ -0,0 +1,2 @@
Add test for nested queues when using ``multiprocessing`` shared objects
``AutoProxy[Queue]`` inside ``ListProxy`` and ``DictProxy``