| from multiprocessing import freeze_support |
| from multiprocessing.managers import BaseManager, BaseProxy |
| import operator |
| |
| ## |
| |
| class Foo: |
| def f(self): |
| print('you called Foo.f()') |
| def g(self): |
| print('you called Foo.g()') |
| def _h(self): |
| print('you called Foo._h()') |
| |
| # A simple generator function |
| def baz(): |
| for i in range(10): |
| yield i*i |
| |
| # Proxy type for generator objects |
| class GeneratorProxy(BaseProxy): |
| _exposed_ = ['__next__'] |
| def __iter__(self): |
| return self |
| def __next__(self): |
| return self._callmethod('__next__') |
| |
| # Function to return the operator module |
| def get_operator_module(): |
| return operator |
| |
| ## |
| |
| class MyManager(BaseManager): |
| pass |
| |
| # register the Foo class; make `f()` and `g()` accessible via proxy |
| MyManager.register('Foo1', Foo) |
| |
| # register the Foo class; make `g()` and `_h()` accessible via proxy |
| MyManager.register('Foo2', Foo, exposed=('g', '_h')) |
| |
| # register the generator function baz; use `GeneratorProxy` to make proxies |
| MyManager.register('baz', baz, proxytype=GeneratorProxy) |
| |
| # register get_operator_module(); make public functions accessible via proxy |
| MyManager.register('operator', get_operator_module) |
| |
| ## |
| |
| def test(): |
| manager = MyManager() |
| manager.start() |
| |
| print('-' * 20) |
| |
| f1 = manager.Foo1() |
| f1.f() |
| f1.g() |
| assert not hasattr(f1, '_h') |
| assert sorted(f1._exposed_) == sorted(['f', 'g']) |
| |
| print('-' * 20) |
| |
| f2 = manager.Foo2() |
| f2.g() |
| f2._h() |
| assert not hasattr(f2, 'f') |
| assert sorted(f2._exposed_) == sorted(['g', '_h']) |
| |
| print('-' * 20) |
| |
| it = manager.baz() |
| for i in it: |
| print('<%d>' % i, end=' ') |
| print() |
| |
| print('-' * 20) |
| |
| op = manager.operator() |
| print('op.add(23, 45) =', op.add(23, 45)) |
| print('op.pow(2, 94) =', op.pow(2, 94)) |
| print('op._exposed_ =', op._exposed_) |
| |
| ## |
| |
| if __name__ == '__main__': |
| freeze_support() |
| test() |