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
#
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。