During the sprints at PyCon 2008 we were able to take some time and head downtown. Jannis Leidel was wanting to buy an iPhone which took us to the Apple Store in downtown Chicago. Like any group of Python programmers we ended up running the Django test suite on a 3.2 GHz Dual-Proc Quad-Core Mac Pro. We weren't too impressed with its results as it was only able to utilize one core at a time. Eric Florenzano and I began brainstorming how this can be improved.
Well, after doing a little bit of brainstorming myself I have written up some code that allows the Django test suite to use as many cores as possible. I am not going to say I am expert with this kind of programming, but it really intriques me. The code currently has a dependancy on the absolutely sweet Python module called processing which implements a similar interface to the built-in threading module. The difference between the two is that the former uses processes and the latter is threads. This allows the code to run as fast as the machine can handle. It is a nice workaround to the GIL when the overhead of the spawned processes is not a killer. Here is the code:
def run_tests_faster(test_labels, verbosity=1, interactive=True, extra_tests=[]):
from Queue import Empty
from processing import Queue, cpuCount, enableLogging
from processing import currentProcess, Process, freezeSupport
#enableLogging()
# the number of processes to run in parallel.
num_processes = cpuCount()
def process_print(msg):
print "%s: %s" % (currentProcess().getName(), msg)
def worker(old_name, verbosity, interactive, ready_queue, app_queue, done_queue):
process_print("setting up environment")
setup_test_environment()
create_test_db(verbosity, autoclobber=not interactive)
ready_queue.put(1)
for app_label in iter(app_queue.get, "STOP"):
process_print(app_label)
suite = build_suite(get_app(app_label))
result = unittest.TestResult()
suite.run(result)
done_queue.put((app_label, len(result.failures), len(result.errors)))
# done_queue.put((app_label, 0, 0)) # FOR TESTING
process_print("tearing down environment")
destroy_test_db(old_name, verbosity)
teardown_test_environment()
ready_queue = Queue()
app_queue = Queue()
done_queue = Queue()
apps = []
for installed_app in settings.INSTALLED_APPS:
apps.append(installed_app.split(".")[-1])
for i in range(num_processes):
Process(target=worker, args=[settings.DATABASE_NAME, verbosity, interactive, ready_queue, app_queue, done_queue]).start()
ready_count = 0
done_apps = []
total_failures = 0
total_errors = 0
while ready_count != num_processes:
ready_queue.get()
ready_count += 1
print "start timer"
start_time = time.time()
app_queue.putmany(apps)
while len(apps) != len(done_apps):
app_label, failures, errors = done_queue.get()
total_failures += failures
total_errors += errors
done_apps.append(app_label)
stop_time = time.time()
print "timer ended"
time_taken = stop_time - start_time
for i in range(num_processes):
app_queue.put("STOP")
print "time: %.3fs, failures: %d, errors: %d" % (time_taken, total_failures, total_errors)
# give the workers some time to finish
time.sleep(2)
return 0
There are a few things I would like to optimize and clean up. Ultimately, this can become a nice thing to submit to djangosnippets.org once more finalized. I wanted to get this up so that others can look at it and critique me and make this even better. I have also pasted this to dpaste for syntax highlighting and a plain text version.
Brian,
Out of interest, do you have some simple comparisons as to what sort of impact the changes would have ?
Al.
Posted by Al on Mar 26, 2008 at 6:16 AM