1 import time 2 import random 3 4 from multiprocessing import Process, Queue, current_process, freeze_support 5 6 # 7 # Function run by worker processes 8 # 9 10 def worker(input, output): 11 for func, args in iter(input.get, 'STOP'): 12 result = calculate(func, args) 13 output.put(result) 14 15 # 16 # Function used to calculate result 17 # 18 19 def calculate(func, args): 20 result = func(*args) 21 return '%s says that %s%s = %s' % \ 22 (current_process().name, func.__name__, args, result) 23 24 # 25 # Functions referenced by tasks 26 # 27 28 def mul(a, b): 29 time.sleep(0.5*random.random()) 30 return a * b 31 32 def plus(a, b): 33 time.sleep(0.5*random.random()) 34 return a + b 35 36 # 37 # 38 # 39 40 def test(): 41 NUMBER_OF_PROCESSES = 4 42 TASKS1 = [(mul, (i, 7)) for i in range(20)] 43 TASKS2 = [(plus, (i, 8)) for i in range(10)] 44 45 # Create queues 46 task_queue = Queue() 47 done_queue = Queue() 48 49 # Submit tasks 50 for task in TASKS1: 51 task_queue.put(task) 52 53 # Start worker processes 54 for i in range(NUMBER_OF_PROCESSES): 55 Process(target=worker, args=(task_queue, done_queue)).start() 56 57 # Get and print results 58 print('Unordered results:') 59 for i in range(len(TASKS1)): 60 print('\t', done_queue.get()) 61 62 # Add more tasks using `put()` 63 for task in TASKS2: 64 task_queue.put(task) 65 66 # Get and print some more results 67 for i in range(len(TASKS2)): 68 print('\t', done_queue.get()) 69 70 # Tell child processes to stop 71 for i in range(NUMBER_OF_PROCESSES): 72 task_queue.put('STOP') 73 74 75 if __name__ == '__main__': 76 freeze_support() 77 test() 78
