Comment
Author: Admin | 2025-04-27
Self def next(self): return self._callmethod('next') def __next__(self): return self._callmethod('__next__')# Function to return the operator moduledef get_operator_module(): return operator##class MyManager(BaseManager): pass# register the Foo class; make `f()` and `g()` accessible via proxyMyManager.register('Foo1', Foo)# register the Foo class; make `g()` and `_h()` accessible via proxyMyManager.register('Foo2', Foo, exposed=('g', '_h'))# register the generator function baz; use `GeneratorProxy` to make proxiesMyManager.register('baz', baz, proxytype=GeneratorProxy)# register get_operator_module(); make public functions accessible via proxyMyManager.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, 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.getslice(range(10), 2, 6) =', op.getslice(range(10), 2, 6) print 'op.repeat(range(5), 3) =', op.repeat(range(5), 3) print 'op._exposed_ =', op._exposed_##if __name__ == '__main__': freeze_support() test()Using Pool:## A test of `multiprocessing.Pool` class## Copyright (c) 2006-2008, R Oudkerk# All rights reserved.#import multiprocessingimport timeimport randomimport sys## Functions used by test code#def calculate(func, args): result = func(*args) return '%s says that %s%s = %s' % ( multiprocessing.current_process().name, func.__name__, args, result )def calculatestar(args): return calculate(*args)def mul(a, b): time.sleep(0.5*random.random()) return a * bdef plus(a, b): time.sleep(0.5*random.random()) return a + bdef f(x): return 1.0 / (x-5.0)def pow3(x): return x**3def noop(x): pass## Test code#def test(): print 'cpu_count() = %d\n' % multiprocessing.cpu_count() # # Create pool # PROCESSES = 4 print 'Creating pool with %d processes\n' % PROCESSES pool = multiprocessing.Pool(PROCESSES) print 'pool = %s' % pool print # # Tests # TASKS = [(mul, (i, 7)) for i in range(10)] + \ [(plus, (i, 8)) for i in range(10)] results = [pool.apply_async(calculate, t) for t in TASKS] imap_it = pool.imap(calculatestar, TASKS) imap_unordered_it = pool.imap_unordered(calculatestar, TASKS) print 'Ordered results using pool.apply_async():' for r in results: print '\t', r.get() print print 'Ordered results using pool.imap():' for x in imap_it: print '\t', x print print 'Unordered results using pool.imap_unordered():' for x in imap_unordered_it: print '\t', x print print 'Ordered results
Add Comment