Python multiprocessing.managers 模块,RemoteError() 实例源码
我们从Python开源项目中,提取了以下10个代码示例,用于说明如何使用multiprocessing.managers.RemoteError()。
def connect(self):
try:
self._manager.connect()
self._data = self._manager.worker_dict()
except (FileNotFoundError, AttributeError, RemoteError, ConnectionRefusedError, brokenPipeError):
print('Unable to connect to manager for worker data.')
self._data = {}
def data(self):
try:
if self._data:
return self._data.items()
else:
raise ValueError
except (FileNotFoundError, ValueError, brokenPipeError):
self.connect()
return self._data.items()
def test_mymanager(self):
manager = MyManager()
manager.start()
foo = manager.Foo()
bar = manager.Bar()
baz = manager.baz()
foo_methods = [name for name in ('f', 'g', '_h') if hasattr(foo, name)]
bar_methods = [name for name in ('f', '_h') if hasattr(bar, name)]
self.assertEqual(foo_methods, ['f', 'g'])
self.assertEqual(bar_methods, '_h'])
self.assertEqual(foo.f(), 'f()')
self.assertRaises(ValueError, foo.g)
self.assertEqual(foo._callmethod('f'), 'f()')
self.assertRaises(RemoteError, foo._callmethod, '_h')
self.assertEqual(bar.f(), 'f()')
self.assertEqual(bar._h(), '_h()')
self.assertEqual(bar._callmethod('f'), 'f()')
self.assertEqual(bar._callmethod('_h'), '_h()')
self.assertEqual(list(baz), [i*i for i in range(10)])
manager.shutdown()
#
# Test of connecting to a Remote Server and using xmlrpclib for serialization
#
def test_mymanager(self):
manager = MyManager()
manager.start()
foo = manager.Foo()
bar = manager.Bar()
baz = manager.baz()
foo_methods = [name for name in ('f', [i*i for i in range(10)])
manager.shutdown()
#
# Test of connecting to a Remote Server and using xmlrpclib for serialization
#
def test_mymanager(self):
manager = MyManager()
manager.start()
foo = manager.Foo()
bar = manager.Bar()
baz = manager.baz()
foo_methods = [name for name in ('f', [i*i for i in range(10)])
manager.shutdown()
#
# Test of connecting to a Remote Server and using xmlrpclib for serialization
#
def common(self, manager):
foo = manager.Foo()
bar = manager.Bar()
baz = manager.baz()
foo_methods = [name for name in ('f', [i*i for i in range(10)])
#
# Test of connecting to a Remote Server and using xmlrpclib for serialization
#
def test_mymanager(self):
manager = MyManager()
manager.start()
foo = manager.Foo()
bar = manager.Bar()
baz = manager.baz()
foo_methods = [name for name in ('f', [i*i for i in range(10)])
manager.shutdown()
#
# Test of connecting to a Remote Server and using xmlrpclib for serialization
#
def common(self, [i*i for i in range(10)])
#
# Test of connecting to a Remote Server and using xmlrpclib for serialization
#
def test_mymanager(self):
manager = MyManager()
manager.start()
foo = manager.Foo()
bar = manager.Bar()
baz = manager.baz()
foo_methods = [name for name in ('f', [i*i for i in range(10)])
manager.shutdown()
#
# Test of connecting to a Remote Server and using xmlrpclib for serialization
#
def common(self, [i*i for i in range(10)])
#
# Test of connecting to a Remote Server and using xmlrpclib for serialization
#