Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
branch: master
Fetching contributors…

Cannot retrieve contributors at this time

79 lines (65 sloc) 2.578 kb
#
# Copyright 2008 Google Inc.
# Released under the GPLv2
import threading, Queue
class ThreadPool:
""" A generic threading class for use in the CLI
ThreadPool class takes the function to be executed as an argument and
optionally number of threads. It then creates multiple threads for
faster execution. """
def __init__(self, function, numthreads=40):
assert(numthreads > 0)
self.threads = Queue.Queue(0)
self.function = function
self.numthreads = 0
self.queue = Queue.Queue(0)
self._start_threads(numthreads)
def wait(self):
""" Checks to see if any threads are still working and
blocks until worker threads all complete. """
for x in xrange(self.numthreads):
self.queue.put('die')
# As only spawned threads are allowed to add new ones,
# we can safely wait for the thread queue to be empty
# (if we're at the last thread and it creates a new one,
# it will get queued before it finishes).
dead = 0
while True:
try:
thread = self.threads.get(block=True, timeout=1)
if thread.isAlive():
thread.join()
dead += 1
except Queue.Empty:
assert(dead == self.numthreads)
return
def queue_work(self, data):
""" Takes a list of items and appends them to the
work queue. """
[self.queue.put(item) for item in data]
def add_one_thread_post_wait(self):
# Only a spawned thread (not the main one)
# should call this (see wait() for details)
self._start_threads(1)
self.queue.put('die')
def _start_threads(self, nthreads):
""" Start up threads to spawn workers. """
self.numthreads += nthreads
for i in xrange(nthreads):
thread = threading.Thread(target=self._new_worker)
thread.setDaemon(True)
self.threads.put(thread)
thread.start()
def _new_worker(self):
""" Spawned worker threads. These threads loop until queue is empty."""
while True:
# Blocking call
data = self.queue.get()
if data == 'die':
return
try:
self.function(data)
except Exception, full_error:
# Put a catch all here.
print ('Unexpected failure in the thread calling %s: %s' %
(self.function.__name__, full_error))
Jump to Line
Something went wrong with that request. Please try again.