1 """Easy to use object-oriented thread pool framework.
3 A thread pool is an object that maintains a pool of worker threads to perform
4 time consuming operations in parallel. It assigns jobs to the threads
5 by putting them in a work request queue, where they are picked up by the
6 next available thread. This then performs the requested operation in the
7 background and puts the results in a another queue.
9 The thread pool object can then collect the results from all threads from
10 this queue as soon as they become available or after all threads have
11 finished their work. It's also possible, to define callbacks to handle
12 each result as it comes in.
14 The basic concept and some code was taken from the book "Python in a Nutshell"
15 by Alex Martelli, copyright 2003, ISBN 0-596-00188-6, from section 14.5
16 "Threaded Program Architecture". I wrapped the main program logic in the
17 ThreadPool class, added the WorkRequest class and the callback system and
18 tweaked the code here and there. Kudos also to Florent Aide for the exception
23 >>> pool = TreadPool(poolsize)
24 >>> requests = makeRequests(some_callable, list_of_args, callback)
25 >>> [pool.putRequest(req) for req in requests]
28 See the end of the module code for a brief, annotated usage example.
30 Website : http://chrisarndt.de/en/software/python/threadpool/
42 __author__ = "Christopher Arndt"
44 __revision__ = "$Revision: 1.5 $"
45 __date__ = "$Date: 2006/06/23 12:32:25 $"
46 __license__ = 'Python license'
48 # standard library modules
54 class NoResultsPending(Exception):
55 """All work requests have been processed."""
58 class NoWorkersAvailable(Exception):
59 """No worker threads available to process remaining requests."""
63 class WorkerThread(threading.Thread):
64 """Background thread connected to the requests/results queues.
66 A worker thread sits in the background and picks up work requests from
67 one queue and puts the results in another until it is dismissed.
70 def __init__(self, requestsQueue, resultsQueue, **kwds):
71 """Set up thread in daemonic mode and start it immediatedly.
73 requestsQueue and resultQueue are instances of Queue.Queue passed
74 by the ThreadPool class when it creates a new worker thread.
77 threading.Thread.__init__(self, **kwds)
79 self.workRequestQueue = requestsQueue
80 self.resultQueue = resultsQueue
81 self._dismissed = threading.Event()
85 """Repeatedly process the job queue until told to exit."""
87 while not self._dismissed.isSet():
88 # thread blocks here, if queue empty
89 request = self.workRequestQueue.get()
90 if self._dismissed.isSet():
91 # if told to exit, return the work request we just picked up
92 self.workRequestQueue.put(request)
96 (request, request.callable(*request.args, **request.kwds))
99 request.exception = True
100 self.resultQueue.put((request, sys.exc_info()))
103 """Sets a flag to tell the thread to exit when done with current job.
106 self._dismissed.set()
110 """A request to execute a callable for putting in the request queue later.
112 See the module function makeRequests() for the common case
113 where you want to build several WorkRequests for the same callable
114 but with different arguments for each call.
117 def __init__(self, callable, args=None, kwds=None, requestID=None,
118 callback=None, exc_callback=None):
119 """Create a work request for a callable and attach callbacks.
121 A work request consists of the a callable to be executed by a
122 worker thread, a list of positional arguments, a dictionary
123 of keyword arguments.
125 A callback function can be specified, that is called when the results
126 of the request are picked up from the result queue. It must accept
127 two arguments, the request object and the results of the callable,
128 in that order. If you want to pass additional information to the
129 callback, just stick it on the request object.
131 You can also give a callback for when an exception occurs. It should
132 also accept two arguments, the work request and a tuple with the
133 exception details as returned by sys.exc_info().
135 requestID, if given, must be hashable since it is used by the
136 ThreadPool object to store the results of that work request in a
137 dictionary. It defaults to the return value of id(self).
140 if requestID is None:
141 self.requestID = id(self)
146 raise TypeError("requestID must be hashable.")
147 self.requestID = requestID
148 self.exception = False
149 self.callback = callback
150 self.exc_callback = exc_callback
151 self.callable = callable
152 self.args = args or []
153 self.kwds = kwds or {}
157 """A thread pool, distributing work requests and collecting results.
159 See the module doctring for more information.
162 def __init__(self, num_workers, q_size=0):
163 """Set up the thread pool and start num_workers worker threads.
165 num_workers is the number of worker threads to start initialy.
166 If q_size > 0 the size of the work request queue is limited and
167 the thread pool blocks when the queue is full and it tries to put
168 more work requests in it (see putRequest method).
171 self.requestsQueue = Queue.Queue(q_size)
172 self.resultsQueue = Queue.Queue()
174 self.workRequests = {}
175 self.createWorkers(num_workers)
177 def createWorkers(self, num_workers):
178 """Add num_workers worker threads to the pool."""
180 for i in range(num_workers):
181 self.workers.append(WorkerThread(self.requestsQueue,
184 def dismissWorkers(self, num_workers):
185 """Tell num_workers worker threads to quit after their current task.
188 for i in range(min(num_workers, len(self.workers))):
189 worker = self.workers.pop()
192 def putRequest(self, request, block=True, timeout=0):
193 """Put work request into work queue and save its id for later."""
195 assert isinstance(request, WorkRequest)
196 self.requestsQueue.put(request, block, timeout)
197 self.workRequests[request.requestID] = request
199 def poll(self, block=False):
200 """Process any new results in the queue."""
203 # still results pending?
204 if not self.workRequests:
205 raise NoResultsPending
206 # are there still workers to process remaining requests?
207 elif block and not self.workers:
208 raise NoWorkersAvailable
210 # get back next results
211 request, result = self.resultsQueue.get(block=block)
212 # has an exception occured?
213 if request.exception and request.exc_callback:
214 request.exc_callback(request, result)
215 # hand results to callback, if any
216 if request.callback and not \
217 (request.exception and request.exc_callback):
218 request.callback(request, result)
219 del self.workRequests[request.requestID]
225 """Wait for results, blocking until all have arrived."""
230 except NoResultsPending:
234 def makeRequests(callable, args_list, callback=None, exc_callback=None):
235 """Create several work requests for same callable with different arguments.
237 Convenience function for creating several work requests for the same
238 callable where each invocation of the callable receives different values
241 args_list contains the parameters for each invocation of callable.
242 Each item in 'args_list' should be either a 2-item tuple of the list of
243 positional arguments and a dictionary of keyword arguments or a single,
246 See docstring for WorkRequest for info on callback and exc_callback.
250 for item in args_list:
251 if isinstance(item, tuple):
253 WorkRequest(callable, item[0], item[1], callback=callback,
254 exc_callback=exc_callback)
258 WorkRequest(callable, [item], None, callback=callback,
259 exc_callback=exc_callback)
267 if __name__ == '__main__':
271 # the work the threads will have to do (rather trivial in our example)
272 def do_something(data):
273 time.sleep(random.randint(1,5))
274 result = round(random.random() * data, 5)
275 # just to show off, we throw an exception once in a while
277 raise RuntimeError("Something extraordinary happened!")
280 # this will be called each time a result is available
281 def print_result(request, result):
282 print "**Result: %s from request #%s" % (result, request.requestID)
284 # this will be called when an exception occurs within a thread
285 def handle_exception(request, exc_info):
286 print "Exception occured in request #%s: %s" % \
287 (request.requestID, exc_info[1])
289 # assemble the arguments for each job to a list...
290 data = [random.randint(1,10) for i in range(20)]
291 # ... and build a WorkRequest object for each item in data
292 requests = makeRequests(do_something, data, print_result, handle_exception)
294 # or the other form of args_lists accepted by makeRequests: ((,), {})
295 data = [((random.randint(1,10),), {}) for i in range(20)]
297 makeRequests(do_something, data, print_result, handle_exception)
300 # we create a pool of 3 worker threads
303 # then we put the work requests in the queue...
306 print "Work request #%s added." % req.requestID
308 # [main.putRequest(req) for req in requests]
310 # ...and wait for the results to arrive in the result queue
311 # by using ThreadPool.wait(). This would block until results for
312 # all work requests have arrived:
315 # instead we can poll for results while doing something else:
320 print "Main thread working..."
323 print "Adding 3 more worker threads..."
324 main.createWorkers(3)
326 except KeyboardInterrupt:
329 except NoResultsPending:
330 print "All results collected."