Benchmark single-process, threads, processes in python procedure

multi_bench.py

import time
import random
import itertools

from fastset.bitvector import bitvector
from Queue import Queue
from threading import Thread

from multiprocessing import Process


class Worker(Thread):
    """Thread executing tasks from a given tasks queue"""
    def __init__(self, tasks):
        Thread.__init__(self)
        self.tasks = tasks
        self.daemon = True
        self.start()

    def run(self):
        while True:
            func, args, kargs = self.tasks.get()
            try:
                func(*args, **kargs)
            except Exception, e:
                print e
            self.tasks.task_done()


class ThreadPool:
    """Pool of threads consuming tasks from a queue"""
    def __init__(self, num_threads):
        self.tasks = Queue(num_threads)
        for _ in range(num_threads):
            Worker(self.tasks)

    def add_task(self, func, *args, **kargs):
        """Add a task to the queue"""
        self.tasks.put((func, args, kargs))

    def wait_completion(self):
        """Wait for completion of all the tasks in the queue"""
        self.tasks.join()


def time_it(function):
    """
    Decorator whichs times a function execution.
    """
    def wrap(*arg):
        start = time.time()
        r = function(*arg)
        end = time.time()
        print "%s (%0.3f ms)" % (function.func_name, (end - start) * 1000)
        return r
    return wrap


def union(index, bitvectors, res):
    res.update({index: reduce(lambda x, y: x & y, bitvectors)})


@time_it
def single_computing(res, bitvectors):
    for counter, b in enumerate(bitvectors):
        union(counter, b, res)


@time_it
def threads(res, bitvectors, pool):
    for counter, b in enumerate(bitvectors):
        pool.add_task(union, str(counter), b, res)

    pool.wait_completion()


@time_it
def parralel_processing(res, bitvectors):
    processes = []
    for counter, b in enumerate(bitvectors):
        p = Process(target=union, args=(str(counter), b, res))
        processes.append(p)

    [p.start() for p in processes]
    [p.join() for p in processes]


if __name__ == '__main__':
    pool = ThreadPool(10)
    res = {}

    global_bitvectors = []
    i = 0

    print "Generating bitvectors"
    while i < 10:
        process_bitvectors = [bitvector()] * 300
        for b in process_bitvectors:
            for x in xrange(150000):
                b[random.randint(0, 300000)] = 1
        global_bitvectors.append(process_bitvectors)
        i += 1

    print "--- SINGLE ---"
    print "Computing"
    for _ in itertools.repeat(None, 5):
        single_computing(res, global_bitvectors)
    print "--- DONE ---\n"

    print "--- THREADS ---"
    print "Computing"
    for _ in itertools.repeat(None, 5):
        threads(res, global_bitvectors, pool)
    print "--- DONE ---"

    print "--- MULTI-PROCESS ---"
    print "Computing"
    for _ in itertools.repeat(None, 5):
        parralel_processing(res, global_bitvectors)
    print "--- DONE ---"