source: python/trunk/Doc/includes/mp_workers.py

Last change on this file was 2, checked in by Yuri Dario, 15 years ago

Initial import for vendor code.

  • Property svn:eol-style set to native
File size: 2.0 KB
Line 
1#
2# Simple example which uses a pool of workers to carry out some tasks.
3#
4# Notice that the results will probably not come out of the output
5# queue in the same in the same order as the corresponding tasks were
6# put on the input queue. If it is important to get the results back
7# in the original order then consider using `Pool.map()` or
8# `Pool.imap()` (which will save on the amount of code needed anyway).
9#
10# Copyright (c) 2006-2008, R Oudkerk
11# All rights reserved.
12#
13
14import time
15import random
16
17from multiprocessing import Process, Queue, current_process, freeze_support
18
19#
20# Function run by worker processes
21#
22
23def worker(input, output):
24 for func, args in iter(input.get, 'STOP'):
25 result = calculate(func, args)
26 output.put(result)
27
28#
29# Function used to calculate result
30#
31
32def calculate(func, args):
33 result = func(*args)
34 return '%s says that %s%s = %s' % \
35 (current_process().name, func.__name__, args, result)
36
37#
38# Functions referenced by tasks
39#
40
41def mul(a, b):
42 time.sleep(0.5*random.random())
43 return a * b
44
45def plus(a, b):
46 time.sleep(0.5*random.random())
47 return a + b
48
49#
50#
51#
52
53def test():
54 NUMBER_OF_PROCESSES = 4
55 TASKS1 = [(mul, (i, 7)) for i in range(20)]
56 TASKS2 = [(plus, (i, 8)) for i in range(10)]
57
58 # Create queues
59 task_queue = Queue()
60 done_queue = Queue()
61
62 # Submit tasks
63 for task in TASKS1:
64 task_queue.put(task)
65
66 # Start worker processes
67 for i in range(NUMBER_OF_PROCESSES):
68 Process(target=worker, args=(task_queue, done_queue)).start()
69
70 # Get and print results
71 print 'Unordered results:'
72 for i in range(len(TASKS1)):
73 print '\t', done_queue.get()
74
75 # Add more tasks using `put()`
76 for task in TASKS2:
77 task_queue.put(task)
78
79 # Get and print some more results
80 for i in range(len(TASKS2)):
81 print '\t', done_queue.get()
82
83 # Tell child processes to stop
84 for i in range(NUMBER_OF_PROCESSES):
85 task_queue.put('STOP')
86
87
88if __name__ == '__main__':
89 freeze_support()
90 test()
Note: See TracBrowser for help on using the repository browser.